diff --git a/rust/README.md b/rust/README.md index 99bb78a7..2b5aad89 100644 --- a/rust/README.md +++ b/rust/README.md @@ -38,14 +38,18 @@ embedded NULs. Passwords and other C strings reject embedded NULs. Callbacks are owned `Send + 'static` closures, not borrowed `&mut` closures. Use `move` captures and channels or `Arc` for shared application state. Native -handles are thread-bound; send copied data through channels rather than moving -a runner or value handle between threads. Handles own stable native pointers -and can be moved out of or back into their Rust boxes on the same thread. -Callback panics are contained at the -FFI boundary; a panicking value callback stops its operation. +runner and crypto handles remain thread-bound. `Value` is `Send`, not `Sync`: +it can move between threads, and mutation requires exclusive `&mut` access. +Get/listen callbacks receive ordinary owned `Box` objects and can move +them directly through channels. Callback panics are contained at the FFI +boundary; a panicking value callback stops its operation. -`Value::boxed()` creates an independent copy, including payload and metadata, -so a retained value does not alias mutable native storage or in-flight puts. +The same `Value` type is used for construction, callback delivery, async streams, +and publication. `as_slice()` borrows the payload; `user_type()` and `push_type()` +borrow strings. `owner()` returns a retained public-key handle without exporting +it. `as_bytes()` explicitly copies the payload into a Rust `Vec`. +Received values can be edited directly and passed to `put` without converting +or cloning them. `boxed()` explicitly creates an additional independent copy. ```rust use opendht::{DhtRunner, InfoHash, Value}; @@ -56,11 +60,12 @@ runner.run(0)?; let hash = InfoHash::get("example"); let (sent, received) = mpsc::channel(); let token = runner.listen(&hash, move |value, expired| { - if !expired { let _ = sent.send(value.as_bytes()); } + if !expired { let _ = sent.send(value); } true })?; runner.put(&hash, Value::new("hello"), |_| {}, false); -let bytes = received.recv_timeout(std::time::Duration::from_secs(5))?; +let value = received.recv_timeout(std::time::Duration::from_secs(5))?; +assert_eq!(value.as_slice(), b"hello"); runner.cancel_listen(&hash, token); runner.join()?; # Ok::<(), Box>(()) @@ -120,13 +125,13 @@ runner.join()?; concurrently and succeeds when any peer responds. Resolve hostnames through the application's runtime before calling it. - `get`, `get_filtered`, and `query` return fallible streams. Get/listen values - are independent `ValueSnapshot`s containing payload, ID, type, sequence, - priority, user/push types, signer public-key bytes, and recipient. Listener - `ValueEvent`s also preserve the expiration flag. + are the same owned `Box` objects delivered by the callback API, + with borrowed access to payload and metadata. Listener `ValueEvent`s also + preserve the expiration flag. - Streams buffer at most 64 items by default; `with_capacity` changes this limit. A full buffer stops delivery and yields `WouldBlock` after buffered items, then ends. Restart the operation to obtain a fresh snapshot. Native - failures, invalid metadata, and runner termination also yield errors rather + failures and runner termination also yield errors rather than successful EOF. Duplicate native deliveries are not suppressed. - Dropping a listener immediately queues native cancellation and releases its token. Listeners retain the runner until dropped. Dropping get/query streams @@ -134,11 +139,25 @@ runner.join()?; Dropping a completion future abandons its result but does not cancel the native operation or undo a permanent put. - Runner adapters and listeners remain thread-bound. Use `block_on`, a local - executor, or Tokio `LocalSet`, not `tokio::spawn`. Snapshots and query rows can + executor, or Tokio `LocalSet`, not `tokio::spawn`. Received values and query rows can be sent to other threads. Non-threaded runners still require independent `loop_once()` calls while waiting. `join()` remains blocking; await graceful `shutdown()` first when persistence is required. +Received values are copied once by the native callback bridge to isolate them +from mutable DHT storage. Async streams move those exact boxes through their +channels without copying payloads or rebuilding metadata. Borrowed data remains +valid for the value's lifetime, even after the callback returns or the runner +stops. Put operations consume their box, so Rust cannot continue mutating a value +after native code retains it. Malformed UTF-8 metadata is reported when its +string accessor is called, rather than eagerly failing the entire stream. + +`value.owner()` returns the existing `PublicKey` type, not a second key class or +DER bytes. Call `pack()` only when exported bytes are actually needed. Shared +const owner-key fingerprint caches are initialized on the callback thread before +transfer. Replacing a returned key through `unpack()` changes that handle, not +the value's owner key. + ## Typed Records Enable `serde` for typed MessagePack operations on `AsyncDhtRunner`. This feature @@ -175,7 +194,7 @@ let record = Service { }; runner.put_typed(&hash, &record, false).await?; if let Some(event) = updates.try_next().await? { - println!("{}: {:?}", event.value.raw.id, event.value.data); + println!("{}: {:?}", event.value.raw.id(), event.value.data); } drop(updates); let records: Vec<_> = runner.get_typed::(&hash).try_collect().await?; @@ -184,8 +203,10 @@ let records: Vec<_> = runner.get_typed::(&hash).try_collect().await?; Typed puts accept `Serialize`, including unsized values. Typed reads require `DeserializeOwned`, without imposing `Clone`, `Send`, or `Unpin` on the record. Decoding happens when the consumer polls, not in native callbacks. Each -`TypedValue` exposes decoded `data` and the original `raw` snapshot, retaining -payload bytes and all metadata. `TypedEvent` preserves the expiration flag. +`TypedValue` exposes decoded `data` and the original owned `raw: Box`, retaining +payload bytes and all metadata. Decoding borrows the raw payload directly, with +no intermediate byte-buffer copy; allocations required by the decoded record +still apply. `TypedEvent` preserves the expiration flag. `TypedError` distinguishes `Serialize`, `Deserialize { id, source }`, and native `Operation` errors, preserving the underlying error through `Error::source()`. @@ -199,7 +220,7 @@ only when polled. Serialization failure does not start a native put. Dropping the future after publication starts does not undo the put. Typed puts use the default native metadata; for custom IDs/types or signed/encrypted publication, serialize into a `Value` and use the existing untyped methods. Raw get/listen -snapshots can also be decoded with `TypedValue::::try_from(snapshot)` or +values can also be decoded with `TypedValue::::try_from(value)` or `TypedEvent::::try_from(event)`. The wire payload is compact MessagePack, following OpenDHT's serialization diff --git a/rust/examples/dhtnode_async.rs b/rust/examples/dhtnode_async.rs index cc95499a..f0c0a58e 100644 --- a/rust/examples/dhtnode_async.rs +++ b/rust/examples/dhtnode_async.rs @@ -26,8 +26,8 @@ fn main() -> Result<(), Box> { .ok_or_else(|| io::Error::other("listener ended before receiving a value"))??; println!( "LISTEN {}: {} (expired: {})", - event.value.id, - String::from_utf8_lossy(&event.value.data), + event.value.id(), + String::from_utf8_lossy(event.value.as_slice()), event.expired, ); drop(updates); @@ -60,7 +60,8 @@ fn main() -> Result<(), Box> { })??; println!( "TYPED LISTEN {}: {:?}", - event.value.raw.id, event.value.data + event.value.raw.id(), + event.value.data ); drop(updates); let values: Vec<_> = destination diff --git a/rust/src/aio.rs b/rust/src/aio.rs index fa08cd5f..af9b8ac2 100644 --- a/rust/src/aio.rs +++ b/rust/src/aio.rs @@ -1,7 +1,8 @@ //! Runtime-independent futures and bounded streams, enabled by the `async` feature. //! //! Runners and listeners stay on their creating thread. Use a local executor or -//! Tokio `LocalSet`; streamed snapshots contain only owned, thread-safe data. +//! Tokio `LocalSet`; received values own their native storage and can be +//! transferred between threads without another payload copy. use crate::{DhtRunner, InfoHash, OpToken, PkId, QueryResult, Value}; use futures_channel::{mpsc, oneshot}; @@ -16,43 +17,9 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; -/// An independent value snapshot containing no native handles. -pub struct ValueSnapshot { - pub data: Vec, - pub id: u64, - pub value_type: u16, - pub seq: u16, - pub priority: u32, - pub user_type: String, - pub push_type: String, - /// DER-encoded public key of the signer, if present. - pub owner: Option>, - pub recipient: InfoHash, -} - -impl ValueSnapshot { - fn copy(value: &Value) -> io::Result { - Ok(Self { - data: value.as_bytes(), - id: value.id(), - value_type: value.value_type(), - seq: value.seq(), - priority: value.priority(), - user_type: value - .user_type() - .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?, - push_type: value - .push_type() - .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?, - owner: value.owner().map(|key| key.pack()).transpose()?, - recipient: value.recipient(), - }) - } -} - /// A listener update; expiration events retain the expired value's metadata. pub struct ValueEvent { - pub value: ValueSnapshot, + pub value: Box, pub expired: bool, } @@ -142,7 +109,7 @@ impl StreamSender { /// A bounded get or query stream. /// /// Buffered items precede a terminal error. Native failure, runner shutdown, -/// invalid metadata, and buffer overflow are never reported as successful EOF. +/// and buffer overflow are never reported as successful EOF. /// Dropping the stream stops delivery; native work stops at its next callback /// or when the runner is joined. #[must_use = "streams must be polled to receive values and completion errors"] @@ -374,12 +341,12 @@ impl AsyncDhtRunner { pending } - pub fn get(&self, hash: &InfoHash) -> OperationStream { + pub fn get(&self, hash: &InfoHash) -> OperationStream> { let (mut sender, stream) = stream_channel(self.capacity); let completion = sender.completion.clone(); self.state.runner.borrow_mut().get( hash, - move |value| sender.send(ValueSnapshot::copy(&value)), + move |value| sender.send(Ok(value)), move |success| completion.finish(outcome(success)), ); stream @@ -389,13 +356,13 @@ impl AsyncDhtRunner { &self, hash: &InfoHash, filter: &str, - ) -> io::Result> { + ) -> io::Result>> { let (mut sender, stream) = stream_channel(self.capacity); let completion = sender.completion.clone(); self.state.runner.borrow_mut().get_filtered( hash, filter, - move |value| sender.send(ValueSnapshot::copy(&value)), + move |value| sender.send(Ok(value)), move |success| completion.finish(outcome(success)), )?; Ok(stream) @@ -422,9 +389,7 @@ impl AsyncDhtRunner { let token = self.state.runner.borrow_mut().listen_filtered( hash, filter, - move |value, expired| { - sender.send(ValueSnapshot::copy(&value).map(|value| ValueEvent { value, expired })) - }, + move |value, expired| sender.send(Ok(ValueEvent { value, expired })), )?; Ok(Listener { state: self.state.clone(), diff --git a/rust/src/dhtrunner.rs b/rust/src/dhtrunner.rs index 3dcc2ffd..d144a513 100644 --- a/rust/src/dhtrunner.rs +++ b/rust/src/dhtrunner.rs @@ -272,7 +272,7 @@ extern "C" fn get_handler_cb(value: *const ffi::Value, ptr: *mut c_void) -> bool } callback(false, || unsafe { let handler = ptr as *mut GetHandler; - (*handler).get_cb(Value::from_raw(ffi::dht_value_copy(value)).expect("value copy failed")) + (*handler).get_cb(Value::copy_from_received(value).expect("value copy failed")) }) } @@ -308,7 +308,7 @@ extern "C" fn listen_handler(value: *const ffi::Value, expired: bool, ptr: *mut callback(false, || unsafe { let handler = ptr as *mut ListenHandler; (*handler).cb( - Value::from_raw(ffi::dht_value_copy(value)).expect("value copy failed"), + Value::copy_from_received(value).expect("value copy failed"), expired, ) }) @@ -537,6 +537,7 @@ impl DhtRunner { result(unsafe { ffi::dht_runner_connectivity_changed(self.as_ptr()) }) } + /// Deliver owned values that callbacks may mutate or move to another thread. pub fn get( &mut self, h: &InfoHash, @@ -676,7 +677,8 @@ impl DhtRunner { unsafe { ffi::dht_runner_cancel_put(self.as_ptr(), h, vid) } } - /// Registers an owned callback. It is released after cancellation or `join()`. + /// Registers an owned callback receiving ordinary owned `Box` objects. + /// It is released after cancellation or `join()`. /// Borrowed callbacks cannot outlive their stack frame: /// /// ```compile_fail diff --git a/rust/src/typed.rs b/rust/src/typed.rs index eff39778..2d3d3596 100644 --- a/rust/src/typed.rs +++ b/rust/src/typed.rs @@ -30,7 +30,7 @@ //! }; //! runner.put_typed(key, &record, false).await?; //! if let Some(event) = updates.try_next().await? { -//! println!("{}: {:?}", event.value.raw.id, event.value.data); +//! println!("{}: {:?}", event.value.raw.id(), event.value.data); //! } //! drop(updates); //! let records: Vec<_> = runner.get_typed::(key).try_collect().await?; @@ -38,7 +38,7 @@ //! } //! ``` -use crate::aio::{AsyncDhtRunner, ValueEvent, ValueSnapshot}; +use crate::aio::{AsyncDhtRunner, ValueEvent}; use crate::{InfoHash, Value}; use futures_core::Stream; use serde::{de::DeserializeOwned, Serialize}; @@ -51,7 +51,7 @@ use std::task::{Context, Poll}; /// A decoded record together with its original payload and native metadata. pub struct TypedValue { pub data: T, - pub raw: ValueSnapshot, + pub raw: Box, } impl fmt::Debug for TypedValue { @@ -59,8 +59,8 @@ impl fmt::Debug for TypedValue { formatter .debug_struct("TypedValue") .field("data", &self.data) - .field("id", &self.raw.id) - .field("seq", &self.raw.seq) + .field("id", &self.raw.id()) + .field("seq", &self.raw.seq()) .finish_non_exhaustive() } } @@ -106,12 +106,15 @@ impl Error for TypedError { } } -impl TryFrom for TypedValue { +impl TryFrom> for TypedValue { type Error = TypedError; - fn try_from(raw: ValueSnapshot) -> Result { - let data = rmp_serde::from_slice(&raw.data) - .map_err(|source| TypedError::Deserialize { id: raw.id, source })?; + fn try_from(raw: Box) -> Result { + let data = + rmp_serde::from_slice(raw.as_slice()).map_err(|source| TypedError::Deserialize { + id: raw.id(), + source, + })?; Ok(Self { data, raw }) } } diff --git a/rust/src/value.rs b/rust/src/value.rs index 3c976bc8..e8a842f9 100644 --- a/rust/src/value.rs +++ b/rust/src/value.rs @@ -14,7 +14,40 @@ pub use crate::ffi::DataView; native_handle!(Value, dht_value_unref); +/// Rust owns an unaliased native value: constructors allocate it, received +/// values are deep-copied once, and boxed() deep-copies explicitly. Puts consume +/// the handle before native code retains it. Shared owner keys are const and +/// their lazy fingerprints are initialized before received values cross threads. +/// Mutation requires exclusive access; native storage may be freed on any thread. +/// Value is Send, not Sync. +unsafe impl Send for Value {} + +impl AsRef<[u8]> for Value { + fn as_ref(&self) -> &[u8] { + self.as_slice() + } +} + +impl fmt::Debug for Value { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("Value") + .field("id", &self.id()) + .field("seq", &self.seq()) + .finish_non_exhaustive() + } +} + impl Value { + pub(crate) unsafe fn copy_from_received(value: *const ffi::Value) -> io::Result> { + let value = unsafe { Self::from_raw(ffi::dht_value_copy(value))? }; + if let Some(owner) = value.owner() { + owner.id(); + owner.long_id(); + } + Ok(value) + } + pub fn with_id(data: &[u8], id: u64) -> Box { unsafe { Self::from_raw(ffi::dht_value_with_id_new(data.as_ptr(), data.len(), id)) @@ -41,13 +74,16 @@ impl Value { } pub fn as_bytes(&self) -> Vec { - unsafe { - let dv = ffi::dht_value_get_data(self.as_ptr()); - if dv.size == 0 { - return Vec::new(); - } - let slice = slice::from_raw_parts(dv.data, dv.size); - slice.to_vec() + self.as_slice().to_vec() + } + + /// Borrow the payload without allocating or copying it. + pub fn as_slice(&self) -> &[u8] { + let view = self.dataview(); + if view.size == 0 { + &[] + } else { + unsafe { slice::from_raw_parts(view.data, view.size) } } } @@ -63,12 +99,8 @@ impl Value { unsafe { PublicKey::from_raw(ffi::dht_value_get_owner(self.as_ptr())).ok() } } - pub fn user_type(&self) -> Result { - unsafe { - Ok(CStr::from_ptr(ffi::dht_value_get_user_type(self.as_ptr())) - .to_str()? - .to_owned()) - } + pub fn user_type(&self) -> Result<&str, str::Utf8Error> { + unsafe { CStr::from_ptr(ffi::dht_value_get_user_type(self.as_ptr())).to_str() } } pub fn set_user_type(&mut self, value: &str) -> io::Result<()> { @@ -103,12 +135,8 @@ impl Value { ffi::dht_value_set_priority(self.as_ptr(), value); } } - pub fn push_type(&self) -> Result { - unsafe { - Ok(CStr::from_ptr(ffi::dht_value_get_push_type(self.as_ptr())) - .to_str()? - .to_owned()) - } + pub fn push_type(&self) -> Result<&str, str::Utf8Error> { + unsafe { CStr::from_ptr(ffi::dht_value_get_push_type(self.as_ptr())).to_str() } } pub fn set_push_type(&mut self, value: &str) -> io::Result<()> { let value = CString::new(value)?;