[Notes] [Git][BuildStream/buildstream][tristan/cache-management] 5 commits: _artifactcache.py: Add status messages in cache management operations

[Tristan Van Berkom <gitlab mg gitlab com>]

Title: GitLab

Tristan Van Berkom pushed to branch tristan/cache-management at BuildStream / buildstream

Commits:

• 78340e80
  by Tristan Van Berkom at 2019-01-23T15:19:07Z

  _artifactcache.py: Add status messages in cache management operations
  
  Added some useful status messages when:
  
    * Calculating a new artifact cache usage size
  
    * Starting a cleanup
  
    * Finishing a cleanup
  
  Also enhanced messaging about what was cleaned up so far when
  aborting a cleanup.
  

• f93b0420
  by Tristan Van Berkom at 2019-01-23T15:19:07Z

  _artifactcache.py: Added client progress callback to ArtifactCache.clean()
  

• 83f06d55
  by Tristan Van Berkom at 2019-01-23T15:19:07Z

  _scheduler/jobs/job.py: Allow subclasses to message the frontend
  

• 985647fd
  by Tristan Van Berkom at 2019-01-23T15:19:07Z

  _scheduler/jobs/cleanupjob.py: Update cache size while processing
  
  Updates the known cache size in the main process while the cleanup
  process is ongoing, so that the status indicators update live
  while the cleanup happens.
  

• c07e04ea
  by Tristan Van Berkom at 2019-01-23T15:19:07Z

  _artifactcache.py: Correcting API documenting comment for remove()
  
  This seems to have been copy/pasted from cascache, and
  documents the function to possibly return None if defer_prune
  was specified, but this function does not expose defer_prune.
  

3 changed files:

• buildstream/_artifactcache.py
• buildstream/_scheduler/jobs/cleanupjob.py
• buildstream/_scheduler/jobs/job.py

Changes:

buildstream/_artifactcache.py

---

...	...	@@ -250,11 +250,33 @@ class ArtifactCache():
250	250	#
251	251	# Clean the artifact cache as much as possible.
252	252	#
	253	+ # Args:
	254	+ # progress (callable): A callback to call when a ref is removed
	255	+ #
253	256	# Returns:
254	257	# (int): The size of the cache after having cleaned up
255	258	#
256		- def clean(self):
	259	+ def clean(self, progress=None):
257	260	artifacts = self.list_artifacts()
	261	+ context = self.context
	262	+
	263	+ # Some accumulative statistics
	264	+ removed_ref_count = 0
	265	+ space_saved = 0
	266	+
	267	+ # Start off with an announcement with as much info as possible
	268	+ volume_size, volume_avail = self._get_cache_volume_size()
	269	+ self._message(MessageType.STATUS, "Starting cache cleanup",
	270	+ detail=("Elements required by the current build plan: {}\n" +
	271	+ "User specified quota: {} ({})\n" +
	272	+ "Cache usage: {}\n" +
	273	+ "Cache volume: {} total, {} available")
	274	+ .format(len(self._required_elements),
	275	+ context.config_cache_quota,
	276	+ utils._pretty_size(self._cache_quota_original, dec_places=2),
	277	+ utils._pretty_size(self.get_cache_size(), dec_places=2),
	278	+ utils._pretty_size(volume_size, dec_places=2),
	279	+ utils._pretty_size(volume_avail, dec_places=2)))
258	280
259	281	# Build a set of the cache keys which are required
260	282	# based on the required elements at cleanup time
...	...	@@ -279,11 +301,15 @@ class ArtifactCache():
279	301	# can't remove them, we have to abort the build.
280	302	#
281	303	# FIXME: Asking the user what to do may be neater
	304	+ #
282	305	default_conf = os.path.join(os.environ['XDG_CONFIG_HOME'],
283	306	'buildstream.conf')
284		- detail = ("There is not enough space to complete the build.\n"
285		- "Please increase the cache-quota in {}."
286		- .format(self.context.config_origin or default_conf))
	307	+ detail = ("Aborted after removing {} refs and saving {} disk space.\n\n"
	308	+ "There is not enough space to complete the build.\n"
	309	+ "Please increase the cache-quota in {} and/or make more disk space."
	310	+ .format(removed_ref_count,
	311	+ utils._pretty_size(space_saved, dec_places=2),
	312	+ (context.config_origin or default_conf)))
