1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327
//! Remote channels.
//!
//! This module contains channels that can be used to exchange data of
//! arbitrary type with a remote endpoint.
//!
//! The data type must be serializable and deserializable by [serde] and
//! [sendable across thread boundaries](Send).
//! The [RemoteSend](crate::RemoteSend) trait is implemented automatically
//! for all types that fulfill these requirements.
//!
//! # Establishing a channel connection with a remote endpoint
//!
//! Except for a [base channel](base), a channel connection is established by sending
//! the Sender or Receiver half of a channel to a remote endpoint over an already
//! existing channel.
//! The transmitted channel-half can also be part of a larger object, such as a struct, tuple or enum.
//! Most channel types can even be forwarded over multiple connections.
//!
//! The primary purpose of a [base channel](base) is to provide an initial channel after
//! establishing a connection over a physical transport.
//! In most cases there is no need to create it directly and you will probably only use it
//! after it has been returned from an [initial connect function](crate::Connect).
//!
//! # Channel types
//!
//! [Broadcast](broadcast), [MPSC](mpsc), [oneshot] and [watch] channels are closely
//! modelled after the channels found in [tokio::sync].
//! They also work when both halves of the channel are local and can be forwarded
//! over multiple connections.
//!
//! An [local/remote channel](lr) is a more restricted version of an [MPSC](mpsc) channel.
//! It does not support forwarding and exactly one half of it must be on a remote endpoint.
//! Its benefit is, that it does not spawn an async task and thus uses less resources.
//! When in doubt use an [MPSC channel](mpsc) instead.
//!
//! A [binary channel](bin) can be used to exchange binary data over a channel.
//! It skips serialization and deserialization and thus is more efficient for binary data,
//! especially when using text codecs such as JSON.
//! It does support forwarding.
//! However, at least one half of it must be on a remote endpoint.
//!
//! # Acknowledgements and connection latency
//!
//! The channels do not wait for acknowledgement of transmitted values.
//! Thus, the next value can be queued for transmission before the previous value
//! has been received by the remote endpoint.
//! Hence, throughput is unaffected by the connection roundtrip time.
//!
//! A successful send does not guarantee that the transmitted value will be received
//! by the remote endpoint.
//! However, values cannot be lost intermittently, i.e. if a transmission error
//! occurs on the underlying physical connection the whole [chmux] connection will
//! be terminated eventually and sending will fail.
//!
//! If you need confirmation for every sent value, consider sending a [`oneshot::Sender`]`<()>`
//! along with it (for example as a tuple).
//! The remote endpoint can then send back an empty message as confirmation over the oneshot channel
//! after it has processed the received value.
//!
//! # Size considerations
//!
//! The size of the objects exchanged is not limited.
//! If the object is small enough it is first serialized into a buffer and then sent as
//! one message to the remote endpoint.
//! For larger objects, a serialization thread is spawned and the message is sent
//! chunk-by-chunk to avoid the need for a large temporary memory buffer.
//! Deserialization is also performed on-the-fly as data is received for large objects.
//!
//! Large values are always sent in chunks, so that one channel does not block other channels
//! for a long period of time.
//!
//! To avoid denial of service attacks when you exchange data with untrusted remote endpoints,
//! make sure that an object cannot grow to infinite size during deserialization.
//! For example, avoid using containers of unlimited size like [Vec] or [HashMap](std::collections::HashMap)
//! in your data structure or limit their maximum size during deserialization by applying
//! the `#[serde(deserialize_with="...")]` attribute to the fields containing them.
//!
//! # Static buffer size configuration for received channel-halves
//!
//! This is used to specify the local buffer size (in items) via a const generic type parameter
//! when the sender or receiver half of a channel *is received*.
//!
//! The default buffer size is [DEFAULT_BUFFER], which is currently 2 items.
//! It can be increased to improve performance, but this will also increase
//! the maximum amount of memory used per channel.
//! Setting the buffer size too high must be avoided, since this can lead to
//! the program running out of memory.
//!
//! # Error handling
//!
//! During sending it is often useful to treat errors that occur due to the disconnection, either graceful
//! or non-graceful, of the remote endpoint as an indication that the receiver is not
//! interested anymore in the transmitted data, rather than a hard error.
//! To avoid complicated error checks in your code, you can use [SendResultExt::into_disconnected]
//! to query whether a send error occurred due to the above cause and exit the sending process gracefully.
//!
use serde::{Deserialize, Serialize};
use std::{error::Error, fmt};
use crate::chmux;
mod interlock;
pub mod base;
pub mod bin;
pub mod broadcast;
//pub mod buffer;
pub mod lr;
pub mod mpsc;
pub mod oneshot;
pub mod watch;
/// Error connecting a remote channel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ConnectError {
/// The corresponding sender or receiver has been dropped.
Dropped,
/// Error initiating chmux connection.
Connect(chmux::ConnectError),
/// Error listening for or accepting chmux connection.
Listen(chmux::ListenerError),
}
impl fmt::Display for ConnectError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Self::Dropped => write!(f, "other part was dropped"),
Self::Connect(err) => write!(f, "connect error: {err}"),
Self::Listen(err) => write!(f, "listen error: {err}"),
}
}
}
impl Error for ConnectError {}
/// Reason for closure of sender.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum ClosedReason {
/// Channel was closed because receiver has been closed.
Closed,
/// Channel was closed because receiver has been dropped.
Dropped,
/// Channel was closed because connection between sender and receiver failed.
Failed,
}
/// Back channel message that receiver has been closed.
pub(crate) const BACKCHANNEL_MSG_CLOSE: u8 = 0x01;
/// Back channel message that error has occurred.
pub(crate) const BACKCHANNEL_MSG_ERROR: u8 = 0x02;
/// Remote sending error.
#[derive(Clone)]
pub(crate) enum RemoteSendError {
/// Send error.
Send(base::SendErrorKind),
/// Connect error.
Connect(chmux::ConnectError),
/// Listen for connect error.
Listen(chmux::ListenerError),
/// Forwarding error.
Forward,
/// Receiver was closed.
Closed,
}
/// Common functions to query send errors for details.
///
/// This is implemented for all send errors in this module.
///
/// Since [watch] channels have no method to close them,
/// [is_closed](Self::is_closed) and [is_disconnected](Self::is_disconnected) are
/// equivalent for them.
pub trait SendErrorExt {
/// Whether the remote endpoint closed the channel.
fn is_closed(&self) -> bool;
/// Whether the remote endpoint closed the channel, was dropped or the connection failed.
fn is_disconnected(&self) -> bool;
/// Whether the error is final, i.e. no further send operation can succeed.
fn is_final(&self) -> bool;
/// Whether the error is caused by the item to be sent.
fn is_item_specific(&self) -> bool;
}
/// Common functions to query results of send operations for details.
///
/// This is implemented by all results from send operations in this module.
pub trait SendResultExt {
/// The error type of this result.
type Err: SendErrorExt;
/// Whether the remote endpoint closed the channel.
///
/// Returns `Ok(false)` if the send was successful and `Ok(true)` if the send failed for the
/// above reason.
/// The original error is returned if the failure had another cause.
///
/// # Example
///
/// In the following example the client counts upwards and the server closes the connections once
/// it receives the number 5.
/// The client stops counting as soon as sending fails.
/// Note that the panic error path in the client would only be triggered by another error,
/// such as a serialization error.
///
/// ```
/// use remoc::prelude::*;
///
/// // This would be run on the client.
/// async fn client(mut tx: rch::base::Sender<u32>) {
/// let mut n = 0;
///
/// loop {
/// if tx.send(n).await.into_closed().unwrap() {
/// break;
/// }
/// n += 1;
/// }
///
/// assert!(n > 5);
/// }
///
/// // This would be run on the server.
/// async fn server(mut rx: rch::base::Receiver<u32>) {
/// while let Some(n) = rx.recv().await.unwrap() {
/// if n == 5 {
/// rx.close().await;
/// }
/// }
/// }
/// # tokio_test::block_on(remoc::doctest::client_server(client, server));
/// ```
fn into_closed(self) -> Result<bool, Self::Err>;
/// Whether the remote endpoint closed the channel, was dropped or the connection failed.
///
/// Returns `Ok(false)` if the send was successful and `Ok(true)` if the send failed for the
/// above reasons.
/// The original error is returned if the failure had another cause.
///
/// # Example
///
/// In the following example the client counts upwards and the server drops the connections once
/// it receives the number 5.
/// The client stops counting as soon as sending fails.
/// Note that the panic error path in the client would only be triggered by another error,
/// such as a serialization error.
///
/// ```
/// use remoc::prelude::*;
///
/// // This would be run on the client.
/// async fn client(mut tx: rch::base::Sender<u32>) {
/// let mut n = 0;
///
/// loop {
/// if tx.send(n).await.into_disconnected().unwrap() {
/// break;
/// }
/// n += 1;
/// }
///
/// assert!(n > 5);
/// }
///
/// // This would be run on the server.
/// async fn server(mut rx: rch::base::Receiver<u32>) {
/// while let Some(n) = rx.recv().await.unwrap() {
/// if n == 5 {
/// break;
/// }
/// }
/// }
/// # tokio_test::block_on(remoc::doctest::client_server(client, server));
/// ```
fn into_disconnected(self) -> Result<bool, Self::Err>;
}
impl<E> SendResultExt for Result<(), E>
where
E: SendErrorExt,
{
type Err = E;
fn into_closed(self) -> Result<bool, E> {
match self {
Ok(()) => Ok(false),
Err(err) => {
if err.is_closed() {
Ok(true)
} else {
Err(err)
}
}
}
}
fn into_disconnected(self) -> Result<bool, E> {
match self {
Ok(()) => Ok(false),
Err(err) => {
if err.is_disconnected() {
Ok(true)
} else {
Err(err)
}
}
}
}
}
/// Default buffer size in items, when the sender or receiver half of a channel *is received*.
///
/// This can be changed via a const generic type parameter of a sender or receiver.
///
/// The current default buffer size is 2.
pub const DEFAULT_BUFFER: usize = 2;
/// Default maximum allowed item size for sending and receiving items over remote channels.
///
/// The current default maximum allowed item size is 16 MB.
pub const DEFAULT_MAX_ITEM_SIZE: usize = 16_777_216;