libhv/event/hloop.c

#include "hloop.h"
#include "hevent.h"
#include "iowatcher.h"

#include "hdef.h"
#include "hlog.h"
#include "hmath.h"

#define PAUSE_TIME              10      // ms
#define MIN_BLOCK_TIME          1       // ms
#define MAX_BLOCK_TIME          10000   // ms

#define IO_ARRAY_INIT_SIZE      64

static int timer_minheap_compare(const struct heap_node* lhs, const struct heap_node* rhs) {
    return TIMER_HEAP_ENTRY(lhs)->timeout < TIMER_HEAP_ENTRY(rhs)->timeout;
}

static int hloop_process_idles(hloop_t* loop) {
    int nidles = 0;
    struct list_node* node = loop->idles.next;
    hidle_t* idle = NULL;
    while (node != &loop->idles) {
        idle = IDLE_ENTRY(node);
        if (idle->destroy) goto destroy;
        if (!idle->active) goto next;
        if (idle->repeat == 0) {
            hidle_del(idle);
            //goto next;
            goto destroy;
        }
        if (idle->repeat != INFINITE) {
            --idle->repeat;
        }
        EVENT_PENDING(idle);
        ++nidles;
next:
        node = node->next;
        continue;
destroy:
        node = node->next;
        list_del(node->prev);
        free(idle);
    }
    return nidles;
}

static int hloop_process_timers(hloop_t* loop) {
    int ntimers = 0;
    struct list_node* node = loop->timers.next;
    htimer_t* timer = NULL;
    while (node != &loop->timers) {
        timer = TIMER_ENTRY(node);
        if (timer->destroy) goto destroy;
        if (!timer->active) goto next;
        if (timer->repeat == 0) {
            htimer_del(timer);
            //goto next;
            goto destroy;
        }
        if (loop->cur_hrtime > timer->next_timeout) {
            if (timer->repeat != INFINITE) {
                --timer->repeat;
            }
            timer->next_timeout += timer->timeout*1000;
            EVENT_PENDING(timer);
            ++ntimers;
        }
next:
        node = node->next;
        continue;
destroy:
        node = node->next;
        list_del(node->prev);
        free(timer);
    }
    return ntimers;
}

static int hloop_process_ios(hloop_t* loop, int timeout) {
    int nevents = iowatcher_poll_events(loop, timeout);
    if (nevents < 0) {
        hloge("poll_events error=%d", -nevents);
    }
    return nevents < 0 ? 0 : nevents;
}

static int hloop_process_pendings(hloop_t* loop) {
    if (loop->npendings == 0) return 0;

    hevent_t* prev = NULL;
    hevent_t* next = NULL;
    int ncbs = 0;
    for (int i = HEVENT_PRIORITY_SIZE-1; i >= 0; --i) {
        next = loop->pendings[i];
        while (next) {
            if (next->active && next->cb) {
                next->cb(next);
                ++ncbs;
            }
            prev = next;
            next = next->pending_next;
            prev->pending = 0;
            prev->pending_next = NULL;
        }
        loop->pendings[i] = NULL;
    }
    loop->npendings = 0;
    return ncbs;
}

static int hloop_process_events(hloop_t* loop) {
    // ios -> timers -> idles
    int nios, ntimers, nidles;
    nios = ntimers = nidles = 0;

    int blocktime = MAX_BLOCK_TIME;
    if (loop->timer_minheap.root) {
        // if have timers, blocktime = min_timeout
        blocktime = TIMER_HEAP_ENTRY(loop->timer_minheap.root)->timeout;
        //if (!list_empty(&loop->idles))
            blocktime /= 10;
    }
    blocktime = LIMIT(MIN_BLOCK_TIME, blocktime, MAX_BLOCK_TIME);
    // if you want timer more precise, reduce blocktime

    uint64_t last_hrtime = loop->cur_hrtime;
    nios = hloop_process_ios(loop, blocktime);
    hloop_update_time(loop);
    ntimers = hloop_process_timers(loop);
    if (loop->npendings == 0) {
        loop->idle_time += last_hrtime - loop->cur_hrtime;
        // avoid frequent call idles
        if (loop->cur_hrtime - loop->last_idle_hrtime > 1e6) {
            loop->last_idle_hrtime = loop->cur_hrtime;
            nidles= hloop_process_idles(loop);
        }
        else {
            // hloop_process_ios maybe nonblock, so sleep here
            msleep(blocktime);
        }
    }
    printd("blocktime=%d nios=%d ntimers=%d nidles=%d nactives=%d npendings=%d\n", blocktime, nios, ntimers, nidles, loop->nactives, loop->npendings);
    return hloop_process_pendings(loop);
}

int hloop_init(hloop_t* loop) {
    memset(loop, 0, sizeof(hloop_t));
    loop->status = HLOOP_STATUS_STOP;
    // idles
    list_init(&loop->idles);
    // timers
    list_init(&loop->timers);
    heap_init(&loop->timer_minheap, timer_minheap_compare);
    // iowatcher
    //iowatcher_init(loop);
    return 0;
}

