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

[Qinusty <gitlab mg gitlab com>]

Title: GitLab

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

Commits:

• e8defe24
  by Josh Smith at 2018-08-17T16:07:21Z

  Overhaul Message API
  
  _context.py: message API standardisation, all messages should
  come through context. Added skipped() for MessageType.SKIPPED.
  Also added context.report_unhandled_exceptions() to standardise
  catching unhandled exceptions and sending a BUG message.
  
  job.py: Add context attribute for referencing the context during messaging
  calls. Messaging helpers added to use context helpers with task_id kwarg.
  _child_action() has been overhauled to no longer use parts of
  timed_activity, it now takes advantage of timed_activity and helper
  functions.
  
  queue.py: Now takes advantage of context helper functions and uses
  report_unhandled_exceptions instead of MessageType.BUG in _job_done().
  
  Others: Removing redundant imports and changing to use context message
  API.
  

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

buildstream/_artifactcache/cascache.py

---

...	...	@@ -33,7 +33,6 @@ from .._protos.google.bytestream import bytestream_pb2, bytestream_pb2_grpc
33	33	from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
34	34	from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
35	35
36		-from .._message import MessageType, Message
37	36	from .. import _signals, utils
38	37	from .._exceptions import ArtifactError
39	38
...	...	@@ -348,12 +347,10 @@ class CASCache(ArtifactCache):
348	347	raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
349	348
350	349	if skipped_remote:
351		- self.context.message(Message(
352		- None,
353		- MessageType.SKIPPED,
	350	+ self.context.skipped(
354	351	"Remote ({}) already has {} cached".format(
355	352	remote.spec.url, element._get_brief_display_key())
356		- ))
	353	+ )
357	354	return pushed
358	355
359	356	################################################

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
...	...	@@ -401,7 +402,7 @@ class Context():
401	402	return True
402	403	return False
403	404
404		- # message():
	405	+ # _send_message():
405	406	#
406	407	# Proxies a message back to the caller, this is the central
407	408	# point through which all messages pass.
...	...	@@ -438,35 +439,55 @@ class Context():
438	439
439	440	self._send_message(Message(plugin_id, msg_type, str(text), **kwargs))
440	441
441		- def start(self, text, *, plugin=None, **kwargs):
442		- self._message(text, plugin=plugin, msg_type=MessageType.START, **kwargs)
443		-
444		- def success(self, text, *, plugin=None, **kwargs):
445		- self._message(text, plugin=plugin, msg_type=MessageType.SUCCESS, **kwargs)
446		-
447		- def failure(self, text, *, plugin=None, **kwargs):
448		- self._message(text, plugin=plugin, msg_type=MessageType.FAIL, **kwargs)
449		-
450		- def debug(self, text, *, plugin=None, **kwargs):
451		- self._message(text, plugin=plugin, msg_type=MessageType.DEBUG, **kwargs)
	442	+ # skipped():
	443	+ #
	444	+ # Produce and send a skipped message through the context.
	445	+ #
	446	+ def skipped(self, text, **kwargs):
	447	+ self._message(text, msg_type=MessageType.SKIPPED, **kwargs)
452	448
453		- def status(self, text, *, plugin=None, **kwargs):
454		- self._message(text, plugin=plugin, msg_type=MessageType.STATUS, **kwargs)
	449	+ # debug():
	450	+ #
	451	+ # Produce and send a debug message through the context.
	452	+ #
	453	+ def debug(self, text, **kwargs):
	454	+ if self.log_debug:
	455	+ self._message(text, msg_type=MessageType.DEBUG, **kwargs)
455	456
456		- def info(self, text, *, plugin=None, **kwargs):
457		- self._message(text, plugin=plugin, msg_type=MessageType.INFO, **kwargs)
	457	+ # status():
	458	+ #
	459	+ # Produce and send a status message through the context.
	460	+ #
	461	+ def status(self, text, **kwargs):
	462	+ self._message(text, msg_type=MessageType.STATUS, **kwargs)
