Change to priority queue of pointers to Callback objects

NEWS.md

@@ -8,6 +8,8 @@
 
 * Fixed [issue #36](https://github.com/r-lib/later/issues/36): Failure to build on OS X <=10.12 (thanks @mingwandroid). [PR #21](https://github.com/r-lib/later/pull/21)
 
+* Fixed [issue #39](https://github.com/r-lib/later/issues/39): Calling the C++ function `later::later()` from a different thread could cause an R GC event to occur on that thread, leading to memory corruption. [PR #40](https://github.com/r-lib/later/pull/40)
+
 ## later 0.6
 
 * Fix a hang on address sanitized (ASAN) builds of R. [Issue #16](https://github.com/r-lib/later/issues/16), [PR #17](https://github.com/r-lib/later/pull/17)

src/callback_registry.cpp

@@ -1,20 +1,22 @@
 #include <boost/bind.hpp>
+#include <boost/shared_ptr.hpp>
+#include <boost/make_shared.hpp>
 #include "callback_registry.h"
 
 CallbackRegistry::CallbackRegistry() : mutex(mtx_recursive), condvar(mutex) {
 }
 
 void CallbackRegistry::add(Rcpp::Function func, double secs) {
   Timestamp when(secs);
-  Callback cb(when, func);
+  Callback_sp cb = boost::make_shared<Callback>(when, func);
   Guard guard(mutex);
   queue.push(cb);
   condvar.signal();
 }
 
 void CallbackRegistry::add(void (*func)(void*), void* data, double secs) {
   Timestamp when(secs);
-  Callback cb(when, boost::bind(func, data));
+  Callback_sp cb = boost::make_shared<Callback>(when, boost::bind(func, data));
   Guard guard(mutex);
   queue.push(cb);
   condvar.signal();
@@ -27,7 +29,7 @@ Optional<Timestamp> CallbackRegistry::nextTimestamp() const {
   if (this->queue.empty()) {
     return Optional<Timestamp>();
   } else {
-    return Optional<Timestamp>(this->queue.top().when);
+    return Optional<Timestamp>(this->queue.top()->when);
   }
 }
 
@@ -39,12 +41,12 @@ bool CallbackRegistry::empty() const {
 // Returns true if the smallest timestamp exists and is not in the future.
 bool CallbackRegistry::due(const Timestamp& time) const {
   Guard guard(mutex);
-  return !this->queue.empty() && !(this->queue.top().when > time);
+  return !this->queue.empty() && !(this->queue.top()->when > time);
 }
 
-std::vector<Callback> CallbackRegistry::take(size_t max, const Timestamp& time) {
+std::vector<Callback_sp> CallbackRegistry::take(size_t max, const Timestamp& time) {
   Guard guard(mutex);
-  std::vector<Callback> results;
+  std::vector<Callback_sp> results;
   while (this->due(time) && (max <= 0 || results.size() < max)) {
     results.push_back(this->queue.top());
     this->queue.pop();

src/callback_registry.h

@@ -4,6 +4,7 @@
 #include <Rcpp.h>
 #include <queue>
 #include <boost/function.hpp>
+#include <boost/shared_ptr.hpp>
 #include "timestamp.h"
 #include "optional.h"
 #include "threadutils.h"
@@ -33,10 +34,24 @@ class Callback {
   Task func;
 };
 
+typedef boost::shared_ptr<Callback> Callback_sp;
+
+template <typename T>
+struct pointer_greater_than {
+  const bool operator()(const T a, const T b) const {
+    return *a > *b;
+  }
+};
+
+
 // Stores R function callbacks, ordered by timestamp.
 class CallbackRegistry {
 private:
-  std::priority_queue<Callback,std::vector<Callback>,std::greater<Callback> > queue;
+  // This is a priority queue of shared pointers to Callback objects. The
+  // reason it is not a priority_queue<Callback> is because that can cause
+  // objects to be copied on the wrong thread, and even trigger an R GC event
+  // on the wrong thread. https://github.com/r-lib/later/issues/39
+  std::priority_queue<Callback_sp, std::vector<Callback_sp>, pointer_greater_than<Callback_sp> > queue;
   mutable Mutex mutex;
   mutable ConditionVariable condvar;
 
@@ -62,7 +77,7 @@ class CallbackRegistry {
   bool due(const Timestamp& time = Timestamp()) const;
 
   // Pop and return an ordered list of functions to execute now.
-  std::vector<Callback> take(size_t max = -1, const Timestamp& time = Timestamp());
+  std::vector<Callback_sp> take(size_t max = -1, const Timestamp& time = Timestamp());
 
   bool wait(double timeoutSecs) const;
 };

src/later.cpp

@@ -74,12 +74,12 @@ bool execCallbacks(double timeoutSecs) {
   while (true) {
     // We only take one at a time, because we don't want to lose callbacks if 
     // one of the callbacks throws an error
-    std::vector<Callback> callbacks = callbackRegistry.take(1, now);
+    std::vector<Callback_sp> callbacks = callbackRegistry.take(1, now);
     if (callbacks.size() == 0) {
       break;
     }
     // This line may throw errors!
-    callbacks[0]();
+    (*callbacks[0])();
   }
   return true;
 }