Better suspend for continuation

src/ripple/rpc/impl/Coroutine.cpp

@@ -24,60 +24,56 @@
 
 namespace ripple {
 namespace RPC {
+namespace {
 
 using CoroutineType = Continuation;
-using CoroutinePull = boost::coroutines::coroutine <CoroutineType>::pull_type;
-using CoroutinePush = boost::coroutines::coroutine <CoroutineType>::push_type;
+using BoostCoroutine = boost::coroutines::asymmetric_coroutine<CoroutineType>;
+using Pull = BoostCoroutine::pull_type;
+using Push = BoostCoroutine::push_type;
 
-struct Coroutine::Impl : public std::enable_shared_from_this <Coroutine::Impl>
+void runOnCoroutineImpl(std::shared_ptr<Pull> pull)
 {
-    Impl (CoroutinePull&& pull_) : pull (std::move (pull_))
+    while (*pull)
     {
-    }
+        (*pull)();
 
-    CoroutinePull pull;
+        if (! *pull)
+            return;
 
-    void run()
-    {
-        while (pull)
+        if (auto continuation = pull->get())
         {
-            pull();
-
-            if (! pull)
-                return;
-
-            if (auto continuation = pull.get())
-            {
-                auto that = shared_from_this();
-                continuation ([that] () { that->run(); });
-                return;
-            }
+            continuation ([pull] () { runOnCoroutineImpl(pull); });
+            return;
         }
     }
-};
-
-Coroutine::Coroutine (SuspendCallback const& suspendCallback)
-{
-    CoroutinePull pull ([suspendCallback] (CoroutinePush& push)
-    {
-        Suspend suspend = [&push] (CoroutineType const& cbc) {
-            push (cbc);
-        };
-        suspend ({});
-        suspendCallback (suspend);
-    });
-
-    impl_ = std::make_shared<Impl> (std::move (pull));
 }
 
-Coroutine::~Coroutine() = default;
+} // namespace
 
-void Coroutine::run()
+void runOnCoroutine(Coroutine const& coroutine)
 {
-    assert (impl_);
-    if (impl_)
-        impl_->run();
-    impl_.reset();
+    auto pullFunction = [coroutine] (Push& push) {
+        Suspend suspend = [&push] (CoroutineType const& cbc) {
+            if (push)
+                push (cbc);
+        };
+
+        // Run once doing nothing, to get the other side started.
+        suspend([] (Callback const& callback) { callback(); });
+
+        // Now run the coroutine.
+        coroutine(suspend);
+    };
+
+    runOnCoroutineImpl(std::make_shared<Pull>(pullFunction));
+}
+
+void runOnCoroutine(UseCoroutines useCoroutines, Coroutine const& coroutine)
+{
+    if (useCoroutines == UseCoroutines::yes)
+        runOnCoroutine(coroutine);
+    else
+        coroutine(dontSuspend);
 }
 
 } // RPC

src/ripple/rpc/impl/Yield.cpp

@@ -18,6 +18,7 @@
 //==============================================================================
 
 #include <BeastConfig.h>
+#include <ripple/app/main/Application.h>
 #include <ripple/basics/BasicConfig.h>
 #include <ripple/rpc/Yield.h>
 #include <ripple/rpc/tests/TestOutputSuite.test.h>
@@ -25,6 +26,34 @@
 namespace ripple {
 namespace RPC {
 
+static
+UseCoroutines defaultUseCoroutines = UseCoroutines::no;
+
+Suspend const dontSuspend = [] (Continuation const& continuation)
+{
+    continuation([] () {});
+};
+
+namespace {
+
+void runOnJobQueue(std::string const& name, Callback const& callback)
+{
+    boost::function <void (Job&)> cb([callback] (Job&) { callback(); });
+    getApp().getJobQueue().addJob(jtCLIENT, name, cb);
+};
+
+Callback suspendForJobQueue(Suspend const& suspend, std::string const& jobName)
+{
+    assert(suspend);
+    return Callback( [suspend, jobName] () {
+        suspend([jobName] (Callback const& callback) {
+            runOnJobQueue(jobName, callback);
+        });
+    });
+}
+
+} // namespace
+
 Json::Output chunkedYieldingOutput (
     Json::Output const& output, Callback const& yield, std::size_t chunkSize)
 {
@@ -44,7 +73,6 @@ Json::Output chunkedYieldingOutput (
     };
 }
 
-
 CountedYield::CountedYield (std::size_t yieldCount, Callback const& yield)
         : yieldCount_ (yieldCount), yield_ (yield)
 {
@@ -52,10 +80,7 @@ CountedYield::CountedYield (std::size_t yieldCount, Callback const& yield)
 
 void CountedYield::yield()
 {
-    if (!yield_)
-        return;
-
-    if (yieldCount_)
+    if (yieldCount_ && yield_)
     {
         if (++count_ >= yieldCount_)
         {
@@ -65,28 +90,38 @@ void CountedYield::yield()
     }
 }
 
-YieldStrategy makeYieldStrategy (Section const& s)
+UseCoroutines useCoroutines(BasicConfig const& config)
 {
+    if (auto use = config["section"].get<bool>("use_coroutines"))
+        return *use ? UseCoroutines::yes : UseCoroutines::no;
+    return defaultUseCoroutines;
+}
+
+YieldStrategy makeYieldStrategy (BasicConfig const& config)
+{
+    auto s = config["section"];
     YieldStrategy ys;
     ys.streaming = get<bool> (s, "streaming") ?
             YieldStrategy::Streaming::yes :
             YieldStrategy::Streaming::no;
-    ys.useCoroutines = get<bool> (s, "use_coroutines") ?
-            YieldStrategy::UseCoroutines::yes :
-            YieldStrategy::UseCoroutines::no;
-    ys.byteYieldCount = get<std::size_t> (s, "byte_yield_count");
+    ys.useCoroutines = useCoroutines(config);
     ys.accountYieldCount = get<std::size_t> (s, "account_yield_count");
     ys.transactionYieldCount = get<std::size_t> (s, "transaction_yield_count");
 
     return ys;
 }
 
-Continuation callbackOnJobQueue (
-    JobQueue& jobQueue, std::string const& name, JobType jobType)
+JobQueueSuspender::JobQueueSuspender(
+    Suspend const& susp, std::string const& jobName)
+        : suspend(susp ? susp : dontSuspend),
+          yield(suspendForJobQueue(suspend, jobName))
+{
+    // There's a non-empty jobName exactly if there's a non-empty Suspend.
+    assert(!(susp && jobName.empty()));
+}
+
+JobQueueSuspender::JobQueueSuspender() : JobQueueSuspender({}, {})
 {
-    return Continuation ([name, jobType, &jobQueue] (Callback const& cb) {
-        jobQueue.addJob (jobType, name, [cb] (Job&) { cb(); });
-    });
 }
 
 } // RPC