[Notes] [Git][BuildGrid/buildgrid][mablanch/117-job-scheduler-refactoring] buildgrid/server/job.py

[Martin Blanchard <gitlab mg gitlab com>]

Title: GitLab

Martin Blanchard pushed to branch mablanch/117-job-scheduler-refactoring at BuildGrid / buildgrid

Commits:

• 859993e8
  by Martin Blanchard at 2018-10-11T17:49:37Z

  buildgrid/server/job.py
  

5 changed files:

• buildgrid/server/bots/instance.py
• buildgrid/server/execution/instance.py
• buildgrid/server/job.py
• buildgrid/server/scheduler.py
• tests/integration/operations_service.py

Changes:

buildgrid/server/bots/instance.py

---

...	...	@@ -109,7 +109,7 @@ class BotsInterface:
109	109	if server_state == LeaseState.PENDING:
110	110
111	111	if client_state == LeaseState.ACTIVE:
112		- self._scheduler.update_job_lease_state(client_lease.id, client_lease.state)
	112	+ self._scheduler.update_job_lease_state(client_lease.id, client_state)
113	113	elif client_state == LeaseState.COMPLETED:
114	114	# TODO: Lease was rejected
115	115	raise NotImplementedError("'Not Accepted' is unsupported")
...	...	@@ -122,8 +122,7 @@ class BotsInterface:
122	122	pass
123	123
124	124	elif client_state == LeaseState.COMPLETED:
125		- self._scheduler.update_job_lease_state(client_lease.id, client_lease.state)
126		- self._scheduler.job_complete(client_lease.id, client_lease.result, client_lease.status)
	125	+ self._scheduler.update_job_lease_state(client_lease.id, client_state, lease_status=client_lease.status, lease_result=client_lease.result)
127	126	return None
128	127
129	128	else:

buildgrid/server/execution/instance.py

---

...	...	@@ -53,7 +53,7 @@ class ExecutionInstance:
53	53
54	54	self._scheduler.append_job(job, skip_cache_lookup)
55	55
56		- return job.get_operation()
	56	+ return job.operation
57	57
58	58	def register_message_client(self, name, queue):
59	59	try:

buildgrid/server/job.py

---

...	...	@@ -53,18 +53,23 @@ class LeaseState(Enum):
53	53	class Job:
54	54
55	55	def __init__(self, action_digest, do_not_cache=False, message_queue=None):
56		- self.lease = None
57	56	self.logger = logging.getLogger(__name__)
58		- self.n_tries = 0
59		- self.result = None
60		- self.result_cached = False
61	57
	58	+ self._name = str(uuid.uuid4())
62	59	self._action_digest = action_digest
63	60	self._do_not_cache = do_not_cache
64		- self._execute_stage = OperationStage.UNKNOWN
65		- self._name = str(uuid.uuid4())
66		- self._operation = operations_pb2.Operation(name=self._name)
	61	+ self._operation = operations_pb2.Operation()
	62	+ self._lease = None
	63	+ self._n_tries = 0
	64	+
	65	+ self.__operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
	66	+
	67	+ self.__operation_metadata.action_digest.CopyFrom(self._action_digest)
	68	+ self.__operation_metadata.stage = OperationStage.UNKNOWN.value
	69	+
67	70	self._operation_update_queues = []
	71	+ self._operation.name = self._name
	72	+ self._operation.done = False
68	73
69	74	if message_queue is not None:
70	75	self.register_client(message_queue)
...	...	@@ -81,47 +86,72 @@ class Job:
81	86	def do_not_cache(self):
82	87	return self._do_not_cache
83	88
84		- def check_job_finished(self):
85		- if not self._operation_update_queues:
86		- return self._operation.done
87		- return False
	89	+ @property
	90	+ def operation(self):
	91	+ return self._operation
	92	+
	93	+ @property
	94	+ def operation_stage(self):
	95	+ return OperationStage(self.__operation_metadata.state)
	96	+
	97	+ @property
	98	+ def lease(self):
	99	+ return self._lease
	100	+
	101	+ @property
	102	+ def lease_state(self):
	103	+ return LeaseState(self._lease.state)
	104	+
	105	+ @property
	106	+ def n_tries(self):
	107	+ return self._n_tries
