From 2fbd9d4c4c5cce4e73b45a6f2d1477dba44d698c Mon Sep 17 00:00:00 2001
From: Wez Furlong <wez@wezfurlong.org>
Date: Wed, 27 Mar 2024 16:22:25 -0700
Subject: [PATCH] add provisional knob to control how aggressive we are at open
 connections

I'm not sure if I want to keep this, but it is currently useful for
testing purposes.

The previous logic was to skip opening an SMTP connection in a
newly constructed Dispatcher if no messages were immediately
available in the ready queue.

This logic makes a lot of sense in a multi-node deployment
with shared throttles, so that other nodes have an opportunity
to open connections to drain their own queues, but when testing
a single node it leaves some throughput on the table.

This boolean option allows selecting whether we aggressively
continue to open an SMTP connection even if we may not have
a message ready to send yet, or, if left at the default `false`
value, continue with the original more conservative logic.
---
 crates/kumo-api-types/src/egress_path.rs |  5 +++++
 crates/kumod/src/ready_queue.rs          | 20 ++++++++++++++------
 2 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/crates/kumo-api-types/src/egress_path.rs b/crates/kumo-api-types/src/egress_path.rs
index bd2afe15..fc0c5f01 100644
--- a/crates/kumo-api-types/src/egress_path.rs
+++ b/crates/kumo-api-types/src/egress_path.rs
@@ -98,6 +98,10 @@ pub struct EgressPathConfig {
 
     #[serde(default)]
     pub suspended: bool,
+
+    // TODO: decide if we want to keep this and then document
+    #[serde(default)]
+    pub aggressive_connection_opening: bool,
 }
 
 #[cfg(feature = "lua")]
@@ -125,6 +129,7 @@ impl Default for EgressPathConfig {
             smtp_auth_plain_username: None,
             smtp_auth_plain_password: None,
             suspended: false,
+            aggressive_connection_opening: false,
         }
     }
 }
diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs
index c926abe8..8a6b0ae8 100644
--- a/crates/kumod/src/ready_queue.rs
+++ b/crates/kumod/src/ready_queue.rs
@@ -628,12 +628,20 @@ impl Dispatcher {
             }
         };
 
-        dispatcher.obtain_message().await;
-        if dispatcher.msg.is_none() {
-            // We raced with another dispatcher and there is no
-            // more work to be done; no need to open a new connection.
-            dispatcher.lease.release().await;
-            return Ok(());
+        // We get better throughput by being more aggressive with establishing
+        // connections.
+        if !dispatcher
+            .path_config
+            .borrow()
+            .aggressive_connection_opening
+        {
+            dispatcher.obtain_message().await;
+            if dispatcher.msg.is_none() {
+                // We raced with another dispatcher and there is no
+                // more work to be done; no need to open a new connection.
+                dispatcher.lease.release().await;
+                return Ok(());
+            }
         }
 
         let mut connection_failures = vec![];