Switch from async-lock to tokio primitives

crates/subspace-farmer-components/Cargo.toml

@@ -16,7 +16,6 @@ include = [
 bench = false
 
 [dependencies]
-async-lock = "3.3.0"
 async-trait = "0.1.81"
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 bitvec = "1.0.1"

crates/subspace-farmer-components/src/plotting.rs

@@ -12,7 +12,6 @@ use crate::sector::{
 };
 use crate::segment_reconstruction::recover_missing_piece;
 use crate::{FarmerProtocolInfo, PieceGetter};
-use async_lock::Mutex as AsyncMutex;
 use backoff::future::retry;
 use backoff::{Error as BackoffError, ExponentialBackoff};
 use futures::stream::FuturesUnordered;
@@ -34,7 +33,7 @@ use subspace_core_primitives::{
 use subspace_erasure_coding::ErasureCoding;
 use subspace_proof_of_space::{Table, TableGenerator};
 use thiserror::Error;
-use tokio::sync::{AcquireError, Semaphore};
+use tokio::sync::{AcquireError, Mutex as AsyncMutex, Semaphore};
 use tracing::{debug, trace, warn};
 
 const RECONSTRUCTION_CONCURRENCY_LIMIT: usize = 1;
@@ -397,7 +396,7 @@ where
 
                         loop {
                             // Take mutex briefly to make sure encoding is allowed right now
-                            global_mutex.lock_blocking();
+                            let _ = global_mutex.blocking_lock();
 
                             // This instead of `while` above because otherwise mutex will be held
                             // for the duration of the loop and will limit concurrency to 1 record

crates/subspace-farmer/Cargo.toml

@@ -17,7 +17,6 @@ required-features = ["binary"]
 
 [dependencies]
 anyhow = "1.0.86"
-async-lock = "3.3.0"
 async-nats = { version = "0.35.1", optional = true }
 async-trait = "0.1.81"
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
@@ -65,7 +64,7 @@ supports-color = { version = "3.0.0", optional = true }
 tempfile = "3.12.0"
 thiserror = "1.0.63"
 thread-priority = "1.1.0"
-tokio = { version = "1.39.2", features = ["macros", "parking_lot", "rt-multi-thread", "signal", "time"] }
+tokio = { version = "1.39.2", features = ["macros", "parking_lot", "rt-multi-thread", "signal", "sync", "time"] }
 tokio-stream = { version = "0.1.15", features = ["sync"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"], optional = true }

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/controller.rs

@@ -6,7 +6,6 @@ use crate::commands::cluster::controller::farms::{maintain_farms, FarmIndex};
 use crate::commands::shared::derive_libp2p_keypair;
 use crate::commands::shared::network::{configure_network, NetworkArgs};
 use anyhow::anyhow;
-use async_lock::RwLock as AsyncRwLock;
 use backoff::ExponentialBackoff;
 use clap::{Parser, ValueHint};
 use futures::stream::FuturesUnordered;
@@ -31,6 +30,7 @@ use subspace_farmer::node_client::NodeClient;
 use subspace_farmer::single_disk_farm::identity::Identity;
 use subspace_farmer::utils::{run_future_in_dedicated_thread, AsyncJoinOnDrop};
 use subspace_networking::utils::piece_provider::PieceProvider;
+use tokio::sync::RwLock as AsyncRwLock;
 use tracing::info;
 
 /// Get piece retry attempts number.

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/controller/farms.rs

@@ -6,7 +6,6 @@
 
 use crate::commands::cluster::farmer::FARMER_IDENTIFICATION_BROADCAST_INTERVAL;
 use anyhow::anyhow;
-use async_lock::RwLock as AsyncRwLock;
 use futures::channel::oneshot;
 use futures::future::FusedFuture;
 use futures::stream::FuturesUnordered;
@@ -25,6 +24,7 @@ use subspace_farmer::cluster::farmer::{ClusterFarm, ClusterFarmerIdentifyFarmBro
 use subspace_farmer::cluster::nats_client::NatsClient;
 use subspace_farmer::farm::plotted_pieces::PlottedPieces;
 use subspace_farmer::farm::{Farm, FarmId, SectorPlottingDetails, SectorUpdate};
+use tokio::sync::RwLock as AsyncRwLock;
 use tokio::task;
 use tokio::time::MissedTickBehavior;
 use tracing::{error, info, trace, warn};
@@ -178,7 +178,7 @@ pub(super) async fn maintain_farms(
                     let plotted_pieces = Arc::clone(plotted_pieces);
 
                     let delete_farm_fut = task::spawn_blocking(move || {
-                        plotted_pieces.write_blocking().delete_farm(farm_index);
+                        plotted_pieces.blocking_write().delete_farm(farm_index);
                     });
                     if let Err(error) = delete_farm_fut.await {
                         error!(
@@ -235,7 +235,7 @@ pub(super) async fn maintain_farms(
                         let plotted_pieces = Arc::clone(plotted_pieces);
 
                         let delete_farm_fut = task::spawn_blocking(move || {
-                            plotted_pieces.write_blocking().delete_farm(farm_index);
+                            plotted_pieces.blocking_write().delete_farm(farm_index);
                         });
                         if let Err(error) = delete_farm_fut.await {
                             error!(
@@ -322,7 +322,7 @@ fn process_farm_identify_message<'a>(
             let plotted_pieces = Arc::clone(plotted_pieces);
 
             let delete_farm_fut = task::spawn_blocking(move || {
-                plotted_pieces.write_blocking().delete_farm(farm_index);
+                plotted_pieces.blocking_write().delete_farm(farm_index);
             });
             if let Err(error) = delete_farm_fut.await {
                 error!(
@@ -434,7 +434,7 @@ async fn initialize_farm(
     drop(sector_update_handler);
     let plotted_sectors_buffer = mem::take(&mut *plotted_sectors_buffer.lock());
     let add_buffered_sectors_fut = task::spawn_blocking(move || {
-        let mut plotted_pieces = plotted_pieces.write_blocking();
+        let mut plotted_pieces = plotted_pieces.blocking_write();
 
         for (plotted_sector, old_plotted_sector) in plotted_sectors_buffer {
             if let Some(old_plotted_sector) = old_plotted_sector {

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/farmer.rs

@@ -2,7 +2,6 @@
 
 use crate::commands::shared::DiskFarm;
 use anyhow::anyhow;
-use async_lock::Mutex as AsyncMutex;
 use backoff::ExponentialBackoff;
 use bytesize::ByteSize;
 use clap::Parser;
@@ -35,7 +34,7 @@ use subspace_farmer::utils::{
 };
 use subspace_farmer_components::reading::ReadSectorRecordChunksMode;
 use subspace_proof_of_space::Table;
-use tokio::sync::{Barrier, Semaphore};
+use tokio::sync::{Barrier, Mutex as AsyncMutex, Semaphore};
 use tracing::{error, info, info_span, warn, Instrument};
 
 const FARM_ERROR_PRINT_INTERVAL: Duration = Duration::from_secs(30);

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/plotter.rs

@@ -1,6 +1,5 @@
 use crate::commands::shared::PlottingThreadPriority;
 use anyhow::anyhow;
-use async_lock::Mutex as AsyncMutex;
 use clap::Parser;
 use futures::{select, FutureExt};
 use prometheus_client::registry::Registry;
@@ -27,7 +26,7 @@ use subspace_farmer::utils::{
 };
 use subspace_farmer_components::PieceGetter;
 use subspace_proof_of_space::Table;
-use tokio::sync::Semaphore;
+use tokio::sync::{Mutex as AsyncMutex, Semaphore};
 use tracing::info;
 
 const PLOTTING_RETRY_INTERVAL: Duration = Duration::from_secs(5);

crates/subspace-farmer/src/bin/subspace-farmer/commands/farm.rs

@@ -2,7 +2,6 @@ use crate::commands::shared::network::{configure_network, NetworkArgs};
 use crate::commands::shared::{derive_libp2p_keypair, DiskFarm, PlottingThreadPriority};
 use crate::utils::shutdown_signal;
 use anyhow::anyhow;
-use async_lock::{Mutex as AsyncMutex, RwLock as AsyncRwLock};
 use backoff::ExponentialBackoff;
 use bytesize::ByteSize;
 use clap::{Parser, ValueHint};
@@ -51,7 +50,8 @@ use subspace_farmer_components::PieceGetter;
 use subspace_metrics::{start_prometheus_metrics_server, RegistryAdapter};
 use subspace_networking::utils::piece_provider::PieceProvider;
 use subspace_proof_of_space::Table;
-use tokio::sync::{Barrier, Semaphore};
+use tokio::sync::{Barrier, Mutex as AsyncMutex, RwLock as AsyncRwLock, Semaphore};
+use tokio::task;
 use tracing::{error, info, info_span, warn, Instrument};
 
 /// Get piece retry attempts number.
@@ -713,7 +713,8 @@ where
                 {
                     let _span_guard = span.enter();
 
-                    let mut plotted_pieces = plotted_pieces.write_blocking();
+                    let mut plotted_pieces =
+                        task::block_in_place(|| plotted_pieces.blocking_write());
 
                     if let Some(old_plotted_sector) = &old_plotted_sector {
                         plotted_pieces.delete_sector(farm_index, old_plotted_sector);

crates/subspace-farmer/src/bin/subspace-farmer/commands/shared/network.rs

@@ -1,4 +1,3 @@
-use async_lock::RwLock as AsyncRwLock;
 use clap::Parser;
 use prometheus_client::registry::Registry;
 use std::collections::HashSet;
@@ -22,6 +21,7 @@ use subspace_networking::{
     SegmentHeaderBySegmentIndexesRequestHandler, SegmentHeaderRequest, SegmentHeaderResponse,
 };
 use subspace_rpc_primitives::MAX_SEGMENT_HEADERS_PER_REQUEST;
+use tokio::sync::RwLock as AsyncRwLock;
 use tracing::{debug, error, info, Instrument};
 
 /// How many segment headers can be requested at a time.
@@ -139,7 +139,8 @@ where
 
                         let read_piece_fut = match weak_plotted_pieces.upgrade() {
                             Some(plotted_pieces) => plotted_pieces
-                                .try_read()?
+                                .try_read()
+                                .ok()?
                                 .read_piece(piece_index)?
                                 .in_current_span(),
                             None => {

crates/subspace-farmer/src/cluster/controller.rs

@@ -14,7 +14,6 @@ use crate::farm::{PieceCacheId, PieceCacheOffset};
 use crate::farmer_cache::FarmerCache;
 use crate::node_client::{Error as NodeClientError, NodeClient};
 use anyhow::anyhow;
-use async_lock::Semaphore;
 use async_nats::HeaderValue;
 use async_trait::async_trait;
 use futures::{select, FutureExt, Stream, StreamExt};
@@ -29,6 +28,7 @@ use subspace_farmer_components::PieceGetter;
 use subspace_rpc_primitives::{
     FarmerAppInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
+use tokio::sync::Semaphore;
 use tracing::{debug, trace, warn};
 
 /// Broadcast sent by controllers requesting farmers to identify themselves

crates/subspace-farmer/src/farmer_cache.rs

@@ -13,7 +13,6 @@ use crate::farmer_cache::metrics::FarmerCacheMetrics;
 use crate::farmer_cache::piece_cache_state::PieceCachesState;
 use crate::node_client::NodeClient;
 use crate::utils::run_future_in_dedicated_thread;
-use async_lock::RwLock as AsyncRwLock;
 use event_listener_primitives::{Bag, HandlerId};
 use futures::stream::{FuturesOrdered, FuturesUnordered};
 use futures::{select, FutureExt, StreamExt};
@@ -32,7 +31,7 @@ use subspace_networking::libp2p::PeerId;
 use subspace_networking::utils::multihash::ToMultihash;
 use subspace_networking::{KeyWithDistance, LocalRecordProvider};
 use tokio::runtime::Handle;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, RwLock as AsyncRwLock};
 use tokio::task::{block_in_place, yield_now};
 use tracing::{debug, error, info, trace, warn};
 
@@ -1208,7 +1207,8 @@ where
         let distance_key = KeyWithDistance::new(self.peer_id, key.clone());
         if self
             .piece_caches
-            .try_read()?
+            .try_read()
+            .ok()?
             .contains_stored_piece(&distance_key)
         {
             // Note: We store our own provider records locally without local addresses
@@ -1225,7 +1225,8 @@ where
         let found_fut = self
             .plot_caches
             .caches
-            .try_read()?
+            .try_read()
+            .ok()?
             .iter()
             .map(|plot_cache| {
                 let plot_cache = Arc::clone(plot_cache);

crates/subspace-farmer/src/farmer_piece_getter.rs

@@ -3,9 +3,6 @@
 use crate::farm::plotted_pieces::PlottedPieces;
 use crate::farmer_cache::FarmerCache;
 use crate::node_client::NodeClient;
-use async_lock::{
-    Mutex as AsyncMutex, MutexGuardArc as AsyncMutexGuardArc, RwLock as AsyncRwLock, Semaphore,
-};
 use async_trait::async_trait;
 use backoff::backoff::Backoff;
 use backoff::future::retry;
@@ -22,6 +19,9 @@ use subspace_core_primitives::{Piece, PieceIndex};
 use subspace_farmer_components::PieceGetter;
 use subspace_networking::utils::multihash::ToMultihash;
 use subspace_networking::utils::piece_provider::{PieceProvider, PieceValidator};
+use tokio::sync::{
+    Mutex as AsyncMutex, OwnedMutexGuard as AsyncOwnedMutexGuard, RwLock as AsyncRwLock, Semaphore,
+};
 use tracing::{debug, error, trace};
 
 pub mod piece_validator;
@@ -30,7 +30,7 @@ const MAX_RANDOM_WALK_ROUNDS: usize = 15;
 
 struct InProgressPieceGetting<'a> {
     piece_index: PieceIndex,
-    in_progress_piece: AsyncMutexGuardArc<Option<Piece>>,
+    in_progress_piece: AsyncOwnedMutexGuard<Option<Piece>>,
     in_progress_pieces: &'a Mutex<HashMap<PieceIndex, Arc<AsyncMutex<Option<Piece>>>>>,
 }
 
@@ -63,8 +63,8 @@ impl<'a> InProgressPiece<'a> {
         // Take lock before anything else, set to `None` when another piece getting is already in
         // progress
         let mut local_in_progress_piece_guard = Some(
-            in_progress_piece_mutex
-                .try_lock_arc()
+            Arc::clone(&in_progress_piece_mutex)
+                .try_lock_owned()
                 .expect("Just created; qed"),
         );
         let in_progress_piece_mutex = in_progress_pieces
@@ -282,6 +282,7 @@ where
         let maybe_read_piece_fut = inner
             .plotted_pieces
             .try_read()
+            .ok()
             .and_then(|plotted_pieces| plotted_pieces.read_piece(piece_index));
 
         if let Some(read_piece_fut) = maybe_read_piece_fut {

crates/subspace-farmer/src/node_client/caching_proxy_node_client.rs

@@ -3,7 +3,6 @@
 
 use crate::node_client::{Error as RpcError, Error, NodeClient, NodeClientExt};
 use crate::utils::AsyncJoinOnDrop;
-use async_lock::{Mutex as AsyncMutex, RwLock as AsyncRwLock};
 use async_trait::async_trait;
 use futures::{select, FutureExt, Stream, StreamExt};
 use std::pin::Pin;
@@ -14,7 +13,7 @@ use subspace_rpc_primitives::{
     FarmerAppInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
     MAX_SEGMENT_HEADERS_PER_REQUEST,
 };
-use tokio::sync::watch;
+use tokio::sync::{watch, Mutex as AsyncMutex, RwLock as AsyncRwLock};
 use tokio_stream::wrappers::WatchStream;
 use tracing::{info, trace, warn};
 

crates/subspace-farmer/src/node_client/rpc_node_client.rs

@@ -1,7 +1,6 @@
 //! Node client implementation that connects to node via RPC (WebSockets)
 
 use crate::node_client::{Error as RpcError, Error, NodeClient, NodeClientExt};
-use async_lock::Semaphore;
 use async_trait::async_trait;
 use futures::{Stream, StreamExt};
 use jsonrpsee::core::client::{ClientT, Error as JsonError, SubscriptionClientT};
@@ -13,6 +12,7 @@ use subspace_core_primitives::{Piece, PieceIndex, SegmentHeader, SegmentIndex};
 use subspace_rpc_primitives::{
     FarmerAppInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
+use tokio::sync::Semaphore;
 
 /// TODO: Node is having a hard time responding for many piece requests, specifically this results
 ///  in subscriptions become broken on the node: https://github.com/paritytech/jsonrpsee/issues/1409

crates/subspace-farmer/src/plotter/cpu.rs

@@ -6,7 +6,6 @@ use crate::plotter::cpu::metrics::CpuPlotterMetrics;
 use crate::plotter::{Plotter, SectorPlottingProgress};
 use crate::thread_pool_manager::PlottingThreadPoolManager;
 use crate::utils::AsyncJoinOnDrop;
-use async_lock::Mutex as AsyncMutex;
 use async_trait::async_trait;
 use event_listener_primitives::{Bag, HandlerId};
 use futures::channel::mpsc;
@@ -32,7 +31,7 @@ use subspace_farmer_components::plotting::{
 };
 use subspace_farmer_components::{FarmerProtocolInfo, PieceGetter};
 use subspace_proof_of_space::Table;
-use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+use tokio::sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore};
 use tokio::task::yield_now;
 use tracing::{warn, Instrument};
 
@@ -302,7 +301,7 @@ where
                     }
 
                     // Take mutex briefly to make sure plotting is allowed right now
-                    global_mutex.lock().await;
+                    let _ = global_mutex.lock().await;
 
                     let downloading_start = Instant::now();