xprtsock.c 43 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657
  1. /*
  2. * linux/net/sunrpc/xprtsock.c
  3. *
  4. * Client-side transport implementation for sockets.
  5. *
  6. * TCP callback races fixes (C) 1998 Red Hat Software <alan@redhat.com>
  7. * TCP send fixes (C) 1998 Red Hat Software <alan@redhat.com>
  8. * TCP NFS related read + write fixes
  9. * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
  10. *
  11. * Rewrite of larges part of the code in order to stabilize TCP stuff.
  12. * Fix behaviour when socket buffer is full.
  13. * (C) 1999 Trond Myklebust <trond.myklebust@fys.uio.no>
  14. *
  15. * IP socket transport implementation, (C) 2005 Chuck Lever <cel@netapp.com>
  16. */
  17. #include <linux/types.h>
  18. #include <linux/slab.h>
  19. #include <linux/capability.h>
  20. #include <linux/pagemap.h>
  21. #include <linux/errno.h>
  22. #include <linux/socket.h>
  23. #include <linux/in.h>
  24. #include <linux/net.h>
  25. #include <linux/mm.h>
  26. #include <linux/udp.h>
  27. #include <linux/tcp.h>
  28. #include <linux/sunrpc/clnt.h>
  29. #include <linux/sunrpc/sched.h>
  30. #include <linux/file.h>
  31. #include <net/sock.h>
  32. #include <net/checksum.h>
  33. #include <net/udp.h>
  34. #include <net/tcp.h>
  35. /*
  36. * xprtsock tunables
  37. */
  38. unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
  39. unsigned int xprt_tcp_slot_table_entries = RPC_DEF_SLOT_TABLE;
  40. unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
  41. unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
  42. /*
  43. * We can register our own files under /proc/sys/sunrpc by
  44. * calling register_sysctl_table() again. The files in that
  45. * directory become the union of all files registered there.
  46. *
  47. * We simply need to make sure that we don't collide with
  48. * someone else's file names!
  49. */
  50. #ifdef RPC_DEBUG
  51. static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
  52. static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
  53. static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
  54. static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
  55. static struct ctl_table_header *sunrpc_table_header;
  56. /*
  57. * FIXME: changing the UDP slot table size should also resize the UDP
  58. * socket buffers for existing UDP transports
  59. */
  60. static ctl_table xs_tunables_table[] = {
  61. {
  62. .ctl_name = CTL_SLOTTABLE_UDP,
  63. .procname = "udp_slot_table_entries",
  64. .data = &xprt_udp_slot_table_entries,
  65. .maxlen = sizeof(unsigned int),
  66. .mode = 0644,
  67. .proc_handler = &proc_dointvec_minmax,
  68. .strategy = &sysctl_intvec,
  69. .extra1 = &min_slot_table_size,
  70. .extra2 = &max_slot_table_size
  71. },
  72. {
  73. .ctl_name = CTL_SLOTTABLE_TCP,
  74. .procname = "tcp_slot_table_entries",
  75. .data = &xprt_tcp_slot_table_entries,
  76. .maxlen = sizeof(unsigned int),
  77. .mode = 0644,
  78. .proc_handler = &proc_dointvec_minmax,
  79. .strategy = &sysctl_intvec,
  80. .extra1 = &min_slot_table_size,
  81. .extra2 = &max_slot_table_size
  82. },
  83. {
  84. .ctl_name = CTL_MIN_RESVPORT,
  85. .procname = "min_resvport",
  86. .data = &xprt_min_resvport,
  87. .maxlen = sizeof(unsigned int),
  88. .mode = 0644,
  89. .proc_handler = &proc_dointvec_minmax,
  90. .strategy = &sysctl_intvec,
  91. .extra1 = &xprt_min_resvport_limit,
  92. .extra2 = &xprt_max_resvport_limit
  93. },
  94. {
  95. .ctl_name = CTL_MAX_RESVPORT,
  96. .procname = "max_resvport",
  97. .data = &xprt_max_resvport,
  98. .maxlen = sizeof(unsigned int),
  99. .mode = 0644,
  100. .proc_handler = &proc_dointvec_minmax,
  101. .strategy = &sysctl_intvec,
  102. .extra1 = &xprt_min_resvport_limit,
  103. .extra2 = &xprt_max_resvport_limit
  104. },
  105. {
  106. .ctl_name = 0,
  107. },
  108. };
  109. static ctl_table sunrpc_table[] = {
  110. {
  111. .ctl_name = CTL_SUNRPC,
  112. .procname = "sunrpc",
  113. .mode = 0555,
  114. .child = xs_tunables_table
  115. },
  116. {
  117. .ctl_name = 0,
  118. },
  119. };
  120. #endif
  121. /*
  122. * How many times to try sending a request on a socket before waiting
  123. * for the socket buffer to clear.
  124. */
  125. #define XS_SENDMSG_RETRY (10U)
  126. /*
  127. * Time out for an RPC UDP socket connect. UDP socket connects are
  128. * synchronous, but we set a timeout anyway in case of resource
  129. * exhaustion on the local host.
  130. */
  131. #define XS_UDP_CONN_TO (5U * HZ)
  132. /*
  133. * Wait duration for an RPC TCP connection to be established. Solaris
  134. * NFS over TCP uses 60 seconds, for example, which is in line with how
  135. * long a server takes to reboot.
  136. */
  137. #define XS_TCP_CONN_TO (60U * HZ)
  138. /*
  139. * Wait duration for a reply from the RPC portmapper.
  140. */
  141. #define XS_BIND_TO (60U * HZ)
  142. /*
  143. * Delay if a UDP socket connect error occurs. This is most likely some
  144. * kind of resource problem on the local host.
  145. */
  146. #define XS_UDP_REEST_TO (2U * HZ)
  147. /*
  148. * The reestablish timeout allows clients to delay for a bit before attempting
  149. * to reconnect to a server that just dropped our connection.
  150. *
  151. * We implement an exponential backoff when trying to reestablish a TCP
  152. * transport connection with the server. Some servers like to drop a TCP
  153. * connection when they are overworked, so we start with a short timeout and
  154. * increase over time if the server is down or not responding.
  155. */
  156. #define XS_TCP_INIT_REEST_TO (3U * HZ)
  157. #define XS_TCP_MAX_REEST_TO (5U * 60 * HZ)
  158. /*
  159. * TCP idle timeout; client drops the transport socket if it is idle
  160. * for this long. Note that we also timeout UDP sockets to prevent
  161. * holding port numbers when there is no RPC traffic.
  162. */
  163. #define XS_IDLE_DISC_TO (5U * 60 * HZ)
  164. #ifdef RPC_DEBUG
  165. # undef RPC_DEBUG_DATA
  166. # define RPCDBG_FACILITY RPCDBG_TRANS
  167. #endif
  168. #ifdef RPC_DEBUG_DATA
  169. static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
  170. {
  171. u8 *buf = (u8 *) packet;
  172. int j;
  173. dprintk("RPC: %s\n", msg);
  174. for (j = 0; j < count && j < 128; j += 4) {
  175. if (!(j & 31)) {
  176. if (j)
  177. dprintk("\n");
  178. dprintk("0x%04x ", j);
  179. }
  180. dprintk("%02x%02x%02x%02x ",
  181. buf[j], buf[j+1], buf[j+2], buf[j+3]);
  182. }
  183. dprintk("\n");
  184. }
  185. #else
  186. static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
  187. {
  188. /* NOP */
  189. }
  190. #endif
  191. struct sock_xprt {
  192. struct rpc_xprt xprt;
  193. /*
  194. * Network layer
  195. */
  196. struct socket * sock;
  197. struct sock * inet;
  198. /*
  199. * State of TCP reply receive
  200. */
  201. __be32 tcp_fraghdr,
  202. tcp_xid;
  203. u32 tcp_offset,
  204. tcp_reclen;
  205. unsigned long tcp_copied,
  206. tcp_flags;
  207. /*
  208. * Connection of transports
  209. */
  210. struct delayed_work connect_worker;
  211. unsigned short port;
  212. /*
  213. * UDP socket buffer size parameters
  214. */
  215. size_t rcvsize,
  216. sndsize;
  217. /*
  218. * Saved socket callback addresses
  219. */
  220. void (*old_data_ready)(struct sock *, int);
  221. void (*old_state_change)(struct sock *);
  222. void (*old_write_space)(struct sock *);
  223. };
  224. /*
  225. * TCP receive state flags
  226. */
  227. #define TCP_RCV_LAST_FRAG (1UL << 0)
  228. #define TCP_RCV_COPY_FRAGHDR (1UL << 1)
  229. #define TCP_RCV_COPY_XID (1UL << 2)
  230. #define TCP_RCV_COPY_DATA (1UL << 3)
  231. static void xs_format_peer_addresses(struct rpc_xprt *xprt)
  232. {
  233. struct sockaddr_in *addr = (struct sockaddr_in *) &xprt->addr;
  234. char *buf;
  235. buf = kzalloc(20, GFP_KERNEL);
  236. if (buf) {
  237. snprintf(buf, 20, "%u.%u.%u.%u",
  238. NIPQUAD(addr->sin_addr.s_addr));
  239. }
  240. xprt->address_strings[RPC_DISPLAY_ADDR] = buf;
  241. buf = kzalloc(8, GFP_KERNEL);
  242. if (buf) {
  243. snprintf(buf, 8, "%u",
  244. ntohs(addr->sin_port));
  245. }
  246. xprt->address_strings[RPC_DISPLAY_PORT] = buf;
  247. if (xprt->prot == IPPROTO_UDP)
  248. xprt->address_strings[RPC_DISPLAY_PROTO] = "udp";
  249. else
  250. xprt->address_strings[RPC_DISPLAY_PROTO] = "tcp";
  251. buf = kzalloc(48, GFP_KERNEL);
  252. if (buf) {
  253. snprintf(buf, 48, "addr=%u.%u.%u.%u port=%u proto=%s",
  254. NIPQUAD(addr->sin_addr.s_addr),
  255. ntohs(addr->sin_port),
  256. xprt->prot == IPPROTO_UDP ? "udp" : "tcp");
  257. }
  258. xprt->address_strings[RPC_DISPLAY_ALL] = buf;
  259. }
  260. static void xs_free_peer_addresses(struct rpc_xprt *xprt)
  261. {
  262. kfree(xprt->address_strings[RPC_DISPLAY_ADDR]);
  263. kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
  264. kfree(xprt->address_strings[RPC_DISPLAY_ALL]);
  265. }
  266. #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
  267. static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
  268. {
  269. struct msghdr msg = {
  270. .msg_name = addr,
  271. .msg_namelen = addrlen,
  272. .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
  273. };
  274. struct kvec iov = {
  275. .iov_base = vec->iov_base + base,
  276. .iov_len = vec->iov_len - base,
  277. };
  278. if (iov.iov_len != 0)
  279. return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
  280. return kernel_sendmsg(sock, &msg, NULL, 0, 0);
  281. }
  282. static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more)
  283. {
  284. struct page **ppage;
  285. unsigned int remainder;
  286. int err, sent = 0;
  287. remainder = xdr->page_len - base;
  288. base += xdr->page_base;
  289. ppage = xdr->pages + (base >> PAGE_SHIFT);
  290. base &= ~PAGE_MASK;
  291. for(;;) {
  292. unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
  293. int flags = XS_SENDMSG_FLAGS;
  294. remainder -= len;
  295. if (remainder != 0 || more)
  296. flags |= MSG_MORE;
  297. err = sock->ops->sendpage(sock, *ppage, base, len, flags);
  298. if (remainder == 0 || err != len)
  299. break;
  300. sent += err;
  301. ppage++;
  302. base = 0;
  303. }
  304. if (sent == 0)
  305. return err;
  306. if (err > 0)
  307. sent += err;
  308. return sent;
  309. }
  310. /**
  311. * xs_sendpages - write pages directly to a socket
  312. * @sock: socket to send on
  313. * @addr: UDP only -- address of destination
  314. * @addrlen: UDP only -- length of destination address
  315. * @xdr: buffer containing this request
  316. * @base: starting position in the buffer
  317. *
  318. */
  319. static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base)
  320. {
  321. unsigned int remainder = xdr->len - base;
  322. int err, sent = 0;
  323. if (unlikely(!sock))
  324. return -ENOTCONN;
  325. clear_bit(SOCK_ASYNC_NOSPACE, &sock->flags);
  326. if (base != 0) {
  327. addr = NULL;
  328. addrlen = 0;
  329. }
  330. if (base < xdr->head[0].iov_len || addr != NULL) {
  331. unsigned int len = xdr->head[0].iov_len - base;
  332. remainder -= len;
  333. err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
  334. if (remainder == 0 || err != len)
  335. goto out;
  336. sent += err;
  337. base = 0;
  338. } else
  339. base -= xdr->head[0].iov_len;
  340. if (base < xdr->page_len) {
  341. unsigned int len = xdr->page_len - base;
  342. remainder -= len;
  343. err = xs_send_pagedata(sock, xdr, base, remainder != 0);
  344. if (remainder == 0 || err != len)
  345. goto out;
  346. sent += err;
  347. base = 0;
  348. } else
  349. base -= xdr->page_len;
  350. if (base >= xdr->tail[0].iov_len)
  351. return sent;
  352. err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
  353. out:
  354. if (sent == 0)
  355. return err;
  356. if (err > 0)
  357. sent += err;
  358. return sent;
  359. }
  360. /**
  361. * xs_nospace - place task on wait queue if transmit was incomplete
  362. * @task: task to put to sleep
  363. *
  364. */
  365. static void xs_nospace(struct rpc_task *task)
  366. {
  367. struct rpc_rqst *req = task->tk_rqstp;
  368. struct rpc_xprt *xprt = req->rq_xprt;
  369. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  370. dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
  371. task->tk_pid, req->rq_slen - req->rq_bytes_sent,
  372. req->rq_slen);
  373. if (test_bit(SOCK_ASYNC_NOSPACE, &transport->sock->flags)) {
  374. /* Protect against races with write_space */
  375. spin_lock_bh(&xprt->transport_lock);
  376. /* Don't race with disconnect */
  377. if (!xprt_connected(xprt))
  378. task->tk_status = -ENOTCONN;
  379. else if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
  380. xprt_wait_for_buffer_space(task);
  381. spin_unlock_bh(&xprt->transport_lock);
  382. } else
  383. /* Keep holding the socket if it is blocked */
  384. rpc_delay(task, HZ>>4);
  385. }
  386. /**
  387. * xs_udp_send_request - write an RPC request to a UDP socket
  388. * @task: address of RPC task that manages the state of an RPC request
  389. *
  390. * Return values:
  391. * 0: The request has been sent
  392. * EAGAIN: The socket was blocked, please call again later to
  393. * complete the request
  394. * ENOTCONN: Caller needs to invoke connect logic then call again
  395. * other: Some other error occured, the request was not sent
  396. */
  397. static int xs_udp_send_request(struct rpc_task *task)
  398. {
  399. struct rpc_rqst *req = task->tk_rqstp;
  400. struct rpc_xprt *xprt = req->rq_xprt;
  401. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  402. struct xdr_buf *xdr = &req->rq_snd_buf;
  403. int status;
  404. xs_pktdump("packet data:",
  405. req->rq_svec->iov_base,
  406. req->rq_svec->iov_len);
  407. req->rq_xtime = jiffies;
  408. status = xs_sendpages(transport->sock,
  409. (struct sockaddr *) &xprt->addr,
  410. xprt->addrlen, xdr,
  411. req->rq_bytes_sent);
  412. dprintk("RPC: xs_udp_send_request(%u) = %d\n",
  413. xdr->len - req->rq_bytes_sent, status);
  414. if (likely(status >= (int) req->rq_slen))
  415. return 0;
  416. /* Still some bytes left; set up for a retry later. */
  417. if (status > 0)
  418. status = -EAGAIN;
  419. switch (status) {
  420. case -ENETUNREACH:
  421. case -EPIPE:
  422. case -ECONNREFUSED:
  423. /* When the server has died, an ICMP port unreachable message
  424. * prompts ECONNREFUSED. */
  425. break;
  426. case -EAGAIN:
  427. xs_nospace(task);
  428. break;
  429. default:
  430. dprintk("RPC: sendmsg returned unrecognized error %d\n",
  431. -status);
  432. break;
  433. }
  434. return status;
  435. }
  436. static inline void xs_encode_tcp_record_marker(struct xdr_buf *buf)
  437. {
  438. u32 reclen = buf->len - sizeof(rpc_fraghdr);
  439. rpc_fraghdr *base = buf->head[0].iov_base;
  440. *base = htonl(RPC_LAST_STREAM_FRAGMENT | reclen);
  441. }
  442. /**
  443. * xs_tcp_send_request - write an RPC request to a TCP socket
  444. * @task: address of RPC task that manages the state of an RPC request
  445. *
  446. * Return values:
  447. * 0: The request has been sent
  448. * EAGAIN: The socket was blocked, please call again later to
  449. * complete the request
  450. * ENOTCONN: Caller needs to invoke connect logic then call again
  451. * other: Some other error occured, the request was not sent
  452. *
  453. * XXX: In the case of soft timeouts, should we eventually give up
  454. * if sendmsg is not able to make progress?
  455. */
  456. static int xs_tcp_send_request(struct rpc_task *task)
  457. {
  458. struct rpc_rqst *req = task->tk_rqstp;
  459. struct rpc_xprt *xprt = req->rq_xprt;
  460. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  461. struct xdr_buf *xdr = &req->rq_snd_buf;
  462. int status, retry = 0;
  463. xs_encode_tcp_record_marker(&req->rq_snd_buf);
  464. xs_pktdump("packet data:",
  465. req->rq_svec->iov_base,
  466. req->rq_svec->iov_len);
  467. /* Continue transmitting the packet/record. We must be careful
  468. * to cope with writespace callbacks arriving _after_ we have
  469. * called sendmsg(). */
  470. while (1) {
  471. req->rq_xtime = jiffies;
  472. status = xs_sendpages(transport->sock,
  473. NULL, 0, xdr, req->rq_bytes_sent);
  474. dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
  475. xdr->len - req->rq_bytes_sent, status);
  476. if (unlikely(status < 0))
  477. break;
  478. /* If we've sent the entire packet, immediately
  479. * reset the count of bytes sent. */
  480. req->rq_bytes_sent += status;
  481. task->tk_bytes_sent += status;
  482. if (likely(req->rq_bytes_sent >= req->rq_slen)) {
  483. req->rq_bytes_sent = 0;
  484. return 0;
  485. }
  486. status = -EAGAIN;
  487. if (retry++ > XS_SENDMSG_RETRY)
  488. break;
  489. }
  490. switch (status) {
  491. case -EAGAIN:
  492. xs_nospace(task);
  493. break;
  494. case -ECONNREFUSED:
  495. case -ECONNRESET:
  496. case -ENOTCONN:
  497. case -EPIPE:
  498. status = -ENOTCONN;
  499. break;
  500. default:
  501. dprintk("RPC: sendmsg returned unrecognized error %d\n",
  502. -status);
  503. xprt_disconnect(xprt);
  504. break;
  505. }
  506. return status;
  507. }
  508. /**
  509. * xs_tcp_release_xprt - clean up after a tcp transmission
  510. * @xprt: transport
  511. * @task: rpc task
  512. *
  513. * This cleans up if an error causes us to abort the transmission of a request.
  514. * In this case, the socket may need to be reset in order to avoid confusing
  515. * the server.
  516. */
  517. static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
  518. {
  519. struct rpc_rqst *req;
  520. if (task != xprt->snd_task)
  521. return;
  522. if (task == NULL)
  523. goto out_release;
  524. req = task->tk_rqstp;
  525. if (req->rq_bytes_sent == 0)
  526. goto out_release;
  527. if (req->rq_bytes_sent == req->rq_snd_buf.len)
  528. goto out_release;
  529. set_bit(XPRT_CLOSE_WAIT, &task->tk_xprt->state);
  530. out_release:
  531. xprt_release_xprt(xprt, task);
  532. }
  533. /**
  534. * xs_close - close a socket
  535. * @xprt: transport
  536. *
  537. * This is used when all requests are complete; ie, no DRC state remains
  538. * on the server we want to save.
  539. */
  540. static void xs_close(struct rpc_xprt *xprt)
  541. {
  542. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  543. struct socket *sock = transport->sock;
  544. struct sock *sk = transport->inet;
  545. if (!sk)
  546. goto clear_close_wait;
  547. dprintk("RPC: xs_close xprt %p\n", xprt);
  548. write_lock_bh(&sk->sk_callback_lock);
  549. transport->inet = NULL;
  550. transport->sock = NULL;
  551. sk->sk_user_data = NULL;
  552. sk->sk_data_ready = transport->old_data_ready;
  553. sk->sk_state_change = transport->old_state_change;
  554. sk->sk_write_space = transport->old_write_space;
  555. write_unlock_bh(&sk->sk_callback_lock);
  556. sk->sk_no_check = 0;
  557. sock_release(sock);
  558. clear_close_wait:
  559. smp_mb__before_clear_bit();
  560. clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
  561. smp_mb__after_clear_bit();
  562. }
  563. /**
  564. * xs_destroy - prepare to shutdown a transport
  565. * @xprt: doomed transport
  566. *
  567. */
  568. static void xs_destroy(struct rpc_xprt *xprt)
  569. {
  570. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  571. dprintk("RPC: xs_destroy xprt %p\n", xprt);
  572. cancel_delayed_work(&transport->connect_worker);
  573. flush_scheduled_work();
  574. xprt_disconnect(xprt);
  575. xs_close(xprt);
  576. xs_free_peer_addresses(xprt);
  577. kfree(xprt->slot);
  578. kfree(xprt);
  579. }
  580. static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
  581. {
  582. return (struct rpc_xprt *) sk->sk_user_data;
  583. }
  584. /**
  585. * xs_udp_data_ready - "data ready" callback for UDP sockets
  586. * @sk: socket with data to read
  587. * @len: how much data to read
  588. *
  589. */
  590. static void xs_udp_data_ready(struct sock *sk, int len)
  591. {
  592. struct rpc_task *task;
  593. struct rpc_xprt *xprt;
  594. struct rpc_rqst *rovr;
  595. struct sk_buff *skb;
  596. int err, repsize, copied;
  597. u32 _xid;
  598. __be32 *xp;
  599. read_lock(&sk->sk_callback_lock);
  600. dprintk("RPC: xs_udp_data_ready...\n");
  601. if (!(xprt = xprt_from_sock(sk)))
  602. goto out;
  603. if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
  604. goto out;
  605. if (xprt->shutdown)
  606. goto dropit;
  607. repsize = skb->len - sizeof(struct udphdr);
  608. if (repsize < 4) {
  609. dprintk("RPC: impossible RPC reply size %d!\n", repsize);
  610. goto dropit;
  611. }
  612. /* Copy the XID from the skb... */
  613. xp = skb_header_pointer(skb, sizeof(struct udphdr),
  614. sizeof(_xid), &_xid);
  615. if (xp == NULL)
  616. goto dropit;
  617. /* Look up and lock the request corresponding to the given XID */
  618. spin_lock(&xprt->transport_lock);
  619. rovr = xprt_lookup_rqst(xprt, *xp);
  620. if (!rovr)
  621. goto out_unlock;
  622. task = rovr->rq_task;
  623. if ((copied = rovr->rq_private_buf.buflen) > repsize)
  624. copied = repsize;
  625. /* Suck it into the iovec, verify checksum if not done by hw. */
  626. if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb))
  627. goto out_unlock;
  628. /* Something worked... */
  629. dst_confirm(skb->dst);
  630. xprt_adjust_cwnd(task, copied);
  631. xprt_update_rtt(task);
  632. xprt_complete_rqst(task, copied);
  633. out_unlock:
  634. spin_unlock(&xprt->transport_lock);
  635. dropit:
  636. skb_free_datagram(sk, skb);
  637. out:
  638. read_unlock(&sk->sk_callback_lock);
  639. }
  640. static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
  641. {
  642. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  643. size_t len, used;
  644. char *p;
  645. p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
  646. len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
  647. used = xdr_skb_read_bits(desc, p, len);
  648. transport->tcp_offset += used;
  649. if (used != len)
  650. return;
  651. transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
  652. if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
  653. transport->tcp_flags |= TCP_RCV_LAST_FRAG;
  654. else
  655. transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
  656. transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
  657. transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
  658. transport->tcp_offset = 0;
  659. /* Sanity check of the record length */
  660. if (unlikely(transport->tcp_reclen < 4)) {
  661. dprintk("RPC: invalid TCP record fragment length\n");
  662. xprt_disconnect(xprt);
  663. return;
  664. }
  665. dprintk("RPC: reading TCP record fragment of length %d\n",
  666. transport->tcp_reclen);
  667. }
  668. static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
  669. {
  670. if (transport->tcp_offset == transport->tcp_reclen) {
  671. transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
  672. transport->tcp_offset = 0;
  673. if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
  674. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  675. transport->tcp_flags |= TCP_RCV_COPY_XID;
  676. transport->tcp_copied = 0;
  677. }
  678. }
  679. }
  680. static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
  681. {
  682. size_t len, used;
  683. char *p;
  684. len = sizeof(transport->tcp_xid) - transport->tcp_offset;
  685. dprintk("RPC: reading XID (%Zu bytes)\n", len);
  686. p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
  687. used = xdr_skb_read_bits(desc, p, len);
  688. transport->tcp_offset += used;
  689. if (used != len)
  690. return;
  691. transport->tcp_flags &= ~TCP_RCV_COPY_XID;
  692. transport->tcp_flags |= TCP_RCV_COPY_DATA;
  693. transport->tcp_copied = 4;
  694. dprintk("RPC: reading reply for XID %08x\n",
  695. ntohl(transport->tcp_xid));
  696. xs_tcp_check_fraghdr(transport);
  697. }
  698. static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
  699. {
  700. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  701. struct rpc_rqst *req;
  702. struct xdr_buf *rcvbuf;
  703. size_t len;
  704. ssize_t r;
  705. /* Find and lock the request corresponding to this xid */
  706. spin_lock(&xprt->transport_lock);
  707. req = xprt_lookup_rqst(xprt, transport->tcp_xid);
  708. if (!req) {
  709. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  710. dprintk("RPC: XID %08x request not found!\n",
  711. ntohl(transport->tcp_xid));
  712. spin_unlock(&xprt->transport_lock);
  713. return;
  714. }
  715. rcvbuf = &req->rq_private_buf;
  716. len = desc->count;
  717. if (len > transport->tcp_reclen - transport->tcp_offset) {
  718. struct xdr_skb_reader my_desc;
  719. len = transport->tcp_reclen - transport->tcp_offset;
  720. memcpy(&my_desc, desc, sizeof(my_desc));
  721. my_desc.count = len;
  722. r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
  723. &my_desc, xdr_skb_read_bits);
  724. desc->count -= r;
  725. desc->offset += r;
  726. } else
  727. r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
  728. desc, xdr_skb_read_bits);
  729. if (r > 0) {
  730. transport->tcp_copied += r;
  731. transport->tcp_offset += r;
  732. }
  733. if (r != len) {
  734. /* Error when copying to the receive buffer,
  735. * usually because we weren't able to allocate
  736. * additional buffer pages. All we can do now
  737. * is turn off TCP_RCV_COPY_DATA, so the request
  738. * will not receive any additional updates,
  739. * and time out.
  740. * Any remaining data from this record will
  741. * be discarded.
  742. */
  743. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  744. dprintk("RPC: XID %08x truncated request\n",
  745. ntohl(transport->tcp_xid));
  746. dprintk("RPC: xprt = %p, tcp_copied = %lu, "
  747. "tcp_offset = %u, tcp_reclen = %u\n",
  748. xprt, transport->tcp_copied,
  749. transport->tcp_offset, transport->tcp_reclen);
  750. goto out;
  751. }
  752. dprintk("RPC: XID %08x read %Zd bytes\n",
  753. ntohl(transport->tcp_xid), r);
  754. dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
  755. "tcp_reclen = %u\n", xprt, transport->tcp_copied,
  756. transport->tcp_offset, transport->tcp_reclen);
  757. if (transport->tcp_copied == req->rq_private_buf.buflen)
  758. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  759. else if (transport->tcp_offset == transport->tcp_reclen) {
  760. if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
  761. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  762. }
  763. out:
  764. if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
  765. xprt_complete_rqst(req->rq_task, transport->tcp_copied);
  766. spin_unlock(&xprt->transport_lock);
  767. xs_tcp_check_fraghdr(transport);
  768. }
  769. static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
  770. {
  771. size_t len;
  772. len = transport->tcp_reclen - transport->tcp_offset;
  773. if (len > desc->count)
  774. len = desc->count;
  775. desc->count -= len;
  776. desc->offset += len;
  777. transport->tcp_offset += len;
  778. dprintk("RPC: discarded %Zu bytes\n", len);
  779. xs_tcp_check_fraghdr(transport);
  780. }
  781. static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
  782. {
  783. struct rpc_xprt *xprt = rd_desc->arg.data;
  784. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  785. struct xdr_skb_reader desc = {
  786. .skb = skb,
  787. .offset = offset,
  788. .count = len,
  789. };
  790. dprintk("RPC: xs_tcp_data_recv started\n");
  791. do {
  792. /* Read in a new fragment marker if necessary */
  793. /* Can we ever really expect to get completely empty fragments? */
  794. if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
  795. xs_tcp_read_fraghdr(xprt, &desc);
  796. continue;
  797. }
  798. /* Read in the xid if necessary */
  799. if (transport->tcp_flags & TCP_RCV_COPY_XID) {
  800. xs_tcp_read_xid(transport, &desc);
  801. continue;
  802. }
  803. /* Read in the request data */
  804. if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
  805. xs_tcp_read_request(xprt, &desc);
  806. continue;
  807. }
  808. /* Skip over any trailing bytes on short reads */
  809. xs_tcp_read_discard(transport, &desc);
  810. } while (desc.count);
  811. dprintk("RPC: xs_tcp_data_recv done\n");
  812. return len - desc.count;
  813. }
  814. /**
  815. * xs_tcp_data_ready - "data ready" callback for TCP sockets
  816. * @sk: socket with data to read
  817. * @bytes: how much data to read
  818. *
  819. */
  820. static void xs_tcp_data_ready(struct sock *sk, int bytes)
  821. {
  822. struct rpc_xprt *xprt;
  823. read_descriptor_t rd_desc;
  824. dprintk("RPC: xs_tcp_data_ready...\n");
  825. read_lock(&sk->sk_callback_lock);
  826. if (!(xprt = xprt_from_sock(sk)))
  827. goto out;
  828. if (xprt->shutdown)
  829. goto out;
  830. /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
  831. rd_desc.arg.data = xprt;
  832. rd_desc.count = 65536;
  833. tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
  834. out:
  835. read_unlock(&sk->sk_callback_lock);
  836. }
  837. /**
  838. * xs_tcp_state_change - callback to handle TCP socket state changes
  839. * @sk: socket whose state has changed
  840. *
  841. */
  842. static void xs_tcp_state_change(struct sock *sk)
  843. {
  844. struct rpc_xprt *xprt;
  845. read_lock(&sk->sk_callback_lock);
  846. if (!(xprt = xprt_from_sock(sk)))
  847. goto out;
  848. dprintk("RPC: xs_tcp_state_change client %p...\n", xprt);
  849. dprintk("RPC: state %x conn %d dead %d zapped %d\n",
  850. sk->sk_state, xprt_connected(xprt),
  851. sock_flag(sk, SOCK_DEAD),
  852. sock_flag(sk, SOCK_ZAPPED));
  853. switch (sk->sk_state) {
  854. case TCP_ESTABLISHED:
  855. spin_lock_bh(&xprt->transport_lock);
  856. if (!xprt_test_and_set_connected(xprt)) {
  857. struct sock_xprt *transport = container_of(xprt,
  858. struct sock_xprt, xprt);
  859. /* Reset TCP record info */
  860. transport->tcp_offset = 0;
  861. transport->tcp_reclen = 0;
  862. transport->tcp_copied = 0;
  863. transport->tcp_flags =
  864. TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
  865. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  866. xprt_wake_pending_tasks(xprt, 0);
  867. }
  868. spin_unlock_bh(&xprt->transport_lock);
  869. break;
  870. case TCP_SYN_SENT:
  871. case TCP_SYN_RECV:
  872. break;
  873. case TCP_CLOSE_WAIT:
  874. /* Try to schedule an autoclose RPC calls */
  875. set_bit(XPRT_CLOSE_WAIT, &xprt->state);
  876. if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
  877. schedule_work(&xprt->task_cleanup);
  878. default:
  879. xprt_disconnect(xprt);
  880. }
  881. out:
  882. read_unlock(&sk->sk_callback_lock);
  883. }
  884. /**
  885. * xs_udp_write_space - callback invoked when socket buffer space
  886. * becomes available
  887. * @sk: socket whose state has changed
  888. *
  889. * Called when more output buffer space is available for this socket.
  890. * We try not to wake our writers until they can make "significant"
  891. * progress, otherwise we'll waste resources thrashing kernel_sendmsg
  892. * with a bunch of small requests.
  893. */
  894. static void xs_udp_write_space(struct sock *sk)
  895. {
  896. read_lock(&sk->sk_callback_lock);
  897. /* from net/core/sock.c:sock_def_write_space */
  898. if (sock_writeable(sk)) {
  899. struct socket *sock;
  900. struct rpc_xprt *xprt;
  901. if (unlikely(!(sock = sk->sk_socket)))
  902. goto out;
  903. if (unlikely(!(xprt = xprt_from_sock(sk))))
  904. goto out;
  905. if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
  906. goto out;
  907. xprt_write_space(xprt);
  908. }
  909. out:
  910. read_unlock(&sk->sk_callback_lock);
  911. }
  912. /**
  913. * xs_tcp_write_space - callback invoked when socket buffer space
  914. * becomes available
  915. * @sk: socket whose state has changed
  916. *
  917. * Called when more output buffer space is available for this socket.
  918. * We try not to wake our writers until they can make "significant"
  919. * progress, otherwise we'll waste resources thrashing kernel_sendmsg
  920. * with a bunch of small requests.
  921. */
  922. static void xs_tcp_write_space(struct sock *sk)
  923. {
  924. read_lock(&sk->sk_callback_lock);
  925. /* from net/core/stream.c:sk_stream_write_space */
  926. if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
  927. struct socket *sock;
  928. struct rpc_xprt *xprt;
  929. if (unlikely(!(sock = sk->sk_socket)))
  930. goto out;
  931. if (unlikely(!(xprt = xprt_from_sock(sk))))
  932. goto out;
  933. if (unlikely(!test_and_clear_bit(SOCK_NOSPACE, &sock->flags)))
  934. goto out;
  935. xprt_write_space(xprt);
  936. }
  937. out:
  938. read_unlock(&sk->sk_callback_lock);
  939. }
  940. static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
  941. {
  942. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  943. struct sock *sk = transport->inet;
  944. if (transport->rcvsize) {
  945. sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
  946. sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
  947. }
  948. if (transport->sndsize) {
  949. sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
  950. sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
  951. sk->sk_write_space(sk);
  952. }
  953. }
  954. /**
  955. * xs_udp_set_buffer_size - set send and receive limits
  956. * @xprt: generic transport
  957. * @sndsize: requested size of send buffer, in bytes
  958. * @rcvsize: requested size of receive buffer, in bytes
  959. *
  960. * Set socket send and receive buffer size limits.
  961. */
  962. static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
  963. {
  964. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  965. transport->sndsize = 0;
  966. if (sndsize)
  967. transport->sndsize = sndsize + 1024;
  968. transport->rcvsize = 0;
  969. if (rcvsize)
  970. transport->rcvsize = rcvsize + 1024;
  971. xs_udp_do_set_buffer_size(xprt);
  972. }
  973. /**
  974. * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
  975. * @task: task that timed out
  976. *
  977. * Adjust the congestion window after a retransmit timeout has occurred.
  978. */
  979. static void xs_udp_timer(struct rpc_task *task)
  980. {
  981. xprt_adjust_cwnd(task, -ETIMEDOUT);
  982. }
  983. static unsigned short xs_get_random_port(void)
  984. {
  985. unsigned short range = xprt_max_resvport - xprt_min_resvport;
  986. unsigned short rand = (unsigned short) net_random() % range;
  987. return rand + xprt_min_resvport;
  988. }
  989. /**
  990. * xs_set_port - reset the port number in the remote endpoint address
  991. * @xprt: generic transport
  992. * @port: new port number
  993. *
  994. */
  995. static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
  996. {
  997. struct sockaddr_in *sap = (struct sockaddr_in *) &xprt->addr;
  998. dprintk("RPC: setting port for xprt %p to %u\n", xprt, port);
  999. sap->sin_port = htons(port);
  1000. }
  1001. static int xs_bindresvport(struct sock_xprt *transport, struct socket *sock)
  1002. {
  1003. struct sockaddr_in myaddr = {
  1004. .sin_family = AF_INET,
  1005. };
  1006. int err;
  1007. unsigned short port = transport->port;
  1008. do {
  1009. myaddr.sin_port = htons(port);
  1010. err = kernel_bind(sock, (struct sockaddr *) &myaddr,
  1011. sizeof(myaddr));
  1012. if (err == 0) {
  1013. transport->port = port;
  1014. dprintk("RPC: xs_bindresvport bound to port %u\n",
  1015. port);
  1016. return 0;
  1017. }
  1018. if (port <= xprt_min_resvport)
  1019. port = xprt_max_resvport;
  1020. else
  1021. port--;
  1022. } while (err == -EADDRINUSE && port != transport->port);
  1023. dprintk("RPC: can't bind to reserved port (%d).\n", -err);
  1024. return err;
  1025. }
  1026. #ifdef CONFIG_DEBUG_LOCK_ALLOC
  1027. static struct lock_class_key xs_key[2];
  1028. static struct lock_class_key xs_slock_key[2];
  1029. static inline void xs_reclassify_socket(struct socket *sock)
  1030. {
  1031. struct sock *sk = sock->sk;
  1032. BUG_ON(sk->sk_lock.owner != NULL);
  1033. switch (sk->sk_family) {
  1034. case AF_INET:
  1035. sock_lock_init_class_and_name(sk, "slock-AF_INET-NFS",
  1036. &xs_slock_key[0], "sk_lock-AF_INET-NFS", &xs_key[0]);
  1037. break;
  1038. case AF_INET6:
  1039. sock_lock_init_class_and_name(sk, "slock-AF_INET6-NFS",
  1040. &xs_slock_key[1], "sk_lock-AF_INET6-NFS", &xs_key[1]);
  1041. break;
  1042. default:
  1043. BUG();
  1044. }
  1045. }
  1046. #else
  1047. static inline void xs_reclassify_socket(struct socket *sock)
  1048. {
  1049. }
  1050. #endif
  1051. /**
  1052. * xs_udp_connect_worker - set up a UDP socket
  1053. * @work: RPC transport to connect
  1054. *
  1055. * Invoked by a work queue tasklet.
  1056. */
  1057. static void xs_udp_connect_worker(struct work_struct *work)
  1058. {
  1059. struct sock_xprt *transport =
  1060. container_of(work, struct sock_xprt, connect_worker.work);
  1061. struct rpc_xprt *xprt = &transport->xprt;
  1062. struct socket *sock = transport->sock;
  1063. int err, status = -EIO;
  1064. if (xprt->shutdown || !xprt_bound(xprt))
  1065. goto out;
  1066. /* Start by resetting any existing state */
  1067. xs_close(xprt);
  1068. if ((err = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock)) < 0) {
  1069. dprintk("RPC: can't create UDP transport socket (%d).\n", -err);
  1070. goto out;
  1071. }
  1072. xs_reclassify_socket(sock);
  1073. if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
  1074. sock_release(sock);
  1075. goto out;
  1076. }
  1077. dprintk("RPC: worker connecting xprt %p to address: %s\n",
  1078. xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
  1079. if (!transport->inet) {
  1080. struct sock *sk = sock->sk;
  1081. write_lock_bh(&sk->sk_callback_lock);
  1082. sk->sk_user_data = xprt;
  1083. transport->old_data_ready = sk->sk_data_ready;
  1084. transport->old_state_change = sk->sk_state_change;
  1085. transport->old_write_space = sk->sk_write_space;
  1086. sk->sk_data_ready = xs_udp_data_ready;
  1087. sk->sk_write_space = xs_udp_write_space;
  1088. sk->sk_no_check = UDP_CSUM_NORCV;
  1089. sk->sk_allocation = GFP_ATOMIC;
  1090. xprt_set_connected(xprt);
  1091. /* Reset to new socket */
  1092. transport->sock = sock;
  1093. transport->inet = sk;
  1094. write_unlock_bh(&sk->sk_callback_lock);
  1095. }
  1096. xs_udp_do_set_buffer_size(xprt);
  1097. status = 0;
  1098. out:
  1099. xprt_wake_pending_tasks(xprt, status);
  1100. xprt_clear_connecting(xprt);
  1101. }
  1102. /*
  1103. * We need to preserve the port number so the reply cache on the server can
  1104. * find our cached RPC replies when we get around to reconnecting.
  1105. */
  1106. static void xs_tcp_reuse_connection(struct rpc_xprt *xprt)
  1107. {
  1108. int result;
  1109. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1110. struct sockaddr any;
  1111. dprintk("RPC: disconnecting xprt %p to reuse port\n", xprt);
  1112. /*
  1113. * Disconnect the transport socket by doing a connect operation
  1114. * with AF_UNSPEC. This should return immediately...
  1115. */
  1116. memset(&any, 0, sizeof(any));
  1117. any.sa_family = AF_UNSPEC;
  1118. result = kernel_connect(transport->sock, &any, sizeof(any), 0);
  1119. if (result)
  1120. dprintk("RPC: AF_UNSPEC connect return code %d\n",
  1121. result);
  1122. }
  1123. /**
  1124. * xs_tcp_connect_worker - connect a TCP socket to a remote endpoint
  1125. * @work: RPC transport to connect
  1126. *
  1127. * Invoked by a work queue tasklet.
  1128. */
  1129. static void xs_tcp_connect_worker(struct work_struct *work)
  1130. {
  1131. struct sock_xprt *transport =
  1132. container_of(work, struct sock_xprt, connect_worker.work);
  1133. struct rpc_xprt *xprt = &transport->xprt;
  1134. struct socket *sock = transport->sock;
  1135. int err, status = -EIO;
  1136. if (xprt->shutdown || !xprt_bound(xprt))
  1137. goto out;
  1138. if (!sock) {
  1139. /* start from scratch */
  1140. if ((err = sock_create_kern(PF_INET, SOCK_STREAM, IPPROTO_TCP, &sock)) < 0) {
  1141. dprintk("RPC: can't create TCP transport "
  1142. "socket (%d).\n", -err);
  1143. goto out;
  1144. }
  1145. xs_reclassify_socket(sock);
  1146. if (xprt->resvport && xs_bindresvport(transport, sock) < 0) {
  1147. sock_release(sock);
  1148. goto out;
  1149. }
  1150. } else
  1151. /* "close" the socket, preserving the local port */
  1152. xs_tcp_reuse_connection(xprt);
  1153. dprintk("RPC: worker connecting xprt %p to address: %s\n",
  1154. xprt, xprt->address_strings[RPC_DISPLAY_ALL]);
  1155. if (!transport->inet) {
  1156. struct sock *sk = sock->sk;
  1157. write_lock_bh(&sk->sk_callback_lock);
  1158. sk->sk_user_data = xprt;
  1159. transport->old_data_ready = sk->sk_data_ready;
  1160. transport->old_state_change = sk->sk_state_change;
  1161. transport->old_write_space = sk->sk_write_space;
  1162. sk->sk_data_ready = xs_tcp_data_ready;
  1163. sk->sk_state_change = xs_tcp_state_change;
  1164. sk->sk_write_space = xs_tcp_write_space;
  1165. sk->sk_allocation = GFP_ATOMIC;
  1166. /* socket options */
  1167. sk->sk_userlocks |= SOCK_BINDPORT_LOCK;
  1168. sock_reset_flag(sk, SOCK_LINGER);
  1169. tcp_sk(sk)->linger2 = 0;
  1170. tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
  1171. xprt_clear_connected(xprt);
  1172. /* Reset to new socket */
  1173. transport->sock = sock;
  1174. transport->inet = sk;
  1175. write_unlock_bh(&sk->sk_callback_lock);
  1176. }
  1177. /* Tell the socket layer to start connecting... */
  1178. xprt->stat.connect_count++;
  1179. xprt->stat.connect_start = jiffies;
  1180. status = kernel_connect(sock, (struct sockaddr *) &xprt->addr,
  1181. xprt->addrlen, O_NONBLOCK);
  1182. dprintk("RPC: %p connect status %d connected %d sock state %d\n",
  1183. xprt, -status, xprt_connected(xprt),
  1184. sock->sk->sk_state);
  1185. if (status < 0) {
  1186. switch (status) {
  1187. case -EINPROGRESS:
  1188. case -EALREADY:
  1189. goto out_clear;
  1190. case -ECONNREFUSED:
  1191. case -ECONNRESET:
  1192. /* retry with existing socket, after a delay */
  1193. break;
  1194. default:
  1195. /* get rid of existing socket, and retry */
  1196. xs_close(xprt);
  1197. break;
  1198. }
  1199. }
  1200. out:
  1201. xprt_wake_pending_tasks(xprt, status);
  1202. out_clear:
  1203. xprt_clear_connecting(xprt);
  1204. }
  1205. /**
  1206. * xs_connect - connect a socket to a remote endpoint
  1207. * @task: address of RPC task that manages state of connect request
  1208. *
  1209. * TCP: If the remote end dropped the connection, delay reconnecting.
  1210. *
  1211. * UDP socket connects are synchronous, but we use a work queue anyway
  1212. * to guarantee that even unprivileged user processes can set up a
  1213. * socket on a privileged port.
  1214. *
  1215. * If a UDP socket connect fails, the delay behavior here prevents
  1216. * retry floods (hard mounts).
  1217. */
  1218. static void xs_connect(struct rpc_task *task)
  1219. {
  1220. struct rpc_xprt *xprt = task->tk_xprt;
  1221. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1222. if (xprt_test_and_set_connecting(xprt))
  1223. return;
  1224. if (transport->sock != NULL) {
  1225. dprintk("RPC: xs_connect delayed xprt %p for %lu "
  1226. "seconds\n",
  1227. xprt, xprt->reestablish_timeout / HZ);
  1228. schedule_delayed_work(&transport->connect_worker,
  1229. xprt->reestablish_timeout);
  1230. xprt->reestablish_timeout <<= 1;
  1231. if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
  1232. xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
  1233. } else {
  1234. dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
  1235. schedule_delayed_work(&transport->connect_worker, 0);
  1236. /* flush_scheduled_work can sleep... */
  1237. if (!RPC_IS_ASYNC(task))
  1238. flush_scheduled_work();
  1239. }
  1240. }
  1241. /**
  1242. * xs_udp_print_stats - display UDP socket-specifc stats
  1243. * @xprt: rpc_xprt struct containing statistics
  1244. * @seq: output file
  1245. *
  1246. */
  1247. static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  1248. {
  1249. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1250. seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %Lu %Lu\n",
  1251. transport->port,
  1252. xprt->stat.bind_count,
  1253. xprt->stat.sends,
  1254. xprt->stat.recvs,
  1255. xprt->stat.bad_xids,
  1256. xprt->stat.req_u,
  1257. xprt->stat.bklog_u);
  1258. }
  1259. /**
  1260. * xs_tcp_print_stats - display TCP socket-specifc stats
  1261. * @xprt: rpc_xprt struct containing statistics
  1262. * @seq: output file
  1263. *
  1264. */
  1265. static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  1266. {
  1267. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1268. long idle_time = 0;
  1269. if (xprt_connected(xprt))
  1270. idle_time = (long)(jiffies - xprt->last_used) / HZ;
  1271. seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu\n",
  1272. transport->port,
  1273. xprt->stat.bind_count,
  1274. xprt->stat.connect_count,
  1275. xprt->stat.connect_time,
  1276. idle_time,
  1277. xprt->stat.sends,
  1278. xprt->stat.recvs,
  1279. xprt->stat.bad_xids,
  1280. xprt->stat.req_u,
  1281. xprt->stat.bklog_u);
  1282. }
  1283. static struct rpc_xprt_ops xs_udp_ops = {
  1284. .set_buffer_size = xs_udp_set_buffer_size,
  1285. .reserve_xprt = xprt_reserve_xprt_cong,
  1286. .release_xprt = xprt_release_xprt_cong,
  1287. .rpcbind = rpc_getport,
  1288. .set_port = xs_set_port,
  1289. .connect = xs_connect,
  1290. .buf_alloc = rpc_malloc,
  1291. .buf_free = rpc_free,
  1292. .send_request = xs_udp_send_request,
  1293. .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
  1294. .timer = xs_udp_timer,
  1295. .release_request = xprt_release_rqst_cong,
  1296. .close = xs_close,
  1297. .destroy = xs_destroy,
  1298. .print_stats = xs_udp_print_stats,
  1299. };
  1300. static struct rpc_xprt_ops xs_tcp_ops = {
  1301. .reserve_xprt = xprt_reserve_xprt,
  1302. .release_xprt = xs_tcp_release_xprt,
  1303. .rpcbind = rpc_getport,
  1304. .set_port = xs_set_port,
  1305. .connect = xs_connect,
  1306. .buf_alloc = rpc_malloc,
  1307. .buf_free = rpc_free,
  1308. .send_request = xs_tcp_send_request,
  1309. .set_retrans_timeout = xprt_set_retrans_timeout_def,
  1310. .close = xs_close,
  1311. .destroy = xs_destroy,
  1312. .print_stats = xs_tcp_print_stats,
  1313. };
  1314. static struct rpc_xprt *xs_setup_xprt(struct sockaddr *addr, size_t addrlen, unsigned int slot_table_size)
  1315. {
  1316. struct rpc_xprt *xprt;
  1317. struct sock_xprt *new;
  1318. if (addrlen > sizeof(xprt->addr)) {
  1319. dprintk("RPC: xs_setup_xprt: address too large\n");
  1320. return ERR_PTR(-EBADF);
  1321. }
  1322. new = kzalloc(sizeof(*new), GFP_KERNEL);
  1323. if (new == NULL) {
  1324. dprintk("RPC: xs_setup_xprt: couldn't allocate "
  1325. "rpc_xprt\n");
  1326. return ERR_PTR(-ENOMEM);
  1327. }
  1328. xprt = &new->xprt;
  1329. xprt->max_reqs = slot_table_size;
  1330. xprt->slot = kcalloc(xprt->max_reqs, sizeof(struct rpc_rqst), GFP_KERNEL);
  1331. if (xprt->slot == NULL) {
  1332. kfree(xprt);
  1333. dprintk("RPC: xs_setup_xprt: couldn't allocate slot "
  1334. "table\n");
  1335. return ERR_PTR(-ENOMEM);
  1336. }
  1337. memcpy(&xprt->addr, addr, addrlen);
  1338. xprt->addrlen = addrlen;
  1339. new->port = xs_get_random_port();
  1340. return xprt;
  1341. }
  1342. /**
  1343. * xs_setup_udp - Set up transport to use a UDP socket
  1344. * @addr: address of remote server
  1345. * @addrlen: length of address in bytes
  1346. * @to: timeout parameters
  1347. *
  1348. */
  1349. struct rpc_xprt *xs_setup_udp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
  1350. {
  1351. struct rpc_xprt *xprt;
  1352. struct sock_xprt *transport;
  1353. xprt = xs_setup_xprt(addr, addrlen, xprt_udp_slot_table_entries);
  1354. if (IS_ERR(xprt))
  1355. return xprt;
  1356. transport = container_of(xprt, struct sock_xprt, xprt);
  1357. if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
  1358. xprt_set_bound(xprt);
  1359. xprt->prot = IPPROTO_UDP;
  1360. xprt->tsh_size = 0;
  1361. /* XXX: header size can vary due to auth type, IPv6, etc. */
  1362. xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
  1363. INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_connect_worker);
  1364. xprt->bind_timeout = XS_BIND_TO;
  1365. xprt->connect_timeout = XS_UDP_CONN_TO;
  1366. xprt->reestablish_timeout = XS_UDP_REEST_TO;
  1367. xprt->idle_timeout = XS_IDLE_DISC_TO;
  1368. xprt->ops = &xs_udp_ops;
  1369. if (to)
  1370. xprt->timeout = *to;
  1371. else
  1372. xprt_set_timeout(&xprt->timeout, 5, 5 * HZ);
  1373. xs_format_peer_addresses(xprt);
  1374. dprintk("RPC: set up transport to address %s\n",
  1375. xprt->address_strings[RPC_DISPLAY_ALL]);
  1376. return xprt;
  1377. }
  1378. /**
  1379. * xs_setup_tcp - Set up transport to use a TCP socket
  1380. * @addr: address of remote server
  1381. * @addrlen: length of address in bytes
  1382. * @to: timeout parameters
  1383. *
  1384. */
  1385. struct rpc_xprt *xs_setup_tcp(struct sockaddr *addr, size_t addrlen, struct rpc_timeout *to)
  1386. {
  1387. struct rpc_xprt *xprt;
  1388. struct sock_xprt *transport;
  1389. xprt = xs_setup_xprt(addr, addrlen, xprt_tcp_slot_table_entries);
  1390. if (IS_ERR(xprt))
  1391. return xprt;
  1392. transport = container_of(xprt, struct sock_xprt, xprt);
  1393. if (ntohs(((struct sockaddr_in *)addr)->sin_port) != 0)
  1394. xprt_set_bound(xprt);
  1395. xprt->prot = IPPROTO_TCP;
  1396. xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
  1397. xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
  1398. INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_connect_worker);
  1399. xprt->bind_timeout = XS_BIND_TO;
  1400. xprt->connect_timeout = XS_TCP_CONN_TO;
  1401. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  1402. xprt->idle_timeout = XS_IDLE_DISC_TO;
  1403. xprt->ops = &xs_tcp_ops;
  1404. if (to)
  1405. xprt->timeout = *to;
  1406. else
  1407. xprt_set_timeout(&xprt->timeout, 2, 60 * HZ);
  1408. xs_format_peer_addresses(xprt);
  1409. dprintk("RPC: set up transport to address %s\n",
  1410. xprt->address_strings[RPC_DISPLAY_ALL]);
  1411. return xprt;
  1412. }
  1413. /**
  1414. * init_socket_xprt - set up xprtsock's sysctls
  1415. *
  1416. */
  1417. int init_socket_xprt(void)
  1418. {
  1419. #ifdef RPC_DEBUG
  1420. if (!sunrpc_table_header)
  1421. sunrpc_table_header = register_sysctl_table(sunrpc_table);
  1422. #endif
  1423. return 0;
  1424. }
  1425. /**
  1426. * cleanup_socket_xprt - remove xprtsock's sysctls
  1427. *
  1428. */
  1429. void cleanup_socket_xprt(void)
  1430. {
  1431. #ifdef RPC_DEBUG
  1432. if (sunrpc_table_header) {
  1433. unregister_sysctl_table(sunrpc_table_header);
  1434. sunrpc_table_header = NULL;
  1435. }
  1436. #endif
  1437. }