Add ability to shutdown the server from the outside.

[dpc]

Add ability to shutdown the server from the outside.

[dpc]

Last time I though about it, it seemed to me that shutting down from the outside is unnecessary. One can always notify a coroutine to do it from the inside. This simplifies the API. mioco::start is blocking call anyway, so can't work as posted in the issue description.

[dpc]

I'm working on some basic implementation.

[troplin]

Last time I though about it, it seemed to me that shutting down from the outside is unnecessary. One can always notify a coroutine to do it from the inside. This simplifies the API. mioco::start is blocking call anyway, so can't work as posted in the issue description.

@dpc Yes it is possible but tedious, because you can no longer just use blocking APIs.
You need some wakeup object (socket, pipe, mailbox...) that you can select on together with the socket that you are actually interested in, just to listen for an external event.
Or as an alternative just use async io, but that's not the point of coroutines.

Thanks to lifetimes it probably wouldn't even be possible to use the same object for all coroutines. Or is it?

So the most important point is the possibility to interrupt blocking calls. They could just fail with EINTR, I think that would be fine.

Another idea that goes a bit further would be some kind of shutdown routine for every coroutine that is executed in case of shutdown. Like mioco::{start, spawn}, but with an additional parameter.
And then one should be able to define regions that cannot be interrupted by a shutdown.

Hm, that would actually be like unix signals for coroutines. Not really surprising though...

EDIT:
The advantage of the "unix signal" solution would be better composability. For example, BufReader::read_line just ignores EINTR and thus will still block.
This is actually a problem with all solutions that are not built into mioco. Basically everything that uses multiple read calls whithout a possiblity to interrupt in between is a problem.

[dpc]

@troplin : I did not really understand what you're saying.

Just to give you some overview on how I am implementing right now:

let (terminate_send, terminate_recv) = mioco::sync::mpsc::channel();
mioco::spawn(move || {
    mioco::spawn(move || {
           let _ = terminiate_recv.recv();
           mioco::shutdown();
    });

    (...) // normal inside mioco  logic
});

(...) // normal outsie mioco logic
terminate_send.send(()).unwrap();

mioco::spawn creates a mioco instance without blocking, first spawned coroutine is just sitting and waiting for a channel message to terminiate the whole instance.

Let me know if that would work for you.

[troplin]

Am 02.03.2016 um 01:07 schrieb Dawid Ciężarkiewicz [email protected]:

@troplin : I did not really understand what you're saying.

@dpc yeah I know I'm bad at explaining...

What I'm looking for is graceful shutdown, not just killing everything.

Just to give you some overview on how I am implementing right now:

let (terminate_send, terminate_recv) = mioco::sync::mpsc::channel();
mioco::spawn(move || {
mioco::spawn(move || {
let _ = terminiate_recv.recv();
mioco::shutdown();
});

(...) // normal inside mioco  logic

});

(...) // normal outsie mioco logic
terminate_send.send(()).unwrap();
mioco::spawn creates a mioco instance without blocking, first spawned coroutine is just sitting and waiting for a channel message to terminiate the whole instance.

Let me know if that would work for you.

I'm fine with the way how to trigger the shutdown, but there's no way to react on it.

What I'm really after is sending a goodbye message to every open connection.
Actually I'm even fine with using mioco::start instead of mioco::spawn because I want to trigger shutdown from a unix signal anyway, not from the main thread.

[dpc]

The forceful shutdown should work for you. Every coroutine will be unwinded, so just created an struct implementing Drop that will send "goodbye" message.

[dpc]

You can mioco::spawn, handle signals the way you wish on a native thread, and then send channel message to mioco instance to shutdown itself. Then each coroutine will drop everything on unwinding so you can reuse that as a reactor.

[dpc]

This got closed automatically - reopening for more discussion.

[troplin]

Thanks for reminding me about Drop.
So this means that streams are still fully operable after shutdown is triggered?

Is there a way of protecting certain regions from a shutdown?
I.e. if I'm just sending a reply I don't want it to be interrupted.

[dpc]

Hmm...

I was actually thinking about it, and shutdown will just call the panic from withing whatever blocking mioco call coroutine did last. So one can always std::panic::recover to catch it and potentially ignore it... which is going to confuse the current code.

Streams itself will work OK, as long as try_write is used. Should be enough to print a simple goodbye message. Blocking on any mioco io would lead to the case in which panicking coroutine switches out during the panic - again that would confuse mioco.

The above two corner cases are a little worrying. I'm not sure what exact behavior should I standardize. It seems to me - not supporting them at all is the way to go.

[troplin]

Yeah it sounds very brittle... I'd rather not rely on panics and unwinding.

So back to my original suggestion?
Maybe a mioco::interrupt() call like your mioco::shutdown() that doesn't call panic! but instead just interrupts every blocking call?
By interrupting I mean just let it fail with EINTR (io::ErrorKind::Interrupted).
That way I can check some global state and decide myself what to do.

[dpc]

@troplin: Global call to interrupt all blocked operations with EINTR is so disruptive, that I don't really find it useful in general case.

It seems to me you are looking for general notification/broadcast mechanism. It seems to me, that you want to something like the following scheme:

