這幾天寫網路爬蟲, 結果伺服器長時間阻塞在connect上導致不可容忍的併發問題, 於是著手增加一個非同步connect介面.

  常規的實現手段為配合select進行檢測, 不過其效能對於高併發時會有些問題,  如果想做到結構簡單合理採用one peer one thread的處理方式還會引發過多的執行緒上下文切換導致不必要的效能浪費, 故放棄使用select來實現.

  由於伺服器網路庫採用epoll實現, 故此介面也基於epoll實現. 查閱了一些資料, 總結一下:

  1. 設定socket為nonblocking

  2. 呼叫connect

  3. 將socket加入epoll監聽

  4. 檢測epollout 或 epollhup, 回撥上層處理函式做相應的處理, 如果socket ready(僅檢測到epollout) 還需額外通過getsockopt判斷sock_error值是否正常

  5. 將該fd從epoll中清除

  ok, 流程明確後, 就開工吧.

 介面需求:

  1. 可以設定超時, 方便的控制時間, 精度上定為妙級即可

  2. 對於連線成功, 錯誤, 超時均會回撥上層介面進行通知

 資料結構和介面:

 #define LNET_CONN_SUCESS 0 #define LNET_CONN_TIMEOUT 1 #define LNET_CONN_ERROR 2 typedef union{ int u32; uint64_t u64; void* ptr; }conn_arg_t; typedef void (*pconn)(int fd, int ev, conn_arg_t arg); typedef struct conn_event { int fd; int timeout; int is_block; int conn_time; //start connect time pconn pc; conn_arg_t arg; }conn_ev;

 

 實現:

  我們知道, 如果呼叫connect立刻就成功, 那麼我們也無需將其加入epoll進行等待, 這樣沒有必要, 所以, 可以稍微的優化下 :), 僅當connect的errno為EINPROGRESS時才將其加入epoll

  // async connect // return: // > 0 : sucess connect , you can use it // -1 : error // 0 : connect has in process // // arg: // timeout: >= 0 as normal, < 0 infinite int lnet_conn_a(char* ip, int port, int is_block, int timeout, pconn pfunc, conn_arg_t arg) { int sockfd = -1; int s = net_conn_a(ip, port, &sockfd); if( s == 0 ){ // connect sucess return sockfd; } else if( s == -1 ){ // connect error return -1; } else{ // connect has in process conn_ev* cev = (conn_ev*)malloc(sizeof(conn_ev)); cev->fd = sockfd; cev->timeout = timeout; cev->is_block = is_block; cev->conn_time = time(NULL); cev->pc = pfunc; cev->arg.u64 = arg.u64; //copy from arg int vfd = lnet_gen_vfd(); hash_set_int(pnif->ev_pool, vfd, cev); uint64_t event_data = lnet_make_eventdata(LNET_EVENT_TYPE_CONN, vfd); net_epoll_add(pnif->c_epfd, sockfd, FREAD | FWRITE, (void*)event_data); return 0; } }

  上面, 我們成功的將處於處理中的連結加入epoll等待通知, 那麼接下來看看epoll_wait如何處理:

  void* lnet_conn_base(void* arg) { int nfds = 0, i = 0; int epfd = pnif->c_epfd; int last_check_time = 0; struct epoll_event* events = (struct epoll_event*)malloc(sizeof(events) *EPOLL_QUEUE_NUM); printf("start conn thread/n"); while(1){ last_check_time = time(NULL); nfds = epoll_wait(epfd, events, EPOLL_QUEUE_NUM, LNET_CONN_TIME_OUT); for (i=0; i<nfds; ++i){ uint64_t data = events[i].data.u64; int vfd = lnet_get_event_fd((uint64_t)data); // connect close if( events[i].events & (EPOLLHUP | EPOLLERR) ){ uint64_t event_data = lnet_make_eventdata(LNET_EVENT_TYPE_CONN_ERROR, vfd); lnet_push_queue(pnif->r_pool, vfd, (void*)event_data); } if( events[i].events & EPOLLOUT ){ uint64_t event_data = lnet_make_eventdata(LNET_EVENT_TYPE_CONN_SUCESS, vfd); lnet_push_queue(pnif->r_pool, vfd, (void*)event_data); } } if( time(NULL) - last_check_time >= LNET_CONN_CHECK_ALTER ){ lnet_check_conn_timeout(); } } free(events); return NULL; }

  這裡的timeout處理的優點粗糙, 沒有使用time_wheel方式(具體方案詳見陳碩的博文 http://blog.csdn.net/Solstice/archive/2011/05/04/6395098.aspx), 而是採用的蠻力進行輪詢, 對於併發連線小時還能忍受.. 有待改進.

  接下來, 我們需要處理的就是具體的事件了: 成功, 異常, 超時.

  這裡簡單列舉一下成功時的處理:

  ... 上面省略其他不相關事件處理... else if( event_type == LNET_EVENT_TYPE_CONN_SUCESS ){ conn_ev* cev = hash_get_int(pnif->ev_pool, vfd); if( cev ){ int error = 0; socklen_t len = sizeof(int); if (( 0 == getsockopt(cev->fd, SOL_SOCKET, SO_ERROR, &error, &len) )){ if( 0 == error ){ cev->pc(cev->fd, LNET_CONN_SUCESS, cev->arg); } else{ printf("connect fd has not ready! fd=%d/n", cev->fd); cev->pc(cev->fd, LNET_CONN_ERROR, cev->arg); } free( hash_del_int(pnif->ev_pool, vfd) ); net_epoll_del(pnif->c_epfd, cev->fd); } } lnet_pop_queue(pnif->r_pool, vfd); } ...

  這裡可以看到, 具體的回撥通知呼叫和getsockopt的檢測, so.. 至此, 便完成了非同步connect的介面.

  當然, 這裡面timeout的部分還有待改進, 程式碼有些部分有些凌亂... 不過重點的在於原理配合程式碼的理解過程, 如有不正之處, 歡迎批評指出 :)