458	463
459		- def warn(self, text, *, plugin=None, **kwargs):
460		- self._message(text, plugin=plugin, msg_type=MessageType.WARN, **kwargs)
	464	+ # info():
	465	+ #
	466	+ # Produce and send a info message through the context.
	467	+ #
	468	+ def info(self, text, **kwargs):
	469	+ self._message(text, msg_type=MessageType.INFO, **kwargs)
461	470
462		- def error(self, text, *, plugin=None, **kwargs):
463		- self._message(text, plugin=plugin, msg_type=MessageType.ERROR, **kwargs)
	471	+ # warn():
	472	+ #
	473	+ # Produce and send a warning message through the context.
	474	+ #
	475	+ def warn(self, text, **kwargs):
	476	+ self._message(text, msg_type=MessageType.WARN, **kwargs)
464	477
465		- def bug(self, text, *, plugin=None, **kwargs):
466		- self._message(text, plugin=plugin, msg_type=MessageType.BUG, **kwargs)
	478	+ # error():
	479	+ #
	480	+ # Produce and send a error message through the context.
	481	+ #
	482	+ def error(self, text, **kwargs):
	483	+ self._message(text, msg_type=MessageType.ERROR, **kwargs)
467	484
468		- def log(self, text, *, plugin=None, **kwargs):
469		- self._message(text, plugin=plugin, msg_type=MessageType.LOG, **kwargs)
	485	+ # log():
	486	+ #
	487	+ # Produce and send a log message through the context.
	488	+ #
	489	+ def log(self, text, **kwargs):
	490	+ self._message(text, msg_type=MessageType.LOG, **kwargs)
470	491
471	492	# silence()
472	493	#
...	...	@@ -483,6 +504,14 @@ class Context():
483	504	finally:
484	505	self._pop_message_depth()
485	506
	507	+ @contextmanager
	508	+ def report_unhandled_exceptions(self, brief="An unhandled exception occured", *, unique_id=None, **kwargs):
	509	+ try:
	510	+ yield
	511	+ except Exception: # pylint: disable=broad-except
	512	+ self._message(brief, plugin=unique_id, detail=traceback.format_exc(),
	513	+ msg_type=MessageType.BUG, **kwargs)
	514	+
486	515	# timed_activity()
487	516	#
488	517	# Context manager for performing timed activities and logging those
...	...	@@ -512,7 +541,7 @@ class Context():
512	541	with _signals.suspendable(stop_time, resume_time):
513	542	try:
514	543	# Push activity depth for status messages
515		- self.start(activity_name, detail=detail, plugin=unique_id)
	544	+ self.__start(activity_name, detail=detail, plugin=unique_id)
516	545	self._push_message_depth(silent_nested)
517	546	yield
518	547
...	...	@@ -521,13 +550,13 @@ class Context():
521	550	# expects an error when there is an error.
522	551	elapsed = datetime.datetime.now() - starttime
523	552	self._pop_message_depth()
524		- self.failure(activity_name, detail=detail, elapsed=elapsed, plugin=unique_id)
	553	+ self.__failure(activity_name, detail=detail, elapsed=elapsed, plugin=unique_id)
525	554	raise
526	555
527	556	elapsed = datetime.datetime.now() - starttime
528	557	self._pop_message_depth()
529		- self.success(activity_name, detail=detail,
530		- elapsed=elapsed, plugin=unique_id)
	558	+ self.__success(activity_name, detail=detail,
	559	+ elapsed=elapsed, plugin=unique_id)
531	560
532	561	# recorded_messages()
533	562	#
...	...	@@ -683,3 +712,16 @@ class Context():
683	712	os.environ['XDG_CONFIG_HOME'] = os.path.expanduser('~/.config')
684	713	if not os.environ.get('XDG_DATA_HOME'):
685	714	os.environ['XDG_DATA_HOME'] = os.path.expanduser('~/.local/share')
	715	+
	716	+ ##########################################################
	717	+ # Local private messaging API, used by timed_activity() #
	718	+ ##########################################################
	719	+
	720	+ def __start(self, text, **kwargs):
	721	+ self._message(text, msg_type=MessageType.START, **kwargs)
	722	+
	723	+ def __success(self, text, **kwargs):
	724	+ self._message(text, msg_type=MessageType.SUCCESS, **kwargs)
	725	+
	726	+ def __failure(self, text, **kwargs):
	727	+ self._message(text, msg_type=MessageType.FAIL, **kwargs)

