[libco] libco 工作流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
int fcntl(int fildes, int cmd, ...) {
    HOOK_SYS_FUNC(fcntl);
    ...
    switch (cmd) {
        ...
        case F_SETFL: {
            int param = va_arg(arg_list, int);
            int flag = param;
            if (co_is_enable_sys_hook() && lp) {
                /* 设置非阻塞。 */
                flag |= O_NONBLOCK;
            }
            ret = g_sys_fcntl_func(fildes, cmd, flag);
            if (0 == ret && lp) {
                lp->user_flag = param;
            }
            break;
        }
        ...
    };
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
int connect(int fd, const struct sockaddr *address, socklen_t address_len) {
    HOOK_SYS_FUNC(connect);
    ...
    //1.sys call
    int ret = g_sys_connect_func(fd, address, address_len);
    ...
    //2.wait
    int pollret = 0;
    struct pollfd pf = {0};

    for (int i = 0; i < 3; i++) {
        memset(&pf, 0, sizeof(pf));
        pf.fd = fd;
        pf.events = (POLLOUT | POLLERR | POLLHUP);

        /* 关联事件驱动，获取事件处理结果。 */
        pollret = poll(&pf, 1, 25000);

        if (1 == pollret) {
            break;
        }
    }
    ...
}

int poll(struct pollfd fds[], nfds_t nfds, int timeout) {
    HOOK_SYS_FUNC(poll);
    ...
    if (nfds_merge == nfds || nfds == 1) {
        /* 关联事件驱动，切换当前协程。 */
        ret = co_poll_inner(co_get_epoll_ct(), fds, nfds, timeout, g_sys_poll_func);
    }
    ...
}

int co_poll_inner(stCoEpoll_t *ctx, struct pollfd fds[], nfds_t nfds, int timeout, poll_pfn_t pollfunc) {
    ...
    int epfd = ctx->iEpollFd;
    stCoRoutine_t *self = co_self();

    //1.struct change
    /* fd 关联协程。 */
    stPoll_t &arg = *((stPoll_t *)malloc(sizeof(stPoll_t)));
    memset(&arg, 0, sizeof(arg));

    arg.iEpollFd = epfd;
    arg.fds = (pollfd *)calloc(nfds, sizeof(pollfd));
    arg.nfds = nfds;
    ...
    arg.pfnProcess = OnPollProcessEvent;
    arg.pArg = GetCurrCo(co_get_curr_thread_env());

    //2. add epoll
    for (nfds_t i = 0; i < nfds; i++) {
        ...
        if (fds[i].fd > -1) {
            ev.data.ptr = arg.pPollItems + i;
            ev.events = PollEvent2Epoll(fds[i].events);
            /* 向事件驱动添加关注 fd 事件。 */
            int ret = co_epoll_ctl(epfd, EPOLL_CTL_ADD, fds[i].fd, &ev);
            ...
        }
        //if fail,the timeout would work
    }

    //3.add timeout
    /* 事件添加时钟，避免超时。 */
    unsigned long long now = GetTickMS();
    arg.ullExpireTime = now + timeout;
    int ret = AddTimeout(ctx->pTimeout, &arg, now);
    int iRaiseCnt = 0;
    if (ret != 0) {
        co_log_err("CO_ERR: AddTimeout ret %d now %lld timeout %d arg.ullExpireTime %lld",
                   ret, now, timeout, arg.ullExpireTime);
        errno = EINVAL;
        iRaiseCnt = -1;
    } else {
        /* 将当前协程挂起，唤醒其它协程。等待事件驱动事件通知或者时钟过期。 */
        co_yield_env(co_get_curr_thread_env());
        /* 协程挂起后被唤醒，从这行代码开始执行。 */
        iRaiseCnt = arg.iRaiseCnt;
    }
    ...
    {
        //clear epoll status and memory
        RemoveFromLink<stTimeoutItem_t, stTimeoutItemLink_t>(&arg);
        for (nfds_t i = 0; i < nfds; i++) {
            int fd = fds[i].fd;
            if (fd > -1) {
                /* 已获得结果，清除事件数据。 */
                co_epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &arg.pPollItems[i].stEvent);
            }
            /* 填充获得的数据。 */
            fds[i].revents = arg.fds[i].revents;
        }
        ...
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
void OnPollProcessEvent(stTimeoutItem_t *ap) {
    stCoRoutine_t *co = (stCoRoutine_t *)ap->pArg;
    /* 唤醒协程。 */
    co_resume(co);
}

void co_eventloop(stCoEpoll_t *ctx, pfn_co_eventloop_t pfn, void *arg) {
    ...
    for (;;) {
        /* 主协程通过 epoll_wait 捞出就绪事件。 */
        int ret = co_epoll_wait(ctx->iEpollFd, result, stCoEpoll_t::_EPOLL_SIZE, 1);

        stTimeoutItemLink_t *active = (ctx->pstActiveList);
        stTimeoutItemLink_t *timeout = (ctx->pstTimeoutList);
        ...
        /* 主协程捞出过期事件。 */
        TakeAllTimeout(ctx->pTimeout, now, timeout);
        ...
        Join<stTimeoutItem_t, stTimeoutItemLink_t>(active, timeout);

        lp = active->head;
        while (lp) {
            ...
            /* 主协程挂起当前协程，切换到对应子协程，处理到期事件和就绪事件结果。 */
            if (lp->pfnProcess) {
                lp->pfnProcess(lp);
            }
            lp = active->head;
        }
        ...
    }
}