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