Added HEFT scheduler #35
Conversation
|
@saurav-c, the tests on the Travis build are failing. |
|
Forgot to change policy in the tests. |
| arg_list = schedule.arguments[fname] | ||
| arg_list.values.extend(args) | ||
|
|
||
| # for fname in dag.functions: |
There was a problem hiding this comment.
Why is all this code commented out?
| self.unique_executors.add(max_ip) | ||
| return max_ip | ||
|
|
||
| def pin_function(self, dag_name, function_name): |
There was a problem hiding this comment.
Is this any different from the pin_function method in the default scheduler? If not, it might make sense to move this into the base scheduler to avoid being repetitive.
|
|
||
| continue | ||
|
|
||
| def commit_dag(self, dag_name): |
There was a problem hiding this comment.
Same with the rest of these functions...
| @@ -29,8 +29,8 @@ | |||
| create_function, | |||
| delete_dag | |||
| ) | |||
| from cloudburst.server.scheduler.policy.default_policy import ( | |||
| DefaultCloudburstSchedulerPolicy | |||
| from cloudburst.server.scheduler.policy.heft_policy import ( | |||
There was a problem hiding this comment.
Can we make this something that’s read in from the YAML conf instead?
| @@ -48,7 +48,7 @@ def setUp(self): | |||
| self.kvs_client = kvs_client.MockAnnaClient() | |||
| self.ip = '127.0.0.1' | |||
|
|
|||
| self.policy = DefaultCloudburstSchedulerPolicy(self.pin_socket, | |||
| self.policy = HeftCloudburstSchedulerPolicy(self.pin_socket, | |||
There was a problem hiding this comment.
Hm. It seems like we would: (1) want to have HEFT specific tests; (2) not want to remove the existing policy tests. At the very minimum, we should duplicate the existing tests to have HEFT versions instead of replacing the existing tests.
Makes the default function scheduler use HEFT DAG scheduling algorithm.