async.c是uwsgi提供的异步io的模块，相关的文档 uwsgi async doc虽然花了不少时间看，但是涉及的细节过多，所以也只是简单过了下整个流程,注释了一些自己的见解
greenlet之类的框架也只是粗略过了下，所以关于和协程交互的那块就还没有详细的注释

相关说明

async 把请求分为几种状态

async_waiting_fd_table：等待io处理的请求
async_proto_fd_table：io处理完成，正在解析的请求
async_runqueue：协议处理完成，处理业务逻辑的请求

请求的处理状态

UWSGI_OK 请求处理完成
UWSGI_AGAIN 请求还没处理完成，需要存在async_runqueue中等待下次循环处理

#include <uwsgi.h>
extern struct uwsgi_server uwsgi;
/*
	This is a general-purpose async loop engine (it expects a coroutine-based approach)
	You can see it as an hub holding the following structures:
	1) the runqueue, cores ready to be run are appended to this list
	2) the fd list, this is a list of monitored file descriptors, a core can wait for all the file descriptors it needs
	3) the timeout value, if set, the current core will timeout after the specified number of seconds (unless an event cancels it)
	IMPORTANT: this is not a callback-based engine !!!
*/
// https://uwsgi-docs.readthedocs.io/en/latest/Async.html
// this is called whenever a new connection is ready, but there are no cores to handle it
//主要作用是当async_queue满的时候 打个日志出来
void uwsgi_async_queue_is_full(time_t now) {
	if (now > uwsgi.async_queue_is_full && uwsgi.async_warn_if_queue_full) {
		uwsgi_log_verbose("[DANGER] async queue is full !!!\n");
		uwsgi.async_queue_is_full = now;
	}
}
//初始化async_queue
void uwsgi_async_init() {
	//event_queue_init初始化epoll之类的 并返回相关的fd
	uwsgi.async_queue = event_queue_init();
//初始化失败结束程序
	if (uwsgi.async_queue < 0) {
		exit(1);
	}
	//添加已经监听的fd到epoll 如与systemd交互的 接受请求的
	uwsgi_add_sockets_to_queue(uwsgi.async_queue, -1);
	//初始化一棵红黑树用于定时器 主要检测超时之类的
	uwsgi.rb_async_timeouts = uwsgi_init_rb_timer();
	// optimization, this array maps file descriptor to requests
		// 初始化请求等待队列的大小
        uwsgi.async_waiting_fd_table = uwsgi_calloc(sizeof(struct wsgi_request *) * uwsgi.max_fd);
        // 初始化解析协议的队列
        uwsgi.async_proto_fd_table = uwsgi_calloc(sizeof(struct wsgi_request *) * uwsgi.max_fd);
}
//找出对应解析协议队列中对应fd的请求
struct wsgi_request *find_wsgi_req_proto_by_fd(int fd) {
	return uwsgi.async_proto_fd_table[fd];
}
// 从等待队列中获取req
struct wsgi_request *find_wsgi_req_by_fd(int fd) {
	return uwsgi.async_waiting_fd_table[fd];
}
// 请求处理完毕 移除队列 释放资源
static void runqueue_remove(struct uwsgi_async_request *u_request) {
	struct uwsgi_async_request *parent = u_request->prev;
	struct uwsgi_async_request *child = u_request->next;
	if (parent) {
		parent->next = child;
	}
	if (child) {
		child->prev = parent;
	}
	if (u_request == uwsgi.async_runqueue) {
		uwsgi.async_runqueue = child;
	}
	if (u_request == uwsgi.async_runqueue_last) {
		uwsgi.async_runqueue_last = parent;
	}
	free(u_request);
}
//把请求加入run queue
static void runqueue_push(struct wsgi_request *wsgi_req) {
	// do not push the same request in the runqueue
	struct uwsgi_async_request *uar = uwsgi.async_runqueue;
	while(uar) {
		if (uar->wsgi_req == wsgi_req) return;
		uar = uar->next;
	}
	uar = uwsgi_malloc(sizeof(struct uwsgi_async_request));
	uar->prev = NULL;
	uar->next = NULL;
	uar->wsgi_req = wsgi_req;
	if (uwsgi.async_runqueue == NULL) {
		uwsgi.async_runqueue = uar;
	}
	else {
		uar->prev = uwsgi.async_runqueue_last;	
	}
	if (uwsgi.async_runqueue_last) {
		uwsgi.async_runqueue_last->next = uar;
	}
	uwsgi.async_runqueue_last = uar;
}
//获取async_queue_unused中的req 这是一个可以反复使用的数组
struct wsgi_request *find_first_available_wsgi_req() {
	struct wsgi_request *wsgi_req;
	if (uwsgi.async_queue_unused_ptr < 0) {
		return NULL;
	}
	wsgi_req = uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr];
	uwsgi.async_queue_unused_ptr--;
	return wsgi_req;
}
//重置超时请求
void async_reset_request(struct wsgi_request *wsgi_req) {
	if (wsgi_req->async_timeout) {
		uwsgi_del_rb_timer(uwsgi.rb_async_timeouts, wsgi_req->async_timeout);
		free(wsgi_req->async_timeout);
		wsgi_req->async_timeout = NULL;
	}
	struct uwsgi_async_fd *uaf = wsgi_req->waiting_fds;
	while (uaf) {
        	event_queue_del_fd(uwsgi.async_queue, uaf->fd, uaf->event);
                uwsgi.async_waiting_fd_table[uaf->fd] = NULL;
                struct uwsgi_async_fd *current_uaf = uaf;
                uaf = current_uaf->next;
                free(current_uaf);
	}
	wsgi_req->waiting_fds = NULL;
}
//计算超时请求
static void async_expire_timeouts(uint64_t now) {
	struct wsgi_request *wsgi_req;
	struct uwsgi_rb_timer *urbt;
	for (;;) {
		urbt = uwsgi_min_rb_timer(uwsgi.rb_async_timeouts, NULL);
		if (urbt == NULL)
			return;
		if (urbt->value <= now) {
			wsgi_req = (struct wsgi_request *) urbt->data;
			// timeout expired
			wsgi_req->async_timed_out = 1;
			// reset the request
			async_reset_request(wsgi_req);
			// push it in the runqueue
			runqueue_push(wsgi_req);
			continue;
		}
		break;
	}
}
int async_add_fd_read(struct wsgi_request *wsgi_req, int fd, int timeout) {
	if (uwsgi.async < 1 || !uwsgi.async_waiting_fd_table){ 
		uwsgi_log_verbose("ASYNC call without async mode !!!\n");
		return -1;
	}
	struct uwsgi_async_fd *last_uad = NULL, *uad = wsgi_req->waiting_fds;
	if (fd < 0)
		return -1;
	// find last slot
	while (uad) {
		last_uad = uad;
		uad = uad->next;
	}
	uad = uwsgi_malloc(sizeof(struct uwsgi_async_fd));
	uad->fd = fd;
	uad->event = event_queue_read();
	uad->prev = last_uad;
	uad->next = NULL;
	if (last_uad) {
		last_uad->next = uad;
	}
	else {
		wsgi_req->waiting_fds = uad;
	}
	if (timeout > 0) {
		async_add_timeout(wsgi_req, timeout);
	}
	uwsgi.async_waiting_fd_table[fd] = wsgi_req;
	wsgi_req->async_force_again = 1;
	return event_queue_add_fd_read(uwsgi.async_queue, fd);
}
static int async_wait_fd_read(int fd, int timeout) {
	struct wsgi_request *wsgi_req = current_wsgi_req();
	wsgi_req->async_ready_fd = 0;
	if (async_add_fd_read(wsgi_req, fd, timeout)) {
		return -1;
	}
	if (uwsgi.schedule_to_main) {
		uwsgi.schedule_to_main(wsgi_req);
	}
	if (wsgi_req->async_timed_out) {
		wsgi_req->async_timed_out = 0;
		return 0;
	}
	return 1;
}
static int async_wait_fd_read2(int fd0, int fd1, int timeout, int *fd) {
        struct wsgi_request *wsgi_req = current_wsgi_req();
        wsgi_req->async_ready_fd = 0;
        if (async_add_fd_read(wsgi_req, fd0, timeout)) {
                return -1;
        }
        if (async_add_fd_read(wsgi_req, fd1, timeout)) {
		// reset already registered fd
		async_reset_request(wsgi_req);
                return -1;
        }
        if (uwsgi.schedule_to_main) {
                uwsgi.schedule_to_main(wsgi_req);
        }
        if (wsgi_req->async_timed_out) {
                wsgi_req->async_timed_out = 0;
                return 0;
        }
	if (wsgi_req->async_ready_fd) {
		*fd = wsgi_req->async_last_ready_fd;
		return 1;
	}
        return -1;
}
void async_add_timeout(struct wsgi_request *wsgi_req, int timeout) {
	if (uwsgi.async < 1 || !uwsgi.rb_async_timeouts) {
		uwsgi_log_verbose("ASYNC call without async mode !!!\n");
		return;
	}
	wsgi_req->async_ready_fd = 0;
	if (timeout > 0 && wsgi_req->async_timeout == NULL) {
		wsgi_req->async_timeout = uwsgi_add_rb_timer(uwsgi.rb_async_timeouts, uwsgi_now() + timeout, wsgi_req);
	}
}
int async_add_fd_write(struct wsgi_request *wsgi_req, int fd, int timeout) {
	if (uwsgi.async < 1 || !uwsgi.async_waiting_fd_table) {
		uwsgi_log_verbose("ASYNC call without async mode !!!\n");
		return -1;
	}
	struct uwsgi_async_fd *last_uad = NULL, *uad = wsgi_req->waiting_fds;
	if (fd < 0)
		return -1;
	// find last slot
	while (uad) {
		last_uad = uad;
		uad = uad->next;
	}
	uad = uwsgi_malloc(sizeof(struct uwsgi_async_fd));
	uad->fd = fd;
	uad->event = event_queue_write();
	uad->prev = last_uad;
	uad->next = NULL;
	if (last_uad) {
		last_uad->next = uad;
	}
	else {
		wsgi_req->waiting_fds = uad;
	}
	if (timeout > 0) {
		async_add_timeout(wsgi_req, timeout);
	}
	uwsgi.async_waiting_fd_table[fd] = wsgi_req;
	wsgi_req->async_force_again = 1;
	return event_queue_add_fd_write(uwsgi.async_queue, fd);
}
static int async_wait_fd_write(int fd, int timeout) {
	struct wsgi_request *wsgi_req = current_wsgi_req();
	wsgi_req->async_ready_fd = 0;
	if (async_add_fd_write(wsgi_req, fd, timeout)) {
		return -1;
	}
	if (uwsgi.schedule_to_main) {
		uwsgi.schedule_to_main(wsgi_req);
	}
	if (wsgi_req->async_timed_out) {
		wsgi_req->async_timed_out = 0;
		return 0;
	}
	return 1;
}
//默认处理请求的方法 没有配合协程
void async_schedule_to_req(void) {
#ifdef UWSGI_ROUTING
        if (uwsgi_apply_routes(uwsgi.wsgi_req) == UWSGI_ROUTE_BREAK) {
		goto end;
        }
	// a trick to avoid calling routes again
	uwsgi.wsgi_req->is_routing = 1;
#endif
	// 估计这里就是调用脚本的地方
	uwsgi.wsgi_req->async_status = uwsgi.p[uwsgi.wsgi_req->uh->modifier1]->request(uwsgi.wsgi_req);
        if (uwsgi.wsgi_req->async_status <= UWSGI_OK) goto end;
        // uwsgi初始化设置的 比如greenlet_schedule_to_main
	if (uwsgi.schedule_to_main) {
        	uwsgi.schedule_to_main(uwsgi.wsgi_req);
	}
	return;
//返还资源
end:
	async_reset_request(uwsgi.wsgi_req);
	uwsgi_close_request(uwsgi.wsgi_req);
	uwsgi.wsgi_req->async_status = UWSGI_OK;	
	uwsgi.async_queue_unused_ptr++;
        uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req;
}
//greenlet 处理req的方法 这个已经交给协程调度
void async_schedule_to_req_green(void) {
	struct wsgi_request *wsgi_req = uwsgi.wsgi_req;
#ifdef UWSGI_ROUTING
        if (uwsgi_apply_routes(wsgi_req) == UWSGI_ROUTE_BREAK) {
                goto end;
        }
#endif
        //也就是说起码有一个请求处理完成 或者报错才会从循环跳出来
        for(;;) {
		wsgi_req->async_status = uwsgi.p[wsgi_req->uh->modifier1]->request(wsgi_req);
                if (wsgi_req->async_status <= UWSGI_OK) {
                        break;
                }
                wsgi_req->switches++;
		if (uwsgi.schedule_fix) {
			uwsgi.schedule_fix(wsgi_req);
		}
                // switch after each yield
		if (uwsgi.schedule_to_main)
                	uwsgi.schedule_to_main(wsgi_req);
        }
#ifdef UWSGI_ROUTING
end:
#endif
	// re-set the global state
	uwsgi.wsgi_req = wsgi_req;
        async_reset_request(wsgi_req);
        uwsgi_close_request(wsgi_req);
	// re-set the global state (routing could have changed it)
	uwsgi.wsgi_req = wsgi_req;
        wsgi_req->async_status = UWSGI_OK;
	uwsgi.async_queue_unused_ptr++;
        uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = wsgi_req;
}
static int uwsgi_async_wait_milliseconds_hook(int timeout) {
	struct wsgi_request *wsgi_req = current_wsgi_req();
	timeout = timeout / 1000;
	if (!timeout) timeout = 1;
	async_add_timeout(wsgi_req, timeout);
	wsgi_req->async_force_again = 1;
	if (uwsgi.schedule_to_main) {
                uwsgi.schedule_to_main(wsgi_req);
        }
        if (wsgi_req->async_timed_out) {
                wsgi_req->async_timed_out = 0;
                return 0;
        }
	return -1;
}
void async_loop() {
	if (uwsgi.async < 1) {
		uwsgi_log("the async loop engine requires async mode (--async <n>)\n");
		exit(1);
	}
	int interesting_fd, i;
	struct uwsgi_rb_timer *min_timeout;
	int timeout;
	int is_a_new_connection;
	int proto_parser_status;
	uint64_t now;
	struct uwsgi_async_request *current_request = NULL;
	void *events = event_queue_alloc(64);
	struct uwsgi_socket *uwsgi_sock;
	uwsgi_async_init();
	uwsgi.async_runqueue = NULL;
	//设置读写事件的hook函数
	uwsgi.wait_write_hook = async_wait_fd_write;
        uwsgi.wait_read_hook = async_wait_fd_read;
        uwsgi.wait_read2_hook = async_wait_fd_read2;
	uwsgi.wait_milliseconds_hook = uwsgi_async_wait_milliseconds_hook;
	//内部进程交互的socket
	if (uwsgi.signal_socket > -1) {
		event_queue_add_fd_read(uwsgi.async_queue, uwsgi.signal_socket);
		event_queue_add_fd_read(uwsgi.async_queue, uwsgi.my_signal_socket);
	}
	// set a default request manager 
	//如果没有设置类似greenlet的协程库的话 有的话就启用greenlet_schedule_to_req  之类的回调
	if (!uwsgi.schedule_to_req)
		uwsgi.schedule_to_req = async_schedule_to_req;
	if (!uwsgi.schedule_to_main) {
		uwsgi_log("*** DANGER *** async mode without coroutine/greenthread engine loaded !!!\n");
	}
   //manage_next_request 为0的时候则表示停止处理请求
	while (uwsgi.workers[uwsgi.mywid].manage_next_request) {
		now = (uint64_t) uwsgi_now();
		if (uwsgi.async_runqueue) {
			timeout = 0;
		}
		else {
			min_timeout = uwsgi_min_rb_timer(uwsgi.rb_async_timeouts, NULL);
			if (min_timeout) {
				timeout = min_timeout->value - now;
				if (timeout <= 0) {
					 //检查红黑树中过期的请求并重置
					async_expire_timeouts(now);
					timeout = 0;
				}
			}
			else {
				timeout = -1;
			}
		}
		//从epoll之类的拿取就绪事件 并没有看到类似nginx加锁那样子解决惊群问题 
		//搜索后发现 epoll已经解决了 可以在多个进程内wait了
		uwsgi.async_nevents = event_queue_wait_multi(uwsgi.async_queue, timeout, events, 64);
		now = (uint64_t) uwsgi_now();
		// timeout ???
		if (uwsgi.async_nevents == 0) {
			async_expire_timeouts(now);
		}
		for (i = 0; i < uwsgi.async_nevents; i++) {
			// manage events
			interesting_fd = event_queue_interesting_fd(events, i);
			//内部进程通信socket
			// signals are executed in the main stack... 
			//in the future we could have dedicated stacks for them
			if (uwsgi.signal_socket > -1 && (interesting_fd == uwsgi.signal_socket || interesting_fd == uwsgi.my_signal_socket)) {
				uwsgi.wsgi_req = find_first_available_wsgi_req();
                                if (uwsgi.wsgi_req == NULL) {
                                	uwsgi_async_queue_is_full((time_t)now);
                                        continue; 
                                }
				uwsgi_receive_signal(uwsgi.wsgi_req, interesting_fd, "worker", uwsgi.mywid);
				continue;
			}
			is_a_new_connection = 0;
			// new request coming in ?
			//uwsgi.c 39
			uwsgi_sock = uwsgi.sockets;
			while (uwsgi_sock) {
				//说明这个fd 是已经监听的端口 说明是新连接
				if (interesting_fd == uwsgi_sock->fd) {
					//分配給心情求资源
					is_a_new_connection = 1;
					uwsgi.wsgi_req = find_first_available_wsgi_req();
					//说明服务已经满载了 退出处理新请求的循环
					if (uwsgi.wsgi_req == NULL) {
						uwsgi_async_queue_is_full((time_t)now);
						break;
					}
					// on error re-insert the request in the queue 初始化请求
					wsgi_req_setup(uwsgi.wsgi_req, uwsgi.wsgi_req->async_id, uwsgi_sock);
					//accept新请求 失败则返还资源
					if (wsgi_req_simple_accept(uwsgi.wsgi_req, interesting_fd)) {
						uwsgi.async_queue_unused_ptr++;
						uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req;
						break;
					}
					//把新请求加入epoll 监听 失败则返还资源
					if (wsgi_req_async_recv(uwsgi.wsgi_req)) {
						uwsgi.async_queue_unused_ptr++;
						uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req;
						break;
					}
					// by default the core is in UWSGI_AGAIN mode 
					//UWSGI_AGAIN表示请求还没完成 uwsgi会保存这个req 等到下次循环到来在处理
					uwsgi.wsgi_req->async_status = UWSGI_AGAIN;
					// some protocol (like zeromq) do not need additional parsing, just push it in the runqueue 
					if (uwsgi.wsgi_req->do_not_add_to_async_queue) {
						runqueue_push(uwsgi.wsgi_req);
					}
					break;
				}
				uwsgi_sock = uwsgi_sock->next;
			}
			if (!is_a_new_connection) {
				// proto event
				uwsgi.wsgi_req = find_wsgi_req_proto_by_fd(interesting_fd);
				if (uwsgi.wsgi_req) {
					proto_parser_status = uwsgi.wsgi_req->socket->proto(uwsgi.wsgi_req);
					// reset timeout
					async_reset_request(uwsgi.wsgi_req);
					// parsing complete
					if (!proto_parser_status) {
						// remove fd from event poll and fd proto table 
						uwsgi.async_proto_fd_table[interesting_fd] = NULL;
						event_queue_del_fd(uwsgi.async_queue, interesting_fd, event_queue_read());
						// put request in the runqueue (set it as UWSGI_OK to signal the first run)
						uwsgi.wsgi_req->async_status = UWSGI_OK;
						runqueue_push(uwsgi.wsgi_req);
						continue;
					}
					else if (proto_parser_status < 0) {
						//解析失败 所以释放资源
						uwsgi.async_proto_fd_table[interesting_fd] = NULL;
						close(interesting_fd);
						uwsgi.async_queue_unused_ptr++;
						uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req;
						continue;
					}
					// re-add timer
					async_add_timeout(uwsgi.wsgi_req, uwsgi.socket_timeout);
					continue;
				}
				// app-registered event
				uwsgi.wsgi_req = find_wsgi_req_by_fd(interesting_fd);
				// unknown fd, remove it (for safety)
				if (uwsgi.wsgi_req == NULL) {
					close(interesting_fd);
					continue;
				}
				// remove all the fd monitors and timeout
				async_reset_request(uwsgi.wsgi_req);
				uwsgi.wsgi_req->async_ready_fd = 1;
				uwsgi.wsgi_req->async_last_ready_fd = interesting_fd;
				// put the request in the runqueue again
				runqueue_push(uwsgi.wsgi_req);
			}
		}
		// event queue managed, give cpu to runqueue
		//run queue是解析协议之后 需要处理业务的队列
		current_request = uwsgi.async_runqueue;
		//消费此队列
		while(current_request) {
			// current_request could be nulled on error/end of request
			struct uwsgi_async_request *next_request = current_request->next;
			uwsgi.wsgi_req = current_request->wsgi_req;
			uwsgi.schedule_to_req();
			uwsgi.wsgi_req->switches++;
			// request ended ?
			if (uwsgi.wsgi_req->async_status <= UWSGI_OK ||
				uwsgi.wsgi_req->waiting_fds || uwsgi.wsgi_req->async_timeout) {
				// remove from the runqueue
				runqueue_remove(current_request);
			}
			current_request = next_request;
		}
	}
}

总结

aysnc只是在接受解析请求的时候异步了，但是具体到调用脚本时，还需要脚本具体脚本的框架配合。

虽然不能提高单个请求的速度，但是async带来的不堵塞使得线程上下文切换的开销少了不少，提高了吞吐量。