linux內核中實現了nfs，nfs具體是用rpc來實現的，于是linux內核實現了rpc，rpc到底是什么，以及協議細節本文不討論，網上書上多的是，包括協議編碼規范也不說，本文僅僅描述一下linux內核的rpc實現框架。
linux內核的rpc模塊實現涉及了大致三個小模塊：一是rpc與用戶層的接口；二是rpc的邏輯控制框架；三是rpc的通信框架。在這三個小模塊里，rpc協議細節貫穿前后，畢竟就是由協議規范來規定具體行為的。這三個模塊中除了第二個是邏輯控制必要的之外，另外兩個都是可插拔可替換的，邏輯控制模塊實際上你自己可以有更好的實現，它無非就是在rpc協議規范下將數據完成XDR二進制編碼然后發送到rpc的通信子模塊，那么這個通信子模塊就是很隨意的了，只要是可以傳輸二進制數據的網絡協議都可以作為rpc的通信協議，當然用的最多的還是TCP/IP了，表現出來就是inet的socket；至于第一個子模塊的可替換就更好理解了，nfs就是一個rpc應用，因此nfs可以作為一個用戶接口子模塊，當然另外一個應用，只要基于rpc的，都可以作為一個rpc用戶接口子模塊。
nfs最開始是vfs層的工作，無非就是一些回調函數，file_operations之類的：
static ssize_t nfs_file_read(struct kiocb *iocb, const struct iovec *iov, unsigned long nr_segs, loff_t pos)
{
...
if (iocb->ki_filp->f_flags & O_DIRECT)
return nfs_file_direct_read(iocb, iov, nr_segs, pos);
...
}
static ssize_t nfs_direct_read(struct kiocb *iocb, const struct iovec *iov, unsigned long nr_segs, loff_t pos)
{
ssize_t result = 0;
struct inode *inode = iocb->ki_filp->f_mapping->host;
struct nfs_direct_req *dreq;
dreq = nfs_direct_req_alloc();
if (!dreq)
...
dreq->inode = inode;
dreq->ctx = get_nfs_open_context(nfs_file_open_context(iocb->ki_filp));
if (!is_sync_kiocb(iocb))
dreq->iocb = iocb;
result = nfs_direct_read_schedule_iovec(dreq, iov, nr_segs, pos);
...
}
以上的函數調用都是在vfs層次，接下來的這個函數負責兩個模塊的交接，就是vfs模塊和rpc邏輯控制模塊，其實可以將nfs的vfs層作為linux內核中rpc實現的第一個小模塊，即與用戶層的接口。linux內核是高度模塊化的，這種模塊化實現的好想有點玄乎，大內核但是模塊化，這其實很正常。
static ssize_t nfs_direct_read_schedule_segment(struct nfs_direct_req *dreq, const struct iovec *iov, loff_t pos)
{
struct nfs_open_context *ctx = dreq->ctx;
struct inode *inode = ctx->path.dentry->d_inode;
unsigned long user_addr = (unsigned long)iov->iov_base;
size_t count = iov->iov_len;
size_t rsize = NFS_SERVER(inode)->rsize;
struct rpc_task *task;
struct rpc_message msg = {
.rpc_cred = ctx->cred,
};
struct rpc_task_setup task_setup_data = {
.rpc_client = NFS_CLIENT(inode), //屬性參數
.rpc_message = &msg, //數據參數，很重要
.callback_ops = &nfs_read_direct_ops, //控制參數，執行過程的回調函數
.workqueue = nfsiod_workqueue,
.flags = RPC_TASK_ASYNC,
};
unsigned int pgbase;
int result;
ssize_t started = 0;
do {
struct nfs_read_data *data;
size_t bytes;
pgbase = user_addr & ~PAGE_MASK;
bytes = min(rsize,count);
result = -ENOMEM;
data = nfs_readdata_alloc(nfs_page_array_len(pgbase, bytes));
if (unlikely(!data))
break;
down_read(¤t->mm->mmap_sem);
result = get_user_pages(current, current->mm, user_addr, data->npages, 1, 0, data->pagevec, NULL);
up_read(¤t->mm->mmap_sem);
...
get_dreq(dreq);
data->req = (struct nfs_page *) dreq;
data->inode = inode;
data->cred = msg.rpc_cred;
data->args.fh = NFS_FH(inode);
data->args.context = get_nfs_open_context(ctx);
data->args.offset = pos;
data->args.pgbase = pgbase;
data->args.pages = data->pagevec;
data->args.count = bytes;
data->res.fattr = &data->fattr;
data->res.eof = 0;
data->res.count = bytes;
msg.rpc_argp = &data->args;
msg.rpc_resp = &data->res;
task_setup_data.task = &data->task; //這里用task_setup_data將數據參數向rpc模塊傳遞
task_setup_data.callback_data = data;
NFS_PROTO(inode)->read_setup(data, &msg);
task = rpc_run_task(&task_setup_data); //從vfs層進入rpc模塊
...//更新文件指針偏移
} while (count != 0);
if (started)
return started;
return result < 0 ? (ssize_t) result : -EFAULT;
}
struct rpc_task *rpc_run_task(const struct rpc_task_setup *task_setup_data)
{
struct rpc_task *task, *ret;
task = rpc_new_task(task_setup_data); //初始化一個rpc任務
...//出錯處理，省略
atomic_inc(&task->tk_count); //設置引用計數
rpc_execute(task); //執行這個rpc任務
ret = task;
...
}
以上過程中rpc_task_setup結構體的作用就是將數據從vfs模塊傳遞到rpc模塊，這個結構體封裝了很多信息，這些信息都是在執行rpc的過程中要用到的或者初始化任務的時候要用到的。rpc_execute是一個很重要的函數，它將rpc的執行過程表示為一個狀態機，rpc的不同執行過程表示不同的狀態，另外rpc_task結構體封裝了一個rpc任務，里面有表示rpc任務當前狀態的字段，控制字段，一些維護管理模塊要用到的鏈表，總之在linux中這樣的結構很多，比如task_struct等等，這些結構體均封裝了一個實體，也就是一個對象，理解了OO就好理解這些了。另外還有一類結構體是專門為了在不同的模塊之間傳遞參數的，比如上面的rpc_task_setup就是，另外還有iovec，kiocb等等，在linux內核中，有些這樣的結構可以在不同的模塊或者層次之間傳遞，上述的實體結構只在一個模塊內被使用，這二類結構體的意義并不同。linux內核不是OO的勝似OO的，這座建筑宏偉又靈活，各類結構體比如實體結構，控制結構（iovec，scan_control）還有連接結構（kobject，list_head）相互作用，可謂各抱地勢，勾心斗角啊，它們之間你可以說是并列獨立的關系，你也可以說是不同的層次，按照OO的說法，list_head和kobject就是基類了，實體結構就是最小面的派生類了，控制結構包含了實體結構，但是如果不按OO的說法，說法就更多了，管它怎么說，總之怎么說都行，linux內核就是棒，linux的靈活性正在這里體現。下面看一下rpc的狀態機：
void rpc_execute(struct rpc_task *task)
{
rpc_set_active(task);
rpc_set_running(task);
__rpc_execute(task);
}
static void __rpc_execute(struct rpc_task *task)
{
int status = 0;
for (;;) {
if (task->tk_callback) {
void (*save_callback)(struct rpc_task *);
save_callback = task->tk_callback;
task->tk_callback = NULL;
save_callback(task);
}
if (!RPC_IS_QUEUED(task)) { //注意tk_action回調函數時刻在更改，代表不同狀態下的rpc任務的不同回調函數。
if (task->tk_action == NULL)
break;
task->tk_action(task); //調用回調函數
}
if (!RPC_IS_QUEUED(task))
continue;
rpc_clear_running(task);
if (RPC_IS_ASYNC(task)) { //如果是異步的，那么退出狀態機，接下來的任務由工作隊列來完成。
if (RPC_IS_QUEUED(task))
return;
if (rpc_test_and_set_running(task))
return;
continue;
}
status = out_of_line_wait_on_bit(&task->tk_runstate, RPC_TASK_QUEUED,
rpc_wait_bit_killable, TASK_KILLABLE);//如果是同步的，那么睡眠在這里，繼續狀態機的運轉。
...//信號異常處理
rpc_set_running(task);
}
rpc_release_task(task);
}
在linux內核實現的rpc中，異步rpc是靠工作隊列來完成的，在老版本中是靠一個內核線程完成的，在新的內核中，工作隊列擔當了一個很重要的角色，還記得AIO嗎，也是工作隊列完成的，這么看來在新內核中工作隊列實現了異步IO和異步rpc以及...形式看起來更加統一了，不用再像以前那樣為每一個特殊內核任務都創建一個獨立的內核線程了，統一用工作隊列完成，2.6內核就是不錯。具體說來，如果當前的rpc傳輸任務沒有完成，那么直接返回到__rpc_execute函數，然后判斷后返回，但是這個時候rpc還沒有完成，具體的完成工作就要工作隊列完成了，大致過程和AIO一樣，就是不睡眠而是直接返回，待到該任務被wakeup“喚醒”（加上引號是因為根本沒有真正睡眠何談真正喚醒）的時候將任務加入到工作隊列中去，工作隊列會調度任務的執行的。
在上述的狀態機中，并沒有設置所謂的狀態，而是通過回調函數的形式，在狀態要改變的時候更新回調函數，這樣就免去了一個大的switch-case了，不過這只是編程上的技巧。涉及到具體過程上，最終是要傳輸數據的，在call_allocate以后的狀態回調函數大致演化順序為：call_allocate->call_bind->call_connect->call_transmit->rpc_exit_task，這中間省略了狀態相關的函數，以下看一下兩個最重要的:
static void call_transmit(struct rpc_task *task)
{
dprint_status(task);
task->tk_action = call_status;
if (task->tk_status < 0)
return;
task->tk_status = xprt_prepare_transmit(task);
if (task->tk_status != 0)
return;
task->tk_action = call_transmit_status;
if (rpc_task_need_encode(task)) {
BUG_ON(task->tk_rqstp->rq_bytes_sent != 0);
call_encode(task); //編碼，rpc的底層規范
if (task->tk_status != 0)
return;
}
xprt_transmit(task); //實際傳輸數據
if (task->tk_status < 0)
return;
call_transmit_status(task);
if (task->tk_msg.rpc_proc->p_decode != NULL)
return;
task->tk_action = rpc_exit_task; //傳輸完畢
rpc_wake_up_queued_task(&task->tk_xprt->pending, task);
}
void rpc_exit_task(struct rpc_task *task)
{
task->tk_action = NULL;
if (task->tk_ops->rpc_call_done != NULL) {
lock_kernel();
task->tk_ops->rpc_call_done(task, task->tk_calldata);
unlock_kernel();
if (task->tk_action != NULL) {
WARN_ON(RPC_ASSASSINATED(task));
/* Always release the RPC slot and buffer memory */
xprt_release(task);
}
}
}
在實際傳輸之前要用XDR規范將數據進行編碼，這是rpc的約定。看一下xprt_transmit就會發現，底層的rpc使用socket將數據傳給服務器的，當然也可以用別的機制，比如任何底層鏈路協議，只要能進行網絡傳輸的就可以，在socket實現的rpc中，socket結構是怎樣傳遞給rpc的xprt_transmit的呢？還記得上面說的linux內核的結構類型吧，實際上rpc_task內就包含了足夠的信息，而rpc_task在初始化的時候，從vfs層傳遞而來的數據結構已經將數據參數賦給了rpc_task了，而這些參數是在open的時候被創建的，這樣的話，從一個rpc_task很容易的得到了需要的數據，比如socket。linux內核中的數據結構耦合性彼此都很小，并且數據結構本身大多數也都是小型的，這種特性使得不同模塊的數據結構之間的協作相當容易，也正因為如此，一個數據結構才得以在不同的模塊穿梭，方便得傳遞參數。
以上就是linux內核中關于nfs的rpc客戶端的大致流程，那么服務器是如何實現的呢？很簡單，考慮以下C/S模型的結構，最基本的就是服務器只有一個，客戶端隨意，也就是說服務器是確定的，而客戶端不確定因素較多，這么說來，服務器就比客戶端要簡單不少，這就好像一個web服務器在機房里面靜靜地運行著，大不了整個集群啥的，但是這個web服務器的客戶端就五花八門了，pc臺式機，筆記本，手機，PDA，教師，明星，流氓，馬加爵...稍微具體來說服務器就是啟動一個守護內核線程，然后循環處理收到的請求，其實就是nfsd，在linux中主要由nfs用到了rpc。nfsd在自己的所有服務器套接字上讀取數據請求，然后查找遠程rpc客戶機調用的過程，隨之調用這個過程，并且將結果返回遠程rpc客戶機，就是這么簡單。
?本文轉自 dog250 51CTO博客，原文鏈接:http://blog.51cto.com/dog250/1273942
版权声明：本站所有资料均为网友推荐收集整理而来,仅供学习和研究交流使用。
工作时间：8:00-18:00
客服电话
电子邮件
admin@qq.com
扫码二维码
获取最新动态
