From 36ccce5e24691014c54ef6e0721291b1d762087b Mon Sep 17 00:00:00 2001 From: Ville Tuulos Date: Sat, 5 Aug 2023 23:28:10 -0700 Subject: [PATCH 1/9] [current.card.refresh] refactor cards to support runtime updates --- metaflow/plugins/cards/card_cli.py | 100 +++++++++++++--- metaflow/plugins/cards/card_datastore.py | 40 +++++-- metaflow/plugins/cards/card_decorator.py | 107 ++++++++++++++---- metaflow/plugins/cards/card_modules/card.py | 12 +- .../plugins/cards/component_serializer.py | 77 ++++++++++++- 5 files changed, 274 insertions(+), 62 deletions(-) diff --git a/metaflow/plugins/cards/card_cli.py b/metaflow/plugins/cards/card_cli.py index 2c0c8017744..ce2509b0d6f 100644 --- a/metaflow/plugins/cards/card_cli.py +++ b/metaflow/plugins/cards/card_cli.py @@ -6,7 +6,9 @@ from metaflow._vendor import click import os import json +import uuid import signal +import inspect import random from contextlib import contextmanager from functools import wraps @@ -375,14 +377,20 @@ def wrapper(*args, **kwargs): return wrapper -def render_card(mf_card, task, timeout_value=None): - rendered_info = None +def update_card(mf_card, mode, task, data, timeout_value=None): + def _call(): + # compatibility with old render()-method that doesn't accept the data arg + new_render = "data" in inspect.getfullargspec(mf_card.render).args + if mode == "render" and not new_render: + return mf_card.render(task) + else: + return getattr(mf_card, mode)(task, data=data) + if timeout_value is None or timeout_value < 0: - rendered_info = mf_card.render(task) + return _call() else: with timeout(timeout_value): - rendered_info = mf_card.render(task) - return rendered_info + return _call() @card.command(help="create a HTML card") @@ -414,29 +422,61 @@ def render_card(mf_card, task, timeout_value=None): is_flag=True, help="Upon failing to render a card, render a card holding the stack trace", ) +@click.option( + "--id", + default=None, + show_default=True, + type=str, + help="ID of the card", +) @click.option( "--component-file", default=None, show_default=True, type=str, - help="JSON File with Pre-rendered components.(internal)", + help="JSON File with Pre-rendered components. (internal)", ) @click.option( - "--id", + "--mode", + default="render", + show_default=True, + type=str, + help="Rendering mode. (internal)", +) +@click.option( + "--data-file", default=None, show_default=True, type=str, - help="ID of the card", + help="JSON file containing data to be updated. (internal)", +) +@click.option( + "--card-uuid", + default=None, + show_default=True, + type=str, + help="Card UUID. (internal)", +) +@click.option( + "--delete-input-files", + default=False, + is_flag=True, + show_default=True, + help="Delete data-file and compontent-file after reading. (internal)", ) @click.pass_context def create( ctx, pathspec, + mode=None, type=None, options=None, timeout=None, component_file=None, + data_file=None, render_error_card=False, + card_uuid=None, + delete_input_files=None, id=None, ): card_id = id @@ -452,11 +492,26 @@ def create( graph_dict, _ = ctx.obj.graph.output_steps() + if card_uuid is None: + card_uuid = str(uuid.uuid4()).replace("-", "") + # Components are rendered in a Step and added via `current.card.append` are added here. component_arr = [] if component_file is not None: with open(component_file, "r") as f: component_arr = json.load(f) + # data is passed in as temporary files which can be deleted after use + if delete_input_files: + os.remove(component_file) + + # Load data to be refreshed for runtime cards + data = {} + if data_file is not None: + with open(data_file, "r") as f: + data = json.load(f) + # data is passed in as temporary files which can be deleted after use + if delete_input_files: + os.remove(data_file) task = Task(full_pathspec) from metaflow.plugins import CARDS @@ -500,7 +555,9 @@ def create( if mf_card: try: - rendered_info = render_card(mf_card, task, timeout_value=timeout) + rendered_info = update_card( + mf_card, mode, task, data, timeout_value=timeout + ) except: if render_error_card: error_stack_trace = str(UnrenderableCardException(type, options)) @@ -508,10 +565,10 @@ def create( raise UnrenderableCardException(type, options) # - if error_stack_trace is not None: + if error_stack_trace is not None and mode != "refresh": rendered_info = error_card().render(task, stack_trace=error_stack_trace) - if rendered_info is None and render_error_card: + if rendered_info is None and render_error_card and mode != "refresh": rendered_info = error_card().render( task, stack_trace="No information rendered From card of type %s" % type ) @@ -532,12 +589,20 @@ def create( card_id = None if rendered_info is not None: - card_info = card_datastore.save_card(save_type, rendered_info, card_id=card_id) - ctx.obj.echo( - "Card created with type: %s and hash: %s" - % (card_info.type, card_info.hash[:NUM_SHORT_HASH_CHARS]), - fg="green", - ) + if mode == "refresh": + card_datastore.save_data( + card_uuid, save_type, rendered_info, card_id=card_id + ) + ctx.obj.echo("Data updated", fg="green") + else: + card_info = card_datastore.save_card( + card_uuid, save_type, rendered_info, card_id=card_id + ) + ctx.obj.echo( + "Card created with type: %s and hash: %s" + % (card_info.type, card_info.hash[:NUM_SHORT_HASH_CHARS]), + fg="green", + ) @card.command() @@ -655,7 +720,6 @@ def list( as_json=False, file=None, ): - card_id = id if pathspec is None: list_many_cards( diff --git a/metaflow/plugins/cards/card_datastore.py b/metaflow/plugins/cards/card_datastore.py index 59931592878..42ec99ca6ab 100644 --- a/metaflow/plugins/cards/card_datastore.py +++ b/metaflow/plugins/cards/card_datastore.py @@ -6,6 +6,7 @@ from hashlib import sha1 from io import BytesIO import os +import json import shutil from metaflow.plugins.datastores.local_storage import LocalStorage @@ -88,15 +89,15 @@ def __init__(self, flow_datastore, pathspec=None): self._temp_card_save_path = self._get_write_path(base_pth=TEMP_DIR_NAME) @classmethod - def get_card_location(cls, base_path, card_name, card_html, card_id=None): - chash = sha1(bytes(card_html, "utf-8")).hexdigest() + def get_card_location(cls, base_path, card_name, uuid, card_id=None, suffix="html"): + chash = uuid if card_id is None: - card_file_name = "%s-%s.html" % (card_name, chash) + card_file_name = "%s-%s.%s" % (card_name, chash, suffix) else: - card_file_name = "%s-%s-%s.html" % (card_name, card_id, chash) + card_file_name = "%s-%s-%s.%s" % (card_name, card_id, chash, suffix) return os.path.join(base_path, card_file_name) - def _make_path(self, base_pth, pathspec=None, with_steps=False): + def _make_path(self, base_pth, pathspec=None, with_steps=False, suffix="cards"): sysroot = base_pth if pathspec is not None: # since most cards are at a task level there will always be 4 non-none values returned @@ -121,7 +122,7 @@ def _make_path(self, base_pth, pathspec=None, with_steps=False): step_name, "tasks", task_id, - "cards", + suffix, ] else: pth_arr = [ @@ -131,14 +132,16 @@ def _make_path(self, base_pth, pathspec=None, with_steps=False): run_id, "tasks", task_id, - "cards", + suffix, ] if sysroot == "" or sysroot is None: pth_arr.pop(0) return os.path.join(*pth_arr) - def _get_write_path(self, base_pth=""): - return self._make_path(base_pth, pathspec=self._pathspec, with_steps=True) + def _get_write_path(self, base_pth="", suffix="cards"): + return self._make_path( + base_pth, pathspec=self._pathspec, with_steps=True, suffix=suffix + ) def _get_read_path(self, base_pth="", with_steps=False): return self._make_path(base_pth, pathspec=self._pathspec, with_steps=with_steps) @@ -173,7 +176,20 @@ def card_info_from_path(path): card_hash = card_hash.split(".html")[0] return CardInfo(card_type, card_hash, card_id, card_file_name) - def save_card(self, card_type, card_html, card_id=None, overwrite=True): + def save_data(self, uuid, card_type, json_data, card_id=None): + card_file_name = card_type + loc = self.get_card_location( + self._get_write_path(suffix="runtime"), + card_file_name, + uuid, + card_id=card_id, + suffix="data.json", + ) + self._backend.save_bytes( + [(loc, BytesIO(json.dumps(json_data).encode("utf-8")))], overwrite=True + ) + + def save_card(self, uuid, card_type, card_html, card_id=None, overwrite=True): card_file_name = card_type # TEMPORARY_WORKAROUND: FIXME (LATER) : Fix the duplication of below block in a few months. # Check file blame to understand the age of this temporary workaround. @@ -193,7 +209,7 @@ def save_card(self, card_type, card_html, card_id=None, overwrite=True): # It will also easily end up breaking the metaflow-ui (which maybe using a client from an older version). # Hence, we are writing cards to both paths so that we can introduce breaking changes later in the future. card_path_with_steps = self.get_card_location( - self._get_write_path(), card_file_name, card_html, card_id=card_id + self._get_write_path(), card_file_name, uuid, card_id=card_id ) if SKIP_CARD_DUALWRITE: self._backend.save_bytes( @@ -204,7 +220,7 @@ def save_card(self, card_type, card_html, card_id=None, overwrite=True): card_path_without_steps = self.get_card_location( self._get_read_path(with_steps=False), card_file_name, - card_html, + uuid, card_id=card_id, ) for cp in [card_path_with_steps, card_path_without_steps]: diff --git a/metaflow/plugins/cards/card_decorator.py b/metaflow/plugins/cards/card_decorator.py index efa13a1ec90..c629d260c00 100644 --- a/metaflow/plugins/cards/card_decorator.py +++ b/metaflow/plugins/cards/card_decorator.py @@ -2,6 +2,7 @@ import os import tempfile import sys +import time import json from typing import Dict, Any @@ -17,6 +18,8 @@ from .exception import CARD_ID_PATTERN, TYPE_CHECK_REGEX +ASYNC_TIMEOUT = 30 + def warning_message(message, logger=None, ts=False): msg = "[@card WARNING] %s" % message @@ -69,6 +72,7 @@ def __init__(self, *args, **kwargs): self._is_editable = False self._card_uuid = None self._user_set_card_id = None + self._async_proc = None def _is_event_registered(self, evt_name): return evt_name in self._called_once @@ -89,7 +93,6 @@ def _increment_step_counter(cls): def step_init( self, flow, graph, step_name, decorators, environment, flow_datastore, logger ): - self._flow_datastore = flow_datastore self._environment = environment self._logger = logger @@ -131,6 +134,8 @@ def task_pre_step( if card_class is not None: # Card type was not found if card_class.ALLOW_USER_COMPONENTS: self._is_editable = True + self._is_runtime_card = card_class.IS_RUNTIME_CARD + # We have a step counter to ensure that on calling the final card decorator's `task_pre_step` # we call a `finalize` function in the `CardComponentCollector`. # This can help ensure the behaviour of the `current.card` object is according to specification. @@ -155,7 +160,9 @@ def task_pre_step( # we need to ensure that `current.card` has `CardComponentCollector` instantiated only once. if not self._is_event_registered("pre-step"): self._register_event("pre-step") - current._update_env({"card": CardComponentCollector(self._logger)}) + current._update_env( + {"card": CardComponentCollector(self._logger, self._card_proc)} + ) # this line happens because of decospecs parsing. customize = False @@ -184,9 +191,22 @@ def task_finished( ): if not is_task_ok: return - component_strings = current.card._serialize_components(self._card_uuid) + return self._card_proc("render") + + def _card_proc(self, mode): + if mode != "render" and not self._is_runtime_card: + # silently ignore runtime updates for cards that don't support them + return + elif mode == "refresh": + # don't serialize components, which can be a somewhat expensive operation, + # if we are just updating data + component_strings = [] + else: + component_strings = current.card._serialize_components(self._card_uuid) + + data = current.card._get_latest_data(self._card_uuid) runspec = "/".join([current.run_id, current.step_name, current.task_id]) - self._run_cards_subprocess(runspec, component_strings) + self._run_cards_subprocess(mode, runspec, component_strings, data) @staticmethod def _options(mapping): @@ -200,7 +220,6 @@ def _options(mapping): yield to_unicode(value) def _create_top_level_args(self): - top_level_options = { "quiet": True, "metadata": self._metadata.TYPE, @@ -215,12 +234,23 @@ def _create_top_level_args(self): } return list(self._options(top_level_options)) - def _run_cards_subprocess(self, runspec, component_strings): - temp_file = None + def _run_cards_subprocess(self, mode, runspec, component_strings, data=None): + components_file = data_file = None + wait = mode == "render" + if len(component_strings) > 0: - temp_file = tempfile.NamedTemporaryFile("w", suffix=".json") - json.dump(component_strings, temp_file) - temp_file.seek(0) + # note that we can't delete temporary files here when calling the subprocess + # async due to a race condition. The subprocess must delete them + components_file = tempfile.NamedTemporaryFile( + "w", suffix=".json", delete=False + ) + json.dump(component_strings, components_file) + compotents_file.seek(0) + if data is not None: + data_file = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + json.dump(data, data_file) + data_file.seek(0) + executable = sys.executable cmd = [ executable, @@ -230,6 +260,11 @@ def _run_cards_subprocess(self, runspec, component_strings): "card", "create", runspec, + "--delete-input-files", + "--card-uuid", + self._card_uuid, + "--mode", + mode, "--type", self.attributes["type"], # Add the options relating to card arguments. @@ -248,11 +283,14 @@ def _run_cards_subprocess(self, runspec, component_strings): if self.attributes["save_errors"]: cmd += ["--render-error-card"] - if temp_file is not None: - cmd += ["--component-file", temp_file.name] + if components_file is not None: + cmd += ["--component-file", components_file.name] + + if data_file is not None: + cmd += ["--data-file", data_file.name] response, fail = self._run_command( - cmd, os.environ, timeout=self.attributes["timeout"] + cmd, os.environ, timeout=self.attributes["timeout"], wait=wait ) if fail: resp = "" if response is None else response.decode("utf-8") @@ -262,19 +300,38 @@ def _run_cards_subprocess(self, runspec, component_strings): bad=True, ) - def _run_command(self, cmd, env, timeout=None): + def _run_command(self, cmd, env, wait=True, timeout=None): fail = False timeout_args = {} + async_timeout = ASYNC_TIMEOUT if timeout is not None: + async_timeout = int(timeout) + 10 timeout_args = dict(timeout=int(timeout) + 10) - try: - rep = subprocess.check_output( - cmd, env=env, stderr=subprocess.STDOUT, **timeout_args - ) - except subprocess.CalledProcessError as e: - rep = e.output - fail = True - except subprocess.TimeoutExpired as e: - rep = e.output - fail = True - return rep, fail + + if wait: + try: + rep = subprocess.check_output( + cmd, env=env, stderr=subprocess.STDOUT, **timeout_args + ) + except subprocess.CalledProcessError as e: + rep = e.output + fail = True + except subprocess.TimeoutExpired as e: + rep = e.output + fail = True + return rep, fail + else: + if self._async_proc and self._async_proc.poll() is None: + if time.time() - self._async_started > async_timeout: + self._async_proc.kill() + else: + # silently refuse to run an async process if a previous one is still running + # and timeout hasn't been reached + return "", False + else: + #print("CARD CMD", " ".join(cmd)) + self._async_proc = subprocess.Popen( + cmd, env=env, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL + ) + self._async_started = time.time() + return "", False diff --git a/metaflow/plugins/cards/card_modules/card.py b/metaflow/plugins/cards/card_modules/card.py index 35a639299dd..a011d4e4f88 100644 --- a/metaflow/plugins/cards/card_modules/card.py +++ b/metaflow/plugins/cards/card_modules/card.py @@ -35,6 +35,7 @@ class MetaflowCard(object): type = None ALLOW_USER_COMPONENTS = False + IS_RUNTIME_CARD = False scope = "task" # can be task | run @@ -49,7 +50,8 @@ def _get_mustache(self): except ImportError: return None - def render(self, task) -> str: + # FIXME document data + def render(self, task, data=None) -> str: """ Produce custom card contents in HTML. @@ -68,6 +70,14 @@ def render(self, task) -> str: """ return NotImplementedError() + # FIXME document + def render_runtime(self, task, data): + return + + # FIXME document + def refresh(self, task, data): + return + class MetaflowCardComponent(object): def render(self): diff --git a/metaflow/plugins/cards/component_serializer.py b/metaflow/plugins/cards/component_serializer.py index 9451cd67406..afcc8bd1def 100644 --- a/metaflow/plugins/cards/component_serializer.py +++ b/metaflow/plugins/cards/component_serializer.py @@ -3,9 +3,14 @@ from .card_modules.components import UserComponent import uuid import json +import time _TYPE = type +# TODO move these to config +RUNTIME_CARD_MIN_REFRESH_INTERVAL = 5 +RUNTIME_CARD_RENDER_INTERVAL = 60 + def get_card_class(card_type): from metaflow.plugins import CARDS @@ -21,6 +26,51 @@ def __init__(self, warning_message): super().__init__("@card WARNING", warning_message) +class CardComponents: + def __init__(self, card_proc, components=None): + self._card_proc = card_proc + self._latest_user_data = None + self._last_refresh = 0 + self._last_render = 0 + + if components is None: + self._components = [] + else: + self._components = list(components) + + def append(self, component): + self._components.append(component) + + def extend(self, components): + self._components.extend(components) + + def clear(self): + self._components.clear() + + def refresh(self, data=None, force=False): + # todo make this a configurable variable + self._latest_user_data = data + nu = time.time() + if nu - self._last_refresh < RUNTIME_CARD_MIN_REFRESH_INTERVAL: + # rate limit refreshes: silently ignore requests that + # happen too frequently + return + self._last_refresh = nu + # FIXME force render if components have changed + if force or nu - self._last_render > RUNTIME_CARD_RENDER_INTERVAL: + self._card_proc("render_runtime") + self._last_render = nu + else: + self._card_proc("refresh") + + def _get_latest_data(self): + # FIXME add component data + return {"user": self._latest_user_data, "components": []} + + def __iter__(self): + return iter(self._components) + + class CardComponentCollector: """ This class helps collect `MetaflowCardComponent`s during runtime execution @@ -42,17 +92,18 @@ class CardComponentCollector: - [x] by looking it up by its type, e.g. `current.card.get(type='pytorch')`. """ - def __init__(self, logger=None): + def __init__(self, logger=None, card_proc=None): from metaflow.metaflow_config import CARD_NO_WARNING self._cards_components = ( {} - ) # a dict with key as uuid and value as a list of MetaflowCardComponent. + ) # a dict with key as uuid and value as CardComponents, holding a list of MetaflowCardComponents. self._cards_meta = ( {} ) # a `dict` of (card_uuid, `dict)` holding all metadata about all @card decorators on the `current` @step. self._card_id_map = {} # card_id to uuid map for all cards with ids self._logger = logger + self._card_proc = card_proc # `self._default_editable_card` holds the uuid of the card that is default editable. This card has access to `append`/`extend` methods of `self` self._default_editable_card = None self._warned_once = {"__getitem__": {}, "append": False, "extend": False} @@ -60,7 +111,7 @@ def __init__(self, logger=None): @staticmethod def create_uuid(): - return str(uuid.uuid4()) + return str(uuid.uuid4()).replace("-", "") def _log(self, *args, **kwargs): if self._logger: @@ -97,7 +148,7 @@ def _add_card( suppress_warnings=suppress_warnings, ) self._cards_meta[card_uuid] = card_metadata - self._cards_components[card_uuid] = [] + self._cards_components[card_uuid] = CardComponents(self._card_proc) return card_metadata def _warning(self, message): @@ -229,7 +280,7 @@ def __getitem__(self, key): Returns ------- - CardComponentCollector + CardComponents An object with `append` and `extend` calls which allow you to add components to the chosen card. """ @@ -275,7 +326,7 @@ def __setitem__(self, key, value): ) self._warning(_warning_msg) return - self._cards_components[card_uuid] = value + self._cards_components[card_uuid] = CardComponents(self._card_proc, value) return self._warning( @@ -359,6 +410,20 @@ def extend(self, components): self._cards_components[self._default_editable_card].extend(components) + def clear(self): + if self._default_editable_card is not None: + self._cards_components[self._default_editable_card].clear() + + def refresh(self, *args, **kwargs): + if self._default_editable_card is not None: + self._cards_components[self._default_editable_card].refresh(*args, **kwargs) + + def _get_latest_data(self, card_uuid): + """ + Returns latest data so it can be used in the final render() call + """ + return self._cards_components[card_uuid]._get_latest_data() + def _serialize_components(self, card_uuid): """ This method renders components present in a card to strings/json. From 256601941d2813e5271f6088e848920f70fcd8bd Mon Sep 17 00:00:00 2001 From: Ville Tuulos Date: Sun, 6 Aug 2023 13:15:20 -0700 Subject: [PATCH 2/9] [card-refresh] implement card reload policy --- metaflow/plugins/cards/card_cli.py | 33 ++++++++++++++++--- metaflow/plugins/cards/card_decorator.py | 16 +++++---- metaflow/plugins/cards/card_modules/card.py | 28 ++++++++++++++++ .../plugins/cards/component_serializer.py | 13 +++++--- 4 files changed, 74 insertions(+), 16 deletions(-) diff --git a/metaflow/plugins/cards/card_cli.py b/metaflow/plugins/cards/card_cli.py index ce2509b0d6f..d9f9867ddec 100644 --- a/metaflow/plugins/cards/card_cli.py +++ b/metaflow/plugins/cards/card_cli.py @@ -378,13 +378,38 @@ def wrapper(*args, **kwargs): def update_card(mf_card, mode, task, data, timeout_value=None): + def _reload_token(): + if data["render_seq"] == "final": + # final data update should always trigger a card reload to show + # the final card, hence a different token for the final update + return "final" + elif mf_card.RELOAD_POLICY == mf_card.RELOAD_POLICY_ALWAYS: + return "render-seq-%s" % data["render_seq"] + elif mf_card.RELOAD_POLICY == mf_card.RELOAD_POLICY_NEVER: + return "never" + elif mf_card.RELOAD_POLICY == mf_card.RELOAD_POLICY_ONCHANGE: + return mf_card.reload_content_token(task, data) + + def _add_token_html(html): + if html is not None: + return html.replace(mf_card.RELOAD_POLICY_TOKEN, _reload_token()) + + def _add_token_json(json_msg): + if json_msg is not None: + return {"reload_token": _reload_token(), "data": json_msg} + def _call(): # compatibility with old render()-method that doesn't accept the data arg new_render = "data" in inspect.getfullargspec(mf_card.render).args - if mode == "render" and not new_render: - return mf_card.render(task) - else: - return getattr(mf_card, mode)(task, data=data) + if mode == "render": + if new_render: + return _add_token_html(mf_card.render(task, data)) + else: + return _add_token_html(mf_card.render(task)) + elif mode == "render_runtime": + return _add_token_html(mf_card.render_runtime(task, data)) + elif mode == "refresh": + return _add_token_json(mf_card.refresh(task, data)) if timeout_value is None or timeout_value < 0: return _call() diff --git a/metaflow/plugins/cards/card_decorator.py b/metaflow/plugins/cards/card_decorator.py index c629d260c00..dbeb83522a5 100644 --- a/metaflow/plugins/cards/card_decorator.py +++ b/metaflow/plugins/cards/card_decorator.py @@ -131,6 +131,8 @@ def task_pre_step( ): card_type = self.attributes["type"] card_class = get_card_class(card_type) + + self._is_runtime_card = False if card_class is not None: # Card type was not found if card_class.ALLOW_USER_COMPONENTS: self._is_editable = True @@ -189,11 +191,11 @@ def task_pre_step( def task_finished( self, step_name, flow, graph, is_task_ok, retry_count, max_user_code_retries ): - if not is_task_ok: - return - return self._card_proc("render") + if is_task_ok: + self._card_proc("render") + self._card_proc("refresh", final=True) - def _card_proc(self, mode): + def _card_proc(self, mode, final=False): if mode != "render" and not self._is_runtime_card: # silently ignore runtime updates for cards that don't support them return @@ -204,7 +206,7 @@ def _card_proc(self, mode): else: component_strings = current.card._serialize_components(self._card_uuid) - data = current.card._get_latest_data(self._card_uuid) + data = current.card._get_latest_data(self._card_uuid, final=final) runspec = "/".join([current.run_id, current.step_name, current.task_id]) self._run_cards_subprocess(mode, runspec, component_strings, data) @@ -245,7 +247,7 @@ def _run_cards_subprocess(self, mode, runspec, component_strings, data=None): "w", suffix=".json", delete=False ) json.dump(component_strings, components_file) - compotents_file.seek(0) + components_file.seek(0) if data is not None: data_file = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) json.dump(data, data_file) @@ -329,7 +331,7 @@ def _run_command(self, cmd, env, wait=True, timeout=None): # and timeout hasn't been reached return "", False else: - #print("CARD CMD", " ".join(cmd)) + # print("CARD CMD", " ".join(cmd)) self._async_proc = subprocess.Popen( cmd, env=env, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL ) diff --git a/metaflow/plugins/cards/card_modules/card.py b/metaflow/plugins/cards/card_modules/card.py index a011d4e4f88..c305c8fdd3d 100644 --- a/metaflow/plugins/cards/card_modules/card.py +++ b/metaflow/plugins/cards/card_modules/card.py @@ -32,10 +32,34 @@ class MetaflowCard(object): JSON-encodable dictionary containing user-definable options for the class. """ + # RELOAD_POLICY determines whether UIs should + # reload intermediate cards produced by render_runtime + # or whether they can just rely on data updates + + # the UI may keep using the same card + # until the final card is produced + RELOAD_POLICY_NEVER = "never" + + # the UI should reload card every time + # render_runtime() has produced a new card + RELOAD_POLICY_ALWAYS = "always" + + # derive reload token from data and component + # content - force reload only when the content + # changes. The actual policy is card-specific, + # defined by the method reload_content_token() + RELOAD_POLICY_ONCHANGE = "onchange" + + # this token will get replaced in the html with a unique + # string that is used to ensure that data updates and the + # card content matches + RELOAD_POLICY_TOKEN = "[METAFLOW_RELOAD_TOKEN]" + type = None ALLOW_USER_COMPONENTS = False IS_RUNTIME_CARD = False + RELOAD_POLICY = RELOAD_POLICY_NEVER scope = "task" # can be task | run @@ -78,6 +102,10 @@ def render_runtime(self, task, data): def refresh(self, task, data): return + # FIXME document + def reload_content_token(self, task, data): + return "content-token" + class MetaflowCardComponent(object): def render(self): diff --git a/metaflow/plugins/cards/component_serializer.py b/metaflow/plugins/cards/component_serializer.py index afcc8bd1def..7cd5e3e93d5 100644 --- a/metaflow/plugins/cards/component_serializer.py +++ b/metaflow/plugins/cards/component_serializer.py @@ -32,6 +32,7 @@ def __init__(self, card_proc, components=None): self._latest_user_data = None self._last_refresh = 0 self._last_render = 0 + self._render_seq = 0 if components is None: self._components = [] @@ -58,14 +59,16 @@ def refresh(self, data=None, force=False): self._last_refresh = nu # FIXME force render if components have changed if force or nu - self._last_render > RUNTIME_CARD_RENDER_INTERVAL: - self._card_proc("render_runtime") + self._render_seq += 1 self._last_render = nu + self._card_proc("render_runtime") else: self._card_proc("refresh") - def _get_latest_data(self): + def _get_latest_data(self, final=False): # FIXME add component data - return {"user": self._latest_user_data, "components": []} + seq = 'final' if final else self._render_seq + return {"user": self._latest_user_data, "components": [], "render_seq": seq} def __iter__(self): return iter(self._components) @@ -418,11 +421,11 @@ def refresh(self, *args, **kwargs): if self._default_editable_card is not None: self._cards_components[self._default_editable_card].refresh(*args, **kwargs) - def _get_latest_data(self, card_uuid): + def _get_latest_data(self, card_uuid, final=False): """ Returns latest data so it can be used in the final render() call """ - return self._cards_components[card_uuid]._get_latest_data() + return self._cards_components[card_uuid]._get_latest_data(final=final) def _serialize_components(self, card_uuid): """ From bc4064a75070cb7aafaee0c79404f5b934e2eaee Mon Sep 17 00:00:00 2001 From: Valay Dave Date: Tue, 26 Sep 2023 21:25:47 +0000 Subject: [PATCH 3/9] [current.card.components] card component refresh - added interface to MetaflowCardComponent for making it REALTIME_UPDATABLE - add realtime component rendering capabiity. - Changed ville's abstraction of CardComponents to CardComponentManager - introduced the `current.card.components` / `current.card['abc'].components` interface - `current.card.components` interface helps access/remove components --- metaflow/plugins/cards/card_modules/card.py | 25 ++ .../plugins/cards/component_serializer.py | 345 +++++++++++++++--- 2 files changed, 318 insertions(+), 52 deletions(-) diff --git a/metaflow/plugins/cards/card_modules/card.py b/metaflow/plugins/cards/card_modules/card.py index c305c8fdd3d..30002e5ef15 100644 --- a/metaflow/plugins/cards/card_modules/card.py +++ b/metaflow/plugins/cards/card_modules/card.py @@ -108,6 +108,31 @@ def reload_content_token(self, task, data): class MetaflowCardComponent(object): + + # Setting REALTIME_UPDATABLE as True will make the card component + # updatable via the `current.card.update` method for realtime updates + REALTIME_UPDATABLE = False + + _component_id = None + + @property + def id(self): + return self._component_id + + @id.setter + def id(self, value): + if not isinstance(value, str): + raise TypeError("id must be a string") + self._component_id = value + + def update(self, *args, **kwargs): + """ + Gets called when the user calls `current.card.update(id="abc", data, myval=123)`. + The logic of the update method will be component specific. Some components may + update the contents of the component, while others can just append to the data. + """ + raise NotImplementedError() + def render(self): """ `render` returns a string or dictionary. This class can be called on the client side to dynamically add components to the `MetaflowCard` diff --git a/metaflow/plugins/cards/component_serializer.py b/metaflow/plugins/cards/component_serializer.py index 7cd5e3e93d5..f455cd7502d 100644 --- a/metaflow/plugins/cards/component_serializer.py +++ b/metaflow/plugins/cards/component_serializer.py @@ -21,26 +21,194 @@ def get_card_class(card_type): return filtered_cards[0] +def _component_is_valid(component): + """ + Validates if the component is of the correct class. + """ + if not issubclass(type(component), MetaflowCardComponent): + return False + return True + + +def warning_message(message, logger=None, ts=False): + msg = "[@card WARNING] %s" % message + if logger: + logger(msg, timestamp=ts, bad=True) + + class WarningComponent(ErrorComponent): def __init__(self, warning_message): super().__init__("@card WARNING", warning_message) -class CardComponents: - def __init__(self, card_proc, components=None): +class ComponentStore: + """ + The `ComponentStore` class handles the in-memory storage of the components for a single card. + This class has combination of a array/dictionary like interface to access the components. + + It exposes the `append` /`extend` methods like an array to add components. + It also exposes the `__getitem__`/`__setitem__` methods like a dictionary to access the components by thier Ids. + + The reason this has dual behavior is because components cannot be stored entirely as a map/dictionary because + the order of the components matter. The order of the components will visually affect the order of the components seen on the browser. + + """ + + def __init__(self, logger, components=None): + self._component_map = {} + self._components = [] + self._logger = logger + if components is not None: + for c in list(components): + self._store_component(c, component_id=None) + + def _realtime_updateable_components(self): + for c in self._components: + if c.REALTIME_UPDATABLE: + yield c + + def _create_component_id(self, component): + uuid_bit = "".join(uuid.uuid4().hex.split("-"))[:6] + return type(component).__name__.lower() + "_" + uuid_bit + + def _store_component(self, component, component_id=None): + if not _component_is_valid(component): + warning_message( + "Component (%s) is not a valid MetaflowCardComponent. It will not be stored." + % str(component), + self._logger, + ) + return + if component_id is not None: + component.id = component_id + elif component.id is None: + component.id = self._create_component_id(component) + self._components.append(component) + self._component_map[component.id] = self._components[-1] + + def _remove_component(self, component_id): + self._components.remove(self._component_map[component_id]) + del self._component_map[component_id] + + def __iter__(self): + return iter(self._components) + + def __setitem__(self, key, value): + if self._component_map.get(key) is not None: + # FIXME: what happens to this codepath + # Component Exists in the store + # What happens we relace a realtime component with a non realtime one. + # We have to ensure that layout change has take place so that the card should get re-rendered. + pass + else: + self._store_component(value, component_id=key) + + def __getitem__(self, key): + if key not in self._component_map: + raise KeyError( + "MetaflowCardComponent with id `%s` not found. Available components for the cards include : %s" + % (key, ", ".join(self.keys())) + ) + return self._component_map[key] + + def __delitem__(self, key): + if key not in self._component_map: + raise KeyError( + "MetaflowCardComponent with id `%s` not found. Available components for the cards include : %s" + % (key, ", ".join(self.keys())) + ) + self._remove_component(key) + + def __contains__(self, key): + return key in self._component_map + + def append(self, component, id=None): + self._store_component(component, component_id=id) + + def extend(self, components): + for c in components: + self._store_component(c, component_id=None) + + def clear(self): + self._components.clear() + self._component_map.clear() + + def keys(self): + return list(self._component_map.keys()) + + def values(self): + return self._components + + def __str__(self): + return "Card components present in the card: `%s` " % ("`, `".join(self.keys())) + + def __len__(self): + return len(self._components) + + +class CardComponentManager: + """ + This class manages the components for a single card. + It uses the `ComponentStore` to manage the storage of the components + and exposes methods to add, remove and access the components. + + It also exposes a `refresh` method that will allow refreshing a card with new data + for realtime(ish) updates. + + The `CardComponentCollector` class helps manage interaction with individual cards. The `CardComponentManager` + class helps manage the components for a single card. + The `CardComponentCollector` class uses this class to manage the components for a single card. + + The `CardComponentCollector` exposes convinience methods similar to this class for a default editable card. + `CardComponentCollector` resolves the default editable card at the time of task initialization and + exposes component manipulation methods for that card. These methods include : + + - `append` + - `extend` + - `clear` + - `refresh` + - `components` + - `__iter__` + + Under the hood these common methods will call the corresponding methods on the `CardComponentManager`. + `CardComponentManager` leverages the `ComponentStore` under the hood to actually add/update/remove components + under the hood. + + ## Usage Patterns : + + ```python + current.card["mycardid"].append(component, id="comp123") + current.card["mycardid"].extend([component]) + current.card["mycardid"].refresh(data) # refreshes the card with new data + current.card["mycardid"].components["comp123"] # returns the component with id "comp123" + current.card["mycardid"].components["comp123"].update() + current.card["mycardid"].components.clear() # Wipe all the components + del current.card["mycardid"].components["mycomponentid"] # Delete a component + current.card["mycardid"].components["mynewcomponent"] = Markdown("## New Component") # Set a new component + ``` + """ + + def __init__(self, card_proc, components=None, logger=None, no_warnings=False): self._card_proc = card_proc self._latest_user_data = None self._last_refresh = 0 self._last_render = 0 self._render_seq = 0 - + self._logger = logger + self._no_warnings = no_warnings + self._warn_once = { + "update": {}, + "not_implemented": {}, + } if components is None: - self._components = [] + self._components = ComponentStore(logger=self._logger, components=None) else: - self._components = list(components) + self._components = ComponentStore( + logger=self._logger, components=list(components) + ) - def append(self, component): - self._components.append(component) + def append(self, component, id=None): + self._components.append(component, id=id) def extend(self, components): self._components.extend(components) @@ -52,6 +220,7 @@ def refresh(self, data=None, force=False): # todo make this a configurable variable self._latest_user_data = data nu = time.time() + if nu - self._last_refresh < RUNTIME_CARD_MIN_REFRESH_INTERVAL: # rate limit refreshes: silently ignore requests that # happen too frequently @@ -65,10 +234,27 @@ def refresh(self, data=None, force=False): else: self._card_proc("refresh") + @property + def components(self): + return self._components + + def _warning(self, message): + msg = "[@card WARNING] %s" % message + self._logger(msg, timestamp=False, bad=True) + def _get_latest_data(self, final=False): - # FIXME add component data - seq = 'final' if final else self._render_seq - return {"user": self._latest_user_data, "components": [], "render_seq": seq} + seq = "final" if final else self._render_seq + component_dict = {} + for component in self._components._realtime_updateable_components(): + rendered_comp = _render_card_component(component) + if rendered_comp is not None: + component_dict.update({component.id: rendered_comp}) + # FIXME: Verify _latest_user_data is json serializable + return { + "user": self._latest_user_data, + "components": component_dict, + "render_seq": seq, + } def __iter__(self): return iter(self._components) @@ -98,9 +284,11 @@ class CardComponentCollector: def __init__(self, logger=None, card_proc=None): from metaflow.metaflow_config import CARD_NO_WARNING - self._cards_components = ( + self._card_component_store = ( + # Each key in the dictionary is the UUID of an individual card. + # value is of type `CardComponentManager`, holding a list of MetaflowCardComponents for that particular card {} - ) # a dict with key as uuid and value as CardComponents, holding a list of MetaflowCardComponents. + ) self._cards_meta = ( {} ) # a `dict` of (card_uuid, `dict)` holding all metadata about all @card decorators on the `current` @step. @@ -109,7 +297,13 @@ def __init__(self, logger=None, card_proc=None): self._card_proc = card_proc # `self._default_editable_card` holds the uuid of the card that is default editable. This card has access to `append`/`extend` methods of `self` self._default_editable_card = None - self._warned_once = {"__getitem__": {}, "append": False, "extend": False} + self._warned_once = { + "__getitem__": {}, + "append": False, + "extend": False, + "update": False, + "update_no_id": False, + } self._no_warnings = True if CARD_NO_WARNING else False @staticmethod @@ -151,7 +345,12 @@ def _add_card( suppress_warnings=suppress_warnings, ) self._cards_meta[card_uuid] = card_metadata - self._cards_components[card_uuid] = CardComponents(self._card_proc) + self._card_component_store[card_uuid] = CardComponentManager( + self._card_proc, + components=None, + logger=self._logger, + no_warnings=self._no_warnings, + ) return card_metadata def _warning(self, message): @@ -161,9 +360,9 @@ def _warning(self, message): def _add_warning_to_cards(self, warn_msg): if self._no_warnings: return - for card_id in self._cards_components: + for card_id in self._card_component_store: if not self._cards_meta[card_id]["suppress_warnings"]: - self._cards_components[card_id].append(WarningComponent(warn_msg)) + self._card_component_store[card_id].append(WarningComponent(warn_msg)) def get(self, type=None): """`get` @@ -182,7 +381,7 @@ def get(self, type=None): for card_meta in self._cards_meta.values() if card_meta["type"] == card_type ] - return [self._cards_components[uuid] for uuid in card_uuids] + return [self._card_component_store[uuid] for uuid in card_uuids] def _finalize(self): """ @@ -283,13 +482,13 @@ def __getitem__(self, key): Returns ------- - CardComponents + CardComponentManager An object with `append` and `extend` calls which allow you to add components to the chosen card. """ if key in self._card_id_map: card_uuid = self._card_id_map[key] - return self._cards_components[card_uuid] + return self._card_component_store[card_uuid] if key not in self._warned_once["__getitem__"]: _warn_msg = [ "`current.card['%s']` is not present. Please set the `id` argument in @card to '%s' to access `current.card['%s']`." @@ -300,6 +499,7 @@ def __getitem__(self, key): self._warning(" ".join(_warn_msg)) self._add_warning_to_cards("\n".join(_warn_msg)) self._warned_once["__getitem__"][key] = True + return [] def __setitem__(self, key, value): @@ -317,7 +517,7 @@ def __setitem__(self, key, value): key: str Card ID. - value: List[CardComponent] + value: List[MetaflowCardComponent] List of card components to assign to this card. """ if key in self._card_id_map: @@ -329,7 +529,12 @@ def __setitem__(self, key, value): ) self._warning(_warning_msg) return - self._cards_components[card_uuid] = CardComponents(self._card_proc, value) + self._card_component_store[card_uuid] = CardComponentManager( + self._card_proc, + components=value, + logger=self._logger, + no_warnings=self._no_warnings, + ) return self._warning( @@ -337,18 +542,18 @@ def __setitem__(self, key, value): % (key, key, key) ) - def append(self, component): + def append(self, component, id=None): """ Appends a component to the current card. Parameters ---------- - component : CardComponent + component : MetaflowCardComponent Card component to add to this card. """ if self._default_editable_card is None: if ( - len(self._cards_components) == 1 + len(self._card_component_store) == 1 ): # if there is one card which is not the _default_editable_card then the card is not editable card_type = list(self._cards_meta.values())[0]["type"] if list(self._cards_meta.values())[0]["exists"]: @@ -378,7 +583,7 @@ def append(self, component): self._warned_once["append"] = True return - self._cards_components[self._default_editable_card].append(component) + self._card_component_store[self._default_editable_card].append(component, id=id) def extend(self, components): """ @@ -386,12 +591,12 @@ def extend(self, components): Parameters ---------- - component : Iterator[CardComponent] + component : Iterator[MetaflowCardComponent] Card components to add to this card. """ if self._default_editable_card is None: # if there is one card which is not the _default_editable_card then the card is not editable - if len(self._cards_components) == 1: + if len(self._card_component_store) == 1: card_type = list(self._cards_meta.values())[0]["type"] _warning_msg = [ "Card of type `%s` is not an editable card." % card_type, @@ -411,21 +616,50 @@ def extend(self, components): return - self._cards_components[self._default_editable_card].extend(components) + self._card_component_store[self._default_editable_card].extend(components) + + @property + def components(self): + # FIXME: document + if self._default_editable_card is None: + if len(self._card_component_store) == 1: + card_type = list(self._cards_meta.values())[0]["type"] + _warning_msg = [ + "Card of type `%s` is not an editable card." % card_type, + "Components list will not be updated and `current.card.components` will not work for any call during this runtime execution.", + "Please use an editable card", # todo : link to documentation + ] + else: + _warning_msg = [ + "`current.card.components` cannot disambiguate between multiple @card decorators.", + "Components list will not be accessible and `current.card.components` will not work for any call during this runtime execution.", + "To fix this set the `id` argument in all @card when using multiple @card decorators over a single @step and reference `current.card[ID].components`", # todo : Add Link to documentation + "to update/access the appropriate card component.", + ] + if not self._warned_once["components"]: + self._warning(" ".join(_warning_msg)) + self._warned_once["components"] = True + return + + return self._card_component_store[self._default_editable_card].components def clear(self): + # FIXME: document if self._default_editable_card is not None: - self._cards_components[self._default_editable_card].clear() + self._card_component_store[self._default_editable_card].clear() def refresh(self, *args, **kwargs): + # FIXME: document if self._default_editable_card is not None: - self._cards_components[self._default_editable_card].refresh(*args, **kwargs) + self._card_component_store[self._default_editable_card].refresh( + *args, **kwargs + ) def _get_latest_data(self, card_uuid, final=False): """ Returns latest data so it can be used in the final render() call """ - return self._cards_components[card_uuid]._get_latest_data(final=final) + return self._card_component_store[card_uuid]._get_latest_data(final=final) def _serialize_components(self, card_uuid): """ @@ -434,36 +668,43 @@ def _serialize_components(self, card_uuid): don't render safely then we don't add them to the final list of serialized functions """ serialized_components = [] - if card_uuid not in self._cards_components: + if card_uuid not in self._card_component_store: return [] has_user_components = any( [ issubclass(type(component), UserComponent) - for component in self._cards_components[card_uuid] + for component in self._card_component_store[card_uuid] ] ) - for component in self._cards_components[card_uuid]: - if not issubclass(type(component), MetaflowCardComponent): - continue - try: - rendered_obj = component.render() - except: + for component in self._card_component_store[card_uuid]: + rendered_obj = _render_card_component(component) + if rendered_obj is None: continue - else: - if not (type(rendered_obj) == str or type(rendered_obj) == dict): - continue - else: - # Since `UserComponent`s are safely_rendered using render_tools.py - # we don't need to check JSON serialization as @render_tools.render_safely - # decorator ensures this check so there is no need to re-serialize - if not issubclass(type(component), UserComponent): - try: # check if rendered object is json serializable. - json.dumps(rendered_obj) - except (TypeError, OverflowError) as e: - continue - serialized_components.append(rendered_obj) + serialized_components.append(rendered_obj) if has_user_components and len(serialized_components) > 0: serialized_components = [ SectionComponent(contents=serialized_components).render() ] return serialized_components + + +def _render_card_component(component): + if not _component_is_valid(component): + return None + try: + rendered_obj = component.render() + except: + return None + else: + if not (type(rendered_obj) == str or type(rendered_obj) == dict): + return None + else: + # Since `UserComponent`s are safely_rendered using render_tools.py + # we don't need to check JSON serialization as @render_tools.render_safely + # decorator ensures this check so there is no need to re-serialize + if not issubclass(type(component), UserComponent): + try: # check if rendered object is json serializable. + json.dumps(rendered_obj) + except (TypeError, OverflowError) as e: + return None + return rendered_obj From befd68963918203bdecabe8bb0b4826539f4b8cb Mon Sep 17 00:00:00 2001 From: Valay Dave Date: Tue, 26 Sep 2023 21:24:39 +0000 Subject: [PATCH 4/9] [card-creator] decouple decorator code from card creation - abstraction to handle card creation for decorator (and in future outside code) - Create CardProcessManager class that helps manage the card processes running - Remove `_card_proc` logic from code. - fix card-refresh-bug caused by cardproc instance-method refactor - the `_card_proc` method is an instance method of a card decorator which is passed to componentCollector. This is done so that ComponentCollector can call the method when a refresh is called for an individual card. - Since there is only one copy of the ComponentCollector it created an issue when other cards were trying to call refresh (since ComponentCollector was instantiated with a **single card decorator's** `_card_proc`) - This commit refactored the code to handle multiple cards calling refresh --- metaflow/plugins/cards/card_creator.py | 208 ++++++++++++++++++ metaflow/plugins/cards/card_decorator.py | 154 +++---------- .../plugins/cards/component_serializer.py | 51 ++++- 3 files changed, 281 insertions(+), 132 deletions(-) create mode 100644 metaflow/plugins/cards/card_creator.py diff --git a/metaflow/plugins/cards/card_creator.py b/metaflow/plugins/cards/card_creator.py new file mode 100644 index 00000000000..b6a33c7e5a2 --- /dev/null +++ b/metaflow/plugins/cards/card_creator.py @@ -0,0 +1,208 @@ +import time +import subprocess +import tempfile +import json +import sys +import os +from metaflow import current + +ASYNC_TIMEOUT = 30 + + +class CardProcessManager: + """ + This class is responsible for managing the card creation processes. + + """ + + async_card_processes = { + # "carduuid": { + # "proc": subprocess.Popen, + # "started": time.time() + # } + } + + @classmethod + def _register_card_process(cls, carduuid, proc): + cls.async_card_processes[carduuid] = { + "proc": proc, + "started": time.time(), + } + + @classmethod + def _get_card_process(cls, carduuid): + proc_dict = cls.async_card_processes.get(carduuid, None) + if proc_dict is not None: + return proc_dict["proc"], proc_dict["started"] + return None, None + + @classmethod + def _remove_card_process(cls, carduuid): + if carduuid in cls.async_card_processes: + cls.async_card_processes[carduuid]["proc"].kill() + del cls.async_card_processes[carduuid] + + +class CardCreator: + def __init__(self, top_level_options): + self._top_level_options = top_level_options + + def create( + self, + card_uuid=None, + user_set_card_id=None, + runtime_card=False, + decorator_attributes=None, + card_options=None, + logger=None, + mode="render", + final=False, + ): + # warning_message("calling proc for uuid %s" % self._card_uuid, self._logger) + if mode != "render" and not runtime_card: + # silently ignore runtime updates for cards that don't support them + return + elif mode == "refresh": + # don't serialize components, which can be a somewhat expensive operation, + # if we are just updating data + component_strings = [] + else: + component_strings = current.card._serialize_components(card_uuid) + + data = current.card._get_latest_data(card_uuid, final=final) + runspec = "/".join([current.run_id, current.step_name, current.task_id]) + self._run_cards_subprocess( + card_uuid, + user_set_card_id, + mode, + runspec, + decorator_attributes, + card_options, + component_strings, + logger, + data, + ) + + def _run_cards_subprocess( + self, + card_uuid, + user_set_card_id, + mode, + runspec, + decorator_attributes, + card_options, + component_strings, + logger, + data=None, + ): + components_file = data_file = None + wait = mode == "render" + + if len(component_strings) > 0: + # note that we can't delete temporary files here when calling the subprocess + # async due to a race condition. The subprocess must delete them + components_file = tempfile.NamedTemporaryFile( + "w", suffix=".json", delete=False + ) + json.dump(component_strings, components_file) + components_file.seek(0) + if data is not None: + data_file = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + json.dump(data, data_file) + data_file.seek(0) + + executable = sys.executable + cmd = [ + executable, + sys.argv[0], + ] + cmd += self._top_level_options + [ + "card", + "create", + runspec, + "--delete-input-files", + "--card-uuid", + card_uuid, + "--mode", + mode, + "--type", + decorator_attributes["type"], + # Add the options relating to card arguments. + # todo : add scope as a CLI arg for the create method. + ] + if card_options is not None and len(card_options) > 0: + cmd += ["--options", json.dumps(card_options)] + # set the id argument. + + if decorator_attributes["timeout"] is not None: + cmd += ["--timeout", str(decorator_attributes["timeout"])] + + if user_set_card_id is not None: + cmd += ["--id", str(user_set_card_id)] + + if decorator_attributes["save_errors"]: + cmd += ["--render-error-card"] + + if components_file is not None: + cmd += ["--component-file", components_file.name] + + if data_file is not None: + cmd += ["--data-file", data_file.name] + + response, fail = self._run_command( + cmd, + card_uuid, + os.environ, + timeout=decorator_attributes["timeout"], + wait=wait, + ) + if fail: + resp = "" if response is None else response.decode("utf-8") + logger( + "Card render failed with error : \n\n %s" % resp, + timestamp=False, + bad=True, + ) + + def _run_command(self, cmd, card_uuid, env, wait=True, timeout=None): + fail = False + timeout_args = {} + async_timeout = ASYNC_TIMEOUT + if timeout is not None: + async_timeout = int(timeout) + 10 + timeout_args = dict(timeout=int(timeout) + 10) + + if wait: + try: + rep = subprocess.check_output( + cmd, env=env, stderr=subprocess.STDOUT, **timeout_args + ) + except subprocess.CalledProcessError as e: + rep = e.output + fail = True + except subprocess.TimeoutExpired as e: + rep = e.output + fail = True + return rep, fail + else: + _async_proc, _async_started = CardProcessManager._get_card_process( + card_uuid + ) + if _async_proc and _async_proc.poll() is None: + if time.time() - _async_started > async_timeout: + CardProcessManager._remove_card_process(card_uuid) + else: + # silently refuse to run an async process if a previous one is still running + # and timeout hasn't been reached + return "".encode(), False + else: + CardProcessManager._register_card_process( + card_uuid, + subprocess.Popen( + cmd, + env=env, + stderr=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + ), + ) + return "".encode(), False diff --git a/metaflow/plugins/cards/card_decorator.py b/metaflow/plugins/cards/card_decorator.py index dbeb83522a5..4fa20cf08f7 100644 --- a/metaflow/plugins/cards/card_decorator.py +++ b/metaflow/plugins/cards/card_decorator.py @@ -11,6 +11,7 @@ from metaflow.current import current from metaflow.util import to_unicode from .component_serializer import CardComponentCollector, get_card_class +from .card_creator import CardCreator # from metaflow import get_metadata @@ -63,6 +64,8 @@ class CardDecorator(StepDecorator): _called_once = {} + card_creator = None + def __init__(self, *args, **kwargs): super(CardDecorator, self).__init__(*args, **kwargs) self._task_datastore = None @@ -72,7 +75,10 @@ def __init__(self, *args, **kwargs): self._is_editable = False self._card_uuid = None self._user_set_card_id = None - self._async_proc = None + + @classmethod + def _set_card_creator(cls, card_creator): + cls.card_creator = card_creator def _is_event_registered(self, evt_name): return evt_name in self._called_once @@ -129,6 +135,9 @@ def task_pre_step( ubf_context, inputs, ): + self._task_datastore = task_datastore + self._metadata = metadata + card_type = self.attributes["type"] card_class = get_card_class(card_type) @@ -162,8 +171,10 @@ def task_pre_step( # we need to ensure that `current.card` has `CardComponentCollector` instantiated only once. if not self._is_event_registered("pre-step"): self._register_event("pre-step") + self._set_card_creator(CardCreator(self._create_top_level_args())) + current._update_env( - {"card": CardComponentCollector(self._logger, self._card_proc)} + {"card": CardComponentCollector(self._logger, self.card_creator)} ) # this line happens because of decospecs parsing. @@ -174,8 +185,11 @@ def task_pre_step( card_metadata = current.card._add_card( self.attributes["type"], self._user_set_card_id, - self._is_editable, - customize, + self.attributes, + self.card_options, + editable=self._is_editable, + customize=customize, + runtime_card=self._is_runtime_card, ) self._card_uuid = card_metadata["uuid"] @@ -185,30 +199,20 @@ def task_pre_step( if self.step_counter == self.total_decos_on_step[step_name]: current.card._finalize() - self._task_datastore = task_datastore - self._metadata = metadata - def task_finished( self, step_name, flow, graph, is_task_ok, retry_count, max_user_code_retries ): + create_options = dict( + card_uuid=self._card_uuid, + user_set_card_id=self._user_set_card_id, + runtime_card=self._is_runtime_card, + decorator_attributes=self.attributes, + card_options=self.card_options, + logger=self._logger, + ) if is_task_ok: - self._card_proc("render") - self._card_proc("refresh", final=True) - - def _card_proc(self, mode, final=False): - if mode != "render" and not self._is_runtime_card: - # silently ignore runtime updates for cards that don't support them - return - elif mode == "refresh": - # don't serialize components, which can be a somewhat expensive operation, - # if we are just updating data - component_strings = [] - else: - component_strings = current.card._serialize_components(self._card_uuid) - - data = current.card._get_latest_data(self._card_uuid, final=final) - runspec = "/".join([current.run_id, current.step_name, current.task_id]) - self._run_cards_subprocess(mode, runspec, component_strings, data) + self.card_creator.create(mode="render", **create_options) + self.card_creator.create(mode="refresh", final=True, **create_options) @staticmethod def _options(mapping): @@ -235,105 +239,3 @@ def _create_top_level_args(self): # the context of the main process } return list(self._options(top_level_options)) - - def _run_cards_subprocess(self, mode, runspec, component_strings, data=None): - components_file = data_file = None - wait = mode == "render" - - if len(component_strings) > 0: - # note that we can't delete temporary files here when calling the subprocess - # async due to a race condition. The subprocess must delete them - components_file = tempfile.NamedTemporaryFile( - "w", suffix=".json", delete=False - ) - json.dump(component_strings, components_file) - components_file.seek(0) - if data is not None: - data_file = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) - json.dump(data, data_file) - data_file.seek(0) - - executable = sys.executable - cmd = [ - executable, - sys.argv[0], - ] - cmd += self._create_top_level_args() + [ - "card", - "create", - runspec, - "--delete-input-files", - "--card-uuid", - self._card_uuid, - "--mode", - mode, - "--type", - self.attributes["type"], - # Add the options relating to card arguments. - # todo : add scope as a CLI arg for the create method. - ] - if self.card_options is not None and len(self.card_options) > 0: - cmd += ["--options", json.dumps(self.card_options)] - # set the id argument. - - if self.attributes["timeout"] is not None: - cmd += ["--timeout", str(self.attributes["timeout"])] - - if self._user_set_card_id is not None: - cmd += ["--id", str(self._user_set_card_id)] - - if self.attributes["save_errors"]: - cmd += ["--render-error-card"] - - if components_file is not None: - cmd += ["--component-file", components_file.name] - - if data_file is not None: - cmd += ["--data-file", data_file.name] - - response, fail = self._run_command( - cmd, os.environ, timeout=self.attributes["timeout"], wait=wait - ) - if fail: - resp = "" if response is None else response.decode("utf-8") - self._logger( - "Card render failed with error : \n\n %s" % resp, - timestamp=False, - bad=True, - ) - - def _run_command(self, cmd, env, wait=True, timeout=None): - fail = False - timeout_args = {} - async_timeout = ASYNC_TIMEOUT - if timeout is not None: - async_timeout = int(timeout) + 10 - timeout_args = dict(timeout=int(timeout) + 10) - - if wait: - try: - rep = subprocess.check_output( - cmd, env=env, stderr=subprocess.STDOUT, **timeout_args - ) - except subprocess.CalledProcessError as e: - rep = e.output - fail = True - except subprocess.TimeoutExpired as e: - rep = e.output - fail = True - return rep, fail - else: - if self._async_proc and self._async_proc.poll() is None: - if time.time() - self._async_started > async_timeout: - self._async_proc.kill() - else: - # silently refuse to run an async process if a previous one is still running - # and timeout hasn't been reached - return "", False - else: - # print("CARD CMD", " ".join(cmd)) - self._async_proc = subprocess.Popen( - cmd, env=env, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL - ) - self._async_started = time.time() - return "", False diff --git a/metaflow/plugins/cards/component_serializer.py b/metaflow/plugins/cards/component_serializer.py index f455cd7502d..c8bed702b13 100644 --- a/metaflow/plugins/cards/component_serializer.py +++ b/metaflow/plugins/cards/component_serializer.py @@ -1,6 +1,7 @@ from .card_modules import MetaflowCardComponent from .card_modules.basic import ErrorComponent, SectionComponent from .card_modules.components import UserComponent +from functools import partial import uuid import json import time @@ -188,8 +189,27 @@ class helps manage the components for a single card. ``` """ - def __init__(self, card_proc, components=None, logger=None, no_warnings=False): - self._card_proc = card_proc + def __init__( + self, + card_uuid, + decorator_attributes, + card_creator, + components=None, + logger=None, + no_warnings=False, + user_set_card_id=None, + runtime_card=False, + card_options=None, + ): + self._card_creator_args = dict( + card_uuid=card_uuid, + user_set_card_id=user_set_card_id, + runtime_card=runtime_card, + decorator_attributes=decorator_attributes, + card_options=card_options, + logger=logger, + ) + self._card_creator = card_creator self._latest_user_data = None self._last_refresh = 0 self._last_render = 0 @@ -216,6 +236,9 @@ def extend(self, components): def clear(self): self._components.clear() + def _card_proc(self, mode): + self._card_creator.create(**self._card_creator_args, mode=mode) + def refresh(self, data=None, force=False): # todo make this a configurable variable self._latest_user_data = data @@ -281,7 +304,7 @@ class CardComponentCollector: - [x] by looking it up by its type, e.g. `current.card.get(type='pytorch')`. """ - def __init__(self, logger=None, card_proc=None): + def __init__(self, logger=None, card_creator=None): from metaflow.metaflow_config import CARD_NO_WARNING self._card_component_store = ( @@ -294,7 +317,7 @@ def __init__(self, logger=None, card_proc=None): ) # a `dict` of (card_uuid, `dict)` holding all metadata about all @card decorators on the `current` @step. self._card_id_map = {} # card_id to uuid map for all cards with ids self._logger = logger - self._card_proc = card_proc + self._card_creator = card_creator # `self._default_editable_card` holds the uuid of the card that is default editable. This card has access to `append`/`extend` methods of `self` self._default_editable_card = None self._warned_once = { @@ -318,9 +341,12 @@ def _add_card( self, card_type, card_id, + decorator_attributes, + card_options, editable=False, customize=False, suppress_warnings=False, + runtime_card=False, ): """ This function helps collect cards from all the card decorators. @@ -343,13 +369,21 @@ def _add_card( editable=editable, customize=customize, suppress_warnings=suppress_warnings, + runtime_card=runtime_card, + decorator_attributes=decorator_attributes, + card_options=card_options, ) self._cards_meta[card_uuid] = card_metadata self._card_component_store[card_uuid] = CardComponentManager( - self._card_proc, + card_uuid, + decorator_attributes, + self._card_creator, components=None, logger=self._logger, no_warnings=self._no_warnings, + user_set_card_id=card_id, + runtime_card=runtime_card, + card_options=card_options, ) return card_metadata @@ -530,10 +564,15 @@ def __setitem__(self, key, value): self._warning(_warning_msg) return self._card_component_store[card_uuid] = CardComponentManager( - self._card_proc, + card_uuid, + self._cards_meta[card_uuid]["decorator_attributes"], + self._card_creator, components=value, logger=self._logger, no_warnings=self._no_warnings, + user_set_card_id=key, + card_options=self._cards_meta[card_uuid]["card_options"], + runtime_card=self._cards_meta[card_uuid]["runtime_card"], ) return From eba36be133bec4ee6fdd776c656c1ad13f3e5d9e Mon Sep 17 00:00:00 2001 From: Valay Dave Date: Tue, 26 Sep 2023 23:11:20 +0000 Subject: [PATCH 5/9] [default-card][ui] default card realtime updatable - cleanup the ui component code and add naive progress-bar --- metaflow/plugins/cards/card_modules/base.html | 1 + .../plugins/cards/card_modules/bundle.css | 112 +- metaflow/plugins/cards/card_modules/main.js | 9 +- metaflow/plugins/cards/ui/.gitignore | 581 ++ metaflow/plugins/cards/ui/.prettierrc.json | 3 + metaflow/plugins/cards/ui/.yarnrc.yml | 1 + .../plugins/cards/ui/public/card-example.json | 66 +- metaflow/plugins/cards/ui/public/index.html | 22 + metaflow/plugins/cards/ui/rollup.config.js | 2 +- metaflow/plugins/cards/ui/src/App.svelte | 6 +- .../cards/ui/src/components/artifacts.svelte | 5 +- .../cards/ui/src/components/bar-chart.svelte | 4 +- .../components/card-component-renderer.svelte | 7 +- .../cards/ui/src/components/heading.svelte | 2 +- .../cards/ui/src/components/image.svelte | 3 +- .../cards/ui/src/components/line-chart.svelte | 4 +- .../cards/ui/src/components/modal.svelte | 1 + .../cards/ui/src/components/page.svelte | 7 +- .../ui/src/components/progress-bar.svelte | 14 + .../cards/ui/src/components/section.svelte | 2 +- .../cards/ui/src/components/subtitle.svelte | 2 +- .../ui/src/components/table-horizontal.svelte | 2 +- .../ui/src/components/table-vertical.svelte | 2 +- .../cards/ui/src/components/table.svelte | 2 +- .../cards/ui/src/components/text.svelte | 3 +- .../cards/ui/src/components/title.svelte | 3 +- metaflow/plugins/cards/ui/src/store.ts | 50 +- metaflow/plugins/cards/ui/src/types.ts | 38 +- metaflow/plugins/cards/ui/tsconfig.json | 24 +- metaflow/plugins/cards/ui/yarn.lock | 8686 ++++++++++------- 30 files changed, 6217 insertions(+), 3447 deletions(-) create mode 100644 metaflow/plugins/cards/ui/.gitignore create mode 100644 metaflow/plugins/cards/ui/.prettierrc.json create mode 100644 metaflow/plugins/cards/ui/.yarnrc.yml create mode 100644 metaflow/plugins/cards/ui/src/components/progress-bar.svelte diff --git a/metaflow/plugins/cards/card_modules/base.html b/metaflow/plugins/cards/card_modules/base.html index 38a066179ea..91c05ca1f4b 100644 --- a/metaflow/plugins/cards/card_modules/base.html +++ b/metaflow/plugins/cards/card_modules/base.html @@ -18,6 +18,7 @@
diff --git a/metaflow/plugins/cards/ui/rollup.config.js b/metaflow/plugins/cards/ui/rollup.config.js index 5d4b7502aef..c3054cd1756 100644 --- a/metaflow/plugins/cards/ui/rollup.config.js +++ b/metaflow/plugins/cards/ui/rollup.config.js @@ -43,7 +43,7 @@ export default { input: "src/main.ts", output: { dir: process.env.OUTPUT_DIR ?? "public/build", - sourcemap: true, + sourcemap: !production, format: "iife", name: "app", }, diff --git a/metaflow/plugins/cards/ui/src/App.svelte b/metaflow/plugins/cards/ui/src/App.svelte index bbb390ebbca..881d3e0234e 100644 --- a/metaflow/plugins/cards/ui/src/App.svelte +++ b/metaflow/plugins/cards/ui/src/App.svelte @@ -3,7 +3,7 @@ import "./prism"; import "./global.css"; import "./prism.css"; - import "./app.css" + import "./app.css"; import { cardData, setCardData, modal } from "./store"; import * as utils from "./utils"; import Aside from "./components/aside.svelte"; @@ -16,11 +16,11 @@ // Get the data from the element in `windows.__MF_DATA__` corresponding to `cardDataId`. This allows multiple sets of // data to exist on a single page - setCardData(cardDataId) + setCardData(cardDataId); // Set the `embed` class to hide the `aside` if specified in the URL const urlParams = new URLSearchParams(window?.location.search); - let embed = Boolean(urlParams.get('embed')) + let embed = Boolean(urlParams.get("embed"));
diff --git a/metaflow/plugins/cards/ui/src/components/artifacts.svelte b/metaflow/plugins/cards/ui/src/components/artifacts.svelte index 72309d41225..3e4de16f06c 100644 --- a/metaflow/plugins/cards/ui/src/components/artifacts.svelte +++ b/metaflow/plugins/cards/ui/src/components/artifacts.svelte @@ -4,10 +4,9 @@ import ArtifactRow from "./artifact-row.svelte"; export let componentData: types.ArtifactsComponent; - const { data } = componentData; // we can't guarantee the data is sorted from the source, so we sort before render - const sortedData = data.sort((a, b) => { + const sortedData = componentData?.data.sort((a, b) => { // nulls first if (a.name && b.name) { if (a.name > b.name) { @@ -24,7 +23,7 @@ {#each sortedData as artifact} - + {/each}
diff --git a/metaflow/plugins/cards/ui/src/components/bar-chart.svelte b/metaflow/plugins/cards/ui/src/components/bar-chart.svelte index 7db0febe9a7..0d88591fcff 100644 --- a/metaflow/plugins/cards/ui/src/components/bar-chart.svelte +++ b/metaflow/plugins/cards/ui/src/components/bar-chart.svelte @@ -17,11 +17,11 @@ BarController, LinearScale, CategoryScale, - PointElement + PointElement, ); export let componentData: types.BarChartComponent; - const { config, data, labels } = componentData; + $: ({ config, data, labels } = componentData); let el: HTMLCanvasElement; diff --git a/metaflow/plugins/cards/ui/src/components/card-component-renderer.svelte b/metaflow/plugins/cards/ui/src/components/card-component-renderer.svelte index 6d9e05f9a14..3965d5ce183 100644 --- a/metaflow/plugins/cards/ui/src/components/card-component-renderer.svelte +++ b/metaflow/plugins/cards/ui/src/components/card-component-renderer.svelte @@ -10,6 +10,7 @@ import Log from "./log.svelte"; import Markdown from "./markdown.svelte"; import Page from "./page.svelte"; + import ProgressBar from "./progress-bar.svelte"; import Section from "./section.svelte"; import Subtitle from "./subtitle.svelte"; import Table from "./table.svelte"; @@ -29,6 +30,7 @@ log: Log, markdown: Markdown, page: Page, + progressBar: ProgressBar, section: Section, subtitle: Subtitle, table: Table, @@ -36,11 +38,12 @@ title: Title, }; - let component = typesMap?.[componentData.type] + let component = typesMap?.[componentData.type]; if (!component) { - console.error("Unknown component type: ", componentData.type) + console.error("Unknown component type: ", componentData.type); } + {#if component} {#if (componentData.type === "page" || componentData.type === "section") && componentData?.contents} diff --git a/metaflow/plugins/cards/ui/src/components/heading.svelte b/metaflow/plugins/cards/ui/src/components/heading.svelte index 03983575467..b41cd0bed21 100644 --- a/metaflow/plugins/cards/ui/src/components/heading.svelte +++ b/metaflow/plugins/cards/ui/src/components/heading.svelte @@ -5,7 +5,7 @@ import Subtitle from "./subtitle.svelte"; export let componentData: types.HeadingComponent; - const { title, subtitle } = componentData; + $: ({ title, subtitle } = componentData);
diff --git a/metaflow/plugins/cards/ui/src/components/image.svelte b/metaflow/plugins/cards/ui/src/components/image.svelte index 3a247455907..e28dfac31a8 100644 --- a/metaflow/plugins/cards/ui/src/components/image.svelte +++ b/metaflow/plugins/cards/ui/src/components/image.svelte @@ -4,9 +4,10 @@ import { modal } from "../store"; export let componentData: types.ImageComponent; - const { src, label, description } = componentData; + $: ({ src, label, description } = componentData); +
modal.set(componentData)} data-component="image">
{label diff --git a/metaflow/plugins/cards/ui/src/components/line-chart.svelte b/metaflow/plugins/cards/ui/src/components/line-chart.svelte index 1defb02e457..1f5173309b9 100644 --- a/metaflow/plugins/cards/ui/src/components/line-chart.svelte +++ b/metaflow/plugins/cards/ui/src/components/line-chart.svelte @@ -17,11 +17,11 @@ LinearScale, LineController, CategoryScale, - PointElement + PointElement, ); export let componentData: types.LineChartComponent; - const { config, data, labels } = componentData; + $: ({ config, data, labels } = componentData); let el: HTMLCanvasElement; diff --git a/metaflow/plugins/cards/ui/src/components/modal.svelte b/metaflow/plugins/cards/ui/src/components/modal.svelte index 3dff91f914d..a9adf1bbbea 100644 --- a/metaflow/plugins/cards/ui/src/components/modal.svelte +++ b/metaflow/plugins/cards/ui/src/components/modal.svelte @@ -20,6 +20,7 @@ {#if componentData && $modal} +
diff --git a/metaflow/plugins/cards/ui/src/components/page.svelte b/metaflow/plugins/cards/ui/src/components/page.svelte index 8f2736c9a39..0eb96f25ae8 100644 --- a/metaflow/plugins/cards/ui/src/components/page.svelte +++ b/metaflow/plugins/cards/ui/src/components/page.svelte @@ -2,10 +2,13 @@ -
+
diff --git a/metaflow/plugins/cards/ui/src/components/progress-bar.svelte b/metaflow/plugins/cards/ui/src/components/progress-bar.svelte new file mode 100644 index 00000000000..51b8354c41a --- /dev/null +++ b/metaflow/plugins/cards/ui/src/components/progress-bar.svelte @@ -0,0 +1,14 @@ + + + +
+ {#if label} + + {/if} + {value}{unit || ""} +
diff --git a/metaflow/plugins/cards/ui/src/components/section.svelte b/metaflow/plugins/cards/ui/src/components/section.svelte index e43a6f34d3f..8377e19cdab 100644 --- a/metaflow/plugins/cards/ui/src/components/section.svelte +++ b/metaflow/plugins/cards/ui/src/components/section.svelte @@ -2,7 +2,7 @@ -

{componentData.text}

+

{componentData?.text || ""}

+ + + + + +
+ +
+ + + \ No newline at end of file From 17ec1662ea5e11cfb6ba57678fec6a65475f759d Mon Sep 17 00:00:00 2001 From: Valay Dave Date: Wed, 27 Sep 2023 07:16:07 +0000 Subject: [PATCH 8/9] [sidecard][unfinished] current.card in sidecar - Created classes that will help dump current.card state - created functions that will help retrieve current.card in a separate subprocess. What is missing : 1. Multi-process update safety 2. Constraints on sharing / not sharing components with main process --- metaflow/plugins/cards/card_creator.py | 23 +++- metaflow/plugins/cards/card_decorator.py | 41 +++++- .../plugins/cards/component_serializer.py | 123 ++++++++++++++---- .../plugins/cards/runtime_collector_state.py | 68 ++++++++++ 4 files changed, 221 insertions(+), 34 deletions(-) create mode 100644 metaflow/plugins/cards/runtime_collector_state.py diff --git a/metaflow/plugins/cards/card_creator.py b/metaflow/plugins/cards/card_creator.py index b6a33c7e5a2..64197552f00 100644 --- a/metaflow/plugins/cards/card_creator.py +++ b/metaflow/plugins/cards/card_creator.py @@ -44,8 +44,20 @@ def _remove_card_process(cls, carduuid): class CardCreator: - def __init__(self, top_level_options): + def __init__( + self, + top_level_options=None, + pathspec=None, + # TODO Add a pathspec somewhere here so that it can instantiate correctly. + ): self._top_level_options = top_level_options + self._pathspec = pathspec + + def _dump_state_to_dict(self): + return { + "top_level_options": self._top_level_options, + "pathspec": self._pathspec, + } def create( self, @@ -57,6 +69,8 @@ def create( logger=None, mode="render", final=False, + component_serialzer=None, + fetch_latest_data=None, ): # warning_message("calling proc for uuid %s" % self._card_uuid, self._logger) if mode != "render" and not runtime_card: @@ -67,10 +81,9 @@ def create( # if we are just updating data component_strings = [] else: - component_strings = current.card._serialize_components(card_uuid) - - data = current.card._get_latest_data(card_uuid, final=final) - runspec = "/".join([current.run_id, current.step_name, current.task_id]) + component_strings = component_serialzer() + data = fetch_latest_data(final=final) + runspec = "/".join(self._pathspec.split("/")[1:]) self._run_cards_subprocess( card_uuid, user_set_card_id, diff --git a/metaflow/plugins/cards/card_decorator.py b/metaflow/plugins/cards/card_decorator.py index 4fa20cf08f7..ed89c1274b1 100644 --- a/metaflow/plugins/cards/card_decorator.py +++ b/metaflow/plugins/cards/card_decorator.py @@ -2,6 +2,7 @@ import os import tempfile import sys +from functools import partial import time import json @@ -10,8 +11,10 @@ from metaflow.decorators import StepDecorator, flow_decorators from metaflow.current import current from metaflow.util import to_unicode +from metaflow.metaflow_config import DATASTORE_LOCAL_DIR from .component_serializer import CardComponentCollector, get_card_class from .card_creator import CardCreator +from .runtime_collector_state import CardStateManager # from metaflow import get_metadata @@ -66,6 +69,8 @@ class CardDecorator(StepDecorator): card_creator = None + _collector_state_dump_file = None + def __init__(self, *args, **kwargs): super(CardDecorator, self).__init__(*args, **kwargs) self._task_datastore = None @@ -75,6 +80,12 @@ def __init__(self, *args, **kwargs): self._is_editable = False self._card_uuid = None self._user_set_card_id = None + self._current_step = None + self._state_manager = None # This is set on a per TASK basis + + @classmethod + def _set_collector_state_dump_file(cls, file): + cls._collector_state_dump_file = file @classmethod def _set_card_creator(cls, card_creator): @@ -135,6 +146,7 @@ def task_pre_step( ubf_context, inputs, ): + self._task_datastore = task_datastore self._metadata = metadata @@ -167,11 +179,17 @@ def task_pre_step( warning_message(wrn_msg, self._logger) self._user_set_card_id = None + pathspec = "/".join([flow.name, str(run_id), step_name, str(task_id)]) # As we have multiple decorators, # we need to ensure that `current.card` has `CardComponentCollector` instantiated only once. if not self._is_event_registered("pre-step"): self._register_event("pre-step") - self._set_card_creator(CardCreator(self._create_top_level_args())) + self._set_card_creator( + CardCreator( + top_level_options=self._create_top_level_args(), + pathspec=pathspec, + ) + ) current._update_env( {"card": CardComponentCollector(self._logger, self.card_creator)} @@ -198,6 +216,21 @@ def task_pre_step( # This will set up the `current.card` object for usage inside `@step` code. if self.step_counter == self.total_decos_on_step[step_name]: current.card._finalize() + self._dump_state_of_collector_to_file(pathspec) + # Set env var here so that any process spawned + # by this process can access the card state. + os.environ["METAFLOW_CARD_PATHSPEC"] = pathspec + + def task_exception( + self, exception, step_name, flow, graph, retry_count, max_user_code_retries + ): + if self._state_manager is not None: + self._state_manager.done() + + def _dump_state_of_collector_to_file(self, pathspec): + # Create a directory in the DATASTORE_LOCAL_DIR like `DATASTORE_LOCAL_DIR/` + self._state_manager = CardStateManager(pathspec) + self._state_manager.save(current.card._dump_state_to_dict()) def task_finished( self, step_name, flow, graph, is_task_ok, retry_count, max_user_code_retries @@ -209,10 +242,16 @@ def task_finished( decorator_attributes=self.attributes, card_options=self.card_options, logger=self._logger, + fetch_latest_data=partial(current.card._get_latest_data, self._card_uuid), + component_serialzer=partial( + current.card._serialize_components, self._card_uuid + ), ) if is_task_ok: self.card_creator.create(mode="render", **create_options) self.card_creator.create(mode="refresh", final=True, **create_options) + if self._state_manager is not None: + self._state_manager.done() @staticmethod def _options(mapping): diff --git a/metaflow/plugins/cards/component_serializer.py b/metaflow/plugins/cards/component_serializer.py index c8bed702b13..62f326be1cd 100644 --- a/metaflow/plugins/cards/component_serializer.py +++ b/metaflow/plugins/cards/component_serializer.py @@ -191,9 +191,9 @@ class helps manage the components for a single card. def __init__( self, - card_uuid, - decorator_attributes, - card_creator, + card_uuid=None, + decorator_attributes=None, + card_creator=None, components=None, logger=None, no_warnings=False, @@ -207,7 +207,6 @@ def __init__( runtime_card=runtime_card, decorator_attributes=decorator_attributes, card_options=card_options, - logger=logger, ) self._card_creator = card_creator self._latest_user_data = None @@ -227,6 +226,13 @@ def __init__( logger=self._logger, components=list(components) ) + def _dump_state_to_dict(self): + return {**self._card_creator_args, "no_warnings": self._no_warnings} + + @classmethod + def _load_from_dict(cls, card_dict, card_creator, logger=None): + return cls(**card_dict, card_creator=card_creator, logger=logger) + def append(self, component, id=None): self._components.append(component, id=id) @@ -237,7 +243,13 @@ def clear(self): self._components.clear() def _card_proc(self, mode): - self._card_creator.create(**self._card_creator_args, mode=mode) + self._card_creator.create( + **self._card_creator_args, + fetch_latest_data=self._get_latest_data, + component_serialzer=self._serialize_components, + logger=self._logger, + mode=mode, + ) def refresh(self, data=None, force=False): # todo make this a configurable variable @@ -279,6 +291,25 @@ def _get_latest_data(self, final=False): "render_seq": seq, } + def _serialize_components(self): + serialized_components = [] + has_user_components = any( + [ + issubclass(type(component), UserComponent) + for component in self._components + ] + ) + for component in self._components: + rendered_obj = _render_card_component(component) + if rendered_obj is None: + continue + serialized_components.append(rendered_obj) + if has_user_components and len(serialized_components) > 0: + serialized_components = [ + SectionComponent(contents=serialized_components).render() + ] + return serialized_components + def __iter__(self): return iter(self._components) @@ -304,6 +335,56 @@ class CardComponentCollector: - [x] by looking it up by its type, e.g. `current.card.get(type='pytorch')`. """ + def _dump_state_to_dict(self): + """ + Dump the following object's state to file: + - `self._card_component_store` + - `self._cards_meta` + - `self._card_id_map` + - `self._default_editable_card` + - `self._card_creator` + + """ + + def _serialize_store(card_manager_dict): + serialized_store = {} + for card_uuid, card_manager in card_manager_dict.items(): + serialized_store[card_uuid] = card_manager._dump_state_to_dict() + return serialized_store + + state_object = { + "card_component_store": _serialize_store(self._card_component_store), + "cards_meta": self._cards_meta, + "card_id_map": self._card_id_map, + "default_editable_card": self._default_editable_card, # UUId of defautl editable card + "card_creator": self._card_creator._dump_state_to_dict(), + "no_warnings": self._no_warnings, + } + return state_object + + @classmethod + def _load_state(cls, state_dict, logger): + from .card_creator import CardCreator + + def _load_stores(card_manager_dict): + loaded_store = {} + for card_uuid, card_manager in card_manager_dict.items(): + loaded_store[card_uuid] = CardComponentManager._load_from_dict( + card_manager, card_creator, logger=logger + ) + return loaded_store + + card_creator = CardCreator(**state_dict["card_creator"]) + collector_object = cls(logger=logger, card_creator=card_creator) + collector_object._card_component_store = _load_stores( + state_dict["card_component_store"] + ) + collector_object._cards_meta = state_dict["cards_meta"] + collector_object._card_id_map = state_dict["card_id_map"] + collector_object._default_editable_card = state_dict["default_editable_card"] + collector_object._no_warnings = state_dict["no_warnings"] + return collector_object + def __init__(self, logger=None, card_creator=None): from metaflow.metaflow_config import CARD_NO_WARNING @@ -375,9 +456,9 @@ def _add_card( ) self._cards_meta[card_uuid] = card_metadata self._card_component_store[card_uuid] = CardComponentManager( - card_uuid, - decorator_attributes, - self._card_creator, + card_uuid=card_uuid, + decorator_attributes=decorator_attributes, + card_creator=self._card_creator, components=None, logger=self._logger, no_warnings=self._no_warnings, @@ -564,9 +645,11 @@ def __setitem__(self, key, value): self._warning(_warning_msg) return self._card_component_store[card_uuid] = CardComponentManager( - card_uuid, - self._cards_meta[card_uuid]["decorator_attributes"], - self._card_creator, + card_uuid=card_uuid, + decorator_attributes=self._cards_meta[card_uuid][ + "decorator_attributes" + ], + card_creator=self._card_creator, components=value, logger=self._logger, no_warnings=self._no_warnings, @@ -706,25 +789,9 @@ def _serialize_components(self, card_uuid): Components exposed by metaflow ensure that they render safely. If components don't render safely then we don't add them to the final list of serialized functions """ - serialized_components = [] if card_uuid not in self._card_component_store: return [] - has_user_components = any( - [ - issubclass(type(component), UserComponent) - for component in self._card_component_store[card_uuid] - ] - ) - for component in self._card_component_store[card_uuid]: - rendered_obj = _render_card_component(component) - if rendered_obj is None: - continue - serialized_components.append(rendered_obj) - if has_user_components and len(serialized_components) > 0: - serialized_components = [ - SectionComponent(contents=serialized_components).render() - ] - return serialized_components + return self._card_component_store[card_uuid]._serialize_components() def _render_card_component(component): diff --git a/metaflow/plugins/cards/runtime_collector_state.py b/metaflow/plugins/cards/runtime_collector_state.py new file mode 100644 index 00000000000..2d07ffd81a3 --- /dev/null +++ b/metaflow/plugins/cards/runtime_collector_state.py @@ -0,0 +1,68 @@ +from metaflow.metaflow_config import DATASTORE_LOCAL_DIR +import os +import hashlib +import tempfile +import json + +CARD_STATE_ROOT_DIR = os.path.join(DATASTORE_LOCAL_DIR, "mf.card_state") + + +class CardStateManager: + def __init__(self, pathspec) -> None: + self._pathspec = pathspec + # Create a directory that will store the card state. + os.makedirs(CARD_STATE_ROOT_DIR, exist_ok=True) + self._card_state_directory = _get_card_state_directory(pathspec) + self._card_state_file = _get_card_state_file(pathspec) + self._state_file_created = False + + def load(self): + with open(self._card_state_file, "r") as _state_file: + state = json.load(_state_file) + return state + + def save(self, state): + os.makedirs(self._card_state_directory, exist_ok=True) + _state_file = open(self._card_state_file, "w") + json.dump(state, _state_file) + _state_file.close() + self._state_file_created = True + + def done(self): + if self._state_file_created: + os.remove(self._card_state_file) + os.rmdir(self._card_state_directory) + + +def _get_card_state_file(pathspec): + return os.path.join(_get_card_state_directory(pathspec), "card_state.json") + + +def _get_card_state_directory(pathspec): + pathspec_hash = hashlib.md5(pathspec.encode()).hexdigest() + return os.path.join(CARD_STATE_ROOT_DIR, pathspec_hash) + + +def get_realtime_cards( + pathspec=None, +): + from .component_serializer import CardComponentCollector + from metaflow import current + from metaflow.cli import logger + + if pathspec is None: + # if getattr(current, "card", None): + # return current.card + # else: + if "METAFLOW_CARD_PATHSPEC" not in os.environ: + raise ValueError( + "It seems like `get_realtime_cards` is being called from outside a Metaflow step process. " + "In such cases a `pathspec` argument is required or `METAFLOW_CARD_PATHSPEC` needs to be set " + "as an environment variable." + ) + pathspec = os.environ["METAFLOW_CARD_PATHSPEC"] + + card_state_manager = CardStateManager(pathspec) + return CardComponentCollector._load_state( + state_dict=card_state_manager.load(), logger=logger + ) From d12cd98334bbe23133acb9cf6100c97cc29d92e1 Mon Sep 17 00:00:00 2001 From: Valay Dave Date: Mon, 25 Sep 2023 20:06:31 +0000 Subject: [PATCH 9/9] [unfinished] card locking for safe sidecar updates - created a locking mechanism to ensure cards are locked to individual subprocesses. - introduced `LockableCardDict` which helps lock cards per subprocess - introduced `CardStateManager` to save the state of components/data updates during task execution ## life-cycle of cards 1. Pre-task execution : - @card decorator instantiates a CardComponentCollector with a LockableCardDict set to `pre_task_startup=True` and empty dictionary for it's `card_component_store` - This ensures that new CardComponentManagers can be trivially added by the main process. - Once main process is done adding all cards, we run `finalize` method : - it will set `pre_task_startup=False` and set post_task_startup=True - it will instantiate the locks. Won't check if they are already possessed by others since the code is being run in the main process and task execution has not started yet; (All lock alteration happens once user code starts executing). - Upon finishing finalize `current.card` will also dump the state of `current.card` for being available to any outside subprocesses. 2. Task Execution: - Main process : - Which ever cards the main thread update's will be locked by the main thread. - The cards which are not accessed / updated by the main thread, can be locked by other processes. - Locked cards will provide the same type of object back to the user, but the `card_proc` method will be replaced with a method that will warn the user that the card is locked. - Subprocesses: - calls the `get_runtime_card` method to get the `current.card` object - the method will seek out the `current.card`'s state location and load the state. - the state of the card specifies what ever task spec's component collector's state - will be able to call render_runtime and refresh from the cli - `refresh` will also dump the state of the components to disk. - [todo] when finished will call `current.card.done()` which will dump the state of the components/data-updates to disk - [todo] no other processes will also be able to lock "done" cards - [todo] If a subprocess is gracefully shutdown, then: - [todo] the card manager will call `current.card.done()` which will avoid writes from other processes 3. Post Task Execution: - sub processes: - Should have **ideally** ended and released all locks and ran the finalize method. - If not then what ever is the last component state dumped in the subprocess will be the components loaded for the card. - Main process: - For each card, - call the render method. - [todo] If the card was locked by an outside process then seek it's component state and load it. - if the card --- metaflow/plugins/cards/card_creator.py | 33 ++- metaflow/plugins/cards/card_decorator.py | 16 +- metaflow/plugins/cards/card_locks.py | 223 ++++++++++++++++++ .../plugins/cards/component_serializer.py | 58 ++++- .../plugins/cards/runtime_collector_state.py | 80 +++++-- 5 files changed, 363 insertions(+), 47 deletions(-) create mode 100644 metaflow/plugins/cards/card_locks.py diff --git a/metaflow/plugins/cards/card_creator.py b/metaflow/plugins/cards/card_creator.py index 64197552f00..1a4ab3e5908 100644 --- a/metaflow/plugins/cards/card_creator.py +++ b/metaflow/plugins/cards/card_creator.py @@ -9,6 +9,12 @@ ASYNC_TIMEOUT = 30 +def warning_message(message, logger=None, ts=False): + msg = "[@card WARNING] %s" % message + if logger: + logger(msg, timestamp=ts, bad=True) + + class CardProcessManager: """ This class is responsible for managing the card creation processes. @@ -44,18 +50,25 @@ def _remove_card_process(cls, carduuid): class CardCreator: + + state_manager = None # of type `CardStateManager` + def __init__( self, - top_level_options=None, + base_command=None, pathspec=None, # TODO Add a pathspec somewhere here so that it can instantiate correctly. ): - self._top_level_options = top_level_options + self._base_command = base_command self._pathspec = pathspec + @property + def pathspec(self): + return self._pathspec + def _dump_state_to_dict(self): return { - "top_level_options": self._top_level_options, + "base_command": self._base_command, "pathspec": self._pathspec, } @@ -124,12 +137,13 @@ def _run_cards_subprocess( json.dump(data, data_file) data_file.seek(0) - executable = sys.executable - cmd = [ - executable, - sys.argv[0], - ] - cmd += self._top_level_options + [ + if self.state_manager is not None: + self.state_manager._save_card_state( + card_uuid, components=component_strings, data=data + ) + + cmd = [] + cmd += self._base_command + [ "card", "create", runspec, @@ -169,6 +183,7 @@ def _run_cards_subprocess( timeout=decorator_attributes["timeout"], wait=wait, ) + # warning_message(str(cmd), logger) if fail: resp = "" if response is None else response.decode("utf-8") logger( diff --git a/metaflow/plugins/cards/card_decorator.py b/metaflow/plugins/cards/card_decorator.py index ed89c1274b1..29bddf798ba 100644 --- a/metaflow/plugins/cards/card_decorator.py +++ b/metaflow/plugins/cards/card_decorator.py @@ -83,10 +83,6 @@ def __init__(self, *args, **kwargs): self._current_step = None self._state_manager = None # This is set on a per TASK basis - @classmethod - def _set_collector_state_dump_file(cls, file): - cls._collector_state_dump_file = file - @classmethod def _set_card_creator(cls, card_creator): cls.card_creator = card_creator @@ -186,7 +182,7 @@ def task_pre_step( self._register_event("pre-step") self._set_card_creator( CardCreator( - top_level_options=self._create_top_level_args(), + base_command=self._create_base_command(), pathspec=pathspec, ) ) @@ -230,7 +226,8 @@ def task_exception( def _dump_state_of_collector_to_file(self, pathspec): # Create a directory in the DATASTORE_LOCAL_DIR like `DATASTORE_LOCAL_DIR/` self._state_manager = CardStateManager(pathspec) - self._state_manager.save(current.card._dump_state_to_dict()) + self.card_creator.state_manager = self._state_manager + self._state_manager.component_collector.save(current.card._dump_state_to_dict()) def task_finished( self, step_name, flow, graph, is_task_ok, retry_count, max_user_code_retries @@ -250,6 +247,7 @@ def task_finished( if is_task_ok: self.card_creator.create(mode="render", **create_options) self.card_creator.create(mode="refresh", final=True, **create_options) + current.card._finished() if self._state_manager is not None: self._state_manager.done() @@ -264,7 +262,7 @@ def _options(mapping): if not isinstance(value, bool): yield to_unicode(value) - def _create_top_level_args(self): + def _create_base_command(self): top_level_options = { "quiet": True, "metadata": self._metadata.TYPE, @@ -277,4 +275,6 @@ def _create_top_level_args(self): # We don't provide --with as all execution is taking place in # the context of the main process } - return list(self._options(top_level_options)) + executable = sys.executable + cmd = [executable, sys.argv[0]] + return cmd + list(self._options(top_level_options)) diff --git a/metaflow/plugins/cards/card_locks.py b/metaflow/plugins/cards/card_locks.py new file mode 100644 index 00000000000..202fa3e03ce --- /dev/null +++ b/metaflow/plugins/cards/card_locks.py @@ -0,0 +1,223 @@ +from .runtime_collector_state import _get_card_state_directory +import fcntl +import os +import functools + + +class CardLock: + + POSSIBLE_STATES = ["locked_by_me", "unlocked", "locked_by_other"] + + lock_base_path = None + + def __init__(self, carduuid): + self._carduuid = carduuid + self._lock_path = os.path.join(self.lock_base_path, self._carduuid + ".lock") + self._lock_file = None + self._current_state = "unlocked" + + def _try_to_acquire_lock(self): + try: + lock_file = open(self._lock_path, "w") + fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + return lock_file + except BlockingIOError: + return False + + @property + def owned_by_others(self): + return self._current_state == "locked_by_other" + + @property + def owns_the_lock(self): + return self._current_state == "locked_by_me" + + @property + def seems_unlocked(self): + return self._current_state == "unlocked" + + def test_lockability(self): + if self.owns_the_lock: + return True + elif self.seems_unlocked: + _lock_file = self._try_to_acquire_lock() + if _lock_file: + _lock_file.close() + return True + self._current_state = "locked_by_other" + return False + else: + return False + + def lock(self): + if self.owns_the_lock: + return True + elif self.seems_unlocked: + self._lock_file = self._try_to_acquire_lock() + if self._lock_file: + self._current_state = "locked_by_me" + return True + else: + self._current_state = "locked_by_other" + return False + else: + return False + + def unlock(self): + if self.owns_the_lock: + fcntl.flock(self._lock_file, fcntl.LOCK_UN) + self._lock_file.close() + self._current_state = "unlocked" + return True + else: + return False + + def __del__(self): + if self.owns_the_lock: + self.unlock() + + +class LockableCardDict: + """ + This is a class that will help us ensure that when `current.card` interface is called in an outside subprocess, + then we are able to call the `current.card` interface in outside subprocesses without any issues. + + Some constraints which the `LockableCardDict` aims to fulfill: + + > No two processes can lock the same card at the same time. Once a card is locked by a process, no other process can lock it. + (Meaning the first process to lock the card locks it for-ever. No other process can lock it after that). + + The life-cycle of cards is as follows: + 1. Pre-task execution : + - @card decorator instantiates a CardComponentCollector with a LockableCardDict set to `pre_task_startup=True` and empty dictionary for it's `card_component_store` + - This ensures that new CardComponentManagers can be trivially added by the main process. + - Once main process is done adding all cards, we run `finalize` method : + - it will set `pre_task_startup=False` and set post_task_startup=True + - it will instantiate the locks. Won't check if they are already possessed by others since the code is being run in the main process and + task execution has not started yet; (All lock alteration happens once user code starts executing). + - Upon finishing finalize `current.card` will also dump the state of `current.card` for being available to any outside subprocesses. + 2. Task Execution: + - Main process : + - Which ever cards the main thread update's will be locked by the main thread. + - The cards which are not accessed / updated by the main thread, can be locked by other processes. + - Locked cards will provide the same type of object back to the user, but the `card_proc` method will be replaced with a + method that will warn the user that the card is locked. + - Subprocesses: + - calls the `get_runtime_card` method to get the `current.card` object + - the method will seek out the `current.card`'s state location and load the state. + - the state of the card specifies what ever task spec's component collector's state + - will be able to call render_runtime and refresh from the cli + - `refresh` will also dump the state of the components to disk. + - [todo] when finished will call `current.card.done()` which will dump the state of the components/data-updates to disk + - [todo] no other processes will also be able to lock "done" cards + - [todo] If a subprocess is gracefully shutdown, then: + - [todo] the card manager will call `current.card.done()` which will avoid writes from other processes + 3. Post Task Execution: + - sub processes: + - Should have **ideally** ended and released all locks and ran the finalize method. + - If not then what ever is the last component state dumped in the subprocess will be the components loaded for the card. + - Main process: + - For each card, + - call the render method. + - [todo] If the card was locked by an outside process then seek it's component state and load it. + - if the card was not locked by an outside process then use the in-memory component state of the card. + """ + + def __init__( + self, + card_component_store, + lock_base_path, + pre_task_startup=False, + post_task_startup=True, + ): + self._card_component_store = card_component_store + self._card_locks = {} + self._lock_base_path = lock_base_path + self._pre_task_startup = pre_task_startup + self._post_task_startup = post_task_startup + self._init_locks() + + def _init_locks(self): + if self._pre_task_startup: + return + + CardLock.lock_base_path = self._lock_base_path + for carduuid in self._card_component_store: + self._card_locks[carduuid] = CardLock(carduuid) + + if not self._post_task_startup: + return + + # Since locks are initializing here we can test lockability here. + # As this code may be instantiated in a subprocess outside the main process, + # we need to ensure that when then LockableCardDict initializes, it can keeps only + # unlocked cards in the card_component_store. + # At this point when the code is running, the owning process of this object will + # not have any locks that it owns since all objects are being initialized. + for carduuid in self._card_component_store: + self._card_locks[carduuid].test_lockability() + if not self._card_locks[carduuid].seems_unlocked: + self._disable_rendering_for_locked_card(carduuid) + + def finalize(self): + # This method will only be called once by the CardComponent collector in scope of the + # @card decorator + self._pre_task_startup = False + self._init_locks() + self._post_task_startup = True + + def _disable_rendering_for_locked_card(self, carduuid): + def _wrap_method(instance, method_name): + def wrapper(*args, **kwargs): + instance._warning( + "Card is locked by another process. " + "This card will not be rendered/refreshed." + ) + return None + + # Bind the new method to the instance + setattr(instance, method_name, wrapper) + + _wrap_method(self._card_component_store[carduuid], "_card_proc") + + def _lock_check_for_getter(self, key): + # Here the key is uuid, so it's an internal things. + # Users will never have access to this object + assert key in self._card_locks, "Card not found in card store." + if self._post_task_startup: + if self._card_locks[key].seems_unlocked: + if not self._card_locks[key].lock(): + # If a card was locked then the in-memory status + # here will also flip to locked. + self._disable_rendering_for_locked_card(key) + + def _lock_check_for_setter(self, key): + if self._post_task_startup: + if self._card_locks[key].seems_unlocked: + if not self._card_locks[key].lock(): + return False + elif self._card_locks[key].owned_by_others: + return False + return True + + def unlock_everything(self): + for key in self._card_locks: + self._card_locks[key].unlock() + + def __contains__(self, key): + return key in self._card_component_store + + def __getitem__(self, key): + self._lock_check_for_getter(key) + return self._card_component_store[key] + + def __setitem__(self, key, value): + self._card_component_store[key] = value + if not self._lock_check_for_setter(key): + self._disable_rendering_for_locked_card(key) + + def items(self): + return self._card_component_store.items() + + def __len__(self): + return len(self._card_component_store) diff --git a/metaflow/plugins/cards/component_serializer.py b/metaflow/plugins/cards/component_serializer.py index 62f326be1cd..7c868c7a287 100644 --- a/metaflow/plugins/cards/component_serializer.py +++ b/metaflow/plugins/cards/component_serializer.py @@ -1,6 +1,8 @@ from .card_modules import MetaflowCardComponent from .card_modules.basic import ErrorComponent, SectionComponent from .card_modules.components import UserComponent +from .card_locks import LockableCardDict +from .runtime_collector_state import _get_card_state_directory from functools import partial import uuid import json @@ -251,7 +253,11 @@ def _card_proc(self, mode): mode=mode, ) - def refresh(self, data=None, force=False): + def refresh( + self, + data=None, + force=False, + ): # todo make this a configurable variable self._latest_user_data = data nu = time.time() @@ -363,7 +369,7 @@ def _serialize_store(card_manager_dict): return state_object @classmethod - def _load_state(cls, state_dict, logger): + def _load_state(cls, state_dict, logger, state_manager): from .card_creator import CardCreator def _load_stores(card_manager_dict): @@ -375,9 +381,15 @@ def _load_stores(card_manager_dict): return loaded_store card_creator = CardCreator(**state_dict["card_creator"]) - collector_object = cls(logger=logger, card_creator=card_creator) - collector_object._card_component_store = _load_stores( - state_dict["card_component_store"] + card_creator.state_manager = state_manager + collector_object = cls( + logger=logger, card_creator=card_creator, inside_main_process=False + ) + collector_object._card_component_store = LockableCardDict( + _load_stores(state_dict["card_component_store"]), + _get_card_state_directory(card_creator.pathspec), + pre_task_startup=False, + post_task_startup=True, ) collector_object._cards_meta = state_dict["cards_meta"] collector_object._card_id_map = state_dict["card_id_map"] @@ -385,14 +397,18 @@ def _load_stores(card_manager_dict): collector_object._no_warnings = state_dict["no_warnings"] return collector_object - def __init__(self, logger=None, card_creator=None): + def __init__(self, logger=None, card_creator=None, inside_main_process=True): from metaflow.metaflow_config import CARD_NO_WARNING - self._card_component_store = ( + self._prefinalize_cards_dict = {} + self._card_component_store = { # Each key in the dictionary is the UUID of an individual card. # value is of type `CardComponentManager`, holding a list of MetaflowCardComponents for that particular card - {} - ) + # When the @card decorator instantiates the CardComponentCollector, it will first have a {} for this variable. + # Once the decorator adds all the cards to the collector and calls `_finalize`, this variable will turn into a lockable dictionary. + # This lockable dictionary will help ensure that this same object can be instantiated outside metaflow + # and we can dis-allow concurrent access to the same card. + } self._cards_meta = ( {} ) # a `dict` of (card_uuid, `dict)` holding all metadata about all @card decorators on the `current` @step. @@ -409,6 +425,7 @@ def __init__(self, logger=None, card_creator=None): "update_no_id": False, } self._no_warnings = True if CARD_NO_WARNING else False + self._inside_main_process = inside_main_process @staticmethod def create_uuid(): @@ -455,7 +472,7 @@ def _add_card( card_options=card_options, ) self._cards_meta[card_uuid] = card_metadata - self._card_component_store[card_uuid] = CardComponentManager( + self._prefinalize_cards_dict[card_uuid] = CardComponentManager( card_uuid=card_uuid, decorator_attributes=decorator_attributes, card_creator=self._card_creator, @@ -473,6 +490,8 @@ def _warning(self, message): self._log(msg, timestamp=False, bad=True) def _add_warning_to_cards(self, warn_msg): + # FIXME: This should behave based on the type of card it is updating. + # All cards shouldnt get a warning component. if self._no_warnings: return for card_id in self._card_component_store: @@ -506,6 +525,7 @@ def _finalize(self): - The `self._default_editable_card` holds the uuid of the card that will have access to the `append`/`extend` methods. 2. Resolving edge cases where @card `id` argument may be `None` or have a duplicate `id` when there are more than one editable cards. 3. Resolving the `self._default_editable_card` to the card with the`customize=True` argument. + 4. Finalizing all the locks in the `self._card_component_store` """ all_card_meta = list(self._cards_meta.values()) for c in all_card_meta: @@ -518,7 +538,7 @@ def _finalize(self): editable_cards_meta = [c for c in all_card_meta if c["editable"]] if len(editable_cards_meta) == 0: - return + return self._init_lockable_card_store() # Create the `self._card_id_map` lookup table which maps card `id` to `uuid`. # This table has access to all cards with `id`s set to them. @@ -531,7 +551,7 @@ def _finalize(self): # If there is only one editable card then this card becomes `self._default_editable_card` if len(editable_cards_meta) == 1: self._default_editable_card = editable_cards_meta[0]["uuid"] - return + return self._init_lockable_card_store() # Segregate cards which have id as none and those which don't. not_none_id_cards = [c for c in editable_cards_meta if c["card_id"] is not None] @@ -579,6 +599,17 @@ def _finalize(self): # since `editable_cards_meta` hold only `editable=True` by default we can just set this card here. self._default_editable_card = customize_cards[0]["uuid"] + return self._init_lockable_card_store() + + def _init_lockable_card_store(self): + self._card_component_store = LockableCardDict( + self._prefinalize_cards_dict, + _get_card_state_directory(self._card_creator.pathspec), + pre_task_startup=True, + post_task_startup=False, + ) + self._card_component_store.finalize() + def __getitem__(self, key): """ Choose a specific card for manipulation. @@ -793,6 +824,9 @@ def _serialize_components(self, card_uuid): return [] return self._card_component_store[card_uuid]._serialize_components() + def _finished(self): + self._card_component_store.unlock_everything() + def _render_card_component(component): if not _component_is_valid(component): diff --git a/metaflow/plugins/cards/runtime_collector_state.py b/metaflow/plugins/cards/runtime_collector_state.py index 2d07ffd81a3..1bbfb5fc9d1 100644 --- a/metaflow/plugins/cards/runtime_collector_state.py +++ b/metaflow/plugins/cards/runtime_collector_state.py @@ -1,40 +1,79 @@ from metaflow.metaflow_config import DATASTORE_LOCAL_DIR import os import hashlib +import shutil import tempfile import json CARD_STATE_ROOT_DIR = os.path.join(DATASTORE_LOCAL_DIR, "mf.card_state") +class StateContainer: + def __init__( + self, + file_name, + ): + os.makedirs(os.path.dirname(file_name), exist_ok=True) + self._file_name = file_name + self._file_created = os.path.exists(self._file_name) + + @property + def is_present(self): + return os.path.exists(self._file_name) + + def load(self): + with open(self._file_name, "r") as _file: + state = json.load(_file) + return state + + def save(self, state): + with open(self._file_name, "w") as _file: + json.dump(state, _file) + self._file_created = True + + def done(self): + if self._file_created: + os.remove(self._file_name) + + class CardStateManager: + + card_state = {} + def __init__(self, pathspec) -> None: self._pathspec = pathspec - # Create a directory that will store the card state. - os.makedirs(CARD_STATE_ROOT_DIR, exist_ok=True) self._card_state_directory = _get_card_state_directory(pathspec) - self._card_state_file = _get_card_state_file(pathspec) + self._component_collector_state = StateContainer( + _get_card_collector_state_file_path(pathspec) + ) self._state_file_created = False - def load(self): - with open(self._card_state_file, "r") as _state_file: - state = json.load(_state_file) - return state - - def save(self, state): - os.makedirs(self._card_state_directory, exist_ok=True) - _state_file = open(self._card_state_file, "w") - json.dump(state, _state_file) - _state_file.close() - self._state_file_created = True + @property + def component_collector(self): + return self._component_collector_state def done(self): if self._state_file_created: - os.remove(self._card_state_file) - os.rmdir(self._card_state_directory) + self._component_collector_state.done() + shutil.rmtree(self._card_state_directory) + + def _save_card_state(self, uuid, components=None, data=None): + if uuid not in self.card_state: + self.card_state[uuid] = { + "components": StateContainer( + os.path.join(self._card_state_directory, f"{uuid}.components.json") + ), + "data": StateContainer( + os.path.join(self._card_state_directory, f"{uuid}.data.json") + ), + } + if components is not None: + self.card_state[uuid]["components"].save(components) + if data is not None: + self.card_state[uuid]["data"].save(data) -def _get_card_state_file(pathspec): +def _get_card_collector_state_file_path(pathspec): return os.path.join(_get_card_state_directory(pathspec), "card_state.json") @@ -46,6 +85,9 @@ def _get_card_state_directory(pathspec): def get_realtime_cards( pathspec=None, ): + # FIXME : What is a way we can gaurentee that `CardStateManager` + # Loads the correct path based on what ever it's CWD is set in the subprocess + # calling this. from .component_serializer import CardComponentCollector from metaflow import current from metaflow.cli import logger @@ -64,5 +106,7 @@ def get_realtime_cards( card_state_manager = CardStateManager(pathspec) return CardComponentCollector._load_state( - state_dict=card_state_manager.load(), logger=logger + state_dict=card_state_manager.component_collector.load(), + logger=logger, + state_manager=card_state_manager, )