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