[Notes] [Git][BuildGrid/buildgrid][finn/async] Added messaging queues for operation updates

[finnball <gitlab mg gitlab com>]

Title: GitLab

finnball pushed to branch finn/async at BuildGrid / buildgrid

Commits:

• 6a903877
  by finn at 2018-08-07T09:23:13Z

  Added messaging queues for operation updates
  
  ExecutionService now registers for message updates.
  Once all messages have been read and job is completed,
  job is removed.
  

6 changed files:

• buildgrid/server/execution/execution_instance.py
• buildgrid/server/execution/execution_service.py
• buildgrid/server/job.py
• buildgrid/server/scheduler.py
• tests/integration/bots_service.py
• tests/integration/execution_service.py

Changes:

buildgrid/server/execution/execution_instance.py

---

...	...	@@ -34,12 +34,12 @@ class ExecutionInstance():
34	34	self.logger = logging.getLogger(__name__)
35	35	self._scheduler = scheduler
36	36
37		- def execute(self, action_digest, skip_cache_lookup):
	37	+ def execute(self, action_digest, skip_cache_lookup, message_queue=None):
38	38	""" Sends a job for execution.
39	39	Queues an action and creates an Operation instance to be associated with
40	40	this action.
41	41	"""
42		- job = Job(action_digest)
	42	+ job = Job(action_digest, message_queue)
43	43	self.logger.info("Operation name: {}".format(job.name))
44	44
45	45	if not skip_cache_lookup:
...	...	@@ -70,3 +70,15 @@ class ExecutionInstance():
70	70	def cancel_operation(self, name):
71	71	# TODO: Cancel leases
72	72	raise NotImplementedError("Cancelled operations not supported")
	73	+
	74	+ def register_message_client(self, name, queue):
	75	+ try:
	76	+ self._scheduler.register_client(name, queue)
	77	+ except KeyError:
	78	+ raise InvalidArgumentError("Operation name does not exist: {}".format(name))
	79	+
	80	+ def unregister_message_client(self, name, queue):
	81	+ try:
	82	+ self._scheduler.unregister_client(name, queue)
	83	+ except KeyError:
	84	+ raise InvalidArgumentError("Operation name does not exist: {}".format(name))

buildgrid/server/execution/execution_service.py

---