287	313
288	314	if self.has_quota_exceeded():
289	315	raise ArtifactError("Cache too full. Aborting.",
...	...	@@ -298,10 +324,33 @@ class ArtifactCache():
298	324	# Remove the actual artifact, if it's not required.
299	325	size = self.remove(to_remove)
300	326
	327	+ removed_ref_count += 1
	328	+ space_saved += size
	329	+
	330	+ self._message(MessageType.STATUS,
	331	+ "Freed {: <7} {}".format(
	332	+ utils._pretty_size(size, dec_places=2),
	333	+ to_remove))
	334	+
301	335	# Remove the size from the removed size
302	336	self.set_cache_size(self._cache_size - size)
303	337
304		- # This should be O(1) if implemented correctly
	338	+ # User callback
	339	+ #
	340	+ # Currently this process is fairly slow, but we should
	341	+ # think about throttling this progress() callback if this
	342	+ # becomes too intense.
	343	+ if progress:
	344	+ progress()
	345	+
	346	+ # Informational message about the side effects of the cleanup
	347	+ self._message(MessageType.INFO, "Cleanup completed",
	348	+ detail=("Removed {} refs and saving {} disk space.\n" +
	349	+ "Cache usage is now: {}")
	350	+ .format(removed_ref_count,
	351	+ utils._pretty_size(space_saved, dec_places=2),
	352	+ utils._pretty_size(self.get_cache_size(), dec_places=2)))
	353	+
305	354	return self.get_cache_size()
306	355
307	356	# compute_cache_size()
...	...	@@ -313,7 +362,14 @@ class ArtifactCache():
313	362	# (int): The size of the artifact cache.
314	363	#
315	364	def compute_cache_size(self):
316		- self._cache_size = self.cas.calculate_cache_size()
	365	+ old_cache_size = self._cache_size
	366	+ new_cache_size = self.cas.calculate_cache_size()
	367	+
	368	+ if old_cache_size != new_cache_size:
	369	+ self._cache_size = new_cache_size
	370	+
	371	+ usage = ArtifactCacheUsage(self)
	372	+ self._message(MessageType.STATUS, "Cache usage recomputed: {}".format(usage))
317	373
318	374	return self._cache_size
319	375
...	...	@@ -493,8 +549,7 @@ class ArtifactCache():
493	549	# `ArtifactCache.get_artifact_fullname`)
494	550	#
495	551	# Returns:
496		- # (int|None) The amount of space pruned from the repository in
497		- # Bytes, or None if defer_prune is True
	552	+ # (int): The amount of space recovered in the cache, in bytes
498	553	#
499	554	def remove(self, ref):
500	555

buildstream/_scheduler/jobs/cleanupjob.py

---

...	...	@@ -28,7 +28,20 @@ class CleanupJob(Job):
28	28	self._artifacts = context.artifactcache
29	29
30	30	def child_process(self):
31		- return self._artifacts.clean()
	31	+ def progress():
	32	+ self.send_message('update-cache-size',
	33	+ self._artifacts.get_cache_size())
	34	+ return self._artifacts.clean(progress)
	35	+
	36	+ def handle_message(self, message_type, message):
	37	+
	38	+ # Update the cache size in the main process as we go,
	39	+ # this provides better feedback in the UI.
	40	+ if message_type == 'update-cache-size':
	41	+ self._artifacts.set_cache_size(message)
	42	+ return True
	43	+
	44	+ return False
32	45
33	46	def parent_complete(self, status, result):
34	47	if status == JobStatus.OK:

buildstream/_scheduler/jobs/job.py

---

...	...	@@ -58,10 +58,10 @@ class JobStatus():
58	58
59	59
60	60	# Used to distinguish between status messages and return values
61		-class Envelope():
	61	+class _Envelope():
62	62	def __init__(self, message_type, message):
63		- self._message_type = message_type
64		- self._message = message
	63	+ self.message_type = message_type
	64	+ self.message = message
65	65
66	66
67	67	# Process class that doesn't call waitpid on its own.
...	...	@@ -275,10 +275,37 @@ class Job():
275	275	def set_task_id(self, task_id):
276	276	self._task_id = task_id
277	277
	278	+ # send_message()
	279	+ #
	280	+ # To be called from inside Job.child_process() implementations
	281	+ # to send messages to the main process during processing.
	282	+ #
	283	+ # These messages will be processed by the class's Job.handle_message()
	284	+ # implementation.
	285	+ #
	286	+ def send_message(self, message_type, message):
	287	+ self._queue.put(_Envelope(message_type, message))
	288	+
278	289	#######################################################
279	290	# Abstract Methods #
280	291	#######################################################
281	292
	293	+ # handle_message()
	294	+ #
	295	+ # Handle a custom message. This will be called in the main process in
	296	+ # response to any messages sent to the main proces using the
	297	+ # Job.send_message() API from inside a Job.child_process() implementation
	298	+ #
	299	+ # Args:
	300	+ # message_type (str): A string to identify the message type
	301	+ # message (any): A simple serializable object
	302	+ #
	303	+ # Returns:
	304	+ # (bool): Should return a truthy value if message_type is handled.
	305	+ #
	306	+ def handle_message(self, message_type, message):
	307	+ return False
	308	+
282	309	# parent_complete()
283	310	#
284	311	# This will be executed after the job finishes, and is expected to
...	...	@@ -416,7 +443,7 @@ class Job():
416	443	elapsed=elapsed, detail=e.detail,
417	444	logfile=filename, sandbox=e.sandbox)
418	445
419		- self._queue.put(Envelope('child_data', self.child_process_data()))
	446	+ self._queue.put(_Envelope('child_data', self.child_process_data()))
