update msg

This commit is contained in:
holmes1412
2021-06-24 00:33:27 +08:00
parent 1d2a61fb55
commit 604eb72382
4 changed files with 54 additions and 24 deletions
+28 -12
View File
@@ -32,9 +32,17 @@ class __WFCondition : public WFCondition
public:
__WFCondition(const std::string& str) :
name(str)
{ }
{
this->node.ptr = this;
}
public:
struct entry
{
struct rb_node rb;
__WFCondition *ptr;
} node;
struct rb_node rb;
std::string name;
};
@@ -132,13 +140,19 @@ __ConditionMap::~__ConditionMap()
WFWaitTask *task;
struct list_head *pos;
struct list_head *tmp;
struct WFSemaphoreTask::entry *node;
struct __WFCondition::entry *cond_node;
while (this->condition_map.rb_node)
{
cond = rb_entry(this->condition_map.rb_node, __WFCondition, rb);
cond_node = rb_entry(this->condition_map.rb_node,
struct __WFCondition::entry, rb);
cond = cond_node->ptr;
list_for_each_safe(pos, tmp, &cond->waiter_list)
{
task = list_entry(pos, WFWaitTask, list);
node = list_entry(pos, struct WFSemaphoreTask::entry, list);
task = (WFWaitTask *)node->ptr;
list_del(pos);
delete task;
}
@@ -150,15 +164,17 @@ __ConditionMap::~__ConditionMap()
__WFCondition *__ConditionMap::find_condition(const std::string& name)
{
__WFCondition *cond;
struct __WFCondition::entry *cond_node;
struct rb_node **p = &this->condition_map.rb_node;
struct rb_node *parent = NULL;
__WFCondition *cond;
this->mutex.lock();
while (*p)
{
parent = *p;
cond = rb_entry(*p, __WFCondition, rb);
cond_node = rb_entry(*p, struct __WFCondition::entry, rb);
cond = cond_node->ptr;
if (name < cond->name)
p = &(*p)->rb_left;
@@ -171,8 +187,8 @@ __WFCondition *__ConditionMap::find_condition(const std::string& name)
if (*p == NULL)
{
cond = new __WFCondition(name);
rb_link_node(&cond->rb, parent, p);
rb_insert_color(&cond->rb, &this->condition_map);
rb_link_node(&cond->node.rb, parent, p);
rb_insert_color(&cond->node.rb, &this->condition_map);
}
this->mutex.unlock();
@@ -236,7 +252,7 @@ WFMailboxTask *WFSemTaskFactory::create_wait_task(WFCondition *cond,
WFWaitTask *task = new WFWaitTask(std::move(callback));
cond->mutex.lock();
list_add_tail(&task->list, &cond->waiter_list);
list_add_tail(&task->node.list, &cond->waiter_list);
cond->mutex.unlock();
return task;
@@ -253,7 +269,7 @@ WFMailboxTask *WFSemTaskFactory::create_timedwait_task(WFCondition *cond,
waiter->set_timer(task);
cond->mutex.lock();
list_add_tail(&waiter->list, &cond->waiter_list);
list_add_tail(&waiter->node.list, &cond->waiter_list);
cond->mutex.unlock();
return waiter;
@@ -265,7 +281,7 @@ WFMailboxTask *WFSemTaskFactory::create_switch_wait_task(WFCondition *cond,
WFWaitTask *task = new WFSwitchWaitTask(std::move(callback));
cond->mutex.lock();
list_add_tail(&task->list, &cond->waiter_list);
list_add_tail(&task->node.list, &cond->waiter_list);
cond->mutex.unlock();
return task;
@@ -282,7 +298,7 @@ WFMailboxTask *WFSemTaskFactory::create_switch_timedwait_task(WFCondition *cond,
waiter->set_timer(task);
cond->mutex.lock();
list_add_tail(&waiter->list, &cond->waiter_list);
list_add_tail(&waiter->node.list, &cond->waiter_list);
cond->mutex.unlock();
return waiter;
+12 -6
View File
@@ -38,7 +38,7 @@ WFMailboxTask *WFSemaphore::acquire(std::function<void (WFMailboxTask *)> cb)
else
{
this->mutex.lock();
list_add_tail(&task->list, &this->waiter_list);
list_add_tail(&task->node.list, &this->waiter_list);
this->mutex.unlock();
}
@@ -49,13 +49,15 @@ void WFSemaphore::release(void *msg)
{
WFSemaphoreTask *task;
struct list_head *pos;
struct WFSemaphoreTask::entry *node;
this->mutex.lock();
if (++this->concurrency <= 0)// && !list_empty(&this->waiter_list))
{
pos = this->waiter_list.next;
task = list_entry(pos, WFSemaphoreTask, list);
node = list_entry(pos, struct WFSemaphoreTask::entry, list);
task = node->ptr;
list_del(pos);
task->send(msg);
}
@@ -82,9 +84,9 @@ void WFWaitTask::clear_timer_waiter()
SubTask *WFTimedWaitTask::done()
{
this->mutex->lock();
if (this->wait_task && this->wait_task->list.next)
if (this->wait_task && this->wait_task->node.list.next)
{
list_del(&this->wait_task->list);
list_del(&this->wait_task->node.list);
this->wait_task->set_error(ETIMEDOUT);
this->wait_task->count();
this->wait_task = NULL;
@@ -101,13 +103,15 @@ void WFCondition::signal(void *msg)
{
WFWaitTask *task;
struct list_head *pos;
struct WFSemaphoreTask::entry *node;
this->mutex.lock();
if (!list_empty(&this->waiter_list))
{
pos = this->waiter_list.next;
task = list_entry(pos, WFWaitTask, list);
node = list_entry(pos, struct WFSemaphoreTask::entry, list);
task = (WFWaitTask *)node->ptr;
list_del(pos);
task->clear_timer_waiter();
task->send(msg);
@@ -120,13 +124,15 @@ void WFCondition::broadcast(void *msg)
{
WFWaitTask *task;
struct list_head *pos, *tmp;
struct WFSemaphoreTask::entry *node;
this->mutex.lock();
if (!list_empty(&this->waiter_list))
{
list_for_each_safe(pos, tmp, &this->waiter_list)
{
task = list_entry(pos, WFWaitTask, list);
node = list_entry(pos, struct WFSemaphoreTask::entry, list);
task = (WFWaitTask *)node->ptr;
list_del(pos);
task->send(msg);
}
+9 -5
View File
@@ -68,19 +68,23 @@ class WFSemaphoreTask : public WFMailboxTask
{
public:
WFSemaphoreTask(std::function<void (WFMailboxTask *)>&& cb) :
WFMailboxTask(std::move(cb))
// WFMailboxTask(&this->msg, 1, std::move(cb))
WFMailboxTask(&this->msg, 1, std::move(cb))
{
this->list.next = NULL;
this->node.list.next = NULL;
this->node.ptr = this;
}
virtual ~WFSemaphoreTask() { }
public:
struct list_head list;
struct entry
{
struct list_head list;
WFSemaphoreTask *ptr;
} node;
private:
void *msg; // TODO: user_data instead
void *msg;
};
class WFTimedWaitTask;
+5 -1
View File
@@ -101,7 +101,11 @@ TEST(condition_unittest, timedwait)
auto *task2 = WFSemTaskFactory::create_timedwait_task("timedwait2", &ts,
[&wait_group](WFMailboxTask *task) {
EXPECT_EQ(task->get_error(), 0);
EXPECT_TRUE(strcmp((char *)task->user_data, "wake up!!") == 0);
void **msg;
size_t n;
msg = task->get_mailbox(&n);
EXPECT_EQ(n, 1);
EXPECT_TRUE(strcmp((char *)*msg, "wake up!!") == 0);
wait_group.done();
});