Syncable._param_change not robust

[MarcSkovMadsen]

I'm on Panel 0.11.3

I have a complicated app where I show the currently running Job. The job runs in background thread. When the job is finished its no longer shown in the app.

I can see that sometimes (random) Syncable._param_change fails with a

  File "/opt/conda/lib/python3.8/site-packages/panel/reactive.py", line 237, in _param_change
    for ref, (model, parent) in self._models.items():
RuntimeError: dictionary changed size during iteration

The code of Syncable._param_change is

 def _param_change(self, *events):
        msgs = []
        for event in events:
            msg = self._process_param_change({event.name: event.new})
            if msg:
                msgs.append(msg)

        events = {event.name: event for event in events}
        msg = {k: v for msg in msgs for k, v in msg.items()}
        if not msg:
            return

        for ref, (model, parent) in self._models.items():
                self._apply_update(events, msg, model, ref)

I can see that when it fails the value of model is CheckboxGroup(id='7669', ...). I can see that the value of self._models.items() is dict_items([]). My understanding is what triggered _param_change was that I set job.running=False. But during _param_change the job is removed from the application leading to self._models_items() changing.

My guess is that this should be supported and somehow the iteration should be done on a copy of self._models.items(). Or alternatively error handling should be introduced???

[MarcSkovMadsen]

Example

changed-size-during.mp4

The code below triggers the exception at random times when clicking "Run" button multiple times.

"""Contains functionality for running Jobs in the Background"""
import datetime
import threading
import time
import traceback
from queue import Queue

import panel as pn
import param


class Logger(param.Parameterized):
    """A component for logging messages and for viewing the log


    >>> logger=Logger()
    >>> logger.info("Started Job")
    >>> logger.error("Job Crashed")

    You can add logger.param.log or logger.view to your application if you want to see the log
    value updating.
    """

    log = param.String(doc="""The text send to the log""", constant=True)
    view = param.Parameter(
        precedence=-1,
        doc="""A view of the log. You can add this to your application""",
        constant=True,
    )

    _widgets = {
        # Should eventually be the Panel 0.12 Terminal
        "log": {"widget_type": pn.widgets.Ace, "sizing_mode": "stretch_both", "readonly": True}
    }

    def __init__(self, **params):
        super().__init__(**params)

        with param.edit_constant(self):
            self.view = pn.Param(self, widgets=self._widgets, height=200, name="Log")

    def info(self, message):
        """Adds an *info* message to the log"""
        with param.edit_constant(self):
            self.log += (
                datetime.datetime.utcnow().strftime("%Y%m%d %H:%M:%S") + " INFO - " + message + "\n"
            )

    def error(self, message):
        """Adds an *error* message to the log"""
        with param.edit_constant(self):
            self.log += (
                datetime.datetime.utcnow().strftime("%Y%m%d %H:%M:%S")
                + " ERROR - "
                + message
                + "\n"
            )

    def clear(self):
        """Clears the log, i.e. sets .log=''"""
        with param.edit_constant(self):
            self.log = ""


class Job(param.Parameterized):
    """A Job Component

    >>> import time
    >>> def func():
    ...     time.sleep(2)
    ...     return "result set"
    >>> job=Job(target=func, name="My Job")
    >>> job.run()
    >>> job.result
    'result set'

    Add job.param.result to your application if you want to view the result
    """

    target = param.Parameter(precedence=-1, doc="The target function to run.", constant=True)
    panels_loading = param.List(
        precedence=-1,
        doc="A list of components to set .loading to True for while running the target function",
        constant=True,
    )

    run = param.Action(doc="Runs the job, i.e. the target function.", constant=True)
    running = param.Boolean(doc="True if the job is currently running.", constant=True)

    result = param.Parameter(
        doc="The return value of the target function. Or the exception raised if any.",
        constant=True
    )
    last_run_utc = param.Date(doc="The date and time (UTC) the job was last run.", constant=True)

    logger = param.ClassSelector(class_=Logger, precedence=-1, doc="An optional Logger")

    view = param.Parameter(precedence=-1, doc="""Add this to your app for viewing the component""")

    _widgets = {
        "running": {"disabled": True},
        "result": {"disabled": True},
        "last_run_utc": {"disabled": True},
    }

    def __init__(self, **params):
        super().__init__(**params)

        with param.edit_constant(self):
            self.run = self._run
            self.view = pn.Param(self, widgets=self._widgets)

    def _run(self, *_):
        with param.edit_constant(self):
            self._start()
            try:
                self.result = self.target()
            except Exception as ex:  # pylint: disable=broad-except
                self._log_error(traceback.format_exc())
                self.result = ex
            self._finish()

    def _start(self):
        self._log_info("Started")
        self._set_panels_loading(True)
        self.running = True

    def _finish(self):
        try:
            self.running = False
        except RuntimeError:
            # C.f. https://github.com/holoviz/panel/issues/2631
            pass
        self._set_panels_loading(False)
        self.last_run_utc = datetime.datetime.utcnow()
        self._log_info("Finished")

    def _set_panels_loading(self, value: bool):
        if self.panels_loading:
            for panel in self.panels_loading.copy():
                try:
                    panel.loading = value
                except:  # pylint: disable=bare-except
                    pass

    def _log_info(self, message):
        if self.logger:
            self.logger.info(self.name + " - " + message)

    def _log_error(self, message):
        if self.logger:
            self.logger.error(self.name + " - " + message)

    def __str__(self):
        return self.name


