1 2 345 6 7 8 / 8 页下一页

[C] tbox新增stackless协程支持 [复制链接]

waruqi

白手起家

论坛徽章:: 1

31楼 [报告]

发表于 2017-01-29 15:37 |只看该作者

回复 29# yulihua49

我试下。。

static tb_void_t tb_demo_coroutine_client(tb_cpointer_t priv)
{
// check
tb_socket_ref_t sock = (tb_socket_ref_t)priv;
tb_assert_and_check_return(sock);
// trace
tb_trace_d("[%p]: sending %s ..", sock, g_filepath);
// init file
tb_file_ref_t file = tb_file_init(g_filepath, TB_FILE_MODE_RO | TB_FILE_MODE_BINARY);
tb_assert_and_check_return(file);
// send data
tb_hize_t send = 0;
tb_hize_t size = tb_file_size(file);
tb_long_t wait = 0;
tb_hong_t time = tb_mclock();
while (send < size)
{
// 直接发送文件，利用sendfile优化文件传输
tb_hong_t real = tb_socket_sendf(sock, file, send, size - send);
// trace
tb_trace_d("[%p]: send: %ld", sock, real);
// has data?
if (real > 0)
{
send += real;
wait = 0;
}
// no data? wait it
else if (!real && !wait)
{
// 如果在协程模式下，会把socket添加到io调度器中，进行自动调度
wait = tb_socket_wait(sock, TB_SOCKET_EVENT_SEND, TB_DEMO_TIMEOUT);
tb_assert_and_check_break(wait >= 0);
}
// failed or end?
else break;
}
// trace
tb_trace_i("[%p]: send: %lld bytes %lld ms", sock, send, tb_mclock() - time);
// exit file
tb_file_exit(file);
// exit socket
tb_socket_exit(sock);
}
static tb_void_t tb_demo_coroutine_listen(tb_cpointer_t priv)
{
// done
tb_socket_ref_t sock = tb_null;
do
{
// init socket
sock = tb_socket_init(TB_SOCKET_TYPE_TCP, TB_IPADDR_FAMILY_IPV4);
tb_assert_and_check_break(sock);
// bind socket
tb_ipaddr_t addr;
tb_ipaddr_set(&addr, tb_null, TB_DEMO_PORT, TB_IPADDR_FAMILY_IPV4);
if (!tb_socket_bind(sock, &addr)) break;
// listen socket
if (!tb_socket_listen(sock, 1000)) break;
// trace
tb_trace_i("listening ..");
// 此处也会进入协程io调度，等待accept事件
while (tb_socket_wait(sock, TB_SOCKET_EVENT_ACPT, -1) > 0)
{
// accept client sockets
tb_size_t count = 0;
tb_socket_ref_t client = tb_null;
while ((client = tb_socket_accept(sock, tb_null)))
{
// start client connection
if (!tb_coroutine_start(tb_null, tb_demo_coroutine_client, client, 0)) break;
count++;
}
// trace
tb_trace_i("listened %lu", count);
}
} while (0);
// exit socket
if (sock) tb_socket_exit(sock);
sock = tb_null;
}
tb_int_t main(tb_int_t argc, tb_char_t** argv)
{
// check
tb_assert_and_check_return_val(argc == 2 && argv[1], -1);
// the file path
tb_char_t const* filepath = argv[1];
tb_assert_and_check_return_val(filepath, -1);
// save the file path
tb_strlcpy(g_filepath, filepath, sizeof(g_filepath));
// init scheduler
tb_co_scheduler_ref_t scheduler = tb_co_scheduler_init();
if (scheduler)
{
// start listening
tb_coroutine_start(scheduler, tb_demo_coroutine_listen, tb_null, 0);
// run scheduler
tb_co_scheduler_loop(scheduler, tb_true);
// exit scheduler
tb_co_scheduler_exit(scheduler);
}
return 0;
}

复制代码

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

waruqi

白手起家

论坛徽章:: 1

32楼 [报告]

发表于 2017-01-29 15:38 |只看该作者

回复 29# yulihua49

可以贴代码啊。。

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

