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);elsethread_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; #endifassert(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)); #endifret = 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);} #endifthread_call(&thread);} }

?

?

?

?

?

?

?

轉載于:https://www.cnblogs.com/davidwang456/p/3543937.html

《新程序員》：云原生和全面數字化實踐50位技術專家共同創作，文字、視頻、音頻交互閱讀

總結

以上是生活随笔為你收集整理的深入学习keepalived之预备工作--线程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。