xprtsock.c 84 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235
  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
  7. * TCP send fixes (C) 1998 Red Hat
  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. * IPv6 support contributed by Gilles Quillard, Bull Open Source, 2005.
  18. * <gilles.quillard@bull.net>
  19. */
  20. #include <linux/types.h>
  21. #include <linux/string.h>
  22. #include <linux/slab.h>
  23. #include <linux/module.h>
  24. #include <linux/capability.h>
  25. #include <linux/pagemap.h>
  26. #include <linux/errno.h>
  27. #include <linux/socket.h>
  28. #include <linux/in.h>
  29. #include <linux/net.h>
  30. #include <linux/mm.h>
  31. #include <linux/un.h>
  32. #include <linux/udp.h>
  33. #include <linux/tcp.h>
  34. #include <linux/sunrpc/clnt.h>
  35. #include <linux/sunrpc/addr.h>
  36. #include <linux/sunrpc/sched.h>
  37. #include <linux/sunrpc/svcsock.h>
  38. #include <linux/sunrpc/xprtsock.h>
  39. #include <linux/file.h>
  40. #ifdef CONFIG_SUNRPC_BACKCHANNEL
  41. #include <linux/sunrpc/bc_xprt.h>
  42. #endif
  43. #include <net/sock.h>
  44. #include <net/checksum.h>
  45. #include <net/udp.h>
  46. #include <net/tcp.h>
  47. #include <trace/events/sunrpc.h>
  48. #include "sunrpc.h"
  49. static void xs_close(struct rpc_xprt *xprt);
  50. /*
  51. * xprtsock tunables
  52. */
  53. static unsigned int xprt_udp_slot_table_entries = RPC_DEF_SLOT_TABLE;
  54. static unsigned int xprt_tcp_slot_table_entries = RPC_MIN_SLOT_TABLE;
  55. static unsigned int xprt_max_tcp_slot_table_entries = RPC_MAX_SLOT_TABLE;
  56. static unsigned int xprt_min_resvport = RPC_DEF_MIN_RESVPORT;
  57. static unsigned int xprt_max_resvport = RPC_DEF_MAX_RESVPORT;
  58. #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  59. #define XS_TCP_LINGER_TO (15U * HZ)
  60. static unsigned int xs_tcp_fin_timeout __read_mostly = XS_TCP_LINGER_TO;
  61. /*
  62. * We can register our own files under /proc/sys/sunrpc by
  63. * calling register_sysctl_table() again. The files in that
  64. * directory become the union of all files registered there.
  65. *
  66. * We simply need to make sure that we don't collide with
  67. * someone else's file names!
  68. */
  69. static unsigned int min_slot_table_size = RPC_MIN_SLOT_TABLE;
  70. static unsigned int max_slot_table_size = RPC_MAX_SLOT_TABLE;
  71. static unsigned int max_tcp_slot_table_limit = RPC_MAX_SLOT_TABLE_LIMIT;
  72. static unsigned int xprt_min_resvport_limit = RPC_MIN_RESVPORT;
  73. static unsigned int xprt_max_resvport_limit = RPC_MAX_RESVPORT;
  74. static struct ctl_table_header *sunrpc_table_header;
  75. /*
  76. * FIXME: changing the UDP slot table size should also resize the UDP
  77. * socket buffers for existing UDP transports
  78. */
  79. static struct ctl_table xs_tunables_table[] = {
  80. {
  81. .procname = "udp_slot_table_entries",
  82. .data = &xprt_udp_slot_table_entries,
  83. .maxlen = sizeof(unsigned int),
  84. .mode = 0644,
  85. .proc_handler = proc_dointvec_minmax,
  86. .extra1 = &min_slot_table_size,
  87. .extra2 = &max_slot_table_size
  88. },
  89. {
  90. .procname = "tcp_slot_table_entries",
  91. .data = &xprt_tcp_slot_table_entries,
  92. .maxlen = sizeof(unsigned int),
  93. .mode = 0644,
  94. .proc_handler = proc_dointvec_minmax,
  95. .extra1 = &min_slot_table_size,
  96. .extra2 = &max_slot_table_size
  97. },
  98. {
  99. .procname = "tcp_max_slot_table_entries",
  100. .data = &xprt_max_tcp_slot_table_entries,
  101. .maxlen = sizeof(unsigned int),
  102. .mode = 0644,
  103. .proc_handler = proc_dointvec_minmax,
  104. .extra1 = &min_slot_table_size,
  105. .extra2 = &max_tcp_slot_table_limit
  106. },
  107. {
  108. .procname = "min_resvport",
  109. .data = &xprt_min_resvport,
  110. .maxlen = sizeof(unsigned int),
  111. .mode = 0644,
  112. .proc_handler = proc_dointvec_minmax,
  113. .extra1 = &xprt_min_resvport_limit,
  114. .extra2 = &xprt_max_resvport
  115. },
  116. {
  117. .procname = "max_resvport",
  118. .data = &xprt_max_resvport,
  119. .maxlen = sizeof(unsigned int),
  120. .mode = 0644,
  121. .proc_handler = proc_dointvec_minmax,
  122. .extra1 = &xprt_min_resvport,
  123. .extra2 = &xprt_max_resvport_limit
  124. },
  125. {
  126. .procname = "tcp_fin_timeout",
  127. .data = &xs_tcp_fin_timeout,
  128. .maxlen = sizeof(xs_tcp_fin_timeout),
  129. .mode = 0644,
  130. .proc_handler = proc_dointvec_jiffies,
  131. },
  132. { },
  133. };
  134. static struct ctl_table sunrpc_table[] = {
  135. {
  136. .procname = "sunrpc",
  137. .mode = 0555,
  138. .child = xs_tunables_table
  139. },
  140. { },
  141. };
  142. #endif
  143. /*
  144. * Wait duration for a reply from the RPC portmapper.
  145. */
  146. #define XS_BIND_TO (60U * HZ)
  147. /*
  148. * Delay if a UDP socket connect error occurs. This is most likely some
  149. * kind of resource problem on the local host.
  150. */
  151. #define XS_UDP_REEST_TO (2U * HZ)
  152. /*
  153. * The reestablish timeout allows clients to delay for a bit before attempting
  154. * to reconnect to a server that just dropped our connection.
  155. *
  156. * We implement an exponential backoff when trying to reestablish a TCP
  157. * transport connection with the server. Some servers like to drop a TCP
  158. * connection when they are overworked, so we start with a short timeout and
  159. * increase over time if the server is down or not responding.
  160. */
  161. #define XS_TCP_INIT_REEST_TO (3U * HZ)
  162. #define XS_TCP_MAX_REEST_TO (5U * 60 * HZ)
  163. /*
  164. * TCP idle timeout; client drops the transport socket if it is idle
  165. * for this long. Note that we also timeout UDP sockets to prevent
  166. * holding port numbers when there is no RPC traffic.
  167. */
  168. #define XS_IDLE_DISC_TO (5U * 60 * HZ)
  169. #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  170. # undef RPC_DEBUG_DATA
  171. # define RPCDBG_FACILITY RPCDBG_TRANS
  172. #endif
  173. #ifdef RPC_DEBUG_DATA
  174. static void xs_pktdump(char *msg, u32 *packet, unsigned int count)
  175. {
  176. u8 *buf = (u8 *) packet;
  177. int j;
  178. dprintk("RPC: %s\n", msg);
  179. for (j = 0; j < count && j < 128; j += 4) {
  180. if (!(j & 31)) {
  181. if (j)
  182. dprintk("\n");
  183. dprintk("0x%04x ", j);
  184. }
  185. dprintk("%02x%02x%02x%02x ",
  186. buf[j], buf[j+1], buf[j+2], buf[j+3]);
  187. }
  188. dprintk("\n");
  189. }
  190. #else
  191. static inline void xs_pktdump(char *msg, u32 *packet, unsigned int count)
  192. {
  193. /* NOP */
  194. }
  195. #endif
  196. static inline struct rpc_xprt *xprt_from_sock(struct sock *sk)
  197. {
  198. return (struct rpc_xprt *) sk->sk_user_data;
  199. }
  200. static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
  201. {
  202. return (struct sockaddr *) &xprt->addr;
  203. }
  204. static inline struct sockaddr_un *xs_addr_un(struct rpc_xprt *xprt)
  205. {
  206. return (struct sockaddr_un *) &xprt->addr;
  207. }
  208. static inline struct sockaddr_in *xs_addr_in(struct rpc_xprt *xprt)
  209. {
  210. return (struct sockaddr_in *) &xprt->addr;
  211. }
  212. static inline struct sockaddr_in6 *xs_addr_in6(struct rpc_xprt *xprt)
  213. {
  214. return (struct sockaddr_in6 *) &xprt->addr;
  215. }
  216. static void xs_format_common_peer_addresses(struct rpc_xprt *xprt)
  217. {
  218. struct sockaddr *sap = xs_addr(xprt);
  219. struct sockaddr_in6 *sin6;
  220. struct sockaddr_in *sin;
  221. struct sockaddr_un *sun;
  222. char buf[128];
  223. switch (sap->sa_family) {
  224. case AF_LOCAL:
  225. sun = xs_addr_un(xprt);
  226. strlcpy(buf, sun->sun_path, sizeof(buf));
  227. xprt->address_strings[RPC_DISPLAY_ADDR] =
  228. kstrdup(buf, GFP_KERNEL);
  229. break;
  230. case AF_INET:
  231. (void)rpc_ntop(sap, buf, sizeof(buf));
  232. xprt->address_strings[RPC_DISPLAY_ADDR] =
  233. kstrdup(buf, GFP_KERNEL);
  234. sin = xs_addr_in(xprt);
  235. snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
  236. break;
  237. case AF_INET6:
  238. (void)rpc_ntop(sap, buf, sizeof(buf));
  239. xprt->address_strings[RPC_DISPLAY_ADDR] =
  240. kstrdup(buf, GFP_KERNEL);
  241. sin6 = xs_addr_in6(xprt);
  242. snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
  243. break;
  244. default:
  245. BUG();
  246. }
  247. xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
  248. }
  249. static void xs_format_common_peer_ports(struct rpc_xprt *xprt)
  250. {
  251. struct sockaddr *sap = xs_addr(xprt);
  252. char buf[128];
  253. snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
  254. xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
  255. snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
  256. xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
  257. }
  258. static void xs_format_peer_addresses(struct rpc_xprt *xprt,
  259. const char *protocol,
  260. const char *netid)
  261. {
  262. xprt->address_strings[RPC_DISPLAY_PROTO] = protocol;
  263. xprt->address_strings[RPC_DISPLAY_NETID] = netid;
  264. xs_format_common_peer_addresses(xprt);
  265. xs_format_common_peer_ports(xprt);
  266. }
  267. static void xs_update_peer_port(struct rpc_xprt *xprt)
  268. {
  269. kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]);
  270. kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
  271. xs_format_common_peer_ports(xprt);
  272. }
  273. static void xs_free_peer_addresses(struct rpc_xprt *xprt)
  274. {
  275. unsigned int i;
  276. for (i = 0; i < RPC_DISPLAY_MAX; i++)
  277. switch (i) {
  278. case RPC_DISPLAY_PROTO:
  279. case RPC_DISPLAY_NETID:
  280. continue;
  281. default:
  282. kfree(xprt->address_strings[i]);
  283. }
  284. }
  285. #define XS_SENDMSG_FLAGS (MSG_DONTWAIT | MSG_NOSIGNAL)
  286. static int xs_send_kvec(struct socket *sock, struct sockaddr *addr, int addrlen, struct kvec *vec, unsigned int base, int more)
  287. {
  288. struct msghdr msg = {
  289. .msg_name = addr,
  290. .msg_namelen = addrlen,
  291. .msg_flags = XS_SENDMSG_FLAGS | (more ? MSG_MORE : 0),
  292. };
  293. struct kvec iov = {
  294. .iov_base = vec->iov_base + base,
  295. .iov_len = vec->iov_len - base,
  296. };
  297. if (iov.iov_len != 0)
  298. return kernel_sendmsg(sock, &msg, &iov, 1, iov.iov_len);
  299. return kernel_sendmsg(sock, &msg, NULL, 0, 0);
  300. }
  301. static int xs_send_pagedata(struct socket *sock, struct xdr_buf *xdr, unsigned int base, int more, bool zerocopy, int *sent_p)
  302. {
  303. ssize_t (*do_sendpage)(struct socket *sock, struct page *page,
  304. int offset, size_t size, int flags);
  305. struct page **ppage;
  306. unsigned int remainder;
  307. int err;
  308. remainder = xdr->page_len - base;
  309. base += xdr->page_base;
  310. ppage = xdr->pages + (base >> PAGE_SHIFT);
  311. base &= ~PAGE_MASK;
  312. do_sendpage = sock->ops->sendpage;
  313. if (!zerocopy)
  314. do_sendpage = sock_no_sendpage;
  315. for(;;) {
  316. unsigned int len = min_t(unsigned int, PAGE_SIZE - base, remainder);
  317. int flags = XS_SENDMSG_FLAGS;
  318. remainder -= len;
  319. if (more)
  320. flags |= MSG_MORE;
  321. if (remainder != 0)
  322. flags |= MSG_SENDPAGE_NOTLAST | MSG_MORE;
  323. err = do_sendpage(sock, *ppage, base, len, flags);
  324. if (remainder == 0 || err != len)
  325. break;
  326. *sent_p += err;
  327. ppage++;
  328. base = 0;
  329. }
  330. if (err > 0) {
  331. *sent_p += err;
  332. err = 0;
  333. }
  334. return err;
  335. }
  336. /**
  337. * xs_sendpages - write pages directly to a socket
  338. * @sock: socket to send on
  339. * @addr: UDP only -- address of destination
  340. * @addrlen: UDP only -- length of destination address
  341. * @xdr: buffer containing this request
  342. * @base: starting position in the buffer
  343. * @zerocopy: true if it is safe to use sendpage()
  344. * @sent_p: return the total number of bytes successfully queued for sending
  345. *
  346. */
  347. static int xs_sendpages(struct socket *sock, struct sockaddr *addr, int addrlen, struct xdr_buf *xdr, unsigned int base, bool zerocopy, int *sent_p)
  348. {
  349. unsigned int remainder = xdr->len - base;
  350. int err = 0;
  351. int sent = 0;
  352. if (unlikely(!sock))
  353. return -ENOTSOCK;
  354. if (base != 0) {
  355. addr = NULL;
  356. addrlen = 0;
  357. }
  358. if (base < xdr->head[0].iov_len || addr != NULL) {
  359. unsigned int len = xdr->head[0].iov_len - base;
  360. remainder -= len;
  361. err = xs_send_kvec(sock, addr, addrlen, &xdr->head[0], base, remainder != 0);
  362. if (remainder == 0 || err != len)
  363. goto out;
  364. *sent_p += err;
  365. base = 0;
  366. } else
  367. base -= xdr->head[0].iov_len;
  368. if (base < xdr->page_len) {
  369. unsigned int len = xdr->page_len - base;
  370. remainder -= len;
  371. err = xs_send_pagedata(sock, xdr, base, remainder != 0, zerocopy, &sent);
  372. *sent_p += sent;
  373. if (remainder == 0 || sent != len)
  374. goto out;
  375. base = 0;
  376. } else
  377. base -= xdr->page_len;
  378. if (base >= xdr->tail[0].iov_len)
  379. return 0;
  380. err = xs_send_kvec(sock, NULL, 0, &xdr->tail[0], base, 0);
  381. out:
  382. if (err > 0) {
  383. *sent_p += err;
  384. err = 0;
  385. }
  386. return err;
  387. }
  388. static void xs_nospace_callback(struct rpc_task *task)
  389. {
  390. struct sock_xprt *transport = container_of(task->tk_rqstp->rq_xprt, struct sock_xprt, xprt);
  391. transport->inet->sk_write_pending--;
  392. }
  393. /**
  394. * xs_nospace - place task on wait queue if transmit was incomplete
  395. * @task: task to put to sleep
  396. *
  397. */
  398. static int xs_nospace(struct rpc_task *task)
  399. {
  400. struct rpc_rqst *req = task->tk_rqstp;
  401. struct rpc_xprt *xprt = req->rq_xprt;
  402. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  403. struct sock *sk = transport->inet;
  404. int ret = -EAGAIN;
  405. dprintk("RPC: %5u xmit incomplete (%u left of %u)\n",
  406. task->tk_pid, req->rq_slen - req->rq_bytes_sent,
  407. req->rq_slen);
  408. /* Protect against races with write_space */
  409. spin_lock_bh(&xprt->transport_lock);
  410. /* Don't race with disconnect */
  411. if (xprt_connected(xprt)) {
  412. /* wait for more buffer space */
  413. sk->sk_write_pending++;
  414. xprt_wait_for_buffer_space(task, xs_nospace_callback);
  415. } else
  416. ret = -ENOTCONN;
  417. spin_unlock_bh(&xprt->transport_lock);
  418. /* Race breaker in case memory is freed before above code is called */
  419. sk->sk_write_space(sk);
  420. return ret;
  421. }
  422. /*
  423. * Construct a stream transport record marker in @buf.
  424. */
  425. static inline void xs_encode_stream_record_marker(struct xdr_buf *buf)
  426. {
  427. u32 reclen = buf->len - sizeof(rpc_fraghdr);
  428. rpc_fraghdr *base = buf->head[0].iov_base;
  429. *base = cpu_to_be32(RPC_LAST_STREAM_FRAGMENT | reclen);
  430. }
  431. /**
  432. * xs_local_send_request - write an RPC request to an AF_LOCAL socket
  433. * @task: RPC task that manages the state of an RPC request
  434. *
  435. * Return values:
  436. * 0: The request has been sent
  437. * EAGAIN: The socket was blocked, please call again later to
  438. * complete the request
  439. * ENOTCONN: Caller needs to invoke connect logic then call again
  440. * other: Some other error occured, the request was not sent
  441. */
  442. static int xs_local_send_request(struct rpc_task *task)
  443. {
  444. struct rpc_rqst *req = task->tk_rqstp;
  445. struct rpc_xprt *xprt = req->rq_xprt;
  446. struct sock_xprt *transport =
  447. container_of(xprt, struct sock_xprt, xprt);
  448. struct xdr_buf *xdr = &req->rq_snd_buf;
  449. int status;
  450. int sent = 0;
  451. xs_encode_stream_record_marker(&req->rq_snd_buf);
  452. xs_pktdump("packet data:",
  453. req->rq_svec->iov_base, req->rq_svec->iov_len);
  454. status = xs_sendpages(transport->sock, NULL, 0, xdr, req->rq_bytes_sent,
  455. true, &sent);
  456. dprintk("RPC: %s(%u) = %d\n",
  457. __func__, xdr->len - req->rq_bytes_sent, status);
  458. if (status == -EAGAIN && sock_writeable(transport->inet))
  459. status = -ENOBUFS;
  460. if (likely(sent > 0) || status == 0) {
  461. req->rq_bytes_sent += sent;
  462. req->rq_xmit_bytes_sent += sent;
  463. if (likely(req->rq_bytes_sent >= req->rq_slen)) {
  464. req->rq_bytes_sent = 0;
  465. return 0;
  466. }
  467. status = -EAGAIN;
  468. }
  469. switch (status) {
  470. case -ENOBUFS:
  471. break;
  472. case -EAGAIN:
  473. status = xs_nospace(task);
  474. break;
  475. default:
  476. dprintk("RPC: sendmsg returned unrecognized error %d\n",
  477. -status);
  478. case -EPIPE:
  479. xs_close(xprt);
  480. status = -ENOTCONN;
  481. }
  482. return status;
  483. }
  484. /**
  485. * xs_udp_send_request - write an RPC request to a UDP socket
  486. * @task: address of RPC task that manages the state of an RPC request
  487. *
  488. * Return values:
  489. * 0: The request has been sent
  490. * EAGAIN: The socket was blocked, please call again later to
  491. * complete the request
  492. * ENOTCONN: Caller needs to invoke connect logic then call again
  493. * other: Some other error occurred, the request was not sent
  494. */
  495. static int xs_udp_send_request(struct rpc_task *task)
  496. {
  497. struct rpc_rqst *req = task->tk_rqstp;
  498. struct rpc_xprt *xprt = req->rq_xprt;
  499. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  500. struct xdr_buf *xdr = &req->rq_snd_buf;
  501. int sent = 0;
  502. int status;
  503. xs_pktdump("packet data:",
  504. req->rq_svec->iov_base,
  505. req->rq_svec->iov_len);
  506. if (!xprt_bound(xprt))
  507. return -ENOTCONN;
  508. status = xs_sendpages(transport->sock, xs_addr(xprt), xprt->addrlen,
  509. xdr, req->rq_bytes_sent, true, &sent);
  510. dprintk("RPC: xs_udp_send_request(%u) = %d\n",
  511. xdr->len - req->rq_bytes_sent, status);
  512. /* firewall is blocking us, don't return -EAGAIN or we end up looping */
  513. if (status == -EPERM)
  514. goto process_status;
  515. if (status == -EAGAIN && sock_writeable(transport->inet))
  516. status = -ENOBUFS;
  517. if (sent > 0 || status == 0) {
  518. req->rq_xmit_bytes_sent += sent;
  519. if (sent >= req->rq_slen)
  520. return 0;
  521. /* Still some bytes left; set up for a retry later. */
  522. status = -EAGAIN;
  523. }
  524. process_status:
  525. switch (status) {
  526. case -ENOTSOCK:
  527. status = -ENOTCONN;
  528. /* Should we call xs_close() here? */
  529. break;
  530. case -EAGAIN:
  531. status = xs_nospace(task);
  532. break;
  533. case -ENETUNREACH:
  534. case -ENOBUFS:
  535. case -EPIPE:
  536. case -ECONNREFUSED:
  537. case -EPERM:
  538. /* When the server has died, an ICMP port unreachable message
  539. * prompts ECONNREFUSED. */
  540. break;
  541. default:
  542. dprintk("RPC: sendmsg returned unrecognized error %d\n",
  543. -status);
  544. }
  545. return status;
  546. }
  547. /**
  548. * xs_tcp_send_request - write an RPC request to a TCP socket
  549. * @task: address of RPC task that manages the state of an RPC request
  550. *
  551. * Return values:
  552. * 0: The request has been sent
  553. * EAGAIN: The socket was blocked, please call again later to
  554. * complete the request
  555. * ENOTCONN: Caller needs to invoke connect logic then call again
  556. * other: Some other error occurred, the request was not sent
  557. *
  558. * XXX: In the case of soft timeouts, should we eventually give up
  559. * if sendmsg is not able to make progress?
  560. */
  561. static int xs_tcp_send_request(struct rpc_task *task)
  562. {
  563. struct rpc_rqst *req = task->tk_rqstp;
  564. struct rpc_xprt *xprt = req->rq_xprt;
  565. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  566. struct xdr_buf *xdr = &req->rq_snd_buf;
  567. bool zerocopy = true;
  568. bool vm_wait = false;
  569. int status;
  570. int sent;
  571. xs_encode_stream_record_marker(&req->rq_snd_buf);
  572. xs_pktdump("packet data:",
  573. req->rq_svec->iov_base,
  574. req->rq_svec->iov_len);
  575. /* Don't use zero copy if this is a resend. If the RPC call
  576. * completes while the socket holds a reference to the pages,
  577. * then we may end up resending corrupted data.
  578. */
  579. if (task->tk_flags & RPC_TASK_SENT)
  580. zerocopy = false;
  581. /* Continue transmitting the packet/record. We must be careful
  582. * to cope with writespace callbacks arriving _after_ we have
  583. * called sendmsg(). */
  584. while (1) {
  585. sent = 0;
  586. status = xs_sendpages(transport->sock, NULL, 0, xdr,
  587. req->rq_bytes_sent, zerocopy, &sent);
  588. dprintk("RPC: xs_tcp_send_request(%u) = %d\n",
  589. xdr->len - req->rq_bytes_sent, status);
  590. /* If we've sent the entire packet, immediately
  591. * reset the count of bytes sent. */
  592. req->rq_bytes_sent += sent;
  593. req->rq_xmit_bytes_sent += sent;
  594. if (likely(req->rq_bytes_sent >= req->rq_slen)) {
  595. req->rq_bytes_sent = 0;
  596. return 0;
  597. }
  598. WARN_ON_ONCE(sent == 0 && status == 0);
  599. if (status == -EAGAIN ) {
  600. /*
  601. * Return EAGAIN if we're sure we're hitting the
  602. * socket send buffer limits.
  603. */
  604. if (test_bit(SOCK_NOSPACE, &transport->sock->flags))
  605. break;
  606. /*
  607. * Did we hit a memory allocation failure?
  608. */
  609. if (sent == 0) {
  610. status = -ENOBUFS;
  611. if (vm_wait)
  612. break;
  613. /* Retry, knowing now that we're below the
  614. * socket send buffer limit
  615. */
  616. vm_wait = true;
  617. }
  618. continue;
  619. }
  620. if (status < 0)
  621. break;
  622. vm_wait = false;
  623. }
  624. switch (status) {
  625. case -ENOTSOCK:
  626. status = -ENOTCONN;
  627. /* Should we call xs_close() here? */
  628. break;
  629. case -EAGAIN:
  630. status = xs_nospace(task);
  631. break;
  632. case -ECONNRESET:
  633. case -ECONNREFUSED:
  634. case -ENOTCONN:
  635. case -EADDRINUSE:
  636. case -ENOBUFS:
  637. case -EPIPE:
  638. break;
  639. default:
  640. dprintk("RPC: sendmsg returned unrecognized error %d\n",
  641. -status);
  642. }
  643. return status;
  644. }
  645. /**
  646. * xs_tcp_release_xprt - clean up after a tcp transmission
  647. * @xprt: transport
  648. * @task: rpc task
  649. *
  650. * This cleans up if an error causes us to abort the transmission of a request.
  651. * In this case, the socket may need to be reset in order to avoid confusing
  652. * the server.
  653. */
  654. static void xs_tcp_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
  655. {
  656. struct rpc_rqst *req;
  657. if (task != xprt->snd_task)
  658. return;
  659. if (task == NULL)
  660. goto out_release;
  661. req = task->tk_rqstp;
  662. if (req == NULL)
  663. goto out_release;
  664. if (req->rq_bytes_sent == 0)
  665. goto out_release;
  666. if (req->rq_bytes_sent == req->rq_snd_buf.len)
  667. goto out_release;
  668. set_bit(XPRT_CLOSE_WAIT, &xprt->state);
  669. out_release:
  670. xprt_release_xprt(xprt, task);
  671. }
  672. static void xs_save_old_callbacks(struct sock_xprt *transport, struct sock *sk)
  673. {
  674. transport->old_data_ready = sk->sk_data_ready;
  675. transport->old_state_change = sk->sk_state_change;
  676. transport->old_write_space = sk->sk_write_space;
  677. transport->old_error_report = sk->sk_error_report;
  678. }
  679. static void xs_restore_old_callbacks(struct sock_xprt *transport, struct sock *sk)
  680. {
  681. sk->sk_data_ready = transport->old_data_ready;
  682. sk->sk_state_change = transport->old_state_change;
  683. sk->sk_write_space = transport->old_write_space;
  684. sk->sk_error_report = transport->old_error_report;
  685. }
  686. static void xs_sock_reset_state_flags(struct rpc_xprt *xprt)
  687. {
  688. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  689. clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
  690. }
  691. static void xs_sock_reset_connection_flags(struct rpc_xprt *xprt)
  692. {
  693. smp_mb__before_atomic();
  694. clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
  695. clear_bit(XPRT_CLOSING, &xprt->state);
  696. xs_sock_reset_state_flags(xprt);
  697. smp_mb__after_atomic();
  698. }
  699. static void xs_sock_mark_closed(struct rpc_xprt *xprt)
  700. {
  701. xs_sock_reset_connection_flags(xprt);
  702. /* Mark transport as closed and wake up all pending tasks */
  703. xprt_disconnect_done(xprt);
  704. }
  705. /**
  706. * xs_error_report - callback to handle TCP socket state errors
  707. * @sk: socket
  708. *
  709. * Note: we don't call sock_error() since there may be a rpc_task
  710. * using the socket, and so we don't want to clear sk->sk_err.
  711. */
  712. static void xs_error_report(struct sock *sk)
  713. {
  714. struct rpc_xprt *xprt;
  715. int err;
  716. read_lock_bh(&sk->sk_callback_lock);
  717. if (!(xprt = xprt_from_sock(sk)))
  718. goto out;
  719. err = -sk->sk_err;
  720. if (err == 0)
  721. goto out;
  722. /* Is this a reset event? */
  723. if (sk->sk_state == TCP_CLOSE)
  724. xs_sock_mark_closed(xprt);
  725. dprintk("RPC: xs_error_report client %p, error=%d...\n",
  726. xprt, -err);
  727. trace_rpc_socket_error(xprt, sk->sk_socket, err);
  728. xprt_wake_pending_tasks(xprt, err);
  729. out:
  730. read_unlock_bh(&sk->sk_callback_lock);
  731. }
  732. static void xs_reset_transport(struct sock_xprt *transport)
  733. {
  734. struct socket *sock = transport->sock;
  735. struct sock *sk = transport->inet;
  736. struct rpc_xprt *xprt = &transport->xprt;
  737. if (sk == NULL)
  738. return;
  739. if (atomic_read(&transport->xprt.swapper))
  740. sk_clear_memalloc(sk);
  741. kernel_sock_shutdown(sock, SHUT_RDWR);
  742. mutex_lock(&transport->recv_mutex);
  743. write_lock_bh(&sk->sk_callback_lock);
  744. transport->inet = NULL;
  745. transport->sock = NULL;
  746. sk->sk_user_data = NULL;
  747. xs_restore_old_callbacks(transport, sk);
  748. xprt_clear_connected(xprt);
  749. write_unlock_bh(&sk->sk_callback_lock);
  750. xs_sock_reset_connection_flags(xprt);
  751. mutex_unlock(&transport->recv_mutex);
  752. trace_rpc_socket_close(xprt, sock);
  753. sock_release(sock);
  754. }
  755. /**
  756. * xs_close - close a socket
  757. * @xprt: transport
  758. *
  759. * This is used when all requests are complete; ie, no DRC state remains
  760. * on the server we want to save.
  761. *
  762. * The caller _must_ be holding XPRT_LOCKED in order to avoid issues with
  763. * xs_reset_transport() zeroing the socket from underneath a writer.
  764. */
  765. static void xs_close(struct rpc_xprt *xprt)
  766. {
  767. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  768. dprintk("RPC: xs_close xprt %p\n", xprt);
  769. xs_reset_transport(transport);
  770. xprt->reestablish_timeout = 0;
  771. xprt_disconnect_done(xprt);
  772. }
  773. static void xs_inject_disconnect(struct rpc_xprt *xprt)
  774. {
  775. dprintk("RPC: injecting transport disconnect on xprt=%p\n",
  776. xprt);
  777. xprt_disconnect_done(xprt);
  778. }
  779. static void xs_xprt_free(struct rpc_xprt *xprt)
  780. {
  781. xs_free_peer_addresses(xprt);
  782. xprt_free(xprt);
  783. }
  784. /**
  785. * xs_destroy - prepare to shutdown a transport
  786. * @xprt: doomed transport
  787. *
  788. */
  789. static void xs_destroy(struct rpc_xprt *xprt)
  790. {
  791. struct sock_xprt *transport = container_of(xprt,
  792. struct sock_xprt, xprt);
  793. dprintk("RPC: xs_destroy xprt %p\n", xprt);
  794. cancel_delayed_work_sync(&transport->connect_worker);
  795. xs_close(xprt);
  796. cancel_work_sync(&transport->recv_worker);
  797. xs_xprt_free(xprt);
  798. module_put(THIS_MODULE);
  799. }
  800. static int xs_local_copy_to_xdr(struct xdr_buf *xdr, struct sk_buff *skb)
  801. {
  802. struct xdr_skb_reader desc = {
  803. .skb = skb,
  804. .offset = sizeof(rpc_fraghdr),
  805. .count = skb->len - sizeof(rpc_fraghdr),
  806. };
  807. if (xdr_partial_copy_from_skb(xdr, 0, &desc, xdr_skb_read_bits) < 0)
  808. return -1;
  809. if (desc.count)
  810. return -1;
  811. return 0;
  812. }
  813. /**
  814. * xs_local_data_read_skb
  815. * @xprt: transport
  816. * @sk: socket
  817. * @skb: skbuff
  818. *
  819. * Currently this assumes we can read the whole reply in a single gulp.
  820. */
  821. static void xs_local_data_read_skb(struct rpc_xprt *xprt,
  822. struct sock *sk,
  823. struct sk_buff *skb)
  824. {
  825. struct rpc_task *task;
  826. struct rpc_rqst *rovr;
  827. int repsize, copied;
  828. u32 _xid;
  829. __be32 *xp;
  830. repsize = skb->len - sizeof(rpc_fraghdr);
  831. if (repsize < 4) {
  832. dprintk("RPC: impossible RPC reply size %d\n", repsize);
  833. return;
  834. }
  835. /* Copy the XID from the skb... */
  836. xp = skb_header_pointer(skb, sizeof(rpc_fraghdr), sizeof(_xid), &_xid);
  837. if (xp == NULL)
  838. return;
  839. /* Look up and lock the request corresponding to the given XID */
  840. spin_lock_bh(&xprt->transport_lock);
  841. rovr = xprt_lookup_rqst(xprt, *xp);
  842. if (!rovr)
  843. goto out_unlock;
  844. task = rovr->rq_task;
  845. copied = rovr->rq_private_buf.buflen;
  846. if (copied > repsize)
  847. copied = repsize;
  848. if (xs_local_copy_to_xdr(&rovr->rq_private_buf, skb)) {
  849. dprintk("RPC: sk_buff copy failed\n");
  850. goto out_unlock;
  851. }
  852. xprt_complete_rqst(task, copied);
  853. out_unlock:
  854. spin_unlock_bh(&xprt->transport_lock);
  855. }
  856. static void xs_local_data_receive(struct sock_xprt *transport)
  857. {
  858. struct sk_buff *skb;
  859. struct sock *sk;
  860. int err;
  861. mutex_lock(&transport->recv_mutex);
  862. sk = transport->inet;
  863. if (sk == NULL)
  864. goto out;
  865. for (;;) {
  866. skb = skb_recv_datagram(sk, 0, 1, &err);
  867. if (skb != NULL) {
  868. xs_local_data_read_skb(&transport->xprt, sk, skb);
  869. skb_free_datagram(sk, skb);
  870. continue;
  871. }
  872. if (!test_and_clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
  873. break;
  874. }
  875. out:
  876. mutex_unlock(&transport->recv_mutex);
  877. }
  878. static void xs_local_data_receive_workfn(struct work_struct *work)
  879. {
  880. struct sock_xprt *transport =
  881. container_of(work, struct sock_xprt, recv_worker);
  882. xs_local_data_receive(transport);
  883. }
  884. /**
  885. * xs_udp_data_read_skb - receive callback for UDP sockets
  886. * @xprt: transport
  887. * @sk: socket
  888. * @skb: skbuff
  889. *
  890. */
  891. static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
  892. struct sock *sk,
  893. struct sk_buff *skb)
  894. {
  895. struct rpc_task *task;
  896. struct rpc_rqst *rovr;
  897. int repsize, copied;
  898. u32 _xid;
  899. __be32 *xp;
  900. repsize = skb->len;
  901. if (repsize < 4) {
  902. dprintk("RPC: impossible RPC reply size %d!\n", repsize);
  903. return;
  904. }
  905. /* Copy the XID from the skb... */
  906. xp = skb_header_pointer(skb, 0, sizeof(_xid), &_xid);
  907. if (xp == NULL)
  908. return;
  909. /* Look up and lock the request corresponding to the given XID */
  910. spin_lock_bh(&xprt->transport_lock);
  911. rovr = xprt_lookup_rqst(xprt, *xp);
  912. if (!rovr)
  913. goto out_unlock;
  914. task = rovr->rq_task;
  915. if ((copied = rovr->rq_private_buf.buflen) > repsize)
  916. copied = repsize;
  917. /* Suck it into the iovec, verify checksum if not done by hw. */
  918. if (csum_partial_copy_to_xdr(&rovr->rq_private_buf, skb)) {
  919. __UDPX_INC_STATS(sk, UDP_MIB_INERRORS);
  920. goto out_unlock;
  921. }
  922. __UDPX_INC_STATS(sk, UDP_MIB_INDATAGRAMS);
  923. xprt_adjust_cwnd(xprt, task, copied);
  924. xprt_complete_rqst(task, copied);
  925. out_unlock:
  926. spin_unlock_bh(&xprt->transport_lock);
  927. }
  928. static void xs_udp_data_receive(struct sock_xprt *transport)
  929. {
  930. struct sk_buff *skb;
  931. struct sock *sk;
  932. int err;
  933. mutex_lock(&transport->recv_mutex);
  934. sk = transport->inet;
  935. if (sk == NULL)
  936. goto out;
  937. for (;;) {
  938. skb = skb_recv_datagram(sk, 0, 1, &err);
  939. if (skb != NULL) {
  940. xs_udp_data_read_skb(&transport->xprt, sk, skb);
  941. skb_free_datagram(sk, skb);
  942. continue;
  943. }
  944. if (!test_and_clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
  945. break;
  946. }
  947. out:
  948. mutex_unlock(&transport->recv_mutex);
  949. }
  950. static void xs_udp_data_receive_workfn(struct work_struct *work)
  951. {
  952. struct sock_xprt *transport =
  953. container_of(work, struct sock_xprt, recv_worker);
  954. xs_udp_data_receive(transport);
  955. }
  956. /**
  957. * xs_data_ready - "data ready" callback for UDP sockets
  958. * @sk: socket with data to read
  959. *
  960. */
  961. static void xs_data_ready(struct sock *sk)
  962. {
  963. struct rpc_xprt *xprt;
  964. read_lock_bh(&sk->sk_callback_lock);
  965. dprintk("RPC: xs_data_ready...\n");
  966. xprt = xprt_from_sock(sk);
  967. if (xprt != NULL) {
  968. struct sock_xprt *transport = container_of(xprt,
  969. struct sock_xprt, xprt);
  970. transport->old_data_ready(sk);
  971. /* Any data means we had a useful conversation, so
  972. * then we don't need to delay the next reconnect
  973. */
  974. if (xprt->reestablish_timeout)
  975. xprt->reestablish_timeout = 0;
  976. if (!test_and_set_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
  977. queue_work(xprtiod_workqueue, &transport->recv_worker);
  978. }
  979. read_unlock_bh(&sk->sk_callback_lock);
  980. }
  981. /*
  982. * Helper function to force a TCP close if the server is sending
  983. * junk and/or it has put us in CLOSE_WAIT
  984. */
  985. static void xs_tcp_force_close(struct rpc_xprt *xprt)
  986. {
  987. xprt_force_disconnect(xprt);
  988. }
  989. static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
  990. {
  991. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  992. size_t len, used;
  993. char *p;
  994. p = ((char *) &transport->tcp_fraghdr) + transport->tcp_offset;
  995. len = sizeof(transport->tcp_fraghdr) - transport->tcp_offset;
  996. used = xdr_skb_read_bits(desc, p, len);
  997. transport->tcp_offset += used;
  998. if (used != len)
  999. return;
  1000. transport->tcp_reclen = ntohl(transport->tcp_fraghdr);
  1001. if (transport->tcp_reclen & RPC_LAST_STREAM_FRAGMENT)
  1002. transport->tcp_flags |= TCP_RCV_LAST_FRAG;
  1003. else
  1004. transport->tcp_flags &= ~TCP_RCV_LAST_FRAG;
  1005. transport->tcp_reclen &= RPC_FRAGMENT_SIZE_MASK;
  1006. transport->tcp_flags &= ~TCP_RCV_COPY_FRAGHDR;
  1007. transport->tcp_offset = 0;
  1008. /* Sanity check of the record length */
  1009. if (unlikely(transport->tcp_reclen < 8)) {
  1010. dprintk("RPC: invalid TCP record fragment length\n");
  1011. xs_tcp_force_close(xprt);
  1012. return;
  1013. }
  1014. dprintk("RPC: reading TCP record fragment of length %d\n",
  1015. transport->tcp_reclen);
  1016. }
  1017. static void xs_tcp_check_fraghdr(struct sock_xprt *transport)
  1018. {
  1019. if (transport->tcp_offset == transport->tcp_reclen) {
  1020. transport->tcp_flags |= TCP_RCV_COPY_FRAGHDR;
  1021. transport->tcp_offset = 0;
  1022. if (transport->tcp_flags & TCP_RCV_LAST_FRAG) {
  1023. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  1024. transport->tcp_flags |= TCP_RCV_COPY_XID;
  1025. transport->tcp_copied = 0;
  1026. }
  1027. }
  1028. }
  1029. static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_reader *desc)
  1030. {
  1031. size_t len, used;
  1032. char *p;
  1033. len = sizeof(transport->tcp_xid) - transport->tcp_offset;
  1034. dprintk("RPC: reading XID (%Zu bytes)\n", len);
  1035. p = ((char *) &transport->tcp_xid) + transport->tcp_offset;
  1036. used = xdr_skb_read_bits(desc, p, len);
  1037. transport->tcp_offset += used;
  1038. if (used != len)
  1039. return;
  1040. transport->tcp_flags &= ~TCP_RCV_COPY_XID;
  1041. transport->tcp_flags |= TCP_RCV_READ_CALLDIR;
  1042. transport->tcp_copied = 4;
  1043. dprintk("RPC: reading %s XID %08x\n",
  1044. (transport->tcp_flags & TCP_RPC_REPLY) ? "reply for"
  1045. : "request with",
  1046. ntohl(transport->tcp_xid));
  1047. xs_tcp_check_fraghdr(transport);
  1048. }
  1049. static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
  1050. struct xdr_skb_reader *desc)
  1051. {
  1052. size_t len, used;
  1053. u32 offset;
  1054. char *p;
  1055. /*
  1056. * We want transport->tcp_offset to be 8 at the end of this routine
  1057. * (4 bytes for the xid and 4 bytes for the call/reply flag).
  1058. * When this function is called for the first time,
  1059. * transport->tcp_offset is 4 (after having already read the xid).
  1060. */
  1061. offset = transport->tcp_offset - sizeof(transport->tcp_xid);
  1062. len = sizeof(transport->tcp_calldir) - offset;
  1063. dprintk("RPC: reading CALL/REPLY flag (%Zu bytes)\n", len);
  1064. p = ((char *) &transport->tcp_calldir) + offset;
  1065. used = xdr_skb_read_bits(desc, p, len);
  1066. transport->tcp_offset += used;
  1067. if (used != len)
  1068. return;
  1069. transport->tcp_flags &= ~TCP_RCV_READ_CALLDIR;
  1070. /*
  1071. * We don't yet have the XDR buffer, so we will write the calldir
  1072. * out after we get the buffer from the 'struct rpc_rqst'
  1073. */
  1074. switch (ntohl(transport->tcp_calldir)) {
  1075. case RPC_REPLY:
  1076. transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
  1077. transport->tcp_flags |= TCP_RCV_COPY_DATA;
  1078. transport->tcp_flags |= TCP_RPC_REPLY;
  1079. break;
  1080. case RPC_CALL:
  1081. transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
  1082. transport->tcp_flags |= TCP_RCV_COPY_DATA;
  1083. transport->tcp_flags &= ~TCP_RPC_REPLY;
  1084. break;
  1085. default:
  1086. dprintk("RPC: invalid request message type\n");
  1087. xs_tcp_force_close(&transport->xprt);
  1088. }
  1089. xs_tcp_check_fraghdr(transport);
  1090. }
  1091. static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
  1092. struct xdr_skb_reader *desc,
  1093. struct rpc_rqst *req)
  1094. {
  1095. struct sock_xprt *transport =
  1096. container_of(xprt, struct sock_xprt, xprt);
  1097. struct xdr_buf *rcvbuf;
  1098. size_t len;
  1099. ssize_t r;
  1100. rcvbuf = &req->rq_private_buf;
  1101. if (transport->tcp_flags & TCP_RCV_COPY_CALLDIR) {
  1102. /*
  1103. * Save the RPC direction in the XDR buffer
  1104. */
  1105. memcpy(rcvbuf->head[0].iov_base + transport->tcp_copied,
  1106. &transport->tcp_calldir,
  1107. sizeof(transport->tcp_calldir));
  1108. transport->tcp_copied += sizeof(transport->tcp_calldir);
  1109. transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
  1110. }
  1111. len = desc->count;
  1112. if (len > transport->tcp_reclen - transport->tcp_offset) {
  1113. struct xdr_skb_reader my_desc;
  1114. len = transport->tcp_reclen - transport->tcp_offset;
  1115. memcpy(&my_desc, desc, sizeof(my_desc));
  1116. my_desc.count = len;
  1117. r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
  1118. &my_desc, xdr_skb_read_bits);
  1119. desc->count -= r;
  1120. desc->offset += r;
  1121. } else
  1122. r = xdr_partial_copy_from_skb(rcvbuf, transport->tcp_copied,
  1123. desc, xdr_skb_read_bits);
  1124. if (r > 0) {
  1125. transport->tcp_copied += r;
  1126. transport->tcp_offset += r;
  1127. }
  1128. if (r != len) {
  1129. /* Error when copying to the receive buffer,
  1130. * usually because we weren't able to allocate
  1131. * additional buffer pages. All we can do now
  1132. * is turn off TCP_RCV_COPY_DATA, so the request
  1133. * will not receive any additional updates,
  1134. * and time out.
  1135. * Any remaining data from this record will
  1136. * be discarded.
  1137. */
  1138. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  1139. dprintk("RPC: XID %08x truncated request\n",
  1140. ntohl(transport->tcp_xid));
  1141. dprintk("RPC: xprt = %p, tcp_copied = %lu, "
  1142. "tcp_offset = %u, tcp_reclen = %u\n",
  1143. xprt, transport->tcp_copied,
  1144. transport->tcp_offset, transport->tcp_reclen);
  1145. return;
  1146. }
  1147. dprintk("RPC: XID %08x read %Zd bytes\n",
  1148. ntohl(transport->tcp_xid), r);
  1149. dprintk("RPC: xprt = %p, tcp_copied = %lu, tcp_offset = %u, "
  1150. "tcp_reclen = %u\n", xprt, transport->tcp_copied,
  1151. transport->tcp_offset, transport->tcp_reclen);
  1152. if (transport->tcp_copied == req->rq_private_buf.buflen)
  1153. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  1154. else if (transport->tcp_offset == transport->tcp_reclen) {
  1155. if (transport->tcp_flags & TCP_RCV_LAST_FRAG)
  1156. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  1157. }
  1158. }
  1159. /*
  1160. * Finds the request corresponding to the RPC xid and invokes the common
  1161. * tcp read code to read the data.
  1162. */
  1163. static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
  1164. struct xdr_skb_reader *desc)
  1165. {
  1166. struct sock_xprt *transport =
  1167. container_of(xprt, struct sock_xprt, xprt);
  1168. struct rpc_rqst *req;
  1169. dprintk("RPC: read reply XID %08x\n", ntohl(transport->tcp_xid));
  1170. /* Find and lock the request corresponding to this xid */
  1171. spin_lock_bh(&xprt->transport_lock);
  1172. req = xprt_lookup_rqst(xprt, transport->tcp_xid);
  1173. if (!req) {
  1174. dprintk("RPC: XID %08x request not found!\n",
  1175. ntohl(transport->tcp_xid));
  1176. spin_unlock_bh(&xprt->transport_lock);
  1177. return -1;
  1178. }
  1179. xs_tcp_read_common(xprt, desc, req);
  1180. if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
  1181. xprt_complete_rqst(req->rq_task, transport->tcp_copied);
  1182. spin_unlock_bh(&xprt->transport_lock);
  1183. return 0;
  1184. }
  1185. #if defined(CONFIG_SUNRPC_BACKCHANNEL)
  1186. /*
  1187. * Obtains an rpc_rqst previously allocated and invokes the common
  1188. * tcp read code to read the data. The result is placed in the callback
  1189. * queue.
  1190. * If we're unable to obtain the rpc_rqst we schedule the closing of the
  1191. * connection and return -1.
  1192. */
  1193. static int xs_tcp_read_callback(struct rpc_xprt *xprt,
  1194. struct xdr_skb_reader *desc)
  1195. {
  1196. struct sock_xprt *transport =
  1197. container_of(xprt, struct sock_xprt, xprt);
  1198. struct rpc_rqst *req;
  1199. /* Look up and lock the request corresponding to the given XID */
  1200. spin_lock_bh(&xprt->transport_lock);
  1201. req = xprt_lookup_bc_request(xprt, transport->tcp_xid);
  1202. if (req == NULL) {
  1203. spin_unlock_bh(&xprt->transport_lock);
  1204. printk(KERN_WARNING "Callback slot table overflowed\n");
  1205. xprt_force_disconnect(xprt);
  1206. return -1;
  1207. }
  1208. dprintk("RPC: read callback XID %08x\n", ntohl(req->rq_xid));
  1209. xs_tcp_read_common(xprt, desc, req);
  1210. if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
  1211. xprt_complete_bc_request(req, transport->tcp_copied);
  1212. spin_unlock_bh(&xprt->transport_lock);
  1213. return 0;
  1214. }
  1215. static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
  1216. struct xdr_skb_reader *desc)
  1217. {
  1218. struct sock_xprt *transport =
  1219. container_of(xprt, struct sock_xprt, xprt);
  1220. return (transport->tcp_flags & TCP_RPC_REPLY) ?
  1221. xs_tcp_read_reply(xprt, desc) :
  1222. xs_tcp_read_callback(xprt, desc);
  1223. }
  1224. static int xs_tcp_bc_up(struct svc_serv *serv, struct net *net)
  1225. {
  1226. int ret;
  1227. ret = svc_create_xprt(serv, "tcp-bc", net, PF_INET, 0,
  1228. SVC_SOCK_ANONYMOUS);
  1229. if (ret < 0)
  1230. return ret;
  1231. return 0;
  1232. }
  1233. static size_t xs_tcp_bc_maxpayload(struct rpc_xprt *xprt)
  1234. {
  1235. return PAGE_SIZE;
  1236. }
  1237. #else
  1238. static inline int _xs_tcp_read_data(struct rpc_xprt *xprt,
  1239. struct xdr_skb_reader *desc)
  1240. {
  1241. return xs_tcp_read_reply(xprt, desc);
  1242. }
  1243. #endif /* CONFIG_SUNRPC_BACKCHANNEL */
  1244. /*
  1245. * Read data off the transport. This can be either an RPC_CALL or an
  1246. * RPC_REPLY. Relay the processing to helper functions.
  1247. */
  1248. static void xs_tcp_read_data(struct rpc_xprt *xprt,
  1249. struct xdr_skb_reader *desc)
  1250. {
  1251. struct sock_xprt *transport =
  1252. container_of(xprt, struct sock_xprt, xprt);
  1253. if (_xs_tcp_read_data(xprt, desc) == 0)
  1254. xs_tcp_check_fraghdr(transport);
  1255. else {
  1256. /*
  1257. * The transport_lock protects the request handling.
  1258. * There's no need to hold it to update the tcp_flags.
  1259. */
  1260. transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
  1261. }
  1262. }
  1263. static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)
  1264. {
  1265. size_t len;
  1266. len = transport->tcp_reclen - transport->tcp_offset;
  1267. if (len > desc->count)
  1268. len = desc->count;
  1269. desc->count -= len;
  1270. desc->offset += len;
  1271. transport->tcp_offset += len;
  1272. dprintk("RPC: discarded %Zu bytes\n", len);
  1273. xs_tcp_check_fraghdr(transport);
  1274. }
  1275. static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, unsigned int offset, size_t len)
  1276. {
  1277. struct rpc_xprt *xprt = rd_desc->arg.data;
  1278. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1279. struct xdr_skb_reader desc = {
  1280. .skb = skb,
  1281. .offset = offset,
  1282. .count = len,
  1283. };
  1284. dprintk("RPC: xs_tcp_data_recv started\n");
  1285. do {
  1286. trace_xs_tcp_data_recv(transport);
  1287. /* Read in a new fragment marker if necessary */
  1288. /* Can we ever really expect to get completely empty fragments? */
  1289. if (transport->tcp_flags & TCP_RCV_COPY_FRAGHDR) {
  1290. xs_tcp_read_fraghdr(xprt, &desc);
  1291. continue;
  1292. }
  1293. /* Read in the xid if necessary */
  1294. if (transport->tcp_flags & TCP_RCV_COPY_XID) {
  1295. xs_tcp_read_xid(transport, &desc);
  1296. continue;
  1297. }
  1298. /* Read in the call/reply flag */
  1299. if (transport->tcp_flags & TCP_RCV_READ_CALLDIR) {
  1300. xs_tcp_read_calldir(transport, &desc);
  1301. continue;
  1302. }
  1303. /* Read in the request data */
  1304. if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
  1305. xs_tcp_read_data(xprt, &desc);
  1306. continue;
  1307. }
  1308. /* Skip over any trailing bytes on short reads */
  1309. xs_tcp_read_discard(transport, &desc);
  1310. } while (desc.count);
  1311. trace_xs_tcp_data_recv(transport);
  1312. dprintk("RPC: xs_tcp_data_recv done\n");
  1313. return len - desc.count;
  1314. }
  1315. static void xs_tcp_data_receive(struct sock_xprt *transport)
  1316. {
  1317. struct rpc_xprt *xprt = &transport->xprt;
  1318. struct sock *sk;
  1319. read_descriptor_t rd_desc = {
  1320. .count = 2*1024*1024,
  1321. .arg.data = xprt,
  1322. };
  1323. unsigned long total = 0;
  1324. int read = 0;
  1325. mutex_lock(&transport->recv_mutex);
  1326. sk = transport->inet;
  1327. if (sk == NULL)
  1328. goto out;
  1329. /* We use rd_desc to pass struct xprt to xs_tcp_data_recv */
  1330. for (;;) {
  1331. lock_sock(sk);
  1332. read = tcp_read_sock(sk, &rd_desc, xs_tcp_data_recv);
  1333. if (read <= 0) {
  1334. clear_bit(XPRT_SOCK_DATA_READY, &transport->sock_state);
  1335. release_sock(sk);
  1336. if (!test_bit(XPRT_SOCK_DATA_READY, &transport->sock_state))
  1337. break;
  1338. } else {
  1339. release_sock(sk);
  1340. total += read;
  1341. }
  1342. rd_desc.count = 65536;
  1343. }
  1344. out:
  1345. mutex_unlock(&transport->recv_mutex);
  1346. trace_xs_tcp_data_ready(xprt, read, total);
  1347. }
  1348. static void xs_tcp_data_receive_workfn(struct work_struct *work)
  1349. {
  1350. struct sock_xprt *transport =
  1351. container_of(work, struct sock_xprt, recv_worker);
  1352. xs_tcp_data_receive(transport);
  1353. }
  1354. /**
  1355. * xs_tcp_state_change - callback to handle TCP socket state changes
  1356. * @sk: socket whose state has changed
  1357. *
  1358. */
  1359. static void xs_tcp_state_change(struct sock *sk)
  1360. {
  1361. struct rpc_xprt *xprt;
  1362. struct sock_xprt *transport;
  1363. read_lock_bh(&sk->sk_callback_lock);
  1364. if (!(xprt = xprt_from_sock(sk)))
  1365. goto out;
  1366. dprintk("RPC: xs_tcp_state_change client %p...\n", xprt);
  1367. dprintk("RPC: state %x conn %d dead %d zapped %d sk_shutdown %d\n",
  1368. sk->sk_state, xprt_connected(xprt),
  1369. sock_flag(sk, SOCK_DEAD),
  1370. sock_flag(sk, SOCK_ZAPPED),
  1371. sk->sk_shutdown);
  1372. transport = container_of(xprt, struct sock_xprt, xprt);
  1373. trace_rpc_socket_state_change(xprt, sk->sk_socket);
  1374. switch (sk->sk_state) {
  1375. case TCP_ESTABLISHED:
  1376. spin_lock(&xprt->transport_lock);
  1377. if (!xprt_test_and_set_connected(xprt)) {
  1378. /* Reset TCP record info */
  1379. transport->tcp_offset = 0;
  1380. transport->tcp_reclen = 0;
  1381. transport->tcp_copied = 0;
  1382. transport->tcp_flags =
  1383. TCP_RCV_COPY_FRAGHDR | TCP_RCV_COPY_XID;
  1384. xprt->connect_cookie++;
  1385. clear_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
  1386. xprt_clear_connecting(xprt);
  1387. xprt_wake_pending_tasks(xprt, -EAGAIN);
  1388. }
  1389. spin_unlock(&xprt->transport_lock);
  1390. break;
  1391. case TCP_FIN_WAIT1:
  1392. /* The client initiated a shutdown of the socket */
  1393. xprt->connect_cookie++;
  1394. xprt->reestablish_timeout = 0;
  1395. set_bit(XPRT_CLOSING, &xprt->state);
  1396. smp_mb__before_atomic();
  1397. clear_bit(XPRT_CONNECTED, &xprt->state);
  1398. clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
  1399. smp_mb__after_atomic();
  1400. break;
  1401. case TCP_CLOSE_WAIT:
  1402. /* The server initiated a shutdown of the socket */
  1403. xprt->connect_cookie++;
  1404. clear_bit(XPRT_CONNECTED, &xprt->state);
  1405. xs_tcp_force_close(xprt);
  1406. case TCP_CLOSING:
  1407. /*
  1408. * If the server closed down the connection, make sure that
  1409. * we back off before reconnecting
  1410. */
  1411. if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
  1412. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  1413. break;
  1414. case TCP_LAST_ACK:
  1415. set_bit(XPRT_CLOSING, &xprt->state);
  1416. smp_mb__before_atomic();
  1417. clear_bit(XPRT_CONNECTED, &xprt->state);
  1418. smp_mb__after_atomic();
  1419. break;
  1420. case TCP_CLOSE:
  1421. if (test_and_clear_bit(XPRT_SOCK_CONNECTING,
  1422. &transport->sock_state))
  1423. xprt_clear_connecting(xprt);
  1424. xs_sock_mark_closed(xprt);
  1425. }
  1426. out:
  1427. read_unlock_bh(&sk->sk_callback_lock);
  1428. }
  1429. static void xs_write_space(struct sock *sk)
  1430. {
  1431. struct socket_wq *wq;
  1432. struct rpc_xprt *xprt;
  1433. if (!sk->sk_socket)
  1434. return;
  1435. clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
  1436. if (unlikely(!(xprt = xprt_from_sock(sk))))
  1437. return;
  1438. rcu_read_lock();
  1439. wq = rcu_dereference(sk->sk_wq);
  1440. if (!wq || test_and_clear_bit(SOCKWQ_ASYNC_NOSPACE, &wq->flags) == 0)
  1441. goto out;
  1442. xprt_write_space(xprt);
  1443. out:
  1444. rcu_read_unlock();
  1445. }
  1446. /**
  1447. * xs_udp_write_space - callback invoked when socket buffer space
  1448. * becomes available
  1449. * @sk: socket whose state has changed
  1450. *
  1451. * Called when more output buffer space is available for this socket.
  1452. * We try not to wake our writers until they can make "significant"
  1453. * progress, otherwise we'll waste resources thrashing kernel_sendmsg
  1454. * with a bunch of small requests.
  1455. */
  1456. static void xs_udp_write_space(struct sock *sk)
  1457. {
  1458. read_lock_bh(&sk->sk_callback_lock);
  1459. /* from net/core/sock.c:sock_def_write_space */
  1460. if (sock_writeable(sk))
  1461. xs_write_space(sk);
  1462. read_unlock_bh(&sk->sk_callback_lock);
  1463. }
  1464. /**
  1465. * xs_tcp_write_space - callback invoked when socket buffer space
  1466. * becomes available
  1467. * @sk: socket whose state has changed
  1468. *
  1469. * Called when more output buffer space is available for this socket.
  1470. * We try not to wake our writers until they can make "significant"
  1471. * progress, otherwise we'll waste resources thrashing kernel_sendmsg
  1472. * with a bunch of small requests.
  1473. */
  1474. static void xs_tcp_write_space(struct sock *sk)
  1475. {
  1476. read_lock_bh(&sk->sk_callback_lock);
  1477. /* from net/core/stream.c:sk_stream_write_space */
  1478. if (sk_stream_is_writeable(sk))
  1479. xs_write_space(sk);
  1480. read_unlock_bh(&sk->sk_callback_lock);
  1481. }
  1482. static void xs_udp_do_set_buffer_size(struct rpc_xprt *xprt)
  1483. {
  1484. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1485. struct sock *sk = transport->inet;
  1486. if (transport->rcvsize) {
  1487. sk->sk_userlocks |= SOCK_RCVBUF_LOCK;
  1488. sk->sk_rcvbuf = transport->rcvsize * xprt->max_reqs * 2;
  1489. }
  1490. if (transport->sndsize) {
  1491. sk->sk_userlocks |= SOCK_SNDBUF_LOCK;
  1492. sk->sk_sndbuf = transport->sndsize * xprt->max_reqs * 2;
  1493. sk->sk_write_space(sk);
  1494. }
  1495. }
  1496. /**
  1497. * xs_udp_set_buffer_size - set send and receive limits
  1498. * @xprt: generic transport
  1499. * @sndsize: requested size of send buffer, in bytes
  1500. * @rcvsize: requested size of receive buffer, in bytes
  1501. *
  1502. * Set socket send and receive buffer size limits.
  1503. */
  1504. static void xs_udp_set_buffer_size(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize)
  1505. {
  1506. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1507. transport->sndsize = 0;
  1508. if (sndsize)
  1509. transport->sndsize = sndsize + 1024;
  1510. transport->rcvsize = 0;
  1511. if (rcvsize)
  1512. transport->rcvsize = rcvsize + 1024;
  1513. xs_udp_do_set_buffer_size(xprt);
  1514. }
  1515. /**
  1516. * xs_udp_timer - called when a retransmit timeout occurs on a UDP transport
  1517. * @task: task that timed out
  1518. *
  1519. * Adjust the congestion window after a retransmit timeout has occurred.
  1520. */
  1521. static void xs_udp_timer(struct rpc_xprt *xprt, struct rpc_task *task)
  1522. {
  1523. xprt_adjust_cwnd(xprt, task, -ETIMEDOUT);
  1524. }
  1525. static unsigned short xs_get_random_port(void)
  1526. {
  1527. unsigned short range = xprt_max_resvport - xprt_min_resvport + 1;
  1528. unsigned short rand = (unsigned short) prandom_u32() % range;
  1529. return rand + xprt_min_resvport;
  1530. }
  1531. /**
  1532. * xs_set_reuseaddr_port - set the socket's port and address reuse options
  1533. * @sock: socket
  1534. *
  1535. * Note that this function has to be called on all sockets that share the
  1536. * same port, and it must be called before binding.
  1537. */
  1538. static void xs_sock_set_reuseport(struct socket *sock)
  1539. {
  1540. int opt = 1;
  1541. kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEPORT,
  1542. (char *)&opt, sizeof(opt));
  1543. }
  1544. static unsigned short xs_sock_getport(struct socket *sock)
  1545. {
  1546. struct sockaddr_storage buf;
  1547. int buflen;
  1548. unsigned short port = 0;
  1549. if (kernel_getsockname(sock, (struct sockaddr *)&buf, &buflen) < 0)
  1550. goto out;
  1551. switch (buf.ss_family) {
  1552. case AF_INET6:
  1553. port = ntohs(((struct sockaddr_in6 *)&buf)->sin6_port);
  1554. break;
  1555. case AF_INET:
  1556. port = ntohs(((struct sockaddr_in *)&buf)->sin_port);
  1557. }
  1558. out:
  1559. return port;
  1560. }
  1561. /**
  1562. * xs_set_port - reset the port number in the remote endpoint address
  1563. * @xprt: generic transport
  1564. * @port: new port number
  1565. *
  1566. */
  1567. static void xs_set_port(struct rpc_xprt *xprt, unsigned short port)
  1568. {
  1569. dprintk("RPC: setting port for xprt %p to %u\n", xprt, port);
  1570. rpc_set_port(xs_addr(xprt), port);
  1571. xs_update_peer_port(xprt);
  1572. }
  1573. static void xs_set_srcport(struct sock_xprt *transport, struct socket *sock)
  1574. {
  1575. if (transport->srcport == 0)
  1576. transport->srcport = xs_sock_getport(sock);
  1577. }
  1578. static unsigned short xs_get_srcport(struct sock_xprt *transport)
  1579. {
  1580. unsigned short port = transport->srcport;
  1581. if (port == 0 && transport->xprt.resvport)
  1582. port = xs_get_random_port();
  1583. return port;
  1584. }
  1585. static unsigned short xs_next_srcport(struct sock_xprt *transport, unsigned short port)
  1586. {
  1587. if (transport->srcport != 0)
  1588. transport->srcport = 0;
  1589. if (!transport->xprt.resvport)
  1590. return 0;
  1591. if (port <= xprt_min_resvport || port > xprt_max_resvport)
  1592. return xprt_max_resvport;
  1593. return --port;
  1594. }
  1595. static int xs_bind(struct sock_xprt *transport, struct socket *sock)
  1596. {
  1597. struct sockaddr_storage myaddr;
  1598. int err, nloop = 0;
  1599. unsigned short port = xs_get_srcport(transport);
  1600. unsigned short last;
  1601. /*
  1602. * If we are asking for any ephemeral port (i.e. port == 0 &&
  1603. * transport->xprt.resvport == 0), don't bind. Let the local
  1604. * port selection happen implicitly when the socket is used
  1605. * (for example at connect time).
  1606. *
  1607. * This ensures that we can continue to establish TCP
  1608. * connections even when all local ephemeral ports are already
  1609. * a part of some TCP connection. This makes no difference
  1610. * for UDP sockets, but also doens't harm them.
  1611. *
  1612. * If we're asking for any reserved port (i.e. port == 0 &&
  1613. * transport->xprt.resvport == 1) xs_get_srcport above will
  1614. * ensure that port is non-zero and we will bind as needed.
  1615. */
  1616. if (port == 0)
  1617. return 0;
  1618. memcpy(&myaddr, &transport->srcaddr, transport->xprt.addrlen);
  1619. do {
  1620. rpc_set_port((struct sockaddr *)&myaddr, port);
  1621. err = kernel_bind(sock, (struct sockaddr *)&myaddr,
  1622. transport->xprt.addrlen);
  1623. if (err == 0) {
  1624. transport->srcport = port;
  1625. break;
  1626. }
  1627. last = port;
  1628. port = xs_next_srcport(transport, port);
  1629. if (port > last)
  1630. nloop++;
  1631. } while (err == -EADDRINUSE && nloop != 2);
  1632. if (myaddr.ss_family == AF_INET)
  1633. dprintk("RPC: %s %pI4:%u: %s (%d)\n", __func__,
  1634. &((struct sockaddr_in *)&myaddr)->sin_addr,
  1635. port, err ? "failed" : "ok", err);
  1636. else
  1637. dprintk("RPC: %s %pI6:%u: %s (%d)\n", __func__,
  1638. &((struct sockaddr_in6 *)&myaddr)->sin6_addr,
  1639. port, err ? "failed" : "ok", err);
  1640. return err;
  1641. }
  1642. /*
  1643. * We don't support autobind on AF_LOCAL sockets
  1644. */
  1645. static void xs_local_rpcbind(struct rpc_task *task)
  1646. {
  1647. xprt_set_bound(task->tk_xprt);
  1648. }
  1649. static void xs_local_set_port(struct rpc_xprt *xprt, unsigned short port)
  1650. {
  1651. }
  1652. #ifdef CONFIG_DEBUG_LOCK_ALLOC
  1653. static struct lock_class_key xs_key[2];
  1654. static struct lock_class_key xs_slock_key[2];
  1655. static inline void xs_reclassify_socketu(struct socket *sock)
  1656. {
  1657. struct sock *sk = sock->sk;
  1658. sock_lock_init_class_and_name(sk, "slock-AF_LOCAL-RPC",
  1659. &xs_slock_key[1], "sk_lock-AF_LOCAL-RPC", &xs_key[1]);
  1660. }
  1661. static inline void xs_reclassify_socket4(struct socket *sock)
  1662. {
  1663. struct sock *sk = sock->sk;
  1664. sock_lock_init_class_and_name(sk, "slock-AF_INET-RPC",
  1665. &xs_slock_key[0], "sk_lock-AF_INET-RPC", &xs_key[0]);
  1666. }
  1667. static inline void xs_reclassify_socket6(struct socket *sock)
  1668. {
  1669. struct sock *sk = sock->sk;
  1670. sock_lock_init_class_and_name(sk, "slock-AF_INET6-RPC",
  1671. &xs_slock_key[1], "sk_lock-AF_INET6-RPC", &xs_key[1]);
  1672. }
  1673. static inline void xs_reclassify_socket(int family, struct socket *sock)
  1674. {
  1675. if (WARN_ON_ONCE(!sock_allow_reclassification(sock->sk)))
  1676. return;
  1677. switch (family) {
  1678. case AF_LOCAL:
  1679. xs_reclassify_socketu(sock);
  1680. break;
  1681. case AF_INET:
  1682. xs_reclassify_socket4(sock);
  1683. break;
  1684. case AF_INET6:
  1685. xs_reclassify_socket6(sock);
  1686. break;
  1687. }
  1688. }
  1689. #else
  1690. static inline void xs_reclassify_socket(int family, struct socket *sock)
  1691. {
  1692. }
  1693. #endif
  1694. static void xs_dummy_setup_socket(struct work_struct *work)
  1695. {
  1696. }
  1697. static struct socket *xs_create_sock(struct rpc_xprt *xprt,
  1698. struct sock_xprt *transport, int family, int type,
  1699. int protocol, bool reuseport)
  1700. {
  1701. struct socket *sock;
  1702. int err;
  1703. err = __sock_create(xprt->xprt_net, family, type, protocol, &sock, 1);
  1704. if (err < 0) {
  1705. dprintk("RPC: can't create %d transport socket (%d).\n",
  1706. protocol, -err);
  1707. goto out;
  1708. }
  1709. xs_reclassify_socket(family, sock);
  1710. if (reuseport)
  1711. xs_sock_set_reuseport(sock);
  1712. err = xs_bind(transport, sock);
  1713. if (err) {
  1714. sock_release(sock);
  1715. goto out;
  1716. }
  1717. return sock;
  1718. out:
  1719. return ERR_PTR(err);
  1720. }
  1721. static int xs_local_finish_connecting(struct rpc_xprt *xprt,
  1722. struct socket *sock)
  1723. {
  1724. struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
  1725. xprt);
  1726. if (!transport->inet) {
  1727. struct sock *sk = sock->sk;
  1728. write_lock_bh(&sk->sk_callback_lock);
  1729. xs_save_old_callbacks(transport, sk);
  1730. sk->sk_user_data = xprt;
  1731. sk->sk_data_ready = xs_data_ready;
  1732. sk->sk_write_space = xs_udp_write_space;
  1733. sock_set_flag(sk, SOCK_FASYNC);
  1734. sk->sk_error_report = xs_error_report;
  1735. sk->sk_allocation = GFP_NOIO;
  1736. xprt_clear_connected(xprt);
  1737. /* Reset to new socket */
  1738. transport->sock = sock;
  1739. transport->inet = sk;
  1740. write_unlock_bh(&sk->sk_callback_lock);
  1741. }
  1742. /* Tell the socket layer to start connecting... */
  1743. xprt->stat.connect_count++;
  1744. xprt->stat.connect_start = jiffies;
  1745. return kernel_connect(sock, xs_addr(xprt), xprt->addrlen, 0);
  1746. }
  1747. /**
  1748. * xs_local_setup_socket - create AF_LOCAL socket, connect to a local endpoint
  1749. * @transport: socket transport to connect
  1750. */
  1751. static int xs_local_setup_socket(struct sock_xprt *transport)
  1752. {
  1753. struct rpc_xprt *xprt = &transport->xprt;
  1754. struct socket *sock;
  1755. int status = -EIO;
  1756. status = __sock_create(xprt->xprt_net, AF_LOCAL,
  1757. SOCK_STREAM, 0, &sock, 1);
  1758. if (status < 0) {
  1759. dprintk("RPC: can't create AF_LOCAL "
  1760. "transport socket (%d).\n", -status);
  1761. goto out;
  1762. }
  1763. xs_reclassify_socket(AF_LOCAL, sock);
  1764. dprintk("RPC: worker connecting xprt %p via AF_LOCAL to %s\n",
  1765. xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
  1766. status = xs_local_finish_connecting(xprt, sock);
  1767. trace_rpc_socket_connect(xprt, sock, status);
  1768. switch (status) {
  1769. case 0:
  1770. dprintk("RPC: xprt %p connected to %s\n",
  1771. xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
  1772. xprt_set_connected(xprt);
  1773. case -ENOBUFS:
  1774. break;
  1775. case -ENOENT:
  1776. dprintk("RPC: xprt %p: socket %s does not exist\n",
  1777. xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
  1778. break;
  1779. case -ECONNREFUSED:
  1780. dprintk("RPC: xprt %p: connection refused for %s\n",
  1781. xprt, xprt->address_strings[RPC_DISPLAY_ADDR]);
  1782. break;
  1783. default:
  1784. printk(KERN_ERR "%s: unhandled error (%d) connecting to %s\n",
  1785. __func__, -status,
  1786. xprt->address_strings[RPC_DISPLAY_ADDR]);
  1787. }
  1788. out:
  1789. xprt_clear_connecting(xprt);
  1790. xprt_wake_pending_tasks(xprt, status);
  1791. return status;
  1792. }
  1793. static void xs_local_connect(struct rpc_xprt *xprt, struct rpc_task *task)
  1794. {
  1795. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1796. int ret;
  1797. if (RPC_IS_ASYNC(task)) {
  1798. /*
  1799. * We want the AF_LOCAL connect to be resolved in the
  1800. * filesystem namespace of the process making the rpc
  1801. * call. Thus we connect synchronously.
  1802. *
  1803. * If we want to support asynchronous AF_LOCAL calls,
  1804. * we'll need to figure out how to pass a namespace to
  1805. * connect.
  1806. */
  1807. rpc_exit(task, -ENOTCONN);
  1808. return;
  1809. }
  1810. ret = xs_local_setup_socket(transport);
  1811. if (ret && !RPC_IS_SOFTCONN(task))
  1812. msleep_interruptible(15000);
  1813. }
  1814. #if IS_ENABLED(CONFIG_SUNRPC_SWAP)
  1815. /*
  1816. * Note that this should be called with XPRT_LOCKED held (or when we otherwise
  1817. * know that we have exclusive access to the socket), to guard against
  1818. * races with xs_reset_transport.
  1819. */
  1820. static void xs_set_memalloc(struct rpc_xprt *xprt)
  1821. {
  1822. struct sock_xprt *transport = container_of(xprt, struct sock_xprt,
  1823. xprt);
  1824. /*
  1825. * If there's no sock, then we have nothing to set. The
  1826. * reconnecting process will get it for us.
  1827. */
  1828. if (!transport->inet)
  1829. return;
  1830. if (atomic_read(&xprt->swapper))
  1831. sk_set_memalloc(transport->inet);
  1832. }
  1833. /**
  1834. * xs_enable_swap - Tag this transport as being used for swap.
  1835. * @xprt: transport to tag
  1836. *
  1837. * Take a reference to this transport on behalf of the rpc_clnt, and
  1838. * optionally mark it for swapping if it wasn't already.
  1839. */
  1840. static int
  1841. xs_enable_swap(struct rpc_xprt *xprt)
  1842. {
  1843. struct sock_xprt *xs = container_of(xprt, struct sock_xprt, xprt);
  1844. if (atomic_inc_return(&xprt->swapper) != 1)
  1845. return 0;
  1846. if (wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_KILLABLE))
  1847. return -ERESTARTSYS;
  1848. if (xs->inet)
  1849. sk_set_memalloc(xs->inet);
  1850. xprt_release_xprt(xprt, NULL);
  1851. return 0;
  1852. }
  1853. /**
  1854. * xs_disable_swap - Untag this transport as being used for swap.
  1855. * @xprt: transport to tag
  1856. *
  1857. * Drop a "swapper" reference to this xprt on behalf of the rpc_clnt. If the
  1858. * swapper refcount goes to 0, untag the socket as a memalloc socket.
  1859. */
  1860. static void
  1861. xs_disable_swap(struct rpc_xprt *xprt)
  1862. {
  1863. struct sock_xprt *xs = container_of(xprt, struct sock_xprt, xprt);
  1864. if (!atomic_dec_and_test(&xprt->swapper))
  1865. return;
  1866. if (wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_KILLABLE))
  1867. return;
  1868. if (xs->inet)
  1869. sk_clear_memalloc(xs->inet);
  1870. xprt_release_xprt(xprt, NULL);
  1871. }
  1872. #else
  1873. static void xs_set_memalloc(struct rpc_xprt *xprt)
  1874. {
  1875. }
  1876. static int
  1877. xs_enable_swap(struct rpc_xprt *xprt)
  1878. {
  1879. return -EINVAL;
  1880. }
  1881. static void
  1882. xs_disable_swap(struct rpc_xprt *xprt)
  1883. {
  1884. }
  1885. #endif
  1886. static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
  1887. {
  1888. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1889. if (!transport->inet) {
  1890. struct sock *sk = sock->sk;
  1891. write_lock_bh(&sk->sk_callback_lock);
  1892. xs_save_old_callbacks(transport, sk);
  1893. sk->sk_user_data = xprt;
  1894. sk->sk_data_ready = xs_data_ready;
  1895. sk->sk_write_space = xs_udp_write_space;
  1896. sock_set_flag(sk, SOCK_FASYNC);
  1897. sk->sk_allocation = GFP_NOIO;
  1898. xprt_set_connected(xprt);
  1899. /* Reset to new socket */
  1900. transport->sock = sock;
  1901. transport->inet = sk;
  1902. xs_set_memalloc(xprt);
  1903. write_unlock_bh(&sk->sk_callback_lock);
  1904. }
  1905. xs_udp_do_set_buffer_size(xprt);
  1906. }
  1907. static void xs_udp_setup_socket(struct work_struct *work)
  1908. {
  1909. struct sock_xprt *transport =
  1910. container_of(work, struct sock_xprt, connect_worker.work);
  1911. struct rpc_xprt *xprt = &transport->xprt;
  1912. struct socket *sock = transport->sock;
  1913. int status = -EIO;
  1914. sock = xs_create_sock(xprt, transport,
  1915. xs_addr(xprt)->sa_family, SOCK_DGRAM,
  1916. IPPROTO_UDP, false);
  1917. if (IS_ERR(sock))
  1918. goto out;
  1919. dprintk("RPC: worker connecting xprt %p via %s to "
  1920. "%s (port %s)\n", xprt,
  1921. xprt->address_strings[RPC_DISPLAY_PROTO],
  1922. xprt->address_strings[RPC_DISPLAY_ADDR],
  1923. xprt->address_strings[RPC_DISPLAY_PORT]);
  1924. xs_udp_finish_connecting(xprt, sock);
  1925. trace_rpc_socket_connect(xprt, sock, 0);
  1926. status = 0;
  1927. out:
  1928. xprt_unlock_connect(xprt, transport);
  1929. xprt_clear_connecting(xprt);
  1930. xprt_wake_pending_tasks(xprt, status);
  1931. }
  1932. /**
  1933. * xs_tcp_shutdown - gracefully shut down a TCP socket
  1934. * @xprt: transport
  1935. *
  1936. * Initiates a graceful shutdown of the TCP socket by calling the
  1937. * equivalent of shutdown(SHUT_RDWR);
  1938. */
  1939. static void xs_tcp_shutdown(struct rpc_xprt *xprt)
  1940. {
  1941. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1942. struct socket *sock = transport->sock;
  1943. if (sock == NULL)
  1944. return;
  1945. if (xprt_connected(xprt)) {
  1946. kernel_sock_shutdown(sock, SHUT_RDWR);
  1947. trace_rpc_socket_shutdown(xprt, sock);
  1948. } else
  1949. xs_reset_transport(transport);
  1950. }
  1951. static int xs_tcp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
  1952. {
  1953. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  1954. int ret = -ENOTCONN;
  1955. if (!transport->inet) {
  1956. struct sock *sk = sock->sk;
  1957. unsigned int keepidle = xprt->timeout->to_initval / HZ;
  1958. unsigned int keepcnt = xprt->timeout->to_retries + 1;
  1959. unsigned int opt_on = 1;
  1960. unsigned int timeo;
  1961. /* TCP Keepalive options */
  1962. kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
  1963. (char *)&opt_on, sizeof(opt_on));
  1964. kernel_setsockopt(sock, SOL_TCP, TCP_KEEPIDLE,
  1965. (char *)&keepidle, sizeof(keepidle));
  1966. kernel_setsockopt(sock, SOL_TCP, TCP_KEEPINTVL,
  1967. (char *)&keepidle, sizeof(keepidle));
  1968. kernel_setsockopt(sock, SOL_TCP, TCP_KEEPCNT,
  1969. (char *)&keepcnt, sizeof(keepcnt));
  1970. /* TCP user timeout (see RFC5482) */
  1971. timeo = jiffies_to_msecs(xprt->timeout->to_initval) *
  1972. (xprt->timeout->to_retries + 1);
  1973. kernel_setsockopt(sock, SOL_TCP, TCP_USER_TIMEOUT,
  1974. (char *)&timeo, sizeof(timeo));
  1975. write_lock_bh(&sk->sk_callback_lock);
  1976. xs_save_old_callbacks(transport, sk);
  1977. sk->sk_user_data = xprt;
  1978. sk->sk_data_ready = xs_data_ready;
  1979. sk->sk_state_change = xs_tcp_state_change;
  1980. sk->sk_write_space = xs_tcp_write_space;
  1981. sock_set_flag(sk, SOCK_FASYNC);
  1982. sk->sk_error_report = xs_error_report;
  1983. sk->sk_allocation = GFP_NOIO;
  1984. /* socket options */
  1985. sock_reset_flag(sk, SOCK_LINGER);
  1986. tcp_sk(sk)->nonagle |= TCP_NAGLE_OFF;
  1987. xprt_clear_connected(xprt);
  1988. /* Reset to new socket */
  1989. transport->sock = sock;
  1990. transport->inet = sk;
  1991. write_unlock_bh(&sk->sk_callback_lock);
  1992. }
  1993. if (!xprt_bound(xprt))
  1994. goto out;
  1995. xs_set_memalloc(xprt);
  1996. /* Tell the socket layer to start connecting... */
  1997. xprt->stat.connect_count++;
  1998. xprt->stat.connect_start = jiffies;
  1999. set_bit(XPRT_SOCK_CONNECTING, &transport->sock_state);
  2000. ret = kernel_connect(sock, xs_addr(xprt), xprt->addrlen, O_NONBLOCK);
  2001. switch (ret) {
  2002. case 0:
  2003. xs_set_srcport(transport, sock);
  2004. case -EINPROGRESS:
  2005. /* SYN_SENT! */
  2006. if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
  2007. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  2008. break;
  2009. case -EADDRNOTAVAIL:
  2010. /* Source port number is unavailable. Try a new one! */
  2011. transport->srcport = 0;
  2012. }
  2013. out:
  2014. return ret;
  2015. }
  2016. /**
  2017. * xs_tcp_setup_socket - create a TCP socket and connect to a remote endpoint
  2018. *
  2019. * Invoked by a work queue tasklet.
  2020. */
  2021. static void xs_tcp_setup_socket(struct work_struct *work)
  2022. {
  2023. struct sock_xprt *transport =
  2024. container_of(work, struct sock_xprt, connect_worker.work);
  2025. struct socket *sock = transport->sock;
  2026. struct rpc_xprt *xprt = &transport->xprt;
  2027. int status = -EIO;
  2028. if (!sock) {
  2029. sock = xs_create_sock(xprt, transport,
  2030. xs_addr(xprt)->sa_family, SOCK_STREAM,
  2031. IPPROTO_TCP, true);
  2032. if (IS_ERR(sock)) {
  2033. status = PTR_ERR(sock);
  2034. goto out;
  2035. }
  2036. }
  2037. dprintk("RPC: worker connecting xprt %p via %s to "
  2038. "%s (port %s)\n", xprt,
  2039. xprt->address_strings[RPC_DISPLAY_PROTO],
  2040. xprt->address_strings[RPC_DISPLAY_ADDR],
  2041. xprt->address_strings[RPC_DISPLAY_PORT]);
  2042. status = xs_tcp_finish_connecting(xprt, sock);
  2043. trace_rpc_socket_connect(xprt, sock, status);
  2044. dprintk("RPC: %p connect status %d connected %d sock state %d\n",
  2045. xprt, -status, xprt_connected(xprt),
  2046. sock->sk->sk_state);
  2047. switch (status) {
  2048. default:
  2049. printk("%s: connect returned unhandled error %d\n",
  2050. __func__, status);
  2051. case -EADDRNOTAVAIL:
  2052. /* We're probably in TIME_WAIT. Get rid of existing socket,
  2053. * and retry
  2054. */
  2055. xs_tcp_force_close(xprt);
  2056. break;
  2057. case 0:
  2058. case -EINPROGRESS:
  2059. case -EALREADY:
  2060. xprt_unlock_connect(xprt, transport);
  2061. return;
  2062. case -EINVAL:
  2063. /* Happens, for instance, if the user specified a link
  2064. * local IPv6 address without a scope-id.
  2065. */
  2066. case -ECONNREFUSED:
  2067. case -ECONNRESET:
  2068. case -ENETUNREACH:
  2069. case -EADDRINUSE:
  2070. case -ENOBUFS:
  2071. /* retry with existing socket, after a delay */
  2072. xs_tcp_force_close(xprt);
  2073. goto out;
  2074. }
  2075. status = -EAGAIN;
  2076. out:
  2077. xprt_unlock_connect(xprt, transport);
  2078. xprt_clear_connecting(xprt);
  2079. xprt_wake_pending_tasks(xprt, status);
  2080. }
  2081. /**
  2082. * xs_connect - connect a socket to a remote endpoint
  2083. * @xprt: pointer to transport structure
  2084. * @task: address of RPC task that manages state of connect request
  2085. *
  2086. * TCP: If the remote end dropped the connection, delay reconnecting.
  2087. *
  2088. * UDP socket connects are synchronous, but we use a work queue anyway
  2089. * to guarantee that even unprivileged user processes can set up a
  2090. * socket on a privileged port.
  2091. *
  2092. * If a UDP socket connect fails, the delay behavior here prevents
  2093. * retry floods (hard mounts).
  2094. */
  2095. static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task)
  2096. {
  2097. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  2098. WARN_ON_ONCE(!xprt_lock_connect(xprt, task, transport));
  2099. if (transport->sock != NULL) {
  2100. dprintk("RPC: xs_connect delayed xprt %p for %lu "
  2101. "seconds\n",
  2102. xprt, xprt->reestablish_timeout / HZ);
  2103. /* Start by resetting any existing state */
  2104. xs_reset_transport(transport);
  2105. queue_delayed_work(xprtiod_workqueue,
  2106. &transport->connect_worker,
  2107. xprt->reestablish_timeout);
  2108. xprt->reestablish_timeout <<= 1;
  2109. if (xprt->reestablish_timeout < XS_TCP_INIT_REEST_TO)
  2110. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  2111. if (xprt->reestablish_timeout > XS_TCP_MAX_REEST_TO)
  2112. xprt->reestablish_timeout = XS_TCP_MAX_REEST_TO;
  2113. } else {
  2114. dprintk("RPC: xs_connect scheduled xprt %p\n", xprt);
  2115. queue_delayed_work(xprtiod_workqueue,
  2116. &transport->connect_worker, 0);
  2117. }
  2118. }
  2119. /**
  2120. * xs_local_print_stats - display AF_LOCAL socket-specifc stats
  2121. * @xprt: rpc_xprt struct containing statistics
  2122. * @seq: output file
  2123. *
  2124. */
  2125. static void xs_local_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  2126. {
  2127. long idle_time = 0;
  2128. if (xprt_connected(xprt))
  2129. idle_time = (long)(jiffies - xprt->last_used) / HZ;
  2130. seq_printf(seq, "\txprt:\tlocal %lu %lu %lu %ld %lu %lu %lu "
  2131. "%llu %llu %lu %llu %llu\n",
  2132. xprt->stat.bind_count,
  2133. xprt->stat.connect_count,
  2134. xprt->stat.connect_time,
  2135. idle_time,
  2136. xprt->stat.sends,
  2137. xprt->stat.recvs,
  2138. xprt->stat.bad_xids,
  2139. xprt->stat.req_u,
  2140. xprt->stat.bklog_u,
  2141. xprt->stat.max_slots,
  2142. xprt->stat.sending_u,
  2143. xprt->stat.pending_u);
  2144. }
  2145. /**
  2146. * xs_udp_print_stats - display UDP socket-specifc stats
  2147. * @xprt: rpc_xprt struct containing statistics
  2148. * @seq: output file
  2149. *
  2150. */
  2151. static void xs_udp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  2152. {
  2153. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  2154. seq_printf(seq, "\txprt:\tudp %u %lu %lu %lu %lu %llu %llu "
  2155. "%lu %llu %llu\n",
  2156. transport->srcport,
  2157. xprt->stat.bind_count,
  2158. xprt->stat.sends,
  2159. xprt->stat.recvs,
  2160. xprt->stat.bad_xids,
  2161. xprt->stat.req_u,
  2162. xprt->stat.bklog_u,
  2163. xprt->stat.max_slots,
  2164. xprt->stat.sending_u,
  2165. xprt->stat.pending_u);
  2166. }
  2167. /**
  2168. * xs_tcp_print_stats - display TCP socket-specifc stats
  2169. * @xprt: rpc_xprt struct containing statistics
  2170. * @seq: output file
  2171. *
  2172. */
  2173. static void xs_tcp_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  2174. {
  2175. struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
  2176. long idle_time = 0;
  2177. if (xprt_connected(xprt))
  2178. idle_time = (long)(jiffies - xprt->last_used) / HZ;
  2179. seq_printf(seq, "\txprt:\ttcp %u %lu %lu %lu %ld %lu %lu %lu "
  2180. "%llu %llu %lu %llu %llu\n",
  2181. transport->srcport,
  2182. xprt->stat.bind_count,
  2183. xprt->stat.connect_count,
  2184. xprt->stat.connect_time,
  2185. idle_time,
  2186. xprt->stat.sends,
  2187. xprt->stat.recvs,
  2188. xprt->stat.bad_xids,
  2189. xprt->stat.req_u,
  2190. xprt->stat.bklog_u,
  2191. xprt->stat.max_slots,
  2192. xprt->stat.sending_u,
  2193. xprt->stat.pending_u);
  2194. }
  2195. /*
  2196. * Allocate a bunch of pages for a scratch buffer for the rpc code. The reason
  2197. * we allocate pages instead doing a kmalloc like rpc_malloc is because we want
  2198. * to use the server side send routines.
  2199. */
  2200. static void *bc_malloc(struct rpc_task *task, size_t size)
  2201. {
  2202. struct page *page;
  2203. struct rpc_buffer *buf;
  2204. WARN_ON_ONCE(size > PAGE_SIZE - sizeof(struct rpc_buffer));
  2205. if (size > PAGE_SIZE - sizeof(struct rpc_buffer))
  2206. return NULL;
  2207. page = alloc_page(GFP_KERNEL);
  2208. if (!page)
  2209. return NULL;
  2210. buf = page_address(page);
  2211. buf->len = PAGE_SIZE;
  2212. return buf->data;
  2213. }
  2214. /*
  2215. * Free the space allocated in the bc_alloc routine
  2216. */
  2217. static void bc_free(void *buffer)
  2218. {
  2219. struct rpc_buffer *buf;
  2220. if (!buffer)
  2221. return;
  2222. buf = container_of(buffer, struct rpc_buffer, data);
  2223. free_page((unsigned long)buf);
  2224. }
  2225. /*
  2226. * Use the svc_sock to send the callback. Must be called with svsk->sk_mutex
  2227. * held. Borrows heavily from svc_tcp_sendto and xs_tcp_send_request.
  2228. */
  2229. static int bc_sendto(struct rpc_rqst *req)
  2230. {
  2231. int len;
  2232. struct xdr_buf *xbufp = &req->rq_snd_buf;
  2233. struct rpc_xprt *xprt = req->rq_xprt;
  2234. struct sock_xprt *transport =
  2235. container_of(xprt, struct sock_xprt, xprt);
  2236. struct socket *sock = transport->sock;
  2237. unsigned long headoff;
  2238. unsigned long tailoff;
  2239. xs_encode_stream_record_marker(xbufp);
  2240. tailoff = (unsigned long)xbufp->tail[0].iov_base & ~PAGE_MASK;
  2241. headoff = (unsigned long)xbufp->head[0].iov_base & ~PAGE_MASK;
  2242. len = svc_send_common(sock, xbufp,
  2243. virt_to_page(xbufp->head[0].iov_base), headoff,
  2244. xbufp->tail[0].iov_base, tailoff);
  2245. if (len != xbufp->len) {
  2246. printk(KERN_NOTICE "Error sending entire callback!\n");
  2247. len = -EAGAIN;
  2248. }
  2249. return len;
  2250. }
  2251. /*
  2252. * The send routine. Borrows from svc_send
  2253. */
  2254. static int bc_send_request(struct rpc_task *task)
  2255. {
  2256. struct rpc_rqst *req = task->tk_rqstp;
  2257. struct svc_xprt *xprt;
  2258. int len;
  2259. dprintk("sending request with xid: %08x\n", ntohl(req->rq_xid));
  2260. /*
  2261. * Get the server socket associated with this callback xprt
  2262. */
  2263. xprt = req->rq_xprt->bc_xprt;
  2264. /*
  2265. * Grab the mutex to serialize data as the connection is shared
  2266. * with the fore channel
  2267. */
  2268. if (!mutex_trylock(&xprt->xpt_mutex)) {
  2269. rpc_sleep_on(&xprt->xpt_bc_pending, task, NULL);
  2270. if (!mutex_trylock(&xprt->xpt_mutex))
  2271. return -EAGAIN;
  2272. rpc_wake_up_queued_task(&xprt->xpt_bc_pending, task);
  2273. }
  2274. if (test_bit(XPT_DEAD, &xprt->xpt_flags))
  2275. len = -ENOTCONN;
  2276. else
  2277. len = bc_sendto(req);
  2278. mutex_unlock(&xprt->xpt_mutex);
  2279. if (len > 0)
  2280. len = 0;
  2281. return len;
  2282. }
  2283. /*
  2284. * The close routine. Since this is client initiated, we do nothing
  2285. */
  2286. static void bc_close(struct rpc_xprt *xprt)
  2287. {
  2288. }
  2289. /*
  2290. * The xprt destroy routine. Again, because this connection is client
  2291. * initiated, we do nothing
  2292. */
  2293. static void bc_destroy(struct rpc_xprt *xprt)
  2294. {
  2295. dprintk("RPC: bc_destroy xprt %p\n", xprt);
  2296. xs_xprt_free(xprt);
  2297. module_put(THIS_MODULE);
  2298. }
  2299. static struct rpc_xprt_ops xs_local_ops = {
  2300. .reserve_xprt = xprt_reserve_xprt,
  2301. .release_xprt = xs_tcp_release_xprt,
  2302. .alloc_slot = xprt_alloc_slot,
  2303. .rpcbind = xs_local_rpcbind,
  2304. .set_port = xs_local_set_port,
  2305. .connect = xs_local_connect,
  2306. .buf_alloc = rpc_malloc,
  2307. .buf_free = rpc_free,
  2308. .send_request = xs_local_send_request,
  2309. .set_retrans_timeout = xprt_set_retrans_timeout_def,
  2310. .close = xs_close,
  2311. .destroy = xs_destroy,
  2312. .print_stats = xs_local_print_stats,
  2313. .enable_swap = xs_enable_swap,
  2314. .disable_swap = xs_disable_swap,
  2315. };
  2316. static struct rpc_xprt_ops xs_udp_ops = {
  2317. .set_buffer_size = xs_udp_set_buffer_size,
  2318. .reserve_xprt = xprt_reserve_xprt_cong,
  2319. .release_xprt = xprt_release_xprt_cong,
  2320. .alloc_slot = xprt_alloc_slot,
  2321. .rpcbind = rpcb_getport_async,
  2322. .set_port = xs_set_port,
  2323. .connect = xs_connect,
  2324. .buf_alloc = rpc_malloc,
  2325. .buf_free = rpc_free,
  2326. .send_request = xs_udp_send_request,
  2327. .set_retrans_timeout = xprt_set_retrans_timeout_rtt,
  2328. .timer = xs_udp_timer,
  2329. .release_request = xprt_release_rqst_cong,
  2330. .close = xs_close,
  2331. .destroy = xs_destroy,
  2332. .print_stats = xs_udp_print_stats,
  2333. .enable_swap = xs_enable_swap,
  2334. .disable_swap = xs_disable_swap,
  2335. .inject_disconnect = xs_inject_disconnect,
  2336. };
  2337. static struct rpc_xprt_ops xs_tcp_ops = {
  2338. .reserve_xprt = xprt_reserve_xprt,
  2339. .release_xprt = xs_tcp_release_xprt,
  2340. .alloc_slot = xprt_lock_and_alloc_slot,
  2341. .rpcbind = rpcb_getport_async,
  2342. .set_port = xs_set_port,
  2343. .connect = xs_connect,
  2344. .buf_alloc = rpc_malloc,
  2345. .buf_free = rpc_free,
  2346. .send_request = xs_tcp_send_request,
  2347. .set_retrans_timeout = xprt_set_retrans_timeout_def,
  2348. .close = xs_tcp_shutdown,
  2349. .destroy = xs_destroy,
  2350. .print_stats = xs_tcp_print_stats,
  2351. .enable_swap = xs_enable_swap,
  2352. .disable_swap = xs_disable_swap,
  2353. .inject_disconnect = xs_inject_disconnect,
  2354. #ifdef CONFIG_SUNRPC_BACKCHANNEL
  2355. .bc_setup = xprt_setup_bc,
  2356. .bc_up = xs_tcp_bc_up,
  2357. .bc_maxpayload = xs_tcp_bc_maxpayload,
  2358. .bc_free_rqst = xprt_free_bc_rqst,
  2359. .bc_destroy = xprt_destroy_bc,
  2360. #endif
  2361. };
  2362. /*
  2363. * The rpc_xprt_ops for the server backchannel
  2364. */
  2365. static struct rpc_xprt_ops bc_tcp_ops = {
  2366. .reserve_xprt = xprt_reserve_xprt,
  2367. .release_xprt = xprt_release_xprt,
  2368. .alloc_slot = xprt_alloc_slot,
  2369. .buf_alloc = bc_malloc,
  2370. .buf_free = bc_free,
  2371. .send_request = bc_send_request,
  2372. .set_retrans_timeout = xprt_set_retrans_timeout_def,
  2373. .close = bc_close,
  2374. .destroy = bc_destroy,
  2375. .print_stats = xs_tcp_print_stats,
  2376. .enable_swap = xs_enable_swap,
  2377. .disable_swap = xs_disable_swap,
  2378. .inject_disconnect = xs_inject_disconnect,
  2379. };
  2380. static int xs_init_anyaddr(const int family, struct sockaddr *sap)
  2381. {
  2382. static const struct sockaddr_in sin = {
  2383. .sin_family = AF_INET,
  2384. .sin_addr.s_addr = htonl(INADDR_ANY),
  2385. };
  2386. static const struct sockaddr_in6 sin6 = {
  2387. .sin6_family = AF_INET6,
  2388. .sin6_addr = IN6ADDR_ANY_INIT,
  2389. };
  2390. switch (family) {
  2391. case AF_LOCAL:
  2392. break;
  2393. case AF_INET:
  2394. memcpy(sap, &sin, sizeof(sin));
  2395. break;
  2396. case AF_INET6:
  2397. memcpy(sap, &sin6, sizeof(sin6));
  2398. break;
  2399. default:
  2400. dprintk("RPC: %s: Bad address family\n", __func__);
  2401. return -EAFNOSUPPORT;
  2402. }
  2403. return 0;
  2404. }
  2405. static struct rpc_xprt *xs_setup_xprt(struct xprt_create *args,
  2406. unsigned int slot_table_size,
  2407. unsigned int max_slot_table_size)
  2408. {
  2409. struct rpc_xprt *xprt;
  2410. struct sock_xprt *new;
  2411. if (args->addrlen > sizeof(xprt->addr)) {
  2412. dprintk("RPC: xs_setup_xprt: address too large\n");
  2413. return ERR_PTR(-EBADF);
  2414. }
  2415. xprt = xprt_alloc(args->net, sizeof(*new), slot_table_size,
  2416. max_slot_table_size);
  2417. if (xprt == NULL) {
  2418. dprintk("RPC: xs_setup_xprt: couldn't allocate "
  2419. "rpc_xprt\n");
  2420. return ERR_PTR(-ENOMEM);
  2421. }
  2422. new = container_of(xprt, struct sock_xprt, xprt);
  2423. mutex_init(&new->recv_mutex);
  2424. memcpy(&xprt->addr, args->dstaddr, args->addrlen);
  2425. xprt->addrlen = args->addrlen;
  2426. if (args->srcaddr)
  2427. memcpy(&new->srcaddr, args->srcaddr, args->addrlen);
  2428. else {
  2429. int err;
  2430. err = xs_init_anyaddr(args->dstaddr->sa_family,
  2431. (struct sockaddr *)&new->srcaddr);
  2432. if (err != 0) {
  2433. xprt_free(xprt);
  2434. return ERR_PTR(err);
  2435. }
  2436. }
  2437. return xprt;
  2438. }
  2439. static const struct rpc_timeout xs_local_default_timeout = {
  2440. .to_initval = 10 * HZ,
  2441. .to_maxval = 10 * HZ,
  2442. .to_retries = 2,
  2443. };
  2444. /**
  2445. * xs_setup_local - Set up transport to use an AF_LOCAL socket
  2446. * @args: rpc transport creation arguments
  2447. *
  2448. * AF_LOCAL is a "tpi_cots_ord" transport, just like TCP
  2449. */
  2450. static struct rpc_xprt *xs_setup_local(struct xprt_create *args)
  2451. {
  2452. struct sockaddr_un *sun = (struct sockaddr_un *)args->dstaddr;
  2453. struct sock_xprt *transport;
  2454. struct rpc_xprt *xprt;
  2455. struct rpc_xprt *ret;
  2456. xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
  2457. xprt_max_tcp_slot_table_entries);
  2458. if (IS_ERR(xprt))
  2459. return xprt;
  2460. transport = container_of(xprt, struct sock_xprt, xprt);
  2461. xprt->prot = 0;
  2462. xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
  2463. xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
  2464. xprt->bind_timeout = XS_BIND_TO;
  2465. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  2466. xprt->idle_timeout = XS_IDLE_DISC_TO;
  2467. xprt->ops = &xs_local_ops;
  2468. xprt->timeout = &xs_local_default_timeout;
  2469. INIT_WORK(&transport->recv_worker, xs_local_data_receive_workfn);
  2470. INIT_DELAYED_WORK(&transport->connect_worker,
  2471. xs_dummy_setup_socket);
  2472. switch (sun->sun_family) {
  2473. case AF_LOCAL:
  2474. if (sun->sun_path[0] != '/') {
  2475. dprintk("RPC: bad AF_LOCAL address: %s\n",
  2476. sun->sun_path);
  2477. ret = ERR_PTR(-EINVAL);
  2478. goto out_err;
  2479. }
  2480. xprt_set_bound(xprt);
  2481. xs_format_peer_addresses(xprt, "local", RPCBIND_NETID_LOCAL);
  2482. ret = ERR_PTR(xs_local_setup_socket(transport));
  2483. if (ret)
  2484. goto out_err;
  2485. break;
  2486. default:
  2487. ret = ERR_PTR(-EAFNOSUPPORT);
  2488. goto out_err;
  2489. }
  2490. dprintk("RPC: set up xprt to %s via AF_LOCAL\n",
  2491. xprt->address_strings[RPC_DISPLAY_ADDR]);
  2492. if (try_module_get(THIS_MODULE))
  2493. return xprt;
  2494. ret = ERR_PTR(-EINVAL);
  2495. out_err:
  2496. xs_xprt_free(xprt);
  2497. return ret;
  2498. }
  2499. static const struct rpc_timeout xs_udp_default_timeout = {
  2500. .to_initval = 5 * HZ,
  2501. .to_maxval = 30 * HZ,
  2502. .to_increment = 5 * HZ,
  2503. .to_retries = 5,
  2504. };
  2505. /**
  2506. * xs_setup_udp - Set up transport to use a UDP socket
  2507. * @args: rpc transport creation arguments
  2508. *
  2509. */
  2510. static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
  2511. {
  2512. struct sockaddr *addr = args->dstaddr;
  2513. struct rpc_xprt *xprt;
  2514. struct sock_xprt *transport;
  2515. struct rpc_xprt *ret;
  2516. xprt = xs_setup_xprt(args, xprt_udp_slot_table_entries,
  2517. xprt_udp_slot_table_entries);
  2518. if (IS_ERR(xprt))
  2519. return xprt;
  2520. transport = container_of(xprt, struct sock_xprt, xprt);
  2521. xprt->prot = IPPROTO_UDP;
  2522. xprt->tsh_size = 0;
  2523. /* XXX: header size can vary due to auth type, IPv6, etc. */
  2524. xprt->max_payload = (1U << 16) - (MAX_HEADER << 3);
  2525. xprt->bind_timeout = XS_BIND_TO;
  2526. xprt->reestablish_timeout = XS_UDP_REEST_TO;
  2527. xprt->idle_timeout = XS_IDLE_DISC_TO;
  2528. xprt->ops = &xs_udp_ops;
  2529. xprt->timeout = &xs_udp_default_timeout;
  2530. INIT_WORK(&transport->recv_worker, xs_udp_data_receive_workfn);
  2531. INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
  2532. switch (addr->sa_family) {
  2533. case AF_INET:
  2534. if (((struct sockaddr_in *)addr)->sin_port != htons(0))
  2535. xprt_set_bound(xprt);
  2536. xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP);
  2537. break;
  2538. case AF_INET6:
  2539. if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
  2540. xprt_set_bound(xprt);
  2541. xs_format_peer_addresses(xprt, "udp", RPCBIND_NETID_UDP6);
  2542. break;
  2543. default:
  2544. ret = ERR_PTR(-EAFNOSUPPORT);
  2545. goto out_err;
  2546. }
  2547. if (xprt_bound(xprt))
  2548. dprintk("RPC: set up xprt to %s (port %s) via %s\n",
  2549. xprt->address_strings[RPC_DISPLAY_ADDR],
  2550. xprt->address_strings[RPC_DISPLAY_PORT],
  2551. xprt->address_strings[RPC_DISPLAY_PROTO]);
  2552. else
  2553. dprintk("RPC: set up xprt to %s (autobind) via %s\n",
  2554. xprt->address_strings[RPC_DISPLAY_ADDR],
  2555. xprt->address_strings[RPC_DISPLAY_PROTO]);
  2556. if (try_module_get(THIS_MODULE))
  2557. return xprt;
  2558. ret = ERR_PTR(-EINVAL);
  2559. out_err:
  2560. xs_xprt_free(xprt);
  2561. return ret;
  2562. }
  2563. static const struct rpc_timeout xs_tcp_default_timeout = {
  2564. .to_initval = 60 * HZ,
  2565. .to_maxval = 60 * HZ,
  2566. .to_retries = 2,
  2567. };
  2568. /**
  2569. * xs_setup_tcp - Set up transport to use a TCP socket
  2570. * @args: rpc transport creation arguments
  2571. *
  2572. */
  2573. static struct rpc_xprt *xs_setup_tcp(struct xprt_create *args)
  2574. {
  2575. struct sockaddr *addr = args->dstaddr;
  2576. struct rpc_xprt *xprt;
  2577. struct sock_xprt *transport;
  2578. struct rpc_xprt *ret;
  2579. unsigned int max_slot_table_size = xprt_max_tcp_slot_table_entries;
  2580. if (args->flags & XPRT_CREATE_INFINITE_SLOTS)
  2581. max_slot_table_size = RPC_MAX_SLOT_TABLE_LIMIT;
  2582. xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
  2583. max_slot_table_size);
  2584. if (IS_ERR(xprt))
  2585. return xprt;
  2586. transport = container_of(xprt, struct sock_xprt, xprt);
  2587. xprt->prot = IPPROTO_TCP;
  2588. xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
  2589. xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
  2590. xprt->bind_timeout = XS_BIND_TO;
  2591. xprt->reestablish_timeout = XS_TCP_INIT_REEST_TO;
  2592. xprt->idle_timeout = XS_IDLE_DISC_TO;
  2593. xprt->ops = &xs_tcp_ops;
  2594. xprt->timeout = &xs_tcp_default_timeout;
  2595. INIT_WORK(&transport->recv_worker, xs_tcp_data_receive_workfn);
  2596. INIT_DELAYED_WORK(&transport->connect_worker, xs_tcp_setup_socket);
  2597. switch (addr->sa_family) {
  2598. case AF_INET:
  2599. if (((struct sockaddr_in *)addr)->sin_port != htons(0))
  2600. xprt_set_bound(xprt);
  2601. xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP);
  2602. break;
  2603. case AF_INET6:
  2604. if (((struct sockaddr_in6 *)addr)->sin6_port != htons(0))
  2605. xprt_set_bound(xprt);
  2606. xs_format_peer_addresses(xprt, "tcp", RPCBIND_NETID_TCP6);
  2607. break;
  2608. default:
  2609. ret = ERR_PTR(-EAFNOSUPPORT);
  2610. goto out_err;
  2611. }
  2612. if (xprt_bound(xprt))
  2613. dprintk("RPC: set up xprt to %s (port %s) via %s\n",
  2614. xprt->address_strings[RPC_DISPLAY_ADDR],
  2615. xprt->address_strings[RPC_DISPLAY_PORT],
  2616. xprt->address_strings[RPC_DISPLAY_PROTO]);
  2617. else
  2618. dprintk("RPC: set up xprt to %s (autobind) via %s\n",
  2619. xprt->address_strings[RPC_DISPLAY_ADDR],
  2620. xprt->address_strings[RPC_DISPLAY_PROTO]);
  2621. if (try_module_get(THIS_MODULE))
  2622. return xprt;
  2623. ret = ERR_PTR(-EINVAL);
  2624. out_err:
  2625. xs_xprt_free(xprt);
  2626. return ret;
  2627. }
  2628. /**
  2629. * xs_setup_bc_tcp - Set up transport to use a TCP backchannel socket
  2630. * @args: rpc transport creation arguments
  2631. *
  2632. */
  2633. static struct rpc_xprt *xs_setup_bc_tcp(struct xprt_create *args)
  2634. {
  2635. struct sockaddr *addr = args->dstaddr;
  2636. struct rpc_xprt *xprt;
  2637. struct sock_xprt *transport;
  2638. struct svc_sock *bc_sock;
  2639. struct rpc_xprt *ret;
  2640. xprt = xs_setup_xprt(args, xprt_tcp_slot_table_entries,
  2641. xprt_tcp_slot_table_entries);
  2642. if (IS_ERR(xprt))
  2643. return xprt;
  2644. transport = container_of(xprt, struct sock_xprt, xprt);
  2645. xprt->prot = IPPROTO_TCP;
  2646. xprt->tsh_size = sizeof(rpc_fraghdr) / sizeof(u32);
  2647. xprt->max_payload = RPC_MAX_FRAGMENT_SIZE;
  2648. xprt->timeout = &xs_tcp_default_timeout;
  2649. /* backchannel */
  2650. xprt_set_bound(xprt);
  2651. xprt->bind_timeout = 0;
  2652. xprt->reestablish_timeout = 0;
  2653. xprt->idle_timeout = 0;
  2654. xprt->ops = &bc_tcp_ops;
  2655. switch (addr->sa_family) {
  2656. case AF_INET:
  2657. xs_format_peer_addresses(xprt, "tcp",
  2658. RPCBIND_NETID_TCP);
  2659. break;
  2660. case AF_INET6:
  2661. xs_format_peer_addresses(xprt, "tcp",
  2662. RPCBIND_NETID_TCP6);
  2663. break;
  2664. default:
  2665. ret = ERR_PTR(-EAFNOSUPPORT);
  2666. goto out_err;
  2667. }
  2668. dprintk("RPC: set up xprt to %s (port %s) via %s\n",
  2669. xprt->address_strings[RPC_DISPLAY_ADDR],
  2670. xprt->address_strings[RPC_DISPLAY_PORT],
  2671. xprt->address_strings[RPC_DISPLAY_PROTO]);
  2672. /*
  2673. * Once we've associated a backchannel xprt with a connection,
  2674. * we want to keep it around as long as the connection lasts,
  2675. * in case we need to start using it for a backchannel again;
  2676. * this reference won't be dropped until bc_xprt is destroyed.
  2677. */
  2678. xprt_get(xprt);
  2679. args->bc_xprt->xpt_bc_xprt = xprt;
  2680. xprt->bc_xprt = args->bc_xprt;
  2681. bc_sock = container_of(args->bc_xprt, struct svc_sock, sk_xprt);
  2682. transport->sock = bc_sock->sk_sock;
  2683. transport->inet = bc_sock->sk_sk;
  2684. /*
  2685. * Since we don't want connections for the backchannel, we set
  2686. * the xprt status to connected
  2687. */
  2688. xprt_set_connected(xprt);
  2689. if (try_module_get(THIS_MODULE))
  2690. return xprt;
  2691. args->bc_xprt->xpt_bc_xprt = NULL;
  2692. args->bc_xprt->xpt_bc_xps = NULL;
  2693. xprt_put(xprt);
  2694. ret = ERR_PTR(-EINVAL);
  2695. out_err:
  2696. xs_xprt_free(xprt);
  2697. return ret;
  2698. }
  2699. static struct xprt_class xs_local_transport = {
  2700. .list = LIST_HEAD_INIT(xs_local_transport.list),
  2701. .name = "named UNIX socket",
  2702. .owner = THIS_MODULE,
  2703. .ident = XPRT_TRANSPORT_LOCAL,
  2704. .setup = xs_setup_local,
  2705. };
  2706. static struct xprt_class xs_udp_transport = {
  2707. .list = LIST_HEAD_INIT(xs_udp_transport.list),
  2708. .name = "udp",
  2709. .owner = THIS_MODULE,
  2710. .ident = XPRT_TRANSPORT_UDP,
  2711. .setup = xs_setup_udp,
  2712. };
  2713. static struct xprt_class xs_tcp_transport = {
  2714. .list = LIST_HEAD_INIT(xs_tcp_transport.list),
  2715. .name = "tcp",
  2716. .owner = THIS_MODULE,
  2717. .ident = XPRT_TRANSPORT_TCP,
  2718. .setup = xs_setup_tcp,
  2719. };
  2720. static struct xprt_class xs_bc_tcp_transport = {
  2721. .list = LIST_HEAD_INIT(xs_bc_tcp_transport.list),
  2722. .name = "tcp NFSv4.1 backchannel",
  2723. .owner = THIS_MODULE,
  2724. .ident = XPRT_TRANSPORT_BC_TCP,
  2725. .setup = xs_setup_bc_tcp,
  2726. };
  2727. /**
  2728. * init_socket_xprt - set up xprtsock's sysctls, register with RPC client
  2729. *
  2730. */
  2731. int init_socket_xprt(void)
  2732. {
  2733. #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  2734. if (!sunrpc_table_header)
  2735. sunrpc_table_header = register_sysctl_table(sunrpc_table);
  2736. #endif
  2737. xprt_register_transport(&xs_local_transport);
  2738. xprt_register_transport(&xs_udp_transport);
  2739. xprt_register_transport(&xs_tcp_transport);
  2740. xprt_register_transport(&xs_bc_tcp_transport);
  2741. return 0;
  2742. }
  2743. /**
  2744. * cleanup_socket_xprt - remove xprtsock's sysctls, unregister
  2745. *
  2746. */
  2747. void cleanup_socket_xprt(void)
  2748. {
  2749. #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  2750. if (sunrpc_table_header) {
  2751. unregister_sysctl_table(sunrpc_table_header);
  2752. sunrpc_table_header = NULL;
  2753. }
  2754. #endif
  2755. xprt_unregister_transport(&xs_local_transport);
  2756. xprt_unregister_transport(&xs_udp_transport);
  2757. xprt_unregister_transport(&xs_tcp_transport);
  2758. xprt_unregister_transport(&xs_bc_tcp_transport);
  2759. }
  2760. static int param_set_uint_minmax(const char *val,
  2761. const struct kernel_param *kp,
  2762. unsigned int min, unsigned int max)
  2763. {
  2764. unsigned int num;
  2765. int ret;
  2766. if (!val)
  2767. return -EINVAL;
  2768. ret = kstrtouint(val, 0, &num);
  2769. if (ret == -EINVAL || num < min || num > max)
  2770. return -EINVAL;
  2771. *((unsigned int *)kp->arg) = num;
  2772. return 0;
  2773. }
  2774. static int param_set_portnr(const char *val, const struct kernel_param *kp)
  2775. {
  2776. if (kp->arg == &xprt_min_resvport)
  2777. return param_set_uint_minmax(val, kp,
  2778. RPC_MIN_RESVPORT,
  2779. xprt_max_resvport);
  2780. return param_set_uint_minmax(val, kp,
  2781. xprt_min_resvport,
  2782. RPC_MAX_RESVPORT);
  2783. }
  2784. static const struct kernel_param_ops param_ops_portnr = {
  2785. .set = param_set_portnr,
  2786. .get = param_get_uint,
  2787. };
  2788. #define param_check_portnr(name, p) \
  2789. __param_check(name, p, unsigned int);
  2790. module_param_named(min_resvport, xprt_min_resvport, portnr, 0644);
  2791. module_param_named(max_resvport, xprt_max_resvport, portnr, 0644);
  2792. static int param_set_slot_table_size(const char *val,
  2793. const struct kernel_param *kp)
  2794. {
  2795. return param_set_uint_minmax(val, kp,
  2796. RPC_MIN_SLOT_TABLE,
  2797. RPC_MAX_SLOT_TABLE);
  2798. }
  2799. static const struct kernel_param_ops param_ops_slot_table_size = {
  2800. .set = param_set_slot_table_size,
  2801. .get = param_get_uint,
  2802. };
  2803. #define param_check_slot_table_size(name, p) \
  2804. __param_check(name, p, unsigned int);
  2805. static int param_set_max_slot_table_size(const char *val,
  2806. const struct kernel_param *kp)
  2807. {
  2808. return param_set_uint_minmax(val, kp,
  2809. RPC_MIN_SLOT_TABLE,
  2810. RPC_MAX_SLOT_TABLE_LIMIT);
  2811. }
  2812. static const struct kernel_param_ops param_ops_max_slot_table_size = {
  2813. .set = param_set_max_slot_table_size,
  2814. .get = param_get_uint,
  2815. };
  2816. #define param_check_max_slot_table_size(name, p) \
  2817. __param_check(name, p, unsigned int);
  2818. module_param_named(tcp_slot_table_entries, xprt_tcp_slot_table_entries,
  2819. slot_table_size, 0644);
  2820. module_param_named(tcp_max_slot_table_entries, xprt_max_tcp_slot_table_entries,
  2821. max_slot_table_size, 0644);
  2822. module_param_named(udp_slot_table_entries, xprt_udp_slot_table_entries,
  2823. slot_table_size, 0644);