void hloop_cleanup(hloop_t* loop) {
    // pendings
    for (int i = 0; i < HEVENT_PRIORITY_SIZE; ++i) {
        loop->pendings[i] = NULL;
    }
    // idles
    struct list_node* node = loop->idles.next;
    hidle_t* idle;
    while (node != &loop->idles) {
        idle = IDLE_ENTRY(node);
        node = node->next;
        free(idle);
    }
    list_init(&loop->idles);
    // timers
    node = loop->timers.next;
    htimer_t* timer;
    while (node != &loop->timers) {
        timer = TIMER_ENTRY(node);
        node = node->next;
        free(timer);
    }
    list_init(&loop->timers);
    heap_init(&loop->timer_minheap, NULL);
    // iowatcher
    //iowatcher_cleanup(loop);
};

int hloop_run(hloop_t* loop) {
    time(&loop->start_time);
    loop->start_hrtime = gethrtime();
    loop->loop_cnt = 0;
    loop->status = HLOOP_STATUS_RUNNING;
    while (loop->status != HLOOP_STATUS_STOP) {
        if (loop->status == HLOOP_STATUS_PAUSE) {
            msleep(PAUSE_TIME);
            hloop_update_time(loop);
            continue;
        }
        ++loop->loop_cnt;
        if (loop->nactives == 0) break;
        hloop_process_events(loop);
    }
    loop->status = HLOOP_STATUS_STOP;
    loop->end_hrtime = gethrtime();
    hloop_cleanup(loop);
    return 0;
};

int hloop_stop(hloop_t* loop) {
    loop->status = HLOOP_STATUS_STOP;
    return 0;
}

int hloop_pause(hloop_t* loop) {
    if (loop->status == HLOOP_STATUS_RUNNING) {
        loop->status = HLOOP_STATUS_PAUSE;
    }
    return 0;
}

int hloop_resume(hloop_t* loop) {
    if (loop->status == HLOOP_STATUS_PAUSE) {
        loop->status = HLOOP_STATUS_RUNNING;
    }
    return 0;
}

hidle_t* hidle_add(hloop_t* loop, hidle_cb cb, uint32_t repeat) {
    hidle_t* idle = (hidle_t*)malloc(sizeof(hidle_t));
    memset(idle, 0, sizeof(hidle_t));
    idle->event_type = HEVENT_TYPE_IDLE;
    idle->repeat = repeat;
    list_add(&idle->node, &loop->idles);
    EVENT_ADD(loop, idle, cb);
    return idle;
}

void hidle_del(hidle_t* idle) {
    EVENT_DEL(idle);
}

htimer_t* htimer_add(hloop_t* loop, htimer_cb cb, uint64_t timeout, uint32_t repeat) {
    htimer_t* timer = (htimer_t*)malloc(sizeof(htimer_t));
    memset(timer, 0, sizeof(htimer_t));
    timer->event_type = HEVENT_TYPE_TIMER;
    timer->repeat = repeat;
    timer->timeout = timeout;
    timer->next_timeout = gethrtime() + timeout*1000;
    list_add(&timer->node, &loop->timers);
    heap_insert(&loop->timer_minheap, &timer->hnode);
    EVENT_ADD(loop, timer, cb);
    return timer;
}

void htimer_del(htimer_t* timer) {
    heap_remove(&timer->loop->timer_minheap, &timer->hnode);
    EVENT_DEL(timer);
}

void hio_init(hio_t* io) {
    memset(io, 0, sizeof(hio_t));
    io->event_type = HEVENT_TYPE_IO;
    io->event_index[0] = io->event_index[1] = -1;
    // move to hwrite
    //write_queue_init(&io->write_queue, 4);;
};

void hio_cleanup(hio_t* io) {
    offset_buf_t* pbuf = NULL;
    while (!write_queue_empty(&io->write_queue)) {
        pbuf = write_queue_front(&io->write_queue);
        SAFE_FREE(pbuf->base);
        write_queue_pop_front(&io->write_queue);
    }
    write_queue_cleanup(&io->write_queue);
}

hio_t* hio_add(hloop_t* loop, hio_cb cb, int fd, int events) {
    if (loop->ios.maxsize == 0) {
        io_array_init(&loop->ios, IO_ARRAY_INIT_SIZE);
    }

    if (fd > loop->ios.maxsize) {
        io_array_resize(&loop->ios, ceil2e(fd));
    }

    hio_t* io = loop->ios.ptr[fd];
    if (io == NULL) {
        io = (hio_t*)malloc(sizeof(hio_t));
        memset(io, 0, sizeof(hio_t));
        loop->ios.ptr[fd] = io;
    }

    if (!io->active || io->destroy) {
        hio_init(io);
        EVENT_ADD(loop, io, cb);
    }

    io->fd = fd;
    if (cb) {
        io->cb = (hevent_cb)cb;
    }
    iowatcher_add_event(loop, fd, events);
    io->events |= events;
    return io;
}

void hio_del(hio_t* io, int events) {
    iowatcher_del_event(io->loop, io->fd, events);
    io->events &= ~events;
    if (io->events == 0) {
        hio_cleanup(io);
        EVENT_DEL(io);
    }
}