88	108
89	109	def register_client(self, queue):
90	110	self._operation_update_queues.append(queue)
91		- queue.put(self.get_operation())
	111	+ queue.put(self._operation)
92	112
93	113	def unregister_client(self, queue):
94	114	self._operation_update_queues.remove(queue)
95	115
96		- def get_operation(self):
97		- self._operation.metadata.Pack(self.get_operation_meta())
98		- if self.result is not None:
99		- self._operation.done = True
100		- response = remote_execution_pb2.ExecuteResponse(result=self.result,
101		- cached_result=self.result_cached)
	116	+ def create_lease(self):
	117	+ self._lease = bots_pb2.Lease()
	118	+ self._lease.id = self._name
	119	+ self._lease.payload.Pack(self._action_digest)
	120	+ self._lease.state = LeaseState.PENDING.value
	121	+
	122	+ return self._lease
	123	+
	124	+ def update_lease_state(self, state, status=None, result=None):
	125	+ if state.value == self._lease.state:
	126	+ return
	127	+
	128	+ self._lease.state = state.value
102	129
103		- if not self.result_cached:
104		- response.status.CopyFrom(self.lease.status)
	130	+ if self._lease.state == LeaseState.COMPLETED.value:
	131	+ response = remote_execution_pb2.ExecuteResponse()
	132	+ response.result.CopyFrom(result)
	133	+ response.cached_result = False
	134	+ response.status.CopyFrom(status)
105	135
106	136	self._operation.response.Pack(response)
	137	+ self._operation.done = True
107	138
108		- return self._operation
	139	+ def cancel_lease(self):
	140	+ pass
109	141
110		- def get_operation_meta(self):
111		- meta = remote_execution_pb2.ExecuteOperationMetadata()
112		- meta.stage = self._execute_stage.value
113		- meta.action_digest.CopyFrom(self._action_digest)
	142	+ def update_operation_stage(self, stage):
	143	+ if stage.value == self.__operation_metadata.stage:
	144	+ return
114	145
115		- return meta
	146	+ self.__operation_metadata.stage = stage.value
116	147
117		- def create_lease(self):
118		- lease = bots_pb2.Lease(id=self.name, state=LeaseState.PENDING.value)
119		- lease.payload.Pack(self._action_digest)
	148	+ if self.__operation_metadata.stage == OperationStage.QUEUED.value:
	149	+ self._n_tries += 1
120	150
121		- self.lease = lease
122		- return lease
	151	+ self._operation.metadata.Pack(self.__operation_metadata)
123	152
124		- def update_operation_stage(self, stage):
125		- self._execute_stage = stage
126	153	for queue in self._operation_update_queues:
127		- queue.put(self.get_operation())
	154	+ queue.put(self._operation)
	155	+
	156	+ def cancel_operation(self):
	157	+ pass

buildgrid/server/scheduler.py

---

...	...	@@ -45,7 +45,7 @@ class Scheduler:
45	45	def unregister_client(self, name, queue):
46	46	job = self.jobs[name]
47	47	job.unregister_client(queue)
48		- if job.check_job_finished():
	48	+ if job.operation.done:
49	49	del self.jobs[name]
50	50
51	51	def append_job(self, job, skip_cache_lookup=False):
...	...	@@ -58,61 +58,52 @@ class Scheduler:
58	58	job.update_operation_stage(OperationStage.QUEUED)
59	59
60	60	else:
61		- job.result = cached_result
62		- job.result_cached = True
63	61	job.update_operation_stage(OperationStage.COMPLETED)
64	62
65	63	else:
66	64	self.queue.append(job)
67	65	job.update_operation_stage(OperationStage.QUEUED)
68	66
69		- def retry_job(self, name):
70		- if name in self.jobs:
71		- job = self.jobs[name]
	67	+ def retry_job(self, job_name):
	68	+ if job_name in self.jobs:
	69	+ job = self.jobs[job_name]
