RUST-585 Refactor Topology implementation to actor model #628

patrickfreed · 2022-04-14T20:01:25Z

This PR refactors the Topology implementation to follow an actor model (similar to the connection pool) rather than a lock based one, hopefully making the code easier to maintain and reason about.

monitor, basic updates

possible deadlock?

This partially reverts commit d7ac989.

This reverts commit 7f1e962.

patrickfreed · 2022-04-15T21:06:34Z

Cargo.toml

@@ -119,7 +119,7 @@ version = "0.11.5"
 optional = true

 [dependencies.tokio]
-version = "1.4.0"
+version = "1.17.0"


This was done to get some newer functionality in the sync::watch channel. It's not strictly necessary, but I figured there wasn't any risk in this, as I doubt any users have strict tokio dependency requirements.

patrickfreed · 2022-04-15T21:07:14Z

src/client/mod.rs

@@ -103,12 +103,6 @@ struct ClientInner {
    session_pool: ServerSessionPool,
 }

-impl Drop for ClientInner {


This will happen automatically now.

patrickfreed · 2022-04-15T21:08:42Z

src/client/executor.rs

@@ -446,8 +434,10 @@ impl Client {
            return Err(first_error);
        }

+        let txn_number = prior_txn_number.or_else(|| get_txn_number(session, retryability));


I'm not sure why these changes surfaced this problem, but basically if a commitTransaction retry failed during connection acquisition with a retryable error, the txnNumber would never get set and the user would see an error. This now checks to see if there was a txnNumber from before and if not, gets a new one from the session.

So this was actually unrelated, but it seems to be fixed now. Filed RUST-1274 to track this separately.

should we backport this fix to 2.2?

Yep, I added a comment to RUST-1274 as a reminder. Once this is merged I'll do the backport.

patrickfreed · 2022-04-15T21:09:57Z

src/cmap/worker.rs

@@ -696,46 +701,6 @@ impl From<PoolManagementRequest> for PoolTask {
    }
 }

