~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/ceph/messenger.c

Version: ~ [ linux-5.7-rc7 ] ~ [ linux-5.6.14 ] ~ [ linux-5.5.19 ] ~ [ linux-5.4.42 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.124 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.181 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.224 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.224 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.84 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

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

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp