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