libcurl官方libevent驱动例子的研究

介绍

最近需要做一个高并发Http请求的功能，在此基础上，可以进一步做服务端高并发的压力测试。之前做了个Http的同步请求功能，使用的是libcurl，所以这次还是使用libcur+libevent的方式来实现高并发Http请求的功能。

例子代码

正好libcurl官方例子里面有个libcurl+libevent的例子代码，直接拿过来看下流程，看不懂的地方结合下官方文档的资料，理解起来还是比较容易的

代码执行流程图

详解

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <sys/poll.h>
#include <curl/curl.h>
#include <event2/event.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>


#define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */


/* 定义了三个结构体：
    1. GlobalInfo：全局结构体，记录一些全局信息
    2. ConnInfo：一个请求绑定一个该结构体
    3. SockInfo：这个结构体绑定在CURL结构体上，最有用的就是SockInfo.ev这个libevent指针，其他没什么用，在我封装的C++项目中，弃用了这个结构体，直接在CURL结构体上绑定了一个struct event*指针
*/
/* Global information, common to all connections */
typedef struct _GlobalInfo
{
	struct event_base *evbase;
	struct event *fifo_event;       // 监听管道的event对象，该例子是通过管道的方式输入请求
	struct event *timer_event;      // 时间event，这个很重要，来驱动所有请求流程
	CURLM *multi;                   // multi curl对象
	int still_running;              // 存储当前还有几个请求在运行
	FILE *input;                    // 监听的管道文件
} GlobalInfo;


/* Information associated with a specific easy handle */
typedef struct _ConnInfo
{
	CURL *easy;
	char *url;                      // 请求地址，使用了strdup生成，需要释放内存;
	GlobalInfo *global;
	char error[CURL_ERROR_SIZE];    // 存放错误信息;
} ConnInfo;


/* Information associated with a specific socket */
typedef struct _SockInfo
{
	curl_socket_t sockfd;
	CURL *easy;
	int action;
	long timeout;
	struct event *ev;
	int evset;
	GlobalInfo *global;
} SockInfo;

/* Update the event timer after curl_multi library calls */
static int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo *g)
{
	struct timeval timeout;
	(void)multi; /* unused */

	timeout.tv_sec = timeout_ms / 1000;
	timeout.tv_usec = (timeout_ms % 1000) * 1000;
	fprintf(MSG_OUT, "multi_timer_cb: Setting timeout to %ld ms\n", timeout_ms);

	/* TODO
	*
	* if timeout_ms is 0, call curl_multi_socket_action() at once!
	*
	* if timeout_ms is -1, just delete the timer
	*
	* for all other values of timeout_ms, this should set or *update*
	* the timer to the new value
	*/
	// 接收到libcurl的设置定时器回调，利用libevent设置定时器，这里需要注意两种情况，例子代码中设置成了TODO
	evtimer_add(g->timer_event, &timeout);
	return 0;
}

/* Die if we get a bad CURLMcode somewhere */
static void mcode_or_die(const char *where, CURLMcode code)
{
	if (CURLM_OK != code) {
		const char *s;
		switch (code) {
		case     CURLM_BAD_HANDLE:         s = "CURLM_BAD_HANDLE";         break;
		case     CURLM_BAD_EASY_HANDLE:    s = "CURLM_BAD_EASY_HANDLE";    break;
		case     CURLM_OUT_OF_MEMORY:      s = "CURLM_OUT_OF_MEMORY";      break;
		case     CURLM_INTERNAL_ERROR:     s = "CURLM_INTERNAL_ERROR";     break;
		case     CURLM_UNKNOWN_OPTION:     s = "CURLM_UNKNOWN_OPTION";     break;
		case     CURLM_LAST:               s = "CURLM_LAST";               break;
		default: s = "CURLM_unknown";
			break;
		case     CURLM_BAD_SOCKET:         s = "CURLM_BAD_SOCKET";
			fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s);
			/* ignore this error */
			return;
		}
		fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s);
		exit(code);
	}
}



/* Check for completed transfers, and remove their easy handles */
static void check_multi_info(GlobalInfo *g)
{
	char *eff_url;
	CURLMsg *msg;
	int msgs_left;
	ConnInfo *conn;
	CURL *easy;
	CURLcode res;

	fprintf(MSG_OUT, "REMAINING: %d\n", g->still_running);
	while ((msg = curl_multi_info_read(g->multi, &msgs_left))) {
		if (msg->msg == CURLMSG_DONE) {
		    // 该easy curl已经执行完成，进行清理操作;
			easy = msg->easy_handle;
			res = msg->data.result;
			curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
			curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url);
			fprintf(MSG_OUT, "DONE: %s => (%d) %s\n", eff_url, res, conn->error);
			curl_multi_remove_handle(g->multi, easy);
			free(conn->url);
			curl_easy_cleanup(easy);
			free(conn);
		}
	}
}



