Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheBlueMatt merged 4 commits into from
Feb 2, 2021
Merged

Update tokio to 1.0 #781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5a403bd
Update tokio to 1.0
TheBlueMatt Jan 26, 2021
07aff06
Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc
TheBlueMatt Jan 26, 2021
11f5e23
Fix codecov by calling the new binary paths from rust 1.45
TheBlueMatt Jan 26, 2021
32abba7
Drop spurious semicolon that new rustc complains about
TheBlueMatt Jan 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions .github/workflows/build.yml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
1.30.0,
# 1.34.2 is Debian stable
1.34.2,
# 1.39.0 is MSRV for lightning-net-tokio and generates coverage
1.39.0]
# 1.45.2 is MSRV for lightning-net-tokio and generates coverage
1.45.2]
include:
- toolchain: stable
build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
build-net-tokio: true
- toolchain: beta
build-net-tokio: true
- toolchain: 1.39.0
- toolchain: 1.45.2
build-net-tokio: true
coverage: true
runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
- name: Generate coverage report
if: matrix.coverage
run: |
for file in target/debug/lightning-*; do
for file in target/debug/deps/lightning*; do
[ -x "${file}" ] || continue;
mkdir -p "target/cov/$(basename $file)";
./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
linting:
runs-on: ubuntu-latest
env:
TOOLCHAIN: 1.39.0
TOOLCHAIN: 1.45.2
steps:
- name: Checkout source code
uses: actions/checkout@v2
Expand Down
4 changes: 2 additions & 2 deletions lightning-net-tokio/Cargo.toml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }
35 changes: 22 additions & 13 deletions lightning-net-tokio/src/lib.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
// In this case, we do need to call peer_manager.socket_disconnected() to inform
// Rust-Lightning that the socket is gone.
PeerDisconnected
};
}
let disconnect_type = loop {
macro_rules! shutdown_socket {
($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
}
}

fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
// We only ever need a channel of depth 1 here: if we returned a non-full write to the
// PeerManager, we will eventually get notified that there is room in the socket to write
// new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
// we shove a value into the channel which comes after we've reset the read_paused bool to
// false.
let (read_waker, read_receiver) = mpsc::channel(1);
let (reader, writer) = io::split(stream);
stream.set_nonblocking(true).unwrap();
let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());
Comment on lines +233 to +234

@jkczyzjkczyzJan 27, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.

Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.

@TheBlueMattTheBlueMattJan 27, 2021

Copy link
Copy Markdown
CollaboratorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.


(reader, write_receiver, read_receiver,
Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
CMH: ChannelMessageHandler + 'static,
RMH: RoutingMessageHandler + 'static,
L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
CMH: ChannelMessageHandler + 'static,
RMH: RoutingMessageHandler + 'static,
L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
CMH: ChannelMessageHandler + 'static,
RMH: RoutingMessageHandler + 'static,
L: Logger + 'static + ?Sized {
if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
let mut sender = unsafe { (*sender_ptr).clone() };
let sender = unsafe { (*sender_ptr).clone() };
let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
use tokio::sync::mpsc;

use std::mem;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
expected_pubkey: PublicKey,
pubkey_connected: mpsc::Sender<()>,
pubkey_disconnected: mpsc::Sender<()>,
disconnected_flag: AtomicBool,
msg_events: Mutex<Vec<MessageSendEvent>>,
}
impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
if *their_node_id == self.expected_pubkey {
self.disconnected_flag.store(true, Ordering::SeqCst);
self.pubkey_disconnected.clone().try_send(()).unwrap();
}
}
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
expected_pubkey: b_pub,
pubkey_connected: a_connected_sender,
pubkey_disconnected: a_disconnected_sender,
disconnected_flag: AtomicBool::new(false),
msg_events: Mutex::new(Vec::new()),
});
let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
expected_pubkey: a_pub,
pubkey_connected: b_connected_sender,
pubkey_disconnected: b_disconnected_sender,
disconnected_flag: AtomicBool::new(false),
msg_events: Mutex::new(Vec::new()),
});
let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
} else { panic!("Failed to bind to v4 localhost on common ports"); };

let (sender, _receiver) = mpsc::channel(2);
let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
let fut_b = super::setup_inbound(b_manager, sender, conn_b);

tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
});
assert!(a_disconnected.try_recv().is_err());
assert!(b_disconnected.try_recv().is_err());
assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

a_manager.process_events();
tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

fut_a.await;
fut_b.await;
}

#[tokio::test(threaded_scheduler)]
#[tokio::test(flavor = "multi_thread")]
async fn basic_threaded_connection_test() {
do_basic_connection_test().await;
}
Expand Down
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
Update tokio to 1.0#781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

TheBlueMatt
merged 4 commits into

from

Feb 2, 2021
 Merged


