Fix config cannot be updated

golem/envs/__init__.py

@@ -3,10 +3,10 @@
 from copy import deepcopy
 from enum import Enum
 from logging import Logger, getLogger
-from threading import RLock
+from threading import Lock, RLock
 
 from typing import Any, Callable, Dict, List, Optional, NamedTuple, Union, \
-    Sequence, Iterable, ContextManager, Set, Tuple, TYPE_CHECKING
+    Sequence, Iterable, ContextManager, Set, Tuple, Type, TYPE_CHECKING
 
 from dataclasses import dataclass, field
 from twisted.internet.threads import deferToThread
@@ -599,3 +599,50 @@ def listen(
             listener: EnvEventListener
     ) -> None:
         self._event_listeners.setdefault(event_type, set()).add(listener)
+
+
+def delayed_config(cls: Type[Environment]) -> Type[Environment]:
+    """
+    This class decorator allows to save config update and apply it, when env is
+    disabled.
+
+    Mutex prevents the following scenario
+    Thread 1                Thread 2
+    call apply_next_config
+                            call update_config
+                            status is disabled, so calls
+                                super().update_config
+    cls.update_config
+        with _next_config
+    """
+    # FIXME workaround https://github.com/python/mypy/issues/5865
+    cls2: Any = cls
+
+    class DelayedConfigWrapper(cls2):
+        def __init__(self, *args, **kwargs) -> None:
+            super().__init__(*args, **kwargs)
+            self._next_config: Optional[EnvConfig] = None
+            self._config_lock = Lock()
+
+            def apply_next_config(_):
+                with self._config_lock:
+                    if self._next_config is None:
+                        return
+                    self._logger.debug("Applying saved config")
+                    config = self._next_config
+                    self._next_config = None
+                    cls.update_config(self, config)
+
+            self.listen(EnvEventType.DISABLED, apply_next_config)
+
+        def update_config(self, new_config: EnvConfig) -> None:
+            with self._config_lock:
+                if self._status == EnvStatus.DISABLED:
+                    self._logger.debug("Config applied immediately")
+                    super().update_config(new_config)
+                    return
+
+                self._logger.debug("Config saved for later")
+                self._next_config = new_config
+
+    return DelayedConfigWrapper

golem/envs/docker/cpu.py

@@ -41,7 +41,8 @@
     RuntimePayload,
     RuntimeStatus,
     UsageCounter,
-    UsageCounterValues
+    UsageCounterValues,
+    delayed_config,
 )
 from golem.envs.docker import DockerRuntimePayload, DockerPrerequisites
 from golem.envs.docker.whitelist import Whitelist
@@ -496,6 +497,7 @@ def usage_counter_values(self) -> UsageCounterValues:
         return deepcopy(self._counters)
 
 
+@delayed_config
 class DockerCPUEnvironment(EnvironmentBase):
 
     MIN_MEMORY_MB: ClassVar[int] = 1024

golem/task/taskcomputer.py

@@ -77,7 +77,7 @@ def __init__(
         )
         self._new_computer = NewTaskComputer(
             env_manager=env_manager,
-            work_dir=task_server.get_task_computer_root(),
+            work_dir=Path(task_server.get_task_computer_root()),
             stats_keeper=self.stats
         )
 
@@ -263,10 +263,8 @@ def change_config(
             config_desc: 'ClientConfigDescriptor',
             in_background: bool = True
     ) -> defer.Deferred:
-        work_dir = Path(self._task_server.get_task_computer_root())
-        yield self._new_computer.change_config(
-            config_desc=config_desc,
-            work_dir=work_dir)
+        self._new_computer.change_config(
+            config_desc=config_desc)
         return (yield self._old_computer.change_config(
             config_desc=config_desc,
             in_background=in_background))