yulihua49

巨富豪门

论坛徽章:: 15

33楼 [报告]

发表于 2017-01-29 19:46 |只看该作者

waruqi 发表于 2017-01-29 15:38
回复 29# yulihua49

可以贴代码啊。。

贴了3遍都没了，还白说了好多话，伤心了。

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

waruqi

白手起家

论坛徽章:: 1

34楼 [报告]

发表于 2017-01-30 10:35 |只看该作者

额。。好吧。。

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

yulihua49

巨富豪门

论坛徽章:: 15

35楼 [报告]

发表于 2017-02-01 20:06 |只看该作者

waruqi 发表于 2017-01-30 10:35
额。。好吧。。

static void *thread_work(void *param)

{

resource *rs=(resource *)param;

int ret,fds;

TCB *task=NULL;

struct epoll_event event;

ShowLog(2,"%s:thread %lx start!",__FUNCTION__,pthread_self());

getcontext(&rs->tc);

if(task) pthread_mutex_unlock(&task->lock);

while(1) {

//从就绪队列取一个任务

pthread_mutex_lock(&rpool.mut);

while(!(task=rdy_get())) {

if(rpool.flg >= tpool.rdy_num) break;

rpool.flg++;

ret=pthread_cond_wait(&rpool.cond,&rpool.mut); //没有任务，等待

rpool.flg--;

}

pthread_mutex_unlock(&rpool.mut);

if(task) {

if(!task->AIO_flg && !task->call_back) {

task->fd=task->conn.Socket;

ShowLog(5,"%s:tid=%lx,TCB_no=%d from rdy_queue",__FUNCTION__,

pthread_self(),task->sv.TCB_no);

if(task->fd>=0) {

do_epoll(task,0,0);

}

continue;

}

} else {

fds = epoll_wait(g_epoll_fd, &event, 1 , -1);

if(fds < 0){

ShowLog(1,"%s:epoll_wait err=%d,%s",__FUNCTION__,errno,strerror(errno));

usleep(30000000);

continue;

}

task = (TCB *)event.data.ptr;

if(task->events) {

ShowLog(1,"%s:tid=%lx,TCB_no=%d,task->events=%08X,conflict!",__FUNCTION__,

pthread_self(),task->sv.TCB_no,task->events);//发现惊群

task=NULL;

continue;//丢掉它

}

task->events=event.events;

}

rs->timestamp=now_usec();

if(task->status>0) set_showid(task->ctx);//Showid 应该在会话上下文结构里

if(task->AIO_flg) {//fiber task

task->uc.uc_link=&rs->tc;

rs->tc.uc_link=(ucontext_t *)task;

ShowLog(5,"%s:tid=%lx,resume to TCB_no=%d",__FUNCTION__,pthread_self(),task->sv.TCB_no);

pthread_mutex_lock(&task->lock);//防止其他线程提前闯入

setcontext(&task->uc); //== longjmp()

continue;//no action,logic only

}

if(task->uc.uc_stack.ss_size>0) {//call_back模式，抢入了，进入同步模式

rs->tc.uc_link=NULL;

ShowLog(5,"%s:tid %lx 抢入 SYNC",__FUNCTION__,pthread_self());

do_work(task->sv.TCB_no);

continue;

}

if(!rs->tc.uc_stack.ss_sp) {

ShowLog(5,"%s:%lx create fiber for TCB_no=%d",__FUNCTION__,rs->tid,task->sv.TCB_no);

task->uc.uc_stack.ss_sp=mmap(0, use_stack_size,

PROT_READ | PROT_WRITE | PROT_EXEC,

MAP_PRIVATE | MAP_ANON | MAP_GROWSDOWN, -1, 0);

if(task->uc.uc_stack.ss_sp==MAP_FAILED) {

task->uc.uc_stack.ss_sp=NULL;

do_work(task->sv.TCB_no); //进行你的服务,不使用AIO

continue;

}

} else {

//ShowLog(5,"%s:%lx reuse fiber for TCB_no=%d",__FUNCTION__,rs->tid,task->sv.TCB_no);

task->uc.uc_stack.ss_sp=rs->tc.uc_stack.ss_sp;

rs->tc.uc_stack.ss_sp=NULL;

rs->tc.uc_stack.ss_size=0;

}

task->uc.uc_stack.ss_size=use_stack_size;

task->uc.uc_link=&rs->tc;

rs->tc.uc_link=(ucontext_t *)task;

makecontext(&task->uc,(void (*)())do_work,1,task->sv.TCB_no);

ret=swapcontext(&rs->tc,&task->uc);

if(ret<0) {

ShowLog(1,"%s:swapcontext fault TCB_NO=%d,tid=%lx,errno=%d,%s",

__FUNCTION__,task->sv.TCB_no,pthread_self(),ret,strerror(abs(ret)));

rs->tc.uc_link=NULL;

task->uc.uc_link=NULL;

if(task->uc.uc_stack.ss_sp)

munmap(task->uc.uc_stack.ss_sp,task->uc.uc_stack.ss_size);

task->uc.uc_stack.ss_sp=NULL;

task->uc.uc_stack.ss_size=0;

do_work(task->sv.TCB_no);

mthr_showid_del(rs->tid);

continue;

}

if(!task) {

ShowLog(1,"%s:aft swapcontext task is NULL",__FUNCTION__);

continue;

}

if(!task->AIO_flg) {//service complate

if(!rs->tc.uc_stack.ss_size) {//回收fiber stack

//ShowLog(5,"%s:%lx release fiber from TCB_no=%d",__FUNCTION__,rs->tid,task->sv.TCB_no);

rs->tc.uc_stack.ss_sp=task->uc.uc_stack.ss_sp;

if(rs->tc.uc_stack.ss_sp)

rs->tc.uc_stack.ss_size=use_stack_size;

else rs->tc.uc_stack.ss_size=0;

} else {

ShowLog(5,"%s:%lx destroy fiber from TCB_no=%d",__FUNCTION__,rs->tid,task->sv.TCB_no);

if(task->uc.uc_stack.ss_sp)

munmap(task->uc.uc_stack.ss_sp,task->uc.uc_stack.ss_size);

}

task->uc.uc_stack.ss_sp=NULL;

rs->tc.uc_link=NULL;

task->uc.uc_link=NULL;

task->uc.uc_stack.ss_size=0;//mark fiber cpmplate

//ShowLog(5,"%s:TCB_no=%d,tid=%lx,timeout=%d,conn.timeout=%d",__FUNCTION__,task->sv.TCB_no,rs->tid,task->timeout,task->conn.timeout);

} else {

pthread_mutex_unlock(&task->lock);

ShowLog(5,"%s:tid=%lx,fiber yield from TCB_no=%d",

__FUNCTION__,pthread_self(),task->sv.TCB_no);

}

mthr_showid_del(rs->tid);

}

ShowLog(1,"%s:tid=%lx canceled",__FUNCTION__,pthread_self());

mthr_showid_del(rs->tid);

rs->timestamp=now_usec();

rs->status=0;

rs->tid=0;

if(rs->tc.uc_stack.ss_sp) {

munmap(rs->tc.uc_stack.ss_sp,rs->tc.uc_stack.ss_size);

rs->tc.uc_stack.ss_sp=NULL;

rs->tc.uc_stack.ss_size=0;

}

return NULL;

}

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