/* Called by libevent when we get action on a multi socket */
static void event_cb(int fd, short kind, void *userp)
{
	GlobalInfo *g = (GlobalInfo*)userp;
	CURLMcode rc;

    // libevent事件转libcurl事件
	int action =
		(kind & EV_READ ? CURL_CSELECT_IN : 0) |
		(kind & EV_WRITE ? CURL_CSELECT_OUT : 0);

    // 驱动fd的状态机继续执行;
	rc = curl_multi_socket_action(g->multi, fd, action, &g->still_running);
	mcode_or_die("event_cb: curl_multi_socket_action", rc);

    // 查看libcurl的执行情况，不能根据still_running数量少一个来判断当前的fd已经执行完成，一定要调用curl_multi_info_read来判断;
	check_multi_info(g);
	if (g->still_running <= 0) {
		fprintf(MSG_OUT, "last transfer done, kill timeout\n");
		if (evtimer_pending(g->timer_event, NULL)) {
			evtimer_del(g->timer_event);
		}
	}
}



/* Called by libevent when our timeout expires */
static void timer_cb(int fd, short kind, void *userp)
{
	GlobalInfo *g = (GlobalInfo *)userp;
	CURLMcode rc;
	(void)fd;
	(void)kind;

    // 定时器超时后，重新驱动libcurl，在curl获取到hostname对应的ip地址后，驱动该函数会回调到sock_cb中;
	rc = curl_multi_socket_action(g->multi,
		CURL_SOCKET_TIMEOUT, 0, &g->still_running);
	mcode_or_die("timer_cb: curl_multi_socket_action", rc);
	
	// 查看libcurl的执行情况;
	check_multi_info(g);
}



/* Clean up the SockInfo structure */
static void remsock(SockInfo *f)
{
	if (f) {
		if (f->evset)
			event_free(f->ev);
		free(f);
	}
}



/* Assign information to a SockInfo structure */
static void setsock(SockInfo *f, curl_socket_t s, CURL *e, int act,
	GlobalInfo *g)
{
    // libcurl事件转libevent事件
	int kind =
		(act&CURL_POLL_IN ? EV_READ : 0) | (act&CURL_POLL_OUT ? EV_WRITE : 0) | EV_PERSIST;

	f->sockfd = s;
	f->action = act;
	f->easy = e;
	if (f->evset)
		event_free(f->ev);
	// 创建关联的event对象，并加入到libevent循环中去监听;
	f->ev = event_new(g->evbase, f->sockfd, kind, event_cb, g);
	f->evset = 1;
	event_add(f->ev, NULL);
}



/* Initialize a new SockInfo structure */
static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo *g)
{
	SockInfo *fdp = calloc(sizeof(SockInfo), 1);

	fdp->global = g;
	setsock(fdp, s, easy, action, g);
	curl_multi_assign(g->multi, s, fdp);        // 设置socket fd关联的指针
}

/* CURLMOPT_SOCKETFUNCTION */
static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp)
{
	GlobalInfo *g = (GlobalInfo*)cbp;
	SockInfo *fdp = (SockInfo*)sockp;
	const char *whatstr[] = { "none", "IN", "OUT", "INOUT", "REMOVE" };

	fprintf(MSG_OUT,
		"socket callback: s=%d e=%p what=%s ", s, e, whatstr[what]);
	if (what == CURL_POLL_REMOVE) {
		fprintf(MSG_OUT, "\n");
		remsock(fdp);
	}
	else {
		if (!fdp) {
			fprintf(MSG_OUT, "Adding data: %s\n", whatstr[what]);
			addsock(s, e, what, g);
		}
		else {
			fprintf(MSG_OUT,
				"Changing action from %s to %s\n",
				whatstr[fdp->action], whatstr[what]);
			setsock(fdp, s, e, what, g);
		}
	}
	return 0;
}



/* CURLOPT_WRITEFUNCTION */
static size_t write_cb(void *ptr, size_t size, size_t nmemb, void *data)
{
	size_t realsize = size * nmemb;
	ConnInfo *conn = (ConnInfo*)data;
	(void)ptr;
	(void)conn;
	return realsize;
}


/* CURLOPT_PROGRESSFUNCTION */
static int prog_cb(void *p, double dltotal, double dlnow, double ult,
	double uln)
{
	ConnInfo *conn = (ConnInfo *)p;
	(void)ult;
	(void)uln;

	fprintf(MSG_OUT, "Progress: %s (%g/%g)\n", conn->url, dlnow, dltotal);
	return 0;
}


