refactor!: Make Sender clone (#25)

This makes the `Sender` be `Clone` and take `&self` in the methods. For the remote sender, this is achieved by putting the actual send stream and buffer into a `tokio::sync::Mutex`. We also ensure that if sending failed on one sender, subsequent sends on cloned senders will fail as well. And if a `send` future is dropped before completion, further sends will fail as well, as we cannot guarantee integrity of the send stream anymore if a send future was dropped.

## Breaking changes

The methods from `Sender` all take `&self` instead of `&mut self`.

## Notes and open questions

My quick benchmark shows some weird numbers:
```
Local bench
RPC seq 128_996 rps
RPC par 1_093_166 rps
bidi seq 4_281_373 rps
Remote bench
RPC seq 11_393 rps
RPC par 60_073 rps
bidi seq 1_480_826 rps
Reference bench
Reference seq 139_889 rps
Reference par 2_784_308 rps
```

Compared to #24 the `bidi seq` for remote got *much* faster. I can't really explain that? But I ran quite a few times, without much change on either `Frando/improve-bench` or `Frando/clone-sender`.

Fixes #23

Author: Frando

Cargo.lock

@@ -57,6 +57,7 @@ tokio = { workspace = true, features = ["full"] }
 thousands = "0.2.0"
 # macro tests
 trybuild = "1.0.104"
+testresult = "0.4.1"
 
 [features]
 # enable the remote transport

Cargo.toml

@@ -123,7 +123,7 @@ impl ComputeActor {
                     tx, inner, span, ..
                 } = fib;
                 let _entered = span.enter();
-                let mut sender = tx;
+                let sender = tx;
                 let mut a = 0u64;
                 let mut b = 1u64;
                 while a <= inner.max {
@@ -144,7 +144,7 @@ impl ComputeActor {
                 } = mult;
                 let _entered = span.enter();
                 let mut receiver = rx;
-                let mut sender = tx;
+                let sender = tx;
                 let multiplier = inner.initial;
                 while let Some(num) = receiver.recv().await? {
                     sender.send(multiplier * num).await?;
@@ -260,7 +260,7 @@ async fn local() -> anyhow::Result<()> {
     println!("Local: 5^2 = {}", rx.await?);
 
     // Test Sum
-    let (mut tx, rx) = api.sum().await?;
+    let (tx, rx) = api.sum().await?;
     tx.send(1).await?;
     tx.send(2).await?;
     tx.send(3).await?;
@@ -276,7 +276,7 @@ async fn local() -> anyhow::Result<()> {
     println!();
 
     // Test Multiply
-    let (mut in_tx, mut out_rx) = api.multiply(3).await?;
+    let (in_tx, mut out_rx) = api.multiply(3).await?;
     in_tx.send(2).await?;
     in_tx.send(4).await?;
     in_tx.send(6).await?;
@@ -311,7 +311,7 @@ async fn remote() -> anyhow::Result<()> {
     println!("Remote: 4^2 = {}", rx.await?);
 
     // Test Sum
-    let (mut tx, rx) = api.sum().await?;
+    let (tx, rx) = api.sum().await?;
     tx.send(4).await?;
     tx.send(5).await?;
     tx.send(6).await?;
@@ -327,7 +327,7 @@ async fn remote() -> anyhow::Result<()> {
     println!();
 
     // Test Multiply
-    let (mut in_tx, mut out_rx) = api.multiply(5).await?;
+    let (in_tx, mut out_rx) = api.multiply(5).await?;
     in_tx.send(1).await?;
     in_tx.send(2).await?;
     in_tx.send(3).await?;
@@ -380,7 +380,7 @@ async fn bench(api: ComputeApi, n: u64) -> anyhow::Result<()> {
     // Sequential streaming (using Multiply instead of MultiplyUpdate)
     {
         let t0 = std::time::Instant::now();
-        let (mut send, mut recv) = api.multiply(2).await?;
+        let (send, mut recv) = api.multiply(2).await?;
         let handle = tokio::task::spawn(async move {
             for i in 0..n {
                 send.send(i).await?;

examples/compute.rs

@@ -111,7 +111,7 @@ impl StorageActor {
             }
             StorageMessage::List(list) => {
                 info!("list {:?}", list);
-                let WithChannels { mut tx, .. } = list;
+                let WithChannels { tx, .. } = list;
                 for (key, value) in &self.state {
                     if tx.send(format!("{key}={value}")).await.is_err() {
                         break;
@@ -172,7 +172,7 @@ async fn client_demo(api: StorageApi) -> Result<()> {
     let value = api.get("hello".to_string()).await?;
     println!("get: hello = {:?}", value);
 
-    let (mut tx, rx) = api.set_many().await?;
+    let (tx, rx) = api.set_many().await?;
     for i in 0..3 {
         tx.send((format!("key{i}"), format!("value{i}"))).await?;
     }

examples/derive.rs

@@ -104,7 +104,7 @@ impl StorageActor {
             }
             StorageMessage::List(list) => {
                 info!("list {:?}", list);
-                let WithChannels { mut tx, .. } = list;
+                let WithChannels { tx, .. } = list;
                 for (key, value) in &self.state {
                     if tx.send(format!("{key}={value}")).await.is_err() {
                         break;

examples/storage.rs

@@ -218,7 +218,7 @@ mod storage {
                 }
                 StorageMessage::List(list) => {
                     info!("list {:?}", list);
-                    let WithChannels { mut tx, .. } = list;
+                    let WithChannels { tx, .. } = list;
                     let values = {
                         let state = self.state.lock().unwrap();
                         // TODO: use async lock to not clone here.

irpc-iroh/examples/auth.rs

@@ -141,7 +141,7 @@ mod storage {
                 }
                 StorageMessage::List(list) => {
                     info!("list {:?}", list);
-                    let WithChannels { mut tx, .. } = list;
+                    let WithChannels { tx, .. } = list;
                     for (key, value) in &self.state {
                         if tx.send(format!("{key}={value}")).await.is_err() {
                             break;