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 auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry); 1397 if (IS_ERR(auth)) 1398 return auth; 1399 1400 con->auth_reply_buf = auth->authorizer_reply_buf; 1401 con->auth_reply_buf_len = auth->authorizer_reply_buf_len; 1402 return auth; 1403 } 1404 1405 /* 1406 * We connected to a peer and are saying hello. 1407 */ 1408 static void prepare_write_banner(struct ceph_connection *con) 1409 { 1410 con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER); 1411 con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr), 1412 &con->msgr->my_enc_addr); 1413 1414 con->out_more = 0; 1415 con_flag_set(con, CON_FLAG_WRITE_PENDING); 1416 } 1417 1418 static int prepare_write_connect(struct ceph_connection *con) 1419 { 1420 unsigned int global_seq = get_global_seq(con->msgr, 0); 1421 int proto; 1422 int auth_proto; 1423 struct ceph_auth_handshake *auth; 1424 1425 switch (con->peer_name.type) { 1426 case CEPH_ENTITY_TYPE_MON: 1427 proto = CEPH_MONC_PROTOCOL; 1428 break; 1429 case CEPH_ENTITY_TYPE_OSD: 1430 proto = CEPH_OSDC_PROTOCOL; 1431 break; 1432 case CEPH_ENTITY_TYPE_MDS: 1433 proto = CEPH_MDSC_PROTOCOL; 1434 break; 1435 default: 1436 BUG(); 1437 } 1438 1439 dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con, 1440 con->connect_seq, global_seq, proto); 1441 1442 con->out_connect.features = 1443 cpu_to_le64(from_msgr(con->msgr)->supported_features); 1444 con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT); 1445 con->out_connect.connect_seq = cpu_to_le32(con->connect_seq); 1446 con->out_connect.global_seq = cpu_to_le32(global_seq); 1447 con->out_connect.protocol_version = cpu_to_le32(proto); 1448 con->out_connect.flags = 0; 1449 1450 auth_proto = CEPH_AUTH_UNKNOWN; 1451 auth = get_connect_authorizer(con, &auth_proto); 1452 if (IS_ERR(auth)) 1453 return PTR_ERR(auth); 1454 1455 con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto); 1456 con->out_connect.authorizer_len = auth ? 1457 cpu_to_le32(auth->authorizer_buf_len) : 0; 1458 1459 con_out_kvec_add(con, sizeof (con->out_connect), 1460 &con->out_connect); 1461 if (auth && auth->authorizer_buf_len) 1462 con_out_kvec_add(con, auth->authorizer_buf_len, 1463 auth->authorizer_buf); 1464 1465 con->out_more = 0; 1466 con_flag_set(con, CON_FLAG_WRITE_PENDING); 1467 1468 return 0; 1469 } 1470 1471 /* 1472 * write as much of pending kvecs to the socket as we can. 1473 * 1 -> done 1474 * 0 -> socket full, but more to do 1475 * <0 -> error 1476 */ 1477 static int write_partial_kvec(struct ceph_connection *con) 1478 { 1479 int ret; 1480 1481 dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes); 1482 while (con->out_kvec_bytes > 0) { 1483 ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur, 1484 con->out_kvec_left, con->out_kvec_bytes, 1485 con->out_more); 1486 if (ret <= 0) 1487 goto out; 1488 con->out_kvec_bytes -= ret; 1489 if (con->out_kvec_bytes == 0) 1490 break; /* done */ 1491 1492 /* account for full iov entries consumed */ 1493 while (ret >= con->out_kvec_cur->iov_len) { 1494 BUG_ON(!con->out_kvec_left); 1495 ret -= con->out_kvec_cur->iov_len; 1496 con->out_kvec_cur++; 1497 con->out_kvec_left--; 1498 } 1499 /* and for a partially-consumed entry */ 1500 if (ret) { 1501 con->out_kvec_cur->iov_len -= ret; 1502 con->out_kvec_cur->iov_base += ret; 1503 } 1504 } 1505 con->out_kvec_left = 0; 1506 ret = 1; 1507 out: 1508 dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con, 1509 con->out_kvec_bytes, con->out_kvec_left, ret); 1510 return ret; /* done! */ 1511 } 1512 1513 static u32 ceph_crc32c_page(u32 crc, struct page *page, 1514 unsigned int page_offset, 1515 unsigned int length) 1516 { 1517 char *kaddr; 1518 1519 kaddr = kmap(page); 1520 BUG_ON(kaddr == NULL); 1521 crc = crc32c(crc, kaddr + page_offset, length); 1522 kunmap(page); 1523 1524 return crc; 1525 } 1526 /* 1527 * Write as much message data payload as we can. If we finish, queue 1528 * up the footer. 1529 * 1 -> done, footer is now queued in out_kvec[]. 1530 * 0 -> socket full, but more to do 1531 * <0 -> error 1532 */ 1533 static int write_partial_message_data(struct ceph_connection *con) 1534 { 1535 struct ceph_msg *msg = con->out_msg; 1536 struct ceph_msg_data_cursor *cursor = &msg->cursor; 1537 bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC); 1538 u32 crc; 1539 1540 dout("%s %p msg %p\n", __func__, con, msg); 1541 1542 if (list_empty(&msg->data)) 1543 return -EINVAL; 1544 1545 /* 1546 * Iterate through each page that contains data to be 1547 * written, and send as much as possible for each. 1548 * 1549 * If we are calculating the data crc (the default), we will 1550 * need to map the page. If we have no pages, they have 1551 * been revoked, so use the zero page. 1552 */ 1553 crc = do_datacrc ? le32_to_cpu(msg->footer.data_crc) : 0; 1554 while (cursor->resid) { 1555 struct page *page; 1556 size_t page_offset; 1557 size_t length; 1558 bool last_piece; 1559 bool need_crc; 1560 int ret; 1561 1562 page = ceph_msg_data_next(cursor, &page_offset, &length, 1563 &last_piece); 1564 ret = ceph_tcp_sendpage(con->sock, page, page_offset, 1565 length, !last_piece); 1566 if (ret <= 0) { 1567 if (do_datacrc) 1568 msg->footer.data_crc = cpu_to_le32(crc); 1569 1570 return ret; 1571 } 1572 if (do_datacrc && cursor->need_crc) 1573 crc = ceph_crc32c_page(crc, page, page_offset, length); 1574 need_crc = ceph_msg_data_advance(cursor, (size_t)ret); 1575 } 1576 1577 dout("%s %p msg %p done\n", __func__, con, msg); 1578 1579 /* prepare and queue up footer, too */ 1580 if (do_datacrc) 1581 msg->footer.data_crc = cpu_to_le32(crc); 1582 else 1583 msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC; 1584 con_out_kvec_reset(con); 1585 prepare_write_message_footer(con); 1586 1587 return 1; /* must return > 0 to indicate success */ 1588 } 1589 1590 /* 1591 * write some zeros 1592 */ 1593 static int write_partial_skip(struct ceph_connection *con) 1594 { 1595 int ret; 1596 1597 dout("%s %p %d left\n", __func__, con, con->out_skip); 1598 while (con->out_skip > 0) { 1599 size_t size = min(con->out_skip, (int) PAGE_SIZE); 1600 1601 ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true); 1602 if (ret <= 0) 1603 goto out; 1604 con->out_skip -= ret; 1605 } 1606 ret = 1; 1607 out: 1608 return ret; 1609 } 1610 1611 /* 1612 * Prepare to read connection handshake, or an ack. 1613 */ 1614 static void prepare_read_banner(struct ceph_connection *con) 1615 { 1616 dout("prepare_read_banner %p\n", con); 1617 con->in_base_pos = 0; 1618 } 1619 1620 static void prepare_read_connect(struct ceph_connection *con) 1621 { 1622 dout("prepare_read_connect %p\n", con); 1623 con->in_base_pos = 0; 1624 } 1625 1626 static void prepare_read_ack(struct ceph_connection *con) 1627 { 1628 dout("prepare_read_ack %p\n", con); 1629 con->in_base_pos = 0; 1630 } 1631 1632 static void prepare_read_seq(struct ceph_connection *con) 1633 { 1634 dout("prepare_read_seq %p\n", con); 1635 con->in_base_pos = 0; 1636 con->in_tag = CEPH_MSGR_TAG_SEQ; 1637 } 1638 1639 static void prepare_read_tag(struct ceph_connection *con) 1640 { 1641 dout("prepare_read_tag %p\n", con); 1642 con->in_base_pos = 0; 1643 con->in_tag = CEPH_MSGR_TAG_READY; 1644 } 1645 1646 static void prepare_read_keepalive_ack(struct ceph_connection *con) 1647 { 1648 dout("prepare_read_keepalive_ack %p\n", con); 1649 con->in_base_pos = 0; 1650 } 1651 1652 /* 1653 * Prepare to read a message. 1654 */ 1655 static int prepare_read_message(struct ceph_connection *con) 1656 { 1657 dout("prepare_read_message %p\n", con); 1658 BUG_ON(con->in_msg != NULL); 1659 con->in_base_pos = 0; 1660 con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0; 1661 return 0; 1662 } 1663 1664 1665 static int read_partial(struct ceph_connection *con, 1666 int end, int size, void *object) 1667 { 1668 while (con->in_base_pos < end) { 1669 int left = end - con->in_base_pos; 1670 int have = size - left; 1671 int ret = ceph_tcp_recvmsg(con->sock, object + have, left); 1672 if (ret <= 0) 1673 return ret; 1674 con->in_base_pos += ret; 1675 } 1676 return 1; 1677 } 1678 1679 1680 /* 1681 * Read all or part of the connect-side handshake on a new connection 1682 */ 1683 static int read_partial_banner(struct ceph_connection *con) 1684 { 1685 int size; 1686 int end; 1687 int ret; 1688 1689 dout("read_partial_banner %p at %d\n", con, con->in_base_pos); 1690 1691 /* peer's banner */ 1692 size = strlen(CEPH_BANNER); 1693 end = size; 1694 ret = read_partial(con, end, size, con->in_banner); 1695 if (ret <= 0) 1696 goto out; 1697 1698 size = sizeof (con->actual_peer_addr); 1699 end += size; 1700 ret = read_partial(con, end, size, &con->actual_peer_addr); 1701 if (ret <= 0) 1702 goto out; 1703 1704 size = sizeof (con->peer_addr_for_me); 1705 end += size; 1706 ret = read_partial(con, end, size, &con->peer_addr_for_me); 1707 if (ret <= 0) 1708 goto out; 1709 1710 out: 1711 return ret; 1712 } 1713 1714 static int read_partial_connect(struct ceph_connection *con) 1715 { 1716 int size; 1717 int end; 1718 int ret; 1719 1720 dout("read_partial_connect %p at %d\n", con, con->in_base_pos); 1721 1722 size = sizeof (con->in_reply); 1723 end = size; 1724 ret = read_partial(con, end, size, &con->in_reply); 1725 if (ret <= 0) 1726 goto out; 1727 1728 size = le32_to_cpu(con->in_reply.authorizer_len); 1729 end += size; 1730 ret = read_partial(con, end, size, con->auth_reply_buf); 1731 if (ret <= 0) 1732 goto out; 1733 1734 dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n", 1735 con, (int)con->in_reply.tag, 1736 le32_to_cpu(con->in_reply.connect_seq), 1737 le32_to_cpu(con->in_reply.global_seq)); 1738 out: 1739 return ret; 1740 1741 } 1742 1743 /* 1744 * Verify the hello banner looks okay. 1745 */ 1746 static int verify_hello(struct ceph_connection *con) 1747 { 1748 if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) { 1749 pr_err("connect to %s got bad banner\n", 1750 ceph_pr_addr(&con->peer_addr.in_addr)); 1751 con->error_msg = "protocol error, bad banner"; 1752 return -1; 1753 } 1754 return 0; 1755 } 1756 1757 static bool addr_is_blank(struct sockaddr_storage *ss) 1758 { 1759 struct in_addr *addr = &((struct sockaddr_in *)ss)->sin_addr; 1760 struct in6_addr *addr6 = &((struct sockaddr_in6 *)ss)->sin6_addr; 1761 1762 switch (ss->ss_family) { 1763 case AF_INET: 1764 return addr->s_addr == htonl(INADDR_ANY); 1765 case AF_INET6: 1766 return ipv6_addr_any(addr6); 1767 default: 1768 return true; 1769 } 1770 } 1771 1772 static int addr_port(struct sockaddr_storage *ss) 1773 { 1774 switch (ss->ss_family) { 1775 case AF_INET: 1776 return ntohs(((struct sockaddr_in *)ss)->sin_port); 1777 case AF_INET6: 1778 return ntohs(((struct sockaddr_in6 *)ss)->sin6_port); 1779 } 1780 return 0; 1781 } 1782 1783 static void addr_set_port(struct sockaddr_storage *ss, int p) 1784 { 1785 switch (ss->ss_family) { 1786 case AF_INET: 1787 ((struct sockaddr_in *)ss)->sin_port = htons(p); 1788 break; 1789 case AF_INET6: 1790 ((struct sockaddr_in6 *)ss)->sin6_port = htons(p); 1791 break; 1792 } 1793 } 1794 1795 /* 1796 * Unlike other *_pton function semantics, zero indicates success. 1797 */ 1798 static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss, 1799 char delim, const char **ipend) 1800 { 1801 struct sockaddr_in *in4 = (struct sockaddr_in *) ss; 1802 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss; 1803 1804 memset(ss, 0, sizeof(*ss)); 1805 1806 if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) { 1807 ss->ss_family = AF_INET; 1808 return 0; 1809 } 1810 1811 if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) { 1812 ss->ss_family = AF_INET6; 1813 return 0; 1814 } 1815 1816 return -EINVAL; 1817 } 1818 1819 /* 1820 * Extract hostname string and resolve using kernel DNS facility. 1821 */ 1822 #ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER 1823 static int ceph_dns_resolve_name(const char *name, size_t namelen, 1824 struct sockaddr_storage *ss, char delim, const char **ipend) 1825 { 1826 const char *end, *delim_p; 1827 char *colon_p, *ip_addr = NULL; 1828 int ip_len, ret; 1829 1830 /* 1831 * The end of the hostname occurs immediately preceding the delimiter or 1832 * the port marker (':') where the delimiter takes precedence. 1833 */ 1834 delim_p = memchr(name, delim, namelen); 1835 colon_p = memchr(name, ':', namelen); 1836 1837 if (delim_p && colon_p) 1838 end = delim_p < colon_p ? delim_p : colon_p; 1839 else if (!delim_p && colon_p) 1840 end = colon_p; 1841 else { 1842 end = delim_p; 1843 if (!end) /* case: hostname:/ */ 1844 end = name + namelen; 1845 } 1846 1847 if (end <= name) 1848 return -EINVAL; 1849 1850 /* do dns_resolve upcall */ 1851 ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL); 1852 if (ip_len > 0) 1853 ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL); 1854 else 1855 ret = -ESRCH; 1856 1857 kfree(ip_addr); 1858 1859 *ipend = end; 1860 1861 pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name, 1862 ret, ret ? "failed" : ceph_pr_addr(ss)); 1863 1864 return ret; 1865 } 1866 #else 1867 static inline int ceph_dns_resolve_name(const char *name, size_t namelen, 1868 struct sockaddr_storage *ss, char delim, const char **ipend) 1869 { 1870 return -EINVAL; 1871 } 1872 #endif 1873 1874 /* 1875 * Parse a server name (IP or hostname). If a valid IP address is not found 1876 * then try to extract a hostname to resolve using userspace DNS upcall. 1877 */ 1878 static int ceph_parse_server_name(const char *name, size_t namelen, 1879 struct sockaddr_storage *ss, char delim, const char **ipend) 1880 { 1881 int ret; 1882 1883 ret = ceph_pton(name, namelen, ss, delim, ipend); 1884 if (ret) 1885 ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend); 1886 1887 return ret; 1888 } 1889 1890 /* 1891 * Parse an ip[:port] list into an addr array. Use the default 1892 * monitor port if a port isn't specified. 1893 */ 1894 int ceph_parse_ips(const char *c, const char *end, 1895 struct ceph_entity_addr *addr, 1896 int max_count, int *count) 1897 { 1898 int i, ret = -EINVAL; 1899 const char *p = c; 1900 1901 dout("parse_ips on '%.*s'\n", (int)(end-c), c); 1902 for (i = 0; i < max_count; i++) { 1903 const char *ipend; 1904 struct sockaddr_storage *ss = &addr[i].in_addr; 1905 int port; 1906 char delim = ','; 1907 1908 if (*p == '[') { 1909 delim = ']'; 1910 p++; 1911 } 1912 1913 ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend); 1914 if (ret) 1915 goto bad; 1916 ret = -EINVAL; 1917 1918 p = ipend; 1919 1920 if (delim == ']') { 1921 if (*p != ']') { 1922 dout("missing matching ']'\n"); 1923 goto bad; 1924 } 1925 p++; 1926 } 1927 1928 /* port? */ 1929 if (p < end && *p == ':') { 1930 port = 0; 1931 p++; 1932 while (p < end && *p >= '0' && *p <= '9') { 1933 port = (port * 10) + (*p - '0'); 1934 p++; 1935 } 1936 if (port == 0) 1937 port = CEPH_MON_PORT; 1938 else if (port > 65535) 1939 goto bad; 1940 } else { 1941 port = CEPH_MON_PORT; 1942 } 1943 1944 addr_set_port(ss, port); 1945 1946 dout("parse_ips got %s\n", ceph_pr_addr(ss)); 1947 1948 if (p == end) 1949 break; 1950 if (*p != ',') 1951 goto bad; 1952 p++; 1953 } 1954 1955 if (p != end) 1956 goto bad; 1957 1958 if (count) 1959 *count = i + 1; 1960 return 0; 1961 1962 bad: 1963 pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c); 1964 return ret; 1965 } 1966 EXPORT_SYMBOL(ceph_parse_ips); 1967 1968 static int process_banner(struct ceph_connection *con) 1969 { 1970 dout("process_banner on %p\n", con); 1971 1972 if (verify_hello(con) < 0) 1973 return -1; 1974 1975 ceph_decode_addr(&con->actual_peer_addr); 1976 ceph_decode_addr(&con->peer_addr_for_me); 1977 1978 /* 1979 * Make sure the other end is who we wanted. note that the other 1980 * end may not yet know their ip address, so if it's 0.0.0.0, give 1981 * them the benefit of the doubt. 1982 */ 1983 if (memcmp(&con->peer_addr, &con->actual_peer_addr, 1984 sizeof(con->peer_addr)) != 0 && 1985 !(addr_is_blank(&con->actual_peer_addr.in_addr) && 1986 con->actual_peer_addr.nonce == con->peer_addr.nonce)) { 1987 pr_warn("wrong peer, want %s/%d, got %s/%d\n", 1988 ceph_pr_addr(&con->peer_addr.in_addr), 1989 (int)le32_to_cpu(con->peer_addr.nonce), 1990 ceph_pr_addr(&con->actual_peer_addr.in_addr), 1991 (int)le32_to_cpu(con->actual_peer_addr.nonce)); 1992 con->error_msg = "wrong peer at address"; 1993 return -1; 1994 } 1995 1996 /* 1997 * did we learn our address? 1998 */ 1999 if (addr_is_blank(&con->msgr->inst.addr.in_addr)) { 2000 int port = addr_port(&con->msgr->inst.addr.in_addr); 2001 2002 memcpy(&con->msgr->inst.addr.in_addr, 2003 &con->peer_addr_for_me.in_addr, 2004 sizeof(con->peer_addr_for_me.in_addr)); 2005 addr_set_port(&con->msgr->inst.addr.in_addr, port); 2006 encode_my_addr(con->msgr); 2007 dout("process_banner learned my addr is %s\n", 2008 ceph_pr_addr(&con->msgr->inst.addr.in_addr)); 2009 } 2010 2011 return 0; 2012 } 2013 2014 static int process_connect(struct ceph_connection *con) 2015 { 2016 u64 sup_feat = from_msgr(con->msgr)->supported_features; 2017 u64 req_feat = from_msgr(con->msgr)->required_features; 2018 u64 server_feat = ceph_sanitize_features( 2019 le64_to_cpu(con->in_reply.features)); 2020 int ret; 2021 2022 dout("process_connect on %p tag %d\n", con, (int)con->in_tag); 2023 2024 if (con->auth_reply_buf) { 2025 /* 2026 * Any connection that defines ->get_authorizer() 2027 * should also define ->verify_authorizer_reply(). 2028 * See get_connect_authorizer(). 2029 */ 2030 ret = con->ops->verify_authorizer_reply(con); 2031 if (ret < 0) { 2032 con->error_msg = "bad authorize reply"; 2033 return ret; 2034 } 2035 } 2036 2037 switch (con->in_reply.tag) { 2038 case CEPH_MSGR_TAG_FEATURES: 2039 pr_err("%s%lld %s feature set mismatch," 2040 " my %llx < server's %llx, missing %llx\n", 2041 ENTITY_NAME(con->peer_name), 2042 ceph_pr_addr(&con->peer_addr.in_addr), 2043 sup_feat, server_feat, server_feat & ~sup_feat); 2044 con->error_msg = "missing required protocol features"; 2045 reset_connection(con); 2046 return -1; 2047 2048 case CEPH_MSGR_TAG_BADPROTOVER: 2049 pr_err("%s%lld %s protocol version mismatch," 2050 " my %d != server's %d\n", 2051 ENTITY_NAME(con->peer_name), 2052 ceph_pr_addr(&con->peer_addr.in_addr), 2053 le32_to_cpu(con->out_connect.protocol_version), 2054 le32_to_cpu(con->in_reply.protocol_version)); 2055 con->error_msg = "protocol version mismatch"; 2056 reset_connection(con); 2057 return -1; 2058 2059 case CEPH_MSGR_TAG_BADAUTHORIZER: 2060 con->auth_retry++; 2061 dout("process_connect %p got BADAUTHORIZER attempt %d\n", con, 2062 con->auth_retry); 2063 if (con->auth_retry == 2) { 2064 con->error_msg = "connect authorization failure"; 2065 return -1; 2066 } 2067 con_out_kvec_reset(con); 2068 ret = prepare_write_connect(con); 2069 if (ret < 0) 2070 return ret; 2071 prepare_read_connect(con); 2072 break; 2073 2074 case CEPH_MSGR_TAG_RESETSESSION: 2075 /* 2076 * If we connected with a large connect_seq but the peer 2077 * has no record of a session with us (no connection, or 2078 * connect_seq == 0), they will send RESETSESION to indicate 2079 * that they must have reset their session, and may have 2080 * dropped messages. 2081 */ 2082 dout("process_connect got RESET peer seq %u\n", 2083 le32_to_cpu(con->in_reply.connect_seq)); 2084 pr_err("%s%lld %s connection reset\n", 2085 ENTITY_NAME(con->peer_name), 2086 ceph_pr_addr(&con->peer_addr.in_addr)); 2087 reset_connection(con); 2088 con_out_kvec_reset(con); 2089 ret = prepare_write_connect(con); 2090 if (ret < 0) 2091 return ret; 2092 prepare_read_connect(con); 2093 2094 /* Tell ceph about it. */ 2095 mutex_unlock(&con->mutex); 2096 pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name)); 2097 if (con->ops->peer_reset) 2098 con->ops->peer_reset(con); 2099 mutex_lock(&con->mutex); 2100 if (con->state != CON_STATE_NEGOTIATING) 2101 return -EAGAIN; 2102 break; 2103 2104 case CEPH_MSGR_TAG_RETRY_SESSION: 2105 /* 2106 * If we sent a smaller connect_seq than the peer has, try 2107 * again with a larger value. 2108 */ 2109 dout("process_connect got RETRY_SESSION my seq %u, peer %u\n", 2110 le32_to_cpu(con->out_connect.connect_seq), 2111 le32_to_cpu(con->in_reply.connect_seq)); 2112 con->connect_seq = le32_to_cpu(con->in_reply.connect_seq); 2113 con_out_kvec_reset(con); 2114 ret = prepare_write_connect(con); 2115 if (ret < 0) 2116 return ret; 2117 prepare_read_connect(con); 2118 break; 2119 2120 case CEPH_MSGR_TAG_RETRY_GLOBAL: 2121 /* 2122 * If we sent a smaller global_seq than the peer has, try 2123 * again with a larger value. 2124 */ 2125 dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n", 2126 con->peer_global_seq, 2127 le32_to_cpu(con->in_reply.global_seq)); 2128 get_global_seq(con->msgr, 2129 le32_to_cpu(con->in_reply.global_seq)); 2130 con_out_kvec_reset(con); 2131 ret = prepare_write_connect(con); 2132 if (ret < 0) 2133 return ret; 2134 prepare_read_connect(con); 2135 break; 2136 2137 case CEPH_MSGR_TAG_SEQ: 2138 case CEPH_MSGR_TAG_READY: 2139 if (req_feat & ~server_feat) { 2140 pr_err("%s%lld %s protocol feature mismatch," 2141 " my required %llx > server's %llx, need %llx\n", 2142 ENTITY_NAME(con->peer_name), 2143 ceph_pr_addr(&con->peer_addr.in_addr), 2144 req_feat, server_feat, req_feat & ~server_feat); 2145 con->error_msg = "missing required protocol features"; 2146 reset_connection(con); 2147 return -1; 2148 } 2149 2150 WARN_ON(con->state != CON_STATE_NEGOTIATING); 2151 con->state = CON_STATE_OPEN; 2152 con->auth_retry = 0; /* we authenticated; clear flag */ 2153 con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq); 2154 con->connect_seq++; 2155 con->peer_features = server_feat; 2156 dout("process_connect got READY gseq %d cseq %d (%d)\n", 2157 con->peer_global_seq, 2158 le32_to_cpu(con->in_reply.connect_seq), 2159 con->connect_seq); 2160 WARN_ON(con->connect_seq != 2161 le32_to_cpu(con->in_reply.connect_seq)); 2162 2163 if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY) 2164 con_flag_set(con, CON_FLAG_LOSSYTX); 2165 2166 con->delay = 0; /* reset backoff memory */ 2167 2168 if (con->in_reply.tag == CEPH_MSGR_TAG_SEQ) { 2169 prepare_write_seq(con); 2170 prepare_read_seq(con); 2171 } else { 2172 prepare_read_tag(con); 2173 } 2174 break; 2175 2176 case CEPH_MSGR_TAG_WAIT: 2177 /* 2178 * If there is a connection race (we are opening 2179 * connections to each other), one of us may just have 2180 * to WAIT. This shouldn't happen if we are the 2181 * client. 2182 */ 2183 con->error_msg = "protocol error, got WAIT as client"; 2184 return -1; 2185 2186 default: 2187 con->error_msg = "protocol error, garbage tag during connect"; 2188 return -1; 2189 } 2190 return 0; 2191 } 2192 2193 2194 /* 2195 * read (part of) an ack 2196 */ 2197 static int read_partial_ack(struct ceph_connection *con) 2198 { 2199 int size = sizeof (con->in_temp_ack); 2200 int end = size; 2201 2202 return read_partial(con, end, size, &con->in_temp_ack); 2203 } 2204 2205 /* 2206 * We can finally discard anything that's been acked. 2207 */ 2208 static void process_ack(struct ceph_connection *con) 2209 { 2210 struct ceph_msg *m; 2211 u64 ack = le64_to_cpu(con->in_temp_ack); 2212 u64 seq; 2213 2214 while (!list_empty(&con->out_sent)) { 2215 m = list_first_entry(&con->out_sent, struct ceph_msg, 2216 list_head); 2217 seq = le64_to_cpu(m->hdr.seq); 2218 if (seq > ack) 2219 break; 2220 dout("got ack for seq %llu type %d at %p\n", seq, 2221 le16_to_cpu(m->hdr.type), m); 2222 m->ack_stamp = jiffies; 2223 ceph_msg_remove(m); 2224 } 2225 prepare_read_tag(con); 2226 } 2227 2228 2229 static int read_partial_message_section(struct ceph_connection *con, 2230 struct kvec *section, 2231 unsigned int sec_len, u32 *crc) 2232 { 2233 int ret, left; 2234 2235 BUG_ON(!section); 2236 2237 while (section->iov_len < sec_len) { 2238 BUG_ON(section->iov_base == NULL); 2239 left = sec_len - section->iov_len; 2240 ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base + 2241 section->iov_len, left); 2242 if (ret <= 0) 2243 return ret; 2244 section->iov_len += ret; 2245 } 2246 if (section->iov_len == sec_len) 2247 *crc = crc32c(0, section->iov_base, section->iov_len); 2248 2249 return 1; 2250 } 2251 2252 static int read_partial_msg_data(struct ceph_connection *con) 2253 { 2254 struct ceph_msg *msg = con->in_msg; 2255 struct ceph_msg_data_cursor *cursor = &msg->cursor; 2256 bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC); 2257 struct page *page; 2258 size_t page_offset; 2259 size_t length; 2260 u32 crc = 0; 2261 int ret; 2262 2263 BUG_ON(!msg); 2264 if (list_empty(&msg->data)) 2265 return -EIO; 2266 2267 if (do_datacrc) 2268 crc = con->in_data_crc; 2269 while (cursor->resid) { 2270 page = ceph_msg_data_next(cursor, &page_offset, &length, NULL); 2271 ret = ceph_tcp_recvpage(con->sock, page, page_offset, length); 2272 if (ret <= 0) { 2273 if (do_datacrc) 2274 con->in_data_crc = crc; 2275 2276 return ret; 2277 } 2278 2279 if (do_datacrc) 2280 crc = ceph_crc32c_page(crc, page, page_offset, ret); 2281 (void) ceph_msg_data_advance(cursor, (size_t)ret); 2282 } 2283 if (do_datacrc) 2284 con->in_data_crc = crc; 2285 2286 return 1; /* must return > 0 to indicate success */ 2287 } 2288 2289 /* 2290 * read (part of) a message. 2291 */ 2292 static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip); 2293 2294 static int read_partial_message(struct ceph_connection *con) 2295 { 2296 struct ceph_msg *m = con->in_msg; 2297 int size; 2298 int end; 2299 int ret; 2300 unsigned int front_len, middle_len, data_len; 2301 bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC); 2302 bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH); 2303 u64 seq; 2304 u32 crc; 2305 2306 dout("read_partial_message con %p msg %p\n", con, m); 2307 2308 /* header */ 2309 size = sizeof (con->in_hdr); 2310 end = size; 2311 ret = read_partial(con, end, size, &con->in_hdr); 2312 if (ret <= 0) 2313 return ret; 2314 2315 crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc)); 2316 if (cpu_to_le32(crc) != con->in_hdr.crc) { 2317 pr_err("read_partial_message bad hdr crc %u != expected %u\n", 2318 crc, con->in_hdr.crc); 2319 return -EBADMSG; 2320 } 2321 2322 front_len = le32_to_cpu(con->in_hdr.front_len); 2323 if (front_len > CEPH_MSG_MAX_FRONT_LEN) 2324 return -EIO; 2325 middle_len = le32_to_cpu(con->in_hdr.middle_len); 2326 if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN) 2327 return -EIO; 2328 data_len = le32_to_cpu(con->in_hdr.data_len); 2329 if (data_len > CEPH_MSG_MAX_DATA_LEN) 2330 return -EIO; 2331 2332 /* verify seq# */ 2333 seq = le64_to_cpu(con->in_hdr.seq); 2334 if ((s64)seq - (s64)con->in_seq < 1) { 2335 pr_info("skipping %s%lld %s seq %lld expected %lld\n", 2336 ENTITY_NAME(con->peer_name), 2337 ceph_pr_addr(&con->peer_addr.in_addr), 2338 seq, con->in_seq + 1); 2339 con->in_base_pos = -front_len - middle_len - data_len - 2340 sizeof_footer(con); 2341 con->in_tag = CEPH_MSGR_TAG_READY; 2342 return 1; 2343 } else if ((s64)seq - (s64)con->in_seq > 1) { 2344 pr_err("read_partial_message bad seq %lld expected %lld\n", 2345 seq, con->in_seq + 1); 2346 con->error_msg = "bad message sequence # for incoming message"; 2347 return -EBADE; 2348 } 2349 2350 /* allocate message? */ 2351 if (!con->in_msg) { 2352 int skip = 0; 2353 2354 dout("got hdr type %d front %d data %d\n", con->in_hdr.type, 2355 front_len, data_len); 2356 ret = ceph_con_in_msg_alloc(con, &skip); 2357 if (ret < 0) 2358 return ret; 2359 2360 BUG_ON(!con->in_msg ^ skip); 2361 if (skip) { 2362 /* skip this message */ 2363 dout("alloc_msg said skip message\n"); 2364 con->in_base_pos = -front_len - middle_len - data_len - 2365 sizeof_footer(con); 2366 con->in_tag = CEPH_MSGR_TAG_READY; 2367 con->in_seq++; 2368 return 1; 2369 } 2370 2371 BUG_ON(!con->in_msg); 2372 BUG_ON(con->in_msg->con != con); 2373 m = con->in_msg; 2374 m->front.iov_len = 0; /* haven't read it yet */ 2375 if (m->middle) 2376 m->middle->vec.iov_len = 0; 2377 2378 /* prepare for data payload, if any */ 2379 2380 if (data_len) 2381 prepare_message_data(con->in_msg, data_len); 2382 } 2383 2384 /* front */ 2385 ret = read_partial_message_section(con, &m->front, front_len, 2386 &con->in_front_crc); 2387 if (ret <= 0) 2388 return ret; 2389 2390 /* middle */ 2391 if (m->middle) { 2392 ret = read_partial_message_section(con, &m->middle->vec, 2393 middle_len, 2394 &con->in_middle_crc); 2395 if (ret <= 0) 2396 return ret; 2397 } 2398 2399 /* (page) data */ 2400 if (data_len) { 2401 ret = read_partial_msg_data(con); 2402 if (ret <= 0) 2403 return ret; 2404 } 2405 2406 /* footer */ 2407 size = sizeof_footer(con); 2408 end += size; 2409 ret = read_partial(con, end, size, &m->footer); 2410 if (ret <= 0) 2411 return ret; 2412 2413 if (!need_sign) { 2414 m->footer.flags = m->old_footer.flags; 2415 m->footer.sig = 0; 2416 } 2417 2418 dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n", 2419 m, front_len, m->footer.front_crc, middle_len, 2420 m->footer.middle_crc, data_len, m->footer.data_crc); 2421 2422 /* crc ok? */ 2423 if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) { 2424 pr_err("read_partial_message %p front crc %u != exp. %u\n", 2425 m, con->in_front_crc, m->footer.front_crc); 2426 return -EBADMSG; 2427 } 2428 if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) { 2429 pr_err("read_partial_message %p middle crc %u != exp %u\n", 2430 m, con->in_middle_crc, m->footer.middle_crc); 2431 return -EBADMSG; 2432 } 2433 if (do_datacrc && 2434 (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 && 2435 con->in_data_crc != le32_to_cpu(m->footer.data_crc)) { 2436 pr_err("read_partial_message %p data crc %u != exp. %u\n", m, 2437 con->in_data_crc, le32_to_cpu(m->footer.data_crc)); 2438 return -EBADMSG; 2439 } 2440 2441 if (need_sign && con->ops->check_message_signature && 2442 con->ops->check_message_signature(m)) { 2443 pr_err("read_partial_message %p signature check failed\n", m); 2444 return -EBADMSG; 2445 } 2446 2447 return 1; /* done! */ 2448 } 2449 2450 /* 2451 * Process message. This happens in the worker thread. The callback should 2452 * be careful not to do anything that waits on other incoming messages or it 2453 * may deadlock. 2454 */ 2455 static void process_message(struct ceph_connection *con) 2456 { 2457 struct ceph_msg *msg = con->in_msg; 2458 2459 BUG_ON(con->in_msg->con != con); 2460 con->in_msg = NULL; 2461 2462 /* if first message, set peer_name */ 2463 if (con->peer_name.type == 0) 2464 con->peer_name = msg->hdr.src; 2465 2466 con->in_seq++; 2467 mutex_unlock(&con->mutex); 2468 2469 dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n", 2470 msg, le64_to_cpu(msg->hdr.seq), 2471 ENTITY_NAME(msg->hdr.src), 2472 le16_to_cpu(msg->hdr.type), 2473 ceph_msg_type_name(le16_to_cpu(msg->hdr.type)), 2474 le32_to_cpu(msg->hdr.front_len), 2475 le32_to_cpu(msg->hdr.data_len), 2476 con->in_front_crc, con->in_middle_crc, con->in_data_crc); 2477 con->ops->dispatch(con, msg); 2478 2479 mutex_lock(&con->mutex); 2480 } 2481 2482 static int read_keepalive_ack(struct ceph_connection *con) 2483 { 2484 struct ceph_timespec ceph_ts; 2485 size_t size = sizeof(ceph_ts); 2486 int ret = read_partial(con, size, size, &ceph_ts); 2487 if (ret <= 0) 2488 return ret; 2489 ceph_decode_timespec(&con->last_keepalive_ack, &ceph_ts); 2490 prepare_read_tag(con); 2491 return 1; 2492 } 2493 2494 /* 2495 * Write something to the socket. Called in a worker thread when the 2496 * socket appears to be writeable and we have something ready to send. 2497 */ 2498 static int try_write(struct ceph_connection *con) 2499 { 2500 int ret = 1; 2501 2502 dout("try_write start %p state %lu\n", con, con->state); 2503 2504 more: 2505 dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes); 2506 2507 /* open the socket first? */ 2508 if (con->state == CON_STATE_PREOPEN) { 2509 BUG_ON(con->sock); 2510 con->state = CON_STATE_CONNECTING; 2511 2512 con_out_kvec_reset(con); 2513 prepare_write_banner(con); 2514 prepare_read_banner(con); 2515 2516 BUG_ON(con->in_msg); 2517 con->in_tag = CEPH_MSGR_TAG_READY; 2518 dout("try_write initiating connect on %p new state %lu\n", 2519 con, con->state); 2520 ret = ceph_tcp_connect(con); 2521 if (ret < 0) { 2522 con->error_msg = "connect error"; 2523 goto out; 2524 } 2525 } 2526 2527 more_kvec: 2528 /* kvec data queued? */ 2529 if (con->out_kvec_left) { 2530 ret = write_partial_kvec(con); 2531 if (ret <= 0) 2532 goto out; 2533 } 2534 if (con->out_skip) { 2535 ret = write_partial_skip(con); 2536 if (ret <= 0) 2537 goto out; 2538 } 2539 2540 /* msg pages? */ 2541 if (con->out_msg) { 2542 if (con->out_msg_done) { 2543 ceph_msg_put(con->out_msg); 2544 con->out_msg = NULL; /* we're done with this one */ 2545 goto do_next; 2546 } 2547 2548 ret = write_partial_message_data(con); 2549 if (ret == 1) 2550 goto more_kvec; /* we need to send the footer, too! */ 2551 if (ret == 0) 2552 goto out; 2553 if (ret < 0) { 2554 dout("try_write write_partial_message_data err %d\n", 2555 ret); 2556 goto out; 2557 } 2558 } 2559 2560 do_next: 2561 if (con->state == CON_STATE_OPEN) { 2562 if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) { 2563 prepare_write_keepalive(con); 2564 goto more; 2565 } 2566 /* is anything else pending? */ 2567 if (!list_empty(&con->out_queue)) { 2568 prepare_write_message(con); 2569 goto more; 2570 } 2571 if (con->in_seq > con->in_seq_acked) { 2572 prepare_write_ack(con); 2573 goto more; 2574 } 2575 } 2576 2577 /* Nothing to do! */ 2578 con_flag_clear(con, CON_FLAG_WRITE_PENDING); 2579 dout("try_write nothing else to write.\n"); 2580 ret = 0; 2581 out: 2582 dout("try_write done on %p ret %d\n", con, ret); 2583 return ret; 2584 } 2585 2586 2587 2588 /* 2589 * Read what we can from the socket. 2590 */ 2591 static int try_read(struct ceph_connection *con) 2592 { 2593 int ret = -1; 2594 2595 more: 2596 dout("try_read start on %p state %lu\n", con, con->state); 2597 if (con->state != CON_STATE_CONNECTING && 2598 con->state != CON_STATE_NEGOTIATING && 2599 con->state != CON_STATE_OPEN) 2600 return 0; 2601 2602 BUG_ON(!con->sock); 2603 2604 dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag, 2605 con->in_base_pos); 2606 2607 if (con->state == CON_STATE_CONNECTING) { 2608 dout("try_read connecting\n"); 2609 ret = read_partial_banner(con); 2610 if (ret <= 0) 2611 goto out; 2612 ret = process_banner(con); 2613 if (ret < 0) 2614 goto out; 2615 2616 con->state = CON_STATE_NEGOTIATING; 2617 2618 /* 2619 * Received banner is good, exchange connection info. 2620 * Do not reset out_kvec, as sending our banner raced 2621 * with receiving peer banner after connect completed. 2622 */ 2623 ret = prepare_write_connect(con); 2624 if (ret < 0) 2625 goto out; 2626 prepare_read_connect(con); 2627 2628 /* Send connection info before awaiting response */ 2629 goto out; 2630 } 2631 2632 if (con->state == CON_STATE_NEGOTIATING) { 2633 dout("try_read negotiating\n"); 2634 ret = read_partial_connect(con); 2635 if (ret <= 0) 2636 goto out; 2637 ret = process_connect(con); 2638 if (ret < 0) 2639 goto out; 2640 goto more; 2641 } 2642 2643 WARN_ON(con->state != CON_STATE_OPEN); 2644 2645 if (con->in_base_pos < 0) { 2646 /* 2647 * skipping + discarding content. 2648 * 2649 * FIXME: there must be a better way to do this! 2650 */ 2651 static char buf[SKIP_BUF_SIZE]; 2652 int skip = min((int) sizeof (buf), -con->in_base_pos); 2653 2654 dout("skipping %d / %d bytes\n", skip, -con->in_base_pos); 2655 ret = ceph_tcp_recvmsg(con->sock, buf, skip); 2656 if (ret <= 0) 2657 goto out; 2658 con->in_base_pos += ret; 2659 if (con->in_base_pos) 2660 goto more; 2661 } 2662 if (con->in_tag == CEPH_MSGR_TAG_READY) { 2663 /* 2664 * what's next? 2665 */ 2666 ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1); 2667 if (ret <= 0) 2668 goto out; 2669 dout("try_read got tag %d\n", (int)con->in_tag); 2670 switch (con->in_tag) { 2671 case CEPH_MSGR_TAG_MSG: 2672 prepare_read_message(con); 2673 break; 2674 case CEPH_MSGR_TAG_ACK: 2675 prepare_read_ack(con); 2676 break; 2677 case CEPH_MSGR_TAG_KEEPALIVE2_ACK: 2678 prepare_read_keepalive_ack(con); 2679 break; 2680 case CEPH_MSGR_TAG_CLOSE: 2681 con_close_socket(con); 2682 con->state = CON_STATE_CLOSED; 2683 goto out; 2684 default: 2685 goto bad_tag; 2686 } 2687 } 2688 if (con->in_tag == CEPH_MSGR_TAG_MSG) { 2689 ret = read_partial_message(con); 2690 if (ret <= 0) { 2691 switch (ret) { 2692 case -EBADMSG: 2693 con->error_msg = "bad crc/signature"; 2694 /* fall through */ 2695 case -EBADE: 2696 ret = -EIO; 2697 break; 2698 case -EIO: 2699 con->error_msg = "io error"; 2700 break; 2701 } 2702 goto out; 2703 } 2704 if (con->in_tag == CEPH_MSGR_TAG_READY) 2705 goto more; 2706 process_message(con); 2707 if (con->state == CON_STATE_OPEN) 2708 prepare_read_tag(con); 2709 goto more; 2710 } 2711 if (con->in_tag == CEPH_MSGR_TAG_ACK || 2712 con->in_tag == CEPH_MSGR_TAG_SEQ) { 2713 /* 2714 * the final handshake seq exchange is semantically 2715 * equivalent to an ACK 2716 */ 2717 ret = read_partial_ack(con); 2718 if (ret <= 0) 2719 goto out; 2720 process_ack(con); 2721 goto more; 2722 } 2723 if (con->in_tag == CEPH_MSGR_TAG_KEEPALIVE2_ACK) { 2724 ret = read_keepalive_ack(con); 2725 if (ret <= 0) 2726 goto out; 2727 goto more; 2728 } 2729 2730 out: 2731 dout("try_read done on %p ret %d\n", con, ret); 2732 return ret; 2733 2734 bad_tag: 2735 pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag); 2736 con->error_msg = "protocol error, garbage tag"; 2737 ret = -1; 2738 goto out; 2739 } 2740 2741 2742 /* 2743 * Atomically queue work on a connection after the specified delay. 2744 * Bump @con reference to avoid races with connection teardown. 2745 * Returns 0 if work was queued, or an error code otherwise. 2746 */ 2747 static int queue_con_delay(struct ceph_connection *con, unsigned long delay) 2748 { 2749 if (!con->ops->get(con)) { 2750 dout("%s %p ref count 0\n", __func__, con); 2751 return -ENOENT; 2752 } 2753 2754 if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) { 2755 dout("%s %p - already queued\n", __func__, con); 2756 con->ops->put(con); 2757 return -EBUSY; 2758 } 2759 2760 dout("%s %p %lu\n", __func__, con, delay); 2761 return 0; 2762 } 2763 2764 static void queue_con(struct ceph_connection *con) 2765 { 2766 (void) queue_con_delay(con, 0); 2767 } 2768 2769 static void cancel_con(struct ceph_connection *con) 2770 { 2771 if (cancel_delayed_work(&con->work)) { 2772 dout("%s %p\n", __func__, con); 2773 con->ops->put(con); 2774 } 2775 } 2776 2777 static bool con_sock_closed(struct ceph_connection *con) 2778 { 2779 if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED)) 2780 return false; 2781 2782 #define CASE(x) \ 2783 case CON_STATE_ ## x: \ 2784 con->error_msg = "socket closed (con state " #x ")"; \ 2785 break; 2786 2787 switch (con->state) { 2788 CASE(CLOSED); 2789 CASE(PREOPEN); 2790 CASE(CONNECTING); 2791 CASE(NEGOTIATING); 2792 CASE(OPEN); 2793 CASE(STANDBY); 2794 default: 2795 pr_warn("%s con %p unrecognized state %lu\n", 2796 __func__, con, con->state); 2797 con->error_msg = "unrecognized con state"; 2798 BUG(); 2799 break; 2800 } 2801 #undef CASE 2802 2803 return true; 2804 } 2805 2806 static bool con_backoff(struct ceph_connection *con) 2807 { 2808 int ret; 2809 2810 if (!con_flag_test_and_clear(con, CON_FLAG_BACKOFF)) 2811 return false; 2812 2813 ret = queue_con_delay(con, round_jiffies_relative(con->delay)); 2814 if (ret) { 2815 dout("%s: con %p FAILED to back off %lu\n", __func__, 2816 con, con->delay); 2817 BUG_ON(ret == -ENOENT); 2818 con_flag_set(con, CON_FLAG_BACKOFF); 2819 } 2820 2821 return true; 2822 } 2823 2824 /* Finish fault handling; con->mutex must *not* be held here */ 2825 2826 static void con_fault_finish(struct ceph_connection *con) 2827 { 2828 dout("%s %p\n", __func__, con); 2829 2830 /* 2831 * in case we faulted due to authentication, invalidate our 2832 * current tickets so that we can get new ones. 2833 */ 2834 if (con->auth_retry) { 2835 dout("auth_retry %d, invalidating\n", con->auth_retry); 2836 if (con->ops->invalidate_authorizer) 2837 con->ops->invalidate_authorizer(con); 2838 con->auth_retry = 0; 2839 } 2840 2841 if (con->ops->fault) 2842 con->ops->fault(con); 2843 } 2844 2845 /* 2846 * Do some work on a connection. Drop a connection ref when we're done. 2847 */ 2848 static void ceph_con_workfn(struct work_struct *work) 2849 { 2850 struct ceph_connection *con = container_of(work, struct ceph_connection, 2851 work.work); 2852 bool fault; 2853 2854 mutex_lock(&con->mutex); 2855 while (true) { 2856 int ret; 2857 2858 if ((fault = con_sock_closed(con))) { 2859 dout("%s: con %p SOCK_CLOSED\n", __func__, con); 2860 break; 2861 } 2862 if (con_backoff(con)) { 2863 dout("%s: con %p BACKOFF\n", __func__, con); 2864 break; 2865 } 2866 if (con->state == CON_STATE_STANDBY) { 2867 dout("%s: con %p STANDBY\n", __func__, con); 2868 break; 2869 } 2870 if (con->state == CON_STATE_CLOSED) { 2871 dout("%s: con %p CLOSED\n", __func__, con); 2872 BUG_ON(con->sock); 2873 break; 2874 } 2875 if (con->state == CON_STATE_PREOPEN) { 2876 dout("%s: con %p PREOPEN\n", __func__, con); 2877 BUG_ON(con->sock); 2878 } 2879 2880 ret = try_read(con); 2881 if (ret < 0) { 2882 if (ret == -EAGAIN) 2883 continue; 2884 if (!con->error_msg) 2885 con->error_msg = "socket error on read"; 2886 fault = true; 2887 break; 2888 } 2889 2890 ret = try_write(con); 2891 if (ret < 0) { 2892 if (ret == -EAGAIN) 2893 continue; 2894 if (!con->error_msg) 2895 con->error_msg = "socket error on write"; 2896 fault = true; 2897 } 2898 2899 break; /* If we make it to here, we're done */ 2900 } 2901 if (fault) 2902 con_fault(con); 2903 mutex_unlock(&con->mutex); 2904 2905 if (fault) 2906 con_fault_finish(con); 2907 2908 con->ops->put(con); 2909 } 2910 2911 /* 2912 * Generic error/fault handler. A retry mechanism is used with 2913 * exponential backoff 2914 */ 2915 static void con_fault(struct ceph_connection *con) 2916 { 2917 dout("fault %p state %lu to peer %s\n", 2918 con, con->state, ceph_pr_addr(&con->peer_addr.in_addr)); 2919 2920 pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name), 2921 ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg); 2922 con->error_msg = NULL; 2923 2924 WARN_ON(con->state != CON_STATE_CONNECTING && 2925 con->state != CON_STATE_NEGOTIATING && 2926 con->state != CON_STATE_OPEN); 2927 2928 con_close_socket(con); 2929 2930 if (con_flag_test(con, CON_FLAG_LOSSYTX)) { 2931 dout("fault on LOSSYTX channel, marking CLOSED\n"); 2932 con->state = CON_STATE_CLOSED; 2933 return; 2934 } 2935 2936 if (con->in_msg) { 2937 BUG_ON(con->in_msg->con != con); 2938 ceph_msg_put(con->in_msg); 2939 con->in_msg = NULL; 2940 } 2941 2942 /* Requeue anything that hasn't been acked */ 2943 list_splice_init(&con->out_sent, &con->out_queue); 2944 2945 /* If there are no messages queued or keepalive pending, place 2946 * the connection in a STANDBY state */ 2947 if (list_empty(&con->out_queue) && 2948 !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) { 2949 dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con); 2950 con_flag_clear(con, CON_FLAG_WRITE_PENDING); 2951 con->state = CON_STATE_STANDBY; 2952 } else { 2953 /* retry after a delay. */ 2954 con->state = CON_STATE_PREOPEN; 2955 if (con->delay == 0) 2956 con->delay = BASE_DELAY_INTERVAL; 2957 else if (con->delay < MAX_DELAY_INTERVAL) 2958 con->delay *= 2; 2959 con_flag_set(con, CON_FLAG_BACKOFF); 2960 queue_con(con); 2961 } 2962 } 2963 2964 2965 2966 /* 2967 * initialize a new messenger instance 2968 */ 2969 void ceph_messenger_init(struct ceph_messenger *msgr, 2970 struct ceph_entity_addr *myaddr) 2971 { 2972 spin_lock_init(&msgr->global_seq_lock); 2973 2974 if (myaddr) 2975 msgr->inst.addr = *myaddr; 2976 2977 /* select a random nonce */ 2978 msgr->inst.addr.type = 0; 2979 get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce)); 2980 encode_my_addr(msgr); 2981 2982 atomic_set(&msgr->stopping, 0); 2983 write_pnet(&msgr->net, get_net(current->nsproxy->net_ns)); 2984 2985 dout("%s %p\n", __func__, msgr); 2986 } 2987 EXPORT_SYMBOL(ceph_messenger_init); 2988 2989 void ceph_messenger_fini(struct ceph_messenger *msgr) 2990 { 2991 put_net(read_pnet(&msgr->net)); 2992 } 2993 EXPORT_SYMBOL(ceph_messenger_fini); 2994 2995 static void msg_con_set(struct ceph_msg *msg, struct ceph_connection *con) 2996 { 2997 if (msg->con) 2998 msg->con->ops->put(msg->con); 2999 3000 msg->con = con ? con->ops->get(con) : NULL; 3001 BUG_ON(msg->con != con); 3002 } 3003 3004 static void clear_standby(struct ceph_connection *con) 3005 { 3006 /* come back from STANDBY? */ 3007 if (con->state == CON_STATE_STANDBY) { 3008 dout("clear_standby %p and ++connect_seq\n", con); 3009 con->state = CON_STATE_PREOPEN; 3010 con->connect_seq++; 3011 WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING)); 3012 WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)); 3013 } 3014 } 3015 3016 /* 3017 * Queue up an outgoing message on the given connection. 3018 */ 3019 void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg) 3020 { 3021 /* set src+dst */ 3022 msg->hdr.src = con->msgr->inst.name; 3023 BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len)); 3024 msg->needs_out_seq = true; 3025 3026 mutex_lock(&con->mutex); 3027 3028 if (con->state == CON_STATE_CLOSED) { 3029 dout("con_send %p closed, dropping %p\n", con, msg); 3030 ceph_msg_put(msg); 3031 mutex_unlock(&con->mutex); 3032 return; 3033 } 3034 3035 msg_con_set(msg, con); 3036 3037 BUG_ON(!list_empty(&msg->list_head)); 3038 list_add_tail(&msg->list_head, &con->out_queue); 3039 dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg, 3040 ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type), 3041 ceph_msg_type_name(le16_to_cpu(msg->hdr.type)), 3042 le32_to_cpu(msg->hdr.front_len), 3043 le32_to_cpu(msg->hdr.middle_len), 3044 le32_to_cpu(msg->hdr.data_len)); 3045 3046 clear_standby(con); 3047 mutex_unlock(&con->mutex); 3048 3049 /* if there wasn't anything waiting to send before, queue 3050 * new work */ 3051 if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0) 3052 queue_con(con); 3053 } 3054 EXPORT_SYMBOL(ceph_con_send); 3055 3056 /* 3057 * Revoke a message that was previously queued for send 3058 */ 3059 void ceph_msg_revoke(struct ceph_msg *msg) 3060 { 3061 struct ceph_connection *con = msg->con; 3062 3063 if (!con) { 3064 dout("%s msg %p null con\n", __func__, msg); 3065 return; /* Message not in our possession */ 3066 } 3067 3068 mutex_lock(&con->mutex); 3069 if (!list_empty(&msg->list_head)) { 3070 dout("%s %p msg %p - was on queue\n", __func__, con, msg); 3071 list_del_init(&msg->list_head); 3072 msg->hdr.seq = 0; 3073 3074 ceph_msg_put(msg); 3075 } 3076 if (con->out_msg == msg) { 3077 BUG_ON(con->out_skip); 3078 /* footer */ 3079 if (con->out_msg_done) { 3080 con->out_skip += con_out_kvec_skip(con); 3081 } else { 3082 BUG_ON(!msg->data_length); 3083 con->out_skip += sizeof_footer(con); 3084 } 3085 /* data, middle, front */ 3086 if (msg->data_length) 3087 con->out_skip += msg->cursor.total_resid; 3088 if (msg->middle) 3089 con->out_skip += con_out_kvec_skip(con); 3090 con->out_skip += con_out_kvec_skip(con); 3091 3092 dout("%s %p msg %p - was sending, will write %d skip %d\n", 3093 __func__, con, msg, con->out_kvec_bytes, con->out_skip); 3094 msg->hdr.seq = 0; 3095 con->out_msg = NULL; 3096 ceph_msg_put(msg); 3097 } 3098 3099 mutex_unlock(&con->mutex); 3100 } 3101 3102 /* 3103 * Revoke a message that we may be reading data into 3104 */ 3105 void ceph_msg_revoke_incoming(struct ceph_msg *msg) 3106 { 3107 struct ceph_connection *con = msg->con; 3108 3109 if (!con) { 3110 dout("%s msg %p null con\n", __func__, msg); 3111 return; /* Message not in our possession */ 3112 } 3113 3114 mutex_lock(&con->mutex); 3115 if (con->in_msg == msg) { 3116 unsigned int front_len = le32_to_cpu(con->in_hdr.front_len); 3117 unsigned int middle_len = le32_to_cpu(con->in_hdr.middle_len); 3118 unsigned int data_len = le32_to_cpu(con->in_hdr.data_len); 3119 3120 /* skip rest of message */ 3121 dout("%s %p msg %p revoked\n", __func__, con, msg); 3122 con->in_base_pos = con->in_base_pos - 3123 sizeof(struct ceph_msg_header) - 3124 front_len - 3125 middle_len - 3126 data_len - 3127 sizeof(struct ceph_msg_footer); 3128 ceph_msg_put(con->in_msg); 3129 con->in_msg = NULL; 3130 con->in_tag = CEPH_MSGR_TAG_READY; 3131 con->in_seq++; 3132 } else { 3133 dout("%s %p in_msg %p msg %p no-op\n", 3134 __func__, con, con->in_msg, msg); 3135 } 3136 mutex_unlock(&con->mutex); 3137 } 3138 3139 /* 3140 * Queue a keepalive byte to ensure the tcp connection is alive. 3141 */ 3142 void ceph_con_keepalive(struct ceph_connection *con) 3143 { 3144 dout("con_keepalive %p\n", con); 3145 mutex_lock(&con->mutex); 3146 clear_standby(con); 3147 mutex_unlock(&con->mutex); 3148 if (con_flag_test_and_set(con, CON_FLAG_KEEPALIVE_PENDING) == 0 && 3149 con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0) 3150 queue_con(con); 3151 } 3152 EXPORT_SYMBOL(ceph_con_keepalive); 3153 3154 bool ceph_con_keepalive_expired(struct ceph_connection *con, 3155 unsigned long interval) 3156 { 3157 if (interval > 0 && 3158 (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2)) { 3159 struct timespec now = CURRENT_TIME; 3160 struct timespec ts; 3161 jiffies_to_timespec(interval, &ts); 3162 ts = timespec_add(con->last_keepalive_ack, ts); 3163 return timespec_compare(&now, &ts) >= 0; 3164 } 3165 return false; 3166 } 3167 3168 static struct ceph_msg_data *ceph_msg_data_create(enum ceph_msg_data_type type) 3169 { 3170 struct ceph_msg_data *data; 3171 3172 if (WARN_ON(!ceph_msg_data_type_valid(type))) 3173 return NULL; 3174 3175 data = kmem_cache_zalloc(ceph_msg_data_cache, GFP_NOFS); 3176 if (data) 3177 data->type = type; 3178 INIT_LIST_HEAD(&data->links); 3179 3180 return data; 3181 } 3182 3183 static void ceph_msg_data_destroy(struct ceph_msg_data *data) 3184 { 3185 if (!data) 3186 return; 3187 3188 WARN_ON(!list_empty(&data->links)); 3189 if (data->type == CEPH_MSG_DATA_PAGELIST) 3190 ceph_pagelist_release(data->pagelist); 3191 kmem_cache_free(ceph_msg_data_cache, data); 3192 } 3193 3194 void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages, 3195 size_t length, size_t alignment) 3196 { 3197 struct ceph_msg_data *data; 3198 3199 BUG_ON(!pages); 3200 BUG_ON(!length); 3201 3202 data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES); 3203 BUG_ON(!data); 3204 data->pages = pages; 3205 data->length = length; 3206 data->alignment = alignment & ~PAGE_MASK; 3207 3208 list_add_tail(&data->links, &msg->data); 3209 msg->data_length += length; 3210 } 3211 EXPORT_SYMBOL(ceph_msg_data_add_pages); 3212 3213 void ceph_msg_data_add_pagelist(struct ceph_msg *msg, 3214 struct ceph_pagelist *pagelist) 3215 { 3216 struct ceph_msg_data *data; 3217 3218 BUG_ON(!pagelist); 3219 BUG_ON(!pagelist->length); 3220 3221 data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST); 3222 BUG_ON(!data); 3223 data->pagelist = pagelist; 3224 3225 list_add_tail(&data->links, &msg->data); 3226 msg->data_length += pagelist->length; 3227 } 3228 EXPORT_SYMBOL(ceph_msg_data_add_pagelist); 3229 3230 #ifdef CONFIG_BLOCK 3231 void ceph_msg_data_add_bio(struct ceph_msg *msg, struct bio *bio, 3232 size_t length) 3233 { 3234 struct ceph_msg_data *data; 3235 3236 BUG_ON(!bio); 3237 3238 data = ceph_msg_data_create(CEPH_MSG_DATA_BIO); 3239 BUG_ON(!data); 3240 data->bio = bio; 3241 data->bio_length = length; 3242 3243 list_add_tail(&data->links, &msg->data); 3244 msg->data_length += length; 3245 } 3246 EXPORT_SYMBOL(ceph_msg_data_add_bio); 3247 #endif /* CONFIG_BLOCK */ 3248 3249 /* 3250 * construct a new message with given type, size 3251 * the new msg has a ref count of 1. 3252 */ 3253 struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags, 3254 bool can_fail) 3255 { 3256 struct ceph_msg *m; 3257 3258 m = kmem_cache_zalloc(ceph_msg_cache, flags); 3259 if (m == NULL) 3260 goto out; 3261 3262 m->hdr.type = cpu_to_le16(type); 3263 m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT); 3264 m->hdr.front_len = cpu_to_le32(front_len); 3265 3266 INIT_LIST_HEAD(&m->list_head); 3267 kref_init(&m->kref); 3268 INIT_LIST_HEAD(&m->data); 3269 3270 /* front */ 3271 if (front_len) { 3272 m->front.iov_base = ceph_kvmalloc(front_len, flags); 3273 if (m->front.iov_base == NULL) { 3274 dout("ceph_msg_new can't allocate %d bytes\n", 3275 front_len); 3276 goto out2; 3277 } 3278 } else { 3279 m->front.iov_base = NULL; 3280 } 3281 m->front_alloc_len = m->front.iov_len = front_len; 3282 3283 dout("ceph_msg_new %p front %d\n", m, front_len); 3284 return m; 3285 3286 out2: 3287 ceph_msg_put(m); 3288 out: 3289 if (!can_fail) { 3290 pr_err("msg_new can't create type %d front %d\n", type, 3291 front_len); 3292 WARN_ON(1); 3293 } else { 3294 dout("msg_new can't create type %d front %d\n", type, 3295 front_len); 3296 } 3297 return NULL; 3298 } 3299 EXPORT_SYMBOL(ceph_msg_new); 3300 3301 /* 3302 * Allocate "middle" portion of a message, if it is needed and wasn't 3303 * allocated by alloc_msg. This allows us to read a small fixed-size 3304 * per-type header in the front and then gracefully fail (i.e., 3305 * propagate the error to the caller based on info in the front) when 3306 * the middle is too large. 3307 */ 3308 static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg) 3309 { 3310 int type = le16_to_cpu(msg->hdr.type); 3311 int middle_len = le32_to_cpu(msg->hdr.middle_len); 3312 3313 dout("alloc_middle %p type %d %s middle_len %d\n", msg, type, 3314 ceph_msg_type_name(type), middle_len); 3315 BUG_ON(!middle_len); 3316 BUG_ON(msg->middle); 3317 3318 msg->middle = ceph_buffer_new(middle_len, GFP_NOFS); 3319 if (!msg->middle) 3320 return -ENOMEM; 3321 return 0; 3322 } 3323 3324 /* 3325 * Allocate a message for receiving an incoming message on a 3326 * connection, and save the result in con->in_msg. Uses the 3327 * connection's private alloc_msg op if available. 3328 * 3329 * Returns 0 on success, or a negative error code. 3330 * 3331 * On success, if we set *skip = 1: 3332 * - the next message should be skipped and ignored. 3333 * - con->in_msg == NULL 3334 * or if we set *skip = 0: 3335 * - con->in_msg is non-null. 3336 * On error (ENOMEM, EAGAIN, ...), 3337 * - con->in_msg == NULL 3338 */ 3339 static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip) 3340 { 3341 struct ceph_msg_header *hdr = &con->in_hdr; 3342 int middle_len = le32_to_cpu(hdr->middle_len); 3343 struct ceph_msg *msg; 3344 int ret = 0; 3345 3346 BUG_ON(con->in_msg != NULL); 3347 BUG_ON(!con->ops->alloc_msg); 3348 3349 mutex_unlock(&con->mutex); 3350 msg = con->ops->alloc_msg(con, hdr, skip); 3351 mutex_lock(&con->mutex); 3352 if (con->state != CON_STATE_OPEN) { 3353 if (msg) 3354 ceph_msg_put(msg); 3355 return -EAGAIN; 3356 } 3357 if (msg) { 3358 BUG_ON(*skip); 3359 msg_con_set(msg, con); 3360 con->in_msg = msg; 3361 } else { 3362 /* 3363 * Null message pointer means either we should skip 3364 * this message or we couldn't allocate memory. The 3365 * former is not an error. 3366 */ 3367 if (*skip) 3368 return 0; 3369 3370 con->error_msg = "error allocating memory for incoming message"; 3371 return -ENOMEM; 3372 } 3373 memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr)); 3374 3375 if (middle_len && !con->in_msg->middle) { 3376 ret = ceph_alloc_middle(con, con->in_msg); 3377 if (ret < 0) { 3378 ceph_msg_put(con->in_msg); 3379 con->in_msg = NULL; 3380 } 3381 } 3382 3383 return ret; 3384 } 3385 3386 3387 /* 3388 * Free a generically kmalloc'd message. 3389 */ 3390 static void ceph_msg_free(struct ceph_msg *m) 3391 { 3392 dout("%s %p\n", __func__, m); 3393 kvfree(m->front.iov_base); 3394 kmem_cache_free(ceph_msg_cache, m); 3395 } 3396 3397 static void ceph_msg_release(struct kref *kref) 3398 { 3399 struct ceph_msg *m = container_of(kref, struct ceph_msg, kref); 3400 struct ceph_msg_data *data, *next; 3401 3402 dout("%s %p\n", __func__, m); 3403 WARN_ON(!list_empty(&m->list_head)); 3404 3405 msg_con_set(m, NULL); 3406 3407 /* drop middle, data, if any */ 3408 if (m->middle) { 3409 ceph_buffer_put(m->middle); 3410 m->middle = NULL; 3411 } 3412 3413 list_for_each_entry_safe(data, next, &m->data, links) { 3414 list_del_init(&data->links); 3415 ceph_msg_data_destroy(data); 3416 } 3417 m->data_length = 0; 3418 3419 if (m->pool) 3420 ceph_msgpool_put(m->pool, m); 3421 else 3422 ceph_msg_free(m); 3423 } 3424 3425 struct ceph_msg *ceph_msg_get(struct ceph_msg *msg) 3426 { 3427 dout("%s %p (was %d)\n", __func__, msg, 3428 atomic_read(&msg->kref.refcount)); 3429 kref_get(&msg->kref); 3430 return msg; 3431 } 3432 EXPORT_SYMBOL(ceph_msg_get); 3433 3434 void ceph_msg_put(struct ceph_msg *msg) 3435 { 3436 dout("%s %p (was %d)\n", __func__, msg, 3437 atomic_read(&msg->kref.refcount)); 3438 kref_put(&msg->kref, ceph_msg_release); 3439 } 3440 EXPORT_SYMBOL(ceph_msg_put); 3441 3442 void ceph_msg_dump(struct ceph_msg *msg) 3443 { 3444 pr_debug("msg_dump %p (front_alloc_len %d length %zd)\n", msg, 3445 msg->front_alloc_len, msg->data_length); 3446 print_hex_dump(KERN_DEBUG, "header: ", 3447 DUMP_PREFIX_OFFSET, 16, 1, 3448 &msg->hdr, sizeof(msg->hdr), true); 3449 print_hex_dump(KERN_DEBUG, " front: ", 3450 DUMP_PREFIX_OFFSET, 16, 1, 3451 msg->front.iov_base, msg->front.iov_len, true); 3452 if (msg->middle) 3453 print_hex_dump(KERN_DEBUG, "middle: ", 3454 DUMP_PREFIX_OFFSET, 16, 1, 3455 msg->middle->vec.iov_base, 3456 msg->middle->vec.iov_len, true); 3457 print_hex_dump(KERN_DEBUG, "footer: ", 3458 DUMP_PREFIX_OFFSET, 16, 1, 3459 &msg->footer, sizeof(msg->footer), true); 3460 } 3461 EXPORT_SYMBOL(ceph_msg_dump); 3462