深入学习keepalived之预备工作--线程

1. 线程的定义

1.1 线程定义在scheduler.h文件中，其定义如下所示

/* Thread itself. */
typedef struct _thread {
    unsigned long id;          /*identify*/
    unsigned char type;        /* thread type */
    struct _thread *next;        /* next pointer of the thread */
    struct _thread *prev;        /* previous pointer of the thread */
    struct _thread_master *master;    /* pointer to the struct thread_master. */
    int (*func) (struct _thread *);    /* event function */
    void *arg;            /* event argument */
    timeval_t sands;        /* rest of time sands value. */
    union {
        int val;        /* second argument of the event. */
        int fd;            /* file descriptor in case of read/write. */
        struct {
            pid_t pid;    /* process id a child thread is wanting. */
            int status;    /* return status of the process */
        } c;
    } u;
} thread_t;

1.2. 线程链表定义

/* Linked list of thread. */
typedef struct _thread_list {
    thread_t *head;
    thread_t *tail;
    int count;
} thread_list_t;

线程类型的定义如下：

/* Thread types. */
#define THREAD_READ        0             //读线程
#define THREAD_WRITE        1            //写线程
#define THREAD_TIMER        2            //计时器线程
#define THREAD_EVENT        3            //事件线程
#define THREAD_CHILD        4            //子线程
#define THREAD_READY        5            //就绪线程
#define THREAD_UNUSED        6           //未使用线程
#define THREAD_WRITE_TIMEOUT    7        //写超时线程
#define THREAD_READ_TIMEOUT    8         //读超时线程
#define THREAD_CHILD_TIMEOUT    9        //子超时线程
#define THREAD_TERMINATE    10            //停止线程
#define THREAD_READY_FD        11        

1.3.主线程定义

/* Master of the theads. */
typedef struct _thread_master {
    thread_list_t read;
    thread_list_t write;
    thread_list_t timer;
    thread_list_t child;
    thread_list_t event;
    thread_list_t ready;
    thread_list_t unuse;
    fd_set readfd;
    fd_set writefd;
    fd_set exceptfd;
    unsigned long alloc;
} thread_master_t;

2. 线程操作

2.1 生成主线程

/* global vars */
thread_master_t *master = NULL;

/* Make thread master. */
thread_master_t *
thread_make_master(void)
{
    thread_master_t *new;

    new = (thread_master_t *) MALLOC(sizeof (thread_master_t));
    return new;
}

2.2 销毁一个主线程

    

/* Stop thread scheduler. */
void
thread_destroy_master(thread_master_t * m)
{
    thread_cleanup_master(m);
    FREE(m);
}

//调用子函数，清空主线程的内容
/* Cleanup master */
static void
thread_cleanup_master(thread_master_t * m)
{
    /* Unuse current thread lists */
    thread_destroy_list(m, m->read);
    thread_destroy_list(m, m->write);
    thread_destroy_list(m, m->timer);
    thread_destroy_list(m, m->event);
    thread_destroy_list(m, m->ready);

    /* Clear all FDs */
    FD_ZERO(&m->readfd);
    FD_ZERO(&m->writefd);
    FD_ZERO(&m->exceptfd);

    /* Clean garbage */
    thread_clean_unuse(m);
}
//回收主线程内存
FREE(m);

2.3 增加一个简单的事件线程

/* Add simple event thread. */
thread_t *
thread_add_terminate_event(thread_master_t * m)
{
    thread_t *thread;

    assert(m != NULL);

    thread = thread_new(m);
    thread->type = THREAD_TERMINATE;
    thread->id = 0;
    thread->master = m;
    thread->func = NULL;
    thread->arg = NULL;
    thread->u.val = 0;
    thread_list_add(&m->event, thread);

    return thread;
}

2.4 创建不同类型的线程，并加入主线程中的对应线程链表，如读线程为例介绍

/* Add new read thread. */
thread_t *
thread_add_read(thread_master_t * m, int (*func) (thread_t *)
        , void *arg, int fd, long timer)
{
    thread_t *thread;

    assert(m != NULL);

    if (FD_ISSET(fd, &m->readfd)) {
        log_message(LOG_WARNING, "There is already read fd [%d]", fd);
        return NULL;
    }

    thread = thread_new(m);
    thread->type = THREAD_READ;
    thread->id = 0;
    thread->master = m;
    thread->func = func;
    thread->arg = arg;
    FD_SET(fd, &m->readfd);
    thread->u.fd = fd;

    /* Compute read timeout value */
    set_time_now();
    thread->sands = timer_add_long(time_now, timer);

    /* Sort the thread. */
    thread_list_add_timeval(&m->read, thread);

    return thread;
}

2.4.1 创建一个新的线程

/* Make new thread. */
thread_t *
thread_new(thread_master_t * m)
{
    thread_t *new;

    /* If one thread is already allocated return it */
    if (m->unuse.head) {
        new = thread_trim_head(&m->unuse);
        memset(new, 0, sizeof (thread_t));
        return new;
    }

    new = (thread_t *) MALLOC(sizeof (thread_t));
    m->alloc++;
    return new;
}