...	...	@@ -22,10 +22,9 @@ ExecutionService
22	22	Serves remote execution requests.
23	23	"""
24	24
25		-import copy
26	25	import grpc
27	26	import logging
28		-import time
	27	+import queue
29	28
30	29	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
31	30	from buildgrid._protos.google.longrunning import operations_pb2_grpc, operations_pb2
...	...	@@ -35,17 +34,23 @@ from ._exceptions import InvalidArgumentError
35	34	class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
36	35
37	36	def __init__(self, instance):
38		- self._instance = instance
39	37	self.logger = logging.getLogger(__name__)
	38	+ self._instance = instance
40	39
41	40	def Execute(self, request, context):
42	41	# Ignore request.instance_name for now
43	42	# Have only one instance
44	43	try:
	44	+ message_queue = queue.Queue()
45	45	operation = self._instance.execute(request.action_digest,
46		- request.skip_cache_lookup)
	46	+ request.skip_cache_lookup,
	47	+ message_queue)
47	48
48		- yield from self._stream_operation_updates(operation.name)
	49	+ remove_client = lambda : self._remove_client(operation.name, message_queue)
	50	+ context.add_callback(remove_client)
	51	+
	52	+ yield from self._stream_operation_updates(message_queue,
	53	+ operation.name)
49	54
50	55	except InvalidArgumentError as e:
51	56	self.logger.error(e)
...	...	@@ -59,19 +64,28 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
59	64
60	65	def WaitExecution(self, request, context):
61	66	try:
62		- yield from self._stream_operation_updates(request.name)
	67	+ message_queue = queue.Queue()
	68	+ operation_name = request.name
	69	+
	70	+ self._instance.register_message_client(operation_name, message_queue)
	71	+
	72	+ remove_client = lambda : self._remove_client(operation_name, message_queue)
	73	+ context.add_callback(remove_client)
	74	+
	75	+ yield from self._stream_operation_updates(message_queue,
	76	+ operation_name)
63	77
64	78	except InvalidArgumentError as e:
65	79	self.logger.error(e)
66	80	context.set_details(str(e))
67	81	context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
68	82
69		- def _stream_operation_updates(self, name):
70		- stream_previous = None
71		- while True:
72		- stream = self._instance.get_operation(name)
73		- if stream != stream_previous:
74		- yield stream
75		- if stream.done == True: break
76		- stream_previous = copy.deepcopy(stream)
77		- time.sleep(1)
	83	+ def _remove_client(self, operation_name, message_queue):
	84	+ self._instance.unregister_message_client(operation_name, message_queue)
	85	+
	86	+ def _stream_operation_updates(self, message_queue, operation_name):
	87	+ operation = message_queue.get()
	88	+ while not operation.done:
	89	+ yield operation
	90	+ operation = message_queue.get()
	91	+ yield operation

buildgrid/server/job.py

---

...	...	@@ -51,21 +51,39 @@ class LeaseState(Enum):
51	51
52	52	class Job():
53	53
54		- def __init__(self, action):
55		- self.action = action
56		- self.bot_status = BotStatus.BOT_STATUS_UNSPECIFIED
57		- self.execute_stage = ExecuteStage.UNKNOWN
	54	+ def __init__(self, action_digest, message_queue=None):
58	55	self.lease = None
59	56	self.logger = logging.getLogger(__name__)
60		- self.name = str(uuid.uuid4())
61	57	self.result = None
62	58
	59	+ self._action_digest = action_digest
	60	+ self._execute_stage = ExecuteStage.UNKNOWN
63	61	self._n_tries = 0
64		- self._operation = operations_pb2.Operation(name = self.name)
	62	+ self._name = str(uuid.uuid4())
	63	+ self._operation = operations_pb2.Operation(name = self._name)
	64	+ self._operation_update_queues = []
	65	+
	66	+ if message_queue is not None:
	67	+ self.register_client(message_queue)
	68	+
	69	+ @property
	70	+ def name(self):
	71	+ return self._name
	72	+
	73	+ def check_job_finished(self):
	74	+ if not self._operation_update_queues:
	75	+ return self._operation.done
	76	+ return False
	77	+
	78	+ def register_client(self, queue):
	79	+ queue.put(self.get_operation())
	80	+ self._operation_update_queues.append(queue)
	81	+
	82	+ def unregister_client(self, queue):
	83	+ self._operation_update_queues.remove(queue)
65	84
66	85	def get_operation(self):
67	86	self._operation.metadata.CopyFrom(self._pack_any(self.get_operation_meta()))
68		-
69	87	if self.result is not None:
70	88	self._operation.done = True
71	89	response = ExecuteResponse()
...	...	@@ -76,15 +94,15 @@ class Job():
76	94
77	95	def get_operation_meta(self):
78	96	meta = ExecuteOperationMetadata()
79		- meta.stage = self.execute_stage.value
	97	+ meta.stage = self._execute_stage.value
80	98
81	99	return meta
82	100
83	101	def create_lease(self):
84		- action = self._pack_any(self.action)
	102	+ action_digest = self._pack_any(self._action_digest)
85	103
86	104	lease = bots_pb2.Lease(id = self.name,
87		- payload = action,
	105	+ payload = action_digest,
88	106	state = LeaseState.PENDING.value)
89	107	self.lease = lease
90	108	return lease
...	...	@@ -92,6 +110,11 @@ class Job():
92	110	def get_operations(self):
93	111	return operations_pb2.ListOperationsResponse(operations = [self.get_operation()])
94	112
	113	+ def update_execute_stage(self, stage):
	114	+ self._execute_stage = stage
	115	+ for queue in self._operation_update_queues:
	116	+ queue.put(self.get_operation())
	117	+
95	118	def _pack_any(self, pack):
96	119	any = any_pb2.Any()
97	120	any.Pack(pack)

buildgrid/server/scheduler.py

---

...	...	@@ -35,8 +35,17 @@ class Scheduler():
35	35	self.jobs = {}
36	36	self.queue = deque()
37	37
	38	+ def register_client(self, name, queue):
	39	+ self.jobs[name].register_client(queue)
	40	+
	41	+ def unregister_client(self, name, queue):
	42	+ job = self.jobs[name]
	43	+ job.unregister_client(queue)
	44	+ if job.check_job_finished():
	45	+ del self.jobs[name]
	46	+
38	47	def append_job(self, job):
39		- job.execute_stage = ExecuteStage.QUEUED
	48	+ job.update_execute_stage(ExecuteStage.QUEUED)
40	49	self.jobs[job.name] = job
41	50	self.queue.append(job)
42	51
...	...	@@ -45,9 +54,9 @@ class Scheduler():
45	54
46	55	if job.n_tries >= self.MAX_N_TRIES:
47	56	# TODO: Decide what to do with these jobs
48		- job.execute_stage = ExecuteStage.COMPLETED
	57	+ job.update_execute_stage(ExecuteStage.COMPLETED)
49	58	else:
50		- job.execute_stage = ExecuteStage.QUEUED
	59	+ job.update_execute_stage(ExecuteStage.QUEUED)
51	60	job.n_tries += 1
52	61	self.queue.appendleft(job)
53	62
...	...	@@ -56,15 +65,14 @@ class Scheduler():
56	65	def create_job(self):
57	66	if len(self.queue) > 0:
58	67	job = self.queue.popleft()
59		- job.execute_stage = ExecuteStage.EXECUTING
	68	+ job.update_execute_stage(ExecuteStage.EXECUTING)
60	69	self.jobs[job.name] = job
61	70	return job
62		- return None
63	71
64	72	def job_complete(self, name, result):
65	73	job = self.jobs[name]
66		- job.execute_stage = ExecuteStage.COMPLETED
67	74	job.result = result
	75	+ job.update_execute_stage(ExecuteStage.COMPLETED)
68	76	self.jobs[name] = job
69	77
70	78	def get_operations(self):
...	...	@@ -85,7 +93,7 @@ class Scheduler():
85	93	return lease
86	94	else:
87	95	job = create_job
88		- job.lease = job.create_lease()
	96	+ job.create_lease()
89	97
90	98	elif state == LeaseState.PENDING.value:
91	99	job.lease = lease

tests/integration/bots_service.py

---

...	...	@@ -187,7 +187,7 @@ def test_update_leases_work_complete(bot_session, context, instance):
187	187	# Simulated the severed binding between client and server
188	188	response = copy.deepcopy(instance.UpdateBotSession(request, context))
189	189	assert isinstance(response, bots_pb2.BotSession)
190		- assert instance._instance._scheduler.jobs[operation_name].execute_stage == ExecuteStage.COMPLETED
	190	+ assert instance._instance._scheduler.jobs[operation_name]._execute_stage == ExecuteStage.COMPLETED
191	191
192	192	def test_post_bot_event_temp(context, instance):
193	193	request = bots_pb2.PostBotEventTempRequest()

tests/integration/execution_service.py

---

...	...	@@ -67,19 +67,25 @@ def test_execute(skip_cache_lookup, instance, context):
67	67	assert metadata.stage == job.ExecuteStage.QUEUED.value
68	68	assert uuid.UUID(result.name, version=4)
69	69	assert result.done is False
70		-
	70	+"""
