You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2024/08/22 16:24:24 UTC
(beam) 01/01: Move RunInference GC to finalize_bundle
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch users/damccorm/inference_gc
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a476a927a5e37eb9c6cbed1a1a45a9153aad303d
Author: Danny McCormick <da...@google.com>
AuthorDate: Thu Aug 22 17:24:04 2024 +0100
Move RunInference GC to finalize_bundle
---
sdks/python/apache_beam/ml/inference/base.py | 20 ++++++++++----------
1 file changed, 10 insertions(+), 10 deletions(-)
diff --git a/sdks/python/apache_beam/ml/inference/base.py b/sdks/python/apache_beam/ml/inference/base.py
index b075f22f15b..5cc38cd842c 100644
--- a/sdks/python/apache_beam/ml/inference/base.py
+++ b/sdks/python/apache_beam/ml/inference/base.py
@@ -1795,16 +1795,6 @@ class _RunInferenceDoFn(beam.DoFn, Generic[ExampleT, PredictionT]):
side input is empty or the model has not been updated, the method
simply runs inference on the batch of data.
"""
- # Do garbage collection of old models before starting inference.
- tags_to_gc = self._model_metadata.get_tags_for_garbage_collection()
- if len(tags_to_gc) > 0:
- for unprefixed_tag in tags_to_gc:
- for tag in _get_tags_for_copies(unprefixed_tag,
- self._model_handler.model_copies()):
- multi_process_shared.MultiProcessShared(lambda: None,
- tag).unsafe_hard_delete()
- self._model_metadata.mark_tags_deleted(tags_to_gc)
-
if not si_model_metadata:
if (not self._model_metadata.is_valid_tag(self._cur_tag) or
self._model is None):
@@ -1834,6 +1824,16 @@ class _RunInferenceDoFn(beam.DoFn, Generic[ExampleT, PredictionT]):
if self._metrics_collector:
self._metrics_collector.update_metrics_with_cache()
+ # Do garbage collection of old models
+ tags_to_gc = self._model_metadata.get_tags_for_garbage_collection()
+ if len(tags_to_gc) > 0:
+ for unprefixed_tag in tags_to_gc:
+ for tag in _get_tags_for_copies(unprefixed_tag,
+ self._model_handler.model_copies()):
+ multi_process_shared.MultiProcessShared(lambda: None,
+ tag).unsafe_hard_delete()
+ self._model_metadata.mark_tags_deleted(tags_to_gc)
+
def _is_darwin() -> bool:
return sys.platform == 'darwin'