Implementation of asynchronous request pool

Synchronous asynchronous and blocking non blocking

First, make clear the difference between synchronous asynchronous and blocking non blocking . Synchronous asynchronism is the relationship between the two , Is the mechanism . Synchronization is a request and a return , Equivalent to serial operation ; Asynchrony means that you don't have to wait for the result to return after the request , Continue to do other operations , Just send the request , There's no need to wait for the results , It is equivalent to working in parallel . Blocking nonblocking yes fd The state of , It's characteristics . yes fd Whether to return immediately when the data is not ready . Synchronous asynchrony is not necessarily related to blocking and non blocking .

The practice of asynchronous request pool

1. Asynchronous requests use multiple connections （ That is many fd）.
2. Based on the Internet IO Send the corresponding protocol .
3. Multiple connections ： When the request is sent , Before returning results , This fd Can't release , Need to be stored in epoll in , Listen to read events , from epoll Conduct management .
4. Non blocking ： because fd Set non-blocking ,send() and recv(), Request and response data , Can't be in the same thread , Because these asynchronous interfaces are provided for external use .

Design of asynchronous request pool

The design of asynchronous request pool has a routine , There are four steps .

1.init

Initializes the context of an asynchronous operation . There are two operations ,epoll_create() And open a thread for recv_cb. The context here refers to epfd And thread id.

2.commit

Set up the connection first , Organize the corresponding agreements , Again send() Send to the corresponding server , Then the fd Join in epoll, Listen for readable status .

3. Thread entry function callback()

epoll_wait(), Monitor what fd Can be read ,recv(fd), Read all the data and parse it .

4.destroy() Close and recycle resources

close(epfd),pthread_cancel()
Let's say DNS Request as an example

static void* dns_async_client_proc(void *arg) 
{
    
	struct async_context *ctx = (struct async_context*)arg;
	int epfd = ctx->epfd;
	while (1) 
{
    
		struct epoll_event events[ASYNC_CLIENT_NUM] = {
    0};
		int nready = epoll_wait(epfd, events, ASYNC_CLIENT_NUM, -1);
		if (nready < 0) 
        {
    
			if (errno == EINTR || errno == EAGAIN) 
            {
    
				continue;
			} else 
            {
    
				break;
			}
		} else if (nready == 0) 
        {
    
			continue;
		}
		printf("nready:%d\n", nready);
		int i = 0;
		for (i = 0;i < nready;i ++) 
        {
    

			struct ep_arg *data = (struct ep_arg*)events[i].data.ptr;
			int sockfd = data->sockfd;
			char buffer[1024] = {
    0};
			struct sockaddr_in addr;
			size_t addr_len = sizeof(struct sockaddr_in);
			int n = recvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr*)&addr, (socklen_t*)&addr_len);
			struct dns_item *domain_list = NULL;
			int count = dns_parse_response(buffer, &domain_list);
			data->cb(domain_list, count); //call cb
			int ret = epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
			//printf("epoll_ctl DEL --> sockfd:%d\n", sockfd);
			close(sockfd); 
			dns_async_client_free_domains(domain_list, count);
			free(data);
		}
	}
}

//dns_async_client_init()
//epoll init
//thread init
struct async_context *dns_async_client_init(void) 
{
    
	int epfd = epoll_create(1); // 
	if (epfd < 0) return NULL;
	struct async_context *ctx = calloc(1, sizeof(struct async_context));
	if (ctx == NULL) 
    {
    
		close(epfd);
		return NULL;
	}
	ctx->epfd = epfd;
	pthread_t thread_id;
	int ret = pthread_create(&thread_id, NULL, dns_async_client_proc, ctx);
	if (ret) 
    {
    
		perror("pthread_create");
		return NULL;
	}
	usleep(1); //child go first
	return ctx;
}

//dns_async_client_commit(ctx, domain)
//socket init
//dns_request
//sendto dns send
int dns_async_client_commit(struct async_context* ctx, const char *domain, async_result_cb cb) 
{
    
	int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
	if (sockfd < 0) 
    {
    
		perror("create socket failed\n");
		exit(-1);
	}
	printf("url:%s\n", domain);
	set_block(sockfd, 0); //nonblock
	struct sockaddr_in dest;
	bzero(&dest, sizeof(dest));
	dest.sin_family = AF_INET;
	dest.sin_port = htons(53);
	dest.sin_addr.s_addr = inet_addr(DNS_SVR);
	int ret = connect(sockfd, (struct sockaddr*)&dest, sizeof(dest));
	//printf("connect :%d\n", ret);
	struct dns_header header = {
    0};
	dns_create_header(&header);
	struct dns_question question = {
    0};
	dns_create_question(&question, domain);
	char request[1024] = {
    0};
	int req_len = dns_build_request(&header, &question, request);
	int slen = sendto(sockfd, request, req_len, 0, (struct sockaddr*)&dest, sizeof(struct sockaddr));
	struct ep_arg *eparg = (struct ep_arg*)calloc(1, sizeof(struct ep_arg));
	if (eparg == NULL) return -1;
	eparg->sockfd = sockfd;
	eparg->cb = cb;

	struct epoll_event ev;
	ev.data.ptr = eparg;
	ev.events = EPOLLIN;
	ret = epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev); 
	//printf(" epoll_ctl ADD: sockfd->%d, ret:%d\n", sockfd, ret);
	return ret;	
}

static void dns_async_client_result_callback(struct dns_item *list, int count) 
{
    
	int i = 0;
	for (i = 0;i < count;i ++) 
    {
    
		printf("name:%s, ip:%s\n", list[i].domain, list[i].ip);
	}
}

Here are some tips . One is to prevent the main thread from exiting , Resulting in incomplete data reception , Can be in main Function to add getchar() The function is blocking . The second is to minimize the use of global variables , Because global variables are difficult to maintain , Poor readability . Third ,epoll No TCP The exclusive product of , here DNS Although it is UDP, But it can also be used epoll, because epoll Is based on fd It's done , It's not just web-based IO Ability to use epoll, therefore UDP、eventfd、timerfd wait , Both can be used. epoll. Fourth , After receiving data , It can be turned off fd, also EPOLL_CTL_DEL, It can also be reused fd,EPOLL_CTL_MOD Listen to writable events , Prepare for the next data transmission , If fd It's no use for a long time , To do timeout management , Otherwise, the server may actively disconnect , There are generally two ways , One is to send heartbeat packets regularly , The second is that it doesn't work , Next time send() Turn off fd, Reassign another fd. Both of these approaches cost very little , All work . Last , There is a problem in the code , If the server has no data to return , that epoll The corresponding... Will not be triggered fd Reading events ,fd Won't close(), So you need to join fd Timer （timerfd）, If you time out, you will resend .