71	71	def test_wait_execution(instance, context):
	72	+ # TODO: Figure out why next(response) hangs on the .get()
	73	+ # method when running in pytest.
72	74	action_digest = remote_execution_pb2.Digest()
73	75	action_digest.hash = 'zhora'
74	76
75		- execution_request = remote_execution_pb2.ExecuteRequest(instance_name = '',
76		- action_digest = action_digest,
77		- skip_cache_lookup = True)
78		- execution_response = next(instance.Execute(execution_request, context))
	77	+ j = job.Job(action_digest, None)
	78	+ j._operation.done = True
	79	+
	80	+ request = remote_execution_pb2.WaitExecutionRequest(name=j.name)
79	81
	82	+ instance._instance._scheduler.jobs[j.name] = j
80	83
81		- request = remote_execution_pb2.WaitExecutionRequest(name=execution_response.name)
	84	+ action_result_any = any_pb2.Any()
	85	+ action_result = remote_execution_pb2.ActionResult()
	86	+ action_result_any.Pack(action_result)
82	87
83		- response = next(instance.WaitExecution(request, context))
	88	+ instance._instance._scheduler._update_execute_stage(j, job.ExecuteStage.COMPLETED)
84	89
85		- assert response == execution_response
	90	+ response = instance.WaitExecution(request, context)
	91	+"""

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