Merge pull request #1389 from Barenboim/master

Add cancelable timer task, update thread/go task's state when timeout.

docs/about-go-task.md

@@ -62,7 +62,7 @@ int main(void)
 class WFTaskFactory
 {
     /* Create 'Go' task with running time limit in seconds plus nanoseconds.
-     * If time exceeded, state WFT_STATE_ABORTED will be got in callback. */
+     * If time exceeded, state WFT_STATE_SYS_ERROR and error ETIMEDOUT will be got in callback. */
     template<class FUNC, class... ARGS>
     static WFGoTask *create_timedgo_task(time_t seconds, long nanoseconds,
                                          const std::string& queue_name,
@@ -70,7 +70,7 @@ class WFTaskFactory
 };
 ~~~
 相比创建普通的go task，create_timedgo_task函数需要多传两个参数，seconds和nanoseconds。  
-如果func的运行时间到达seconds+nanosconds时限，task直接callback，且state为WFT_STATE_ABORTED。  
+如果func的运行时间到达seconds+nanosconds时限，task直接callback，且state为WFT_STATE_SYS_ERROR，error为ETIMEDOUT。  
 注意，框架无法中断用户执行中的任务。func依然会继续执行到结束，但不会再次callback。另外，nanoseconds取值区间在\[0,10亿）。  
 另外，当我们给go task加上了运行时间限制，callback的时机可能会先于func函数的结束，任务所在series可能也会先于func结束。  
 如果我们在func里访问series，可能就是一个错误了。例如：
@@ -99,7 +99,7 @@ int main()
         {
              ...
         }
-        else // state == WFT_STATE_ABORTED.         // 超过运行时间限制
+        else // state == WFT_STATE_SYS_ERROR && error == ETIMEDOUT  // 超过运行时间限制
         {
              ...
         }
@@ -120,7 +120,7 @@ int main()
         {
 		    int result = (int)task->user_data;
         }
-        else // state == WFT_STATE_ABORTED.         // 超过运行时间限制
+        else // state == WFT_STATE_SYS_ERROR && error == ETIMEDOUT    // 超过运行时间限制
         {
 		    ...
         }

docs/en/about-go-task.md

@@ -57,14 +57,14 @@ You may create a go task with running time limit by calling WFTaskFactory::creat
 class WFTaskFactory
 {
     /* Create 'Go' task with running time limit in seconds plus nanoseconds.
-     * If time exceeded, state WFT_STATE_ABORTED will be got in callback. */
+     * If time exceeded, state WFT_STATE_SYS_ERROR and error ETIMEDOUT will be got in callback. */
     template<class FUNC, class... ARGS>
     static WFGoTask *create_timedgo_task(time_t seconds, long nanoseconds,
                                          const std::string& queue_name,
                                          FUNC&& func, ARGS&&... args);
 };
 ~~~
-Compared with creating a normal go task, the ``create_timedgo_task`` function needs to pass two more parameters, seconds and nanoseconds. If the running time of ``func`` reaches the seconds+nanosconds time limit, the task callback directly, and the state is WFT_STATE_ABORTED.  
+Compared with creating a normal go task, the ``create_timedgo_task`` function needs to pass two more parameters, seconds and nanoseconds. If the running time of ``func`` reaches the seconds+nanosconds time limit, the task callback directly, and the state is WFT_STATE_SYS_ERROR and the error is ETIMEDOUT.  
 Note that the framework cannot interrupt the user's ongoing task. ``func`` will still continue to execute to the end, but will not callback again. In addition, the value range of nanoseconds is [0,1 billion).
 
 

docs/en/tutorial-08-matrix_multiply.md

@@ -172,7 +172,7 @@ Obviously, our framework can not interrupt a computing task because it's a user
      ...
  };
  ~~~
-This create_thread_task function needs to pass two more parameters, seconds and nanoseconds. If the running time of func reaches the seconds+nanosconds time limit, the task callback directly, and the state is WFT_STATE_ABORTED. But the task routine will continue to run till the end.
+This create_thread_task function needs to pass two more parameters, seconds and nanoseconds. If the running time of func reaches the seconds+nanosconds time limit, the task callback directly, and the state is WFT_STATE_SYS_ERROR and the error is ETIMEDOUT. But the task routine will continue to run till the end.
 
 # Symmetry of the algorithm and the protocol
 

docs/tutorial-08-matrix_multiply.md

@@ -164,7 +164,7 @@ public:
 };
 ~~~
 参数seconds和nanoseconds构成了运行时限。在这里，nanoseconds的取值范围在\[0,1000000000)。  
