开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 嵌入式/单片机编程 > 嵌入式Linux > 查看源码
xprt.c
资源名称：SBC-2410X_kernel.tar.gz　[点击查看]
上传用户：lgb322
上传日期：2013-02-24
资源大小：30529k
文件大小：38k
源码类别：

嵌入式Linux

开发平台：

Unix_Linux

xprt.c：源码内容
  1. /*
  2.  *  linux/net/sunrpc/xprt.c
  3.  *
  4.  *  This is a generic RPC call interface supporting congestion avoidance,
  5.  *  and asynchronous calls.
  6.  *
  7.  *  The interface works like this:
  8.  *
  9.  *  - When a process places a call, it allocates a request slot if
  10.  * one is available. Otherwise, it sleeps on the backlog queue
  11.  * (xprt_reserve).
  12.  *  - Next, the caller puts together the RPC message, stuffs it into
  13.  * the request struct, and calls xprt_call().
  14.  *  - xprt_call transmits the message and installs the caller on the
  15.  * socket's wait list. At the same time, it installs a timer that
  16.  * is run after the packet's timeout has expired.
  17.  *  - When a packet arrives, the data_ready handler walks the list of
  18.  * pending requests for that socket. If a matching XID is found, the
  19.  * caller is woken up, and the timer removed.
  20.  *  - When no reply arrives within the timeout interval, the timer is
  21.  * fired by the kernel and runs xprt_timer(). It either adjusts the
  22.  * timeout values (minor timeout) or wakes up the caller with a status
  23.  * of -ETIMEDOUT.
  24.  *  - When the caller receives a notification from RPC that a reply arrived,
  25.  * it should release the RPC slot, and process the reply.
  26.  * If the call timed out, it may choose to retry the operation by
  27.  * adjusting the initial timeout value, and simply calling rpc_call
  28.  * again.
  29.  *
  30.  *  Support for async RPC is done through a set of RPC-specific scheduling
  31.  *  primitives that `transparently' work for processes as well as async
  32.  *  tasks that rely on callbacks.
  33.  *
  34.  *  Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de>
  35.  *
  36.  *  TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com>
  37.  *  TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
  38.  *  TCP NFS related read + write fixes
  39.  *   (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
  40.  *
  41.  *  Rewrite of larges part of the code in order to stabilize TCP stuff.
  42.  *  Fix behaviour when socket buffer is full.
  43.  *   (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
  44.  */
  46. #define __KERNEL_SYSCALLS__
  48. #include <linux/version.h>
  49. #include <linux/types.h>
  50. #include <linux/slab.h>
  51. #include <linux/capability.h>
  52. #include <linux/sched.h>
  53. #include <linux/errno.h>
  54. #include <linux/socket.h>
  55. #include <linux/in.h>
  56. #include <linux/net.h>
  57. #include <linux/mm.h>
  58. #include <linux/udp.h>
  59. #include <linux/unistd.h>
  60. #include <linux/sunrpc/clnt.h>
  61. #include <linux/file.h>
  63. #include <net/sock.h>
  64. #include <net/checksum.h>
  65. #include <net/udp.h>
  67. #include <asm/uaccess.h>
  69. /* Following value should be > 32k + RPC overhead */
  70. #define XPRT_MIN_WRITE_SPACE (35000 + SOCK_MIN_WRITE_SPACE)
  72. extern spinlock_t rpc_queue_lock;
  74. /*
  75.  * Local variables
  76.  */
  78. #ifdef RPC_DEBUG
  79. # undef  RPC_DEBUG_DATA
  80. # define RPCDBG_FACILITY RPCDBG_XPRT
  81. #endif
  83. /*
  84.  * Local functions
  85.  */
  86. static void xprt_request_init(struct rpc_task *, struct rpc_xprt *);
  87. static void do_xprt_transmit(struct rpc_task *);
  88. static void xprt_reserve_status(struct rpc_task *task);
  89. static void xprt_disconnect(struct rpc_xprt *);
  90. static void xprt_reconn_status(struct rpc_task *task);
  91. static struct socket *xprt_create_socket(int, struct rpc_timeout *);
  92. static int xprt_bind_socket(struct rpc_xprt *, struct socket *);
  93. static void xprt_remove_pending(struct rpc_xprt *);
  95. #ifdef RPC_DEBUG_DATA
  96. /*
  97.  * Print the buffer contents (first 128 bytes only--just enough for
  98.  * diropres return).
  99.  */
  100. static void
  101. xprt_pktdump(char *msg, u32 *packet, unsigned int count)
  102. {
  103. u8 *buf = (u8 *) packet;
  104. int j;
  106. dprintk("RPC:      %sn", msg);
  107. for (j = 0; j < count && j < 128; j += 4) {
  108. if (!(j & 31)) {
  109. if (j)
  110. dprintk("n");
  111. dprintk("0x%04x ", j);
  112. }
  113. dprintk("%02x%02x%02x%02x ",
  114. buf[j], buf[j+1], buf[j+2], buf[j+3]);
  115. }
  116. dprintk("n");
  117. }
  118. #else
  119. static inline void
  120. xprt_pktdump(char *msg, u32 *packet, unsigned int count)
  121. {
  122. /* NOP */
  123. }
  124. #endif
  126. /*
  127.  * Look up RPC transport given an INET socket
  128.  */
  129. static inline struct rpc_xprt *
  130. xprt_from_sock(struct sock *sk)
  131. {
  132. return (struct rpc_xprt *) sk->user_data;
  133. }
  135. /*
  136.  * Adjust the iovec to move on 'n' bytes
  137.  */
  138.  
  139. extern inline void
  140. xprt_move_iov(struct msghdr *msg, struct iovec *niv, unsigned amount)
  141. {
  142. struct iovec *iv=msg->msg_iov;
  143. int i;
  145. /*
  146.  * Eat any sent iovecs
  147.  */
  148. while (iv->iov_len <= amount) {
  149. amount -= iv->iov_len;
  150. iv++;
  151. msg->msg_iovlen--;
  152. }
  154. /*
  155.  * And chew down the partial one
  156.  */
  157. niv[0].iov_len = iv->iov_len-amount;
  158. niv[0].iov_base =((unsigned char *)iv->iov_base)+amount;
  159. iv++;
  161. /*
  162.  * And copy any others
  163.  */
  164. for(i = 1; i < msg->msg_iovlen; i++)
  165. niv[i]=*iv++;
  167. msg->msg_iov=niv;
  168. }
  170. /*
  171.  * Serialize write access to sockets, in order to prevent different
  172.  * requests from interfering with each other.
  173.  * Also prevents TCP socket reconnections from colliding with writes.
  174.  */
  175. static int
  176. xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
  177. {
  178. int retval;
  179. spin_lock_bh(&xprt->sock_lock);
  180. if (!xprt->snd_task)
  181. xprt->snd_task = task;
  182. else if (xprt->snd_task != task) {
  183. dprintk("RPC: %4d TCP write queue full (task %d)n",
  184. task->tk_pid, xprt->snd_task->tk_pid);
  185. task->tk_timeout = 0;
  186. task->tk_status = -EAGAIN;
  187. rpc_sleep_on(&xprt->sending, task, NULL, NULL);
  188. }
  189. retval = xprt->snd_task == task;
  190. spin_unlock_bh(&xprt->sock_lock);
  191. return retval;
  192. }
  194. /*
  195.  * Releases the socket for use by other requests.
  196.  */
  197. static void
  198. xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
  199. {
  200. spin_lock_bh(&xprt->sock_lock);
  201. if (xprt->snd_task == task) {
  202. xprt->snd_task = NULL;
  203. rpc_wake_up_next(&xprt->sending);
  204. }
  205. spin_unlock_bh(&xprt->sock_lock);
  206. }
  208. /*
  209.  * Write data to socket.
  210.  */
  211. static inline int
  212. xprt_sendmsg(struct rpc_xprt *xprt, struct rpc_rqst *req)
  213. {
  214. struct socket *sock = xprt->sock;
  215. struct msghdr msg;
  216. mm_segment_t oldfs;
  217. int result;
  218. int slen = req->rq_slen - req->rq_bytes_sent;
  219. struct iovec niv[MAX_IOVEC];
  221. if (slen <= 0)
  222. return 0;
  224. if (!sock)
  225. return -ENOTCONN;
  227. xprt_pktdump("packet data:",
  228. req->rq_svec->iov_base,
  229. req->rq_svec->iov_len);
  231. msg.msg_flags   = MSG_DONTWAIT|MSG_NOSIGNAL;
  232. msg.msg_iov = req->rq_svec;
  233. msg.msg_iovlen = req->rq_snr;
  234. msg.msg_name = (struct sockaddr *) &xprt->addr;
  235. msg.msg_namelen = sizeof(xprt->addr);
  236. msg.msg_control = NULL;
  237. msg.msg_controllen = 0;
  239. /* Dont repeat bytes */
  240. if (req->rq_bytes_sent)
  241. xprt_move_iov(&msg, niv, req->rq_bytes_sent);
  243. oldfs = get_fs(); set_fs(get_ds());
  244. result = sock_sendmsg(sock, &msg, slen);
  245. set_fs(oldfs);
  247. dprintk("RPC:      xprt_sendmsg(%d) = %dn", slen, result);
  249. if (result >= 0)
  250. return result;
  252. switch (result) {
  253. case -ECONNREFUSED:
  254. /* When the server has died, an ICMP port unreachable message
  255.  * prompts ECONNREFUSED.
  256.  */
  257. break;
  258. case -EAGAIN:
  259. if (test_bit(SOCK_NOSPACE, &sock->flags))
  260. result = -ENOMEM;
  261. break;
  262. case -ENOTCONN:
  263. case -EPIPE:
  264. /* connection broken */
  265. if (xprt->stream)
  266. result = -ENOTCONN;
  267. break;
  268. default:
  269. printk(KERN_NOTICE "RPC: sendmsg returned error %dn", -result);
  270. }
  271. return result;
  272. }
  274. /*
  275.  * Read data from socket
  276.  */
  277. static int
  278. xprt_recvmsg(struct rpc_xprt *xprt, struct iovec *iov, int nr, unsigned len, unsigned shift)
  279. {
  280. struct socket *sock = xprt->sock;
  281. struct msghdr msg;
  282. mm_segment_t oldfs;
  283. struct iovec niv[MAX_IOVEC];
  284. int result;
  286. if (!sock)
  287. return -ENOTCONN;
  289. msg.msg_flags   = MSG_DONTWAIT|MSG_NOSIGNAL;
  290. msg.msg_iov = iov;
  291. msg.msg_iovlen = nr;
  292. msg.msg_name = NULL;
  293. msg.msg_namelen = 0;
  294. msg.msg_control = NULL;
  295. msg.msg_controllen = 0;
  297. /* Adjust the iovec if we've already filled it */
  298. if (shift)
  299. xprt_move_iov(&msg, niv, shift);
  301. oldfs = get_fs(); set_fs(get_ds());
  302. result = sock_recvmsg(sock, &msg, len, MSG_DONTWAIT);
  303. set_fs(oldfs);
  305. dprintk("RPC:      xprt_recvmsg(iov %p, len %d) = %dn",
  306. iov, len, result);
  307. return result;
  308. }
  311. /*
  312.  * Adjust RPC congestion window
  313.  * We use a time-smoothed congestion estimator to avoid heavy oscillation.
  314.  */
  315. static void
  316. xprt_adjust_cwnd(struct rpc_xprt *xprt, int result)
  317. {
  318. unsigned long cwnd;
  320. if (xprt->nocong)
  321. return;
  322. /*
  323.  * Note: we're in a BH context
  324.  */
  325. spin_lock(&xprt->xprt_lock);
  326. cwnd = xprt->cwnd;
  327. if (result >= 0) {
  328. if (xprt->cong < cwnd || time_before(jiffies, xprt->congtime))
  329. goto out;
  330. /* The (cwnd >> 1) term makes sure
  331.  * the result gets rounded properly. */
  332. cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd;
  333. if (cwnd > RPC_MAXCWND)
  334. cwnd = RPC_MAXCWND;
  335. else
  336. pprintk("RPC: %lu %ld cwndn", jiffies, cwnd);
  337. xprt->congtime = jiffies + ((cwnd * HZ) << 2) / RPC_CWNDSCALE;
  338. dprintk("RPC:      cong %08lx, cwnd was %08lx, now %08lx, "
  339. "time %ld msn", xprt->cong, xprt->cwnd, cwnd,
  340. (xprt->congtime-jiffies)*1000/HZ);
  341. } else if (result == -ETIMEDOUT) {
  342. if ((cwnd >>= 1) < RPC_CWNDSCALE)
  343. cwnd = RPC_CWNDSCALE;
  344. xprt->congtime = jiffies + ((cwnd * HZ) << 3) / RPC_CWNDSCALE;
  345. dprintk("RPC:      cong %ld, cwnd was %ld, now %ld, "
  346. "time %ld msn", xprt->cong, xprt->cwnd, cwnd,
  347. (xprt->congtime-jiffies)*1000/HZ);
  348. pprintk("RPC: %lu %ld cwndn", jiffies, cwnd);
  349. }
  351. xprt->cwnd = cwnd;
  352.  out:
  353. spin_unlock(&xprt->xprt_lock);
  354. }
  356. /*
  357.  * Adjust timeout values etc for next retransmit
  358.  */
  359. int
  360. xprt_adjust_timeout(struct rpc_timeout *to)
  361. {
  362. if (to->to_retries > 0) {
  363. if (to->to_exponential)
  364. to->to_current <<= 1;
  365. else
  366. to->to_current += to->to_increment;
  367. if (to->to_maxval && to->to_current >= to->to_maxval)
  368. to->to_current = to->to_maxval;
  369. } else {
  370. if (to->to_exponential)
  371. to->to_initval <<= 1;
  372. else
  373. to->to_initval += to->to_increment;
  374. if (to->to_maxval && to->to_initval >= to->to_maxval)
  375. to->to_initval = to->to_maxval;
  376. to->to_current = to->to_initval;
  377. }
  379. if (!to->to_current) {
  380. printk(KERN_WARNING "xprt_adjust_timeout: to_current = 0!n");
  381. to->to_current = 5 * HZ;
  382. }
  383. pprintk("RPC: %lu %sn", jiffies,
  384. to->to_retries? "retrans" : "timeout");
  385. return to->to_retries-- > 0;
  386. }
  388. /*
  389.  * Close down a transport socket
  390.  */
  391. static void
  392. xprt_close(struct rpc_xprt *xprt)
  393. {
  394. struct socket *sock = xprt->sock;
  395. struct sock *sk = xprt->inet;
  397. if (!sk)
  398. return;
  400. xprt->inet = NULL;
  401. xprt->sock = NULL;
  403. sk->user_data    = NULL;
  404. sk->data_ready   = xprt->old_data_ready;
  405. sk->state_change = xprt->old_state_change;
  406. sk->write_space  = xprt->old_write_space;
  408. xprt_disconnect(xprt);
  409. sk->no_check  = 0;
  411. sock_release(sock);
  412. /*
  413.  * TCP doesnt require the rpciod now - other things may
  414.  * but rpciod handles that not us.
  415.  */
  416. if(xprt->stream)
  417. rpciod_down();
  418. }
  420. /*
  421.  * Mark a transport as disconnected
  422.  */
  423. static void
  424. xprt_disconnect(struct rpc_xprt *xprt)
  425. {
  426. dprintk("RPC:      disconnected transport %pn", xprt);
  427. xprt_clear_connected(xprt);
  428. xprt_remove_pending(xprt);
  429. rpc_wake_up_status(&xprt->pending, -ENOTCONN);
  430. }
  432. /*
  433.  * Reconnect a broken TCP connection.
  434.  *
  435.  * Note: This cannot collide with the TCP reads, as both run from rpciod
  436.  */
  437. void
  438. xprt_reconnect(struct rpc_task *task)
  439. {
  440. struct rpc_xprt *xprt = task->tk_xprt;
  441. struct socket *sock = xprt->sock;
  442. struct sock *inet = xprt->inet;
  443. int status;
  445. dprintk("RPC: %4d xprt_reconnect %p connected %dn",
  446. task->tk_pid, xprt, xprt_connected(xprt));
  447. if (xprt->shutdown)
  448. return;
  450. if (!xprt->stream)
  451. return;
  453. if (!xprt->addr.sin_port) {
  454. task->tk_status = -EIO;
  455. return;
  456. }
  458. if (!xprt_lock_write(xprt, task))
  459. return;
  460. if (xprt_connected(xprt))
  461. goto out_write;
  463. status = -ENOTCONN;
  464. if (!inet) {
  465. /* Create an unconnected socket */
  466. if (!(sock = xprt_create_socket(xprt->prot, &xprt->timeout)))
  467. goto defer;
  468. xprt_bind_socket(xprt, sock);
  469. inet = sock->sk;
  470. }
  472. xprt_disconnect(xprt);
  474. /* Reset TCP record info */
  475. xprt->tcp_offset = 0;
  476. xprt->tcp_reclen = 0;
  477. xprt->tcp_copied = 0;
  478. xprt->tcp_more = 0;
  480. /* Now connect it asynchronously. */
  481. dprintk("RPC: %4d connecting new socketn", task->tk_pid);
  482. status = sock->ops->connect(sock, (struct sockaddr *) &xprt->addr,
  483. sizeof(xprt->addr), O_NONBLOCK);
  485. if (status < 0) {
  486. switch (status) {
  487. case -EALREADY:
  488. case -EINPROGRESS:
  489. status = 0;
  490. break;
  491. case -EISCONN:
  492. case -EPIPE:
  493. status = 0;
  494. xprt_close(xprt);
  495. goto defer;
  496. default:
  497. printk("RPC: TCP connect error %d!n", -status);
  498. xprt_close(xprt);
  499. goto defer;
  500. }
  502. dprintk("RPC: %4d connect status %d connected %dn",
  503. task->tk_pid, status, xprt_connected(xprt));
  505. spin_lock_bh(&xprt->sock_lock);
  506. if (!xprt_connected(xprt)) {
  507. task->tk_timeout = xprt->timeout.to_maxval;
  508. rpc_sleep_on(&xprt->sending, task, xprt_reconn_status, NULL);
  509. spin_unlock_bh(&xprt->sock_lock);
  510. return;
  511. }
  512. spin_unlock_bh(&xprt->sock_lock);
  513. }
  514. defer:
  515. if (status < 0) {
  516. rpc_delay(task, 5*HZ);
  517. task->tk_status = -ENOTCONN;
  518. }
  519.  out_write:
  520. xprt_release_write(xprt, task);
  521. }
  523. /*
  524.  * Reconnect timeout. We just mark the transport as not being in the
  525.  * process of reconnecting, and leave the rest to the upper layers.
  526.  */
  527. static void
  528. xprt_reconn_status(struct rpc_task *task)
  529. {
  530. struct rpc_xprt *xprt = task->tk_xprt;
  532. dprintk("RPC: %4d xprt_reconn_timeout %dn",
  533. task->tk_pid, task->tk_status);
  535. xprt_release_write(xprt, task);
  536. }
  538. /*
  539.  * Look up the RPC request corresponding to a reply, and then lock it.
  540.  */
  541. static inline struct rpc_rqst *
  542. xprt_lookup_rqst(struct rpc_xprt *xprt, u32 xid)
  543. {
  544. struct rpc_task *head, *task;
  545. struct rpc_rqst *req;
  546. int safe = 0;
  548. spin_lock_bh(&rpc_queue_lock);
  549. if ((head = xprt->pending.task) != NULL) {
  550. task = head;
  551. do {
  552. if ((req = task->tk_rqstp) && req->rq_xid == xid)
  553. goto out;
  554. task = task->tk_next;
  555. if (++safe > 100) {
  556. printk("xprt_lookup_rqst: loop in Q!n");
  557. goto out_bad;
  558. }
  559. } while (task != head);
  560. }
  561. dprintk("RPC:      unknown XID %08x in reply.n", xid);
  562.  out_bad:
  563. req = NULL;
  564.  out:
  565. if (req && !__rpc_lock_task(req->rq_task))
  566. req = NULL;
  567. spin_unlock_bh(&rpc_queue_lock);
  568. return req;
  569. }
  571. /*
  572.  * Complete reply received.
  573.  * The TCP code relies on us to remove the request from xprt->pending.
  574.  */
  575. static inline void
  576. xprt_complete_rqst(struct rpc_xprt *xprt, struct rpc_rqst *req, int copied)
  577. {
  578. struct rpc_task *task = req->rq_task;
  580. /* Adjust congestion window */
  581. xprt_adjust_cwnd(xprt, copied);
  583. #ifdef RPC_PROFILE
  584. /* Profile only reads for now */
  585. if (copied > 1024) {
  586. static unsigned long nextstat = 0;
  587. static unsigned long pkt_rtt = 0, pkt_len = 0, pkt_cnt = 0;
  589. pkt_cnt++;
  590. pkt_len += req->rq_slen + copied;
  591. pkt_rtt += jiffies - req->rq_xtime;
  592. if (time_before(nextstat, jiffies)) {
  593. printk("RPC: %lu %ld cwndn", jiffies, xprt->cwnd);
  594. printk("RPC: %ld %ld %ld %ld statn",
  595. jiffies, pkt_cnt, pkt_len, pkt_rtt);
  596. pkt_rtt = pkt_len = pkt_cnt = 0;
  597. nextstat = jiffies + 5 * HZ;
  598. }
  599. }
  600. #endif
  602. dprintk("RPC: %4d has input (%d bytes)n", task->tk_pid, copied);
  603. task->tk_status = copied;
  604. req->rq_received = 1;
  606. /* ... and wake up the process. */
  607. rpc_wake_up_task(task);
  608. return;
  609. }
  611. /*
  612.  * We have set things up such that we perform the checksum of the UDP
  613.  * packet in parallel with the copies into the RPC client iovec.  -DaveM
  614.  */
  615. static int csum_partial_copy_to_page_cache(struct iovec *iov,
  616.    struct sk_buff *skb,
  617.    int copied)
  618. {
  619. int offset = sizeof(struct udphdr);
  620. __u8 *cur_ptr = iov->iov_base;
  621. __kernel_size_t cur_len = iov->iov_len;
  622. unsigned int csum = skb->csum;
  623. int need_csum = (skb->ip_summed != CHECKSUM_UNNECESSARY);
  624. int slack = skb->len - copied - sizeof(struct udphdr);
  626. if (need_csum)
  627. csum = csum_partial(skb->data, sizeof(struct udphdr), csum);
  628. while (copied > 0) {
  629. if (cur_len) {
  630. int to_move = cur_len;
  631. if (to_move > copied)
  632. to_move = copied;
  633. if (need_csum) {
  634. unsigned int csum2;
  636. csum2 = skb_copy_and_csum_bits(skb, offset,
  637.        cur_ptr,
  638.        to_move, 0);
  639. csum = csum_block_add(csum, csum2, offset);
  640. } else
  641. skb_copy_bits(skb, offset, cur_ptr, to_move);
  642. offset += to_move;
  643. copied -= to_move;
  644. cur_ptr += to_move;
  645. cur_len -= to_move;
  646. }
  647. if (cur_len <= 0) {
  648. iov++;
  649. cur_len = iov->iov_len;
  650. cur_ptr = iov->iov_base;
  651. }
  652. }
  653. if (need_csum) {
  654. if (slack > 0) {
  655. unsigned int csum2;
  657. csum2 = skb_checksum(skb, offset, slack, 0);
  658. csum = csum_block_add(csum, csum2, offset);
  659. }
  660. if ((unsigned short)csum_fold(csum))
  661. return -1;
  662. }
  663. return 0;
  664. }
  666. /*
  667.  * Input handler for RPC replies. Called from a bottom half and hence
  668.  * atomic.
  669.  */
  670. static void
  671. udp_data_ready(struct sock *sk, int len)
  672. {
  673. struct rpc_task *task;
  674. struct rpc_xprt *xprt;
  675. struct rpc_rqst *rovr;
  676. struct sk_buff *skb;
  677. int err, repsize, copied;
  679. dprintk("RPC:      udp_data_ready...n");
  680. if (!(xprt = xprt_from_sock(sk))) {
  681. printk("RPC:      udp_data_ready request not found!n");
  682. goto out;
  683. }
  685. dprintk("RPC:      udp_data_ready client %pn", xprt);
  687. if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
  688. goto out;
  690. if (xprt->shutdown)
  691. goto dropit;
  693. repsize = skb->len - sizeof(struct udphdr);
  694. if (repsize < 4) {
  695. printk("RPC: impossible RPC reply size %d!n", repsize);
  696. goto dropit;
  697. }
  699. /* Look up and lock the request corresponding to the given XID */
  700. rovr = xprt_lookup_rqst(xprt, *(u32 *) (skb->h.raw + sizeof(struct udphdr)));
  701. if (!rovr)
  702. goto dropit;
  703. task = rovr->rq_task;
  705. dprintk("RPC: %4d received replyn", task->tk_pid);
  706. xprt_pktdump("packet data:",
  707.      (u32 *) (skb->h.raw+sizeof(struct udphdr)), repsize);
  709. if ((copied = rovr->rq_rlen) > repsize)
  710. copied = repsize;
  712. /* Suck it into the iovec, verify checksum if not done by hw. */
  713. if (csum_partial_copy_to_page_cache(rovr->rq_rvec, skb, copied))
  714. goto out_unlock;
  716. /* Something worked... */
  717. dst_confirm(skb->dst);
  719. xprt_complete_rqst(xprt, rovr, copied);
  721.  out_unlock:
  722. rpc_unlock_task(task);
  724.  dropit:
  725. skb_free_datagram(sk, skb);
  726.  out:
  727. if (sk->sleep && waitqueue_active(sk->sleep))
  728. wake_up_interruptible(sk->sleep);
  729. }
  731. /*
  732.  * TCP read fragment marker
  733.  */
  734. static inline int
  735. tcp_read_fraghdr(struct rpc_xprt *xprt)
  736. {
  737. struct iovec riov;
  738. int want, result;
  740. if (xprt->tcp_offset >= sizeof(xprt->tcp_recm))
  741. goto done;
  743. want = sizeof(xprt->tcp_recm) - xprt->tcp_offset;
  744. dprintk("RPC:      reading header (%d bytes)n", want);
  745. do {
  746. riov.iov_base = ((u8*) &xprt->tcp_recm) + xprt->tcp_offset;
  747. riov.iov_len  = want;
  748. result = xprt_recvmsg(xprt, &riov, 1, want, 0);
  749. if (result < 0)
  750. return result;
  751. xprt->tcp_offset += result;
  752. want -= result;
  753. } while (want);
  755. /* Get the record length and mask out the last fragment bit */
  756. xprt->tcp_reclen = ntohl(xprt->tcp_recm);
  757. xprt->tcp_more = (xprt->tcp_reclen & 0x80000000) ? 0 : 1;
  758. xprt->tcp_reclen &= 0x7fffffff;
  760. dprintk("RPC:      New record reclen %d morefrags %dn",
  761.    xprt->tcp_reclen, xprt->tcp_more);
  762.  done:
  763. return xprt->tcp_reclen + sizeof(xprt->tcp_recm) - xprt->tcp_offset;
  764. }
  766. /*
  767.  * TCP read xid
  768.  */
  769. static inline int
  770. tcp_read_xid(struct rpc_xprt *xprt, int avail)
  771. {
  772. struct iovec riov;
  773. int want, result;
  775. if (xprt->tcp_copied >= sizeof(xprt->tcp_xid) || !avail)
  776. goto done;
  777. want = min_t(unsigned int, sizeof(xprt->tcp_xid) - xprt->tcp_copied, avail);
  778. do {
  779. dprintk("RPC:      reading xid (%d bytes)n", want);
  780. riov.iov_base = ((u8*) &xprt->tcp_xid) + xprt->tcp_copied;
  781. riov.iov_len  = want;
  782. result = xprt_recvmsg(xprt, &riov, 1, want, 0);
  783. if (result < 0)
  784. return result;
  785. xprt->tcp_copied += result;
  786. xprt->tcp_offset += result;
  787. want  -= result;
  788. avail -= result;
  789. } while (want);
  790.  done:
  791. return avail;
  792. }
  794. /*
  795.  * TCP read and complete request
  796.  */
  797. static inline int
  798. tcp_read_request(struct rpc_xprt *xprt, struct rpc_rqst *req, int avail)
  799. {
  800. int want, result;
  802. if (req->rq_rlen <= xprt->tcp_copied || !avail)
  803. goto done;
  804. want = min_t(unsigned int, req->rq_rlen - xprt->tcp_copied, avail);
  805. do {
  806. dprintk("RPC: %4d TCP receiving %d bytesn",
  807. req->rq_task->tk_pid, want);
  809. result = xprt_recvmsg(xprt, req->rq_rvec, req->rq_rnr, want, xprt->tcp_copied);
  810. if (result < 0)
  811. return result;
  812. xprt->tcp_copied += result;
  813. xprt->tcp_offset += result;
  814. avail  -= result;
  815. want   -= result;
  816. } while (want);
  818.  done:
  819. if (req->rq_rlen > xprt->tcp_copied && xprt->tcp_more)
  820. return avail;
  821. dprintk("RPC: %4d received reply completen", req->rq_task->tk_pid);
  822. xprt_complete_rqst(xprt, req, xprt->tcp_copied);
  824. return avail;
  825. }
  827. /*
  828.  * TCP discard extra bytes from a short read
  829.  */
  830. static inline int
  831. tcp_read_discard(struct rpc_xprt *xprt, int avail)
  832. {
  833. struct iovec riov;
  834. static u8 dummy[64];
  835. int want, result = 0;
  837. while (avail) {
  838. want = min_t(unsigned int, avail, sizeof(dummy));
  839. riov.iov_base = dummy;
  840. riov.iov_len  = want;
  841. dprintk("RPC:      TCP skipping %d bytesn", want);
  842. result = xprt_recvmsg(xprt, &riov, 1, want, 0);
  843. if (result < 0)
  844. return result;
  845. xprt->tcp_offset += result;
  846. avail  -= result;
  847. }
  848. return avail;
  849. }
  851. /*
  852.  * TCP record receive routine
  853.  * This is not the most efficient code since we call recvfrom thrice--
  854.  * first receiving the record marker, then the XID, then the data.
  855.  * 
  856.  * The optimal solution would be a RPC support in the TCP layer, which
  857.  * would gather all data up to the next record marker and then pass us
  858.  * the list of all TCP segments ready to be copied.
  859.  */
  860. static int
  861. tcp_input_record(struct rpc_xprt *xprt)
  862. {
  863. struct rpc_rqst *req = NULL;
  864. struct rpc_task *task = NULL;
  865. int avail, result;
  867. dprintk("RPC:      tcp_input_recordn");
  869. if (xprt->shutdown)
  870. return -EIO;
  871. if (!xprt_connected(xprt))
  872. return -ENOTCONN;
  874. /* Read in a new fragment marker if necessary */
  875. /* Can we ever really expect to get completely empty fragments? */
  876. if ((result = tcp_read_fraghdr(xprt)) < 0)
  877. return result;
  878. avail = result;
  880. /* Read in the xid if necessary */
  881. if ((result = tcp_read_xid(xprt, avail)) < 0)
  882. return result;
  883. if (!(avail = result))
  884. goto out_ok;
  886. /* Find and lock the request corresponding to this xid */
  887. req = xprt_lookup_rqst(xprt, xprt->tcp_xid);
  888. if (req) {
  889. task = req->rq_task;
  890. /* Read in the request data */
  891. result = tcp_read_request(xprt,  req, avail);
  892. rpc_unlock_task(task);
  893. if (result < 0)
  894. return result;
  895. avail = result;
  896. }
  898. /* Skip over any trailing bytes on short reads */
  899. if ((result = tcp_read_discard(xprt, avail)) < 0)
  900. return result;
  902.  out_ok:
  903. dprintk("RPC:      tcp_input_record done (off %d reclen %d copied %d)n",
  904. xprt->tcp_offset, xprt->tcp_reclen, xprt->tcp_copied);
  905. result = xprt->tcp_reclen;
  906. xprt->tcp_reclen = 0;
  907. xprt->tcp_offset = 0;
  908. if (!xprt->tcp_more)
  909. xprt->tcp_copied = 0;
  910. return result;
  911. }
  913. /*
  914.  * TCP task queue stuff
  915.  */
  916. LIST_HEAD(rpc_xprt_pending); /* List of xprts having pending tcp requests */
  918. static inline
  919. void tcp_rpciod_queue(void)
  920. {
  921. rpciod_wake_up();
  922. }
  924. int xprt_tcp_pending(void)
  925. {
  926. int retval;
  928. spin_lock_bh(&rpc_queue_lock);
  929. retval = !list_empty(&rpc_xprt_pending);
  930. spin_unlock_bh(&rpc_queue_lock);
  931. return retval;
  932. }
  934. static inline
  935. void xprt_append_pending(struct rpc_xprt *xprt)
  936. {
  937. spin_lock_bh(&rpc_queue_lock);
  938. if (list_empty(&xprt->rx_pending)) {
  939. list_add(&xprt->rx_pending, rpc_xprt_pending.prev);
  940. dprintk("RPC:     xprt queue %pn", xprt);
  941. tcp_rpciod_queue();
  942. }
  943. spin_unlock_bh(&rpc_queue_lock);
  944. }
  946. static
  947. void xprt_remove_pending(struct rpc_xprt *xprt)
  948. {
  949. spin_lock_bh(&rpc_queue_lock);
  950. if (!list_empty(&xprt->rx_pending)) {
  951. list_del(&xprt->rx_pending);
  952. INIT_LIST_HEAD(&xprt->rx_pending);
  953. }
  954. spin_unlock_bh(&rpc_queue_lock);
  955. }
  957. static inline
  958. struct rpc_xprt *xprt_remove_pending_next(void)
  959. {
  960. struct rpc_xprt *xprt = NULL;
  962. spin_lock_bh(&rpc_queue_lock);
  963. if (!list_empty(&rpc_xprt_pending)) {
  964. xprt = list_entry(rpc_xprt_pending.next, struct rpc_xprt, rx_pending);
  965. list_del(&xprt->rx_pending);
  966. INIT_LIST_HEAD(&xprt->rx_pending);
  967. }
  968. spin_unlock_bh(&rpc_queue_lock);
  969. return xprt;
  970. }
  972. /*
  973.  * This is protected from tcp_data_ready and the stack as its run
  974.  * inside of the RPC I/O daemon
  975.  */
  976. void
  977. __rpciod_tcp_dispatcher(void)
  978. {
  979. struct rpc_xprt *xprt;
  980. int safe_retry = 0, result;
  982. dprintk("rpciod_tcp_dispatcher: Queue Runningn");
  984. /*
  985.  * Empty each pending socket
  986.  */
  987. while ((xprt = xprt_remove_pending_next()) != NULL) {
  988. dprintk("rpciod_tcp_dispatcher: Processing %pn", xprt);
  990. do {
  991. result = tcp_input_record(xprt);
  992. } while (result >= 0);
  994. if (safe_retry++ > 200) {
  995. schedule();
  996. safe_retry = 0;
  997. }
  998. }
  999. }
  1001. /*
  1002.  * data_ready callback for TCP. We can't just jump into the
  1003.  * tcp recvmsg functions inside of the network receive bh or
  1004.  *  bad things occur. We queue it to pick up after networking
  1005.  * is done.
  1006.  */
  1007.  
  1008. static void tcp_data_ready(struct sock *sk, int len)
  1009. {
  1010. struct rpc_xprt *xprt;
  1012. dprintk("RPC:      tcp_data_ready...n");
  1013. if (!(xprt = xprt_from_sock(sk)))
  1014. {
  1015. printk("Not a socket with xprt %pn", sk);
  1016. goto out;
  1017. }
  1019. if (xprt->shutdown)
  1020. goto out;
  1022. xprt_append_pending(xprt);
  1024. dprintk("RPC:      tcp_data_ready client %pn", xprt);
  1025. dprintk("RPC:      state %x conn %d dead %d zapped %dn",
  1026. sk->state, xprt_connected(xprt),
  1027. sk->dead, sk->zapped);
  1028.  out:
  1029. if (sk->sleep && waitqueue_active(sk->sleep))
  1030. wake_up_interruptible(sk->sleep);
  1031. }
  1034. static void
  1035. tcp_state_change(struct sock *sk)
  1036. {
  1037. struct rpc_xprt *xprt;
  1039. if (!(xprt = xprt_from_sock(sk)))
  1040. goto out;
  1041. dprintk("RPC:      tcp_state_change client %p...n", xprt);
  1042. dprintk("RPC:      state %x conn %d dead %d zapped %dn",
  1043. sk->state, xprt_connected(xprt),
  1044. sk->dead, sk->zapped);
  1046. switch (sk->state) {
  1047. case TCP_ESTABLISHED:
  1048. if (xprt_test_and_set_connected(xprt))
  1049. break;
  1050. spin_lock(&xprt->sock_lock);
  1051. if (xprt->snd_task && xprt->snd_task->tk_rpcwait == &xprt->sending)
  1052. rpc_wake_up_task(xprt->snd_task);
  1053. spin_unlock(&xprt->sock_lock);
  1054. break;
  1055. case TCP_SYN_SENT:
  1056. case TCP_SYN_RECV:
  1057. break;
  1058. default:
  1059. xprt_disconnect(xprt);
  1060. break;
  1061. }
  1062.  out:
  1063. if (sk->sleep && waitqueue_active(sk->sleep))
  1064. wake_up_interruptible_all(sk->sleep);
  1065. }
  1067. /*
  1068.  * The following 2 routines allow a task to sleep while socket memory is
  1069.  * low.
  1070.  */
  1071. static void
  1072. tcp_write_space(struct sock *sk)
  1073. {
  1074. struct rpc_xprt *xprt;
  1075. struct socket *sock;
  1077. if (!(xprt = xprt_from_sock(sk)) || !(sock = sk->socket))
  1078. return;
  1079. if (xprt->shutdown)
  1080. return;
  1082. /* Wait until we have enough socket memory */
  1083. if (!sock_writeable(sk))
  1084. return;
  1086. if (!xprt_test_and_set_wspace(xprt)) {
  1087. spin_lock(&xprt->sock_lock);
  1088. if (xprt->snd_task && xprt->snd_task->tk_rpcwait == &xprt->sending)
  1089. rpc_wake_up_task(xprt->snd_task);
  1090. spin_unlock(&xprt->sock_lock);
  1091. }
  1093. if (test_bit(SOCK_NOSPACE, &sock->flags)) {
  1094. if (sk->sleep && waitqueue_active(sk->sleep)) {
  1095. clear_bit(SOCK_NOSPACE, &sock->flags);
  1096. wake_up_interruptible(sk->sleep);
  1097. }
  1098. }
  1099. }
  1101. static void
  1102. udp_write_space(struct sock *sk)
  1103. {
  1104. struct rpc_xprt *xprt;
  1106. if (!(xprt = xprt_from_sock(sk)))
  1107. return;
  1108. if (xprt->shutdown)
  1109. return;
  1112. /* Wait until we have enough socket memory */
  1113. if (sock_wspace(sk) < min_t(int, sk->sndbuf,XPRT_MIN_WRITE_SPACE))
  1114. return;
  1116. if (!xprt_test_and_set_wspace(xprt)) {
  1117. spin_lock(&xprt->sock_lock);
  1118. if (xprt->snd_task && xprt->snd_task->tk_rpcwait == &xprt->sending)
  1119. rpc_wake_up_task(xprt->snd_task);
  1120. spin_unlock(&xprt->sock_lock);
  1121. }
  1123. if (sk->sleep && waitqueue_active(sk->sleep))
  1124. wake_up_interruptible(sk->sleep);
  1125. }
  1127. /*
  1128.  * RPC receive timeout handler.
  1129.  */
  1130. static void
  1131. xprt_timer(struct rpc_task *task)
  1132. {
  1133. struct rpc_rqst *req = task->tk_rqstp;
  1135. if (req)
  1136. xprt_adjust_cwnd(task->tk_xprt, -ETIMEDOUT);
  1138. dprintk("RPC: %4d xprt_timer (%s request)n",
  1139. task->tk_pid, req ? "pending" : "backlogged");
  1141. task->tk_status  = -ETIMEDOUT;
  1142. task->tk_timeout = 0;
  1143. rpc_wake_up_task(task);
  1144. }
  1146. /*
  1147.  * Place the actual RPC call.
  1148.  * We have to copy the iovec because sendmsg fiddles with its contents.
  1149.  */
  1150. void
  1151. xprt_transmit(struct rpc_task *task)
  1152. {
  1153. struct rpc_rqst *req = task->tk_rqstp;
  1154. struct rpc_xprt *xprt = req->rq_xprt;
  1156. dprintk("RPC: %4d xprt_transmit(%x)n", task->tk_pid, 
  1157. *(u32 *)(req->rq_svec[0].iov_base));
  1159. if (xprt->shutdown)
  1160. task->tk_status = -EIO;
  1162. if (!xprt_connected(xprt))
  1163. task->tk_status = -ENOTCONN;
  1165. if (task->tk_status < 0)
  1166. return;
  1168. if (task->tk_rpcwait)
  1169. rpc_remove_wait_queue(task);
  1171. /* set up everything as needed. */
  1172. /* Write the record marker */
  1173. if (xprt->stream) {
  1174. u32 *marker = req->rq_svec[0].iov_base;
  1176. *marker = htonl(0x80000000|(req->rq_slen-sizeof(*marker)));
  1177. }
  1179. if (!xprt_lock_write(xprt, task))
  1180. return;
  1182. #ifdef RPC_PROFILE
  1183. req->rq_xtime = jiffies;
  1184. #endif
  1185. do_xprt_transmit(task);
  1186. }
  1188. static void
  1189. do_xprt_transmit(struct rpc_task *task)
  1190. {
  1191. struct rpc_rqst *req = task->tk_rqstp;
  1192. struct rpc_xprt *xprt = req->rq_xprt;
  1193. int status, retry = 0;
  1196. /* For fast networks/servers we have to put the request on
  1197.  * the pending list now:
  1198.  * Note that we don't want the task timing out during the
  1199.  * call to xprt_sendmsg(), so we initially disable the timeout,
  1200.  * and then reset it later...
  1201.  */
  1202. xprt_receive(task);
  1204. /* Continue transmitting the packet/record. We must be careful
  1205.  * to cope with writespace callbacks arriving _after_ we have
  1206.  * called xprt_sendmsg().
  1207.  */
  1208. while (1) {
  1209. xprt_clear_wspace(xprt);
  1210. status = xprt_sendmsg(xprt, req);
  1212. if (status < 0)
  1213. break;
  1215. if (xprt->stream) {
  1216. req->rq_bytes_sent += status;
  1218. if (req->rq_bytes_sent >= req->rq_slen)
  1219. goto out_receive;
  1220. } else {
  1221. if (status >= req->rq_slen)
  1222. goto out_receive;
  1223. status = -ENOMEM;
  1224. break;
  1225. }
  1227. dprintk("RPC: %4d xmit incomplete (%d left of %d)n",
  1228. task->tk_pid, req->rq_slen - req->rq_bytes_sent,
  1229. req->rq_slen);
  1231. status = -EAGAIN;
  1232. if (retry++ > 50)
  1233. break;
  1234. }
  1235. rpc_unlock_task(task);
  1237. /* Note: at this point, task->tk_sleeping has not yet been set,
  1238.  *  hence there is no danger of the waking up task being put on
  1239.  *  schedq, and being picked up by a parallel run of rpciod().
  1240.  */
  1241. rpc_wake_up_task(task);
  1242. if (!RPC_IS_RUNNING(task))
  1243. goto out_release;
  1244. if (req->rq_received)
  1245. goto out_release;
  1247. task->tk_status = status;
  1249. switch (status) {
  1250. case -ENOMEM:
  1251. /* Protect against (udp|tcp)_write_space */
  1252. spin_lock_bh(&xprt->sock_lock);
  1253. if (!xprt_wspace(xprt)) {
  1254. task->tk_timeout = req->rq_timeout.to_current;
  1255. rpc_sleep_on(&xprt->sending, task, NULL, NULL);
  1256. }
  1257. spin_unlock_bh(&xprt->sock_lock);
  1258. return;
  1259. case -EAGAIN:
  1260. /* Keep holding the socket if it is blocked */
  1261. rpc_delay(task, HZ>>4);
  1262. return;
  1263. case -ECONNREFUSED:
  1264. case -ENOTCONN:
  1265. if (!xprt->stream)
  1266. return;
  1267. default:
  1268. if (xprt->stream)
  1269. xprt_disconnect(xprt);
  1270. req->rq_bytes_sent = 0;
  1271. goto out_release;
  1272. }
  1274.  out_receive:
  1275. dprintk("RPC: %4d xmit completen", task->tk_pid);
  1276. /* Set the task's receive timeout value */
  1277. task->tk_timeout = req->rq_timeout.to_current;
  1278. rpc_add_timer(task, xprt_timer);
  1279. rpc_unlock_task(task);
  1280.  out_release:
  1281. xprt_release_write(xprt, task);
  1282. }
  1284. /*
  1285.  * Queue the task for a reply to our call.
  1286.  * When the callback is invoked, the congestion window should have
  1287.  * been updated already.
  1288.  */
  1289. void
  1290. xprt_receive(struct rpc_task *task)
  1291. {
  1292. struct rpc_rqst *req = task->tk_rqstp;
  1293. struct rpc_xprt *xprt = req->rq_xprt;
  1295. dprintk("RPC: %4d xprt_receiven", task->tk_pid);
  1297. req->rq_received = 0;
  1298. task->tk_timeout = 0;
  1299. rpc_sleep_locked(&xprt->pending, task, NULL, NULL);
  1300. }
  1302. /*
  1303.  * Reserve an RPC call slot.
  1304.  */
  1305. int
  1306. xprt_reserve(struct rpc_task *task)
  1307. {
  1308. struct rpc_xprt *xprt = task->tk_xprt;
  1310. /* We already have an initialized request. */
  1311. if (task->tk_rqstp)
  1312. return 0;
  1314. dprintk("RPC: %4d xprt_reserve cong = %ld cwnd = %ldn",
  1315. task->tk_pid, xprt->cong, xprt->cwnd);
  1316. spin_lock_bh(&xprt->xprt_lock);
  1317. xprt_reserve_status(task);
  1318. if (task->tk_rqstp) {
  1319. task->tk_timeout = 0;
  1320. } else if (!task->tk_timeout) {
  1321. task->tk_status = -ENOBUFS;
  1322. } else {
  1323. dprintk("RPC:      xprt_reserve waiting on backlogn");
  1324. task->tk_status = -EAGAIN;
  1325. rpc_sleep_on(&xprt->backlog, task, NULL, NULL);
  1326. }
  1327. spin_unlock_bh(&xprt->xprt_lock);
  1328. dprintk("RPC: %4d xprt_reserve returns %dn",
  1329. task->tk_pid, task->tk_status);
  1330. return task->tk_status;
  1331. }
  1333. /*
  1334.  * Reservation callback
  1335.  */
  1336. static void
  1337. xprt_reserve_status(struct rpc_task *task)
  1338. {
  1339. struct rpc_xprt *xprt = task->tk_xprt;
  1340. struct rpc_rqst *req;
  1342. if (xprt->shutdown) {
  1343. task->tk_status = -EIO;
  1344. } else if (task->tk_status < 0) {
  1345. /* NOP */
  1346. } else if (task->tk_rqstp) {
  1347. /* We've already been given a request slot: NOP */
  1348. } else {
  1349. if (RPCXPRT_CONGESTED(xprt) || !(req = xprt->free))
  1350. goto out_nofree;
  1351. /* OK: There's room for us. Grab a free slot and bump
  1352.  * congestion value */
  1353. xprt->free     = req->rq_next;
  1354. req->rq_next   = NULL;
  1355. xprt->cong    += RPC_CWNDSCALE;
  1356. task->tk_rqstp = req;
  1357. xprt_request_init(task, xprt);
  1359. if (xprt->free)
  1360. xprt_clear_backlog(xprt);
  1361. }
  1363. return;
  1365. out_nofree:
  1366. task->tk_status = -EAGAIN;
  1367. }
  1369. /*
  1370.  * Initialize RPC request
  1371.  */
  1372. static void
  1373. xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
  1374. {
  1375. struct rpc_rqst *req = task->tk_rqstp;
  1376. static u32 xid = 0;
  1378. if (!xid)
  1379. xid = CURRENT_TIME << 12;
  1381. dprintk("RPC: %4d reserved req %p xid %08xn", task->tk_pid, req, xid);
  1382. task->tk_status = 0;
  1383. req->rq_timeout = xprt->timeout;
  1384. req->rq_task = task;
  1385. req->rq_xprt    = xprt;
  1386. req->rq_xid     = xid++;
  1387. if (!xid)
  1388. xid++;
  1389. }
  1391. /*
  1392.  * Release an RPC call slot
  1393.  */
  1394. void
  1395. xprt_release(struct rpc_task *task)
  1396. {
  1397. struct rpc_xprt *xprt = task->tk_xprt;
  1398. struct rpc_rqst *req;
  1400. if (xprt->snd_task == task) {
  1401. if (xprt->stream)
  1402. xprt_disconnect(xprt);
  1403. xprt_release_write(xprt, task);
  1404. }
  1405. if (!(req = task->tk_rqstp))
  1406. return;
  1407. task->tk_rqstp = NULL;
  1408. memset(req, 0, sizeof(*req)); /* mark unused */
  1410. dprintk("RPC: %4d release request %pn", task->tk_pid, req);
  1412. spin_lock_bh(&xprt->xprt_lock);
  1413. req->rq_next = xprt->free;
  1414. xprt->free   = req;
  1416. /* Decrease congestion value. */
  1417. xprt->cong -= RPC_CWNDSCALE;
  1419. xprt_clear_backlog(xprt);
  1420. spin_unlock_bh(&xprt->xprt_lock);
  1421. }
  1423. /*
  1424.  * Set default timeout parameters
  1425.  */
  1426. void
  1427. xprt_default_timeout(struct rpc_timeout *to, int proto)
  1428. {
  1429. if (proto == IPPROTO_UDP)
  1430. xprt_set_timeout(to, 5,  5 * HZ);
  1431. else
  1432. xprt_set_timeout(to, 5, 60 * HZ);
  1433. }
  1435. /*
  1436.  * Set constant timeout
  1437.  */
  1438. void
  1439. xprt_set_timeout(struct rpc_timeout *to, unsigned int retr, unsigned long incr)
  1440. {
  1441. to->to_current   = 
  1442. to->to_initval   = 
  1443. to->to_increment = incr;
  1444. to->to_maxval    = incr * retr;
  1445. to->to_resrvval  = incr * retr;
  1446. to->to_retries   = retr;
  1447. to->to_exponential = 0;
  1448. }
  1450. /*
  1451.  * Initialize an RPC client
  1452.  */
  1453. static struct rpc_xprt *
  1454. xprt_setup(struct socket *sock, int proto,
  1455. struct sockaddr_in *ap, struct rpc_timeout *to)
  1456. {
  1457. struct rpc_xprt *xprt;
  1458. struct rpc_rqst *req;
  1459. int i;
  1461. dprintk("RPC:      setting up %s transport...n",
  1462. proto == IPPROTO_UDP? "UDP" : "TCP");
  1464. if ((xprt = kmalloc(sizeof(struct rpc_xprt), GFP_KERNEL)) == NULL)
  1465. return NULL;
  1466. memset(xprt, 0, sizeof(*xprt)); /* Nnnngh! */
  1468. xprt->addr = *ap;
  1469. xprt->prot = proto;
  1470. xprt->stream = (proto == IPPROTO_TCP)? 1 : 0;
  1471. if (xprt->stream) {
  1472. xprt->cwnd = RPC_MAXCWND;
  1473. xprt->nocong = 1;
  1474. } else
  1475. xprt->cwnd = RPC_INITCWND;
  1476. xprt->congtime = jiffies;
  1477. spin_lock_init(&xprt->sock_lock);
  1478. spin_lock_init(&xprt->xprt_lock);
  1479. init_waitqueue_head(&xprt->cong_wait);
  1481. /* Set timeout parameters */
  1482. if (to) {
  1483. xprt->timeout = *to;
  1484. xprt->timeout.to_current = to->to_initval;
  1485. xprt->timeout.to_resrvval = to->to_maxval << 1;
  1486. } else
  1487. xprt_default_timeout(&xprt->timeout, xprt->prot);
  1489. xprt->pending = RPC_INIT_WAITQ("xprt_pending");
  1490. xprt->sending = RPC_INIT_WAITQ("xprt_sending");
  1491. xprt->backlog = RPC_INIT_WAITQ("xprt_backlog");
  1493. /* initialize free list */
  1494. for (i = 0, req = xprt->slot; i < RPC_MAXREQS-1; i++, req++)
  1495. req->rq_next = req + 1;
  1496. req->rq_next = NULL;
  1497. xprt->free = xprt->slot;
  1499. INIT_LIST_HEAD(&xprt->rx_pending);
  1501. dprintk("RPC:      created transport %pn", xprt);
  1503. xprt_bind_socket(xprt, sock);
  1504. return xprt;
  1505. }
  1507. /*
  1508.  * Bind to a reserved port
  1509.  */
  1510. static inline int
  1511. xprt_bindresvport(struct socket *sock)
  1512. {
  1513. struct sockaddr_in myaddr;
  1514. int err, port;
  1516. memset(&myaddr, 0, sizeof(myaddr));
  1517. myaddr.sin_family = AF_INET;
  1518. port = 800;
  1519. do {
  1520. myaddr.sin_port = htons(port);
  1521. err = sock->ops->bind(sock, (struct sockaddr *) &myaddr,
  1522. sizeof(myaddr));
  1523. } while (err == -EADDRINUSE && --port > 0);
  1525. if (err < 0)
  1526. printk("RPC: Can't bind to reserved port (%d).n", -err);
  1528. return err;
  1529. }
  1531. static int 
  1532. xprt_bind_socket(struct rpc_xprt *xprt, struct socket *sock)
  1533. {
  1534. struct sock *sk = sock->sk;
  1536. if (xprt->inet)
  1537. return -EBUSY;
  1539. sk->user_data = xprt;
  1540. xprt->old_data_ready = sk->data_ready;
  1541. xprt->old_state_change = sk->state_change;
  1542. xprt->old_write_space = sk->write_space;
  1543. if (xprt->prot == IPPROTO_UDP) {
  1544. sk->data_ready = udp_data_ready;
  1545. sk->write_space = udp_write_space;
  1546. sk->no_check = UDP_CSUM_NORCV;
  1547. xprt_set_connected(xprt);
  1548. } else {
  1549. sk->data_ready = tcp_data_ready;
  1550. sk->state_change = tcp_state_change;
  1551. sk->write_space = tcp_write_space;
  1552. xprt_clear_connected(xprt);
  1553. }
  1555. /* Reset to new socket */
  1556. xprt->sock = sock;
  1557. xprt->inet = sk;
  1558. /*
  1559.  * TCP requires the rpc I/O daemon is present
  1560.  */
  1561. if(xprt->stream)
  1562. rpciod_up();
  1564. return 0;
  1565. }
  1567. /*
  1568.  * Create a client socket given the protocol and peer address.
  1569.  */
  1570. static struct socket *
  1571. xprt_create_socket(int proto, struct rpc_timeout *to)
  1572. {
  1573. struct socket *sock;
  1574. int type, err;
  1576. dprintk("RPC:      xprt_create_socket(%s %d)n",
  1577.    (proto == IPPROTO_UDP)? "udp" : "tcp", proto);
  1579. type = (proto == IPPROTO_UDP)? SOCK_DGRAM : SOCK_STREAM;
  1581. if ((err = sock_create(PF_INET, type, proto, &sock)) < 0) {
  1582. printk("RPC: can't create socket (%d).n", -err);
  1583. goto failed;
  1584. }
  1586. /* If the caller has the capability, bind to a reserved port */
  1587. if (capable(CAP_NET_BIND_SERVICE) && xprt_bindresvport(sock) < 0)
  1588. goto failed;
  1590. return sock;
  1592. failed:
  1593. sock_release(sock);
  1594. return NULL;
  1595. }
  1597. /*
  1598.  * Create an RPC client transport given the protocol and peer address.
  1599.  */
  1600. struct rpc_xprt *
  1601. xprt_create_proto(int proto, struct sockaddr_in *sap, struct rpc_timeout *to)
  1602. {
  1603. struct socket *sock;
  1604. struct rpc_xprt *xprt;
  1606. dprintk("RPC:      xprt_create_proto calledn");
  1608. if (!(sock = xprt_create_socket(proto, to)))
  1609. return NULL;
  1611. if (!(xprt = xprt_setup(sock, proto, sap, to)))
  1612. sock_release(sock);
  1614. return xprt;
  1615. }
  1617. /*
  1618.  * Prepare for transport shutdown.
  1619.  */
  1620. void
  1621. xprt_shutdown(struct rpc_xprt *xprt)
  1622. {
  1623. xprt->shutdown = 1;
  1624. rpc_wake_up(&xprt->sending);
  1625. rpc_wake_up(&xprt->pending);
  1626. rpc_wake_up(&xprt->backlog);
  1627. if (waitqueue_active(&xprt->cong_wait))
  1628. wake_up(&xprt->cong_wait);
  1629. }
  1631. /*
  1632.  * Clear the xprt backlog queue
  1633.  */
  1634. int
  1635. xprt_clear_backlog(struct rpc_xprt *xprt) {
  1636. if (RPCXPRT_CONGESTED(xprt))
  1637. return 0;
  1638. rpc_wake_up_next(&xprt->backlog);
  1639. if (waitqueue_active(&xprt->cong_wait))
  1640. wake_up(&xprt->cong_wait);
  1641. return 1;
  1642. }
  1644. /*
  1645.  * Destroy an RPC transport, killing off all requests.
  1646.  */
  1647. int
  1648. xprt_destroy(struct rpc_xprt *xprt)
  1649. {
  1650. dprintk("RPC:      destroying transport %pn", xprt);
  1651. xprt_shutdown(xprt);
  1652. xprt_close(xprt);
  1653. kfree(xprt);
  1655. return 0;
  1656. }