buildstream/_frontend/app.py

---

...	...	@@ -33,10 +33,11 @@ from .. import Scope
33	33
34	34	# Import various buildstream internals
35	35	from .._context import Context
	36	+from .. import utils
36	37	from .._platform import Platform
37	38	from .._project import Project
38	39	from .._exceptions import BstError, StreamError, LoadError, LoadErrorReason, AppError
39		-from .._message import Message, MessageType, unconditional_messages
	40	+from .._message import MessageType, unconditional_messages
40	41	from .._stream import Stream
41	42	from .._versions import BST_FORMAT_VERSION
42	43	from .. import _yaml
...	...	@@ -251,40 +252,33 @@ class App():
251	252	self._success_profile, self._error_profile,
252	253	self.stream, colors=self.colors)
253	254
254		- # Mark the beginning of the session
255		- if session_name:
256		- self.context.start(session_name)
257		-
258		- # Run the body of the session here, once everything is loaded
259		- try:
260		- yield
261		- except BstError as e:
262		-
263		- # Print a nice summary if this is a session
264		- if session_name:
265		- elapsed = self.stream.elapsed_time
266		-
267		- if isinstance(e, StreamError) and e.terminated: # pylint: disable=no-member
268		- self.context.warn(session_name + ' Terminated', elapsed=elapsed)
269		- else:
270		- self.context.failure(session_name, elapsed=elapsed)
271		-
272		- if self._started:
273		- self._print_summary()
274		-
	255	+ with (self.context.timed_activity(session_name) if session_name else utils._none_context()):
	256	+ error_occurred = None
	257	+ # Run the body of the session here, once everything is loaded
	258	+ try:
	259	+ yield
	260	+ except BstError as e:
	261	+
	262	+ # Print a nice summary if this is a session
	263	+ if session_name:
	264	+ elapsed = self.stream.elapsed_time
	265	+
	266	+ if isinstance(e, StreamError) and e.terminated: # pylint: disable=no-member
	267	+ self.context.warn(session_name + ' Terminated', elapsed=elapsed)
	268	+ else:
	269	+ raise
	270	+ error_occurred = e
	271	+ except RecursionError:
	272	+ click.echo("RecursionError: Depency depth is too large. Maximum recursion depth exceeded.",
	273	+ err=True)
	274	+ sys.exit(-1)
	275	+
	276	+ if self._started and session_name:
	277	+ self._print_summary()
	278	+
	279	+ if error_occurred:
275	280	# Exit with the error
276		- self._error_exit(e)
277		- except RecursionError:
278		- click.echo("RecursionError: Depency depth is too large. Maximum recursion depth exceeded.",
279		- err=True)
280		- sys.exit(-1)
281		-
282		- else:
283		- # No exceptions occurred, print session time and summary
284		- if session_name:
285		- self.context.success(session_name, elapsed=self.stream.elapsed_time)
286		- if self._started:
287		- self._print_summary()
	281	+ self._error_exit(error_occurred)
288	282
289	283	# init_project()
290	284	#

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

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

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
...	...	@@ -492,13 +490,9 @@ class Project():
492	490
493	491	# Deprecation check
494	492	if fail_on_overlap is not None:
495		- self._context.message(
496		- Message(
497		- None,
498		- MessageType.WARN,
499		- "Use of fail-on-overlap within project.conf " +
500		- "is deprecated. Consider using fatal-warnings instead."
501		- )
	493	+ self._context.warn(
	494	+ "Use of fail-on-overlap within project.conf " +
	495	+ "is deprecated. Consider using fatal-warnings instead."
502	496	)
503	497
504	498	# 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)
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

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 = multiprocessing.Queue() # A message passing queue
113	114	self._process = None # The Process object
114	115	self._watcher = None # Child process watcher
...	...	@@ -179,7 +180,7 @@ class Job():
179	180	# First resume the job if it's suspended
180	181	self.resume(silent=True)
181	182
182		- self._status("{} terminating".format(self.action_name), self.element)
	183	+ self._status("{} terminating".format(self.action_name))
183	184
184	185	# Make sure there is no garbage on the queue
185	186	self._parent_stop_listening()
...	...	@@ -211,7 +212,7 @@ class Job():
211	212
212	213	# Force kill
213	214	self._warn("{} did not terminate gracefully, killing"
214		- .format(self.action_name), self.element)
	215	+ .format(self.action_name))
