一個TCP非阻塞client端簡單的例子

如果我們要產(chǎn)生一個非阻塞的socket,在C語言中如下代碼所示:

// 創(chuàng)建socket

intsock_fd=socket(AF_INET,SOCK_STREAM,0);

...

// 更改socket為nonblock

fcntl(sock_fd,F_SETFL,fdflags|O_NONBLOCK);

// connect

....

while(1){

intrecvlen=recv(sock_fd,recvbuf,RECV_BUF_SIZE);

......

}

...

由于網(wǎng)絡(luò)協(xié)議非常復(fù)雜，內(nèi)核里面用到了大量的面向?qū)ο蟮募记?，所以我們從?chuàng)建連接開始，一步一步追述到最后代碼的調(diào)用點。

socket的創(chuàng)建

很明顯，內(nèi)核的第一步應(yīng)該是通過AF_INET、SOCK_STREAM以及最后一個參數(shù)0定位到需要創(chuàng)建一個TCP的socket,如下圖綠線所示:

我們跟蹤源碼調(diào)用

socket(AF_INET,SOCK_STREAM,0)

|->sys_socket進(jìn)入系統(tǒng)調(diào)用

|->sock_create

|->__sock_create

進(jìn)一步分析__sock_create的代碼判斷:

conststructnet_proto_family*pf;

// RCU(Read-Copy Update)是linux的一種內(nèi)核同步方法，在此不闡述

// family=INET

pf=rcu_dereference(net_families[family]);

err=pf->create(net,sock,protocol);

conststructnet_proto_family*pf;// RCU(Read-Copy Update)是linux的一種內(nèi)核同步方法，在此不闡述// family=INETpf = rcu_dereference(net_families[family]); err = pf->create(net, sock, protocol);

則通過源碼可知，由于是AF_INET(PF_INET),所以net_families[PF_INET].create=inet_create(以后我們都用PF_INET表示)，即pf->create = inet_create; 進(jìn)一步追溯調(diào)用:

inet_create(structnet*net,structsocket*sock,intprotocol){

Sock*sock;

......

// 此處是尋找對應(yīng)協(xié)議處理器的過程

lookup_protocol:

// 迭代尋找protocol==answer->protocol的情況

list_for_each_rcu(p,&inetsw[sock->type])answer=list_entry(p,structinet_protosw,list);

/* Check the non-wild match. */

if(protocol==answer->protocol){

if(protocol!=IPPROTO_IP)

break;

}

......

// 這邊answer指的是SOCK_STREAM

sock->ops=answer->ops;

answer_no_check=answer->no_check;

// 這邊sk->prot就是answer_prot=>tcp_prot

sk=sk_alloc(net,PF_INET,GFP_KERNEL,answer_prot);

sock_init_data(sock,sk);

......

}

上面的代碼就是在INET中尋找SOCK_STREAM的過程了 我們再看一下inetsw[SOCK_STREAM]的具體配置:

staticstructinet_protoswinetsw_array[]=

{

{

.type=SOCK_STREAM,

.protocol=IPPROTO_TCP,

.prot= &tcp_prot,

.ops=&inet_stream_ops,

.capability= -1,

.no_check=0,

.flags=INET_PROTOSW_PERMANENT|

INET_PROTOSW_ICSK,

},

......

}

這邊也用了重載，AF_INET有TCP、UDP以及Raw三種:

從上述代碼，我們可以清楚的發(fā)現(xiàn)sock->ops=&inet_stream_ops;

conststructproto_opsinet_stream_ops={

.family =PF_INET,

.owner =THIS_MODULE,

......

.sendmsg =tcp_sendmsg,

.recvmsg =sock_common_recvmsg,

......

}

即sock->ops->recvmsg = sock_common_recvmsg;

同時sock->sk->sk_prot = tcp_prot;

我們再看下tcp_prot中的各個函數(shù)重載的定義:

structprototcp_prot={

.name ="TCP",

.close =tcp_close,

.connect =tcp_v4_connect,

.disconnect =tcp_disconnect,

.accept =inet_csk_accept,

......

// 我們重點考察tcp的讀

.recvmsg =tcp_recvmsg,

......

}

fcntl控制socket的阻塞\非阻塞狀態(tài)

