aa71a67450afa4d41f32cfaed09776b5b5552c49
[linux-2.6.git] / net / ceph / messenger.c
1 #include <linux/ceph/ceph_debug.h>
2
3 #include <linux/crc32c.h>
4 #include <linux/ctype.h>
5 #include <linux/highmem.h>
6 #include <linux/inet.h>
7 #include <linux/kthread.h>
8 #include <linux/net.h>
9 #include <linux/slab.h>
10 #include <linux/socket.h>
11 #include <linux/string.h>
12 #include <linux/bio.h>
13 #include <linux/blkdev.h>
14 #include <linux/dns_resolver.h>
15 #include <net/tcp.h>
16
17 #include <linux/ceph/libceph.h>
18 #include <linux/ceph/messenger.h>
19 #include <linux/ceph/decode.h>
20 #include <linux/ceph/pagelist.h>
21 #include <linux/export.h>
22
23 /*
24  * Ceph uses the messenger to exchange ceph_msg messages with other
25  * hosts in the system.  The messenger provides ordered and reliable
26  * delivery.  We tolerate TCP disconnects by reconnecting (with
27  * exponential backoff) in the case of a fault (disconnection, bad
28  * crc, protocol error).  Acks allow sent messages to be discarded by
29  * the sender.
30  */
31
32 /*
33  * We track the state of the socket on a given connection using
34  * values defined below.  The transition to a new socket state is
35  * handled by a function which verifies we aren't coming from an
36  * unexpected state.
37  *
38  *      --------
39  *      | NEW* |  transient initial state
40  *      --------
41  *          | con_sock_state_init()
42  *          v
43  *      ----------
44  *      | CLOSED |  initialized, but no socket (and no
45  *      ----------  TCP connection)
46  *       ^      \
47  *       |       \ con_sock_state_connecting()
48  *       |        ----------------------
49  *       |                              \
50  *       + con_sock_state_closed()       \
51  *       |+---------------------------    \
52  *       | \                          \    \
53  *       |  -----------                \    \
54  *       |  | CLOSING |  socket event;  \    \
55  *       |  -----------  await close     \    \
56  *       |       ^                        \   |
57  *       |       |                         \  |
58  *       |       + con_sock_state_closing() \ |
59  *       |      / \                         | |
60  *       |     /   ---------------          | |
61  *       |    /                   \         v v
62  *       |   /                    --------------
63  *       |  /    -----------------| CONNECTING |  socket created, TCP
64  *       |  |   /                 --------------  connect initiated
65  *       |  |   | con_sock_state_connected()
66  *       |  |   v
67  *      -------------
68  *      | CONNECTED |  TCP connection established
69  *      -------------
70  *
71  * State values for ceph_connection->sock_state; NEW is assumed to be 0.
72  */
73
74 #define CON_SOCK_STATE_NEW              0       /* -> CLOSED */
75 #define CON_SOCK_STATE_CLOSED           1       /* -> CONNECTING */
76 #define CON_SOCK_STATE_CONNECTING       2       /* -> CONNECTED or -> CLOSING */
77 #define CON_SOCK_STATE_CONNECTED        3       /* -> CLOSING or -> CLOSED */
78 #define CON_SOCK_STATE_CLOSING          4       /* -> CLOSED */
79
80 /*
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_add(con, sizeof (con->out_connect),
915                                         &con->out_connect);
916         if (auth && auth->authorizer_buf_len)
917                 con_out_kvec_add(con, auth->authorizer_buf_len,
918                                         auth->authorizer_buf);
919
920         con->out_more = 0;
921         set_bit(CON_FLAG_WRITE_PENDING, &con->flags);
922
923         return 0;
924 }
925
926 /*
927  * write as much of pending kvecs to the socket as we can.
928  *  1 -> done
929  *  0 -> socket full, but more to do
930  * <0 -> error
931  */
932 static int write_partial_kvec(struct ceph_connection *con)
933 {
934         int ret;
935
936         dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
937         while (con->out_kvec_bytes > 0) {
938                 ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
939                                        con->out_kvec_left, con->out_kvec_bytes,
940                                        con->out_more);
941                 if (ret <= 0)
942                         goto out;
943                 con->out_kvec_bytes -= ret;
944                 if (con->out_kvec_bytes == 0)
945                         break;            /* done */
946
947                 /* account for full iov entries consumed */
948                 while (ret >= con->out_kvec_cur->iov_len) {
949                         BUG_ON(!con->out_kvec_left);
950                         ret -= con->out_kvec_cur->iov_len;
951                         con->out_kvec_cur++;
952                         con->out_kvec_left--;
953                 }
954                 /* and for a partially-consumed entry */
955                 if (ret) {
956                         con->out_kvec_cur->iov_len -= ret;
957                         con->out_kvec_cur->iov_base += ret;
958                 }
959         }
960         con->out_kvec_left = 0;
961         con->out_kvec_is_msg = false;
962         ret = 1;
963 out:
964         dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
965              con->out_kvec_bytes, con->out_kvec_left, ret);
966         return ret;  /* done! */
967 }
968
969 static void out_msg_pos_next(struct ceph_connection *con, struct page *page,
970                         size_t len, size_t sent, bool in_trail)
971 {
972         struct ceph_msg *msg = con->out_msg;
973
974         BUG_ON(!msg);
975         BUG_ON(!sent);
976
977         con->out_msg_pos.data_pos += sent;
978         con->out_msg_pos.page_pos += sent;
979         if (sent < len)
980                 return;
981
982         BUG_ON(sent != len);
983         con->out_msg_pos.page_pos = 0;
984         con->out_msg_pos.page++;
985         con->out_msg_pos.did_page_crc = false;
986         if (in_trail)
987                 list_move_tail(&page->lru,
988                                &msg->trail->head);
989         else if (msg->pagelist)
990                 list_move_tail(&page->lru,
991                                &msg->pagelist->head);
992 #ifdef CONFIG_BLOCK
993         else if (msg->bio)
994                 iter_bio_next(&msg->bio_iter, &msg->bio_seg);
995 #endif
996 }
997
998 /*
999  * Write as much message data payload as we can.  If we finish, queue
1000  * up the footer.
1001  *  1 -> done, footer is now queued in out_kvec[].
1002  *  0 -> socket full, but more to do
1003  * <0 -> error
1004  */
1005 static int write_partial_msg_pages(struct ceph_connection *con)
1006 {
1007         struct ceph_msg *msg = con->out_msg;
1008         unsigned data_len = le32_to_cpu(msg->hdr.data_len);
1009         size_t len;
1010         bool do_datacrc = !con->msgr->nocrc;
1011         int ret;
1012         int total_max_write;
1013         bool in_trail = false;
1014         const size_t trail_len = (msg->trail ? msg->trail->length : 0);
1015         const size_t trail_off = data_len - trail_len;
1016
1017         dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
1018              con, msg, con->out_msg_pos.page, msg->nr_pages,
1019              con->out_msg_pos.page_pos);
1020
1021         /*
1022          * Iterate through each page that contains data to be
1023          * written, and send as much as possible for each.
1024          *
1025          * If we are calculating the data crc (the default), we will
1026          * need to map the page.  If we have no pages, they have
1027          * been revoked, so use the zero page.
1028          */
1029         while (data_len > con->out_msg_pos.data_pos) {
1030                 struct page *page = NULL;
1031                 int max_write = PAGE_SIZE;
1032                 int bio_offset = 0;
1033
1034                 in_trail = in_trail || con->out_msg_pos.data_pos >= trail_off;
1035                 if (!in_trail)
1036                         total_max_write = trail_off - con->out_msg_pos.data_pos;
1037
1038                 if (in_trail) {
1039                         total_max_write = data_len - con->out_msg_pos.data_pos;
1040
1041                         page = list_first_entry(&msg->trail->head,
1042                                                 struct page, lru);
1043                 } else if (msg->pages) {
1044                         page = msg->pages[con->out_msg_pos.page];
1045                 } else if (msg->pagelist) {
1046                         page = list_first_entry(&msg->pagelist->head,
1047                                                 struct page, lru);
1048 #ifdef CONFIG_BLOCK
1049                 } else if (msg->bio) {
1050                         struct bio_vec *bv;
1051
1052                         bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
1053                         page = bv->bv_page;
1054                         bio_offset = bv->bv_offset;
1055                         max_write = bv->bv_len;
1056 #endif
1057                 } else {
1058                         page = zero_page;
1059                 }
1060                 len = min_t(int, max_write - con->out_msg_pos.page_pos,
1061                             total_max_write);
1062
1063                 if (do_datacrc && !con->out_msg_pos.did_page_crc) {
1064                         void *base;
1065                         u32 crc = le32_to_cpu(msg->footer.data_crc);
1066                         char *kaddr;
1067
1068                         kaddr = kmap(page);
1069                         BUG_ON(kaddr == NULL);
1070                         base = kaddr + con->out_msg_pos.page_pos + bio_offset;
1071                         crc = crc32c(crc, base, len);
1072                         kunmap(page);
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                 if (ret <= 0)
1080                         goto out;
1081
1082                 out_msg_pos_next(con, page, len, (size_t) ret, in_trail);
1083         }
1084
1085         dout("write_partial_msg_pages %p msg %p done\n", con, msg);
1086
1087         /* prepare and queue up footer, too */
1088         if (!do_datacrc)
1089                 msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1090         con_out_kvec_reset(con);
1091         prepare_write_message_footer(con);
1092         ret = 1;
1093 out:
1094         return ret;
1095 }
1096
1097 /*
1098  * write some zeros
1099  */
1100 static int write_partial_skip(struct ceph_connection *con)
1101 {
1102         int ret;
1103
1104         while (con->out_skip > 0) {
1105                 size_t size = min(con->out_skip, (int) PAGE_CACHE_SIZE);
1106
1107                 ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, 1);
1108                 if (ret <= 0)
1109                         goto out;
1110                 con->out_skip -= ret;
1111         }
1112         ret = 1;
1113 out:
1114         return ret;
1115 }
1116
1117 /*
1118  * Prepare to read connection handshake, or an ack.
1119  */
1120 static void prepare_read_banner(struct ceph_connection *con)
1121 {
1122         dout("prepare_read_banner %p\n", con);
1123         con->in_base_pos = 0;
1124 }
1125
1126 static void prepare_read_connect(struct ceph_connection *con)
1127 {
1128         dout("prepare_read_connect %p\n", con);
1129         con->in_base_pos = 0;
1130 }
1131
1132 static void prepare_read_ack(struct ceph_connection *con)
1133 {
1134         dout("prepare_read_ack %p\n", con);
1135         con->in_base_pos = 0;
1136 }
1137
1138 static void prepare_read_tag(struct ceph_connection *con)
1139 {
1140         dout("prepare_read_tag %p\n", con);
1141         con->in_base_pos = 0;
1142         con->in_tag = CEPH_MSGR_TAG_READY;
1143 }
1144
1145 /*
1146  * Prepare to read a message.
1147  */
1148 static int prepare_read_message(struct ceph_connection *con)
1149 {
1150         dout("prepare_read_message %p\n", con);
1151         BUG_ON(con->in_msg != NULL);
1152         con->in_base_pos = 0;
1153         con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
1154         return 0;
1155 }
1156
1157
1158 static int read_partial(struct ceph_connection *con,
1159                         int end, int size, void *object)
1160 {
1161         while (con->in_base_pos < end) {
1162                 int left = end - con->in_base_pos;
1163                 int have = size - left;
1164                 int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
1165                 if (ret <= 0)
1166                         return ret;
1167                 con->in_base_pos += ret;
1168         }
1169         return 1;
1170 }
1171
1172
1173 /*
1174  * Read all or part of the connect-side handshake on a new connection
1175  */
1176 static int read_partial_banner(struct ceph_connection *con)
1177 {
1178         int size;
1179         int end;
1180         int ret;
1181
1182         dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
1183
1184         /* peer's banner */
1185         size = strlen(CEPH_BANNER);
1186         end = size;
1187         ret = read_partial(con, end, size, con->in_banner);
1188         if (ret <= 0)
1189                 goto out;
1190
1191         size = sizeof (con->actual_peer_addr);
1192         end += size;
1193         ret = read_partial(con, end, size, &con->actual_peer_addr);
1194         if (ret <= 0)
1195                 goto out;
1196
1197         size = sizeof (con->peer_addr_for_me);
1198         end += size;
1199         ret = read_partial(con, end, size, &con->peer_addr_for_me);
1200         if (ret <= 0)
1201                 goto out;
1202
1203 out:
1204         return ret;
1205 }
1206
1207 static int read_partial_connect(struct ceph_connection *con)
1208 {
1209         int size;
1210         int end;
1211         int ret;
1212
1213         dout("read_partial_connect %p at %d\n", con, con->in_base_pos);
1214
1215         size = sizeof (con->in_reply);
1216         end = size;
1217         ret = read_partial(con, end, size, &con->in_reply);
1218         if (ret <= 0)
1219                 goto out;
1220
1221         size = le32_to_cpu(con->in_reply.authorizer_len);
1222         end += size;
1223         ret = read_partial(con, end, size, con->auth_reply_buf);
1224         if (ret <= 0)
1225                 goto out;
1226
1227         dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
1228              con, (int)con->in_reply.tag,
1229              le32_to_cpu(con->in_reply.connect_seq),
1230              le32_to_cpu(con->in_reply.global_seq));
1231 out:
1232         return ret;
1233
1234 }
1235
1236 /*
1237  * Verify the hello banner looks okay.
1238  */
1239 static int verify_hello(struct ceph_connection *con)
1240 {
1241         if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1242                 pr_err("connect to %s got bad banner\n",
1243                        ceph_pr_addr(&con->peer_addr.in_addr));
1244                 con->error_msg = "protocol error, bad banner";
1245                 return -1;
1246         }
1247         return 0;
1248 }
1249
1250 static bool addr_is_blank(struct sockaddr_storage *ss)
1251 {
1252         switch (ss->ss_family) {
1253         case AF_INET:
1254                 return ((struct sockaddr_in *)ss)->sin_addr.s_addr == 0;
1255         case AF_INET6:
1256                 return
1257                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[0] == 0 &&
1258                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[1] == 0 &&
1259                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[2] == 0 &&
1260                      ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[3] == 0;
1261         }
1262         return false;
1263 }
1264
1265 static int addr_port(struct sockaddr_storage *ss)
1266 {
1267         switch (ss->ss_family) {
1268         case AF_INET:
1269                 return ntohs(((struct sockaddr_in *)ss)->sin_port);
1270         case AF_INET6:
1271                 return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
1272         }
1273         return 0;
1274 }
1275
1276 static void addr_set_port(struct sockaddr_storage *ss, int p)
1277 {
1278         switch (ss->ss_family) {
1279         case AF_INET:
1280                 ((struct sockaddr_in *)ss)->sin_port = htons(p);
1281                 break;
1282         case AF_INET6:
1283                 ((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1284                 break;
1285         }
1286 }
1287
1288 /*
1289  * Unlike other *_pton function semantics, zero indicates success.
1290  */
1291 static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
1292                 char delim, const char **ipend)
1293 {
1294         struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
1295         struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
1296
1297         memset(ss, 0, sizeof(*ss));
1298
1299         if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) {
1300                 ss->ss_family = AF_INET;
1301                 return 0;
1302         }
1303
1304         if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) {
1305                 ss->ss_family = AF_INET6;
1306                 return 0;
1307         }
1308
1309         return -EINVAL;
1310 }
1311
1312 /*
1313  * Extract hostname string and resolve using kernel DNS facility.
1314  */
1315 #ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER
1316 static int ceph_dns_resolve_name(const char *name, size_t namelen,
1317                 struct sockaddr_storage *ss, char delim, const char **ipend)
1318 {
1319         const char *end, *delim_p;
1320         char *colon_p, *ip_addr = NULL;
1321         int ip_len, ret;
1322
1323         /*
1324          * The end of the hostname occurs immediately preceding the delimiter or
1325          * the port marker (':') where the delimiter takes precedence.
1326          */
1327         delim_p = memchr(name, delim, namelen);
1328         colon_p = memchr(name, ':', namelen);
1329
1330         if (delim_p && colon_p)
1331                 end = delim_p < colon_p ? delim_p : colon_p;
1332         else if (!delim_p && colon_p)
1333                 end = colon_p;
1334         else {
1335                 end = delim_p;
1336                 if (!end) /* case: hostname:/ */
1337                         end = name + namelen;
1338         }
1339
1340         if (end <= name)
1341                 return -EINVAL;
1342
1343         /* do dns_resolve upcall */
1344         ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL);
1345         if (ip_len > 0)
1346                 ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL);
1347         else
1348                 ret = -ESRCH;
1349
1350         kfree(ip_addr);
1351
1352         *ipend = end;
1353
1354         pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name,
1355                         ret, ret ? "failed" : ceph_pr_addr(ss));
1356
1357         return ret;
1358 }
1359 #else
1360 static inline int ceph_dns_resolve_name(const char *name, size_t namelen,
1361                 struct sockaddr_storage *ss, char delim, const char **ipend)
1362 {
1363         return -EINVAL;
1364 }
1365 #endif
1366
1367 /*
1368  * Parse a server name (IP or hostname). If a valid IP address is not found
1369  * then try to extract a hostname to resolve using userspace DNS upcall.
1370  */
1371 static int ceph_parse_server_name(const char *name, size_t namelen,
1372                         struct sockaddr_storage *ss, char delim, const char **ipend)
1373 {
1374         int ret;
1375
1376         ret = ceph_pton(name, namelen, ss, delim, ipend);
1377         if (ret)
1378                 ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend);
1379
1380         return ret;
1381 }
1382
1383 /*
1384  * Parse an ip[:port] list into an addr array.  Use the default
1385  * monitor port if a port isn't specified.
1386  */
1387 int ceph_parse_ips(const char *c, const char *end,
1388                    struct ceph_entity_addr *addr,
1389                    int max_count, int *count)
1390 {
1391         int i, ret = -EINVAL;
1392         const char *p = c;
1393
1394         dout("parse_ips on '%.*s'\n", (int)(end-c), c);
1395         for (i = 0; i < max_count; i++) {
1396                 const char *ipend;
1397                 struct sockaddr_storage *ss = &addr[i].in_addr;
1398                 int port;
1399                 char delim = ',';
1400
1401                 if (*p == '[') {
1402                         delim = ']';
1403                         p++;
1404                 }
1405
1406                 ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
1407                 if (ret)
1408                         goto bad;
1409                 ret = -EINVAL;
1410
1411                 p = ipend;
1412
1413                 if (delim == ']') {
1414                         if (*p != ']') {
1415                                 dout("missing matching ']'\n");
1416                                 goto bad;
1417                         }
1418                         p++;
1419                 }
1420
1421                 /* port? */
1422                 if (p < end && *p == ':') {
1423                         port = 0;
1424                         p++;
1425                         while (p < end && *p >= '0' && *p <= '9') {
1426                                 port = (port * 10) + (*p - '0');
1427                                 p++;
1428                         }
1429                         if (port > 65535 || port == 0)
1430                                 goto bad;
1431                 } else {
1432                         port = CEPH_MON_PORT;
1433                 }
1434
1435                 addr_set_port(ss, port);
1436
1437                 dout("parse_ips got %s\n", ceph_pr_addr(ss));
1438
1439                 if (p == end)
1440                         break;
1441                 if (*p != ',')
1442                         goto bad;
1443                 p++;
1444         }
1445
1446         if (p != end)
1447                 goto bad;
1448
1449         if (count)
1450                 *count = i + 1;
1451         return 0;
1452
1453 bad:
1454         pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
1455         return ret;
1456 }
1457 EXPORT_SYMBOL(ceph_parse_ips);
1458
1459 static int process_banner(struct ceph_connection *con)
1460 {
1461         dout("process_banner on %p\n", con);
1462
1463         if (verify_hello(con) < 0)
1464                 return -1;
1465
1466         ceph_decode_addr(&con->actual_peer_addr);
1467         ceph_decode_addr(&con->peer_addr_for_me);
1468
1469         /*
1470          * Make sure the other end is who we wanted.  note that the other
1471          * end may not yet know their ip address, so if it's 0.0.0.0, give
1472          * them the benefit of the doubt.
1473          */
1474         if (memcmp(&con->peer_addr, &con->actual_peer_addr,
1475                    sizeof(con->peer_addr)) != 0 &&
1476             !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
1477               con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
1478                 pr_warning("wrong peer, want %s/%d, got %s/%d\n",
1479                            ceph_pr_addr(&con->peer_addr.in_addr),
1480                            (int)le32_to_cpu(con->peer_addr.nonce),
1481                            ceph_pr_addr(&con->actual_peer_addr.in_addr),
1482                            (int)le32_to_cpu(con->actual_peer_addr.nonce));
1483                 con->error_msg = "wrong peer at address";
1484                 return -1;
1485         }
1486
1487         /*
1488          * did we learn our address?
1489          */
1490         if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
1491                 int port = addr_port(&con->msgr->inst.addr.in_addr);
1492
1493                 memcpy(&con->msgr->inst.addr.in_addr,
1494                        &con->peer_addr_for_me.in_addr,
1495                        sizeof(con->peer_addr_for_me.in_addr));
1496                 addr_set_port(&con->msgr->inst.addr.in_addr, port);
1497                 encode_my_addr(con->msgr);
1498                 dout("process_banner learned my addr is %s\n",
1499                      ceph_pr_addr(&con->msgr->inst.addr.in_addr));
1500         }
1501
1502         return 0;
1503 }
1504
1505 static void fail_protocol(struct ceph_connection *con)
1506 {
1507         reset_connection(con);
1508         BUG_ON(con->state != CON_STATE_NEGOTIATING);
1509         con->state = CON_STATE_CLOSED;
1510 }
1511
1512 static int process_connect(struct ceph_connection *con)
1513 {
1514         u64 sup_feat = con->msgr->supported_features;
1515         u64 req_feat = con->msgr->required_features;
1516         u64 server_feat = le64_to_cpu(con->in_reply.features);
1517         int ret;
1518
1519         dout("process_connect on %p tag %d\n", con, (int)con->in_tag);
1520
1521         switch (con->in_reply.tag) {
1522         case CEPH_MSGR_TAG_FEATURES:
1523                 pr_err("%s%lld %s feature set mismatch,"
1524                        " my %llx < server's %llx, missing %llx\n",
1525                        ENTITY_NAME(con->peer_name),
1526                        ceph_pr_addr(&con->peer_addr.in_addr),
1527                        sup_feat, server_feat, server_feat & ~sup_feat);
1528                 con->error_msg = "missing required protocol features";
1529                 fail_protocol(con);
1530                 return -1;
1531
1532         case CEPH_MSGR_TAG_BADPROTOVER:
1533                 pr_err("%s%lld %s protocol version mismatch,"
1534                        " my %d != server's %d\n",
1535                        ENTITY_NAME(con->peer_name),
1536                        ceph_pr_addr(&con->peer_addr.in_addr),
1537                        le32_to_cpu(con->out_connect.protocol_version),
1538                        le32_to_cpu(con->in_reply.protocol_version));
1539                 con->error_msg = "protocol version mismatch";
1540                 fail_protocol(con);
1541                 return -1;
1542
1543         case CEPH_MSGR_TAG_BADAUTHORIZER:
1544                 con->auth_retry++;
1545                 dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
1546                      con->auth_retry);
1547                 if (con->auth_retry == 2) {
1548                         con->error_msg = "connect authorization failure";
1549                         return -1;
1550                 }
1551                 con->auth_retry = 1;
1552                 con_out_kvec_reset(con);
1553                 ret = prepare_write_connect(con);
1554                 if (ret < 0)
1555                         return ret;
1556                 prepare_read_connect(con);
1557                 break;
1558
1559         case CEPH_MSGR_TAG_RESETSESSION:
1560                 /*
1561                  * If we connected with a large connect_seq but the peer
1562                  * has no record of a session with us (no connection, or
1563                  * connect_seq == 0), they will send RESETSESION to indicate
1564                  * that they must have reset their session, and may have
1565                  * dropped messages.
1566                  */
1567                 dout("process_connect got RESET peer seq %u\n",
1568                      le32_to_cpu(con->in_reply.connect_seq));
1569                 pr_err("%s%lld %s connection reset\n",
1570                        ENTITY_NAME(con->peer_name),
1571                        ceph_pr_addr(&con->peer_addr.in_addr));
1572                 reset_connection(con);
1573                 con_out_kvec_reset(con);
1574                 ret = prepare_write_connect(con);
1575                 if (ret < 0)
1576                         return ret;
1577                 prepare_read_connect(con);
1578
1579                 /* Tell ceph about it. */
1580                 mutex_unlock(&con->mutex);
1581                 pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
1582                 if (con->ops->peer_reset)
1583                         con->ops->peer_reset(con);
1584                 mutex_lock(&con->mutex);
1585                 if (con->state != CON_STATE_NEGOTIATING)
1586                         return -EAGAIN;
1587                 break;
1588
1589         case CEPH_MSGR_TAG_RETRY_SESSION:
1590                 /*
1591                  * If we sent a smaller connect_seq than the peer has, try
1592                  * again with a larger value.
1593                  */
1594                 dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
1595                      le32_to_cpu(con->out_connect.connect_seq),
1596                      le32_to_cpu(con->in_reply.connect_seq));
1597                 con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
1598                 con_out_kvec_reset(con);
1599                 ret = prepare_write_connect(con);
1600                 if (ret < 0)
1601                         return ret;
1602                 prepare_read_connect(con);
1603                 break;
1604
1605         case CEPH_MSGR_TAG_RETRY_GLOBAL:
1606                 /*
1607                  * If we sent a smaller global_seq than the peer has, try
1608                  * again with a larger value.
1609                  */
1610                 dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
1611                      con->peer_global_seq,
1612                      le32_to_cpu(con->in_reply.global_seq));
1613                 get_global_seq(con->msgr,
1614                                le32_to_cpu(con->in_reply.global_seq));
1615                 con_out_kvec_reset(con);
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 int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);
1737
1738 static int read_partial_message_pages(struct ceph_connection *con,
1739                                       struct page **pages,
1740                                       unsigned data_len, bool do_datacrc)
1741 {
1742         void *p;
1743         int ret;
1744         int left;
1745
1746         left = min((int)(data_len - con->in_msg_pos.data_pos),
1747                    (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
1748         /* (page) data */
1749         BUG_ON(pages == NULL);
1750         p = kmap(pages[con->in_msg_pos.page]);
1751         ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1752                                left);
1753         if (ret > 0 && do_datacrc)
1754                 con->in_data_crc =
1755                         crc32c(con->in_data_crc,
1756                                   p + con->in_msg_pos.page_pos, ret);
1757         kunmap(pages[con->in_msg_pos.page]);
1758         if (ret <= 0)
1759                 return ret;
1760         con->in_msg_pos.data_pos += ret;
1761         con->in_msg_pos.page_pos += ret;
1762         if (con->in_msg_pos.page_pos == PAGE_SIZE) {
1763                 con->in_msg_pos.page_pos = 0;
1764                 con->in_msg_pos.page++;
1765         }
1766
1767         return ret;
1768 }
1769
1770 #ifdef CONFIG_BLOCK
1771 static int read_partial_message_bio(struct ceph_connection *con,
1772                                     struct bio **bio_iter, int *bio_seg,
1773                                     unsigned data_len, bool do_datacrc)
1774 {
1775         struct bio_vec *bv = bio_iovec_idx(*bio_iter, *bio_seg);
1776         void *p;
1777         int ret, left;
1778
1779         left = min((int)(data_len - con->in_msg_pos.data_pos),
1780                    (int)(bv->bv_len - con->in_msg_pos.page_pos));
1781
1782         p = kmap(bv->bv_page) + bv->bv_offset;
1783
1784         ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1785                                left);
1786         if (ret > 0 && do_datacrc)
1787                 con->in_data_crc =
1788                         crc32c(con->in_data_crc,
1789                                   p + con->in_msg_pos.page_pos, ret);
1790         kunmap(bv->bv_page);
1791         if (ret <= 0)
1792                 return ret;
1793         con->in_msg_pos.data_pos += ret;
1794         con->in_msg_pos.page_pos += ret;
1795         if (con->in_msg_pos.page_pos == bv->bv_len) {
1796                 con->in_msg_pos.page_pos = 0;
1797                 iter_bio_next(bio_iter, bio_seg);
1798         }
1799
1800         return ret;
1801 }
1802 #endif
1803
1804 /*
1805  * read (part of) a message.
1806  */
1807 static int read_partial_message(struct ceph_connection *con)
1808 {
1809         struct ceph_msg *m = con->in_msg;
1810         int size;
1811         int end;
1812         int ret;
1813         unsigned front_len, middle_len, data_len;
1814         bool do_datacrc = !con->msgr->nocrc;
1815         u64 seq;
1816         u32 crc;
1817
1818         dout("read_partial_message con %p msg %p\n", con, m);
1819
1820         /* header */
1821         size = sizeof (con->in_hdr);
1822         end = size;
1823         ret = read_partial(con, end, size, &con->in_hdr);
1824         if (ret <= 0)
1825                 return ret;
1826
1827         crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
1828         if (cpu_to_le32(crc) != con->in_hdr.crc) {
1829                 pr_err("read_partial_message bad hdr "
1830                        " crc %u != expected %u\n",
1831                        crc, con->in_hdr.crc);
1832                 return -EBADMSG;
1833         }
1834
1835         front_len = le32_to_cpu(con->in_hdr.front_len);
1836         if (front_len > CEPH_MSG_MAX_FRONT_LEN)
1837                 return -EIO;
1838         middle_len = le32_to_cpu(con->in_hdr.middle_len);
1839         if (middle_len > CEPH_MSG_MAX_DATA_LEN)
1840                 return -EIO;
1841         data_len = le32_to_cpu(con->in_hdr.data_len);
1842         if (data_len > CEPH_MSG_MAX_DATA_LEN)
1843                 return -EIO;
1844
1845         /* verify seq# */
1846         seq = le64_to_cpu(con->in_hdr.seq);
1847         if ((s64)seq - (s64)con->in_seq < 1) {
1848                 pr_info("skipping %s%lld %s seq %lld expected %lld\n",
1849                         ENTITY_NAME(con->peer_name),
1850                         ceph_pr_addr(&con->peer_addr.in_addr),
1851                         seq, con->in_seq + 1);
1852                 con->in_base_pos = -front_len - middle_len - data_len -
1853                         sizeof(m->footer);
1854                 con->in_tag = CEPH_MSGR_TAG_READY;
1855                 return 0;
1856         } else if ((s64)seq - (s64)con->in_seq > 1) {
1857                 pr_err("read_partial_message bad seq %lld expected %lld\n",
1858                        seq, con->in_seq + 1);
1859                 con->error_msg = "bad message sequence # for incoming message";
1860                 return -EBADMSG;
1861         }
1862
1863         /* allocate message? */
1864         if (!con->in_msg) {
1865                 int skip = 0;
1866
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                 ret = ceph_con_in_msg_alloc(con, &skip);
1870                 if (ret < 0)
1871                         return ret;
1872                 if (skip) {
1873                         /* skip this message */
1874                         dout("alloc_msg said skip message\n");
1875                         BUG_ON(con->in_msg);
1876                         con->in_base_pos = -front_len - middle_len - data_len -
1877                                 sizeof(m->footer);
1878                         con->in_tag = CEPH_MSGR_TAG_READY;
1879                         con->in_seq++;
1880                         return 0;
1881                 }
1882
1883                 BUG_ON(!con->in_msg);
1884                 BUG_ON(con->in_msg->con != con);
1885                 m = con->in_msg;
1886                 m->front.iov_len = 0;    /* haven't read it yet */
1887                 if (m->middle)
1888                         m->middle->vec.iov_len = 0;
1889
1890                 con->in_msg_pos.page = 0;
1891                 if (m->pages)
1892                         con->in_msg_pos.page_pos = m->page_alignment;
1893                 else
1894                         con->in_msg_pos.page_pos = 0;
1895                 con->in_msg_pos.data_pos = 0;
1896
1897 #ifdef CONFIG_BLOCK
1898                 if (m->bio)
1899                         init_bio_iter(m->bio, &m->bio_iter, &m->bio_seg);
1900 #endif
1901         }
1902
1903         /* front */
1904         ret = read_partial_message_section(con, &m->front, front_len,
1905                                            &con->in_front_crc);
1906         if (ret <= 0)
1907                 return ret;
1908
1909         /* middle */
1910         if (m->middle) {
1911                 ret = read_partial_message_section(con, &m->middle->vec,
1912                                                    middle_len,
1913                                                    &con->in_middle_crc);
1914                 if (ret <= 0)
1915                         return ret;
1916         }
1917
1918         /* (page) data */
1919         while (con->in_msg_pos.data_pos < data_len) {
1920                 if (m->pages) {
1921                         ret = read_partial_message_pages(con, m->pages,
1922                                                  data_len, do_datacrc);
1923                         if (ret <= 0)
1924                                 return ret;
1925 #ifdef CONFIG_BLOCK
1926                 } else if (m->bio) {
1927                         BUG_ON(!m->bio_iter);
1928                         ret = read_partial_message_bio(con,
1929                                                  &m->bio_iter, &m->bio_seg,
1930                                                  data_len, do_datacrc);
1931                         if (ret <= 0)
1932                                 return ret;
1933 #endif
1934                 } else {
1935                         BUG_ON(1);
1936                 }
1937         }
1938
1939         /* footer */
1940         size = sizeof (m->footer);
1941         end += size;
1942         ret = read_partial(con, end, size, &m->footer);
1943         if (ret <= 0)
1944                 return ret;
1945
1946         dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
1947              m, front_len, m->footer.front_crc, middle_len,
1948              m->footer.middle_crc, data_len, m->footer.data_crc);
1949
1950         /* crc ok? */
1951         if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
1952                 pr_err("read_partial_message %p front crc %u != exp. %u\n",
1953                        m, con->in_front_crc, m->footer.front_crc);
1954                 return -EBADMSG;
1955         }
1956         if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
1957                 pr_err("read_partial_message %p middle crc %u != exp %u\n",
1958                        m, con->in_middle_crc, m->footer.middle_crc);
1959                 return -EBADMSG;
1960         }
1961         if (do_datacrc &&
1962             (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
1963             con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
1964                 pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
1965                        con->in_data_crc, le32_to_cpu(m->footer.data_crc));
1966                 return -EBADMSG;
1967         }
1968
1969         return 1; /* done! */
1970 }
1971
1972 /*
1973  * Process message.  This happens in the worker thread.  The callback should
1974  * be careful not to do anything that waits on other incoming messages or it
1975  * may deadlock.
1976  */
1977 static void process_message(struct ceph_connection *con)
1978 {
1979         struct ceph_msg *msg;
1980
1981         BUG_ON(con->in_msg->con != con);
1982         con->in_msg->con = NULL;
1983         msg = con->in_msg;
1984         con->in_msg = NULL;
1985         con->ops->put(con);
1986
1987         /* if first message, set peer_name */
1988         if (con->peer_name.type == 0)
1989                 con->peer_name = msg->hdr.src;
1990
1991         con->in_seq++;
1992         mutex_unlock(&con->mutex);
1993
1994         dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
1995              msg, le64_to_cpu(msg->hdr.seq),
1996              ENTITY_NAME(msg->hdr.src),
1997              le16_to_cpu(msg->hdr.type),
1998              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
1999              le32_to_cpu(msg->hdr.front_len),
2000              le32_to_cpu(msg->hdr.data_len),
2001              con->in_front_crc, con->in_middle_crc, con->in_data_crc);
2002         con->ops->dispatch(con, msg);
2003
2004         mutex_lock(&con->mutex);
2005 }
2006
2007
2008 /*
2009  * Write something to the socket.  Called in a worker thread when the
2010  * socket appears to be writeable and we have something ready to send.
2011  */
2012 static int try_write(struct ceph_connection *con)
2013 {
2014         int ret = 1;
2015
2016         dout("try_write start %p state %lu\n", con, con->state);
2017
2018 more:
2019         dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
2020
2021         /* open the socket first? */
2022         if (con->state == CON_STATE_PREOPEN) {
2023                 BUG_ON(con->sock);
2024                 con->state = CON_STATE_CONNECTING;
2025
2026                 con_out_kvec_reset(con);
2027                 prepare_write_banner(con);
2028                 prepare_read_banner(con);
2029
2030                 BUG_ON(con->in_msg);
2031                 con->in_tag = CEPH_MSGR_TAG_READY;
2032                 dout("try_write initiating connect on %p new state %lu\n",
2033                      con, con->state);
2034                 ret = ceph_tcp_connect(con);
2035                 if (ret < 0) {
2036                         con->error_msg = "connect error";
2037                         goto out;
2038                 }
2039         }
2040
2041 more_kvec:
2042         /* kvec data queued? */
2043         if (con->out_skip) {
2044                 ret = write_partial_skip(con);
2045                 if (ret <= 0)
2046                         goto out;
2047         }
2048         if (con->out_kvec_left) {
2049                 ret = write_partial_kvec(con);
2050                 if (ret <= 0)
2051                         goto out;
2052         }
2053
2054         /* msg pages? */
2055         if (con->out_msg) {
2056                 if (con->out_msg_done) {
2057                         ceph_msg_put(con->out_msg);
2058                         con->out_msg = NULL;   /* we're done with this one */
2059                         goto do_next;
2060                 }
2061
2062                 ret = write_partial_msg_pages(con);
2063                 if (ret == 1)
2064                         goto more_kvec;  /* we need to send the footer, too! */
2065                 if (ret == 0)
2066                         goto out;
2067                 if (ret < 0) {
2068                         dout("try_write write_partial_msg_pages err %d\n",
2069                              ret);
2070                         goto out;
2071                 }
2072         }
2073
2074 do_next:
2075         if (con->state == CON_STATE_OPEN) {
2076                 /* is anything else pending? */
2077                 if (!list_empty(&con->out_queue)) {
2078                         prepare_write_message(con);
2079                         goto more;
2080                 }
2081                 if (con->in_seq > con->in_seq_acked) {
2082                         prepare_write_ack(con);
2083                         goto more;
2084                 }
2085                 if (test_and_clear_bit(CON_FLAG_KEEPALIVE_PENDING,
2086                                        &con->flags)) {
2087                         prepare_write_keepalive(con);
2088                         goto more;
2089                 }
2090         }
2091
2092         /* Nothing to do! */
2093         clear_bit(CON_FLAG_WRITE_PENDING, &con->flags);
2094         dout("try_write nothing else to write.\n");
2095         ret = 0;
2096 out:
2097         dout("try_write done on %p ret %d\n", con, ret);
2098         return ret;
2099 }
2100
2101
2102
2103 /*
2104  * Read what we can from the socket.
2105  */
2106 static int try_read(struct ceph_connection *con)
2107 {
2108         int ret = -1;
2109
2110 more:
2111         dout("try_read start on %p state %lu\n", con, con->state);
2112         if (con->state != CON_STATE_CONNECTING &&
2113             con->state != CON_STATE_NEGOTIATING &&
2114             con->state != CON_STATE_OPEN)
2115                 return 0;
2116
2117         BUG_ON(!con->sock);
2118
2119         dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
2120              con->in_base_pos);
2121
2122         if (con->state == CON_STATE_CONNECTING) {
2123                 dout("try_read connecting\n");
2124                 ret = read_partial_banner(con);
2125                 if (ret <= 0)
2126                         goto out;
2127                 ret = process_banner(con);
2128                 if (ret < 0)
2129                         goto out;
2130
2131                 BUG_ON(con->state != CON_STATE_CONNECTING);
2132                 con->state = CON_STATE_NEGOTIATING;
2133
2134                 /*
2135                  * Received banner is good, exchange connection info.
2136                  * Do not reset out_kvec, as sending our banner raced
2137                  * with receiving peer banner after connect completed.
2138                  */
2139                 ret = prepare_write_connect(con);
2140                 if (ret < 0)
2141                         goto out;
2142                 prepare_read_connect(con);
2143
2144                 /* Send connection info before awaiting response */
2145                 goto out;
2146         }
2147
2148         if (con->state == CON_STATE_NEGOTIATING) {
2149                 dout("try_read negotiating\n");
2150                 ret = read_partial_connect(con);
2151                 if (ret <= 0)
2152                         goto out;
2153                 ret = process_connect(con);
2154                 if (ret < 0)
2155                         goto out;
2156                 goto more;
2157         }
2158
2159         BUG_ON(con->state != CON_STATE_OPEN);
2160
2161         if (con->in_base_pos < 0) {
2162                 /*
2163                  * skipping + discarding content.
2164                  *
2165                  * FIXME: there must be a better way to do this!
2166                  */
2167                 static char buf[SKIP_BUF_SIZE];
2168                 int skip = min((int) sizeof (buf), -con->in_base_pos);
2169
2170                 dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
2171                 ret = ceph_tcp_recvmsg(con->sock, buf, skip);
2172                 if (ret <= 0)
2173                         goto out;
2174                 con->in_base_pos += ret;
2175                 if (con->in_base_pos)
2176                         goto more;
2177         }
2178         if (con->in_tag == CEPH_MSGR_TAG_READY) {
2179                 /*
2180                  * what's next?
2181                  */
2182                 ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
2183                 if (ret <= 0)
2184                         goto out;
2185                 dout("try_read got tag %d\n", (int)con->in_tag);
2186                 switch (con->in_tag) {
2187                 case CEPH_MSGR_TAG_MSG:
2188                         prepare_read_message(con);
2189                         break;
2190                 case CEPH_MSGR_TAG_ACK:
2191                         prepare_read_ack(con);
2192                         break;
2193                 case CEPH_MSGR_TAG_CLOSE:
2194                         con_close_socket(con);
2195                         con->state = CON_STATE_CLOSED;
2196                         goto out;
2197                 default:
2198                         goto bad_tag;
2199                 }
2200         }
2201         if (con->in_tag == CEPH_MSGR_TAG_MSG) {
2202                 ret = read_partial_message(con);
2203                 if (ret <= 0) {
2204                         switch (ret) {
2205                         case -EBADMSG:
2206                                 con->error_msg = "bad crc";
2207                                 ret = -EIO;
2208                                 break;
2209                         case -EIO:
2210                                 con->error_msg = "io error";
2211                                 break;
2212                         }
2213                         goto out;
2214                 }
2215                 if (con->in_tag == CEPH_MSGR_TAG_READY)
2216                         goto more;
2217                 process_message(con);
2218                 if (con->state == CON_STATE_OPEN)
2219                         prepare_read_tag(con);
2220                 goto more;
2221         }
2222         if (con->in_tag == CEPH_MSGR_TAG_ACK) {
2223                 ret = read_partial_ack(con);
2224                 if (ret <= 0)
2225                         goto out;
2226                 process_ack(con);
2227                 goto more;
2228         }
2229
2230 out:
2231         dout("try_read done on %p ret %d\n", con, ret);
2232         return ret;
2233
2234 bad_tag:
2235         pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
2236         con->error_msg = "protocol error, garbage tag";
2237         ret = -1;
2238         goto out;
2239 }
2240
2241
2242 /*
2243  * Atomically queue work on a connection.  Bump @con reference to
2244  * avoid races with connection teardown.
2245  */
2246 static void queue_con(struct ceph_connection *con)
2247 {
2248         if (!con->ops->get(con)) {
2249                 dout("queue_con %p ref count 0\n", con);
2250                 return;
2251         }
2252
2253         if (!queue_delayed_work(ceph_msgr_wq, &con->work, 0)) {
2254                 dout("queue_con %p - already queued\n", con);
2255                 con->ops->put(con);
2256         } else {
2257                 dout("queue_con %p\n", con);
2258         }
2259 }
2260
2261 /*
2262  * Do some work on a connection.  Drop a connection ref when we're done.
2263  */
2264 static void con_work(struct work_struct *work)
2265 {
2266         struct ceph_connection *con = container_of(work, struct ceph_connection,
2267                                                    work.work);
2268         int ret;
2269
2270         mutex_lock(&con->mutex);
2271 restart:
2272         if (test_and_clear_bit(CON_FLAG_SOCK_CLOSED, &con->flags)) {
2273                 switch (con->state) {
2274                 case CON_STATE_CONNECTING:
2275                         con->error_msg = "connection failed";
2276                         break;
2277                 case CON_STATE_NEGOTIATING:
2278                         con->error_msg = "negotiation failed";
2279                         break;
2280                 case CON_STATE_OPEN:
2281                         con->error_msg = "socket closed";
2282                         break;
2283                 default:
2284                         dout("unrecognized con state %d\n", (int)con->state);
2285                         con->error_msg = "unrecognized con state";
2286                         BUG();
2287                 }
2288                 goto fault;
2289         }
2290
2291         if (test_and_clear_bit(CON_FLAG_BACKOFF, &con->flags)) {
2292                 dout("con_work %p backing off\n", con);
2293                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
2294                                        round_jiffies_relative(con->delay))) {
2295                         dout("con_work %p backoff %lu\n", con, con->delay);
2296                         mutex_unlock(&con->mutex);
2297                         return;
2298                 } else {
2299                         dout("con_work %p FAILED to back off %lu\n", con,
2300                              con->delay);
2301                         set_bit(CON_FLAG_BACKOFF, &con->flags);
2302                 }
2303                 goto done;
2304         }
2305
2306         if (con->state == CON_STATE_STANDBY) {
2307                 dout("con_work %p STANDBY\n", con);
2308                 goto done;
2309         }
2310         if (con->state == CON_STATE_CLOSED) {
2311                 dout("con_work %p CLOSED\n", con);
2312                 BUG_ON(con->sock);
2313                 goto done;
2314         }
2315         if (con->state == CON_STATE_PREOPEN) {
2316                 dout("con_work OPENING\n");
2317                 BUG_ON(con->sock);
2318         }
2319
2320         ret = try_read(con);
2321         if (ret == -EAGAIN)
2322                 goto restart;
2323         if (ret < 0) {
2324                 con->error_msg = "socket error on read";
2325                 goto fault;
2326         }
2327
2328         ret = try_write(con);
2329         if (ret == -EAGAIN)
2330                 goto restart;
2331         if (ret < 0) {
2332                 con->error_msg = "socket error on write";
2333                 goto fault;
2334         }
2335
2336 done:
2337         mutex_unlock(&con->mutex);
2338 done_unlocked:
2339         con->ops->put(con);
2340         return;
2341
2342 fault:
2343         ceph_fault(con);     /* error/fault path */
2344         goto done_unlocked;
2345 }
2346
2347
2348 /*
2349  * Generic error/fault handler.  A retry mechanism is used with
2350  * exponential backoff
2351  */
2352 static void ceph_fault(struct ceph_connection *con)
2353         __releases(con->mutex)
2354 {
2355         pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2356                ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
2357         dout("fault %p state %lu to peer %s\n",
2358              con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
2359
2360         BUG_ON(con->state != CON_STATE_CONNECTING &&
2361                con->state != CON_STATE_NEGOTIATING &&
2362                con->state != CON_STATE_OPEN);
2363
2364         con_close_socket(con);
2365
2366         if (test_bit(CON_FLAG_LOSSYTX, &con->flags)) {
2367                 dout("fault on LOSSYTX channel, marking CLOSED\n");
2368                 con->state = CON_STATE_CLOSED;
2369                 goto out_unlock;
2370         }
2371
2372         if (con->in_msg) {
2373                 BUG_ON(con->in_msg->con != con);
2374                 con->in_msg->con = NULL;
2375                 ceph_msg_put(con->in_msg);
2376                 con->in_msg = NULL;
2377                 con->ops->put(con);
2378         }
2379
2380         /* Requeue anything that hasn't been acked */
2381         list_splice_init(&con->out_sent, &con->out_queue);
2382
2383         /* If there are no messages queued or keepalive pending, place
2384          * the connection in a STANDBY state */
2385         if (list_empty(&con->out_queue) &&
2386             !test_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags)) {
2387                 dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2388                 clear_bit(CON_FLAG_WRITE_PENDING, &con->flags);
2389                 con->state = CON_STATE_STANDBY;
2390         } else {
2391                 /* retry after a delay. */
2392                 con->state = CON_STATE_PREOPEN;
2393                 if (con->delay == 0)
2394                         con->delay = BASE_DELAY_INTERVAL;
2395                 else if (con->delay < MAX_DELAY_INTERVAL)
2396                         con->delay *= 2;
2397                 con->ops->get(con);
2398                 if (queue_delayed_work(ceph_msgr_wq, &con->work,
2399                                        round_jiffies_relative(con->delay))) {
2400                         dout("fault queued %p delay %lu\n", con, con->delay);
2401                 } else {
2402                         con->ops->put(con);
2403                         dout("fault failed to queue %p delay %lu, backoff\n",
2404                              con, con->delay);
2405                         /*
2406                          * In many cases we see a socket state change
2407                          * while con_work is running and end up
2408                          * queuing (non-delayed) work, such that we
2409                          * can't backoff with a delay.  Set a flag so
2410                          * that when con_work restarts we schedule the
2411                          * delay then.
2412                          */
2413                         set_bit(CON_FLAG_BACKOFF, &con->flags);
2414                 }
2415         }
2416
2417 out_unlock:
2418         mutex_unlock(&con->mutex);
2419         /*
2420          * in case we faulted due to authentication, invalidate our
2421          * current tickets so that we can get new ones.
2422          */
2423         if (con->auth_retry && con->ops->invalidate_authorizer) {
2424                 dout("calling invalidate_authorizer()\n");
2425                 con->ops->invalidate_authorizer(con);
2426         }
2427
2428         if (con->ops->fault)
2429                 con->ops->fault(con);
2430 }
2431
2432
2433
2434 /*
2435  * initialize a new messenger instance
2436  */
2437 void ceph_messenger_init(struct ceph_messenger *msgr,
2438                         struct ceph_entity_addr *myaddr,
2439                         u32 supported_features,
2440                         u32 required_features,
2441                         bool nocrc)
2442 {
2443         msgr->supported_features = supported_features;
2444         msgr->required_features = required_features;
2445
2446         spin_lock_init(&msgr->global_seq_lock);
2447
2448         if (myaddr)
2449                 msgr->inst.addr = *myaddr;
2450
2451         /* select a random nonce */
2452         msgr->inst.addr.type = 0;
2453         get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2454         encode_my_addr(msgr);
2455         msgr->nocrc = nocrc;
2456
2457         atomic_set(&msgr->stopping, 0);
2458
2459         dout("%s %p\n", __func__, msgr);
2460 }
2461 EXPORT_SYMBOL(ceph_messenger_init);
2462
2463 static void clear_standby(struct ceph_connection *con)
2464 {
2465         /* come back from STANDBY? */
2466         if (con->state == CON_STATE_STANDBY) {
2467                 dout("clear_standby %p and ++connect_seq\n", con);
2468                 con->state = CON_STATE_PREOPEN;
2469                 con->connect_seq++;
2470                 WARN_ON(test_bit(CON_FLAG_WRITE_PENDING, &con->flags));
2471                 WARN_ON(test_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags));
2472         }
2473 }
2474
2475 /*
2476  * Queue up an outgoing message on the given connection.
2477  */
2478 void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
2479 {
2480         /* set src+dst */
2481         msg->hdr.src = con->msgr->inst.name;
2482         BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
2483         msg->needs_out_seq = true;
2484
2485         mutex_lock(&con->mutex);
2486
2487         if (con->state == CON_STATE_CLOSED) {
2488                 dout("con_send %p closed, dropping %p\n", con, msg);
2489                 ceph_msg_put(msg);
2490                 mutex_unlock(&con->mutex);
2491                 return;
2492         }
2493
2494         BUG_ON(msg->con != NULL);
2495         msg->con = con->ops->get(con);
2496         BUG_ON(msg->con == NULL);
2497
2498         BUG_ON(!list_empty(&msg->list_head));
2499         list_add_tail(&msg->list_head, &con->out_queue);
2500         dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
2501              ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
2502              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2503              le32_to_cpu(msg->hdr.front_len),
2504              le32_to_cpu(msg->hdr.middle_len),
2505              le32_to_cpu(msg->hdr.data_len));
2506
2507         clear_standby(con);
2508         mutex_unlock(&con->mutex);
2509
2510         /* if there wasn't anything waiting to send before, queue
2511          * new work */
2512         if (test_and_set_bit(CON_FLAG_WRITE_PENDING, &con->flags) == 0)
2513                 queue_con(con);
2514 }
2515 EXPORT_SYMBOL(ceph_con_send);
2516
2517 /*
2518  * Revoke a message that was previously queued for send
2519  */
2520 void ceph_msg_revoke(struct ceph_msg *msg)
2521 {
2522         struct ceph_connection *con = msg->con;
2523
2524         if (!con)
2525                 return;         /* Message not in our possession */
2526
2527         mutex_lock(&con->mutex);
2528         if (!list_empty(&msg->list_head)) {
2529                 dout("%s %p msg %p - was on queue\n", __func__, con, msg);
2530                 list_del_init(&msg->list_head);
2531                 BUG_ON(msg->con == NULL);
2532                 msg->con->ops->put(msg->con);
2533                 msg->con = NULL;
2534                 msg->hdr.seq = 0;
2535
2536                 ceph_msg_put(msg);
2537         }
2538         if (con->out_msg == msg) {
2539                 dout("%s %p msg %p - was sending\n", __func__, con, msg);
2540                 con->out_msg = NULL;
2541                 if (con->out_kvec_is_msg) {
2542                         con->out_skip = con->out_kvec_bytes;
2543                         con->out_kvec_is_msg = false;
2544                 }
2545                 msg->hdr.seq = 0;
2546
2547                 ceph_msg_put(msg);
2548         }
2549         mutex_unlock(&con->mutex);
2550 }
2551
2552 /*
2553  * Revoke a message that we may be reading data into
2554  */
2555 void ceph_msg_revoke_incoming(struct ceph_msg *msg)
2556 {
2557         struct ceph_connection *con;
2558
2559         BUG_ON(msg == NULL);
2560         if (!msg->con) {
2561                 dout("%s msg %p null con\n", __func__, msg);
2562
2563                 return;         /* Message not in our possession */
2564         }
2565
2566         con = msg->con;
2567         mutex_lock(&con->mutex);
2568         if (con->in_msg == msg) {
2569                 unsigned front_len = le32_to_cpu(con->in_hdr.front_len);
2570                 unsigned middle_len = le32_to_cpu(con->in_hdr.middle_len);
2571                 unsigned data_len = le32_to_cpu(con->in_hdr.data_len);
2572
2573                 /* skip rest of message */
2574                 dout("%s %p msg %p revoked\n", __func__, con, msg);
2575                 con->in_base_pos = con->in_base_pos -
2576                                 sizeof(struct ceph_msg_header) -
2577                                 front_len -
2578                                 middle_len -
2579                                 data_len -
2580                                 sizeof(struct ceph_msg_footer);
2581                 ceph_msg_put(con->in_msg);
2582                 con->in_msg = NULL;
2583                 con->in_tag = CEPH_MSGR_TAG_READY;
2584                 con->in_seq++;
2585         } else {
2586                 dout("%s %p in_msg %p msg %p no-op\n",
2587                      __func__, con, con->in_msg, msg);
2588         }
2589         mutex_unlock(&con->mutex);
2590 }
2591
2592 /*
2593  * Queue a keepalive byte to ensure the tcp connection is alive.
2594  */
2595 void ceph_con_keepalive(struct ceph_connection *con)
2596 {
2597         dout("con_keepalive %p\n", con);
2598         mutex_lock(&con->mutex);
2599         clear_standby(con);
2600         mutex_unlock(&con->mutex);
2601         if (test_and_set_bit(CON_FLAG_KEEPALIVE_PENDING, &con->flags) == 0 &&
2602             test_and_set_bit(CON_FLAG_WRITE_PENDING, &con->flags) == 0)
2603                 queue_con(con);
2604 }
2605 EXPORT_SYMBOL(ceph_con_keepalive);
2606
2607
2608 /*
2609  * construct a new message with given type, size
2610  * the new msg has a ref count of 1.
2611  */
2612 struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
2613                               bool can_fail)
2614 {
2615         struct ceph_msg *m;
2616
2617         m = kmalloc(sizeof(*m), flags);
2618         if (m == NULL)
2619                 goto out;
2620         kref_init(&m->kref);
2621
2622         m->con = NULL;
2623         INIT_LIST_HEAD(&m->list_head);
2624
2625         m->hdr.tid = 0;
2626         m->hdr.type = cpu_to_le16(type);
2627         m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
2628         m->hdr.version = 0;
2629         m->hdr.front_len = cpu_to_le32(front_len);
2630         m->hdr.middle_len = 0;
2631         m->hdr.data_len = 0;
2632         m->hdr.data_off = 0;
2633         m->hdr.reserved = 0;
2634         m->footer.front_crc = 0;
2635         m->footer.middle_crc = 0;
2636         m->footer.data_crc = 0;
2637         m->footer.flags = 0;
2638         m->front_max = front_len;
2639         m->front_is_vmalloc = false;
2640         m->more_to_follow = false;
2641         m->ack_stamp = 0;
2642         m->pool = NULL;
2643
2644         /* middle */
2645         m->middle = NULL;
2646
2647         /* data */
2648         m->nr_pages = 0;
2649         m->page_alignment = 0;
2650         m->pages = NULL;
2651         m->pagelist = NULL;
2652         m->bio = NULL;
2653         m->bio_iter = NULL;
2654         m->bio_seg = 0;
2655         m->trail = NULL;
2656
2657         /* front */
2658         if (front_len) {
2659                 if (front_len > PAGE_CACHE_SIZE) {
2660                         m->front.iov_base = __vmalloc(front_len, flags,
2661                                                       PAGE_KERNEL);
2662                         m->front_is_vmalloc = true;
2663                 } else {
2664                         m->front.iov_base = kmalloc(front_len, flags);
2665                 }
2666                 if (m->front.iov_base == NULL) {
2667                         dout("ceph_msg_new can't allocate %d bytes\n",
2668                              front_len);
2669                         goto out2;
2670                 }
2671         } else {
2672                 m->front.iov_base = NULL;
2673         }
2674         m->front.iov_len = front_len;
2675
2676         dout("ceph_msg_new %p front %d\n", m, front_len);
2677         return m;
2678
2679 out2:
2680         ceph_msg_put(m);
2681 out:
2682         if (!can_fail) {
2683                 pr_err("msg_new can't create type %d front %d\n", type,
2684                        front_len);
2685                 WARN_ON(1);
2686         } else {
2687                 dout("msg_new can't create type %d front %d\n", type,
2688                      front_len);
2689         }
2690         return NULL;
2691 }
2692 EXPORT_SYMBOL(ceph_msg_new);
2693
2694 /*
2695  * Allocate "middle" portion of a message, if it is needed and wasn't
2696  * allocated by alloc_msg.  This allows us to read a small fixed-size
2697  * per-type header in the front and then gracefully fail (i.e.,
2698  * propagate the error to the caller based on info in the front) when
2699  * the middle is too large.
2700  */
2701 static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
2702 {
2703         int type = le16_to_cpu(msg->hdr.type);
2704         int middle_len = le32_to_cpu(msg->hdr.middle_len);
2705
2706         dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
2707              ceph_msg_type_name(type), middle_len);
2708         BUG_ON(!middle_len);
2709         BUG_ON(msg->middle);
2710
2711         msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
2712         if (!msg->middle)
2713                 return -ENOMEM;
2714         return 0;
2715 }
2716
2717 /*
2718  * Allocate a message for receiving an incoming message on a
2719  * connection, and save the result in con->in_msg.  Uses the
2720  * connection's private alloc_msg op if available.
2721  *
2722  * Returns 0 on success, or a negative error code.
2723  *
2724  * On success, if we set *skip = 1:
2725  *  - the next message should be skipped and ignored.
2726  *  - con->in_msg == NULL
2727  * or if we set *skip = 0:
2728  *  - con->in_msg is non-null.
2729  * On error (ENOMEM, EAGAIN, ...),
2730  *  - con->in_msg == NULL
2731  */
2732 static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
2733 {
2734         struct ceph_msg_header *hdr = &con->in_hdr;
2735         int type = le16_to_cpu(hdr->type);
2736         int front_len = le32_to_cpu(hdr->front_len);
2737         int middle_len = le32_to_cpu(hdr->middle_len);
2738         int ret = 0;
2739
2740         BUG_ON(con->in_msg != NULL);
2741
2742         if (con->ops->alloc_msg) {
2743                 struct ceph_msg *msg;
2744
2745                 mutex_unlock(&con->mutex);
2746                 msg = con->ops->alloc_msg(con, hdr, skip);
2747                 mutex_lock(&con->mutex);
2748                 if (con->state != CON_STATE_OPEN) {
2749                         if (msg)
2750                                 ceph_msg_put(msg);
2751                         return -EAGAIN;
2752                 }
2753                 con->in_msg = msg;
2754                 if (con->in_msg) {
2755                         con->in_msg->con = con->ops->get(con);
2756                         BUG_ON(con->in_msg->con == NULL);
2757                 }
2758                 if (*skip) {
2759                         con->in_msg = NULL;
2760                         return 0;
2761                 }
2762                 if (!con->in_msg) {
2763                         con->error_msg =
2764                                 "error allocating memory for incoming message";
2765                         return -ENOMEM;
2766                 }
2767         }
2768         if (!con->in_msg) {
2769                 con->in_msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
2770                 if (!con->in_msg) {
2771                         pr_err("unable to allocate msg type %d len %d\n",
2772                                type, front_len);
2773                         return -ENOMEM;
2774                 }
2775                 con->in_msg->con = con->ops->get(con);
2776                 BUG_ON(con->in_msg->con == NULL);
2777                 con->in_msg->page_alignment = le16_to_cpu(hdr->data_off);
2778         }
2779         memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2780
2781         if (middle_len && !con->in_msg->middle) {
2782                 ret = ceph_alloc_middle(con, con->in_msg);
2783                 if (ret < 0) {
2784                         ceph_msg_put(con->in_msg);
2785                         con->in_msg = NULL;
2786                 }
2787         }
2788
2789         return ret;
2790 }
2791
2792
2793 /*
2794  * Free a generically kmalloc'd message.
2795  */
2796 void ceph_msg_kfree(struct ceph_msg *m)
2797 {
2798         dout("msg_kfree %p\n", m);
2799         if (m->front_is_vmalloc)
2800                 vfree(m->front.iov_base);
2801         else
2802                 kfree(m->front.iov_base);
2803         kfree(m);
2804 }
2805
2806 /*
2807  * Drop a msg ref.  Destroy as needed.
2808  */
2809 void ceph_msg_last_put(struct kref *kref)
2810 {
2811         struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
2812
2813         dout("ceph_msg_put last one on %p\n", m);
2814         WARN_ON(!list_empty(&m->list_head));
2815
2816         /* drop middle, data, if any */
2817         if (m->middle) {
2818                 ceph_buffer_put(m->middle);
2819                 m->middle = NULL;
2820         }
2821         m->nr_pages = 0;
2822         m->pages = NULL;
2823
2824         if (m->pagelist) {
2825                 ceph_pagelist_release(m->pagelist);
2826                 kfree(m->pagelist);
2827                 m->pagelist = NULL;
2828         }
2829
2830         m->trail = NULL;
2831
2832         if (m->pool)
2833                 ceph_msgpool_put(m->pool, m);
2834         else
2835                 ceph_msg_kfree(m);
2836 }
2837 EXPORT_SYMBOL(ceph_msg_last_put);
2838
2839 void ceph_msg_dump(struct ceph_msg *msg)
2840 {
2841         pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
2842                  msg->front_max, msg->nr_pages);
2843         print_hex_dump(KERN_DEBUG, "header: ",
2844                        DUMP_PREFIX_OFFSET, 16, 1,
2845                        &msg->hdr, sizeof(msg->hdr), true);
2846         print_hex_dump(KERN_DEBUG, " front: ",
2847                        DUMP_PREFIX_OFFSET, 16, 1,
2848                        msg->front.iov_base, msg->front.iov_len, true);
2849         if (msg->middle)
2850                 print_hex_dump(KERN_DEBUG, "middle: ",
2851                                DUMP_PREFIX_OFFSET, 16, 1,
2852                                msg->middle->vec.iov_base,
2853                                msg->middle->vec.iov_len, true);
2854         print_hex_dump(KERN_DEBUG, "footer: ",
2855                        DUMP_PREFIX_OFFSET, 16, 1,
2856                        &msg->footer, sizeof(msg->footer), true);
2857 }
2858 EXPORT_SYMBOL(ceph_msg_dump);