[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:

• fe719bfa
  by Martin Blanchard at 2018-10-11T17:24:09Z

  buildgrid/server/job.py
  

4 changed files:

• buildgrid/server/bots/instance.py
• buildgrid/server/execution/instance.py
• buildgrid/server/job.py
• buildgrid/server/scheduler.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,25 @@ 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	57	self.result = None
60	58	self.result_cached = False
61	59
	60	+ self._name = str(uuid.uuid4())
62	61	self._action_digest = action_digest
63	62	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)
	63	+ self._operation = operations_pb2.Operation()
	64	+ self._lease = None
	65	+ self._n_tries = 0
	66	+
	67	+ self.__operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
	68	+
	69	+ self.__operation_metadata.action_digest.CopyFrom(self._action_digest)
	70	+ self.__operation_metadata.stage = OperationStage.UNKNOWN.value
	71	+
67	72	self._operation_update_queues = []
	73	+ self._operation.name = self._name
	74	+ self._operation.done = False
68	75
69	76	if message_queue is not None:
70	77	self.register_client(message_queue)
...	...	@@ -81,47 +88,72 @@ class Job:
81	88	def do_not_cache(self):
82	89	return self._do_not_cache
83	90
84		- def check_job_finished(self):
85		- if not self._operation_update_queues:
86		- return self._operation.done
87		- return False
	91	+ @property
	92	+ def operation(self):
	93	+ return self._operation
	94	+
	95	+ @property
	96	+ def operation_stage(self):
	97	+ return OperationStage(self.__operation_metadata.state)
	98	+
	99	+ @property
	100	+ def lease(self):
	101	+ return self._lease
	102	+
	103	+ @property
	104	+ def lease_state(self):
	105	+ return LeaseState(self._lease.state)
	106	+
	107	+ @property
	108	+ def n_tries(self):
	109	+ return self._n_tries
88	110
89	111	def register_client(self, queue):
90	112	self._operation_update_queues.append(queue)
91		- queue.put(self.get_operation())
	113	+ queue.put(self._operation)
92	114
93	115	def unregister_client(self, queue):
94	116	self._operation_update_queues.remove(queue)
95	117
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)
	118	+ def create_lease(self):
	119	+ self._lease = bots_pb2.Lease()
	120	+ self._lease.id = self._name
	121	+ self._lease.payload.Pack(self._action_digest)
	122	+ self._lease.state = LeaseState.PENDING.value
	123	+
	124	+ return self._lease
	125	+
	126	+ def update_lease_state(self, state, status=None, result=None):
	127	+ if state.value == self._lease.state:
	128	+ return
	129	+
	130	+ self._lease.state = state.value
102	131
103		- if not self.result_cached:
104		- response.status.CopyFrom(self.lease.status)
	132	+ if self._lease.state == LeaseState.COMPLETED.value:
	133	+ response = remote_execution_pb2.ExecuteResponse()
	134	+ response.result.CopyFrom(result)
	135	+ response.cached_result = False
	136	+ response.status.CopyFrom(status)
105	137
106	138	self._operation.response.Pack(response)
	139	+ self._operation.done = True
107	140
108		- return self._operation
	141	+ def cancel_lease(self):
	142	+ pass
109	143
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)
	144	+ def update_operation_stage(self, stage):
	145	+ if stage.value == self.__operation_metadata.stage:
	146	+ return
114	147
115		- return meta
	148	+ self.__operation_metadata.stage = stage.value
116	149
117		- def create_lease(self):
118		- lease = bots_pb2.Lease(id=self.name, state=LeaseState.PENDING.value)
119		- lease.payload.Pack(self._action_digest)
	150	+ if self.__operation_metadata.stage == OperationStage.QUEUED.value:
	151	+ self._n_tries += 1
120	152
121		- self.lease = lease
122		- return lease
	153	+ self._operation.metadata.Pack(self.__operation_metadata)
123	154
124		- def update_operation_stage(self, stage):
125		- self._execute_stage = stage
126	155	for queue in self._operation_update_queues:
127		- queue.put(self.get_operation())
	156	+ queue.put(self._operation)
	157	+
	158	+ def cancel_operation(self):
	159	+ 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):
...	...	@@ -66,24 +66,24 @@ class Scheduler:
66	66	self.queue.append(job)
67	67	job.update_operation_stage(OperationStage.QUEUED)
68	68
69		- def retry_job(self, name):
70		- if name in self.jobs:
71		- job = self.jobs[name]
	69	+ def retry_job(self, job_name):
	70	+ if job_name in self.jobs:
	71	+ job = self.jobs[job_name]
72	72	if job.n_tries >= self.MAX_N_TRIES:
73	73	# TODO: Decide what to do with these jobs
74	74	job.update_operation_stage(OperationStage.COMPLETED)
75	75	# TODO: Mark these jobs as done
76	76	else:
77	77	job.update_operation_stage(OperationStage.QUEUED)
78		- job.n_tries += 1
79	78	self.queue.appendleft(job)
80	79
81		- def job_complete(self, name, result, status):
82		- job = self.jobs[name]
83		- job.lease.status.CopyFrom(status)
	80	+ def job_complete(self, job_name, result, status):
	81	+ job = self.jobs[job_name]
	82	+ # job.lease.status.CopyFrom(status)
84	83	action_result = remote_execution_pb2.ActionResult()
85	84	result.Unpack(action_result)
86		- job.result = action_result
	85	+ # job.result = action_result
	86	+ job.update_lease_state(LeaseState.COMPLETED, status=status, action_result=action_result)
87	87	if not job.do_not_cache and self._action_cache is not None:
88	88	if not job.lease.status.code:
89	89	self._action_cache.update_action_result(job.action_digest, action_result)
...	...	@@ -92,27 +92,32 @@ class Scheduler:
92	92	def get_operations(self):
93	93	response = operations_pb2.ListOperationsResponse()
94	94	for v in self.jobs.values():
95		- response.operations.extend([v.get_operation()])
	95	+ response.operations.extend([v.operation])
96	96	return response
97	97
98		- def update_job_lease_state(self, name, state):
99		- job = self.jobs[name]
100		- job.lease.state = state
	98	+ def update_job_lease_state(self, job_name, lease_state, lease_status=None, lease_result=None):
	99	+ job = self.jobs[job_name]
	100	+ if lease_state != LeaseState.COMPLETED:
	101	+ job.update_lease_state(lease_state)
	102	+ else:
	103	+ action_result = remote_execution_pb2.ActionResult()
	104	+ lease_result.Unpack(action_result)
101	105
102		- def get_job_lease(self, name):
103		- return self.jobs[name].lease
	106	+ job.update_lease_state(lease_state, status=lease_status, result=action_result)
104	107
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)
	108	+ if not job.do_not_cache and self._action_cache is not None:
	109	+ if not job.lease.status.code:
	110	+ self._action_cache.update_action_result(job.action_digest, action_result)
	111	+
	112	+ job.update_operation_stage(OperationStage.COMPLETED)
	113	+
	114	+ def get_job_lease(self, job_name):
	115	+ return self.jobs[job_name].lease
110	116
111	117	def create_lease(self):
112	118	if self.queue:
113	119	job = self.queue.popleft()
114	120	job.update_operation_stage(OperationStage.EXECUTING)
115	121	job.create_lease()
116		- job.lease.state = LeaseState.PENDING.value
117	122	return job.lease
118	123	return None

—
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.