72	70	if job.n_tries >= self.MAX_N_TRIES:
73	71	# TODO: Decide what to do with these jobs
74	72	job.update_operation_stage(OperationStage.COMPLETED)
75	73	# TODO: Mark these jobs as done
76	74	else:
77	75	job.update_operation_stage(OperationStage.QUEUED)
78		- job.n_tries += 1
79	76	self.queue.appendleft(job)
80	77
81		- def job_complete(self, name, result, status):
82		- job = self.jobs[name]
83		- job.lease.status.CopyFrom(status)
84		- action_result = remote_execution_pb2.ActionResult()
85		- result.Unpack(action_result)
86		- job.result = action_result
87		- if not job.do_not_cache and self._action_cache is not None:
88		- if not job.lease.status.code:
89		- self._action_cache.update_action_result(job.action_digest, action_result)
90		- job.update_operation_stage(OperationStage.COMPLETED)
91		-
92	78	def get_operations(self):
93	79	response = operations_pb2.ListOperationsResponse()
94	80	for v in self.jobs.values():
95		- response.operations.extend([v.get_operation()])
	81	+ response.operations.extend([v.operation])
96	82	return response
97	83
98		- def update_job_lease_state(self, name, state):
99		- job = self.jobs[name]
100		- job.lease.state = state
	84	+ def update_job_lease_state(self, job_name, lease_state, lease_status=None, lease_result=None):
	85	+ job = self.jobs[job_name]
	86	+ if lease_state != LeaseState.COMPLETED:
	87	+ job.update_lease_state(lease_state)
	88	+ else:
	89	+ action_result = remote_execution_pb2.ActionResult()
	90	+ lease_result.Unpack(action_result)
101	91
102		- def get_job_lease(self, name):
103		- return self.jobs[name].lease
	92	+ job.update_lease_state(lease_state, status=lease_status, result=action_result)
104	93
105		- def cancel_session(self, name):
106		- job = self.jobs[name]
107		- state = job.lease.state
108		- if state in (LeaseState.PENDING.value, LeaseState.ACTIVE.value):
109		- self.retry_job(name)
	94	+ if not job.do_not_cache and self._action_cache is not None:
	95	+ if not job.lease.status.code:
	96	+ self._action_cache.update_action_result(job.action_digest, action_result)
	97	+
	98	+ job.update_operation_stage(OperationStage.COMPLETED)
	99	+
	100	+ def get_job_lease(self, job_name):
	101	+ return self.jobs[job_name].lease
110	102
111	103	def create_lease(self):
112	104	if self.queue:
113	105	job = self.queue.popleft()
114	106	job.update_operation_stage(OperationStage.EXECUTING)
115	107	job.create_lease()
116		- job.lease.state = LeaseState.PENDING.value
117	108	return job.lease
118	109	return None

tests/integration/operations_service.py

---

...	...	@@ -30,6 +30,7 @@ from buildgrid._protos.google.longrunning import operations_pb2
30	30	from buildgrid._protos.google.rpc import status_pb2
31	31	from buildgrid.server.cas.storage import lru_memory_cache
32	32	from buildgrid.server.controller import ExecutionController
	33	+from buildgrid.server.job import OperationStage
33	34	from buildgrid.server.operations import service
34	35	from buildgrid.server.operations.service import OperationsService
35	36	from buildgrid.utils import create_digest
...	...	@@ -131,9 +132,10 @@ def test_list_operations_with_result(instance, controller, execute_request, cont
131	132	action_result.output_files.extend([output_file])
132	133
133	134	controller.operations_instance._scheduler.jobs[response_execute.name].create_lease()
134		- controller.operations_instance._scheduler.job_complete(response_execute.name,
135		- _pack_any(action_result),
136		- status_pb2.Status())
	135	+ controller.operations_instance._scheduler.update_job_lease_state(response_execute.name,
	136	+ OperationStage.COMPLETED,
	137	+ lease_status=status_pb2.Status(),
	138	+ lease_result=_pack_any(action_result))
137	139
138	140	request = operations_pb2.ListOperationsRequest(name=instance_name)
139	141	response = instance.ListOperations(request, context)

—
View it on GitLab.
You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.