[Notes] [Git][BuildStream/buildstream][Qinusty/message-helpers] Overhaul internal messaging API

[Tristan Maat <gitlab mg gitlab com>]

Title: GitLab

Tristan Maat pushed to branch Qinusty/message-helpers at BuildStream / buildstream

Commits:

• 3230c639
  by Tristan Maat at 2018-09-26T12:55:22Z

  Overhaul internal messaging API
  
  _context.py: Added helpers for the majority of MessageTypes,
  start, failure and success have been avoided to avoid temptation to use
  these throughout the code base. These should be called primarily from
  timed_activity().
  
  job.py, elementjob.py: Add helper functions to populate task_id kwarg. _fail() exists
  purely to be overridden by ElementJob as it requires the scheduler kwarg
  to be set to True.
  
  Other: Switched to use _context.py message helpers throughout the code
  base.
  

14 changed files:

• buildstream/_artifactcache/artifactcache.py
• buildstream/_artifactcache/cascache.py
• buildstream/_context.py
• buildstream/_frontend/app.py
• buildstream/_pipeline.py
• buildstream/_platform/linux.py
• buildstream/_project.py
• buildstream/_scheduler/jobs/elementjob.py
• buildstream/_scheduler/jobs/job.py
• buildstream/_scheduler/queues/buildqueue.py
• buildstream/_scheduler/queues/queue.py
• buildstream/_stream.py
• buildstream/plugin.py
• buildstream/utils.py

Changes:

buildstream/_artifactcache/artifactcache.py

---

...	...	@@ -23,7 +23,6 @@ from collections import Mapping, namedtuple
23	23
24	24	from ..element_enums import _KeyStrength
25	25	from .._exceptions import ArtifactError, ImplError, LoadError, LoadErrorReason
26		-from .._message import Message, MessageType
27	26	from .. import utils
28	27	from .. import _yaml
29	28
...	...	@@ -498,15 +497,6 @@ class ArtifactCache():
498	497	# Local Private Methods #
499	498	################################################
500	499
501		- # _message()
502		- #
503		- # Local message propagator
504		- #
505		- def _message(self, message_type, message, **kwargs):
506		- args = dict(kwargs)
507		- self.context.message(
508		- Message(None, message_type, message, **args))
509		-
510	500	# _set_remotes():
511	501	#
512	502	# Set the list of remote caches. If project is None, the global list of
...	...	@@ -530,7 +520,7 @@ class ArtifactCache():
530	520	#
531	521	def _initialize_remotes(self):
532	522	def remote_failed(url, error):
533		- self._message(MessageType.WARN, "Failed to initialize remote {}: {}".format(url, error))
	523	+ self.context.warn("Failed to fetch remote refs from {}: {}".format(url, error))
534	524
535	525	with self.context.timed_activity("Initializing remote caches", silent_nested=True):
536	526	self.initialize_remotes(on_failure=remote_failed)

buildstream/_artifactcache/cascache.py

---

...	...	@@ -36,10 +36,9 @@ from .._protos.google.bytestream import bytestream_pb2, bytestream_pb2_grpc
36	36	from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
37	37	from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
38	38
39		-from .._message import MessageType, Message
40	39	from .. import _signals, utils
41	40	from .._exceptions import ArtifactError
42		-
	41	+from .._message import MessageType
43	42	from . import ArtifactCache
44	43
45	44
...	...	@@ -250,12 +249,11 @@ class CASCache(ArtifactCache):
250	249	raise ArtifactError("Failed to pull artifact {}: {}".format(
251	250	element._get_brief_display_key(), e)) from e
252	251	else:
253		- self.context.message(Message(
254		- None,
255		- MessageType.SKIPPED,
	252	+ self.context._message(
256	253	"Remote ({}) does not have {} cached".format(
257		- remote.spec.url, element._get_brief_display_key())
258		- ))
	254	+ remote.spec.url, element._get_brief_display_key()),
	255	+ msg_type=MessageType.SKIPPED
	256	+ )
259	257
260	258	return False
261	259
...	...	@@ -361,12 +359,11 @@ class CASCache(ArtifactCache):
361	359	raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
362	360
363	361	if skipped_remote:
364		- self.context.message(Message(
365		- None,
366		- MessageType.SKIPPED,
	362	+ self.context._message(
367	363	"Remote ({}) already has {} cached".format(
368		- remote.spec.url, element._get_brief_display_key())
369		- ))
	364	+ remote.spec.url, element._get_brief_display_key()),
	365	+ msg_type=MessageType.SKIPPED
	366	+ )
370	367	return pushed
371	368
372	369	################################################

buildstream/_context.py

---

...	...	@@ -19,6 +19,7 @@
19	19
20	20	import os
21	21	import datetime
	22	+import traceback
22	23	from collections import deque, Mapping
23	24	from contextlib import contextmanager
24	25	from . import utils
...	...	@@ -26,6 +27,7 @@ from . import _cachekey
26	27	from . import _signals
27	28	from . import _site
28	29	from . import _yaml
	30	+from .plugin import Plugin
