tokio integration

[diwic]

It would be nice if it was easier to use this dbus library in an async fashion.

[brson]

@antoyo, @zeenix and I spent some time hacking on a proof of concept of tokio integration. We managed to create a basic example client, started hacking on an example server, and then began feeling a bit over our heads w/r/t dbus-rs internals. I hope that, at the least, this work can serve as guidance toward the real solution.

Here's the branch: https://github.com/brson/dbus-rs/tree/tokio

The entire implementation is contained in tokio.rs. There is a client example and server.

The comment at the top of tokio.rs describes the integration strategy, which is derived from tokio-curl.

This patch integrates directly into dbus-rs for simplicity, but it would be just as well as a standalone crate.

One unfortunate thing about this integration is that it needs to know directly about tokio-core and mio in order to communicate information about the fds that come out of dbus. So it's not completely abstracted from the underlying event loop, and will need specific integration into future tokio event loop implementations. I figure solutions there will shake out in time.

Since we didn't make much progress on the server implementation, we just wrote the server example to demonstrate what might be a plausible API. In the client code we took a strategy of encapsulating Connection in TokioConnection; for the server we might similarly encapsulate it in TokioServer, or possibly give TokioConnection a method that returns a server future of some kind, which can then be used to drive an event loop (this seems like it may be more consistent with the existing Connection design).

The big thing we got stuck on was understanding the parameterization of Factory over a function abstraction, and in particular what MTFn does. I suspect we need some serious guidance at this point on how to proceed in a way that is congruent with the existing design.

FYI, I don't personally expect to continue working on this, but I'm hopeful @antoyo and @zeenix can continue in collaboration with @diwic.

[diwic]

MTFn, MTFnMut and MTSync represent methods that are Fn, FnMut and Fn + Send + Sync. This was needed because someone asked for being able to run the tree from several threads in parallel. Then someone else asked for the tree methods to be able to mutate its environment. So I made the tree generic over different function types. But it does complicate matters quite a bit, so I'm not totally happy about it. Not sure how to integrate async-ness into that...

[diwic]

@brson Looking at the server example, I see you added an f.async_method which is not implemented. Could you answer:

1. This function takes a callback. Does this callback return a AsyncMethodResult? Is an AsyncMethodResult a Future of some sort?
2. When dbus-rs gets this AsyncMethodResult back from a method, I guess it's supposed to hand it over to an event loop somehow, wait for it to complete into a normal MethodResult, and then handle that MethodResult. But I'm lost on how to do that, i e, how to hand an AsyncMethodResult over to the event loop and how it then hands me a MethodResult later?

[brson]

This function takes a callback. Does this callback return a AsyncMethodResult? Is an AsyncMethodResult a Future of some sort?

Yes, that was the intent.

When dbus-rs gets this AsyncMethodResult back from a method, I guess it's supposed to hand it over to an event loop somehow, wait for it to complete into a normal MethodResult, and then handle that MethodResult. But I'm lost on how to do that, i e, how to hand an AsyncMethodResult over to the event loop and how it then hands me a MethodResult later?

So async_method itself does not return a future, it's only the callback provided to async_method that returns a future. That callback can be considered a method call future factory. The idea is that the 'engine' itself calls that callback (factory) to construct a new future, then itself is responsible for scheduling a task based on that future.

I cannot say that this is a correct idea, but it felt right at a first pass.

[diwic]

then itself is responsible for scheduling a task based on that future.