215	216
216	217	try:
217	218	utils._kill_process_tree(self._process.pid)
...	...	@@ -226,7 +227,7 @@ class Job():
226	227	#
227	228	def suspend(self):
228	229	if not self._suspended:
229		- self._status("{} suspending".format(self.action_name), self.element)
	230	+ self._status("{} suspending".format(self.action_name))
230	231
231	232	try:
232	233	# Use SIGTSTP so that child processes may handle and propagate
...	...	@@ -250,7 +251,7 @@ class Job():
250	251	def resume(self, silent=False):
251	252	if self._suspended:
252	253	if not silent:
253		- self._status("{} resuming".format(self.action_name), self.element)
	254	+ self._status("{} resuming".format(self.action_name))
254	255
255	256	os.kill(self._process.pid, signal.SIGCONT)
256	257	self._suspended = False
...	...	@@ -303,26 +304,6 @@ class Job():
303	304	raise ImplError("Job '{kind}' does not implement child_process()"
304	305	.format(kind=type(self).__name__))
305	306
306		- # message():
307		- #
308		- # Logs a message, this will be logged in the task's logfile and
309		- # conditionally also be sent to the frontend.
310		- #
311		- # Args:
312		- # message (str): The message
313		- # message_type (MessageType): The type of message to send
314		- # plugin (Plugin): The plugin sending the message.
315		- # kwargs: Remaining Message() constructor arguments
316		- #
317		- def _message(self, text, plugin, *, msg_type=None, **kwargs):
318		- self._scheduler.context._message(
319		- text,
320		- plugin=plugin._get_unique_id(),
321		- msg_type=msg_type,
322		- scheduler=True,
323		- **kwargs
324		- )
325		-
326	307	# child_process_data()
327	308	#
328	309	# Abstract method to retrieve additional data that should be
...	...	@@ -348,35 +329,24 @@ class Job():
348	329	# Other methods can be called in both child or parent processes
349	330	#
350	331	#######################################################
351		- def _start(self, *args, **kwargs):
352		- self._message(*args, msg_type=MessageType.START, **kwargs)
353		-
354		- def _success(self, *args, **kwargs):
355		- self._message(*args, msg_type=MessageType.SUCCESS, **kwargs)
356	332
357		- def _failure(self, *args, **kwargs):
358		- self._message(*args, msg_type=MessageType.FAIL, **kwargs)
	333	+ def _debug(self, text, **kwargs):
	334	+ self._context.debug(text, task_id=self._task_id, **kwargs)
359	335
360		- def _debug(self, *args, **kwargs):
361		- self._message(*args, msg_type=MessageType.DEBUG, **kwargs)
	336	+ def _status(self, text, **kwargs):
	337	+ self._context.status(text, task_id=self._task_id, **kwargs)
362	338
363		- def _status(self, *args, **kwargs):
364		- self._message(*args, msg_type=MessageType.STATUS, **kwargs)
	339	+ def _info(self, text, **kwargs):
	340	+ self._context.info(text, task_id=self._task_id, **kwargs)
365	341
366		- def _info(self, *args, **kwargs):
367		- self._message(*args, msg_type=MessageType.INFO, **kwargs)
	342	+ def _warn(self, text, **kwargs):
	343	+ self._context.warn(text, task_id=self._task_id, **kwargs)
368	344
369		- def _warn(self, *args, **kwargs):
370		- self._message(*args, msg_type=MessageType.WARN, **kwargs)
	345	+ def _error(self, text, **kwargs):
	346	+ self._context.error(text, task_id=self._task_id, **kwargs)
371	347
372		- def _error(self, *args, **kwargs):
373		- self._message(*args, msg_type=MessageType.ERROR, **kwargs)
374		-
375		- def _bug(self, *args, **kwargs):
376		- self._message(*args, msg_type=MessageType.BUG, **kwargs)
377		-
378		- def _log(self, *args, **kwargs):
379		- self._message(*args, msg_type=MessageType.LOG, **kwargs)
	348	+ def _log(self, text, **kwargs):
	349	+ self._context.log(text, task_id=self._task_id, **kwargs)
