mirrors/ntex
1
0
Fork 0
mirror of https://github.com/ntex-rs/ntex.git synced 2025-04-03 21:07:39 +03:00
Code Issues Projects Releases Packages Wiki Activity

replace tokio channels

Browse source
This commit is contained in:
Nikolay Kim 2021-12-18 12:04:29 +06:00
parent b8a8e98c1c
commit 66524a89a8
4 changed files with 51 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

4
ntex-rt/Cargo.toml
View file

@ -25,7 +25,9 @@ tokio = ["tok-io", "ntex-io/tokio"]
ntex-bytes = "0.1.7"
ntex-io = "0.1.0"
ntex-util = "0.1.2"
async-oneshot = "0.5.0"
async-channel = "1.6.1"
log = "0.4"
pin-project-lite = "0.2"
tok-io = { version = "1", package = "tokio", default-features = false, features = ["rt", "signal", "sync"], optional = true }
tok-io = { version = "1", package = "tokio", default-features = false, features = ["rt", "net", "signal"], optional = true }

59
ntex-rt/src/arbiter.rs
View file

@ -3,8 +3,12 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll};
use std::{cell::RefCell, collections::HashMap, fmt, future::Future, pin::Pin, thread};
use tok_io::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
use tok_io::sync::oneshot::{channel, error::RecvError, Sender};
use async_channel::{unbounded, Receiver, Sender};
use async_oneshot as oneshot;
use ntex_util::Stream;
//use tok_io::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
//use tok_io::sync::oneshot::{channel, error::RecvError, Sender};
use crate::{system::System, Runtime};
@ -26,7 +30,7 @@ pub(super) enum ArbiterCommand {
/// and futures. When an Arbiter is created, it spawns a new OS thread, and
/// hosts an event loop. Some Arbiter functions execute on the current thread.
pub struct Arbiter {
sender: UnboundedSender<ArbiterCommand>,
sender: Sender<ArbiterCommand>,
thread_handle: Option<thread::JoinHandle<()>>,
}
@ -49,8 +53,9 @@ impl Clone for Arbiter {
}
impl Arbiter {
#[allow(clippy::borrowed_box)]
pub(super) fn new_system(rt: &Box<dyn Runtime>) -> Self {
let (tx, rx) = unbounded_channel();
let (tx, rx) = unbounded();
let arb = Arbiter::with_sender(tx);
ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
@ -72,7 +77,7 @@ impl Arbiter {
/// Stop arbiter from continuing it's event loop.
pub fn stop(&self) {
let _ = self.sender.send(ArbiterCommand::Stop);
let _ = self.sender.try_send(ArbiterCommand::Stop);
}
/// Spawn new thread and run event loop in spawned thread.
@ -81,7 +86,7 @@ impl Arbiter {
let id = COUNT.fetch_add(1, Ordering::Relaxed);
let name = format!("ntex-rt:worker:{}", id);
let sys = System::current();
let (arb_tx, arb_rx) = unbounded_channel();
let (arb_tx, arb_rx) = unbounded();
let arb_tx2 = arb_tx.clone();
let handle = thread::Builder::new()
@ -90,7 +95,7 @@ impl Arbiter {
let rt = crate::create_runtime();
let arb = Arbiter::with_sender(arb_tx);
let (stop, stop_rx) = channel();
let (stop, stop_rx) = oneshot::oneshot();
STORAGE.with(|cell| cell.borrow_mut().clear());
System::set_current(sys);
@ -105,7 +110,7 @@ impl Arbiter {
// register arbiter
let _ = System::current()
.sys()
.send(SystemCommand::RegisterArbiter(id, arb));
.try_send(SystemCommand::RegisterArbiter(id, arb));
// run loop
rt.block_on(Box::pin(async move {
@ -115,7 +120,7 @@ impl Arbiter {
// unregister arbiter
let _ = System::current()
.sys()
.send(SystemCommand::UnregisterArbiter(id));
.try_send(SystemCommand::UnregisterArbiter(id));
})
.unwrap_or_else(|err| {
panic!("Cannot spawn an arbiter's thread {:?}: {:?}", &name, err)
@ -132,21 +137,23 @@ impl Arbiter {
where
F: Future<Output = ()> + Send + Unpin + 'static,
{
let _ = self.sender.send(ArbiterCommand::Execute(Box::new(future)));
let _ = self
.sender
.try_send(ArbiterCommand::Execute(Box::new(future)));
}
/// Send a function to the Arbiter's thread. This function will be executed asynchronously.
/// A future is created, and when resolved will contain the result of the function sent
/// to the Arbiters thread.
pub fn exec<F, R>(&self, f: F) -> impl Future<Output = Result<R, RecvError>>
pub fn exec<F, R>(&self, f: F) -> impl Future<Output = Result<R, oneshot::Closed>>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
R: Sync + Send + 'static,
{
let (tx, rx) = channel();
let (mut tx, rx) = oneshot::oneshot();
let _ = self
.sender
.send(ArbiterCommand::ExecuteFn(Box::new(move || {
.try_send(ArbiterCommand::ExecuteFn(Box::new(move || {
if !tx.is_closed() {
let _ = tx.send(f());
}
@ -162,7 +169,7 @@ impl Arbiter {
{
let _ = self
.sender
.send(ArbiterCommand::ExecuteFn(Box::new(move || {
.try_send(ArbiterCommand::ExecuteFn(Box::new(move || {
f();
})));
}
@ -215,7 +222,7 @@ impl Arbiter {
})
}
fn with_sender(sender: UnboundedSender<ArbiterCommand>) -> Self {
fn with_sender(sender: Sender<ArbiterCommand>) -> Self {
Self {
sender,
thread_handle: None,
@ -233,8 +240,8 @@ impl Arbiter {
}
struct ArbiterController {
stop: Option<Sender<i32>>,
rx: UnboundedReceiver<ArbiterCommand>,
stop: Option<oneshot::Sender<i32>>,
rx: Receiver<ArbiterCommand>,
}
impl Drop for ArbiterController {
@ -255,11 +262,11 @@ impl Future for ArbiterController {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
match Pin::new(&mut self.rx).poll_recv(cx) {
match Pin::new(&mut self.rx).poll_next(cx) {
Poll::Ready(None) => return Poll::Ready(()),
Poll::Ready(Some(item)) => match item {
ArbiterCommand::Stop => {
if let Some(stop) = self.stop.take() {
if let Some(mut stop) = self.stop.take() {
let _ = stop.send(0);
};
return Poll::Ready(());
@ -286,15 +293,15 @@ pub(super) enum SystemCommand {
#[derive(Debug)]
pub(super) struct SystemArbiter {
stop: Option<Sender<i32>>,
commands: UnboundedReceiver<SystemCommand>,
stop: Option<oneshot::Sender<i32>>,
commands: Receiver<SystemCommand>,
arbiters: HashMap<usize, Arbiter>,
}
impl SystemArbiter {
pub(super) fn new(
stop: Sender<i32>,
commands: UnboundedReceiver<SystemCommand>,
stop: oneshot::Sender<i32>,
commands: Receiver<SystemCommand>,
) -> Self {
SystemArbiter {
commands,
@ -309,7 +316,7 @@ impl Future for SystemArbiter {
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
match Pin::new(&mut self.commands).poll_recv(cx) {
match Pin::new(&mut self.commands).poll_next(cx) {
Poll::Ready(None) => return Poll::Ready(()),
Poll::Ready(Some(cmd)) => match cmd {
SystemCommand::Exit(code) => {
@ -318,7 +325,7 @@ impl Future for SystemArbiter {
arb.stop();
}
// stop event loop
if let Some(stop) = self.stop.take() {
if let Some(mut stop) = self.stop.take() {
let _ = stop.send(code);
}
}

26
ntex-rt/src/builder.rs
View file

@ -1,8 +1,8 @@
use std::{cell::RefCell, future::Future, io, rc::Rc};
use async_channel::unbounded;
use async_oneshot as oneshot;
use ntex_util::future::lazy;
use tok_io::sync::mpsc::unbounded_channel;
use tok_io::sync::oneshot::{channel, Receiver};
use crate::arbiter::{Arbiter, SystemArbiter};
use crate::{create_runtime, Runtime, System};
@ -63,8 +63,8 @@ impl Builder {
where
F: FnOnce() + 'static,
{
let (stop_tx, stop) = channel();
let (sys_sender, sys_receiver) = unbounded_channel();
let (stop_tx, stop) = oneshot::oneshot();
let (sys_sender, sys_receiver) = unbounded();
let rt = create_runtime();
@ -86,7 +86,7 @@ impl Builder {
#[must_use = "SystemRunner must be run"]
pub struct SystemRunner {
rt: Box<dyn Runtime>,
stop: Receiver<i32>,
stop: oneshot::Receiver<i32>,
_system: System,
}
@ -108,7 +108,7 @@ impl SystemRunner {
Ok(())
}
}
Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)),
Err(_) => Err(io::Error::new(io::ErrorKind::Other, "Closed")),
}
}
@ -142,6 +142,7 @@ impl<T> BlockResult<T> {
}
#[inline]
#[allow(clippy::borrowed_box)]
fn block_on<F, R>(rt: &Box<dyn Runtime>, fut: F) -> BlockResult<R>
where
F: Future<Output = R> + 'static,
@ -168,19 +169,12 @@ mod tests {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let rt = tok_io::runtime::Builder::new_current_thread()
.build()
.unwrap();
let local = tok_io::task::LocalSet::new();
let runner = crate::System::build()
.stop_on_panic(true)
.finish_with(&local);
let runner = crate::System::build().stop_on_panic(true).finish();
tx.send(System::current()).unwrap();
let _ = rt.block_on(local.run_until(runner.run()));
let _ = runner.run();
});
let mut s = System::new("test");
let s = System::new("test");
let sys = rx.recv().unwrap();
let id = sys.id();

10
ntex-rt/src/system.rs
View file

@ -1,5 +1,5 @@
use async_channel::Sender;
use std::{cell::RefCell, io, sync::atomic::AtomicUsize, sync::atomic::Ordering};
use tok_io::sync::mpsc::UnboundedSender;
use super::arbiter::{Arbiter, SystemCommand};
use super::builder::{Builder, SystemRunner};
@ -10,7 +10,7 @@ static SYSTEM_COUNT: AtomicUsize = AtomicUsize::new(0);
#[derive(Clone, Debug)]
pub struct System {
id: usize,
sys: UnboundedSender<SystemCommand>,
sys: Sender<SystemCommand>,
arbiter: Arbiter,
stop_on_panic: bool,
}
@ -22,7 +22,7 @@ thread_local!(
impl System {
/// Constructs new system and sets it as current
pub(super) fn construct(
sys: UnboundedSender<SystemCommand>,
sys: Sender<SystemCommand>,
arbiter: Arbiter,
stop_on_panic: bool,
) -> Self {
@ -80,10 +80,10 @@ impl System {
/// Stop the system with a particular exit code.
pub fn stop_with_code(&self, code: i32) {
let _ = self.sys.send(SystemCommand::Exit(code));
let _ = self.sys.try_send(SystemCommand::Exit(code));
}
pub(super) fn sys(&self) -> &UnboundedSender<SystemCommand> {
pub(super) fn sys(&self) -> &Sender<SystemCommand> {
&self.sys
}