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

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

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

~ [ 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