Сцены

crontab в Linux играет важную роль в реализации задач с синхронизацией, так как же мы можем это сделать, если реализуем contab с синхронизацией? Как это может быть эффективно? В этой статье будет представлена реализация таймера внутри Nginx и реализация таймера Tick в golang.

традиционное мышление

В бесконечном цикле каждую секунду (с точностью до секунд) для опроса, чтобы узнать, есть ли задачи, которые нужно выполнить.

Вопрос первый

Нужно ли опрашивать каждую секунду? В следующую секунду, в следующую секунду, в следующую секунду, в следующую секунду и в следующую секунду может не быть задач, которые должны быть выполнены, поэтому это будет пустой тратой ресурсов ЦП, почему бы не спать до момента времени последнего выполнить задачу, а затем перейти к разделу Что насчет обхода?

Вопрос второй

Какую структуру данных следует использовать для хранения всех задач для эффективного опроса? с массивом? Проходить каждый раз? Понятно, что не эффективно. В Nginx используется красно-черное дерево, а в golang — небольшая куча.

реализация галочки в голанге

type timer struct {
    i int // heap index
    when   int64
    period int64
    f      func(interface{}, uintptr)
    arg    interface{}
    seq    uintptr
}

Golang использует структуру таймера для представления задачи по времени, где when представляет собой абсолютную отметку времени истечения срока действия, которая также является ключевым полем для сортировки и создания небольшой кучи. Таким образом, первой задачей, срок действия которой истечет, будет узел во главе небольшой кучи. Поле периода используется, чтобы указать, выполнять ли запланированную задачу циклически.Если это циклическая задача, она будет повторно изменять момент времени следующего выполнения после выполнения и корректировать небольшую кучу. В противном случае удалите запланированную задачу из небольшой кучи.

func addtimerLocked(t *timer) {
    if t.when < 0 {
        t.when = 1<<63 - 1
    }
    t.i = len(timers.t)
    timers.t = append(timers.t, t) // 插入到小堆尾部
    siftupTimer(t.i)  // 堆调整
    if t.i == 0 {
        // 如果新加入的timer比之前最早到期的timer还早
        // 更新后台运行的检查到期timer 的 goroutine 的睡眠时间
        if timers.sleeping {
            timers.sleeping = false
            notewakeup(&timers.waitnote)
        }
        if timers.rescheduling {
            timers.rescheduling = false
            goready(timers.gp, 0)
        }
    }
    if !timers.created {
        // 如果是第一次创建的定时器维护
        // 启动一条后台goroutine来维护定时器
        timers.created = true
        go timerproc()
    }
}

func timerproc() {
    timers.gp = getg()
    for {
        lock(&timers.lock)
        timers.sleeping = false
        now := nanotime()
        delta := int64(-1)
        for {
            if len(timers.t) == 0 {
                delta = -1
                break
            }
            t := timers.t[0] // 获取小堆第一个定时任务，即最早到期的任务
            delta = t.when - now
            if delta > 0 {
                break
            }
            if t.period > 0 {
                // 如果是循环执行的定时任务，重新计算下次执行时间，重新调整堆内位置
                t.when += t.period * (1 + -delta/t.period)
                siftdownTimer(0)
            } else {
                // 如果是只执行一次，直接把它从堆内删除
                last := len(timers.t) - 1
                if last > 0 {
                    timers.t[0] = timers.t[last]
                    timers.t[0].i = 0
                }
                timers.t[last] = nil
                timers.t = timers.t[:last]
                if last > 0 {
                    siftdownTimer(0)
                }
                t.i = -1 // mark as removed
            }
            f := t.f
            arg := t.arg
            seq := t.seq
            unlock(&timers.lock)
            if raceenabled {
                raceacquire(unsafe.Pointer(t))
            }
            // 找到到期的任务，并执行
            f(arg, seq)
            lock(&timers.lock)
        }
        if delta < 0 || faketime > 0 {
            // No timers left - put goroutine to sleep.
            timers.rescheduling = true
            goparkunlock(&timers.lock, "timer goroutine (idle)", traceEvGoBlock, 1)
            continue
        }
        // 休眠到最早到期的定时任务
        timers.sleeping = true
        noteclear(&timers.waitnote)
        unlock(&timers.lock)
        notetsleepg(&timers.waitnote, delta)
    }
}

