perf: reduce memory usage

crates/libtortillas/examples/live.rs

@@ -1,10 +1,15 @@
-use std::path::PathBuf;
+use std::{
+   path::PathBuf,
+   time::{Duration, Instant},
+};
 
 use libtortillas::prelude::{
    Engine, EngineEventKind, EventStreamError, TorrentEventKind, TorrentSource, TorrentState,
 };
 use tracing::{error, info, warn};
 
+const METRICS_LOG_INTERVAL: Duration = Duration::from_secs(1);
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt()
@@ -22,16 +27,43 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let engine = Engine::default();
    let mut listener = engine.listener();
    let event_task = tokio::spawn(async move {
+      let mut last_metrics_log = None;
       loop {
          match listener.recv().await {
             Ok(event) => {
                let view = listener.view();
-               info!(
-                  sequence = event.sequence,
-                  torrent_count = view.torrent_count(),
-                  ?event.kind,
-                  "received an engine event"
-               );
+               match &event.kind {
+                  EngineEventKind::Torrent {
+                     torrent,
+                     event: TorrentEventKind::MetricsChanged(metrics),
+                  } => {
+                     let complete = metrics
+                        .progress
+                        .remaining_bytes
+                        .is_some_and(|bytes| bytes.0 == 0);
+                     let now = Instant::now();
+                     if complete
+                        || last_metrics_log
+                           .is_none_or(|last| now.duration_since(last) >= METRICS_LOG_INTERVAL)
+                     {
+                        last_metrics_log = Some(now);
+                        info!(
+                           sequence = event.sequence,
+                           torrent_id = %torrent.info_hash(),
+                           downloaded_bytes = metrics.traffic.totals.downloaded.0,
+                           verified_bytes = metrics.progress.verified_bytes.0,
+                           total_bytes = ?metrics.progress.total_bytes.map(|bytes| bytes.0),
+                           "received torrent metrics"
+                        );
+                     }
+                  }
+                  _ => info!(
+                     sequence = event.sequence,
+                     torrent_count = view.torrent_count(),
+                     ?event.kind,
+                     "received an engine event"
+                  ),
+               }
                if matches!(event.kind, EngineEventKind::Shutdown(_)) {
                   break;
                }

crates/libtortillas/examples/peer_transfer.rs

@@ -0,0 +1,65 @@
+use std::{error::Error, time::Instant};
+
+use bytes::Bytes;
+use libtortillas::protocol::{
+   messages::PeerMessages,
+   stream::{PeerRecv, PeerSend, PeerStream},
+};
+use tokio::net::{TcpListener, TcpStream};
+use tracing::info;
+
+const BLOCK_LENGTH: usize = 16 * 1024;
+const BLOCK_COUNT: usize = 8 * 1024;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+   tracing_subscriber::fmt().init();
+
+   let listener = TcpListener::bind("127.0.0.1:0").await?;
+   let address = listener.local_addr()?;
+   let receiver = tokio::spawn(async move {
+      let (stream, _) = listener.accept().await?;
+      let mut stream = PeerStream::tcp(stream);
+      let mut bytes_received = 0usize;
+
+      for expected_index in 0..BLOCK_COUNT {
+         match stream.recv().await? {
+            PeerMessages::Piece(index, 0, block)
+               if index == expected_index as u32
+                  && block.len() == BLOCK_LENGTH
+                  && block.iter().all(|byte| *byte == 0xa5) =>
+            {
+               bytes_received += block.len();
+            }
+            message => {
+               return Err(format!("received unexpected message: {message}").into());
+            }
+         }
+      }
+
+      Ok::<_, Box<dyn Error + Send + Sync>>(bytes_received)
+   });
+
+   let block = Bytes::from(vec![0xa5; BLOCK_LENGTH]);
+   let mut sender = PeerStream::tcp(TcpStream::connect(address).await?);
+   let started_at = Instant::now();
+
+   for index in 0..BLOCK_COUNT {
+      sender
+         .send(PeerMessages::Piece(index as u32, 0, block.clone()))
+         .await?;
+   }
+
+   let bytes_transferred = receiver.await??;
+   let elapsed = started_at.elapsed();
+   let mebibytes = bytes_transferred as f64 / (1024.0 * 1024.0);
+   let throughput = mebibytes / elapsed.as_secs_f64();
+   info!(
+      bytes_transferred,
+      ?elapsed,
+      throughput_mib_per_second = throughput,
+      "completed local peer transfer"
+   );
+
+   Ok(())
+}

crates/libtortillas/src/engine/messages.rs

@@ -85,7 +85,7 @@ pub(crate) mod commands {
             return;
          }
 
-         let info_hash = *handshake.info_hash;
+         let info_hash = handshake.info_hash;
          let mut peer = Peer::from_socket_addr(peer_addr);
 
          // Populate peer fields from parsed handshake.
@@ -151,7 +151,7 @@ pub(crate) mod commands {
          Ok(torrent)
       }
 
-      /// Creates a new [`Torrent`](crate::torrent::Torrent) actor.
+      /// Creates a new [`Torrent`] actor.
       #[message]
       pub(crate) async fn create_torrent(
          &mut self, request: CreateTorrentRequest,

crates/libtortillas/src/lib.rs

@@ -194,6 +194,7 @@
 //! transport-agnostic live views and event streams. Durable state is
 //! represented by [`EngineSnapshot`](engine::EngineSnapshot) and
 //! [`TorrentSnapshot`](torrent::TorrentSnapshot), never by live views.
+
 // `cfg!` type-checks both branches; this drops disabled live code before name
 // resolution.
 macro_rules! live_only {
@@ -263,6 +264,7 @@ pub(crate) mod testing {
    pub(crate) const BIG_BUCK_BUNNY_NAME: &str = "Big Buck Bunny";
    pub(crate) const BIG_BUCK_BUNNY_INFO_HASH: &str = "dd8255ecdc7ca55fb0bbf81323d87062db1f6d1c";
    pub(crate) const BIG_BUCK_BUNNY_TORRENT_FILE: &str = "big-buck-bunny.torrent";
+   pub(crate) const WIRED_CD_TORRENT_FILE: &str = "wired-cd.torrent";
    pub(crate) const KNOPPIX_TORRENT_FILE: &str = "KNOPPIX_V9.1DVD-2021-01-25-EN.torrent";
 
    pub(crate) fn fixture_path(relative_path: &str) -> PathBuf {
@@ -562,7 +564,7 @@ pub(crate) mod testing {
       let handshake = stream.recv_handshake_message().await?;
       handshakes.lock().await.push(handshake.clone());
 
-      let response = Handshake::new(handshake.info_hash.clone(), peer_id);
+      let response = Handshake::new(handshake.info_hash, peer_id);
       stream.write_all(&response.to_bytes()).await?;
 
       for message in messages.iter() {
@@ -590,7 +592,7 @@ pub(crate) mod testing {
 
    #[cfg(test)]
    mod tests {
-      use std::{net::Ipv4Addr, sync::Arc};
+      use std::net::Ipv4Addr;
 
       use tokio::time::{Duration, timeout};
 
@@ -623,11 +625,8 @@ pub(crate) mod testing {
          let mut stream = PeerStream::connect(local_peer.peer().socket_addr(), None)
             .await
             .unwrap();
-         let info_hash = Arc::new(test_info_hash());
-         stream
-            .send_handshake(peer_id(), info_hash.clone())
-            .await
-            .unwrap();
+         let info_hash = test_info_hash();
+         stream.send_handshake(peer_id(), info_hash).await.unwrap();
 
          let (received_peer_id, _) = stream.recv_handshake().await.unwrap();
          let message = timeout(Duration::from_secs(1), stream.recv())

crates/libtortillas/src/metainfo/file.rs

@@ -40,10 +40,12 @@ impl TorrentFile {
    }
 
    pub fn announce_list(&self) -> Vec<Tracker> {
-      let mut announce_list: Vec<Tracker> = self.announce.clone().into_iter().collect();
-      if let Some(list) = self.announce_list.clone() {
-         for tracker in list.into_iter().flatten() {
-            announce_list.push(tracker);
+      let mut announce_list: Vec<Tracker> = self.announce.iter().cloned().collect();
+      if let Some(list) = &self.announce_list {
+         for tracker in list.iter().flatten() {
+            if !announce_list.contains(tracker) {
+               announce_list.push(tracker.clone());
+            }
          }
       }
       announce_list
@@ -185,4 +187,21 @@ mod tests {
       assert!(torrent.announce.is_none());
       assert!(torrent.announce_list().is_empty());
    }
+
+   #[tokio::test]
+   async fn torrent_file_when_announce_is_repeated_then_returns_it_once() {
+      let metainfo = testing::read_torrent_fixture(testing::WIRED_CD_TORRENT_FILE).await;
+      let MetaInfo::Torrent(torrent) = metainfo else {
+         panic!("Expected Torrent");
+      };
+      let trackers = torrent.announce_list();
+
+      assert_eq!(
+         trackers
+            .iter()
+            .filter(|tracker| Some(*tracker) == torrent.announce.as_ref())
+            .count(),
+         1
+      );
+   }
 }

crates/libtortillas/src/peer/actor.rs

@@ -584,6 +584,16 @@ impl Message<PeerMessages> for PeerActor {
       &mut self, msg: PeerMessages, _: &mut KameoContext<Self, Self::Reply>,
    ) -> Self::Reply {
       self.peer.update_last_message_received();
+      #[cfg(feature = "live")]
+      let publish_live_state = matches!(
+         &msg,
+         PeerMessages::Choke
+            | PeerMessages::Unchoke
+            | PeerMessages::Interested
+            | PeerMessages::NotInterested
+            | PeerMessages::Have(_)
+            | PeerMessages::Bitfield(_)
+      );
       match msg {
          PeerMessages::Piece(index, offset, data) => {
             trace!(
@@ -742,7 +752,8 @@ impl Message<PeerMessages> for PeerActor {
             warn!("Received unexpected handshake from peer");
          }
       }
-      crate::live_only! {
+      #[cfg(feature = "live")]
+      if publish_live_state {
          let samples = self.live_handle.view().metrics.transfer.samples;
          self
             .live_handle