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

reformat

Browse Source
This commit is contained in:
Monica Moniot
2023-08-16 15:44:42 -04:00
parent fcef733b88
commit 281edbc4e2
6 changed files with 51 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+24 -15
View File
@@ -3,21 +3,21 @@
version = 3
[[package]]
name = "futex"
version = "0.1.3"
name = "cfg-if"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d370e15a8972dee506ad50638e331183109e85ff99f349f19e04f288dc6cef3"
dependencies = [
"integer-atomics",
"libc",
"lock-wrappers",
]
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "integer-atomics"
version = "1.0.2"
name = "getrandom"
version = "0.2.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5c33cd4d18b4ade167caace0e92364e8568c1e47c193738397b4b48a3e414139"
checksum = "be4136b2a15dd319360be1c07d9933517ccf0be8f16bf62a3bee4f0d618df427"
dependencies = [
"cfg-if",
"libc",
"wasi",
]
[[package]]
name = "libc"
@@ -26,14 +26,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4668fb0ea861c1df094127ac5f1da3409a82116a4ba74fca2e58ef927159bb3"
[[package]]
name = "lock-wrappers"
version = "0.1.2"
name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d654f44a90e266c873afdcf93f4506d9d6fb036b211e2be63a09695ecb0a07a"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom",
]
[[package]]
name = "seq_ex"
version = "0.1.0"
dependencies = [
"futex",
"rand_core",
]
[[package]]
name = "wasi"
version = "0.11.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
Cargo.toml
+3 -3
View File
@@ -11,7 +11,7 @@ doc = true
[features]
default = ["std"]
std = ["futex"]
std = []
[dependencies]
futex = {version = "0.1.3", optional = true}
[dev-dependencies]
rand_core = { version = "0.6.4", features = ["getrandom"]}
examples/calculator.rs
+4 -12
View File
@@ -1,8 +1,4 @@
use std::{
sync::{mpsc::Receiver, Mutex},
thread,
time::Duration,
};
use std::{sync::mpsc::Receiver, thread, time::Duration};
use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync};
@@ -16,12 +12,8 @@ enum Packet {
}
fn drop_packet() -> bool {
static RNG: Mutex<u32> = Mutex::new(43);
let mut rng = RNG.lock().unwrap();
*rng ^= *rng << 13;
*rng ^= *rng >> 17;
*rng ^= *rng << 5;
*rng & 1 == 0
use rand_core::RngCore;
rand_core::OsRng.next_u32() & 1 > 0
}
fn process(_: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, _: Option<Packet>, value: &mut f32) {
@@ -48,7 +40,7 @@ fn receive<'a>(
let _ = seq.receive_ack(reply_no);
}
PacketType::Payload { seq_no, reply_no, payload } => {
for RecvSuccess { guard, packet, send_data } in seq.receive_iter(transport, seq_no, reply_no, payload) {
for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) {
process(guard, packet, send_data, value);
}
}
examples/hello_world.rs
+1 -1
View File
@@ -47,7 +47,7 @@ fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTran
}
}
PacketType::Payload { seq_no, reply_no, payload } => {
for RecvSuccess {guard, packet, send_data} in seq.receive_iter(transport, seq_no, reply_no, payload) {
for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) {
process(guard, packet, send_data)
}
}
src/single_thread.rs
+8 -8
View File
@@ -1,6 +1,6 @@
use crate::{Error, SeqEx, SeqNo, TransportLayer};
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
pub struct ReplyGuard<'a, TL: TransportLayer>(&'a mut SeqEx<TL>, TL, SeqNo);
pub struct ReplyGuard<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a mut SeqEx<TL, CAP>, TL, SeqNo);
impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
/// If you need to reply more than once, say to fragment a large file, then include in your
/// first reply some identifier, and then `send` all fragments with the same included identifier.
@@ -12,30 +12,30 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
impl<'a, TL: TransportLayer, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> {
fn drop(&mut self) {
self.0.ack_raw(self.1.clone(), self.2)
}
}
pub struct RecvSuccess<'a, TL: TransportLayer, P> {
pub guard: ReplyGuard<'a, TL>,
pub struct RecvSuccess<'a, TL: TransportLayer, P, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub guard: ReplyGuard<'a, TL, CAP>,
pub packet: P,
pub send_data: Option<TL::SendData>,
}
impl<TL: TransportLayer> SeqEx<TL> {
impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
pub fn receive<P: Into<TL::RecvData>>(
&mut self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P>, Error> {
) -> Result<RecvSuccess<'_, TL, P, CAP>, Error> {
self.receive_raw(app.clone(), seq_no, reply_no, packet)
.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn pump(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData>, Error> {
pub fn pump(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData, CAP>, Error> {
self.pump_raw()
.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
src/sync.rs
+11 -19
View File
@@ -8,9 +8,7 @@ use std::{
time::Instant,
};
use crate::{
Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_WINDOW_CAP, DEFAULT_RESEND_INTERVAL_MS,
};
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
pub struct SeqExSync<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq_ex: Mutex<SeqEx<TL, CAP>>,
@@ -46,6 +44,12 @@ pub struct RecvSuccess<'a, TL: TransportLayer, P: Into<TL::RecvData>, const CAP:
pub send_data: Option<TL::SendData>,
}
pub struct ReplyIter<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExSync<TL, CAP>>,
app: TL,
first: Option<RecvSuccess<'a, TL, TL::RecvData, CAP>>,
}
impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
Self {
@@ -74,13 +78,7 @@ impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
self.unblock(ret.is_ok());
ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn receive_iter(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: TL::RecvData,
) -> ReplyIter<'_, TL, CAP> {
pub fn receive_all(&self, app: TL, seq_no: SeqNo, reply_no: Option<SeqNo>, packet: TL::RecvData) -> ReplyIter<'_, TL, CAP> {
if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) {
ReplyIter { origin: Some(self), app, first: Some(g) }
} else {
@@ -121,20 +119,14 @@ impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
self.seq_ex.lock().unwrap()
}
}
impl<TL: TransportLayer> Default for SeqExSync<TL> {
impl<TL: TransportLayer, const CAP: usize> Default for SeqExSync<TL, CAP> {
fn default() -> Self {
Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
}
}
pub struct ReplyIter<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExSync<TL, CAP>>,
app: TL,
first: Option<RecvSuccess<'a, TL, TL::RecvData, CAP>>
}
impl<'a, TL: TransportLayer> Iterator for ReplyIter<'a, TL> {
type Item = RecvSuccess<'a, TL, TL::RecvData>;
impl<'a, TL: TransportLayer, const CAP: usize> Iterator for ReplyIter<'a, TL, CAP> {
type Item = RecvSuccess<'a, TL, TL::RecvData, CAP>;
fn next(&mut self) -> Option<Self::Item> {
if let Some(g) = self.first.take() {
Some(g)