Remve dead ChannelMaps, ViewQueue, and SendToTask

crates/hotshot/examples/combined/types.rs

@@ -1,7 +1,7 @@
 use crate::infra::CombinedDARun;
 use hotshot::traits::implementations::{CombinedCommChannel, MemoryStorage};
 use hotshot_testing::state_types::TestTypes;
-use hotshot_types::traits::node_implementation::{ChannelMaps, NodeImplementation, NodeType};
+use hotshot_types::traits::node_implementation::NodeImplementation;
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 
@@ -22,12 +22,6 @@ impl NodeImplementation<TestTypes> for NodeImpl {
     type Storage = MemoryStorage<TestTypes>;
     type QuorumNetwork = QuorumNetwork;
     type CommitteeNetwork = DANetwork;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (ChannelMaps::new(start_view), None)
-    }
 }
 /// convenience type alias
 pub type ThisRun = CombinedDARun<TestTypes>;

crates/hotshot/examples/libp2p/types.rs

@@ -1,7 +1,7 @@
 use crate::infra::Libp2pDARun;
 use hotshot::traits::implementations::{Libp2pCommChannel, MemoryStorage};
 use hotshot_testing::state_types::TestTypes;
-use hotshot_types::traits::node_implementation::{ChannelMaps, NodeImplementation, NodeType};
+use hotshot_types::traits::node_implementation::NodeImplementation;
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 
@@ -22,12 +22,6 @@ impl NodeImplementation<TestTypes> for NodeImpl {
     type Storage = MemoryStorage<TestTypes>;
     type QuorumNetwork = QuorumNetwork;
     type CommitteeNetwork = DANetwork;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (ChannelMaps::new(start_view), None)
-    }
 }
 /// convenience type alias
 pub type ThisRun = Libp2pDARun<TestTypes>;

crates/hotshot/examples/webserver/types.rs

@@ -1,7 +1,7 @@
 use crate::infra::WebServerDARun;
 use hotshot::traits::implementations::{MemoryStorage, WebCommChannel};
 use hotshot_testing::state_types::TestTypes;
-use hotshot_types::traits::node_implementation::{ChannelMaps, NodeImplementation, NodeType};
+use hotshot_types::traits::node_implementation::NodeImplementation;
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 
@@ -22,12 +22,6 @@ impl NodeImplementation<TestTypes> for NodeImpl {
     type Storage = MemoryStorage<TestTypes>;
     type CommitteeNetwork = DANetwork;
     type QuorumNetwork = QuorumNetwork;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (ChannelMaps::new(start_view), None)
-    }
 }
 /// convenience type alias
 pub type ThisRun = WebServerDARun<TestTypes>;

crates/hotshot/src/lib.rs

@@ -25,7 +25,7 @@ use async_compatibility_layer::{
     art::{async_spawn, async_spawn_local},
     channel::UnboundedSender,
 };
-use async_lock::{RwLock, RwLockUpgradableReadGuard, RwLockWriteGuard};
+use async_lock::RwLock;
 use async_trait::async_trait;
 use commit::Committable;
 use custom_debug::Debug;
@@ -37,11 +37,8 @@ use hotshot_task::{
 };
 use hotshot_task_impls::{events::HotShotEvent, network::NetworkTaskKind};
 
