feat(txpool|p2p): use seqlock instead of small copy-able RwLocks

CHANGELOG.md

@@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2416](https://github.com/FuelLabs/fuel-core/issues/2416): Define the `GasPriceServiceV1` task.
 - [2033](https://github.com/FuelLabs/fuel-core/pull/2033): Remove `Option<BlockHeight>` in favor of `BlockHeightQuery` where applicable.
 - [2472](https://github.com/FuelLabs/fuel-core/pull/2472): Added the `amountU128` field to the `Balance` GraphQL schema, providing the total balance as a `U128`. The existing `amount` field clamps any balance exceeding `U64` to `u64::MAX`.
+- [2524](https://github.com/FuelLabs/fuel-core/pull/2524): Adds a new lock type which is optimized for certain workloads to the txpool and p2p services.
 
 ### Fixed
 - [2365](https://github.com/FuelLabs/fuel-core/pull/2365): Fixed the error during dry run in the case of race condition.

crates/services/p2p/src/config.rs

@@ -6,6 +6,12 @@ use crate::{
 };
 use fuel_core_types::blockchain::consensus::Genesis;
 
+use self::{
+    connection_tracker::ConnectionTracker,
+    fuel_authenticated::FuelAuthenticated,
+    fuel_upgrade::Checksum,
+};
+use fuel_core_services::seqlock::SeqLock;
 use libp2p::{
     gossipsub,
     identity::{
@@ -22,18 +28,9 @@ use std::{
         IpAddr,
         Ipv4Addr,
     },
-    sync::{
-        Arc,
-        RwLock,
-    },
+    sync::Arc,
     time::Duration,
 };
-
-use self::{
-    connection_tracker::ConnectionTracker,
-    fuel_authenticated::FuelAuthenticated,
-    fuel_upgrade::Checksum,
-};
 mod connection_tracker;
 mod fuel_authenticated;
 pub(crate) mod fuel_upgrade;
@@ -260,7 +257,7 @@ pub(crate) fn build_transport_function(
     p2p_config: &Config,
 ) -> (
     impl FnOnce(&Keypair) -> Result<FuelAuthenticated<ConnectionTracker>, ()> + '_,
-    Arc<RwLock<ConnectionState>>,
+    Arc<SeqLock<ConnectionState>>,
 ) {
     let connection_state = ConnectionState::new();
     let kept_connection_state = connection_state.clone();

crates/services/p2p/src/config/connection_tracker.rs

@@ -3,30 +3,28 @@ use super::{
     peer_ids_set_from,
 };
 use crate::peer_manager::ConnectionState;
+use fuel_core_services::seqlock::SeqLock;
 use libp2p::{
     Multiaddr,
     PeerId,
 };
 use std::{
     collections::HashSet,
-    sync::{
-        Arc,
-        RwLock,
-    },
+    sync::Arc,
 };
 
 /// A `ConnectionTracker` allows either Reserved Peers or other peers if there is an available slot.
 /// It is synced with `PeerManager` which keeps track of the `ConnectionState`.
 #[derive(Debug, Clone)]
 pub(crate) struct ConnectionTracker {
     reserved_nodes: HashSet<PeerId>,
-    connection_state: Option<Arc<RwLock<ConnectionState>>>,
+    connection_state: Option<Arc<SeqLock<ConnectionState>>>,
 }
 
 impl ConnectionTracker {
     pub(crate) fn new(
         reserved_nodes: &[Multiaddr],
-        connection_state: Option<Arc<RwLock<ConnectionState>>>,
+        connection_state: Option<Arc<SeqLock<ConnectionState>>>,
     ) -> Self {
         Self {
             reserved_nodes: peer_ids_set_from(reserved_nodes),
@@ -42,9 +40,7 @@ impl Approver for ConnectionTracker {
         }
 
         if let Some(connection_state) = &self.connection_state {
-            if let Ok(connection_state) = connection_state.read() {
-                return connection_state.available_slot()
-            }
+            return connection_state.read().available_slot();
         }
 
         false

crates/services/p2p/src/peer_manager.rs

@@ -1,3 +1,8 @@
+use crate::{
+    gossipsub_config::GRAYLIST_THRESHOLD,
+    peer_manager::heartbeat_data::HeartbeatData,
+};
+use fuel_core_services::seqlock::SeqLock;
 use fuel_core_types::{
     fuel_types::BlockHeight,
     services::p2p::peer_reputation::{
@@ -18,21 +23,13 @@ use std::{
         HashMap,
         HashSet,
     },
-    sync::{
-        Arc,
-        RwLock,
-    },
+    sync::Arc,
 };
 use tracing::{
     debug,
     info,
 };
 
-use crate::{
-    gossipsub_config::GRAYLIST_THRESHOLD,
-    peer_manager::heartbeat_data::HeartbeatData,
-};
-
 pub mod heartbeat_data;
 
 /// At this point we better just ban the peer
@@ -65,7 +62,7 @@ pub struct PeerManager {
     non_reserved_connected_peers: HashMap<PeerId, PeerInfo>,
     reserved_connected_peers: HashMap<PeerId, PeerInfo>,
     reserved_peers: HashSet<PeerId>,
-    connection_state: Arc<RwLock<ConnectionState>>,
+    connection_state: Arc<SeqLock<ConnectionState>>,
     max_non_reserved_peers: usize,
     reserved_peers_updates: tokio::sync::broadcast::Sender<usize>,
 }
@@ -74,7 +71,7 @@ impl PeerManager {
     pub fn new(
         reserved_peers_updates: tokio::sync::broadcast::Sender<usize>,
         reserved_peers: HashSet<PeerId>,
-        connection_state: Arc<RwLock<ConnectionState>>,
+        connection_state: Arc<SeqLock<ConnectionState>>,
         max_non_reserved_peers: usize,
     ) -> Self {
         Self {
@@ -210,9 +207,9 @@ impl PeerManager {
             {
                 // since all the slots were full prior to this disconnect
                 // let's allow new peer non-reserved peers connections
-                if let Ok(mut connection_state) = self.connection_state.write() {
-                    connection_state.allow_new_peers();
-                }
+                self.connection_state.write(|data| {
+                    data.allow_new_peers();
+                });
             }
 
             false
@@ -258,9 +255,9 @@ impl PeerManager {
                 == self.max_non_reserved_peers
             {
                 // this is the last non-reserved peer allowed
-                if let Ok(mut connection_state) = self.connection_state.write() {
-                    connection_state.deny_new_peers();
-                }
+                self.connection_state.write(|data| {
+                    data.deny_new_peers();
+                });
             }
 
             self.non_reserved_connected_peers
@@ -313,8 +310,8 @@ pub struct ConnectionState {
 }
 
 impl ConnectionState {
-    pub fn new() -> Arc<RwLock<Self>> {
-        Arc::new(RwLock::new(Self {
+    pub fn new() -> Arc<SeqLock<Self>> {
+        Arc::new(SeqLock::new(Self {
             peers_allowed: true,
         }))
     }

crates/services/src/lib.rs

@@ -6,6 +6,7 @@
 #![deny(warnings)]
 
 mod async_processor;
+pub mod seqlock;
 mod service;
 mod state;
 mod sync;

crates/services/src/seqlock.rs

@@ -0,0 +1,126 @@
+//! A simple implementation of a sequential lock.
+//! More details: <https://docs.kernel.org/locking/seqlock.html>
+//! Optimized for occasional writes and frequent reads
+
+use std::{
+    cell::UnsafeCell,
+    sync::atomic::{
+        AtomicU64,
+        Ordering,
+    },
+};
+
+/// A simple implementation of a sequential lock.
+/// some usage of unsafe, T must be Copy
+#[derive(Debug)]
+pub struct SeqLock<T> {
+    sequence: AtomicU64,
+    data: UnsafeCell<T>,
+}
+
+unsafe impl<T: Send> Sync for SeqLock<T> {}
+
+impl<T> SeqLock<T> {
+    /// Create a new SeqLock with the given data
+    pub fn new(data: T) -> Self {
+        Self {
+            sequence: AtomicU64::new(0),
+            data: UnsafeCell::new(data),
+        }
+    }
+
+    /// Write the data
+    pub fn write<F>(&self, f: F)
+    where
+        F: FnOnce(&mut T),
+    {
+        // starting guard
+        self.sequence.fetch_add(1, Ordering::Release);
+
+        // Modify the data
+        unsafe {
+            f(&mut *self.data.get());
+        }
+
+        // ending guard
+        self.sequence.fetch_add(1, Ordering::Release);
+    }
+
+    /// Read the data
+    pub fn read(&self) -> T
+    where
+        T: Copy,
+    {
+        loop {
+            // check starting guard
+            let start = self.sequence.load(Ordering::Acquire);
+
+            // if odd, write in progress
+            if start % 2 != 0 {
+                continue;
+            }
+
+            let data = unsafe { *self.data.get() };
+
+            // check starting/ending guard
+            let end = self.sequence.load(Ordering::Acquire);
+
+            // if value changed, retry
+            if start == end && start % 2 == 0 {
+                return data;
+            }
+        }
+    }
+}
+
+#[allow(non_snake_case)]
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{
+        sync::Arc,
+        thread,
+    };
+
+    #[test]
+    fn test_seqlock__provides_correct_values_in_order() {
+        let lock = Arc::new(SeqLock::new(42));
+        let iterations = 100;
+
+        let writer = {
+            let lock = lock.clone();
+            thread::spawn(move || {
+                for i in 0..iterations {
+                    lock.write(|data| *data = i);
+                }
+            })
+        };
+
+        let reader = {
+            let lock = lock.clone();
+            thread::spawn(move || {
+                let seen = 0;
+
+                for _ in 0..iterations {
+                    let value = lock.read();
+                    assert!(value >= seen);
+                }
+            })
+        };
+
+        writer.join().unwrap();
+        reader.join().unwrap();
+    }
+
+    #[test]
+    fn test_seqlock__single_threaded() {
+        let lock = SeqLock::new(42);
+
+        lock.write(|data| {
+            *data = 100;
+        });
+
+        let value = lock.read();
+        assert_eq!(value, 100);
+    }
+}

crates/services/txpool_v2/src/service.rs

@@ -5,6 +5,7 @@ use fuel_core_services::TaskNextAction;
 
 use fuel_core_metrics::txpool_metrics::txpool_metrics;
 use fuel_core_services::{
+    seqlock::SeqLock,
     AsyncProcessor,
     RunnableService,
     RunnableTask,
@@ -188,7 +189,7 @@ pub struct Task<View> {
     p2p_sync_process: AsyncProcessor,
     pruner: TransactionPruner,
     pool: Shared<TxPool>,
-    current_height: Shared<BlockHeight>,
+    current_height: Arc<SeqLock<BlockHeight>>,
     tx_sync_history: Shared<HashSet<PeerId>>,
     shared_state: SharedState,
     metrics: bool,
@@ -320,8 +321,9 @@ where
         }
 
         {
-            let mut block_height = self.current_height.write();
-            *block_height = new_height;
+            self.current_height.write(|data| {
+                *data = new_height;
+            });
         }
     }
 
@@ -396,7 +398,7 @@ where
         let utxo_validation = self.utxo_validation;
 
         let insert_transaction_thread_pool_op = move || {
-            let current_height = *current_height.read();
+            let current_height = current_height.read();
 
             // TODO: This should be removed if the checked transactions
             //  can work with Arc in it
@@ -802,7 +804,7 @@ where
         p2p_sync_process,
         pruner,
         p2p: Arc::new(p2p),
-        current_height: Arc::new(RwLock::new(current_height)),
+        current_height: Arc::new(SeqLock::new(current_height)),
         pool: Arc::new(RwLock::new(txpool)),
         shared_state,
         metrics,