net/ceph/messenger.c

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