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