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

[Qinusty <gitlab mg gitlab com>]

Title: GitLab

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

Commits:

• 5f094da3
  by Josh Smith at 2018-08-20T21:45:37Z

  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(). report_unhandled_exceptions() has been added as a
  context manager to reduce the amount of places BUG messages are
  generated from.
  
  plugin.py: Added skipped() to the API.
  
  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.
  

• 05ec44c5
  by Josh Smith at 2018-08-20T21:46:19Z

  Refactor App.initialized() to use timed_activity()
  

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

buildstream/_artifactcache/cascache.py

---

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

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
...	...	@@ -318,7 +319,7 @@ class Context():
318	319	# the context.
319	320	#
320	321	# The message handler should have the same signature as
321		- # the message() method
	322	+ # the _send_message() method
322	323	def set_message_handler(self, handler):
323	324	self._message_handler = handler
324	325
...	...	@@ -333,9 +334,9 @@ class Context():
333	334	return True
334	335	return False
335	336
336		- # message():
	337	+ # _send_message():
337	338	#
338		- # Proxies a message back to the caller, this is the central
	339	+ # Proxies a message back through the message handler, this is the central
339	340	# point through which all messages pass.
340	341	#
341	342	# Args:
...	...	@@ -363,6 +364,8 @@ class Context():
363	364	# final message.
364	365	#
365	366	def _message(self, text, *, plugin=None, msg_type=None, **kwargs):
	367	+ assert msg_type is not None
	368	+
366	369	if isinstance(plugin, Plugin):
367	370	plugin_id = plugin._get_unique_id()
368	371	else:
...	...	@@ -370,35 +373,55 @@ class Context():
370	373
371	374	self._send_message(Message(plugin_id, msg_type, str(text), **kwargs))
372	375
373		- def start(self, text, *, plugin=None, **kwargs):
374		- self._message(text, plugin=plugin, msg_type=MessageType.START, **kwargs)
375		-
376		- def success(self, text, *, plugin=None, **kwargs):
377		- self._message(text, plugin=plugin, msg_type=MessageType.SUCCESS, **kwargs)
378		-
379		- def failure(self, text, *, plugin=None, **kwargs):
380		- self._message(text, plugin=plugin, msg_type=MessageType.FAIL, **kwargs)
381		-
382		- def debug(self, text, *, plugin=None, **kwargs):
383		- self._message(text, plugin=plugin, msg_type=MessageType.DEBUG, **kwargs)
	376	+ # skipped():
	377	+ #
	378	+ # Produce and send a skipped message through the context.
	379	+ #
	380	+ def skipped(self, text, **kwargs):
	381	+ self._message(text, msg_type=MessageType.SKIPPED, **kwargs)
384	382
385		- def status(self, text, *, plugin=None, **kwargs):
386		- self._message(text, plugin=plugin, msg_type=MessageType.STATUS, **kwargs)
	383	+ # debug():
	384	+ #
	385	+ # Produce and send a debug message through the context.
	386	+ #
	387	+ def debug(self, text, **kwargs):
	388	+ if self.log_debug:
	389	+ self._message(text, msg_type=MessageType.DEBUG, **kwargs)
387	390
388		- def info(self, text, *, plugin=None, **kwargs):
389		- self._message(text, plugin=plugin, msg_type=MessageType.INFO, **kwargs)
	391	+ # status():
	392	+ #
	393	+ # Produce and send a status message through the context.
	394	+ #
	395	+ def status(self, text, **kwargs):
	396	+ self._message(text, msg_type=MessageType.STATUS, **kwargs)
390	397
391		- def warn(self, text, *, plugin=None, **kwargs):
392		- self._message(text, plugin=plugin, msg_type=MessageType.WARN, **kwargs)
	398	+ # info():
	399	+ #
	400	+ # Produce and send a info message through the context.
	401	+ #
	402	+ def info(self, text, **kwargs):
	403	+ self._message(text, msg_type=MessageType.INFO, **kwargs)
393	404
394		- def error(self, text, *, plugin=None, **kwargs):
395		- self._message(text, plugin=plugin, msg_type=MessageType.ERROR, **kwargs)
	405	+ # warn():
	406	+ #
	407	+ # Produce and send a warning message through the context.
	408	+ #
	409	+ def warn(self, text, **kwargs):
	410	+ self._message(text, msg_type=MessageType.WARN, **kwargs)
396	411
397		- def bug(self, text, *, plugin=None, **kwargs):
398		- self._message(text, plugin=plugin, msg_type=MessageType.BUG, **kwargs)
	412	+ # error():
	413	+ #
	414	+ # Produce and send a error message through the context.
	415	+ #
	416	+ def error(self, text, **kwargs):
	417	+ self._message(text, msg_type=MessageType.ERROR, **kwargs)
399	418
400		- def log(self, text, *, plugin=None, **kwargs):
401		- self._message(text, plugin=plugin, msg_type=MessageType.LOG, **kwargs)
	419	+ # log():
	420	+ #
	421	+ # Produce and send a log message through the context.
	422	+ #
	423	+ def log(self, text, **kwargs):
	424	+ self._message(text, msg_type=MessageType.LOG, **kwargs)
