Finished parallel submap implementation, experimental setup.

sparta/makefile

@@ -0,0 +1,24 @@
+HEAP_SIZE = 0x100000000
+STACK_SIZE = 0x400000
+MAX_THREADS = 9
+
+BUILD_DIR = target/x86_64-fortanix-unknown-sgx/release
+
+BINARY = $(BUILD_DIR)/sparta
+TARGET = $(BUILD_DIR)/sparta.sgxs
+SIGNATURE = $(BUILD_DIR)/sparta.sig
+
+.PHONY: clean run
+
+$(TARGET): $(BINARY)
+	ftxsgx-elf2sgxs -o $(TARGET) $(BINARY) --heap-size $(HEAP_SIZE) --stack-size $(STACK_SIZE) --threads $(MAX_THREADS)
+	sgxs-sign --key private.pem $(TARGET) $(SIGNATURE)
+
+$(BINARY):
+	cargo build --release
+
+run: $(TARGET)
+	ftxsgx-runner $(TARGET) 5 3 3 8 4
+
+clean:
+	rm -rf $(BUILD_DIR)

sparta/private.pem

@@ -0,0 +1,39 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIG4gIBAAKCAYEAuod+uoYcb9c2Xm4P7NxVwKBBavYbri4etwTs+V4g5zZ+RoZT
+AjlNYEDl33m+UE/3azOCvu5sHCMUjeRsPPAvcnA1o5+XTde1zR4oqK6zkPYC3kp4
+x5l2hVbxdACwn4/HlDM/DF8U2hxQ1c+vp3W0xSn1YJmOHYdHkRJBS14oZcVr7k4i
+kCfx9vSNqP6ALTV3OpQZgtS+lbCFiu5YX3A2COVKcxwS/bzrcFYsS5yI/s53CGDL
+IzAegcEVXftHGZHHR523CWo27KBtMQPsoLA0w2XLFQYiKYPfXV1yfrD2x4Wwx8TD
+NiOv21bV1lP2C6J7pBpj28Bre6+Db/faQ7IUqAQqqlqiYnc9KIkRAlfSYtL8KkzL
+O1Jd8Kxh3BCGIsa3ZJp+5MkKlkG11CmJA/YTreqHMR+DgdRUMYmmx4NC4z3KrR2c
+FkysHoatsNMI34q7WbZr/gzGnS2rkzHRnlWRejqyBIMPdFcSGBXEek4KYUkpK3FK
+NALuo906JGvzJ5iJAgEDAoIBgHxaVHxZaEqPeZRJX/M9joBq1kdOvR7Jac9YnfuU
+Fe95qYRZjKwmM5WAmT+mfuA1T5zNAdSe8r1suF6YSCigH6GgI8JqZN6Pzoi+xcXJ
+zQtOrJQxpdpmTwOPS6KrIGpf2mLM1LLqDea9izk1H8T5Iy4b+OsRCWkE2mC21jI+
+xZkuR/Q0FwrFS/n4XnCpqsjOT3xiu6yN1GPLA7H0OupKzrCY3Ey9Yf598krkHYe9
+sKne+gWV3MIgFFaAuOlSL2ZhLmG20HuC+yttQDUC7e+lGaXGEcLDQCj07SOQ3X0I
+Ws1YoOISqsndDEAfjA8Fevx/2+pOH4wqzg7tBBzxvJGI4KmD+rkl3BJWYWjCXQuC
+VQ4FXLxhQLQNU7TqAhZxa5eeO3geDNSEZhZfVH4p6Rim/vxbvIDH+pe1HSXqJvFu
+Zfy9sX9SMShgWeprf/RdULprF0iwBO+USjHuZY4lnaRTtrwa5tJ8QCknO6EUy35u
+xA+oQJeS+yuLIsyLF2Lx24knYwKBwQDbRjhHeQfAVZhW6yKOGmFMwkeAztsqRvbJ
+welT1jW6cBjzdbnaJv83GHTBNfEzUh06s3xZmXN4TKEWrZjLdnlhGZvvyJnxtj58
+gyI0iUErtN5s4Ma04ThQczxcm8lHmSzvoWi+cZakv7D3GRxVZ+FJQkmVVW3G3KMT
+KnPcCsAdv2M22Y42aWNA5PZ2vL/XYDvmIHpUer+wgVfllAumUqNmaYZbU2cy7Tmt
+tt8G+yILYDGTIktCpEmbq6epaY2IYjcCgcEA2cVGCKy2ayb0ipRlKx4s/fpo8BJm
+wc2E3kI870442RCmgTMJLTEeRA8O/pXMXZCUifG4L+bC7P9cLd+mne5KSKSWpNk+
+VT9sjuJjPIsRohYSotojeF0oENoiXcqbd8AxQMwC55Daz/MGH9GPCZOz6h6i2p4d
+12gFZX/4gVLpymHdN0jEI1LB/gB3S4pP8Incho3qQjOXoca7WGVZgYVb+ajRbXVI
+A67++L+0NazCPmyWtPKGMA3w1WDvZpecUXs/AoHBAJIu0C+mBSrjuuScwbQRljMs
+L6s0khwvTzEr8OKOznxKu0z5JpFvVM9loyt5S3eME3x3qDu7olAzFg8eZdz5pkC7
+vUqFu/Z5fv2swXhbgMfN6Z3rLyNA0DWiKD29MNpmHfUWRdRLucMqdfoQvY5FQNuB
+hmOOSS89wgzG9+gHKr5/l3nmXs7w7NXt+aR91TpAJ+7AUY2nKnWrj+5ism7hwkRG
+WZI3miHze8kklK9SFrJAIQzBh4HC270dGnDxCQWWzwKBwQCRLi6wcyRHb02xuENy
+FB3+pvCgDESBM63pgX303tCQtcRWIgYeIL7YCgn/Dog+YGMGoSV1RIHzVOgelRm+
+ntwwbbnDO37jf520luzTB2EWuWHB5sJQPhq15sGT3GelKsuAiAHvtec1TK6/4QoG
+YnfxacHnFBPk8AOY//sA4fExlpN6MILCNyv+qvoyXDVLBpMEXpwsImUWhHzlmOZW
+WOf7xeDzo4VXyf9Qf815HdbUSGR4oa7Ks/XjlfTvD72Lp38CgcBj4cZu9aeR6jQ/
+z6XPi6M/iiVxhrW5qL2pWegSYBB5TsLnAafNs7j/WVtKIErJWe+QzGk7fKqH2SEJ
+YkZAWv01d+UPojd2pfvQY/EwWe62oo4NpDCHMS6ql/PwIaxSZS7lR535AKiHGVJJ
+3AJi3vvfVRY1Re+p7LRkaDlVJQNDfZONKg1VG+3Y2MMgA1NyA7w54t8i1VlM5FjS
+r8FDDHKRGwNvLINus54JWj4StKec4MlHA8ZUww37zz5GzbsTrx0=
+-----END RSA PRIVATE KEY-----

