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