29	31	from ._exceptions import LoadError, LoadErrorReason, BstError
30	32	from ._message import Message, MessageType
31	33	from ._profile import Topics, profile_start, profile_end
...	...	@@ -317,7 +319,7 @@ class Context():
317	319	# the context.
318	320	#
319	321	# The message handler should have the same signature as
320		- # the message() method
	322	+ # the _send_message() method
321	323	def set_message_handler(self, handler):
322	324	self._message_handler = handler
323	325
...	...	@@ -332,16 +334,15 @@ class Context():
332	334	return True
333	335	return False
334	336
335		- # message():
	337	+ # _send_message():
336	338	#
337		- # Proxies a message back to the caller, this is the central
	339	+ # Proxies a message back through the message handler, this is the central
338	340	# point through which all messages pass.
339	341	#
340	342	# Args:
341	343	# message: A Message object
342	344	#
343		- def message(self, message):
344		-
	345	+ def _send_message(self, message):
345	346	# Tag message only once
346	347	if message.depth is None:
347	348	message.depth = len(list(self._message_depth))
...	...	@@ -355,7 +356,87 @@ class Context():
355	356	assert self._message_handler
356	357
357	358	self._message_handler(message, context=self)
358		- return
	359	+
	360	+ # message():
	361	+ #
	362	+ # The global message API. Any message-sending functions should go
	363	+ # through here. This will call `_send_message` to deliver the
	364	+ # final message.
	365	+ #
	366	+ # Args:
	367	+ # text (str): The text of the message.
	368	+ #
	369	+ # Kwargs:
	370	+ # msg_type (MessageType): The type of the message (required).
	371	+ # plugin (Plugin|str|None): The id of the plugin
	372	+ # (i.e. Element, Source subclass
	373	+ # instance) sending the message. If
	374	+ # a plugin is given, this will be
	375	+ # determined automatically, if
	376	+ # omitted the message will be sent
	377	+ # without a plugin context.
	378	+ #
	379	+ # For other kwargs, see `Message`.
	380	+ #
	381	+ def message(self, text, *, plugin=None, msg_type=None, **kwargs):
	382	+ assert msg_type is not None
	383	+
	384	+ if isinstance(plugin, Plugin):
	385	+ plugin_id = plugin._get_unique_id()
	386	+ else:
	387	+ plugin_id = plugin
	388	+
	389	+ self._send_message(Message(plugin_id, msg_type, str(text), **kwargs))
	390	+
	391	+ # skipped():
	392	+ #
	393	+ # Produce and send a skipped message through the context.
	394	+ #
	395	+ def skipped(self, text, **kwargs):
	396	+ self._message(text, msg_type=MessageType.SKIPPED, **kwargs)
	397	+
	398	+ # debug():
	399	+ #
	400	+ # Produce and send a debug message through the context.
	401	+ #
	402	+ def debug(self, text, **kwargs):
	403	+ if self.log_debug:
	404	+ self._message(text, msg_type=MessageType.DEBUG, **kwargs)
	405	+
	406	+ # status():
	407	+ #
	408	+ # Produce and send a status message through the context.
	409	+ #
	410	+ def status(self, text, **kwargs):
	411	+ self._message(text, msg_type=MessageType.STATUS, **kwargs)
	412	+
	413	+ # info():
	414	+ #
	415	+ # Produce and send a info message through the context.
	416	+ #
	417	+ def info(self, text, **kwargs):
	418	+ self._message(text, msg_type=MessageType.INFO, **kwargs)
	419	+
	420	+ # warn():
	421	+ #
	422	+ # Produce and send a warning message through the context.
	423	+ #
	424	+ def warn(self, text, **kwargs):
	425	+ self._message(text, msg_type=MessageType.WARN, **kwargs)
	426	+
	427	+ # error():
	428	+ #
	429	+ # Produce and send a error message through the context.
	430	+ #
	431	+ def error(self, text, **kwargs):
	432	+ self._message(text, msg_type=MessageType.ERROR, **kwargs)
	433	+
	434	+ # log():
	435	+ #
	436	+ # Produce and send a log message through the context.
	437	+ #
	438	+ def log(self, text, **kwargs):
	439	+ self._message(text, msg_type=MessageType.LOG, **kwargs)
359	440
360	441	# silence()
361	442	#
...	...	@@ -401,8 +482,8 @@ class Context():
401	482	with _signals.suspendable(stop_time, resume_time):
402	483	try:
403	484	# Push activity depth for status messages
404		- message = Message(unique_id, MessageType.START, activity_name, detail=detail)
405		- self.message(message)
	485	+ self._message(activity_name, detail=detail, plugin=unique_id,
	486	+ msg_type=MessageType.START)
