ClamAV 0.95线程池分析

ClamAV的强大之处不在于其他的，而是ClamAV提供了一个很好的线程池架构。线程池用队列或链表将工作保存在工作队列，工作用需要处理的任务的特征数据代表。线程池使用预创建技术，它创建一定数量的空闲线程，这些线程处于阻塞状态。当有工作任务时，缓冲池选择一个空闲线程，把任务附在这个线程运行，即线程执行工作处理函数，任务完成后成为空闲线程，线程不退出，而是回到线程池。线程池维持着一定数量的线程。线程池技术减少了线程创建和销毁的次数，但也占用了少量资源。因此，线程池技术适合于单位时间内处理任务多且线程创建销毁频繁的工作。在这里主要是分析ClamAV
0.95中的thrmgr.c和thrmgr.h这两个文件。

ClamAV服务器使用了一个效果较好的线程池，这个线程池只使用了处理函数scanner_thread，即每个线程都调用处理函数scanner_thread。线程池以多个同样的线程分别独立进行病毒的扫描。线程池一般使用了线程池结构、工作队列结构和工作队列成员结构来描述，每个工作以某一特征数据代表，将工作放入工作队列中，clamd服务器的线程池结构、工作队列结构和工作队列成员结构在源码中的thrmgr.h中，主要如下：

工作队列结构：

//工作队列中项的结构
typedef struct work_item_tag {
struct work_item_tag *next;
void *data;
struct timeval time_queued;
} work_item_t;
//工作队列结构
typedef struct work_queue_tag {
work_item_t *head;
work_item_t *tail;
int item_count;	//总的工作项个数
int popped;		//已经取出的工作项个数
} work_queue_t;

工作队列链表的管理函数有work_queue_new、work_queue_add和work_queue_pop，分别用来创建工作队列链表、向链表增加成员、从链表中取出成员。

work_queue_new函数如下：

static work_queue_t *work_queue_new(void)
{
work_queue_t *work_q;
//给工作队列分配内存
work_q = (work_queue_t *) malloc(sizeof(work_queue_t));
if (!work_q) {
return NULL;
}

work_q->head = work_q->tail = NULL;
work_q->item_count = 0;
work_q->popped = 0;
return work_q;
}

work_queue_add函数如下：

static int work_queue_add(work_queue_t *work_q, void *data)
{
work_item_t *work_item;

if (!work_q) {
return FALSE;
}
work_item = (work_item_t *) malloc(sizeof(work_item_t));
if (!work_item) {
return FALSE;
}

work_item->next = NULL;
work_item->data = data;
gettimeofday(&(work_item->time_queued), NULL);	//获取当前精确计时

if (work_q->head == NULL) {
work_q->head = work_q->tail = work_item;
work_q->item_count = 1;
} else {
work_q->tail->next = work_item;
work_q->tail = work_item;
work_q->item_count++;
}
return TRUE;
}

work_queue_pop函数如下：

//从工作队列中取出来一个data
static void *work_queue_pop(work_queue_t *work_q)
{
work_item_t *work_item;
void *data;

if (!work_q || !work_q->head) {
return NULL;
}
work_item = work_q->head;
data = work_item->data;
work_q->head = work_item->next;
if (work_q->head == NULL) {
work_q->tail = NULL;
}
free(work_item);
work_q->item_count--;
return data;
}

附加timeval结构如下：

struct timeval {
long inttv_sec;		// 秒数
long inttv_usec; 	// 微秒数
}

至此工作队列的结构完成了，工作队列的操作函数比较简单，主要是着重分析ClamAV的线程结构。

线程池结构分析：

说到线程池，不得不提线程池的状态。在clamav中主要使用下面这个枚举类型来表示，如下：

//线程池的状态，主要包括无效、有效、退出
typedef enum {
POOL_INVALID,
POOL_VALID,
POOL_EXIT
} pool_state_t;

在ClamAV 0.95中使用下面这个结构来表示线程管理程序的退出状态，如下：

//线程管理退出的状态，分别是正常退出、错误退出、其他原因退出
enum thrmgr_exit {
EXIT_OK,
EXIT_ERROR,
EXIT_OTHER
};

现在开始拿出线程池的结构了，线程池的结构主要如下：

typedef struct threadpool_tag {
pthread_mutex_t pool_mutex;	//线程池的互斥量
pthread_cond_t pool_cond;	//线程池的条件变量
pthread_attr_t pool_attr;	//线程池的属性

pthread_cond_t  idle_cond;	//空闲线程的条件变量
pthread_cond_t  queueable_cond;	//可用队列的条件变量

pool_state_t state;	//线程池的状态，主要有无效、有效、退出
int thr_max;		//最大线程数量
int queue_max;		//最大工作队列数量
int thr_alive;		//激活线程的数量
int thr_idle;		//空闲线程的数量
int idle_timeout;	//空闲时间
struct task_desc *tasks;	//指向工作任务的task_desc结构

void (*handler)(void *);	//线程所用的函数，参数是void类型的指针，返回值是void型指针

work_queue_t *bulk_queue;
work_queue_t *single_queue;
} threadpool_t;

任务结构task_desc如下：

struct task_desc {
const char *filename;	//扫描的文件名称
const char *command;	//扫描命令
struct timeval tv;		//扫描任务的时间
struct task_desc *prv;	//指向前一个task_desc
struct task_desc *nxt;	//指向后一个task_desc
const struct cl_engine *engine;	//病毒扫描引擎结构
};

结合前面介绍的work_queue_t、work_item_t，那么整个线程池的数据结构图如下：



在clamAV 0.95中，针对线程池使用了线程池链表。主要结构如下：

static struct threadpool_list {
threadpool_t *pool;
struct threadpool_list *nxt;
} *pools = NULL;

每次操作线程池的时候，为了避免出现竞争使用了pools_lock互斥量。对于互斥量的初始化有两种，一种是使用函数pthread_mutex_init进行动态初始化，另外一种是使用静态初始化，即下面这种形式：

static pthread_mutex_t pools_lock = PTHREAD_MUTEX_INITIALIZER;

在分析线程池链表时，注意新建立的pools是指向前一个的，程序主要如下：

static void add_topools(threadpool_t *t)
{
struct threadpool_list *new = malloc(sizeof(*new));
if(!new) {
logg("!Unable to add threadpool to list\n");
return;
}
new->pool = t;
pthread_mutex_lock(&pools_lock);
new->nxt = pools;
pools = new;
pthread_mutex_unlock(&pools_lock);
}

下面通过图来展示：



结合上面两个图，对于函数remove_frompools的理解就更容易了。

static void remove_frompools(threadpool_t *t)
{
struct threadpool_list *l, *prev;
struct task_desc *desc;
pthread_mutex_lock(&pools_lock);
prev = NULL;
l = pools;
while(l && l->pool != t) {
prev = l;
l = l->nxt;
}
if(!l)
return;
if(prev)
prev->nxt = l->nxt;
if(l == pools)
pools = l->nxt;
free(l);
desc = t->tasks;
while(desc) {
struct task_desc *q = desc;
desc = desc->nxt;
free(q);
}
t->tasks = NULL;
pthread_mutex_unlock(&pools_lock);
}

分析完clamAV 0.95中的数据结构之后，就进入开始分析线程池里面使用的函数。创建线程池函数：