/* Create a new easy handle, and add it to the global curl_multi */
static void new_conn(char *url, GlobalInfo *g)
{
	ConnInfo *conn;
	CURLMcode rc;

	conn = calloc(1, sizeof(ConnInfo));
	memset(conn, 0, sizeof(ConnInfo));
	conn->error[0] = '\0';

	conn->easy = curl_easy_init();
	if (!conn->easy) {
		fprintf(MSG_OUT, "curl_easy_init() failed, exiting!\n");
		exit(2);
	}
	conn->global = g;
	conn->url = strdup(url);
	curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url);
	curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb);
	curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, conn);
	curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L);
	curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error);
	curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn);
	curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 0L);
	curl_easy_setopt(conn->easy, CURLOPT_PROGRESSFUNCTION, prog_cb);
	curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn);
	fprintf(MSG_OUT,
		"Adding easy %p to multi %p (%s)\n", conn->easy, g->multi, url);
		
	// 把easy curl加入到muti handle中，加入之后会立即回调一次multi_timer_cb
	rc = curl_multi_add_handle(g->multi, conn->easy);
	mcode_or_die("new_conn: curl_multi_add_handle", rc);

	/* note that the add_handle() will set a time-out to trigger very soon so
	that the necessary socket_action() call will be called by this app */
}

/* This gets called whenever data is received from the fifo */
static void fifo_cb(int fd, short event, void *arg)
{
    // 这里接收到了发起请求的信息，进行请求流程;
	char s[1024];
	long int rv = 0;
	int n = 0;
	GlobalInfo *g = (GlobalInfo *)arg;
	(void)fd; /* unused */
	(void)event; /* unused */

	do {
		s[0] = '\0';
		rv = fscanf(g->input, "%1023s%n", s, &n);
		s[n] = '\0';
		if (n && s[0]) {
			new_conn(s, arg);  /* if we read a URL, go get it! */
		}
		else
			break;
	} while (rv != EOF);
}

/* Create a named pipe and tell libevent to monitor it */
static const char *fifo = "hiper.fifo";
static int init_fifo(GlobalInfo *g)
{
	struct stat st;
	curl_socket_t sockfd;

	fprintf(MSG_OUT, "Creating named pipe \"%s\"\n", fifo);
	if (lstat(fifo, &st) == 0) {
		if ((st.st_mode & S_IFMT) == S_IFREG) {
			errno = EEXIST;
			perror("lstat");
			exit(1);
		}
	}
	unlink(fifo);
	if (mkfifo(fifo, 0600) == -1) {
		perror("mkfifo");
		exit(1);
	}
	sockfd = open(fifo, O_RDWR | O_NONBLOCK, 0);
	if (sockfd == -1) {
		perror("open");
		exit(1);
	}
	g->input = fdopen(sockfd, "r");

	fprintf(MSG_OUT, "Now, pipe some URL's into > %s\n", fifo);
	g->fifo_event = event_new(g->evbase, sockfd, EV_READ | EV_PERSIST, fifo_cb, g);
	event_add(g->fifo_event, NULL);
	return (0);
}

static void clean_fifo(GlobalInfo *g)
{
	event_free(g->fifo_event);
	fclose(g->input);
	unlink(fifo);
}

int main(int argc, char **argv)
{
	GlobalInfo g;
	(void)argc;
	(void)argv;

	memset(&g, 0, sizeof(GlobalInfo));
	g.evbase = event_base_new();
	
	// 初始化管道输入监听事件;
	init_fifo(&g);
	
	// 创建multi curl与事件event
	g.multi = curl_multi_init();
	g.timer_event = evtimer_new(g.evbase, timer_cb, &g);

    /* 设置multi curl的回调事件
        1. CURLMOPT_SOCKETFUNCTION: 该回调会被libcur在socket需要进行下一步操作的时候进行回调，例如：socket需要进行读操作，socekt需要进行写操作，socket需要删除
        2. CURLMOPT_TIMERFUNCTION: libcurl内部使用了多线程和状态机，需要上层设置一个超时定时器，一定时间后去驱动curl的状态机的执行
	/* setup the generic multi interface options we want */
	curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
	curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g);
	curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
	curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g);

	/* we don't call any curl_multi_socket*() function yet as we have no handles
	added! */

    // libevent事件循环;
	event_base_dispatch(g.evbase);

	/* this, of course, won't get called since only way to stop this program is
	via ctrl-C, but it is here to show how cleanup /would/ be done. */
	clean_fifo(&g);
	event_free(g.timer_event);
	event_base_free(g.evbase);
	curl_multi_cleanup(g.multi);
	return 0;
}