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