leyoda/rqbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Box<[u8]> instead of Vec<u8> for ByteBufOwned

Browse source
This commit is contained in:
Igor Katson 2024-03-29 11:00:58 +00:00
parent 3cdf6d4cfc
commit fa05fe8376
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
24 changed files with 118 additions and 109 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Makefile
View file

@ -113,4 +113,4 @@ release-all: release-windows release-linux release-macos-universal
cp ./target/x86_64-pc-windows-gnu/release-github/rqbit.exe /tmp/rqbit-release
cp ./target/x86_64-apple-darwin/release-github/rqbit-osx-universal /tmp/rqbit-release
cp ./target/x86_64-unknown-linux-gnu/release-github/rqbit /tmp/rqbit-release/rqbit-linux-x86_64
echo "The release was built in /tmp/rqbit-release"
echo "The release was built in /tmp/rqbit-release"

4
crates/bencode/src/bencode_value.rs
View file

@ -1,6 +1,6 @@
use std::{collections::HashMap, marker::PhantomData};
use buffers::{ByteBuf, ByteString};
use buffers::{ByteBuf, ByteBufOwned};
use clone_to_owned::CloneToOwned;
use serde::Deserializer;
@ -133,7 +133,7 @@ where
}
pub type BencodeValueBorrowed<'a> = BencodeValue<ByteBuf<'a>>;
pub type BencodeValueOwned = BencodeValue<ByteString>;
pub type BencodeValueOwned = BencodeValue<ByteBufOwned>;
#[cfg(test)]
mod tests {

2
crates/bencode/src/lib.rs
View file

@ -6,4 +6,4 @@ pub use bencode_value::*;
pub use serde_bencode_de::*;
pub use serde_bencode_ser::*;
pub use buffers::{ByteBuf, ByteString};
pub use buffers::{ByteBuf, ByteBufOwned};

14
crates/bencode/src/serde_bencode_ser.rs
View file

@ -2,7 +2,7 @@ use std::collections::BTreeMap;
use serde::{ser::Impossible, Serialize, Serializer};
use buffers::ByteString;
use buffers::ByteBufOwned;
#[derive(Debug)]
pub enum SerErrorKind {
@ -136,8 +136,8 @@ impl<'ser, W: std::io::Write> serde::ser::SerializeTuple for SerializeTuple<'ser
struct SerializeMap<'ser, W: std::io::Write> {
ser: &'ser mut BencodeSerializer<W>,
tmp: BTreeMap<ByteString, ByteString>,
last_key: Option<ByteString>,
tmp: BTreeMap<ByteBufOwned, ByteBufOwned>,
last_key: Option<ByteBufOwned>,
}
impl<'ser, W: std::io::Write> serde::ser::SerializeMap for SerializeMap<'ser, W> {
type Ok = ();
@ -152,7 +152,7 @@ impl<'ser, W: std::io::Write> serde::ser::SerializeMap for SerializeMap<'ser, W>
let mut ser = BencodeSerializer::new(&mut buf);
ser.hack_no_bytestring_prefix = true;
key.serialize(&mut ser)?;
self.last_key.replace(ByteString::from(buf));
self.last_key.replace(ByteBufOwned::from(buf));
Ok(())
// key.serialize(&mut *self.ser);
}
@ -165,7 +165,7 @@ impl<'ser, W: std::io::Write> serde::ser::SerializeMap for SerializeMap<'ser, W>
let mut ser = BencodeSerializer::new(&mut buf);
value.serialize(&mut ser)?;
self.tmp
.insert(self.last_key.take().unwrap(), ByteString::from(buf));
.insert(self.last_key.take().unwrap(), ByteBufOwned::from(buf));
Ok(())
}
@ -180,7 +180,7 @@ impl<'ser, W: std::io::Write> serde::ser::SerializeMap for SerializeMap<'ser, W>
struct SerializeStruct<'ser, W: std::io::Write> {
ser: &'ser mut BencodeSerializer<W>,
tmp: BTreeMap<&'static str, ByteString>,
tmp: BTreeMap<&'static str, ByteBufOwned>,
}
impl<'ser, W: std::io::Write> serde::ser::SerializeStruct for SerializeStruct<'ser, W> {
type Ok = ();
@ -198,7 +198,7 @@ impl<'ser, W: std::io::Write> serde::ser::SerializeStruct for SerializeStruct<'s
let mut buf = Vec::new();
let mut ser = BencodeSerializer::new(&mut buf);
value.serialize(&mut ser)?;
self.tmp.insert(key, ByteString::from(buf));
self.tmp.insert(key, ByteBufOwned::from(buf));
Ok(())
}

38
crates/buffers/src/lib.rs
View file