-当任务无法在运行时限内结束，会直接回到callback，并且任务的状态为WFT_STATE_ABORTED。  
+当任务无法在运行时限内结束，会直接回到callback，并且任务的状态为WFT_STATE_SYS_ERROR且错误码为ETIMEDOUT。  
 还是用matrix_multiply的例子，我们可以这样写：
 ~~~cpp
 void callback(MMTask *task)     // MMtask = WFThreadTask<MMInput, MMOutput>
@@ -172,12 +172,12 @@ void callback(MMTask *task)     // MMtask = WFThreadTask<MMInput, MMOutput>
     MMInput *input = task->get_input();
     MMOutput *output = task->get_output();
 
-    if (task->get_state() == WFT_STATE_ABORTED)
+    if (task->get_state() == WFT_STATE_SYS_ERROR && task->get_error() == ETIMEDOUT)
     {
         printf("Run out of time.\n");
         return;
     }
-	
+
     assert(task->get_state() == WFT_STATE_SUCCESS)
 
     if (output->error)
@@ -210,13 +210,13 @@ int main()
     ...
 }
 ~~~
-上面的示例，限制了任务运行时间不超过1毫秒，否则，以WFT_STATE_ABORTD的状态返回。  
+上面的示例，限制了任务运行时间不超过1毫秒，否则，以WFT_STATE_SYS_ERROR的状态返回。  
 再次提醒，我们并不会中断用户的实际运行函数。当任务超时并callback，计算函数还会一直运行直到结束。  
 如果用户希望函数不再继续执行，需要在代码中自行加入检查点来实现这样的功能。可以在INPUT里加入flag，例如：
 ~~~cpp
 void callback(MMTask *task)     // MMtask = WFThreadTask<MMInput, MMOutput>
 {
-    if (task->get_state() == WFT_STATE_ABORTED)
+    if (task->get_state() == WFT_STATE_SYS_ERROR && task->get_error() == ETIMEDOUT)
     {
         task->get_input()->flag = true;
         printf("Run out of time.\n");

src/factory/WFTaskFactory.cc

@@ -14,11 +14,12 @@
   limitations under the License.
 
   Authors: Xie Han (xiehan@sogou-inc.com)
-           Wu Jiaxu (wujiaxu@sogou-inc.com)
 */
 
 #include <sys/types.h>
+#include <errno.h>
 #include <time.h>
+#include <utility>
 #include <string>
 #include <mutex>
 #include "list.h"
@@ -41,7 +42,7 @@ protected:
 	long nanoseconds;
 
 public:
-	__WFTimerTask(CommScheduler *scheduler, time_t seconds, long nanoseconds,
+	__WFTimerTask(time_t seconds, long nanoseconds, CommScheduler *scheduler,
 				  timer_callback_t&& cb) :
 		WFTimerTask(scheduler, std::move(cb))
 	{
@@ -53,7 +54,7 @@ public:
 WFTimerTask *WFTaskFactory::create_timer_task(time_t seconds, long nanoseconds,
 											  timer_callback_t callback)
 {
-	return new __WFTimerTask(WFGlobal::get_scheduler(), seconds, nanoseconds,
+	return new __WFTimerTask(seconds, nanoseconds, WFGlobal::get_scheduler(),
 							 std::move(callback));
 }
 
@@ -61,10 +62,9 @@ WFTimerTask *WFTaskFactory::create_timer_task(time_t seconds, long nanoseconds,
 WFTimerTask *WFTaskFactory::create_timer_task(unsigned int microseconds,
 											  timer_callback_t callback)
 {
-	return new __WFTimerTask(WFGlobal::get_scheduler(),
-							 (time_t)(microseconds / 1000000),
-							 (long)(microseconds % 1000000 * 1000),
-							 std::move(callback));
+	return WFTaskFactory::create_timer_task(microseconds / 1000000,
+											microseconds % 1000000 * 1000,
+											std::move(callback));
 }
 
 /****************** Named Tasks ******************/
@@ -88,9 +88,14 @@ struct __NamedObjectList
 		return list_empty(&this->head);
 	}
 
-	void del(T *node)
+	void del(T *node, rb_root *root)
 	{
 		list_del(&node->list);
+		if (this->empty())
+		{
+			rb_erase(&this->rb, root);
+			delete this;
+		}
 	}
 
 	struct rb_node rb;
@@ -129,6 +134,170 @@ static T *__get_object_list(const std::string& name, struct rb_root *root,
 	return NULL;
 }
 
+/****************** Named Timer ******************/
+
+class __WFNamedTimerTask;
+
+struct __timer_node
+{
+	struct list_head list;
+	__WFNamedTimerTask *task;
+};
+
+static class __NamedTimerMap
+{
+public:
+	using TimerList = struct __NamedObjectList<struct __timer_node>;
+
+public:
+	WFTimerTask *create(const std::string& name,
+						time_t seconds, long nanoseconds,
+						CommScheduler *scheduler,
+						timer_callback_t&& cb);
+
+public:
+	void cancel(const std::string& name, size_t max);
+
+private:
+	struct rb_root root_;
+	std::mutex mutex_;
+
+public:
+	__NamedTimerMap()
+	{
+		root_.rb_node = NULL;
+	}
+
+	friend class __WFNamedTimerTask;
+} __timer_map;
+
+class __WFNamedTimerTask : public __WFTimerTask
+{
+public:
+	__WFNamedTimerTask(time_t seconds, long nanoseconds,
+					   __NamedTimerMap::TimerList *timers,
+					   CommScheduler *scheduler,
+					   timer_callback_t&& cb) :
+		__WFTimerTask(seconds, nanoseconds, scheduler, std::move(cb))
+	{
+		dispatched_ = false;
+		node_.task = this;
+		timers->push_back(&node_);
+		timers_ = timers;
+	}
+
+	virtual ~__WFNamedTimerTask()
+	{
+		if (!dispatched_)
+		{
+			__timer_map.mutex_.lock();
+			if (node_.task)
+				timers_->del(&node_, &__timer_map.root_);
+
+			__timer_map.mutex_.unlock();
+		}
+	}
+
+protected:
+	virtual void dispatch();
+	virtual void handle(int state, int error);
+
+private:
+	bool dispatched_;
+	struct __timer_node node_;
+	__NamedTimerMap::TimerList *timers_;
+	friend class __NamedTimerMap;
+};
+
+void __WFNamedTimerTask::dispatch()
+{
+	int ret;
+
+	__timer_map.mutex_.lock();
+	ret = this->scheduler->sleep(this);
+	if (ret < 0)
+	{
+		if (node_.task)
+			timers_->del(&node_, &__timer_map.root_);
+	}
+	else
+	{
+		if (!node_.task)
+			this->cancel();
+	}
+
+	dispatched_ = true;
+	__timer_map.mutex_.unlock();
+	if (ret < 0)
+		this->__WFTimerTask::handle(SS_STATE_ERROR, errno);
+}
+
+void __WFNamedTimerTask::handle(int state, int error)
+{
+	__timer_map.mutex_.lock();
+	if (node_.task)
+		timers_->del(&node_, &__timer_map.root_);
+
+	__timer_map.mutex_.unlock();
+	this->state = state;
+	this->error = error;
+	this->subtask_done();
+}
+
+WFTimerTask *__NamedTimerMap::create(const std::string& name,
+									 time_t seconds, long nanoseconds,
+									 CommScheduler *scheduler,
+									 timer_callback_t&& cb)
+{
+	TimerList *timers;
+	std::lock_guard<std::mutex> lock(mutex_);
+	timers = __get_object_list<TimerList>(name, &root_, true);
+	return new __WFNamedTimerTask(seconds, nanoseconds, timers,
+								  scheduler, std::move(cb));
+}
+
+void __NamedTimerMap::cancel(const std::string& name, size_t max)
+{
+	struct __timer_node *node;
+	TimerList *timers;
+
+	std::lock_guard<std::mutex> lock(mutex_);
+	timers = __get_object_list<TimerList>(name, &root_, false);
+	if (timers)
+	{
+		do
+		{
+			if (max == 0)
+				return;
+
+			node = list_entry(timers->head.next, struct __timer_node, list);
+			list_del(&node->list);
+			if (node->task->dispatched_)
+				node->task->cancel();
+
+			node->task = NULL;
+			max--;
+		} while (!timers->empty());
+
+		rb_erase(&timers->rb, &root_);
+		delete timers;
+	}
+}
+
+WFTimerTask *WFTaskFactory::create_timer_task(const std::string& timer_name,
+											  time_t seconds, long nanoseconds,
+											  timer_callback_t callback)
+{
+	return __timer_map.create(timer_name, seconds, nanoseconds,
+							  WFGlobal::get_scheduler(),
+							  std::move(callback));
+}
+
+void WFTaskFactory::cancel_by_name(const std::string& timer_name, size_t max)
+{
+	__timer_map.cancel(timer_name, max);
+}
+
 /****************** Named Counter ******************/
 
 class __WFNamedCounterTask;
@@ -147,11 +316,17 @@ public:
 
 public:
 	WFCounterTask *create(const std::string& name, unsigned int target_value,
-						  std::function<void (WFCounterTask *)>&& cb);
+						  counter_callback_t&& cb);
 
 	void count_n(const std::string& name, unsigned int n);
 	void count(CounterList *counters, struct __counter_node *node);
-	void remove(CounterList *counters, struct __counter_node *node);
+
+	void remove(CounterList *counters, struct __counter_node *node)
+	{
+		mutex_.lock();
+		counters->del(node, &root_);
+		mutex_.unlock();
+	}
 
 private:
 	void count_n_locked(CounterList *counters, unsigned int n,
@@ -171,7 +346,7 @@ class __WFNamedCounterTask : public WFCounterTask
 public:
 	__WFNamedCounterTask(unsigned int target_value,
 					 __NamedCounterMap::CounterList *counters,
-					 std::function<void (WFCounterTask *)>&& cb) :
+					 counter_callback_t&& cb) :
 		WFCounterTask(1, std::move(cb))
 	{
 		node_.target_value = target_value;
@@ -198,7 +373,7 @@ private:
 
 WFCounterTask *__NamedCounterMap::create(const std::string& name,
 								unsigned int target_value,
-								std::function<void (WFCounterTask *)>&& cb)
+								counter_callback_t&& cb)
 {
 	CounterList *counters;
 
@@ -214,33 +389,28 @@ void __NamedCounterMap::count_n_locked(CounterList *counters, unsigned int n,
 									   struct list_head *task_list)
 {
 	struct __counter_node *node;
-	struct list_head *pos;
-	struct list_head *tmp;
 
-	list_for_each_safe(pos, tmp, &counters->head)
+	do
 	{
 		if (n == 0)
 			return;
 
-		node = list_entry(pos, struct __counter_node, list);
+		node = list_entry(counters->head.next, struct __counter_node, list);
 		if (n >= node->target_value)
 		{
 			n -= node->target_value;
 			node->target_value = 0;
-			list_move_tail(pos, task_list);
-			if (counters->empty())
-			{
-				rb_erase(&counters->rb, &root_);
-				delete counters;
-				return;
-			}
+			list_move_tail(&node->list, task_list);
 		}
 		else
 		{
 			node->target_value -= n;
 			n = 0;
 		}
-	}
+	} while (!counters->empty());
+
+	rb_erase(&counters->rb, &root_);
+	delete counters;
 }
 
 void __NamedCounterMap::count_n(const std::string& name, unsigned int n)
@@ -272,12 +442,7 @@ void __NamedCounterMap::count(CounterList *counters,
 	if (--node->target_value == 0)
 	{
 		task = node->task;
-		counters->del(node);
-		if (counters->empty())
-		{
-			rb_erase(&counters->rb, &root_);
-			delete counters;
-		}
+		counters->del(node, &root_);
 	}
 
 	mutex_.unlock();
@@ -285,20 +450,6 @@ void __NamedCounterMap::count(CounterList *counters,
 		task->WFCounterTask::count();
 }
 
-void __NamedCounterMap::remove(CounterList *counters,
-							   struct __counter_node *node)
-{
-	mutex_.lock();
-	counters->del(node);
-	if (counters->empty())
-	{
-		rb_erase(&counters->rb, &root_);
-		delete counters;
-	}
-
-	mutex_.unlock();
-}
-
 WFCounterTask *WFTaskFactory::create_counter_task(const std::string& counter_name,
 												  unsigned int target_value,
 												  counter_callback_t callback)
@@ -334,7 +485,13 @@ public:
 	void signal(const std::string& name, void *msg, size_t max);
 	void signal(ConditionalList *conds, struct __conditional_node *node,
 				void *msg);
-	void remove(ConditionalList *conds, struct __conditional_node *node);
+
+	void remove(ConditionalList *conds, struct __conditional_node *node)
+	{
+		mutex_.lock();
+		conds->del(node, &root_);
+		mutex_.unlock();
+	}
 
 private:
 	void signal_max_locked(ConditionalList *conds, void *msg, size_t max,
@@ -408,20 +565,18 @@ void __NamedConditionalMap::signal_max_locked(ConditionalList *conds,
 											  void *msg, size_t max,
 											  struct list_head *cond_list)
 {
-	struct list_head *pos, *tmp;
-
 	if (max == (size_t)-1)
 		list_splice(&conds->head, cond_list);
 	else
 	{
-		list_for_each_safe(pos, tmp, &conds->head)
+		do
 		{
 			if (max == 0)
 				return;
 
-			list_move_tail(pos, cond_list);
+			list_move_tail(conds->head.next, cond_list);
 			max--;
-		}
+		} while (!conds->empty());
 	}
 
 	rb_erase(&conds->rb, &root_);
@@ -453,31 +608,11 @@ void __NamedConditionalMap::signal(ConditionalList *conds,
 								   void *msg)
 {
 	mutex_.lock();
-	conds->del(node);
-	if (conds->empty())
-	{
-		rb_erase(&conds->rb, &root_);
-		delete conds;
-	}
-
+	conds->del(node, &root_);
 	mutex_.unlock();
 	node->cond->WFConditional::signal(msg);
 }
 
-void __NamedConditionalMap::remove(ConditionalList *conds,
-								   struct __conditional_node *node)
-{
-	mutex_.lock();
-	conds->del(node);
-	if (conds->empty())
-	{
-		rb_erase(&conds->rb, &root_);
-		delete conds;
-	}
-
-	mutex_.unlock();
-}
-
 WFConditional *WFTaskFactory::create_conditional(const std::string& cond_name,
 												 SubTask *task, void **msgbuf)
 {
@@ -537,8 +672,8 @@ void __WFTimedGoTask::timer_callback(WFTimerTask *timer)
 
 	if (--task->ref == 3)
 	{
-		task->state = WFT_STATE_ABORTED;
-		task->error = 0;
+		task->state = WFT_STATE_SYS_ERROR;
+		task->error = ETIMEDOUT;
 		task->subtask_done();
 	}
 

src/factory/WFTaskFactory.h

@@ -13,8 +13,8 @@
   See the License for the specific language governing permissions and
   limitations under the License.
 
-  Authors: Wu Jiaxu (wujiaxu@sogou-inc.com)
-           Xie Han (xiehan@sogou-inc.com)
+  Authors: Xie Han (xiehan@sogou-inc.com)
+           Wu Jiaxu (wujiaxu@sogou-inc.com)
 */
 
 #ifndef _WFTASKFACTORY_H_
@@ -220,7 +220,21 @@ public:
 	static WFTimerTask *create_timer_task(time_t seconds, long nanoseconds,
 										  timer_callback_t callback);
 
-	/* Deprecated. */
+	/* create a named timer. */
+	static WFTimerTask *create_timer_task(const std::string& timer_name,
+										  time_t seconds, long nanoseconds,
+										  timer_callback_t callback);
+
+	/* cancel all timers under the name. */
+	static void cancel_by_name(const std::string& timer_name)
+	{
+		WFTaskFactory::cancel_by_name(timer_name, (size_t)-1);
+	}
+
+	/* cancel at most 'max' timers under the name. */
+	static void cancel_by_name(const std::string& timer_name, size_t max);
+
+	/* timer in microseconds (deprecated) */
 	static WFTimerTask *create_timer_task(unsigned int microseconds,
 										  timer_callback_t callback);
 
@@ -299,7 +313,8 @@ public:
 									FUNC&& func, ARGS&&... args);
 
 	/* Create 'Go' task with running time limit in seconds plus nanoseconds.
-	 * If time exceeded, state WFT_STATE_ABORTED will be got in callback. */
+	 * If time exceeded, state WFT_STATE_SYS_ERROR and error ETIMEDOUT
+	 * will be got in callback. */
 	template<class FUNC, class... ARGS>
 	static WFGoTask *create_timedgo_task(time_t seconds, long nanoseconds,
 										 const std::string& queue_name,

src/factory/WFTaskFactory.inl

@@ -822,8 +822,8 @@ void __WFTimedThreadTask<INPUT, OUTPUT>::timer_callback(WFTimerTask *timer)
 
 	if (--task->ref == 3)
 	{
-		task->state = WFT_STATE_ABORTED;
-		task->error = 0;
+		task->state = WFT_STATE_SYS_ERROR;
+		task->error = ETIMEDOUT;
 		task->subtask_done();
 	}
 

src/kernel/CommScheduler.h

@@ -170,6 +170,12 @@ public:
 		return this->comm.sleep(session);
 	}
 
+	/* Call 'unsleep' only before 'handle()' returns. */
+	int unsleep(SleepSession *session)
+	{
+		return this->comm.unsleep(session);
+	}
+
 	/* for file aio services. */
 	int io_bind(IOService *service)
 	{

src/kernel/Communicator.cc

@@ -1098,12 +1098,22 @@ void Communicator::handle_sleep_result(struct poller_result *res)
 	SleepSession *session = (SleepSession *)res->data.context;
 	int state;
 
-	if (res->state == PR_ST_STOPPED)
-		state = SS_STATE_DISRUPTED;
-	else
+	switch (res->state)
+	{
+	case PR_ST_FINISHED:
 		state = SS_STATE_COMPLETE;
+		break;
+	case PR_ST_DELETED:
+		res->error = ECANCELED;
+	case PR_ST_ERROR:
+		state = SS_STATE_ERROR;
+		break;
+	case PR_ST_STOPPED:
+		state = SS_STATE_DISRUPTED;
+		break;
+	}
 
-	session->handle(state, 0);
+	session->handle(state, res->error);
 }
 
 void Communicator::handle_aio_result(struct poller_result *res)
@@ -1836,16 +1846,28 @@ int Communicator::shutdown(CommSession *session)
 int Communicator::sleep(SleepSession *session)
 {
 	struct timespec value;
+	void *timer;
+	int index;
 
 	if (session->duration(&value) >= 0)
 	{
-		if (mpoller_add_timer(&value, session, this->mpoller) >= 0)
+		timer = mpoller_add_timer(&value, session, &index, this->mpoller);
+		if (timer)
+		{
+			session->timer = timer;
+			session->index = index;
 			return 0;
+		}
 	}
 
 	return -1;
 }
 
+int Communicator::unsleep(SleepSession *session)
+{
+	return mpoller_del_timer(session->timer, session->index, this->mpoller);
+}
+
 int Communicator::is_handler_thread() const
 {
 	return thrdpool_in_pool(this->thrdpool);

src/kernel/Communicator.h

@@ -246,6 +246,10 @@ private:
 	virtual int duration(struct timespec *value) = 0;
 	virtual void handle(int state, int error) = 0;
 
+private:
+	void *timer;
+	int index;
+
 public:
 	virtual ~SleepSession() { }
 	friend class Communicator;
@@ -274,6 +278,7 @@ public:
 	void unbind(CommService *service);
 
 	int sleep(SleepSession *session);
+	int unsleep(SleepSession *session);
 
 	int io_bind(IOService *service);
 	void io_unbind(IOService *service);

src/kernel/SleepRequest.h

@@ -39,6 +39,12 @@ public:
 			this->handle(SS_STATE_ERROR, errno);
 	}
 
+protected:
+	int cancel()
+	{
+		return this->scheduler->unsleep(this);
+	}
+
 protected:
 	int state;
 	int error;

src/kernel/mpoller.h

@@ -48,35 +48,41 @@ struct __mpoller
 static inline int mpoller_add(const struct poller_data *data, int timeout,
 							  mpoller_t *mpoller)
 {
-	unsigned int index = (unsigned int)data->fd % mpoller->nthreads;
+	int index = (unsigned int)data->fd % mpoller->nthreads;
 	return poller_add(data, timeout, mpoller->poller[index]);
 }
 
 static inline int mpoller_del(int fd, mpoller_t *mpoller)
 {
-	unsigned int index = (unsigned int)fd % mpoller->nthreads;
+	int index = (unsigned int)fd % mpoller->nthreads;
 	return poller_del(fd, mpoller->poller[index]);
 }
 
 static inline int mpoller_mod(const struct poller_data *data, int timeout,
 							  mpoller_t *mpoller)
 {
-	unsigned int index = (unsigned int)data->fd % mpoller->nthreads;
+	int index = (unsigned int)data->fd % mpoller->nthreads;
 	return poller_mod(data, timeout, mpoller->poller[index]);
 }
 
 static inline int mpoller_set_timeout(int fd, int timeout, mpoller_t *mpoller)
 {
-	unsigned int index = (unsigned int)fd % mpoller->nthreads;
+	int index = (unsigned int)fd % mpoller->nthreads;
 	return poller_set_timeout(fd, timeout, mpoller->poller[index]);
 }
 
-static inline int mpoller_add_timer(const struct timespec *value, void *context,
-									mpoller_t *mpoller)
+static inline void *mpoller_add_timer(const struct timespec *value,
+									  void *context, int *index,
+									  mpoller_t *mpoller)
 {
 	static unsigned int n = 0;
-	unsigned int index = n++ % mpoller->nthreads;
-	return poller_add_timer(value, context, mpoller->poller[index]);
+	*index = n++ % mpoller->nthreads;
+	return poller_add_timer(value, context, mpoller->poller[*index]);
+}
+
+static inline int mpoller_del_timer(void *timer, int index, mpoller_t *mpoller)
+{
+	return poller_del_timer(timer, mpoller->poller[index]);
 }
 
 #endif

src/kernel/poller.c

@@ -930,6 +930,8 @@ static void __poller_handle_timeout(const struct __poller_node *time_node,
 				poller->nodes[node->data.fd] = NULL;
 				__poller_del_fd(node->data.fd, node->event, poller);
 			}
+			else
+				node->removed = 1;
 
 			list_move_tail(pos, &timeo_list);
 		}
@@ -949,6 +951,8 @@ static void __poller_handle_timeout(const struct __poller_node *time_node,
 					poller->nodes[node->data.fd] = NULL;
 					__poller_del_fd(node->data.fd, node->event, poller);
 				}
+				else
+					node->removed = 1;
 
 				poller->tree_first = rb_next(poller->tree_first);
 				rb_erase(&node->rb, &poller->timeo_tree);
@@ -968,9 +972,17 @@ static void __poller_handle_timeout(const struct __poller_node *time_node,
 	{
 		node = list_entry(timeo_list.next, struct __poller_node, list);
 		list_del(&node->list);
+		if (node->data.fd >= 0)
+		{
+			node->error = ETIMEDOUT;
+			node->state = PR_ST_ERROR;
+		}
+		else
+		{
+			node->error = 0;
+			node->state = PR_ST_FINISHED;
+		}
 
-		node->error = ETIMEDOUT;
-		node->state = PR_ST_ERROR;
 		free(node->res);
 		poller->cb((struct poller_result *)node, poller->ctx);
 	}
@@ -1526,8 +1538,8 @@ int poller_set_timeout(int fd, int timeout, poller_t *poller)
 	return -!node;
 }
 
-int poller_add_timer(const struct timespec *value, void *context,
-					 poller_t *poller)
+void *poller_add_timer(const struct timespec *value, void *context,
+					   poller_t *poller)
 {
 	struct __poller_node *node;
 
@@ -1554,10 +1566,39 @@ int poller_add_timer(const struct timespec *value, void *context,
 		pthread_mutex_lock(&poller->mutex);
 		__poller_insert_node(node, poller);
 		pthread_mutex_unlock(&poller->mutex);
-		return 0;
 	}
 
-	return -1;
+	return node;
+}
+
+int poller_del_timer(void *timer, poller_t *poller)
+{
+	struct __poller_node *node = (struct __poller_node *)timer;
+	int removed;
+
+	pthread_mutex_lock(&poller->mutex);
+	removed = node->removed;
+	if (!removed)
+	{
+		node->removed = 1;
+
+		if (node->in_rbtree)
+			__poller_tree_erase(node, poller);
+		else
+			list_del(&node->list);
+
+		node->error = 0;
+		node->state = PR_ST_DELETED;
+		if (poller->stopped)
+			poller->cb((struct poller_result *)node, poller->ctx);
+		else
+			write(poller->pipe_wr, &node, sizeof (void *));
+	}
+	else
+		errno = ENOENT;
+
+	pthread_mutex_unlock(&poller->mutex);
+	return -removed;
 }
 
 void poller_stop(poller_t *poller)

src/kernel/poller.h

@@ -103,8 +103,9 @@ int poller_add(const struct poller_data *data, int timeout, poller_t *poller);
 int poller_del(int fd, poller_t *poller);
 int poller_mod(const struct poller_data *data, int timeout, poller_t *poller);
 int poller_set_timeout(int fd, int timeout, poller_t *poller);
-int poller_add_timer(const struct timespec *value, void *context,
-					 poller_t *poller);
+void *poller_add_timer(const struct timespec *value, void *context,
+					   poller_t *poller);
+int poller_del_timer(void *timer, poller_t *poller);
 void poller_stop(poller_t *poller);
 void poller_destroy(poller_t *poller);