-#[cfg(feature = "hotshot-testing")]
-use hotshot_types::traits::node_implementation::ChannelMaps;
-
 use hotshot_types::{
-    consensus::{Consensus, ConsensusMetricsValue, View, ViewInner, ViewQueue},
+    consensus::{Consensus, ConsensusMetricsValue, View, ViewInner},
     data::Leaf,
     error::StorageSnafu,
     event::EventType,
@@ -52,7 +49,7 @@ use hotshot_types::{
     traits::{
         consensus_api::ConsensusApi,
         network::{CommunicationChannel, NetworkError},
-        node_implementation::{ConsensusTime, NodeType, SendToTasks},
+        node_implementation::{ConsensusTime, NodeType},
         signature_key::SignatureKey,
         states::ValidatedState,
         storage::StoredView,
@@ -148,11 +145,6 @@ pub struct SystemContextInner<TYPES: NodeType, I: NodeImplementation<TYPES>> {
     /// The hotstuff implementation
     consensus: Arc<RwLock<Consensus<TYPES>>>,
 
-    /// Channels for sending/recv-ing proposals and votes for quorum and committee exchanges, the
-    /// latter of which is only applicable for sequencing consensus.
-    #[cfg(feature = "hotshot-testing")]
-    channel_maps: (ChannelMaps<TYPES>, Option<ChannelMaps<TYPES>>),
-
     // global_registry: GlobalRegistry,
     /// Access to the output event stream.
     output_event_stream: ChannelStream<Event<TYPES>>,
@@ -253,7 +245,6 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
         let inner: Arc<SystemContextInner<TYPES, I>> = Arc::new(SystemContextInner {
             id: nonce,
             #[cfg(feature = "hotshot-testing")]
-            channel_maps: I::new_channel_maps(start_view),
             consensus,
             public_key,
             private_key,
@@ -316,7 +307,6 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
     async fn send_external_event(&self, event: Event<TYPES>) {
         debug!(?event, "send_external_event");
         self.inner.output_event_stream.publish(event).await;
-
     }
 
     /// Publishes a transaction asynchronously to the network
@@ -528,41 +518,6 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
     pub fn get_next_view_timeout(&self) -> u64 {
         self.inner.config.next_view_timeout
     }
-
-    /// given a view number and a upgradable read lock on a channel map, inserts entry into map if it
-    /// doesn't exist, or creates entry. Then returns a clone of the entry
-    pub async fn create_or_obtain_chan_from_read(
-        view_num: TYPES::Time,
-        channel_map: RwLockUpgradableReadGuard<'_, SendToTasks<TYPES>>,
-    ) -> ViewQueue<TYPES> {
-        // check if we have the entry
-        // if we don't, insert
-        if let Some(vq) = channel_map.channel_map.get(&view_num) {
-            vq.clone()
-        } else {
-            let mut channel_map =
-                RwLockUpgradableReadGuard::<'_, SendToTasks<TYPES>>::upgrade(channel_map).await;
-            let new_view_queue = ViewQueue::default();
-            let vq = new_view_queue.clone();
-            // NOTE: the read lock is held until all other read locks are DROPPED and
-            // the read lock may be turned into a write lock.
-            // This means that the `channel_map` will not change. So we don't need
-            // to check again to see if a channel was added
-
-            channel_map.channel_map.insert(view_num, new_view_queue);
-            vq
-        }
-    }
-
-    /// given a view number and a write lock on a channel map, inserts entry into map if it
-    /// doesn't exist, or creates entry. Then returns a clone of the entry
-    #[allow(clippy::unused_async)] // async for API compatibility reasons
-    pub async fn create_or_obtain_chan_from_write(
-        view_num: TYPES::Time,
-        mut channel_map: RwLockWriteGuard<'_, SendToTasks<TYPES>>,
-    ) -> ViewQueue<TYPES> {
-        channel_map.channel_map.entry(view_num).or_default().clone()
-    }
 }
 
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {

crates/hotshot/src/types/handle.rs

@@ -255,40 +255,4 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES> + 'static> SystemContextHandl
             .send_direct_message(MessageKind::from_consensus_message(msg), recipient)
             .await;
     }
-
-    /// Get length of the replica's receiver channel
-    #[cfg(feature = "hotshot-testing")]
-    pub async fn get_replica_receiver_channel_len(
-        &self,
-        view_number: TYPES::Time,
-    ) -> Option<usize> {
-        use async_compatibility_layer::channel::UnboundedReceiver;
-
-        let channel_map = self.hotshot.inner.channel_maps.0.vote_channel.read().await;
-        let chan = channel_map.channel_map.get(&view_number)?;
-        let receiver = chan.receiver_chan.lock().await;
-        UnboundedReceiver::len(&*receiver)
-    }
-
-    /// Get length of the next leaders's receiver channel
-    #[cfg(feature = "hotshot-testing")]
-    pub async fn get_next_leader_receiver_channel_len(
-        &self,
-        view_number: TYPES::Time,
-    ) -> Option<usize> {
-        use async_compatibility_layer::channel::UnboundedReceiver;
-
-        let channel_map = self
-            .hotshot
-            .inner
-            .channel_maps
-            .0
-            .proposal_channel
-            .read()
-            .await;
-        let chan = channel_map.channel_map.get(&view_number)?;
-
-        let receiver = chan.receiver_chan.lock().await;
-        UnboundedReceiver::len(&*receiver)
-    }
 }

crates/testing/src/node_types.rs

@@ -13,9 +13,7 @@ use hotshot::traits::{
     NodeImplementation,
 };
 use hotshot_types::{
-    data::ViewNumber,
-    signature_key::BLSPubKey,
-    traits::node_implementation::{ChannelMaps, NodeType},
+    data::ViewNumber, signature_key::BLSPubKey, traits::node_implementation::NodeType,
 };
 use serde::{Deserialize, Serialize};
 
@@ -100,58 +98,22 @@ impl NodeImplementation<TestTypes> for Libp2pImpl {
     type Storage = MemoryStorage<TestTypes>;
     type QuorumNetwork = StaticLibp2pQuorumComm;
     type CommitteeNetwork = StaticLibp2pDAComm;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (
-            ChannelMaps::new(start_view),
-            Some(ChannelMaps::new(start_view)),
-        )
-    }
 }
 
 impl NodeImplementation<TestTypes> for MemoryImpl {
     type Storage = MemoryStorage<TestTypes>;
     type QuorumNetwork = StaticMemoryQuorumComm;
     type CommitteeNetwork = StaticMemoryDAComm;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (
-            ChannelMaps::new(start_view),
-            Some(ChannelMaps::new(start_view)),
-        )
-    }
 }
 
 impl NodeImplementation<TestTypes> for WebImpl {
     type Storage = MemoryStorage<TestTypes>;
     type QuorumNetwork = StaticWebQuorumComm;
     type CommitteeNetwork = StaticWebDAComm;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (
-            ChannelMaps::new(start_view),
-            Some(ChannelMaps::new(start_view)),
-        )
-    }
 }
 
 impl NodeImplementation<TestTypes> for CombinedImpl {
     type Storage = MemoryStorage<TestTypes>;
     type QuorumNetwork = StaticCombinedQuorumComm;
     type CommitteeNetwork = StaticCombinedDAComm;
-
-    fn new_channel_maps(
-        start_view: <TestTypes as NodeType>::Time,
-    ) -> (ChannelMaps<TestTypes>, Option<ChannelMaps<TestTypes>>) {
-        (
-            ChannelMaps::new(start_view),
-            Some(ChannelMaps::new(start_view)),
-        )
-    }
 }

crates/testing/tests/memory_network.rs

@@ -19,7 +19,7 @@ use hotshot_types::message::Message;
 use hotshot_types::signature_key::BLSPubKey;
 use hotshot_types::traits::network::TestableNetworkingImplementation;
 use hotshot_types::traits::network::{ConnectedNetwork, TransmitType};
-use hotshot_types::traits::node_implementation::{ChannelMaps, ConsensusTime, NodeType};
+use hotshot_types::traits::node_implementation::{ConsensusTime, NodeType};
 use hotshot_types::{
     data::ViewNumber,
     message::{DataMessage, MessageKind},
@@ -70,12 +70,6 @@ impl NodeImplementation<Test> for TestImpl {
     type Storage = MemoryStorage<Test>;
     type QuorumNetwork = QuorumNetwork;
     type CommitteeNetwork = DANetwork;
-
-    fn new_channel_maps(
-        start_view: <Test as NodeType>::Time,
-    ) -> (ChannelMaps<Test>, Option<ChannelMaps<Test>>) {
-        (ChannelMaps::new(start_view), None)
-    }
 }
 
 /// fake Eq

crates/types/src/consensus.rs

@@ -1,9 +1,6 @@
 //! Provides the core consensus types
 
-pub use crate::{
-    traits::node_implementation::ViewQueue,
-    utils::{View, ViewInner},
-};
+pub use crate::utils::{View, ViewInner};
 use displaydoc::Display;
 
 use crate::{