Faster matching with threads (#1)

teddav · web-flow · commit cc6b5338280f · 2025-05-07T23:34:08.000+02:00
* server: adding threads for faster matching

* threads working
diff --git a/mpc-server/Cargo.toml b/mpc-server/Cargo.toml
@@ -37,3 +37,4 @@ dotenv = "0.15.0"
 toml = "0.8.22"
 hex = "0.4.3"
 axum-server = { version = "0.7.2", features = ["tls-rustls"] }
+rayon = "1.10.0"
diff --git a/mpc-server/src/db.rs b/mpc-server/src/db.rs
@@ -6,12 +6,14 @@ use crate::matching::DATA_DIR;
 #[derive(Debug, Clone)]
 pub struct User {
     pub id: String,
+    #[allow(dead_code)]
     pub twitter_handle: String,
     pub checked: Vec<String>,
 }
 
 #[derive(Debug, Clone)]
 pub struct Match {
+    #[allow(dead_code)]
     pub id: u32,
     pub user_id1: String,
     pub user_id2: String,
@@ -117,6 +119,22 @@ pub fn update_checked(
     Ok(())
 }
 
+pub fn update_checked_many(
+    conn: &Connection,
+    user_ids: Vec<String>,
+    new_checked: Vec<String>,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let mut stmt = conn.prepare("UPDATE users SET checked = ?1 WHERE id = ?2")?;
+
+    for user_id in user_ids {
+        let user = get_user(conn, &user_id)?;
+        let mut checked: HashSet<String> = user.checked.into_iter().collect();
+        checked.extend(new_checked.clone());
+        stmt.execute((serde_json::to_string(&checked)?, user_id))?;
+    }
+    Ok(())
+}
+
 pub fn insert_matches(
     conn: &Connection,
     matches: Vec<(String, String)>,
diff --git a/mpc-server/src/main.rs b/mpc-server/src/main.rs
@@ -5,7 +5,7 @@ use axum::{
     routing::{get, post},
 };
 use axum_server::tls_rustls::RustlsConfig;
-use co_noir::{Address, Bn254, CrsParser, NetworkParty, PartyID, Utils};
+use co_noir::{Bn254, CrsParser, Utils};
 use co_ultrahonk::prelude::{ProverCrs, ZeroKnowledge};
 use rustls::pki_types::CertificateDer;
 use serde::Deserialize;
@@ -61,22 +61,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 
     setup_db()?;
 
-    let parties = vec![
-        NetworkParty::new(
-            PartyID::ID0.into(),
-            Address::new("localhost".to_string(), 10000),
-            CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert0.der"))?).into_owned(),
-        ),
-        NetworkParty::new(
-            PartyID::ID1.into(),
-            Address::new("localhost".to_string(), 10001),
-            CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert1.der"))?).into_owned(),
-        ),
-        NetworkParty::new(
-            PartyID::ID2.into(),
-            Address::new("localhost".to_string(), 10002),
-            CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert2.der"))?).into_owned(),
-        ),
+    let parties_certs = [
+        CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert0.der"))?).into_owned(),
+        CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert1.der"))?).into_owned(),
+        CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert2.der"))?).into_owned(),
     ];
 
     let program_artifact = Utils::get_program_artifact_from_file(DATA_DIR.join("circuit.json"))?;
@@ -128,9 +116,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
             post(move |token: Token| async move {
                 match run_matches(
                     token.user_id,
-                    parties,
+                    parties_certs,
                     &program_artifact,
-                    constraint_system,
+                    constraint_system.clone(),
                     recursive,
                     has_zk,
                     prover_crs,
@@ -201,8 +189,8 @@ mod tests {
     use super::*;
     use crate::{matching::run_match, shares::split_input};
 
-    #[tokio::test]
-    async fn test_match() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    #[test]
+    fn test_match() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         let prover_toml = r#"[user1]
 age = 30
 region = 1
@@ -233,22 +221,10 @@ gender = 0"#;
             .install_default()
             .unwrap();
 
-        let parties = vec![
-            NetworkParty::new(
-                PartyID::ID0.into(),
-                Address::new("localhost".to_string(), 10000),
-                CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert0.der"))?).into_owned(),
-            ),
-            NetworkParty::new(
-                PartyID::ID1.into(),
-                Address::new("localhost".to_string(), 10001),
-                CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert1.der"))?).into_owned(),
-            ),
-            NetworkParty::new(
-                PartyID::ID2.into(),
-                Address::new("localhost".to_string(), 10002),
-                CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert2.der"))?).into_owned(),
-            ),
+        let parties_certs = [
+            CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert0.der"))?).into_owned(),
+            CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert1.der"))?).into_owned(),
+            CertificateDer::from(std::fs::read(CONFIG_DIR.join("cert2.der"))?).into_owned(),
         ];
 
         let program_artifact =
@@ -274,19 +250,20 @@ gender = 0"#;
 
         let shares = split_input(PathBuf::from("Prover.toml"), &program_artifact)?;
 
-        let result = run_match(
-            shares,
-            parties.clone(),
-            &program_artifact,
-            constraint_system.clone(),
-            recursive,
-            has_zk,
-            prover_crs.clone(),
-            verifier_crs.clone(),
-        )
-        .await;
-
-        println!("result: {:?}", result);
+        for i in 0..5 {
+            let result = run_match(
+                i,
+                shares.clone(),
+                parties_certs.clone(),
+                &program_artifact,
+                constraint_system.clone(),
+                recursive,
+                has_zk,
+                prover_crs.clone(),
+                verifier_crs.clone(),
+            );
+            println!("result: {:?}", result);
+        }
 
         std::fs::remove_file("Prover.toml").unwrap();
 
diff --git a/mpc-server/src/matching.rs b/mpc-server/src/matching.rs
@@ -1,19 +1,23 @@
 use co_noir::{
-    AcirFormat, Bn254, NetworkConfig, NetworkParty, PartyID, Poseidon2Sponge, Rep3CoUltraHonk,
-    Rep3MpcNet, UltraHonk, merge_input_shares,
+    AcirFormat, Address, Bn254, NetworkConfig, NetworkParty, PartyID, Poseidon2Sponge,
+    Rep3CoUltraHonk, Rep3MpcNet, UltraHonk, merge_input_shares,
 };
 use co_ultrahonk::prelude::{ProverCrs, ZeroKnowledge};
 use noirc_artifacts::program::ProgramArtifact;
 use once_cell::sync::Lazy;
+use rayon::prelude::*;
+use rustls::pki_types::CertificateDer;
 use rustls::pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer};
-use std::sync::Arc;
-use std::thread;
 use std::{
     path::PathBuf,
+    sync::Arc,
+    thread,
     time::{Duration, Instant},
 };
 
-use crate::db::{connect_db, get_all_users, get_user, insert_matches, update_checked};
+use crate::db::{
+    connect_db, get_all_users, get_user, insert_matches, update_checked, update_checked_many,
+};
 use crate::shares::{Share, get_shares};
 
 pub const DATA_DIR: Lazy<PathBuf> =
@@ -25,7 +29,7 @@ pub const SHARES_DIR_2: Lazy<PathBuf> = Lazy::new(|| DATA_DIR.join("user2"));
 
 pub async fn run_matches(
     user_id: String,
-    parties: Vec<NetworkParty>,
+    parties_certs: [CertificateDer<'static>; 3],
     program_artifact: &ProgramArtifact,
     constraint_system: Arc<AcirFormat<ark_bn254::Fr>>,
     recursive: bool,
@@ -52,38 +56,45 @@ pub async fn run_matches(
             .collect::<Vec<String>>(),
     )?;
 
-    let mut verified_matches = Vec::new();
-
-    for user2 in all_users {
-        update_checked(&conn, &user2.id, vec![user_id.clone()])?;
+    let users2 = all_users
+        .iter()
+        .map(|u| u.id.clone())
+        .collect::<Vec<String>>();
 
-        let shares_user1 = get_shares(&user1.id, true)?;
-        let shares_user2 = get_shares(&user2.id, false)?;
+    let verified_matches = all_users
+        .into_par_iter()
+        .enumerate()
+        .map(
+            |(thread_id, user2)| -> Result<String, Box<dyn std::error::Error + Send + Sync + 'static>> {
+                let shares_user1 = get_shares(&user1.id, true)?;
+                let shares_user2 = get_shares(&user2.id, false)?;
 
-        let share0 = merge_shares(shares_user1[0].clone(), shares_user2[0].clone())?;
-        let share1 = merge_shares(shares_user1[1].clone(), shares_user2[1].clone())?;
-        let share2 = merge_shares(shares_user1[2].clone(), shares_user2[2].clone())?;
+                let share0 = merge_shares(shares_user1[0].clone(), shares_user2[0].clone())?;
+                let share1 = merge_shares(shares_user1[1].clone(), shares_user2[1].clone())?;
+                let share2 = merge_shares(shares_user1[2].clone(), shares_user2[2].clone())?;
 
-        match run_match(
-            [share0, share1, share2],
-            parties.clone(),
-            program_artifact,
-            constraint_system.clone(),
-            recursive,
-            has_zk,
-            prover_crs.clone(),
-            verifier_crs.clone(),
+                match run_match(
+                    thread_id,
+                    [share0, share1, share2],
+                    parties_certs.clone(),
+                    program_artifact,
+                    constraint_system.clone(),
+                    recursive,
+                    has_zk,
+                    prover_crs.clone(),
+                    verifier_crs.clone(),
+                ) {
+                    Ok(_) => Ok(user2.id),
+                    Err(e) => Err(e),
+                }
+            },
         )
-        .await
-        {
-            Ok(_) => {
-                verified_matches.push(user2.id);
-            }
-            Err(e) => {
-                println!("Error: {:?}", e);
-            }
-        }
-    }
+        .filter(|m| m.is_ok())
+        .collect::<Result<Vec<String>, _>>()?;
+
+    println!("verified matches: {verified_matches:?}");
+
+    update_checked_many(&conn, users2, vec![user_id.clone()])?;
 
     insert_matches(
         &conn,
@@ -95,9 +106,10 @@ pub async fn run_matches(
     Ok(())
 }
 
-pub async fn run_match(
+pub fn run_match(
+    thread_id: usize,
     [share0, share1, share2]: [Share; 3],
-    parties: Vec<NetworkParty>,
+    parties_certs: [CertificateDer<'static>; 3],
     program_artifact: &ProgramArtifact,
     constraint_system: Arc<AcirFormat<ark_bn254::Fr>>,
     recursive: bool,
@@ -107,9 +119,31 @@ pub async fn run_match(
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
     let match_time = Instant::now();
 
+    let party0_port = 10000 + thread_id as u16;
+    let party1_port = 11000 + thread_id as u16;
+    let party2_port = 12000 + thread_id as u16;
+
+    let parties = vec![
+        NetworkParty::new(
+            PartyID::ID0.into(),
+            Address::new("localhost".to_string(), party0_port),
+            parties_certs[0].clone(),
+        ),
+        NetworkParty::new(
+            PartyID::ID1.into(),
+            Address::new("localhost".to_string(), party1_port),
+            parties_certs[1].clone(),
+        ),
+        NetworkParty::new(
+            PartyID::ID2.into(),
+            Address::new("localhost".to_string(), party2_port),
+            parties_certs[2].clone(),
+        ),
+    ];
+
     let data0 = DataForThread {
         id: PartyID::ID0,
-        port: 10000,
+        port: party0_port,
         key: PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(std::fs::read(
             CONFIG_DIR.join("key0.der"),
         )?))
@@ -125,7 +159,7 @@ pub async fn run_match(
     };
     let data1 = DataForThread {
         id: PartyID::ID1,
-        port: 10001,
+        port: party1_port,
         key: PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(std::fs::read(
             CONFIG_DIR.join("key1.der"),
         )?))
@@ -141,7 +175,7 @@ pub async fn run_match(
     };
     let data2 = DataForThread {
         id: PartyID::ID2,
-        port: 10002,
+        port: party2_port,
         key: PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(std::fs::read(
             CONFIG_DIR.join("key2.der"),
         )?))
@@ -237,7 +271,6 @@ fn spawn_party(
     println!("pk time: {:?}", pk_time.elapsed());
 
     let proof_time = Instant::now();
-    // let (proof, _) = Rep3CoUltraHonk::<_, _, Poseidon2Sponge>::prove(net, pk, &prover_crs, has_zk)?;
     let (proof, _) = Rep3CoUltraHonk::<_, _, Poseidon2Sponge>::prove(net, pk, &prover_crs, has_zk)?;
     println!("proof time: {:?}", proof_time.elapsed());
 
diff --git a/web-app/src/pages/Matches.tsx b/web-app/src/pages/Matches.tsx
@@ -73,7 +73,7 @@ export default function Matches() {
         </h2>
         <div className="p-4 bg-gradient-to-r from-pink-50 to-purple-50 rounded-xl border border-pink-100 mb-6">
           <p className="text-gray-600 mb-2">
-            This might take up to 45 seconds - we&apos;re doing some heavy cryptographic lifting behind the scenes! 🔐
+            This might take up to 15 seconds - we&apos;re doing some heavy cryptographic lifting behind the scenes! 🔐
           </p>
           <p className="text-sm text-gray-500 mb-2">While you wait, here&apos;s what&apos;s happening:</p>
           <ul className="list-disc list-inside text-sm text-gray-500 mt-2 space-y-1">