sparta/src/load_balancer.rs

@@ -2,7 +2,7 @@ use crate::omap::ObliviousMap;
 pub use crate::record::{IndexRecord, Record, RecordType, SubmapRecord};
 use fastapprox::fast;
 use otils::{self, ObliviousOps};
-use std::{cmp, f64::consts::E};
+use std::{cmp, f64::consts::E, thread};
 
 const LAMBDA: usize = 128;
 
@@ -41,7 +41,7 @@ impl LoadBalancer {
 
         LoadBalancer {
             num_users,
-            num_threads,
+            num_threads: num_threads / num_submaps,
             num_submaps,
             user_store,
             submaps,
@@ -171,10 +171,29 @@ impl LoadBalancer {
             .map(|r| r.0)
             .collect();
 
-        for idx in 0..self.num_submaps {
-            let batch: Vec<Record> = requests.drain(0..submap_size).collect();
-            self.submaps[idx].batch_send(batch);
-        }
+        let mut remaining_submaps = &mut self.submaps[..];
+        let mut remaining_requests = &mut requests[..self.num_submaps * submap_size];
+
+        thread::scope(|s| {
+            for _ in 0..self.num_submaps - 1 {
+                let (submap, rest_submaps) = remaining_submaps.split_at_mut(1);
+                remaining_submaps = rest_submaps;
+
+                let (requests, rest_requests) = remaining_requests.split_at_mut(submap_size);
+                remaining_requests = rest_requests;
+                s.spawn(|| submap[0].batch_send(requests.to_vec()));
+            }
+
+            let (submap, rest_submaps) = remaining_submaps.split_at_mut(1);
+            remaining_submaps = rest_submaps;
+
+            let (requests, rest_requests) = remaining_requests.split_at_mut(submap_size);
+            remaining_requests = rest_requests;
+            submap[0].batch_send(requests.to_vec());
+        });
+
+        // parallelize
+        for _ in 0..self.num_submaps {}
     }
 
     fn update_with_fetches(&mut self, fetches: Vec<IndexRecord>, num_fetches: usize) {
@@ -252,14 +271,39 @@ impl LoadBalancer {
             .map(|r| r.0)
             .collect();
 
-        let mut responses: Vec<IndexRecord> = Vec::new();
-        responses.reserve(submap_size * self.num_submaps);
+        let mut responses: Vec<IndexRecord> = Vec::with_capacity(submap_size * self.num_submaps);
+        (0..submap_size * self.num_submaps)
+            .for_each(|_| responses.push(IndexRecord::new(0, RecordType::Dummy)));
 
-        // parallelize
-        for idx in 0..self.num_submaps {
-            let batch: Vec<Record> = requests.drain(0..submap_size).collect();
-            responses.extend(self.submaps[idx].batch_fetch(batch));
-        }
+        let mut remaining_submaps = &mut self.submaps[..];
+        let mut remaining_requests = &mut requests[..self.num_submaps * submap_size];
+        let mut remaining_responses = &mut responses[..];
+
+        thread::scope(|s| {
+            for _ in 0..self.num_submaps - 1 {
+                let (submap, rest_submaps) = remaining_submaps.split_at_mut(1);
+                remaining_submaps = rest_submaps;
+                // println!("submaps: {}", remaining_submaps.len());
+
+                let (batch, rest_requests) = remaining_requests.split_at_mut(submap_size);
+                remaining_requests = rest_requests;
+                // println!("");
+
+                let (responses, rest_responses) = remaining_responses.split_at_mut(submap_size);
+                remaining_responses = rest_responses;
+                s.spawn(|| submap[0].batch_fetch(batch.to_vec(), responses));
+            }
+
+            let (submap, rest_submaps) = remaining_submaps.split_at_mut(1);
+            remaining_submaps = rest_submaps;
+
+            let (batch, rest_requests) = remaining_requests.split_at_mut(submap_size);
+            remaining_requests = rest_requests;
+
+            let (responses, rest_responses) = remaining_responses.split_at_mut(submap_size);
+            remaining_responses = rest_responses;
+            submap[0].batch_fetch(batch.to_vec(), responses);
+        });
 
         // this only really needs to be a shuffle
         responses = otils::sort(responses, self.num_threads);

sparta/src/main.rs

@@ -2,24 +2,43 @@ mod load_balancer;
 mod omap;
 mod record;
 
+use std::time::UNIX_EPOCH;
+
 use load_balancer::LoadBalancer;
-// use omap::ObliviousMap;
 use record::Record;
 
 fn main() {
-    let mut l = LoadBalancer::new(5, 8, 4);
-    let sends: Vec<Record> = (0..3)
-        .map(|x| Record::send(x, x.try_into().unwrap()))
+    let args: Vec<String> = std::env::args().collect();
+    let num_users = args[1].parse::<usize>().unwrap();
+    let num_sends = args[2].parse::<usize>().unwrap();
+    let num_fetches = args[3].parse::<usize>().unwrap();
+    let num_threads = args[4].parse::<usize>().unwrap();
+    let num_maps = args[5].parse::<usize>().unwrap();
+
+    let mut l = LoadBalancer::new(num_users as i64, num_threads, num_maps);
+    let sends: Vec<Record> = (0..num_sends)
+        .map(|x| Record::send(x as i64, x.try_into().unwrap()))
         .collect();
 
-    println!("--- SEND ---\n");
     l.batch_send(sends);
 
-    let fetches: Vec<Record> = vec![Record::fetch(0, 3)];
+    let results: Vec<u128> = (0..15)
+        .map(|_| {
+            let start = std::time::SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_nanos();
+            let _ = l.batch_fetch(vec![Record::fetch(0, num_fetches as u64)]);
+            let end = std::time::SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_nanos();
+            end - start
+        })
+        .collect();
 
-    println!("--- FETCH ---\n");
-    let responses = l.batch_fetch(fetches);
-    for response in responses.iter() {
-        println!("{:?}", response);
+    for idx in 5..results.len() {
+        print!("{}\t", results[idx]);
     }
+    println!();
 }

sparta/src/omap.rs

@@ -41,6 +41,7 @@ impl Max for MapRecord {
     }
 }
 
+#[derive(Default)]
 pub struct ObliviousMap {
     num_threads: usize,
     message_store: Vec<MapRecord>,
@@ -76,7 +77,7 @@ impl ObliviousMap {
             .extend(requests.into_iter().map(|r| MapRecord(r)));
     }
 
-    pub fn batch_fetch(&mut self, requests: Vec<Record>) -> Vec<IndexRecord> {
+    pub fn batch_fetch(&mut self, requests: Vec<Record>, response: &mut [IndexRecord]) {
         let final_size = self.message_store.len();
         let num_requests = requests.len();
 
@@ -100,11 +101,11 @@ impl ObliviousMap {
             self.num_threads,
         );
 
-        let response: Vec<IndexRecord> = self
-            .message_store
-            .drain(0..num_requests)
-            .map(|r| IndexRecord(r.0))
-            .collect();
+        for idx in 0..num_requests {
+            response[idx] = IndexRecord(self.message_store[idx].0.clone());
+        }
+
+        self.message_store.drain(0..num_requests);
 
         otils::compact(
             &mut self.message_store[..],
@@ -112,7 +113,5 @@ impl ObliviousMap {
             self.num_threads,
         );
         self.message_store.truncate(final_size);
-
-        response
     }
 }

sparta/src/record.rs

@@ -46,8 +46,8 @@ impl Record {
         Record::new(uid, RecordType::Send, message, 0, 0)
     }
 
-    pub fn fetch(uid: i64, message: u64) -> Self {
-        Record::new(uid, RecordType::Fetch, message, 0, 0)
+    pub fn fetch(uid: i64, volume: u64) -> Self {
+        Record::new(uid, RecordType::Fetch, volume, 0, 0)
     }
 
     pub fn is_user_store(&self) -> bool {