402	425
403	426	# silence()
404	427	#
...	...	@@ -415,6 +438,14 @@ class Context():
415	438	finally:
416	439	self._pop_message_depth()
417	440
	441	+ @contextmanager
	442	+ def report_unhandled_exceptions(self, brief="An unhandled exception occured", *, unique_id=None, **kwargs):
	443	+ try:
	444	+ yield
	445	+ except Exception: # pylint: disable=broad-except
	446	+ self._message(brief, plugin=unique_id, detail=traceback.format_exc(),
	447	+ msg_type=MessageType.BUG, **kwargs)
	448	+
418	449	# timed_activity()
419	450	#
420	451	# Context manager for performing timed activities and logging those
...	...	@@ -444,7 +475,8 @@ class Context():
444	475	with _signals.suspendable(stop_time, resume_time):
445	476	try:
446	477	# Push activity depth for status messages
447		- self.start(activity_name, detail=detail, plugin=unique_id)
	478	+ self._message(activity_name, detail=detail, plugin=unique_id,
	479	+ msg_type=MessageType.START)
448	480	self._push_message_depth(silent_nested)
449	481	yield
450	482
...	...	@@ -453,13 +485,15 @@ class Context():
453	485	# expects an error when there is an error.
454	486	elapsed = datetime.datetime.now() - starttime
455	487	self._pop_message_depth()
456		- self.failure(activity_name, detail=detail, elapsed=elapsed, plugin=unique_id)
	488	+ self._message(activity_name, detail=detail, elapsed=elapsed, plugin=unique_id,
	489	+ msg_type=MessageType.FAIL)
457	490	raise
458	491
459	492	elapsed = datetime.datetime.now() - starttime
460	493	self._pop_message_depth()
461		- self.success(activity_name, detail=detail,
462		- elapsed=elapsed, plugin=unique_id)
	494	+ self._message(activity_name, detail=detail,
	495	+ elapsed=elapsed, plugin=unique_id,
	496	+ msg_type=MessageType.SUCCESS)
463	497
464	498	# recorded_messages()
465	499	#

buildstream/_frontend/app.py

---

...	...	@@ -37,6 +37,7 @@ from .._platform import Platform
37	37	from .._project import Project
38	38	from .._exceptions import BstError, StreamError, LoadError, LoadErrorReason, AppError
39	39	from .._message import Message, MessageType, unconditional_messages
	40	+from .. import utils
40	41	from .._stream import Stream
41	42	from .._versions import BST_FORMAT_VERSION
42	43	from .. import _yaml
...	...	@@ -252,48 +253,38 @@ class App():
252	253	self._content_profile, self._format_profile,
253	254	self._success_profile, self._error_profile,
254	255	self.stream, colors=self.colors)
255		-
256		- # Mark the beginning of the session
257		- if session_name:
258		- self.context.start(session_name)
259		-
260		- # Run the body of the session here, once everything is loaded
	256	+ last_err = None
261	257	try:
262		- yield
263		- except BstError as e:
264		-
265		- # Print a nice summary if this is a session
266		- if session_name:
267		- elapsed = self.stream.elapsed_time
268		-
269		- if isinstance(e, StreamError) and e.terminated: # pylint: disable=no-member
270		- self.context.warn(session_name + ' Terminated', elapsed=elapsed)
271		- else:
272		- self.context.failure(session_name, elapsed=elapsed)
273		-
274		- # Notify session failure
275		- self._notify("{} failed".format(session_name), "{}".format(e))
276		-
277		- if self._started:
278		- self._print_summary()
279		-
280		- # Exit with the error
281		- self._error_exit(e)
282		- except RecursionError:
283		- click.echo("RecursionError: Depency depth is too large. Maximum recursion depth exceeded.",
284		- err=True)
285		- sys.exit(-1)
286		-
287		- else:
288		- # No exceptions occurred, print session time and summary
289		- if session_name:
290		- self.context.success(session_name, elapsed=self.stream.elapsed_time)
291		- if self._started:
292		- self._print_summary()
293		-
	258	+ with (self.context.timed_activity(session_name) if session_name else utils._none_context()):
	259	+ # Run the body of the session here, once everything is loaded
	260	+ try:
	261	+ yield
	262	+ except BstError as e:
	263	+ last_err = e
	264	+ # Check for Stream error on termination
	265	+ if session_name and isinstance(e, StreamError) and e.terminated: # pylint: disable=no-member
	266	+ elapsed = self.stream.elapsed_time
	267	+ self.context.warn(session_name + ' Terminated', elapsed=elapsed)
	268	+ else:
	269	+ raise # Raise to timed_activity for failure.
	270	+ except RecursionError:
	271	+ click.echo("RecursionError: Depency depth is too large. Maximum recursion depth exceeded.",
	272	+ err=True)
	273	+ sys.exit(-1)
	274	+ except BstError as e: # Catch from timed_activity()
	275	+ # Notify session failure
	276	+ self._notify("{} failed".format(session_name), "{}".format(e))
	277	+
	278	+ # No exceptions occurred, print session time and summary
	279	+ if session_name and self._started:
	280	+ self._print_summary()
	281	+ if not last_err:
