messenger.c 71 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860
  1. #include <linux/ceph/ceph_debug.h>
  2. #include <linux/crc32c.h>
  3. #include <linux/ctype.h>
  4. #include <linux/highmem.h>
  5. #include <linux/inet.h>
  6. #include <linux/kthread.h>
  7. #include <linux/net.h>
  8. #include <linux/slab.h>
  9. #include <linux/socket.h>
  10. #include <linux/string.h>
  11. #include <linux/bio.h>
  12. #include <linux/blkdev.h>
  13. #include <linux/dns_resolver.h>
  14. #include <net/tcp.h>
  15. #include <linux/ceph/libceph.h>
  16. #include <linux/ceph/messenger.h>
  17. #include <linux/ceph/decode.h>
  18. #include <linux/ceph/pagelist.h>
  19. #include <linux/export.h>
  20. /*
  21. * Ceph uses the messenger to exchange ceph_msg messages with other
  22. * hosts in the system. The messenger provides ordered and reliable
  23. * delivery. We tolerate TCP disconnects by reconnecting (with
  24. * exponential backoff) in the case of a fault (disconnection, bad
  25. * crc, protocol error). Acks allow sent messages to be discarded by
  26. * the sender.
  27. */
  28. /*
  29. * We track the state of the socket on a given connection using
  30. * values defined below. The transition to a new socket state is
  31. * handled by a function which verifies we aren't coming from an
  32. * unexpected state.
  33. *
  34. * --------
  35. * | NEW* | transient initial state
  36. * --------
  37. * | con_sock_state_init()
  38. * v
  39. * ----------
  40. * | CLOSED | initialized, but no socket (and no
  41. * ---------- TCP connection)
  42. * ^ \
  43. * | \ con_sock_state_connecting()
  44. * | ----------------------
  45. * | \
  46. * + con_sock_state_closed() \
  47. * |+--------------------------- \
  48. * | \ \ \
  49. * | ----------- \ \
  50. * | | CLOSING | socket event; \ \
  51. * | ----------- await close \ \
  52. * | ^ \ |
  53. * | | \ |
  54. * | + con_sock_state_closing() \ |
  55. * | / \ | |
  56. * | / --------------- | |
  57. * | / \ v v
  58. * | / --------------
  59. * | / -----------------| CONNECTING | socket created, TCP
  60. * | | / -------------- connect initiated
  61. * | | | con_sock_state_connected()
  62. * | | v
  63. * -------------
  64. * | CONNECTED | TCP connection established
  65. * -------------
  66. *
  67. * State values for ceph_connection->sock_state; NEW is assumed to be 0.
  68. */
  69. #define CON_SOCK_STATE_NEW 0 /* -> CLOSED */
  70. #define CON_SOCK_STATE_CLOSED 1 /* -> CONNECTING */
  71. #define CON_SOCK_STATE_CONNECTING 2 /* -> CONNECTED or -> CLOSING */
  72. #define CON_SOCK_STATE_CONNECTED 3 /* -> CLOSING or -> CLOSED */
  73. #define CON_SOCK_STATE_CLOSING 4 /* -> CLOSED */
  74. /*
  75. * connection states
  76. */
  77. #define CON_STATE_CLOSED 1 /* -> PREOPEN */
  78. #define CON_STATE_PREOPEN 2 /* -> CONNECTING, CLOSED */
  79. #define CON_STATE_CONNECTING 3 /* -> NEGOTIATING, CLOSED */
  80. #define CON_STATE_NEGOTIATING 4 /* -> OPEN, CLOSED */
  81. #define CON_STATE_OPEN 5 /* -> STANDBY, CLOSED */
  82. #define CON_STATE_STANDBY 6 /* -> PREOPEN, CLOSED */
  83. /*
  84. * ceph_connection flag bits
  85. */
  86. #define CON_FLAG_LOSSYTX 0 /* we can close channel or drop
  87. * messages on errors */
  88. #define CON_FLAG_KEEPALIVE_PENDING 1 /* we need to send a keepalive */
  89. #define CON_FLAG_WRITE_PENDING 2 /* we have data ready to send */
  90. #define CON_FLAG_SOCK_CLOSED 3 /* socket state changed to closed */
  91. #define CON_FLAG_BACKOFF 4 /* need to retry queuing delayed work */
  92. /* static tag bytes (protocol control messages) */
  93. static char tag_msg = CEPH_MSGR_TAG_MSG;
  94. static char tag_ack = CEPH_MSGR_TAG_ACK;
  95. static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
  96. #ifdef CONFIG_LOCKDEP
  97. static struct lock_class_key socket_class;
  98. #endif
  99. /*
  100. * When skipping (ignoring) a block of input we read it into a "skip
  101. * buffer," which is this many bytes in size.
  102. */
  103. #define SKIP_BUF_SIZE 1024
  104. static void queue_con(struct ceph_connection *con);
  105. static void con_work(struct work_struct *);
  106. static void ceph_fault(struct ceph_connection *con);
  107. /*
  108. * Nicely render a sockaddr as a string. An array of formatted
  109. * strings is used, to approximate reentrancy.
  110. */
  111. #define ADDR_STR_COUNT_LOG 5 /* log2(# address strings in array) */
  112. #define ADDR_STR_COUNT (1 << ADDR_STR_COUNT_LOG)
  113. #define ADDR_STR_COUNT_MASK (ADDR_STR_COUNT - 1)
  114. #define MAX_ADDR_STR_LEN 64 /* 54 is enough */
  115. static char addr_str[ADDR_STR_COUNT][MAX_ADDR_STR_LEN];
  116. static atomic_t addr_str_seq = ATOMIC_INIT(0);
  117. static struct page *zero_page; /* used in certain error cases */
  118. const char *ceph_pr_addr(const struct sockaddr_storage *ss)
  119. {
  120. int i;
  121. char *s;
  122. struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
  123. struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
  124. i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
  125. s = addr_str[i];
  126. switch (ss->ss_family) {
  127. case AF_INET:
  128. snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
  129. ntohs(in4->sin_port));
  130. break;
  131. case AF_INET6:
  132. snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
  133. ntohs(in6->sin6_port));
  134. break;
  135. default:
  136. snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
  137. ss->ss_family);
  138. }
  139. return s;
  140. }
  141. EXPORT_SYMBOL(ceph_pr_addr);
  142. static void encode_my_addr(struct ceph_messenger *msgr)
  143. {
  144. memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
  145. ceph_encode_addr(&msgr->my_enc_addr);
  146. }
  147. /*
  148. * work queue for all reading and writing to/from the socket.
  149. */
  150. static struct workqueue_struct *ceph_msgr_wq;
  151. void _ceph_msgr_exit(void)
  152. {
  153. if (ceph_msgr_wq) {
  154. destroy_workqueue(ceph_msgr_wq);
  155. ceph_msgr_wq = NULL;
  156. }
  157. BUG_ON(zero_page == NULL);
  158. kunmap(zero_page);
  159. page_cache_release(zero_page);
  160. zero_page = NULL;
  161. }
  162. int ceph_msgr_init(void)
  163. {
  164. BUG_ON(zero_page != NULL);
  165. zero_page = ZERO_PAGE(0);
  166. page_cache_get(zero_page);
  167. ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_NON_REENTRANT, 0);
  168. if (ceph_msgr_wq)
  169. return 0;
  170. pr_err("msgr_init failed to create workqueue\n");
  171. _ceph_msgr_exit();
  172. return -ENOMEM;
  173. }
  174. EXPORT_SYMBOL(ceph_msgr_init);
  175. void ceph_msgr_exit(void)
  176. {
  177. BUG_ON(ceph_msgr_wq == NULL);
  178. _ceph_msgr_exit();
  179. }
  180. EXPORT_SYMBOL(ceph_msgr_exit);
  181. void ceph_msgr_flush(void)
  182. {
  183. flush_workqueue(ceph_msgr_wq);
  184. }
  185. EXPORT_SYMBOL(ceph_msgr_flush);
  186. /* Connection socket state transition functions */
  187. static void con_sock_state_init(struct ceph_connection *con)
  188. {
  189. int old_state;
  190. old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
  191. if (WARN_ON(old_state != CON_SOCK_STATE_NEW))
  192. printk("%s: unexpected old state %d\n", __func__, old_state);
  193. dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
  194. CON_SOCK_STATE_CLOSED);
  195. }
  196. static void con_sock_state_connecting(struct ceph_connection *con)
  197. {
  198. int old_state;
  199. old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTING);
  200. if (WARN_ON(old_state != CON_SOCK_STATE_CLOSED))
  201. printk("%s: unexpected old state %d\n", __func__, old_state);
  202. dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
  203. CON_SOCK_STATE_CONNECTING);
  204. }
  205. static void con_sock_state_connected(struct ceph_connection *con)
  206. {
  207. int old_state;
  208. old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTED);
  209. if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING))
  210. printk("%s: unexpected old state %d\n", __func__, old_state);
  211. dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
  212. CON_SOCK_STATE_CONNECTED);
  213. }
  214. static void con_sock_state_closing(struct ceph_connection *con)
  215. {
  216. int old_state;
  217. old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSING);
  218. if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING &&
  219. old_state != CON_SOCK_STATE_CONNECTED &&
  220. old_state != CON_SOCK_STATE_CLOSING))
  221. printk("%s: unexpected old state %d\n", __func__, old_state);
  222. dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
  223. CON_SOCK_STATE_CLOSING);
  224. }
  225. static void con_sock_state_closed(struct ceph_connection *con)
  226. {
  227. int old_state;
  228. old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
  229. if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTED &&
  230. old_state != CON_SOCK_STATE_CLOSING &&
  231. old_state != CON_SOCK_STATE_CONNECTING &&
  232. old_state != CON_SOCK_STATE_CLOSED))
  233. printk("%s: unexpected old state %d\n", __func__, old_state);
  234. dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
  235. CON_SOCK_STATE_CLOSED);
  236. }
  237. /*
  238. * socket callback functions
  239. */
  240. /* data available on socket, or listen socket received a connect */
  241. static void ceph_sock_data_ready(struct sock *sk, int count_unused)
  242. {
  243. struct ceph_connection *con = sk->sk_user_data;
  244. if (atomic_read(&con->msgr->stopping)) {
  245. return;
  246. }
  247. if (sk->sk_state != TCP_CLOSE_WAIT) {
  248. dout("%s on %p state = %lu, queueing work\n", __func__,
  249. con, con->state);
  250. queue_con(con);
  251. }
  252. }
  253. /* socket has buffer space for writing */
  254. static void ceph_sock_write_space(struct sock *sk)
  255. {
  256. struct ceph_connection *con = sk->sk_user_data;
  257. /* only queue to workqueue if there is data we want to write,
  258. * and there is sufficient space in the socket buffer to accept
  259. * more data. clear SOCK_NOSPACE so that ceph_sock_write_space()
  260. * doesn't get called again until try_write() fills the socket
  261. * buffer. See net/ipv4/tcp_input.c:tcp_check_space()
  262. * and net/core/stream.c:sk_stream_write_space().
  263. */
  264. if (test_bit(CON_FLAG_WRITE_PENDING, &con->flags)) {
  265. if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk)) {
  266. dout("%s %p queueing write work\n", __func__, con);
  267. clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
  268. queue_con(con);
  269. }
  270. } else {
  271. dout("%s %p nothing to write\n", __func__, con);
  272. }
  273. }
  274. /* socket's state has changed */
  275. static void ceph_sock_state_change(struct sock *sk)
  276. {
  277. struct ceph_connection *con = sk->sk_user_data;
  278. dout("%s %p state = %lu sk_state = %u\n", __func__,
  279. con, con->state, sk->sk_state);
  280. switch (sk->sk_state) {
  281. case TCP_CLOSE:
  282. dout("%s TCP_CLOSE\n", __func__);
  283. case TCP_CLOSE_WAIT:
  284. dout("%s TCP_CLOSE_WAIT\n", __func__);
  285. con_sock_state_closing(con);
  286. set_bit(CON_FLAG_SOCK_CLOSED, &con->flags);
  287. queue_con(con);
  288. break;
  289. case TCP_ESTABLISHED:
  290. dout("%s TCP_ESTABLISHED\n", __func__);
  291. con_sock_state_connected(con);
  292. queue_con(con);
  293. break;
  294. default: /* Everything else is uninteresting */
  295. break;
  296. }
  297. }
  298. /*
  299. * set up socket callbacks
  300. */
  301. static void set_sock_callbacks(struct socket *sock,
  302. struct ceph_connection *con)
  303. {
  304. struct sock *sk = sock->sk;
  305. sk->sk_user_data = con;
  306. sk->sk_data_ready = ceph_sock_data_ready;
  307. sk->sk_write_space = ceph_sock_write_space;
  308. sk->sk_state_change = ceph_sock_state_change;
  309. }
  310. /*
  311. * socket helpers
  312. */
  313. /*
  314. * initiate connection to a remote socket.
  315. */
  316. static int ceph_tcp_connect(struct ceph_connection *con)
  317. {
  318. struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
  319. struct socket *sock;
  320. int ret;
  321. BUG_ON(con->sock);
  322. ret = sock_create_kern(con->peer_addr.in_addr.ss_family, SOCK_STREAM,
  323. IPPROTO_TCP, &sock);
  324. if (ret)
  325. return ret;
  326. sock->sk->sk_allocation = GFP_NOFS;
  327. #ifdef CONFIG_LOCKDEP
  328. lockdep_set_class(&sock->sk->sk_lock, &socket_class);
  329. #endif
  330. set_sock_callbacks(sock, con);
  331. dout("connect %s\n", ceph_pr_addr(&con->peer_addr.in_addr));
  332. con_sock_state_connecting(con);
  333. ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
  334. O_NONBLOCK);
  335. if (ret == -EINPROGRESS) {
  336. dout("connect %s EINPROGRESS sk_state = %u\n",
  337. ceph_pr_addr(&con->peer_addr.in_addr),
  338. sock->sk->sk_state);
  339. } else if (ret < 0) {
  340. pr_err("connect %s error %d\n",
  341. ceph_pr_addr(&con->peer_addr.in_addr), ret);
  342. sock_release(sock);
  343. con->error_msg = "connect error";
  344. return ret;
  345. }
  346. con->sock = sock;
  347. return 0;
  348. }
  349. static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
  350. {
  351. struct kvec iov = {buf, len};
  352. struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
  353. int r;
  354. r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
  355. if (r == -EAGAIN)
  356. r = 0;
  357. return r;
  358. }
  359. /*
  360. * write something. @more is true if caller will be sending more data
  361. * shortly.
  362. */
  363. static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
  364. size_t kvlen, size_t len, int more)
  365. {
  366. struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
  367. int r;
  368. if (more)
  369. msg.msg_flags |= MSG_MORE;
  370. else
  371. msg.msg_flags |= MSG_EOR; /* superfluous, but what the hell */
  372. r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
  373. if (r == -EAGAIN)
  374. r = 0;
  375. return r;
  376. }
  377. static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
  378. int offset, size_t size, int more)
  379. {
  380. int flags = MSG_DONTWAIT | MSG_NOSIGNAL | (more ? MSG_MORE : MSG_EOR);
  381. int ret;
  382. ret = kernel_sendpage(sock, page, offset, size, flags);
  383. if (ret == -EAGAIN)
  384. ret = 0;
  385. return ret;
  386. }
  387. /*
  388. * Shutdown/close the socket for the given connection.
  389. */
  390. static int con_close_socket(struct ceph_connection *con)
  391. {
  392. int rc = 0;
  393. dout("con_close_socket on %p sock %p\n", con, con->sock);
  394. if (con->sock) {
  395. rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
  396. sock_release(con->sock);
  397. con->sock = NULL;
  398. }
  399. /*
  400. * Forcibly clear the SOCK_CLOSED flag. It gets set
  401. * independent of the connection mutex, and we could have
  402. * received a socket close event before we had the chance to
  403. * shut the socket down.
  404. */
  405. clear_bit(CON_FLAG_SOCK_CLOSED, &con->flags);
  406. con_sock_state_closed(con);
  407. return rc;
  408. }
  409. /*
  410. * Reset a connection. Discard all incoming and outgoing messages
  411. * and clear *_seq state.
  412. */
  413. static void ceph_msg_remove(struct ceph_msg *msg)
  414. {
  415. list_del_init(&msg->list_head);
  416. BUG_ON(msg->con == NULL);
  417. msg->con->ops->put(msg->con);
  418. msg->con = NULL;
  419. ceph_msg_put(msg);
  420. }
  421. static void ceph_msg_remove_list(struct list_head *head)
  422. {
  423. while (!list_empty(head)) {
  424. struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
  425. list_head);
  426. ceph_msg_remove(msg);
  427. }
  428. }
  429. static void reset_connection(struct ceph_connection *con)
  430. {
  431. /* reset connection, out_queue, msg_ and connect_seq */
  432. /* discard existing out_queue and msg_seq */
  433. ceph_msg_remove_list(&con->out_queue);
  434. ceph_msg_remove_list(&con->out_sent);
  435. if (con->in_msg) {
  436. BUG_ON(con->in_msg->con != con);
  437. con->in_msg->con = NULL;
  438. ceph_msg_put(con->in_msg);
  439. con->in_msg = NULL;
  440. con->ops->put(con);
  441. }
  442. con->connect_seq = 0;
  443. con->out_seq = 0;
  444. if (con->out_msg) {
  445. ceph_msg_put(con->out_msg);
  446. con->out_msg = NULL;
  447. }
  448. con->in_seq = 0;
  449. con->in_seq_acked = 0;
  450. }
  451. /*
  452. * mark a peer down. drop any open connections.
  453. */
  454. void ceph_con_close(struct ceph_connection *con)
  455. {
  456. mutex_lock(&con->mutex);
  457. dout("con_close %p peer %s\n", con,
  458. ceph_pr_addr(&con->peer_addr.in_addr));
  459. con->state = CON_STATE_CLOSED;
  460. clear_bit(CON_FLAG_LOSSYTX, &con->flags); /* so we retry next connect */
  461. clear_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags);
  462. clear_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  463. clear_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags);
  464. clear_bit(CON_FLAG_BACKOFF, &con->flags);
  465. reset_connection(con);
  466. con->peer_global_seq = 0;
  467. cancel_delayed_work(&con->work);
  468. con_close_socket(con);
  469. mutex_unlock(&con->mutex);
  470. }
  471. EXPORT_SYMBOL(ceph_con_close);
  472. /*
  473. * Reopen a closed connection, with a new peer address.
  474. */
  475. void ceph_con_open(struct ceph_connection *con,
  476. __u8 entity_type, __u64 entity_num,
  477. struct ceph_entity_addr *addr)
  478. {
  479. mutex_lock(&con->mutex);
  480. dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
  481. BUG_ON(con->state != CON_STATE_CLOSED);
  482. con->state = CON_STATE_PREOPEN;
  483. con->peer_name.type = (__u8) entity_type;
  484. con->peer_name.num = cpu_to_le64(entity_num);
  485. memcpy(&con->peer_addr, addr, sizeof(*addr));
  486. con->delay = 0; /* reset backoff memory */
  487. mutex_unlock(&con->mutex);
  488. queue_con(con);
  489. }
  490. EXPORT_SYMBOL(ceph_con_open);
  491. /*
  492. * return true if this connection ever successfully opened
  493. */
  494. bool ceph_con_opened(struct ceph_connection *con)
  495. {
  496. return con->connect_seq > 0;
  497. }
  498. /*
  499. * initialize a new connection.
  500. */
  501. void ceph_con_init(struct ceph_connection *con, void *private,
  502. const struct ceph_connection_operations *ops,
  503. struct ceph_messenger *msgr)
  504. {
  505. dout("con_init %p\n", con);
  506. memset(con, 0, sizeof(*con));
  507. con->private = private;
  508. con->ops = ops;
  509. con->msgr = msgr;
  510. con_sock_state_init(con);
  511. mutex_init(&con->mutex);
  512. INIT_LIST_HEAD(&con->out_queue);
  513. INIT_LIST_HEAD(&con->out_sent);
  514. INIT_DELAYED_WORK(&con->work, con_work);
  515. con->state = CON_STATE_CLOSED;
  516. }
  517. EXPORT_SYMBOL(ceph_con_init);
  518. /*
  519. * We maintain a global counter to order connection attempts. Get
  520. * a unique seq greater than @gt.
  521. */
  522. static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
  523. {
  524. u32 ret;
  525. spin_lock(&msgr->global_seq_lock);
  526. if (msgr->global_seq < gt)
  527. msgr->global_seq = gt;
  528. ret = ++msgr->global_seq;
  529. spin_unlock(&msgr->global_seq_lock);
  530. return ret;
  531. }
  532. static void con_out_kvec_reset(struct ceph_connection *con)
  533. {
  534. con->out_kvec_left = 0;
  535. con->out_kvec_bytes = 0;
  536. con->out_kvec_cur = &con->out_kvec[0];
  537. }
  538. static void con_out_kvec_add(struct ceph_connection *con,
  539. size_t size, void *data)
  540. {
  541. int index;
  542. index = con->out_kvec_left;
  543. BUG_ON(index >= ARRAY_SIZE(con->out_kvec));
  544. con->out_kvec[index].iov_len = size;
  545. con->out_kvec[index].iov_base = data;
  546. con->out_kvec_left++;
  547. con->out_kvec_bytes += size;
  548. }
  549. #ifdef CONFIG_BLOCK
  550. static void init_bio_iter(struct bio *bio, struct bio **iter, int *seg)
  551. {
  552. if (!bio) {
  553. *iter = NULL;
  554. *seg = 0;
  555. return;
  556. }
  557. *iter = bio;
  558. *seg = bio->bi_idx;
  559. }
  560. static void iter_bio_next(struct bio **bio_iter, int *seg)
  561. {
  562. if (*bio_iter == NULL)
  563. return;
  564. BUG_ON(*seg >= (*bio_iter)->bi_vcnt);
  565. (*seg)++;
  566. if (*seg == (*bio_iter)->bi_vcnt)
  567. init_bio_iter((*bio_iter)->bi_next, bio_iter, seg);
  568. }
  569. #endif
  570. static void prepare_write_message_data(struct ceph_connection *con)
  571. {
  572. struct ceph_msg *msg = con->out_msg;
  573. BUG_ON(!msg);
  574. BUG_ON(!msg->hdr.data_len);
  575. /* initialize page iterator */
  576. con->out_msg_pos.page = 0;
  577. if (msg->pages)
  578. con->out_msg_pos.page_pos = msg->page_alignment;
  579. else
  580. con->out_msg_pos.page_pos = 0;
  581. #ifdef CONFIG_BLOCK
  582. if (msg->bio)
  583. init_bio_iter(msg->bio, &msg->bio_iter, &msg->bio_seg);
  584. #endif
  585. con->out_msg_pos.data_pos = 0;
  586. con->out_msg_pos.did_page_crc = false;
  587. con->out_more = 1; /* data + footer will follow */
  588. }
  589. /*
  590. * Prepare footer for currently outgoing message, and finish things
  591. * off. Assumes out_kvec* are already valid.. we just add on to the end.
  592. */
  593. static void prepare_write_message_footer(struct ceph_connection *con)
  594. {
  595. struct ceph_msg *m = con->out_msg;
  596. int v = con->out_kvec_left;
  597. m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;
  598. dout("prepare_write_message_footer %p\n", con);
  599. con->out_kvec_is_msg = true;
  600. con->out_kvec[v].iov_base = &m->footer;
  601. con->out_kvec[v].iov_len = sizeof(m->footer);
  602. con->out_kvec_bytes += sizeof(m->footer);
  603. con->out_kvec_left++;
  604. con->out_more = m->more_to_follow;
  605. con->out_msg_done = true;
  606. }
  607. /*
  608. * Prepare headers for the next outgoing message.
  609. */
  610. static void prepare_write_message(struct ceph_connection *con)
  611. {
  612. struct ceph_msg *m;
  613. u32 crc;
  614. con_out_kvec_reset(con);
  615. con->out_kvec_is_msg = true;
  616. con->out_msg_done = false;
  617. /* Sneak an ack in there first? If we can get it into the same
  618. * TCP packet that's a good thing. */
  619. if (con->in_seq > con->in_seq_acked) {
  620. con->in_seq_acked = con->in_seq;
  621. con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
  622. con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
  623. con_out_kvec_add(con, sizeof (con->out_temp_ack),
  624. &con->out_temp_ack);
  625. }
  626. BUG_ON(list_empty(&con->out_queue));
  627. m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
  628. con->out_msg = m;
  629. BUG_ON(m->con != con);
  630. /* put message on sent list */
  631. ceph_msg_get(m);
  632. list_move_tail(&m->list_head, &con->out_sent);
  633. /*
  634. * only assign outgoing seq # if we haven't sent this message
  635. * yet. if it is requeued, resend with it's original seq.
  636. */
  637. if (m->needs_out_seq) {
  638. m->hdr.seq = cpu_to_le64(++con->out_seq);
  639. m->needs_out_seq = false;
  640. }
  641. #ifdef CONFIG_BLOCK
  642. else
  643. m->bio_iter = NULL;
  644. #endif
  645. dout("prepare_write_message %p seq %lld type %d len %d+%d+%d %d pgs\n",
  646. m, con->out_seq, le16_to_cpu(m->hdr.type),
  647. le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
  648. le32_to_cpu(m->hdr.data_len),
  649. m->nr_pages);
  650. BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);
  651. /* tag + hdr + front + middle */
  652. con_out_kvec_add(con, sizeof (tag_msg), &tag_msg);
  653. con_out_kvec_add(con, sizeof (m->hdr), &m->hdr);
  654. con_out_kvec_add(con, m->front.iov_len, m->front.iov_base);
  655. if (m->middle)
  656. con_out_kvec_add(con, m->middle->vec.iov_len,
  657. m->middle->vec.iov_base);
  658. /* fill in crc (except data pages), footer */
  659. crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
  660. con->out_msg->hdr.crc = cpu_to_le32(crc);
  661. con->out_msg->footer.flags = 0;
  662. crc = crc32c(0, m->front.iov_base, m->front.iov_len);
  663. con->out_msg->footer.front_crc = cpu_to_le32(crc);
  664. if (m->middle) {
  665. crc = crc32c(0, m->middle->vec.iov_base,
  666. m->middle->vec.iov_len);
  667. con->out_msg->footer.middle_crc = cpu_to_le32(crc);
  668. } else
  669. con->out_msg->footer.middle_crc = 0;
  670. dout("%s front_crc %u middle_crc %u\n", __func__,
  671. le32_to_cpu(con->out_msg->footer.front_crc),
  672. le32_to_cpu(con->out_msg->footer.middle_crc));
  673. /* is there a data payload? */
  674. con->out_msg->footer.data_crc = 0;
  675. if (m->hdr.data_len)
  676. prepare_write_message_data(con);
  677. else
  678. /* no, queue up footer too and be done */
  679. prepare_write_message_footer(con);
  680. set_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  681. }
  682. /*
  683. * Prepare an ack.
  684. */
  685. static void prepare_write_ack(struct ceph_connection *con)
  686. {
  687. dout("prepare_write_ack %p %llu -> %llu\n", con,
  688. con->in_seq_acked, con->in_seq);
  689. con->in_seq_acked = con->in_seq;
  690. con_out_kvec_reset(con);
  691. con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
  692. con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
  693. con_out_kvec_add(con, sizeof (con->out_temp_ack),
  694. &con->out_temp_ack);
  695. con->out_more = 1; /* more will follow.. eventually.. */
  696. set_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  697. }
  698. /*
  699. * Prepare to write keepalive byte.
  700. */
  701. static void prepare_write_keepalive(struct ceph_connection *con)
  702. {
  703. dout("prepare_write_keepalive %p\n", con);
  704. con_out_kvec_reset(con);
  705. con_out_kvec_add(con, sizeof (tag_keepalive), &tag_keepalive);
  706. set_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  707. }
  708. /*
  709. * Connection negotiation.
  710. */
  711. static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
  712. int *auth_proto)
  713. {
  714. struct ceph_auth_handshake *auth;
  715. if (!con->ops->get_authorizer) {
  716. con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
  717. con->out_connect.authorizer_len = 0;
  718. return NULL;
  719. }
  720. /* Can't hold the mutex while getting authorizer */
  721. mutex_unlock(&con->mutex);
  722. auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
  723. mutex_lock(&con->mutex);
  724. if (IS_ERR(auth))
  725. return auth;
  726. if (con->state != CON_STATE_NEGOTIATING)
  727. return ERR_PTR(-EAGAIN);
  728. con->auth_reply_buf = auth->authorizer_reply_buf;
  729. con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
  730. return auth;
  731. }
  732. /*
  733. * We connected to a peer and are saying hello.
  734. */
  735. static void prepare_write_banner(struct ceph_connection *con)
  736. {
  737. con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
  738. con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
  739. &con->msgr->my_enc_addr);
  740. con->out_more = 0;
  741. set_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  742. }
  743. static int prepare_write_connect(struct ceph_connection *con)
  744. {
  745. unsigned int global_seq = get_global_seq(con->msgr, 0);
  746. int proto;
  747. int auth_proto;
  748. struct ceph_auth_handshake *auth;
  749. switch (con->peer_name.type) {
  750. case CEPH_ENTITY_TYPE_MON:
  751. proto = CEPH_MONC_PROTOCOL;
  752. break;
  753. case CEPH_ENTITY_TYPE_OSD:
  754. proto = CEPH_OSDC_PROTOCOL;
  755. break;
  756. case CEPH_ENTITY_TYPE_MDS:
  757. proto = CEPH_MDSC_PROTOCOL;
  758. break;
  759. default:
  760. BUG();
  761. }
  762. dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
  763. con->connect_seq, global_seq, proto);
  764. con->out_connect.features = cpu_to_le64(con->msgr->supported_features);
  765. con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
  766. con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
  767. con->out_connect.global_seq = cpu_to_le32(global_seq);
  768. con->out_connect.protocol_version = cpu_to_le32(proto);
  769. con->out_connect.flags = 0;
  770. auth_proto = CEPH_AUTH_UNKNOWN;
  771. auth = get_connect_authorizer(con, &auth_proto);
  772. if (IS_ERR(auth))
  773. return PTR_ERR(auth);
  774. con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
  775. con->out_connect.authorizer_len = auth ?
  776. cpu_to_le32(auth->authorizer_buf_len) : 0;
  777. con_out_kvec_add(con, sizeof (con->out_connect),
  778. &con->out_connect);
  779. if (auth && auth->authorizer_buf_len)
  780. con_out_kvec_add(con, auth->authorizer_buf_len,
  781. auth->authorizer_buf);
  782. con->out_more = 0;
  783. set_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  784. return 0;
  785. }
  786. /*
  787. * write as much of pending kvecs to the socket as we can.
  788. * 1 -> done
  789. * 0 -> socket full, but more to do
  790. * <0 -> error
  791. */
  792. static int write_partial_kvec(struct ceph_connection *con)
  793. {
  794. int ret;
  795. dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
  796. while (con->out_kvec_bytes > 0) {
  797. ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
  798. con->out_kvec_left, con->out_kvec_bytes,
  799. con->out_more);
  800. if (ret <= 0)
  801. goto out;
  802. con->out_kvec_bytes -= ret;
  803. if (con->out_kvec_bytes == 0)
  804. break; /* done */
  805. /* account for full iov entries consumed */
  806. while (ret >= con->out_kvec_cur->iov_len) {
  807. BUG_ON(!con->out_kvec_left);
  808. ret -= con->out_kvec_cur->iov_len;
  809. con->out_kvec_cur++;
  810. con->out_kvec_left--;
  811. }
  812. /* and for a partially-consumed entry */
  813. if (ret) {
  814. con->out_kvec_cur->iov_len -= ret;
  815. con->out_kvec_cur->iov_base += ret;
  816. }
  817. }
  818. con->out_kvec_left = 0;
  819. con->out_kvec_is_msg = false;
  820. ret = 1;
  821. out:
  822. dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
  823. con->out_kvec_bytes, con->out_kvec_left, ret);
  824. return ret; /* done! */
  825. }
  826. static void out_msg_pos_next(struct ceph_connection *con, struct page *page,
  827. size_t len, size_t sent, bool in_trail)
  828. {
  829. struct ceph_msg *msg = con->out_msg;
  830. BUG_ON(!msg);
  831. BUG_ON(!sent);
  832. con->out_msg_pos.data_pos += sent;
  833. con->out_msg_pos.page_pos += sent;
  834. if (sent < len)
  835. return;
  836. BUG_ON(sent != len);
  837. con->out_msg_pos.page_pos = 0;
  838. con->out_msg_pos.page++;
  839. con->out_msg_pos.did_page_crc = false;
  840. if (in_trail)
  841. list_move_tail(&page->lru,
  842. &msg->trail->head);
  843. else if (msg->pagelist)
  844. list_move_tail(&page->lru,
  845. &msg->pagelist->head);
  846. #ifdef CONFIG_BLOCK
  847. else if (msg->bio)
  848. iter_bio_next(&msg->bio_iter, &msg->bio_seg);
  849. #endif
  850. }
  851. /*
  852. * Write as much message data payload as we can. If we finish, queue
  853. * up the footer.
  854. * 1 -> done, footer is now queued in out_kvec[].
  855. * 0 -> socket full, but more to do
  856. * <0 -> error
  857. */
  858. static int write_partial_msg_pages(struct ceph_connection *con)
  859. {
  860. struct ceph_msg *msg = con->out_msg;
  861. unsigned int data_len = le32_to_cpu(msg->hdr.data_len);
  862. size_t len;
  863. bool do_datacrc = !con->msgr->nocrc;
  864. int ret;
  865. int total_max_write;
  866. bool in_trail = false;
  867. const size_t trail_len = (msg->trail ? msg->trail->length : 0);
  868. const size_t trail_off = data_len - trail_len;
  869. dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
  870. con, msg, con->out_msg_pos.page, msg->nr_pages,
  871. con->out_msg_pos.page_pos);
  872. /*
  873. * Iterate through each page that contains data to be
  874. * written, and send as much as possible for each.
  875. *
  876. * If we are calculating the data crc (the default), we will
  877. * need to map the page. If we have no pages, they have
  878. * been revoked, so use the zero page.
  879. */
  880. while (data_len > con->out_msg_pos.data_pos) {
  881. struct page *page = NULL;
  882. int max_write = PAGE_SIZE;
  883. int bio_offset = 0;
  884. in_trail = in_trail || con->out_msg_pos.data_pos >= trail_off;
  885. if (!in_trail)
  886. total_max_write = trail_off - con->out_msg_pos.data_pos;
  887. if (in_trail) {
  888. total_max_write = data_len - con->out_msg_pos.data_pos;
  889. page = list_first_entry(&msg->trail->head,
  890. struct page, lru);
  891. } else if (msg->pages) {
  892. page = msg->pages[con->out_msg_pos.page];
  893. } else if (msg->pagelist) {
  894. page = list_first_entry(&msg->pagelist->head,
  895. struct page, lru);
  896. #ifdef CONFIG_BLOCK
  897. } else if (msg->bio) {
  898. struct bio_vec *bv;
  899. bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
  900. page = bv->bv_page;
  901. bio_offset = bv->bv_offset;
  902. max_write = bv->bv_len;
  903. #endif
  904. } else {
  905. page = zero_page;
  906. }
  907. len = min_t(int, max_write - con->out_msg_pos.page_pos,
  908. total_max_write);
  909. if (do_datacrc && !con->out_msg_pos.did_page_crc) {
  910. void *base;
  911. u32 crc = le32_to_cpu(msg->footer.data_crc);
  912. char *kaddr;
  913. kaddr = kmap(page);
  914. BUG_ON(kaddr == NULL);
  915. base = kaddr + con->out_msg_pos.page_pos + bio_offset;
  916. crc = crc32c(crc, base, len);
  917. kunmap(page);
  918. msg->footer.data_crc = cpu_to_le32(crc);
  919. con->out_msg_pos.did_page_crc = true;
  920. }
  921. ret = ceph_tcp_sendpage(con->sock, page,
  922. con->out_msg_pos.page_pos + bio_offset,
  923. len, 1);
  924. if (ret <= 0)
  925. goto out;
  926. out_msg_pos_next(con, page, len, (size_t) ret, in_trail);
  927. }
  928. dout("write_partial_msg_pages %p msg %p done\n", con, msg);
  929. /* prepare and queue up footer, too */
  930. if (!do_datacrc)
  931. msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
  932. con_out_kvec_reset(con);
  933. prepare_write_message_footer(con);
  934. ret = 1;
  935. out:
  936. return ret;
  937. }
  938. /*
  939. * write some zeros
  940. */
  941. static int write_partial_skip(struct ceph_connection *con)
  942. {
  943. int ret;
  944. while (con->out_skip > 0) {
  945. size_t size = min(con->out_skip, (int) PAGE_CACHE_SIZE);
  946. ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, 1);
  947. if (ret <= 0)
  948. goto out;
  949. con->out_skip -= ret;
  950. }
  951. ret = 1;
  952. out:
  953. return ret;
  954. }
  955. /*
  956. * Prepare to read connection handshake, or an ack.
  957. */
  958. static void prepare_read_banner(struct ceph_connection *con)
  959. {
  960. dout("prepare_read_banner %p\n", con);
  961. con->in_base_pos = 0;
  962. }
  963. static void prepare_read_connect(struct ceph_connection *con)
  964. {
  965. dout("prepare_read_connect %p\n", con);
  966. con->in_base_pos = 0;
  967. }
  968. static void prepare_read_ack(struct ceph_connection *con)
  969. {
  970. dout("prepare_read_ack %p\n", con);
  971. con->in_base_pos = 0;
  972. }
  973. static void prepare_read_tag(struct ceph_connection *con)
  974. {
  975. dout("prepare_read_tag %p\n", con);
  976. con->in_base_pos = 0;
  977. con->in_tag = CEPH_MSGR_TAG_READY;
  978. }
  979. /*
  980. * Prepare to read a message.
  981. */
  982. static int prepare_read_message(struct ceph_connection *con)
  983. {
  984. dout("prepare_read_message %p\n", con);
  985. BUG_ON(con->in_msg != NULL);
  986. con->in_base_pos = 0;
  987. con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
  988. return 0;
  989. }
  990. static int read_partial(struct ceph_connection *con,
  991. int end, int size, void *object)
  992. {
  993. while (con->in_base_pos < end) {
  994. int left = end - con->in_base_pos;
  995. int have = size - left;
  996. int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
  997. if (ret <= 0)
  998. return ret;
  999. con->in_base_pos += ret;
  1000. }
  1001. return 1;
  1002. }
  1003. /*
  1004. * Read all or part of the connect-side handshake on a new connection
  1005. */
  1006. static int read_partial_banner(struct ceph_connection *con)
  1007. {
  1008. int size;
  1009. int end;
  1010. int ret;
  1011. dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
  1012. /* peer's banner */
  1013. size = strlen(CEPH_BANNER);
  1014. end = size;
  1015. ret = read_partial(con, end, size, con->in_banner);
  1016. if (ret <= 0)
  1017. goto out;
  1018. size = sizeof (con->actual_peer_addr);
  1019. end += size;
  1020. ret = read_partial(con, end, size, &con->actual_peer_addr);
  1021. if (ret <= 0)
  1022. goto out;
  1023. size = sizeof (con->peer_addr_for_me);
  1024. end += size;
  1025. ret = read_partial(con, end, size, &con->peer_addr_for_me);
  1026. if (ret <= 0)
  1027. goto out;
  1028. out:
  1029. return ret;
  1030. }
  1031. static int read_partial_connect(struct ceph_connection *con)
  1032. {
  1033. int size;
  1034. int end;
  1035. int ret;
  1036. dout("read_partial_connect %p at %d\n", con, con->in_base_pos);
  1037. size = sizeof (con->in_reply);
  1038. end = size;
  1039. ret = read_partial(con, end, size, &con->in_reply);
  1040. if (ret <= 0)
  1041. goto out;
  1042. size = le32_to_cpu(con->in_reply.authorizer_len);
  1043. end += size;
  1044. ret = read_partial(con, end, size, con->auth_reply_buf);
  1045. if (ret <= 0)
  1046. goto out;
  1047. dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
  1048. con, (int)con->in_reply.tag,
  1049. le32_to_cpu(con->in_reply.connect_seq),
  1050. le32_to_cpu(con->in_reply.global_seq));
  1051. out:
  1052. return ret;
  1053. }
  1054. /*
  1055. * Verify the hello banner looks okay.
  1056. */
  1057. static int verify_hello(struct ceph_connection *con)
  1058. {
  1059. if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
  1060. pr_err("connect to %s got bad banner\n",
  1061. ceph_pr_addr(&con->peer_addr.in_addr));
  1062. con->error_msg = "protocol error, bad banner";
  1063. return -1;
  1064. }
  1065. return 0;
  1066. }
  1067. static bool addr_is_blank(struct sockaddr_storage *ss)
  1068. {
  1069. switch (ss->ss_family) {
  1070. case AF_INET:
  1071. return ((struct sockaddr_in *)ss)->sin_addr.s_addr == 0;
  1072. case AF_INET6:
  1073. return
  1074. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[0] == 0 &&
  1075. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[1] == 0 &&
  1076. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[2] == 0 &&
  1077. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[3] == 0;
  1078. }
  1079. return false;
  1080. }
  1081. static int addr_port(struct sockaddr_storage *ss)
  1082. {
  1083. switch (ss->ss_family) {
  1084. case AF_INET:
  1085. return ntohs(((struct sockaddr_in *)ss)->sin_port);
  1086. case AF_INET6:
  1087. return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
  1088. }
  1089. return 0;
  1090. }
  1091. static void addr_set_port(struct sockaddr_storage *ss, int p)
  1092. {
  1093. switch (ss->ss_family) {
  1094. case AF_INET:
  1095. ((struct sockaddr_in *)ss)->sin_port = htons(p);
  1096. break;
  1097. case AF_INET6:
  1098. ((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
  1099. break;
  1100. }
  1101. }
  1102. /*
  1103. * Unlike other *_pton function semantics, zero indicates success.
  1104. */
  1105. static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
  1106. char delim, const char **ipend)
  1107. {
  1108. struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
  1109. struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
  1110. memset(ss, 0, sizeof(*ss));
  1111. if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) {
  1112. ss->ss_family = AF_INET;
  1113. return 0;
  1114. }
  1115. if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) {
  1116. ss->ss_family = AF_INET6;
  1117. return 0;
  1118. }
  1119. return -EINVAL;
  1120. }
  1121. /*
  1122. * Extract hostname string and resolve using kernel DNS facility.
  1123. */
  1124. #ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER
  1125. static int ceph_dns_resolve_name(const char *name, size_t namelen,
  1126. struct sockaddr_storage *ss, char delim, const char **ipend)
  1127. {
  1128. const char *end, *delim_p;
  1129. char *colon_p, *ip_addr = NULL;
  1130. int ip_len, ret;
  1131. /*
  1132. * The end of the hostname occurs immediately preceding the delimiter or
  1133. * the port marker (':') where the delimiter takes precedence.
  1134. */
  1135. delim_p = memchr(name, delim, namelen);
  1136. colon_p = memchr(name, ':', namelen);
  1137. if (delim_p && colon_p)
  1138. end = delim_p < colon_p ? delim_p : colon_p;
  1139. else if (!delim_p && colon_p)
  1140. end = colon_p;
  1141. else {
  1142. end = delim_p;
  1143. if (!end) /* case: hostname:/ */
  1144. end = name + namelen;
  1145. }
  1146. if (end <= name)
  1147. return -EINVAL;
  1148. /* do dns_resolve upcall */
  1149. ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL);
  1150. if (ip_len > 0)
  1151. ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL);
  1152. else
  1153. ret = -ESRCH;
  1154. kfree(ip_addr);
  1155. *ipend = end;
  1156. pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name,
  1157. ret, ret ? "failed" : ceph_pr_addr(ss));
  1158. return ret;
  1159. }
  1160. #else
  1161. static inline int ceph_dns_resolve_name(const char *name, size_t namelen,
  1162. struct sockaddr_storage *ss, char delim, const char **ipend)
  1163. {
  1164. return -EINVAL;
  1165. }
  1166. #endif
  1167. /*
  1168. * Parse a server name (IP or hostname). If a valid IP address is not found
  1169. * then try to extract a hostname to resolve using userspace DNS upcall.
  1170. */
  1171. static int ceph_parse_server_name(const char *name, size_t namelen,
  1172. struct sockaddr_storage *ss, char delim, const char **ipend)
  1173. {
  1174. int ret;
  1175. ret = ceph_pton(name, namelen, ss, delim, ipend);
  1176. if (ret)
  1177. ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend);
  1178. return ret;
  1179. }
  1180. /*
  1181. * Parse an ip[:port] list into an addr array. Use the default
  1182. * monitor port if a port isn't specified.
  1183. */
  1184. int ceph_parse_ips(const char *c, const char *end,
  1185. struct ceph_entity_addr *addr,
  1186. int max_count, int *count)
  1187. {
  1188. int i, ret = -EINVAL;
  1189. const char *p = c;
  1190. dout("parse_ips on '%.*s'\n", (int)(end-c), c);
  1191. for (i = 0; i < max_count; i++) {
  1192. const char *ipend;
  1193. struct sockaddr_storage *ss = &addr[i].in_addr;
  1194. int port;
  1195. char delim = ',';
  1196. if (*p == '[') {
  1197. delim = ']';
  1198. p++;
  1199. }
  1200. ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
  1201. if (ret)
  1202. goto bad;
  1203. ret = -EINVAL;
  1204. p = ipend;
  1205. if (delim == ']') {
  1206. if (*p != ']') {
  1207. dout("missing matching ']'\n");
  1208. goto bad;
  1209. }
  1210. p++;
  1211. }
  1212. /* port? */
  1213. if (p < end && *p == ':') {
  1214. port = 0;
  1215. p++;
  1216. while (p < end && *p >= '0' && *p <= '9') {
  1217. port = (port * 10) + (*p - '0');
  1218. p++;
  1219. }
  1220. if (port > 65535 || port == 0)
  1221. goto bad;
  1222. } else {
  1223. port = CEPH_MON_PORT;
  1224. }
  1225. addr_set_port(ss, port);
  1226. dout("parse_ips got %s\n", ceph_pr_addr(ss));
  1227. if (p == end)
  1228. break;
  1229. if (*p != ',')
  1230. goto bad;
  1231. p++;
  1232. }
  1233. if (p != end)
  1234. goto bad;
  1235. if (count)
  1236. *count = i + 1;
  1237. return 0;
  1238. bad:
  1239. pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
  1240. return ret;
  1241. }
  1242. EXPORT_SYMBOL(ceph_parse_ips);
  1243. static int process_banner(struct ceph_connection *con)
  1244. {
  1245. dout("process_banner on %p\n", con);
  1246. if (verify_hello(con) < 0)
  1247. return -1;
  1248. ceph_decode_addr(&con->actual_peer_addr);
  1249. ceph_decode_addr(&con->peer_addr_for_me);
  1250. /*
  1251. * Make sure the other end is who we wanted. note that the other
  1252. * end may not yet know their ip address, so if it's 0.0.0.0, give
  1253. * them the benefit of the doubt.
  1254. */
  1255. if (memcmp(&con->peer_addr, &con->actual_peer_addr,
  1256. sizeof(con->peer_addr)) != 0 &&
  1257. !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
  1258. con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
  1259. pr_warning("wrong peer, want %s/%d, got %s/%d\n",
  1260. ceph_pr_addr(&con->peer_addr.in_addr),
  1261. (int)le32_to_cpu(con->peer_addr.nonce),
  1262. ceph_pr_addr(&con->actual_peer_addr.in_addr),
  1263. (int)le32_to_cpu(con->actual_peer_addr.nonce));
  1264. con->error_msg = "wrong peer at address";
  1265. return -1;
  1266. }
  1267. /*
  1268. * did we learn our address?
  1269. */
  1270. if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
  1271. int port = addr_port(&con->msgr->inst.addr.in_addr);
  1272. memcpy(&con->msgr->inst.addr.in_addr,
  1273. &con->peer_addr_for_me.in_addr,
  1274. sizeof(con->peer_addr_for_me.in_addr));
  1275. addr_set_port(&con->msgr->inst.addr.in_addr, port);
  1276. encode_my_addr(con->msgr);
  1277. dout("process_banner learned my addr is %s\n",
  1278. ceph_pr_addr(&con->msgr->inst.addr.in_addr));
  1279. }
  1280. return 0;
  1281. }
  1282. static void fail_protocol(struct ceph_connection *con)
  1283. {
  1284. reset_connection(con);
  1285. BUG_ON(con->state != CON_STATE_NEGOTIATING);
  1286. con->state = CON_STATE_CLOSED;
  1287. }
  1288. static int process_connect(struct ceph_connection *con)
  1289. {
  1290. u64 sup_feat = con->msgr->supported_features;
  1291. u64 req_feat = con->msgr->required_features;
  1292. u64 server_feat = le64_to_cpu(con->in_reply.features);
  1293. int ret;
  1294. dout("process_connect on %p tag %d\n", con, (int)con->in_tag);
  1295. switch (con->in_reply.tag) {
  1296. case CEPH_MSGR_TAG_FEATURES:
  1297. pr_err("%s%lld %s feature set mismatch,"
  1298. " my %llx < server's %llx, missing %llx\n",
  1299. ENTITY_NAME(con->peer_name),
  1300. ceph_pr_addr(&con->peer_addr.in_addr),
  1301. sup_feat, server_feat, server_feat & ~sup_feat);
  1302. con->error_msg = "missing required protocol features";
  1303. fail_protocol(con);
  1304. return -1;
  1305. case CEPH_MSGR_TAG_BADPROTOVER:
  1306. pr_err("%s%lld %s protocol version mismatch,"
  1307. " my %d != server's %d\n",
  1308. ENTITY_NAME(con->peer_name),
  1309. ceph_pr_addr(&con->peer_addr.in_addr),
  1310. le32_to_cpu(con->out_connect.protocol_version),
  1311. le32_to_cpu(con->in_reply.protocol_version));
  1312. con->error_msg = "protocol version mismatch";
  1313. fail_protocol(con);
  1314. return -1;
  1315. case CEPH_MSGR_TAG_BADAUTHORIZER:
  1316. con->auth_retry++;
  1317. dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
  1318. con->auth_retry);
  1319. if (con->auth_retry == 2) {
  1320. con->error_msg = "connect authorization failure";
  1321. return -1;
  1322. }
  1323. con->auth_retry = 1;
  1324. con_out_kvec_reset(con);
  1325. ret = prepare_write_connect(con);
  1326. if (ret < 0)
  1327. return ret;
  1328. prepare_read_connect(con);
  1329. break;
  1330. case CEPH_MSGR_TAG_RESETSESSION:
  1331. /*
  1332. * If we connected with a large connect_seq but the peer
  1333. * has no record of a session with us (no connection, or
  1334. * connect_seq == 0), they will send RESETSESION to indicate
  1335. * that they must have reset their session, and may have
  1336. * dropped messages.
  1337. */
  1338. dout("process_connect got RESET peer seq %u\n",
  1339. le32_to_cpu(con->in_reply.connect_seq));
  1340. pr_err("%s%lld %s connection reset\n",
  1341. ENTITY_NAME(con->peer_name),
  1342. ceph_pr_addr(&con->peer_addr.in_addr));
  1343. reset_connection(con);
  1344. con_out_kvec_reset(con);
  1345. ret = prepare_write_connect(con);
  1346. if (ret < 0)
  1347. return ret;
  1348. prepare_read_connect(con);
  1349. /* Tell ceph about it. */
  1350. mutex_unlock(&con->mutex);
  1351. pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
  1352. if (con->ops->peer_reset)
  1353. con->ops->peer_reset(con);
  1354. mutex_lock(&con->mutex);
  1355. if (con->state != CON_STATE_NEGOTIATING)
  1356. return -EAGAIN;
  1357. break;
  1358. case CEPH_MSGR_TAG_RETRY_SESSION:
  1359. /*
  1360. * If we sent a smaller connect_seq than the peer has, try
  1361. * again with a larger value.
  1362. */
  1363. dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
  1364. le32_to_cpu(con->out_connect.connect_seq),
  1365. le32_to_cpu(con->in_reply.connect_seq));
  1366. con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
  1367. con_out_kvec_reset(con);
  1368. ret = prepare_write_connect(con);
  1369. if (ret < 0)
  1370. return ret;
  1371. prepare_read_connect(con);
  1372. break;
  1373. case CEPH_MSGR_TAG_RETRY_GLOBAL:
  1374. /*
  1375. * If we sent a smaller global_seq than the peer has, try
  1376. * again with a larger value.
  1377. */
  1378. dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
  1379. con->peer_global_seq,
  1380. le32_to_cpu(con->in_reply.global_seq));
  1381. get_global_seq(con->msgr,
  1382. le32_to_cpu(con->in_reply.global_seq));
  1383. con_out_kvec_reset(con);
  1384. ret = prepare_write_connect(con);
  1385. if (ret < 0)
  1386. return ret;
  1387. prepare_read_connect(con);
  1388. break;
  1389. case CEPH_MSGR_TAG_READY:
  1390. if (req_feat & ~server_feat) {
  1391. pr_err("%s%lld %s protocol feature mismatch,"
  1392. " my required %llx > server's %llx, need %llx\n",
  1393. ENTITY_NAME(con->peer_name),
  1394. ceph_pr_addr(&con->peer_addr.in_addr),
  1395. req_feat, server_feat, req_feat & ~server_feat);
  1396. con->error_msg = "missing required protocol features";
  1397. fail_protocol(con);
  1398. return -1;
  1399. }
  1400. BUG_ON(con->state != CON_STATE_NEGOTIATING);
  1401. con->state = CON_STATE_OPEN;
  1402. con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
  1403. con->connect_seq++;
  1404. con->peer_features = server_feat;
  1405. dout("process_connect got READY gseq %d cseq %d (%d)\n",
  1406. con->peer_global_seq,
  1407. le32_to_cpu(con->in_reply.connect_seq),
  1408. con->connect_seq);
  1409. WARN_ON(con->connect_seq !=
  1410. le32_to_cpu(con->in_reply.connect_seq));
  1411. if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
  1412. set_bit(CON_FLAG_LOSSYTX, &con->flags);
  1413. con->delay = 0; /* reset backoff memory */
  1414. prepare_read_tag(con);
  1415. break;
  1416. case CEPH_MSGR_TAG_WAIT:
  1417. /*
  1418. * If there is a connection race (we are opening
  1419. * connections to each other), one of us may just have
  1420. * to WAIT. This shouldn't happen if we are the
  1421. * client.
  1422. */
  1423. pr_err("process_connect got WAIT as client\n");
  1424. con->error_msg = "protocol error, got WAIT as client";
  1425. return -1;
  1426. default:
  1427. pr_err("connect protocol error, will retry\n");
  1428. con->error_msg = "protocol error, garbage tag during connect";
  1429. return -1;
  1430. }
  1431. return 0;
  1432. }
  1433. /*
  1434. * read (part of) an ack
  1435. */
  1436. static int read_partial_ack(struct ceph_connection *con)
  1437. {
  1438. int size = sizeof (con->in_temp_ack);
  1439. int end = size;
  1440. return read_partial(con, end, size, &con->in_temp_ack);
  1441. }
  1442. /*
  1443. * We can finally discard anything that's been acked.
  1444. */
  1445. static void process_ack(struct ceph_connection *con)
  1446. {
  1447. struct ceph_msg *m;
  1448. u64 ack = le64_to_cpu(con->in_temp_ack);
  1449. u64 seq;
  1450. while (!list_empty(&con->out_sent)) {
  1451. m = list_first_entry(&con->out_sent, struct ceph_msg,
  1452. list_head);
  1453. seq = le64_to_cpu(m->hdr.seq);
  1454. if (seq > ack)
  1455. break;
  1456. dout("got ack for seq %llu type %d at %p\n", seq,
  1457. le16_to_cpu(m->hdr.type), m);
  1458. m->ack_stamp = jiffies;
  1459. ceph_msg_remove(m);
  1460. }
  1461. prepare_read_tag(con);
  1462. }
  1463. static int read_partial_message_section(struct ceph_connection *con,
  1464. struct kvec *section,
  1465. unsigned int sec_len, u32 *crc)
  1466. {
  1467. int ret, left;
  1468. BUG_ON(!section);
  1469. while (section->iov_len < sec_len) {
  1470. BUG_ON(section->iov_base == NULL);
  1471. left = sec_len - section->iov_len;
  1472. ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
  1473. section->iov_len, left);
  1474. if (ret <= 0)
  1475. return ret;
  1476. section->iov_len += ret;
  1477. }
  1478. if (section->iov_len == sec_len)
  1479. *crc = crc32c(0, section->iov_base, section->iov_len);
  1480. return 1;
  1481. }
  1482. static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);
  1483. static int read_partial_message_pages(struct ceph_connection *con,
  1484. struct page **pages,
  1485. unsigned int data_len, bool do_datacrc)
  1486. {
  1487. void *p;
  1488. int ret;
  1489. int left;
  1490. left = min((int)(data_len - con->in_msg_pos.data_pos),
  1491. (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
  1492. /* (page) data */
  1493. BUG_ON(pages == NULL);
  1494. p = kmap(pages[con->in_msg_pos.page]);
  1495. ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
  1496. left);
  1497. if (ret > 0 && do_datacrc)
  1498. con->in_data_crc =
  1499. crc32c(con->in_data_crc,
  1500. p + con->in_msg_pos.page_pos, ret);
  1501. kunmap(pages[con->in_msg_pos.page]);
  1502. if (ret <= 0)
  1503. return ret;
  1504. con->in_msg_pos.data_pos += ret;
  1505. con->in_msg_pos.page_pos += ret;
  1506. if (con->in_msg_pos.page_pos == PAGE_SIZE) {
  1507. con->in_msg_pos.page_pos = 0;
  1508. con->in_msg_pos.page++;
  1509. }
  1510. return ret;
  1511. }
  1512. #ifdef CONFIG_BLOCK
  1513. static int read_partial_message_bio(struct ceph_connection *con,
  1514. struct bio **bio_iter, int *bio_seg,
  1515. unsigned int data_len, bool do_datacrc)
  1516. {
  1517. struct bio_vec *bv = bio_iovec_idx(*bio_iter, *bio_seg);
  1518. void *p;
  1519. int ret, left;
  1520. left = min((int)(data_len - con->in_msg_pos.data_pos),
  1521. (int)(bv->bv_len - con->in_msg_pos.page_pos));
  1522. p = kmap(bv->bv_page) + bv->bv_offset;
  1523. ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
  1524. left);
  1525. if (ret > 0 && do_datacrc)
  1526. con->in_data_crc =
  1527. crc32c(con->in_data_crc,
  1528. p + con->in_msg_pos.page_pos, ret);
  1529. kunmap(bv->bv_page);
  1530. if (ret <= 0)
  1531. return ret;
  1532. con->in_msg_pos.data_pos += ret;
  1533. con->in_msg_pos.page_pos += ret;
  1534. if (con->in_msg_pos.page_pos == bv->bv_len) {
  1535. con->in_msg_pos.page_pos = 0;
  1536. iter_bio_next(bio_iter, bio_seg);
  1537. }
  1538. return ret;
  1539. }
  1540. #endif
  1541. /*
  1542. * read (part of) a message.
  1543. */
  1544. static int read_partial_message(struct ceph_connection *con)
  1545. {
  1546. struct ceph_msg *m = con->in_msg;
  1547. int size;
  1548. int end;
  1549. int ret;
  1550. unsigned int front_len, middle_len, data_len;
  1551. bool do_datacrc = !con->msgr->nocrc;
  1552. u64 seq;
  1553. u32 crc;
  1554. dout("read_partial_message con %p msg %p\n", con, m);
  1555. /* header */
  1556. size = sizeof (con->in_hdr);
  1557. end = size;
  1558. ret = read_partial(con, end, size, &con->in_hdr);
  1559. if (ret <= 0)
  1560. return ret;
  1561. crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
  1562. if (cpu_to_le32(crc) != con->in_hdr.crc) {
  1563. pr_err("read_partial_message bad hdr "
  1564. " crc %u != expected %u\n",
  1565. crc, con->in_hdr.crc);
  1566. return -EBADMSG;
  1567. }
  1568. front_len = le32_to_cpu(con->in_hdr.front_len);
  1569. if (front_len > CEPH_MSG_MAX_FRONT_LEN)
  1570. return -EIO;
  1571. middle_len = le32_to_cpu(con->in_hdr.middle_len);
  1572. if (middle_len > CEPH_MSG_MAX_DATA_LEN)
  1573. return -EIO;
  1574. data_len = le32_to_cpu(con->in_hdr.data_len);
  1575. if (data_len > CEPH_MSG_MAX_DATA_LEN)
  1576. return -EIO;
  1577. /* verify seq# */
  1578. seq = le64_to_cpu(con->in_hdr.seq);
  1579. if ((s64)seq - (s64)con->in_seq < 1) {
  1580. pr_info("skipping %s%lld %s seq %lld expected %lld\n",
  1581. ENTITY_NAME(con->peer_name),
  1582. ceph_pr_addr(&con->peer_addr.in_addr),
  1583. seq, con->in_seq + 1);
  1584. con->in_base_pos = -front_len - middle_len - data_len -
  1585. sizeof(m->footer);
  1586. con->in_tag = CEPH_MSGR_TAG_READY;
  1587. return 0;
  1588. } else if ((s64)seq - (s64)con->in_seq > 1) {
  1589. pr_err("read_partial_message bad seq %lld expected %lld\n",
  1590. seq, con->in_seq + 1);
  1591. con->error_msg = "bad message sequence # for incoming message";
  1592. return -EBADMSG;
  1593. }
  1594. /* allocate message? */
  1595. if (!con->in_msg) {
  1596. int skip = 0;
  1597. dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
  1598. con->in_hdr.front_len, con->in_hdr.data_len);
  1599. ret = ceph_con_in_msg_alloc(con, &skip);
  1600. if (ret < 0)
  1601. return ret;
  1602. if (skip) {
  1603. /* skip this message */
  1604. dout("alloc_msg said skip message\n");
  1605. BUG_ON(con->in_msg);
  1606. con->in_base_pos = -front_len - middle_len - data_len -
  1607. sizeof(m->footer);
  1608. con->in_tag = CEPH_MSGR_TAG_READY;
  1609. con->in_seq++;
  1610. return 0;
  1611. }
  1612. BUG_ON(!con->in_msg);
  1613. BUG_ON(con->in_msg->con != con);
  1614. m = con->in_msg;
  1615. m->front.iov_len = 0; /* haven't read it yet */
  1616. if (m->middle)
  1617. m->middle->vec.iov_len = 0;
  1618. con->in_msg_pos.page = 0;
  1619. if (m->pages)
  1620. con->in_msg_pos.page_pos = m->page_alignment;
  1621. else
  1622. con->in_msg_pos.page_pos = 0;
  1623. con->in_msg_pos.data_pos = 0;
  1624. #ifdef CONFIG_BLOCK
  1625. if (m->bio)
  1626. init_bio_iter(m->bio, &m->bio_iter, &m->bio_seg);
  1627. #endif
  1628. }
  1629. /* front */
  1630. ret = read_partial_message_section(con, &m->front, front_len,
  1631. &con->in_front_crc);
  1632. if (ret <= 0)
  1633. return ret;
  1634. /* middle */
  1635. if (m->middle) {
  1636. ret = read_partial_message_section(con, &m->middle->vec,
  1637. middle_len,
  1638. &con->in_middle_crc);
  1639. if (ret <= 0)
  1640. return ret;
  1641. }
  1642. /* (page) data */
  1643. while (con->in_msg_pos.data_pos < data_len) {
  1644. if (m->pages) {
  1645. ret = read_partial_message_pages(con, m->pages,
  1646. data_len, do_datacrc);
  1647. if (ret <= 0)
  1648. return ret;
  1649. #ifdef CONFIG_BLOCK
  1650. } else if (m->bio) {
  1651. BUG_ON(!m->bio_iter);
  1652. ret = read_partial_message_bio(con,
  1653. &m->bio_iter, &m->bio_seg,
  1654. data_len, do_datacrc);
  1655. if (ret <= 0)
  1656. return ret;
  1657. #endif
  1658. } else {
  1659. BUG_ON(1);
  1660. }
  1661. }
  1662. /* footer */
  1663. size = sizeof (m->footer);
  1664. end += size;
  1665. ret = read_partial(con, end, size, &m->footer);
  1666. if (ret <= 0)
  1667. return ret;
  1668. dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
  1669. m, front_len, m->footer.front_crc, middle_len,
  1670. m->footer.middle_crc, data_len, m->footer.data_crc);
  1671. /* crc ok? */
  1672. if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
  1673. pr_err("read_partial_message %p front crc %u != exp. %u\n",
  1674. m, con->in_front_crc, m->footer.front_crc);
  1675. return -EBADMSG;
  1676. }
  1677. if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
  1678. pr_err("read_partial_message %p middle crc %u != exp %u\n",
  1679. m, con->in_middle_crc, m->footer.middle_crc);
  1680. return -EBADMSG;
  1681. }
  1682. if (do_datacrc &&
  1683. (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
  1684. con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
  1685. pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
  1686. con->in_data_crc, le32_to_cpu(m->footer.data_crc));
  1687. return -EBADMSG;
  1688. }
  1689. return 1; /* done! */
  1690. }
  1691. /*
  1692. * Process message. This happens in the worker thread. The callback should
  1693. * be careful not to do anything that waits on other incoming messages or it
  1694. * may deadlock.
  1695. */
  1696. static void process_message(struct ceph_connection *con)
  1697. {
  1698. struct ceph_msg *msg;
  1699. BUG_ON(con->in_msg->con != con);
  1700. con->in_msg->con = NULL;
  1701. msg = con->in_msg;
  1702. con->in_msg = NULL;
  1703. con->ops->put(con);
  1704. /* if first message, set peer_name */
  1705. if (con->peer_name.type == 0)
  1706. con->peer_name = msg->hdr.src;
  1707. con->in_seq++;
  1708. mutex_unlock(&con->mutex);
  1709. dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
  1710. msg, le64_to_cpu(msg->hdr.seq),
  1711. ENTITY_NAME(msg->hdr.src),
  1712. le16_to_cpu(msg->hdr.type),
  1713. ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
  1714. le32_to_cpu(msg->hdr.front_len),
  1715. le32_to_cpu(msg->hdr.data_len),
  1716. con->in_front_crc, con->in_middle_crc, con->in_data_crc);
  1717. con->ops->dispatch(con, msg);
  1718. mutex_lock(&con->mutex);
  1719. }
  1720. /*
  1721. * Write something to the socket. Called in a worker thread when the
  1722. * socket appears to be writeable and we have something ready to send.
  1723. */
  1724. static int try_write(struct ceph_connection *con)
  1725. {
  1726. int ret = 1;
  1727. dout("try_write start %p state %lu\n", con, con->state);
  1728. more:
  1729. dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
  1730. /* open the socket first? */
  1731. if (con->state == CON_STATE_PREOPEN) {
  1732. BUG_ON(con->sock);
  1733. con->state = CON_STATE_CONNECTING;
  1734. con_out_kvec_reset(con);
  1735. prepare_write_banner(con);
  1736. prepare_read_banner(con);
  1737. BUG_ON(con->in_msg);
  1738. con->in_tag = CEPH_MSGR_TAG_READY;
  1739. dout("try_write initiating connect on %p new state %lu\n",
  1740. con, con->state);
  1741. ret = ceph_tcp_connect(con);
  1742. if (ret < 0) {
  1743. con->error_msg = "connect error";
  1744. goto out;
  1745. }
  1746. }
  1747. more_kvec:
  1748. /* kvec data queued? */
  1749. if (con->out_skip) {
  1750. ret = write_partial_skip(con);
  1751. if (ret <= 0)
  1752. goto out;
  1753. }
  1754. if (con->out_kvec_left) {
  1755. ret = write_partial_kvec(con);
  1756. if (ret <= 0)
  1757. goto out;
  1758. }
  1759. /* msg pages? */
  1760. if (con->out_msg) {
  1761. if (con->out_msg_done) {
  1762. ceph_msg_put(con->out_msg);
  1763. con->out_msg = NULL; /* we're done with this one */
  1764. goto do_next;
  1765. }
  1766. ret = write_partial_msg_pages(con);
  1767. if (ret == 1)
  1768. goto more_kvec; /* we need to send the footer, too! */
  1769. if (ret == 0)
  1770. goto out;
  1771. if (ret < 0) {
  1772. dout("try_write write_partial_msg_pages err %d\n",
  1773. ret);
  1774. goto out;
  1775. }
  1776. }
  1777. do_next:
  1778. if (con->state == CON_STATE_OPEN) {
  1779. /* is anything else pending? */
  1780. if (!list_empty(&con->out_queue)) {
  1781. prepare_write_message(con);
  1782. goto more;
  1783. }
  1784. if (con->in_seq > con->in_seq_acked) {
  1785. prepare_write_ack(con);
  1786. goto more;
  1787. }
  1788. if (test_and_clear_bit(CON_FLAG_KEEPALIVE_PENDING,
  1789. &con->flags)) {
  1790. prepare_write_keepalive(con);
  1791. goto more;
  1792. }
  1793. }
  1794. /* Nothing to do! */
  1795. clear_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  1796. dout("try_write nothing else to write.\n");
  1797. ret = 0;
  1798. out:
  1799. dout("try_write done on %p ret %d\n", con, ret);
  1800. return ret;
  1801. }
  1802. /*
  1803. * Read what we can from the socket.
  1804. */
  1805. static int try_read(struct ceph_connection *con)
  1806. {
  1807. int ret = -1;
  1808. more:
  1809. dout("try_read start on %p state %lu\n", con, con->state);
  1810. if (con->state != CON_STATE_CONNECTING &&
  1811. con->state != CON_STATE_NEGOTIATING &&
  1812. con->state != CON_STATE_OPEN)
  1813. return 0;
  1814. BUG_ON(!con->sock);
  1815. dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
  1816. con->in_base_pos);
  1817. if (con->state == CON_STATE_CONNECTING) {
  1818. dout("try_read connecting\n");
  1819. ret = read_partial_banner(con);
  1820. if (ret <= 0)
  1821. goto out;
  1822. ret = process_banner(con);
  1823. if (ret < 0)
  1824. goto out;
  1825. BUG_ON(con->state != CON_STATE_CONNECTING);
  1826. con->state = CON_STATE_NEGOTIATING;
  1827. /*
  1828. * Received banner is good, exchange connection info.
  1829. * Do not reset out_kvec, as sending our banner raced
  1830. * with receiving peer banner after connect completed.
  1831. */
  1832. ret = prepare_write_connect(con);
  1833. if (ret < 0)
  1834. goto out;
  1835. prepare_read_connect(con);
  1836. /* Send connection info before awaiting response */
  1837. goto out;
  1838. }
  1839. if (con->state == CON_STATE_NEGOTIATING) {
  1840. dout("try_read negotiating\n");
  1841. ret = read_partial_connect(con);
  1842. if (ret <= 0)
  1843. goto out;
  1844. ret = process_connect(con);
  1845. if (ret < 0)
  1846. goto out;
  1847. goto more;
  1848. }
  1849. BUG_ON(con->state != CON_STATE_OPEN);
  1850. if (con->in_base_pos < 0) {
  1851. /*
  1852. * skipping + discarding content.
  1853. *
  1854. * FIXME: there must be a better way to do this!
  1855. */
  1856. static char buf[SKIP_BUF_SIZE];
  1857. int skip = min((int) sizeof (buf), -con->in_base_pos);
  1858. dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
  1859. ret = ceph_tcp_recvmsg(con->sock, buf, skip);
  1860. if (ret <= 0)
  1861. goto out;
  1862. con->in_base_pos += ret;
  1863. if (con->in_base_pos)
  1864. goto more;
  1865. }
  1866. if (con->in_tag == CEPH_MSGR_TAG_READY) {
  1867. /*
  1868. * what's next?
  1869. */
  1870. ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
  1871. if (ret <= 0)
  1872. goto out;
  1873. dout("try_read got tag %d\n", (int)con->in_tag);
  1874. switch (con->in_tag) {
  1875. case CEPH_MSGR_TAG_MSG:
  1876. prepare_read_message(con);
  1877. break;
  1878. case CEPH_MSGR_TAG_ACK:
  1879. prepare_read_ack(con);
  1880. break;
  1881. case CEPH_MSGR_TAG_CLOSE:
  1882. con_close_socket(con);
  1883. con->state = CON_STATE_CLOSED;
  1884. goto out;
  1885. default:
  1886. goto bad_tag;
  1887. }
  1888. }
  1889. if (con->in_tag == CEPH_MSGR_TAG_MSG) {
  1890. ret = read_partial_message(con);
  1891. if (ret <= 0) {
  1892. switch (ret) {
  1893. case -EBADMSG:
  1894. con->error_msg = "bad crc";
  1895. ret = -EIO;
  1896. break;
  1897. case -EIO:
  1898. con->error_msg = "io error";
  1899. break;
  1900. }
  1901. goto out;
  1902. }
  1903. if (con->in_tag == CEPH_MSGR_TAG_READY)
  1904. goto more;
  1905. process_message(con);
  1906. if (con->state == CON_STATE_OPEN)
  1907. prepare_read_tag(con);
  1908. goto more;
  1909. }
  1910. if (con->in_tag == CEPH_MSGR_TAG_ACK) {
  1911. ret = read_partial_ack(con);
  1912. if (ret <= 0)
  1913. goto out;
  1914. process_ack(con);
  1915. goto more;
  1916. }
  1917. out:
  1918. dout("try_read done on %p ret %d\n", con, ret);
  1919. return ret;
  1920. bad_tag:
  1921. pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
  1922. con->error_msg = "protocol error, garbage tag";
  1923. ret = -1;
  1924. goto out;
  1925. }
  1926. /*
  1927. * Atomically queue work on a connection. Bump @con reference to
  1928. * avoid races with connection teardown.
  1929. */
  1930. static void queue_con(struct ceph_connection *con)
  1931. {
  1932. if (!con->ops->get(con)) {
  1933. dout("queue_con %p ref count 0\n", con);
  1934. return;
  1935. }
  1936. if (!queue_delayed_work(ceph_msgr_wq, &con->work, 0)) {
  1937. dout("queue_con %p - already queued\n", con);
  1938. con->ops->put(con);
  1939. } else {
  1940. dout("queue_con %p\n", con);
  1941. }
  1942. }
  1943. /*
  1944. * Do some work on a connection. Drop a connection ref when we're done.
  1945. */
  1946. static void con_work(struct work_struct *work)
  1947. {
  1948. struct ceph_connection *con = container_of(work, struct ceph_connection,
  1949. work.work);
  1950. int ret;
  1951. mutex_lock(&con->mutex);
  1952. restart:
  1953. if (test_and_clear_bit(CON_FLAG_SOCK_CLOSED, &con->flags)) {
  1954. switch (con->state) {
  1955. case CON_STATE_CONNECTING:
  1956. con->error_msg = "connection failed";
  1957. break;
  1958. case CON_STATE_NEGOTIATING:
  1959. con->error_msg = "negotiation failed";
  1960. break;
  1961. case CON_STATE_OPEN:
  1962. con->error_msg = "socket closed";
  1963. break;
  1964. default:
  1965. dout("unrecognized con state %d\n", (int)con->state);
  1966. con->error_msg = "unrecognized con state";
  1967. BUG();
  1968. }
  1969. goto fault;
  1970. }
  1971. if (test_and_clear_bit(CON_FLAG_BACKOFF, &con->flags)) {
  1972. dout("con_work %p backing off\n", con);
  1973. if (queue_delayed_work(ceph_msgr_wq, &con->work,
  1974. round_jiffies_relative(con->delay))) {
  1975. dout("con_work %p backoff %lu\n", con, con->delay);
  1976. mutex_unlock(&con->mutex);
  1977. return;
  1978. } else {
  1979. con->ops->put(con);
  1980. dout("con_work %p FAILED to back off %lu\n", con,
  1981. con->delay);
  1982. }
  1983. }
  1984. if (con->state == CON_STATE_STANDBY) {
  1985. dout("con_work %p STANDBY\n", con);
  1986. goto done;
  1987. }
  1988. if (con->state == CON_STATE_CLOSED) {
  1989. dout("con_work %p CLOSED\n", con);
  1990. BUG_ON(con->sock);
  1991. goto done;
  1992. }
  1993. if (con->state == CON_STATE_PREOPEN) {
  1994. dout("con_work OPENING\n");
  1995. BUG_ON(con->sock);
  1996. }
  1997. ret = try_read(con);
  1998. if (ret == -EAGAIN)
  1999. goto restart;
  2000. if (ret < 0) {
  2001. con->error_msg = "socket error on read";
  2002. goto fault;
  2003. }
  2004. ret = try_write(con);
  2005. if (ret == -EAGAIN)
  2006. goto restart;
  2007. if (ret < 0) {
  2008. con->error_msg = "socket error on write";
  2009. goto fault;
  2010. }
  2011. done:
  2012. mutex_unlock(&con->mutex);
  2013. done_unlocked:
  2014. con->ops->put(con);
  2015. return;
  2016. fault:
  2017. ceph_fault(con); /* error/fault path */
  2018. goto done_unlocked;
  2019. }
  2020. /*
  2021. * Generic error/fault handler. A retry mechanism is used with
  2022. * exponential backoff
  2023. */
  2024. static void ceph_fault(struct ceph_connection *con)
  2025. __releases(con->mutex)
  2026. {
  2027. pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
  2028. ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
  2029. dout("fault %p state %lu to peer %s\n",
  2030. con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
  2031. BUG_ON(con->state != CON_STATE_CONNECTING &&
  2032. con->state != CON_STATE_NEGOTIATING &&
  2033. con->state != CON_STATE_OPEN);
  2034. con_close_socket(con);
  2035. if (test_bit(CON_FLAG_LOSSYTX, &con->flags)) {
  2036. dout("fault on LOSSYTX channel, marking CLOSED\n");
  2037. con->state = CON_STATE_CLOSED;
  2038. goto out_unlock;
  2039. }
  2040. if (con->in_msg) {
  2041. BUG_ON(con->in_msg->con != con);
  2042. con->in_msg->con = NULL;
  2043. ceph_msg_put(con->in_msg);
  2044. con->in_msg = NULL;
  2045. con->ops->put(con);
  2046. }
  2047. /* Requeue anything that hasn't been acked */
  2048. list_splice_init(&con->out_sent, &con->out_queue);
  2049. /* If there are no messages queued or keepalive pending, place
  2050. * the connection in a STANDBY state */
  2051. if (list_empty(&con->out_queue) &&
  2052. !test_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags)) {
  2053. dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
  2054. clear_bit(CON_FLAG_WRITE_PENDING, &con->flags);
  2055. con->state = CON_STATE_STANDBY;
  2056. } else {
  2057. /* retry after a delay. */
  2058. con->state = CON_STATE_PREOPEN;
  2059. if (con->delay == 0)
  2060. con->delay = BASE_DELAY_INTERVAL;
  2061. else if (con->delay < MAX_DELAY_INTERVAL)
  2062. con->delay *= 2;
  2063. con->ops->get(con);
  2064. if (queue_delayed_work(ceph_msgr_wq, &con->work,
  2065. round_jiffies_relative(con->delay))) {
  2066. dout("fault queued %p delay %lu\n", con, con->delay);
  2067. } else {
  2068. con->ops->put(con);
  2069. dout("fault failed to queue %p delay %lu, backoff\n",
  2070. con, con->delay);
  2071. /*
  2072. * In many cases we see a socket state change
  2073. * while con_work is running and end up
  2074. * queuing (non-delayed) work, such that we
  2075. * can't backoff with a delay. Set a flag so
  2076. * that when con_work restarts we schedule the
  2077. * delay then.
  2078. */
  2079. set_bit(CON_FLAG_BACKOFF, &con->flags);
  2080. }
  2081. }
  2082. out_unlock:
  2083. mutex_unlock(&con->mutex);
  2084. /*
  2085. * in case we faulted due to authentication, invalidate our
  2086. * current tickets so that we can get new ones.
  2087. */
  2088. if (con->auth_retry && con->ops->invalidate_authorizer) {
  2089. dout("calling invalidate_authorizer()\n");
  2090. con->ops->invalidate_authorizer(con);
  2091. }
  2092. if (con->ops->fault)
  2093. con->ops->fault(con);
  2094. }
  2095. /*
  2096. * initialize a new messenger instance
  2097. */
  2098. void ceph_messenger_init(struct ceph_messenger *msgr,
  2099. struct ceph_entity_addr *myaddr,
  2100. u32 supported_features,
  2101. u32 required_features,
  2102. bool nocrc)
  2103. {
  2104. msgr->supported_features = supported_features;
  2105. msgr->required_features = required_features;
  2106. spin_lock_init(&msgr->global_seq_lock);
  2107. if (myaddr)
  2108. msgr->inst.addr = *myaddr;
  2109. /* select a random nonce */
  2110. msgr->inst.addr.type = 0;
  2111. get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
  2112. encode_my_addr(msgr);
  2113. msgr->nocrc = nocrc;
  2114. atomic_set(&msgr->stopping, 0);
  2115. dout("%s %p\n", __func__, msgr);
  2116. }
  2117. EXPORT_SYMBOL(ceph_messenger_init);
  2118. static void clear_standby(struct ceph_connection *con)
  2119. {
  2120. /* come back from STANDBY? */
  2121. if (con->state == CON_STATE_STANDBY) {
  2122. dout("clear_standby %p and ++connect_seq\n", con);
  2123. con->state = CON_STATE_PREOPEN;
  2124. con->connect_seq++;
  2125. WARN_ON(test_bit(CON_FLAG_WRITE_PENDING, &con->flags));
  2126. WARN_ON(test_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags));
  2127. }
  2128. }
  2129. /*
  2130. * Queue up an outgoing message on the given connection.
  2131. */
  2132. void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
  2133. {
  2134. /* set src+dst */
  2135. msg->hdr.src = con->msgr->inst.name;
  2136. BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
  2137. msg->needs_out_seq = true;
  2138. mutex_lock(&con->mutex);
  2139. if (con->state == CON_STATE_CLOSED) {
  2140. dout("con_send %p closed, dropping %p\n", con, msg);
  2141. ceph_msg_put(msg);
  2142. mutex_unlock(&con->mutex);
  2143. return;
  2144. }
  2145. BUG_ON(msg->con != NULL);
  2146. msg->con = con->ops->get(con);
  2147. BUG_ON(msg->con == NULL);
  2148. BUG_ON(!list_empty(&msg->list_head));
  2149. list_add_tail(&msg->list_head, &con->out_queue);
  2150. dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
  2151. ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
  2152. ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
  2153. le32_to_cpu(msg->hdr.front_len),
  2154. le32_to_cpu(msg->hdr.middle_len),
  2155. le32_to_cpu(msg->hdr.data_len));
  2156. clear_standby(con);
  2157. mutex_unlock(&con->mutex);
  2158. /* if there wasn't anything waiting to send before, queue
  2159. * new work */
  2160. if (test_and_set_bit(CON_FLAG_WRITE_PENDING, &con->flags) == 0)
  2161. queue_con(con);
  2162. }
  2163. EXPORT_SYMBOL(ceph_con_send);
  2164. /*
  2165. * Revoke a message that was previously queued for send
  2166. */
  2167. void ceph_msg_revoke(struct ceph_msg *msg)
  2168. {
  2169. struct ceph_connection *con = msg->con;
  2170. if (!con)
  2171. return; /* Message not in our possession */
  2172. mutex_lock(&con->mutex);
  2173. if (!list_empty(&msg->list_head)) {
  2174. dout("%s %p msg %p - was on queue\n", __func__, con, msg);
  2175. list_del_init(&msg->list_head);
  2176. BUG_ON(msg->con == NULL);
  2177. msg->con->ops->put(msg->con);
  2178. msg->con = NULL;
  2179. msg->hdr.seq = 0;
  2180. ceph_msg_put(msg);
  2181. }
  2182. if (con->out_msg == msg) {
  2183. dout("%s %p msg %p - was sending\n", __func__, con, msg);
  2184. con->out_msg = NULL;
  2185. if (con->out_kvec_is_msg) {
  2186. con->out_skip = con->out_kvec_bytes;
  2187. con->out_kvec_is_msg = false;
  2188. }
  2189. msg->hdr.seq = 0;
  2190. ceph_msg_put(msg);
  2191. }
  2192. mutex_unlock(&con->mutex);
  2193. }
  2194. /*
  2195. * Revoke a message that we may be reading data into
  2196. */
  2197. void ceph_msg_revoke_incoming(struct ceph_msg *msg)
  2198. {
  2199. struct ceph_connection *con;
  2200. BUG_ON(msg == NULL);
  2201. if (!msg->con) {
  2202. dout("%s msg %p null con\n", __func__, msg);
  2203. return; /* Message not in our possession */
  2204. }
  2205. con = msg->con;
  2206. mutex_lock(&con->mutex);
  2207. if (con->in_msg == msg) {
  2208. unsigned int front_len = le32_to_cpu(con->in_hdr.front_len);
  2209. unsigned int middle_len = le32_to_cpu(con->in_hdr.middle_len);
  2210. unsigned int data_len = le32_to_cpu(con->in_hdr.data_len);
  2211. /* skip rest of message */
  2212. dout("%s %p msg %p revoked\n", __func__, con, msg);
  2213. con->in_base_pos = con->in_base_pos -
  2214. sizeof(struct ceph_msg_header) -
  2215. front_len -
  2216. middle_len -
  2217. data_len -
  2218. sizeof(struct ceph_msg_footer);
  2219. ceph_msg_put(con->in_msg);
  2220. con->in_msg = NULL;
  2221. con->in_tag = CEPH_MSGR_TAG_READY;
  2222. con->in_seq++;
  2223. } else {
  2224. dout("%s %p in_msg %p msg %p no-op\n",
  2225. __func__, con, con->in_msg, msg);
  2226. }
  2227. mutex_unlock(&con->mutex);
  2228. }
  2229. /*
  2230. * Queue a keepalive byte to ensure the tcp connection is alive.
  2231. */
  2232. void ceph_con_keepalive(struct ceph_connection *con)
  2233. {
  2234. dout("con_keepalive %p\n", con);
  2235. mutex_lock(&con->mutex);
  2236. clear_standby(con);
  2237. mutex_unlock(&con->mutex);
  2238. if (test_and_set_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags) == 0 &&
  2239. test_and_set_bit(CON_FLAG_WRITE_PENDING, &con->flags) == 0)
  2240. queue_con(con);
  2241. }
  2242. EXPORT_SYMBOL(ceph_con_keepalive);
  2243. /*
  2244. * construct a new message with given type, size
  2245. * the new msg has a ref count of 1.
  2246. */
  2247. struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
  2248. bool can_fail)
  2249. {
  2250. struct ceph_msg *m;
  2251. m = kmalloc(sizeof(*m), flags);
  2252. if (m == NULL)
  2253. goto out;
  2254. kref_init(&m->kref);
  2255. m->con = NULL;
  2256. INIT_LIST_HEAD(&m->list_head);
  2257. m->hdr.tid = 0;
  2258. m->hdr.type = cpu_to_le16(type);
  2259. m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
  2260. m->hdr.version = 0;
  2261. m->hdr.front_len = cpu_to_le32(front_len);
  2262. m->hdr.middle_len = 0;
  2263. m->hdr.data_len = 0;
  2264. m->hdr.data_off = 0;
  2265. m->hdr.reserved = 0;
  2266. m->footer.front_crc = 0;
  2267. m->footer.middle_crc = 0;
  2268. m->footer.data_crc = 0;
  2269. m->footer.flags = 0;
  2270. m->front_max = front_len;
  2271. m->front_is_vmalloc = false;
  2272. m->more_to_follow = false;
  2273. m->ack_stamp = 0;
  2274. m->pool = NULL;
  2275. /* middle */
  2276. m->middle = NULL;
  2277. /* data */
  2278. m->nr_pages = 0;
  2279. m->page_alignment = 0;
  2280. m->pages = NULL;
  2281. m->pagelist = NULL;
  2282. m->bio = NULL;
  2283. m->bio_iter = NULL;
  2284. m->bio_seg = 0;
  2285. m->trail = NULL;
  2286. /* front */
  2287. if (front_len) {
  2288. if (front_len > PAGE_CACHE_SIZE) {
  2289. m->front.iov_base = __vmalloc(front_len, flags,
  2290. PAGE_KERNEL);
  2291. m->front_is_vmalloc = true;
  2292. } else {
  2293. m->front.iov_base = kmalloc(front_len, flags);
  2294. }
  2295. if (m->front.iov_base == NULL) {
  2296. dout("ceph_msg_new can't allocate %d bytes\n",
  2297. front_len);
  2298. goto out2;
  2299. }
  2300. } else {
  2301. m->front.iov_base = NULL;
  2302. }
  2303. m->front.iov_len = front_len;
  2304. dout("ceph_msg_new %p front %d\n", m, front_len);
  2305. return m;
  2306. out2:
  2307. ceph_msg_put(m);
  2308. out:
  2309. if (!can_fail) {
  2310. pr_err("msg_new can't create type %d front %d\n", type,
  2311. front_len);
  2312. WARN_ON(1);
  2313. } else {
  2314. dout("msg_new can't create type %d front %d\n", type,
  2315. front_len);
  2316. }
  2317. return NULL;
  2318. }
  2319. EXPORT_SYMBOL(ceph_msg_new);
  2320. /*
  2321. * Allocate "middle" portion of a message, if it is needed and wasn't
  2322. * allocated by alloc_msg. This allows us to read a small fixed-size
  2323. * per-type header in the front and then gracefully fail (i.e.,
  2324. * propagate the error to the caller based on info in the front) when
  2325. * the middle is too large.
  2326. */
  2327. static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
  2328. {
  2329. int type = le16_to_cpu(msg->hdr.type);
  2330. int middle_len = le32_to_cpu(msg->hdr.middle_len);
  2331. dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
  2332. ceph_msg_type_name(type), middle_len);
  2333. BUG_ON(!middle_len);
  2334. BUG_ON(msg->middle);
  2335. msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
  2336. if (!msg->middle)
  2337. return -ENOMEM;
  2338. return 0;
  2339. }
  2340. /*
  2341. * Allocate a message for receiving an incoming message on a
  2342. * connection, and save the result in con->in_msg. Uses the
  2343. * connection's private alloc_msg op if available.
  2344. *
  2345. * Returns 0 on success, or a negative error code.
  2346. *
  2347. * On success, if we set *skip = 1:
  2348. * - the next message should be skipped and ignored.
  2349. * - con->in_msg == NULL
  2350. * or if we set *skip = 0:
  2351. * - con->in_msg is non-null.
  2352. * On error (ENOMEM, EAGAIN, ...),
  2353. * - con->in_msg == NULL
  2354. */
  2355. static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
  2356. {
  2357. struct ceph_msg_header *hdr = &con->in_hdr;
  2358. int type = le16_to_cpu(hdr->type);
  2359. int front_len = le32_to_cpu(hdr->front_len);
  2360. int middle_len = le32_to_cpu(hdr->middle_len);
  2361. int ret = 0;
  2362. BUG_ON(con->in_msg != NULL);
  2363. if (con->ops->alloc_msg) {
  2364. struct ceph_msg *msg;
  2365. mutex_unlock(&con->mutex);
  2366. msg = con->ops->alloc_msg(con, hdr, skip);
  2367. mutex_lock(&con->mutex);
  2368. if (con->state != CON_STATE_OPEN) {
  2369. ceph_msg_put(msg);
  2370. return -EAGAIN;
  2371. }
  2372. con->in_msg = msg;
  2373. if (con->in_msg) {
  2374. con->in_msg->con = con->ops->get(con);
  2375. BUG_ON(con->in_msg->con == NULL);
  2376. }
  2377. if (*skip) {
  2378. con->in_msg = NULL;
  2379. return 0;
  2380. }
  2381. if (!con->in_msg) {
  2382. con->error_msg =
  2383. "error allocating memory for incoming message";
  2384. return -ENOMEM;
  2385. }
  2386. }
  2387. if (!con->in_msg) {
  2388. con->in_msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
  2389. if (!con->in_msg) {
  2390. pr_err("unable to allocate msg type %d len %d\n",
  2391. type, front_len);
  2392. return -ENOMEM;
  2393. }
  2394. con->in_msg->con = con->ops->get(con);
  2395. BUG_ON(con->in_msg->con == NULL);
  2396. con->in_msg->page_alignment = le16_to_cpu(hdr->data_off);
  2397. }
  2398. memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
  2399. if (middle_len && !con->in_msg->middle) {
  2400. ret = ceph_alloc_middle(con, con->in_msg);
  2401. if (ret < 0) {
  2402. ceph_msg_put(con->in_msg);
  2403. con->in_msg = NULL;
  2404. }
  2405. }
  2406. return ret;
  2407. }
  2408. /*
  2409. * Free a generically kmalloc'd message.
  2410. */
  2411. void ceph_msg_kfree(struct ceph_msg *m)
  2412. {
  2413. dout("msg_kfree %p\n", m);
  2414. if (m->front_is_vmalloc)
  2415. vfree(m->front.iov_base);
  2416. else
  2417. kfree(m->front.iov_base);
  2418. kfree(m);
  2419. }
  2420. /*
  2421. * Drop a msg ref. Destroy as needed.
  2422. */
  2423. void ceph_msg_last_put(struct kref *kref)
  2424. {
  2425. struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
  2426. dout("ceph_msg_put last one on %p\n", m);
  2427. WARN_ON(!list_empty(&m->list_head));
  2428. /* drop middle, data, if any */
  2429. if (m->middle) {
  2430. ceph_buffer_put(m->middle);
  2431. m->middle = NULL;
  2432. }
  2433. m->nr_pages = 0;
  2434. m->pages = NULL;
  2435. if (m->pagelist) {
  2436. ceph_pagelist_release(m->pagelist);
  2437. kfree(m->pagelist);
  2438. m->pagelist = NULL;
  2439. }
  2440. m->trail = NULL;
  2441. if (m->pool)
  2442. ceph_msgpool_put(m->pool, m);
  2443. else
  2444. ceph_msg_kfree(m);
  2445. }
  2446. EXPORT_SYMBOL(ceph_msg_last_put);
  2447. void ceph_msg_dump(struct ceph_msg *msg)
  2448. {
  2449. pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
  2450. msg->front_max, msg->nr_pages);
  2451. print_hex_dump(KERN_DEBUG, "header: ",
  2452. DUMP_PREFIX_OFFSET, 16, 1,
  2453. &msg->hdr, sizeof(msg->hdr), true);
  2454. print_hex_dump(KERN_DEBUG, " front: ",
  2455. DUMP_PREFIX_OFFSET, 16, 1,
  2456. msg->front.iov_base, msg->front.iov_len, true);
  2457. if (msg->middle)
  2458. print_hex_dump(KERN_DEBUG, "middle: ",
  2459. DUMP_PREFIX_OFFSET, 16, 1,
  2460. msg->middle->vec.iov_base,
  2461. msg->middle->vec.iov_len, true);
  2462. print_hex_dump(KERN_DEBUG, "footer: ",
  2463. DUMP_PREFIX_OFFSET, 16, 1,
  2464. &msg->footer, sizeof(msg->footer), true);
  2465. }
  2466. EXPORT_SYMBOL(ceph_msg_dump);