GithubStar/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-10-21 06:36:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

remove WFSwitchWaitTask

Browse Source
This commit is contained in:
holmes1412
2021-07-26 18:44:31 +08:00
parent 8d2ab28211
commit f1cf95c379
8 changed files with 50 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/factory/WFChannel.inl
+4 -4
View File
@@ -249,8 +249,8 @@ void WFComplexChannel<MSG>::handle_terminated()
this->sending = true;
shutdown = true;
} else {
waiter = WFCondTaskFactory::create_swait_task(&this->condition,
nullptr);
waiter = WFCondTaskFactory::create_wait_task(&this->condition,
nullptr);
series_of(this)->push_front(this);
series_of(this)->push_front(waiter);
}
@@ -353,7 +353,7 @@ void ComplexChannelOutTask<MSG>::dispatch()
}
else
{
waiter = WFCondTaskFactory::create_swait_task(&channel->condition,
waiter = WFCondTaskFactory::create_wait_task(&channel->condition,
[this](WFMailboxTask *task)
{
auto *channel = (WFComplexChannel<MSG> *)this->get_request_channel();
@@ -374,7 +374,7 @@ void ComplexChannelOutTask<MSG>::dispatch()
}
else
{
waiter = WFCondTaskFactory::create_swait_task(&channel->condition,
waiter = WFCondTaskFactory::create_wait_task(&channel->condition,
[this](WFMailboxTask *task)
{
auto *channel = (WFComplexChannel<MSG> *)this->get_request_channel();
src/factory/WFCondTask.h
+3
View File
@@ -34,6 +34,9 @@ class WFCondWaitTask : public WFMailboxTask
public:
virtual void clear_locked() { }
protected:
virtual SubTask *done();
private:
struct list_head list;
src/factory/WFCondTaskFactory.cc
+12 -52
View File
@@ -50,8 +50,6 @@ public:
WFWaitTask *create(const std::string& name,
const struct timespec *timeout,
wait_callback_t&& cb);
WFWaitTask *create_switch(const std::string& name,
wait_callback_t&& cb);
public:
static __ConditionMap *get_instance()
@@ -105,14 +103,6 @@ WFWaitTask *__ConditionMap::create(const std::string& name,
std::move(cb));
}
WFWaitTask *__ConditionMap::create_switch(const std::string& name,
wait_callback_t&& cb)
{
__WFCondition *cond = this->find_condition(name);
return WFCondTaskFactory::create_swait_task(cond, std::move(cb));
}
__ConditionMap::~__ConditionMap()
{
__WFCondition *cond;
@@ -235,16 +225,20 @@ public:
virtual ~__WFWaitTimerTask();
};
class WFSwitchWaitTask : public WFCondWaitTask
SubTask *WFCondWaitTask::done()
{
public:
WFSwitchWaitTask(wait_callback_t&& cb) :
WFCondWaitTask(std::move(cb))
{ }
SeriesWork *series = series_of(this);
protected:
SubTask *done();
};
WFTimerTask *switch_task = WFTaskFactory::create_timer_task(0,
[this](WFTimerTask *task) {
if (this->callback)
this->callback(this);
delete this;
});
series->push_front(switch_task);
return series->pop();
}
void WFTimedWaitTask::clear_locked()
{
@@ -276,21 +270,6 @@ WFTimedWaitTask::~WFTimedWaitTask()
delete this->timer;
}
SubTask *WFSwitchWaitTask::done()
{
SeriesWork *series = series_of(this);
WFTimerTask *switch_task = WFTaskFactory::create_timer_task(0,
[this](WFTimerTask *task) {
if (this->callback)
this->callback(this);
delete this;
});
series->push_front(switch_task);
return series->pop();
}
SubTask *__WFWaitTimerTask::done()
{
WFTimedWaitTask *waiter = NULL;
@@ -339,13 +318,6 @@ WFWaitTask *WFCondTaskFactory::create_wait_task(const std::string& name,
return __ConditionMap::get_instance()->create(name, std::move(callback));
}
WFWaitTask *WFCondTaskFactory::create_swait_task(const std::string& name,
wait_callback_t callback)
{
return __ConditionMap::get_instance()->create_switch(name,
std::move(callback));
}
WFWaitTask *WFCondTaskFactory::create_timedwait_task(const std::string& name,
const struct timespec *timeout,
wait_callback_t callback)
@@ -383,15 +355,3 @@ WFWaitTask *WFCondTaskFactory::create_timedwait_task(WFCondition *cond,
return waiter;
}
WFWaitTask *WFCondTaskFactory::create_swait_task(WFCondition *cond,
wait_callback_t callback)
{
WFSwitchWaitTask *task = new WFSwitchWaitTask(std::move(callback));
cond->mutex->lock();
list_add_tail(&task->list, &cond->wait_list);
cond->mutex->unlock();
return task;
}
src/factory/WFCondTaskFactory.h
+4 -6
View File
@@ -32,6 +32,10 @@
class WFCondTaskFactory
{
public:
// simple timedwait task
// static WFWaitTask *create_timedwait_task(const struct timespec *timeout,
// wait_callback_t callback);
// use condition by name
static void signal_by_name(const std::string& name, void *msg);
@@ -44,9 +48,6 @@ public:
const struct timespec *timeout,
wait_callback_t callback);
static WFWaitTask *create_swait_task(const std::string& name,
wait_callback_t callback);
// use condition by ptr
static WFWaitTask *create_wait_task(WFCondition *cond,
wait_callback_t callback);
@@ -55,9 +56,6 @@ public:
const struct timespec *timeout,
wait_callback_t callback);
static WFWaitTask *create_swait_task(WFCondition *cond,
wait_callback_t callback);
};
#endif
src/factory/WFResourcePool.cc
+16 -5
View File
@@ -17,6 +17,7 @@
Xie Han (xiehan@sogou-inc.com)
*/
#include <string.h>
#include "list.h"
#include "WFTask.h"
#include "WFResourcePool.h"
@@ -32,9 +33,9 @@ public:
virtual void signal(void *res) { }
public:
__WFConditional(SubTask *task, void **pres,
__WFConditional(SubTask *task, void **resbuf,
struct WFResourcePool::Data *data) :
WFConditional(task, pres)
WFConditional(task, resbuf)
{
this->data = data;
}
@@ -54,9 +55,19 @@ void __WFConditional::dispatch()
this->WFConditional::dispatch();
}
WFConditional *WFResourcePool::get(SubTask *task, void **pres)
WFConditional *WFResourcePool::get(SubTask *task, void **resbuf)
{
return new __WFConditional(task, pres, &this->data);
return new __WFConditional(task, resbuf, &this->data);
}
WFResourcePool::WFResourcePool(void *const *res, size_t n)
{
this->data.res = new void *[n];
memcpy(this->data.res, res, n * sizeof (void *));
this->data.value = n;
this->data.index = 0;
INIT_LIST_HEAD(&this->data.wait_list);
this->data.pool = this;
}
void WFResourcePool::post(void *res)
@@ -73,7 +84,7 @@ void WFResourcePool::post(void *res)
else
{
cond = NULL;
data->push(res);
this->push(res);
}
data->mutex.unlock();
src/factory/WFResourcePool.h
+9 -26
View File
@@ -21,35 +21,27 @@
#define _WFRESOURCEPOOL_H_
#include <mutex>
#include <atomic>
#include "list.h"
#include "WFTask.h"
class WFResourcePool
{
public:
WFConditional *get(SubTask *task, void **pres);
WFConditional *get(SubTask *task, void **resbuf);
void post(void *res);
public:
struct Data
{
void *pop() { return this->pool->pop(); }
void push(void *res) { this->pool->push(res); }
void **res;
std::atomic<int> value;
std::atomic<int> index;
long value;
size_t index;
struct list_head wait_list;
std::mutex mutex;
WFResourcePool *ptr;
virtual void *pop()
{
return this->ptr->pop();
}
virtual void push(void *res)
{
ptr->push(res);
}
WFResourcePool *pool;
};
private:
@@ -67,17 +59,8 @@ private:
struct Data data;
public:
WFResourcePool(void **res, int n)
{
this->data.ptr = this;
this->data.res = new void *[n];
memcpy(this->data.res, res, n * sizeof(void *));
this->data.value = n;
this->data.index = 0;
INIT_LIST_HEAD(&this->data.wait_list);
}
virtual ~WFResourcePool() { delete[] this->data.res; }
WFResourcePool(void *const *res, size_t n);
virtual ~WFResourcePool() { delete []this->data.res; }
};
#endif
src/manager/WFCondition.h
+1
View File
@@ -84,6 +84,7 @@ public:
INIT_LIST_HEAD(&this->get_list);
INIT_LIST_HEAD(&this->wait_list);
}
virtual ~WFCondition();
};
test/condition_unittest.cc
+1 -84
View File
@@ -125,7 +125,7 @@ TEST(condition_unittest, resource_pool)
int res_concurrency = 3;
int task_concurrency = 10;
const char *words[3] = {"workflow", "srpc", "pyworkflow"};
WFResourcePool res_pool((void **)words, res_concurrency);
WFResourcePool res_pool((void *const *)words, res_concurrency);
WFFacilities::WaitGroup wg(task_concurrency);
for (int i = 0; i < task_concurrency; i++)
@@ -148,86 +148,3 @@ TEST(condition_unittest, resource_pool)
wg.wait();
}
class TestResource : public WFCondition::BaseResource
{
public:
TestResource(std::string msg) : msg(msg) { }
void *get() { return &this->msg; }
void set(std::string&& msg) { this->msg = std::move(msg);}
private:
std::string msg;
};
TestResource res("");
void wait_callback(WFWaitTask *task)
{
void **msg;
size_t n;
msg = task->get_mailbox(&n);
std::string *str = *msg;
fprintf(stderr, "waiter get msg:%s\n", str->c_str());
series->set_context(*msg);
}
void work(std::string *msg, WFFacilities::WaitGroup &wg)
{
fprintf(stderr, "working on message:%s\n", msg->c_str());
wg.done();
}
SubTask *create_router_task(WFCondition *cond, std::function<WFTimerTask *task> cb)
{
auto *router = WFTaskFactory::create_timer_task(100000, std::move(cb));
router->user_data = cond;
}
void router_callback(WFTimerTask *task)
{
fprintf(stderr, "finish routing and should broadcast every one else\n");
((WFCondition *)(task->user_data))->broadcast();
}
TEST(condition_unittest, res_condition)
{
int task_concurrency = 10;
WFFacilities::WaitGroup wg(task_concurrency);
WFCondition cond((WFCondition::BaseResource *)&res);
for (int i = 0; i < task_concurrency; i++)
{
auto *timer = WFTaskFactory::create_timer_task(i * 100000,
[&cond](WFTimerTask *task) {
std::string *msg;
int ret = cond.get(&msg);
if (ret == -1)
{
auto *waiter = cond.create_wait_task(wait_callback);
series_of(task)->push_front(waiter);
fprintf(stderr, "task-%lu waiting\n", id);
}
else if (ret == 0)
{
auto *router = create_router_task(route_callback);
series_of(task)->push_front(router);
fprintf(stderr, "task-%lu routing\n", id);
}
else
{
fprintf(stderr, "task-%lu get [%s]\n", id, (*msg).c_str());
}
});
SeriesWork *series = Workflow::create_series_work(timer, nullptr);
auto *worker = WFTaskFactory::create_go_task(work,
(std::string *)series->get_context(),
&wg);
series->push_back(worker);
series->start();
}
wg.wait();
}