-
Notifications
You must be signed in to change notification settings - Fork 50
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: ideal functionality synchronization (#201)
- Loading branch information
Showing
17 changed files
with
246 additions
and
254 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,191 +1,58 @@ | ||
| //! Ideal functionality utilities. | ||
| use futures::channel::oneshot; | ||
| use std::{ | ||
| any::Any, | ||
| collections::HashMap, | ||
| sync::{Arc, Mutex, MutexGuard}, | ||
| }; | ||
| use std::sync::Arc; | ||
| use tokio::sync::Barrier; | ||
|
|
||
| use crate::{Context, ThreadId}; | ||
|
|
||
| type BoxAny = Box<dyn Any + Send + 'static>; | ||
|
|
||
| #[derive(Debug, Default)] | ||
| struct Buffer { | ||
| alice: HashMap<ThreadId, (BoxAny, oneshot::Sender<BoxAny>)>, | ||
| bob: HashMap<ThreadId, (BoxAny, oneshot::Sender<BoxAny>)>, | ||
| } | ||
|
|
||
| /// The ideal functionality from the perspective of Alice. | ||
| #[derive(Debug)] | ||
| pub struct Alice<F> { | ||
| f: Arc<Mutex<F>>, | ||
| buffer: Arc<Mutex<Buffer>>, | ||
| } | ||
|
|
||
| impl<F> Clone for Alice<F> { | ||
| fn clone(&self) -> Self { | ||
| Self { | ||
| f: self.f.clone(), | ||
| buffer: self.buffer.clone(), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl<F> Alice<F> { | ||
| /// Returns a lock to the ideal functionality. | ||
| pub fn get_mut(&mut self) -> MutexGuard<'_, F> { | ||
| self.f.lock().unwrap() | ||
| } | ||
|
|
||
| /// Calls the ideal functionality. | ||
| pub async fn call<Ctx, C, IA, IB, OA, OB>(&mut self, ctx: &mut Ctx, input: IA, call: C) -> OA | ||
| where | ||
| Ctx: Context, | ||
| C: FnOnce(&mut F, IA, IB) -> (OA, OB), | ||
| IA: Send + 'static, | ||
| IB: Send + 'static, | ||
| OA: Send + 'static, | ||
| OB: Send + 'static, | ||
| { | ||
| let receiver = { | ||
| let mut buffer = self.buffer.lock().unwrap(); | ||
| if let Some((input_bob, ret_bob)) = buffer.bob.remove(ctx.id()) { | ||
| let input_bob = *input_bob | ||
| .downcast() | ||
| .expect("alice received correct input type for bob"); | ||
|
|
||
| let (output_alice, output_bob) = | ||
| call(&mut self.f.lock().unwrap(), input, input_bob); | ||
|
|
||
| _ = ret_bob.send(Box::new(output_bob)); | ||
|
|
||
| return output_alice; | ||
| } | ||
|
|
||
| let (sender, receiver) = oneshot::channel(); | ||
| buffer | ||
| .alice | ||
| .insert(ctx.id().clone(), (Box::new(input), sender)); | ||
| receiver | ||
| }; | ||
|
|
||
| let output_alice = receiver.await.expect("bob did not drop the channel"); | ||
| *output_alice | ||
| .downcast() | ||
| .expect("bob sent correct output type for alice") | ||
| } | ||
| /// Creates a new call synchronizer between two parties. | ||
| pub fn call_sync() -> (CallSync, CallSync) { | ||
| let barrier = Arc::new(Barrier::new(2)); | ||
| ( | ||
| CallSync { | ||
| barrier: Arc::clone(&barrier), | ||
| }, | ||
| CallSync { barrier }, | ||
| ) | ||
| } | ||
|
|
||
| /// The ideal functionality from the perspective of Bob. | ||
| /// Synchronizes function calls between two parties. | ||
| #[derive(Debug)] | ||
| pub struct Bob<F> { | ||
| f: Arc<Mutex<F>>, | ||
| buffer: Arc<Mutex<Buffer>>, | ||
| } | ||
|
|
||
| impl<F> Clone for Bob<F> { | ||
| fn clone(&self) -> Self { | ||
| Self { | ||
| f: self.f.clone(), | ||
| buffer: self.buffer.clone(), | ||
| } | ||
| } | ||
| pub struct CallSync { | ||
| barrier: Arc<Barrier>, | ||
| } | ||
|
|
||
| impl<F> Bob<F> { | ||
| /// Returns a lock to the ideal functionality. | ||
| pub fn get_mut(&mut self) -> MutexGuard<'_, F> { | ||
| self.f.lock().unwrap() | ||
| } | ||
|
|
||
| /// Calls the ideal functionality. | ||
| pub async fn call<Ctx, C, IA, IB, OA, OB>(&mut self, ctx: &mut Ctx, input: IB, call: C) -> OB | ||
| where | ||
| Ctx: Context, | ||
| C: FnOnce(&mut F, IA, IB) -> (OA, OB), | ||
| IA: Send + 'static, | ||
| IB: Send + 'static, | ||
| OA: Send + 'static, | ||
| OB: Send + 'static, | ||
| { | ||
| let receiver = { | ||
| let mut buffer = self.buffer.lock().unwrap(); | ||
| if let Some((input_alice, ret_alice)) = buffer.alice.remove(ctx.id()) { | ||
| let input_alice = *input_alice | ||
| .downcast() | ||
| .expect("bob received correct input type for alice"); | ||
| impl CallSync { | ||
| /// Synchronizes a call. | ||
| pub async fn call<F: FnMut() -> R, R>(&mut self, mut f: F) -> Option<R> { | ||
| // Wait for both parties to call. | ||
| let is_leader = self.barrier.wait().await.is_leader(); | ||
|
|
||
| let (output_alice, output_bob) = | ||
| call(&mut self.f.lock().unwrap(), input_alice, input); | ||
| let ret = if is_leader { Some(f()) } else { None }; | ||
|
|
||
| _ = ret_alice.send(Box::new(output_alice)); | ||
| // Wait for the call to return. | ||
| self.barrier.wait().await; | ||
|
|
||
| return output_bob; | ||
| } | ||
|
|
||
| let (sender, receiver) = oneshot::channel(); | ||
| buffer | ||
| .bob | ||
| .insert(ctx.id().clone(), (Box::new(input), sender)); | ||
| receiver | ||
| }; | ||
|
|
||
| let output_bob = receiver.await.expect("alice did not drop the channel"); | ||
| *output_bob | ||
| .downcast() | ||
| .expect("alice sent correct output type for bob") | ||
| ret | ||
| } | ||
| } | ||
|
|
||
| /// Creates an ideal functionality, returning the perspectives of Alice and Bob. | ||
| pub fn ideal_f2p<F>(f: F) -> (Alice<F>, Bob<F>) { | ||
| let f = Arc::new(Mutex::new(f)); | ||
| let buffer = Arc::new(Mutex::new(Buffer::default())); | ||
|
|
||
| ( | ||
| Alice { | ||
| f: f.clone(), | ||
| buffer: buffer.clone(), | ||
| }, | ||
| Bob { f, buffer }, | ||
| ) | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod test { | ||
| use crate::executor::test_st_executor; | ||
| use std::sync::Mutex; | ||
|
|
||
| use super::*; | ||
|
|
||
| #[test] | ||
| fn test_ideal() { | ||
| let (mut alice, mut bob) = ideal_f2p(()); | ||
| let (mut ctx_a, mut ctx_b) = test_st_executor(8); | ||
| #[tokio::test] | ||
| async fn test_call_sync() { | ||
| let x = Arc::new(Mutex::new(0)); | ||
|
|
||
| let (output_a, output_b) = futures::executor::block_on(async { | ||
| futures::join!( | ||
| alice.call(&mut ctx_a, 1u8, |&mut (), a: u8, b: u8| (a + b, a + b)), | ||
| bob.call(&mut ctx_b, 2u8, |&mut (), a: u8, b: u8| (a + b, a + b)), | ||
| ) | ||
| }); | ||
| let (mut sync_0, mut sync_1) = call_sync(); | ||
|
|
||
| assert_eq!(output_a, 3); | ||
| assert_eq!(output_b, 3); | ||
| } | ||
| let add_one = || { | ||
| *x.lock().unwrap() += 1; | ||
| }; | ||
|
|
||
| #[test] | ||
| #[should_panic] | ||
| fn test_ideal_wrong_input_type() { | ||
| let (mut alice, mut bob) = ideal_f2p(()); | ||
| let (mut ctx_a, mut ctx_b) = test_st_executor(8); | ||
| futures::join!(sync_0.call(add_one.clone()), sync_1.call(add_one)); | ||
|
|
||
| futures::executor::block_on(async { | ||
| futures::join!( | ||
| alice.call(&mut ctx_a, 1u16, |&mut (), a: u16, b: u16| (a + b, a + b)), | ||
| bob.call(&mut ctx_b, 2u8, |&mut (), a: u8, b: u8| (a + b, a + b)), | ||
| ) | ||
| }); | ||
| assert_eq!(*x.lock().unwrap(), 1); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.