update codes and count()

src/factory/WFChannel.inl

@@ -449,8 +449,8 @@ public:
 	void set_size_limit(size_t size_limit) { this->size_limit = size_limit; }
 
 protected:
-	CommMessageIn *message_in();
-	void handle_in(CommMessageIn *in);
+	virtual CommMessageIn *message_in();
+	virtual void handle_in(CommMessageIn *in);
 	virtual int first_timeout();
 	virtual WFWebSocketTask *new_session();
 

src/manager/WFCondition.cc

@@ -109,14 +109,20 @@ void WFWaitTask::clear_timer_waiter()
 
 SubTask *WFTimedWaitTask::done()
 {
+	WFWaitTask *tmp = NULL;
+
 	this->mutex->lock();
 	if (this->wait_task && this->wait_task->entry.list.next)
 	{
 		list_del(&this->wait_task->entry.list);
-		this->wait_task->count();
+		tmp = this->wait_task;
+		this->wait_task = NULL;
 	}
 	this->mutex->unlock();
 
+	if (tmp)
+		tmp->count();
+
 	SeriesWork *series = series_of(this);
 	
 	if (this->callback)
@@ -162,7 +168,7 @@ WFCounterTask *WFCondition::create_timedwait_task(const struct timespec *abstime
 
 void WFCondition::signal()
 {
-	WFWaitTask *task;
+	WFWaitTask *task = NULL;
 	struct list_head *pos;
 	struct WFWaitTask::task_entry *entry;
 
@@ -175,10 +181,12 @@ void WFCondition::signal()
 		task = entry->ptr;
 		list_del(pos);
 		task->clear_timer_waiter();
-		task->count();
 	}
 
 	this->mutex.unlock();
+
+	if (task)
+		task->count();
 }
 
 void WFCondition::broadcast()
@@ -186,6 +194,7 @@ void WFCondition::broadcast()
 	WFWaitTask *task;
 	struct list_head *pos, *tmp;
 	struct WFWaitTask::task_entry *entry;
+	LIST_HEAD(tmp_list);
 
 	this->mutex.lock();
 	if (!list_empty(&this->waiter_list))
@@ -193,11 +202,17 @@ void WFCondition::broadcast()
 		list_for_each_safe(pos, tmp, &this->waiter_list)
 		{
 			entry = list_entry(pos, struct WFWaitTask::task_entry, list);
-			task = entry->ptr;
-			list_del(pos);
-			task->count();
+			list_move_tail(pos, &tmp_list);
 		}
 	}
 	this->mutex.unlock();
+
+	while (!list_empty(&tmp_list))
+	{
+		entry = list_entry(tmp_list.next, struct WFWaitTask::task_entry, list);
+		task = entry->ptr;
+		list_del(&entry->list);
+		task->count();
+	}
 }
 

src/manager/WFCondition.h

@@ -28,19 +28,18 @@
 class WFCondition
 {
 public:
-	WFCondition()
-	{
-		INIT_LIST_HEAD(&this->waiter_list);
-	}
-
 	WFCounterTask *create_wait_task(std::function<void (WFCounterTask *)> cb);
 	WFCounterTask *create_timedwait_task(const struct timespec *abstime,
 										 std::function<void (WFCounterTask *)> cb);
 	void signal();
 	void broadcast();
 
+public:
+	WFCondition() { INIT_LIST_HEAD(&this->waiter_list); }
+
 public:
 	std::mutex mutex;
+
 private:
 	struct list_head waiter_list;
 };

tutorial/tutorial-14-websocket_cli.cc

@@ -67,10 +67,12 @@ int main(int argc, char *argv[])
 			return;
 		}
 
+		auto *timer_task = WFTaskFactory::create_timer_task(3000000 /* 3s */, nullptr);
 		auto *close_task = client.create_close_task([&wg] (WFWebSocketTask *task) {
 			wg.done();
 		});
 
+		series_of(task)->push_back(timer_task);
 		series_of(task)->push_back(close_task);
 	});