380	350
381	351	# _child_action()
382	352	#
...	...	@@ -406,74 +376,40 @@ class Job():
406	376	self._queue = queue
407	377	self._scheduler.context.set_message_handler(self._child_message_handler)
408	378
409		- starttime = datetime.datetime.now()
410		- stopped_time = None
411		-
412		- def stop_time():
413		- nonlocal stopped_time
414		- stopped_time = datetime.datetime.now()
415		-
416		- def resume_time():
417		- nonlocal stopped_time
418		- nonlocal starttime
419		- starttime += (datetime.datetime.now() - stopped_time)
420		-
421	379	# Time, log and and run the action function
422	380	#
423		- with _signals.suspendable(stop_time, resume_time), \
424		- self._scheduler.context.recorded_messages(self._logfile) as filename:
425		-
426		- self._start(self.action_name, element, logfile=filename)
427		-
428		- try:
429		- # Try the task action
430		- result = self.child_process()
431		- except BstError as e:
432		- elapsed = datetime.datetime.now() - starttime
433		- self._retry_flag = e.temporary
434		-
435		- if self._retry_flag and (self._tries <= self._max_retries):
436		- self._failure("Try #{} failed, retrying".format(self._tries),
437		- element, elapsed=elapsed)
	381	+ with self._context.recorded_messages(self._logfile) as filename, \
	382	+ self._context.timed_activity(self.action_name, unique_id=self._task_id):
	383	+
	384	+ returncode = RC_FAIL # Default to RC_FAIL so that the else: statement produces RC_OK
	385	+ with self._context.report_unhandled_exceptions(self.action_name, logfile=filename):
	386	+ try:
	387	+ # Try the task action
	388	+ result = self.child_process()
	389	+ except BstError as e:
	390	+ self._retry_flag = e.temporary
	391	+
	392	+ if self._retry_flag and (self._tries <= self._max_retries):
	393	+ self._failure("Try #{} failed, retrying".format(self._tries), elapsed=elapsed)
	394	+ else:
	395	+ raise
	396	+
	397	+ self._queue.put(Envelope('child_data', self.child_process_data()))
	398	+
	399	+ # Report the exception to the parent (for internal testing purposes)
	400	+ self._child_send_error(e)
	401	+
	402	+ # Set return code based on whether or not the error was temporary.
	403	+ #
	404	+ returncode = RC_FAIL if self._retry_flag else RC_PERM_FAIL
438	405	else:
439		- self._failure(str(e), element, elapsed=elapsed,
440		- detail=e.detail, logfile=filename,
441		- sandbox=e.sandbox)
442		-
443		- self._queue.put(Envelope('child_data', self.child_process_data()))
444		-
445		- # Report the exception to the parent (for internal testing purposes)
446		- self._child_send_error(e)
447		-
448		- # Set return code based on whether or not the error was temporary.
449		- #
450		- self._child_shutdown(RC_FAIL if self._retry_flag else RC_PERM_FAIL)
451		-
452		- except Exception as e: # pylint: disable=broad-except
453		-
454		- # If an unhandled (not normalized to BstError) occurs, that's a bug,
455		- # send the traceback and formatted exception back to the frontend
456		- # and print it to the log file.
457		- #
458		- elapsed = datetime.datetime.now() - starttime
459		- detail = "An unhandled exception occured:\n\n{}".format(traceback.format_exc())
460		-
461		- self._bug(self.action_name, element, elapsed=elapsed,
462		- detail=detail, logfile=filename)
463		- self._child_shutdown(RC_FAIL)
464		- else:
465		- # No exception occurred in the action
466		- self._queue.put(Envelope('child_data', self.child_process_data()))
467		- self._child_send_result(result)
	406	+ # No exception occurred in the action
	407	+ self._queue.put(Envelope('child_data', self.child_process_data()))
	408	+ self._child_send_result(result)
468	409
469		- elapsed = datetime.datetime.now() - starttime
470		- self._success(self.action_name, element,
471		- elapsed=elapsed, logfile=filename)
	410	+ returncode = RC_OK
