Is it possible for wamp to process one RPC call while a different RPC call has not returned yet?

[jamespwebster]

I'm trying to use cppwamp in a multi-thread program, and would like each thread to be able to make RPC calls that might take a long time to return. The behavior I have seen so far indicates that WAMP is waiting for one RPC to return before it executes the next RCP. This happens even though the functions are being called from different threads.

In the code below, a RPC is made to a function that takes 10 seconds to return. One second later an RPC is made on a different thread to a different function that returns immediately. I would like it if the second function would return to the caller immediately, but what I see when I run it is that this function blocks until the slow function finishes.

Is there some other way to set this program up that would get me where I want to go? Or is this a fundamental limitation of WAMP?

I'm sorry if this is the incorrect forum to ask a question like this, but I figured it might be an attribute of cppwamp rather than of crossbar, and I was unsure where else to go to talk about it.

Thanks,

////////////////CALLER CODE/////////////////////////
#include <ctime>
#include <iostream>
#include <cppwamp/conversion.hpp>
#include <cppwamp/corosession.hpp>
#include <cppwamp/json.hpp>
#include <cppwamp/tcp.hpp>
#include <cppwamp/unpacker.hpp>
#include <thread>
#include <boost/thread/thread.hpp>
const std::string realm = "cppwamp.demo.time";
const std::string address = "localhost";
const short port = 12345;
void quick_job(){
    wamp::AsioService iosvc;
    auto tcp = wamp::connector<wamp::Json>(iosvc, wamp::TcpHost(address,port));
    using namespace wamp;
    auto session = CoroSession<>::create(tcp);
    boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
    {
        session->connect(yield);
        session->join(Realm(realm), yield);
        session->call(Rpc("quick_return"), yield);
        std::cout << "Quick return\n";
        Reason aReason;
        session->leave(aReason, yield);
        session->disconnect();
    });
    iosvc.run();
}
void slow_job(){
    wamp::AsioService iosvc;
    auto tcp = wamp::connector<wamp::Json>(iosvc, wamp::TcpHost(address,port));
    using namespace wamp;
    auto session = CoroSession<>::create(tcp);
    boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
    {
        session->connect(yield);
        session->join(Realm(realm), yield);
        session->call(Rpc("slow_return"), yield);
        std::cout << "Slow return\n";
        Reason aReason;
        session->leave(aReason, yield);
        session->disconnect();
    });
    iosvc.run();
}
int main()
{
    boost::thread s(slow_job);
    boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
    boost::thread q(quick_job);
    s.join();
    q.join();
    return 0;
}


////////////////SERVICE CODE/////////////////////////
#include <chrono>
#include <ctime>
#include <iostream>
#include <cppwamp/conversion.hpp>
#include <cppwamp/corosession.hpp>
#include <cppwamp/msgpack.hpp>
#include <cppwamp/json.hpp>
#include <cppwamp/tcp.hpp>
#include <boost/thread/thread.hpp>
const std::string realm = "cppwamp.demo.time";
const std::string address = "localhost";
const short port = 12345;
wamp::Outcome quick_return(wamp::Invocation){
    return wamp::Result().withArgs("Done");
}
wamp::Outcome slow_return(wamp::Invocation){
    boost::this_thread::sleep(boost::posix_time::milliseconds(10000));
    return wamp::Result().withArgs("Done");
}
int main()
{
    program_on = true;
    wamp::AsioService iosvc;
    auto tcp = wamp::connector<wamp::Json>(iosvc, wamp::TcpHost(address,port));
    using namespace wamp;
    auto session = CoroSession<>::create(tcp);
    boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
    {
        session->connect(yield);
        session->join(Realm(realm), yield);
        session->enroll(Procedure("quick_return"), &quick_return, yield);
        session->enroll(Procedure("slow_return"), &slow_return, yield);
    });
    iosvc.run();
    return 0;
}

[ecorm]

@jamespwebster Please use a Markdown fenced code block and fix the indentation so that the code is readable.

[jamespwebster]

:-) Thanks for the tip

[ecorm]

Thank you for the minimal, complete, and verifiable example.

Your RPC handlers on the callee side should never block. Otherwise, they'll prevent the io_service from executing other pending tasks (like responding to another RPC call).

If your RPC handler cannot quickly return a result, it should return an Outcome::deferred(), then use Invocation::yield() at a later time when the RPC has completed its work. This is illustrated in Deferred Outcomes in the reference documentation:

Outcome Client::addPerson(Invocation inv, Object person)
{
    // We need to issue another RPC to the database service before we can
    // fulfill this RPC.
    boost::asio::spawn(iosvc_, [this, inv](boost::asio::yield_context yield)
    {
        auto dbResult = session_->call( Rpc("db.add").withArgs(person),
                                        yield );
        // Manually send the result back to the caller
        auto personId = dbResult[0].to<int>();
        inv.yield({personId});
    });
    // We don't know the result yet as this point. Return a deferred outcome
    // to indicate that we'll send the result manually in a different
    // asynchronous context.
    return Outcome::deferred();
}

Your slow_return RPC should do it's work asynchronously via the io_service, or in a separate thread, so that it doesn't block the CoroSession object from processing incoming/outgoing WAMP messages. If you tell me what is the nature of the work you're doing in slow_return, then I can give you more detailed guidance on how you should proceed. What exactly makes slow_return slow? Is it performing a long computation? Is it calling blocking OS functions? Does it need to introduce a sleep delay?