406	487	self._push_message_depth(silent_nested)
407	488	yield
408	489
...	...	@@ -410,15 +491,16 @@ class Context():
410	491	# Note the failure in status messages and reraise, the scheduler
411	492	# expects an error when there is an error.
412	493	elapsed = datetime.datetime.now() - starttime
413		- message = Message(unique_id, MessageType.FAIL, activity_name, elapsed=elapsed)
414	494	self._pop_message_depth()
415		- self.message(message)
	495	+ self._message(activity_name, detail=detail, elapsed=elapsed, plugin=unique_id,
	496	+ msg_type=MessageType.FAIL)
416	497	raise
417	498
418	499	elapsed = datetime.datetime.now() - starttime
419		- message = Message(unique_id, MessageType.SUCCESS, activity_name, elapsed=elapsed)
420	500	self._pop_message_depth()
421		- self.message(message)
	501	+ self._message(activity_name, detail=detail,
	502	+ elapsed=elapsed, plugin=unique_id,
	503	+ msg_type=MessageType.SUCCESS)
422	504
423	505	# recorded_messages()
424	506	#

buildstream/_frontend/app.py

---

...	...	@@ -255,7 +255,7 @@ class App():
255	255
256	256	# Mark the beginning of the session
257	257	if session_name:
258		- self._message(MessageType.START, session_name)
	258	+ self.context._message(session_name, msg_type=MessageType.START)
259	259
260	260	# Run the body of the session here, once everything is loaded
261	261	try:
...	...	@@ -267,9 +267,9 @@ class App():
267	267	elapsed = self.stream.elapsed_time
268	268
269	269	if isinstance(e, StreamError) and e.terminated: # pylint: disable=no-member
270		- self._message(MessageType.WARN, session_name + ' Terminated', elapsed=elapsed)
	270	+ self.context.warn(session_name + ' Terminated', elapsed=elapsed)
271	271	else:
272		- self._message(MessageType.FAIL, session_name, elapsed=elapsed)
	272	+ self.context._message(session_name, elapsed=elapsed, msg_type=MessageType.FAIL)
273	273
274	274	# Notify session failure
275	275	self._notify("{} failed".format(session_name), "{}".format(e))
...	...	@@ -287,7 +287,7 @@ class App():
287	287	else:
288	288	# No exceptions occurred, print session time and summary
289	289	if session_name:
290		- self._message(MessageType.SUCCESS, session_name, elapsed=self.stream.elapsed_time)
	290	+ self.context._message(session_name, elapsed=self.stream.elapsed_time, msg_type=MessageType.SUCCESS)
291	291	if self._started:
292	292	self._print_summary()
293	293
...	...	@@ -433,21 +433,13 @@ class App():
433	433	if self.interactive:
434	434	self.notify(title, text)
435	435
436		- # Local message propagator
437		- #
438		- def _message(self, message_type, message, **kwargs):
439		- args = dict(kwargs)
440		- self.context.message(
441		- Message(None, message_type, message, **args))
442		-
443	436	# Exception handler
444	437	#
445	438	def _global_exception_handler(self, etype, value, tb):
446	439
447	440	# Print the regular BUG message
448	441	formatted = "".join(traceback.format_exception(etype, value, tb))
449		- self._message(MessageType.BUG, str(value),
450		- detail=formatted)
	442	+ self.context._message(str(value), detail=formatted, msg_type=MessageType.BUG)
451	443
452	444	# If the scheduler has started, try to terminate all jobs gracefully,
453	445	# otherwise exit immediately.

buildstream/_pipeline.py

---

...	...	@@ -24,7 +24,6 @@ import itertools
24	24	from operator import itemgetter
25	25
26	26	from ._exceptions import PipelineError
27		-from ._message import Message, MessageType
28	27	from ._profile import Topics, profile_start, profile_end
29	28	from . import Scope, Consistency
30	29	from ._project import ProjectRefStorage
...	...	@@ -201,8 +200,8 @@ class Pipeline():
201	200	for t in targets:
202	201	new_elm = t._get_source_element()
203	202	if new_elm != t and not silent:
204		- self._message(MessageType.INFO, "Element '{}' redirected to '{}'"
205		- .format(t.name, new_elm.name))
	203	+ self._context.info("Element '{}' redirected to '{}'"
	204	+ .format(t.name, new_elm.name))
206	205	if new_elm not in elements:
207	206	elements.append(new_elm)
208	207	elif mode == PipelineSelection.PLAN:
...	...	@@ -433,15 +432,6 @@ class Pipeline():
433	432
434	433	raise PipelineError("Untrackable sources", detail=detail, reason="untrackable-sources")
435	434
436		- # _message()
437		- #
438		- # Local message propagator
439		- #
440		- def _message(self, message_type, message, **kwargs):
441		- args = dict(kwargs)
442		- self._context.message(
443		- Message(None, message_type, message, **args))
444		-
445	435
446	436	# _Planner()
447	437	#

buildstream/_platform/linux.py

---