class BackgroundWorker(param.Parameterized):
    """A BackgroundWorker Component

    The BackgroundWorker executes jobs from a queue in a background thread.

    This is nice if you want to avoid blocking your application while running a long running
    function.

    >>> import time
    >>> worker=BackgroundWorker()
    >>> def func():
    >>>     time.sleep(2)
    >>>     return "result set"
    >>> job=Job(target=func, name="My Job")
    >>> worker.submit(job)


    The code above will add the job to a queue and .run the job in the background when resources
    are available.

    You can submit as many jobs as you would like to the BackgroundWorker

    Add the .view to your application if you want to see the progress.
    """

    current = param.ClassSelector(class_=Job, doc="""The currently running job""")

    panels_loading = param.List(
        precedence=-1, doc="""A list of components to set .loading to True while working"""
    )

    index = param.Integer(doc="""The index of the current job. Used for progress reporting""")
    total = param.Integer(
        doc="""The total number of jobs appended to the queue. Used for progress reporting"""
    )

    logger = param.ClassSelector(class_=Logger, precedence=-1, doc="""An optional Logger""")
    view = param.ClassSelector(
        class_=pn.indicators.Progress,
        precedence=-1,
        doc="""Add this to your app for viewing the component""",
    )

    def __init__(self, **params):
        super().__init__(**params)
        self.queue = Queue()
        self.view = pn.indicators.Progress(
            value=-1, name="Background Worker", sizing_mode="stretch_width"
        )
        if not self.logger:
            self.logger=Logger()

        def callback():
            while True:
                if self.queue.empty():
                    self.current = None
                    self.index = 0
                    self.total = 0
                    self._set_panels_loading(False)

                job = self.queue.get()

                self._set_panels_loading(True)
                self.current = job
                self.index = self.index + 1

                job.run()

                self.queue.task_done()

        thread = threading.Thread(target=callback)
        thread.deamon = True
        thread.start()

    def submit(self, job: Job):
        "Submit the job to the queue and run it in the background when resources become available"
        if not job.logger:
            job.logger = self.logger
        self.queue.put(job)
        self.total += 1

    def _set_panels_loading(self, value: bool):
        if self.panels_loading:
            for panel in self.panels_loading.copy():
                try:
                    panel.loading = value
                except:  # pylint: disable=bare-except
                    pass

    @param.depends("total", "index", watch=True)
    def _update_progress_max(self):
        self.view.max = self.total
        if self.total == 0:
            # Hack to not have progres rotating when not needed
            self.view.value = -1
        else:
            self.view.value = min(self.index, self.total)

def test_app():
    pn.extension(sizing_mode="stretch_width")

    logger = Logger()

    SLEEP = 0.1

    def _func():
        time.sleep(SLEEP)
        return "result set"

    def _func_with_error():
        time.sleep(SLEEP)
        raise NotImplementedError("Not Implemented")

    panel = pn.Spacer(height=200, background="#111111")
    worker = BackgroundWorker(logger=logger, panels_loading=[panel])

    @pn.depends(current=worker.param.current)
    def get_current(current):
        column = pn.Column("#### Current Job")
        if current:
            column.append(pn.Param(current))
        return column

    run_button = pn.widgets.Button(name="Run")

    def _run_job(*_):
        job1 = Job(name="My job", target=_func)
        worker.submit(job1)

        job2 = Job(name="Job 2", target=_func)
        worker.submit(job2)
        worker.submit(job2)

        job3 = Job(name="Job with Error", target=_func_with_error)
        worker.submit(job3)
        worker.submit(job3)

    run_button.on_click(_run_job)

    return pn.Column(
        "#### Panel",
        panel,
        pn.Param(worker),
        worker.view,
        run_button,
        logger.view,
        get_current,
        sizing_mode="stretch_both",
    )

if __name__.startswith("bokeh"):
    test_app().servable()

[MarcSkovMadsen]

Workaround

If I wrap self._models.items() in a list I don't get an exception.

# pylint: disable=protected-access
# Fix Bug: https://github.com/holoviz/panel/issues/2631
def _param_change(self, *events):
    msgs = []
    for event in events:
        msg = self._process_param_change({event.name: event.new})
        if msg:
            msgs.append(msg)

    events = {event.name: event for event in events}
    msg = {k: v for msg in msgs for k, v in msg.items()}
    if not msg:
        return

    for ref, (model, _) in list(self._models.items()):
        self._apply_update(events, msg, model, ref)


pn.reactive.Syncable._param_change = _param_change
# pylint: enable=protected-access

[philippjfr]

Suspect there are probably a few such instances of things not being entirely thread safe. Hopefully we can catch them all before the 0.13 release.