我們用fcntl修改socket的阻塞\非阻塞狀態(tài)。 事實上: fcntl的作用就是將O_NONBLOCK標(biāo)志位存儲在sock_fd對應(yīng)的filp結(jié)構(gòu)的f_lags里,如下圖所示。

fcntl(sock_fd,F_SETFL,fdflags|O_NONBLOCK);

|->setfl

追蹤setfl代碼:

staticintsetfl(intfd,structfile*filp,unsignedlongarg){

......

filp->f_flags=(arg&SETFL_MASK)|(filp->f_flags& ~SETFL_MASK);

......

}

上圖中，由sock_fd在task_struct(進(jìn)程結(jié)構(gòu)體)->files_struct->fd_array中找到對應(yīng)的socket的file描述符，再修改file->flags

在調(diào)用socket.recv的時候

我們跟蹤源碼調(diào)用:

socket.recv

|->sys_recv

|->sys_recvfrom

|->sock_recvmsg

|->__sock_recvmsg

|->sock->ops->recvmsg

由上文可知: sock->ops->recvmsg = sock_common_recvmsg;

sock

值得注意的是,在sock_recmsg中,有對標(biāo)識O_NONBLOCK的處理

if(sock->file->f_flags&O_NONBLOCK)

flags|=MSG_DONTWAIT;

上述代碼中sock關(guān)聯(lián)的file中獲取其f_flags,如果flags有O_NONBLOCK標(biāo)識，那么就設(shè)置msg_flags為MSG_DONTWAIT(不等待)。

fcntl與socket就是通過其共同操作File結(jié)構(gòu)關(guān)聯(lián)起來的。

繼續(xù)跟蹤調(diào)用

sock_common_recvmsg

intsock_common_recvmsg(structkiocb *iocb,structsocket *sock,

structmsghdr *msg,size_tsize,intflags){

......

// 如果flags的MSG_DONTWAIT標(biāo)識置位，則傳給recvmsg的第5個參數(shù)為正,否則為0

err=sk->sk_prot->recvmsg(iocb,sk,msg,size,flags&MSG_DONTWAIT,

flags& ~MSG_DONTWAIT,&addr_len);

.....

}

由上文可知: sk->sk_prot->recvmsg 其中sk_prot=tcp_prot,即最終調(diào)用的是tcp_prot->tcp_recvmsg,上面的代碼可以看出，如果fcntl(O_NONBLOCK)=>MSG_DONTWAIT置位=>(flags & MSG_DONTWAIT)>0, 再結(jié)合tcp_recvmsg的函數(shù)簽名,即如果設(shè)置了O_NONBLOCK的話，設(shè)置給tcp_recvmsg的nonblock參數(shù)>0,關(guān)系如下圖所示:

最終的調(diào)用邏輯tcp_recvmsg

首先我們看下tcp_recvmsg的函數(shù)簽名:

inttcp_recvmsg(structkiocb*iocb,structsock*sk,structmsghdr*msg,

size_tlen,intnonblock,intflags,int*addr_len)

顯然我們關(guān)注焦點在(int nonblock這個參數(shù)上):

