mirror of
https://github.com/zerotier/sequential-exchange.git
synced 2026-05-22 16:28:28 -07:00
finished tokio support
This commit is contained in:
Monica Moniot
Generated
+296
@@ -2,6 +2,63 @@
|
||||
# It is not intended for manual editing.
|
||||
version = 3
|
||||
|
||||
[[package]]
|
||||
name = "addr2line"
|
||||
version = "0.20.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f4fa78e18c64fce05e902adecd7a5eed15a5e0a3439f7b0e169f0252214865e3"
|
||||
dependencies = [
|
||||
"gimli",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "adler"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
|
||||
|
||||
[[package]]
|
||||
name = "autocfg"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
|
||||
|
||||
[[package]]
|
||||
name = "backtrace"
|
||||
version = "0.3.68"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4319208da049c43661739c5fade2ba182f09d1dc2299b32298d3a31692b17e12"
|
||||
dependencies = [
|
||||
"addr2line",
|
||||
"cc",
|
||||
"cfg-if",
|
||||
"libc",
|
||||
"miniz_oxide",
|
||||
"object",
|
||||
"rustc-demangle",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bitflags"
|
||||
version = "1.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
|
||||
|
||||
[[package]]
|
||||
name = "bytes"
|
||||
version = "1.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "89b2fd2a0dcf38d7971e2194b6b6eebab45ae01067456a7fd93d5547a61b70be"
|
||||
|
||||
[[package]]
|
||||
name = "cc"
|
||||
version = "1.0.82"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "305fe645edc1442a0fa8b6726ba61d422798d37a52e12eaecf4b022ebbb88f01"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cfg-if"
|
||||
version = "1.0.0"
|
||||
@@ -19,18 +76,114 @@ dependencies = [
|
||||
"wasi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gimli"
|
||||
version = "0.27.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b6c80984affa11d98d1b88b66ac8853f143217b399d3c74116778ff8fdb4ed2e"
|
||||
|
||||
[[package]]
|
||||
name = "half"
|
||||
version = "1.8.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "eabb4a44450da02c90444cf74558da904edde8fb4e9035a9a6a4e15445af0bd7"
|
||||
|
||||
[[package]]
|
||||
name = "hermit-abi"
|
||||
version = "0.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "443144c8cdadd93ebf52ddb4056d257f5b52c04d3c804e657d19eb73fc33668b"
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
version = "0.2.147"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b4668fb0ea861c1df094127ac5f1da3409a82116a4ba74fca2e58ef927159bb3"
|
||||
|
||||
[[package]]
|
||||
name = "lock_api"
|
||||
version = "0.4.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c1cc9717a20b1bb222f333e6a92fd32f7d8a18ddc5a3191a11af45dcbf4dcd16"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"scopeguard",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "memchr"
|
||||
version = "2.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
|
||||
|
||||
[[package]]
|
||||
name = "miniz_oxide"
|
||||
version = "0.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e7810e0be55b428ada41041c41f32c9f1a42817901b4ccf45fa3d4b6561e74c7"
|
||||
dependencies = [
|
||||
"adler",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "0.8.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "927a765cd3fc26206e66b296465fa9d3e5ab003e651c1b3c060e7956d96b19d2"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"wasi",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_cpus"
|
||||
version = "1.16.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "object"
|
||||
version = "0.31.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8bda667d9f2b5051b8833f59f3bf748b28ef54f850f4fcb389a252aa383866d1"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot"
|
||||
version = "0.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
|
||||
dependencies = [
|
||||
"lock_api",
|
||||
"parking_lot_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot_core"
|
||||
version = "0.9.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "93f00c865fe7cabf650081affecd3871070f26767e7b2070a3ffae14c654b447"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"libc",
|
||||
"redox_syscall",
|
||||
"smallvec",
|
||||
"windows-targets",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-lite"
|
||||
version = "0.2.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "12cc1b0bf1727a77a54b6654e7b5f1af8604923edc8b81885f8ec92f9e3f0a05"
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro2"
|
||||
version = "1.0.66"
|
||||
@@ -58,6 +211,27 @@ dependencies = [
|
||||
"getrandom",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_syscall"
|
||||
version = "0.3.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "567664f262709473930a4bf9e51bf2ebf3348f2e748ccc50dea20646858f8f29"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustc-demangle"
|
||||
version = "0.1.23"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76"
|
||||
|
||||
[[package]]
|
||||
name = "scopeguard"
|
||||
version = "1.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
|
||||
|
||||
[[package]]
|
||||
name = "seq_ex"
|
||||
version = "0.1.0"
|
||||
@@ -65,6 +239,7 @@ dependencies = [
|
||||
"rand_core",
|
||||
"serde",
|
||||
"serde_cbor",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -97,6 +272,31 @@ dependencies = [
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "signal-hook-registry"
|
||||
version = "1.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d8229b473baa5980ac72ef434c4415e70c4b5e71b423043adb4ba059f89c99a1"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "smallvec"
|
||||
version = "1.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "62bb4feee49fdd9f707ef802e22365a35de4b7b299de4763d44bfea899442ff9"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.5.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2538b18701741680e0322a2302176d3253a35388e2e62f172f64f4f16605f877"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.28"
|
||||
@@ -108,6 +308,36 @@ dependencies = [
|
||||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio"
|
||||
version = "1.32.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "17ed6077ed6cd6c74735e21f37eb16dc3935f96878b1fe961074089cc80893f9"
|
||||
dependencies = [
|
||||
"backtrace",
|
||||
"bytes",
|
||||
"libc",
|
||||
"mio",
|
||||
"num_cpus",
|
||||
"parking_lot",
|
||||
"pin-project-lite",
|
||||
"signal-hook-registry",
|
||||
"socket2",
|
||||
"tokio-macros",
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-macros"
|
||||
version = "2.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "unicode-ident"
|
||||
version = "1.0.11"
|
||||
@@ -119,3 +349,69 @@ name = "wasi"
|
||||
version = "0.11.0+wasi-snapshot-preview1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
|
||||
|
||||
[[package]]
|
||||
name = "windows-sys"
|
||||
version = "0.48.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
|
||||
dependencies = [
|
||||
"windows-targets",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-targets"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d1eeca1c172a285ee6c2c84c341ccea837e7c01b12fbb2d0fe3c9e550ce49ec8"
|
||||
dependencies = [
|
||||
"windows_aarch64_gnullvm",
|
||||
"windows_aarch64_msvc",
|
||||
"windows_i686_gnu",
|
||||
"windows_i686_msvc",
|
||||
"windows_x86_64_gnu",
|
||||
"windows_x86_64_gnullvm",
|
||||
"windows_x86_64_msvc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_gnullvm"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b10d0c968ba7f6166195e13d593af609ec2e3d24f916f081690695cf5eaffb2f"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_msvc"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "571d8d4e62f26d4932099a9efe89660e8bd5087775a2ab5cdd8b747b811f1058"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnu"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2229ad223e178db5fbbc8bd8d3835e51e566b8474bfca58d2e6150c48bb723cd"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_msvc"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "600956e2d840c194eedfc5d18f8242bc2e17c7775b6684488af3a9fff6fe3287"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnu"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ea99ff3f8b49fb7a8e0d305e5aec485bd068c2ba691b6e277d29eaeac945868a"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnullvm"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8f1a05a1ece9a7a0d5a7ccf30ba2c33e3a61a30e042ffd247567d1de1d94120d"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_msvc"
|
||||
version = "0.48.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d419259aba16b663966e29e6d7c6ecfa0bb8425818bb96f6f1f3c3eb71a6e7b9"
|
||||
|
||||
+4
-2
@@ -10,13 +10,15 @@ path = "src/lib.rs"
|
||||
doc = true
|
||||
|
||||
[features]
|
||||
default = ["std", "serde"]
|
||||
default = ["serde", "tokio"]
|
||||
tokio = ["std", "dep:tokio"]
|
||||
std = []
|
||||
|
||||
[dependencies]
|
||||
serde = { version = "1.0.183", default-features = false, features = ["derive"], optional = true }
|
||||
#tokio = { version = "1.31.0", default-features = false, features = ["sync", "rt", "time", "macros"], optional = true }
|
||||
tokio = { version = "1.31.0", default-features = false, features = ["sync", "time"], optional = true }
|
||||
|
||||
[dev-dependencies]
|
||||
rand_core = { version = "0.6.4", features = ["getrandom"]}
|
||||
serde_cbor = { version = "0.11.2" }
|
||||
tokio = { version = "1.31.0", default-features = false, features = ["full"] }
|
||||
|
||||
@@ -1,250 +0,0 @@
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
fs::{read_dir, File},
|
||||
io::{Read, Write},
|
||||
ops::Deref,
|
||||
path::PathBuf,
|
||||
sync::{
|
||||
mpsc::{channel, Receiver, Sender},
|
||||
Arc, RwLock,
|
||||
},
|
||||
thread,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use rand_core::{OsRng, RngCore};
|
||||
use seq_ex::{SeqNo, TransportLayer, Packet};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// serde_cbor minimal format is both smaller and faster than default format.
|
||||
/// The overhead is ~33% faster.
|
||||
fn to_writer_minimal(value: &impl serde::Serialize, w: &mut impl serde_cbor::ser::Write) -> serde_cbor::Result<()> {
|
||||
value.serialize(&mut serde_cbor::Serializer::new(w).packed_format().legacy_enums())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn drop_packet() -> bool {
|
||||
OsRng.next_u32() >= (u32::MAX / 4 * 3)
|
||||
}
|
||||
|
||||
const FILE_CHUNK_SIZE: usize = 1000;
|
||||
const DOWNLOAD_LIMIT: u64 = 1000000;
|
||||
|
||||
#[derive(Debug)]
|
||||
enum SendData {
|
||||
RequestFile { filename: String },
|
||||
ConfirmFileSize { file: File },
|
||||
ConfirmDownload,
|
||||
FileDownload,
|
||||
ReadDir { download_missing: bool },
|
||||
DirContents,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
enum Payload<'a> {
|
||||
RequestFile { filename: &'a str },
|
||||
ConfirmFileSize { filesize: u64 },
|
||||
ConfirmDownload { fileid: u64 },
|
||||
FileDownload { fileid: u64, file_chunk: &'a [u8] },
|
||||
FileDownloadComplete { fileid: u64 },
|
||||
ReadDir,
|
||||
DirContents { filenames: Vec<&'a str> },
|
||||
}
|
||||
|
||||
const PACKET_TYPE_PAYLOAD: u8 = 0;
|
||||
const PACKET_TYPE_LOCK_PAYLOAD: u8 = 1;
|
||||
const PACKET_TYPE_REPLY: u8 = 2;
|
||||
const PACKET_TYPE_LOCK_REPLY: u8 = 3;
|
||||
const PACKET_TYPE_ACK: u8 = 4;
|
||||
|
||||
fn create_payload(seq_no: SeqNo, packet: &Payload<'_>) -> Vec<u8> {
|
||||
let mut p = vec![PACKET_TYPE_PAYLOAD];
|
||||
p.extend(&seq_no.to_be_bytes());
|
||||
to_writer_minimal(packet, &mut p);
|
||||
p
|
||||
}
|
||||
fn create_reply(seq_no: SeqNo, reply_no: SeqNo, packet: &Payload<'_>) -> Vec<u8> {
|
||||
let mut p = vec![PACKET_TYPE_REPLY];
|
||||
p.extend(&seq_no.to_be_bytes());
|
||||
p.extend(&reply_no.to_be_bytes());
|
||||
to_writer_minimal(packet, &mut p);
|
||||
p
|
||||
}
|
||||
macro_rules! reply {
|
||||
($guard:expr, $send_data:expr, $packet: expr) => {
|
||||
$guard.reply_with(|seq_no, reply_no| ($send_data, create_reply(seq_no, reply_no, &$packet)))
|
||||
};
|
||||
}
|
||||
macro_rules! send {
|
||||
($peer:expr, $trans:expr, $send_data:expr, $packet: expr) => {
|
||||
$peer.seqex.send_with($trans, |seq_no| ($send_data, create_payload(seq_no, &$packet)))
|
||||
};
|
||||
}
|
||||
|
||||
type SeqEx = seq_ex::sync::SeqExSync<(SendData, Vec<u8>), Vec<u8>>;
|
||||
type ReplyGuard<'a> = seq_ex::sync::ReplyGuard<'a, &'a Transport, (SendData, Vec<u8>), Vec<u8>>;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct Transport {
|
||||
sender: Sender<Vec<u8>>,
|
||||
time: Instant,
|
||||
}
|
||||
|
||||
struct Peer {
|
||||
downloads_in_progress: HashMap<u64, File>,
|
||||
home_dir: PathBuf,
|
||||
seqex: SeqEx,
|
||||
}
|
||||
|
||||
impl TransportLayer<(SendData, Vec<u8>)> for &Transport {
|
||||
fn time(&mut self) -> i64 {
|
||||
self.time.elapsed().as_millis() as i64
|
||||
}
|
||||
|
||||
fn send(&mut self, packet: Packet<&(SendData, Vec<u8>)>) {
|
||||
let p = match packet.consume() {
|
||||
Ok((_, packet)) => packet.clone(),
|
||||
Err(reply_no) => {
|
||||
let mut p = Vec::with_capacity(5);
|
||||
p.push(PACKET_TYPE_ACK);
|
||||
p.extend(&reply_no.to_be_bytes());
|
||||
p
|
||||
}
|
||||
};
|
||||
let _ = self.sender.send(p);
|
||||
}
|
||||
}
|
||||
|
||||
fn process(peer: &Arc<Peer>, transport: &Transport, guard: ReplyGuard<'_>, payload: Payload<'_>, data: Option<SendData>) -> Option<()> {
|
||||
match (payload, data) {
|
||||
(Payload::RequestFile { filename }, None) => {
|
||||
let file = File::open(peer.home_dir.join(filename)).ok()?;
|
||||
let metadata = file.metadata().ok()?;
|
||||
let filesize = metadata.len();
|
||||
reply!(
|
||||
guard,
|
||||
SendData::ConfirmFileSize { file },
|
||||
Payload::ConfirmFileSize { filesize }
|
||||
);
|
||||
}
|
||||
(Payload::ConfirmFileSize { filesize }, Some(SendData::RequestFile { filename })) => {
|
||||
let path = peer.home_dir.join(filename);
|
||||
if filesize > DOWNLOAD_LIMIT {
|
||||
return None;
|
||||
}
|
||||
let file = File::create(&path).ok()?;
|
||||
let fileid = OsRng.next_u64();
|
||||
peer.downloads_in_progress.insert(fileid, file);
|
||||
reply!(guard, SendData::ConfirmDownload, Payload::ConfirmDownload { fileid });
|
||||
}
|
||||
(Payload::ConfirmDownload { fileid }, Some(SendData::ConfirmFileSize { mut file })) => {
|
||||
let peer = peer.clone();
|
||||
let transport = transport.clone();
|
||||
thread::spawn(move || {
|
||||
const BUFFERED_CHUNKS: usize = 10;
|
||||
let mut buffer = [0u8; BUFFERED_CHUNKS * FILE_CHUNK_SIZE];
|
||||
while let Ok(n) = file.read(&mut buffer) {
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
let mut i = 0;
|
||||
while i < n {
|
||||
let j = n.min(i + FILE_CHUNK_SIZE);
|
||||
send!(
|
||||
peer,
|
||||
&transport,
|
||||
SendData::FileDownload,
|
||||
Payload::FileDownload { fileid, file_chunk: &buffer[i..j] }
|
||||
);
|
||||
i = j;
|
||||
}
|
||||
}
|
||||
send!(
|
||||
peer,
|
||||
&transport,
|
||||
SendData::FileDownload,
|
||||
Payload::FileDownloadComplete { fileid }
|
||||
);
|
||||
});
|
||||
}
|
||||
(Payload::FileDownload { fileid, file_chunk }, None) => {
|
||||
let mut file = peer.downloads_in_progress.get(&fileid)?;
|
||||
let result = file.write_all(file_chunk);
|
||||
}
|
||||
(Payload::FileDownloadComplete { fileid }, None) => {
|
||||
|
||||
}
|
||||
(Payload::ReadDir, None) => {
|
||||
let dir = read_dir(&peer.home_dir).ok()?;
|
||||
let mut filenames = Vec::new();
|
||||
for entry in dir {
|
||||
if let Ok(entry) = entry {
|
||||
if entry.file_type().map_or(false, |f| f.is_file()) {
|
||||
if let Ok(filename) = entry.file_name().into_string() {
|
||||
filenames.push(filename)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
let filenames: Vec<&str> = filenames.iter().map(|f| f.as_str()).collect();
|
||||
reply!(guard, SendData::DirContents, Payload::DirContents { filenames });
|
||||
}
|
||||
(Payload::DirContents { filenames }, Some(SendData::ReadDir { download_missing: true })) => {
|
||||
drop(guard);
|
||||
for filename in filenames {}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
Some(())
|
||||
}
|
||||
|
||||
fn receive(peer: &Arc<Peer>, transport: &Transport, receiver: &Receiver<Vec<u8>>) -> Option<()> {
|
||||
while let Ok(packet) = receiver.try_recv() {
|
||||
if drop_packet() {
|
||||
continue;
|
||||
}
|
||||
let pt = *packet.get(0)?;
|
||||
let (mut parsed_packet, payload_offset) = match pt {
|
||||
PACKET_TYPE_ACK => {
|
||||
let reply_no = SeqNo::from_be_bytes(packet.get(1..5)?.try_into().ok()?);
|
||||
(Packet::Ack(reply_no), 5)
|
||||
}
|
||||
PACKET_TYPE_PAYLOAD | PACKET_TYPE_LOCK_PAYLOAD => {
|
||||
let seq_no = SeqNo::from_be_bytes(packet.get(1..5)?.try_into().ok()?);
|
||||
(Packet::Payload(seq_no, packet), 5)
|
||||
}
|
||||
PACKET_TYPE_REPLY | PACKET_TYPE_LOCK_REPLY => {
|
||||
let seq_no = SeqNo::from_be_bytes(packet.get(1..5)?.try_into().ok()?);
|
||||
let reply_no = SeqNo::from_be_bytes(packet.get(5..9)?.try_into().ok()?);
|
||||
(Packet::Reply(seq_no, reply_no, packet), 9)
|
||||
}
|
||||
_ => return None,
|
||||
};
|
||||
if pt == PACKET_TYPE_LOCK_PAYLOAD || pt == PACKET_TYPE_LOCK_REPLY {
|
||||
parsed_packet.set_locking(true);
|
||||
}
|
||||
for recv_data in peer.seqex.receive_all(transport, parsed_packet) {
|
||||
let recv_data = recv_data.map(|packet| serde_cbor::from_slice::<Payload>(&packet[payload_offset..]));
|
||||
if let Ok(parsed_payload) = serde_cbor::from_slice::<Payload>(packet.get(payload_offset..)?) {
|
||||
process(peer, transport, guard, parsed_packet, send_data.map(|d| d.0));
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(())
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let alice_root = std::path::Path::new("examples").join("alice_home");
|
||||
let bob_root = std::path::Path::new("examples").join("bob_home");
|
||||
let (s, r) = channel::<i32>();
|
||||
thread::spawn(move || {
|
||||
s.send(0);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test() {
|
||||
main()
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
use std::{sync::mpsc::Receiver, thread, time::Duration};
|
||||
|
||||
use seq_ex::{
|
||||
sync::{MpscSeqEx, MpscTransport},
|
||||
sync::{MpscTransport, SeqExSync},
|
||||
Packet,
|
||||
};
|
||||
|
||||
@@ -19,7 +19,7 @@ fn drop_packet() -> bool {
|
||||
rand_core::OsRng.next_u32() & 1 > 0
|
||||
}
|
||||
|
||||
fn receive(recv: &Receiver<Packet<Payload>>, seq: &MpscSeqEx<Payload>, transport: &MpscTransport<Payload>, value: &mut f32) {
|
||||
fn receive(recv: &Receiver<Packet<Payload>>, seq: &SeqExSync<Payload, Payload>, transport: &MpscTransport<Payload>, value: &mut f32) {
|
||||
while let Ok(packet) = recv.try_recv() {
|
||||
if drop_packet() {
|
||||
continue;
|
||||
@@ -42,20 +42,20 @@ fn receive(recv: &Receiver<Packet<Payload>>, seq: &MpscSeqEx<Payload>, transport
|
||||
fn main() {
|
||||
let (transport1, recv2) = MpscTransport::new();
|
||||
let (transport2, recv1) = MpscTransport::new();
|
||||
let seq1 = MpscSeqEx::new(5, 1);
|
||||
let seq2 = MpscSeqEx::new(5, 1);
|
||||
let seq1 = SeqExSync::new(5, 1);
|
||||
let seq2 = SeqExSync::new(5, 1);
|
||||
let mut value = 0.0;
|
||||
let mut remote_value = value;
|
||||
|
||||
seq1.send(&transport1, Payload::Add(1.0));
|
||||
seq1.send(&transport1, true, Payload::Add(1.0));
|
||||
value += 1.0;
|
||||
seq1.send(&transport1, Payload::Sub(2.0));
|
||||
seq1.send(&transport1, true, Payload::Sub(2.0));
|
||||
value -= 2.0;
|
||||
seq1.send(&transport1, Payload::Mul(3.0));
|
||||
seq1.send(&transport1, true, Payload::Mul(3.0));
|
||||
value *= 3.0;
|
||||
seq1.send(&transport1, Payload::Div(4.0));
|
||||
seq1.send(&transport1, true, Payload::Div(4.0));
|
||||
value /= 4.0;
|
||||
seq1.send(&transport1, Payload::Mod(5.0));
|
||||
seq1.send(&transport1, true, Payload::Mod(5.0));
|
||||
value %= 5.0;
|
||||
|
||||
for _ in 0..16 {
|
||||
|
||||
+15
-10
@@ -60,7 +60,7 @@ fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payl
|
||||
let transport = peer.transport.clone();
|
||||
let seqex = peer.seqex.clone();
|
||||
if let Some(file) = filesystem.read().unwrap().get(&filename) {
|
||||
guard.reply(ConfirmRequestFile { filesize: file.len() as u64 });
|
||||
guard.reply(true, ConfirmRequestFile { filesize: file.len() as u64 });
|
||||
}
|
||||
thread::spawn(move || {
|
||||
let filesystem = filesystem.read().unwrap();
|
||||
@@ -70,18 +70,22 @@ fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payl
|
||||
let mut i = 0;
|
||||
while i < file.len() {
|
||||
let j = file.len().min(i + FILE_CHUNK_SIZE);
|
||||
seqex.send_locked(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() });
|
||||
seqex.send(
|
||||
&transport,
|
||||
true,
|
||||
FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() },
|
||||
);
|
||||
i = j;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
(Some((_, ConfirmRequestFile { filesize })), Some(RequestFile { filename })) => {
|
||||
(Some((_g, ConfirmRequestFile { filesize })), Some(RequestFile { filename })) => {
|
||||
let mut filesystem = peer.filesystem.write().unwrap();
|
||||
let file = Vec::with_capacity(filesize as usize);
|
||||
filesystem.insert(filename, file);
|
||||
}
|
||||
(Some((_, FileDownload { filename, file_chunk })), None) => {
|
||||
(Some((_g, 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() {
|
||||
@@ -102,7 +106,7 @@ fn receive(peer: &Peer) {
|
||||
continue;
|
||||
}
|
||||
if let Ok(parsed_packet) = serde_cbor::from_slice::<Packet<Payload>>(&packet) {
|
||||
for recv_data in peer.seqex.receive_all(&peer.transport, parsed_packet) {
|
||||
for recv_data in peer.seqex.try_receive_all(&peer.transport, parsed_packet) {
|
||||
process(peer, recv_data);
|
||||
}
|
||||
}
|
||||
@@ -137,14 +141,15 @@ fn main() {
|
||||
receiver: recv2,
|
||||
};
|
||||
|
||||
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() });
|
||||
let tl = &peer1.transport;
|
||||
peer1.seqex.send(tl, false, Payload::RequestFile { filename: "File1".to_string() });
|
||||
peer1.seqex.send(tl, false, Payload::RequestFile { filename: "File3".to_string() });
|
||||
peer1.seqex.send(tl, false, Payload::RequestFile { filename: "File2".to_string() });
|
||||
|
||||
for _ in 0..500 {
|
||||
for _ in 0..400 {
|
||||
receive(&peer1);
|
||||
receive(&peer2);
|
||||
thread::sleep(Duration::from_millis(1));
|
||||
thread::sleep(Duration::from_millis(2));
|
||||
peer1.seqex.service(&peer1.transport);
|
||||
peer2.seqex.service(&peer2.transport);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use std::sync::mpsc::Receiver;
|
||||
|
||||
use seq_ex::{
|
||||
sync::{MpscSeqEx, MpscTransport, RecvOk},
|
||||
sync::{MpscTransport, RecvOk, SeqExSync},
|
||||
Packet,
|
||||
};
|
||||
|
||||
@@ -14,23 +14,23 @@ enum Payload {
|
||||
}
|
||||
use Payload::*;
|
||||
|
||||
fn receive(recv: &Receiver<Packet<Payload>>, seq: &MpscSeqEx<Payload>, transport: &MpscTransport<Payload>) {
|
||||
fn receive(recv: &Receiver<Packet<Payload>>, seq: &SeqExSync<Payload, 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);
|
||||
guard.reply(false, Space);
|
||||
}
|
||||
(Some((guard, Space)), Some(Hello)) => {
|
||||
print!(" ");
|
||||
guard.reply(World);
|
||||
guard.reply(false, World);
|
||||
}
|
||||
(Some((guard, World)), Some(Space)) => {
|
||||
print!("World");
|
||||
guard.reply(Exclamation);
|
||||
guard.reply(false, Exclamation);
|
||||
}
|
||||
(Some((_, Exclamation)), Some(World)) => {
|
||||
(Some((_g, Exclamation)), Some(World)) => {
|
||||
print!("!");
|
||||
}
|
||||
(None, Some(Exclamation)) => {
|
||||
@@ -48,11 +48,11 @@ fn receive(recv: &Receiver<Packet<Payload>>, seq: &MpscSeqEx<Payload>, transport
|
||||
fn main() {
|
||||
let (transport1, recv2) = MpscTransport::new();
|
||||
let (transport2, recv1) = MpscTransport::new();
|
||||
let seq1 = MpscSeqEx::default();
|
||||
let seq2 = MpscSeqEx::default();
|
||||
let seq1 = SeqExSync::default();
|
||||
let seq2 = SeqExSync::default();
|
||||
|
||||
// We begin a "Hello World" exchange right here.
|
||||
seq1.send(&transport1, Payload::Hello);
|
||||
seq1.send(&transport1, false, Payload::Hello);
|
||||
|
||||
receive(&recv2, &seq2, &transport2);
|
||||
receive(&recv1, &seq1, &transport1);
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use tokio::{sync::mpsc, task};
|
||||
|
||||
use seq_ex::{
|
||||
tokio::{AsyncRecvError, MpscTransport, ReplyGuard, SeqExTokio},
|
||||
Packet,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
enum Payload {
|
||||
Hello,
|
||||
Space,
|
||||
World,
|
||||
Exclamation,
|
||||
NewLine,
|
||||
}
|
||||
use Payload::*;
|
||||
|
||||
async fn receive(reply_guard: ReplyGuard<'_, &MpscTransport<Payload>, Payload>, payload: Payload) -> Option<()> {
|
||||
match payload {
|
||||
Hello => {
|
||||
print!("Hello");
|
||||
|
||||
let (reply_guard, payload) = reply_guard.reply(false, Space).await.ok()?;
|
||||
if payload != World {
|
||||
return None;
|
||||
}
|
||||
print!("World");
|
||||
|
||||
let (reply_guard, payload) = reply_guard.reply(false, Exclamation).await.ok()?;
|
||||
if payload != NewLine {
|
||||
return None;
|
||||
}
|
||||
print!("\n");
|
||||
drop(reply_guard);
|
||||
}
|
||||
_ => {
|
||||
assert!(false, "Unsolicited payload received: {:?}", payload);
|
||||
}
|
||||
}
|
||||
Some(())
|
||||
}
|
||||
|
||||
async fn say_hello(seq: &SeqExTokio<Payload>, transport: &MpscTransport<Payload>) -> Option<()> {
|
||||
let (reply_guard, payload) = seq.send(transport, false, Hello).await.ok()?;
|
||||
if payload != Space {
|
||||
return None;
|
||||
}
|
||||
print!(" ");
|
||||
|
||||
let (reply_guard, payload) = reply_guard.reply(false, World).await.ok()?;
|
||||
if payload != Exclamation {
|
||||
return None;
|
||||
}
|
||||
print!("!");
|
||||
|
||||
let _ = reply_guard.reply(false, NewLine).await;
|
||||
Some(())
|
||||
}
|
||||
|
||||
async fn pump_all(peer: Arc<SeqExTokio<Payload>>, transport: MpscTransport<Payload>) {
|
||||
if let Some((g, payload)) = peer.pump(&transport).await {
|
||||
spawn_pump(peer.clone(), transport.clone());
|
||||
receive(g, payload).await;
|
||||
}
|
||||
}
|
||||
fn spawn_pump(peer: Arc<SeqExTokio<Payload>>, transport: MpscTransport<Payload>) {
|
||||
task::spawn(async move {
|
||||
pump_all(peer, transport).await;
|
||||
});
|
||||
}
|
||||
|
||||
fn peer_main(transport: MpscTransport<Payload>, mut recv: mpsc::Receiver<Packet<Payload>>) -> Arc<SeqExTokio<Payload>> {
|
||||
let (seq, mut service) = SeqExTokio::<Payload>::new_default();
|
||||
let peer = Arc::new(seq);
|
||||
let peer_weak = Arc::downgrade(&peer);
|
||||
let tl = transport.clone();
|
||||
task::spawn(async move {
|
||||
while let Some(peer) = peer_weak.upgrade() {
|
||||
peer.service_task(&tl, &mut service).await;
|
||||
}
|
||||
});
|
||||
let peer_weak = Arc::downgrade(&peer);
|
||||
task::spawn(async move {
|
||||
while let Some(packet) = recv.recv().await {
|
||||
if let Some(peer) = peer_weak.upgrade() {
|
||||
let tl = transport.clone();
|
||||
task::spawn(async move {
|
||||
let result = peer.receive(&tl, packet);
|
||||
if matches!(&result, Ok(_) | Err(AsyncRecvError::AsyncReply)) {
|
||||
spawn_pump(peer.clone(), tl.clone());
|
||||
}
|
||||
if let Ok((g, payload)) = result {
|
||||
receive(g, payload).await;
|
||||
}
|
||||
});
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
peer
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
let (transport2, recv1) = MpscTransport::new(32);
|
||||
let (transport1, recv2) = MpscTransport::new(32);
|
||||
let peer1 = peer_main(transport1.clone(), recv1);
|
||||
let _peer2 = peer_main(transport2, recv2);
|
||||
|
||||
say_hello(&peer1, &transport1).await;
|
||||
}
|
||||
#[test]
|
||||
fn test() {
|
||||
main()
|
||||
}
|
||||
+217
-164
File diff suppressed because it is too large
Load Diff
+79
-67
@@ -1,10 +1,10 @@
|
||||
use crate::{DirectError, Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
|
||||
use crate::{DirectRecvError, Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
|
||||
|
||||
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,
|
||||
is_holding_lock: 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
|
||||
@@ -12,31 +12,25 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
|
||||
/// The identifier will tell the remote peer which packets contain fragments of the file,
|
||||
/// 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.reply_inner(false, |_, _| packet_data)
|
||||
pub fn reply(self, seq_cst: bool, packet_data: SendData) {
|
||||
self.reply_with(seq_cst, |_, _| 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);
|
||||
fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
|
||||
let app = self.app.take();
|
||||
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);
|
||||
self.seq.reply_raw(
|
||||
app.unwrap(),
|
||||
self.reply_no,
|
||||
self.is_holding_lock,
|
||||
seq_cst,
|
||||
packet_data(seq_no, self.reply_no),
|
||||
);
|
||||
}
|
||||
}
|
||||
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
|
||||
fn drop(&mut self) {
|
||||
if let Some(app) = &mut self.app {
|
||||
if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no, self.locked) {
|
||||
if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no, self.is_holding_lock) {
|
||||
app.send(p)
|
||||
}
|
||||
}
|
||||
@@ -44,20 +38,33 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Dro
|
||||
}
|
||||
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()
|
||||
f.debug_struct("ReplyGuard")
|
||||
.field("reply_no", &self.reply_no)
|
||||
.field("is_holding_lock", &self.is_holding_lock)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[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 RecvError {
|
||||
DroppedTooEarly,
|
||||
DroppedDuplicate,
|
||||
WaitingForRecv,
|
||||
WaitingForReply,
|
||||
}
|
||||
#[cfg(feature = "std")]
|
||||
impl std::fmt::Display for RecvError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
RecvError::DroppedTooEarly => write!(f, "packet arrived too early"),
|
||||
RecvError::DroppedDuplicate => write!(f, "packet was a duplicate"),
|
||||
RecvError::WaitingForRecv => write!(f, "can't process until another packet is received"),
|
||||
RecvError::WaitingForReply => write!(f, "can't process until a reply is finished"),
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "std")]
|
||||
impl std::error::Error for RecvError {}
|
||||
|
||||
pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
|
||||
Payload {
|
||||
@@ -97,14 +104,14 @@ macro_rules! impl_recvok {
|
||||
}
|
||||
}
|
||||
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 {
|
||||
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 },
|
||||
RecvOkRaw::Payload { reply_no, seq_cst, recv_data } => Self::Payload {
|
||||
reply_guard: ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock: seq_cst },
|
||||
recv_data,
|
||||
},
|
||||
RecvOkRaw::Reply { reply_no, locked, recv_data, send_data } => Self::Reply {
|
||||
reply_guard: ReplyGuard { seq, app: Some(app), reply_no, locked },
|
||||
RecvOkRaw::Reply { reply_no, seq_cst, recv_data, send_data } => Self::Reply {
|
||||
reply_guard: ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock: seq_cst },
|
||||
recv_data,
|
||||
send_data,
|
||||
},
|
||||
@@ -126,13 +133,6 @@ macro_rules! impl_recvok {
|
||||
(None, None) => None,
|
||||
}
|
||||
}
|
||||
pub fn map<R>(self, f: impl FnOnce(P) -> R) -> $recv<'a, TL, R, SendData, RecvData, CAP> {
|
||||
match self {
|
||||
Self::Payload { reply_guard, recv_data } => $recv::Payload { reply_guard, recv_data: f(recv_data) },
|
||||
Self::Reply { reply_guard, recv_data, send_data } => $recv::Reply { reply_guard, recv_data: f(recv_data), send_data },
|
||||
Self::Ack { send_data } => $recv::Ack { send_data },
|
||||
}
|
||||
}
|
||||
}
|
||||
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> {
|
||||
@@ -149,25 +149,25 @@ 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);
|
||||
/// Can mutate `next_service_timestamp`.
|
||||
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, seq_cst: bool, packet_data: SendData) -> Result<(), SendData> {
|
||||
match self.try_send_direct(app.time(), seq_cst, packet_data) {
|
||||
Ok(p) => {
|
||||
app.send(p);
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
pub fn try_send_with(
|
||||
/// Can mutate `next_service_timestamp`.
|
||||
pub fn try_send_with<F: FnOnce(SeqNo) -> SendData>(
|
||||
&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);
|
||||
seq_cst: bool,
|
||||
packet_data: F,
|
||||
) -> Result<(), F> {
|
||||
match self.try_send_direct_with(app.time(), seq_cst, packet_data) {
|
||||
Ok(p) => {
|
||||
app.send(p);
|
||||
Ok(())
|
||||
}
|
||||
@@ -179,29 +179,41 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
|
||||
&mut self,
|
||||
mut app: impl TransportLayer<SendData>,
|
||||
packet: Packet<P>,
|
||||
) -> Result<crate::seq_queue::RecvOkRaw<SendData, P>, Error<P>> {
|
||||
) -> Result<crate::seq_queue::RecvOkRaw<SendData, P>, RecvError> {
|
||||
match self.receive_raw_and_direct(packet) {
|
||||
Ok(a) => Ok(a),
|
||||
Err(DirectError::ResendAck(reply_no)) => {
|
||||
Err(DirectRecvError::DroppedDuplicateResendAck(reply_no)) => {
|
||||
app.send(Packet::Ack(reply_no));
|
||||
Err(Error::OutOfSequence)
|
||||
Err(RecvError::DroppedDuplicate)
|
||||
}
|
||||
Err(DirectError::OutOfSequence) => Err(Error::OutOfSequence),
|
||||
Err(DirectError::WindowIsFull(p)) => Err(Error::WindowIsFull(p)),
|
||||
Err(DirectError::WindowIsLocked(p)) => Err(Error::WindowIsLocked(p)),
|
||||
Err(DirectRecvError::DroppedTooEarly) => Err(RecvError::DroppedTooEarly),
|
||||
Err(DirectRecvError::DroppedDuplicate) => Err(RecvError::DroppedDuplicate),
|
||||
Err(DirectRecvError::WaitingForRecv) => Err(RecvError::WaitingForRecv),
|
||||
Err(DirectRecvError::WaitingForReply) => Err(RecvError::WaitingForReply),
|
||||
}
|
||||
}
|
||||
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)
|
||||
/// Can mutate `next_service_timestamp`.
|
||||
/// If `unlock` is true and the return value is true pump may return new values.
|
||||
///
|
||||
/// Only returns false if the reply number was incorrect or used twice.
|
||||
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool, seq_cst: bool, packet_data: SendData) -> bool {
|
||||
if let Some(p) = self.reply_raw_and_direct(app.time(), reply_no, unlock, seq_cst, packet_data) {
|
||||
app.send(p);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool) {
|
||||
/// If `unlock` is true and the return value is true pump may return new values.
|
||||
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool) -> bool {
|
||||
if let Some(p) = self.ack_raw_and_direct(reply_no, unlock) {
|
||||
app.send(p)
|
||||
app.send(p);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
/// Can mutate `next_service_timestamp`.
|
||||
pub fn service(&mut self, mut app: impl TransportLayer<SendData>) -> i64 {
|
||||
let current_time = app.time();
|
||||
let mut iter = None;
|
||||
@@ -214,10 +226,10 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
|
||||
&mut self,
|
||||
app: TL,
|
||||
packet: Packet<P>,
|
||||
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, Error<P>> {
|
||||
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, RecvError> {
|
||||
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<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
|
||||
self.pump_raw().map(|r| RecvOk::from_raw(self, app, r))
|
||||
pub fn try_pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
|
||||
self.try_pump_raw().map(|r| RecvOk::from_raw(self, app, r))
|
||||
}
|
||||
}
|
||||
|
||||
+127
-134
@@ -1,83 +1,71 @@
|
||||
use std::{
|
||||
ops::{Deref, DerefMut},
|
||||
sync::{
|
||||
mpsc::{channel, Receiver, Sender},
|
||||
Condvar, Mutex, MutexGuard,
|
||||
Condvar, Mutex,
|
||||
},
|
||||
time::Instant,
|
||||
};
|
||||
|
||||
use crate::{Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
|
||||
use crate::{
|
||||
Packet, PumpError, RecvError, 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)>,
|
||||
seq_ex: Mutex<(SeqEx<SendData, RecvData, CAP>, usize, bool)>,
|
||||
send_block: Condvar,
|
||||
recv_lock: Condvar,
|
||||
reply_block: Condvar,
|
||||
}
|
||||
|
||||
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
|
||||
seq: &'a SeqExSync<SendData, RecvData, CAP>,
|
||||
app: Option<TL>,
|
||||
reply_no: SeqNo,
|
||||
locked: bool,
|
||||
is_holding_lock: bool,
|
||||
}
|
||||
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
|
||||
fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
|
||||
let app = self.app.take().unwrap();
|
||||
let mut seq = self.seq.seq_ex.lock().unwrap();
|
||||
let seq_no = seq.0.seq_no();
|
||||
seq.0
|
||||
.reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no));
|
||||
if seq.2 {
|
||||
seq.2 = false;
|
||||
drop(seq);
|
||||
self.seq.reply_block.notify_all();
|
||||
}
|
||||
}
|
||||
/// 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.
|
||||
/// The identifier will tell the remote peer which packets contain fragments of the file,
|
||||
/// 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.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 mut seq = self.seq.lock();
|
||||
let seq_no = seq.seq_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);
|
||||
pub fn reply(self, seq_cst: bool, packet_data: SendData) {
|
||||
self.reply_with(seq_cst, |_, _| packet_data)
|
||||
}
|
||||
}
|
||||
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
|
||||
fn drop(&mut self) {
|
||||
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)
|
||||
if let Some(app) = self.app.take() {
|
||||
let mut seq = self.seq.seq_ex.lock().unwrap();
|
||||
seq.0.ack_raw(app, self.reply_no, self.is_holding_lock);
|
||||
if seq.2 {
|
||||
seq.2 = false;
|
||||
drop(seq);
|
||||
self.seq.reply_block.notify_all();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
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()
|
||||
f.debug_struct("ReplyGuard")
|
||||
.field("reply_no", &self.reply_no)
|
||||
.field("is_holding_lock", &self.is_holding_lock)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[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 enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
|
||||
Payload {
|
||||
reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
|
||||
@@ -94,135 +82,138 @@ pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const C
|
||||
}
|
||||
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> {
|
||||
pub struct RecvIter<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
|
||||
seq: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
|
||||
app: TL,
|
||||
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)>);
|
||||
impl<'a, SendData, RecvData, const CAP: usize> Deref for SeqExGuard<'a, SendData, RecvData, CAP> {
|
||||
type Target = SeqEx<SendData, RecvData, CAP>;
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.0 .0
|
||||
}
|
||||
}
|
||||
impl<'a, SendData, RecvData, const CAP: usize> DerefMut for SeqExGuard<'a, SendData, RecvData, CAP> {
|
||||
fn deref_mut(&mut self) -> &mut Self::Target {
|
||||
&mut self.0 .0
|
||||
}
|
||||
blocking: bool,
|
||||
}
|
||||
|
||||
impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
|
||||
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
|
||||
Self {
|
||||
seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)),
|
||||
seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0, false)),
|
||||
send_block: Condvar::default(),
|
||||
recv_lock: Condvar::default(),
|
||||
reply_block: Condvar::default(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
|
||||
&self,
|
||||
app: TL,
|
||||
mut packet: Packet<P>,
|
||||
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, Error> {
|
||||
packet: Packet<P>,
|
||||
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, RecvError> {
|
||||
let mut seq = self.seq_ex.lock().unwrap();
|
||||
match seq.0.receive_raw(app.clone(), packet) {
|
||||
Ok(r) => {
|
||||
if seq.1 > 0 {
|
||||
seq.1 -= 1;
|
||||
drop(seq);
|
||||
self.send_block.notify_one();
|
||||
}
|
||||
Ok(RecvOk::from_raw(self, app, r))
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
pub fn try_pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
|
||||
let mut seq = self.seq_ex.lock().unwrap();
|
||||
// Enforce that only one thread may pump at a time.
|
||||
if seq.2 {
|
||||
return Err(PumpError::WaitingForReply);
|
||||
}
|
||||
match seq.0.try_pump_raw() {
|
||||
Ok(r) => {
|
||||
if seq.1 > 0 {
|
||||
seq.1 -= 1;
|
||||
drop(seq);
|
||||
self.send_block.notify_one();
|
||||
}
|
||||
Ok(RecvOk::from_raw(self, app, r))
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
pub fn pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Option<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>> {
|
||||
let mut seq = self.seq_ex.lock().unwrap();
|
||||
// Enforce that only one thread may pump at a time.
|
||||
if seq.2 {
|
||||
return None;
|
||||
}
|
||||
loop {
|
||||
match seq.0.receive_raw(app.clone(), packet) {
|
||||
match seq.0.try_pump_raw() {
|
||||
Ok(r) => {
|
||||
if seq.1 > 0 {
|
||||
seq.1 -= 1;
|
||||
drop(seq);
|
||||
self.send_block.notify_one();
|
||||
}
|
||||
return Ok(RecvOk::from_raw(self, app, r));
|
||||
return Some(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;
|
||||
Err(PumpError::WaitingForRecv) => return None,
|
||||
Err(PumpError::WaitingForReply) => {
|
||||
seq.2 = true;
|
||||
seq = self.reply_block.wait(seq).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
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();
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
fn receive_all_inner<TL: TransportLayer<SendData>, P: Into<RecvData>>(
|
||||
&self,
|
||||
app: TL,
|
||||
blocking: bool,
|
||||
packet: Packet<P>,
|
||||
) -> RecvIter<'_, TL, P, SendData, RecvData, CAP> {
|
||||
match self.receive(app.clone(), packet) {
|
||||
Ok(r) => RecvIter { seq: Some(self), app, first: Some(r), blocking },
|
||||
Err(RecvError::WaitingForReply) if blocking => RecvIter { seq: Some(self), app, first: None, blocking },
|
||||
Err(_) => RecvIter { seq: None, app, first: None, blocking },
|
||||
}
|
||||
}
|
||||
pub fn receive_all<TL: TransportLayer<SendData>, P: Into<RecvData>>(
|
||||
&self,
|
||||
app: TL,
|
||||
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 { seq: None, app, first: None }
|
||||
}
|
||||
) -> RecvIter<'_, TL, P, SendData, RecvData, CAP> {
|
||||
self.receive_all_inner(app, true, packet)
|
||||
}
|
||||
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)
|
||||
pub fn try_receive_all<TL: TransportLayer<SendData>, P: Into<RecvData>>(
|
||||
&self,
|
||||
app: TL,
|
||||
packet: Packet<P>,
|
||||
) -> RecvIter<'_, TL, P, SendData, RecvData, CAP> {
|
||||
self.receive_all_inner(app, false, packet)
|
||||
}
|
||||
fn send_with_inner<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, mut packet_data: impl FnMut(SeqNo) -> SendData) {
|
||||
|
||||
pub fn try_send_with<TL: TransportLayer<SendData>, F: FnOnce(SeqNo) -> SendData>(&self, app: TL, seq_cst: bool, packet_data: F) -> Result<(), F> {
|
||||
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;
|
||||
}
|
||||
seq.0.try_send_with(app, seq_cst, packet_data)
|
||||
}
|
||||
fn send_inner<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, mut packet_data: SendData) {
|
||||
pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, packet_data: SendData) -> Result<(), SendData> {
|
||||
let mut seq = self.seq_ex.lock().unwrap();
|
||||
while let Err(p) = seq.0.try_send(app.clone(), locked, packet_data) {
|
||||
seq.0.try_send(app, seq_cst, packet_data)
|
||||
}
|
||||
|
||||
pub fn send_with<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) {
|
||||
let mut seq = self.seq_ex.lock().unwrap();
|
||||
while let Err(p) = seq.0.try_send_with(app.clone(), seq_cst, packet_data) {
|
||||
packet_data = p;
|
||||
seq.1 += 1;
|
||||
seq = self.send_block.wait(seq).unwrap();
|
||||
seq.1 -= 1;
|
||||
}
|
||||
}
|
||||
pub fn send(&self, app: impl TransportLayer<SendData>, packet_data: SendData) {
|
||||
self.send_inner(app, false, packet_data)
|
||||
}
|
||||
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, locked, packet_data(seq_no))
|
||||
}
|
||||
pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
|
||||
self.lock().service(app)
|
||||
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, mut packet_data: SendData) {
|
||||
let mut seq = self.seq_ex.lock().unwrap();
|
||||
while let Err(p) = seq.0.try_send(app.clone(), seq_cst, packet_data) {
|
||||
packet_data = p;
|
||||
seq.1 += 1;
|
||||
seq = self.send_block.wait(seq).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
pub fn lock(&self) -> SeqExGuard<'_, SendData, RecvData, CAP> {
|
||||
SeqExGuard(self.seq_ex.lock().unwrap())
|
||||
pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
|
||||
self.seq_ex.lock().unwrap().0.service(app)
|
||||
}
|
||||
}
|
||||
impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvData, CAP> {
|
||||
@@ -231,20 +222,24 @@ impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvD
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> ReplyIter<'a, TL, P, SendData, RecvData, CAP> {
|
||||
impl<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize> RecvIter<'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>
|
||||
for RecvIter<'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.into())
|
||||
} else if let Some(origin) = self.seq {
|
||||
origin.pump(self.app.clone()).ok()
|
||||
if self.blocking {
|
||||
origin.pump(self.app.clone())
|
||||
} else {
|
||||
origin.try_pump(self.app.clone()).ok()
|
||||
}
|
||||
} else {
|
||||
None
|
||||
}
|
||||
@@ -256,8 +251,6 @@ pub struct MpscTransport<Payload: Clone> {
|
||||
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<Packet<Payload>>) {
|
||||
|
||||
+372
-270
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user