Please note that a CoroSession instance is not thread-safe (only one thread should be invoking it's methods). It should be safe, however, to have separate CoroSession objects in separate threads, like you've done in your "CALLER CODE" snippet.

The main reason why Session/CoroSession is not thread safe is that I reuse WAMP message objects. I'm considering revising Session/CoroSession so that they are thread safe, but that might sacrifice a bit of performance. I haven't done any performance measurements yet, so it's possible the loss of performance would be insignificant.

[jamespwebster]

Thanks, that clears it up. Fortunately this property only affects my test code, and would not actually affect my project.

Specifically, here is what the code I am writing wants to do:

John produces data which must be processed. John starts a thread, which creates a CoroSession and makes an RPC to Mary to send the data over. John starts a timer and goes to sleep. When the thread gets a response from Mary, it wakes John and reports successful completion of the task. If the timer finishes and John wakes up before Mary has replied, John decides that Mary has taken too long, so John starts another thread, which sends the same data to Sally. Mary and Sally do not share a thread.

In the test code I wrote, Mary and Sally were the same function in the same thread, so this blocked.

[ecorm]

You can fix your "SERVICE" test code, without involving multiple threads, by making use of wamp::Outcome::deferred() and boost::asio::steady_timer:

wamp::Outcome quick_return(wamp::Invocation){
    return wamp::Result().withArgs("Done");
}
wamp::Outcome slow_return(wamp::AsioService iosvc, wamp::Invocation inv){
    boost::asio::spawn(iosvc, [&iosvc, inv](boost::asio::yield_context yield)
    {
        boost::asio::steady_timer timer(iosvc);
        timer.expires_from_now(std::chrono::milliseconds(1000));
        timer.wait(yield);
        inv.yield({"Done"});
    });
    return wamp::Outcome::deferred();
}
int main()
{
    program_on = true;
    wamp::AsioService iosvc;
    auto tcp = wamp::connector<wamp::Json>(iosvc, wamp::TcpHost(address,port));
    using namespace wamp;
    auto session = CoroSession<>::create(tcp);
    boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
    {
        session->connect(yield);
        session->join(Realm(realm), yield);
        session->enroll(Procedure("quick_return"), &quick_return, yield);
        session->enroll(Procedure("slow_return"),
                        [&iosvc](wamp::Invocation inv){slow_return(iosvc, inv);},
                        yield);
    });
    iosvc.run();
    return 0;
}

If have not compiled or tested the above, but I hope it helps you anyway.

[jamespwebster]

Thank you, that is helpful. I'm new to c++ and find coroutines to be difficult to understand. Hopefully a little more exposure to it will make it seem more natural.

[ecorm]

I'm new to c++ and find coroutines to be difficult to understand.

Sadly, there is not a lot of literature out there on boost::asio and coroutines. I get the feeling that I'm a pioneer in that style of programming in C++.

John produces data which must be processed. John starts a thread, which creates a CoroSession and makes an RPC to Mary to send the data over. John starts a timer and goes to sleep. When the thread gets a response from Mary, it wakes John and reports successful completion of the task. If the timer finishes and John wakes up before Mary has replied, John decides that Mary has taken too long, so John starts another thread, which sends the same data to Sally. Mary and Sally do not share a thread.

If call timeouts were supported on your WAMP router, this would be easy. To my knowledge, Crossbar does not yet support this feature.

Your proposed way of implementing call timeouts manually (on the "John" side) should be possible without multithreading by making use of boost::asio::steady_timer for the timeout. Here is some rough pseudocode:

boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
{
    session->connect(yield);
    session->join(Realm(realm), yield);
    iosvc.stop(); // Halts the iosvc.run() loop
});
iosvc.run(); // Runs above coroutine to completion
iosvc.reset(); // Required to resume io_service loop

boost::asio::steady_timer timer(iosvc);

// Spawn two parallel coroutines:
boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
{
    session->call(Rpc("mary"), yield);
    timer.cancel();
    std::cout << "Mary return\n";
});

boost::asio::spawn(iosvc, [&](boost::asio::yield_context yield)
{
    timer.expires_from_now(std::chrono::milliseconds(1000));
    boost::system::error_code ec;
    timer.wait(yield[ec]);
    if (!ec)
    {
        std::cout << "Mary call timed out\n";
        session->call(Rpc("sally"), yield);
        std::cout << "Sally return\n";
    }
});

iosvc.run(); // Execute both coroutines

I do plan on supporting client-side timeouts in CppWAMP, but I don't have a timetable of when it's going to be done.

[jamespwebster]

I was surprised that crossbar does not have a timeout! Wamp is still pretty new I guess. Bonefish doesn't have one either as far as I can tell. Have not looked at other routers yet.

No doubt coroutines have some big advantages (e.g. less overhead, no synchronization) once you get comfortable using them.

[ecorm]

I was surprised that wamp does not have a timeout!

The WAMP spec does describe a Call Timeouts feature. It just that the maintainers of the Crossbar router haven't had then chance to implement it yet.

[ecorm]

@jamespwebster You may find this discussion on the CppWAMP Google Groups to be informative, particularly my example at the end of the thread.

[jamespwebster]

Thank you!

[ecorm]

Caller-initiated timeouts were added to CppWAMP v0.8.0. See https://ecorm.github.io/cppwamp/_rpc_tutorial.html#autotoc_md34