b59e64a7b24bfb22bbe1f5f70a8fcfb4c79e9b09
[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                 prepare_read_tag(con);
1633                 break;
1634
1635         case CEPH_MSGR_TAG_WAIT:
1636                 /*
1637                  * If there is a connection race (we are opening
1638                  * connections to each other), one of us may just have
1639                  * to WAIT.  This shouldn't happen if we are the
1640                  * client.
1641                  */
1642                 pr_err("process_connect got WAIT as client\n");
1643                 con->error_msg = "protocol error, got WAIT as client";
1644                 return -1;
1645
1646         default:
1647                 pr_err("connect protocol error, will retry\n");
1648                 con->error_msg = "protocol error, garbage tag during connect";
1649                 return -1;
1650         }
1651         return 0;
1652 }
1653
1654
1655 /*
1656  * read (part of) an ack
1657  */
1658 static int read_partial_ack(struct ceph_connection *con)
1659 {
1660         int size = sizeof (con->in_temp_ack);
1661         int end = size;
1662
1663         return read_partial(con, end, size, &con->in_temp_ack);
1664 }
1665
1666
1667 /*
1668  * We can finally discard anything that's been acked.
1669  */
1670 static void process_ack(struct ceph_connection *con)
1671 {
1672         struct ceph_msg *m;
1673         u64 ack = le64_to_cpu(con->in_temp_ack);
1674         u64 seq;
1675
1676         while (!list_empty(&con->out_sent)) {
1677                 m = list_first_entry(&con->out_sent, struct ceph_msg,
1678                                      list_head);
1679                 seq = le64_to_cpu(m->hdr.seq);
1680                 if (seq > ack)
1681                         break;
1682                 dout("got ack for seq %llu type %d at %p\n", seq,
1683                      le16_to_cpu(m->hdr.type), m);
1684                 m->ack_stamp = jiffies;
1685                 ceph_msg_remove(m);
1686         }
1687         prepare_read_tag(con);
1688 }
1689
1690
1691
1692
1693 static int read_partial_message_section(struct ceph_connection *con,
1694                                         struct kvec *section,
1695                                         unsigned int sec_len, u32 *crc)
1696 {
1697         int ret, left;
1698
1699         BUG_ON(!section);
1700
1701         while (section->iov_len < sec_len) {
1702                 BUG_ON(section->iov_base == NULL);
1703                 left = sec_len - section->iov_len;
1704                 ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
1705                                        section->iov_len, left);
1706                 if (ret <= 0)
1707                         return ret;
1708                 section->iov_len += ret;
1709         }
1710         if (section->iov_len == sec_len)
1711                 *crc = crc32c(0, section->iov_base, section->iov_len);
1712
1713         return 1;
1714 }
1715
1716 static bool ceph_con_in_msg_alloc(struct ceph_connection *con,
1717                                 struct ceph_msg_header *hdr);
1718
1719
1720 static int read_partial_message_pages(struct ceph_connection *con,
1721                                       struct page **pages,
1722                                       unsigned data_len, bool do_datacrc)
1723 {
1724         void *p;
1725         int ret;
1726         int left;
1727
1728         left = min((int)(data_len - con->in_msg_pos.data_pos),
1729                    (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
1730         /* (page) data */
1731         BUG_ON(pages == NULL);
1732         p = kmap(pages[con->in_msg_pos.page]);
1733         ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1734                                left);
1735         if (ret > 0 && do_datacrc)
1736                 con->in_data_crc =
1737                         crc32c(con->in_data_crc,
1738                                   p + con->in_msg_pos.page_pos, ret);
1739         kunmap(pages[con->in_msg_pos.page]);
1740         if (ret <= 0)
1741                 return ret;
1742         con->in_msg_pos.data_pos += ret;
1743         con->in_msg_pos.page_pos += ret;
1744         if (con->in_msg_pos.page_pos == PAGE_SIZE) {
1745                 con->in_msg_pos.page_pos = 0;
1746                 con->in_msg_pos.page++;
1747         }
1748
1749         return ret;
1750 }
1751
1752 #ifdef CONFIG_BLOCK
1753 static int read_partial_message_bio(struct ceph_connection *con,
1754                                     struct bio **bio_iter, int *bio_seg,
1755                                     unsigned data_len, bool do_datacrc)
1756 {
1757         struct bio_vec *bv = bio_iovec_idx(*bio_iter, *bio_seg);
1758         void *p;
1759         int ret, left;
1760
1761         left = min((int)(data_len - con->in_msg_pos.data_pos),
1762                    (int)(bv->bv_len - con->in_msg_pos.page_pos));
1763
1764         p = kmap(bv->bv_page) + bv->bv_offset;
1765
1766         ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1767                                left);
1768         if (ret > 0 && do_datacrc)
1769                 con->in_data_crc =
1770                         crc32c(con->in_data_crc,
1771                                   p + con->in_msg_pos.page_pos, ret);
1772         kunmap(bv->bv_page);
1773         if (ret <= 0)
1774                 return ret;
1775         con->in_msg_pos.data_pos += ret;
1776         con->in_msg_pos.page_pos += ret;
1777         if (con->in_msg_pos.page_pos == bv->bv_len) {
1778                 con->in_msg_pos.page_pos = 0;
1779                 iter_bio_next(bio_iter, bio_seg);
1780         }
1781
1782         return ret;
1783 }
1784 #endif
1785
1786 /*
1787  * read (part of) a message.
1788  */
1789 static int read_partial_message(struct ceph_connection *con)
1790 {
1791         struct ceph_msg *m = con->in_msg;
1792         int size;
1793         int end;
1794         int ret;
1795         unsigned front_len, middle_len, data_len;
1796         bool do_datacrc = !con->msgr->nocrc;
1797         u64 seq;
1798         u32 crc;
1799
1800         dout("read_partial_message con %p msg %p\n", con, m);
1801
1802         /* header */
1803         size = sizeof (con->in_hdr);
1804         end = size;
1805         ret = read_partial(con, end, size, &con->in_hdr);
1806         if (ret <= 0)
1807                 return ret;
1808
1809         crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
1810         if (cpu_to_le32(crc) != con->in_hdr.crc) {
1811                 pr_err("read_partial_message bad hdr "
1812                        " crc %u != expected %u\n",
1813                        crc, con->in_hdr.crc);
1814                 return -EBADMSG;
1815         }
1816
1817         front_len = le32_to_cpu(con->in_hdr.front_len);
1818         if (front_len > CEPH_MSG_MAX_FRONT_LEN)
1819                 return -EIO;
1820         middle_len = le32_to_cpu(con->in_hdr.middle_len);
1821         if (middle_len > CEPH_MSG_MAX_DATA_LEN)
1822                 return -EIO;
1823         data_len = le32_to_cpu(con->in_hdr.data_len);
1824         if (data_len > CEPH_MSG_MAX_DATA_LEN)
1825                 return -EIO;
1826
1827         /* verify seq# */
1828         seq = le64_to_cpu(con->in_hdr.seq);
1829         if ((s64)seq - (s64)con->in_seq < 1) {
1830                 pr_info("skipping %s%lld %s seq %lld expected %lld\n",
1831                         ENTITY_NAME(con->peer_name),
1832                         ceph_pr_addr(&con->peer_addr.in_addr),
1833                         seq, con->in_seq + 1);
1834                 con->in_base_pos = -front_len - middle_len - data_len -
1835                         sizeof(m->footer);
1836                 con->in_tag = CEPH_MSGR_TAG_READY;
1837                 return 0;
1838         } else if ((s64)seq - (s64)con->in_seq > 1) {
1839                 pr_err("read_partial_message bad seq %lld expected %lld\n",
1840                        seq, con->in_seq + 1);
1841                 con->error_msg = "bad message sequence # for incoming message";
1842                 return -EBADMSG;
1843         }
1844
1845         /* allocate message? */
1846         if (!con->in_msg) {
1847                 dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
1848                      con->in_hdr.front_len, con->in_hdr.data_len);
1849                 if (ceph_con_in_msg_alloc(con, &con->in_hdr)) {
1850                         /* skip this message */
1851                         dout("alloc_msg said skip message\n");
1852                         BUG_ON(con->in_msg);
1853                         con->in_base_pos = -front_len - middle_len - data_len -
1854                                 sizeof(m->footer);
1855                         con->in_tag = CEPH_MSGR_TAG_READY;
1856                         con->in_seq++;
1857                         return 0;
1858                 }
1859                 if (!con->in_msg) {
1860                         con->error_msg =
1861                                 "error allocating memory for incoming message";
1862                         return -ENOMEM;
1863                 }
1864
1865                 BUG_ON(con->in_msg->con != con);
1866                 m = con->in_msg;
1867                 m->front.iov_len = 0;    /* haven't read it yet */
1868                 if (m->middle)
1869                         m->middle->vec.iov_len = 0;
1870
1871                 con->in_msg_pos.page = 0;
1872                 if (m->pages)
1873                         con->in_msg_pos.page_pos = m->page_alignment;
1874                 else
1875                         con->in_msg_pos.page_pos = 0;
1876                 con->in_msg_pos.data_pos = 0;
1877
1878 #ifdef CONFIG_BLOCK
1879                 if (m->bio)
1880                         init_bio_iter(m->bio, &m->bio_iter, &m->bio_seg);
1881 #endif
1882         }
1883
1884         /* front */
1885         ret = read_partial_message_section(con, &m->front, front_len,
1886                                            &con->in_front_crc);
1887         if (ret <= 0)
1888                 return ret;
1889
1890         /* middle */
1891         if (m->middle) {
1892                 ret = read_partial_message_section(con, &m->middle->vec,
1893                                                    middle_len,
1894                                                    &con->in_middle_crc);
1895                 if (ret <= 0)
1896                         return ret;
1897         }
1898
1899         /* (page) data */
1900         while (con->in_msg_pos.data_pos < data_len) {
1901                 if (m->pages) {
1902                         ret = read_partial_message_pages(con, m->pages,
1903                                                  data_len, do_datacrc);
1904                         if (ret <= 0)
1905                                 return ret;
1906 #ifdef CONFIG_BLOCK
1907                 } else if (m->bio) {
1908                         BUG_ON(!m->bio_iter);
1909                         ret = read_partial_message_bio(con,
1910                                                  &m->bio_iter, &m->bio_seg,
1911                                                  data_len, do_datacrc);
1912                         if (ret <= 0)
1913                                 return ret;
1914 #endif
1915                 } else {
1916                         BUG_ON(1);
1917                 }
1918         }
1919
1920         /* footer */
1921         size = sizeof (m->footer);
1922         end += size;
1923         ret = read_partial(con, end, size, &m->footer);
1924         if (ret <= 0)
1925                 return ret;
1926
1927         dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
1928              m, front_len, m->footer.front_crc, middle_len,
1929              m->footer.middle_crc, data_len, m->footer.data_crc);
1930
1931         /* crc ok? */
1932         if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
1933                 pr_err("read_partial_message %p front crc %u != exp. %u\n",
1934                        m, con->in_front_crc, m->footer.front_crc);
1935                 return -EBADMSG;
1936         }
1937         if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
1938                 pr_err("read_partial_message %p middle crc %u != exp %u\n",
1939                        m, con->in_middle_crc, m->footer.middle_crc);
1940                 return -EBADMSG;
1941         }
1942         if (do_datacrc &&
1943             (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
1944             con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
1945                 pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
1946                        con->in_data_crc, le32_to_cpu(m->footer.data_crc));
1947                 return -EBADMSG;
1948         }
1949
1950         return 1; /* done! */
1951 }
1952
1953 /*
1954  * Process message.  This happens in the worker thread.  The callback should
1955  * be careful not to do anything that waits on other incoming messages or it
1956  * may deadlock.
1957  */
1958 static void process_message(struct ceph_connection *con)
1959 {
1960         struct ceph_msg *msg;
1961
1962         BUG_ON(con->in_msg->con != con);
1963         con->in_msg->con = NULL;
1964         msg = con->in_msg;
1965         con->in_msg = NULL;
1966         con->ops->put(con);
1967
1968         /* if first message, set peer_name */
1969         if (con->peer_name.type == 0)
1970                 con->peer_name = msg->hdr.src;
1971
1972         con->in_seq++;
1973         mutex_unlock(&con->mutex);
1974
1975         dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
1976              msg, le64_to_cpu(msg->hdr.seq),
1977              ENTITY_NAME(msg->hdr.src),
1978              le16_to_cpu(msg->hdr.type),
1979              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
1980              le32_to_cpu(msg->hdr.front_len),
1981              le32_to_cpu(msg->hdr.data_len),
1982              con->in_front_crc, con->in_middle_crc, con->in_data_crc);
1983         con->ops->dispatch(con, msg);
1984
1985         mutex_lock(&con->mutex);
1986         prepare_read_tag(con);
1987 }
1988
1989
1990 /*
1991  * Write something to the socket.  Called in a worker thread when the
1992  * socket appears to be writeable and we have something ready to send.
1993  */
1994 static int try_write(struct ceph_connection *con)
1995 {
1996         int ret = 1;
1997
1998         dout("try_write start %p state %lu\n", con, con->state);
1999
2000 more:
2001         dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
2002
2003         /* open the socket first? */
2004         if (con->sock == NULL) {
2005                 set_bit(CONNECTING, &con->state);
2006
2007                 con_out_kvec_reset(con);
2008                 prepare_write_banner(con);
2009                 prepare_read_banner(con);
2010
2011                 BUG_ON(con->in_msg);
2012                 con->in_tag = CEPH_MSGR_TAG_READY;
2013                 dout("try_write initiating connect on %p new state %lu\n",
2014                      con, con->state);
2015                 ret = ceph_tcp_connect(con);
2016                 if (ret < 0) {
2017                         con->error_msg = "connect error";
2018                         goto out;
2019                 }
2020         }
2021
2022 more_kvec:
2023         /* kvec data queued? */
2024         if (con->out_skip) {
2025                 ret = write_partial_skip(con);
2026                 if (ret <= 0)
2027                         goto out;
2028         }
2029         if (con->out_kvec_left) {
2030                 ret = write_partial_kvec(con);
2031                 if (ret <= 0)
2032                         goto out;
2033         }
2034
2035         /* msg pages? */
2036         if (con->out_msg) {
2037                 if (con->out_msg_done) {
2038                         ceph_msg_put(con->out_msg);
2039                         con->out_msg = NULL;   /* we're done with this one */
2040                         goto do_next;
2041                 }
2042
2043                 ret = write_partial_msg_pages(con);
2044                 if (ret == 1)
2045                         goto more_kvec;  /* we need to send the footer, too! */
2046                 if (ret == 0)
2047                         goto out;
2048                 if (ret < 0) {
2049                         dout("try_write write_partial_msg_pages err %d\n",
2050                              ret);
2051                         goto out;
2052                 }
2053         }
2054
2055 do_next:
2056         if (!test_bit(CONNECTING, &con->state) &&
2057                         !test_bit(NEGOTIATING, &con->state)) {
2058                 /* is anything else pending? */
2059                 if (!list_empty(&con->out_queue)) {
2060                         prepare_write_message(con);
2061                         goto more;
2062                 }
2063                 if (con->in_seq > con->in_seq_acked) {
2064                         prepare_write_ack(con);
2065                         goto more;
2066                 }
2067                 if (test_and_clear_bit(KEEPALIVE_PENDING, &con->flags)) {
2068                         prepare_write_keepalive(con);
2069                         goto more;
2070                 }
2071         }
2072
2073         /* Nothing to do! */
2074         clear_bit(WRITE_PENDING, &con->flags);
2075         dout("try_write nothing else to write.\n");
2076         ret = 0;
2077 out:
2078         dout("try_write done on %p ret %d\n", con, ret);
2079         return ret;
2080 }
2081
2082
2083
2084 /*
2085  * Read what we can from the socket.
2086  */
2087 static int try_read(struct ceph_connection *con)
2088 {
2089         int ret = -1;
2090
2091         if (!con->sock)
2092                 return 0;
2093
2094         if (test_bit(STANDBY, &con->state))
2095                 return 0;
2096
2097         dout("try_read start on %p\n", con);
2098
2099 more:
2100         dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
2101              con->in_base_pos);
2102
2103         /*
2104          * process_connect and process_message drop and re-take
2105          * con->mutex.  make sure we handle a racing close or reopen.
2106          */
2107         if (test_bit(CLOSED, &con->state) ||
2108             test_bit(OPENING, &con->state)) {
2109                 ret = -EAGAIN;
2110                 goto out;
2111         }
2112
2113         if (test_bit(CONNECTING, &con->state)) {
2114                 dout("try_read connecting\n");
2115                 ret = read_partial_banner(con);
2116                 if (ret <= 0)
2117                         goto out;
2118                 ret = process_banner(con);
2119                 if (ret < 0)
2120                         goto out;
2121
2122                 clear_bit(CONNECTING, &con->state);
2123                 set_bit(NEGOTIATING, &con->state);
2124
2125                 /* Banner is good, exchange connection info */
2126                 ret = prepare_write_connect(con);
2127                 if (ret < 0)
2128                         goto out;
2129                 prepare_read_connect(con);
2130
2131                 /* Send connection info before awaiting response */
2132                 goto out;
2133         }
2134
2135         if (test_bit(NEGOTIATING, &con->state)) {
2136                 dout("try_read negotiating\n");
2137                 ret = read_partial_connect(con);
2138                 if (ret <= 0)
2139                         goto out;
2140                 ret = process_connect(con);
2141                 if (ret < 0)
2142                         goto out;
2143                 goto more;
2144         }
2145
2146         if (con->in_base_pos < 0) {
2147                 /*
2148                  * skipping + discarding content.
2149                  *
2150                  * FIXME: there must be a better way to do this!
2151                  */
2152                 static char buf[SKIP_BUF_SIZE];
2153                 int skip = min((int) sizeof (buf), -con->in_base_pos);
2154
2155                 dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
2156                 ret = ceph_tcp_recvmsg(con->sock, buf, skip);
2157                 if (ret <= 0)
2158                         goto out;
2159                 con->in_base_pos += ret;
2160                 if (con->in_base_pos)
2161                         goto more;
2162         }
2163         if (con->in_tag == CEPH_MSGR_TAG_READY) {
2164                 /*
2165                  * what's next?
2166                  */
2167                 ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
2168                 if (ret <= 0)
2169                         goto out;
2170                 dout("try_read got tag %d\n", (int)con->in_tag);
2171                 switch (con->in_tag) {
2172                 case CEPH_MSGR_TAG_MSG:
2173                         prepare_read_message(con);
2174                         break;
2175                 case CEPH_MSGR_TAG_ACK:
2176                         prepare_read_ack(con);
2177                         break;
2178                 case CEPH_MSGR_TAG_CLOSE:
2179                         clear_bit(CONNECTED, &con->state);
2180                         set_bit(CLOSED, &con->state);   /* fixme */
2181                         goto out;
2182                 default:
2183                         goto bad_tag;
2184                 }
2185         }
2186         if (con->in_tag == CEPH_MSGR_TAG_MSG) {
2187                 ret = read_partial_message(con);
2188                 if (ret <= 0) {
2189                         switch (ret) {
2190                         case -EBADMSG:
2191                                 con->error_msg = "bad crc";
2192                                 ret = -EIO;
2193                                 break;
2194                         case -EIO:
2195                                 con->error_msg = "io error";
2196                                 break;
2197                         }
2198                         goto out;
2199                 }
2200                 if (con->in_tag == CEPH_MSGR_TAG_READY)
2201                         goto more;
2202                 process_message(con);
2203                 goto more;
2204         }
2205         if (con->in_tag == CEPH_MSGR_TAG_ACK) {
2206                 ret = read_partial_ack(con);
2207                 if (ret <= 0)
2208                         goto out;
2209                 process_ack(con);
2210                 goto more;
2211         }
2212
2213 out:
2214         dout("try_read done on %p ret %d\n", con, ret);
2215         return ret;
2216
2217 bad_tag:
2218         pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
2219         con->error_msg = "protocol error, garbage tag";
2220         ret = -1;
2221         goto out;
2222 }
2223
2224
2225 /*
2226  * Atomically queue work on a connection.  Bump @con reference to
2227  * avoid races with connection teardown.
2228  */
2229 static void queue_con(struct ceph_connection *con)
2230 {
2231         if (!con->ops->get(con)) {
2232                 dout("queue_con %p ref count 0\n", con);
2233                 return;
2234         }
2235
2236         if (!queue_delayed_work(ceph_msgr_wq, &con->work, 0)) {
2237                 dout("queue_con %p - already queued\n", con);
2238                 con->ops->put(con);
2239         } else {
2240                 dout("queue_con %p\n", con);
2241         }
2242 }
2243
2244 /*
2245  * Do some work on a connection.  Drop a connection ref when we're done.
2246  */
2247 static void con_work(struct work_struct *work)
2248 {
2249         struct ceph_connection *con = container_of(work, struct ceph_connection,
2250                                                    work.work);
2251         int ret;
2252
2253         mutex_lock(&con->mutex);
2254 restart:
2255         if (test_and_clear_bit(SOCK_CLOSED, &con->flags)) {
2256                 if (test_and_clear_bit(CONNECTED, &con->state))
2257                         con->error_msg = "socket closed";
2258                 else if (test_and_clear_bit(NEGOTIATING, &con->state))
2259                         con->error_msg = "negotiation failed";
2260                 else if (test_and_clear_bit(CONNECTING, &con->state))
2261                         con->error_msg = "connection failed";
2262                 else
2263                         con->error_msg = "unrecognized con state";
2264                 goto fault;
2265         }
2266
2267         if (test_and_clear_bit(BACKOFF, &con->flags)) {
2268                 dout("con_work %p backing off\n", con);
2269                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
2270                                        round_jiffies_relative(con->delay))) {
2271                         dout("con_work %p backoff %lu\n", con, con->delay);
2272                         mutex_unlock(&con->mutex);
2273                         return;
2274                 } else {
2275                         con->ops->put(con);
2276                         dout("con_work %p FAILED to back off %lu\n", con,
2277                              con->delay);
2278                 }
2279         }
2280
2281         if (test_bit(STANDBY, &con->state)) {
2282                 dout("con_work %p STANDBY\n", con);
2283                 goto done;
2284         }
2285         if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
2286                 dout("con_work CLOSED\n");
2287                 con_close_socket(con);
2288                 goto done;
2289         }
2290         if (test_and_clear_bit(OPENING, &con->state)) {
2291                 /* reopen w/ new peer */
2292                 dout("con_work OPENING\n");
2293                 con_close_socket(con);
2294         }
2295
2296         ret = try_read(con);
2297         if (ret == -EAGAIN)
2298                 goto restart;
2299         if (ret < 0) {
2300                 con->error_msg = "socket error on read";
2301                 goto fault;
2302         }
2303
2304         ret = try_write(con);
2305         if (ret == -EAGAIN)
2306                 goto restart;
2307         if (ret < 0) {
2308                 con->error_msg = "socket error on write";
2309                 goto fault;
2310         }
2311
2312 done:
2313         mutex_unlock(&con->mutex);
2314 done_unlocked:
2315         con->ops->put(con);
2316         return;
2317
2318 fault:
2319         mutex_unlock(&con->mutex);
2320         ceph_fault(con);     /* error/fault path */
2321         goto done_unlocked;
2322 }
2323
2324
2325 /*
2326  * Generic error/fault handler.  A retry mechanism is used with
2327  * exponential backoff
2328  */
2329 static void ceph_fault(struct ceph_connection *con)
2330 {
2331         pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2332                ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
2333         dout("fault %p state %lu to peer %s\n",
2334              con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
2335
2336         if (test_bit(LOSSYTX, &con->flags)) {
2337                 dout("fault on LOSSYTX channel\n");
2338                 goto out;
2339         }
2340
2341         mutex_lock(&con->mutex);
2342         if (test_bit(CLOSED, &con->state))
2343                 goto out_unlock;
2344
2345         con_close_socket(con);
2346
2347         if (con->in_msg) {
2348                 BUG_ON(con->in_msg->con != con);
2349                 con->in_msg->con = NULL;
2350                 ceph_msg_put(con->in_msg);
2351                 con->in_msg = NULL;
2352                 con->ops->put(con);
2353         }
2354
2355         /* Requeue anything that hasn't been acked */
2356         list_splice_init(&con->out_sent, &con->out_queue);
2357
2358         /* If there are no messages queued or keepalive pending, place
2359          * the connection in a STANDBY state */
2360         if (list_empty(&con->out_queue) &&
2361             !test_bit(KEEPALIVE_PENDING, &con->flags)) {
2362                 dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2363                 clear_bit(WRITE_PENDING, &con->flags);
2364                 set_bit(STANDBY, &con->state);
2365         } else {
2366                 /* retry after a delay. */
2367                 if (con->delay == 0)
2368                         con->delay = BASE_DELAY_INTERVAL;
2369                 else if (con->delay < MAX_DELAY_INTERVAL)
2370                         con->delay *= 2;
2371                 con->ops->get(con);
2372                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
2373                                        round_jiffies_relative(con->delay))) {
2374                         dout("fault queued %p delay %lu\n", con, con->delay);
2375                 } else {
2376                         con->ops->put(con);
2377                         dout("fault failed to queue %p delay %lu, backoff\n",
2378                              con, con->delay);
2379                         /*
2380                          * In many cases we see a socket state change
2381                          * while con_work is running and end up
2382                          * queuing (non-delayed) work, such that we
2383                          * can't backoff with a delay.  Set a flag so
2384                          * that when con_work restarts we schedule the
2385                          * delay then.
2386                          */
2387                         set_bit(BACKOFF, &con->flags);
2388                 }
2389         }
2390
2391 out_unlock:
2392         mutex_unlock(&con->mutex);
2393 out:
2394         /*
2395          * in case we faulted due to authentication, invalidate our
2396          * current tickets so that we can get new ones.
2397          */
2398         if (con->auth_retry && con->ops->invalidate_authorizer) {
2399                 dout("calling invalidate_authorizer()\n");
2400                 con->ops->invalidate_authorizer(con);
2401         }
2402
2403         if (con->ops->fault)
2404                 con->ops->fault(con);
2405 }
2406
2407
2408
2409 /*
2410  * initialize a new messenger instance
2411  */
2412 void ceph_messenger_init(struct ceph_messenger *msgr,
2413                         struct ceph_entity_addr *myaddr,
2414                         u32 supported_features,
2415                         u32 required_features,
2416                         bool nocrc)
2417 {
2418         msgr->supported_features = supported_features;
2419         msgr->required_features = required_features;
2420
2421         spin_lock_init(&msgr->global_seq_lock);
2422
2423         if (myaddr)
2424                 msgr->inst.addr = *myaddr;
2425
2426         /* select a random nonce */
2427         msgr->inst.addr.type = 0;
2428         get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2429         encode_my_addr(msgr);
2430         msgr->nocrc = nocrc;
2431
2432         atomic_set(&msgr->stopping, 0);
2433
2434         dout("%s %p\n", __func__, msgr);
2435 }
2436 EXPORT_SYMBOL(ceph_messenger_init);
2437
2438 static void clear_standby(struct ceph_connection *con)
2439 {
2440         /* come back from STANDBY? */
2441         if (test_and_clear_bit(STANDBY, &con->state)) {
2442                 mutex_lock(&con->mutex);
2443                 dout("clear_standby %p and ++connect_seq\n", con);
2444                 con->connect_seq++;
2445                 WARN_ON(test_bit(WRITE_PENDING, &con->flags));
2446                 WARN_ON(test_bit(KEEPALIVE_PENDING, &con->flags));
2447                 mutex_unlock(&con->mutex);
2448         }
2449 }
2450
2451 /*
2452  * Queue up an outgoing message on the given connection.
2453  */
2454 void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
2455 {
2456         if (test_bit(CLOSED, &con->state)) {
2457                 dout("con_send %p closed, dropping %p\n", con, msg);
2458                 ceph_msg_put(msg);
2459                 return;
2460         }
2461
2462         /* set src+dst */
2463         msg->hdr.src = con->msgr->inst.name;
2464
2465         BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
2466
2467         msg->needs_out_seq = true;
2468
2469         /* queue */
2470         mutex_lock(&con->mutex);
2471
2472         BUG_ON(msg->con != NULL);
2473         msg->con = con->ops->get(con);
2474         BUG_ON(msg->con == NULL);
2475
2476         BUG_ON(!list_empty(&msg->list_head));
2477         list_add_tail(&msg->list_head, &con->out_queue);
2478         dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
2479              ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
2480              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2481              le32_to_cpu(msg->hdr.front_len),
2482              le32_to_cpu(msg->hdr.middle_len),
2483              le32_to_cpu(msg->hdr.data_len));
2484         mutex_unlock(&con->mutex);
2485
2486         /* if there wasn't anything waiting to send before, queue
2487          * new work */
2488         clear_standby(con);
2489         if (test_and_set_bit(WRITE_PENDING, &con->flags) == 0)
2490                 queue_con(con);
2491 }
2492 EXPORT_SYMBOL(ceph_con_send);
2493
2494 /*
2495  * Revoke a message that was previously queued for send
2496  */
2497 void ceph_msg_revoke(struct ceph_msg *msg)
2498 {
2499         struct ceph_connection *con = msg->con;
2500
2501         if (!con)
2502                 return;         /* Message not in our possession */
2503
2504         mutex_lock(&con->mutex);
2505         if (!list_empty(&msg->list_head)) {
2506                 dout("%s %p msg %p - was on queue\n", __func__, con, msg);
2507                 list_del_init(&msg->list_head);
2508                 BUG_ON(msg->con == NULL);
2509                 msg->con->ops->put(msg->con);
2510                 msg->con = NULL;
2511                 msg->hdr.seq = 0;
2512
2513                 ceph_msg_put(msg);
2514         }
2515         if (con->out_msg == msg) {
2516                 dout("%s %p msg %p - was sending\n", __func__, con, msg);
2517                 con->out_msg = NULL;
2518                 if (con->out_kvec_is_msg) {
2519                         con->out_skip = con->out_kvec_bytes;
2520                         con->out_kvec_is_msg = false;
2521                 }
2522                 msg->hdr.seq = 0;
2523
2524                 ceph_msg_put(msg);
2525         }
2526         mutex_unlock(&con->mutex);
2527 }
2528
2529 /*
2530  * Revoke a message that we may be reading data into
2531  */
2532 void ceph_msg_revoke_incoming(struct ceph_msg *msg)
2533 {
2534         struct ceph_connection *con;
2535
2536         BUG_ON(msg == NULL);
2537         if (!msg->con) {
2538                 dout("%s msg %p null con\n", __func__, msg);
2539
2540                 return;         /* Message not in our possession */
2541         }
2542
2543         con = msg->con;
2544         mutex_lock(&con->mutex);
2545         if (con->in_msg == msg) {
2546                 unsigned front_len = le32_to_cpu(con->in_hdr.front_len);
2547                 unsigned middle_len = le32_to_cpu(con->in_hdr.middle_len);
2548                 unsigned data_len = le32_to_cpu(con->in_hdr.data_len);
2549
2550                 /* skip rest of message */
2551                 dout("%s %p msg %p revoked\n", __func__, con, msg);
2552                 con->in_base_pos = con->in_base_pos -
2553                                 sizeof(struct ceph_msg_header) -
2554                                 front_len -
2555                                 middle_len -
2556                                 data_len -
2557                                 sizeof(struct ceph_msg_footer);
2558                 ceph_msg_put(con->in_msg);
2559                 con->in_msg = NULL;
2560                 con->in_tag = CEPH_MSGR_TAG_READY;
2561                 con->in_seq++;
2562         } else {
2563                 dout("%s %p in_msg %p msg %p no-op\n",
2564                      __func__, con, con->in_msg, msg);
2565         }
2566         mutex_unlock(&con->mutex);
2567 }
2568
2569 /*
2570  * Queue a keepalive byte to ensure the tcp connection is alive.
2571  */
2572 void ceph_con_keepalive(struct ceph_connection *con)
2573 {
2574         dout("con_keepalive %p\n", con);
2575         clear_standby(con);
2576         if (test_and_set_bit(KEEPALIVE_PENDING, &con->flags) == 0 &&
2577             test_and_set_bit(WRITE_PENDING, &con->flags) == 0)
2578                 queue_con(con);
2579 }
2580 EXPORT_SYMBOL(ceph_con_keepalive);
2581
2582
2583 /*
2584  * construct a new message with given type, size
2585  * the new msg has a ref count of 1.
2586  */
2587 struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
2588                               bool can_fail)
2589 {
2590         struct ceph_msg *m;
2591
2592         m = kmalloc(sizeof(*m), flags);
2593         if (m == NULL)
2594                 goto out;
2595         kref_init(&m->kref);
2596
2597         m->con = NULL;
2598         INIT_LIST_HEAD(&m->list_head);
2599
2600         m->hdr.tid = 0;
2601         m->hdr.type = cpu_to_le16(type);
2602         m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
2603         m->hdr.version = 0;
2604         m->hdr.front_len = cpu_to_le32(front_len);
2605         m->hdr.middle_len = 0;
2606         m->hdr.data_len = 0;
2607         m->hdr.data_off = 0;
2608         m->hdr.reserved = 0;
2609         m->footer.front_crc = 0;
2610         m->footer.middle_crc = 0;
2611         m->footer.data_crc = 0;
2612         m->footer.flags = 0;
2613         m->front_max = front_len;
2614         m->front_is_vmalloc = false;
2615         m->more_to_follow = false;
2616         m->ack_stamp = 0;
2617         m->pool = NULL;
2618
2619         /* middle */
2620         m->middle = NULL;
2621
2622         /* data */
2623         m->nr_pages = 0;
2624         m->page_alignment = 0;
2625         m->pages = NULL;
2626         m->pagelist = NULL;
2627         m->bio = NULL;
2628         m->bio_iter = NULL;
2629         m->bio_seg = 0;
2630         m->trail = NULL;
2631
2632         /* front */
2633         if (front_len) {
2634                 if (front_len > PAGE_CACHE_SIZE) {
2635                         m->front.iov_base = __vmalloc(front_len, flags,
2636                                                       PAGE_KERNEL);
2637                         m->front_is_vmalloc = true;
2638                 } else {
2639                         m->front.iov_base = kmalloc(front_len, flags);
2640                 }
2641                 if (m->front.iov_base == NULL) {
2642                         dout("ceph_msg_new can't allocate %d bytes\n",
2643                              front_len);
2644                         goto out2;
2645                 }
2646         } else {
2647                 m->front.iov_base = NULL;
2648         }
2649         m->front.iov_len = front_len;
2650
2651         dout("ceph_msg_new %p front %d\n", m, front_len);
2652         return m;
2653
2654 out2:
2655         ceph_msg_put(m);
2656 out:
2657         if (!can_fail) {
2658                 pr_err("msg_new can't create type %d front %d\n", type,
2659                        front_len);
2660                 WARN_ON(1);
2661         } else {
2662                 dout("msg_new can't create type %d front %d\n", type,
2663                      front_len);
2664         }
2665         return NULL;
2666 }
2667 EXPORT_SYMBOL(ceph_msg_new);
2668
2669 /*
2670  * Allocate "middle" portion of a message, if it is needed and wasn't
2671  * allocated by alloc_msg.  This allows us to read a small fixed-size
2672  * per-type header in the front and then gracefully fail (i.e.,
2673  * propagate the error to the caller based on info in the front) when
2674  * the middle is too large.
2675  */
2676 static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
2677 {
2678         int type = le16_to_cpu(msg->hdr.type);
2679         int middle_len = le32_to_cpu(msg->hdr.middle_len);
2680
2681         dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
2682              ceph_msg_type_name(type), middle_len);
2683         BUG_ON(!middle_len);
2684         BUG_ON(msg->middle);
2685
2686         msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
2687         if (!msg->middle)
2688                 return -ENOMEM;
2689         return 0;
2690 }
2691
2692 /*
2693  * Allocate a message for receiving an incoming message on a
2694  * connection, and save the result in con->in_msg.  Uses the
2695  * connection's private alloc_msg op if available.
2696  *
2697  * Returns true if the message should be skipped, false otherwise.
2698  * If true is returned (skip message), con->in_msg will be NULL.
2699  * If false is returned, con->in_msg will contain a pointer to the
2700  * newly-allocated message, or NULL in case of memory exhaustion.
2701  */
2702 static bool ceph_con_in_msg_alloc(struct ceph_connection *con,
2703                                 struct ceph_msg_header *hdr)
2704 {
2705         int type = le16_to_cpu(hdr->type);
2706         int front_len = le32_to_cpu(hdr->front_len);
2707         int middle_len = le32_to_cpu(hdr->middle_len);
2708         int ret;
2709
2710         BUG_ON(con->in_msg != NULL);
2711
2712         if (con->ops->alloc_msg) {
2713                 int skip = 0;
2714
2715                 mutex_unlock(&con->mutex);
2716                 con->in_msg = con->ops->alloc_msg(con, hdr, &skip);
2717                 mutex_lock(&con->mutex);
2718                 if (con->in_msg) {
2719                         con->in_msg->con = con->ops->get(con);
2720                         BUG_ON(con->in_msg->con == NULL);
2721                 }
2722                 if (skip)
2723                         con->in_msg = NULL;
2724
2725                 if (!con->in_msg)
2726                         return skip != 0;
2727         }
2728         if (!con->in_msg) {
2729                 con->in_msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
2730                 if (!con->in_msg) {
2731                         pr_err("unable to allocate msg type %d len %d\n",
2732                                type, front_len);
2733                         return false;
2734                 }
2735                 con->in_msg->con = con->ops->get(con);
2736                 BUG_ON(con->in_msg->con == NULL);
2737                 con->in_msg->page_alignment = le16_to_cpu(hdr->data_off);
2738         }
2739         memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2740
2741         if (middle_len && !con->in_msg->middle) {
2742                 ret = ceph_alloc_middle(con, con->in_msg);
2743                 if (ret < 0) {
2744                         ceph_msg_put(con->in_msg);
2745                         con->in_msg = NULL;
2746                 }
2747         }
2748
2749         return false;
2750 }
2751
2752
2753 /*
2754  * Free a generically kmalloc'd message.
2755  */
2756 void ceph_msg_kfree(struct ceph_msg *m)
2757 {
2758         dout("msg_kfree %p\n", m);
2759         if (m->front_is_vmalloc)
2760                 vfree(m->front.iov_base);
2761         else
2762                 kfree(m->front.iov_base);
2763         kfree(m);
2764 }
2765
2766 /*
2767  * Drop a msg ref.  Destroy as needed.
2768  */
2769 void ceph_msg_last_put(struct kref *kref)
2770 {
2771         struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
2772
2773         dout("ceph_msg_put last one on %p\n", m);
2774         WARN_ON(!list_empty(&m->list_head));
2775
2776         /* drop middle, data, if any */
2777         if (m->middle) {
2778                 ceph_buffer_put(m->middle);
2779                 m->middle = NULL;
2780         }
2781         m->nr_pages = 0;
2782         m->pages = NULL;
2783
2784         if (m->pagelist) {
2785                 ceph_pagelist_release(m->pagelist);
2786                 kfree(m->pagelist);
2787                 m->pagelist = NULL;
2788         }
2789
2790         m->trail = NULL;
2791
2792         if (m->pool)
2793                 ceph_msgpool_put(m->pool, m);
2794         else
2795                 ceph_msg_kfree(m);
2796 }
2797 EXPORT_SYMBOL(ceph_msg_last_put);
2798
2799 void ceph_msg_dump(struct ceph_msg *msg)
2800 {
2801         pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
2802                  msg->front_max, msg->nr_pages);
2803         print_hex_dump(KERN_DEBUG, "header: ",
2804                        DUMP_PREFIX_OFFSET, 16, 1,
2805                        &msg->hdr, sizeof(msg->hdr), true);
2806         print_hex_dump(KERN_DEBUG, " front: ",
2807                        DUMP_PREFIX_OFFSET, 16, 1,
2808                        msg->front.iov_base, msg->front.iov_len, true);
2809         if (msg->middle)
2810                 print_hex_dump(KERN_DEBUG, "middle: ",
2811                                DUMP_PREFIX_OFFSET, 16, 1,
2812                                msg->middle->vec.iov_base,
2813                                msg->middle->vec.iov_len, true);
2814         print_hex_dump(KERN_DEBUG, "footer: ",
2815                        DUMP_PREFIX_OFFSET, 16, 1,
2816                        &msg->footer, sizeof(msg->footer), true);
2817 }
2818 EXPORT_SYMBOL(ceph_msg_dump);