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