fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-28 09:11:34 +00:00
Code Issues Projects Releases Wiki Activity

Add a KademliaHandler (#580)

Browse Source 
* Rework Kademlia for the new design

* Minor work on protocol.rs

* More work

* Remove QueryTarget::FindValue

* Finish work on query

* Query timeout test

* Work on topology

* More work

* Update protocols/kad/src/topology.rs

Co-Authored-By: tomaka <pierre.krieger1708@gmail.com>

* Fix trailing whitespaces

* Use if let
This commit is contained in:
Pierre Krieger
2018-11-29 12:11:35 +01:00
committed by GitHub
parent ab192cdca7
commit 3aa1fcbdc6
21 changed files with 2357 additions and 1666 deletions
Download Patch File Download Diff File Expand all files Collapse all files

446
protocols/kad/src/protocol.rs
View File

@ -18,19 +18,17 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//! Provides the `KadMsg` enum of all the possible messages transmitted with the Kademlia protocol,
//! and the `KademliaProtocolConfig` connection upgrade whose output is a
//! `Stream<Item = KadMsg> + Sink<SinkItem = KadMsg>`.
//! Provides the `KadRequestMsg` and `KadResponseMsg` enums of all the possible messages
//! transmitted with the Kademlia protocol, and the `KademliaProtocolConfig` connection upgrade.
//!
//! The upgrade's output a `Sink + Stream` of messages.
//!
//! The `Stream` component is used to poll the underlying transport, and the `Sink` component is
//! used to send messages.
use bytes::{Bytes, BytesMut};
use futures::{future, sink, Sink, stream, Stream};
use libp2p_core::{
Multiaddr, PeerId,
upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}
};
use futures::{future, sink, stream, Sink, Stream};
use libp2p_core::{InboundUpgrade, Multiaddr, OutboundUpgrade, PeerId, UpgradeInfo};
use multihash::Multihash;
use protobuf::{self, Message};
use protobuf_structs;
@ -40,6 +38,7 @@ use tokio_codec::Framed;
use tokio_io::{AsyncRead, AsyncWrite};
use unsigned_varint::codec;
/// Status of our connection to a node reported by the Kademlia protocol.
#[derive(Copy, Clone, PartialEq, Eq, Debug, Hash)]
pub enum KadConnectionType {
/// Sender hasn't tried to connect to peer.
@ -81,9 +80,11 @@ impl Into<protobuf_structs::dht::Message_ConnectionType> for KadConnectionType {
/// Information about a peer, as known by the sender.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KadPeer {
/// Identifier of the peer.
pub node_id: PeerId,
/// The multiaddresses that are known for that peer.
/// The multiaddresses that the sender think can be used in order to reach the peer.
pub multiaddrs: Vec<Multiaddr>,
/// How the sender is connected to that remote.
pub connection_ty: KadConnectionType,
}
@ -127,13 +128,16 @@ impl Into<protobuf_structs::dht::Message_Peer> for KadPeer {
}
/// Configuration for a Kademlia connection upgrade. When applied to a connection, turns this
/// connection into a `Stream + Sink` whose items are of type `KadMsg`.
/// connection into a `Stream + Sink` whose items are of type `KadRequestMsg` and `KadResponseMsg`.
// TODO: if, as suspected, we can confirm with Protocol Labs that each open Kademlia substream does
// only one request, then we can change the output of the `InboundUpgrade` and
// `OutboundUpgrade` to be just a single message
#[derive(Debug, Default, Copy, Clone)]
pub struct KademliaProtocolConfig;
impl UpgradeInfo for KademliaProtocolConfig {
type NamesIter = iter::Once<(Bytes, ())>;
type UpgradeId = ();
type NamesIter = iter::Once<(Bytes, Self::UpgradeId)>;
#[inline]
fn protocol_names(&self) -> Self::NamesIter {
@ -143,106 +147,102 @@ impl UpgradeInfo for KademliaProtocolConfig {
impl<C> InboundUpgrade<C> for KademliaProtocolConfig
where
C: AsyncRead + AsyncWrite + 'static, // TODO: 'static :-/
C: AsyncRead + AsyncWrite,
{
type Output = KadStreamSink<C>;
type Output = KadInStreamSink<C>;
type Future = future::FutureResult<Self::Output, IoError>;
type Error = IoError;
type Future = future::FutureResult<Self::Output, Self::Error>;
#[inline]
fn upgrade_inbound(self, incoming: C, _: Self::UpgradeId) -> Self::Future {
future::ok(kademlia_protocol(incoming))
fn upgrade_inbound(self, incoming: C, _: ()) -> Self::Future {
future::ok(
Framed::new(incoming, codec::UviBytes::default())
.from_err::<IoError>()
.with::<_, fn(_) -> _, _>(|response| -> Result<_, IoError> {
let proto_struct = resp_msg_to_proto(response);
proto_struct.write_to_bytes()
.map_err(|err| IoError::new(IoErrorKind::InvalidData, err.to_string()))
})
.and_then::<fn(_) -> _, _>(|bytes: BytesMut| {
let request = protobuf::parse_from_bytes(&bytes)?;
proto_to_req_msg(request)
}),
)
}
}
impl<C> OutboundUpgrade<C> for KademliaProtocolConfig
where
C: AsyncRead + AsyncWrite + 'static, // TODO: 'static :-/
C: AsyncRead + AsyncWrite,
{
type Output = KadStreamSink<C>;
type Output = KadOutStreamSink<C>;
type Future = future::FutureResult<Self::Output, IoError>;
type Error = IoError;
type Future = future::FutureResult<Self::Output, Self::Error>;
#[inline]
fn upgrade_outbound(self, incoming: C, _: Self::UpgradeId) -> Self::Future {
future::ok(kademlia_protocol(incoming))
fn upgrade_outbound(self, incoming: C, _: ()) -> Self::Future {
future::ok(
Framed::new(incoming, codec::UviBytes::default())
.from_err::<IoError>()
.with::<_, fn(_) -> _, _>(|request| -> Result<_, IoError> {
let proto_struct = req_msg_to_proto(request);
match proto_struct.write_to_bytes() {
Ok(msg) => Ok(msg),
Err(err) => Err(IoError::new(IoErrorKind::Other, err.to_string())),
}
})
.and_then::<fn(_) -> _, _>(|bytes: BytesMut| {
let response = protobuf::parse_from_bytes(&bytes)?;
proto_to_resp_msg(response)
}),
)
}
}
type KadStreamSink<S> = stream::AndThen<sink::With<stream::FromErr<Framed<S, codec::UviBytes<Vec<u8>>>, IoError>, KadMsg, fn(KadMsg) -> Result<Vec<u8>, IoError>, Result<Vec<u8>, IoError>>, fn(BytesMut) -> Result<KadMsg, IoError>, Result<KadMsg, IoError>>;
/// Sink of responses and stream of requests.
pub type KadInStreamSink<S> = stream::AndThen<
sink::With<
stream::FromErr<Framed<S, codec::UviBytes<Vec<u8>>>, IoError>,
KadResponseMsg,
fn(KadResponseMsg) -> Result<Vec<u8>, IoError>,
Result<Vec<u8>, IoError>,
>,
fn(BytesMut) -> Result<KadRequestMsg, IoError>,
Result<KadRequestMsg, IoError>,
>;
// Upgrades a socket to use the Kademlia protocol.
fn kademlia_protocol<S>(socket: S) -> KadStreamSink<S>
where
S: AsyncRead + AsyncWrite,
{
Framed::new(socket, codec::UviBytes::default())
.from_err::<IoError>()
.with::<_, fn(_) -> _, _>(|request| -> Result<_, IoError> {
let proto_struct = msg_to_proto(request);
Ok(proto_struct.write_to_bytes().unwrap()) // TODO: error?
})
.and_then::<fn(_) -> _, _>(|bytes| {
let response = protobuf::parse_from_bytes(&bytes)?;
proto_to_msg(response)
})
}
/// Sink of requests and stream of responses.
pub type KadOutStreamSink<S> = stream::AndThen<
sink::With<
stream::FromErr<Framed<S, codec::UviBytes<Vec<u8>>>, IoError>,
KadRequestMsg,
fn(KadRequestMsg) -> Result<Vec<u8>, IoError>,
Result<Vec<u8>, IoError>,
>,
fn(BytesMut) -> Result<KadResponseMsg, IoError>,
Result<KadResponseMsg, IoError>,
>;
/// Message that we can send to a peer or received from a peer.
// TODO: document the rest
/// Request that we can send to a peer or that we received from a peer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum KadMsg {
/// Ping request or response.
pub enum KadRequestMsg {
/// Ping request.
Ping,
/// Target must save the given record, can be queried later with `GetValueReq`.
PutValue {
/// Identifier of the record.
key: Multihash,
/// The record itself.
record: (), //record: protobuf_structs::record::Record, // TODO: no
},
GetValueReq {
/// Identifier of the record.
key: Multihash,
},
GetValueRes {
/// Identifier of the returned record.
key: Multihash,
record: (), //record: Option<protobuf_structs::record::Record>, // TODO: no
closer_peers: Vec<KadPeer>,
},
/// Request for the list of nodes whose IDs are the closest to `key`. The number of nodes
/// returned is not specified, but should be around 20.
FindNodeReq {
FindNode {
/// Identifier of the node.
key: PeerId,
},
/// Response to a `FindNodeReq`.
FindNodeRes {
/// Results of the request.
closer_peers: Vec<KadPeer>,
},
/// Same as `FindNodeReq`, but should also return the entries of the local providers list for
/// Same as `FindNode`, but should also return the entries of the local providers list for
/// this key.
GetProvidersReq {
GetProviders {
/// Identifier being searched.
key: Multihash,
},
/// Response to a `FindNodeReq`.
GetProvidersRes {
/// Nodes closest to the key.
closer_peers: Vec<KadPeer>,
/// Known providers for this key.
provider_peers: Vec<KadPeer>,
},
/// Indicates that this list of providers is known for this key.
AddProvider {
/// Key for which we should add providers.
@ -252,39 +252,69 @@ pub enum KadMsg {
},
}
// Turns a type-safe kadmelia message into the corresponding row protobuf message.
fn msg_to_proto(kad_msg: KadMsg) -> protobuf_structs::dht::Message {
/// Response that we can send to a peer or that we received from a peer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum KadResponseMsg {
/// Ping response.
Pong,
/// Response to a `FindNode`.
FindNode {
/// Results of the request.
closer_peers: Vec<KadPeer>,
},
/// Response to a `GetProviders`.
GetProviders {
/// Nodes closest to the key.
closer_peers: Vec<KadPeer>,
/// Known providers for this key.
provider_peers: Vec<KadPeer>,
},
}
// Turns a type-safe Kadmelia message into the corresponding raw protobuf message.
fn req_msg_to_proto(kad_msg: KadRequestMsg) -> protobuf_structs::dht::Message {
match kad_msg {
KadMsg::Ping => {
KadRequestMsg::Ping => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::PING);
msg
}
KadMsg::PutValue { key, .. } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::PUT_VALUE);
msg.set_key(key.into_bytes());
//msg.set_record(record); // TODO:
msg
}
KadMsg::GetValueReq { key } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::GET_VALUE);
msg.set_key(key.into_bytes());
msg.set_clusterLevelRaw(10);
msg
}
KadMsg::GetValueRes { .. } => unimplemented!(), // TODO:
KadMsg::FindNodeReq { key } => {
KadRequestMsg::FindNode { key } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::FIND_NODE);
msg.set_key(key.into_bytes());
msg.set_clusterLevelRaw(10);
msg
}
KadMsg::FindNodeRes { closer_peers } => {
// TODO: if empty, the remote will think it's a request
// TODO: not good, possibly exposed in the API
KadRequestMsg::GetProviders { key } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::GET_PROVIDERS);
msg.set_key(key.into_bytes());
msg.set_clusterLevelRaw(10);
msg
}
KadRequestMsg::AddProvider { key, provider_peer } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::ADD_PROVIDER);
msg.set_clusterLevelRaw(10);
msg.set_key(key.into_bytes());
msg.mut_providerPeers().push(provider_peer.into());
msg
}
}
}
// Turns a type-safe Kadmelia message into the corresponding raw protobuf message.
fn resp_msg_to_proto(kad_msg: KadResponseMsg) -> protobuf_structs::dht::Message {
match kad_msg {
KadResponseMsg::Pong => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::PING);
msg
}
KadResponseMsg::FindNode { closer_peers } => {
assert!(!closer_peers.is_empty());
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::FIND_NODE);
@ -294,16 +324,10 @@ fn msg_to_proto(kad_msg: KadMsg) -> protobuf_structs::dht::Message {
}
msg
}
KadMsg::GetProvidersReq { key } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::GET_PROVIDERS);
msg.set_key(key.into_bytes());
msg.set_clusterLevelRaw(10);
msg
}
KadMsg::GetProvidersRes { closer_peers, provider_peers } => {
// TODO: if empty, the remote will think it's a request
// TODO: not good, possibly exposed in the API
KadResponseMsg::GetProviders {
closer_peers,
provider_peers,
} => {
assert!(!closer_peers.is_empty());
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::GET_PROVIDERS);
@ -316,94 +340,47 @@ fn msg_to_proto(kad_msg: KadMsg) -> protobuf_structs::dht::Message {
}
msg
}
KadMsg::AddProvider { key, provider_peer } => {
let mut msg = protobuf_structs::dht::Message::new();
msg.set_field_type(protobuf_structs::dht::Message_MessageType::ADD_PROVIDER);
msg.set_clusterLevelRaw(10);
msg.set_key(key.into_bytes());
msg.mut_providerPeers().push(provider_peer.into());
msg
}
}
}
/// Turns a raw Kademlia message into a type-safe message.
fn proto_to_msg(mut message: protobuf_structs::dht::Message) -> Result<KadMsg, IoError> {
fn proto_to_req_msg(mut message: protobuf_structs::dht::Message) -> Result<KadRequestMsg, IoError> {
match message.get_field_type() {
protobuf_structs::dht::Message_MessageType::PING => Ok(KadMsg::Ping),
protobuf_structs::dht::Message_MessageType::PING => Ok(KadRequestMsg::Ping),
protobuf_structs::dht::Message_MessageType::PUT_VALUE => {
let key = Multihash::from_bytes(message.take_key())
.map_err(|err| IoError::new(IoErrorKind::InvalidData, err))?;
let _record = message.take_record();
Ok(KadMsg::PutValue {
key: key,
record: (),
})
Err(IoError::new(
IoErrorKind::InvalidData,
"received a PUT_VALUE message, but this is not supported by rust-libp2p yet",
))
}
protobuf_structs::dht::Message_MessageType::GET_VALUE => {
let key = Multihash::from_bytes(message.take_key())
.map_err(|err| IoError::new(IoErrorKind::InvalidData, err))?;
Ok(KadMsg::GetValueReq { key: key })
Err(IoError::new(
IoErrorKind::InvalidData,
"received a GET_VALUE message, but this is not supported by rust-libp2p yet",
))
}
protobuf_structs::dht::Message_MessageType::FIND_NODE => {
if message.get_closerPeers().is_empty() {
let key = PeerId::from_bytes(message.take_key())
.map_err(|_| IoError::new(IoErrorKind::InvalidData, "invalid peer id in FIND_NODE"))?;
Ok(KadMsg::FindNodeReq {
key,
})
} else {
// TODO: for now we don't parse the peer properly, so it is possible that we get
// parsing errors for peers even when they are valid; we ignore these
// errors for now, but ultimately we should just error altogether
let closer_peers = message.mut_closerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.collect::<Vec<_>>();
Ok(KadMsg::FindNodeRes {
closer_peers,
})
}
let key = PeerId::from_bytes(message.take_key()).map_err(|_| {
IoError::new(IoErrorKind::InvalidData, "invalid peer id in FIND_NODE")
})?;
Ok(KadRequestMsg::FindNode { key })
}
protobuf_structs::dht::Message_MessageType::GET_PROVIDERS => {
if message.get_closerPeers().is_empty() {
let key = Multihash::from_bytes(message.take_key())
.map_err(|err| IoError::new(IoErrorKind::InvalidData, err))?;
Ok(KadMsg::GetProvidersReq {
key,
})
} else {
// TODO: for now we don't parse the peer properly, so it is possible that we get
// parsing errors for peers even when they are valid; we ignore these
// errors for now, but ultimately we should just error altogether
let closer_peers = message.mut_closerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.collect::<Vec<_>>();
let provider_peers = message.mut_providerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.collect::<Vec<_>>();
Ok(KadMsg::GetProvidersRes {
closer_peers,
provider_peers,
})
}
let key = Multihash::from_bytes(message.take_key())
.map_err(|err| IoError::new(IoErrorKind::InvalidData, err))?;
Ok(KadRequestMsg::GetProviders { key })
}
protobuf_structs::dht::Message_MessageType::ADD_PROVIDER => {
// TODO: for now we don't parse the peer properly, so it is possible that we get
// parsing errors for peers even when they are valid; we ignore these
// errors for now, but ultimately we should just error altogether
let provider_peer = message.mut_providerPeers()
let provider_peer = message
.mut_providerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.next();
@ -411,10 +388,7 @@ fn proto_to_msg(mut message: protobuf_structs::dht::Message) -> Result<KadMsg, I
if let Some(provider_peer) = provider_peer {
let key = Multihash::from_bytes(message.take_key())
.map_err(|err| IoError::new(IoErrorKind::InvalidData, err))?;
Ok(KadMsg::AddProvider {
key,
provider_peer,
})
Ok(KadRequestMsg::AddProvider { key, provider_peer })
} else {
Err(IoError::new(
IoErrorKind::InvalidData,
@ -425,64 +399,105 @@ fn proto_to_msg(mut message: protobuf_structs::dht::Message) -> Result<KadMsg, I
}
}
/// Turns a raw Kademlia message into a type-safe message.
fn proto_to_resp_msg(
mut message: protobuf_structs::dht::Message,
) -> Result<KadResponseMsg, IoError> {
match message.get_field_type() {
protobuf_structs::dht::Message_MessageType::PING => Ok(KadResponseMsg::Pong),
protobuf_structs::dht::Message_MessageType::GET_VALUE => {
Err(IoError::new(
IoErrorKind::InvalidData,
"received a GET_VALUE message, but this is not supported by rust-libp2p yet",
))
}
protobuf_structs::dht::Message_MessageType::FIND_NODE => {
let closer_peers = message
.mut_closerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.collect::<Vec<_>>();
Ok(KadResponseMsg::FindNode { closer_peers })
}
protobuf_structs::dht::Message_MessageType::GET_PROVIDERS => {
let closer_peers = message
.mut_closerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.collect::<Vec<_>>();
let provider_peers = message
.mut_providerPeers()
.iter_mut()
.filter_map(|peer| KadPeer::from_peer(peer).ok())
.collect::<Vec<_>>();
Ok(KadResponseMsg::GetProviders {
closer_peers,
provider_peers,
})
}
protobuf_structs::dht::Message_MessageType::PUT_VALUE => Err(IoError::new(
IoErrorKind::InvalidData,
"received an unexpected PUT_VALUE message",
)),
protobuf_structs::dht::Message_MessageType::ADD_PROVIDER => Err(IoError::new(
IoErrorKind::InvalidData,
"received an unexpected ADD_PROVIDER message",
)),
}
}
#[cfg(test)]
mod tests {
extern crate libp2p_tcp_transport;
extern crate tokio;
use self::libp2p_tcp_transport::TcpConfig;
use self::tokio::runtime::current_thread::Runtime;
use futures::{Future, Sink, Stream};
use libp2p_core::{Transport, PeerId};
use libp2p_core::{PeerId, PublicKey, Transport};
use multihash::{encode, Hash};
use protocol::{KadConnectionType, KadMsg, KademliaProtocolConfig, KadPeer};
use protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig};
use std::sync::mpsc;
use std::thread;
use self::tokio::runtime::current_thread::Runtime;
/*// TODO: restore
#[test]
fn correct_transfer() {
// We open a server and a client, send a message between the two, and check that they were
// successfully received.
test_one(KadMsg::Ping);
test_one(KadMsg::PutValue {
key: encode(Hash::SHA2256, &[1, 2, 3, 4]).unwrap(),
record: (),
});
test_one(KadMsg::GetValueReq {
key: encode(Hash::SHA2256, &[10, 11, 12]).unwrap(),
});
test_one(KadMsg::FindNodeReq {
key: PeerId::random()
key: PeerId::random(),
});
test_one(KadMsg::FindNodeRes {
closer_peers: vec![
KadPeer {
node_id: PeerId::random(),
multiaddrs: vec!["/ip4/100.101.102.103/tcp/20105".parse().unwrap()],
connection_ty: KadConnectionType::Connected,
},
],
closer_peers: vec![KadPeer {
node_id: PeerId::random(),
multiaddrs: vec!["/ip4/100.101.102.103/tcp/20105".parse().unwrap()],
connection_ty: KadConnectionType::Connected,
}],
});
test_one(KadMsg::GetProvidersReq {
key: encode(Hash::SHA2256, &[9, 12, 0, 245, 245, 201, 28, 95]).unwrap(),
});
test_one(KadMsg::GetProvidersRes {
closer_peers: vec![
KadPeer {
node_id: PeerId::random(),
multiaddrs: vec!["/ip4/100.101.102.103/tcp/20105".parse().unwrap()],
connection_ty: KadConnectionType::Connected,
},
],
provider_peers: vec![
KadPeer {
node_id: PeerId::random(),
multiaddrs: vec!["/ip4/200.201.202.203/tcp/1999".parse().unwrap()],
connection_ty: KadConnectionType::NotConnected,
},
],
closer_peers: vec![KadPeer {
node_id: PeerId::random(),
multiaddrs: vec!["/ip4/100.101.102.103/tcp/20105".parse().unwrap()],
connection_ty: KadConnectionType::Connected,
}],
provider_peers: vec![KadPeer {
node_id: PeerId::random(),
multiaddrs: vec!["/ip4/200.201.202.203/tcp/1999".parse().unwrap()],
connection_ty: KadConnectionType::NotConnected,
}],
});
test_one(KadMsg::AddProvider {
key: encode(Hash::SHA2256, &[9, 12, 0, 245, 245, 201, 28, 95]).unwrap(),
@ -504,7 +519,6 @@ mod tests {
let (listener, addr) = transport
.listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap())
.unwrap();
tx.send(addr).unwrap();
let future = listener
@ -531,5 +545,5 @@ mod tests {
let _ = rt.block_on(future).unwrap();
bg_thread.join().unwrap();
}
}
}*/
}