struct InterruptableTcpStream {
    interrupt : mioco::sync::broadcast::Receiver, // this type does not exist yet, but could be implemented
    stream : TcpStream,
}

This type would implement read and writes as select! on both real underlying stream and interrupt.

If that would work for you, I could add such broadcast mechanism to mioco. Simple implementation shouldn't be too much work: every receiver would have a reference to shared data: Arc<Spinlock<SharedData>> and when blocking on it, just adding it's mio::Sender + Token information to a vector inside it. Sender would just go through the list of all such pairs and send a wakeup message to each of them.

I'm not sure if you're aware but users can implement their own types implementing mioco::Evented, so it should be entirely possible to implement this all in your code. But I think broadcast mechanism would be so useful that I'd add it to mioco itself.

[troplin]

@dpc OTOH this is excactly what EINTR is designed for and one should probably handle it anyway. I'm not sure how/if the regular std::net::TcpStream handles EINTR though.

A general broadcasting mechanism would certainly help.
I'm not really familiar with yet with the mioco API/Internals so no, I wasn't aware of the possibility to implement my own mioco::Evented types.

[dpc]

As mioco is using only non-blocking IO, it will never receive EINTR, and I don't want to re-introduce any oddities like that to the user.

If you're OK with broadcast, I will create an issue for that, and implement it at some point.

[troplin]

Sure, I'm ok with everything that makes ist easier for me :-)
Almost everything is better than having one channel per coroutine just for signalling shutdown... which I initially thought was the only solution.

[troplin]

@dpc is it intended behavior that a panic in mioco::start crashes the entire thread? And if yes, why?
How is this supposed to work together with mioco::shutdown?

[troplin]

Well, I found a way around it but it's super ugly:

let (terminate_send, terminate_recv) = mioco::sync::mpsc::channel();
let (stopped_send, stopped_recv) = mioco::sync::mpsc::channel::<()>();
mioco::spawn(move || {
    mioco::spawn(move || {
           let _ = stopped_send;
           let _ = terminiate_recv.recv();
           mioco::shutdown();
    });

    (...) // normal inside mioco  logic
});

let _ = stopped_recv.recv();
debug!("Stopped");

[dpc]

@troplin : I don't understand the panic in mioco::start issues. Can you provide full failing code?

The stopped channel proves only that this one particular coroutine has been closed, not the whole instance. If mioco::shutdown was called in the instance started with mioco::start, the thread that called mioco::start should unblock and continue executing after all coroutines has been terminated. So I would use that to know when shutdown completed.

[troplin]

If mioco::shutdown was called in the instance started with mioco::start, the thread that called mioco::start should unblock and continue executing after all coroutines has been terminated.

@dpc That's what I expected, but instead the thread crashes.

Ok here's some "reduced" code:

Cargo.toml

[package]
name = "mioco_test"
version = "0.1.0"

[dependencies]
mioco = "0.3.0"
env_logger = "*"
log = "*"
simple-signal = "1.0.5"

src/main.rs

#[macro_use]
extern crate log;
extern crate env_logger;
extern crate simple_signal;
extern crate mioco;

use mioco::tcp;
use simple_signal::{Signals, Signal};
use std::{net,str};

const DEFAULT_LISTEN_ADDR : &'static str = "0.0.0.0:45678";
fn listend_addr() -> net::SocketAddr {
    str::FromStr::from_str(DEFAULT_LISTEN_ADDR).unwrap()
}

fn main() {
    env_logger::init().unwrap();
    info!("Starting...");
    // Handle signals
    let (shutdown_tx, shutdown_rx) = mioco::sync::mpsc::channel();
    Signals::set_handler(&[Signal::Term, Signal::Int], move |_signals| {
        info!("Recieved signal {:?}, stopping...", _signals);
        shutdown_tx.send(()).unwrap();
    });

    mioco::start(move ||{
        // Handle shutdown
        mioco::spawn(move ||{
            let _ = shutdown_rx.recv();
            mioco::shutdown();
        });

        let addr = listend_addr();
        let listener = tcp::TcpListener::bind(&addr).unwrap();
        info!("Started.");

        let _ = try!(listener.accept());
        Ok(())
    });
    info!("Stopped.");
}

Output:

Protheseus:mioco_test muelleto$ RUST_LOG=mioco_test=debug cargo run
     Running `target/debug/mioco_test`
INFO:mioco_test: Starting...
INFO:mioco_test: Started.
^CINFO:mioco_test: Recieved signal [Int], stopping...
thread '<main>' panicked at 'coroutine::jump_in: wrong state Blocked', /Users/muelleto/.multirust/toolchains/nightly/cargo/registry/src/github.com-88ac128001ac3a9a/mioco-0.3.0/src/coroutine.rs:555
note: Run with `RUST_BACKTRACE=1` for a backtrace.

[dpc]

Looks like something to do with signals. Will take care of it in #110

[dpc]

It was a bug not caught by the test suite. I've released 0.3.1 that fixes it. Thanks for reporting.

[dpc]

#111 Will address the broadcasts.

I'm closing this for now, since shutdown is implemented.

[troplin]

I'm glad it was a bug and not by design