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