2.4.2 设置为读线程

    thread->type = THREAD_READ;
    thread->id = 0;
    thread->master = m;
    thread->func = func;
    thread->arg = arg;
    FD_SET(fd, &m->readfd);
    thread->u.fd = fd;

2.4.3 根据超时时间将读进程加入读进程列表中

/* Add a thread in the list sorted by timeval */
void
thread_list_add_timeval(thread_list_t * list, thread_t * thread)
{
    thread_t *tt;

    for (tt = list->head; tt; tt = tt->next) {
        if (timer_cmp(thread->sands, tt->sands) <= 0)
            break;
    }

    if (tt)
        thread_list_add_before(list, tt, thread);
    else
        thread_list_add(list, thread);
}

2.5 取消线程，从对应类型的线程列表中去除该线程，将它设置为unused类型，并加入unused线程链表。

/* Cancel thread from scheduler. */
void
thread_cancel(thread_t * thread)
{
    switch (thread->type) {
    case THREAD_READ:
        assert(FD_ISSET(thread->u.fd, &thread->master->readfd));
        FD_CLR(thread->u.fd, &thread->master->readfd);
        thread_list_delete(&thread->master->read, thread);
        break;
    case THREAD_WRITE:
        assert(FD_ISSET(thread->u.fd, &thread->master->writefd));
        FD_CLR(thread->u.fd, &thread->master->writefd);
        thread_list_delete(&thread->master->write, thread);
        break;
    case THREAD_TIMER:
        thread_list_delete(&thread->master->timer, thread);
        break;
    case THREAD_CHILD:
        /* Does this need to kill the child, or is that the
         * caller's job?
         * This function is currently unused, so leave it for now.
         */
        thread_list_delete(&thread->master->child, thread);
        break;
    case THREAD_EVENT:
        thread_list_delete(&thread->master->event, thread);
        break;
    case THREAD_READY:
    case THREAD_READY_FD:
        thread_list_delete(&thread->master->ready, thread);
        break;
    default:
        break;
    }

    thread->type = THREAD_UNUSED;
    thread_add_unuse(thread->master, thread);
}

2.6 获取下一个就绪进程

