|
|
@@ -1,15 +1,15 @@
|
|
|
+use std::time::Duration;
|
|
|
+
|
|
|
use async_std::{
|
|
|
sync::{Arc, Mutex},
|
|
|
task::sleep,
|
|
|
};
|
|
|
-use std::time::Duration;
|
|
|
-
|
|
|
-use async_executor::Executor;
|
|
|
use chrono::Utc;
|
|
|
use futures::{select, FutureExt};
|
|
|
use fxhash::FxHashMap;
|
|
|
use log::{debug, error, warn};
|
|
|
use rand::{rngs::OsRng, Rng, RngCore};
|
|
|
+use smol::Executor;
|
|
|
|
|
|
use crate::{
|
|
|
net,
|
|
|
@@ -27,7 +27,7 @@ use super::{
|
|
|
prune_map, DataStore, RaftSettings,
|
|
|
};
|
|
|
|
|
|
-async fn send_loop(sender: async_channel::Sender<()>, timeout: Duration) -> Result<()> {
|
|
|
+async fn send_loop(sender: smol::channel::Sender<()>, timeout: Duration) -> Result<()> {
|
|
|
loop {
|
|
|
sleep(timeout).await;
|
|
|
sender.send(()).await?;
|
|
|
@@ -79,10 +79,10 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
|
|
|
let datastore = DataStore::new(settings.datastore_path.to_str().unwrap())?;
|
|
|
|
|
|
// broadcasting channels
|
|
|
- let msgs_channel = async_channel::unbounded::<T>();
|
|
|
- let commits_channel = async_channel::unbounded::<T>();
|
|
|
+ let msgs_channel = smol::channel::unbounded::<T>();
|
|
|
+ let commits_channel = smol::channel::unbounded::<T>();
|
|
|
|
|
|
- let p2p_sender = async_channel::unbounded::<NetMsg>();
|
|
|
+ let p2p_sender = smol::channel::unbounded::<NetMsg>();
|
|
|
|
|
|
let id = match datastore.id.get_last()? {
|
|
|
Some(_id) => _id,
|
|
|
@@ -122,9 +122,9 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
|
|
|
pub async fn run(
|
|
|
&mut self,
|
|
|
p2p: net::P2pPtr,
|
|
|
- p2p_recv_channel: async_channel::Receiver<NetMsg>,
|
|
|
+ p2p_recv_channel: smol::channel::Receiver<NetMsg>,
|
|
|
executor: Arc<Executor<'_>>,
|
|
|
- stop_signal: async_channel::Receiver<()>,
|
|
|
+ stop_signal: smol::channel::Receiver<()>,
|
|
|
) -> Result<()> {
|
|
|
let p2p_send_task = executor.spawn(p2p_send_loop(self.p2p_sender.1.clone(), p2p.clone()));
|
|
|
|
|
|
@@ -134,9 +134,9 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
|
|
|
let prune_nodes_id_task =
|
|
|
executor.spawn(prune_map::<NodeId>(self.nodes.clone(), self.settings.prun_duration));
|
|
|
|
|
|
- let (id_sx, id_rv) = async_channel::unbounded::<()>();
|
|
|
- let (heartbeat_sx, heartbeat_rv) = async_channel::unbounded::<()>();
|
|
|
- let (timeout_sx, timeout_rv) = async_channel::unbounded::<()>();
|
|
|
+ let (id_sx, id_rv) = smol::channel::unbounded::<()>();
|
|
|
+ let (heartbeat_sx, heartbeat_rv) = smol::channel::unbounded::<()>();
|
|
|
+ let (timeout_sx, timeout_rv) = smol::channel::unbounded::<()>();
|
|
|
|
|
|
let id_timeout = Duration::from_secs(self.settings.id_timeout);
|
|
|
let send_id_task = executor.spawn(send_loop(id_sx, id_timeout));
|
|
|
@@ -190,7 +190,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
|
|
|
/// Return async receiver channel which can be used to receive T Messages
|
|
|
/// from raft consensus
|
|
|
///
|
|
|
- pub fn receiver(&self) -> async_channel::Receiver<T> {
|
|
|
+ pub fn receiver(&self) -> smol::channel::Receiver<T> {
|
|
|
self.commits_channel.1.clone()
|
|
|
}
|
|
|
|
|
|
@@ -198,7 +198,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
|
|
|
/// Return async sender channel which can be used to broadcast T Messages
|
|
|
/// to raft consensus
|
|
|
///
|
|
|
- pub fn sender(&self) -> async_channel::Sender<T> {
|
|
|
+ pub fn sender(&self) -> smol::channel::Sender<T> {
|
|
|
self.msgs_channel.0.clone()
|
|
|
}
|
|
|
|