From cb3bf78ebaea74a9eac006bb282dcf19f17c9736 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Wed, 16 Jun 2021 20:27:49 +0800 Subject: [PATCH] update codes and count() --- src/factory/WFChannel.inl | 4 ++-- src/manager/WFCondition.cc | 27 +++++++++++++++++++++------ src/manager/WFCondition.h | 9 ++++----- tutorial/tutorial-14-websocket_cli.cc | 2 ++ 4 files changed, 29 insertions(+), 13 deletions(-) diff --git a/src/factory/WFChannel.inl b/src/factory/WFChannel.inl index fc4e2aa2..54da3960 100644 --- a/src/factory/WFChannel.inl +++ b/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(); diff --git a/src/manager/WFCondition.cc b/src/manager/WFCondition.cc index 60eb2bf3..52ea7efa 100644 --- a/src/manager/WFCondition.cc +++ b/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(); + } } diff --git a/src/manager/WFCondition.h b/src/manager/WFCondition.h index 2e2e1f3c..9f996dc0 100644 --- a/src/manager/WFCondition.h +++ b/src/manager/WFCondition.h @@ -28,19 +28,18 @@ class WFCondition { public: - WFCondition() - { - INIT_LIST_HEAD(&this->waiter_list); - } - WFCounterTask *create_wait_task(std::function cb); WFCounterTask *create_timedwait_task(const struct timespec *abstime, std::function cb); void signal(); void broadcast(); +public: + WFCondition() { INIT_LIST_HEAD(&this->waiter_list); } + public: std::mutex mutex; + private: struct list_head waiter_list; }; diff --git a/tutorial/tutorial-14-websocket_cli.cc b/tutorial/tutorial-14-websocket_cli.cc index 480dcfad..ba3bee72 100644 --- a/tutorial/tutorial-14-websocket_cli.cc +++ b/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); });