libceph: encapsulate reading message data
[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_message_data(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("%s %p msg %p page %d offset %d\n", __func__,
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("%s %p msg %p done\n", __func__, 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 static int read_partial_msg_data(struct ceph_connection *con)
1873 {
1874         struct ceph_msg *msg = con->in_msg;
1875         struct ceph_msg_pos *msg_pos = &con->in_msg_pos;
1876         const bool do_datacrc = !con->msgr->nocrc;
1877         unsigned int data_len;
1878         int ret;
1879
1880         BUG_ON(!msg);
1881
1882         data_len = le32_to_cpu(con->in_hdr.data_len);
1883         while (msg_pos->data_pos < data_len) {
1884                 if (msg->pages) {
1885                         ret = read_partial_message_pages(con, msg->pages,
1886                                                  data_len, do_datacrc);
1887                         if (ret <= 0)
1888                                 return ret;
1889 #ifdef CONFIG_BLOCK
1890                 } else if (msg->bio) {
1891                         ret = read_partial_message_bio(con,
1892                                                  data_len, do_datacrc);
1893                         if (ret <= 0)
1894                                 return ret;
1895 #endif
1896                 } else {
1897                         BUG_ON(1);
1898                 }
1899         }
1900
1901         return 1;       /* must return > 0 to indicate success */
1902 }
1903
1904 /*
1905  * read (part of) a message.
1906  */
1907 static int read_partial_message(struct ceph_connection *con)
1908 {
1909         struct ceph_msg *m = con->in_msg;
1910         int size;
1911         int end;
1912         int ret;
1913         unsigned int front_len, middle_len, data_len;
1914         bool do_datacrc = !con->msgr->nocrc;
1915         u64 seq;
1916         u32 crc;
1917
1918         dout("read_partial_message con %p msg %p\n", con, m);
1919
1920         /* header */
1921         size = sizeof (con->in_hdr);
1922         end = size;
1923         ret = read_partial(con, end, size, &con->in_hdr);
1924         if (ret <= 0)
1925                 return ret;
1926
1927         crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
1928         if (cpu_to_le32(crc) != con->in_hdr.crc) {
1929                 pr_err("read_partial_message bad hdr "
1930                        " crc %u != expected %u\n",
1931                        crc, con->in_hdr.crc);
1932                 return -EBADMSG;
1933         }
1934
1935         front_len = le32_to_cpu(con->in_hdr.front_len);
1936         if (front_len > CEPH_MSG_MAX_FRONT_LEN)
1937                 return -EIO;
1938         middle_len = le32_to_cpu(con->in_hdr.middle_len);
1939         if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
1940                 return -EIO;
1941         data_len = le32_to_cpu(con->in_hdr.data_len);
1942         if (data_len > CEPH_MSG_MAX_DATA_LEN)
1943                 return -EIO;
1944
1945         /* verify seq# */
1946         seq = le64_to_cpu(con->in_hdr.seq);
1947         if ((s64)seq - (s64)con->in_seq < 1) {
1948                 pr_info("skipping %s%lld %s seq %lld expected %lld\n",
1949                         ENTITY_NAME(con->peer_name),
1950                         ceph_pr_addr(&con->peer_addr.in_addr),
1951                         seq, con->in_seq + 1);
1952                 con->in_base_pos = -front_len - middle_len - data_len -
1953                         sizeof(m->footer);
1954                 con->in_tag = CEPH_MSGR_TAG_READY;
1955                 return 0;
1956         } else if ((s64)seq - (s64)con->in_seq > 1) {
1957                 pr_err("read_partial_message bad seq %lld expected %lld\n",
1958                        seq, con->in_seq + 1);
1959                 con->error_msg = "bad message sequence # for incoming message";
1960                 return -EBADMSG;
1961         }
1962
1963         /* allocate message? */
1964         if (!con->in_msg) {
1965                 int skip = 0;
1966
1967                 dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
1968                      front_len, data_len);
1969                 ret = ceph_con_in_msg_alloc(con, &skip);
1970                 if (ret < 0)
1971                         return ret;
1972                 if (skip) {
1973                         /* skip this message */
1974                         dout("alloc_msg said skip message\n");
1975                         BUG_ON(con->in_msg);
1976                         con->in_base_pos = -front_len - middle_len - data_len -
1977                                 sizeof(m->footer);
1978                         con->in_tag = CEPH_MSGR_TAG_READY;
1979                         con->in_seq++;
1980                         return 0;
1981                 }
1982
1983                 BUG_ON(!con->in_msg);
1984                 BUG_ON(con->in_msg->con != con);
1985                 m = con->in_msg;
1986                 m->front.iov_len = 0;    /* haven't read it yet */
1987                 if (m->middle)
1988                         m->middle->vec.iov_len = 0;
1989
1990                 /* prepare for data payload, if any */
1991
1992                 if (data_len)
1993                         prepare_message_data(con->in_msg, &con->in_msg_pos);
1994         }
1995
1996         /* front */
1997         ret = read_partial_message_section(con, &m->front, front_len,
1998                                            &con->in_front_crc);
1999         if (ret <= 0)
2000                 return ret;
2001
2002         /* middle */
2003         if (m->middle) {
2004                 ret = read_partial_message_section(con, &m->middle->vec,
2005                                                    middle_len,
2006                                                    &con->in_middle_crc);
2007                 if (ret <= 0)
2008                         return ret;
2009         }
2010
2011         /* (page) data */
2012         if (data_len) {
2013                 ret = read_partial_msg_data(con);
2014                 if (ret <= 0)
2015                         return ret;
2016         }
2017
2018         /* footer */
2019         size = sizeof (m->footer);
2020         end += size;
2021         ret = read_partial(con, end, size, &m->footer);
2022         if (ret <= 0)
2023                 return ret;
2024
2025         dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
2026              m, front_len, m->footer.front_crc, middle_len,
2027              m->footer.middle_crc, data_len, m->footer.data_crc);
2028
2029         /* crc ok? */
2030         if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
2031                 pr_err("read_partial_message %p front crc %u != exp. %u\n",
2032                        m, con->in_front_crc, m->footer.front_crc);
2033                 return -EBADMSG;
2034         }
2035         if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
2036                 pr_err("read_partial_message %p middle crc %u != exp %u\n",
2037                        m, con->in_middle_crc, m->footer.middle_crc);
2038                 return -EBADMSG;
2039         }
2040         if (do_datacrc &&
2041             (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
2042             con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
2043                 pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
2044                        con->in_data_crc, le32_to_cpu(m->footer.data_crc));
2045                 return -EBADMSG;
2046         }
2047
2048         return 1; /* done! */
2049 }
2050
2051 /*
2052  * Process message.  This happens in the worker thread.  The callback should
2053  * be careful not to do anything that waits on other incoming messages or it
2054  * may deadlock.
2055  */
2056 static void process_message(struct ceph_connection *con)
2057 {
2058         struct ceph_msg *msg;
2059
2060         BUG_ON(con->in_msg->con != con);
2061         con->in_msg->con = NULL;
2062         msg = con->in_msg;
2063         con->in_msg = NULL;
2064         con->ops->put(con);
2065
2066         /* if first message, set peer_name */
2067         if (con->peer_name.type == 0)
2068                 con->peer_name = msg->hdr.src;
2069
2070         con->in_seq++;
2071         mutex_unlock(&con->mutex);
2072
2073         dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
2074              msg, le64_to_cpu(msg->hdr.seq),
2075              ENTITY_NAME(msg->hdr.src),
2076              le16_to_cpu(msg->hdr.type),
2077              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2078              le32_to_cpu(msg->hdr.front_len),
2079              le32_to_cpu(msg->hdr.data_len),
2080              con->in_front_crc, con->in_middle_crc, con->in_data_crc);
2081         con->ops->dispatch(con, msg);
2082
2083         mutex_lock(&con->mutex);
2084 }
2085
2086
2087 /*
2088  * Write something to the socket.  Called in a worker thread when the
2089  * socket appears to be writeable and we have something ready to send.
2090  */
2091 static int try_write(struct ceph_connection *con)
2092 {
2093         int ret = 1;
2094
2095         dout("try_write start %p state %lu\n", con, con->state);
2096
2097 more:
2098         dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
2099
2100         /* open the socket first? */
2101         if (con->state == CON_STATE_PREOPEN) {
2102                 BUG_ON(con->sock);
2103                 con->state = CON_STATE_CONNECTING;
2104
2105                 con_out_kvec_reset(con);
2106                 prepare_write_banner(con);
2107                 prepare_read_banner(con);
2108
2109                 BUG_ON(con->in_msg);
2110                 con->in_tag = CEPH_MSGR_TAG_READY;
2111                 dout("try_write initiating connect on %p new state %lu\n",
2112                      con, con->state);
2113                 ret = ceph_tcp_connect(con);
2114                 if (ret < 0) {
2115                         con->error_msg = "connect error";
2116                         goto out;
2117                 }
2118         }
2119
2120 more_kvec:
2121         /* kvec data queued? */
2122         if (con->out_skip) {
2123                 ret = write_partial_skip(con);
2124                 if (ret <= 0)
2125                         goto out;
2126         }
2127         if (con->out_kvec_left) {
2128                 ret = write_partial_kvec(con);
2129                 if (ret <= 0)
2130                         goto out;
2131         }
2132
2133         /* msg pages? */
2134         if (con->out_msg) {
2135                 if (con->out_msg_done) {
2136                         ceph_msg_put(con->out_msg);
2137                         con->out_msg = NULL;   /* we're done with this one */
2138                         goto do_next;
2139                 }
2140
2141                 ret = write_partial_message_data(con);
2142                 if (ret == 1)
2143                         goto more_kvec;  /* we need to send the footer, too! */
2144                 if (ret == 0)
2145                         goto out;
2146                 if (ret < 0) {
2147                         dout("try_write write_partial_message_data err %d\n",
2148                              ret);
2149                         goto out;
2150                 }
2151         }
2152
2153 do_next:
2154         if (con->state == CON_STATE_OPEN) {
2155                 /* is anything else pending? */
2156                 if (!list_empty(&con->out_queue)) {
2157                         prepare_write_message(con);
2158                         goto more;
2159                 }
2160                 if (con->in_seq > con->in_seq_acked) {
2161                         prepare_write_ack(con);
2162                         goto more;
2163                 }
2164                 if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
2165                         prepare_write_keepalive(con);
2166                         goto more;
2167                 }
2168         }
2169
2170         /* Nothing to do! */
2171         con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2172         dout("try_write nothing else to write.\n");
2173         ret = 0;
2174 out:
2175         dout("try_write done on %p ret %d\n", con, ret);
2176         return ret;
2177 }
2178
2179
2180
2181 /*
2182  * Read what we can from the socket.
2183  */
2184 static int try_read(struct ceph_connection *con)
2185 {
2186         int ret = -1;
2187
2188 more:
2189         dout("try_read start on %p state %lu\n", con, con->state);
2190         if (con->state != CON_STATE_CONNECTING &&
2191             con->state != CON_STATE_NEGOTIATING &&
2192             con->state != CON_STATE_OPEN)
2193                 return 0;
2194
2195         BUG_ON(!con->sock);
2196
2197         dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
2198              con->in_base_pos);
2199
2200         if (con->state == CON_STATE_CONNECTING) {
2201                 dout("try_read connecting\n");
2202                 ret = read_partial_banner(con);
2203                 if (ret <= 0)
2204                         goto out;
2205                 ret = process_banner(con);
2206                 if (ret < 0)
2207                         goto out;
2208
2209                 con->state = CON_STATE_NEGOTIATING;
2210
2211                 /*
2212                  * Received banner is good, exchange connection info.
2213                  * Do not reset out_kvec, as sending our banner raced
2214                  * with receiving peer banner after connect completed.
2215                  */
2216                 ret = prepare_write_connect(con);
2217                 if (ret < 0)
2218                         goto out;
2219                 prepare_read_connect(con);
2220
2221                 /* Send connection info before awaiting response */
2222                 goto out;
2223         }
2224
2225         if (con->state == CON_STATE_NEGOTIATING) {
2226                 dout("try_read negotiating\n");
2227                 ret = read_partial_connect(con);
2228                 if (ret <= 0)
2229                         goto out;
2230                 ret = process_connect(con);
2231                 if (ret < 0)
2232                         goto out;
2233                 goto more;
2234         }
2235
2236         WARN_ON(con->state != CON_STATE_OPEN);
2237
2238         if (con->in_base_pos < 0) {
2239                 /*
2240                  * skipping + discarding content.
2241                  *
2242                  * FIXME: there must be a better way to do this!
2243                  */
2244                 static char buf[SKIP_BUF_SIZE];
2245                 int skip = min((int) sizeof (buf), -con->in_base_pos);
2246
2247                 dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
2248                 ret = ceph_tcp_recvmsg(con->sock, buf, skip);
2249                 if (ret <= 0)
2250                         goto out;
2251                 con->in_base_pos += ret;
2252                 if (con->in_base_pos)
2253                         goto more;
2254         }
2255         if (con->in_tag == CEPH_MSGR_TAG_READY) {
2256                 /*
2257                  * what's next?
2258                  */
2259                 ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
2260                 if (ret <= 0)
2261                         goto out;
2262                 dout("try_read got tag %d\n", (int)con->in_tag);
2263                 switch (con->in_tag) {
2264                 case CEPH_MSGR_TAG_MSG:
2265                         prepare_read_message(con);
2266                         break;
2267                 case CEPH_MSGR_TAG_ACK:
2268                         prepare_read_ack(con);
2269                         break;
2270                 case CEPH_MSGR_TAG_CLOSE:
2271                         con_close_socket(con);
2272                         con->state = CON_STATE_CLOSED;
2273                         goto out;
2274                 default:
2275                         goto bad_tag;
2276                 }
2277         }
2278         if (con->in_tag == CEPH_MSGR_TAG_MSG) {
2279                 ret = read_partial_message(con);
2280                 if (ret <= 0) {
2281                         switch (ret) {
2282                         case -EBADMSG:
2283                                 con->error_msg = "bad crc";
2284                                 ret = -EIO;
2285                                 break;
2286                         case -EIO:
2287                                 con->error_msg = "io error";
2288                                 break;
2289                         }
2290                         goto out;
2291                 }
2292                 if (con->in_tag == CEPH_MSGR_TAG_READY)
2293                         goto more;
2294                 process_message(con);
2295                 if (con->state == CON_STATE_OPEN)
2296                         prepare_read_tag(con);
2297                 goto more;
2298         }
2299         if (con->in_tag == CEPH_MSGR_TAG_ACK) {
2300                 ret = read_partial_ack(con);
2301                 if (ret <= 0)
2302                         goto out;
2303                 process_ack(con);
2304                 goto more;
2305         }
2306
2307 out:
2308         dout("try_read done on %p ret %d\n", con, ret);
2309         return ret;
2310
2311 bad_tag:
2312         pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
2313         con->error_msg = "protocol error, garbage tag";
2314         ret = -1;
2315         goto out;
2316 }
2317
2318
2319 /*
2320  * Atomically queue work on a connection after the specified delay.
2321  * Bump @con reference to avoid races with connection teardown.
2322  * Returns 0 if work was queued, or an error code otherwise.
2323  */
2324 static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
2325 {
2326         if (!con->ops->get(con)) {
2327                 dout("%s %p ref count 0\n", __func__, con);
2328
2329                 return -ENOENT;
2330         }
2331
2332         if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
2333                 dout("%s %p - already queued\n", __func__, con);
2334                 con->ops->put(con);
2335
2336                 return -EBUSY;
2337         }
2338
2339         dout("%s %p %lu\n", __func__, con, delay);
2340
2341         return 0;
2342 }
2343
2344 static void queue_con(struct ceph_connection *con)
2345 {
2346         (void) queue_con_delay(con, 0);
2347 }
2348
2349 static bool con_sock_closed(struct ceph_connection *con)
2350 {
2351         if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2352                 return false;
2353
2354 #define CASE(x)                                                         \
2355         case CON_STATE_ ## x:                                           \
2356                 con->error_msg = "socket closed (con state " #x ")";    \
2357                 break;
2358
2359         switch (con->state) {
2360         CASE(CLOSED);
2361         CASE(PREOPEN);
2362         CASE(CONNECTING);
2363         CASE(NEGOTIATING);
2364         CASE(OPEN);
2365         CASE(STANDBY);
2366         default:
2367                 pr_warning("%s con %p unrecognized state %lu\n",
2368                         __func__, con, con->state);
2369                 con->error_msg = "unrecognized con state";
2370                 BUG();
2371                 break;
2372         }
2373 #undef CASE
2374
2375         return true;
2376 }
2377
2378 static bool con_backoff(struct ceph_connection *con)
2379 {
2380         int ret;
2381
2382         if (!con_flag_test_and_clear(con, CON_FLAG_BACKOFF))
2383                 return false;
2384
2385         ret = queue_con_delay(con, round_jiffies_relative(con->delay));
2386         if (ret) {
2387                 dout("%s: con %p FAILED to back off %lu\n", __func__,
2388                         con, con->delay);
2389                 BUG_ON(ret == -ENOENT);
2390                 con_flag_set(con, CON_FLAG_BACKOFF);
2391         }
2392
2393         return true;
2394 }
2395
2396 /* Finish fault handling; con->mutex must *not* be held here */
2397
2398 static void con_fault_finish(struct ceph_connection *con)
2399 {
2400         /*
2401          * in case we faulted due to authentication, invalidate our
2402          * current tickets so that we can get new ones.
2403          */
2404         if (con->auth_retry && con->ops->invalidate_authorizer) {
2405                 dout("calling invalidate_authorizer()\n");
2406                 con->ops->invalidate_authorizer(con);
2407         }
2408
2409         if (con->ops->fault)
2410                 con->ops->fault(con);
2411 }
2412
2413 /*
2414  * Do some work on a connection.  Drop a connection ref when we're done.
2415  */
2416 static void con_work(struct work_struct *work)
2417 {
2418         struct ceph_connection *con = container_of(work, struct ceph_connection,
2419                                                    work.work);
2420         bool fault;
2421
2422         mutex_lock(&con->mutex);
2423         while (true) {
2424                 int ret;
2425
2426                 if ((fault = con_sock_closed(con))) {
2427                         dout("%s: con %p SOCK_CLOSED\n", __func__, con);
2428                         break;
2429                 }
2430                 if (con_backoff(con)) {
2431                         dout("%s: con %p BACKOFF\n", __func__, con);
2432                         break;
2433                 }
2434                 if (con->state == CON_STATE_STANDBY) {
2435                         dout("%s: con %p STANDBY\n", __func__, con);
2436                         break;
2437                 }
2438                 if (con->state == CON_STATE_CLOSED) {
2439                         dout("%s: con %p CLOSED\n", __func__, con);
2440                         BUG_ON(con->sock);
2441                         break;
2442                 }
2443                 if (con->state == CON_STATE_PREOPEN) {
2444                         dout("%s: con %p PREOPEN\n", __func__, con);
2445                         BUG_ON(con->sock);
2446                 }
2447
2448                 ret = try_read(con);
2449                 if (ret < 0) {
2450                         if (ret == -EAGAIN)
2451                                 continue;
2452                         con->error_msg = "socket error on read";
2453                         fault = true;
2454                         break;
2455                 }
2456
2457                 ret = try_write(con);
2458                 if (ret < 0) {
2459                         if (ret == -EAGAIN)
2460                                 continue;
2461                         con->error_msg = "socket error on write";
2462                         fault = true;
2463                 }
2464
2465                 break;  /* If we make it to here, we're done */
2466         }
2467         if (fault)
2468                 con_fault(con);
2469         mutex_unlock(&con->mutex);
2470
2471         if (fault)
2472                 con_fault_finish(con);
2473
2474         con->ops->put(con);
2475 }
2476
2477 /*
2478  * Generic error/fault handler.  A retry mechanism is used with
2479  * exponential backoff
2480  */
2481 static void con_fault(struct ceph_connection *con)
2482 {
2483         pr_warning("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
2484                ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
2485         dout("fault %p state %lu to peer %s\n",
2486              con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
2487
2488         WARN_ON(con->state != CON_STATE_CONNECTING &&
2489                con->state != CON_STATE_NEGOTIATING &&
2490                con->state != CON_STATE_OPEN);
2491
2492         con_close_socket(con);
2493
2494         if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2495                 dout("fault on LOSSYTX channel, marking CLOSED\n");
2496                 con->state = CON_STATE_CLOSED;
2497                 return;
2498         }
2499
2500         if (con->in_msg) {
2501                 BUG_ON(con->in_msg->con != con);
2502                 con->in_msg->con = NULL;
2503                 ceph_msg_put(con->in_msg);
2504                 con->in_msg = NULL;
2505                 con->ops->put(con);
2506         }
2507
2508         /* Requeue anything that hasn't been acked */
2509         list_splice_init(&con->out_sent, &con->out_queue);
2510
2511         /* If there are no messages queued or keepalive pending, place
2512          * the connection in a STANDBY state */
2513         if (list_empty(&con->out_queue) &&
2514             !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
2515                 dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2516                 con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2517                 con->state = CON_STATE_STANDBY;
2518         } else {
2519                 /* retry after a delay. */
2520                 con->state = CON_STATE_PREOPEN;
2521                 if (con->delay == 0)
2522                         con->delay = BASE_DELAY_INTERVAL;
2523                 else if (con->delay < MAX_DELAY_INTERVAL)
2524                         con->delay *= 2;
2525                 con_flag_set(con, CON_FLAG_BACKOFF);
2526                 queue_con(con);
2527         }
2528 }
2529
2530
2531
2532 /*
2533  * initialize a new messenger instance
2534  */
2535 void ceph_messenger_init(struct ceph_messenger *msgr,
2536                         struct ceph_entity_addr *myaddr,
2537                         u32 supported_features,
2538                         u32 required_features,
2539                         bool nocrc)
2540 {
2541         msgr->supported_features = supported_features;
2542         msgr->required_features = required_features;
2543
2544         spin_lock_init(&msgr->global_seq_lock);
2545
2546         if (myaddr)
2547                 msgr->inst.addr = *myaddr;
2548
2549         /* select a random nonce */
2550         msgr->inst.addr.type = 0;
2551         get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2552         encode_my_addr(msgr);
2553         msgr->nocrc = nocrc;
2554
2555         atomic_set(&msgr->stopping, 0);
2556
2557         dout("%s %p\n", __func__, msgr);
2558 }
2559 EXPORT_SYMBOL(ceph_messenger_init);
2560
2561 static void clear_standby(struct ceph_connection *con)
2562 {
2563         /* come back from STANDBY? */
2564         if (con->state == CON_STATE_STANDBY) {
2565                 dout("clear_standby %p and ++connect_seq\n", con);
2566                 con->state = CON_STATE_PREOPEN;
2567                 con->connect_seq++;
2568                 WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
2569                 WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
2570         }
2571 }
2572
2573 /*
2574  * Queue up an outgoing message on the given connection.
2575  */
2576 void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
2577 {
2578         /* set src+dst */
2579         msg->hdr.src = con->msgr->inst.name;
2580         BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
2581         msg->needs_out_seq = true;
2582
2583         mutex_lock(&con->mutex);
2584
2585         if (con->state == CON_STATE_CLOSED) {
2586                 dout("con_send %p closed, dropping %p\n", con, msg);
2587                 ceph_msg_put(msg);
2588                 mutex_unlock(&con->mutex);
2589                 return;
2590         }
2591
2592         BUG_ON(msg->con != NULL);
2593         msg->con = con->ops->get(con);
2594         BUG_ON(msg->con == NULL);
2595
2596         BUG_ON(!list_empty(&msg->list_head));
2597         list_add_tail(&msg->list_head, &con->out_queue);
2598         dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
2599              ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
2600              ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2601              le32_to_cpu(msg->hdr.front_len),
2602              le32_to_cpu(msg->hdr.middle_len),
2603              le32_to_cpu(msg->hdr.data_len));
2604
2605         clear_standby(con);
2606         mutex_unlock(&con->mutex);
2607
2608         /* if there wasn't anything waiting to send before, queue
2609          * new work */
2610         if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
2611                 queue_con(con);
2612 }
2613 EXPORT_SYMBOL(ceph_con_send);
2614
2615 /*
2616  * Revoke a message that was previously queued for send
2617  */
2618 void ceph_msg_revoke(struct ceph_msg *msg)
2619 {
2620         struct ceph_connection *con = msg->con;
2621
2622         if (!con)
2623                 return;         /* Message not in our possession */
2624
2625         mutex_lock(&con->mutex);
2626         if (!list_empty(&msg->list_head)) {
2627                 dout("%s %p msg %p - was on queue\n", __func__, con, msg);
2628                 list_del_init(&msg->list_head);
2629                 BUG_ON(msg->con == NULL);
2630                 msg->con->ops->put(msg->con);
2631                 msg->con = NULL;
2632                 msg->hdr.seq = 0;
2633
2634                 ceph_msg_put(msg);
2635         }
2636         if (con->out_msg == msg) {
2637                 dout("%s %p msg %p - was sending\n", __func__, con, msg);
2638                 con->out_msg = NULL;
2639                 if (con->out_kvec_is_msg) {
2640                         con->out_skip = con->out_kvec_bytes;
2641                         con->out_kvec_is_msg = false;
2642                 }
2643                 msg->hdr.seq = 0;
2644
2645                 ceph_msg_put(msg);
2646         }
2647         mutex_unlock(&con->mutex);
2648 }
2649
2650 /*
2651  * Revoke a message that we may be reading data into
2652  */
2653 void ceph_msg_revoke_incoming(struct ceph_msg *msg)
2654 {
2655         struct ceph_connection *con;
2656
2657         BUG_ON(msg == NULL);
2658         if (!msg->con) {
2659                 dout("%s msg %p null con\n", __func__, msg);
2660
2661                 return;         /* Message not in our possession */
2662         }
2663
2664         con = msg->con;
2665         mutex_lock(&con->mutex);
2666         if (con->in_msg == msg) {
2667                 unsigned int front_len = le32_to_cpu(con->in_hdr.front_len);
2668                 unsigned int middle_len = le32_to_cpu(con->in_hdr.middle_len);
2669                 unsigned int data_len = le32_to_cpu(con->in_hdr.data_len);
2670
2671                 /* skip rest of message */
2672                 dout("%s %p msg %p revoked\n", __func__, con, msg);
2673                 con->in_base_pos = con->in_base_pos -
2674                                 sizeof(struct ceph_msg_header) -
2675                                 front_len -
2676                                 middle_len -
2677                                 data_len -
2678                                 sizeof(struct ceph_msg_footer);
2679                 ceph_msg_put(con->in_msg);
2680                 con->in_msg = NULL;
2681                 con->in_tag = CEPH_MSGR_TAG_READY;
2682                 con->in_seq++;
2683         } else {
2684                 dout("%s %p in_msg %p msg %p no-op\n",
2685                      __func__, con, con->in_msg, msg);
2686         }
2687         mutex_unlock(&con->mutex);
2688 }
2689
2690 /*
2691  * Queue a keepalive byte to ensure the tcp connection is alive.
2692  */
2693 void ceph_con_keepalive(struct ceph_connection *con)
2694 {
2695         dout("con_keepalive %p\n", con);
2696         mutex_lock(&con->mutex);
2697         clear_standby(con);
2698         mutex_unlock(&con->mutex);
2699         if (con_flag_test_and_set(con, CON_FLAG_KEEPALIVE_PENDING) == 0 &&
2700             con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
2701                 queue_con(con);
2702 }
2703 EXPORT_SYMBOL(ceph_con_keepalive);
2704
2705 void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,
2706                 size_t length, size_t alignment)
2707 {
2708         BUG_ON(!pages);
2709         BUG_ON(!length);
2710         BUG_ON(msg->pages);
2711         BUG_ON(msg->length);
2712
2713         msg->pages = pages;
2714         msg->length = length;
2715         msg->page_alignment = alignment & ~PAGE_MASK;
2716 }
2717 EXPORT_SYMBOL(ceph_msg_data_set_pages);
2718
2719 void ceph_msg_data_set_pagelist(struct ceph_msg *msg,
2720                                 struct ceph_pagelist *pagelist)
2721 {
2722         BUG_ON(!pagelist);
2723         BUG_ON(!pagelist->length);
2724         BUG_ON(msg->pagelist);
2725
2726         msg->pagelist = pagelist;
2727 }
2728 EXPORT_SYMBOL(ceph_msg_data_set_pagelist);
2729
2730 void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio)
2731 {
2732         BUG_ON(!bio);
2733         BUG_ON(msg->bio);
2734
2735         msg->bio = bio;
2736 }
2737 EXPORT_SYMBOL(ceph_msg_data_set_bio);
2738
2739 void ceph_msg_data_set_trail(struct ceph_msg *msg, struct ceph_pagelist *trail)
2740 {
2741         BUG_ON(!trail);
2742         BUG_ON(!trail->length);
2743         BUG_ON(msg->trail);
2744
2745         msg->trail = trail;
2746 }
2747 EXPORT_SYMBOL(ceph_msg_data_set_trail);
2748
2749 /*
2750  * construct a new message with given type, size
2751  * the new msg has a ref count of 1.
2752  */
2753 struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
2754                               bool can_fail)
2755 {
2756         struct ceph_msg *m;
2757
2758         m = kzalloc(sizeof(*m), flags);
2759         if (m == NULL)
2760                 goto out;
2761
2762         m->hdr.type = cpu_to_le16(type);
2763         m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
2764         m->hdr.front_len = cpu_to_le32(front_len);
2765
2766         INIT_LIST_HEAD(&m->list_head);
2767         kref_init(&m->kref);
2768
2769         /* front */
2770         m->front_max = front_len;
2771         if (front_len) {
2772                 if (front_len > PAGE_CACHE_SIZE) {
2773                         m->front.iov_base = __vmalloc(front_len, flags,
2774                                                       PAGE_KERNEL);
2775                         m->front_is_vmalloc = true;
2776                 } else {
2777                         m->front.iov_base = kmalloc(front_len, flags);
2778                 }
2779                 if (m->front.iov_base == NULL) {
2780                         dout("ceph_msg_new can't allocate %d bytes\n",
2781                              front_len);
2782                         goto out2;
2783                 }
2784         } else {
2785                 m->front.iov_base = NULL;
2786         }
2787         m->front.iov_len = front_len;
2788
2789         dout("ceph_msg_new %p front %d\n", m, front_len);
2790         return m;
2791
2792 out2:
2793         ceph_msg_put(m);
2794 out:
2795         if (!can_fail) {
2796                 pr_err("msg_new can't create type %d front %d\n", type,
2797                        front_len);
2798                 WARN_ON(1);
2799         } else {
2800                 dout("msg_new can't create type %d front %d\n", type,
2801                      front_len);
2802         }
2803         return NULL;
2804 }
2805 EXPORT_SYMBOL(ceph_msg_new);
2806
2807 /*
2808  * Allocate "middle" portion of a message, if it is needed and wasn't
2809  * allocated by alloc_msg.  This allows us to read a small fixed-size
2810  * per-type header in the front and then gracefully fail (i.e.,
2811  * propagate the error to the caller based on info in the front) when
2812  * the middle is too large.
2813  */
2814 static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
2815 {
2816         int type = le16_to_cpu(msg->hdr.type);
2817         int middle_len = le32_to_cpu(msg->hdr.middle_len);
2818
2819         dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
2820              ceph_msg_type_name(type), middle_len);
2821         BUG_ON(!middle_len);
2822         BUG_ON(msg->middle);
2823
2824         msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
2825         if (!msg->middle)
2826                 return -ENOMEM;
2827         return 0;
2828 }
2829
2830 /*
2831  * Allocate a message for receiving an incoming message on a
2832  * connection, and save the result in con->in_msg.  Uses the
2833  * connection's private alloc_msg op if available.
2834  *
2835  * Returns 0 on success, or a negative error code.
2836  *
2837  * On success, if we set *skip = 1:
2838  *  - the next message should be skipped and ignored.
2839  *  - con->in_msg == NULL
2840  * or if we set *skip = 0:
2841  *  - con->in_msg is non-null.
2842  * On error (ENOMEM, EAGAIN, ...),
2843  *  - con->in_msg == NULL
2844  */
2845 static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
2846 {
2847         struct ceph_msg_header *hdr = &con->in_hdr;
2848         int middle_len = le32_to_cpu(hdr->middle_len);
2849         struct ceph_msg *msg;
2850         int ret = 0;
2851
2852         BUG_ON(con->in_msg != NULL);
2853         BUG_ON(!con->ops->alloc_msg);
2854
2855         mutex_unlock(&con->mutex);
2856         msg = con->ops->alloc_msg(con, hdr, skip);
2857         mutex_lock(&con->mutex);
2858         if (con->state != CON_STATE_OPEN) {
2859                 if (msg)
2860                         ceph_msg_put(msg);
2861                 return -EAGAIN;
2862         }
2863         if (msg) {
2864                 BUG_ON(*skip);
2865                 con->in_msg = msg;
2866                 con->in_msg->con = con->ops->get(con);
2867                 BUG_ON(con->in_msg->con == NULL);
2868         } else {
2869                 /*
2870                  * Null message pointer means either we should skip
2871                  * this message or we couldn't allocate memory.  The
2872                  * former is not an error.
2873                  */
2874                 if (*skip)
2875                         return 0;
2876                 con->error_msg = "error allocating memory for incoming message";
2877
2878                 return -ENOMEM;
2879         }
2880         memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2881
2882         if (middle_len && !con->in_msg->middle) {
2883                 ret = ceph_alloc_middle(con, con->in_msg);
2884                 if (ret < 0) {
2885                         ceph_msg_put(con->in_msg);
2886                         con->in_msg = NULL;
2887                 }
2888         }
2889
2890         return ret;
2891 }
2892
2893
2894 /*
2895  * Free a generically kmalloc'd message.
2896  */
2897 void ceph_msg_kfree(struct ceph_msg *m)
2898 {
2899         dout("msg_kfree %p\n", m);
2900         if (m->front_is_vmalloc)
2901                 vfree(m->front.iov_base);
2902         else
2903                 kfree(m->front.iov_base);
2904         kfree(m);
2905 }
2906
2907 /*
2908  * Drop a msg ref.  Destroy as needed.
2909  */
2910 void ceph_msg_last_put(struct kref *kref)
2911 {
2912         struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
2913
2914         dout("ceph_msg_put last one on %p\n", m);
2915         WARN_ON(!list_empty(&m->list_head));
2916
2917         /* drop middle, data, if any */
2918         if (m->middle) {
2919                 ceph_buffer_put(m->middle);
2920                 m->middle = NULL;
2921         }
2922         m->length = 0;
2923         m->pages = NULL;
2924
2925         if (m->pagelist) {
2926                 ceph_pagelist_release(m->pagelist);
2927                 kfree(m->pagelist);
2928                 m->pagelist = NULL;
2929         }
2930
2931         m->trail = NULL;
2932
2933         if (m->pool)
2934                 ceph_msgpool_put(m->pool, m);
2935         else
2936                 ceph_msg_kfree(m);
2937 }
2938 EXPORT_SYMBOL(ceph_msg_last_put);
2939
2940 void ceph_msg_dump(struct ceph_msg *msg)
2941 {
2942         pr_debug("msg_dump %p (front_max %d length %zd)\n", msg,
2943                  msg->front_max, msg->length);
2944         print_hex_dump(KERN_DEBUG, "header: ",
2945                        DUMP_PREFIX_OFFSET, 16, 1,
2946                        &msg->hdr, sizeof(msg->hdr), true);
2947         print_hex_dump(KERN_DEBUG, " front: ",
2948                        DUMP_PREFIX_OFFSET, 16, 1,
2949                        msg->front.iov_base, msg->front.iov_len, true);
2950         if (msg->middle)
2951                 print_hex_dump(KERN_DEBUG, "middle: ",
2952                                DUMP_PREFIX_OFFSET, 16, 1,
2953                                msg->middle->vec.iov_base,
2954                                msg->middle->vec.iov_len, true);
2955         print_hex_dump(KERN_DEBUG, "footer: ",
2956                        DUMP_PREFIX_OFFSET, 16, 1,
2957                        &msg->footer, sizeof(msg->footer), true);
2958 }
2959 EXPORT_SYMBOL(ceph_msg_dump);