472	411
473		- # Shutdown needs to stay outside of the above context manager,
474		- # make sure we dont try to handle SIGTERM while the process
475		- # is already busy in sys.exit()
476		- self._child_shutdown(RC_OK)
	412	+ self._child_shutdown(returncode)
477	413
478	414	# _child_send_error()
479	415	#
...	...	@@ -603,7 +539,7 @@ class Job():
603	539	if envelope._message_type == 'message':
604	540	# Propagate received messages from children
605	541	# back through the context.
606		- self._scheduler.context.message(envelope._message)
	542	+ self._scheduler.context._send_message(envelope._message)
607	543	elif envelope._message_type == 'error':
608	544	# For regression tests only, save the last error domain / reason
609	545	# reported from a child task in the main process, this global state

buildstream/_scheduler/queues/buildqueue.py

---

...	...	@@ -23,7 +23,6 @@ from datetime import timedelta
23	23	from . import Queue, QueueStatus
24	24	from ..jobs import ElementJob
25	25	from ..resources import ResourceType
26		-from ..._message import MessageType
27	26
28	27
29	28	# A queue which assembles elements
...	...	@@ -50,7 +49,7 @@ class BuildQueue(Queue):
50	49	self._tried.add(element)
51	50	_, description, detail = element._get_build_result()
52	51	logfile = element._get_build_log()
53		- self._message(element, MessageType.FAIL, description,
	52	+ self._failure(description, plugin=element,
54	53	detail=detail, action_name=self.action_name,
55	54	elapsed=timedelta(seconds=0),
56	55	logfile=logfile)

buildstream/_scheduler/queues/queue.py

---

...	...	@@ -30,7 +30,6 @@ 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
34	33
35	34
36	35	# Queue status for a given element
...	...	@@ -72,6 +71,7 @@ class Queue():
72	71	# Private members
73	72	#
74	73	self._scheduler = scheduler
	74	+ self._context = scheduler.context
75	75	self._wait_queue = deque()
76	76	self._done_queue = deque()
77	77	self._max_retries = 0
...	...	@@ -274,19 +274,17 @@ class Queue():
274	274	# Handle any workspace modifications now
275	275	#
276	276	if workspace_dict:
277		- context = element._get_context()
278		- workspaces = context.get_workspaces()
	277	+ workspaces = self._context.get_workspaces()
279	278	if workspaces.update_workspace(element._get_full_name(), workspace_dict):
280		- try:
281		- workspaces.save_config()
282		- except BstError as e:
283		- element._get_context().error("Error saving workspaces",
284		- detail=str(e),
285		- plugin=element._get_unique_id())
286		- except Exception as e: # pylint: disable=broad-except
287		- self._message(element, MessageType.BUG,
288		- "Unhandled exception while saving workspaces",
289		- detail=traceback.format_exc())
	279	+ unique_id = element._get_unique_id()
	280	+ with self._context.report_unhandled_exceptions("Unhandled exception while saving workspaces",
	281	+ unique_id=unique_id):
	282	+ try:
	283	+ workspaces.save_config()
	284	+ except BstError as e:
	285	+ self._context.error("Error saving workspaces",
	286	+ detail=str(e),
	287	+ plugin=unique_id)
290	288
291	289	# _job_done()
292	290	#
...	...	@@ -306,54 +304,43 @@ class Queue():
306	304	if job.child_data:
307	305	element._get_artifact_cache().cache_size = job.child_data.get('cache_size')
308	306
309		- # Give the result of the job to the Queue implementor,
310		- # and determine if it should be considered as processed
311		- # or skipped.
312		- try:
313		- processed = self.done(job, element, result, success)
314		-
315		- except BstError as e:
316		-
317		- # Report error and mark as failed
318		- #
319		- self._message(element, MessageType.ERROR, "Post processing error", detail=str(e))
320		- self.failed_elements.append(element)
321		-
322		- # Treat this as a task error as it's related to a task
323		- # even though it did not occur in the task context
324		- #
325		- # This just allows us stronger testing capability
326		- #
327		- set_last_task_error(e.domain, e.reason)
328		-
329		- except Exception as e: # pylint: disable=broad-except
330		-
331		- # Report unhandled exceptions and mark as failed
332		- #
333		- self._message(element, MessageType.BUG,
334		- "Unhandled exception in post processing",
335		- detail=traceback.format_exc())
336		- self.failed_elements.append(element)
337		- else:
338		-
339		- # No exception occured, handle the success/failure state in the normal way
340		- #
341		- self._done_queue.append(job)
342		-
343		- if success:
344		- if processed:
345		- self.processed_elements.append(element)
346		- else:
347		- self.skipped_elements.append(element)
348		- else:
	307	+ with self._context.report_unhandled_exceptions("Unhandled exception in post processing",
	308	+ unique_id=element._get_unique_id()):
	309	+ # Give the result of the job to the Queue implementor,
	310	+ # and determine if it should be considered as processed
	311	+ # or skipped.
	312	+ try:
	313	+ processed = self.done(job, element, result, success)
	314	+ except BstError as e:
	315	+ # Report error and mark as failed
	316	+ #
	317	+ self._context.error("Post processing error",
	318	+ plugin=element,
	319	+ detail=str(e))
349	320	self.failed_elements.append(element)
350	321
351		- # Convenience wrapper for Queue implementations to send
352		- # a message for the element they are processing
353		- def _message(self, element, message_type, brief, **kwargs):
354		- context = element._get_context()
355		- context._message(brief, plugin=element._get_unique_id(),
356		- msg_type=message_type, **kwargs)
	322	+ # Treat this as a task error as it's related to a task
	323	+ # even though it did not occur in the task context
	324	+ #
	325	+ # This just allows us stronger testing capability
	326	+ #
	327	+ set_last_task_error(e.domain, e.reason)
	328	+ except Exception: # pylint: disable=broad-except
	329	+ self.failed_elements.append(element)
	330	+ # Intentional reraise for report_unhandled_exceptions() to log.
	331	+ raise
	332	+ else:
	333	+ # No exception occured, handle the success/failure state in the normal way
	334	+ #
	335	+ self._done_queue.append(job)
	336	+
	337	+ if success:
	338	+ if processed:
	339	+ self.processed_elements.append(element)
	340	+ else:
	341	+ self.skipped_elements.append(element)
	342	+ else:
	343	+ self.failed_elements.append(element)
357	344
358	345	def _element_log_path(self, element):
359	346	project = element._get_project()

buildstream/_stream.py

---

...	...	@@ -29,9 +29,9 @@ from contextlib import contextmanager
29	29	from tempfile import TemporaryDirectory
30	30
31	31	from ._exceptions import StreamError, ImplError, BstError, set_last_task_error
32		-from ._message import Message, MessageType
33	32	from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
34	33	from ._pipeline import Pipeline, PipelineSelection
	34	+from ._message import MessageType
35	35	from ._platform import Platform
36	36	from . import utils, _yaml, _site
37	37	from . import Scope, Consistency
...	...	@@ -1001,16 +1001,19 @@ class Stream():
1001	1001
1002	1002	_, status = self._scheduler.run(self.queues)
1003	1003
1004		- # Force update element states after a run, such that the summary
1005		- # is more coherent
1006		- try:
1007		- for element in self.total_elements:
1008		- element._update_state()
1009		- except BstError as e:
1010		- self._context.error("Error resolving final state", detail=str(e))
1011		- set_last_task_error(e.domain, e.reason)
1012		- except Exception as e: # pylint: disable=broad-except
1013		- self._context.bug("Unhandled exception while resolving final state", detail=str(e))
	1004	+ element = None
	1005	+
	1006	+ # Handle un
	1007	+ with self._context.report_unhandled_exceptions("Unhandled exception while resolving final state",
	1008	+ unique_id=element):
	1009	+ # Force update element states after a run, such that the summary
	1010	+ # is more coherent
	1011	+ try:
	1012	+ for element in self.total_elements:
	1013	+ element._update_state()
	1014	+ except BstError as e:
	1015	+ self._context.error("Error resolving final state", detail=str(e))
	1016	+ set_last_task_error(e.domain, e.reason)
1014	1017
1015	1018	if status == SchedStatus.ERROR:
1016	1019	raise StreamError()

buildstream/plugin.py

---

...	...	@@ -462,8 +462,7 @@ class Plugin():
462	462	brief (str): The brief message
463	463	detail (str): An optional detailed message, can be multiline output
464	464	"""
465		- if self.__context.log_debug:
466		- self.__context.debug(brief, detail=detail)
	465	+ self.__context.debug(brief, detail=detail, plugin=self)
467	466
468	467	def status(self, brief, *, detail=None):
469	468	"""Print a status message
...	...	@@ -472,9 +471,9 @@ class Plugin():
472	471	brief (str): The brief message
473	472	detail (str): An optional detailed message, can be multiline output
474	473
475		- Note: Status messages tell about what a plugin is currently doing
	474	+ Note: Status messages tell the user what a plugin is currently doing
476	475	"""
477		- self.__context.status(brief, detail=detail)
	476	+ self.__context.status(brief, detail=detail, plugin=self)
478	477
479	478	def info(self, brief, *, detail=None):
480	479	"""Print an informative message
...	...	@@ -486,7 +485,7 @@ class Plugin():
486	485	Note: Informative messages tell the user something they might want
487	486	to know, like if refreshing an element caused it to change.
488	487	"""
489		- self.__context.info(brief, detail=detail)
	488	+ self.__context.info(brief, detail=detail, plugin=self)
490	489
491	490	def warn(self, brief, *, detail=None, warning_token=None):
492	491	"""Print a warning message, checks warning_token against project configuration
...	...	@@ -509,7 +508,18 @@ class Plugin():
509	508	if project._warning_is_fatal(warning_token):
510	509	raise PluginError(message="{}\n{}".format(brief, detail), reason=warning_token)
511	510
512		- self.__context.warn(brief, detail=detail)
	511	+ self.__context.warn(brief, detail=detail, plugin=self)
	512	+
	513	+ def skipped(self, brief, *, detail=None):
	514	+ """Prints a message indicating that an action has been skipped.
	515	+
	516	+ Args:
	517	+ brief (str): The brief message
	518	+ detail (str): An optional detailed message, can be multiline output
	519	+
	520	+ (*Since 1.4*)
	521	+ """
	522	+ self.__context.skipped(brief, detail=detail, plugin=self)
513	523
514	524	def log(self, brief, *, detail=None):
515	525	"""Log a message into the plugin's log file
...	...	@@ -521,7 +531,7 @@ class Plugin():
521	531	brief (str): The brief message
522	532	detail (str): An optional detailed message, can be multiline output
523	533	"""
524		- self.__context.log(brief, detail=detail)
	534	+ self.__context.log(brief, detail=detail, plugin=self)
525	535
526	536	@contextmanager
527	537	def timed_activity(self, activity_name, *, detail=None, silent_nested=False):
...	...	@@ -717,14 +727,9 @@ class Plugin():
717	727
718	728	return (exit_code, output)
719	729
720		- def __message(self, message_type, brief, **kwargs):
721		- self.__context._message(brief, plugin=self.__unique_id,
722		- msg_type=message_type, **kwargs)
723		-
724	730	def __note_command(self, output, *popenargs, **kwargs):
725		- workdir = os.getcwd()
726		- if 'cwd' in kwargs:
727		- workdir = kwargs['cwd']
	731	+ workdir = kwargs.get("cwd", os.getcwd())
	732	+
728	733	command = " ".join(popenargs[0])
729	734	output.write('Running host command {}: {}\n'.format(workdir, command))
730	735	output.flush()

buildstream/utils.py

---

...	...	@@ -966,6 +966,17 @@ def _tempdir(suffix="", prefix="tmp", dir=None): # pylint: disable=redefined-bu
966	966	cleanup_tempdir()
967	967
968	968
	969	+# _none_context()
	970	+#
	971	+# An empty context, useful for optional contexts e.g.
	972	+#
	973	+# with (_tempdir() if <value> else _none_context())
	974	+#
	975	+@contextmanager
	976	+def _none_context():
	977	+ yield
	978	+
	979	+
969	980	# _kill_process_tree()
970	981	#
971	982	# 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.