Update tokio to 1.0
#781
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
5a403bd

Update tokio to 1.0


TheBlueMatt Jan 26, 2021
07aff06

Bump MSRV for net-tokio to 1.45 as tokio 1.0 req very recent rustc


TheBlueMatt Jan 26, 2021
11f5e23

Fix codecov by calling the new binary paths from rust 1.45


TheBlueMatt Jan 26, 2021
32abba7

Drop spurious semicolon that new rustc complains about


TheBlueMatt Jan 28, 2021

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 5 additions & 5 deletions
.github/workflows/build.yml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -13,8 +13,8 @@ jobs:
 1.30.0,
 # 1.34.2 is Debian stable
 1.34.2,
 # 1.39.0 is MSRV for lightning-net-tokio and generates coverage
 1.39.0]
 # 1.45.2 is MSRV for lightning-net-tokio and generates coverage
 1.45.2]
 include:
 - toolchain: stable
 build-net-tokio: true
Expand All@@ -26,7 +26,7 @@ jobs:
 build-net-tokio: true
 - toolchain: beta
 build-net-tokio: true
 - toolchain: 1.39.0
 - toolchain: 1.45.2
 build-net-tokio: true
 coverage: true
 runs-on: ${{ matrix.platform }}
Expand DownExpand Up@@ -75,7 +75,7 @@ jobs:
 - name: Generate coverage report
 if: matrix.coverage
 run: |
 for file in target/debug/lightning-*; do
 for file in target/debug/deps/lightning*; do
 [ -x "${file}" ] || continue;
 mkdir -p "target/cov/$(basename $file)";
 ./kcov-build/usr/local/bin/kcov --exclude-pattern=/.cargo,/usr/lib --verify "target/cov/$(basename $file)" "$file";
Expand DownExpand Up@@ -180,7 +180,7 @@ jobs:
 linting:
 runs-on: ubuntu-latest
 env:
 TOOLCHAIN: 1.39.0
 TOOLCHAIN: 1.45.2
 steps:
 - name: Checkout source code
 uses: actions/checkout@v2
Expand Down

4 changes: 2 additions & 2 deletions
lightning-net-tokio/Cargo.toml

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -12,7 +12,7 @@ For Rust-Lightning clients which wish to make direct connections to Lightning P2
[dependencies]
bitcoin = "0.24"
lightning = { version = "0.0.12", path = "../lightning" }
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "sync", "net", "time" ] }