@ -8,7 +8,7 @@ use serde::{Deserialize, Deserializer};
use clone_to_owned::CloneToOwned;
#[derive(Default, PartialEq, Eq, Hash, Clone, PartialOrd, Ord)]
pub struct ByteString(pub Vec<u8>);
pub struct ByteBufOwned(pub Box<[u8]>);
#[derive(Default, Deserialize, PartialEq, Eq, Hash, Clone, PartialOrd, Ord)]
#[serde(transparent)]
@ -18,7 +18,7 @@ pub trait ByteBufT {
fn as_slice(&self) -> &[u8];
}
impl ByteBufT for ByteString {
impl ByteBufT for ByteBufOwned {
fn as_slice(&self) -> &[u8] {
self.as_ref()
}
@ -78,31 +78,31 @@ impl<'a> std::fmt::Display for ByteBuf<'a> {
}
}
impl std::fmt::Debug for ByteString {
impl std::fmt::Debug for ByteBufOwned {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
debug_bytes(&self.0, f, true)
}
}
impl std::fmt::Display for ByteString {
impl std::fmt::Display for ByteBufOwned {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
debug_bytes(&self.0, f, false)
}
}
impl<'a> CloneToOwned for ByteBuf<'a> {
type Target = ByteString;
type Target = ByteBufOwned;
fn clone_to_owned(&self) -> Self::Target {
ByteString(self.as_slice().to_owned())
ByteBufOwned(self.as_slice().to_owned().into_boxed_slice())
}
}
impl CloneToOwned for ByteString {
type Target = ByteString;
impl CloneToOwned for ByteBufOwned {
type Target = ByteBufOwned;
fn clone_to_owned(&self) -> Self::Target {
ByteString(self.as_slice().to_owned())
ByteBufOwned(self.0.clone())
}
}
@ -112,7 +112,7 @@ impl<'a> std::convert::AsRef<[u8]> for ByteBuf<'a> {
}
}
impl std::convert::AsRef<[u8]> for ByteString {
impl std::convert::AsRef<[u8]> for ByteBufOwned {
fn as_ref(&self) -> &[u8] {
&self.0
}
@ -126,7 +126,7 @@ impl<'a> std::ops::Deref for ByteBuf<'a> {
}
}
impl std::ops::Deref for ByteString {
impl std::ops::Deref for ByteBufOwned {
type Target = [u8];
fn deref(&self) -> &Self::Target {
@ -140,15 +140,15 @@ impl<'a> From<&'a [u8]> for ByteBuf<'a> {
}
}
impl<'a> From<&'a [u8]> for ByteString {
impl<'a> From<&'a [u8]> for ByteBufOwned {
fn from(b: &'a [u8]) -> Self {
Self(b.into())
}
}
impl From<Vec<u8>> for ByteString {
impl From<Vec<u8>> for ByteBufOwned {
fn from(b: Vec<u8>) -> Self {
Self(b)
Self(b.into_boxed_slice())
}
}
@ -161,7 +161,7 @@ impl<'a> serde::ser::Serialize for ByteBuf<'a> {
}
}
impl serde::ser::Serialize for ByteString {
impl serde::ser::Serialize for ByteBufOwned {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
@ -170,7 +170,7 @@ impl serde::ser::Serialize for ByteString {
}
}
impl<'de> serde::de::Deserialize<'de> for ByteString {
impl<'de> serde::de::Deserialize<'de> for ByteBufOwned {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
@ -178,7 +178,7 @@ impl<'de> serde::de::Deserialize<'de> for ByteString {
struct Visitor;
impl<'de> serde::de::Visitor<'de> for Visitor {
type Value = Vec<u8>;
type Value = ByteBufOwned;
fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
formatter.write_str("byte string")
@ -187,9 +187,9 @@ impl<'de> serde::de::Deserialize<'de> for ByteString {
where
E: serde::de::Error,
{
Ok(v.to_owned())
Ok(v.to_owned().into())
}
}
Ok(ByteString(deserializer.deserialize_byte_buf(Visitor {})?))
Ok(deserializer.deserialize_byte_buf(Visitor {})?)
}
}

4
crates/dht/src/bprotocol.rs
View file

@ -4,7 +4,7 @@ use std::{
net::{Ipv4Addr, SocketAddrV4},
};
use bencode::{ByteBuf, ByteString};
use bencode::{ByteBuf, ByteBufOwned};
use clone_to_owned::CloneToOwned;
use librqbit_core::hash_id::Id20;
use serde::{
@ -356,7 +356,7 @@ pub struct Message<BufT> {
pub ip: Option<SocketAddrV4>,
}
impl Message<ByteString> {
impl Message<ByteBufOwned> {
// This implies that the transaction id was generated by us.
pub fn get_our_transaction_id(&self) -> Option<u16> {
if self.transaction_id.len() != 2 {

30
crates/dht/src/dht.rs
View file

@ -21,7 +21,7 @@ use crate::{
};
use anyhow::{bail, Context};
use backoff::{backoff::Backoff, ExponentialBackoffBuilder};
use bencode::ByteString;
use bencode::ByteBufOwned;
use dashmap::DashMap;
use futures::{
future::BoxFuture, stream::FuturesUnordered, FutureExt, Stream, StreamExt, TryFutureExt,
@ -59,7 +59,7 @@ struct OutstandingRequest {
pub struct WorkerSendRequest {
// If this is set, we are tracking the response in inflight_by_transaction_id
our_tid: Option<u16>,
message: Message<ByteString>,
message: Message<ByteBufOwned>,
addr: SocketAddr,
}
@ -607,13 +607,13 @@ impl DhtState {
}
}
fn create_request(&self, request: Request) -> (u16, Message<ByteString>) {
fn create_request(&self, request: Request) -> (u16, Message<ByteBufOwned>) {
let transaction_id = self.next_transaction_id.fetch_add(1, Ordering::Relaxed);
let transaction_id_buf = [(transaction_id >> 8) as u8, (transaction_id & 0xff) as u8];
let message = match request {
Request::GetPeers(info_hash) => Message {
transaction_id: ByteString::from(transaction_id_buf.as_ref()),
transaction_id: ByteBufOwned::from(transaction_id_buf.as_ref()),
version: None,
ip: None,
kind: MessageKind::GetPeersRequest(GetPeersRequest {
@ -622,7 +622,7 @@ impl DhtState {
}),
},
Request::FindNode(target) => Message {
transaction_id: ByteString::from(transaction_id_buf.as_ref()),
transaction_id: ByteBufOwned::from(transaction_id_buf.as_ref()),
version: None,
ip: None,
kind: MessageKind::FindNodeRequest(FindNodeRequest {
@ -631,7 +631,7 @@ impl DhtState {
}),
},
Request::Ping => Message {
transaction_id: ByteString::from(transaction_id_buf.as_ref()),
transaction_id: ByteBufOwned::from(transaction_id_buf.as_ref()),
version: None,
ip: None,
kind: MessageKind::PingRequest(PingRequest { id: self.id }),
@ -648,7 +648,7 @@ impl DhtState {
port,
token,
}),
transaction_id: ByteString::from(transaction_id_buf.as_ref()),
transaction_id: ByteBufOwned::from(transaction_id_buf.as_ref()),
version: None,
ip: None,
},
@ -658,7 +658,7 @@ impl DhtState {
fn on_received_message(
self: &Arc<Self>,
msg: Message<ByteString>,
msg: Message<ByteBufOwned>,
addr: SocketAddr,
) -> anyhow::Result<()> {
let generate_compact_nodes = |target| {
@ -770,8 +770,8 @@ impl DhtState {
id: self.id,
nodes: Some(compact_node_info),
values: Some(compact_peer_info),
token: Some(ByteString(
self.peer_store.gen_token_for(req.id, addr).to_vec(),
token: Some(ByteBufOwned::from(
&self.peer_store.gen_token_for(req.id, addr)[..],
)),
}),
};
@ -821,15 +821,15 @@ enum Request {
FindNode(Id20),
Announce {
info_hash: Id20,
token: ByteString,
token: ByteBufOwned,
port: u16,
},
Ping,
}
enum ResponseOrError {
Response(Response<ByteString>),
Error(ErrorDescription<ByteString>),
Response(Response<ByteBufOwned>),
Error(ErrorDescription<ByteBufOwned>),
}
impl core::fmt::Debug for ResponseOrError {
@ -1010,7 +1010,7 @@ impl DhtWorker {
&self,
socket: &UdpSocket,
mut input_rx: UnboundedReceiver<WorkerSendRequest>,
output_tx: Sender<(Message<ByteString>, SocketAddr)>,
output_tx: Sender<(Message<ByteBufOwned>, SocketAddr)>,
) -> anyhow::Result<()> {
let writer = async {
let mut buf = Vec::new();
@ -1050,7 +1050,7 @@ impl DhtWorker {
.recv_from(&mut buf)
.await
.context("error reading from UDP socket")?;
match bprotocol::deserialize_message::<ByteString>(&buf[..size]) {
match bprotocol::deserialize_message::<ByteBufOwned>(&buf[..size]) {
Ok(msg) => match output_tx.send((msg, addr)).await {
Ok(_) => {}
Err(_) => break,

4
crates/dht/src/peer_store.rs
View file

@ -5,7 +5,7 @@ use std::{
sync::atomic::AtomicU32,
};
use bencode::ByteString;
use bencode::ByteBufOwned;
use chrono::{DateTime, Utc};
use librqbit_core::hash_id::Id20;
use parking_lot::RwLock;
@ -134,7 +134,7 @@ impl PeerStore {
token
}
pub fn store_peer(&self, announce: &AnnouncePeer<ByteString>, addr: SocketAddr) -> bool {
pub fn store_peer(&self, announce: &AnnouncePeer<ByteBufOwned>, addr: SocketAddr) -> bool {
// If the info_hash in announce is too far away from us, don't store it.
// If the token doesn't match, don't store it.
// If we are out of capacity, don't store it.

4
crates/librqbit/src/api.rs
View file

@ -1,7 +1,7 @@
use std::{net::SocketAddr, sync::Arc};
use anyhow::Context;
use buffers::ByteString;
use buffers::ByteBufOwned;
use dht::{DhtStats, Id20};
use futures::Stream;
use http::StatusCode;
@ -268,7 +268,7 @@ pub struct ApiAddTorrentResponse {
fn make_torrent_details(
info_hash: &Id20,
info: &TorrentMetaV1Info<ByteString>,
info: &TorrentMetaV1Info<ByteBufOwned>,
only_files: Option<&[usize]>,
) -> Result<TorrentDetailsResponse> {
let files = info

2
crates/librqbit/src/chunk_tracker.rs
View file

@ -35,7 +35,7 @@ pub struct ChunkTracker {
fn compute_chunk_status(lengths: &Lengths, needed_pieces: &BF) -> BF {
let required_size = lengths.chunk_bitfield_bytes();
let vec = vec![0u8; required_size];
let mut chunk_bf = BF::from_vec(vec);
let mut chunk_bf = BF::from_boxed_slice(vec.into_boxed_slice());
for piece_index in needed_pieces
.get(0..lengths.total_pieces() as usize)
.unwrap()

15
crates/librqbit/src/create_torrent_file.rs
View file

@ -5,7 +5,7 @@ use std::path::Path;
use anyhow::Context;
use bencode::bencode_serialize_to_writer;
use buffers::ByteString;
use buffers::ByteBufOwned;
use librqbit_core::torrent_metainfo::{TorrentMetaV1File, TorrentMetaV1Info, TorrentMetaV1Owned};
use librqbit_core::Id20;
use sha1w::{ISha1, Sha1};
@ -44,7 +44,7 @@ fn walk_dir_find_paths(dir: &Path, out: &mut Vec<Cow<'_, Path>>) -> anyhow::Resu
Ok(())
}
fn compute_info_hash(t: &TorrentMetaV1Info<ByteString>) -> anyhow::Result<Id20> {
fn compute_info_hash(t: &TorrentMetaV1Info<ByteBufOwned>) -> anyhow::Result<Id20> {
struct W {
hash: sha1w::Sha1,
}
@ -79,7 +79,7 @@ fn osstr_to_bytes(o: &OsStr) -> Vec<u8> {
async fn create_torrent_raw<'a>(
path: &'a Path,
options: CreateTorrentOptions<'a>,
) -> anyhow::Result<TorrentMetaV1Info<ByteString>> {
) -> anyhow::Result<TorrentMetaV1Info<ByteBufOwned>> {
path.try_exists()
.with_context(|| format!("path {:?} doesn't exist", path))?;
let basename = path
@ -87,7 +87,7 @@ async fn create_torrent_raw<'a>(
.ok_or_else(|| anyhow::anyhow!("cannot determine basename of {:?}", path))?;
let is_dir = path.is_dir();
let single_file_mode = !is_dir;
let name: ByteString = match options.name {
let name: ByteBufOwned = match options.name {
Some(name) => name.as_bytes().into(),
None => osstr_to_bytes(basename).into(),
};
@ -112,7 +112,7 @@ async fn create_torrent_raw<'a>(
let mut remaining_piece_length = piece_length;
let mut piece_checksum = sha1w::Sha1::new();
let mut piece_hashes = Vec::<u8>::new();
let mut output_files: Vec<TorrentMetaV1File<ByteString>> = Vec::new();
let mut output_files: Vec<TorrentMetaV1File<ByteBufOwned>> = Vec::new();
let spawner = BlockingSpawner::default();
@ -216,7 +216,7 @@ pub async fn create_torrent<'a>(
#[cfg(test)]
mod tests {
use buffers::ByteBuf;
use buffers::{ByteBuf, ByteBufOwned};
use librqbit_core::torrent_metainfo::torrent_from_bytes;
use crate::create_torrent;
@ -238,5 +238,8 @@ mod tests {
let deserialized = torrent_from_bytes::<ByteBuf>(&bytes).unwrap();
assert_eq!(torrent.info_hash(), deserialized.info_hash);
let deserialized = torrent_from_bytes::<ByteBufOwned>(&bytes).unwrap();
assert_eq!(torrent.info_hash(), deserialized.info_hash);
}
}

4
crates/librqbit/src/dht_utils.rs
View file

@ -1,7 +1,7 @@
use std::{collections::HashSet, net::SocketAddr};
use anyhow::Context;
use buffers::ByteString;
use buffers::ByteBufOwned;
use futures::{stream::FuturesUnordered, Stream, StreamExt};
use librqbit_core::torrent_metainfo::TorrentMetaV1Info;
use tracing::debug;
@ -14,7 +14,7 @@ use librqbit_core::hash_id::Id20;
#[derive(Debug)]
pub enum ReadMetainfoResult<Rx> {
Found {
info: TorrentMetaV1Info<ByteString>,
info: TorrentMetaV1Info<ByteBufOwned>,
rx: Rx,
seen: HashSet<SocketAddr>,
},

14
crates/librqbit/src/file_ops.rs
View file

@ -9,7 +9,7 @@ use std::{
};
use anyhow::Context;
use buffers::ByteString;
use buffers::ByteBufOwned;
use librqbit_core::{
lengths::{ChunkInfo, Lengths, ValidPieceIndex},
torrent_metainfo::{FileIteratorName, TorrentMetaV1Info},
@ -55,7 +55,7 @@ pub fn update_hash_from_file<Sha1: ISha1>(
}
pub(crate) struct FileOps<'a, Sha1> {
torrent: &'a TorrentMetaV1Info<ByteString>,
torrent: &'a TorrentMetaV1Info<ByteBufOwned>,
files: &'a [Arc<Mutex<File>>],
lengths: &'a Lengths,
phantom_data: PhantomData<Sha1>,
@ -63,7 +63,7 @@ pub(crate) struct FileOps<'a, Sha1> {
impl<'a, Sha1Impl: ISha1> FileOps<'a, Sha1Impl> {
pub fn new(
torrent: &'a TorrentMetaV1Info<ByteString>,
torrent: &'a TorrentMetaV1Info<ByteBufOwned>,
files: &'a [Arc<Mutex<File>>],
lengths: &'a Lengths,
) -> Self {
@ -80,8 +80,10 @@ impl<'a, Sha1Impl: ISha1> FileOps<'a, Sha1Impl> {
only_files: Option<&[usize]>,
progress: &AtomicU64,
) -> anyhow::Result<InitialCheckResults> {
let mut needed_pieces = BF::from_vec(vec![0u8; self.lengths.piece_bitfield_bytes()]);
let mut have_pieces = BF::from_vec(vec![0u8; self.lengths.piece_bitfield_bytes()]);
let mut needed_pieces =
BF::from_boxed_slice(vec![0u8; self.lengths.piece_bitfield_bytes()].into());
let mut have_pieces =
BF::from_boxed_slice(vec![0u8; self.lengths.piece_bitfield_bytes()].into());
let mut have_bytes = 0u64;
let mut needed_bytes = 0u64;
@ -92,7 +94,7 @@ impl<'a, Sha1Impl: ISha1> FileOps<'a, Sha1Impl> {
index: usize,
fd: &'a Arc<Mutex<File>>,
len: u64,
name: FileIteratorName<'a, ByteString>,
name: FileIteratorName<'a, ByteBufOwned>,
full_file_required: bool,
processed_bytes: u64,
is_broken: bool,

6
crates/librqbit/src/peer_connection.rs
View file

@ -4,7 +4,7 @@ use std::{
};
use anyhow::{bail, Context};
use buffers::{ByteBuf, ByteString};
use buffers::{ByteBuf, ByteBufOwned};
use clone_to_owned::CloneToOwned;
use librqbit_core::{hash_id::Id20, lengths::ChunkInfo, peer_id::try_decode_peer_id};
use parking_lot::RwLock;
@ -100,7 +100,7 @@ impl<H: PeerConnectionHandler> PeerConnection<H> {
&self,
outgoing_chan: tokio::sync::mpsc::UnboundedReceiver<WriterRequest>,
read_buf: ReadBuf,
handshake: Handshake<ByteString>,
handshake: Handshake<ByteBufOwned>,
mut conn: tokio::net::TcpStream,
) -> anyhow::Result<()> {
use tokio::io::AsyncWriteExt;
@ -220,7 +220,7 @@ impl<H: PeerConnectionHandler> PeerConnection<H> {
.read_write_timeout
.unwrap_or_else(|| Duration::from_secs(10));
let extended_handshake: RwLock<Option<ExtendedHandshake<ByteString>>> = RwLock::new(None);
let extended_handshake: RwLock<Option<ExtendedHandshake<ByteBufOwned>>> = RwLock::new(None);
let extended_handshake_ref = &extended_handshake;
let supports_extended = handshake_supports_extended;

13
crates/librqbit/src/peer_info_reader/mod.rs
View file

@ -1,7 +1,7 @@
use std::net::SocketAddr;
use bencode::from_bytes;
use buffers::{ByteBuf, ByteString};
use buffers::{ByteBuf, ByteBufOwned};
use librqbit_core::{
constants::CHUNK_SIZE,
hash_id::Id20,
@ -30,9 +30,9 @@ pub(crate) async fn read_metainfo_from_peer(
info_hash: Id20,
peer_connection_options: Option<PeerConnectionOptions>,
spawner: BlockingSpawner,
) -> anyhow::Result<TorrentMetaV1Info<ByteString>> {
) -> anyhow::Result<TorrentMetaV1Info<ByteBufOwned>> {
let (result_tx, result_rx) =
tokio::sync::oneshot::channel::<anyhow::Result<TorrentMetaV1Info<ByteString>>>();
tokio::sync::oneshot::channel::<anyhow::Result<TorrentMetaV1Info<ByteBufOwned>>>();
let (writer_tx, writer_rx) = tokio::sync::mpsc::unbounded_channel::<WriterRequest>();
let handler = Handler {
addr,
@ -131,8 +131,9 @@ struct Handler {
addr: SocketAddr,
info_hash: Id20,
writer_tx: UnboundedSender<WriterRequest>,
result_tx:
Mutex<Option<tokio::sync::oneshot::Sender<anyhow::Result<TorrentMetaV1Info<ByteString>>>>>,
result_tx: Mutex<
Option<tokio::sync::oneshot::Sender<anyhow::Result<TorrentMetaV1Info<ByteBufOwned>>>>,
>,
locked: RwLock<Option<HandlerLocked>>,
}
@ -169,7 +170,7 @@ impl PeerConnectionHandler for Handler {
.record_piece(piece, &data, self.info_hash)?;
if piece_ready {
let buf = self.locked.write().take().unwrap().buffer;
let info = from_bytes::<TorrentMetaV1Info<ByteString>>(&buf);
let info = from_bytes::<TorrentMetaV1Info<ByteBufOwned>>(&buf);
self.result_tx
.lock()
.take()

27
crates/librqbit/src/session.rs
View file

@ -21,7 +21,7 @@ use crate::{
};
use anyhow::{bail, Context};
use bencode::{bencode_serialize_to_writer, BencodeDeserializer};
use buffers::{ByteBuf, ByteBufT, ByteString};
use buffers::{ByteBuf, ByteBufOwned, ByteBufT};
use clone_to_owned::CloneToOwned;
use dht::{Dht, DhtBuilder, DhtConfig, Id20, PersistentDht, PersistentDhtConfig};
use futures::{
@ -126,14 +126,17 @@ struct SerializedTorrent {
serialize_with = "serialize_torrent",
deserialize_with = "deserialize_torrent"
)]
info: TorrentMetaV1Info<ByteString>,
info: TorrentMetaV1Info<ByteBufOwned>,
trackers: HashSet<String>,
output_folder: PathBuf,
only_files: Option<Vec<usize>>,
is_paused: bool,
}
fn serialize_torrent<S>(t: &TorrentMetaV1Info<ByteString>, serializer: S) -> Result<S::Ok, S::Error>
fn serialize_torrent<S>(
t: &TorrentMetaV1Info<ByteBufOwned>,
serializer: S,
) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
@ -145,7 +148,7 @@ where
s.serialize(serializer)
}
fn deserialize_torrent<'de, D>(deserializer: D) -> Result<TorrentMetaV1Info<ByteString>, D::Error>
fn deserialize_torrent<'de, D>(deserializer: D) -> Result<TorrentMetaV1Info<ByteBufOwned>, D::Error>
where
D: Deserializer<'de>,
{
@ -155,7 +158,7 @@ where
let b = general_purpose::STANDARD_NO_PAD
.decode(s)
.map_err(D::Error::custom)?;
TorrentMetaV1Info::<ByteString>::deserialize(&mut BencodeDeserializer::new_from_buf(&b))
TorrentMetaV1Info::<ByteBufOwned>::deserialize(&mut BencodeDeserializer::new_from_buf(&b))
.map_err(D::Error::custom)
}
@ -216,7 +219,7 @@ fn compute_only_files_regex<ByteBuf: AsRef<[u8]>>(
}
fn compute_only_files(
info: &TorrentMetaV1Info<ByteString>,
info: &TorrentMetaV1Info<ByteBufOwned>,
only_files: Option<Vec<usize>>,
only_files_regex: Option<String>,
list_only: bool,
@ -303,7 +306,7 @@ pub struct AddTorrentOptions {
pub struct ListOnlyResponse {
pub info_hash: Id20,
pub info: TorrentMetaV1Info<ByteString>,
pub info: TorrentMetaV1Info<ByteBufOwned>,
pub only_files: Option<Vec<usize>>,
pub output_folder: PathBuf,
pub seen_peers: Vec<SocketAddr>,
@ -426,7 +429,7 @@ pub(crate) struct CheckedIncomingConnection {
pub addr: SocketAddr,
pub stream: tokio::net::TcpStream,
pub read_buf: ReadBuf,
pub handshake: Handshake<ByteString>,
pub handshake: Handshake<ByteBufOwned>,
}
impl Session {
@ -715,10 +718,10 @@ impl Session {
serde_json::from_reader(&mut rdr).context("error deserializing session database")?;
let mut futures = Vec::new();
for (id, storrent) in db.torrents.into_iter() {
let trackers: Vec<ByteString> = storrent
let trackers: Vec<ByteBufOwned> = storrent
.trackers
.into_iter()
.map(|t| ByteString(t.into_bytes()))
.map(|t| ByteBufOwned::from(t.into_bytes()))
.collect();
let info = TorrentMetaV1Owned {
announce: trackers.first().cloned(),
@ -929,7 +932,7 @@ impl Session {
fn get_default_subfolder_for_torrent(
&self,
info: &TorrentMetaV1Info<ByteString>,
info: &TorrentMetaV1Info<ByteBufOwned>,
) -> anyhow::Result<Option<PathBuf>> {
let files = info
.iter_filenames_and_lengths()?
@ -957,7 +960,7 @@ impl Session {
async fn main_torrent_info(
&self,
info_hash: Id20,
info: TorrentMetaV1Info<ByteString>,
info: TorrentMetaV1Info<ByteBufOwned>,
trackers: Vec<String>,
peer_rx: Option<PeerStream>,
initial_peers: Vec<SocketAddr>,

8
crates/librqbit/src/torrent_state/live/mod.rs
View file

@ -57,7 +57,7 @@ use std::{
use anyhow::{bail, Context};
use backoff::backoff::Backoff;
use buffers::{ByteBuf, ByteString};
use buffers::{ByteBuf, ByteBufOwned};
use clone_to_owned::CloneToOwned;
use futures::{stream::FuturesUnordered, StreamExt};
use itertools::Itertools;
@ -130,7 +130,7 @@ fn dummy_file() -> anyhow::Result<std::fs::File> {
}
fn make_piece_bitfield(lengths: &Lengths) -> BF {
BF::from_vec(vec![0; lengths.piece_bitfield_bytes()])
BF::from_boxed_slice(vec![0; lengths.piece_bitfield_bytes()].into_boxed_slice())
}
pub(crate) struct TorrentStateLocked {
@ -485,7 +485,7 @@ impl TorrentStateLive {
&self.meta
}
pub fn info(&self) -> &TorrentMetaV1Info<ByteString> {
pub fn info(&self) -> &TorrentMetaV1Info<ByteBufOwned> {
&self.meta.info
}
pub fn info_hash(&self) -> Id20 {
@ -1047,7 +1047,7 @@ impl PeerHandler {
self.on_bitfield_notify.notify_waiters();
}
fn on_bitfield(&self, bitfield: ByteString) -> anyhow::Result<()> {
fn on_bitfield(&self, bitfield: ByteBufOwned) -> anyhow::Result<()> {
if bitfield.len() != self.state.lengths.piece_bitfield_bytes() {
anyhow::bail!(
"dropping peer as its bitfield has unexpected size. Got {}, expected {}",

2
crates/librqbit/src/torrent_state/live/peer/mod.rs
View file

@ -198,7 +198,7 @@ impl LivePeerState {
LivePeerState {
peer_id,
peer_interested: false,
bitfield: BF::new(),
bitfield: BF::default(),
inflight_requests: Default::default(),
tx,
}

4
crates/librqbit/src/torrent_state/live/peers/mod.rs
View file

@ -81,9 +81,9 @@ impl PeerStates {
prev
})
}
pub fn update_bitfield_from_vec(&self, handle: PeerHandle, bitfield: Vec<u8>) -> Option<()> {
pub fn update_bitfield_from_vec(&self, handle: PeerHandle, bitfield: Box<[u8]>) -> Option<()> {
self.with_live_mut(handle, "update_bitfield_from_vec", |live| {
live.bitfield = BF::from_vec(bitfield);
live.bitfield = BF::from_boxed_slice(bitfield);
})
}
pub fn mark_peer_connecting(&self, h: PeerHandle) -> anyhow::Result<(PeerRx, PeerTx)> {

8
crates/librqbit/src/torrent_state/mod.rs
View file

@ -13,7 +13,7 @@ use std::time::Duration;
use anyhow::bail;
use anyhow::Context;
use buffers::ByteString;
use buffers::ByteBufOwned;
use futures::future::BoxFuture;
use futures::FutureExt;
use librqbit_core::hash_id::Id20;
@ -78,7 +78,7 @@ pub(crate) struct ManagedTorrentOptions {
}
pub struct ManagedTorrentInfo {
pub info: TorrentMetaV1Info<ByteString>,
pub info: TorrentMetaV1Info<ByteBufOwned>,
pub info_hash: Id20,
pub out_dir: PathBuf,
pub(crate) spawner: BlockingSpawner,
@ -410,7 +410,7 @@ impl ManagedTorrent {
}
pub struct ManagedTorrentBuilder {
info: TorrentMetaV1Info<ByteString>,
info: TorrentMetaV1Info<ByteBufOwned>,
info_hash: Id20,
output_folder: PathBuf,
force_tracker_interval: Option<Duration>,
@ -425,7 +425,7 @@ pub struct ManagedTorrentBuilder {
impl ManagedTorrentBuilder {
pub fn new<P: AsRef<Path>>(
info: TorrentMetaV1Info<ByteString>,
info: TorrentMetaV1Info<ByteBufOwned>,
info_hash: Id20,
output_folder: P,
) -> Self {

2
crates/librqbit/src/type_aliases.rs
View file

@ -2,7 +2,7 @@ use std::net::SocketAddr;
use futures::stream::BoxStream;
pub type BF = bitvec::vec::BitVec<u8, bitvec::order::Msb0>;
pub type BF = bitvec::boxed::BitBox<u8, bitvec::order::Msb0>;
pub type PeerHandle = SocketAddr;
pub type PeerStream = BoxStream<'static, SocketAddr>;

12
crates/librqbit_core/Cargo.toml
View file

@ -18,16 +18,16 @@ sha1-rust = ["bencode/sha1-rust"]
[dependencies]
tracing = "0.1.40"
tokio = {version = "1", features = ["rt-multi-thread", "macros", "time"]}
tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
hex = "0.4"
anyhow = "1"
url = "2"
uuid = {version = "1", features = ["v4"]}
uuid = { version = "1", features = ["v4"] }
parking_lot = "0.12"
serde = {version = "1", features=["derive"]}
buffers = {path="../buffers", package="librqbit-buffers", version = "2.2.1"}
bencode = {path = "../bencode", default-features=false, package="librqbit-bencode", version="2.2.1"}
clone_to_owned = {path="../clone_to_owned", package="librqbit-clone-to-owned", version = "2.2.1"}
serde = { version = "1", features = ["derive"] }
buffers = { path = "../buffers", package = "librqbit-buffers", version = "2.2.1" }
bencode = { path = "../bencode", default-features = false, package = "librqbit-bencode", version = "2.2.2" }
clone_to_owned = { path = "../clone_to_owned", package = "librqbit-clone-to-owned", version = "2.2.1" }
itertools = "0.12"
directories = "5"
tokio-util = "0.7.10"

4
crates/librqbit_core/src/torrent_metainfo.rs
View file

@ -2,7 +2,7 @@ use std::{iter::once, path::PathBuf};
use anyhow::Context;
use bencode::BencodeDeserializer;
use buffers::{ByteBuf, ByteString};
use buffers::{ByteBuf, ByteBufOwned};
use clone_to_owned::CloneToOwned;
use itertools::Either;
use serde::{Deserialize, Serialize};
@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize};
use crate::hash_id::Id20;
pub type TorrentMetaV1Borrowed<'a> = TorrentMetaV1<ByteBuf<'a>>;
pub type TorrentMetaV1Owned = TorrentMetaV1<ByteString>;
pub type TorrentMetaV1Owned = TorrentMetaV1<ByteBufOwned>;
/// Parse torrent metainfo from bytes.
pub fn torrent_from_bytes<'de, ByteBuf: Deserialize<'de>>(

4
crates/peer_binary_protocol/src/lib.rs
View file

@ -5,7 +5,7 @@
pub mod extended;
use bincode::Options;
use buffers::{ByteBuf, ByteString};
use buffers::{ByteBuf, ByteBufOwned};
use byteorder::{ByteOrder, BE};
use clone_to_owned::CloneToOwned;
use librqbit_core::{constants::CHUNK_SIZE, hash_id::Id20, lengths::ChunkInfo};
@ -183,7 +183,7 @@ pub enum Message<ByteBuf: std::hash::Hash + Eq> {
}
pub type MessageBorrowed<'a> = Message<ByteBuf<'a>>;
pub type MessageOwned = Message<ByteString>;
pub type MessageOwned = Message<ByteBufOwned>;
pub type BitfieldBorrowed<'a> = &'a bitvec::slice::BitSlice<u8, bitvec::order::Lsb0>;
pub type BitfieldOwned = bitvec::vec::BitVec<u8, bitvec::order::Lsb0>;