kernel/net/ceph/messenger.c

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