windoze

版主

论坛徽章:: 44

36楼 [报告]

发表于 2017-02-03 01:18 |只看该作者

本帖最后由 windoze 于 2017-02-03 01:25 编辑

stackless coroutine里面不能有局部变量，这个东西更像CPS，所有coroutine local state都要放在一个外部的stack frame里面，你可以想象它是这个样子的：

f1(stack_frame *shadow_stack) {
// 创建两个局部变量
shadow_stack->push_local_variable("var1") = 10;
shadow_stack->push_local_variable("var2") = shadow_stack->get_local_variable("var1")+42;
// ...
// 调用f2，f2必须也长这个样子
shadow_stack->push(address_of(return_to_here));
longjmp(f2, shadow_stack); // f2结束时会longjmp回来
// caller负责清理stack
return_to_here:
shadow_stack->pop();
// ...
// 返回前清理局部变量
shadow_stack->pop_local_variable("var1");
shadow_stack->pop_local_variable("var2");
longjmp(shadow_stack->top, shadow_stack);
}

复制代码

整个函数并没有使用machine stack，而是搞了一个shadow stack，所以只需要一个固定尺寸的结构就可以做coroutine。

但是这种方式最讨厌的地方在于coroutine里面绝对不能用machine stack，因为machine stack当前sp之上也可能有有效的数据，比如最开始的起始函数里在第一个continuation之后定义的局部变量，也就是说你不但不能使用局部变量，甚至不能调用其它正常的函数，因为一次正常的函数调用会有入栈出栈的动作。
正确的姿势应该是由编译器做全局CPS变换，C#里面的async/await就是这么搞的，python里面的generator/yield也可以说是这么搞的（话说python本来就没用machine stack，倒是没有不能调用其它函数的麻烦），据说java 9里也要有类似的东西。

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