...	...	@@ -22,7 +22,6 @@ import subprocess
22	22	from .. import _site
23	23	from .. import utils
24	24	from .._artifactcache.cascache import CASCache
25		-from .._message import Message, MessageType
26	25	from ..sandbox import SandboxBwrap
27	26
28	27	from . import Platform
...	...	@@ -75,9 +74,9 @@ class Linux(Platform):
75	74	return True
76	75
77	76	else:
78		- context.message(
79		- Message(None, MessageType.WARN,
80		- "Unable to create user namespaces with bubblewrap, resorting to fallback",
81		- detail="Some builds may not function due to lack of uid / gid 0, " +
82		- "artifacts created will not be trusted for push purposes."))
	77	+ context.warn(
	78	+ "Unable to create user namespaces with bubblewrap, resorting to fallback",
	79	+ detail="Some builds may not function due to lack of uid / gid 0, " +
	80	+ "artifacts created will not be trusted for push purposes."
	81	+ )
83	82	return False

buildstream/_project.py

---

...	...	@@ -36,7 +36,6 @@ from ._projectrefs import ProjectRefs, ProjectRefStorage
36	36	from ._versions import BST_FORMAT_VERSION
37	37	from ._loader import Loader
38	38	from .element import Element
39		-from ._message import Message, MessageType
40	39	from ._includes import Includes
41	40
42	41
...	...	@@ -334,8 +333,7 @@ class Project():
334	333	for source, ref in redundant_refs
335	334	]
336	335	detail += "\n".join(lines)
337		- self._context.message(
338		- Message(None, MessageType.WARN, "Ignoring redundant source references", detail=detail))
	336	+ self._context.warn("Ignoring redundant source references", detail=detail)