Таймер Nginx

Nginx использует таймеры во многих местах, например, при чтении тайм-аута заголовка http. Nginx использует красно-черное дерево для поддержки всех событий запланированных задач.После каждого опроса события процесс будет проверять красно-черное дерево, обрабатывать просроченные события с истекшим сроком действия и устанавливать поле тайм-аута в структуре ngx_event_t равным 1.

// ngx_event_timer.h
ngx_int_t ngx_event_timer_init(ngx_log_t *log); // 初始化定时任务管理器
ngx_msec_t ngx_event_find_timer(void);          // 查找定时恩物
void ngx_event_expire_timers(void);             // 处理过期的定时任务
ngx_int_t ngx_event_no_timers_left(void);
extern ngx_rbtree_t  ngx_event_timer_rbtree;    // 全局变量，维护保存所有定时任务时间

Рабочий процесс Nginx проверяет и обрабатывает события с истекшим сроком действия

// worker的事件和定时任务处理函数
void
ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
    ngx_uint_t  flags;
    ngx_msec_t  timer, delta;
    // ...
    delta = ngx_current_msec;
    (void) ngx_process_events(cycle, timer, flags);
    delta = ngx_current_msec - delta;
    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
                   "timer delta: %M", delta);
    // 事件处理
    ngx_event_process_posted(cycle, &ngx_posted_accept_events);
    if (ngx_accept_mutex_held) {
        ngx_shmtx_unlock(&ngx_accept_mutex);
    }
    if (delta) {
        // 处理过期的定时任务
        ngx_event_expire_timers();
    }
    // ...
}

ngx_event_expire_timers истек срок обработки задачи времени

void
ngx_event_expire_timers(void)
{
    ngx_event_t        *ev;
    ngx_rbtree_node_t  *node, *root, *sentinel;
    sentinel = ngx_event_timer_rbtree.sentinel;
    // 循环查找处理当前所有过期的时间
    for ( ;; ) {
        root = ngx_event_timer_rbtree.root;
        if (root == sentinel) {
            return;
        }
        node = ngx_rbtree_min(root, sentinel);  // 在红黑树种查找当前时间最小的节点
        if ((ngx_msec_int_t) (node->key - ngx_current_msec) > 0) {
            // 如果时间最小的节点都还没过期，直接返回
            return;
        }
        ev = (ngx_event_t *) ((char *) node - offsetof(ngx_event_t, timer));
        ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
                       "event timer del: %d: %M",
                       ngx_event_ident(ev->data), ev->timer.key);
        ngx_rbtree_delete(&ngx_event_timer_rbtree, &ev->timer); // 把时间从红黑树种删除
        ev->timer_set = 0;
        ev->timedout = 1;  // 把时间设为过期
        ev->handler(ev);   // 回调事件处理函数
    }
}

Обработчик событий обратного вызова, в качестве примера возьмем обработчик http-заголовка (ngx_http_process_request_headers)

static void
ngx_http_process_request_headers(ngx_event_t *rev)
{
    // ...
    
    c = rev->data;
    r = c->data;
    ngx_log_debug0(NGX_LOG_DEBUG_HTTP, rev->log, 0,
                   "http process request header line");
    // 如果过期了，在上一步ngx_event_expire_timers中，timedout字段会被设为1，表示头部处理超时了，就给客户端错误提示
    if (rev->timedout) {
        ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, "client timed out");
        c->timedout = 1;
        // 超时，关闭链接，发送request timeout错误
        ngx_http_close_request(r, NGX_HTTP_REQUEST_TIME_OUT);
        return;
    }
    // ...
}

Суммировать

Для реализации запланированного планировщика задач требуется процесс (поток или сопрограмма) + эффективная структура данных (для выполнения эффективного запроса, частой вставки и удаления), например красно-черное дерево, используемое Nginx, небольшая куча, используемая golang, или таблица пропуска (список переходов)), таблица переходов представляет собой упорядоченный связанный список, а эффективность вставки, удаления и поиска неплохая, и ее легко реализовать, поэтому вы можете ее рассмотреть.