Multithreaded Client (#2321)

CHANGELOG.md

@@ -17,6 +17,7 @@ The minor version will be incremented upon a breaking change and the patch versi
 - cli: Add `idl close` command to close a program's IDL account ([#2329](https://github.com/coral-xyz/anchor/pull/2329)).
 - cli: `idl init` now supports very large IDL files ([#2329](https://github.com/coral-xyz/anchor/pull/2329)).
 - spl: Add `transfer_checked` function ([#2353](https://github.com/coral-xyz/anchor/pull/2353)).
+- client: Add support for multithreading to the rust client: use flag `--multithreaded` ([#2321](https://github.com/coral-xyz/anchor/pull/2321)).
 
 ### Fixes
 

client/example/run-test.sh

@@ -41,9 +41,39 @@ main() {
     sleep 5
 
     #
-    # Run Test.
+    # Run single threaded test.
     #
-    cargo run -- --composite-pid $composite_pid --basic-2-pid $basic_2_pid --basic-4-pid $basic_4_pid --events-pid $events_pid --optional-pid $optional_pid
+    cargo run -- \
+        --composite-pid $composite_pid \
+        --basic-2-pid $basic_2_pid \
+        --basic-4-pid $basic_4_pid \
+        --events-pid $events_pid \
+        --optional-pid $optional_pid
+
+    #
+    # Restart validator for multithreaded test
+    #
+    cleanup
+    solana-test-validator -r \
+				--bpf-program $composite_pid ../../tests/composite/target/deploy/composite.so \
+				--bpf-program $basic_2_pid ../../examples/tutorial/basic-2/target/deploy/basic_2.so \
+				--bpf-program $basic_4_pid ../../examples/tutorial/basic-4/target/deploy/basic_4.so \
+				--bpf-program $events_pid ../../tests/events/target/deploy/events.so \
+				--bpf-program $optional_pid ../../tests/optional/target/deploy/optional.so \
+				> test-validator.log &
+    sleep 5
+
+    #
+    # Run multi threaded test.
+    #
+    cargo run -- \
+        --composite-pid $composite_pid \
+        --basic-2-pid $basic_2_pid \
+        --basic-4-pid $basic_4_pid \
+        --events-pid $events_pid \
+        --optional-pid $optional_pid \
+        --multithreaded
+
 }
 
 cleanup() {

client/example/src/main.rs

@@ -24,7 +24,9 @@ use composite::accounts::{Bar, CompositeUpdate, Foo, Initialize};
 use composite::instruction as composite_instruction;
 use composite::{DummyA, DummyB};
 use optional::account::{DataAccount, DataPda};
+use std::ops::Deref;
 use std::rc::Rc;
+use std::sync::Arc;
 use std::time::Duration;
 
 #[derive(Parser, Debug)]
@@ -39,12 +41,15 @@ pub struct Opts {
     events_pid: Pubkey,
     #[clap(long)]
     optional_pid: Pubkey,
+    #[clap(long, default_value = "false")]
+    multithreaded: bool,
 }
 
+type TestFn<C> = &'static (dyn Fn(&Client<C>, Pubkey) -> Result<()> + Send + Sync);
+
 // This example assumes a local validator is running with the programs
 // deployed at the addresses given by the CLI args.
 fn main() -> Result<()> {
-    println!("Starting test...");
     let opts = Opts::parse();
 
     // Wallet and cluster params.
@@ -55,15 +60,48 @@ fn main() -> Result<()> {
         "ws://127.0.0.1:8900".to_string(),
     );
 
-    // Client.
-    let client = Client::new_with_options(url, Rc::new(payer), CommitmentConfig::processed());
-
-    // Run tests.
-    composite(&client, opts.composite_pid)?;
-    basic_2(&client, opts.basic_2_pid)?;
-    basic_4(&client, opts.basic_4_pid)?;
-    events(&client, opts.events_pid)?;
-    optional(&client, opts.optional_pid)?;
+    if !opts.multithreaded {
+        // Client.
+        let payer = Rc::new(payer);
+        let client =
+            Client::new_with_options(url.clone(), payer.clone(), CommitmentConfig::processed());
+
+        // Run tests on single thread with a single client using an Rc
+        println!("\nStarting single thread test...");
+        composite(&client, opts.composite_pid)?;
+        basic_2(&client, opts.basic_2_pid)?;
+        basic_4(&client, opts.basic_4_pid)?;
+
+        // Can also use references, since they deref to a signer
+        let payer: &Keypair = &payer;
+        let client = Client::new_with_options(url, payer, CommitmentConfig::processed());
+        events(&client, opts.events_pid)?;
+        optional(&client, opts.optional_pid)?;
+    } else {
+        // Client.
+        let payer = Arc::new(payer);
+        let client = Client::new_with_options(url, payer, CommitmentConfig::processed());
+        let client = Arc::new(client);
+
+        // Run tests multithreaded while sharing a client
+        println!("\nStarting multithread test...");
+        let client = Arc::new(client);
+        let tests: Vec<(TestFn<Arc<Keypair>>, Pubkey)> = vec![
+            (&composite, opts.composite_pid),
+            (&basic_2, opts.basic_2_pid),
+            (&basic_4, opts.basic_4_pid),
+            (&events, opts.events_pid),
+            (&optional, opts.optional_pid),
+        ];
+        let mut handles = vec![];
+        for (test, arg) in tests {
+            let local_client = Arc::clone(&client);
+            handles.push(std::thread::spawn(move || test(&local_client, arg)));
+        }
+        for handle in handles {
+            assert!(handle.join().unwrap().is_ok());
+        }
+    }
 
     // Success.
     Ok(())
@@ -72,7 +110,10 @@ fn main() -> Result<()> {
 // Runs a client for examples/tutorial/composite.
 //
 // Make sure to run a localnet with the program deploy to run this example.
-fn composite(client: &Client, pid: Pubkey) -> Result<()> {
+fn composite<C: Deref<Target = impl Signer> + Clone>(
+    client: &Client<C>,
+    pid: Pubkey,
+) -> Result<()> {
     // Program client.
     let program = client.program(pid);
 
@@ -143,7 +184,7 @@ fn composite(client: &Client, pid: Pubkey) -> Result<()> {
 // Runs a client for examples/tutorial/basic-2.
 //
 // Make sure to run a localnet with the program deploy to run this example.
-fn basic_2(client: &Client, pid: Pubkey) -> Result<()> {
+fn basic_2<C: Deref<Target = impl Signer> + Clone>(client: &Client<C>, pid: Pubkey) -> Result<()> {
     let program = client.program(pid);
 
     // `Create` parameters.
@@ -172,7 +213,7 @@ fn basic_2(client: &Client, pid: Pubkey) -> Result<()> {
     Ok(())
 }
 
-fn events(client: &Client, pid: Pubkey) -> Result<()> {
+fn events<C: Deref<Target = impl Signer> + Clone>(client: &Client<C>, pid: Pubkey) -> Result<()> {
     let program = client.program(pid);
 
     let (sender, receiver) = std::sync::mpsc::channel();
@@ -204,7 +245,10 @@ fn events(client: &Client, pid: Pubkey) -> Result<()> {
     Ok(())
 }
 
-pub fn basic_4(client: &Client, pid: Pubkey) -> Result<()> {
+pub fn basic_4<C: Deref<Target = impl Signer> + Clone>(
+    client: &Client<C>,
+    pid: Pubkey,
+) -> Result<()> {
     let program = client.program(pid);
     let authority = program.payer();
     let (counter, _) = Pubkey::find_program_address(&[b"counter"], &pid);
@@ -240,7 +284,7 @@ pub fn basic_4(client: &Client, pid: Pubkey) -> Result<()> {
 // Runs a client for tests/optional.
 //
 // Make sure to run a localnet with the program deploy to run this example.
-fn optional(client: &Client, pid: Pubkey) -> Result<()> {
+fn optional<C: Deref<Target = impl Signer> + Clone>(client: &Client<C>, pid: Pubkey) -> Result<()> {
     // Program client.
     let program = client.program(pid);
 

client/src/lib.rs

@@ -23,7 +23,7 @@ use solana_sdk::signature::{Signature, Signer};
 use solana_sdk::transaction::Transaction;
 use std::convert::Into;
 use std::iter::Map;
-use std::rc::Rc;
+use std::ops::Deref;
 use std::vec::IntoIter;
 use thiserror::Error;
 
@@ -43,12 +43,12 @@ pub type EventHandle = PubsubClientSubscription<RpcResponse<RpcLogsResponse>>;
 /// Client defines the base configuration for building RPC clients to
 /// communicate with Anchor programs running on a Solana cluster. It's
 /// primary use is to build a `Program` client via the `program` method.
-pub struct Client {
-    cfg: Config,
+pub struct Client<C> {
+    cfg: Config<C>,
 }
 
-impl Client {
-    pub fn new(cluster: Cluster, payer: Rc<dyn Signer>) -> Self {
+impl<C: Clone + Deref<Target = impl Signer>> Client<C> {
+    pub fn new(cluster: Cluster, payer: C) -> Self {
         Self {
             cfg: Config {
                 cluster,
@@ -58,11 +58,7 @@ impl Client {
         }
     }
 
-    pub fn new_with_options(
-        cluster: Cluster,
-        payer: Rc<dyn Signer>,
-        options: CommitmentConfig,
-    ) -> Self {
+    pub fn new_with_options(cluster: Cluster, payer: C, options: CommitmentConfig) -> Self {
         Self {
             cfg: Config {
                 cluster,
@@ -72,7 +68,7 @@ impl Client {
         }
     }
 
-    pub fn program(&self, program_id: Pubkey) -> Program {
+    pub fn program(&self, program_id: Pubkey) -> Program<C> {
         Program {
             program_id,
             cfg: Config {
@@ -86,26 +82,26 @@ impl Client {
 
 // Internal configuration for a client.
 #[derive(Debug)]
-struct Config {
+struct Config<C> {
     cluster: Cluster,
-    payer: Rc<dyn Signer>,
+    payer: C,
     options: Option<CommitmentConfig>,
 }
 
 /// Program is the primary client handle to be used to build and send requests.
 #[derive(Debug)]
-pub struct Program {
+pub struct Program<C> {
     program_id: Pubkey,
-    cfg: Config,
+    cfg: Config<C>,
 }
 
-impl Program {
+impl<C: Deref<Target = impl Signer> + Clone> Program<C> {
     pub fn payer(&self) -> Pubkey {
         self.cfg.payer.pubkey()
     }
 
     /// Returns a request builder.
-    pub fn request(&self) -> RequestBuilder {
+    pub fn request(&self) -> RequestBuilder<C> {
         RequestBuilder::from(
             self.program_id,
             self.cfg.cluster.url(),
@@ -373,23 +369,23 @@ pub enum ClientError {
 
 /// `RequestBuilder` provides a builder interface to create and send
 /// transactions to a cluster.
-pub struct RequestBuilder<'a> {
+pub struct RequestBuilder<'a, C> {
     cluster: String,
     program_id: Pubkey,
     accounts: Vec<AccountMeta>,
     options: CommitmentConfig,
     instructions: Vec<Instruction>,
-    payer: Rc<dyn Signer>,
+    payer: C,
     // Serialized instruction data for the target RPC.
     instruction_data: Option<Vec<u8>>,
     signers: Vec<&'a dyn Signer>,
 }
 
-impl<'a> RequestBuilder<'a> {
+impl<'a, C: Deref<Target = impl Signer> + Clone> RequestBuilder<'a, C> {
     pub fn from(
         program_id: Pubkey,
         cluster: &str,
-        payer: Rc<dyn Signer>,
+        payer: C,
         options: Option<CommitmentConfig>,
     ) -> Self {
         Self {
@@ -405,7 +401,7 @@ impl<'a> RequestBuilder<'a> {
     }
 
     #[must_use]
-    pub fn payer(mut self, payer: Rc<dyn Signer>) -> Self {
+    pub fn payer(mut self, payer: C) -> Self {
         self.payer = payer;
         self
     }