Right, it's specifically this part I'm wondering about. How do I schedule a task based on a future?
(I'm not even sure about the conceptual difference between a task and a future.)

[diwic]

Having read up a little about futures and tasks, I think maybe it will be better for tree.handle_async (or something like that) to return those futures directly and leave it up to the caller to combine, turning into tasks etc.

[diwic]

@brson @zeenix

Ok, so I tried from from the other end, i e, trying to build a tree with async methods. I figured this would be the easier part, since it does not do much fd related stuff. But I haven't come very far yet. I'm stuck on two things:

1. The example fails to compile here with cannot move out of captured outer variable in an Fn closure. I don't know how to resolve that - and_then takes an FnOnce closure so that error doesn't make sense, but maybe this has to do with that we're inside an Fn closure...?

2. As for AMethodResult (short for brson's AsyncMethodResult), I'm not sure what it should contain? A BoxFuture? Or were we supposed to avoid boxes? Or should the callback return an R: IntoFuture perhaps? And what about the Err type of the future?

[zeenix]

I wish I could reply but when it comes to code, I was only looking over the shoulder of @brson and making observations and suggestions based on my guesses about the code. So i hope @brson will be able to remember and respond.

[antoyo]

@diwic :
For your first issue, it is indeed caused by using a Fn. You see, you move the signal into the FnOnce of the and_then call but this closure (the Fn one) can be called again.
A fix is to clone() it:

                let signal2 = signal.clone();
                sleep_future.and_then(move |_| {
                    let s = format!("Hello {}!", sender);
                    let mret = mret.append1(s);
                    let sig = signal2.msg(&pname, &iname).append1(&*sender);

                    // Two messages will be returned - one is the method return (and should always be there),
                    // and in our case we also have a signal we want to send at the same time.
                    Ok(vec!(mret, sig))
                })

For your other issue, you should most probably avoid BoxFuture as it is confusing (Box<Future> behave more as expected).
In any case, you probably want to avoid boxing the future right away.
One way to avoid that is to use -> impl Future which is currently unstable.
All the ways to return a Future are described here.
You Err type will probably be MethodErr.

[diwic]

@antoyo

You see, you move the signal

Ah, right. It would have been helpful if the compiler could have given a hint about what variable it was complaining about. Fixed now.

As for AMethodResult:

One way to avoid that is to use -> impl Future which is currently unstable.
All the ways to return a Future are described here.

AMethodResult is not a generic type, so -> impl Future makes no sense. From the options in the link, AMethodResult is the "Custom type" option.

You Err type will probably be MethodErr.

Okay, I'll started with this (see just committed code). One piece fell in place when I changed the example to convert a TimerError to a MethodErr. I'll continue later, need to go to work now :-)

[antoyo]

@diwic: I reported an issue for the hint and it was fixed.

For the -> impl Future thing, it is not for generic type, but for types implementing a trait.
So, your AMethodResult will implement Future so that would work.
But since you already have a concrete type, it is better to use it than using the -> impl feature which is unstable.

[diwic]

As you have perhaps noticed, not much has happened during the last few weeks. This is partially due to other things taking priority (and I expect the same for a few weeks more), but also due to the Tokio architecture being hard to understand for me. I've filed tokio-rs/tokio#9 now, let's see if that gives something.

[diwic]

Ok, so with some help in that other thread, I was able to get somewhere, so I have somewhat of a working client. Here's my test case:

let conn = Rc::new(Connection::get_private(BusType::Session).unwrap());
let mut core = Core::new().unwrap();
let awatcher = AWatcher::new(conn.clone(), &core.handle()).unwrap();

let (tx, rx) = futures::sync::oneshot::channel();
let m = Message::new_method_call("org.freedesktop.DBus", "/", "org.freedesktop.DBus", "ListNames").unwrap();
awatcher.append(conn.send_with_reply(m, move |r| {
    let z: Vec<&str> = r.get1().unwrap();
    println!("got reply: {:?}", z);
    tx.send(()).unwrap();
}));

core.handle().spawn(awatcher.for_each(|_| Ok(())));
core.run(rx).unwrap();

Feel free to review / try it out etc. I think the main differences between mine and brson's implementations are:

• Calling a method does not return a future. Instead you append it to an instance of AWatcher as you see in the example. Maybe this can be improved in the future (pun not intended :-) ).
• internally, I have one Future per fd.
• my version does not interfere with original dbus crate internals, dbus-tokio is a separate crate. It instead interacts with the somewhat new MsgHandler trait I've been working on, that can also be used without tokio.

[zeenix]

@diwic cool. :)