/* Fetch next ready thread. */
thread_t *
thread_fetch(thread_master_t * m, thread_t * fetch)
{
    int ret, old_errno;
    thread_t *thread;
    fd_set readfd;
    fd_set writefd;
    fd_set exceptfd;
    timeval_t timer_wait;
    int signal_fd;
#ifdef _WITH_SNMP_
    timeval_t snmp_timer_wait;
    int snmpblock = 0;
    int fdsetsize;
#endif

    assert(m != NULL);

    /* Timer initialization */
    memset(&timer_wait, 0, sizeof (timeval_t));

retry:    /* When thread can't fetch try to find next thread again. */

    /* If there is event process it first. */
    while ((thread = thread_trim_head(&m->event))) {
        *fetch = *thread;

        /* If daemon hanging event is received return NULL pointer */
        if (thread->type == THREAD_TERMINATE) {
            thread->type = THREAD_UNUSED;
            thread_add_unuse(m, thread);
            return NULL;
        }
        thread->type = THREAD_UNUSED;
        thread_add_unuse(m, thread);
        return fetch;
    }

    /* If there is ready threads process them */
    while ((thread = thread_trim_head(&m->ready))) {
        *fetch = *thread;
        thread->type = THREAD_UNUSED;
        thread_add_unuse(m, thread);
        return fetch;
    }

    /*
     * Re-read the current time to get the maximum accuracy.
     * Calculate select wait timer. Take care of timeouted fd.
     */
    set_time_now();
    thread_compute_timer(m, &timer_wait);

    /* Call select function. */
    readfd = m->readfd;
    writefd = m->writefd;
    exceptfd = m->exceptfd;

    signal_fd = signal_rfd();
    FD_SET(signal_fd, &readfd);

#ifdef _WITH_SNMP_
    /* When SNMP is enabled, we may have to select() on additional
     * FD. snmp_select_info() will add them to `readfd'. The trick
     * with this function is its last argument. We need to set it
     * to 0 and we need to use the provided new timer only if it
     * is still set to 0. */
    fdsetsize = FD_SETSIZE;
    snmpblock = 0;
    memcpy(&snmp_timer_wait, &timer_wait, sizeof(timeval_t));
    snmp_select_info(&fdsetsize, &readfd, &snmp_timer_wait, &snmpblock);
    if (snmpblock == 0)
        memcpy(&timer_wait, &snmp_timer_wait, sizeof(timeval_t));
#endif

    ret = select(FD_SETSIZE, &readfd, &writefd, &exceptfd, &timer_wait);

    /* we have to save errno here because the next syscalls will set it */
    old_errno = errno;

       /* Handle SNMP stuff */
#ifdef _WITH_SNMP_
    if (ret > 0)
        snmp_read(&readfd);
    else if (ret == 0)
        snmp_timeout();
#endif

    /* handle signals synchronously, including child reaping */
    if (FD_ISSET(signal_fd, &readfd))
        signal_run_callback();

    /* Update current time */
    set_time_now();

    if (ret < 0) {
        if (old_errno == EINTR)
            goto retry;
        /* Real error. */
        DBG("select error: %s", strerror(old_errno));
        assert(0);
    }

    /* Timeout children */
    thread = m->child.head;
    while (thread) {
        thread_t *t;

        t = thread;
        thread = t->next;

        if (timer_cmp(time_now, t->sands) >= 0) {
            thread_list_delete(&m->child, t);
            thread_list_add(&m->ready, t);
            t->type = THREAD_CHILD_TIMEOUT;
        }
    }

    /* Read thead. */
    thread = m->read.head;
    while (thread) {
        thread_t *t;

        t = thread;
        thread = t->next;

        if (FD_ISSET(t->u.fd, &readfd)) {
            assert(FD_ISSET(t->u.fd, &m->readfd));
            FD_CLR(t->u.fd, &m->readfd);
            thread_list_delete(&m->read, t);
            thread_list_add(&m->ready, t);
            t->type = THREAD_READY_FD;
        } else {
            if (timer_cmp(time_now, t->sands) >= 0) {
                FD_CLR(t->u.fd, &m->readfd);
                thread_list_delete(&m->read, t);
                thread_list_add(&m->ready, t);
                t->type = THREAD_READ_TIMEOUT;
            }
        }
    }

    /* Write thead. */
    thread = m->write.head;
    while (thread) {
        thread_t *t;

        t = thread;
        thread = t->next;

        if (FD_ISSET(t->u.fd, &writefd)) {
            assert(FD_ISSET(t->u.fd, &writefd));
            FD_CLR(t->u.fd, &m->writefd);
            thread_list_delete(&m->write, t);
            thread_list_add(&m->ready, t);
            t->type = THREAD_READY_FD;
        } else {
            if (timer_cmp(time_now, t->sands) >= 0) {
                FD_CLR(t->u.fd, &m->writefd);
                thread_list_delete(&m->write, t);
                thread_list_add(&m->ready, t);
                t->type = THREAD_WRITE_TIMEOUT;
            }
        }
    }
    /* Exception thead. */
    /*... */

    /* Timer update. */
    thread = m->timer.head;
    while (thread) {
        thread_t *t;

        t = thread;
        thread = t->next;

        if (timer_cmp(time_now, t->sands) >= 0) {
            thread_list_delete(&m->timer, t);
            thread_list_add(&m->ready, t);
            t->type = THREAD_READY;
        }
    }

    /* Return one event. */
    thread = thread_trim_head(&m->ready);

#ifdef _WITH_SNMP_
    run_alarms();
    netsnmp_check_outstanding_agent_requests();
#endif

    /* There is no ready thread. */
    if (!thread)
        goto retry;

    *fetch = *thread;
    thread->type = THREAD_UNUSED;
    thread_add_unuse(m, thread);

    return fetch;
}

2.7 子线程处理，便利子线程链表取出子线程，并放入就绪线程链表。

/* Synchronous signal handler to reap child processes */
void
thread_child_handler(void * v, int sig)
{
    thread_master_t * m = v;

    /*
     * This is O(n^2), but there will only be a few entries on
     * this list.
     */
    thread_t *thread;
    pid_t pid;
    int status = 77;
    while ((pid = waitpid(-1, &status, WNOHANG))) {
        if (pid == -1) {
            if (errno == ECHILD)
                return;
            DBG("waitpid error: %s", strerror(errno));
            assert(0);
        } else {
            thread = m->child.head;
            while (thread) {
                thread_t *t;
                t = thread;
                thread = t->next;
                if (pid == t->u.c.pid) {
                    thread_list_delete(&m->child, t);
                    thread_list_add(&m->ready, t);
                    t->u.c.status = status;
                    t->type = THREAD_READY;
                    break;
                }
            }
        }
    }
}

2.8 线程调用

/* Call thread ! */
void
thread_call(thread_t * thread)
{
    thread->id = thread_get_id();
    (*thread->func) (thread);
}

2.9 启动调度器

/* Our infinite scheduling loop */
void
launch_scheduler(void)
{
    thread_t thread;

    signal_set(SIGCHLD, thread_child_handler, master);

    /*
     * Processing the master thread queues,
     * return and execute one ready thread.
     */
    while (thread_fetch(master, &thread)) {
        /* Run until error, used for debuging only */
#ifdef _DEBUG_
        if ((debug & 520) == 520) {
            debug &= ~520;
            thread_add_terminate_event(master);
        }
#endif
        thread_call(&thread);
    }
}