linux内核的nfs实现框架

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 = http://blog.soso.com/qz.q/{ 

　　.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 = http://blog.soso.com/qz.q/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 = http://blog.soso.com/qz.q/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客户机，就是这么简单。 

原文地址：