@@ -480,13 +478,9 @@ def get_current_computing_env(self) -> 'Optional[EnvId]':
     def change_config(
             self,
             config_desc: 'ClientConfigDescriptor',
-            work_dir: Path
-    ) -> defer.Deferred:
-        assert not self._is_computing()
-        self._work_dir = work_dir
-
+    ) -> None:
         config_dict = dict(
-            work_dirs=[work_dir],
+            work_dirs=[self._work_dir],
             cpu_count=config_desc.num_cores,
             memory_mb=scale_memory(
                 config_desc.max_memory_size,
@@ -505,8 +499,6 @@ def change_config(
             # TODO: GPU options in config_dict
             docker_gpu.update_config(DockerGPUConfig(**config_dict))
 
-        return defer.succeed(None)
-
     def quit(self):
         if self.has_assigned_task():
             self.task_interrupted()

scripts/node_integration_tests/nodes/provider/configure_or_die.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+
+import logging
+from unittest.mock import patch
+
+from twisted.internet.defer import inlineCallbacks
+
+from golemapp import main
+
+from golem.client import Client
+from golem.task.taskserver import TaskServer
+
+
+def on_exception():
+    logging.critical("#### Integration test failed ####")
+
+
+client_change_config_orig = Client.change_config
+
+
+def client_change_config(self: Client, *args, **kwargs):
+    try:
+        client_change_config_orig(self, *args, **kwargs)
+    except:  # noqa pylint: disable=broad-except
+        on_exception()
+
+
+task_server_change_config_orig = TaskServer.change_config
+
+
+@inlineCallbacks
+def task_server_change_config(self: TaskServer, *args, **kwargs):
+    try:
+        yield task_server_change_config_orig(self, *args, **kwargs)
+    except:  # noqa pylint: disable=broad-except
+        on_exception()
+
+
+with patch("golem.client.Client.change_config",
+           client_change_config), \
+     patch("golem.task.taskserver.TaskServer.change_config",
+           task_server_change_config):
+    main()

scripts/node_integration_tests/playbooks/golem/reconfigure_provider_while_working/playbook.py

@@ -0,0 +1,73 @@
+import time
+from functools import partial
+
+from scripts.node_integration_tests import helpers
+from ...test_config_base import NodeId
+from ..task_api.playbook import Playbook as BasePlaybook
+
+
+class Playbook(BasePlaybook):
+    def wait_for_computing_task(self):
+        def on_success(result):
+            state = result['provider_state']
+            print(f"provider state: {state}")
+            if state['status'] == 'Computing':
+                self.next()
+            else:
+                time.sleep(10)
+
+        def on_error(_):
+            print(f"failed getting provider stats")
+            self.fail()
+        return self.call(NodeId.provider, 'comp.tasks.stats',
+                         on_success=on_success, on_error=on_error)
+
+    def ui_stop(self, node_id: NodeId):
+        def on_success(_):
+            print(f"stopped {node_id.value}")
+            self.next()
+
+        def on_error(_):
+            print(f"stopping {node_id.value} failed")
+            self.fail()
+        return self.call(node_id, 'ui.stop', on_success=on_success,
+                         on_error=on_error)
+
+    def change_config(self, node_id: NodeId):
+        opts = {
+            "node_name": "a new name",
+        }
+
+        def on_success(_):
+            print(f"reconfigured {node_id.value}")
+            time.sleep(10)  # give time for async operations to process
+            self.next()
+
+        def on_error(_):
+            print(f"reconfiguring {node_id.value} failed")
+            self.fail()
+
+        return self.call(node_id, 'env.opts.update', opts,
+                         on_success=on_success, on_error=on_error)
+
+    def check_if_test_failed(self, node_id: NodeId):
+        test_failed = bool(helpers.search_output(
+            self.output_queues[node_id],
+            ".*#### Integration test failed ####.*"))
+
+        if test_failed:
+            self.fail("found failure marker in log")
+
+        print("no failure marker found in log")
+        self.next()
+
+    steps = BasePlaybook.initial_steps + (
+        BasePlaybook.step_enable_app,
+        BasePlaybook.step_create_task,
+        BasePlaybook.step_get_task_id,
+        BasePlaybook.step_get_task_status,
+        wait_for_computing_task,
+        partial(ui_stop, node_id=NodeId.provider),
+        partial(change_config, node_id=NodeId.provider),
+        partial(check_if_test_failed, node_id=NodeId.provider),
+    )

scripts/node_integration_tests/playbooks/golem/reconfigure_provider_while_working/test_config.py

@@ -0,0 +1,9 @@
+from ...test_config_base import NodeId
+
+from ..task_api.test_config import TestConfig as TestConfigBase
+
+
+class TestConfig(TestConfigBase):
+    def __init__(self):
+        super().__init__()
+        self.nodes[NodeId.provider].script = 'provider/configure_or_die'

tests/golem/envs/docker/cpu/test_env.py

@@ -329,11 +329,6 @@ def test_wrong_type(self):
         with self.assertRaises(AssertionError):
             self.env.update_config(object())
 
-    def test_enabled_status(self):
-        self.env._status = EnvStatus.ENABLED
-        with self.assertRaises(ValueError):
-            self.env.update_config(Mock(spec=DockerCPUConfig))
-
     @patch_env('_validate_config', side_effect=ValueError)
     def test_invalid_config(self, validate):
         config = Mock(spec=DockerCPUConfig)

tests/golem/envs/test_env.py

@@ -2,13 +2,18 @@
 from unittest import TestCase
 from unittest.mock import Mock, patch
 
+from dataclasses import dataclass
+
+from twisted.internet import defer
+
 from golem.envs import (
     EnvConfig,
     EnvEvent,
     EnvEventType,
     EnvironmentBase,
     EnvStatus,
-    Prerequisites
+    Prerequisites,
+    delayed_config,
 )
 
 
@@ -121,3 +126,87 @@ def test_re_register(self):
         self.assertEqual(self.env._event_listeners, {
             EnvEventType.ERROR_OCCURRED: {listener}
         })
+
+
+@dataclass
+class MyConfig(EnvConfig):
+    i: int
+
+    def to_dict(self) -> dict:
+        pass
+
+    @staticmethod
+    def from_dict(data):
+        pass
+
+
+@delayed_config
+class MyEnv(EnvironmentBase):
+    def __init__(self, config: MyConfig) -> None:
+        super().__init__()
+        self._config = config
+
+    def update_config(self, config: EnvConfig) -> None:
+        assert isinstance(config, MyConfig)
+        self._logger.debug("dupa %r", self._event_listeners)
+        if self._status != EnvStatus.DISABLED:
+            raise ValueError
+        self._config = config
+        self._config_updated(config)
+
+    def config(self) -> MyConfig:
+        return self._config
+
+    @classmethod
+    def supported(cls):
+        raise NotImplementedError
+
+    def prepare(self):
+        raise NotImplementedError
+
+    def clean_up(self):
+        raise NotImplementedError
+
+    def run_benchmark(self):
+        raise NotImplementedError
+
+    def parse_prerequisites(self, prerequisites_dict):
+        raise NotImplementedError
+
+    def install_prerequisites(self, prerequisites):
+        raise NotImplementedError
+
+    def parse_config(self, config_dict):
+        raise NotImplementedError
+
+    def supported_usage_counters(self):
+        raise NotImplementedError
+
+    def runtime(self, payload, config=None):
+        raise NotImplementedError
+
+
+def execute(f, *args, **kwargs):
+    try:
+        return defer.succeed(f(*args, **kwargs))
+    except Exception as exc:  # pylint: disable=broad-except
+        return defer.fail(exc)
+
+
+@patch('golem.envs.deferToThread', execute)
+class TestDelayedConfig(TestCase):
+
+    def setUp(self) -> None:
+        config = MyConfig(i=1)
+        self.env = MyEnv(config)
+
+    def test_update_config_when_disabled(self):
+        self.env.update_config(MyConfig(i=2))
+        self.assertEqual(self.env.config().i, 2)
+
+    def test_update_config_when_enabled(self):
+        self.env._env_enabled()
+        self.env.update_config(MyConfig(i=2))
+        self.assertEqual(self.env.config().i, 1)
+        self.env._env_disabled()
+        self.assertEqual(self.env.config().i, 2)