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