messenger.c 57 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265
  1. #include "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 <net/tcp.h>
  12. #include "super.h"
  13. #include "messenger.h"
  14. #include "decode.h"
  15. #include "pagelist.h"
  16. /*
  17. * Ceph uses the messenger to exchange ceph_msg messages with other
  18. * hosts in the system. The messenger provides ordered and reliable
  19. * delivery. We tolerate TCP disconnects by reconnecting (with
  20. * exponential backoff) in the case of a fault (disconnection, bad
  21. * crc, protocol error). Acks allow sent messages to be discarded by
  22. * the sender.
  23. */
  24. /* static tag bytes (protocol control messages) */
  25. static char tag_msg = CEPH_MSGR_TAG_MSG;
  26. static char tag_ack = CEPH_MSGR_TAG_ACK;
  27. static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
  28. #ifdef CONFIG_LOCKDEP
  29. static struct lock_class_key socket_class;
  30. #endif
  31. static void queue_con(struct ceph_connection *con);
  32. static void con_work(struct work_struct *);
  33. static void ceph_fault(struct ceph_connection *con);
  34. /*
  35. * nicely render a sockaddr as a string.
  36. */
  37. #define MAX_ADDR_STR 20
  38. static char addr_str[MAX_ADDR_STR][40];
  39. static DEFINE_SPINLOCK(addr_str_lock);
  40. static int last_addr_str;
  41. const char *pr_addr(const struct sockaddr_storage *ss)
  42. {
  43. int i;
  44. char *s;
  45. struct sockaddr_in *in4 = (void *)ss;
  46. unsigned char *quad = (void *)&in4->sin_addr.s_addr;
  47. struct sockaddr_in6 *in6 = (void *)ss;
  48. spin_lock(&addr_str_lock);
  49. i = last_addr_str++;
  50. if (last_addr_str == MAX_ADDR_STR)
  51. last_addr_str = 0;
  52. spin_unlock(&addr_str_lock);
  53. s = addr_str[i];
  54. switch (ss->ss_family) {
  55. case AF_INET:
  56. sprintf(s, "%u.%u.%u.%u:%u",
  57. (unsigned int)quad[0],
  58. (unsigned int)quad[1],
  59. (unsigned int)quad[2],
  60. (unsigned int)quad[3],
  61. (unsigned int)ntohs(in4->sin_port));
  62. break;
  63. case AF_INET6:
  64. sprintf(s, "%04x:%04x:%04x:%04x:%04x:%04x:%04x:%04x:%u",
  65. in6->sin6_addr.s6_addr16[0],
  66. in6->sin6_addr.s6_addr16[1],
  67. in6->sin6_addr.s6_addr16[2],
  68. in6->sin6_addr.s6_addr16[3],
  69. in6->sin6_addr.s6_addr16[4],
  70. in6->sin6_addr.s6_addr16[5],
  71. in6->sin6_addr.s6_addr16[6],
  72. in6->sin6_addr.s6_addr16[7],
  73. (unsigned int)ntohs(in6->sin6_port));
  74. break;
  75. default:
  76. sprintf(s, "(unknown sockaddr family %d)", (int)ss->ss_family);
  77. }
  78. return s;
  79. }
  80. static void encode_my_addr(struct ceph_messenger *msgr)
  81. {
  82. memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
  83. ceph_encode_addr(&msgr->my_enc_addr);
  84. }
  85. /*
  86. * work queue for all reading and writing to/from the socket.
  87. */
  88. struct workqueue_struct *ceph_msgr_wq;
  89. int __init ceph_msgr_init(void)
  90. {
  91. ceph_msgr_wq = create_workqueue("ceph-msgr");
  92. if (IS_ERR(ceph_msgr_wq)) {
  93. int ret = PTR_ERR(ceph_msgr_wq);
  94. pr_err("msgr_init failed to create workqueue: %d\n", ret);
  95. ceph_msgr_wq = NULL;
  96. return ret;
  97. }
  98. return 0;
  99. }
  100. void ceph_msgr_exit(void)
  101. {
  102. destroy_workqueue(ceph_msgr_wq);
  103. }
  104. /*
  105. * socket callback functions
  106. */
  107. /* data available on socket, or listen socket received a connect */
  108. static void ceph_data_ready(struct sock *sk, int count_unused)
  109. {
  110. struct ceph_connection *con =
  111. (struct ceph_connection *)sk->sk_user_data;
  112. if (sk->sk_state != TCP_CLOSE_WAIT) {
  113. dout("ceph_data_ready on %p state = %lu, queueing work\n",
  114. con, con->state);
  115. queue_con(con);
  116. }
  117. }
  118. /* socket has buffer space for writing */
  119. static void ceph_write_space(struct sock *sk)
  120. {
  121. struct ceph_connection *con =
  122. (struct ceph_connection *)sk->sk_user_data;
  123. /* only queue to workqueue if there is data we want to write. */
  124. if (test_bit(WRITE_PENDING, &con->state)) {
  125. dout("ceph_write_space %p queueing write work\n", con);
  126. queue_con(con);
  127. } else {
  128. dout("ceph_write_space %p nothing to write\n", con);
  129. }
  130. /* since we have our own write_space, clear the SOCK_NOSPACE flag */
  131. clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
  132. }
  133. /* socket's state has changed */
  134. static void ceph_state_change(struct sock *sk)
  135. {
  136. struct ceph_connection *con =
  137. (struct ceph_connection *)sk->sk_user_data;
  138. dout("ceph_state_change %p state = %lu sk_state = %u\n",
  139. con, con->state, sk->sk_state);
  140. if (test_bit(CLOSED, &con->state))
  141. return;
  142. switch (sk->sk_state) {
  143. case TCP_CLOSE:
  144. dout("ceph_state_change TCP_CLOSE\n");
  145. case TCP_CLOSE_WAIT:
  146. dout("ceph_state_change TCP_CLOSE_WAIT\n");
  147. if (test_and_set_bit(SOCK_CLOSED, &con->state) == 0) {
  148. if (test_bit(CONNECTING, &con->state))
  149. con->error_msg = "connection failed";
  150. else
  151. con->error_msg = "socket closed";
  152. queue_con(con);
  153. }
  154. break;
  155. case TCP_ESTABLISHED:
  156. dout("ceph_state_change TCP_ESTABLISHED\n");
  157. queue_con(con);
  158. break;
  159. }
  160. }
  161. /*
  162. * set up socket callbacks
  163. */
  164. static void set_sock_callbacks(struct socket *sock,
  165. struct ceph_connection *con)
  166. {
  167. struct sock *sk = sock->sk;
  168. sk->sk_user_data = (void *)con;
  169. sk->sk_data_ready = ceph_data_ready;
  170. sk->sk_write_space = ceph_write_space;
  171. sk->sk_state_change = ceph_state_change;
  172. }
  173. /*
  174. * socket helpers
  175. */
  176. /*
  177. * initiate connection to a remote socket.
  178. */
  179. static struct socket *ceph_tcp_connect(struct ceph_connection *con)
  180. {
  181. struct sockaddr *paddr = (struct sockaddr *)&con->peer_addr.in_addr;
  182. struct socket *sock;
  183. int ret;
  184. BUG_ON(con->sock);
  185. ret = sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
  186. if (ret)
  187. return ERR_PTR(ret);
  188. con->sock = sock;
  189. sock->sk->sk_allocation = GFP_NOFS;
  190. #ifdef CONFIG_LOCKDEP
  191. lockdep_set_class(&sock->sk->sk_lock, &socket_class);
  192. #endif
  193. set_sock_callbacks(sock, con);
  194. dout("connect %s\n", pr_addr(&con->peer_addr.in_addr));
  195. ret = sock->ops->connect(sock, paddr, sizeof(*paddr), O_NONBLOCK);
  196. if (ret == -EINPROGRESS) {
  197. dout("connect %s EINPROGRESS sk_state = %u\n",
  198. pr_addr(&con->peer_addr.in_addr),
  199. sock->sk->sk_state);
  200. ret = 0;
  201. }
  202. if (ret < 0) {
  203. pr_err("connect %s error %d\n",
  204. pr_addr(&con->peer_addr.in_addr), ret);
  205. sock_release(sock);
  206. con->sock = NULL;
  207. con->error_msg = "connect error";
  208. }
  209. if (ret < 0)
  210. return ERR_PTR(ret);
  211. return sock;
  212. }
  213. static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
  214. {
  215. struct kvec iov = {buf, len};
  216. struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
  217. return kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
  218. }
  219. /*
  220. * write something. @more is true if caller will be sending more data
  221. * shortly.
  222. */
  223. static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
  224. size_t kvlen, size_t len, int more)
  225. {
  226. struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
  227. if (more)
  228. msg.msg_flags |= MSG_MORE;
  229. else
  230. msg.msg_flags |= MSG_EOR; /* superfluous, but what the hell */
  231. return kernel_sendmsg(sock, &msg, iov, kvlen, len);
  232. }
  233. /*
  234. * Shutdown/close the socket for the given connection.
  235. */
  236. static int con_close_socket(struct ceph_connection *con)
  237. {
  238. int rc;
  239. dout("con_close_socket on %p sock %p\n", con, con->sock);
  240. if (!con->sock)
  241. return 0;
  242. set_bit(SOCK_CLOSED, &con->state);
  243. rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
  244. sock_release(con->sock);
  245. con->sock = NULL;
  246. clear_bit(SOCK_CLOSED, &con->state);
  247. return rc;
  248. }
  249. /*
  250. * Reset a connection. Discard all incoming and outgoing messages
  251. * and clear *_seq state.
  252. */
  253. static void ceph_msg_remove(struct ceph_msg *msg)
  254. {
  255. list_del_init(&msg->list_head);
  256. ceph_msg_put(msg);
  257. }
  258. static void ceph_msg_remove_list(struct list_head *head)
  259. {
  260. while (!list_empty(head)) {
  261. struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
  262. list_head);
  263. ceph_msg_remove(msg);
  264. }
  265. }
  266. static void reset_connection(struct ceph_connection *con)
  267. {
  268. /* reset connection, out_queue, msg_ and connect_seq */
  269. /* discard existing out_queue and msg_seq */
  270. ceph_msg_remove_list(&con->out_queue);
  271. ceph_msg_remove_list(&con->out_sent);
  272. if (con->in_msg) {
  273. ceph_msg_put(con->in_msg);
  274. con->in_msg = NULL;
  275. }
  276. con->connect_seq = 0;
  277. con->out_seq = 0;
  278. if (con->out_msg) {
  279. ceph_msg_put(con->out_msg);
  280. con->out_msg = NULL;
  281. }
  282. con->out_keepalive_pending = false;
  283. con->in_seq = 0;
  284. con->in_seq_acked = 0;
  285. }
  286. /*
  287. * mark a peer down. drop any open connections.
  288. */
  289. void ceph_con_close(struct ceph_connection *con)
  290. {
  291. dout("con_close %p peer %s\n", con, pr_addr(&con->peer_addr.in_addr));
  292. set_bit(CLOSED, &con->state); /* in case there's queued work */
  293. clear_bit(STANDBY, &con->state); /* avoid connect_seq bump */
  294. clear_bit(LOSSYTX, &con->state); /* so we retry next connect */
  295. clear_bit(KEEPALIVE_PENDING, &con->state);
  296. clear_bit(WRITE_PENDING, &con->state);
  297. mutex_lock(&con->mutex);
  298. reset_connection(con);
  299. con->peer_global_seq = 0;
  300. cancel_delayed_work(&con->work);
  301. mutex_unlock(&con->mutex);
  302. queue_con(con);
  303. }
  304. /*
  305. * Reopen a closed connection, with a new peer address.
  306. */
  307. void ceph_con_open(struct ceph_connection *con, struct ceph_entity_addr *addr)
  308. {
  309. dout("con_open %p %s\n", con, pr_addr(&addr->in_addr));
  310. set_bit(OPENING, &con->state);
  311. clear_bit(CLOSED, &con->state);
  312. memcpy(&con->peer_addr, addr, sizeof(*addr));
  313. con->delay = 0; /* reset backoff memory */
  314. queue_con(con);
  315. }
  316. /*
  317. * return true if this connection ever successfully opened
  318. */
  319. bool ceph_con_opened(struct ceph_connection *con)
  320. {
  321. return con->connect_seq > 0;
  322. }
  323. /*
  324. * generic get/put
  325. */
  326. struct ceph_connection *ceph_con_get(struct ceph_connection *con)
  327. {
  328. dout("con_get %p nref = %d -> %d\n", con,
  329. atomic_read(&con->nref), atomic_read(&con->nref) + 1);
  330. if (atomic_inc_not_zero(&con->nref))
  331. return con;
  332. return NULL;
  333. }
  334. void ceph_con_put(struct ceph_connection *con)
  335. {
  336. dout("con_put %p nref = %d -> %d\n", con,
  337. atomic_read(&con->nref), atomic_read(&con->nref) - 1);
  338. BUG_ON(atomic_read(&con->nref) == 0);
  339. if (atomic_dec_and_test(&con->nref)) {
  340. BUG_ON(con->sock);
  341. kfree(con);
  342. }
  343. }
  344. /*
  345. * initialize a new connection.
  346. */
  347. void ceph_con_init(struct ceph_messenger *msgr, struct ceph_connection *con)
  348. {
  349. dout("con_init %p\n", con);
  350. memset(con, 0, sizeof(*con));
  351. atomic_set(&con->nref, 1);
  352. con->msgr = msgr;
  353. mutex_init(&con->mutex);
  354. INIT_LIST_HEAD(&con->out_queue);
  355. INIT_LIST_HEAD(&con->out_sent);
  356. INIT_DELAYED_WORK(&con->work, con_work);
  357. }
  358. /*
  359. * We maintain a global counter to order connection attempts. Get
  360. * a unique seq greater than @gt.
  361. */
  362. static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
  363. {
  364. u32 ret;
  365. spin_lock(&msgr->global_seq_lock);
  366. if (msgr->global_seq < gt)
  367. msgr->global_seq = gt;
  368. ret = ++msgr->global_seq;
  369. spin_unlock(&msgr->global_seq_lock);
  370. return ret;
  371. }
  372. /*
  373. * Prepare footer for currently outgoing message, and finish things
  374. * off. Assumes out_kvec* are already valid.. we just add on to the end.
  375. */
  376. static void prepare_write_message_footer(struct ceph_connection *con, int v)
  377. {
  378. struct ceph_msg *m = con->out_msg;
  379. dout("prepare_write_message_footer %p\n", con);
  380. con->out_kvec_is_msg = true;
  381. con->out_kvec[v].iov_base = &m->footer;
  382. con->out_kvec[v].iov_len = sizeof(m->footer);
  383. con->out_kvec_bytes += sizeof(m->footer);
  384. con->out_kvec_left++;
  385. con->out_more = m->more_to_follow;
  386. con->out_msg_done = true;
  387. }
  388. /*
  389. * Prepare headers for the next outgoing message.
  390. */
  391. static void prepare_write_message(struct ceph_connection *con)
  392. {
  393. struct ceph_msg *m;
  394. int v = 0;
  395. con->out_kvec_bytes = 0;
  396. con->out_kvec_is_msg = true;
  397. con->out_msg_done = false;
  398. /* Sneak an ack in there first? If we can get it into the same
  399. * TCP packet that's a good thing. */
  400. if (con->in_seq > con->in_seq_acked) {
  401. con->in_seq_acked = con->in_seq;
  402. con->out_kvec[v].iov_base = &tag_ack;
  403. con->out_kvec[v++].iov_len = 1;
  404. con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
  405. con->out_kvec[v].iov_base = &con->out_temp_ack;
  406. con->out_kvec[v++].iov_len = sizeof(con->out_temp_ack);
  407. con->out_kvec_bytes = 1 + sizeof(con->out_temp_ack);
  408. }
  409. m = list_first_entry(&con->out_queue,
  410. struct ceph_msg, list_head);
  411. con->out_msg = m;
  412. if (test_bit(LOSSYTX, &con->state)) {
  413. list_del_init(&m->list_head);
  414. } else {
  415. /* put message on sent list */
  416. ceph_msg_get(m);
  417. list_move_tail(&m->list_head, &con->out_sent);
  418. }
  419. /*
  420. * only assign outgoing seq # if we haven't sent this message
  421. * yet. if it is requeued, resend with it's original seq.
  422. */
  423. if (m->needs_out_seq) {
  424. m->hdr.seq = cpu_to_le64(++con->out_seq);
  425. m->needs_out_seq = false;
  426. }
  427. dout("prepare_write_message %p seq %lld type %d len %d+%d+%d %d pgs\n",
  428. m, con->out_seq, le16_to_cpu(m->hdr.type),
  429. le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
  430. le32_to_cpu(m->hdr.data_len),
  431. m->nr_pages);
  432. BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);
  433. /* tag + hdr + front + middle */
  434. con->out_kvec[v].iov_base = &tag_msg;
  435. con->out_kvec[v++].iov_len = 1;
  436. con->out_kvec[v].iov_base = &m->hdr;
  437. con->out_kvec[v++].iov_len = sizeof(m->hdr);
  438. con->out_kvec[v++] = m->front;
  439. if (m->middle)
  440. con->out_kvec[v++] = m->middle->vec;
  441. con->out_kvec_left = v;
  442. con->out_kvec_bytes += 1 + sizeof(m->hdr) + m->front.iov_len +
  443. (m->middle ? m->middle->vec.iov_len : 0);
  444. con->out_kvec_cur = con->out_kvec;
  445. /* fill in crc (except data pages), footer */
  446. con->out_msg->hdr.crc =
  447. cpu_to_le32(crc32c(0, (void *)&m->hdr,
  448. sizeof(m->hdr) - sizeof(m->hdr.crc)));
  449. con->out_msg->footer.flags = CEPH_MSG_FOOTER_COMPLETE;
  450. con->out_msg->footer.front_crc =
  451. cpu_to_le32(crc32c(0, m->front.iov_base, m->front.iov_len));
  452. if (m->middle)
  453. con->out_msg->footer.middle_crc =
  454. cpu_to_le32(crc32c(0, m->middle->vec.iov_base,
  455. m->middle->vec.iov_len));
  456. else
  457. con->out_msg->footer.middle_crc = 0;
  458. con->out_msg->footer.data_crc = 0;
  459. dout("prepare_write_message front_crc %u data_crc %u\n",
  460. le32_to_cpu(con->out_msg->footer.front_crc),
  461. le32_to_cpu(con->out_msg->footer.middle_crc));
  462. /* is there a data payload? */
  463. if (le32_to_cpu(m->hdr.data_len) > 0) {
  464. /* initialize page iterator */
  465. con->out_msg_pos.page = 0;
  466. con->out_msg_pos.page_pos =
  467. le16_to_cpu(m->hdr.data_off) & ~PAGE_MASK;
  468. con->out_msg_pos.data_pos = 0;
  469. con->out_msg_pos.did_page_crc = 0;
  470. con->out_more = 1; /* data + footer will follow */
  471. } else {
  472. /* no, queue up footer too and be done */
  473. prepare_write_message_footer(con, v);
  474. }
  475. set_bit(WRITE_PENDING, &con->state);
  476. }
  477. /*
  478. * Prepare an ack.
  479. */
  480. static void prepare_write_ack(struct ceph_connection *con)
  481. {
  482. dout("prepare_write_ack %p %llu -> %llu\n", con,
  483. con->in_seq_acked, con->in_seq);
  484. con->in_seq_acked = con->in_seq;
  485. con->out_kvec[0].iov_base = &tag_ack;
  486. con->out_kvec[0].iov_len = 1;
  487. con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
  488. con->out_kvec[1].iov_base = &con->out_temp_ack;
  489. con->out_kvec[1].iov_len = sizeof(con->out_temp_ack);
  490. con->out_kvec_left = 2;
  491. con->out_kvec_bytes = 1 + sizeof(con->out_temp_ack);
  492. con->out_kvec_cur = con->out_kvec;
  493. con->out_more = 1; /* more will follow.. eventually.. */
  494. set_bit(WRITE_PENDING, &con->state);
  495. }
  496. /*
  497. * Prepare to write keepalive byte.
  498. */
  499. static void prepare_write_keepalive(struct ceph_connection *con)
  500. {
  501. dout("prepare_write_keepalive %p\n", con);
  502. con->out_kvec[0].iov_base = &tag_keepalive;
  503. con->out_kvec[0].iov_len = 1;
  504. con->out_kvec_left = 1;
  505. con->out_kvec_bytes = 1;
  506. con->out_kvec_cur = con->out_kvec;
  507. set_bit(WRITE_PENDING, &con->state);
  508. }
  509. /*
  510. * Connection negotiation.
  511. */
  512. static void prepare_connect_authorizer(struct ceph_connection *con)
  513. {
  514. void *auth_buf;
  515. int auth_len = 0;
  516. int auth_protocol = 0;
  517. mutex_unlock(&con->mutex);
  518. if (con->ops->get_authorizer)
  519. con->ops->get_authorizer(con, &auth_buf, &auth_len,
  520. &auth_protocol, &con->auth_reply_buf,
  521. &con->auth_reply_buf_len,
  522. con->auth_retry);
  523. mutex_lock(&con->mutex);
  524. con->out_connect.authorizer_protocol = cpu_to_le32(auth_protocol);
  525. con->out_connect.authorizer_len = cpu_to_le32(auth_len);
  526. con->out_kvec[con->out_kvec_left].iov_base = auth_buf;
  527. con->out_kvec[con->out_kvec_left].iov_len = auth_len;
  528. con->out_kvec_left++;
  529. con->out_kvec_bytes += auth_len;
  530. }
  531. /*
  532. * We connected to a peer and are saying hello.
  533. */
  534. static void prepare_write_banner(struct ceph_messenger *msgr,
  535. struct ceph_connection *con)
  536. {
  537. int len = strlen(CEPH_BANNER);
  538. con->out_kvec[0].iov_base = CEPH_BANNER;
  539. con->out_kvec[0].iov_len = len;
  540. con->out_kvec[1].iov_base = &msgr->my_enc_addr;
  541. con->out_kvec[1].iov_len = sizeof(msgr->my_enc_addr);
  542. con->out_kvec_left = 2;
  543. con->out_kvec_bytes = len + sizeof(msgr->my_enc_addr);
  544. con->out_kvec_cur = con->out_kvec;
  545. con->out_more = 0;
  546. set_bit(WRITE_PENDING, &con->state);
  547. }
  548. static void prepare_write_connect(struct ceph_messenger *msgr,
  549. struct ceph_connection *con,
  550. int after_banner)
  551. {
  552. unsigned global_seq = get_global_seq(con->msgr, 0);
  553. int proto;
  554. switch (con->peer_name.type) {
  555. case CEPH_ENTITY_TYPE_MON:
  556. proto = CEPH_MONC_PROTOCOL;
  557. break;
  558. case CEPH_ENTITY_TYPE_OSD:
  559. proto = CEPH_OSDC_PROTOCOL;
  560. break;
  561. case CEPH_ENTITY_TYPE_MDS:
  562. proto = CEPH_MDSC_PROTOCOL;
  563. break;
  564. default:
  565. BUG();
  566. }
  567. dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
  568. con->connect_seq, global_seq, proto);
  569. con->out_connect.features = CEPH_FEATURE_SUPPORTED_CLIENT;
  570. con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
  571. con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
  572. con->out_connect.global_seq = cpu_to_le32(global_seq);
  573. con->out_connect.protocol_version = cpu_to_le32(proto);
  574. con->out_connect.flags = 0;
  575. if (!after_banner) {
  576. con->out_kvec_left = 0;
  577. con->out_kvec_bytes = 0;
  578. }
  579. con->out_kvec[con->out_kvec_left].iov_base = &con->out_connect;
  580. con->out_kvec[con->out_kvec_left].iov_len = sizeof(con->out_connect);
  581. con->out_kvec_left++;
  582. con->out_kvec_bytes += sizeof(con->out_connect);
  583. con->out_kvec_cur = con->out_kvec;
  584. con->out_more = 0;
  585. set_bit(WRITE_PENDING, &con->state);
  586. prepare_connect_authorizer(con);
  587. }
  588. /*
  589. * write as much of pending kvecs to the socket as we can.
  590. * 1 -> done
  591. * 0 -> socket full, but more to do
  592. * <0 -> error
  593. */
  594. static int write_partial_kvec(struct ceph_connection *con)
  595. {
  596. int ret;
  597. dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
  598. while (con->out_kvec_bytes > 0) {
  599. ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
  600. con->out_kvec_left, con->out_kvec_bytes,
  601. con->out_more);
  602. if (ret <= 0)
  603. goto out;
  604. con->out_kvec_bytes -= ret;
  605. if (con->out_kvec_bytes == 0)
  606. break; /* done */
  607. while (ret > 0) {
  608. if (ret >= con->out_kvec_cur->iov_len) {
  609. ret -= con->out_kvec_cur->iov_len;
  610. con->out_kvec_cur++;
  611. con->out_kvec_left--;
  612. } else {
  613. con->out_kvec_cur->iov_len -= ret;
  614. con->out_kvec_cur->iov_base += ret;
  615. ret = 0;
  616. break;
  617. }
  618. }
  619. }
  620. con->out_kvec_left = 0;
  621. con->out_kvec_is_msg = false;
  622. ret = 1;
  623. out:
  624. dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
  625. con->out_kvec_bytes, con->out_kvec_left, ret);
  626. return ret; /* done! */
  627. }
  628. /*
  629. * Write as much message data payload as we can. If we finish, queue
  630. * up the footer.
  631. * 1 -> done, footer is now queued in out_kvec[].
  632. * 0 -> socket full, but more to do
  633. * <0 -> error
  634. */
  635. static int write_partial_msg_pages(struct ceph_connection *con)
  636. {
  637. struct ceph_msg *msg = con->out_msg;
  638. unsigned data_len = le32_to_cpu(msg->hdr.data_len);
  639. size_t len;
  640. int crc = con->msgr->nocrc;
  641. int ret;
  642. dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
  643. con, con->out_msg, con->out_msg_pos.page, con->out_msg->nr_pages,
  644. con->out_msg_pos.page_pos);
  645. while (con->out_msg_pos.page < con->out_msg->nr_pages) {
  646. struct page *page = NULL;
  647. void *kaddr = NULL;
  648. /*
  649. * if we are calculating the data crc (the default), we need
  650. * to map the page. if our pages[] has been revoked, use the
  651. * zero page.
  652. */
  653. if (msg->pages) {
  654. page = msg->pages[con->out_msg_pos.page];
  655. if (crc)
  656. kaddr = kmap(page);
  657. } else if (msg->pagelist) {
  658. page = list_first_entry(&msg->pagelist->head,
  659. struct page, lru);
  660. if (crc)
  661. kaddr = kmap(page);
  662. } else {
  663. page = con->msgr->zero_page;
  664. if (crc)
  665. kaddr = page_address(con->msgr->zero_page);
  666. }
  667. len = min((int)(PAGE_SIZE - con->out_msg_pos.page_pos),
  668. (int)(data_len - con->out_msg_pos.data_pos));
  669. if (crc && !con->out_msg_pos.did_page_crc) {
  670. void *base = kaddr + con->out_msg_pos.page_pos;
  671. u32 tmpcrc = le32_to_cpu(con->out_msg->footer.data_crc);
  672. BUG_ON(kaddr == NULL);
  673. con->out_msg->footer.data_crc =
  674. cpu_to_le32(crc32c(tmpcrc, base, len));
  675. con->out_msg_pos.did_page_crc = 1;
  676. }
  677. ret = kernel_sendpage(con->sock, page,
  678. con->out_msg_pos.page_pos, len,
  679. MSG_DONTWAIT | MSG_NOSIGNAL |
  680. MSG_MORE);
  681. if (crc && (msg->pages || msg->pagelist))
  682. kunmap(page);
  683. if (ret <= 0)
  684. goto out;
  685. con->out_msg_pos.data_pos += ret;
  686. con->out_msg_pos.page_pos += ret;
  687. if (ret == len) {
  688. con->out_msg_pos.page_pos = 0;
  689. con->out_msg_pos.page++;
  690. con->out_msg_pos.did_page_crc = 0;
  691. if (msg->pagelist)
  692. list_move_tail(&page->lru,
  693. &msg->pagelist->head);
  694. }
  695. }
  696. dout("write_partial_msg_pages %p msg %p done\n", con, msg);
  697. /* prepare and queue up footer, too */
  698. if (!crc)
  699. con->out_msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
  700. con->out_kvec_bytes = 0;
  701. con->out_kvec_left = 0;
  702. con->out_kvec_cur = con->out_kvec;
  703. prepare_write_message_footer(con, 0);
  704. ret = 1;
  705. out:
  706. return ret;
  707. }
  708. /*
  709. * write some zeros
  710. */
  711. static int write_partial_skip(struct ceph_connection *con)
  712. {
  713. int ret;
  714. while (con->out_skip > 0) {
  715. struct kvec iov = {
  716. .iov_base = page_address(con->msgr->zero_page),
  717. .iov_len = min(con->out_skip, (int)PAGE_CACHE_SIZE)
  718. };
  719. ret = ceph_tcp_sendmsg(con->sock, &iov, 1, iov.iov_len, 1);
  720. if (ret <= 0)
  721. goto out;
  722. con->out_skip -= ret;
  723. }
  724. ret = 1;
  725. out:
  726. return ret;
  727. }
  728. /*
  729. * Prepare to read connection handshake, or an ack.
  730. */
  731. static void prepare_read_banner(struct ceph_connection *con)
  732. {
  733. dout("prepare_read_banner %p\n", con);
  734. con->in_base_pos = 0;
  735. }
  736. static void prepare_read_connect(struct ceph_connection *con)
  737. {
  738. dout("prepare_read_connect %p\n", con);
  739. con->in_base_pos = 0;
  740. }
  741. static void prepare_read_ack(struct ceph_connection *con)
  742. {
  743. dout("prepare_read_ack %p\n", con);
  744. con->in_base_pos = 0;
  745. }
  746. static void prepare_read_tag(struct ceph_connection *con)
  747. {
  748. dout("prepare_read_tag %p\n", con);
  749. con->in_base_pos = 0;
  750. con->in_tag = CEPH_MSGR_TAG_READY;
  751. }
  752. /*
  753. * Prepare to read a message.
  754. */
  755. static int prepare_read_message(struct ceph_connection *con)
  756. {
  757. dout("prepare_read_message %p\n", con);
  758. BUG_ON(con->in_msg != NULL);
  759. con->in_base_pos = 0;
  760. con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
  761. return 0;
  762. }
  763. static int read_partial(struct ceph_connection *con,
  764. int *to, int size, void *object)
  765. {
  766. *to += size;
  767. while (con->in_base_pos < *to) {
  768. int left = *to - con->in_base_pos;
  769. int have = size - left;
  770. int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
  771. if (ret <= 0)
  772. return ret;
  773. con->in_base_pos += ret;
  774. }
  775. return 1;
  776. }
  777. /*
  778. * Read all or part of the connect-side handshake on a new connection
  779. */
  780. static int read_partial_banner(struct ceph_connection *con)
  781. {
  782. int ret, to = 0;
  783. dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
  784. /* peer's banner */
  785. ret = read_partial(con, &to, strlen(CEPH_BANNER), con->in_banner);
  786. if (ret <= 0)
  787. goto out;
  788. ret = read_partial(con, &to, sizeof(con->actual_peer_addr),
  789. &con->actual_peer_addr);
  790. if (ret <= 0)
  791. goto out;
  792. ret = read_partial(con, &to, sizeof(con->peer_addr_for_me),
  793. &con->peer_addr_for_me);
  794. if (ret <= 0)
  795. goto out;
  796. out:
  797. return ret;
  798. }
  799. static int read_partial_connect(struct ceph_connection *con)
  800. {
  801. int ret, to = 0;
  802. dout("read_partial_connect %p at %d\n", con, con->in_base_pos);
  803. ret = read_partial(con, &to, sizeof(con->in_reply), &con->in_reply);
  804. if (ret <= 0)
  805. goto out;
  806. ret = read_partial(con, &to, le32_to_cpu(con->in_reply.authorizer_len),
  807. con->auth_reply_buf);
  808. if (ret <= 0)
  809. goto out;
  810. dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
  811. con, (int)con->in_reply.tag,
  812. le32_to_cpu(con->in_reply.connect_seq),
  813. le32_to_cpu(con->in_reply.global_seq));
  814. out:
  815. return ret;
  816. }
  817. /*
  818. * Verify the hello banner looks okay.
  819. */
  820. static int verify_hello(struct ceph_connection *con)
  821. {
  822. if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
  823. pr_err("connect to %s got bad banner\n",
  824. pr_addr(&con->peer_addr.in_addr));
  825. con->error_msg = "protocol error, bad banner";
  826. return -1;
  827. }
  828. return 0;
  829. }
  830. static bool addr_is_blank(struct sockaddr_storage *ss)
  831. {
  832. switch (ss->ss_family) {
  833. case AF_INET:
  834. return ((struct sockaddr_in *)ss)->sin_addr.s_addr == 0;
  835. case AF_INET6:
  836. return
  837. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[0] == 0 &&
  838. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[1] == 0 &&
  839. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[2] == 0 &&
  840. ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[3] == 0;
  841. }
  842. return false;
  843. }
  844. static int addr_port(struct sockaddr_storage *ss)
  845. {
  846. switch (ss->ss_family) {
  847. case AF_INET:
  848. return ntohs(((struct sockaddr_in *)ss)->sin_port);
  849. case AF_INET6:
  850. return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
  851. }
  852. return 0;
  853. }
  854. static void addr_set_port(struct sockaddr_storage *ss, int p)
  855. {
  856. switch (ss->ss_family) {
  857. case AF_INET:
  858. ((struct sockaddr_in *)ss)->sin_port = htons(p);
  859. case AF_INET6:
  860. ((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
  861. }
  862. }
  863. /*
  864. * Parse an ip[:port] list into an addr array. Use the default
  865. * monitor port if a port isn't specified.
  866. */
  867. int ceph_parse_ips(const char *c, const char *end,
  868. struct ceph_entity_addr *addr,
  869. int max_count, int *count)
  870. {
  871. int i;
  872. const char *p = c;
  873. dout("parse_ips on '%.*s'\n", (int)(end-c), c);
  874. for (i = 0; i < max_count; i++) {
  875. const char *ipend;
  876. struct sockaddr_storage *ss = &addr[i].in_addr;
  877. struct sockaddr_in *in4 = (void *)ss;
  878. struct sockaddr_in6 *in6 = (void *)ss;
  879. int port;
  880. memset(ss, 0, sizeof(*ss));
  881. if (in4_pton(p, end - p, (u8 *)&in4->sin_addr.s_addr,
  882. ',', &ipend)) {
  883. ss->ss_family = AF_INET;
  884. } else if (in6_pton(p, end - p, (u8 *)&in6->sin6_addr.s6_addr,
  885. ',', &ipend)) {
  886. ss->ss_family = AF_INET6;
  887. } else {
  888. goto bad;
  889. }
  890. p = ipend;
  891. /* port? */
  892. if (p < end && *p == ':') {
  893. port = 0;
  894. p++;
  895. while (p < end && *p >= '0' && *p <= '9') {
  896. port = (port * 10) + (*p - '0');
  897. p++;
  898. }
  899. if (port > 65535 || port == 0)
  900. goto bad;
  901. } else {
  902. port = CEPH_MON_PORT;
  903. }
  904. addr_set_port(ss, port);
  905. dout("parse_ips got %s\n", pr_addr(ss));
  906. if (p == end)
  907. break;
  908. if (*p != ',')
  909. goto bad;
  910. p++;
  911. }
  912. if (p != end)
  913. goto bad;
  914. if (count)
  915. *count = i + 1;
  916. return 0;
  917. bad:
  918. pr_err("parse_ips bad ip '%s'\n", c);
  919. return -EINVAL;
  920. }
  921. static int process_banner(struct ceph_connection *con)
  922. {
  923. dout("process_banner on %p\n", con);
  924. if (verify_hello(con) < 0)
  925. return -1;
  926. ceph_decode_addr(&con->actual_peer_addr);
  927. ceph_decode_addr(&con->peer_addr_for_me);
  928. /*
  929. * Make sure the other end is who we wanted. note that the other
  930. * end may not yet know their ip address, so if it's 0.0.0.0, give
  931. * them the benefit of the doubt.
  932. */
  933. if (memcmp(&con->peer_addr, &con->actual_peer_addr,
  934. sizeof(con->peer_addr)) != 0 &&
  935. !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
  936. con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
  937. pr_warning("wrong peer, want %s/%lld, got %s/%lld\n",
  938. pr_addr(&con->peer_addr.in_addr),
  939. le64_to_cpu(con->peer_addr.nonce),
  940. pr_addr(&con->actual_peer_addr.in_addr),
  941. le64_to_cpu(con->actual_peer_addr.nonce));
  942. con->error_msg = "wrong peer at address";
  943. return -1;
  944. }
  945. /*
  946. * did we learn our address?
  947. */
  948. if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
  949. int port = addr_port(&con->msgr->inst.addr.in_addr);
  950. memcpy(&con->msgr->inst.addr.in_addr,
  951. &con->peer_addr_for_me.in_addr,
  952. sizeof(con->peer_addr_for_me.in_addr));
  953. addr_set_port(&con->msgr->inst.addr.in_addr, port);
  954. encode_my_addr(con->msgr);
  955. dout("process_banner learned my addr is %s\n",
  956. pr_addr(&con->msgr->inst.addr.in_addr));
  957. }
  958. set_bit(NEGOTIATING, &con->state);
  959. prepare_read_connect(con);
  960. return 0;
  961. }
  962. static void fail_protocol(struct ceph_connection *con)
  963. {
  964. reset_connection(con);
  965. set_bit(CLOSED, &con->state); /* in case there's queued work */
  966. mutex_unlock(&con->mutex);
  967. if (con->ops->bad_proto)
  968. con->ops->bad_proto(con);
  969. mutex_lock(&con->mutex);
  970. }
  971. static int process_connect(struct ceph_connection *con)
  972. {
  973. u64 sup_feat = CEPH_FEATURE_SUPPORTED_CLIENT;
  974. u64 req_feat = CEPH_FEATURE_REQUIRED_CLIENT;
  975. u64 server_feat = le64_to_cpu(con->in_reply.features);
  976. dout("process_connect on %p tag %d\n", con, (int)con->in_tag);
  977. switch (con->in_reply.tag) {
  978. case CEPH_MSGR_TAG_FEATURES:
  979. pr_err("%s%lld %s feature set mismatch,"
  980. " my %llx < server's %llx, missing %llx\n",
  981. ENTITY_NAME(con->peer_name),
  982. pr_addr(&con->peer_addr.in_addr),
  983. sup_feat, server_feat, server_feat & ~sup_feat);
  984. con->error_msg = "missing required protocol features";
  985. fail_protocol(con);
  986. return -1;
  987. case CEPH_MSGR_TAG_BADPROTOVER:
  988. pr_err("%s%lld %s protocol version mismatch,"
  989. " my %d != server's %d\n",
  990. ENTITY_NAME(con->peer_name),
  991. pr_addr(&con->peer_addr.in_addr),
  992. le32_to_cpu(con->out_connect.protocol_version),
  993. le32_to_cpu(con->in_reply.protocol_version));
  994. con->error_msg = "protocol version mismatch";
  995. fail_protocol(con);
  996. return -1;
  997. case CEPH_MSGR_TAG_BADAUTHORIZER:
  998. con->auth_retry++;
  999. dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
  1000. con->auth_retry);
  1001. if (con->auth_retry == 2) {
  1002. con->error_msg = "connect authorization failure";
  1003. reset_connection(con);
  1004. set_bit(CLOSED, &con->state);
  1005. return -1;
  1006. }
  1007. con->auth_retry = 1;
  1008. prepare_write_connect(con->msgr, con, 0);
  1009. prepare_read_connect(con);
  1010. break;
  1011. case CEPH_MSGR_TAG_RESETSESSION:
  1012. /*
  1013. * If we connected with a large connect_seq but the peer
  1014. * has no record of a session with us (no connection, or
  1015. * connect_seq == 0), they will send RESETSESION to indicate
  1016. * that they must have reset their session, and may have
  1017. * dropped messages.
  1018. */
  1019. dout("process_connect got RESET peer seq %u\n",
  1020. le32_to_cpu(con->in_connect.connect_seq));
  1021. pr_err("%s%lld %s connection reset\n",
  1022. ENTITY_NAME(con->peer_name),
  1023. pr_addr(&con->peer_addr.in_addr));
  1024. reset_connection(con);
  1025. prepare_write_connect(con->msgr, con, 0);
  1026. prepare_read_connect(con);
  1027. /* Tell ceph about it. */
  1028. mutex_unlock(&con->mutex);
  1029. pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
  1030. if (con->ops->peer_reset)
  1031. con->ops->peer_reset(con);
  1032. mutex_lock(&con->mutex);
  1033. break;
  1034. case CEPH_MSGR_TAG_RETRY_SESSION:
  1035. /*
  1036. * If we sent a smaller connect_seq than the peer has, try
  1037. * again with a larger value.
  1038. */
  1039. dout("process_connect got RETRY my seq = %u, peer_seq = %u\n",
  1040. le32_to_cpu(con->out_connect.connect_seq),
  1041. le32_to_cpu(con->in_connect.connect_seq));
  1042. con->connect_seq = le32_to_cpu(con->in_connect.connect_seq);
  1043. prepare_write_connect(con->msgr, con, 0);
  1044. prepare_read_connect(con);
  1045. break;
  1046. case CEPH_MSGR_TAG_RETRY_GLOBAL:
  1047. /*
  1048. * If we sent a smaller global_seq than the peer has, try
  1049. * again with a larger value.
  1050. */
  1051. dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
  1052. con->peer_global_seq,
  1053. le32_to_cpu(con->in_connect.global_seq));
  1054. get_global_seq(con->msgr,
  1055. le32_to_cpu(con->in_connect.global_seq));
  1056. prepare_write_connect(con->msgr, con, 0);
  1057. prepare_read_connect(con);
  1058. break;
  1059. case CEPH_MSGR_TAG_READY:
  1060. if (req_feat & ~server_feat) {
  1061. pr_err("%s%lld %s protocol feature mismatch,"
  1062. " my required %llx > server's %llx, need %llx\n",
  1063. ENTITY_NAME(con->peer_name),
  1064. pr_addr(&con->peer_addr.in_addr),
  1065. req_feat, server_feat, req_feat & ~server_feat);
  1066. con->error_msg = "missing required protocol features";
  1067. fail_protocol(con);
  1068. return -1;
  1069. }
  1070. clear_bit(CONNECTING, &con->state);
  1071. con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
  1072. con->connect_seq++;
  1073. con->peer_features = server_feat;
  1074. dout("process_connect got READY gseq %d cseq %d (%d)\n",
  1075. con->peer_global_seq,
  1076. le32_to_cpu(con->in_reply.connect_seq),
  1077. con->connect_seq);
  1078. WARN_ON(con->connect_seq !=
  1079. le32_to_cpu(con->in_reply.connect_seq));
  1080. if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
  1081. set_bit(LOSSYTX, &con->state);
  1082. prepare_read_tag(con);
  1083. break;
  1084. case CEPH_MSGR_TAG_WAIT:
  1085. /*
  1086. * If there is a connection race (we are opening
  1087. * connections to each other), one of us may just have
  1088. * to WAIT. This shouldn't happen if we are the
  1089. * client.
  1090. */
  1091. pr_err("process_connect peer connecting WAIT\n");
  1092. default:
  1093. pr_err("connect protocol error, will retry\n");
  1094. con->error_msg = "protocol error, garbage tag during connect";
  1095. return -1;
  1096. }
  1097. return 0;
  1098. }
  1099. /*
  1100. * read (part of) an ack
  1101. */
  1102. static int read_partial_ack(struct ceph_connection *con)
  1103. {
  1104. int to = 0;
  1105. return read_partial(con, &to, sizeof(con->in_temp_ack),
  1106. &con->in_temp_ack);
  1107. }
  1108. /*
  1109. * We can finally discard anything that's been acked.
  1110. */
  1111. static void process_ack(struct ceph_connection *con)
  1112. {
  1113. struct ceph_msg *m;
  1114. u64 ack = le64_to_cpu(con->in_temp_ack);
  1115. u64 seq;
  1116. while (!list_empty(&con->out_sent)) {
  1117. m = list_first_entry(&con->out_sent, struct ceph_msg,
  1118. list_head);
  1119. seq = le64_to_cpu(m->hdr.seq);
  1120. if (seq > ack)
  1121. break;
  1122. dout("got ack for seq %llu type %d at %p\n", seq,
  1123. le16_to_cpu(m->hdr.type), m);
  1124. ceph_msg_remove(m);
  1125. }
  1126. prepare_read_tag(con);
  1127. }
  1128. static int read_partial_message_section(struct ceph_connection *con,
  1129. struct kvec *section, unsigned int sec_len,
  1130. u32 *crc)
  1131. {
  1132. int left;
  1133. int ret;
  1134. BUG_ON(!section);
  1135. while (section->iov_len < sec_len) {
  1136. BUG_ON(section->iov_base == NULL);
  1137. left = sec_len - section->iov_len;
  1138. ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
  1139. section->iov_len, left);
  1140. if (ret <= 0)
  1141. return ret;
  1142. section->iov_len += ret;
  1143. if (section->iov_len == sec_len)
  1144. *crc = crc32c(0, section->iov_base,
  1145. section->iov_len);
  1146. }
  1147. return 1;
  1148. }
  1149. static struct ceph_msg *ceph_alloc_msg(struct ceph_connection *con,
  1150. struct ceph_msg_header *hdr,
  1151. int *skip);
  1152. /*
  1153. * read (part of) a message.
  1154. */
  1155. static int read_partial_message(struct ceph_connection *con)
  1156. {
  1157. struct ceph_msg *m = con->in_msg;
  1158. void *p;
  1159. int ret;
  1160. int to, left;
  1161. unsigned front_len, middle_len, data_len, data_off;
  1162. int datacrc = con->msgr->nocrc;
  1163. int skip;
  1164. u64 seq;
  1165. dout("read_partial_message con %p msg %p\n", con, m);
  1166. /* header */
  1167. while (con->in_base_pos < sizeof(con->in_hdr)) {
  1168. left = sizeof(con->in_hdr) - con->in_base_pos;
  1169. ret = ceph_tcp_recvmsg(con->sock,
  1170. (char *)&con->in_hdr + con->in_base_pos,
  1171. left);
  1172. if (ret <= 0)
  1173. return ret;
  1174. con->in_base_pos += ret;
  1175. if (con->in_base_pos == sizeof(con->in_hdr)) {
  1176. u32 crc = crc32c(0, (void *)&con->in_hdr,
  1177. sizeof(con->in_hdr) - sizeof(con->in_hdr.crc));
  1178. if (crc != le32_to_cpu(con->in_hdr.crc)) {
  1179. pr_err("read_partial_message bad hdr "
  1180. " crc %u != expected %u\n",
  1181. crc, con->in_hdr.crc);
  1182. return -EBADMSG;
  1183. }
  1184. }
  1185. }
  1186. front_len = le32_to_cpu(con->in_hdr.front_len);
  1187. if (front_len > CEPH_MSG_MAX_FRONT_LEN)
  1188. return -EIO;
  1189. middle_len = le32_to_cpu(con->in_hdr.middle_len);
  1190. if (middle_len > CEPH_MSG_MAX_DATA_LEN)
  1191. return -EIO;
  1192. data_len = le32_to_cpu(con->in_hdr.data_len);
  1193. if (data_len > CEPH_MSG_MAX_DATA_LEN)
  1194. return -EIO;
  1195. data_off = le16_to_cpu(con->in_hdr.data_off);
  1196. /* verify seq# */
  1197. seq = le64_to_cpu(con->in_hdr.seq);
  1198. if ((s64)seq - (s64)con->in_seq < 1) {
  1199. pr_info("skipping %s%lld %s seq %lld, expected %lld\n",
  1200. ENTITY_NAME(con->peer_name),
  1201. pr_addr(&con->peer_addr.in_addr),
  1202. seq, con->in_seq + 1);
  1203. con->in_base_pos = -front_len - middle_len - data_len -
  1204. sizeof(m->footer);
  1205. con->in_tag = CEPH_MSGR_TAG_READY;
  1206. con->in_seq++;
  1207. return 0;
  1208. } else if ((s64)seq - (s64)con->in_seq > 1) {
  1209. pr_err("read_partial_message bad seq %lld expected %lld\n",
  1210. seq, con->in_seq + 1);
  1211. con->error_msg = "bad message sequence # for incoming message";
  1212. return -EBADMSG;
  1213. }
  1214. /* allocate message? */
  1215. if (!con->in_msg) {
  1216. dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
  1217. con->in_hdr.front_len, con->in_hdr.data_len);
  1218. con->in_msg = ceph_alloc_msg(con, &con->in_hdr, &skip);
  1219. if (skip) {
  1220. /* skip this message */
  1221. dout("alloc_msg said skip message\n");
  1222. con->in_base_pos = -front_len - middle_len - data_len -
  1223. sizeof(m->footer);
  1224. con->in_tag = CEPH_MSGR_TAG_READY;
  1225. con->in_seq++;
  1226. return 0;
  1227. }
  1228. if (!con->in_msg) {
  1229. con->error_msg =
  1230. "error allocating memory for incoming message";
  1231. return -ENOMEM;
  1232. }
  1233. m = con->in_msg;
  1234. m->front.iov_len = 0; /* haven't read it yet */
  1235. if (m->middle)
  1236. m->middle->vec.iov_len = 0;
  1237. con->in_msg_pos.page = 0;
  1238. con->in_msg_pos.page_pos = data_off & ~PAGE_MASK;
  1239. con->in_msg_pos.data_pos = 0;
  1240. }
  1241. /* front */
  1242. ret = read_partial_message_section(con, &m->front, front_len,
  1243. &con->in_front_crc);
  1244. if (ret <= 0)
  1245. return ret;
  1246. /* middle */
  1247. if (m->middle) {
  1248. ret = read_partial_message_section(con, &m->middle->vec, middle_len,
  1249. &con->in_middle_crc);
  1250. if (ret <= 0)
  1251. return ret;
  1252. }
  1253. /* (page) data */
  1254. while (con->in_msg_pos.data_pos < data_len) {
  1255. left = min((int)(data_len - con->in_msg_pos.data_pos),
  1256. (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
  1257. BUG_ON(m->pages == NULL);
  1258. p = kmap(m->pages[con->in_msg_pos.page]);
  1259. ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
  1260. left);
  1261. if (ret > 0 && datacrc)
  1262. con->in_data_crc =
  1263. crc32c(con->in_data_crc,
  1264. p + con->in_msg_pos.page_pos, ret);
  1265. kunmap(m->pages[con->in_msg_pos.page]);
  1266. if (ret <= 0)
  1267. return ret;
  1268. con->in_msg_pos.data_pos += ret;
  1269. con->in_msg_pos.page_pos += ret;
  1270. if (con->in_msg_pos.page_pos == PAGE_SIZE) {
  1271. con->in_msg_pos.page_pos = 0;
  1272. con->in_msg_pos.page++;
  1273. }
  1274. }
  1275. /* footer */
  1276. to = sizeof(m->hdr) + sizeof(m->footer);
  1277. while (con->in_base_pos < to) {
  1278. left = to - con->in_base_pos;
  1279. ret = ceph_tcp_recvmsg(con->sock, (char *)&m->footer +
  1280. (con->in_base_pos - sizeof(m->hdr)),
  1281. left);
  1282. if (ret <= 0)
  1283. return ret;
  1284. con->in_base_pos += ret;
  1285. }
  1286. dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
  1287. m, front_len, m->footer.front_crc, middle_len,
  1288. m->footer.middle_crc, data_len, m->footer.data_crc);
  1289. /* crc ok? */
  1290. if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
  1291. pr_err("read_partial_message %p front crc %u != exp. %u\n",
  1292. m, con->in_front_crc, m->footer.front_crc);
  1293. return -EBADMSG;
  1294. }
  1295. if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
  1296. pr_err("read_partial_message %p middle crc %u != exp %u\n",
  1297. m, con->in_middle_crc, m->footer.middle_crc);
  1298. return -EBADMSG;
  1299. }
  1300. if (datacrc &&
  1301. (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
  1302. con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
  1303. pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
  1304. con->in_data_crc, le32_to_cpu(m->footer.data_crc));
  1305. return -EBADMSG;
  1306. }
  1307. return 1; /* done! */
  1308. }
  1309. /*
  1310. * Process message. This happens in the worker thread. The callback should
  1311. * be careful not to do anything that waits on other incoming messages or it
  1312. * may deadlock.
  1313. */
  1314. static void process_message(struct ceph_connection *con)
  1315. {
  1316. struct ceph_msg *msg;
  1317. msg = con->in_msg;
  1318. con->in_msg = NULL;
  1319. /* if first message, set peer_name */
  1320. if (con->peer_name.type == 0)
  1321. con->peer_name = msg->hdr.src;
  1322. con->in_seq++;
  1323. mutex_unlock(&con->mutex);
  1324. dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
  1325. msg, le64_to_cpu(msg->hdr.seq),
  1326. ENTITY_NAME(msg->hdr.src),
  1327. le16_to_cpu(msg->hdr.type),
  1328. ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
  1329. le32_to_cpu(msg->hdr.front_len),
  1330. le32_to_cpu(msg->hdr.data_len),
  1331. con->in_front_crc, con->in_middle_crc, con->in_data_crc);
  1332. con->ops->dispatch(con, msg);
  1333. mutex_lock(&con->mutex);
  1334. prepare_read_tag(con);
  1335. }
  1336. /*
  1337. * Write something to the socket. Called in a worker thread when the
  1338. * socket appears to be writeable and we have something ready to send.
  1339. */
  1340. static int try_write(struct ceph_connection *con)
  1341. {
  1342. struct ceph_messenger *msgr = con->msgr;
  1343. int ret = 1;
  1344. dout("try_write start %p state %lu nref %d\n", con, con->state,
  1345. atomic_read(&con->nref));
  1346. more:
  1347. dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
  1348. /* open the socket first? */
  1349. if (con->sock == NULL) {
  1350. /*
  1351. * if we were STANDBY and are reconnecting _this_
  1352. * connection, bump connect_seq now. Always bump
  1353. * global_seq.
  1354. */
  1355. if (test_and_clear_bit(STANDBY, &con->state))
  1356. con->connect_seq++;
  1357. prepare_write_banner(msgr, con);
  1358. prepare_write_connect(msgr, con, 1);
  1359. prepare_read_banner(con);
  1360. set_bit(CONNECTING, &con->state);
  1361. clear_bit(NEGOTIATING, &con->state);
  1362. BUG_ON(con->in_msg);
  1363. con->in_tag = CEPH_MSGR_TAG_READY;
  1364. dout("try_write initiating connect on %p new state %lu\n",
  1365. con, con->state);
  1366. con->sock = ceph_tcp_connect(con);
  1367. if (IS_ERR(con->sock)) {
  1368. con->sock = NULL;
  1369. con->error_msg = "connect error";
  1370. ret = -1;
  1371. goto out;
  1372. }
  1373. }
  1374. more_kvec:
  1375. /* kvec data queued? */
  1376. if (con->out_skip) {
  1377. ret = write_partial_skip(con);
  1378. if (ret <= 0)
  1379. goto done;
  1380. if (ret < 0) {
  1381. dout("try_write write_partial_skip err %d\n", ret);
  1382. goto done;
  1383. }
  1384. }
  1385. if (con->out_kvec_left) {
  1386. ret = write_partial_kvec(con);
  1387. if (ret <= 0)
  1388. goto done;
  1389. }
  1390. /* msg pages? */
  1391. if (con->out_msg) {
  1392. if (con->out_msg_done) {
  1393. ceph_msg_put(con->out_msg);
  1394. con->out_msg = NULL; /* we're done with this one */
  1395. goto do_next;
  1396. }
  1397. ret = write_partial_msg_pages(con);
  1398. if (ret == 1)
  1399. goto more_kvec; /* we need to send the footer, too! */
  1400. if (ret == 0)
  1401. goto done;
  1402. if (ret < 0) {
  1403. dout("try_write write_partial_msg_pages err %d\n",
  1404. ret);
  1405. goto done;
  1406. }
  1407. }
  1408. do_next:
  1409. if (!test_bit(CONNECTING, &con->state)) {
  1410. /* is anything else pending? */
  1411. if (!list_empty(&con->out_queue)) {
  1412. prepare_write_message(con);
  1413. goto more;
  1414. }
  1415. if (con->in_seq > con->in_seq_acked) {
  1416. prepare_write_ack(con);
  1417. goto more;
  1418. }
  1419. if (test_and_clear_bit(KEEPALIVE_PENDING, &con->state)) {
  1420. prepare_write_keepalive(con);
  1421. goto more;
  1422. }
  1423. }
  1424. /* Nothing to do! */
  1425. clear_bit(WRITE_PENDING, &con->state);
  1426. dout("try_write nothing else to write.\n");
  1427. done:
  1428. ret = 0;
  1429. out:
  1430. dout("try_write done on %p\n", con);
  1431. return ret;
  1432. }
  1433. /*
  1434. * Read what we can from the socket.
  1435. */
  1436. static int try_read(struct ceph_connection *con)
  1437. {
  1438. int ret = -1;
  1439. if (!con->sock)
  1440. return 0;
  1441. if (test_bit(STANDBY, &con->state))
  1442. return 0;
  1443. dout("try_read start on %p\n", con);
  1444. more:
  1445. dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
  1446. con->in_base_pos);
  1447. if (test_bit(CONNECTING, &con->state)) {
  1448. if (!test_bit(NEGOTIATING, &con->state)) {
  1449. dout("try_read connecting\n");
  1450. ret = read_partial_banner(con);
  1451. if (ret <= 0)
  1452. goto done;
  1453. if (process_banner(con) < 0) {
  1454. ret = -1;
  1455. goto out;
  1456. }
  1457. }
  1458. ret = read_partial_connect(con);
  1459. if (ret <= 0)
  1460. goto done;
  1461. if (process_connect(con) < 0) {
  1462. ret = -1;
  1463. goto out;
  1464. }
  1465. goto more;
  1466. }
  1467. if (con->in_base_pos < 0) {
  1468. /*
  1469. * skipping + discarding content.
  1470. *
  1471. * FIXME: there must be a better way to do this!
  1472. */
  1473. static char buf[1024];
  1474. int skip = min(1024, -con->in_base_pos);
  1475. dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
  1476. ret = ceph_tcp_recvmsg(con->sock, buf, skip);
  1477. if (ret <= 0)
  1478. goto done;
  1479. con->in_base_pos += ret;
  1480. if (con->in_base_pos)
  1481. goto more;
  1482. }
  1483. if (con->in_tag == CEPH_MSGR_TAG_READY) {
  1484. /*
  1485. * what's next?
  1486. */
  1487. ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
  1488. if (ret <= 0)
  1489. goto done;
  1490. dout("try_read got tag %d\n", (int)con->in_tag);
  1491. switch (con->in_tag) {
  1492. case CEPH_MSGR_TAG_MSG:
  1493. prepare_read_message(con);
  1494. break;
  1495. case CEPH_MSGR_TAG_ACK:
  1496. prepare_read_ack(con);
  1497. break;
  1498. case CEPH_MSGR_TAG_CLOSE:
  1499. set_bit(CLOSED, &con->state); /* fixme */
  1500. goto done;
  1501. default:
  1502. goto bad_tag;
  1503. }
  1504. }
  1505. if (con->in_tag == CEPH_MSGR_TAG_MSG) {
  1506. ret = read_partial_message(con);
  1507. if (ret <= 0) {
  1508. switch (ret) {
  1509. case -EBADMSG:
  1510. con->error_msg = "bad crc";
  1511. ret = -EIO;
  1512. goto out;
  1513. case -EIO:
  1514. con->error_msg = "io error";
  1515. goto out;
  1516. default:
  1517. goto done;
  1518. }
  1519. }
  1520. if (con->in_tag == CEPH_MSGR_TAG_READY)
  1521. goto more;
  1522. process_message(con);
  1523. goto more;
  1524. }
  1525. if (con->in_tag == CEPH_MSGR_TAG_ACK) {
  1526. ret = read_partial_ack(con);
  1527. if (ret <= 0)
  1528. goto done;
  1529. process_ack(con);
  1530. goto more;
  1531. }
  1532. done:
  1533. ret = 0;
  1534. out:
  1535. dout("try_read done on %p\n", con);
  1536. return ret;
  1537. bad_tag:
  1538. pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
  1539. con->error_msg = "protocol error, garbage tag";
  1540. ret = -1;
  1541. goto out;
  1542. }
  1543. /*
  1544. * Atomically queue work on a connection. Bump @con reference to
  1545. * avoid races with connection teardown.
  1546. *
  1547. * There is some trickery going on with QUEUED and BUSY because we
  1548. * only want a _single_ thread operating on each connection at any
  1549. * point in time, but we want to use all available CPUs.
  1550. *
  1551. * The worker thread only proceeds if it can atomically set BUSY. It
  1552. * clears QUEUED and does it's thing. When it thinks it's done, it
  1553. * clears BUSY, then rechecks QUEUED.. if it's set again, it loops
  1554. * (tries again to set BUSY).
  1555. *
  1556. * To queue work, we first set QUEUED, _then_ if BUSY isn't set, we
  1557. * try to queue work. If that fails (work is already queued, or BUSY)
  1558. * we give up (work also already being done or is queued) but leave QUEUED
  1559. * set so that the worker thread will loop if necessary.
  1560. */
  1561. static void queue_con(struct ceph_connection *con)
  1562. {
  1563. if (test_bit(DEAD, &con->state)) {
  1564. dout("queue_con %p ignoring: DEAD\n",
  1565. con);
  1566. return;
  1567. }
  1568. if (!con->ops->get(con)) {
  1569. dout("queue_con %p ref count 0\n", con);
  1570. return;
  1571. }
  1572. set_bit(QUEUED, &con->state);
  1573. if (test_bit(BUSY, &con->state)) {
  1574. dout("queue_con %p - already BUSY\n", con);
  1575. con->ops->put(con);
  1576. } else if (!queue_work(ceph_msgr_wq, &con->work.work)) {
  1577. dout("queue_con %p - already queued\n", con);
  1578. con->ops->put(con);
  1579. } else {
  1580. dout("queue_con %p\n", con);
  1581. }
  1582. }
  1583. /*
  1584. * Do some work on a connection. Drop a connection ref when we're done.
  1585. */
  1586. static void con_work(struct work_struct *work)
  1587. {
  1588. struct ceph_connection *con = container_of(work, struct ceph_connection,
  1589. work.work);
  1590. int backoff = 0;
  1591. more:
  1592. if (test_and_set_bit(BUSY, &con->state) != 0) {
  1593. dout("con_work %p BUSY already set\n", con);
  1594. goto out;
  1595. }
  1596. dout("con_work %p start, clearing QUEUED\n", con);
  1597. clear_bit(QUEUED, &con->state);
  1598. mutex_lock(&con->mutex);
  1599. if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
  1600. dout("con_work CLOSED\n");
  1601. con_close_socket(con);
  1602. goto done;
  1603. }
  1604. if (test_and_clear_bit(OPENING, &con->state)) {
  1605. /* reopen w/ new peer */
  1606. dout("con_work OPENING\n");
  1607. con_close_socket(con);
  1608. }
  1609. if (test_and_clear_bit(SOCK_CLOSED, &con->state) ||
  1610. try_read(con) < 0 ||
  1611. try_write(con) < 0) {
  1612. mutex_unlock(&con->mutex);
  1613. backoff = 1;
  1614. ceph_fault(con); /* error/fault path */
  1615. goto done_unlocked;
  1616. }
  1617. done:
  1618. mutex_unlock(&con->mutex);
  1619. done_unlocked:
  1620. clear_bit(BUSY, &con->state);
  1621. dout("con->state=%lu\n", con->state);
  1622. if (test_bit(QUEUED, &con->state)) {
  1623. if (!backoff || test_bit(OPENING, &con->state)) {
  1624. dout("con_work %p QUEUED reset, looping\n", con);
  1625. goto more;
  1626. }
  1627. dout("con_work %p QUEUED reset, but just faulted\n", con);
  1628. clear_bit(QUEUED, &con->state);
  1629. }
  1630. dout("con_work %p done\n", con);
  1631. out:
  1632. con->ops->put(con);
  1633. }
  1634. /*
  1635. * Generic error/fault handler. A retry mechanism is used with
  1636. * exponential backoff
  1637. */
  1638. static void ceph_fault(struct ceph_connection *con)
  1639. {
  1640. pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
  1641. pr_addr(&con->peer_addr.in_addr), con->error_msg);
  1642. dout("fault %p state %lu to peer %s\n",
  1643. con, con->state, pr_addr(&con->peer_addr.in_addr));
  1644. if (test_bit(LOSSYTX, &con->state)) {
  1645. dout("fault on LOSSYTX channel\n");
  1646. goto out;
  1647. }
  1648. mutex_lock(&con->mutex);
  1649. if (test_bit(CLOSED, &con->state))
  1650. goto out_unlock;
  1651. con_close_socket(con);
  1652. if (con->in_msg) {
  1653. ceph_msg_put(con->in_msg);
  1654. con->in_msg = NULL;
  1655. }
  1656. /* Requeue anything that hasn't been acked */
  1657. list_splice_init(&con->out_sent, &con->out_queue);
  1658. /* If there are no messages in the queue, place the connection
  1659. * in a STANDBY state (i.e., don't try to reconnect just yet). */
  1660. if (list_empty(&con->out_queue) && !con->out_keepalive_pending) {
  1661. dout("fault setting STANDBY\n");
  1662. set_bit(STANDBY, &con->state);
  1663. } else {
  1664. /* retry after a delay. */
  1665. if (con->delay == 0)
  1666. con->delay = BASE_DELAY_INTERVAL;
  1667. else if (con->delay < MAX_DELAY_INTERVAL)
  1668. con->delay *= 2;
  1669. dout("fault queueing %p delay %lu\n", con, con->delay);
  1670. con->ops->get(con);
  1671. if (queue_delayed_work(ceph_msgr_wq, &con->work,
  1672. round_jiffies_relative(con->delay)) == 0)
  1673. con->ops->put(con);
  1674. }
  1675. out_unlock:
  1676. mutex_unlock(&con->mutex);
  1677. out:
  1678. /*
  1679. * in case we faulted due to authentication, invalidate our
  1680. * current tickets so that we can get new ones.
  1681. */
  1682. if (con->auth_retry && con->ops->invalidate_authorizer) {
  1683. dout("calling invalidate_authorizer()\n");
  1684. con->ops->invalidate_authorizer(con);
  1685. }
  1686. if (con->ops->fault)
  1687. con->ops->fault(con);
  1688. }
  1689. /*
  1690. * create a new messenger instance
  1691. */
  1692. struct ceph_messenger *ceph_messenger_create(struct ceph_entity_addr *myaddr)
  1693. {
  1694. struct ceph_messenger *msgr;
  1695. msgr = kzalloc(sizeof(*msgr), GFP_KERNEL);
  1696. if (msgr == NULL)
  1697. return ERR_PTR(-ENOMEM);
  1698. spin_lock_init(&msgr->global_seq_lock);
  1699. /* the zero page is needed if a request is "canceled" while the message
  1700. * is being written over the socket */
  1701. msgr->zero_page = __page_cache_alloc(GFP_KERNEL | __GFP_ZERO);
  1702. if (!msgr->zero_page) {
  1703. kfree(msgr);
  1704. return ERR_PTR(-ENOMEM);
  1705. }
  1706. kmap(msgr->zero_page);
  1707. if (myaddr)
  1708. msgr->inst.addr = *myaddr;
  1709. /* select a random nonce */
  1710. msgr->inst.addr.type = 0;
  1711. get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
  1712. encode_my_addr(msgr);
  1713. dout("messenger_create %p\n", msgr);
  1714. return msgr;
  1715. }
  1716. void ceph_messenger_destroy(struct ceph_messenger *msgr)
  1717. {
  1718. dout("destroy %p\n", msgr);
  1719. kunmap(msgr->zero_page);
  1720. __free_page(msgr->zero_page);
  1721. kfree(msgr);
  1722. dout("destroyed messenger %p\n", msgr);
  1723. }
  1724. /*
  1725. * Queue up an outgoing message on the given connection.
  1726. */
  1727. void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
  1728. {
  1729. if (test_bit(CLOSED, &con->state)) {
  1730. dout("con_send %p closed, dropping %p\n", con, msg);
  1731. ceph_msg_put(msg);
  1732. return;
  1733. }
  1734. /* set src+dst */
  1735. msg->hdr.src = con->msgr->inst.name;
  1736. BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
  1737. msg->needs_out_seq = true;
  1738. /* queue */
  1739. mutex_lock(&con->mutex);
  1740. BUG_ON(!list_empty(&msg->list_head));
  1741. list_add_tail(&msg->list_head, &con->out_queue);
  1742. dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
  1743. ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
  1744. ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
  1745. le32_to_cpu(msg->hdr.front_len),
  1746. le32_to_cpu(msg->hdr.middle_len),
  1747. le32_to_cpu(msg->hdr.data_len));
  1748. mutex_unlock(&con->mutex);
  1749. /* if there wasn't anything waiting to send before, queue
  1750. * new work */
  1751. if (test_and_set_bit(WRITE_PENDING, &con->state) == 0)
  1752. queue_con(con);
  1753. }
  1754. /*
  1755. * Revoke a message that was previously queued for send
  1756. */
  1757. void ceph_con_revoke(struct ceph_connection *con, struct ceph_msg *msg)
  1758. {
  1759. mutex_lock(&con->mutex);
  1760. if (!list_empty(&msg->list_head)) {
  1761. dout("con_revoke %p msg %p\n", con, msg);
  1762. list_del_init(&msg->list_head);
  1763. ceph_msg_put(msg);
  1764. msg->hdr.seq = 0;
  1765. if (con->out_msg == msg) {
  1766. ceph_msg_put(con->out_msg);
  1767. con->out_msg = NULL;
  1768. }
  1769. if (con->out_kvec_is_msg) {
  1770. con->out_skip = con->out_kvec_bytes;
  1771. con->out_kvec_is_msg = false;
  1772. }
  1773. } else {
  1774. dout("con_revoke %p msg %p - not queued (sent?)\n", con, msg);
  1775. }
  1776. mutex_unlock(&con->mutex);
  1777. }
  1778. /*
  1779. * Revoke a message that we may be reading data into
  1780. */
  1781. void ceph_con_revoke_message(struct ceph_connection *con, struct ceph_msg *msg)
  1782. {
  1783. mutex_lock(&con->mutex);
  1784. if (con->in_msg && con->in_msg == msg) {
  1785. unsigned front_len = le32_to_cpu(con->in_hdr.front_len);
  1786. unsigned middle_len = le32_to_cpu(con->in_hdr.middle_len);
  1787. unsigned data_len = le32_to_cpu(con->in_hdr.data_len);
  1788. /* skip rest of message */
  1789. dout("con_revoke_pages %p msg %p revoked\n", con, msg);
  1790. con->in_base_pos = con->in_base_pos -
  1791. sizeof(struct ceph_msg_header) -
  1792. front_len -
  1793. middle_len -
  1794. data_len -
  1795. sizeof(struct ceph_msg_footer);
  1796. ceph_msg_put(con->in_msg);
  1797. con->in_msg = NULL;
  1798. con->in_tag = CEPH_MSGR_TAG_READY;
  1799. con->in_seq++;
  1800. } else {
  1801. dout("con_revoke_pages %p msg %p pages %p no-op\n",
  1802. con, con->in_msg, msg);
  1803. }
  1804. mutex_unlock(&con->mutex);
  1805. }
  1806. /*
  1807. * Queue a keepalive byte to ensure the tcp connection is alive.
  1808. */
  1809. void ceph_con_keepalive(struct ceph_connection *con)
  1810. {
  1811. if (test_and_set_bit(KEEPALIVE_PENDING, &con->state) == 0 &&
  1812. test_and_set_bit(WRITE_PENDING, &con->state) == 0)
  1813. queue_con(con);
  1814. }
  1815. /*
  1816. * construct a new message with given type, size
  1817. * the new msg has a ref count of 1.
  1818. */
  1819. struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags)
  1820. {
  1821. struct ceph_msg *m;
  1822. m = kmalloc(sizeof(*m), flags);
  1823. if (m == NULL)
  1824. goto out;
  1825. kref_init(&m->kref);
  1826. INIT_LIST_HEAD(&m->list_head);
  1827. m->hdr.tid = 0;
  1828. m->hdr.type = cpu_to_le16(type);
  1829. m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
  1830. m->hdr.version = 0;
  1831. m->hdr.front_len = cpu_to_le32(front_len);
  1832. m->hdr.middle_len = 0;
  1833. m->hdr.data_len = 0;
  1834. m->hdr.data_off = 0;
  1835. m->hdr.reserved = 0;
  1836. m->footer.front_crc = 0;
  1837. m->footer.middle_crc = 0;
  1838. m->footer.data_crc = 0;
  1839. m->footer.flags = 0;
  1840. m->front_max = front_len;
  1841. m->front_is_vmalloc = false;
  1842. m->more_to_follow = false;
  1843. m->pool = NULL;
  1844. /* front */
  1845. if (front_len) {
  1846. if (front_len > PAGE_CACHE_SIZE) {
  1847. m->front.iov_base = __vmalloc(front_len, flags,
  1848. PAGE_KERNEL);
  1849. m->front_is_vmalloc = true;
  1850. } else {
  1851. m->front.iov_base = kmalloc(front_len, flags);
  1852. }
  1853. if (m->front.iov_base == NULL) {
  1854. pr_err("msg_new can't allocate %d bytes\n",
  1855. front_len);
  1856. goto out2;
  1857. }
  1858. } else {
  1859. m->front.iov_base = NULL;
  1860. }
  1861. m->front.iov_len = front_len;
  1862. /* middle */
  1863. m->middle = NULL;
  1864. /* data */
  1865. m->nr_pages = 0;
  1866. m->pages = NULL;
  1867. m->pagelist = NULL;
  1868. dout("ceph_msg_new %p front %d\n", m, front_len);
  1869. return m;
  1870. out2:
  1871. ceph_msg_put(m);
  1872. out:
  1873. pr_err("msg_new can't create type %d front %d\n", type, front_len);
  1874. return NULL;
  1875. }
  1876. /*
  1877. * Allocate "middle" portion of a message, if it is needed and wasn't
  1878. * allocated by alloc_msg. This allows us to read a small fixed-size
  1879. * per-type header in the front and then gracefully fail (i.e.,
  1880. * propagate the error to the caller based on info in the front) when
  1881. * the middle is too large.
  1882. */
  1883. static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
  1884. {
  1885. int type = le16_to_cpu(msg->hdr.type);
  1886. int middle_len = le32_to_cpu(msg->hdr.middle_len);
  1887. dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
  1888. ceph_msg_type_name(type), middle_len);
  1889. BUG_ON(!middle_len);
  1890. BUG_ON(msg->middle);
  1891. msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
  1892. if (!msg->middle)
  1893. return -ENOMEM;
  1894. return 0;
  1895. }
  1896. /*
  1897. * Generic message allocator, for incoming messages.
  1898. */
  1899. static struct ceph_msg *ceph_alloc_msg(struct ceph_connection *con,
  1900. struct ceph_msg_header *hdr,
  1901. int *skip)
  1902. {
  1903. int type = le16_to_cpu(hdr->type);
  1904. int front_len = le32_to_cpu(hdr->front_len);
  1905. int middle_len = le32_to_cpu(hdr->middle_len);
  1906. struct ceph_msg *msg = NULL;
  1907. int ret;
  1908. if (con->ops->alloc_msg) {
  1909. mutex_unlock(&con->mutex);
  1910. msg = con->ops->alloc_msg(con, hdr, skip);
  1911. mutex_lock(&con->mutex);
  1912. if (!msg || *skip)
  1913. return NULL;
  1914. }
  1915. if (!msg) {
  1916. *skip = 0;
  1917. msg = ceph_msg_new(type, front_len, GFP_NOFS);
  1918. if (!msg) {
  1919. pr_err("unable to allocate msg type %d len %d\n",
  1920. type, front_len);
  1921. return NULL;
  1922. }
  1923. }
  1924. memcpy(&msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
  1925. if (middle_len && !msg->middle) {
  1926. ret = ceph_alloc_middle(con, msg);
  1927. if (ret < 0) {
  1928. ceph_msg_put(msg);
  1929. return NULL;
  1930. }
  1931. }
  1932. return msg;
  1933. }
  1934. /*
  1935. * Free a generically kmalloc'd message.
  1936. */
  1937. void ceph_msg_kfree(struct ceph_msg *m)
  1938. {
  1939. dout("msg_kfree %p\n", m);
  1940. if (m->front_is_vmalloc)
  1941. vfree(m->front.iov_base);
  1942. else
  1943. kfree(m->front.iov_base);
  1944. kfree(m);
  1945. }
  1946. /*
  1947. * Drop a msg ref. Destroy as needed.
  1948. */
  1949. void ceph_msg_last_put(struct kref *kref)
  1950. {
  1951. struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
  1952. dout("ceph_msg_put last one on %p\n", m);
  1953. WARN_ON(!list_empty(&m->list_head));
  1954. /* drop middle, data, if any */
  1955. if (m->middle) {
  1956. ceph_buffer_put(m->middle);
  1957. m->middle = NULL;
  1958. }
  1959. m->nr_pages = 0;
  1960. m->pages = NULL;
  1961. if (m->pagelist) {
  1962. ceph_pagelist_release(m->pagelist);
  1963. kfree(m->pagelist);
  1964. m->pagelist = NULL;
  1965. }
  1966. if (m->pool)
  1967. ceph_msgpool_put(m->pool, m);
  1968. else
  1969. ceph_msg_kfree(m);
  1970. }
  1971. void ceph_msg_dump(struct ceph_msg *msg)
  1972. {
  1973. pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
  1974. msg->front_max, msg->nr_pages);
  1975. print_hex_dump(KERN_DEBUG, "header: ",
  1976. DUMP_PREFIX_OFFSET, 16, 1,
  1977. &msg->hdr, sizeof(msg->hdr), true);
  1978. print_hex_dump(KERN_DEBUG, " front: ",
  1979. DUMP_PREFIX_OFFSET, 16, 1,
  1980. msg->front.iov_base, msg->front.iov_len, true);
  1981. if (msg->middle)
  1982. print_hex_dump(KERN_DEBUG, "middle: ",
  1983. DUMP_PREFIX_OFFSET, 16, 1,
  1984. msg->middle->vec.iov_base,
  1985. msg->middle->vec.iov_len, true);
  1986. print_hex_dump(KERN_DEBUG, "footer: ",
  1987. DUMP_PREFIX_OFFSET, 16, 1,
  1988. &msg->footer, sizeof(msg->footer), true);
  1989. }