waruqi

白手起家

论坛徽章:: 1

37楼 [报告]

发表于 2017-02-07 13:55 |只看该作者

回复 36# windoze

对于stackless协程，一般情况下没必要做成动态栈，如果变量不多的话，可以做成struct，虽然会浪费一些空间，但是可读性、易用性会好很多。。
例如，我用stackless协程写的server:

// the client type
typedef struct __tb_demo_lo_client_t
{
// the client socket
tb_socket_ref_t sock;
// the file
tb_file_ref_t file;
// the data buffer size
tb_hize_t size;
// the send size
tb_hize_t send;
// the real size
tb_hong_t real;
// the wait state
tb_long_t wait;
// the time
tb_hong_t time;
}tb_demo_lo_client_t, *tb_demo_lo_client_ref_t;
// the listen type
typedef struct __tb_demo_lo_listen_t
{
// the listen socket
tb_socket_ref_t sock;
// the address
tb_ipaddr_t addr;
// the client socket
tb_socket_ref_t client;
}tb_demo_lo_listen_t, *tb_demo_lo_listen_ref_t;
static tb_void_t tb_demo_lo_coroutine_client(tb_lo_coroutine_ref_t coroutine, tb_cpointer_t priv)
{
// check
tb_demo_lo_client_ref_t client = (tb_demo_lo_client_ref_t)priv;
tb_assert(client);
// enter coroutine
tb_lo_coroutine_enter(coroutine)
{
// trace
tb_trace_d("[%p]: sending %s ..", client->sock, g_filepath);
// init file
client->file = tb_file_init(g_filepath, TB_FILE_MODE_RO | TB_FILE_MODE_BINARY);
tb_assert(client->file);
// send data
client->size = tb_file_size(client->file);
client->time = tb_mclock();
while (client->send < client->size)
{
// send it
client->real = tb_socket_sendf(client->sock, client->file, client->send, client->size - client->send);
// trace
tb_trace_d("[%p]: send: %ld", client->sock, client->real);
// has data?
if (client->real > 0)
{
client->send += client->real;
client->wait = 0;
}
// no data? wait it
else if (!client->real && !client->wait)
{
// wait it
tb_lo_coroutine_waitio(client->sock, TB_SOCKET_EVENT_SEND, TB_DEMO_TIMEOUT);
// wait ok
client->wait = tb_lo_coroutine_events();
tb_assert_and_check_break(client->wait >= 0);
}
// failed or end?
else break;
}
// trace
tb_trace_i("[%p]: send: %lld bytes %lld ms", client->sock, client->send, tb_mclock() - client->time);
// exit file
tb_file_exit(client->file);
// exit socket
tb_socket_exit(client->sock);
}
}
static tb_void_t tb_demo_lo_coroutine_listen(tb_lo_coroutine_ref_t coroutine, tb_cpointer_t priv)
{
// check
tb_demo_lo_listen_ref_t listen = (tb_demo_lo_listen_ref_t)priv;
tb_assert(listen);
// enter coroutine
tb_lo_coroutine_enter(coroutine)
{
// done
do
{
// init socket
listen->sock = tb_socket_init(TB_SOCKET_TYPE_TCP, TB_IPADDR_FAMILY_IPV4);
tb_assert_and_check_break(listen->sock);
// bind socket
tb_ipaddr_set(&listen->addr, tb_null, TB_DEMO_PORT, TB_IPADDR_FAMILY_IPV4);
if (!tb_socket_bind(listen->sock, &listen->addr)) break;
// listen socket
if (!tb_socket_listen(listen->sock, 1000)) break;
// trace
tb_trace_i("listening ..");
// loop
while (1)
{
// wait accept events
tb_lo_coroutine_waitio(listen->sock, TB_SOCKET_EVENT_ACPT, -1);
// wait ok
if (tb_lo_coroutine_events() > 0)
{
// accept client sockets
while ((listen->client = tb_socket_accept(listen->sock, tb_null)))
{
// start client connection
if (!tb_lo_coroutine_start(tb_lo_scheduler_self(), tb_demo_lo_coroutine_client, tb_lo_coroutine_pass1(tb_demo_lo_client_t, sock, listen->client))) break;
}
}
}
} while (0);
// exit socket
if (listen->sock) tb_socket_exit(listen->sock);
listen->sock = tb_null;
}
}
tb_int_t main(tb_int_t argc, tb_char_t** argv)
{
// check
tb_assert_and_check_return_val(argc == 2 && argv[1], -1);
// init tbox
if (!tb_init(tb_null, tb_null)) return -1;
// the file path
tb_char_t const* filepath = argv[1];
tb_assert_and_check_return_val(filepath, -1);
// save the file path
tb_strlcpy(g_filepath, filepath, sizeof(g_filepath));
// init scheduler
tb_lo_scheduler_ref_t scheduler = tb_lo_scheduler_init();
if (scheduler)
{
// start listening
tb_lo_coroutine_start(scheduler, tb_demo_lo_coroutine_listen, tb_lo_coroutine_pass(tb_demo_lo_listen_t));
// run scheduler
tb_lo_scheduler_loop(scheduler, tb_true);
// exit scheduler
tb_lo_scheduler_exit(scheduler);
}
// exit tbox
tb_exit();
return 0;
}

