Merge

indra/llcommon/llcoros.cpp

@@ -51,14 +51,32 @@ boost::thread_specific_ptr<LLCoros::CoroData>
 LLCoros::sCurrentCoro(LLCoros::no_cleanup);
 
 //static
-LLCoros::coro::self& LLCoros::get_self()
+LLCoros::CoroData& LLCoros::get_CoroData(const std::string& caller)
 {
     CoroData* current = sCurrentCoro.get();
     if (! current)
     {
-        LL_ERRS("LLCoros") << "Calling get_self() from non-coroutine context!" << LL_ENDL;
+        LL_ERRS("LLCoros") << "Calling " << caller << " from non-coroutine context!" << LL_ENDL;
     }
-    return *current->mSelf;
+    return *current;
+}
+
+//static
+LLCoros::coro::self& LLCoros::get_self()
+{
+    return *get_CoroData("get_self()").mSelf;
+}
+
+//static
+void LLCoros::set_consuming(bool consuming)
+{
+    get_CoroData("set_consuming()").mConsuming = consuming;
+}
+
+//static
+bool LLCoros::get_consuming()
+{
+    return get_CoroData("get_consuming()").mConsuming;
 }
 
 llcoro::Suspending::Suspending():
@@ -245,6 +263,8 @@ LLCoros::CoroData::CoroData(CoroData* prev, const std::string& name,
     // Wrap the caller's callable in our toplevel() function so we can manage
     // sCurrentCoro appropriately at startup and shutdown of each coroutine.
     mCoro(boost::bind(toplevel, _1, this, callable), stacksize),
+    // don't consume events unless specifically directed
+    mConsuming(false),
     mSelf(0)
 {
 }

indra/llcommon/llcoros.h

@@ -150,6 +150,18 @@ public:
     /// get the current coro::self& for those who really really care
     static coro::self& get_self();
 
+    /**
+     * Most coroutines, most of the time, don't "consume" the events for which
+     * they're suspending. This way, an arbitrary number of listeners (whether
+     * coroutines or simple callbacks) can be registered on a particular
+     * LLEventPump, every listener responding to each of the events on that
+     * LLEventPump. But a particular coroutine can assert that it will consume
+     * each event for which it suspends. (See also llcoro::postAndSuspend(),
+     * llcoro::VoidListener)
+     */
+    static void set_consuming(bool consuming);
+    static bool get_consuming();
+
 private:
     LLCoros();
     friend class LLSingleton<LLCoros>;
@@ -159,6 +171,7 @@ private:
     struct CoroData;
     static void no_cleanup(CoroData*);
     static void toplevel(coro::self& self, CoroData* data, const callable_t& callable);
+    static CoroData& get_CoroData(const std::string& caller);
 
     S32 mStackSize;
 
@@ -178,6 +191,8 @@ private:
         const std::string mName;
         // the actual coroutine instance
         LLCoros::coro mCoro;
+        // set_consuming() state
+        bool mConsuming;
         // When the dcoroutine library calls a top-level callable, it implicitly
         // passes coro::self& as the first parameter. All our consumer code used
         // to explicitly pass coro::self& down through all levels of call stack,

indra/llcommon/lleventcoro.cpp

@@ -167,7 +167,7 @@ LLSD llcoro::postAndSuspend(const LLSD& event, const LLEventPumpOrPumpName& requ
                  const LLEventPumpOrPumpName& replyPump, const LLSD& replyPumpNamePath)
 {
     // declare the future
-    boost::dcoroutines::future<LLSD> future(LLCoros::get_self());
+    boost::dcoroutines::future<LLSD_consumed> future(LLCoros::get_self());
     // make a callback that will assign a value to the future, and listen on
     // the specified LLEventPump with that callback
     std::string listenerName(listenerNameForCoro());
@@ -193,21 +193,25 @@ LLSD llcoro::postAndSuspend(const LLSD& event, const LLEventPumpOrPumpName& requ
                              << " about to wait on LLEventPump " << replyPump.getPump().getName()
                              << LL_ENDL;
     // trying to dereference ("resolve") the future makes us wait for it
-    LLSD value;
+    LLSD_consumed value;
     {
         // instantiate Suspending to manage the "current" coroutine
         llcoro::Suspending suspended;
         value = *future;
     } // destroy Suspending as soon as we're back
     LL_DEBUGS("lleventcoro") << "postAndSuspend(): coroutine " << listenerName
-                             << " resuming with " << value << LL_ENDL;
+                             << " resuming with " << value.first << LL_ENDL;
+    // immediately set consumed according to consuming
+    *value.second = LLCoros::get_consuming();
     // returning should disconnect the connection
-    return value;
+    return value.first;
 }
 
 namespace
 {
 
+typedef std::pair<LLEventWithID, bool*> LLEventWithID_consumed;
+
 /**
  * This helper is specifically for the two-pump version of suspendUntilEventOn().
  * We use a single future object, but we want to listen on two pumps with it.
@@ -227,13 +231,15 @@ public:
         mListener(listener),
         mDiscrim(discriminator)
     {}
+
     // this signature is required for an LLEventPump listener
     bool operator()(const LLSD& event)
     {
-        // our future object is defined to accept LLEventWithID
-        mListener(LLEventWithID(event, mDiscrim));
-        // don't swallow the event, let other listeners see it
-        return false;
+        bool consumed = false;
+        // our future object is defined to accept LLEventWithID_consumed
+        mListener(LLEventWithID_consumed(LLEventWithID(event, mDiscrim), &consumed));
+        // tell LLEventPump whether or not event was consumed
+        return consumed;
     }
 private:
     LISTENER mListener;
@@ -260,7 +266,7 @@ LLEventWithID postAndSuspend2(const LLSD& event,
                            const LLSD& replyPump1NamePath)
 {
     // declare the future
-    boost::dcoroutines::future<LLEventWithID> future(LLCoros::get_self());
+    boost::dcoroutines::future<LLEventWithID_consumed> future(LLCoros::get_self());
     // either callback will assign a value to this future; listen on
     // each specified LLEventPump with a callback
     std::string name(listenerNameForCoro());
@@ -289,17 +295,19 @@ LLEventWithID postAndSuspend2(const LLSD& event,
                              << " about to wait on LLEventPumps " << replyPump0.getPump().getName()
                              << ", " << replyPump1.getPump().getName() << LL_ENDL;
     // trying to dereference ("resolve") the future makes us wait for it
-    LLEventWithID value;
+    LLEventWithID_consumed value;
     {
         // instantiate Suspending to manage "current" coroutine
         llcoro::Suspending suspended;
         value = *future;
     } // destroy Suspending as soon as we're back
     LL_DEBUGS("lleventcoro") << "postAndSuspend(): coroutine " << name
-                             << " resuming with (" << value.first << ", " << value.second << ")"
-                             << LL_ENDL;
+                             << " resuming with (" << value.first.first
+                             << ", " << value.first.second << ")" << LL_ENDL;
+    // tell LLEventPump whether we're consuming
+    *value.second = LLCoros::get_consuming();
     // returning should disconnect both connections
-    return value;
+    return value.first;
 }
 
 LLSD errorException(const LLEventWithID& result, const std::string& desc)

indra/llcommon/lleventcoro.h

@@ -32,6 +32,7 @@
 #include <boost/optional.hpp>
 #include <string>
 #include <stdexcept>
+#include <utility>                  // std::pair
 #include "llevents.h"
 #include "llerror.h"
 
@@ -75,6 +76,8 @@ private:
 namespace llcoro
 {
 
+typedef std::pair<LLSD, bool*> LLSD_consumed;
+
 /// This is an adapter for a signature like void LISTENER(const LLSD&), which
 /// isn't a valid LLEventPump listener: such listeners should return bool.
 template <typename LISTENER>
@@ -84,11 +87,13 @@ public:
     VoidListener(const LISTENER& listener):
         mListener(listener)
     {}
+
     bool operator()(const LLSD& event)
     {
-        mListener(event);
-        // don't swallow the event, let other listeners see it
-        return false;
+        bool consumed = false;
+        mListener(LLSD_consumed(event, &consumed));
+        // tell upstream LLEventPump whether listener consumed
+        return consumed;
     }
 private:
     LISTENER mListener;

indra/llcommon/tests/lleventcoro_test.cpp

@@ -240,7 +240,7 @@ namespace tut
             // ... do whatever preliminary stuff must happen ...
 
             // declare the future
-            boost::dcoroutines::future<LLSD> future(self);
+            boost::dcoroutines::future<llcoro::LLSD_consumed> future(self);
             // tell the future what to suspend for
             LLTempBoundListener connection(
                 LLEventPumps::instance().obtain("source").listen("coro", voidlistener(boost::dcoroutines::make_callback(future))));
@@ -248,7 +248,7 @@ namespace tut
             // attempting to dereference ("resolve") the future causes the calling
             // coroutine to suspend for it
             debug("about to suspend");
-            result = *future;
+            result = (*future).first;
             ensure("Got it", future);
         }
         END