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