Merge pull request #667 from mikeyang2015/master

support custom executor in execution_queue

src/bthread/execution_queue.cpp

@@ -101,15 +101,22 @@ void ExecutionQueueBase::start_execute(TaskNode* node) {
         }
     }
 
-    bthread_t tid;
-    // We start the execution thread in background instead of foreground as
-    // we can't determine whether the code after execute() is urgent (like
-    // unlock a pthread_mutex_t) in which case implicit context switch may
-    // cause undefined behavior (e.g. deadlock)
-    if (bthread_start_background(&tid, &_options.bthread_attr, 
-                _execute_tasks, node) != 0) {
-        PLOG(FATAL) << "Fail to start bthread";
-        _execute_tasks(node);
+    if (nullptr == _options.executor) {
+        bthread_t tid;
+        // We start the execution thread in background instead of foreground as
+        // we can't determine whether the code after execute() is urgent (like
+        // unlock a pthread_mutex_t) in which case implicit context switch may
+        // cause undefined behavior (e.g. deadlock)
+        if (bthread_start_background(&tid, &_options.bthread_attr,
+                                     _execute_tasks, node) != 0) {
+            PLOG(FATAL) << "Fail to start bthread";
+            _execute_tasks(node);
+        }
+    } else {
+        if (_options.executor->submit(_execute_tasks, node) != 0) {
+            PLOG(FATAL) << "Fail to submit task";
+            _execute_tasks(node);
+        }
     }
 }
 

src/bthread/execution_queue.h

@@ -128,11 +128,24 @@ const static TaskOptions TASK_OPTIONS_NORMAL = TaskOptions(false, false);
 const static TaskOptions TASK_OPTIONS_URGENT = TaskOptions(true, false);
 const static TaskOptions TASK_OPTIONS_INPLACE = TaskOptions(false, true);
 
+class Executor {
+public:
+    virtual ~Executor() {}
+
+    // Return 0 on success.
+    virtual int submit(void * (*fn)(void*), void* args) = 0;
+};
+
 struct ExecutionQueueOptions {
     ExecutionQueueOptions();
     // Attribute of the bthread which execute runs on
     // default: BTHREAD_ATTR_NORMAL
     bthread_attr_t bthread_attr;
+
+    // Executor that tasks run on. bthread will be used when executor = NULL.
+    // Note that TaskOptions.in_place_if_possible = false will not work, if implementation of
+    // Executor is in-place(synchronous).
+    Executor * executor;
 };
 
 // Start a ExecutionQueue. If |options| is NULL, the queue will be created with

src/bthread/execution_queue_inl.h

@@ -317,7 +317,7 @@ public:
 };
 
 inline ExecutionQueueOptions::ExecutionQueueOptions()
-    : bthread_attr(BTHREAD_ATTR_NORMAL)
+    : bthread_attr(BTHREAD_ATTR_NORMAL), executor(NULL)
 {}
 
 template <typename T>