420	447
421	448	# Report the exception to the parent (for internal testing purposes)
422	449	self._child_send_error(e)
...	...	@@ -442,7 +469,7 @@ class Job():
442	469
443	470	else:
444	471	# No exception occurred in the action
445		- self._queue.put(Envelope('child_data', self.child_process_data()))
	472	+ self._queue.put(_Envelope('child_data', self.child_process_data()))
446	473	self._child_send_result(result)
447	474
448	475	elapsed = datetime.datetime.now() - starttime
...	...	@@ -469,7 +496,7 @@ class Job():
469	496	domain = e.domain
470	497	reason = e.reason
471	498
472		- envelope = Envelope('error', {
	499	+ envelope = _Envelope('error', {
473	500	'domain': domain,
474	501	'reason': reason
475	502	})
...	...	@@ -487,7 +514,7 @@ class Job():
487	514	#
488	515	def _child_send_result(self, result):
489	516	if result is not None:
490		- envelope = Envelope('result', result)
	517	+ envelope = _Envelope('result', result)
491	518	self._queue.put(envelope)
492	519
493	520	# _child_shutdown()
...	...	@@ -524,7 +551,7 @@ class Job():
524	551	if message.message_type == MessageType.LOG:
525	552	return
526	553
527		- self._queue.put(Envelope('message', message))
	554	+ self._queue.put(_Envelope('message', message))
528	555
529	556	# _parent_shutdown()
530	557	#
...	...	@@ -588,24 +615,28 @@ class Job():
588	615	if not self._listening:
589	616	return
590	617
591		- if envelope._message_type == 'message':
	618	+ if envelope.message_type == 'message':
592	619	# Propagate received messages from children
593	620	# back through the context.
594		- self._scheduler.context.message(envelope._message)
595		- elif envelope._message_type == 'error':
	621	+ self._scheduler.context.message(envelope.message)
	622	+ elif envelope.message_type == 'error':
596	623	# For regression tests only, save the last error domain / reason
597	624	# reported from a child task in the main process, this global state
598	625	# is currently managed in _exceptions.py
599		- set_last_task_error(envelope._message['domain'],
600		- envelope._message['reason'])
601		- elif envelope._message_type == 'result':
	626	+ set_last_task_error(envelope.message['domain'],
	627	+ envelope.message['reason'])
	628	+ elif envelope.message_type == 'result':
602	629	assert self._result is None
603		- self._result = envelope._message
604		- elif envelope._message_type == 'child_data':
	630	+ self._result = envelope.message
	631	+ elif envelope.message_type == 'child_data':
605	632	# If we retry a job, we assign a new value to this
606		- self.child_data = envelope._message
607		- else:
608		- raise Exception()
	633	+ self.child_data = envelope.message
	634	+
	635	+ # Try Job subclass specific messages now
	636	+ elif not self.handle_message(envelope.message_type,
	637	+ envelope.message):
	638	+ assert 0, "Unhandled message type '{}': {}" \
	639	+ .format(envelope.message_type, envelope.message)
609	640
610	641	# _parent_process_queue()
611	642	#

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