inttcp_recvmsg(structkiocb*iocb,structsock*sk,structmsghdr*msg,

size_tlen,intnonblock,intflags,int*addr_len){

......

// copied是指向用戶空間拷貝了多少字節(jié)，即讀了多少

intcopied;

// target指的是期望多少字節(jié)

inttarget;

// 等效為timo = noblock ? 0 : sk->sk_rcvtimeo;

timeo=sock_rcvtimeo(sk,nonblock);

......

// 如果設(shè)置了MSG_WAITALL標(biāo)識target=需要讀的長度

// 如果未設(shè)置，則為最低低水位值

target=sock_rcvlowat(sk,flags&MSG_WAITALL,len);

......

do{

// 表明讀到數(shù)據(jù)

if(copied){

// 注意，這邊只要!timeo，即nonblock設(shè)置了就會跳出循環(huán)

if(sk->sk_err||

sk->sk_state==TCP_CLOSE||

(sk->sk_shutdown&RCV_SHUTDOWN)||

!timeo||

signal_pending(current)||

(flags&MSG_PEEK))

break;

}else{

// 到這里，表明沒有讀到任何數(shù)據(jù)

// 且nonblock設(shè)置了導(dǎo)致timeo=0，則返回-EAGAIN,符合我們的預(yù)期

if(!timeo){

copied= -EAGAIN;

break;

}

// 這邊如果堵到了期望的數(shù)據(jù)，繼續(xù)，否則當(dāng)前進(jìn)程阻塞在sk_wait_data上

if(copied>=target){

/* Do not sleep, just process backlog. */

release_sock(sk);

lock_sock(sk);

}else

sk_wait_data(sk,&timeo);

}while(len>0);

......

returncopied

}

上面的邏輯歸結(jié)起來就是：

(1)在設(shè)置了nonblock的時候，如果copied>0,則返回讀了多少字節(jié),如果copied=0，則返回-EAGAIN,提示應(yīng)用重復(fù)調(diào)用。

(2)如果沒有設(shè)置nonblock，如果讀取的數(shù)據(jù)>=期望，則返回讀取了多少字節(jié)。如果沒有則用sk_wait_data將當(dāng)前進(jìn)程等待。

如下流程圖所示:

阻塞函數(shù)sk_wait_data

sk_wait_data代碼-函數(shù)為:

// 將進(jìn)程狀態(tài)設(shè)置為可打斷INTERRUPTIBLE

prepare_to_wait(sk->sk_sleep,&wait,TASK_INTERRUPTIBLE);

set_bit(SOCK_ASYNC_WAITDATA,&sk->sk_socket->flags);

// 通過調(diào)用schedule_timeout讓出CPU，然后進(jìn)行睡眠

rc=sk_wait_event(sk,timeo,!skb_queue_empty(&sk->sk_receive_queue));

// 到這里的時候，有網(wǎng)絡(luò)事件或超時事件喚醒了此進(jìn)程，繼續(xù)運行

clear_bit(SOCK_ASYNC_WAITDATA,&sk->sk_socket->flags);

finish_wait(sk->sk_sleep,&wait);

該函數(shù)調(diào)用schedule_timeout進(jìn)入睡眠，其進(jìn)一步調(diào)用了schedule函數(shù)，首先從運行隊列刪除，其次加入到等待隊列，最后調(diào)用和體系結(jié)構(gòu)相關(guān)的switch_to宏來完成進(jìn)程間的切換。

如下圖所示:

阻塞后什么時候恢復(fù)運行呢

情況1:有對應(yīng)的網(wǎng)絡(luò)數(shù)據(jù)到來

首先我們看下網(wǎng)絡(luò)分組到來的內(nèi)核路徑，網(wǎng)卡發(fā)起中斷后調(diào)用netif_rx將事件掛入CPU的等待隊列，并喚起軟中斷(soft_irq)，再通過linux的軟中斷機(jī)制調(diào)用net_rx_action，如下圖所示:

緊接著跟蹤next_rx_action

next_rx_action

|-process_backlog

......

|->packet_type->func在這里我們考慮ip_rcv

|->ipprot->handler在這里ipprot重載為tcp_protocol

(handler即為tcp_v4_rcv)

緊接著tcp_v4_rcv:

tcp_input.c

tcp_v4_rcv

|-tcp_v4_do_rcv

|-tcp_rcv_state_process

|-tcp_data_queue

|-sk->sk_data_ready=sock_def_readable

|-wake_up_interruptible

|-__wake_up

|-__wake_up_common

在這里__wake_up_common將停在當(dāng)前wait_queue_head_t中的進(jìn)程喚醒，即狀態(tài)改為task_running，等待CFS調(diào)度以進(jìn)行下一步的動作,如下圖所示。

情況2:設(shè)定的超時時間到來

在前面調(diào)用sk_wait_event中調(diào)用了schedule_timeout

fastcallsignedlong__sched schedule_timeout(signedlongtimeout){

......

// 設(shè)定超時的回掉函數(shù)為process_timeout

setup_timer(&timer,process_timeout,(unsignedlong)current);

__mod_timer(&timer,expire);

// 這邊讓出CPU

schedule();

del_singleshot_timer_sync(&timer);

timeout=expire-jiffies;

out:

// 返回經(jīng)過了多長事件

returntimeout

}

process_timeout函數(shù)即是將此進(jìn)程重新喚醒

staticvoidprocess_timeout(unsignedlong__data)

{

wake_up_process((structtask_struct*)__data);

}