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