Merge ~cgrabowski/maas:fix_connections_dict_update_while_updating_connection_count into maas:master

[Christian Grabowski]

copy conns dict in rpc connection update to allow concurrent IPC calls

[MAAS Lander (maas-lander)]

UNIT TESTS
-b fix_connections_dict_update_while_updating_connection_count lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/580/consoleText
COMMIT: 72a63c7381dbbf757194e11e8f3a497497cd3fef

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b fix_connections_dict_update_while_updating_connection_count lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 68c26b36716e4010c0fc59ef5d12c971474b790f

review: Approve

[Björn Tillenius (bjornt)]

Your patch probably fixes the traceback. But I'd like to understand a bit what is actually modifying the dictionary?

I wonder if we have another problem here, and we're only fixing the symptoms. I did run the test you added, but that doesn't seem to trigger the problem.

review: Needs Information

[Christian Grabowski (cgrabowski)]

> Your patch probably fixes the traceback. But I'd like to understand a bit what
> is actually modifying the dictionary?
>
> I wonder if we have another problem here, and we're only fixing the symptoms.
> I did run the test you added, but that doesn't seem to trigger the problem.

So the test does pass without the fix _most_ of the time, but so what's happening is `registerWorkerRPC` is being called at the same time as `update`, the way I was able to reproduce this is having a region controller running for awhile, and then restarting the rack controller in a loop, so it has some (or at least previous) connectivity, and new connections are being added at the same time. This becomes more apparent when running with multiple threads and rpc workers, which is a bit difficult to express in the test. So by copying the dict, the original / main connections dict can receive the new connections, while updating the DB based on the copied dict.

[Adam Collard (adam-collard)]

LGTM

review: Approve

[Björn Tillenius (bjornt)]

ok. I think this is ok to merge, but I'd still like to understand the issue a bit more.

I think we have a thread issue here. We modify and read conn["rpc"]["connections"] in different threads. Ideally we shouldn't do this, but assuming that list.copy() is thread-safe, your patch should fix the traceback.

But are we processing all the connections properly when this issue occurs? Can you outline the execution workflow when update() and registerWorkerRPC() are called in parallel?

[Christian Grabowski (cgrabowski)]

Well registerWorkerRPC() doesn't have this issue, it's registerWorkerRPCConnection() that does, which might be splitting hairs, but the big difference is the issue occurred when adding a connection, which is what registerWorkerRPCConnection() is doing, in parallel to updating the DB on the connection state, which is what update() was doing when the stacktrace occurs. Adding RPC workers, which is what registerWorkerRPC() is doing, which update() also iterates over the top level dict that that would affect, but I believe since the workers register on startup, the top level dict doesn't change while update() is called.

[Björn Tillenius (bjornt)]

What calls .update()?

[Christian Grabowski (cgrabowski)]

In the __init__ of this IPCMasterService, there's a LoopingCall(self.update), so it calls update() itself in a loop.

[Björn Tillenius (bjornt)]

Ok. So I do think that we still have a problem. What can happen is that registerWorkerRPCConnection() registers a new connection in the database, but then update() works on the old "connections" dict and removes the connection from the database. Then after 60 seconds update() runs again with the new "connections" dict and adds it back.

Not sure if this causes any actual problems, though.

[Christian Grabowski (cgrabowski)]

Hmm I suppose that is possible, we could use a lock instead, but in order to address that particular issue, I believe we'd have to lock for the entire registerRPCWorkerConnection() and _updateConnection(), which might be fine, but it might be a problem if many connections come in at once.

[Björn Tillenius (bjornt)]

A lock would be one solution, but with Twisted there's another approach. You can basically move the copy of the dict to outside of of _update() and pass it to deferToDatabase(). That way you get an automatic lock for the connections dict, since both the copy and the modifications run in the main thread.

[Adam Collard (adam-collard)]

See LP:1989567

diff --git a/src/maasserver/ipc.py b/src/maasserver/ipc.py
index 9fbb0ef..a199b2b 100644
--- a/src/maasserver/ipc.py
+++ b/src/maasserver/ipc.py
@@ -563,7 +563,7 @@ class IPCMasterService(service.Service):
             else:
                 # RPC is not running, no endpoints.
                 self._updateEndpoints(process, [])
-            self._updateConnections(process, conn["rpc"]["connections"])
+            self._updateConnections(process, conn["rpc"]["connections"].copy())
             previous_process_ids.discard(process.id)
 
         # Delete all the old processes that are dead.
diff --git a/src/maasserver/tests/test_ipc.py b/src/maasserver/tests/test_ipc.py
index af1edba..78355bc 100644
--- a/src/maasserver/tests/test_ipc.py
+++ b/src/maasserver/tests/test_ipc.py
@@ -13,7 +13,7 @@ import uuid
 from fixtures import EnvironmentVariableFixture
 from testtools.matchers import MatchesStructure
 from twisted.internet import reactor
-from twisted.internet.defer import inlineCallbacks, succeed
+from twisted.internet.defer import DeferredList, inlineCallbacks, succeed
 
 from maasserver import ipc, workers
 from maasserver.enum import SERVICE_STATUS
@@ -508,3 +508,57 @@ class TestIPCCommunication(MAASTransactionServerTestCase):
         self.assertEqual(rpc_connections, [])
 
         yield master.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_update_allows_new_connections_while_updating_connections_in_db(
+        self,
+    ):
+        yield deferToDatabase(load_builtin_scripts)
+        master = self.make_IPCMasterService()
+        yield master.startService()
+
+        pid = random.randint(1, 512)
+        port = random.randint(1, 512)
+        ip = factory.make_ip_address()
+        yield master.registerWorker(pid, MagicMock())
+        yield master.registerWorkerRPC(pid, port)
+
+        rack_controller = yield deferToDatabase(factory.make_RackController)
+
+        for _ in range(2):
+            master.registerWorkerRPCConnection(
+                pid,
+                random.randint(1, 512),
+                rack_controller.system_id,
+                ip,
+                random.randint(1, 512),
+            )
+
+        defers = DeferredList(
+            [
+                master.update(),
+                master.registerWorkerRPCConnection(
+                    pid,
+                    random.randint(1, 512),
+                    rack_controller.system_id,
+                    ip,
+                    random.randint(1, 512),
+                ),
+            ]
+        )
+
+        yield defers
+
+        def _get_conn_count():
+            return RegionRackRPCConnection.objects.filter(
+                rack_controller=rack_controller
+            ).count()
+
+        count = yield deferToDatabase(_get_conn_count)
+
+        self.assertEqual(
+            count, len(master.connections[pid]["rpc"]["connections"].values())
+        )
+
+        yield master.stopService()