294	282	# Notify session success
295	283	self._notify("{} succeeded".format(session_name), "")
296	284
	285	+ if last_err:
	286	+ self._error_exit(last_err)
	287	+
297	288	# init_project()
298	289	#
299	290	# Initialize a new BuildStream project, either with the explicitly passed options,
...	...	@@ -439,6 +430,8 @@ class App():
439	430
440	431	# Print the regular BUG message
441	432	formatted = "".join(traceback.format_exception(etype, value, tb))
	433	+ self.context._message(str(value), detail=formatted, msg_type=MessageType.BUG)
	434	+
442	435	# If the scheduler has started, try to terminate all jobs gracefully,
443	436	# otherwise exit immediately.
444	437	if self.stream.running:

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, 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 = 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 and not self._scheduler.terminated:
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,32 @@ 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)
	348	+ def _log(self, text, **kwargs):
	349	+ self._context.log(text, task_id=self._task_id, **kwargs)
374	350
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)
	351	+ # _fail()
	352	+ #
	353	+ # Only exists for sub classes to override and add kwargs to.
	354	+ #
	355	+ def _fail(self, text, **kwargs):
	356	+ self._context._message(text, task_id=self._task_id,
	357	+ msg_type=MessageType.FAIL, **kwargs)
380	358
381	359	# _child_action()
382	360	#
...	...	@@ -404,7 +382,7 @@ class Job():
404	382	# Set the global message handler in this child
405	383	# process to forward messages to the parent process
406	384	self._queue = queue
407		- self._scheduler.context.set_message_handler(self._child_message_handler)
	385	+ self._context.set_message_handler(self._child_message_handler)
408	386
409	387	starttime = datetime.datetime.now()
410	388	stopped_time = None
...	...	@@ -421,9 +399,10 @@ class Job():
421	399	# Time, log and and run the action function
422	400	#
423	401	with _signals.suspendable(stop_time, resume_time), \
424		- self._scheduler.context.recorded_messages(self._logfile) as filename:
	402	+ self._context.recorded_messages(self._logfile) as filename:
425	403
426		- self._start(self.action_name, element, logfile=filename)
	404	+ self._context._message(self.action_name, logfile=filename,
	405	+ msg_type=MessageType.START, task_id=self._task_id)
427	406
428	407	try:
429	408	# Try the task action
...	...	@@ -433,12 +412,12 @@ class Job():
433	412	self._retry_flag = e.temporary
434	413
435	414	if self._retry_flag and (self._tries <= self._max_retries):
436		- self._failure("Try #{} failed, retrying".format(self._tries),
437		- element, elapsed=elapsed)
	415	+ self._fail("Try #{} failed, retrying".format(self._tries),
	416	+ elapsed=elapsed)
438	417	else:
439		- self._failure(str(e), element, elapsed=elapsed,
440		- detail=e.detail, logfile=filename,
441		- sandbox=e.sandbox)
	418	+ self._fail(str(e), elapsed=elapsed,
	419	+ detail=e.detail, logfile=filename,
	420	+ sandbox=e.sandbox)
442	421
443	422	self._queue.put(Envelope('child_data', self.child_process_data()))
444	423
...	...	@@ -458,17 +437,21 @@ class Job():
458	437	elapsed = datetime.datetime.now() - starttime
459	438	detail = "An unhandled exception occured:\n\n{}".format(traceback.format_exc())
460	439
461		- self._bug(self.action_name, element, elapsed=elapsed,
462		- detail=detail, logfile=filename)
	440	+ self._context._message(self.action_name, elapsed=elapsed,
	441	+ detail=detail, logfile=filename,
	442	+ task_id=self._task_id, msg_type=MessageType.BUG)
463	443	self._child_shutdown(RC_FAIL)
	444	+
464	445	else:
465	446	# No exception occurred in the action
466	447	self._queue.put(Envelope('child_data', self.child_process_data()))
467	448	self._child_send_result(result)
468	449
469	450	elapsed = datetime.datetime.now() - starttime
470		- self._success(self.action_name, element,
471		- elapsed=elapsed, logfile=filename)
	451	+ self._context._message(self.action_name,
	452	+ elapsed=elapsed, logfile=filename,
	453	+ msg_type=MessageType.SUCCESS,
	454	+ task_id=self._task_id)
472	455
473	456	# Shutdown needs to stay outside of the above context manager,
474	457	# make sure we dont try to handle SIGTERM while the process
...	...	@@ -603,7 +586,7 @@ class Job():
603	586	if envelope._message_type == 'message':
604	587	# Propagate received messages from children
605	588	# back through the context.
606		- self._scheduler.context.message(envelope._message)
	589	+ self._context._send_message(envelope._message)
607	590	elif envelope._message_type == 'error':
608	591	# For regression tests only, save the last error domain / reason
609	592	# 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,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,7 +29,7 @@ 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
	32	+from ._message import MessageType
33	33	from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
34	34	from ._pipeline import Pipeline, PipelineSelection
35	35	from ._platform import Platform
...	...	@@ -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 unhandled exceptions
	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.