zerotier/sequential-exchange
0
0
Fork 0
mirror of https://github.com/zerotier/sequential-exchange.git synced 2026-05-22 16:28:28 -07:00
Code Issues Packages Projects Releases Wiki Activity

improved multithreaded API

Browse Source
This commit is contained in:
Monica Moniot
2023-08-04 18:14:13 -04:00
parent 1978bc39f8
commit 60eedb123e
10 changed files with 258 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+32
View File
@@ -2,6 +2,38 @@
# It is not intended for manual editing.
version = 3
[[package]]
name = "futex"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d370e15a8972dee506ad50638e331183109e85ff99f349f19e04f288dc6cef3"
dependencies = [
"integer-atomics",
"libc",
"lock-wrappers",
]
[[package]]
name = "integer-atomics"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5c33cd4d18b4ade167caace0e92364e8568c1e47c193738397b4b48a3e414139"
[[package]]
name = "libc"
version = "0.2.147"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4668fb0ea861c1df094127ac5f1da3409a82116a4ba74fca2e58ef927159bb3"
[[package]]
name = "lock-wrappers"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d654f44a90e266c873afdcf93f4506d9d6fb036b211e2be63a09695ecb0a07a"
[[package]]
name = "seq_ex"
version = "0.1.0"
dependencies = [
"futex",
]
Cargo.toml
+4 -1
View File
@@ -11,4 +11,7 @@ doc = true
[features]
default = ["std"]
std = []
std = ["futex"]
[dependencies]
futex = {version = "0.1.3", optional = true}
examples/calculator.rs
+15 -21
View File
@@ -43,8 +43,6 @@ struct Transport {
impl seq_ex::TransportLayer for &Transport {
type RecvData = SendPacket;
type RecvDataRef<'a> = &'a SendPacket;
type RecvReturn = ();
type SendData = RawPacket;
@@ -67,20 +65,16 @@ impl seq_ex::TransportLayer for &Transport {
let _ = self.channel.send(RawPacket::EmptyReply(reply_no));
}
}
fn deserialize<'a>(data: &'a Self::RecvData) -> Self::RecvDataRef<'a> {
data
}
fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option<Self::SendData>) -> Self::RecvReturn {
let mut value = self.value.lock().unwrap();
use SendPacket::*;
match recv_packet {
Add(n) => *value = *value + n,
Sub(n) => *value = *value - n,
Mul(n) => *value = *value * n,
Div(n) => *value = *value / n,
Mod(n) => *value = *value % n,
}
}
fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option<Self::SendData>) -> Self::RecvReturn {
let mut value = self.value.lock().unwrap();
use SendPacket::*;
match recv_packet {
Add(n) => *value = *value + n,
Sub(n) => *value = *value - n,
Mul(n) => *value = *value * n,
Div(n) => *value = *value / n,
Mod(n) => *value = *value % n,
}
}
@@ -107,15 +101,15 @@ fn main() {
let mut seq2 = SeqEx::new(5, 1);
let mut value = 0.0;
assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0))));
assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0))));
value += 1.0;
assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0))));
assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0))));
value -= 2.0;
assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0))));
assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0))));
value *= 3.0;
assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0))));
assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0))));
value /= 4.0;
assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0))));
assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0))));
value %= 5.0;
for _ in 0..30 {
examples/hello_world.rs
+36 -72
View File
@@ -1,94 +1,58 @@
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::mpsc::Receiver;
use seq_ex::{ReplyGuard, SeqEx, SeqNo};
use std::time::Instant;
use seq_ex::sync::{MpscTransport, PacketType};
use seq_ex::{ReplyGuard, SeqEx};
#[derive(Clone, Debug)]
enum Packet {
Ack(SeqNo),
EmptyReply(SeqNo),
Hello(SeqNo),
Reply(SeqNo, SeqNo, ReplyPacket),
}
#[derive(Clone, Debug)]
enum ReplyPacket {
Hello,
Space,
World,
Exclamation,
}
use Packet::*;
struct Transport {
channel: Sender<Packet>,
time: Instant,
}
impl seq_ex::TransportLayer for &Transport {
type RecvData = Packet;
type SendData = Packet;
fn time(&self) -> i64 {
self.time.elapsed().as_millis() as i64
}
fn send(&self, data: &Self::SendData) {
let _ = self.channel.send(data.clone());
}
fn send_ack(&self, reply_no: SeqNo) {
let _ = self.channel.send(Packet::Ack(reply_no));
}
fn send_empty_reply(&self, reply_no: SeqNo) {
let _ = self.channel.send(Packet::EmptyReply(reply_no));
}
}
fn process(guard: ReplyGuard<'_, &Transport>, recv_packet: Packet, send_packet: Option<Packet>) {
use Packet::*;
use ReplyPacket::*;
fn process(guard: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, send_packet: Option<Packet>) {
match (recv_packet, send_packet) {
(Hello(_), None) => {
(Hello, None) => {
print!("Hello");
guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Space));
guard.reply(Space);
}
(Space, Some(Hello)) => {
print!(" ");
guard.reply(World);
}
(World, Some(Space)) => {
print!("World");
guard.reply(Exclamation);
}
(Exclamation, Some(World)) => {
print!("!");
}
(Reply(_, _, r), Some(p)) => match (r, p) {
(Space, Hello(_)) => {
print!(" ");
guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, World));
}
(World, Reply(_, _, Space)) => {
print!("World");
guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Exclamation));
}
(Exclamation, Reply(_, _, World)) => {
println!("!");
}
(a, b) => {
println!("Unsolicited reply received: {:?}, was a reply to: {:?}", a, b);
}
},
(a, None) => {
println!("Unsolicited packet received: {:?}", a);
print!("Unsolicited packet received: {:?}", a);
}
(a, Some(b)) => {
println!("Unsolicited reply received: {:?}, was a reply to: {:?}", a, b);
print!("Incorrect reply received: {:?}, was a reply to: {:?}", a, b);
}
}
}
fn receive<'a>(recv: &Receiver<Packet>, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) {
use Packet::*;
fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &mut SeqEx<&'a MpscTransport<Packet>>, transport: &'a MpscTransport<Packet>) {
let do_pump = {
let result = match recv.recv().unwrap() {
Ack(reply_no) => {
PacketType::Ack { reply_no } => {
seq.receive_ack(reply_no);
return;
}
EmptyReply(reply_no) => {
seq.receive_empty_reply(reply_no);
PacketType::EmptyReply { reply_no } => {
if let Some(Exclamation) = seq.receive_empty_reply(reply_no) {
// Our Hello World exchange ends right here.
print!("\n");
}
return;
}
Hello(seq_no) => seq.receive(transport, seq_no, None, Hello(seq_no)),
Reply(seq_no, reply_no, p) => seq.receive(transport, seq_no, Some(reply_no), Reply(seq_no, reply_no, p)),
PacketType::Data { seq_no, reply_no, payload } => seq.receive(transport, seq_no, reply_no, payload),
};
if let Ok((guard, recv_packet, send_packet)) = result {
process(guard, recv_packet, send_packet);
@@ -105,17 +69,17 @@ fn receive<'a>(recv: &Receiver<Packet>, seq: &mut SeqEx<&'a Transport>, transpor
}
fn main() {
let (send1, recv1) = channel();
let (send2, recv2) = channel();
let transport1 = Transport { channel: send2, time: Instant::now() };
let transport2 = Transport { channel: send1, time: Instant::now() };
let mut seq1 = SeqEx::new(100, 1);
let mut seq2 = SeqEx::new(100, 1);
let (transport1, recv2) = MpscTransport::new();
let (transport2, recv1) = MpscTransport::new();
let mut seq1 = SeqEx::default();
let mut seq2 = SeqEx::default();
assert!(seq1.send_with(&transport1, |seq_no| Packet::Hello(seq_no)));
// We begin a "Hello World" exchange right here.
assert!(seq1.try_send(&transport1, Packet::Hello).is_ok());
receive(&recv2, &mut seq2, &transport2);
receive(&recv1, &mut seq1, &transport1);
receive(&recv2, &mut seq2, &transport2);
receive(&recv1, &mut seq1, &transport1);
receive(&recv2, &mut seq2, &transport2);
}
src/lib.rs
+1 -1
View File
@@ -11,4 +11,4 @@ mod single_thread;
pub use single_thread::*;
#[cfg(feature = "std")]
pub mod multi_thread;
pub mod sync;
src/multi_thread.rs
-62
View File
@@ -1,62 +0,0 @@
use std::sync::{Mutex, MutexGuard};
use crate::{Error, SeqEx, SeqNo, TransportLayer};
pub struct SeqExLock<TL: TransportLayer>(pub Mutex<SeqEx<TL>>);
pub struct ReplyGuard<'a, TL: TransportLayer>(&'a SeqExLock<TL>, TL, SeqNo);
impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) {
let mut seq = self.0 .0.lock().unwrap();
let p = packet_data(seq.seq_no(), self.2);
seq.reply_raw(self.1.clone(), self.2, p);
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
fn drop(&mut self) {
let mut seq = self.0 .0.lock().unwrap();
seq.reply_empty_raw(self.1.clone(), self.2);
}
}
impl<TL: TransportLayer> SeqExLock<TL> {
pub fn receive<P: Into<TL::RecvData>, T>(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<(ReplyGuard<'_, TL>, P, Option<TL::SendData>), Error> {
let mut seq = self.lock();
seq.receive_raw(app.clone(), seq_no, reply_no, packet)
.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
}
pub fn pump(&self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option<TL::SendData>), Error> {
let mut seq = self.lock();
seq.pump_raw()
.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
}
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
Self(Mutex::new(SeqEx::new(retry_interval, initial_seq_no)))
}
pub fn send(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> TL::SendData) -> bool {
let mut seq = self.lock();
let p = packet_data(seq.seq_no());
seq.send(app, p)
}
pub fn receive_ack(&self, reply_no: SeqNo) {
self.lock().receive_ack(reply_no)
}
pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Option<TL::SendData> {
self.lock().receive_empty_reply(reply_no)
}
pub fn service(&self, app: TL) -> i64 {
self.lock().service(app)
}
pub fn lock(&self) -> MutexGuard<SeqEx<TL>> {
self.0.lock().unwrap()
}
}
src/seq_queue.rs
+16 -11
View File
@@ -41,6 +41,11 @@ use crate::TransportLayer;
/// All packets will either have a seq_no, a reply_no, or both.
pub type SeqNo = u32;
/// The resend interval for a default instance of SeqEx.
pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 200;
/// The initial sequence number for a default instance of SeqEx.
pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 1;
const MAX_CONCURRENCY: usize = 24;
pub struct SeqEx<TL: TransportLayer, const SLEN: usize = 64, const RLEN: usize = 32> {
/// The interval at which packets will be resent if they have not yet been acknowledged by the
@@ -151,9 +156,9 @@ impl<TL: TransportLayer> SeqEx<TL> {
/// user would like. However this choice of units must be consistent with the units of the
/// `retry_interval`. `current_time` does not have to be monotonically increasing.
#[must_use = "The queue might be full causing the packet to not be sent"]
pub fn send(&mut self, app: TL, packet_data: TL::SendData) -> bool {
pub fn try_send(&mut self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
if self.is_full() {
return false;
return Err(packet_data);
}
let seq_no = self.next_send_seq_no;
self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1);
@@ -166,13 +171,8 @@ impl<TL: TransportLayer> SeqEx<TL> {
data: packet_data,
});
app.send(&entry.data);
true
}
#[must_use = "The queue might be full causing the packet to not be sent"]
pub fn send_with(&mut self, app: TL, create_data: impl FnOnce(SeqNo) -> TL::SendData) -> bool {
let p = create_data(self.seq_no());
self.send(app, p)
app.send(entry.seq_no, entry.reply_no, &entry.data);
Ok(())
}
pub fn receive_raw<P: Into<TL::RecvData>>(
@@ -303,7 +303,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
if let Some(entry) = item {
if entry.next_resent_time <= current_time {
entry.next_resent_time = next_interval;
app.send(&entry.data);
app.send(entry.seq_no, entry.reply_no, &entry.data);
} else {
next_activity = next_activity.min(entry.next_resent_time);
}
@@ -326,7 +326,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
data: packet_data,
});
app.send(&entry.data);
app.send(entry.seq_no, entry.reply_no, &entry.data);
}
}
pub fn reply_empty_raw(&mut self, app: TL, reply_no: SeqNo) {
@@ -352,6 +352,11 @@ impl<TL: TransportLayer> SeqEx<TL> {
IterMut(self.send_window.iter_mut())
}
}
impl<TL: TransportLayer> Default for SeqEx<TL> {
fn default() -> Self {
Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
}
}
impl<'a, TL: TransportLayer> IntoIterator for &'a SeqEx<TL> {
type Item = &'a TL::SendData;
type IntoIter = Iter<'a, TL>;
src/single_thread.rs
-4
View File
@@ -12,10 +12,6 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
self.0.reply_raw(self.1.clone(), self.2, packet_data);
core::mem::forget(self);
}
pub fn reply_with(self, create_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) {
let p = create_data(self.seq_no(), self.reply_no());
self.reply(p)
}
}
impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
fn drop(&mut self) {
src/sync.rs
+151
View File
@@ -0,0 +1,151 @@
use std::cell::UnsafeCell;
use std::sync::Condvar;
use std::{
sync::{
mpsc::{channel, Receiver, Sender},
Mutex, MutexGuard,
},
time::Instant,
};
use crate::{Error, SeqEx, SeqNo, TransportLayer};
pub struct SeqExSync<TL: TransportLayer> {
seq_ex: Mutex<SeqEx<TL>>,
/// The mutex above is always held when this value changes, hence it is safe to mutate.
/// We don't pack this as a component of the mutex to avoid having to reimplement MutexGuard.
wait_count: UnsafeCell<usize>,
send_block: Condvar,
}
pub struct ReplyGuard<'a, TL: TransportLayer>(&'a SeqExSync<TL>, TL, SeqNo);
impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
pub fn reply(self, packet_data: TL::SendData) {
let mut seq = self.0.seq_ex.lock().unwrap();
seq.reply_raw(self.1.clone(), self.2, packet_data);
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
fn drop(&mut self) {
let mut seq = self.0.seq_ex.lock().unwrap();
seq.reply_empty_raw(self.1.clone(), self.2);
}
}
impl<TL: TransportLayer> SeqExSync<TL> {
pub fn receive<P: Into<TL::RecvData>, T>(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<(ReplyGuard<'_, TL>, P, Option<TL::SendData>), Error> {
let mut seq = self.lock();
let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet);
self.unblock(ret.is_ok());
ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
}
pub fn pump(&self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option<TL::SendData>), Error> {
let mut seq = self.lock();
let ret = seq.pump_raw();
self.unblock(ret.is_ok());
ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data))
}
#[inline]
fn unblock(&self, is_ok: bool) {
let has_waiting = unsafe {
*self.wait_count.get() > 0
};
if has_waiting && is_ok {
self.send_block.notify_one();
}
}
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
Self {
seq_ex: Mutex::new(SeqEx::new(retry_interval, initial_seq_no)),
wait_count: UnsafeCell::new(0),
send_block: Condvar::default(),
}
}
pub fn try_send(&self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> {
let mut seq = self.lock();
seq.try_send(app, packet_data)
}
pub fn send(&self, app: TL, mut packet_data: TL::SendData) {
let mut seq = self.lock();
while let Err(p) = seq.try_send(app.clone(), packet_data) {
packet_data = p;
unsafe { *self.wait_count.get() += 1}
seq = self.send_block.wait(seq).unwrap();
unsafe { *self.wait_count.get() -= 1}
}
}
pub fn receive_ack(&self, reply_no: SeqNo) {
self.lock().receive_ack(reply_no)
}
pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Option<TL::SendData> {
let ret = self.lock().receive_empty_reply(reply_no);
if ret.is_some() {
self.send_block.notify_one();
}
ret
}
pub fn service(&self, app: TL) -> i64 {
self.lock().service(app)
}
pub fn lock(&self) -> MutexGuard<SeqEx<TL>> {
self.seq_ex.lock().unwrap()
}
}
#[derive(Clone)]
pub enum PacketType<Payload: Clone> {
Data {
seq_no: SeqNo,
reply_no: Option<SeqNo>,
payload: Payload,
},
Ack {
reply_no: SeqNo,
},
EmptyReply {
reply_no: SeqNo,
},
}
#[derive(Clone)]
pub struct MpscTransport<Payload: Clone> {
pub channel: Sender<PacketType<Payload>>,
pub time: Instant,
}
impl<Payload: Clone> MpscTransport<Payload> {
pub fn new() -> (Self, Receiver<PacketType<Payload>>) {
let (send, recv) = channel();
(Self { channel: send, time: std::time::Instant::now() }, recv)
}
pub fn from_sender(send: Sender<PacketType<Payload>>) -> Self {
Self { channel: send, time: std::time::Instant::now() }
}
}
impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
type RecvData = Payload;
type SendData = Payload;
fn time(&self) -> i64 {
self.time.elapsed().as_millis() as i64
}
fn send(&self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::SendData) {
let _ = self.channel.send(PacketType::Data { seq_no, reply_no, payload: payload.clone() });
}
fn send_ack(&self, reply_no: SeqNo) {
let _ = self.channel.send(PacketType::Ack { reply_no });
}
fn send_empty_reply(&self, reply_no: SeqNo) {
let _ = self.channel.send(PacketType::EmptyReply { reply_no });
}
}
src/transport_layer.rs
+3 -3
View File
@@ -4,15 +4,15 @@ use crate::SeqNo;
///
/// The implementor is free to choose how to define the generic types based on how they want to
/// manage memory.
/// It is possible through these generics to implement SeqEx to be no-alloc and zero-copy, but otherwise
/// a lot of them are most easily implemented as tuples of custom enums and Vec<u8>.
/// It is possible through these generics to make SeqEx no-alloc and zero-copy, but otherwise
/// they are most easily implemented as some combination of custom enums, `Vec<u8>` and `Arc<[u8]>`.
pub trait TransportLayer: Sized + Clone {
type RecvData;
type SendData;
fn time(&self) -> i64;
fn send(&self, data: &Self::SendData);
fn send(&self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::SendData);
fn send_ack(&self, reply_no: SeqNo);
fn send_empty_reply(&self, reply_no: SeqNo);
}