🐛Fixing nodeports_v2 port.set() race condition (ITISFoundation#2587)

* adding regression test

* semplified and fixed regression test

* added a better method to save the status of the nodeports

* updated docstring

* adding API to save values in batch

* adding tests

* fixing unittests

* refactor to use different interface

* ports is now hashable. refactoring interface

* fixed interface and tets

* added int support

* removed hashing

* port has to be hashable

* not required to update ports before writing

* removed hashable, require by tests

* updating values to ports in parallel

* current refactor

* refactor and added tests

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/nodeports_v2.py

@@ -1,8 +1,10 @@
 import logging
+from collections import deque
 from pathlib import Path
-from typing import Any, Callable, Coroutine, Type
+from typing import Any, Callable, Coroutine, Dict, Type
 
 from pydantic import BaseModel, Field
+from servicelib.utils import logged_gather
 
 from ..node_ports_common.dbmanager import DBManager
 from ..node_ports_common.exceptions import PortNotFound, UnboundPortError
@@ -99,3 +101,22 @@ async def _auto_update_from_db(self) -> None:
             self.internal_inputs[input_key]._node_ports = self
         for output_key in self.internal_outputs:
             self.internal_outputs[output_key]._node_ports = self
+
+    async def set_multiple(self, port_values: Dict[str, ItemConcreteValue]) -> None:
+        """
+        Sets the provided values to the respective input or output ports
+        Only supports port_key by name, not able to distinguish between inputs
+        and outputs using the index.
+        """
+        tasks = deque()
+        for port_key, value in port_values.items():
+            # pylint: disable=protected-access
+            try:
+                tasks.append(self.internal_outputs[port_key]._set(value))
+            except UnboundPortError:
+                # not available try inputs
+                # if this fails it will raise another exception
+                tasks.append(self.internal_inputs[port_key]._set(value))
+
+        await logged_gather(*tasks)
+        await self.save_to_db_cb(self)

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/port.py

@@ -106,7 +106,7 @@ async def get(self) -> Optional[ItemConcreteValue]:
 
         return self._py_value_converter(value)
 
-    async def set(self, new_value: ItemConcreteValue) -> None:
+    async def _set(self, new_value: ItemConcreteValue) -> None:
         log.debug(
             "setting %s[%s] with value %s", self.key, self.property_type, new_value
         )
@@ -129,4 +129,8 @@ async def set(self, new_value: ItemConcreteValue) -> None:
 
         self.value = final_value
         self._used_default_value = False
+
+    async def set(self, new_value: ItemConcreteValue) -> None:
+        """sets a value to the port, by default it is also stored in the database"""
+        await self._set(new_value)
         await self._node_ports.save_to_db_cb(self._node_ports)

packages/simcore-sdk/tests/integration/test_node_ports_v2_nodeports2.py

@@ -7,6 +7,7 @@
 import filecmp
 import tempfile
 import threading
+from asyncio import gather
 from pathlib import Path
 from typing import Any, Callable, Dict, Type, Union
 from uuid import uuid4
@@ -15,6 +16,7 @@
 import pytest
 import sqlalchemy as sa
 from simcore_sdk import node_ports_v2
+from simcore_sdk.node_ports_common.exceptions import UnboundPortError
 from simcore_sdk.node_ports_v2 import exceptions
 from simcore_sdk.node_ports_v2.links import ItemConcreteValue
 from simcore_sdk.node_ports_v2.nodeports_v2 import Nodeports
@@ -554,3 +556,111 @@ async def test_file_mapping(
     assert received_file_link["path"] == file_id
     # received a new eTag
     assert received_file_link["eTag"]
+
+
+@pytest.fixture
+def int_item_value() -> int:
+    return 42
+
+
+@pytest.fixture
+def parallel_int_item_value() -> int:
+    return 142
+
+
+@pytest.fixture
+def port_count() -> int:
+    # the issue manifests from 4 ports onwards
+    # going for many more ports to be sure issue
+    # always occurs in CI or locally
+    return 20
+
+
+async def test_regression_concurrent_port_update_fails(
+    user_id: int,
+    project_id: str,
+    node_uuid: str,
+    special_configuration: Callable,
+    int_item_value: int,
+    parallel_int_item_value: int,
+    port_count: int,
+) -> None:
+    """
+    when using `await PORTS.outputs` test will fail
+    an unexpected status will end up in the database
+    """
+
+    outputs = [(f"value_{i}", "integer", None) for i in range(port_count)]
+    config_dict, _, _ = special_configuration(inputs=[], outputs=outputs)
+
+    PORTS = await node_ports_v2.ports(
+        user_id=user_id, project_id=project_id, node_uuid=node_uuid
+    )
+    await check_config_valid(PORTS, config_dict)
+
+    # when writing in serial these are expected to work
+    for item_key, _, _ in outputs:
+        await (await PORTS.outputs)[item_key].set(int_item_value)
+        assert (await PORTS.outputs)[item_key].value == int_item_value
+
+    # when writing in parallel and reading back,
+    # they fail, with enough concurrency
+    async def _upload_task(item_key: str) -> None:
+        await (await PORTS.outputs)[item_key].set(parallel_int_item_value)
+
+    # updating in parallel creates a race condition
+    results = await gather(*[_upload_task(item_key) for item_key, _, _ in outputs])
+    assert len(results) == port_count
+
+    # since a race condition was created when uploading values in parallel
+    # it is expected to find at least one mismatching value here
+    with pytest.raises(AssertionError) as exc_info:
+        for item_key, _, _ in outputs:
+            assert (await PORTS.outputs)[item_key].value == parallel_int_item_value
+    assert (
+        exc_info.value.args[0]
+        == f"assert {int_item_value} == {parallel_int_item_value}\n  +{int_item_value}\n  -{parallel_int_item_value}"
+    )
+
+
+async def test_batch_update_inputs_outputs(
+    user_id: int,
+    project_id: str,
+    node_uuid: str,
+    special_configuration: Callable,
+    port_count: int,
+) -> None:
+    outputs = [(f"value_out_{i}", "integer", None) for i in range(port_count)]
+    inputs = [(f"value_in_{i}", "integer", None) for i in range(port_count)]
+    config_dict, _, _ = special_configuration(inputs=inputs, outputs=outputs)
+
+    PORTS = await node_ports_v2.ports(
+        user_id=user_id, project_id=project_id, node_uuid=node_uuid
+    )
+    await check_config_valid(PORTS, config_dict)
+
+    await PORTS.set_multiple(
+        {port.key: k for k, port in enumerate((await PORTS.outputs).values())}
+    )
+    await PORTS.set_multiple(
+        {
+            port.key: k
+            for k, port in enumerate((await PORTS.inputs).values(), start=1000)
+        }
+    )
+
+    ports_outputs = await PORTS.outputs
+    ports_inputs = await PORTS.inputs
+    for k, asd in enumerate(outputs):
+        item_key, _, _ = asd
+        assert ports_outputs[item_key].value == k
+        assert await ports_outputs[item_key].get() == k
+
+    for k, asd in enumerate(inputs, start=1000):
+        item_key, _, _ = asd
+        assert ports_inputs[item_key].value == k
+        assert await ports_inputs[item_key].get() == k
+
+    # test missing key raises error
+    with pytest.raises(UnboundPortError):
+        await PORTS.set_multiple({"missing_key_in_both": 123132})

packages/simcore-sdk/tests/unit/test_node_ports_v2_nodeports_v2.py

@@ -125,6 +125,14 @@ async def mock_node_port_creator_cb(*args, **kwargs):
         assert await node_ports.get(port.key) == port.value
         await node_ports.set(port.key, port.value)
 
+    # test batch add
+    await node_ports.set_multiple(
+        {
+            port.key: port.value
+            for port in list(original_inputs.values()) + list(original_outputs.values())
+        }
+    )
+
 
 @pytest.fixture(scope="session")
 def e_tag() -> str: