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

Merge branch 'no-app' into main-fs

Browse Source
This commit is contained in:
Monica Moniot
2023-08-18 12:01:28 -04:00
parent 0c1ef344f3 88618287af
commit a2e76576a7
8 changed files with 734 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/calculator.rs
+23 -28
View File
@@ -1,9 +1,12 @@
use std::{sync::mpsc::Receiver, thread, time::Duration};
use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketType, RecvSuccess};
use seq_ex::{
sync::{MpscSeqEx, MpscTransport},
Packet,
};
#[derive(Clone)]
enum Packet {
enum Payload {
Add(f32),
Sub(f32),
Mul(f32),
@@ -16,28 +19,20 @@ fn drop_packet() -> bool {
rand_core::OsRng.next_u32() & 1 > 0
}
fn process(_: MpscGuard<'_, Packet>, recv_packet: Packet, _: Option<Packet>, value: &mut f32) {
use Packet::*;
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 receive(recv: &Receiver<PacketType<Packet>>, seq: &MpscSeqEx<Packet>, transport: &MpscTransport<Packet>, value: &mut f32) {
fn receive(recv: &Receiver<Packet<Payload>>, seq: &MpscSeqEx<Payload>, transport: &MpscTransport<Payload>, value: &mut f32) {
while let Ok(packet) = recv.try_recv() {
if !drop_packet() {
match packet {
PacketType::Ack(reply_no) => {
let _ = seq.receive_ack(reply_no);
}
PacketType::Payload(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);
}
if drop_packet() {
continue;
}
for recv_data in seq.receive_all(transport, packet) {
use Payload::*;
if let Some((_, recv_packet)) = recv_data.consume().0 {
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,
}
}
}
@@ -52,15 +47,15 @@ fn main() {
let mut value = 0.0;
let mut remote_value = value;
seq1.send(&transport1, Packet::Add(1.0));
seq1.send(&transport1, Payload::Add(1.0));
value += 1.0;
seq1.send(&transport1, Packet::Sub(2.0));
seq1.send(&transport1, Payload::Sub(2.0));
value -= 2.0;
seq1.send(&transport1, Packet::Mul(3.0));
seq1.send(&transport1, Payload::Mul(3.0));
value *= 3.0;
seq1.send(&transport1, Packet::Div(4.0));
seq1.send(&transport1, Payload::Div(4.0));
value /= 4.0;
seq1.send(&transport1, Packet::Mod(5.0));
seq1.send(&transport1, Payload::Mod(5.0));
value %= 5.0;
for _ in 0..16 {
examples/file_download.rs
+24 -40
View File
@@ -11,14 +11,14 @@ use std::{
use rand_core::{OsRng, RngCore};
use seq_ex::{
sync::{PacketType, RecvSuccess, ReplyGuard, SeqExSync},
SeqNo, TransportLayer,
sync::{RecvOk, SeqExSync},
Packet, TransportLayer,
};
use serde::{Deserialize, Serialize};
const FILE_CHUNK_SIZE: usize = 1000;
#[derive(Clone, Debug, Serialize, Deserialize)]
enum Packet {
enum Payload {
RequestFile { filename: String },
ConfirmRequestFile { filesize: u64 },
FileDownload { filename: String, file_chunk: Vec<u8> },
@@ -32,25 +32,17 @@ struct Transport {
struct Peer {
filesystem: Arc<RwLock<HashMap<String, Vec<u8>>>>,
transport: Transport,
seqex: Arc<SeqExSync<Packet, Packet>>,
seqex: Arc<SeqExSync<Payload, Payload>>,
receiver: Receiver<Vec<u8>>,
}
impl TransportLayer<Packet> for &Transport {
impl TransportLayer<Payload> for &Transport {
fn time(&mut self) -> i64 {
self.time.elapsed().as_millis() as i64
}
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Packet) {
let p = PacketType::Payload(seq_no, reply_no, payload.clone());
if let Ok(p) = serde_json::to_vec(&p) {
let _ = self.sender.send(p);
}
}
fn send_ack(&mut self, reply_no: SeqNo) {
let p = PacketType::<Packet>::Ack(reply_no);
if let Ok(p) = serde_json::to_vec(&p) {
fn send(&mut self, packet: Packet<&Payload>) {
if let Ok(p) = serde_json::to_vec(&packet) {
let _ = self.sender.send(p);
}
}
@@ -60,14 +52,15 @@ fn drop_packet() -> bool {
OsRng.next_u32() >= (u32::MAX / 4 * 3)
}
fn process(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_packet: Packet, sent_packet: Option<Packet>) {
match (recv_packet, sent_packet) {
(Packet::RequestFile { filename }, None) => {
fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payload>) {
use Payload::*;
match recv_data.consume() {
(Some((guard, RequestFile { filename })), None) => {
let filesystem = peer.filesystem.clone();
let transport = peer.transport.clone();
let seqex = peer.seqex.clone();
if let Some(file) = filesystem.read().unwrap().get(&filename) {
guard.reply(Packet::ConfirmRequestFile { filesize: file.len() as u64 });
guard.reply(ConfirmRequestFile { filesize: file.len() as u64 });
}
thread::spawn(move || {
let filesystem = filesystem.read().unwrap();
@@ -77,21 +70,18 @@ fn process(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_
let mut i = 0;
while i < file.len() {
let j = file.len().min(i + FILE_CHUNK_SIZE);
seqex.send(
&transport,
Packet::FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() },
);
seqex.send_locked(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() });
i = j;
}
}
});
}
(Packet::ConfirmRequestFile { filesize }, Some(Packet::RequestFile { filename })) => {
(Some((_, ConfirmRequestFile { filesize })), Some(RequestFile { filename })) => {
let mut filesystem = peer.filesystem.write().unwrap();
let file = Vec::with_capacity(filesize as usize);
filesystem.insert(filename, file);
}
(Packet::FileDownload { filename, file_chunk }, None) => {
(Some((_, FileDownload { filename, file_chunk })), None) => {
let mut filesystem = peer.filesystem.write().unwrap();
if let Some(file) = filesystem.get_mut(&filename) {
if file.len() + file_chunk.len() <= file.capacity() {
@@ -99,9 +89,10 @@ fn process(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_
}
}
}
_ => {
assert!(false);
(Some(a), b) => {
print!("Unsolicited packet received: {:?}", RecvOk::new(Some(a), b));
}
_ => {}
}
}
@@ -110,17 +101,10 @@ fn receive(peer: &Peer) {
if drop_packet() {
continue;
}
let parsed_packet = serde_json::from_slice::<PacketType<Packet>>(&packet);
match parsed_packet {
Ok(PacketType::Ack(reply_no)) => {
let _ = peer.seqex.receive_ack(reply_no);
if let Ok(parsed_packet) = serde_json::from_slice::<Packet<Payload>>(&packet) {
for recv_data in peer.seqex.receive_all(&peer.transport, parsed_packet) {
process(peer, recv_data);
}
Ok(PacketType::Payload(seq_no, reply_no, payload)) => {
for RecvSuccess { guard, packet, send_data } in peer.seqex.receive_all(&peer.transport, seq_no, reply_no, payload) {
process(peer, guard, packet, send_data);
}
}
_ => {}
}
}
}
@@ -153,9 +137,9 @@ fn main() {
receiver: recv2,
};
peer1.seqex.send(&peer1.transport, Packet::RequestFile { filename: "File1".to_string() });
peer1.seqex.send(&peer1.transport, Packet::RequestFile { filename: "File3".to_string() });
peer1.seqex.send(&peer1.transport, Packet::RequestFile { filename: "File2".to_string() });
peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File1".to_string() });
peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File3".to_string() });
peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File2".to_string() });
for _ in 0..300 {
receive(&peer1);
examples/hello_world.rs
+30 -39
View File
@@ -1,55 +1,46 @@
use std::sync::mpsc::Receiver;
use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketType, RecvSuccess};
use seq_ex::{
sync::{MpscSeqEx, MpscTransport, RecvOk},
Packet,
};
#[derive(Clone, Debug)]
enum Packet {
enum Payload {
Hello,
Space,
World,
Exclamation,
}
use Packet::*;
use Payload::*;
fn process(guard: MpscGuard<'_, Packet>, recv_packet: Packet, send_packet: Option<Packet>) {
match (recv_packet, send_packet) {
(Hello, None) => {
print!("Hello");
guard.reply(Space);
}
(Space, Some(Hello)) => {
print!(" ");
guard.reply(World);
}
(World, Some(Space)) => {
print!("World");
guard.reply(Exclamation);
}
(Exclamation, Some(World)) => {
print!("!");
}
(a, None) => {
print!("Unsolicited packet received: {:?}", a);
}
(a, Some(b)) => {
print!("Incorrect reply received: {:?}, was a reply to: {:?}", a, b);
}
}
}
fn receive(recv: &Receiver<PacketType<Packet>>, seq: &MpscSeqEx<Packet>, transport: &MpscTransport<Packet>) {
match recv.recv().unwrap() {
PacketType::Ack(reply_no) => {
let result = seq.receive_ack(reply_no);
if let Ok(Exclamation) = result {
fn receive(recv: &Receiver<Packet<Payload>>, seq: &MpscSeqEx<Payload>, transport: &MpscTransport<Payload>) {
let packet = recv.recv().unwrap();
for recv_data in seq.receive_all(transport, packet) {
match recv_data.consume() {
(Some((guard, Hello)), None) => {
print!("Hello");
guard.reply(Space);
}
(Some((guard, Space)), Some(Hello)) => {
print!(" ");
guard.reply(World);
}
(Some((guard, World)), Some(Space)) => {
print!("World");
guard.reply(Exclamation);
}
(Some((_, Exclamation)), Some(World)) => {
print!("!");
}
(None, Some(Exclamation)) => {
// Our Hello World exchange ends right here.
print!("\n");
}
}
PacketType::Payload(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)
(Some(a), b) => {
print!("Unsolicited packet received: {:?}", RecvOk::new(Some(a), b));
}
_ => {}
}
}
}
@@ -61,7 +52,7 @@ fn main() {
let seq2 = MpscSeqEx::default();
// We begin a "Hello World" exchange right here.
seq1.send(&transport1, Packet::Hello);
seq1.send(&transport1, Payload::Hello);
receive(&recv2, &seq2, &transport2);
receive(&recv1, &seq1, &transport1);
src/seq_queue.rs
+293 -124
View File
File diff suppressed because it is too large Load Diff
src/single_thread.rs
+191 -21
View File
@@ -1,10 +1,11 @@
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
use crate::{DirectError, Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(
&'a mut SeqEx<SendData, RecvData, CAP>,
TL,
SeqNo,
);
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a mut SeqEx<SendData, RecvData, CAP>,
app: Option<TL>,
reply_no: SeqNo,
locked: bool,
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
/// 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,35 +13,204 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
/// and since each fragment will be received in order it will be trivial for them to reconstruct
/// the original file.
pub fn reply(self, packet_data: SendData) {
self.0.reply_raw(self.1.clone(), self.2, packet_data);
self.reply_inner(false, |_, _| packet_data)
}
pub fn reply_locked(self, packet_data: SendData) {
self.reply_inner(true, |_, _| packet_data)
}
pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(false, packet_data)
}
pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(true, packet_data)
}
fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let mut app = None;
core::mem::swap(&mut app, &mut self.app);
let seq_no = self.seq.seq_no();
self.seq
.reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no));
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn drop(&mut self) {
self.0.ack_raw(self.1.clone(), self.2)
if let Some(app) = &mut self.app {
if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no, self.locked) {
app.send(p)
}
}
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ReplyGuard").field("reply_no", &self.reply_no).finish()
}
}
pub struct RecvSuccess<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
pub packet: P,
pub send_data: Option<SendData>,
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Error<RecvData> {
/// The packet is out-of-sequence. It was either received too soon or too late and so it would be
/// invalid to process it right now. No action needs to be taken by the caller.
OutOfSequence,
/// The Send Window is currently full. The received packet cannot be processed right now because
/// it could cause the send window to overflow.
WindowIsFull(Packet<RecvData>),
WindowIsLocked(Packet<RecvData>),
}
pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
Payload {
reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
recv_data: P,
},
Reply {
reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
recv_data: P,
send_data: SendData,
},
Ack {
send_data: SendData,
},
}
macro_rules! impl_recvok {
($recv:tt, $seq_ex:ty) => {
#[cfg(feature = "std")]
impl<'a, TL: TransportLayer<SendData>, P: std::fmt::Debug, SendData: std::fmt::Debug, RecvData, const CAP: usize> std::fmt::Debug
for $recv<'a, TL, P, SendData, RecvData, CAP>
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Payload { reply_guard, recv_data } => f
.debug_struct("Payload")
.field("reply_guard", reply_guard)
.field("recv_data", recv_data)
.finish(),
Self::Reply { reply_guard, recv_data, send_data } => f
.debug_struct("Reply")
.field("reply_guard", reply_guard)
.field("recv_data", recv_data)
.field("send_data", send_data)
.finish(),
Self::Ack { send_data } => f.debug_struct("Ack").field("send_data", send_data).finish(),
}
}
}
impl<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> {
pub fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw<SendData, P>) -> Self {
match value {
RecvOkRaw::Payload { reply_no, locked, recv_data } => Self::Payload {
reply_guard: ReplyGuard { seq, app: Some(app), reply_no, locked },
recv_data,
},
RecvOkRaw::Reply { reply_no, locked, recv_data, send_data } => Self::Reply {
reply_guard: ReplyGuard { seq, app: Some(app), reply_no, locked },
recv_data,
send_data,
},
RecvOkRaw::Ack { send_data } => Self::Ack { send_data },
}
}
pub fn consume(self) -> (Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, Option<SendData>) {
match self {
Self::Payload { reply_guard, recv_data } => (Some((reply_guard, recv_data)), None),
Self::Reply { reply_guard, recv_data, send_data } => (Some((reply_guard, recv_data)), Some(send_data)),
Self::Ack { send_data } => (None, Some(send_data)),
}
}
pub fn new(recv_data: Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, send_data: Option<SendData>) -> Option<Self> {
match (recv_data, send_data) {
(Some((reply_guard, recv_data)), None) => Some(Self::Payload { reply_guard, recv_data }),
(Some((reply_guard, recv_data)), Some(send_data)) => Some(Self::Reply { reply_guard, recv_data, send_data }),
(None, Some(send_data)) => Some(Self::Ack { send_data }),
(None, None) => None,
}
}
}
impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> {
pub fn into(self) -> $recv<'a, TL, RecvData, SendData, RecvData, CAP> {
match self {
Self::Payload { reply_guard, recv_data } => $recv::Payload { reply_guard, recv_data: recv_data.into() },
Self::Reply { reply_guard, recv_data, send_data } => $recv::Reply { reply_guard, recv_data: recv_data.into(), send_data },
Self::Ack { send_data } => $recv::Ack { send_data },
}
}
}
};
}
impl_recvok!(RecvOk, &'a mut SeqEx<SendData, RecvData, CAP>);
pub(crate) use impl_recvok;
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, locked: bool, packet_data: SendData) -> Result<(), SendData> {
match self.try_send_direct(packet_data, app.time()) {
Ok(mut p) => {
p.set_locking(locked);
app.send(p);
Ok(())
}
Err(e) => Err(e),
}
}
pub fn try_send_with(
&mut self,
mut app: impl TransportLayer<SendData>,
locked: bool,
packet_data: impl FnOnce(SeqNo) -> SendData,
) -> Result<(), ()> {
match self.try_send_direct_with(packet_data, app.time()) {
Ok(mut p) => {
p.set_locking(locked);
app.send(p);
Ok(())
}
Err(e) => Err(e),
}
}
/// If this returns `Ok` then `try_send` might succeed on next call.
pub fn receive_raw<P: Into<RecvData>>(
&mut self,
mut app: impl TransportLayer<SendData>,
packet: Packet<P>,
) -> Result<crate::seq_queue::RecvOkRaw<SendData, P>, Error<P>> {
match self.receive_raw_and_direct(packet) {
Ok(a) => Ok(a),
Err(DirectError::ResendAck(reply_no)) => {
app.send(Packet::Ack(reply_no));
Err(Error::OutOfSequence)
}
Err(DirectError::OutOfSequence) => Err(Error::OutOfSequence),
Err(DirectError::WindowIsFull(p)) => Err(Error::WindowIsFull(p)),
Err(DirectError::WindowIsLocked(p)) => Err(Error::WindowIsLocked(p)),
}
}
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool, locked_packet: bool, packet_data: SendData) {
if let Some(mut p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) {
p.set_locking(locked_packet);
app.send(p)
}
}
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool) {
if let Some(p) = self.ack_raw_and_direct(reply_no, unlock) {
app.send(p)
}
}
pub fn service(&mut self, mut app: impl TransportLayer<SendData>) -> i64 {
let current_time = app.time();
let mut iter = None;
while let Some(p) = self.service_direct(current_time, &mut iter) {
app.send(p)
}
self.resend_interval.min(self.next_service_timestamp - current_time)
}
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&mut self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, 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 })
packet: Packet<P>,
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, Error<P>> {
self.receive_raw(app.clone(), packet).map(|r| RecvOk::from_raw(self, app, r))
}
pub fn pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
self.pump_raw()
.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
pub fn pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
self.pump_raw().map(|r| RecvOk::from_raw(self, app, r))
}
}
src/sync.rs
+148 -94
View File
@@ -7,17 +7,19 @@ use std::{
time::Instant,
};
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
use crate::{Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
pub struct SeqExSync<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq_ex: Mutex<(SeqEx<SendData, RecvData, CAP>, usize)>,
send_block: Condvar,
recv_lock: Condvar,
}
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: &'a SeqExSync<SendData, RecvData, CAP>,
app: TL,
seq: &'a SeqExSync<SendData, RecvData, CAP>,
app: Option<TL>,
reply_no: SeqNo,
locked: bool,
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
/// If you need to reply more than once, say to fragment a large file, then include in your
@@ -26,34 +28,76 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
/// and since each fragment will be received in order it will be trivial for them to reconstruct
/// the original file.
pub fn reply(self, packet_data: SendData) {
let mut seq = self.origin.lock();
seq.reply_raw(self.app.clone(), self.reply_no, packet_data);
core::mem::forget(self);
self.reply_inner(false, |_, _| packet_data)
}
pub fn reply_locked(self, packet_data: SendData) {
self.reply_inner(true, |_, _| packet_data)
}
pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let mut seq = self.origin.lock();
self.reply_inner(false, packet_data)
}
pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(true, packet_data)
}
fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let mut app = None;
core::mem::swap(&mut app, &mut self.app);
let mut seq = self.seq.lock();
let seq_no = seq.seq_no();
seq.reply_raw(self.app.clone(), self.reply_no, packet_data(seq_no, self.reply_no));
seq.reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no));
drop(seq);
if self.locked {
self.seq.recv_lock.notify_all();
}
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn drop(&mut self) {
let mut seq = self.origin.lock();
seq.ack_raw(self.app.clone(), self.reply_no);
if let Some(app) = self.app.as_mut() {
let mut seq = self.seq.lock();
if let Some(p) = seq.ack_raw_and_direct(self.reply_no, self.locked) {
app.send(p)
}
}
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ReplyGuard").field("reply_no", &self.reply_no).finish()
}
}
pub struct RecvSuccess<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
pub packet: P,
pub send_data: Option<SendData>,
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Error {
/// The packet is out-of-sequence. It was either received too soon or too late and so it would be
/// invalid to process it right now. No action needs to be taken by the caller.
OutOfSequence,
/// The Send Window is currently full. The received packet cannot be processed right now because
/// it could cause the send window to overflow. No action needs to be taken by the caller.
WindowIsFull,
}
pub struct ReplyIter<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
Payload {
reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
recv_data: P,
},
Reply {
reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
recv_data: P,
send_data: SendData,
},
Ack {
send_data: SendData,
},
}
crate::impl_recvok!(RecvOk, &'a SeqExSync<SendData, RecvData, CAP>);
pub struct ReplyIter<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
app: TL,
first: Option<RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>>,
first: Option<RecvOk<'a, TL, P, SendData, RecvData, CAP>>,
}
pub struct SeqExGuard<'a, SendData, RecvData, const CAP: usize>(MutexGuard<'a, (SeqEx<SendData, RecvData, CAP>, usize)>);
@@ -75,90 +119,103 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
Self {
seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)),
send_block: Condvar::default(),
recv_lock: Condvar::default(),
}
}
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, CAP>, Error> {
mut packet: Packet<P>,
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, Error> {
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.receive_raw(app.clone(), seq_no, reply_no, packet);
if seq.1 > 0 && ret.is_ok() {
self.send_block.notify_one();
loop {
match seq.0.receive_raw(app.clone(), packet) {
Ok(r) => {
if seq.1 > 0 {
self.send_block.notify_one();
}
return Ok(RecvOk::from_raw(self, app, r));
}
Err(crate::Error::OutOfSequence) => return Err(Error::OutOfSequence),
Err(crate::Error::WindowIsFull(_)) => return Err(Error::WindowIsFull),
Err(crate::Error::WindowIsLocked(p)) => {
seq = self.recv_lock.wait(seq).unwrap();
packet = p;
}
}
}
ret.map(|(reply_no, packet, send_data)| RecvSuccess {
guard: ReplyGuard { origin: self, app, reply_no },
packet,
send_data,
})
}
pub fn pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
pub fn pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.pump_raw();
if seq.1 > 0 && ret.is_ok() {
self.send_block.notify_one();
loop {
match seq.0.pump_raw() {
Ok(r) => {
if seq.1 > 0 {
self.send_block.notify_one();
}
return Ok(RecvOk::from_raw(self, app, r));
}
Err(PumpError::OutOfSequence) => return Err(Error::OutOfSequence),
Err(PumpError::WindowIsFull) => return Err(Error::WindowIsFull),
Err(PumpError::WindowIsLocked) => {
seq = self.recv_lock.wait(seq).unwrap();
}
}
}
ret.map(|(reply_no, packet, send_data)| RecvSuccess {
guard: ReplyGuard { origin: self, app, reply_no },
packet,
send_data,
})
}
pub fn receive_all<TL: TransportLayer<SendData>>(
pub fn receive_all<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: RecvData,
) -> ReplyIter<'_, TL, SendData, RecvData, CAP> {
if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) {
ReplyIter { origin: Some(self), app, first: Some(g) }
packet: Packet<P>,
) -> ReplyIter<'_, TL, P, SendData, RecvData, CAP> {
if let Ok(r) = self.receive(app.clone(), packet) {
ReplyIter { seq: Some(self), app, first: Some(r) }
} else {
ReplyIter { origin: None, app, first: None }
ReplyIter { seq: None, app, first: None }
}
}
pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> {
let mut seq = self.lock();
seq.try_send(app, packet_data)
pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, packet_data: SendData) -> Result<(), SendData> {
self.try_send_with(app, locked, |_| packet_data)
}
fn send_inner<TL: TransportLayer<SendData>>(
&self,
mut seq: MutexGuard<'_, (SeqEx<SendData, RecvData, CAP>, usize)>,
app: TL,
mut packet_data: SendData,
) {
while let Err(p) = seq.0.try_send(app.clone(), packet_data) {
fn send_with_inner<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, mut packet_data: impl FnMut(SeqNo) -> SendData) {
let mut seq = self.seq_ex.lock().unwrap();
while let Err(()) = seq.0.try_send_with(app.clone(), locked, &mut packet_data) {
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
seq.1 -= 1;
}
}
fn send_inner<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, mut packet_data: SendData) {
let mut seq = self.seq_ex.lock().unwrap();
while let Err(p) = seq.0.try_send(app.clone(), locked, packet_data) {
packet_data = p;
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
seq.1 -= 1;
}
}
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: SendData) {
self.send_inner(self.seq_ex.lock().unwrap(), app, packet_data)
pub fn send(&self, app: impl TransportLayer<SendData>, packet_data: SendData) {
self.send_inner(app, false, packet_data)
}
pub fn try_send_with<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> {
pub fn send_locked(&self, app: impl TransportLayer<SendData>, packet_data: SendData) {
self.send_inner(app, true, packet_data)
}
pub fn send_with(&self, app: impl TransportLayer<SendData>, packet_data: impl FnMut(SeqNo) -> SendData) {
self.send_with_inner(app, false, packet_data)
}
pub fn send_locked_with(&self, app: impl TransportLayer<SendData>, packet_data: impl FnMut(SeqNo) -> SendData) {
self.send_with_inner(app, true, packet_data)
}
pub fn try_send_with<TL: TransportLayer<SendData>>(
&self,
app: TL,
locked: bool,
packet_data: impl FnOnce(SeqNo) -> SendData,
) -> Result<(), SendData> {
let mut seq = self.lock();
let seq_no = seq.seq_no();
seq.try_send(app, packet_data(seq_no))
}
pub fn send_with<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) {
let seq = self.seq_ex.lock().unwrap();
let seq_no = seq.0.seq_no();
self.send_inner(seq, app, packet_data(seq_no))
}
pub fn receive_ack(&self, reply_no: SeqNo) -> Result<SendData, Error> {
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.receive_ack(reply_no);
if seq.1 > 0 && ret.is_ok() {
self.send_block.notify_one();
}
ret
seq.try_send(app, locked, packet_data(seq_no))
}
pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
self.lock().service(app)
@@ -174,12 +231,19 @@ impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvD
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> {
type Item = RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>;
impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> ReplyIter<'a, TL, P, SendData, RecvData, CAP> {
pub fn take_first(&mut self) -> Option<RecvOk<'a, TL, P, SendData, RecvData, CAP>> {
self.first.take()
}
}
impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> Iterator
for ReplyIter<'a, TL, P, SendData, RecvData, CAP>
{
type Item = RecvOk<'a, TL, RecvData, SendData, RecvData, CAP>;
fn next(&mut self) -> Option<Self::Item> {
if let Some(g) = self.first.take() {
Some(g)
} else if let Some(origin) = self.origin {
Some(g.into())
} else if let Some(origin) = self.seq {
origin.pump(self.app.clone()).ok()
} else {
None
@@ -187,27 +251,20 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Ite
}
}
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[derive(Clone)]
pub enum PacketType<Payload: Clone> {
Payload(SeqNo, Option<SeqNo>, Payload),
Ack(SeqNo),
}
#[derive(Clone)]
#[derive(Clone, Debug)]
pub struct MpscTransport<Payload: Clone> {
pub channel: Sender<PacketType<Payload>>,
pub channel: Sender<Packet<Payload>>,
pub time: Instant,
}
pub type MpscGuard<'a, Packet> = ReplyGuard<'a, &'a MpscTransport<Packet>, Packet, Packet>;
pub type MpscSeqEx<Packet> = SeqExSync<Packet, Packet>;
impl<Payload: Clone> MpscTransport<Payload> {
pub fn new() -> (Self, Receiver<PacketType<Payload>>) {
pub fn new() -> (Self, Receiver<Packet<Payload>>) {
let (send, recv) = channel();
(Self { channel: send, time: std::time::Instant::now() }, recv)
}
pub fn from_sender(send: Sender<PacketType<Payload>>) -> Self {
pub fn from_sender(send: Sender<Packet<Payload>>) -> Self {
Self { channel: send, time: std::time::Instant::now() }
}
}
@@ -216,10 +273,7 @@ impl<Payload: Clone> TransportLayer<Payload> for &MpscTransport<Payload> {
self.time.elapsed().as_millis() as i64
}
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Payload) {
let _ = self.channel.send(PacketType::Payload(seq_no, reply_no, payload.clone()));
}
fn send_ack(&mut self, reply_no: SeqNo) {
let _ = self.channel.send(PacketType::Ack(reply_no));
fn send(&mut self, packet: Packet<&Payload>) {
let _ = self.channel.send(packet.cloned());
}
}
src/tokio.rs
+23 -17
View File
@@ -1,9 +1,8 @@
use std::
sync::{
Mutex, MutexGuard,
}
;
use tokio::{task, sync::{Notify, oneshot, mpsc}, time};
use std::sync::{Mutex, MutexGuard};
use tokio::{
sync::{mpsc, oneshot, Notify},
task, time,
};
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
@@ -82,11 +81,7 @@ pub struct TokioTransport<TL: TokioTransportLayer> {
impl<TL: TokioTransportLayer> TokioTransport<TL> {
pub fn new<const CAP: usize, S: AsRef<SeqExTokio<TL::Packet, CAP>> + Send + 'static>(app: TL, seq: S) -> Self {
let (update_queue, mut recv) = mpsc::channel(4);
let ret = TokioTransport {
time: time::Instant::now(),
update_queue,
app,
};
let ret = TokioTransport { time: time::Instant::now(), update_queue, app };
let task_tl = ret.clone();
task::spawn(async move {
let mut update_ts = i64::MAX;
@@ -96,7 +91,7 @@ impl<TL: TokioTransportLayer> TokioTransport<TL> {
let mut do_update = diff <= 0;
if diff > 0 {
let sleep = time::sleep(time::Duration::from_millis(diff as u64));
tokio::select!{
tokio::select! {
Some(up) = recv.recv() => {
update_ts = up;
}
@@ -131,7 +126,7 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
&'a self,
app: &'a TokioTransport<TL>,
mut seq: MutexGuard<'_, (SeqEx<SendData<Packet>, Packet, CAP>, usize)>,
result: Result<(SeqNo, Packet, Option<SendData<Packet>>), Error>
result: Result<(SeqNo, Packet, Option<SendData<Packet>>), Error>,
) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
if let Ok((reply_no, packet, send_data)) = result {
if seq.1 > 0 {
@@ -161,7 +156,10 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
let result = seq.0.receive_raw(app, seq_no, reply_no, packet);
self.process(app, seq, result)
}
pub fn pump<'a, TL: TokioTransportLayer<Packet = Packet>>(&'a self, app: &'a TokioTransport<TL>) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
pub fn pump<'a, TL: TokioTransportLayer<Packet = Packet>>(
&'a self,
app: &'a TokioTransport<TL>,
) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
let mut seq = self.seq_ex.lock().unwrap();
let result = seq.0.pump_raw();
self.process(app, seq, result)
@@ -197,7 +195,7 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
mut seq: MutexGuard<'_, (SeqEx<SendData<Packet>, Packet, CAP>, usize)>,
app: &TokioTransport<TL>,
mut tx: oneshot::Sender<(Packet, SeqNo)>,
mut packet: Packet
mut packet: Packet,
) {
let mut pre_ts = seq.0.next_service_timestamp;
while let Err(e) = seq.0.try_send(app, (tx, packet)) {
@@ -214,7 +212,11 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
}
}
/// If this future is dropped then the remote peer's reply to this packet will also be dropped.
pub async fn send<'a, TL: TokioTransportLayer<Packet = Packet>>(&'a self, app: &'a TokioTransport<TL>, packet: Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
pub async fn send<'a, TL: TokioTransportLayer<Packet = Packet>>(
&'a self,
app: &'a TokioTransport<TL>,
packet: Packet,
) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
self.send_with(app, |_| packet).await
}
//pub fn try_send_with<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> {
@@ -222,7 +224,11 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
// let seq_no = seq.seq_no();
// seq.try_send(app, packet_data(seq_no))
//}
pub async fn send_with<'a, TL: TokioTransportLayer<Packet = Packet>>(&'a self, app: &'a TokioTransport<TL>, packet: impl FnOnce(SeqNo) -> Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
pub async fn send_with<'a, TL: TokioTransportLayer<Packet = Packet>>(
&'a self,
app: &'a TokioTransport<TL>,
packet: impl FnOnce(SeqNo) -> Packet,
) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
let (tx, rx) = oneshot::channel();
let seq = self.seq_ex.lock().unwrap();
let seq_no = seq.0.seq_no();
src/transport_layer.rs
+2 -3
View File
@@ -1,4 +1,4 @@
use crate::SeqNo;
use crate::Packet;
/// A trait for giving an instance of SeqEx access to the transport layer.
///
@@ -9,6 +9,5 @@ use crate::SeqNo;
pub trait TransportLayer<SendData>: Clone {
fn time(&mut self) -> i64;
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &SendData);
fn send_ack(&mut self, reply_no: SeqNo);
fn send(&mut self, packet: Packet<&SendData>);
}