[dev-dependencies]
tokio = { version = ">=0.2.12", features = [ "io-util", "macros", "rt-core", "rt-threaded", "sync", "tcp", "time" ] }
tokio = { version = "1.0", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

35 changes: 22 additions & 13 deletions
lightning-net-tokio/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,7 +24,7 @@
//! The call site should, thus, look something like this:
//! ```
//! use tokio::sync::mpsc;
//! use tokio::net::TcpStream;
//! use std::net::TcpStream;
//! use bitcoin::secp256k1::key::PublicKey;
//! use lightning::util::events::EventsProvider;
//! use std::net::SocketAddr;
Expand DownExpand Up@@ -86,6 +86,7 @@ use lightning::util::logger::Logger;

use std::{task, thread};
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
Expand DownExpand Up@@ -156,7 +157,7 @@ impl Connection {
 // In this case, we do need to call peer_manager.socket_disconnected() to inform
 // Rust-Lightning that the socket is gone.
 PeerDisconnected
 };
 }
 let disconnect_type = loop {
 macro_rules! shutdown_socket {
 ($err: expr, $need_disconnect: expr) => { {
Expand DownExpand Up@@ -218,7 +219,7 @@ impl Connection {
 }
	}

	fn new(event_notify: mpsc::Sender<()>, stream: TcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
	fn new(event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 // We only ever need a channel of depth 1 here: if we returned a non-full write to the
 // PeerManager, we will eventually get notified that there is room in the socket to write
 // new bytes, which will generate an event. That event will be popped off the queue before
Expand All@@ -229,7 +230,8 @@ impl Connection {
 // we shove a value into the channel which comes after we've reset the read_paused bool to
 // false.
 let (read_waker, read_receiver) = mpsc::channel(1);
 let (reader, writer) = io::split(stream);
 stream.set_nonblocking(true).unwrap();
 let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());

Comment on lines

+233
 to 
+234

@jkczyzjkczyzJan 27, 2021

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, the problem was that our test was using std::net::TcpStream::connect without setting the result to non-blocking before converting it to tokio::net::TcpStream.
Could we fix the problem by updating the test to use tokio::net::TcpStream::connect instead? Then a conversion wouldn't be needed here nor in connect_outbound.
@TheBlueMattTheBlueMattJan 27, 2021

Copy link


Copy Markdown

CollaboratorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd initially gone with tokio::net::TcpStream in the public API but then realized its much neater to do the set_nonblocking call in Connection::new given its pretty easy to accidentally do the same issue again, and its simpler for our users to not have to think about it than not. Given that, its fewer conversion calls to just accept an std::net::TcpStream instead of a tokio one. It doesn't really matter, though, just pretty poor API design on Tokio's part.

 (reader, write_receiver, read_receiver,
 Arc::new(Mutex::new(Self {
Expand All@@ -248,7 +250,7 @@ impl Connection {
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -290,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
/// not need to poll the provided future in order to make progress.
///
/// See the module-level documentation for how to handle the event_notify mpsc::Sender.
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
Expand DownExpand Up@@ -366,7 +368,7 @@ pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerM
 CMH: ChannelMessageHandler + 'static,
 RMH: RoutingMessageHandler + 'static,
 L: Logger + 'static + ?Sized {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), TcpStream::connect(&addr)).await {
	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
 Some(setup_outbound(peer_manager, event_notify, their_node_id, stream))
	} else { None }
}
Expand All@@ -388,7 +390,7 @@ fn wake_socket_waker(orig_ptr: *const ()) {
}
fn wake_socket_waker_by_ref(orig_ptr: *const ()) {
	let sender_ptr = orig_ptr as *const mpsc::Sender<()>;
	let mut sender = unsafe { (*sender_ptr).clone() };
	let sender = unsafe { (*sender_ptr).clone() };
	let _ = sender.try_send(());
}
fn drop_socket_waker(orig_ptr: *const ()) {
Expand DownExpand Up@@ -512,6 +514,7 @@ mod tests {
	use tokio::sync::mpsc;

	use std::mem;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::{Arc, Mutex};
	use std::time::Duration;

Expand All@@ -526,6 +529,7 @@ mod tests {
 expected_pubkey: PublicKey,
 pubkey_connected: mpsc::Sender<()>,
 pubkey_disconnected: mpsc::Sender<()>,
 disconnected_flag: AtomicBool,
 msg_events: Mutex<Vec<MessageSendEvent>>,
	}
	impl RoutingMessageHandler for MsgHandler {
Expand DownExpand Up@@ -559,6 +563,7 @@ mod tests {
 fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
 fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
 if *their_node_id == self.expected_pubkey {
 self.disconnected_flag.store(true, Ordering::SeqCst);
 self.pubkey_disconnected.clone().try_send(()).unwrap();
 }
 }
Expand DownExpand Up@@ -591,6 +596,7 @@ mod tests {
 expected_pubkey: b_pub,
 pubkey_connected: a_connected_sender,
 pubkey_disconnected: a_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let a_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -604,6 +610,7 @@ mod tests {
 expected_pubkey: a_pub,
 pubkey_connected: b_connected_sender,
 pubkey_disconnected: b_disconnected_sender,
 disconnected_flag: AtomicBool::new(false),
 msg_events: Mutex::new(Vec::new()),
 });
 let b_manager = Arc::new(PeerManager::new(MessageHandler {
Expand All@@ -624,27 +631,29 @@ mod tests {
 } else { panic!("Failed to bind to v4 localhost on common ports"); };

 let (sender, _receiver) = mpsc::channel(2);
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, tokio::net::TcpStream::from_std(conn_a).unwrap());
 let fut_b = super::setup_inbound(b_manager, sender, tokio::net::TcpStream::from_std(conn_b).unwrap());
 let fut_a = super::setup_outbound(Arc::clone(&a_manager), sender.clone(), b_pub, conn_a);
 let fut_b = super::setup_inbound(b_manager, sender, conn_b);

 tokio::time::timeout(Duration::from_secs(10), a_connected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_connected.recv()).await.unwrap();

 a_handler.msg_events.lock().unwrap().push(MessageSendEvent::HandleError {
 node_id: b_pub, action: ErrorAction::DisconnectPeer { msg: None }
 });
 assert!(a_disconnected.try_recv().is_err());
 assert!(b_disconnected.try_recv().is_err());
 assert!(!a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(!b_handler.disconnected_flag.load(Ordering::SeqCst));

 a_manager.process_events();
 tokio::time::timeout(Duration::from_secs(10), a_disconnected.recv()).await.unwrap();
 tokio::time::timeout(Duration::from_secs(1), b_disconnected.recv()).await.unwrap();
 assert!(a_handler.disconnected_flag.load(Ordering::SeqCst));
 assert!(b_handler.disconnected_flag.load(Ordering::SeqCst));

 fut_a.await;
 fut_b.await;
	}

	#[tokio::test(threaded_scheduler)]
	#[tokio::test(flavor = "multi_thread")]
	async fn basic_threaded_connection_test() {
 do_basic_connection_test().await;
	}
Expand Down