339	337
340	338	return elements
341	339
...	...	@@ -503,13 +501,9 @@ class Project():
503	501
504	502	# Deprecation check
505	503	if fail_on_overlap is not None:
506		- self._context.message(
507		- Message(
508		- None,
509		- MessageType.WARN,
510		- "Use of fail-on-overlap within project.conf " +
511		- "is deprecated. Consider using fatal-warnings instead."
512		- )
	504	+ self._context.warn(
	505	+ "Use of fail-on-overlap within project.conf " +
	506	+ "is deprecated. Consider using fatal-warnings instead."
513	507	)
514	508
515	509	# Load project.refs if it exists, this may be ignored.

buildstream/_scheduler/jobs/elementjob.py

---

...	...	@@ -18,8 +18,6 @@
18	18	#
19	19	from ruamel import yaml
20	20
21		-from ..._message import Message, MessageType
22		-
23	21	from .job import Job
24	22
25	23
...	...	@@ -86,9 +84,8 @@ class ElementJob(Job):
86	84	# This should probably be omitted for non-build tasks but it's harmless here
87	85	elt_env = self._element.get_environment()
88	86	env_dump = yaml.round_trip_dump(elt_env, default_flow_style=False, allow_unicode=True)
89		- self.message(MessageType.LOG,
90		- "Build environment for element {}".format(self._element.name),
91		- detail=env_dump)
	87	+ self._log("Build environment for element {}".format(self._element.name),
	88	+ detail=env_dump, plugin=self.element, scheduler=True)
92	89
93	90	# Run the action
94	91	return self._action_cb(self._element)
...	...	@@ -96,15 +93,6 @@ class ElementJob(Job):
96	93	def parent_complete(self, success, result):
97	94	self._complete_cb(self, self._element, success, self._result)
98	95
99		- def message(self, message_type, message, **kwargs):
100		- args = dict(kwargs)
101		- args['scheduler'] = True
102		- self._scheduler.context.message(
103		- Message(self._element._get_unique_id(),
104		- message_type,
105		- message,
106		- **args))
107		-
108	96	def child_process_data(self):
109	97	data = {}
110	98
...	...	@@ -119,3 +107,10 @@ class ElementJob(Job):
119	107	data['cache_size'] = cache_size
120	108
121	109	return data
	110	+
	111	+ # _fail()
	112	+ #
	113	+ # Override _fail to set scheduler kwarg to true.
	114	+ #
	115	+ def _fail(self, text, **kwargs):
	116	+ super()._fail(text, scheduler=True, **kwargs)

buildstream/_scheduler/jobs/job.py

---

...	...	@@ -32,7 +32,7 @@ import psutil
32	32
33	33	# BuildStream toplevel imports
34	34	from ..._exceptions import ImplError, BstError, set_last_task_error
35		-from ..._message import Message, MessageType, unconditional_messages
	35	+from ..._message import MessageType, unconditional_messages
36	36	from ... import _signals, utils
37	37
38	38	# Return code values shutdown of job handling child processes
...	...	@@ -109,6 +109,7 @@ class Job():
109	109	# Private members
110	110	#
111	111	self._scheduler = scheduler # The scheduler
	112	+ self._context = scheduler.context # The context, used primarily for UI messaging.
112	113	self._queue = None # A message passing queue
113	114	self._process = None # The Process object
114	115	self._watcher = None # Child process watcher
...	...	@@ -181,7 +182,7 @@ class Job():
181	182	# First resume the job if it's suspended
182	183	self.resume(silent=True)
183	184
184		- self.message(MessageType.STATUS, "{} terminating".format(self.action_name))
	185	+ self._status("{} terminating".format(self.action_name))
185	186
186	187	# Make sure there is no garbage on the queue
187	188	self._parent_stop_listening()
...	...	@@ -212,8 +213,8 @@ class Job():
212	213	def kill(self):
213	214
214	215	# Force kill
215		- self.message(MessageType.WARN,
216		- "{} did not terminate gracefully, killing".format(self.action_name))
	216	+ self._warn("{} did not terminate gracefully, killing"
	217	+ .format(self.action_name))
217	218
218	219	try:
219	220	utils._kill_process_tree(self._process.pid)
...	...	@@ -228,8 +229,7 @@ class Job():
228	229	#
229	230	def suspend(self):
230	231	if not self._suspended:
231		- self.message(MessageType.STATUS,
232		- "{} suspending".format(self.action_name))
	232	+ self._status("{} suspending".format(self.action_name))
233	233
234	234	try:
235	235	# Use SIGTSTP so that child processes may handle and propagate
...	...	@@ -253,8 +253,7 @@ class Job():
253	253	def resume(self, silent=False):
254	254	if self._suspended:
255	255	if not silent and not self._scheduler.terminated:
256		- self.message(MessageType.STATUS,
257		- "{} resuming".format(self.action_name))
	256	+ self._status("{} resuming".format(self.action_name))
258	257
259	258	os.kill(self._process.pid, signal.SIGCONT)
260	259	self._suspended = False
...	...	@@ -307,21 +306,6 @@ class Job():
307	306	raise ImplError("Job '{kind}' does not implement child_process()"
308	307	.format(kind=type(self).__name__))
309	308
310		- # message():
311		- #
312		- # Logs a message, this will be logged in the task's logfile and
313		- # conditionally also be sent to the frontend.
314		- #
315		- # Args:
316		- # message_type (MessageType): The type of message to send
317		- # message (str): The message
318		- # kwargs: Remaining Message() constructor arguments
319		- #
320		- def message(self, message_type, message, **kwargs):
321		- args = dict(kwargs)
322		- args['scheduler'] = True
323		- self._scheduler.context.message(Message(None, message_type, message, **args))
324		-
325	309	# child_process_data()
326	310	#
327	311	# Abstract method to retrieve additional data that should be
...	...	@@ -348,6 +332,32 @@ class Job():
348	332	#
349	333	#######################################################
350	334
	335	+ def _debug(self, text, **kwargs):
	336	+ self._context.debug(text, task_id=self._task_id, **kwargs)
	337	+
	338	+ def _status(self, text, **kwargs):
	339	+ self._context.status(text, task_id=self._task_id, **kwargs)
	340	+
	341	+ def _info(self, text, **kwargs):
	342	+ self._context.info(text, task_id=self._task_id, **kwargs)
	343	+
	344	+ def _warn(self, text, **kwargs):
	345	+ self._context.warn(text, task_id=self._task_id, **kwargs)
	346	+
	347	+ def _error(self, text, **kwargs):
	348	+ self._context.error(text, task_id=self._task_id, **kwargs)
	349	+
	350	+ def _log(self, text, **kwargs):
	351	+ self._context.log(text, task_id=self._task_id, **kwargs)
	352	+
	353	+ # _fail()
	354	+ #
	355	+ # Only exists for sub classes to override and add kwargs to.
	356	+ #
	357	+ def _fail(self, text, **kwargs):
	358	+ self._context._message(text, task_id=self._task_id,
	359	+ msg_type=MessageType.FAIL, **kwargs)
	360	+
351	361	# _child_action()
352	362	#
353	363	# Perform the action in the child process, this calls the action_cb.
...	...	@@ -374,7 +384,7 @@ class Job():
374	384	# Set the global message handler in this child
375	385	# process to forward messages to the parent process
376	386	self._queue = queue
377		- self._scheduler.context.set_message_handler(self._child_message_handler)
	387	+ self._context.set_message_handler(self._child_message_handler)
378	388
379	389	starttime = datetime.datetime.now()
380	390	stopped_time = None
...	...	@@ -391,9 +401,10 @@ class Job():
391	401	# Time, log and and run the action function
392	402	#
393	403	with _signals.suspendable(stop_time, resume_time), \
394		- self._scheduler.context.recorded_messages(self._logfile) as filename:
	404	+ self._context.recorded_messages(self._logfile) as filename:
395	405
396		- self.message(MessageType.START, self.action_name, logfile=filename)
	406	+ self._context._message(self.action_name, logfile=filename,
	407	+ msg_type=MessageType.START, task_id=self._task_id)
397	408
398	409	try:
399	410	# Try the task action
...	...	@@ -403,13 +414,12 @@ class Job():
403	414	self._retry_flag = e.temporary
404	415
405	416	if self._retry_flag and (self._tries <= self._max_retries):
406		- self.message(MessageType.FAIL,
407		- "Try #{} failed, retrying".format(self._tries),
408		- elapsed=elapsed, logfile=filename)
	417	+ self._fail("Try #{} failed, retrying".format(self._tries),
	418	+ elapsed=elapsed, logfile=filename)
409	419	else:
410		- self.message(MessageType.FAIL, str(e),
411		- elapsed=elapsed, detail=e.detail,
412		- logfile=filename, sandbox=e.sandbox)
	420	+ self._fail(str(e), elapsed=elapsed,
	421	+ detail=e.detail, logfile=filename,
	422	+ sandbox=e.sandbox)
413	423
414	424	self._queue.put(Envelope('child_data', self.child_process_data()))
415	425
...	...	@@ -429,9 +439,9 @@ class Job():
429	439	elapsed = datetime.datetime.now() - starttime
430	440	detail = "An unhandled exception occured:\n\n{}".format(traceback.format_exc())
431	441
432		- self.message(MessageType.BUG, self.action_name,
433		- elapsed=elapsed, detail=detail,
434		- logfile=filename)
	442	+ self._context._message(self.action_name, elapsed=elapsed,
	443	+ detail=detail, logfile=filename,
	444	+ task_id=self._task_id, msg_type=MessageType.BUG)
435	445	# Unhandled exceptions should permenantly fail
436	446	self._child_shutdown(RC_PERM_FAIL)
437	447
...	...	@@ -441,8 +451,10 @@ class Job():
441	451	self._child_send_result(result)
442	452
443	453	elapsed = datetime.datetime.now() - starttime
444		- self.message(MessageType.SUCCESS, self.action_name, elapsed=elapsed,
445		- logfile=filename)
	454	+ self._context._message(self.action_name,
	455	+ elapsed=elapsed, logfile=filename,
	456	+ msg_type=MessageType.SUCCESS,
	457	+ task_id=self._task_id)
446	458
447	459	# Shutdown needs to stay outside of the above context manager,
448	460	# make sure we dont try to handle SIGTERM while the process
...	...	@@ -575,7 +587,7 @@ class Job():
575	587	if envelope._message_type == 'message':
576	588	# Propagate received messages from children
577	589	# back through the context.
578		- self._scheduler.context.message(envelope._message)
	590	+ self._context._send_message(envelope._message)
579	591	elif envelope._message_type == 'error':
580	592	# For regression tests only, save the last error domain / reason
581	593	# reported from a child task in the main process, this global state

buildstream/_scheduler/queues/buildqueue.py

---

...	...	@@ -50,10 +50,10 @@ class BuildQueue(Queue):
50	50	self._tried.add(element)
51	51	_, description, detail = element._get_build_result()
52	52	logfile = element._get_build_log()
53		- self._message(element, MessageType.FAIL, description,
54		- detail=detail, action_name=self.action_name,
55		- elapsed=timedelta(seconds=0),
56		- logfile=logfile)
	53	+ self._context._message(description, msg_type=MessageType.FAIL, plugin=element,
	54	+ detail=detail, action_name=self.action_name,
	55	+ elapsed=timedelta(seconds=0),
	56	+ logfile=logfile)
57	57	job = ElementJob(self._scheduler, self.action_name,
58	58	logfile, element=element, queue=self,
59	59	resources=self.resources,

buildstream/_scheduler/queues/queue.py

---

...	...	@@ -30,7 +30,7 @@ from ..resources import ResourceType
30	30
31	31	# BuildStream toplevel imports
32	32	from ..._exceptions import BstError, set_last_task_error
33		-from ..._message import Message, MessageType
	33	+from ..._message import MessageType
34	34
35	35
36	36	# Queue status for a given element
...	...	@@ -72,6 +72,7 @@ class Queue():
72	72	# Private members
73	73	#
74	74	self._scheduler = scheduler
	75	+ self._context = scheduler.context
75	76	self._wait_queue = deque()
76	77	self._done_queue = deque()
77	78	self._max_retries = 0
...	...	@@ -274,17 +275,19 @@ class Queue():
274	275	# Handle any workspace modifications now
275	276	#
276	277	if workspace_dict:
277		- context = element._get_context()
278		- workspaces = context.get_workspaces()
	278	+ workspaces = self._context.get_workspaces()
279	279	if workspaces.update_workspace(element._get_full_name(), workspace_dict):
280	280	try:
281	281	workspaces.save_config()
282	282	except BstError as e:
283		- self._message(element, MessageType.ERROR, "Error saving workspaces", detail=str(e))
284		- except Exception as e: # pylint: disable=broad-except
285		- self._message(element, MessageType.BUG,
286		- "Unhandled exception while saving workspaces",
287		- detail=traceback.format_exc())
	283	+ self._context.error("Error saving workspaces",
	284	+ detail=e,
	285	+ plugin=element)
	286	+ except Exception as e: #pylint: disable=broad-except
	287	+ self._context.message("Unhandled exception while saving workspaces",
	288	+ msg_type=MessageType.BUG,
	289	+ detail=traceback.format_exc(),
	290	+ plugin=element)
288	291
289	292	# _job_done()
290	293	#
...	...	@@ -311,10 +314,10 @@ class Queue():
311	314	processed = self.done(job, element, result, success)
312	315
313	316	except BstError as e:
314		-
315	317	# Report error and mark as failed
316	318	#
317		- self._message(element, MessageType.ERROR, "Post processing error", detail=str(e))
	319	+ self._context.error("Post processing error",
	320	+ plugin=element, detail=e)
318	321	self.failed_elements.append(element)
319	322
320	323	# Treat this as a task error as it's related to a task
...	...	@@ -324,16 +327,15 @@ class Queue():
324	327	#
325	328	set_last_task_error(e.domain, e.reason)
326	329
327		- except Exception as e: # pylint: disable=broad-except
328		-
	330	+ except Exception: # pylint: disable=broad-except
329	331	# Report unhandled exceptions and mark as failed
330	332	#
331		- self._message(element, MessageType.BUG,
332		- "Unhandled exception in post processing",
333		- detail=traceback.format_exc())
	333	+ self._context.message("Unhandled exception in post processing",
	334	+ plugin=element, msg_type=MessageType.BUG,
	335	+ detail=traceback.format_exc())
334	336	self.failed_elements.append(element)
335		- else:
336	337
	338	+ else:
337	339	# No exception occured, handle the success/failure state in the normal way
338	340	#
339	341	self._done_queue.append(job)
...	...	@@ -346,13 +348,6 @@ class Queue():
346	348	else:
347	349	self.failed_elements.append(element)
348	350
349		- # Convenience wrapper for Queue implementations to send
350		- # a message for the element they are processing
351		- def _message(self, element, message_type, brief, **kwargs):
352		- context = element._get_context()
353		- message = Message(element._get_unique_id(), message_type, brief, **kwargs)
354		- context.message(message)
355		-
356	351	def _element_log_path(self, element):
357	352	project = element._get_project()
358	353	key = element._get_display_key()[1]

buildstream/_stream.py

---

...	...	@@ -25,11 +25,12 @@ import stat
25	25	import shlex
26	26	import shutil
27	27	import tarfile
	28	+import traceback
28	29	from contextlib import contextmanager
29	30	from tempfile import TemporaryDirectory
30	31
31	32	from ._exceptions import StreamError, ImplError, BstError, set_last_task_error
32		-from ._message import Message, MessageType
	33	+from ._message import MessageType
33	34	from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
34	35	from ._pipeline import Pipeline, PipelineSelection
35	36	from ._platform import Platform
...	...	@@ -512,7 +513,7 @@ class Stream():
512	513	target._open_workspace()
513	514
514	515	workspaces.save_config()
515		- self._message(MessageType.INFO, "Saved workspace configuration")
	516	+ self._context.info("Saved workspace configuration")
516	517
517	518	# workspace_close
518	519	#
...	...	@@ -539,7 +540,7 @@ class Stream():
539	540	# Delete the workspace and save the configuration
540	541	workspaces.delete_workspace(element_name)
541	542	workspaces.save_config()
542		- self._message(MessageType.INFO, "Closed workspace for {}".format(element_name))
	543	+ self._context.info("Closed workspace for {}".format(element_name))
543	544
544	545	# workspace_reset
545	546	#
...	...	@@ -580,8 +581,8 @@ class Stream():
580	581	workspace_path = workspace.get_absolute_path()
581	582	if soft:
582	583	workspace.prepared = False
583		- self._message(MessageType.INFO, "Reset workspace state for {} at: {}"
584		- .format(element.name, workspace_path))
	584	+ self._context.info("Reset workspace state for {} at: {}"
	585	+ .format(element.name, workspace.path))
585	586	continue
586	587
587	588	with element.timed_activity("Removing workspace directory {}"
...	...	@@ -598,9 +599,8 @@ class Stream():
598	599	with element.timed_activity("Staging sources to {}".format(workspace_path)):
599	600	element._open_workspace()
600	601
601		- self._message(MessageType.INFO,
602		- "Reset workspace for {} at: {}".format(element.name,
603		- workspace_path))
	602	+ self._context.info("Reset workspace for {} at: {}"
	603	+ .format(element.name, workspace._path))
604	604
605	605	workspaces.save_config()
606	606
...	...	@@ -676,7 +676,7 @@ class Stream():
676	676	# source-bundle only supports one target
677	677	target = self.targets[0]
678	678
679		- self._message(MessageType.INFO, "Bundling sources for target {}".format(target.name))
	679	+ self._context.info("Bundling sources for target {}".format(target.name))
680	680
681	681	# Find the correct filename for the compression algorithm
682	682	tar_location = os.path.join(directory, target.normal_name + ".tar")
...	...	@@ -958,15 +958,6 @@ class Stream():
958	958
959	959	return selected, track_selected
960	960
961		- # _message()
962		- #
963		- # Local message propagator
964		- #
965		- def _message(self, message_type, message, **kwargs):
966		- args = dict(kwargs)
967		- self._context.message(
968		- Message(None, message_type, message, **args))
969		-
970	961	# _add_queue()
971	962	#
972	963	# Adds a queue to the stream
...	...	@@ -1017,10 +1008,11 @@ class Stream():
1017	1008	for element in self.total_elements:
1018	1009	element._update_state()
1019	1010	except BstError as e:
1020		- self._message(MessageType.ERROR, "Error resolving final state", detail=str(e))
	1011	+ self._context.error("Error resolving final state", detail=e)
1021	1012	set_last_task_error(e.domain, e.reason)
1022		- except Exception as e: # pylint: disable=broad-except
1023		- self._message(MessageType.BUG, "Unhandled exception while resolving final state", detail=str(e))
	1013	+ except Exception as e: # pylint: disable=broad-except
	1014	+ self._context.message("Unhandled exception while resolving final state",
	1015	+ detail=traceback.format_exc())
1024	1016
1025	1017	if status == SchedStatus.ERROR:
1026	1018	raise StreamError()

buildstream/plugin.py

---

...	...	@@ -117,7 +117,6 @@ from weakref import WeakValueDictionary
117	117	from . import _yaml
118	118	from . import utils
119	119	from ._exceptions import PluginError, ImplError
120		-from ._message import Message, MessageType
121	120
122	121
123	122	class Plugin():
...	...	@@ -464,8 +463,7 @@ class Plugin():
464	463	brief (str): The brief message
465	464	detail (str): An optional detailed message, can be multiline output
466	465	"""
467		- if self.__context.log_debug:
468		- self.__message(MessageType.DEBUG, brief, detail=detail)
	466	+ self.__context.debug(brief, detail=detail, plugin=self)
469	467
470	468	def status(self, brief, *, detail=None):
471	469	"""Print a status message
...	...	@@ -474,9 +472,9 @@ class Plugin():
474	472	brief (str): The brief message
475	473	detail (str): An optional detailed message, can be multiline output
476	474
477		- Note: Status messages tell about what a plugin is currently doing
	475	+ Note: Status messages tell the user what a plugin is currently doing
478	476	"""
479		- self.__message(MessageType.STATUS, brief, detail=detail)
	477	+ self.__context.status(brief, detail=detail, plugin=self)
480	478
481	479	def info(self, brief, *, detail=None):
482	480	"""Print an informative message
...	...	@@ -488,7 +486,7 @@ class Plugin():
488	486	Note: Informative messages tell the user something they might want
489	487	to know, like if refreshing an element caused it to change.
490	488	"""
491		- self.__message(MessageType.INFO, brief, detail=detail)
	489	+ self.__context.info(brief, detail=detail, plugin=self)
492	490
493	491	def warn(self, brief, *, detail=None, warning_token=None):
494	492	"""Print a warning message, checks warning_token against project configuration
...	...	@@ -512,7 +510,7 @@ class Plugin():
512	510	detail = detail if detail else ""
513	511	raise PluginError(message="{}\n{}".format(brief, detail), reason=warning_token)
514	512
515		- self.__message(MessageType.WARN, brief=brief, detail=detail)
	513	+ self.__context.warn(brief, detail=detail, plugin=self)
516	514
517	515	def log(self, brief, *, detail=None):
518	516	"""Log a message into the plugin's log file
...	...	@@ -524,7 +522,7 @@ class Plugin():
524	522	brief (str): The brief message
525	523	detail (str): An optional detailed message, can be multiline output
526	524	"""
527		- self.__message(MessageType.LOG, brief, detail=detail)
	525	+ self.__context.log(brief, detail=detail, plugin=self)
528	526
529	527	@contextmanager
530	528	def timed_activity(self, activity_name, *, detail=None, silent_nested=False):
...	...	@@ -746,14 +744,9 @@ class Plugin():
746	744
747	745	return (exit_code, output)
748	746
749		- def __message(self, message_type, brief, **kwargs):
750		- message = Message(self.__unique_id, message_type, brief, **kwargs)
751		- self.__context.message(message)
752		-
753	747	def __note_command(self, output, *popenargs, **kwargs):
754		- workdir = os.getcwd()
755		- if 'cwd' in kwargs:
756		- workdir = kwargs['cwd']
	748	+ workdir = kwargs.get("cwd", os.getcwd())
	749	+
757	750	command = " ".join(popenargs[0])
758	751	output.write('Running host command {}: {}\n'.format(workdir, command))
759	752	output.flush()

buildstream/utils.py

---

...	...	@@ -977,6 +977,17 @@ def _tempdir(suffix="", prefix="tmp", dir=None): # pylint: disable=redefined-bu
977	977	cleanup_tempdir()
978	978
979	979
	980	+# _none_context()
	981	+#
	982	+# An empty context, useful for optional contexts e.g.
	983	+#
	984	+# with (_tempdir() if <value> else _none_context())
	985	+#
	986	+@contextmanager
	987	+def _none_context():
	988	+ yield
	989	+
	990	+
980	991	# _kill_process_tree()
981	992	#
982	993	# Brutally murder a process and all of it's children

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