-/// Constructs a new channel for for monitoring whether this pool still has references
-/// to it.
-fn handle_channel() -> (PoolWorkerHandle, HandleListener) {


this was abstracted to a common type in the runtime module for use with the topology worker

patrickfreed · 2022-04-15T21:13:04Z

src/hello.rs

@@ -133,6 +135,7 @@ where
 pub(crate) struct HelloReply {
    pub server_address: ServerAddress,
    pub command_response: HelloCommandResponse,
+    pub raw_command_response: RawDocumentBuf,


this is included as part of the HelloReply so that the monitors can emit SDAM events instead of doing it in the handshaker or the common functions in the hello module.

patrickfreed · 2022-04-15T21:35:39Z

src/sdam/topology.rs

+
+/// Handle used to request that monitors perform immediate checks of the topology.
+#[derive(Clone, Debug)]
+struct TopologyCheckRequester {


This type wasn't mentioned in the design, but it's a channel used to request topology checks. It was necessary to avoid having to go through the topology worker to request checks (the messages go straight to the monitors)

patrickfreed · 2022-04-15T21:35:53Z

src/sdam/topology.rs

+///
+/// This is used to determine the error handling semantics for certain error types.
+#[derive(Debug, Clone)]
+pub(crate) enum HandshakePhase {


this type was copy/pasted

patrickfreed · 2022-04-18T16:03:56Z

src/sdam/topology.rs

+/// If the topology has been closed, events emitted via this handle will not be sent to
+/// handlers.
+#[derive(Clone)]
+pub(crate) struct SdamEventEmitter {


this struct was required, instead of just emitting events directly from handlers, to ensure no events were emitted after the last TopologyClosedEvent. I originally tried using TopologyWatcher to detect if the topology was still alive before emitting events (similar to the existing implementation), but that proved to be too racy.

this also has the added benefit of preventing users / us from blocking the TopologyWorker via our SdamEventHandler implementations.

patrickfreed · 2022-04-18T16:12:14Z

src/sdam/topology.rs

@@ -0,0 +1,949 @@
+use std::{


To summarize this file, we have the following types:

Topology: a complete handle to the topology. Allows the executor to select servers, update the topology based on appliction errors, request monitor checks, and keep the worker task running. Does so via the other handle types.

TopologyWorker: the aforementioned worker / actor task that process updates to the topology and publishes new states

TopologyUpdater: used to send new server descriptions, application errors, and monitor errors to the topology. Accessed by the executor via Topology and directly by server monitors, SRV polling monitors, and connection pools

TopologyWatcher: used to observe the latest published state of the topology. Used by the executor via Topology and by server monitors directly.

TopologyCheckRequester: used by the executor to request immediate monitor checks when server selection fails

SdamEventEmitter: used by the TopologyWorker and server monitors to publish SDAM events

TopologyState: "plain old data" containing a topology description and a hashmap of the servers. These are published whenever the topology is updated

To summarize the ownership:

Topology

TopologyUpdater

TopologyWatcher

TopologyCheckRequester

WorkerHandle

TopologyWorker

SdamEventEmitter

Server monitors

TopologyWatcher

TopologyUpdater

SdamEventEmitter

SRV polling monitor

TopologyUpdater

Connection pool

TopologyUpdater

patrickfreed · 2022-04-19T17:46:19Z

src/sdam/message_manager.rs

@@ -1,64 +0,0 @@
-use std::time::Duration;


this is no longer needed now that we have the new handles

abr-egn · 2022-04-20T14:28:28Z

src/client/mod.rs


+            let change_occurred = start_time.elapsed() < timeout
+                && watcher
+                    .wait_for_update(timeout - start_time.elapsed())


Is there a potential race condition here? i.e. if the update happens after the request_update call but before the wait_for_update call, will this be waiting until some unrelated future update?

each iteration of the loop marks the topology as "seen" in the clone_latest call, so any update after that point (potentially before the request_update call actually) will be accounted for in wait_for_update.

abr-egn · 2022-04-20T14:38:44Z

src/event/sdam/mod.rs

@@ -208,3 +221,23 @@ pub trait SdamEventHandler: Send + Sync {
    /// a server heartbeat fails.
    fn handle_server_heartbeat_failed_event(&self, _event: ServerHeartbeatFailedEvent) {}
 }
+
+pub(crate) fn handle_sdam_event(handler: &dyn SdamEventHandler, event: SdamEvent) {


Any particular reason this accepts a &dyn SdamEventHandler instead of a generic bound?

We accept handlers in client options as Arc<dyn SdamEventHandler> so we can't get their concrete type from there.

abr-egn · 2022-04-20T14:58:54Z

src/sdam/topology.rs

+    pub(crate) fn watch(&self) -> TopologyWatcher {
+        let mut watcher = self.watcher.clone();
+        // mark the latest topology as seen
+        watcher.receiver.borrow_and_update();


I don't follow why this is needed.

This is to ensure that any calls to wait_for_update on the returned TopologyWatcher will block until a new state is published and not return immediately.

abr-egn · 2022-04-20T15:10:16Z

src/sdam/topology.rs

+
+    /// Clone the latest state, marking it as seen.
+    pub(crate) fn clone_latest(&mut self) -> TopologyState {
+        self.receiver.borrow_and_update().clone()


I don't understand the usage of the receiver's "seen" flag. It's set in clone_latest but not borrow_latest or server_description?

The "seen" flag determines whether the currently observed TopologtState would be considered "new" or not for the purposes of wait_for_update. If we've called clone_latest and then immediately call wait_for_update, it'll block. If we haven't, it'll return immediately.

Not setting it in borrow_latest and server_description was more of an ergonomic choice, since it would require those methods to be &mut self.

To make this a bit clearer, I renamed the methods to observe_latest (formerly clone_latest) and peek_latest (formerly borrow_latest). Let me know if you think these are more helpful / if there are better ones.

Much clearer, thank you :)

abr-egn

LGTM! Definitely much easier to follow, and thank you for explaining the bits I didn't get.

abr-egn · 2022-04-21T14:51:51Z

src/sdam/topology.rs

+
+    /// Clone the latest state, marking it as seen.
+    pub(crate) fn clone_latest(&mut self) -> TopologyState {
+        self.receiver.borrow_and_update().clone()


Much clearer, thank you :)

kmahar

few minor questions but besides those LGTM!

kmahar · 2022-04-22T19:43:44Z

src/client/executor.rs

@@ -446,8 +434,10 @@ impl Client {
            return Err(first_error);
        }

+        let txn_number = prior_txn_number.or_else(|| get_txn_number(session, retryability));


should we backport this fix to 2.2?

src/sdam/test.rs

kmahar · 2022-04-27T21:07:30Z

src/sdam/description/topology/mod.rs

        command: &mut Command<T>,
        criteria: Option<&SelectionCriteria>,
    ) {
+        let server_type = self
+            .get_server_description(address)


when does this return None? would that be an (I think unexpected case) where we somehow selected a server but the TopologyDescription doesn't contain it?

Yeah, this would be really rare, but basically if a server were to be removed from the topology after we had selected it + checked out a connection from it (e.g. if a monitor check comes back that removes it from the topology). This can happen in the normal course of operations, so I think we need to handle it here. We could decide to return an error instead of defaulting to Unknown to try to prevent the operation from executing, but it's not entirely clear if that's what we want or not.

ah that makes sense. I think considering it unknown makes sense then. thanks!

kmahar · 2022-04-27T21:16:36Z

src/sdam/description/topology/test/sdam.rs

@@ -580,19 +586,23 @@ async fn topology_closed_event_last() {
    drop(client);

    subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .wait_for_event(Duration::from_millis(5000), |event| {


was this failing with the lower timeout / any hypotheses as to why?

Oh I think I changed this when I was debugging an earlier version of the test, changed back to 500.

isabelatkinson

looks good, just one question

isabelatkinson · 2022-04-28T21:00:53Z

src/runtime/acknowledged_message.rs

-    /// Borrow the message.
-    pub(crate) fn message(&self) -> &M {
+    /// Send acknowledgement to the receiver.
+    #[allow(dead_code)]


any reason for keeping this around even though it's unused?

Oh good catch, we actually do use this method in the connection pool, so this dead_code ignore can just be removed.

patrickfreed

To try to preserve the review comment / commit history, I did a regular merge commit to fix the merge conflicts. Will still squash it all down to a single commit at the end though.

patrickfreed · 2022-05-02T16:35:59Z

src/runtime/acknowledged_message.rs

-    /// Borrow the message.
-    pub(crate) fn message(&self) -> &M {
+    /// Send acknowledgement to the receiver.
+    #[allow(dead_code)]


Oh good catch, we actually do use this method in the connection pool, so this dead_code ignore can just be removed.

patrickfreed · 2022-05-02T16:37:32Z

src/sdam/description/topology/test/sdam.rs

@@ -580,19 +586,23 @@ async fn topology_closed_event_last() {
    drop(client);

    subscriber
-        .wait_for_event(Duration::from_millis(500), |event| {
+        .wait_for_event(Duration::from_millis(5000), |event| {


Oh I think I changed this when I was debugging an earlier version of the test, changed back to 500.

patrickfreed · 2022-05-02T16:40:35Z

src/sdam/description/topology/mod.rs

        command: &mut Command<T>,
        criteria: Option<&SelectionCriteria>,
    ) {
+        let server_type = self
+            .get_server_description(address)


Yeah, this would be really rare, but basically if a server were to be removed from the topology after we had selected it + checked out a connection from it (e.g. if a monitor check comes back that removes it from the topology). This can happen in the normal course of operations, so I think we need to handle it here. We could decide to return an error instead of defaulting to Unknown to try to prevent the operation from executing, but it's not entirely clear if that's what we want or not.

src/sdam/test.rs

patrickfreed · 2022-05-02T18:58:44Z

src/client/executor.rs

@@ -446,8 +434,10 @@ impl Client {
            return Err(first_error);
        }

+        let txn_number = prior_txn_number.or_else(|| get_txn_number(session, retryability));


Yep, I added a comment to RUST-1274 as a reminder. Once this is merged I'll do the backport.

patrickfreed added 29 commits April 4, 2022 16:47

wip

a02a7b2

wip

4e40828

monitor, basic updates

new topology hooked up to server selection

cf68eed

use new topology everywhere

738e39b

update tests

caf13a3

crud passing

f07192b

wip sdam tests

f779a40

wait for acknowledgment in updater methods, sdam tests more wip

feaf3af

wip fix dropping

8a59dd6

sdam tests passing

d87610c

all standalone tests passing

ace147b

remove debug print, fix srv tests, disable cs doc test

9461a72

possible deadlock?

fix clippy

4e96ecc

add comments, more cleanup

5fd4a1b

delete old sdam/state module

083251c

cleanup

b217db4

only emit events if topology is alive, add better debug

e69c939

drop the broadcaster before emitting closed event

891964a

serialize sdam event emission, get txn number if retrying connection

d7ac989

fix clippy

e25015d

Revert serialize sdam event emission

7f1e962

This partially reverts commit d7ac989.

Revert "Revert serialize sdam event emission"

752418a

This reverts commit 7f1e962.

unify publisher terminology

f2c57b3

acknowledge event emission

d80a86d

dont acknowledge event emission when emitting events from topology

ed4196d

cleanup

9c18b9d

fix racy test

cd418e9

fix sdam auth_error test on replsets

bef78f7

remove unused message manager

873f697

patrickfreed commented Apr 19, 2022

View reviewed changes

patrickfreed marked this pull request as ready for review April 19, 2022 17:49

patrickfreed requested review from kmahar, isabelatkinson and abr-egn April 19, 2022 17:49

abr-egn reviewed Apr 20, 2022

View reviewed changes

make watcher api clearer

3465eea

patrickfreed requested a review from abr-egn April 20, 2022 21:09

abr-egn approved these changes Apr 21, 2022

View reviewed changes

fmt

8c2775e

kmahar reviewed Apr 27, 2022

View reviewed changes

isabelatkinson reviewed Apr 28, 2022

View reviewed changes

kmahar mentioned this pull request Apr 29, 2022

RUST-332 Return an error for replica set name mismatch #648

Merged

patrickfreed added 4 commits May 2, 2022 12:35

remove unneeded dead_code ignore

fda2683

revert event timeout

0d2ff38

verify we saw both types of events

10649dc

Merge branch 'main' into RUST-585/refactor-sdam

f05a507

patrickfreed commented May 2, 2022

View reviewed changes

patrickfreed requested review from isabelatkinson and kmahar May 2, 2022 19:31

kmahar approved these changes May 2, 2022

View reviewed changes

isabelatkinson approved these changes May 2, 2022

View reviewed changes

patrickfreed merged commit 0531022 into mongodb:main May 2, 2022

patrickfreed mentioned this pull request May 3, 2022

RUST-1274 Fix commitTransaction on check out retries #651

Merged

RUST-585 Refactor Topology implementation to actor model #628

RUST-585 Refactor Topology implementation to actor model #628

Uh oh!

Conversation

patrickfreed commented Apr 14, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

abr-egn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kmahar left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

isabelatkinson left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

patrickfreed left a comment

Choose a reason for hiding this comment