复制代码

或者可以看下，我用stackless写的http server，可读性跟stackfull比，差不了多少。。

stackless的http server: https://github.com/tboox/tbox/blob/master/src/demo/coroutine/stackless/http_server.c
stackfull的http server: https://github.com/tboox/tbox/blob/master/src/demo/coroutine/http_server.c

实战分享：从技术角度谈机器学习入门| 【大话IT】RadonDB低门槛向MySQL集群下战书 | ChinaUnix打赏功能已上线！ | 新一代分布式关系型数据库RadonDB知多少？

waruqi

白手起家

论坛徽章:: 1

38楼 [报告]

发表于 2017-02-07 14:03 |只看该作者

回复 36# windoze

一般情况下，就算是stackless协程，也没必要去用动态栈，变量不多的话，直接Struct就行了，虽然会浪费一点空间，但是可读性、易用性会提升不少，跟stackfull的使用，其实差别已经不是很大了
例如，可以看下我用stackless写的server：

// the client type
typedef struct __tb_demo_lo_client_t
{
// the client socket
tb_socket_ref_t sock;
// the file
tb_file_ref_t file;
// the data buffer size
tb_hize_t size;
// the send size
tb_hize_t send;
// the real size
tb_hong_t real;
// the wait state
tb_long_t wait;
// the time
tb_hong_t time;
}tb_demo_lo_client_t, *tb_demo_lo_client_ref_t;
// the listen type
typedef struct __tb_demo_lo_listen_t
{
// the listen socket
tb_socket_ref_t sock;
// the address
tb_ipaddr_t addr;
// the client socket
tb_socket_ref_t client;
}tb_demo_lo_listen_t, *tb_demo_lo_listen_ref_t;
static tb_void_t tb_demo_lo_coroutine_client(tb_lo_coroutine_ref_t coroutine, tb_cpointer_t priv)
{
// check
tb_demo_lo_client_ref_t client = (tb_demo_lo_client_ref_t)priv;
tb_assert(client);
// enter coroutine
tb_lo_coroutine_enter(coroutine)
{
// trace
tb_trace_d("[%p]: sending %s ..", client->sock, g_filepath);
// init file
client->file = tb_file_init(g_filepath, TB_FILE_MODE_RO | TB_FILE_MODE_BINARY);
tb_assert(client->file);
// send data
client->size = tb_file_size(client->file);
client->time = tb_mclock();
while (client->send < client->size)
{
// send it
client->real = tb_socket_sendf(client->sock, client->file, client->send, client->size - client->send);
// trace
tb_trace_d("[%p]: send: %ld", client->sock, client->real);
// has data?
if (client->real > 0)
{
client->send += client->real;
client->wait = 0;
}
// no data? wait it
else if (!client->real && !client->wait)
{
// wait it
tb_lo_coroutine_waitio(client->sock, TB_SOCKET_EVENT_SEND, TB_DEMO_TIMEOUT);
// wait ok
client->wait = tb_lo_coroutine_events();
tb_assert_and_check_break(client->wait >= 0);
}
// failed or end?
else break;
}
// trace
tb_trace_i("[%p]: send: %lld bytes %lld ms", client->sock, client->send, tb_mclock() - client->time);
// exit file
tb_file_exit(client->file);
// exit socket
tb_socket_exit(client->sock);
}
}
static tb_void_t tb_demo_lo_coroutine_listen(tb_lo_coroutine_ref_t coroutine, tb_cpointer_t priv)
{
// check
tb_demo_lo_listen_ref_t listen = (tb_demo_lo_listen_ref_t)priv;
tb_assert(listen);
// enter coroutine
tb_lo_coroutine_enter(coroutine)
{
// done
do
{
// init socket
listen->sock = tb_socket_init(TB_SOCKET_TYPE_TCP, TB_IPADDR_FAMILY_IPV4);
tb_assert_and_check_break(listen->sock);
// bind socket
tb_ipaddr_set(&listen->addr, tb_null, TB_DEMO_PORT, TB_IPADDR_FAMILY_IPV4);
if (!tb_socket_bind(listen->sock, &listen->addr)) break;
// listen socket
if (!tb_socket_listen(listen->sock, 1000)) break;
// trace
tb_trace_i("listening ..");
// loop
while (1)
{
// wait accept events
tb_lo_coroutine_waitio(listen->sock, TB_SOCKET_EVENT_ACPT, -1);
// wait ok
if (tb_lo_coroutine_events() > 0)
{
// accept client sockets
while ((listen->client = tb_socket_accept(listen->sock, tb_null)))
{
// start client connection
if (!tb_lo_coroutine_start(tb_lo_scheduler_self(), tb_demo_lo_coroutine_client, tb_lo_coroutine_pass1(tb_demo_lo_client_t, sock, listen->client))) break;
}
}
}
} while (0);
// exit socket
if (listen->sock) tb_socket_exit(listen->sock);
listen->sock = tb_null;
}
}
tb_int_t main(tb_int_t argc, tb_char_t** argv)
{
// check
tb_assert_and_check_return_val(argc == 2 && argv[1], -1);
// init tbox
if (!tb_init(tb_null, tb_null)) return -1;
// the file path
tb_char_t const* filepath = argv[1];
tb_assert_and_check_return_val(filepath, -1);
// save the file path
tb_strlcpy(g_filepath, filepath, sizeof(g_filepath));
// init scheduler
tb_lo_scheduler_ref_t scheduler = tb_lo_scheduler_init();
if (scheduler)
{
// start listening
tb_lo_coroutine_start(scheduler, tb_demo_lo_coroutine_listen, tb_lo_coroutine_pass(tb_demo_lo_listen_t));
// run scheduler
tb_lo_scheduler_loop(scheduler, tb_true);
// exit scheduler
tb_lo_scheduler_exit(scheduler);
}
// exit tbox
tb_exit();
return 0;
}