1 /* 2 drbd_receiver.c 3 4 This file is part of DRBD by Philipp Reisner and Lars Ellenberg. 5 6 Copyright (C) 2001-2008, LINBIT Information Technologies GmbH. 7 Copyright (C) 1999-2008, Philipp Reisner <philipp.reisner@linbit.com>. 8 Copyright (C) 2002-2008, Lars Ellenberg <lars.ellenberg@linbit.com>. 9 10 drbd is free software; you can redistribute it and/or modify 11 it under the terms of the GNU General Public License as published by 12 the Free Software Foundation; either version 2, or (at your option) 13 any later version. 14 15 drbd is distributed in the hope that it will be useful, 16 but WITHOUT ANY WARRANTY; without even the implied warranty of 17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 18 GNU General Public License for more details. 19 20 You should have received a copy of the GNU General Public License 21 along with drbd; see the file COPYING. If not, write to 22 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA. 23 */ 24 25 26 #include <linux/module.h> 27 28 #include <asm/uaccess.h> 29 #include <net/sock.h> 30 31 #include <linux/version.h> 32 #include <linux/drbd.h> 33 #include <linux/fs.h> 34 #include <linux/file.h> 35 #include <linux/in.h> 36 #include <linux/mm.h> 37 #include <linux/memcontrol.h> 38 #include <linux/mm_inline.h> 39 #include <linux/slab.h> 40 #include <linux/smp_lock.h> 41 #include <linux/pkt_sched.h> 42 #define __KERNEL_SYSCALLS__ 43 #include <linux/unistd.h> 44 #include <linux/vmalloc.h> 45 #include <linux/random.h> 46 #include <linux/mm.h> 47 #include <linux/string.h> 48 #include <linux/scatterlist.h> 49 #include "drbd_int.h" 50 #include "drbd_req.h" 51 52 #include "drbd_vli.h" 53 54 struct flush_work { 55 struct drbd_work w; 56 struct drbd_epoch *epoch; 57 }; 58 59 enum finish_epoch { 60 FE_STILL_LIVE, 61 FE_DESTROYED, 62 FE_RECYCLED, 63 }; 64 65 static int drbd_do_handshake(struct drbd_conf *mdev); 66 static int drbd_do_auth(struct drbd_conf *mdev); 67 68 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *, struct drbd_epoch *, enum epoch_event); 69 static int e_end_block(struct drbd_conf *, struct drbd_work *, int); 70 71 static struct drbd_epoch *previous_epoch(struct drbd_conf *mdev, struct drbd_epoch *epoch) 72 { 73 struct drbd_epoch *prev; 74 spin_lock(&mdev->epoch_lock); 75 prev = list_entry(epoch->list.prev, struct drbd_epoch, list); 76 if (prev == epoch || prev == mdev->current_epoch) 77 prev = NULL; 78 spin_unlock(&mdev->epoch_lock); 79 return prev; 80 } 81 82 #define GFP_TRY (__GFP_HIGHMEM | __GFP_NOWARN) 83 84 static struct page *drbd_pp_first_page_or_try_alloc(struct drbd_conf *mdev) 85 { 86 struct page *page = NULL; 87 88 /* Yes, testing drbd_pp_vacant outside the lock is racy. 89 * So what. It saves a spin_lock. */ 90 if (drbd_pp_vacant > 0) { 91 spin_lock(&drbd_pp_lock); 92 page = drbd_pp_pool; 93 if (page) { 94 drbd_pp_pool = (struct page *)page_private(page); 95 set_page_private(page, 0); /* just to be polite */ 96 drbd_pp_vacant--; 97 } 98 spin_unlock(&drbd_pp_lock); 99 } 100 /* GFP_TRY, because we must not cause arbitrary write-out: in a DRBD 101 * "criss-cross" setup, that might cause write-out on some other DRBD, 102 * which in turn might block on the other node at this very place. */ 103 if (!page) 104 page = alloc_page(GFP_TRY); 105 if (page) 106 atomic_inc(&mdev->pp_in_use); 107 return page; 108 } 109 110 /* kick lower level device, if we have more than (arbitrary number) 111 * reference counts on it, which typically are locally submitted io 112 * requests. don't use unacked_cnt, so we speed up proto A and B, too. */ 113 static void maybe_kick_lo(struct drbd_conf *mdev) 114 { 115 if (atomic_read(&mdev->local_cnt) >= mdev->net_conf->unplug_watermark) 116 drbd_kick_lo(mdev); 117 } 118 119 static void reclaim_net_ee(struct drbd_conf *mdev, struct list_head *to_be_freed) 120 { 121 struct drbd_epoch_entry *e; 122 struct list_head *le, *tle; 123 124 /* The EEs are always appended to the end of the list. Since 125 they are sent in order over the wire, they have to finish 126 in order. As soon as we see the first not finished we can 127 stop to examine the list... */ 128 129 list_for_each_safe(le, tle, &mdev->net_ee) { 130 e = list_entry(le, struct drbd_epoch_entry, w.list); 131 if (drbd_bio_has_active_page(e->private_bio)) 132 break; 133 list_move(le, to_be_freed); 134 } 135 } 136 137 static void drbd_kick_lo_and_reclaim_net(struct drbd_conf *mdev) 138 { 139 LIST_HEAD(reclaimed); 140 struct drbd_epoch_entry *e, *t; 141 142 maybe_kick_lo(mdev); 143 spin_lock_irq(&mdev->req_lock); 144 reclaim_net_ee(mdev, &reclaimed); 145 spin_unlock_irq(&mdev->req_lock); 146 147 list_for_each_entry_safe(e, t, &reclaimed, w.list) 148 drbd_free_ee(mdev, e); 149 } 150 151 /** 152 * drbd_pp_alloc() - Returns a page, fails only if a signal comes in 153 * @mdev: DRBD device. 154 * @retry: whether or not to retry allocation forever (or until signalled) 155 * 156 * Tries to allocate a page, first from our own page pool, then from the 157 * kernel, unless this allocation would exceed the max_buffers setting. 158 * If @retry is non-zero, retry until DRBD frees a page somewhere else. 159 */ 160 static struct page *drbd_pp_alloc(struct drbd_conf *mdev, int retry) 161 { 162 struct page *page = NULL; 163 DEFINE_WAIT(wait); 164 165 if (atomic_read(&mdev->pp_in_use) < mdev->net_conf->max_buffers) { 166 page = drbd_pp_first_page_or_try_alloc(mdev); 167 if (page) 168 return page; 169 } 170 171 for (;;) { 172 prepare_to_wait(&drbd_pp_wait, &wait, TASK_INTERRUPTIBLE); 173 174 drbd_kick_lo_and_reclaim_net(mdev); 175 176 if (atomic_read(&mdev->pp_in_use) < mdev->net_conf->max_buffers) { 177 page = drbd_pp_first_page_or_try_alloc(mdev); 178 if (page) 179 break; 180 } 181 182 if (!retry) 183 break; 184 185 if (signal_pending(current)) { 186 dev_warn(DEV, "drbd_pp_alloc interrupted!\n"); 187 break; 188 } 189 190 schedule(); 191 } 192 finish_wait(&drbd_pp_wait, &wait); 193 194 return page; 195 } 196 197 /* Must not be used from irq, as that may deadlock: see drbd_pp_alloc. 198 * Is also used from inside an other spin_lock_irq(&mdev->req_lock) */ 199 static void drbd_pp_free(struct drbd_conf *mdev, struct page *page) 200 { 201 int free_it; 202 203 spin_lock(&drbd_pp_lock); 204 if (drbd_pp_vacant > (DRBD_MAX_SEGMENT_SIZE/PAGE_SIZE)*minor_count) { 205 free_it = 1; 206 } else { 207 set_page_private(page, (unsigned long)drbd_pp_pool); 208 drbd_pp_pool = page; 209 drbd_pp_vacant++; 210 free_it = 0; 211 } 212 spin_unlock(&drbd_pp_lock); 213 214 atomic_dec(&mdev->pp_in_use); 215 216 if (free_it) 217 __free_page(page); 218 219 wake_up(&drbd_pp_wait); 220 } 221 222 static void drbd_pp_free_bio_pages(struct drbd_conf *mdev, struct bio *bio) 223 { 224 struct page *p_to_be_freed = NULL; 225 struct page *page; 226 struct bio_vec *bvec; 227 int i; 228 229 spin_lock(&drbd_pp_lock); 230 __bio_for_each_segment(bvec, bio, i, 0) { 231 if (drbd_pp_vacant > (DRBD_MAX_SEGMENT_SIZE/PAGE_SIZE)*minor_count) { 232 set_page_private(bvec->bv_page, (unsigned long)p_to_be_freed); 233 p_to_be_freed = bvec->bv_page; 234 } else { 235 set_page_private(bvec->bv_page, (unsigned long)drbd_pp_pool); 236 drbd_pp_pool = bvec->bv_page; 237 drbd_pp_vacant++; 238 } 239 } 240 spin_unlock(&drbd_pp_lock); 241 atomic_sub(bio->bi_vcnt, &mdev->pp_in_use); 242 243 while (p_to_be_freed) { 244 page = p_to_be_freed; 245 p_to_be_freed = (struct page *)page_private(page); 246 set_page_private(page, 0); /* just to be polite */ 247 put_page(page); 248 } 249 250 wake_up(&drbd_pp_wait); 251 } 252 253 /* 254 You need to hold the req_lock: 255 _drbd_wait_ee_list_empty() 256 257 You must not have the req_lock: 258 drbd_free_ee() 259 drbd_alloc_ee() 260 drbd_init_ee() 261 drbd_release_ee() 262 drbd_ee_fix_bhs() 263 drbd_process_done_ee() 264 drbd_clear_done_ee() 265 drbd_wait_ee_list_empty() 266 */ 267 268 struct drbd_epoch_entry *drbd_alloc_ee(struct drbd_conf *mdev, 269 u64 id, 270 sector_t sector, 271 unsigned int data_size, 272 gfp_t gfp_mask) __must_hold(local) 273 { 274 struct request_queue *q; 275 struct drbd_epoch_entry *e; 276 struct page *page; 277 struct bio *bio; 278 unsigned int ds; 279 280 if (FAULT_ACTIVE(mdev, DRBD_FAULT_AL_EE)) 281 return NULL; 282 283 e = mempool_alloc(drbd_ee_mempool, gfp_mask & ~__GFP_HIGHMEM); 284 if (!e) { 285 if (!(gfp_mask & __GFP_NOWARN)) 286 dev_err(DEV, "alloc_ee: Allocation of an EE failed\n"); 287 return NULL; 288 } 289 290 bio = bio_alloc(gfp_mask & ~__GFP_HIGHMEM, div_ceil(data_size, PAGE_SIZE)); 291 if (!bio) { 292 if (!(gfp_mask & __GFP_NOWARN)) 293 dev_err(DEV, "alloc_ee: Allocation of a bio failed\n"); 294 goto fail1; 295 } 296 297 bio->bi_bdev = mdev->ldev->backing_bdev; 298 bio->bi_sector = sector; 299 300 ds = data_size; 301 while (ds) { 302 page = drbd_pp_alloc(mdev, (gfp_mask & __GFP_WAIT)); 303 if (!page) { 304 if (!(gfp_mask & __GFP_NOWARN)) 305 dev_err(DEV, "alloc_ee: Allocation of a page failed\n"); 306 goto fail2; 307 } 308 if (!bio_add_page(bio, page, min_t(int, ds, PAGE_SIZE), 0)) { 309 drbd_pp_free(mdev, page); 310 dev_err(DEV, "alloc_ee: bio_add_page(s=%llu," 311 "data_size=%u,ds=%u) failed\n", 312 (unsigned long long)sector, data_size, ds); 313 314 q = bdev_get_queue(bio->bi_bdev); 315 if (q->merge_bvec_fn) { 316 struct bvec_merge_data bvm = { 317 .bi_bdev = bio->bi_bdev, 318 .bi_sector = bio->bi_sector, 319 .bi_size = bio->bi_size, 320 .bi_rw = bio->bi_rw, 321 }; 322 int l = q->merge_bvec_fn(q, &bvm, 323 &bio->bi_io_vec[bio->bi_vcnt]); 324 dev_err(DEV, "merge_bvec_fn() = %d\n", l); 325 } 326 327 /* dump more of the bio. */ 328 dev_err(DEV, "bio->bi_max_vecs = %d\n", bio->bi_max_vecs); 329 dev_err(DEV, "bio->bi_vcnt = %d\n", bio->bi_vcnt); 330 dev_err(DEV, "bio->bi_size = %d\n", bio->bi_size); 331 dev_err(DEV, "bio->bi_phys_segments = %d\n", bio->bi_phys_segments); 332 333 goto fail2; 334 break; 335 } 336 ds -= min_t(int, ds, PAGE_SIZE); 337 } 338 339 D_ASSERT(data_size == bio->bi_size); 340 341 bio->bi_private = e; 342 e->mdev = mdev; 343 e->sector = sector; 344 e->size = bio->bi_size; 345 346 e->private_bio = bio; 347 e->block_id = id; 348 INIT_HLIST_NODE(&e->colision); 349 e->epoch = NULL; 350 e->flags = 0; 351 352 return e; 353 354 fail2: 355 drbd_pp_free_bio_pages(mdev, bio); 356 bio_put(bio); 357 fail1: 358 mempool_free(e, drbd_ee_mempool); 359 360 return NULL; 361 } 362 363 void drbd_free_ee(struct drbd_conf *mdev, struct drbd_epoch_entry *e) 364 { 365 struct bio *bio = e->private_bio; 366 drbd_pp_free_bio_pages(mdev, bio); 367 bio_put(bio); 368 D_ASSERT(hlist_unhashed(&e->colision)); 369 mempool_free(e, drbd_ee_mempool); 370 } 371 372 int drbd_release_ee(struct drbd_conf *mdev, struct list_head *list) 373 { 374 LIST_HEAD(work_list); 375 struct drbd_epoch_entry *e, *t; 376 int count = 0; 377 378 spin_lock_irq(&mdev->req_lock); 379 list_splice_init(list, &work_list); 380 spin_unlock_irq(&mdev->req_lock); 381 382 list_for_each_entry_safe(e, t, &work_list, w.list) { 383 drbd_free_ee(mdev, e); 384 count++; 385 } 386 return count; 387 } 388 389 390 /* 391 * This function is called from _asender only_ 392 * but see also comments in _req_mod(,barrier_acked) 393 * and receive_Barrier. 394 * 395 * Move entries from net_ee to done_ee, if ready. 396 * Grab done_ee, call all callbacks, free the entries. 397 * The callbacks typically send out ACKs. 398 */ 399 static int drbd_process_done_ee(struct drbd_conf *mdev) 400 { 401 LIST_HEAD(work_list); 402 LIST_HEAD(reclaimed); 403 struct drbd_epoch_entry *e, *t; 404 int ok = (mdev->state.conn >= C_WF_REPORT_PARAMS); 405 406 spin_lock_irq(&mdev->req_lock); 407 reclaim_net_ee(mdev, &reclaimed); 408 list_splice_init(&mdev->done_ee, &work_list); 409 spin_unlock_irq(&mdev->req_lock); 410 411 list_for_each_entry_safe(e, t, &reclaimed, w.list) 412 drbd_free_ee(mdev, e); 413 414 /* possible callbacks here: 415 * e_end_block, and e_end_resync_block, e_send_discard_ack. 416 * all ignore the last argument. 417 */ 418 list_for_each_entry_safe(e, t, &work_list, w.list) { 419 /* list_del not necessary, next/prev members not touched */ 420 ok = e->w.cb(mdev, &e->w, !ok) && ok; 421 drbd_free_ee(mdev, e); 422 } 423 wake_up(&mdev->ee_wait); 424 425 return ok; 426 } 427 428 void _drbd_wait_ee_list_empty(struct drbd_conf *mdev, struct list_head *head) 429 { 430 DEFINE_WAIT(wait); 431 432 /* avoids spin_lock/unlock 433 * and calling prepare_to_wait in the fast path */ 434 while (!list_empty(head)) { 435 prepare_to_wait(&mdev->ee_wait, &wait, TASK_UNINTERRUPTIBLE); 436 spin_unlock_irq(&mdev->req_lock); 437 drbd_kick_lo(mdev); 438 schedule(); 439 finish_wait(&mdev->ee_wait, &wait); 440 spin_lock_irq(&mdev->req_lock); 441 } 442 } 443 444 void drbd_wait_ee_list_empty(struct drbd_conf *mdev, struct list_head *head) 445 { 446 spin_lock_irq(&mdev->req_lock); 447 _drbd_wait_ee_list_empty(mdev, head); 448 spin_unlock_irq(&mdev->req_lock); 449 } 450 451 /* see also kernel_accept; which is only present since 2.6.18. 452 * also we want to log which part of it failed, exactly */ 453 static int drbd_accept(struct drbd_conf *mdev, const char **what, 454 struct socket *sock, struct socket **newsock) 455 { 456 struct sock *sk = sock->sk; 457 int err = 0; 458 459 *what = "listen"; 460 err = sock->ops->listen(sock, 5); 461 if (err < 0) 462 goto out; 463 464 *what = "sock_create_lite"; 465 err = sock_create_lite(sk->sk_family, sk->sk_type, sk->sk_protocol, 466 newsock); 467 if (err < 0) 468 goto out; 469 470 *what = "accept"; 471 err = sock->ops->accept(sock, *newsock, 0); 472 if (err < 0) { 473 sock_release(*newsock); 474 *newsock = NULL; 475 goto out; 476 } 477 (*newsock)->ops = sock->ops; 478 479 out: 480 return err; 481 } 482 483 static int drbd_recv_short(struct drbd_conf *mdev, struct socket *sock, 484 void *buf, size_t size, int flags) 485 { 486 mm_segment_t oldfs; 487 struct kvec iov = { 488 .iov_base = buf, 489 .iov_len = size, 490 }; 491 struct msghdr msg = { 492 .msg_iovlen = 1, 493 .msg_iov = (struct iovec *)&iov, 494 .msg_flags = (flags ? flags : MSG_WAITALL | MSG_NOSIGNAL) 495 }; 496 int rv; 497 498 oldfs = get_fs(); 499 set_fs(KERNEL_DS); 500 rv = sock_recvmsg(sock, &msg, size, msg.msg_flags); 501 set_fs(oldfs); 502 503 return rv; 504 } 505 506 static int drbd_recv(struct drbd_conf *mdev, void *buf, size_t size) 507 { 508 mm_segment_t oldfs; 509 struct kvec iov = { 510 .iov_base = buf, 511 .iov_len = size, 512 }; 513 struct msghdr msg = { 514 .msg_iovlen = 1, 515 .msg_iov = (struct iovec *)&iov, 516 .msg_flags = MSG_WAITALL | MSG_NOSIGNAL 517 }; 518 int rv; 519 520 oldfs = get_fs(); 521 set_fs(KERNEL_DS); 522 523 for (;;) { 524 rv = sock_recvmsg(mdev->data.socket, &msg, size, msg.msg_flags); 525 if (rv == size) 526 break; 527 528 /* Note: 529 * ECONNRESET other side closed the connection 530 * ERESTARTSYS (on sock) we got a signal 531 */ 532 533 if (rv < 0) { 534 if (rv == -ECONNRESET) 535 dev_info(DEV, "sock was reset by peer\n"); 536 else if (rv != -ERESTARTSYS) 537 dev_err(DEV, "sock_recvmsg returned %d\n", rv); 538 break; 539 } else if (rv == 0) { 540 dev_info(DEV, "sock was shut down by peer\n"); 541 break; 542 } else { 543 /* signal came in, or peer/link went down, 544 * after we read a partial message 545 */ 546 /* D_ASSERT(signal_pending(current)); */ 547 break; 548 } 549 }; 550 551 set_fs(oldfs); 552 553 if (rv != size) 554 drbd_force_state(mdev, NS(conn, C_BROKEN_PIPE)); 555 556 return rv; 557 } 558 559 static struct socket *drbd_try_connect(struct drbd_conf *mdev) 560 { 561 const char *what; 562 struct socket *sock; 563 struct sockaddr_in6 src_in6; 564 int err; 565 int disconnect_on_error = 1; 566 567 if (!get_net_conf(mdev)) 568 return NULL; 569 570 what = "sock_create_kern"; 571 err = sock_create_kern(((struct sockaddr *)mdev->net_conf->my_addr)->sa_family, 572 SOCK_STREAM, IPPROTO_TCP, &sock); 573 if (err < 0) { 574 sock = NULL; 575 goto out; 576 } 577 578 sock->sk->sk_rcvtimeo = 579 sock->sk->sk_sndtimeo = mdev->net_conf->try_connect_int*HZ; 580 581 /* explicitly bind to the configured IP as source IP 582 * for the outgoing connections. 583 * This is needed for multihomed hosts and to be 584 * able to use lo: interfaces for drbd. 585 * Make sure to use 0 as port number, so linux selects 586 * a free one dynamically. 587 */ 588 memcpy(&src_in6, mdev->net_conf->my_addr, 589 min_t(int, mdev->net_conf->my_addr_len, sizeof(src_in6))); 590 if (((struct sockaddr *)mdev->net_conf->my_addr)->sa_family == AF_INET6) 591 src_in6.sin6_port = 0; 592 else 593 ((struct sockaddr_in *)&src_in6)->sin_port = 0; /* AF_INET & AF_SCI */ 594 595 what = "bind before connect"; 596 err = sock->ops->bind(sock, 597 (struct sockaddr *) &src_in6, 598 mdev->net_conf->my_addr_len); 599 if (err < 0) 600 goto out; 601 602 /* connect may fail, peer not yet available. 603 * stay C_WF_CONNECTION, don't go Disconnecting! */ 604 disconnect_on_error = 0; 605 what = "connect"; 606 err = sock->ops->connect(sock, 607 (struct sockaddr *)mdev->net_conf->peer_addr, 608 mdev->net_conf->peer_addr_len, 0); 609 610 out: 611 if (err < 0) { 612 if (sock) { 613 sock_release(sock); 614 sock = NULL; 615 } 616 switch (-err) { 617 /* timeout, busy, signal pending */ 618 case ETIMEDOUT: case EAGAIN: case EINPROGRESS: 619 case EINTR: case ERESTARTSYS: 620 /* peer not (yet) available, network problem */ 621 case ECONNREFUSED: case ENETUNREACH: 622 case EHOSTDOWN: case EHOSTUNREACH: 623 disconnect_on_error = 0; 624 break; 625 default: 626 dev_err(DEV, "%s failed, err = %d\n", what, err); 627 } 628 if (disconnect_on_error) 629 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 630 } 631 put_net_conf(mdev); 632 return sock; 633 } 634 635 static struct socket *drbd_wait_for_connect(struct drbd_conf *mdev) 636 { 637 int timeo, err; 638 struct socket *s_estab = NULL, *s_listen; 639 const char *what; 640 641 if (!get_net_conf(mdev)) 642 return NULL; 643 644 what = "sock_create_kern"; 645 err = sock_create_kern(((struct sockaddr *)mdev->net_conf->my_addr)->sa_family, 646 SOCK_STREAM, IPPROTO_TCP, &s_listen); 647 if (err) { 648 s_listen = NULL; 649 goto out; 650 } 651 652 timeo = mdev->net_conf->try_connect_int * HZ; 653 timeo += (random32() & 1) ? timeo / 7 : -timeo / 7; /* 28.5% random jitter */ 654 655 s_listen->sk->sk_reuse = 1; /* SO_REUSEADDR */ 656 s_listen->sk->sk_rcvtimeo = timeo; 657 s_listen->sk->sk_sndtimeo = timeo; 658 659 what = "bind before listen"; 660 err = s_listen->ops->bind(s_listen, 661 (struct sockaddr *) mdev->net_conf->my_addr, 662 mdev->net_conf->my_addr_len); 663 if (err < 0) 664 goto out; 665 666 err = drbd_accept(mdev, &what, s_listen, &s_estab); 667 668 out: 669 if (s_listen) 670 sock_release(s_listen); 671 if (err < 0) { 672 if (err != -EAGAIN && err != -EINTR && err != -ERESTARTSYS) { 673 dev_err(DEV, "%s failed, err = %d\n", what, err); 674 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 675 } 676 } 677 put_net_conf(mdev); 678 679 return s_estab; 680 } 681 682 static int drbd_send_fp(struct drbd_conf *mdev, 683 struct socket *sock, enum drbd_packets cmd) 684 { 685 struct p_header *h = (struct p_header *) &mdev->data.sbuf.header; 686 687 return _drbd_send_cmd(mdev, sock, cmd, h, sizeof(*h), 0); 688 } 689 690 static enum drbd_packets drbd_recv_fp(struct drbd_conf *mdev, struct socket *sock) 691 { 692 struct p_header *h = (struct p_header *) &mdev->data.sbuf.header; 693 int rr; 694 695 rr = drbd_recv_short(mdev, sock, h, sizeof(*h), 0); 696 697 if (rr == sizeof(*h) && h->magic == BE_DRBD_MAGIC) 698 return be16_to_cpu(h->command); 699 700 return 0xffff; 701 } 702 703 /** 704 * drbd_socket_okay() - Free the socket if its connection is not okay 705 * @mdev: DRBD device. 706 * @sock: pointer to the pointer to the socket. 707 */ 708 static int drbd_socket_okay(struct drbd_conf *mdev, struct socket **sock) 709 { 710 int rr; 711 char tb[4]; 712 713 if (!*sock) 714 return FALSE; 715 716 rr = drbd_recv_short(mdev, *sock, tb, 4, MSG_DONTWAIT | MSG_PEEK); 717 718 if (rr > 0 || rr == -EAGAIN) { 719 return TRUE; 720 } else { 721 sock_release(*sock); 722 *sock = NULL; 723 return FALSE; 724 } 725 } 726 727 /* 728 * return values: 729 * 1 yes, we have a valid connection 730 * 0 oops, did not work out, please try again 731 * -1 peer talks different language, 732 * no point in trying again, please go standalone. 733 * -2 We do not have a network config... 734 */ 735 static int drbd_connect(struct drbd_conf *mdev) 736 { 737 struct socket *s, *sock, *msock; 738 int try, h, ok; 739 740 D_ASSERT(!mdev->data.socket); 741 742 if (test_and_clear_bit(CREATE_BARRIER, &mdev->flags)) 743 dev_err(DEV, "CREATE_BARRIER flag was set in drbd_connect - now cleared!\n"); 744 745 if (drbd_request_state(mdev, NS(conn, C_WF_CONNECTION)) < SS_SUCCESS) 746 return -2; 747 748 clear_bit(DISCARD_CONCURRENT, &mdev->flags); 749 750 sock = NULL; 751 msock = NULL; 752 753 do { 754 for (try = 0;;) { 755 /* 3 tries, this should take less than a second! */ 756 s = drbd_try_connect(mdev); 757 if (s || ++try >= 3) 758 break; 759 /* give the other side time to call bind() & listen() */ 760 __set_current_state(TASK_INTERRUPTIBLE); 761 schedule_timeout(HZ / 10); 762 } 763 764 if (s) { 765 if (!sock) { 766 drbd_send_fp(mdev, s, P_HAND_SHAKE_S); 767 sock = s; 768 s = NULL; 769 } else if (!msock) { 770 drbd_send_fp(mdev, s, P_HAND_SHAKE_M); 771 msock = s; 772 s = NULL; 773 } else { 774 dev_err(DEV, "Logic error in drbd_connect()\n"); 775 goto out_release_sockets; 776 } 777 } 778 779 if (sock && msock) { 780 __set_current_state(TASK_INTERRUPTIBLE); 781 schedule_timeout(HZ / 10); 782 ok = drbd_socket_okay(mdev, &sock); 783 ok = drbd_socket_okay(mdev, &msock) && ok; 784 if (ok) 785 break; 786 } 787 788 retry: 789 s = drbd_wait_for_connect(mdev); 790 if (s) { 791 try = drbd_recv_fp(mdev, s); 792 drbd_socket_okay(mdev, &sock); 793 drbd_socket_okay(mdev, &msock); 794 switch (try) { 795 case P_HAND_SHAKE_S: 796 if (sock) { 797 dev_warn(DEV, "initial packet S crossed\n"); 798 sock_release(sock); 799 } 800 sock = s; 801 break; 802 case P_HAND_SHAKE_M: 803 if (msock) { 804 dev_warn(DEV, "initial packet M crossed\n"); 805 sock_release(msock); 806 } 807 msock = s; 808 set_bit(DISCARD_CONCURRENT, &mdev->flags); 809 break; 810 default: 811 dev_warn(DEV, "Error receiving initial packet\n"); 812 sock_release(s); 813 if (random32() & 1) 814 goto retry; 815 } 816 } 817 818 if (mdev->state.conn <= C_DISCONNECTING) 819 goto out_release_sockets; 820 if (signal_pending(current)) { 821 flush_signals(current); 822 smp_rmb(); 823 if (get_t_state(&mdev->receiver) == Exiting) 824 goto out_release_sockets; 825 } 826 827 if (sock && msock) { 828 ok = drbd_socket_okay(mdev, &sock); 829 ok = drbd_socket_okay(mdev, &msock) && ok; 830 if (ok) 831 break; 832 } 833 } while (1); 834 835 msock->sk->sk_reuse = 1; /* SO_REUSEADDR */ 836 sock->sk->sk_reuse = 1; /* SO_REUSEADDR */ 837 838 sock->sk->sk_allocation = GFP_NOIO; 839 msock->sk->sk_allocation = GFP_NOIO; 840 841 sock->sk->sk_priority = TC_PRIO_INTERACTIVE_BULK; 842 msock->sk->sk_priority = TC_PRIO_INTERACTIVE; 843 844 if (mdev->net_conf->sndbuf_size) { 845 sock->sk->sk_sndbuf = mdev->net_conf->sndbuf_size; 846 sock->sk->sk_userlocks |= SOCK_SNDBUF_LOCK; 847 } 848 849 if (mdev->net_conf->rcvbuf_size) { 850 sock->sk->sk_rcvbuf = mdev->net_conf->rcvbuf_size; 851 sock->sk->sk_userlocks |= SOCK_RCVBUF_LOCK; 852 } 853 854 /* NOT YET ... 855 * sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10; 856 * sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT; 857 * first set it to the P_HAND_SHAKE timeout, 858 * which we set to 4x the configured ping_timeout. */ 859 sock->sk->sk_sndtimeo = 860 sock->sk->sk_rcvtimeo = mdev->net_conf->ping_timeo*4*HZ/10; 861 862 msock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10; 863 msock->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ; 864 865 /* we don't want delays. 866 * we use TCP_CORK where apropriate, though */ 867 drbd_tcp_nodelay(sock); 868 drbd_tcp_nodelay(msock); 869 870 mdev->data.socket = sock; 871 mdev->meta.socket = msock; 872 mdev->last_received = jiffies; 873 874 D_ASSERT(mdev->asender.task == NULL); 875 876 h = drbd_do_handshake(mdev); 877 if (h <= 0) 878 return h; 879 880 if (mdev->cram_hmac_tfm) { 881 /* drbd_request_state(mdev, NS(conn, WFAuth)); */ 882 if (!drbd_do_auth(mdev)) { 883 dev_err(DEV, "Authentication of peer failed\n"); 884 return -1; 885 } 886 } 887 888 if (drbd_request_state(mdev, NS(conn, C_WF_REPORT_PARAMS)) < SS_SUCCESS) 889 return 0; 890 891 sock->sk->sk_sndtimeo = mdev->net_conf->timeout*HZ/10; 892 sock->sk->sk_rcvtimeo = MAX_SCHEDULE_TIMEOUT; 893 894 atomic_set(&mdev->packet_seq, 0); 895 mdev->peer_seq = 0; 896 897 drbd_thread_start(&mdev->asender); 898 899 drbd_send_protocol(mdev); 900 drbd_send_sync_param(mdev, &mdev->sync_conf); 901 drbd_send_sizes(mdev, 0); 902 drbd_send_uuids(mdev); 903 drbd_send_state(mdev); 904 clear_bit(USE_DEGR_WFC_T, &mdev->flags); 905 clear_bit(RESIZE_PENDING, &mdev->flags); 906 907 return 1; 908 909 out_release_sockets: 910 if (sock) 911 sock_release(sock); 912 if (msock) 913 sock_release(msock); 914 return -1; 915 } 916 917 static int drbd_recv_header(struct drbd_conf *mdev, struct p_header *h) 918 { 919 int r; 920 921 r = drbd_recv(mdev, h, sizeof(*h)); 922 923 if (unlikely(r != sizeof(*h))) { 924 dev_err(DEV, "short read expecting header on sock: r=%d\n", r); 925 return FALSE; 926 }; 927 h->command = be16_to_cpu(h->command); 928 h->length = be16_to_cpu(h->length); 929 if (unlikely(h->magic != BE_DRBD_MAGIC)) { 930 dev_err(DEV, "magic?? on data m: 0x%lx c: %d l: %d\n", 931 (long)be32_to_cpu(h->magic), 932 h->command, h->length); 933 return FALSE; 934 } 935 mdev->last_received = jiffies; 936 937 return TRUE; 938 } 939 940 static enum finish_epoch drbd_flush_after_epoch(struct drbd_conf *mdev, struct drbd_epoch *epoch) 941 { 942 int rv; 943 944 if (mdev->write_ordering >= WO_bdev_flush && get_ldev(mdev)) { 945 rv = blkdev_issue_flush(mdev->ldev->backing_bdev, NULL); 946 if (rv) { 947 dev_err(DEV, "local disk flush failed with status %d\n", rv); 948 /* would rather check on EOPNOTSUPP, but that is not reliable. 949 * don't try again for ANY return value != 0 950 * if (rv == -EOPNOTSUPP) */ 951 drbd_bump_write_ordering(mdev, WO_drain_io); 952 } 953 put_ldev(mdev); 954 } 955 956 return drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE); 957 } 958 959 static int w_flush(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 960 { 961 struct flush_work *fw = (struct flush_work *)w; 962 struct drbd_epoch *epoch = fw->epoch; 963 964 kfree(w); 965 966 if (!test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags)) 967 drbd_flush_after_epoch(mdev, epoch); 968 969 drbd_may_finish_epoch(mdev, epoch, EV_PUT | 970 (mdev->state.conn < C_CONNECTED ? EV_CLEANUP : 0)); 971 972 return 1; 973 } 974 975 /** 976 * drbd_may_finish_epoch() - Applies an epoch_event to the epoch's state, eventually finishes it. 977 * @mdev: DRBD device. 978 * @epoch: Epoch object. 979 * @ev: Epoch event. 980 */ 981 static enum finish_epoch drbd_may_finish_epoch(struct drbd_conf *mdev, 982 struct drbd_epoch *epoch, 983 enum epoch_event ev) 984 { 985 int finish, epoch_size; 986 struct drbd_epoch *next_epoch; 987 int schedule_flush = 0; 988 enum finish_epoch rv = FE_STILL_LIVE; 989 990 spin_lock(&mdev->epoch_lock); 991 do { 992 next_epoch = NULL; 993 finish = 0; 994 995 epoch_size = atomic_read(&epoch->epoch_size); 996 997 switch (ev & ~EV_CLEANUP) { 998 case EV_PUT: 999 atomic_dec(&epoch->active); 1000 break; 1001 case EV_GOT_BARRIER_NR: 1002 set_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags); 1003 1004 /* Special case: If we just switched from WO_bio_barrier to 1005 WO_bdev_flush we should not finish the current epoch */ 1006 if (test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags) && epoch_size == 1 && 1007 mdev->write_ordering != WO_bio_barrier && 1008 epoch == mdev->current_epoch) 1009 clear_bit(DE_CONTAINS_A_BARRIER, &epoch->flags); 1010 break; 1011 case EV_BARRIER_DONE: 1012 set_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE, &epoch->flags); 1013 break; 1014 case EV_BECAME_LAST: 1015 /* nothing to do*/ 1016 break; 1017 } 1018 1019 if (epoch_size != 0 && 1020 atomic_read(&epoch->active) == 0 && 1021 test_bit(DE_HAVE_BARRIER_NUMBER, &epoch->flags) && 1022 epoch->list.prev == &mdev->current_epoch->list && 1023 !test_bit(DE_IS_FINISHING, &epoch->flags)) { 1024 /* Nearly all conditions are met to finish that epoch... */ 1025 if (test_bit(DE_BARRIER_IN_NEXT_EPOCH_DONE, &epoch->flags) || 1026 mdev->write_ordering == WO_none || 1027 (epoch_size == 1 && test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags)) || 1028 ev & EV_CLEANUP) { 1029 finish = 1; 1030 set_bit(DE_IS_FINISHING, &epoch->flags); 1031 } else if (!test_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags) && 1032 mdev->write_ordering == WO_bio_barrier) { 1033 atomic_inc(&epoch->active); 1034 schedule_flush = 1; 1035 } 1036 } 1037 if (finish) { 1038 if (!(ev & EV_CLEANUP)) { 1039 spin_unlock(&mdev->epoch_lock); 1040 drbd_send_b_ack(mdev, epoch->barrier_nr, epoch_size); 1041 spin_lock(&mdev->epoch_lock); 1042 } 1043 dec_unacked(mdev); 1044 1045 if (mdev->current_epoch != epoch) { 1046 next_epoch = list_entry(epoch->list.next, struct drbd_epoch, list); 1047 list_del(&epoch->list); 1048 ev = EV_BECAME_LAST | (ev & EV_CLEANUP); 1049 mdev->epochs--; 1050 kfree(epoch); 1051 1052 if (rv == FE_STILL_LIVE) 1053 rv = FE_DESTROYED; 1054 } else { 1055 epoch->flags = 0; 1056 atomic_set(&epoch->epoch_size, 0); 1057 /* atomic_set(&epoch->active, 0); is alrady zero */ 1058 if (rv == FE_STILL_LIVE) 1059 rv = FE_RECYCLED; 1060 } 1061 } 1062 1063 if (!next_epoch) 1064 break; 1065 1066 epoch = next_epoch; 1067 } while (1); 1068 1069 spin_unlock(&mdev->epoch_lock); 1070 1071 if (schedule_flush) { 1072 struct flush_work *fw; 1073 fw = kmalloc(sizeof(*fw), GFP_ATOMIC); 1074 if (fw) { 1075 fw->w.cb = w_flush; 1076 fw->epoch = epoch; 1077 drbd_queue_work(&mdev->data.work, &fw->w); 1078 } else { 1079 dev_warn(DEV, "Could not kmalloc a flush_work obj\n"); 1080 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags); 1081 /* That is not a recursion, only one level */ 1082 drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE); 1083 drbd_may_finish_epoch(mdev, epoch, EV_PUT); 1084 } 1085 } 1086 1087 return rv; 1088 } 1089 1090 /** 1091 * drbd_bump_write_ordering() - Fall back to an other write ordering method 1092 * @mdev: DRBD device. 1093 * @wo: Write ordering method to try. 1094 */ 1095 void drbd_bump_write_ordering(struct drbd_conf *mdev, enum write_ordering_e wo) __must_hold(local) 1096 { 1097 enum write_ordering_e pwo; 1098 static char *write_ordering_str[] = { 1099 [WO_none] = "none", 1100 [WO_drain_io] = "drain", 1101 [WO_bdev_flush] = "flush", 1102 [WO_bio_barrier] = "barrier", 1103 }; 1104 1105 pwo = mdev->write_ordering; 1106 wo = min(pwo, wo); 1107 if (wo == WO_bio_barrier && mdev->ldev->dc.no_disk_barrier) 1108 wo = WO_bdev_flush; 1109 if (wo == WO_bdev_flush && mdev->ldev->dc.no_disk_flush) 1110 wo = WO_drain_io; 1111 if (wo == WO_drain_io && mdev->ldev->dc.no_disk_drain) 1112 wo = WO_none; 1113 mdev->write_ordering = wo; 1114 if (pwo != mdev->write_ordering || wo == WO_bio_barrier) 1115 dev_info(DEV, "Method to ensure write ordering: %s\n", write_ordering_str[mdev->write_ordering]); 1116 } 1117 1118 /** 1119 * w_e_reissue() - Worker callback; Resubmit a bio, without BIO_RW_BARRIER set 1120 * @mdev: DRBD device. 1121 * @w: work object. 1122 * @cancel: The connection will be closed anyways (unused in this callback) 1123 */ 1124 int w_e_reissue(struct drbd_conf *mdev, struct drbd_work *w, int cancel) __releases(local) 1125 { 1126 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w; 1127 struct bio *bio = e->private_bio; 1128 1129 /* We leave DE_CONTAINS_A_BARRIER and EE_IS_BARRIER in place, 1130 (and DE_BARRIER_IN_NEXT_EPOCH_ISSUED in the previous Epoch) 1131 so that we can finish that epoch in drbd_may_finish_epoch(). 1132 That is necessary if we already have a long chain of Epochs, before 1133 we realize that BIO_RW_BARRIER is actually not supported */ 1134 1135 /* As long as the -ENOTSUPP on the barrier is reported immediately 1136 that will never trigger. If it is reported late, we will just 1137 print that warning and continue correctly for all future requests 1138 with WO_bdev_flush */ 1139 if (previous_epoch(mdev, e->epoch)) 1140 dev_warn(DEV, "Write ordering was not enforced (one time event)\n"); 1141 1142 /* prepare bio for re-submit, 1143 * re-init volatile members */ 1144 /* we still have a local reference, 1145 * get_ldev was done in receive_Data. */ 1146 bio->bi_bdev = mdev->ldev->backing_bdev; 1147 bio->bi_sector = e->sector; 1148 bio->bi_size = e->size; 1149 bio->bi_idx = 0; 1150 1151 bio->bi_flags &= ~(BIO_POOL_MASK - 1); 1152 bio->bi_flags |= 1 << BIO_UPTODATE; 1153 1154 /* don't know whether this is necessary: */ 1155 bio->bi_phys_segments = 0; 1156 bio->bi_next = NULL; 1157 1158 /* these should be unchanged: */ 1159 /* bio->bi_end_io = drbd_endio_write_sec; */ 1160 /* bio->bi_vcnt = whatever; */ 1161 1162 e->w.cb = e_end_block; 1163 1164 /* This is no longer a barrier request. */ 1165 bio->bi_rw &= ~(1UL << BIO_RW_BARRIER); 1166 1167 drbd_generic_make_request(mdev, DRBD_FAULT_DT_WR, bio); 1168 1169 return 1; 1170 } 1171 1172 static int receive_Barrier(struct drbd_conf *mdev, struct p_header *h) 1173 { 1174 int rv, issue_flush; 1175 struct p_barrier *p = (struct p_barrier *)h; 1176 struct drbd_epoch *epoch; 1177 1178 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE; 1179 1180 rv = drbd_recv(mdev, h->payload, h->length); 1181 ERR_IF(rv != h->length) return FALSE; 1182 1183 inc_unacked(mdev); 1184 1185 if (mdev->net_conf->wire_protocol != DRBD_PROT_C) 1186 drbd_kick_lo(mdev); 1187 1188 mdev->current_epoch->barrier_nr = p->barrier; 1189 rv = drbd_may_finish_epoch(mdev, mdev->current_epoch, EV_GOT_BARRIER_NR); 1190 1191 /* P_BARRIER_ACK may imply that the corresponding extent is dropped from 1192 * the activity log, which means it would not be resynced in case the 1193 * R_PRIMARY crashes now. 1194 * Therefore we must send the barrier_ack after the barrier request was 1195 * completed. */ 1196 switch (mdev->write_ordering) { 1197 case WO_bio_barrier: 1198 case WO_none: 1199 if (rv == FE_RECYCLED) 1200 return TRUE; 1201 break; 1202 1203 case WO_bdev_flush: 1204 case WO_drain_io: 1205 D_ASSERT(rv == FE_STILL_LIVE); 1206 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &mdev->current_epoch->flags); 1207 drbd_wait_ee_list_empty(mdev, &mdev->active_ee); 1208 rv = drbd_flush_after_epoch(mdev, mdev->current_epoch); 1209 if (rv == FE_RECYCLED) 1210 return TRUE; 1211 1212 /* The asender will send all the ACKs and barrier ACKs out, since 1213 all EEs moved from the active_ee to the done_ee. We need to 1214 provide a new epoch object for the EEs that come in soon */ 1215 break; 1216 } 1217 1218 /* receiver context, in the writeout path of the other node. 1219 * avoid potential distributed deadlock */ 1220 epoch = kmalloc(sizeof(struct drbd_epoch), GFP_NOIO); 1221 if (!epoch) { 1222 dev_warn(DEV, "Allocation of an epoch failed, slowing down\n"); 1223 issue_flush = !test_and_set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags); 1224 drbd_wait_ee_list_empty(mdev, &mdev->active_ee); 1225 if (issue_flush) { 1226 rv = drbd_flush_after_epoch(mdev, mdev->current_epoch); 1227 if (rv == FE_RECYCLED) 1228 return TRUE; 1229 } 1230 1231 drbd_wait_ee_list_empty(mdev, &mdev->done_ee); 1232 1233 return TRUE; 1234 } 1235 1236 epoch->flags = 0; 1237 atomic_set(&epoch->epoch_size, 0); 1238 atomic_set(&epoch->active, 0); 1239 1240 spin_lock(&mdev->epoch_lock); 1241 if (atomic_read(&mdev->current_epoch->epoch_size)) { 1242 list_add(&epoch->list, &mdev->current_epoch->list); 1243 mdev->current_epoch = epoch; 1244 mdev->epochs++; 1245 } else { 1246 /* The current_epoch got recycled while we allocated this one... */ 1247 kfree(epoch); 1248 } 1249 spin_unlock(&mdev->epoch_lock); 1250 1251 return TRUE; 1252 } 1253 1254 /* used from receive_RSDataReply (recv_resync_read) 1255 * and from receive_Data */ 1256 static struct drbd_epoch_entry * 1257 read_in_block(struct drbd_conf *mdev, u64 id, sector_t sector, int data_size) __must_hold(local) 1258 { 1259 struct drbd_epoch_entry *e; 1260 struct bio_vec *bvec; 1261 struct page *page; 1262 struct bio *bio; 1263 int dgs, ds, i, rr; 1264 void *dig_in = mdev->int_dig_in; 1265 void *dig_vv = mdev->int_dig_vv; 1266 1267 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ? 1268 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0; 1269 1270 if (dgs) { 1271 rr = drbd_recv(mdev, dig_in, dgs); 1272 if (rr != dgs) { 1273 dev_warn(DEV, "short read receiving data digest: read %d expected %d\n", 1274 rr, dgs); 1275 return NULL; 1276 } 1277 } 1278 1279 data_size -= dgs; 1280 1281 ERR_IF(data_size & 0x1ff) return NULL; 1282 ERR_IF(data_size > DRBD_MAX_SEGMENT_SIZE) return NULL; 1283 1284 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD 1285 * "criss-cross" setup, that might cause write-out on some other DRBD, 1286 * which in turn might block on the other node at this very place. */ 1287 e = drbd_alloc_ee(mdev, id, sector, data_size, GFP_NOIO); 1288 if (!e) 1289 return NULL; 1290 bio = e->private_bio; 1291 ds = data_size; 1292 bio_for_each_segment(bvec, bio, i) { 1293 page = bvec->bv_page; 1294 rr = drbd_recv(mdev, kmap(page), min_t(int, ds, PAGE_SIZE)); 1295 kunmap(page); 1296 if (rr != min_t(int, ds, PAGE_SIZE)) { 1297 drbd_free_ee(mdev, e); 1298 dev_warn(DEV, "short read receiving data: read %d expected %d\n", 1299 rr, min_t(int, ds, PAGE_SIZE)); 1300 return NULL; 1301 } 1302 ds -= rr; 1303 } 1304 1305 if (dgs) { 1306 drbd_csum(mdev, mdev->integrity_r_tfm, bio, dig_vv); 1307 if (memcmp(dig_in, dig_vv, dgs)) { 1308 dev_err(DEV, "Digest integrity check FAILED.\n"); 1309 drbd_bcast_ee(mdev, "digest failed", 1310 dgs, dig_in, dig_vv, e); 1311 drbd_free_ee(mdev, e); 1312 return NULL; 1313 } 1314 } 1315 mdev->recv_cnt += data_size>>9; 1316 return e; 1317 } 1318 1319 /* drbd_drain_block() just takes a data block 1320 * out of the socket input buffer, and discards it. 1321 */ 1322 static int drbd_drain_block(struct drbd_conf *mdev, int data_size) 1323 { 1324 struct page *page; 1325 int rr, rv = 1; 1326 void *data; 1327 1328 page = drbd_pp_alloc(mdev, 1); 1329 1330 data = kmap(page); 1331 while (data_size) { 1332 rr = drbd_recv(mdev, data, min_t(int, data_size, PAGE_SIZE)); 1333 if (rr != min_t(int, data_size, PAGE_SIZE)) { 1334 rv = 0; 1335 dev_warn(DEV, "short read receiving data: read %d expected %d\n", 1336 rr, min_t(int, data_size, PAGE_SIZE)); 1337 break; 1338 } 1339 data_size -= rr; 1340 } 1341 kunmap(page); 1342 drbd_pp_free(mdev, page); 1343 return rv; 1344 } 1345 1346 static int recv_dless_read(struct drbd_conf *mdev, struct drbd_request *req, 1347 sector_t sector, int data_size) 1348 { 1349 struct bio_vec *bvec; 1350 struct bio *bio; 1351 int dgs, rr, i, expect; 1352 void *dig_in = mdev->int_dig_in; 1353 void *dig_vv = mdev->int_dig_vv; 1354 1355 dgs = (mdev->agreed_pro_version >= 87 && mdev->integrity_r_tfm) ? 1356 crypto_hash_digestsize(mdev->integrity_r_tfm) : 0; 1357 1358 if (dgs) { 1359 rr = drbd_recv(mdev, dig_in, dgs); 1360 if (rr != dgs) { 1361 dev_warn(DEV, "short read receiving data reply digest: read %d expected %d\n", 1362 rr, dgs); 1363 return 0; 1364 } 1365 } 1366 1367 data_size -= dgs; 1368 1369 /* optimistically update recv_cnt. if receiving fails below, 1370 * we disconnect anyways, and counters will be reset. */ 1371 mdev->recv_cnt += data_size>>9; 1372 1373 bio = req->master_bio; 1374 D_ASSERT(sector == bio->bi_sector); 1375 1376 bio_for_each_segment(bvec, bio, i) { 1377 expect = min_t(int, data_size, bvec->bv_len); 1378 rr = drbd_recv(mdev, 1379 kmap(bvec->bv_page)+bvec->bv_offset, 1380 expect); 1381 kunmap(bvec->bv_page); 1382 if (rr != expect) { 1383 dev_warn(DEV, "short read receiving data reply: " 1384 "read %d expected %d\n", 1385 rr, expect); 1386 return 0; 1387 } 1388 data_size -= rr; 1389 } 1390 1391 if (dgs) { 1392 drbd_csum(mdev, mdev->integrity_r_tfm, bio, dig_vv); 1393 if (memcmp(dig_in, dig_vv, dgs)) { 1394 dev_err(DEV, "Digest integrity check FAILED. Broken NICs?\n"); 1395 return 0; 1396 } 1397 } 1398 1399 D_ASSERT(data_size == 0); 1400 return 1; 1401 } 1402 1403 /* e_end_resync_block() is called via 1404 * drbd_process_done_ee() by asender only */ 1405 static int e_end_resync_block(struct drbd_conf *mdev, struct drbd_work *w, int unused) 1406 { 1407 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w; 1408 sector_t sector = e->sector; 1409 int ok; 1410 1411 D_ASSERT(hlist_unhashed(&e->colision)); 1412 1413 if (likely(drbd_bio_uptodate(e->private_bio))) { 1414 drbd_set_in_sync(mdev, sector, e->size); 1415 ok = drbd_send_ack(mdev, P_RS_WRITE_ACK, e); 1416 } else { 1417 /* Record failure to sync */ 1418 drbd_rs_failed_io(mdev, sector, e->size); 1419 1420 ok = drbd_send_ack(mdev, P_NEG_ACK, e); 1421 } 1422 dec_unacked(mdev); 1423 1424 return ok; 1425 } 1426 1427 static int recv_resync_read(struct drbd_conf *mdev, sector_t sector, int data_size) __releases(local) 1428 { 1429 struct drbd_epoch_entry *e; 1430 1431 e = read_in_block(mdev, ID_SYNCER, sector, data_size); 1432 if (!e) { 1433 put_ldev(mdev); 1434 return FALSE; 1435 } 1436 1437 dec_rs_pending(mdev); 1438 1439 e->private_bio->bi_end_io = drbd_endio_write_sec; 1440 e->private_bio->bi_rw = WRITE; 1441 e->w.cb = e_end_resync_block; 1442 1443 inc_unacked(mdev); 1444 /* corresponding dec_unacked() in e_end_resync_block() 1445 * respective _drbd_clear_done_ee */ 1446 1447 spin_lock_irq(&mdev->req_lock); 1448 list_add(&e->w.list, &mdev->sync_ee); 1449 spin_unlock_irq(&mdev->req_lock); 1450 1451 drbd_generic_make_request(mdev, DRBD_FAULT_RS_WR, e->private_bio); 1452 /* accounting done in endio */ 1453 1454 maybe_kick_lo(mdev); 1455 return TRUE; 1456 } 1457 1458 static int receive_DataReply(struct drbd_conf *mdev, struct p_header *h) 1459 { 1460 struct drbd_request *req; 1461 sector_t sector; 1462 unsigned int header_size, data_size; 1463 int ok; 1464 struct p_data *p = (struct p_data *)h; 1465 1466 header_size = sizeof(*p) - sizeof(*h); 1467 data_size = h->length - header_size; 1468 1469 ERR_IF(data_size == 0) return FALSE; 1470 1471 if (drbd_recv(mdev, h->payload, header_size) != header_size) 1472 return FALSE; 1473 1474 sector = be64_to_cpu(p->sector); 1475 1476 spin_lock_irq(&mdev->req_lock); 1477 req = _ar_id_to_req(mdev, p->block_id, sector); 1478 spin_unlock_irq(&mdev->req_lock); 1479 if (unlikely(!req)) { 1480 dev_err(DEV, "Got a corrupt block_id/sector pair(1).\n"); 1481 return FALSE; 1482 } 1483 1484 /* hlist_del(&req->colision) is done in _req_may_be_done, to avoid 1485 * special casing it there for the various failure cases. 1486 * still no race with drbd_fail_pending_reads */ 1487 ok = recv_dless_read(mdev, req, sector, data_size); 1488 1489 if (ok) 1490 req_mod(req, data_received); 1491 /* else: nothing. handled from drbd_disconnect... 1492 * I don't think we may complete this just yet 1493 * in case we are "on-disconnect: freeze" */ 1494 1495 return ok; 1496 } 1497 1498 static int receive_RSDataReply(struct drbd_conf *mdev, struct p_header *h) 1499 { 1500 sector_t sector; 1501 unsigned int header_size, data_size; 1502 int ok; 1503 struct p_data *p = (struct p_data *)h; 1504 1505 header_size = sizeof(*p) - sizeof(*h); 1506 data_size = h->length - header_size; 1507 1508 ERR_IF(data_size == 0) return FALSE; 1509 1510 if (drbd_recv(mdev, h->payload, header_size) != header_size) 1511 return FALSE; 1512 1513 sector = be64_to_cpu(p->sector); 1514 D_ASSERT(p->block_id == ID_SYNCER); 1515 1516 if (get_ldev(mdev)) { 1517 /* data is submitted to disk within recv_resync_read. 1518 * corresponding put_ldev done below on error, 1519 * or in drbd_endio_write_sec. */ 1520 ok = recv_resync_read(mdev, sector, data_size); 1521 } else { 1522 if (__ratelimit(&drbd_ratelimit_state)) 1523 dev_err(DEV, "Can not write resync data to local disk.\n"); 1524 1525 ok = drbd_drain_block(mdev, data_size); 1526 1527 drbd_send_ack_dp(mdev, P_NEG_ACK, p); 1528 } 1529 1530 return ok; 1531 } 1532 1533 /* e_end_block() is called via drbd_process_done_ee(). 1534 * this means this function only runs in the asender thread 1535 */ 1536 static int e_end_block(struct drbd_conf *mdev, struct drbd_work *w, int cancel) 1537 { 1538 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w; 1539 sector_t sector = e->sector; 1540 struct drbd_epoch *epoch; 1541 int ok = 1, pcmd; 1542 1543 if (e->flags & EE_IS_BARRIER) { 1544 epoch = previous_epoch(mdev, e->epoch); 1545 if (epoch) 1546 drbd_may_finish_epoch(mdev, epoch, EV_BARRIER_DONE + (cancel ? EV_CLEANUP : 0)); 1547 } 1548 1549 if (mdev->net_conf->wire_protocol == DRBD_PROT_C) { 1550 if (likely(drbd_bio_uptodate(e->private_bio))) { 1551 pcmd = (mdev->state.conn >= C_SYNC_SOURCE && 1552 mdev->state.conn <= C_PAUSED_SYNC_T && 1553 e->flags & EE_MAY_SET_IN_SYNC) ? 1554 P_RS_WRITE_ACK : P_WRITE_ACK; 1555 ok &= drbd_send_ack(mdev, pcmd, e); 1556 if (pcmd == P_RS_WRITE_ACK) 1557 drbd_set_in_sync(mdev, sector, e->size); 1558 } else { 1559 ok = drbd_send_ack(mdev, P_NEG_ACK, e); 1560 /* we expect it to be marked out of sync anyways... 1561 * maybe assert this? */ 1562 } 1563 dec_unacked(mdev); 1564 } 1565 /* we delete from the conflict detection hash _after_ we sent out the 1566 * P_WRITE_ACK / P_NEG_ACK, to get the sequence number right. */ 1567 if (mdev->net_conf->two_primaries) { 1568 spin_lock_irq(&mdev->req_lock); 1569 D_ASSERT(!hlist_unhashed(&e->colision)); 1570 hlist_del_init(&e->colision); 1571 spin_unlock_irq(&mdev->req_lock); 1572 } else { 1573 D_ASSERT(hlist_unhashed(&e->colision)); 1574 } 1575 1576 drbd_may_finish_epoch(mdev, e->epoch, EV_PUT + (cancel ? EV_CLEANUP : 0)); 1577 1578 return ok; 1579 } 1580 1581 static int e_send_discard_ack(struct drbd_conf *mdev, struct drbd_work *w, int unused) 1582 { 1583 struct drbd_epoch_entry *e = (struct drbd_epoch_entry *)w; 1584 int ok = 1; 1585 1586 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C); 1587 ok = drbd_send_ack(mdev, P_DISCARD_ACK, e); 1588 1589 spin_lock_irq(&mdev->req_lock); 1590 D_ASSERT(!hlist_unhashed(&e->colision)); 1591 hlist_del_init(&e->colision); 1592 spin_unlock_irq(&mdev->req_lock); 1593 1594 dec_unacked(mdev); 1595 1596 return ok; 1597 } 1598 1599 /* Called from receive_Data. 1600 * Synchronize packets on sock with packets on msock. 1601 * 1602 * This is here so even when a P_DATA packet traveling via sock overtook an Ack 1603 * packet traveling on msock, they are still processed in the order they have 1604 * been sent. 1605 * 1606 * Note: we don't care for Ack packets overtaking P_DATA packets. 1607 * 1608 * In case packet_seq is larger than mdev->peer_seq number, there are 1609 * outstanding packets on the msock. We wait for them to arrive. 1610 * In case we are the logically next packet, we update mdev->peer_seq 1611 * ourselves. Correctly handles 32bit wrap around. 1612 * 1613 * Assume we have a 10 GBit connection, that is about 1<<30 byte per second, 1614 * about 1<<21 sectors per second. So "worst" case, we have 1<<3 == 8 seconds 1615 * for the 24bit wrap (historical atomic_t guarantee on some archs), and we have 1616 * 1<<9 == 512 seconds aka ages for the 32bit wrap around... 1617 * 1618 * returns 0 if we may process the packet, 1619 * -ERESTARTSYS if we were interrupted (by disconnect signal). */ 1620 static int drbd_wait_peer_seq(struct drbd_conf *mdev, const u32 packet_seq) 1621 { 1622 DEFINE_WAIT(wait); 1623 unsigned int p_seq; 1624 long timeout; 1625 int ret = 0; 1626 spin_lock(&mdev->peer_seq_lock); 1627 for (;;) { 1628 prepare_to_wait(&mdev->seq_wait, &wait, TASK_INTERRUPTIBLE); 1629 if (seq_le(packet_seq, mdev->peer_seq+1)) 1630 break; 1631 if (signal_pending(current)) { 1632 ret = -ERESTARTSYS; 1633 break; 1634 } 1635 p_seq = mdev->peer_seq; 1636 spin_unlock(&mdev->peer_seq_lock); 1637 timeout = schedule_timeout(30*HZ); 1638 spin_lock(&mdev->peer_seq_lock); 1639 if (timeout == 0 && p_seq == mdev->peer_seq) { 1640 ret = -ETIMEDOUT; 1641 dev_err(DEV, "ASSERT FAILED waited 30 seconds for sequence update, forcing reconnect\n"); 1642 break; 1643 } 1644 } 1645 finish_wait(&mdev->seq_wait, &wait); 1646 if (mdev->peer_seq+1 == packet_seq) 1647 mdev->peer_seq++; 1648 spin_unlock(&mdev->peer_seq_lock); 1649 return ret; 1650 } 1651 1652 /* mirrored write */ 1653 static int receive_Data(struct drbd_conf *mdev, struct p_header *h) 1654 { 1655 sector_t sector; 1656 struct drbd_epoch_entry *e; 1657 struct p_data *p = (struct p_data *)h; 1658 int header_size, data_size; 1659 int rw = WRITE; 1660 u32 dp_flags; 1661 1662 header_size = sizeof(*p) - sizeof(*h); 1663 data_size = h->length - header_size; 1664 1665 ERR_IF(data_size == 0) return FALSE; 1666 1667 if (drbd_recv(mdev, h->payload, header_size) != header_size) 1668 return FALSE; 1669 1670 if (!get_ldev(mdev)) { 1671 if (__ratelimit(&drbd_ratelimit_state)) 1672 dev_err(DEV, "Can not write mirrored data block " 1673 "to local disk.\n"); 1674 spin_lock(&mdev->peer_seq_lock); 1675 if (mdev->peer_seq+1 == be32_to_cpu(p->seq_num)) 1676 mdev->peer_seq++; 1677 spin_unlock(&mdev->peer_seq_lock); 1678 1679 drbd_send_ack_dp(mdev, P_NEG_ACK, p); 1680 atomic_inc(&mdev->current_epoch->epoch_size); 1681 return drbd_drain_block(mdev, data_size); 1682 } 1683 1684 /* get_ldev(mdev) successful. 1685 * Corresponding put_ldev done either below (on various errors), 1686 * or in drbd_endio_write_sec, if we successfully submit the data at 1687 * the end of this function. */ 1688 1689 sector = be64_to_cpu(p->sector); 1690 e = read_in_block(mdev, p->block_id, sector, data_size); 1691 if (!e) { 1692 put_ldev(mdev); 1693 return FALSE; 1694 } 1695 1696 e->private_bio->bi_end_io = drbd_endio_write_sec; 1697 e->w.cb = e_end_block; 1698 1699 spin_lock(&mdev->epoch_lock); 1700 e->epoch = mdev->current_epoch; 1701 atomic_inc(&e->epoch->epoch_size); 1702 atomic_inc(&e->epoch->active); 1703 1704 if (mdev->write_ordering == WO_bio_barrier && atomic_read(&e->epoch->epoch_size) == 1) { 1705 struct drbd_epoch *epoch; 1706 /* Issue a barrier if we start a new epoch, and the previous epoch 1707 was not a epoch containing a single request which already was 1708 a Barrier. */ 1709 epoch = list_entry(e->epoch->list.prev, struct drbd_epoch, list); 1710 if (epoch == e->epoch) { 1711 set_bit(DE_CONTAINS_A_BARRIER, &e->epoch->flags); 1712 rw |= (1<<BIO_RW_BARRIER); 1713 e->flags |= EE_IS_BARRIER; 1714 } else { 1715 if (atomic_read(&epoch->epoch_size) > 1 || 1716 !test_bit(DE_CONTAINS_A_BARRIER, &epoch->flags)) { 1717 set_bit(DE_BARRIER_IN_NEXT_EPOCH_ISSUED, &epoch->flags); 1718 set_bit(DE_CONTAINS_A_BARRIER, &e->epoch->flags); 1719 rw |= (1<<BIO_RW_BARRIER); 1720 e->flags |= EE_IS_BARRIER; 1721 } 1722 } 1723 } 1724 spin_unlock(&mdev->epoch_lock); 1725 1726 dp_flags = be32_to_cpu(p->dp_flags); 1727 if (dp_flags & DP_HARDBARRIER) { 1728 dev_err(DEV, "ASSERT FAILED would have submitted barrier request\n"); 1729 /* rw |= (1<<BIO_RW_BARRIER); */ 1730 } 1731 if (dp_flags & DP_RW_SYNC) 1732 rw |= (1<<BIO_RW_SYNCIO) | (1<<BIO_RW_UNPLUG); 1733 if (dp_flags & DP_MAY_SET_IN_SYNC) 1734 e->flags |= EE_MAY_SET_IN_SYNC; 1735 1736 /* I'm the receiver, I do hold a net_cnt reference. */ 1737 if (!mdev->net_conf->two_primaries) { 1738 spin_lock_irq(&mdev->req_lock); 1739 } else { 1740 /* don't get the req_lock yet, 1741 * we may sleep in drbd_wait_peer_seq */ 1742 const int size = e->size; 1743 const int discard = test_bit(DISCARD_CONCURRENT, &mdev->flags); 1744 DEFINE_WAIT(wait); 1745 struct drbd_request *i; 1746 struct hlist_node *n; 1747 struct hlist_head *slot; 1748 int first; 1749 1750 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C); 1751 BUG_ON(mdev->ee_hash == NULL); 1752 BUG_ON(mdev->tl_hash == NULL); 1753 1754 /* conflict detection and handling: 1755 * 1. wait on the sequence number, 1756 * in case this data packet overtook ACK packets. 1757 * 2. check our hash tables for conflicting requests. 1758 * we only need to walk the tl_hash, since an ee can not 1759 * have a conflict with an other ee: on the submitting 1760 * node, the corresponding req had already been conflicting, 1761 * and a conflicting req is never sent. 1762 * 1763 * Note: for two_primaries, we are protocol C, 1764 * so there cannot be any request that is DONE 1765 * but still on the transfer log. 1766 * 1767 * unconditionally add to the ee_hash. 1768 * 1769 * if no conflicting request is found: 1770 * submit. 1771 * 1772 * if any conflicting request is found 1773 * that has not yet been acked, 1774 * AND I have the "discard concurrent writes" flag: 1775 * queue (via done_ee) the P_DISCARD_ACK; OUT. 1776 * 1777 * if any conflicting request is found: 1778 * block the receiver, waiting on misc_wait 1779 * until no more conflicting requests are there, 1780 * or we get interrupted (disconnect). 1781 * 1782 * we do not just write after local io completion of those 1783 * requests, but only after req is done completely, i.e. 1784 * we wait for the P_DISCARD_ACK to arrive! 1785 * 1786 * then proceed normally, i.e. submit. 1787 */ 1788 if (drbd_wait_peer_seq(mdev, be32_to_cpu(p->seq_num))) 1789 goto out_interrupted; 1790 1791 spin_lock_irq(&mdev->req_lock); 1792 1793 hlist_add_head(&e->colision, ee_hash_slot(mdev, sector)); 1794 1795 #define OVERLAPS overlaps(i->sector, i->size, sector, size) 1796 slot = tl_hash_slot(mdev, sector); 1797 first = 1; 1798 for (;;) { 1799 int have_unacked = 0; 1800 int have_conflict = 0; 1801 prepare_to_wait(&mdev->misc_wait, &wait, 1802 TASK_INTERRUPTIBLE); 1803 hlist_for_each_entry(i, n, slot, colision) { 1804 if (OVERLAPS) { 1805 /* only ALERT on first iteration, 1806 * we may be woken up early... */ 1807 if (first) 1808 dev_alert(DEV, "%s[%u] Concurrent local write detected!" 1809 " new: %llus +%u; pending: %llus +%u\n", 1810 current->comm, current->pid, 1811 (unsigned long long)sector, size, 1812 (unsigned long long)i->sector, i->size); 1813 if (i->rq_state & RQ_NET_PENDING) 1814 ++have_unacked; 1815 ++have_conflict; 1816 } 1817 } 1818 #undef OVERLAPS 1819 if (!have_conflict) 1820 break; 1821 1822 /* Discard Ack only for the _first_ iteration */ 1823 if (first && discard && have_unacked) { 1824 dev_alert(DEV, "Concurrent write! [DISCARD BY FLAG] sec=%llus\n", 1825 (unsigned long long)sector); 1826 inc_unacked(mdev); 1827 e->w.cb = e_send_discard_ack; 1828 list_add_tail(&e->w.list, &mdev->done_ee); 1829 1830 spin_unlock_irq(&mdev->req_lock); 1831 1832 /* we could probably send that P_DISCARD_ACK ourselves, 1833 * but I don't like the receiver using the msock */ 1834 1835 put_ldev(mdev); 1836 wake_asender(mdev); 1837 finish_wait(&mdev->misc_wait, &wait); 1838 return TRUE; 1839 } 1840 1841 if (signal_pending(current)) { 1842 hlist_del_init(&e->colision); 1843 1844 spin_unlock_irq(&mdev->req_lock); 1845 1846 finish_wait(&mdev->misc_wait, &wait); 1847 goto out_interrupted; 1848 } 1849 1850 spin_unlock_irq(&mdev->req_lock); 1851 if (first) { 1852 first = 0; 1853 dev_alert(DEV, "Concurrent write! [W AFTERWARDS] " 1854 "sec=%llus\n", (unsigned long long)sector); 1855 } else if (discard) { 1856 /* we had none on the first iteration. 1857 * there must be none now. */ 1858 D_ASSERT(have_unacked == 0); 1859 } 1860 schedule(); 1861 spin_lock_irq(&mdev->req_lock); 1862 } 1863 finish_wait(&mdev->misc_wait, &wait); 1864 } 1865 1866 list_add(&e->w.list, &mdev->active_ee); 1867 spin_unlock_irq(&mdev->req_lock); 1868 1869 switch (mdev->net_conf->wire_protocol) { 1870 case DRBD_PROT_C: 1871 inc_unacked(mdev); 1872 /* corresponding dec_unacked() in e_end_block() 1873 * respective _drbd_clear_done_ee */ 1874 break; 1875 case DRBD_PROT_B: 1876 /* I really don't like it that the receiver thread 1877 * sends on the msock, but anyways */ 1878 drbd_send_ack(mdev, P_RECV_ACK, e); 1879 break; 1880 case DRBD_PROT_A: 1881 /* nothing to do */ 1882 break; 1883 } 1884 1885 if (mdev->state.pdsk == D_DISKLESS) { 1886 /* In case we have the only disk of the cluster, */ 1887 drbd_set_out_of_sync(mdev, e->sector, e->size); 1888 e->flags |= EE_CALL_AL_COMPLETE_IO; 1889 drbd_al_begin_io(mdev, e->sector); 1890 } 1891 1892 e->private_bio->bi_rw = rw; 1893 drbd_generic_make_request(mdev, DRBD_FAULT_DT_WR, e->private_bio); 1894 /* accounting done in endio */ 1895 1896 maybe_kick_lo(mdev); 1897 return TRUE; 1898 1899 out_interrupted: 1900 /* yes, the epoch_size now is imbalanced. 1901 * but we drop the connection anyways, so we don't have a chance to 1902 * receive a barrier... atomic_inc(&mdev->epoch_size); */ 1903 put_ldev(mdev); 1904 drbd_free_ee(mdev, e); 1905 return FALSE; 1906 } 1907 1908 static int receive_DataRequest(struct drbd_conf *mdev, struct p_header *h) 1909 { 1910 sector_t sector; 1911 const sector_t capacity = drbd_get_capacity(mdev->this_bdev); 1912 struct drbd_epoch_entry *e; 1913 struct digest_info *di = NULL; 1914 int size, digest_size; 1915 unsigned int fault_type; 1916 struct p_block_req *p = 1917 (struct p_block_req *)h; 1918 const int brps = sizeof(*p)-sizeof(*h); 1919 1920 if (drbd_recv(mdev, h->payload, brps) != brps) 1921 return FALSE; 1922 1923 sector = be64_to_cpu(p->sector); 1924 size = be32_to_cpu(p->blksize); 1925 1926 if (size <= 0 || (size & 0x1ff) != 0 || size > DRBD_MAX_SEGMENT_SIZE) { 1927 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__, 1928 (unsigned long long)sector, size); 1929 return FALSE; 1930 } 1931 if (sector + (size>>9) > capacity) { 1932 dev_err(DEV, "%s:%d: sector: %llus, size: %u\n", __FILE__, __LINE__, 1933 (unsigned long long)sector, size); 1934 return FALSE; 1935 } 1936 1937 if (!get_ldev_if_state(mdev, D_UP_TO_DATE)) { 1938 if (__ratelimit(&drbd_ratelimit_state)) 1939 dev_err(DEV, "Can not satisfy peer's read request, " 1940 "no local data.\n"); 1941 drbd_send_ack_rp(mdev, h->command == P_DATA_REQUEST ? P_NEG_DREPLY : 1942 P_NEG_RS_DREPLY , p); 1943 return TRUE; 1944 } 1945 1946 /* GFP_NOIO, because we must not cause arbitrary write-out: in a DRBD 1947 * "criss-cross" setup, that might cause write-out on some other DRBD, 1948 * which in turn might block on the other node at this very place. */ 1949 e = drbd_alloc_ee(mdev, p->block_id, sector, size, GFP_NOIO); 1950 if (!e) { 1951 put_ldev(mdev); 1952 return FALSE; 1953 } 1954 1955 e->private_bio->bi_rw = READ; 1956 e->private_bio->bi_end_io = drbd_endio_read_sec; 1957 1958 switch (h->command) { 1959 case P_DATA_REQUEST: 1960 e->w.cb = w_e_end_data_req; 1961 fault_type = DRBD_FAULT_DT_RD; 1962 break; 1963 case P_RS_DATA_REQUEST: 1964 e->w.cb = w_e_end_rsdata_req; 1965 fault_type = DRBD_FAULT_RS_RD; 1966 /* Eventually this should become asynchronously. Currently it 1967 * blocks the whole receiver just to delay the reading of a 1968 * resync data block. 1969 * the drbd_work_queue mechanism is made for this... 1970 */ 1971 if (!drbd_rs_begin_io(mdev, sector)) { 1972 /* we have been interrupted, 1973 * probably connection lost! */ 1974 D_ASSERT(signal_pending(current)); 1975 goto out_free_e; 1976 } 1977 break; 1978 1979 case P_OV_REPLY: 1980 case P_CSUM_RS_REQUEST: 1981 fault_type = DRBD_FAULT_RS_RD; 1982 digest_size = h->length - brps ; 1983 di = kmalloc(sizeof(*di) + digest_size, GFP_NOIO); 1984 if (!di) 1985 goto out_free_e; 1986 1987 di->digest_size = digest_size; 1988 di->digest = (((char *)di)+sizeof(struct digest_info)); 1989 1990 if (drbd_recv(mdev, di->digest, digest_size) != digest_size) 1991 goto out_free_e; 1992 1993 e->block_id = (u64)(unsigned long)di; 1994 if (h->command == P_CSUM_RS_REQUEST) { 1995 D_ASSERT(mdev->agreed_pro_version >= 89); 1996 e->w.cb = w_e_end_csum_rs_req; 1997 } else if (h->command == P_OV_REPLY) { 1998 e->w.cb = w_e_end_ov_reply; 1999 dec_rs_pending(mdev); 2000 break; 2001 } 2002 2003 if (!drbd_rs_begin_io(mdev, sector)) { 2004 /* we have been interrupted, probably connection lost! */ 2005 D_ASSERT(signal_pending(current)); 2006 goto out_free_e; 2007 } 2008 break; 2009 2010 case P_OV_REQUEST: 2011 if (mdev->state.conn >= C_CONNECTED && 2012 mdev->state.conn != C_VERIFY_T) 2013 dev_warn(DEV, "ASSERT FAILED: got P_OV_REQUEST while being %s\n", 2014 drbd_conn_str(mdev->state.conn)); 2015 if (mdev->ov_start_sector == ~(sector_t)0 && 2016 mdev->agreed_pro_version >= 90) { 2017 mdev->ov_start_sector = sector; 2018 mdev->ov_position = sector; 2019 mdev->ov_left = mdev->rs_total - BM_SECT_TO_BIT(sector); 2020 dev_info(DEV, "Online Verify start sector: %llu\n", 2021 (unsigned long long)sector); 2022 } 2023 e->w.cb = w_e_end_ov_req; 2024 fault_type = DRBD_FAULT_RS_RD; 2025 /* Eventually this should become asynchronous. Currently it 2026 * blocks the whole receiver just to delay the reading of a 2027 * resync data block. 2028 * the drbd_work_queue mechanism is made for this... 2029 */ 2030 if (!drbd_rs_begin_io(mdev, sector)) { 2031 /* we have been interrupted, 2032 * probably connection lost! */ 2033 D_ASSERT(signal_pending(current)); 2034 goto out_free_e; 2035 } 2036 break; 2037 2038 2039 default: 2040 dev_err(DEV, "unexpected command (%s) in receive_DataRequest\n", 2041 cmdname(h->command)); 2042 fault_type = DRBD_FAULT_MAX; 2043 } 2044 2045 spin_lock_irq(&mdev->req_lock); 2046 list_add(&e->w.list, &mdev->read_ee); 2047 spin_unlock_irq(&mdev->req_lock); 2048 2049 inc_unacked(mdev); 2050 2051 drbd_generic_make_request(mdev, fault_type, e->private_bio); 2052 maybe_kick_lo(mdev); 2053 2054 return TRUE; 2055 2056 out_free_e: 2057 kfree(di); 2058 put_ldev(mdev); 2059 drbd_free_ee(mdev, e); 2060 return FALSE; 2061 } 2062 2063 static int drbd_asb_recover_0p(struct drbd_conf *mdev) __must_hold(local) 2064 { 2065 int self, peer, rv = -100; 2066 unsigned long ch_self, ch_peer; 2067 2068 self = mdev->ldev->md.uuid[UI_BITMAP] & 1; 2069 peer = mdev->p_uuid[UI_BITMAP] & 1; 2070 2071 ch_peer = mdev->p_uuid[UI_SIZE]; 2072 ch_self = mdev->comm_bm_set; 2073 2074 switch (mdev->net_conf->after_sb_0p) { 2075 case ASB_CONSENSUS: 2076 case ASB_DISCARD_SECONDARY: 2077 case ASB_CALL_HELPER: 2078 dev_err(DEV, "Configuration error.\n"); 2079 break; 2080 case ASB_DISCONNECT: 2081 break; 2082 case ASB_DISCARD_YOUNGER_PRI: 2083 if (self == 0 && peer == 1) { 2084 rv = -1; 2085 break; 2086 } 2087 if (self == 1 && peer == 0) { 2088 rv = 1; 2089 break; 2090 } 2091 /* Else fall through to one of the other strategies... */ 2092 case ASB_DISCARD_OLDER_PRI: 2093 if (self == 0 && peer == 1) { 2094 rv = 1; 2095 break; 2096 } 2097 if (self == 1 && peer == 0) { 2098 rv = -1; 2099 break; 2100 } 2101 /* Else fall through to one of the other strategies... */ 2102 dev_warn(DEV, "Discard younger/older primary did not find a decision\n" 2103 "Using discard-least-changes instead\n"); 2104 case ASB_DISCARD_ZERO_CHG: 2105 if (ch_peer == 0 && ch_self == 0) { 2106 rv = test_bit(DISCARD_CONCURRENT, &mdev->flags) 2107 ? -1 : 1; 2108 break; 2109 } else { 2110 if (ch_peer == 0) { rv = 1; break; } 2111 if (ch_self == 0) { rv = -1; break; } 2112 } 2113 if (mdev->net_conf->after_sb_0p == ASB_DISCARD_ZERO_CHG) 2114 break; 2115 case ASB_DISCARD_LEAST_CHG: 2116 if (ch_self < ch_peer) 2117 rv = -1; 2118 else if (ch_self > ch_peer) 2119 rv = 1; 2120 else /* ( ch_self == ch_peer ) */ 2121 /* Well, then use something else. */ 2122 rv = test_bit(DISCARD_CONCURRENT, &mdev->flags) 2123 ? -1 : 1; 2124 break; 2125 case ASB_DISCARD_LOCAL: 2126 rv = -1; 2127 break; 2128 case ASB_DISCARD_REMOTE: 2129 rv = 1; 2130 } 2131 2132 return rv; 2133 } 2134 2135 static int drbd_asb_recover_1p(struct drbd_conf *mdev) __must_hold(local) 2136 { 2137 int self, peer, hg, rv = -100; 2138 2139 self = mdev->ldev->md.uuid[UI_BITMAP] & 1; 2140 peer = mdev->p_uuid[UI_BITMAP] & 1; 2141 2142 switch (mdev->net_conf->after_sb_1p) { 2143 case ASB_DISCARD_YOUNGER_PRI: 2144 case ASB_DISCARD_OLDER_PRI: 2145 case ASB_DISCARD_LEAST_CHG: 2146 case ASB_DISCARD_LOCAL: 2147 case ASB_DISCARD_REMOTE: 2148 dev_err(DEV, "Configuration error.\n"); 2149 break; 2150 case ASB_DISCONNECT: 2151 break; 2152 case ASB_CONSENSUS: 2153 hg = drbd_asb_recover_0p(mdev); 2154 if (hg == -1 && mdev->state.role == R_SECONDARY) 2155 rv = hg; 2156 if (hg == 1 && mdev->state.role == R_PRIMARY) 2157 rv = hg; 2158 break; 2159 case ASB_VIOLENTLY: 2160 rv = drbd_asb_recover_0p(mdev); 2161 break; 2162 case ASB_DISCARD_SECONDARY: 2163 return mdev->state.role == R_PRIMARY ? 1 : -1; 2164 case ASB_CALL_HELPER: 2165 hg = drbd_asb_recover_0p(mdev); 2166 if (hg == -1 && mdev->state.role == R_PRIMARY) { 2167 self = drbd_set_role(mdev, R_SECONDARY, 0); 2168 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE, 2169 * we might be here in C_WF_REPORT_PARAMS which is transient. 2170 * we do not need to wait for the after state change work either. */ 2171 self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY)); 2172 if (self != SS_SUCCESS) { 2173 drbd_khelper(mdev, "pri-lost-after-sb"); 2174 } else { 2175 dev_warn(DEV, "Successfully gave up primary role.\n"); 2176 rv = hg; 2177 } 2178 } else 2179 rv = hg; 2180 } 2181 2182 return rv; 2183 } 2184 2185 static int drbd_asb_recover_2p(struct drbd_conf *mdev) __must_hold(local) 2186 { 2187 int self, peer, hg, rv = -100; 2188 2189 self = mdev->ldev->md.uuid[UI_BITMAP] & 1; 2190 peer = mdev->p_uuid[UI_BITMAP] & 1; 2191 2192 switch (mdev->net_conf->after_sb_2p) { 2193 case ASB_DISCARD_YOUNGER_PRI: 2194 case ASB_DISCARD_OLDER_PRI: 2195 case ASB_DISCARD_LEAST_CHG: 2196 case ASB_DISCARD_LOCAL: 2197 case ASB_DISCARD_REMOTE: 2198 case ASB_CONSENSUS: 2199 case ASB_DISCARD_SECONDARY: 2200 dev_err(DEV, "Configuration error.\n"); 2201 break; 2202 case ASB_VIOLENTLY: 2203 rv = drbd_asb_recover_0p(mdev); 2204 break; 2205 case ASB_DISCONNECT: 2206 break; 2207 case ASB_CALL_HELPER: 2208 hg = drbd_asb_recover_0p(mdev); 2209 if (hg == -1) { 2210 /* drbd_change_state() does not sleep while in SS_IN_TRANSIENT_STATE, 2211 * we might be here in C_WF_REPORT_PARAMS which is transient. 2212 * we do not need to wait for the after state change work either. */ 2213 self = drbd_change_state(mdev, CS_VERBOSE, NS(role, R_SECONDARY)); 2214 if (self != SS_SUCCESS) { 2215 drbd_khelper(mdev, "pri-lost-after-sb"); 2216 } else { 2217 dev_warn(DEV, "Successfully gave up primary role.\n"); 2218 rv = hg; 2219 } 2220 } else 2221 rv = hg; 2222 } 2223 2224 return rv; 2225 } 2226 2227 static void drbd_uuid_dump(struct drbd_conf *mdev, char *text, u64 *uuid, 2228 u64 bits, u64 flags) 2229 { 2230 if (!uuid) { 2231 dev_info(DEV, "%s uuid info vanished while I was looking!\n", text); 2232 return; 2233 } 2234 dev_info(DEV, "%s %016llX:%016llX:%016llX:%016llX bits:%llu flags:%llX\n", 2235 text, 2236 (unsigned long long)uuid[UI_CURRENT], 2237 (unsigned long long)uuid[UI_BITMAP], 2238 (unsigned long long)uuid[UI_HISTORY_START], 2239 (unsigned long long)uuid[UI_HISTORY_END], 2240 (unsigned long long)bits, 2241 (unsigned long long)flags); 2242 } 2243 2244 /* 2245 100 after split brain try auto recover 2246 2 C_SYNC_SOURCE set BitMap 2247 1 C_SYNC_SOURCE use BitMap 2248 0 no Sync 2249 -1 C_SYNC_TARGET use BitMap 2250 -2 C_SYNC_TARGET set BitMap 2251 -100 after split brain, disconnect 2252 -1000 unrelated data 2253 */ 2254 static int drbd_uuid_compare(struct drbd_conf *mdev, int *rule_nr) __must_hold(local) 2255 { 2256 u64 self, peer; 2257 int i, j; 2258 2259 self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1); 2260 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1); 2261 2262 *rule_nr = 10; 2263 if (self == UUID_JUST_CREATED && peer == UUID_JUST_CREATED) 2264 return 0; 2265 2266 *rule_nr = 20; 2267 if ((self == UUID_JUST_CREATED || self == (u64)0) && 2268 peer != UUID_JUST_CREATED) 2269 return -2; 2270 2271 *rule_nr = 30; 2272 if (self != UUID_JUST_CREATED && 2273 (peer == UUID_JUST_CREATED || peer == (u64)0)) 2274 return 2; 2275 2276 if (self == peer) { 2277 int rct, dc; /* roles at crash time */ 2278 2279 if (mdev->p_uuid[UI_BITMAP] == (u64)0 && mdev->ldev->md.uuid[UI_BITMAP] != (u64)0) { 2280 2281 if (mdev->agreed_pro_version < 91) 2282 return -1001; 2283 2284 if ((mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1)) && 2285 (mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1))) { 2286 dev_info(DEV, "was SyncSource, missed the resync finished event, corrected myself:\n"); 2287 drbd_uuid_set_bm(mdev, 0UL); 2288 2289 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, 2290 mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0); 2291 *rule_nr = 34; 2292 } else { 2293 dev_info(DEV, "was SyncSource (peer failed to write sync_uuid)\n"); 2294 *rule_nr = 36; 2295 } 2296 2297 return 1; 2298 } 2299 2300 if (mdev->ldev->md.uuid[UI_BITMAP] == (u64)0 && mdev->p_uuid[UI_BITMAP] != (u64)0) { 2301 2302 if (mdev->agreed_pro_version < 91) 2303 return -1001; 2304 2305 if ((mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1)) == (mdev->p_uuid[UI_BITMAP] & ~((u64)1)) && 2306 (mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1)) == (mdev->p_uuid[UI_HISTORY_START] & ~((u64)1))) { 2307 dev_info(DEV, "was SyncTarget, peer missed the resync finished event, corrected peer:\n"); 2308 2309 mdev->p_uuid[UI_HISTORY_START + 1] = mdev->p_uuid[UI_HISTORY_START]; 2310 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_BITMAP]; 2311 mdev->p_uuid[UI_BITMAP] = 0UL; 2312 2313 drbd_uuid_dump(mdev, "peer", mdev->p_uuid, mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]); 2314 *rule_nr = 35; 2315 } else { 2316 dev_info(DEV, "was SyncTarget (failed to write sync_uuid)\n"); 2317 *rule_nr = 37; 2318 } 2319 2320 return -1; 2321 } 2322 2323 /* Common power [off|failure] */ 2324 rct = (test_bit(CRASHED_PRIMARY, &mdev->flags) ? 1 : 0) + 2325 (mdev->p_uuid[UI_FLAGS] & 2); 2326 /* lowest bit is set when we were primary, 2327 * next bit (weight 2) is set when peer was primary */ 2328 *rule_nr = 40; 2329 2330 switch (rct) { 2331 case 0: /* !self_pri && !peer_pri */ return 0; 2332 case 1: /* self_pri && !peer_pri */ return 1; 2333 case 2: /* !self_pri && peer_pri */ return -1; 2334 case 3: /* self_pri && peer_pri */ 2335 dc = test_bit(DISCARD_CONCURRENT, &mdev->flags); 2336 return dc ? -1 : 1; 2337 } 2338 } 2339 2340 *rule_nr = 50; 2341 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1); 2342 if (self == peer) 2343 return -1; 2344 2345 *rule_nr = 51; 2346 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1); 2347 if (self == peer) { 2348 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1); 2349 peer = mdev->p_uuid[UI_HISTORY_START + 1] & ~((u64)1); 2350 if (self == peer) { 2351 /* The last P_SYNC_UUID did not get though. Undo the last start of 2352 resync as sync source modifications of the peer's UUIDs. */ 2353 2354 if (mdev->agreed_pro_version < 91) 2355 return -1001; 2356 2357 mdev->p_uuid[UI_BITMAP] = mdev->p_uuid[UI_HISTORY_START]; 2358 mdev->p_uuid[UI_HISTORY_START] = mdev->p_uuid[UI_HISTORY_START + 1]; 2359 return -1; 2360 } 2361 } 2362 2363 *rule_nr = 60; 2364 self = mdev->ldev->md.uuid[UI_CURRENT] & ~((u64)1); 2365 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) { 2366 peer = mdev->p_uuid[i] & ~((u64)1); 2367 if (self == peer) 2368 return -2; 2369 } 2370 2371 *rule_nr = 70; 2372 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1); 2373 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1); 2374 if (self == peer) 2375 return 1; 2376 2377 *rule_nr = 71; 2378 self = mdev->ldev->md.uuid[UI_HISTORY_START] & ~((u64)1); 2379 if (self == peer) { 2380 self = mdev->ldev->md.uuid[UI_HISTORY_START + 1] & ~((u64)1); 2381 peer = mdev->p_uuid[UI_HISTORY_START] & ~((u64)1); 2382 if (self == peer) { 2383 /* The last P_SYNC_UUID did not get though. Undo the last start of 2384 resync as sync source modifications of our UUIDs. */ 2385 2386 if (mdev->agreed_pro_version < 91) 2387 return -1001; 2388 2389 _drbd_uuid_set(mdev, UI_BITMAP, mdev->ldev->md.uuid[UI_HISTORY_START]); 2390 _drbd_uuid_set(mdev, UI_HISTORY_START, mdev->ldev->md.uuid[UI_HISTORY_START + 1]); 2391 2392 dev_info(DEV, "Undid last start of resync:\n"); 2393 2394 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, 2395 mdev->state.disk >= D_NEGOTIATING ? drbd_bm_total_weight(mdev) : 0, 0); 2396 2397 return 1; 2398 } 2399 } 2400 2401 2402 *rule_nr = 80; 2403 peer = mdev->p_uuid[UI_CURRENT] & ~((u64)1); 2404 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) { 2405 self = mdev->ldev->md.uuid[i] & ~((u64)1); 2406 if (self == peer) 2407 return 2; 2408 } 2409 2410 *rule_nr = 90; 2411 self = mdev->ldev->md.uuid[UI_BITMAP] & ~((u64)1); 2412 peer = mdev->p_uuid[UI_BITMAP] & ~((u64)1); 2413 if (self == peer && self != ((u64)0)) 2414 return 100; 2415 2416 *rule_nr = 100; 2417 for (i = UI_HISTORY_START; i <= UI_HISTORY_END; i++) { 2418 self = mdev->ldev->md.uuid[i] & ~((u64)1); 2419 for (j = UI_HISTORY_START; j <= UI_HISTORY_END; j++) { 2420 peer = mdev->p_uuid[j] & ~((u64)1); 2421 if (self == peer) 2422 return -100; 2423 } 2424 } 2425 2426 return -1000; 2427 } 2428 2429 /* drbd_sync_handshake() returns the new conn state on success, or 2430 CONN_MASK (-1) on failure. 2431 */ 2432 static enum drbd_conns drbd_sync_handshake(struct drbd_conf *mdev, enum drbd_role peer_role, 2433 enum drbd_disk_state peer_disk) __must_hold(local) 2434 { 2435 int hg, rule_nr; 2436 enum drbd_conns rv = C_MASK; 2437 enum drbd_disk_state mydisk; 2438 2439 mydisk = mdev->state.disk; 2440 if (mydisk == D_NEGOTIATING) 2441 mydisk = mdev->new_state_tmp.disk; 2442 2443 dev_info(DEV, "drbd_sync_handshake:\n"); 2444 drbd_uuid_dump(mdev, "self", mdev->ldev->md.uuid, mdev->comm_bm_set, 0); 2445 drbd_uuid_dump(mdev, "peer", mdev->p_uuid, 2446 mdev->p_uuid[UI_SIZE], mdev->p_uuid[UI_FLAGS]); 2447 2448 hg = drbd_uuid_compare(mdev, &rule_nr); 2449 2450 dev_info(DEV, "uuid_compare()=%d by rule %d\n", hg, rule_nr); 2451 2452 if (hg == -1000) { 2453 dev_alert(DEV, "Unrelated data, aborting!\n"); 2454 return C_MASK; 2455 } 2456 if (hg == -1001) { 2457 dev_alert(DEV, "To resolve this both sides have to support at least protocol\n"); 2458 return C_MASK; 2459 } 2460 2461 if ((mydisk == D_INCONSISTENT && peer_disk > D_INCONSISTENT) || 2462 (peer_disk == D_INCONSISTENT && mydisk > D_INCONSISTENT)) { 2463 int f = (hg == -100) || abs(hg) == 2; 2464 hg = mydisk > D_INCONSISTENT ? 1 : -1; 2465 if (f) 2466 hg = hg*2; 2467 dev_info(DEV, "Becoming sync %s due to disk states.\n", 2468 hg > 0 ? "source" : "target"); 2469 } 2470 2471 if (hg == 100 || (hg == -100 && mdev->net_conf->always_asbp)) { 2472 int pcount = (mdev->state.role == R_PRIMARY) 2473 + (peer_role == R_PRIMARY); 2474 int forced = (hg == -100); 2475 2476 switch (pcount) { 2477 case 0: 2478 hg = drbd_asb_recover_0p(mdev); 2479 break; 2480 case 1: 2481 hg = drbd_asb_recover_1p(mdev); 2482 break; 2483 case 2: 2484 hg = drbd_asb_recover_2p(mdev); 2485 break; 2486 } 2487 if (abs(hg) < 100) { 2488 dev_warn(DEV, "Split-Brain detected, %d primaries, " 2489 "automatically solved. Sync from %s node\n", 2490 pcount, (hg < 0) ? "peer" : "this"); 2491 if (forced) { 2492 dev_warn(DEV, "Doing a full sync, since" 2493 " UUIDs where ambiguous.\n"); 2494 hg = hg*2; 2495 } 2496 } 2497 } 2498 2499 if (hg == -100) { 2500 if (mdev->net_conf->want_lose && !(mdev->p_uuid[UI_FLAGS]&1)) 2501 hg = -1; 2502 if (!mdev->net_conf->want_lose && (mdev->p_uuid[UI_FLAGS]&1)) 2503 hg = 1; 2504 2505 if (abs(hg) < 100) 2506 dev_warn(DEV, "Split-Brain detected, manually solved. " 2507 "Sync from %s node\n", 2508 (hg < 0) ? "peer" : "this"); 2509 } 2510 2511 if (hg == -100) { 2512 dev_alert(DEV, "Split-Brain detected, dropping connection!\n"); 2513 drbd_khelper(mdev, "split-brain"); 2514 return C_MASK; 2515 } 2516 2517 if (hg > 0 && mydisk <= D_INCONSISTENT) { 2518 dev_err(DEV, "I shall become SyncSource, but I am inconsistent!\n"); 2519 return C_MASK; 2520 } 2521 2522 if (hg < 0 && /* by intention we do not use mydisk here. */ 2523 mdev->state.role == R_PRIMARY && mdev->state.disk >= D_CONSISTENT) { 2524 switch (mdev->net_conf->rr_conflict) { 2525 case ASB_CALL_HELPER: 2526 drbd_khelper(mdev, "pri-lost"); 2527 /* fall through */ 2528 case ASB_DISCONNECT: 2529 dev_err(DEV, "I shall become SyncTarget, but I am primary!\n"); 2530 return C_MASK; 2531 case ASB_VIOLENTLY: 2532 dev_warn(DEV, "Becoming SyncTarget, violating the stable-data" 2533 "assumption\n"); 2534 } 2535 } 2536 2537 if (abs(hg) >= 2) { 2538 dev_info(DEV, "Writing the whole bitmap, full sync required after drbd_sync_handshake.\n"); 2539 if (drbd_bitmap_io(mdev, &drbd_bmio_set_n_write, "set_n_write from sync_handshake")) 2540 return C_MASK; 2541 } 2542 2543 if (hg > 0) { /* become sync source. */ 2544 rv = C_WF_BITMAP_S; 2545 } else if (hg < 0) { /* become sync target */ 2546 rv = C_WF_BITMAP_T; 2547 } else { 2548 rv = C_CONNECTED; 2549 if (drbd_bm_total_weight(mdev)) { 2550 dev_info(DEV, "No resync, but %lu bits in bitmap!\n", 2551 drbd_bm_total_weight(mdev)); 2552 } 2553 } 2554 2555 return rv; 2556 } 2557 2558 /* returns 1 if invalid */ 2559 static int cmp_after_sb(enum drbd_after_sb_p peer, enum drbd_after_sb_p self) 2560 { 2561 /* ASB_DISCARD_REMOTE - ASB_DISCARD_LOCAL is valid */ 2562 if ((peer == ASB_DISCARD_REMOTE && self == ASB_DISCARD_LOCAL) || 2563 (self == ASB_DISCARD_REMOTE && peer == ASB_DISCARD_LOCAL)) 2564 return 0; 2565 2566 /* any other things with ASB_DISCARD_REMOTE or ASB_DISCARD_LOCAL are invalid */ 2567 if (peer == ASB_DISCARD_REMOTE || peer == ASB_DISCARD_LOCAL || 2568 self == ASB_DISCARD_REMOTE || self == ASB_DISCARD_LOCAL) 2569 return 1; 2570 2571 /* everything else is valid if they are equal on both sides. */ 2572 if (peer == self) 2573 return 0; 2574 2575 /* everything es is invalid. */ 2576 return 1; 2577 } 2578 2579 static int receive_protocol(struct drbd_conf *mdev, struct p_header *h) 2580 { 2581 struct p_protocol *p = (struct p_protocol *)h; 2582 int header_size, data_size; 2583 int p_proto, p_after_sb_0p, p_after_sb_1p, p_after_sb_2p; 2584 int p_want_lose, p_two_primaries; 2585 char p_integrity_alg[SHARED_SECRET_MAX] = ""; 2586 2587 header_size = sizeof(*p) - sizeof(*h); 2588 data_size = h->length - header_size; 2589 2590 if (drbd_recv(mdev, h->payload, header_size) != header_size) 2591 return FALSE; 2592 2593 p_proto = be32_to_cpu(p->protocol); 2594 p_after_sb_0p = be32_to_cpu(p->after_sb_0p); 2595 p_after_sb_1p = be32_to_cpu(p->after_sb_1p); 2596 p_after_sb_2p = be32_to_cpu(p->after_sb_2p); 2597 p_want_lose = be32_to_cpu(p->want_lose); 2598 p_two_primaries = be32_to_cpu(p->two_primaries); 2599 2600 if (p_proto != mdev->net_conf->wire_protocol) { 2601 dev_err(DEV, "incompatible communication protocols\n"); 2602 goto disconnect; 2603 } 2604 2605 if (cmp_after_sb(p_after_sb_0p, mdev->net_conf->after_sb_0p)) { 2606 dev_err(DEV, "incompatible after-sb-0pri settings\n"); 2607 goto disconnect; 2608 } 2609 2610 if (cmp_after_sb(p_after_sb_1p, mdev->net_conf->after_sb_1p)) { 2611 dev_err(DEV, "incompatible after-sb-1pri settings\n"); 2612 goto disconnect; 2613 } 2614 2615 if (cmp_after_sb(p_after_sb_2p, mdev->net_conf->after_sb_2p)) { 2616 dev_err(DEV, "incompatible after-sb-2pri settings\n"); 2617 goto disconnect; 2618 } 2619 2620 if (p_want_lose && mdev->net_conf->want_lose) { 2621 dev_err(DEV, "both sides have the 'want_lose' flag set\n"); 2622 goto disconnect; 2623 } 2624 2625 if (p_two_primaries != mdev->net_conf->two_primaries) { 2626 dev_err(DEV, "incompatible setting of the two-primaries options\n"); 2627 goto disconnect; 2628 } 2629 2630 if (mdev->agreed_pro_version >= 87) { 2631 unsigned char *my_alg = mdev->net_conf->integrity_alg; 2632 2633 if (drbd_recv(mdev, p_integrity_alg, data_size) != data_size) 2634 return FALSE; 2635 2636 p_integrity_alg[SHARED_SECRET_MAX-1] = 0; 2637 if (strcmp(p_integrity_alg, my_alg)) { 2638 dev_err(DEV, "incompatible setting of the data-integrity-alg\n"); 2639 goto disconnect; 2640 } 2641 dev_info(DEV, "data-integrity-alg: %s\n", 2642 my_alg[0] ? my_alg : (unsigned char *)"<not-used>"); 2643 } 2644 2645 return TRUE; 2646 2647 disconnect: 2648 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2649 return FALSE; 2650 } 2651 2652 /* helper function 2653 * input: alg name, feature name 2654 * return: NULL (alg name was "") 2655 * ERR_PTR(error) if something goes wrong 2656 * or the crypto hash ptr, if it worked out ok. */ 2657 struct crypto_hash *drbd_crypto_alloc_digest_safe(const struct drbd_conf *mdev, 2658 const char *alg, const char *name) 2659 { 2660 struct crypto_hash *tfm; 2661 2662 if (!alg[0]) 2663 return NULL; 2664 2665 tfm = crypto_alloc_hash(alg, 0, CRYPTO_ALG_ASYNC); 2666 if (IS_ERR(tfm)) { 2667 dev_err(DEV, "Can not allocate \"%s\" as %s (reason: %ld)\n", 2668 alg, name, PTR_ERR(tfm)); 2669 return tfm; 2670 } 2671 if (!drbd_crypto_is_hash(crypto_hash_tfm(tfm))) { 2672 crypto_free_hash(tfm); 2673 dev_err(DEV, "\"%s\" is not a digest (%s)\n", alg, name); 2674 return ERR_PTR(-EINVAL); 2675 } 2676 return tfm; 2677 } 2678 2679 static int receive_SyncParam(struct drbd_conf *mdev, struct p_header *h) 2680 { 2681 int ok = TRUE; 2682 struct p_rs_param_89 *p = (struct p_rs_param_89 *)h; 2683 unsigned int header_size, data_size, exp_max_sz; 2684 struct crypto_hash *verify_tfm = NULL; 2685 struct crypto_hash *csums_tfm = NULL; 2686 const int apv = mdev->agreed_pro_version; 2687 2688 exp_max_sz = apv <= 87 ? sizeof(struct p_rs_param) 2689 : apv == 88 ? sizeof(struct p_rs_param) 2690 + SHARED_SECRET_MAX 2691 : /* 89 */ sizeof(struct p_rs_param_89); 2692 2693 if (h->length > exp_max_sz) { 2694 dev_err(DEV, "SyncParam packet too long: received %u, expected <= %u bytes\n", 2695 h->length, exp_max_sz); 2696 return FALSE; 2697 } 2698 2699 if (apv <= 88) { 2700 header_size = sizeof(struct p_rs_param) - sizeof(*h); 2701 data_size = h->length - header_size; 2702 } else /* apv >= 89 */ { 2703 header_size = sizeof(struct p_rs_param_89) - sizeof(*h); 2704 data_size = h->length - header_size; 2705 D_ASSERT(data_size == 0); 2706 } 2707 2708 /* initialize verify_alg and csums_alg */ 2709 memset(p->verify_alg, 0, 2 * SHARED_SECRET_MAX); 2710 2711 if (drbd_recv(mdev, h->payload, header_size) != header_size) 2712 return FALSE; 2713 2714 mdev->sync_conf.rate = be32_to_cpu(p->rate); 2715 2716 if (apv >= 88) { 2717 if (apv == 88) { 2718 if (data_size > SHARED_SECRET_MAX) { 2719 dev_err(DEV, "verify-alg too long, " 2720 "peer wants %u, accepting only %u byte\n", 2721 data_size, SHARED_SECRET_MAX); 2722 return FALSE; 2723 } 2724 2725 if (drbd_recv(mdev, p->verify_alg, data_size) != data_size) 2726 return FALSE; 2727 2728 /* we expect NUL terminated string */ 2729 /* but just in case someone tries to be evil */ 2730 D_ASSERT(p->verify_alg[data_size-1] == 0); 2731 p->verify_alg[data_size-1] = 0; 2732 2733 } else /* apv >= 89 */ { 2734 /* we still expect NUL terminated strings */ 2735 /* but just in case someone tries to be evil */ 2736 D_ASSERT(p->verify_alg[SHARED_SECRET_MAX-1] == 0); 2737 D_ASSERT(p->csums_alg[SHARED_SECRET_MAX-1] == 0); 2738 p->verify_alg[SHARED_SECRET_MAX-1] = 0; 2739 p->csums_alg[SHARED_SECRET_MAX-1] = 0; 2740 } 2741 2742 if (strcmp(mdev->sync_conf.verify_alg, p->verify_alg)) { 2743 if (mdev->state.conn == C_WF_REPORT_PARAMS) { 2744 dev_err(DEV, "Different verify-alg settings. me=\"%s\" peer=\"%s\"\n", 2745 mdev->sync_conf.verify_alg, p->verify_alg); 2746 goto disconnect; 2747 } 2748 verify_tfm = drbd_crypto_alloc_digest_safe(mdev, 2749 p->verify_alg, "verify-alg"); 2750 if (IS_ERR(verify_tfm)) { 2751 verify_tfm = NULL; 2752 goto disconnect; 2753 } 2754 } 2755 2756 if (apv >= 89 && strcmp(mdev->sync_conf.csums_alg, p->csums_alg)) { 2757 if (mdev->state.conn == C_WF_REPORT_PARAMS) { 2758 dev_err(DEV, "Different csums-alg settings. me=\"%s\" peer=\"%s\"\n", 2759 mdev->sync_conf.csums_alg, p->csums_alg); 2760 goto disconnect; 2761 } 2762 csums_tfm = drbd_crypto_alloc_digest_safe(mdev, 2763 p->csums_alg, "csums-alg"); 2764 if (IS_ERR(csums_tfm)) { 2765 csums_tfm = NULL; 2766 goto disconnect; 2767 } 2768 } 2769 2770 2771 spin_lock(&mdev->peer_seq_lock); 2772 /* lock against drbd_nl_syncer_conf() */ 2773 if (verify_tfm) { 2774 strcpy(mdev->sync_conf.verify_alg, p->verify_alg); 2775 mdev->sync_conf.verify_alg_len = strlen(p->verify_alg) + 1; 2776 crypto_free_hash(mdev->verify_tfm); 2777 mdev->verify_tfm = verify_tfm; 2778 dev_info(DEV, "using verify-alg: \"%s\"\n", p->verify_alg); 2779 } 2780 if (csums_tfm) { 2781 strcpy(mdev->sync_conf.csums_alg, p->csums_alg); 2782 mdev->sync_conf.csums_alg_len = strlen(p->csums_alg) + 1; 2783 crypto_free_hash(mdev->csums_tfm); 2784 mdev->csums_tfm = csums_tfm; 2785 dev_info(DEV, "using csums-alg: \"%s\"\n", p->csums_alg); 2786 } 2787 spin_unlock(&mdev->peer_seq_lock); 2788 } 2789 2790 return ok; 2791 disconnect: 2792 /* just for completeness: actually not needed, 2793 * as this is not reached if csums_tfm was ok. */ 2794 crypto_free_hash(csums_tfm); 2795 /* but free the verify_tfm again, if csums_tfm did not work out */ 2796 crypto_free_hash(verify_tfm); 2797 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2798 return FALSE; 2799 } 2800 2801 static void drbd_setup_order_type(struct drbd_conf *mdev, int peer) 2802 { 2803 /* sorry, we currently have no working implementation 2804 * of distributed TCQ */ 2805 } 2806 2807 /* warn if the arguments differ by more than 12.5% */ 2808 static void warn_if_differ_considerably(struct drbd_conf *mdev, 2809 const char *s, sector_t a, sector_t b) 2810 { 2811 sector_t d; 2812 if (a == 0 || b == 0) 2813 return; 2814 d = (a > b) ? (a - b) : (b - a); 2815 if (d > (a>>3) || d > (b>>3)) 2816 dev_warn(DEV, "Considerable difference in %s: %llus vs. %llus\n", s, 2817 (unsigned long long)a, (unsigned long long)b); 2818 } 2819 2820 static int receive_sizes(struct drbd_conf *mdev, struct p_header *h) 2821 { 2822 struct p_sizes *p = (struct p_sizes *)h; 2823 enum determine_dev_size dd = unchanged; 2824 unsigned int max_seg_s; 2825 sector_t p_size, p_usize, my_usize; 2826 int ldsc = 0; /* local disk size changed */ 2827 enum drbd_conns nconn; 2828 2829 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE; 2830 if (drbd_recv(mdev, h->payload, h->length) != h->length) 2831 return FALSE; 2832 2833 p_size = be64_to_cpu(p->d_size); 2834 p_usize = be64_to_cpu(p->u_size); 2835 2836 if (p_size == 0 && mdev->state.disk == D_DISKLESS) { 2837 dev_err(DEV, "some backing storage is needed\n"); 2838 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2839 return FALSE; 2840 } 2841 2842 /* just store the peer's disk size for now. 2843 * we still need to figure out whether we accept that. */ 2844 mdev->p_size = p_size; 2845 2846 #define min_not_zero(l, r) (l == 0) ? r : ((r == 0) ? l : min(l, r)) 2847 if (get_ldev(mdev)) { 2848 warn_if_differ_considerably(mdev, "lower level device sizes", 2849 p_size, drbd_get_max_capacity(mdev->ldev)); 2850 warn_if_differ_considerably(mdev, "user requested size", 2851 p_usize, mdev->ldev->dc.disk_size); 2852 2853 /* if this is the first connect, or an otherwise expected 2854 * param exchange, choose the minimum */ 2855 if (mdev->state.conn == C_WF_REPORT_PARAMS) 2856 p_usize = min_not_zero((sector_t)mdev->ldev->dc.disk_size, 2857 p_usize); 2858 2859 my_usize = mdev->ldev->dc.disk_size; 2860 2861 if (mdev->ldev->dc.disk_size != p_usize) { 2862 mdev->ldev->dc.disk_size = p_usize; 2863 dev_info(DEV, "Peer sets u_size to %lu sectors\n", 2864 (unsigned long)mdev->ldev->dc.disk_size); 2865 } 2866 2867 /* Never shrink a device with usable data during connect. 2868 But allow online shrinking if we are connected. */ 2869 if (drbd_new_dev_size(mdev, mdev->ldev) < 2870 drbd_get_capacity(mdev->this_bdev) && 2871 mdev->state.disk >= D_OUTDATED && 2872 mdev->state.conn < C_CONNECTED) { 2873 dev_err(DEV, "The peer's disk size is too small!\n"); 2874 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2875 mdev->ldev->dc.disk_size = my_usize; 2876 put_ldev(mdev); 2877 return FALSE; 2878 } 2879 put_ldev(mdev); 2880 } 2881 #undef min_not_zero 2882 2883 if (get_ldev(mdev)) { 2884 dd = drbd_determin_dev_size(mdev); 2885 put_ldev(mdev); 2886 if (dd == dev_size_error) 2887 return FALSE; 2888 drbd_md_sync(mdev); 2889 } else { 2890 /* I am diskless, need to accept the peer's size. */ 2891 drbd_set_my_capacity(mdev, p_size); 2892 } 2893 2894 if (mdev->p_uuid && mdev->state.conn <= C_CONNECTED && get_ldev(mdev)) { 2895 nconn = drbd_sync_handshake(mdev, 2896 mdev->state.peer, mdev->state.pdsk); 2897 put_ldev(mdev); 2898 2899 if (nconn == C_MASK) { 2900 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2901 return FALSE; 2902 } 2903 2904 if (drbd_request_state(mdev, NS(conn, nconn)) < SS_SUCCESS) { 2905 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2906 return FALSE; 2907 } 2908 } 2909 2910 if (get_ldev(mdev)) { 2911 if (mdev->ldev->known_size != drbd_get_capacity(mdev->ldev->backing_bdev)) { 2912 mdev->ldev->known_size = drbd_get_capacity(mdev->ldev->backing_bdev); 2913 ldsc = 1; 2914 } 2915 2916 max_seg_s = be32_to_cpu(p->max_segment_size); 2917 if (max_seg_s != queue_max_segment_size(mdev->rq_queue)) 2918 drbd_setup_queue_param(mdev, max_seg_s); 2919 2920 drbd_setup_order_type(mdev, be32_to_cpu(p->queue_order_type)); 2921 put_ldev(mdev); 2922 } 2923 2924 if (mdev->state.conn > C_WF_REPORT_PARAMS) { 2925 if (be64_to_cpu(p->c_size) != 2926 drbd_get_capacity(mdev->this_bdev) || ldsc) { 2927 /* we have different sizes, probably peer 2928 * needs to know my new size... */ 2929 drbd_send_sizes(mdev, 0); 2930 } 2931 if (test_and_clear_bit(RESIZE_PENDING, &mdev->flags) || 2932 (dd == grew && mdev->state.conn == C_CONNECTED)) { 2933 if (mdev->state.pdsk >= D_INCONSISTENT && 2934 mdev->state.disk >= D_INCONSISTENT) 2935 resync_after_online_grow(mdev); 2936 else 2937 set_bit(RESYNC_AFTER_NEG, &mdev->flags); 2938 } 2939 } 2940 2941 return TRUE; 2942 } 2943 2944 static int receive_uuids(struct drbd_conf *mdev, struct p_header *h) 2945 { 2946 struct p_uuids *p = (struct p_uuids *)h; 2947 u64 *p_uuid; 2948 int i; 2949 2950 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE; 2951 if (drbd_recv(mdev, h->payload, h->length) != h->length) 2952 return FALSE; 2953 2954 p_uuid = kmalloc(sizeof(u64)*UI_EXTENDED_SIZE, GFP_NOIO); 2955 2956 for (i = UI_CURRENT; i < UI_EXTENDED_SIZE; i++) 2957 p_uuid[i] = be64_to_cpu(p->uuid[i]); 2958 2959 kfree(mdev->p_uuid); 2960 mdev->p_uuid = p_uuid; 2961 2962 if (mdev->state.conn < C_CONNECTED && 2963 mdev->state.disk < D_INCONSISTENT && 2964 mdev->state.role == R_PRIMARY && 2965 (mdev->ed_uuid & ~((u64)1)) != (p_uuid[UI_CURRENT] & ~((u64)1))) { 2966 dev_err(DEV, "Can only connect to data with current UUID=%016llX\n", 2967 (unsigned long long)mdev->ed_uuid); 2968 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 2969 return FALSE; 2970 } 2971 2972 if (get_ldev(mdev)) { 2973 int skip_initial_sync = 2974 mdev->state.conn == C_CONNECTED && 2975 mdev->agreed_pro_version >= 90 && 2976 mdev->ldev->md.uuid[UI_CURRENT] == UUID_JUST_CREATED && 2977 (p_uuid[UI_FLAGS] & 8); 2978 if (skip_initial_sync) { 2979 dev_info(DEV, "Accepted new current UUID, preparing to skip initial sync\n"); 2980 drbd_bitmap_io(mdev, &drbd_bmio_clear_n_write, 2981 "clear_n_write from receive_uuids"); 2982 _drbd_uuid_set(mdev, UI_CURRENT, p_uuid[UI_CURRENT]); 2983 _drbd_uuid_set(mdev, UI_BITMAP, 0); 2984 _drbd_set_state(_NS2(mdev, disk, D_UP_TO_DATE, pdsk, D_UP_TO_DATE), 2985 CS_VERBOSE, NULL); 2986 drbd_md_sync(mdev); 2987 } 2988 put_ldev(mdev); 2989 } 2990 2991 /* Before we test for the disk state, we should wait until an eventually 2992 ongoing cluster wide state change is finished. That is important if 2993 we are primary and are detaching from our disk. We need to see the 2994 new disk state... */ 2995 wait_event(mdev->misc_wait, !test_bit(CLUSTER_ST_CHANGE, &mdev->flags)); 2996 if (mdev->state.conn >= C_CONNECTED && mdev->state.disk < D_INCONSISTENT) 2997 drbd_set_ed_uuid(mdev, p_uuid[UI_CURRENT]); 2998 2999 return TRUE; 3000 } 3001 3002 /** 3003 * convert_state() - Converts the peer's view of the cluster state to our point of view 3004 * @ps: The state as seen by the peer. 3005 */ 3006 static union drbd_state convert_state(union drbd_state ps) 3007 { 3008 union drbd_state ms; 3009 3010 static enum drbd_conns c_tab[] = { 3011 [C_CONNECTED] = C_CONNECTED, 3012 3013 [C_STARTING_SYNC_S] = C_STARTING_SYNC_T, 3014 [C_STARTING_SYNC_T] = C_STARTING_SYNC_S, 3015 [C_DISCONNECTING] = C_TEAR_DOWN, /* C_NETWORK_FAILURE, */ 3016 [C_VERIFY_S] = C_VERIFY_T, 3017 [C_MASK] = C_MASK, 3018 }; 3019 3020 ms.i = ps.i; 3021 3022 ms.conn = c_tab[ps.conn]; 3023 ms.peer = ps.role; 3024 ms.role = ps.peer; 3025 ms.pdsk = ps.disk; 3026 ms.disk = ps.pdsk; 3027 ms.peer_isp = (ps.aftr_isp | ps.user_isp); 3028 3029 return ms; 3030 } 3031 3032 static int receive_req_state(struct drbd_conf *mdev, struct p_header *h) 3033 { 3034 struct p_req_state *p = (struct p_req_state *)h; 3035 union drbd_state mask, val; 3036 int rv; 3037 3038 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE; 3039 if (drbd_recv(mdev, h->payload, h->length) != h->length) 3040 return FALSE; 3041 3042 mask.i = be32_to_cpu(p->mask); 3043 val.i = be32_to_cpu(p->val); 3044 3045 if (test_bit(DISCARD_CONCURRENT, &mdev->flags) && 3046 test_bit(CLUSTER_ST_CHANGE, &mdev->flags)) { 3047 drbd_send_sr_reply(mdev, SS_CONCURRENT_ST_CHG); 3048 return TRUE; 3049 } 3050 3051 mask = convert_state(mask); 3052 val = convert_state(val); 3053 3054 rv = drbd_change_state(mdev, CS_VERBOSE, mask, val); 3055 3056 drbd_send_sr_reply(mdev, rv); 3057 drbd_md_sync(mdev); 3058 3059 return TRUE; 3060 } 3061 3062 static int receive_state(struct drbd_conf *mdev, struct p_header *h) 3063 { 3064 struct p_state *p = (struct p_state *)h; 3065 enum drbd_conns nconn, oconn; 3066 union drbd_state ns, peer_state; 3067 enum drbd_disk_state real_peer_disk; 3068 int rv; 3069 3070 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) 3071 return FALSE; 3072 3073 if (drbd_recv(mdev, h->payload, h->length) != h->length) 3074 return FALSE; 3075 3076 peer_state.i = be32_to_cpu(p->state); 3077 3078 real_peer_disk = peer_state.disk; 3079 if (peer_state.disk == D_NEGOTIATING) { 3080 real_peer_disk = mdev->p_uuid[UI_FLAGS] & 4 ? D_INCONSISTENT : D_CONSISTENT; 3081 dev_info(DEV, "real peer disk state = %s\n", drbd_disk_str(real_peer_disk)); 3082 } 3083 3084 spin_lock_irq(&mdev->req_lock); 3085 retry: 3086 oconn = nconn = mdev->state.conn; 3087 spin_unlock_irq(&mdev->req_lock); 3088 3089 if (nconn == C_WF_REPORT_PARAMS) 3090 nconn = C_CONNECTED; 3091 3092 if (mdev->p_uuid && peer_state.disk >= D_NEGOTIATING && 3093 get_ldev_if_state(mdev, D_NEGOTIATING)) { 3094 int cr; /* consider resync */ 3095 3096 /* if we established a new connection */ 3097 cr = (oconn < C_CONNECTED); 3098 /* if we had an established connection 3099 * and one of the nodes newly attaches a disk */ 3100 cr |= (oconn == C_CONNECTED && 3101 (peer_state.disk == D_NEGOTIATING || 3102 mdev->state.disk == D_NEGOTIATING)); 3103 /* if we have both been inconsistent, and the peer has been 3104 * forced to be UpToDate with --overwrite-data */ 3105 cr |= test_bit(CONSIDER_RESYNC, &mdev->flags); 3106 /* if we had been plain connected, and the admin requested to 3107 * start a sync by "invalidate" or "invalidate-remote" */ 3108 cr |= (oconn == C_CONNECTED && 3109 (peer_state.conn >= C_STARTING_SYNC_S && 3110 peer_state.conn <= C_WF_BITMAP_T)); 3111 3112 if (cr) 3113 nconn = drbd_sync_handshake(mdev, peer_state.role, real_peer_disk); 3114 3115 put_ldev(mdev); 3116 if (nconn == C_MASK) { 3117 if (mdev->state.disk == D_NEGOTIATING) { 3118 drbd_force_state(mdev, NS(disk, D_DISKLESS)); 3119 nconn = C_CONNECTED; 3120 } else if (peer_state.disk == D_NEGOTIATING) { 3121 dev_err(DEV, "Disk attach process on the peer node was aborted.\n"); 3122 peer_state.disk = D_DISKLESS; 3123 } else { 3124 D_ASSERT(oconn == C_WF_REPORT_PARAMS); 3125 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 3126 return FALSE; 3127 } 3128 } 3129 } 3130 3131 spin_lock_irq(&mdev->req_lock); 3132 if (mdev->state.conn != oconn) 3133 goto retry; 3134 clear_bit(CONSIDER_RESYNC, &mdev->flags); 3135 ns.i = mdev->state.i; 3136 ns.conn = nconn; 3137 ns.peer = peer_state.role; 3138 ns.pdsk = real_peer_disk; 3139 ns.peer_isp = (peer_state.aftr_isp | peer_state.user_isp); 3140 if ((nconn == C_CONNECTED || nconn == C_WF_BITMAP_S) && ns.disk == D_NEGOTIATING) 3141 ns.disk = mdev->new_state_tmp.disk; 3142 3143 rv = _drbd_set_state(mdev, ns, CS_VERBOSE | CS_HARD, NULL); 3144 ns = mdev->state; 3145 spin_unlock_irq(&mdev->req_lock); 3146 3147 if (rv < SS_SUCCESS) { 3148 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 3149 return FALSE; 3150 } 3151 3152 if (oconn > C_WF_REPORT_PARAMS) { 3153 if (nconn > C_CONNECTED && peer_state.conn <= C_CONNECTED && 3154 peer_state.disk != D_NEGOTIATING ) { 3155 /* we want resync, peer has not yet decided to sync... */ 3156 /* Nowadays only used when forcing a node into primary role and 3157 setting its disk to UpToDate with that */ 3158 drbd_send_uuids(mdev); 3159 drbd_send_state(mdev); 3160 } 3161 } 3162 3163 mdev->net_conf->want_lose = 0; 3164 3165 drbd_md_sync(mdev); /* update connected indicator, la_size, ... */ 3166 3167 return TRUE; 3168 } 3169 3170 static int receive_sync_uuid(struct drbd_conf *mdev, struct p_header *h) 3171 { 3172 struct p_rs_uuid *p = (struct p_rs_uuid *)h; 3173 3174 wait_event(mdev->misc_wait, 3175 mdev->state.conn == C_WF_SYNC_UUID || 3176 mdev->state.conn < C_CONNECTED || 3177 mdev->state.disk < D_NEGOTIATING); 3178 3179 /* D_ASSERT( mdev->state.conn == C_WF_SYNC_UUID ); */ 3180 3181 ERR_IF(h->length != (sizeof(*p)-sizeof(*h))) return FALSE; 3182 if (drbd_recv(mdev, h->payload, h->length) != h->length) 3183 return FALSE; 3184 3185 /* Here the _drbd_uuid_ functions are right, current should 3186 _not_ be rotated into the history */ 3187 if (get_ldev_if_state(mdev, D_NEGOTIATING)) { 3188 _drbd_uuid_set(mdev, UI_CURRENT, be64_to_cpu(p->uuid)); 3189 _drbd_uuid_set(mdev, UI_BITMAP, 0UL); 3190 3191 drbd_start_resync(mdev, C_SYNC_TARGET); 3192 3193 put_ldev(mdev); 3194 } else 3195 dev_err(DEV, "Ignoring SyncUUID packet!\n"); 3196 3197 return TRUE; 3198 } 3199 3200 enum receive_bitmap_ret { OK, DONE, FAILED }; 3201 3202 static enum receive_bitmap_ret 3203 receive_bitmap_plain(struct drbd_conf *mdev, struct p_header *h, 3204 unsigned long *buffer, struct bm_xfer_ctx *c) 3205 { 3206 unsigned num_words = min_t(size_t, BM_PACKET_WORDS, c->bm_words - c->word_offset); 3207 unsigned want = num_words * sizeof(long); 3208 3209 if (want != h->length) { 3210 dev_err(DEV, "%s:want (%u) != h->length (%u)\n", __func__, want, h->length); 3211 return FAILED; 3212 } 3213 if (want == 0) 3214 return DONE; 3215 if (drbd_recv(mdev, buffer, want) != want) 3216 return FAILED; 3217 3218 drbd_bm_merge_lel(mdev, c->word_offset, num_words, buffer); 3219 3220 c->word_offset += num_words; 3221 c->bit_offset = c->word_offset * BITS_PER_LONG; 3222 if (c->bit_offset > c->bm_bits) 3223 c->bit_offset = c->bm_bits; 3224 3225 return OK; 3226 } 3227 3228 static enum receive_bitmap_ret 3229 recv_bm_rle_bits(struct drbd_conf *mdev, 3230 struct p_compressed_bm *p, 3231 struct bm_xfer_ctx *c) 3232 { 3233 struct bitstream bs; 3234 u64 look_ahead; 3235 u64 rl; 3236 u64 tmp; 3237 unsigned long s = c->bit_offset; 3238 unsigned long e; 3239 int len = p->head.length - (sizeof(*p) - sizeof(p->head)); 3240 int toggle = DCBP_get_start(p); 3241 int have; 3242 int bits; 3243 3244 bitstream_init(&bs, p->code, len, DCBP_get_pad_bits(p)); 3245 3246 bits = bitstream_get_bits(&bs, &look_ahead, 64); 3247 if (bits < 0) 3248 return FAILED; 3249 3250 for (have = bits; have > 0; s += rl, toggle = !toggle) { 3251 bits = vli_decode_bits(&rl, look_ahead); 3252 if (bits <= 0) 3253 return FAILED; 3254 3255 if (toggle) { 3256 e = s + rl -1; 3257 if (e >= c->bm_bits) { 3258 dev_err(DEV, "bitmap overflow (e:%lu) while decoding bm RLE packet\n", e); 3259 return FAILED; 3260 } 3261 _drbd_bm_set_bits(mdev, s, e); 3262 } 3263 3264 if (have < bits) { 3265 dev_err(DEV, "bitmap decoding error: h:%d b:%d la:0x%08llx l:%u/%u\n", 3266 have, bits, look_ahead, 3267 (unsigned int)(bs.cur.b - p->code), 3268 (unsigned int)bs.buf_len); 3269 return FAILED; 3270 } 3271 look_ahead >>= bits; 3272 have -= bits; 3273 3274 bits = bitstream_get_bits(&bs, &tmp, 64 - have); 3275 if (bits < 0) 3276 return FAILED; 3277 look_ahead |= tmp << have; 3278 have += bits; 3279 } 3280 3281 c->bit_offset = s; 3282 bm_xfer_ctx_bit_to_word_offset(c); 3283 3284 return (s == c->bm_bits) ? DONE : OK; 3285 } 3286 3287 static enum receive_bitmap_ret 3288 decode_bitmap_c(struct drbd_conf *mdev, 3289 struct p_compressed_bm *p, 3290 struct bm_xfer_ctx *c) 3291 { 3292 if (DCBP_get_code(p) == RLE_VLI_Bits) 3293 return recv_bm_rle_bits(mdev, p, c); 3294 3295 /* other variants had been implemented for evaluation, 3296 * but have been dropped as this one turned out to be "best" 3297 * during all our tests. */ 3298 3299 dev_err(DEV, "receive_bitmap_c: unknown encoding %u\n", p->encoding); 3300 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR)); 3301 return FAILED; 3302 } 3303 3304 void INFO_bm_xfer_stats(struct drbd_conf *mdev, 3305 const char *direction, struct bm_xfer_ctx *c) 3306 { 3307 /* what would it take to transfer it "plaintext" */ 3308 unsigned plain = sizeof(struct p_header) * 3309 ((c->bm_words+BM_PACKET_WORDS-1)/BM_PACKET_WORDS+1) 3310 + c->bm_words * sizeof(long); 3311 unsigned total = c->bytes[0] + c->bytes[1]; 3312 unsigned r; 3313 3314 /* total can not be zero. but just in case: */ 3315 if (total == 0) 3316 return; 3317 3318 /* don't report if not compressed */ 3319 if (total >= plain) 3320 return; 3321 3322 /* total < plain. check for overflow, still */ 3323 r = (total > UINT_MAX/1000) ? (total / (plain/1000)) 3324 : (1000 * total / plain); 3325 3326 if (r > 1000) 3327 r = 1000; 3328 3329 r = 1000 - r; 3330 dev_info(DEV, "%s bitmap stats [Bytes(packets)]: plain %u(%u), RLE %u(%u), " 3331 "total %u; compression: %u.%u%%\n", 3332 direction, 3333 c->bytes[1], c->packets[1], 3334 c->bytes[0], c->packets[0], 3335 total, r/10, r % 10); 3336 } 3337 3338 /* Since we are processing the bitfield from lower addresses to higher, 3339 it does not matter if the process it in 32 bit chunks or 64 bit 3340 chunks as long as it is little endian. (Understand it as byte stream, 3341 beginning with the lowest byte...) If we would use big endian 3342 we would need to process it from the highest address to the lowest, 3343 in order to be agnostic to the 32 vs 64 bits issue. 3344 3345 returns 0 on failure, 1 if we successfully received it. */ 3346 static int receive_bitmap(struct drbd_conf *mdev, struct p_header *h) 3347 { 3348 struct bm_xfer_ctx c; 3349 void *buffer; 3350 enum receive_bitmap_ret ret; 3351 int ok = FALSE; 3352 3353 wait_event(mdev->misc_wait, !atomic_read(&mdev->ap_bio_cnt)); 3354 3355 drbd_bm_lock(mdev, "receive bitmap"); 3356 3357 /* maybe we should use some per thread scratch page, 3358 * and allocate that during initial device creation? */ 3359 buffer = (unsigned long *) __get_free_page(GFP_NOIO); 3360 if (!buffer) { 3361 dev_err(DEV, "failed to allocate one page buffer in %s\n", __func__); 3362 goto out; 3363 } 3364 3365 c = (struct bm_xfer_ctx) { 3366 .bm_bits = drbd_bm_bits(mdev), 3367 .bm_words = drbd_bm_words(mdev), 3368 }; 3369 3370 do { 3371 if (h->command == P_BITMAP) { 3372 ret = receive_bitmap_plain(mdev, h, buffer, &c); 3373 } else if (h->command == P_COMPRESSED_BITMAP) { 3374 /* MAYBE: sanity check that we speak proto >= 90, 3375 * and the feature is enabled! */ 3376 struct p_compressed_bm *p; 3377 3378 if (h->length > BM_PACKET_PAYLOAD_BYTES) { 3379 dev_err(DEV, "ReportCBitmap packet too large\n"); 3380 goto out; 3381 } 3382 /* use the page buff */ 3383 p = buffer; 3384 memcpy(p, h, sizeof(*h)); 3385 if (drbd_recv(mdev, p->head.payload, h->length) != h->length) 3386 goto out; 3387 if (p->head.length <= (sizeof(*p) - sizeof(p->head))) { 3388 dev_err(DEV, "ReportCBitmap packet too small (l:%u)\n", p->head.length); 3389 return FAILED; 3390 } 3391 ret = decode_bitmap_c(mdev, p, &c); 3392 } else { 3393 dev_warn(DEV, "receive_bitmap: h->command neither ReportBitMap nor ReportCBitMap (is 0x%x)", h->command); 3394 goto out; 3395 } 3396 3397 c.packets[h->command == P_BITMAP]++; 3398 c.bytes[h->command == P_BITMAP] += sizeof(struct p_header) + h->length; 3399 3400 if (ret != OK) 3401 break; 3402 3403 if (!drbd_recv_header(mdev, h)) 3404 goto out; 3405 } while (ret == OK); 3406 if (ret == FAILED) 3407 goto out; 3408 3409 INFO_bm_xfer_stats(mdev, "receive", &c); 3410 3411 if (mdev->state.conn == C_WF_BITMAP_T) { 3412 ok = !drbd_send_bitmap(mdev); 3413 if (!ok) 3414 goto out; 3415 /* Omit CS_ORDERED with this state transition to avoid deadlocks. */ 3416 ok = _drbd_request_state(mdev, NS(conn, C_WF_SYNC_UUID), CS_VERBOSE); 3417 D_ASSERT(ok == SS_SUCCESS); 3418 } else if (mdev->state.conn != C_WF_BITMAP_S) { 3419 /* admin may have requested C_DISCONNECTING, 3420 * other threads may have noticed network errors */ 3421 dev_info(DEV, "unexpected cstate (%s) in receive_bitmap\n", 3422 drbd_conn_str(mdev->state.conn)); 3423 } 3424 3425 ok = TRUE; 3426 out: 3427 drbd_bm_unlock(mdev); 3428 if (ok && mdev->state.conn == C_WF_BITMAP_S) 3429 drbd_start_resync(mdev, C_SYNC_SOURCE); 3430 free_page((unsigned long) buffer); 3431 return ok; 3432 } 3433 3434 static int receive_skip(struct drbd_conf *mdev, struct p_header *h) 3435 { 3436 /* TODO zero copy sink :) */ 3437 static char sink[128]; 3438 int size, want, r; 3439 3440 dev_warn(DEV, "skipping unknown optional packet type %d, l: %d!\n", 3441 h->command, h->length); 3442 3443 size = h->length; 3444 while (size > 0) { 3445 want = min_t(int, size, sizeof(sink)); 3446 r = drbd_recv(mdev, sink, want); 3447 ERR_IF(r <= 0) break; 3448 size -= r; 3449 } 3450 return size == 0; 3451 } 3452 3453 static int receive_UnplugRemote(struct drbd_conf *mdev, struct p_header *h) 3454 { 3455 if (mdev->state.disk >= D_INCONSISTENT) 3456 drbd_kick_lo(mdev); 3457 3458 /* Make sure we've acked all the TCP data associated 3459 * with the data requests being unplugged */ 3460 drbd_tcp_quickack(mdev->data.socket); 3461 3462 return TRUE; 3463 } 3464 3465 typedef int (*drbd_cmd_handler_f)(struct drbd_conf *, struct p_header *); 3466 3467 static drbd_cmd_handler_f drbd_default_handler[] = { 3468 [P_DATA] = receive_Data, 3469 [P_DATA_REPLY] = receive_DataReply, 3470 [P_RS_DATA_REPLY] = receive_RSDataReply, 3471 [P_BARRIER] = receive_Barrier, 3472 [P_BITMAP] = receive_bitmap, 3473 [P_COMPRESSED_BITMAP] = receive_bitmap, 3474 [P_UNPLUG_REMOTE] = receive_UnplugRemote, 3475 [P_DATA_REQUEST] = receive_DataRequest, 3476 [P_RS_DATA_REQUEST] = receive_DataRequest, 3477 [P_SYNC_PARAM] = receive_SyncParam, 3478 [P_SYNC_PARAM89] = receive_SyncParam, 3479 [P_PROTOCOL] = receive_protocol, 3480 [P_UUIDS] = receive_uuids, 3481 [P_SIZES] = receive_sizes, 3482 [P_STATE] = receive_state, 3483 [P_STATE_CHG_REQ] = receive_req_state, 3484 [P_SYNC_UUID] = receive_sync_uuid, 3485 [P_OV_REQUEST] = receive_DataRequest, 3486 [P_OV_REPLY] = receive_DataRequest, 3487 [P_CSUM_RS_REQUEST] = receive_DataRequest, 3488 /* anything missing from this table is in 3489 * the asender_tbl, see get_asender_cmd */ 3490 [P_MAX_CMD] = NULL, 3491 }; 3492 3493 static drbd_cmd_handler_f *drbd_cmd_handler = drbd_default_handler; 3494 static drbd_cmd_handler_f *drbd_opt_cmd_handler; 3495 3496 static void drbdd(struct drbd_conf *mdev) 3497 { 3498 drbd_cmd_handler_f handler; 3499 struct p_header *header = &mdev->data.rbuf.header; 3500 3501 while (get_t_state(&mdev->receiver) == Running) { 3502 drbd_thread_current_set_cpu(mdev); 3503 if (!drbd_recv_header(mdev, header)) { 3504 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR)); 3505 break; 3506 } 3507 3508 if (header->command < P_MAX_CMD) 3509 handler = drbd_cmd_handler[header->command]; 3510 else if (P_MAY_IGNORE < header->command 3511 && header->command < P_MAX_OPT_CMD) 3512 handler = drbd_opt_cmd_handler[header->command-P_MAY_IGNORE]; 3513 else if (header->command > P_MAX_OPT_CMD) 3514 handler = receive_skip; 3515 else 3516 handler = NULL; 3517 3518 if (unlikely(!handler)) { 3519 dev_err(DEV, "unknown packet type %d, l: %d!\n", 3520 header->command, header->length); 3521 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR)); 3522 break; 3523 } 3524 if (unlikely(!handler(mdev, header))) { 3525 dev_err(DEV, "error receiving %s, l: %d!\n", 3526 cmdname(header->command), header->length); 3527 drbd_force_state(mdev, NS(conn, C_PROTOCOL_ERROR)); 3528 break; 3529 } 3530 } 3531 } 3532 3533 static void drbd_fail_pending_reads(struct drbd_conf *mdev) 3534 { 3535 struct hlist_head *slot; 3536 struct hlist_node *pos; 3537 struct hlist_node *tmp; 3538 struct drbd_request *req; 3539 int i; 3540 3541 /* 3542 * Application READ requests 3543 */ 3544 spin_lock_irq(&mdev->req_lock); 3545 for (i = 0; i < APP_R_HSIZE; i++) { 3546 slot = mdev->app_reads_hash+i; 3547 hlist_for_each_entry_safe(req, pos, tmp, slot, colision) { 3548 /* it may (but should not any longer!) 3549 * be on the work queue; if that assert triggers, 3550 * we need to also grab the 3551 * spin_lock_irq(&mdev->data.work.q_lock); 3552 * and list_del_init here. */ 3553 D_ASSERT(list_empty(&req->w.list)); 3554 /* It would be nice to complete outside of spinlock. 3555 * But this is easier for now. */ 3556 _req_mod(req, connection_lost_while_pending); 3557 } 3558 } 3559 for (i = 0; i < APP_R_HSIZE; i++) 3560 if (!hlist_empty(mdev->app_reads_hash+i)) 3561 dev_warn(DEV, "ASSERT FAILED: app_reads_hash[%d].first: " 3562 "%p, should be NULL\n", i, mdev->app_reads_hash[i].first); 3563 3564 memset(mdev->app_reads_hash, 0, APP_R_HSIZE*sizeof(void *)); 3565 spin_unlock_irq(&mdev->req_lock); 3566 } 3567 3568 void drbd_flush_workqueue(struct drbd_conf *mdev) 3569 { 3570 struct drbd_wq_barrier barr; 3571 3572 barr.w.cb = w_prev_work_done; 3573 init_completion(&barr.done); 3574 drbd_queue_work(&mdev->data.work, &barr.w); 3575 wait_for_completion(&barr.done); 3576 } 3577 3578 static void drbd_disconnect(struct drbd_conf *mdev) 3579 { 3580 enum drbd_fencing_p fp; 3581 union drbd_state os, ns; 3582 int rv = SS_UNKNOWN_ERROR; 3583 unsigned int i; 3584 3585 if (mdev->state.conn == C_STANDALONE) 3586 return; 3587 if (mdev->state.conn >= C_WF_CONNECTION) 3588 dev_err(DEV, "ASSERT FAILED cstate = %s, expected < WFConnection\n", 3589 drbd_conn_str(mdev->state.conn)); 3590 3591 /* asender does not clean up anything. it must not interfere, either */ 3592 drbd_thread_stop(&mdev->asender); 3593 3594 mutex_lock(&mdev->data.mutex); 3595 drbd_free_sock(mdev); 3596 mutex_unlock(&mdev->data.mutex); 3597 3598 spin_lock_irq(&mdev->req_lock); 3599 _drbd_wait_ee_list_empty(mdev, &mdev->active_ee); 3600 _drbd_wait_ee_list_empty(mdev, &mdev->sync_ee); 3601 _drbd_wait_ee_list_empty(mdev, &mdev->read_ee); 3602 spin_unlock_irq(&mdev->req_lock); 3603 3604 /* We do not have data structures that would allow us to 3605 * get the rs_pending_cnt down to 0 again. 3606 * * On C_SYNC_TARGET we do not have any data structures describing 3607 * the pending RSDataRequest's we have sent. 3608 * * On C_SYNC_SOURCE there is no data structure that tracks 3609 * the P_RS_DATA_REPLY blocks that we sent to the SyncTarget. 3610 * And no, it is not the sum of the reference counts in the 3611 * resync_LRU. The resync_LRU tracks the whole operation including 3612 * the disk-IO, while the rs_pending_cnt only tracks the blocks 3613 * on the fly. */ 3614 drbd_rs_cancel_all(mdev); 3615 mdev->rs_total = 0; 3616 mdev->rs_failed = 0; 3617 atomic_set(&mdev->rs_pending_cnt, 0); 3618 wake_up(&mdev->misc_wait); 3619 3620 /* make sure syncer is stopped and w_resume_next_sg queued */ 3621 del_timer_sync(&mdev->resync_timer); 3622 set_bit(STOP_SYNC_TIMER, &mdev->flags); 3623 resync_timer_fn((unsigned long)mdev); 3624 3625 /* wait for all w_e_end_data_req, w_e_end_rsdata_req, w_send_barrier, 3626 * w_make_resync_request etc. which may still be on the worker queue 3627 * to be "canceled" */ 3628 drbd_flush_workqueue(mdev); 3629 3630 /* This also does reclaim_net_ee(). If we do this too early, we might 3631 * miss some resync ee and pages.*/ 3632 drbd_process_done_ee(mdev); 3633 3634 kfree(mdev->p_uuid); 3635 mdev->p_uuid = NULL; 3636 3637 if (!mdev->state.susp) 3638 tl_clear(mdev); 3639 3640 drbd_fail_pending_reads(mdev); 3641 3642 dev_info(DEV, "Connection closed\n"); 3643 3644 drbd_md_sync(mdev); 3645 3646 fp = FP_DONT_CARE; 3647 if (get_ldev(mdev)) { 3648 fp = mdev->ldev->dc.fencing; 3649 put_ldev(mdev); 3650 } 3651 3652 if (mdev->state.role == R_PRIMARY) { 3653 if (fp >= FP_RESOURCE && mdev->state.pdsk >= D_UNKNOWN) { 3654 enum drbd_disk_state nps = drbd_try_outdate_peer(mdev); 3655 drbd_request_state(mdev, NS(pdsk, nps)); 3656 } 3657 } 3658 3659 spin_lock_irq(&mdev->req_lock); 3660 os = mdev->state; 3661 if (os.conn >= C_UNCONNECTED) { 3662 /* Do not restart in case we are C_DISCONNECTING */ 3663 ns = os; 3664 ns.conn = C_UNCONNECTED; 3665 rv = _drbd_set_state(mdev, ns, CS_VERBOSE, NULL); 3666 } 3667 spin_unlock_irq(&mdev->req_lock); 3668 3669 if (os.conn == C_DISCONNECTING) { 3670 struct hlist_head *h; 3671 wait_event(mdev->misc_wait, atomic_read(&mdev->net_cnt) == 0); 3672 3673 /* we must not free the tl_hash 3674 * while application io is still on the fly */ 3675 wait_event(mdev->misc_wait, atomic_read(&mdev->ap_bio_cnt) == 0); 3676 3677 spin_lock_irq(&mdev->req_lock); 3678 /* paranoia code */ 3679 for (h = mdev->ee_hash; h < mdev->ee_hash + mdev->ee_hash_s; h++) 3680 if (h->first) 3681 dev_err(DEV, "ASSERT FAILED ee_hash[%u].first == %p, expected NULL\n", 3682 (int)(h - mdev->ee_hash), h->first); 3683 kfree(mdev->ee_hash); 3684 mdev->ee_hash = NULL; 3685 mdev->ee_hash_s = 0; 3686 3687 /* paranoia code */ 3688 for (h = mdev->tl_hash; h < mdev->tl_hash + mdev->tl_hash_s; h++) 3689 if (h->first) 3690 dev_err(DEV, "ASSERT FAILED tl_hash[%u] == %p, expected NULL\n", 3691 (int)(h - mdev->tl_hash), h->first); 3692 kfree(mdev->tl_hash); 3693 mdev->tl_hash = NULL; 3694 mdev->tl_hash_s = 0; 3695 spin_unlock_irq(&mdev->req_lock); 3696 3697 crypto_free_hash(mdev->cram_hmac_tfm); 3698 mdev->cram_hmac_tfm = NULL; 3699 3700 kfree(mdev->net_conf); 3701 mdev->net_conf = NULL; 3702 drbd_request_state(mdev, NS(conn, C_STANDALONE)); 3703 } 3704 3705 /* tcp_close and release of sendpage pages can be deferred. I don't 3706 * want to use SO_LINGER, because apparently it can be deferred for 3707 * more than 20 seconds (longest time I checked). 3708 * 3709 * Actually we don't care for exactly when the network stack does its 3710 * put_page(), but release our reference on these pages right here. 3711 */ 3712 i = drbd_release_ee(mdev, &mdev->net_ee); 3713 if (i) 3714 dev_info(DEV, "net_ee not empty, killed %u entries\n", i); 3715 i = atomic_read(&mdev->pp_in_use); 3716 if (i) 3717 dev_info(DEV, "pp_in_use = %u, expected 0\n", i); 3718 3719 D_ASSERT(list_empty(&mdev->read_ee)); 3720 D_ASSERT(list_empty(&mdev->active_ee)); 3721 D_ASSERT(list_empty(&mdev->sync_ee)); 3722 D_ASSERT(list_empty(&mdev->done_ee)); 3723 3724 /* ok, no more ee's on the fly, it is safe to reset the epoch_size */ 3725 atomic_set(&mdev->current_epoch->epoch_size, 0); 3726 D_ASSERT(list_empty(&mdev->current_epoch->list)); 3727 } 3728 3729 /* 3730 * We support PRO_VERSION_MIN to PRO_VERSION_MAX. The protocol version 3731 * we can agree on is stored in agreed_pro_version. 3732 * 3733 * feature flags and the reserved array should be enough room for future 3734 * enhancements of the handshake protocol, and possible plugins... 3735 * 3736 * for now, they are expected to be zero, but ignored. 3737 */ 3738 static int drbd_send_handshake(struct drbd_conf *mdev) 3739 { 3740 /* ASSERT current == mdev->receiver ... */ 3741 struct p_handshake *p = &mdev->data.sbuf.handshake; 3742 int ok; 3743 3744 if (mutex_lock_interruptible(&mdev->data.mutex)) { 3745 dev_err(DEV, "interrupted during initial handshake\n"); 3746 return 0; /* interrupted. not ok. */ 3747 } 3748 3749 if (mdev->data.socket == NULL) { 3750 mutex_unlock(&mdev->data.mutex); 3751 return 0; 3752 } 3753 3754 memset(p, 0, sizeof(*p)); 3755 p->protocol_min = cpu_to_be32(PRO_VERSION_MIN); 3756 p->protocol_max = cpu_to_be32(PRO_VERSION_MAX); 3757 ok = _drbd_send_cmd( mdev, mdev->data.socket, P_HAND_SHAKE, 3758 (struct p_header *)p, sizeof(*p), 0 ); 3759 mutex_unlock(&mdev->data.mutex); 3760 return ok; 3761 } 3762 3763 /* 3764 * return values: 3765 * 1 yes, we have a valid connection 3766 * 0 oops, did not work out, please try again 3767 * -1 peer talks different language, 3768 * no point in trying again, please go standalone. 3769 */ 3770 static int drbd_do_handshake(struct drbd_conf *mdev) 3771 { 3772 /* ASSERT current == mdev->receiver ... */ 3773 struct p_handshake *p = &mdev->data.rbuf.handshake; 3774 const int expect = sizeof(struct p_handshake) 3775 -sizeof(struct p_header); 3776 int rv; 3777 3778 rv = drbd_send_handshake(mdev); 3779 if (!rv) 3780 return 0; 3781 3782 rv = drbd_recv_header(mdev, &p->head); 3783 if (!rv) 3784 return 0; 3785 3786 if (p->head.command != P_HAND_SHAKE) { 3787 dev_err(DEV, "expected HandShake packet, received: %s (0x%04x)\n", 3788 cmdname(p->head.command), p->head.command); 3789 return -1; 3790 } 3791 3792 if (p->head.length != expect) { 3793 dev_err(DEV, "expected HandShake length: %u, received: %u\n", 3794 expect, p->head.length); 3795 return -1; 3796 } 3797 3798 rv = drbd_recv(mdev, &p->head.payload, expect); 3799 3800 if (rv != expect) { 3801 dev_err(DEV, "short read receiving handshake packet: l=%u\n", rv); 3802 return 0; 3803 } 3804 3805 p->protocol_min = be32_to_cpu(p->protocol_min); 3806 p->protocol_max = be32_to_cpu(p->protocol_max); 3807 if (p->protocol_max == 0) 3808 p->protocol_max = p->protocol_min; 3809 3810 if (PRO_VERSION_MAX < p->protocol_min || 3811 PRO_VERSION_MIN > p->protocol_max) 3812 goto incompat; 3813 3814 mdev->agreed_pro_version = min_t(int, PRO_VERSION_MAX, p->protocol_max); 3815 3816 dev_info(DEV, "Handshake successful: " 3817 "Agreed network protocol version %d\n", mdev->agreed_pro_version); 3818 3819 return 1; 3820 3821 incompat: 3822 dev_err(DEV, "incompatible DRBD dialects: " 3823 "I support %d-%d, peer supports %d-%d\n", 3824 PRO_VERSION_MIN, PRO_VERSION_MAX, 3825 p->protocol_min, p->protocol_max); 3826 return -1; 3827 } 3828 3829 #if !defined(CONFIG_CRYPTO_HMAC) && !defined(CONFIG_CRYPTO_HMAC_MODULE) 3830 static int drbd_do_auth(struct drbd_conf *mdev) 3831 { 3832 dev_err(DEV, "This kernel was build without CONFIG_CRYPTO_HMAC.\n"); 3833 dev_err(DEV, "You need to disable 'cram-hmac-alg' in drbd.conf.\n"); 3834 return 0; 3835 } 3836 #else 3837 #define CHALLENGE_LEN 64 3838 static int drbd_do_auth(struct drbd_conf *mdev) 3839 { 3840 char my_challenge[CHALLENGE_LEN]; /* 64 Bytes... */ 3841 struct scatterlist sg; 3842 char *response = NULL; 3843 char *right_response = NULL; 3844 char *peers_ch = NULL; 3845 struct p_header p; 3846 unsigned int key_len = strlen(mdev->net_conf->shared_secret); 3847 unsigned int resp_size; 3848 struct hash_desc desc; 3849 int rv; 3850 3851 desc.tfm = mdev->cram_hmac_tfm; 3852 desc.flags = 0; 3853 3854 rv = crypto_hash_setkey(mdev->cram_hmac_tfm, 3855 (u8 *)mdev->net_conf->shared_secret, key_len); 3856 if (rv) { 3857 dev_err(DEV, "crypto_hash_setkey() failed with %d\n", rv); 3858 rv = 0; 3859 goto fail; 3860 } 3861 3862 get_random_bytes(my_challenge, CHALLENGE_LEN); 3863 3864 rv = drbd_send_cmd2(mdev, P_AUTH_CHALLENGE, my_challenge, CHALLENGE_LEN); 3865 if (!rv) 3866 goto fail; 3867 3868 rv = drbd_recv_header(mdev, &p); 3869 if (!rv) 3870 goto fail; 3871 3872 if (p.command != P_AUTH_CHALLENGE) { 3873 dev_err(DEV, "expected AuthChallenge packet, received: %s (0x%04x)\n", 3874 cmdname(p.command), p.command); 3875 rv = 0; 3876 goto fail; 3877 } 3878 3879 if (p.length > CHALLENGE_LEN*2) { 3880 dev_err(DEV, "expected AuthChallenge payload too big.\n"); 3881 rv = 0; 3882 goto fail; 3883 } 3884 3885 peers_ch = kmalloc(p.length, GFP_NOIO); 3886 if (peers_ch == NULL) { 3887 dev_err(DEV, "kmalloc of peers_ch failed\n"); 3888 rv = 0; 3889 goto fail; 3890 } 3891 3892 rv = drbd_recv(mdev, peers_ch, p.length); 3893 3894 if (rv != p.length) { 3895 dev_err(DEV, "short read AuthChallenge: l=%u\n", rv); 3896 rv = 0; 3897 goto fail; 3898 } 3899 3900 resp_size = crypto_hash_digestsize(mdev->cram_hmac_tfm); 3901 response = kmalloc(resp_size, GFP_NOIO); 3902 if (response == NULL) { 3903 dev_err(DEV, "kmalloc of response failed\n"); 3904 rv = 0; 3905 goto fail; 3906 } 3907 3908 sg_init_table(&sg, 1); 3909 sg_set_buf(&sg, peers_ch, p.length); 3910 3911 rv = crypto_hash_digest(&desc, &sg, sg.length, response); 3912 if (rv) { 3913 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv); 3914 rv = 0; 3915 goto fail; 3916 } 3917 3918 rv = drbd_send_cmd2(mdev, P_AUTH_RESPONSE, response, resp_size); 3919 if (!rv) 3920 goto fail; 3921 3922 rv = drbd_recv_header(mdev, &p); 3923 if (!rv) 3924 goto fail; 3925 3926 if (p.command != P_AUTH_RESPONSE) { 3927 dev_err(DEV, "expected AuthResponse packet, received: %s (0x%04x)\n", 3928 cmdname(p.command), p.command); 3929 rv = 0; 3930 goto fail; 3931 } 3932 3933 if (p.length != resp_size) { 3934 dev_err(DEV, "expected AuthResponse payload of wrong size\n"); 3935 rv = 0; 3936 goto fail; 3937 } 3938 3939 rv = drbd_recv(mdev, response , resp_size); 3940 3941 if (rv != resp_size) { 3942 dev_err(DEV, "short read receiving AuthResponse: l=%u\n", rv); 3943 rv = 0; 3944 goto fail; 3945 } 3946 3947 right_response = kmalloc(resp_size, GFP_NOIO); 3948 if (response == NULL) { 3949 dev_err(DEV, "kmalloc of right_response failed\n"); 3950 rv = 0; 3951 goto fail; 3952 } 3953 3954 sg_set_buf(&sg, my_challenge, CHALLENGE_LEN); 3955 3956 rv = crypto_hash_digest(&desc, &sg, sg.length, right_response); 3957 if (rv) { 3958 dev_err(DEV, "crypto_hash_digest() failed with %d\n", rv); 3959 rv = 0; 3960 goto fail; 3961 } 3962 3963 rv = !memcmp(response, right_response, resp_size); 3964 3965 if (rv) 3966 dev_info(DEV, "Peer authenticated using %d bytes of '%s' HMAC\n", 3967 resp_size, mdev->net_conf->cram_hmac_alg); 3968 3969 fail: 3970 kfree(peers_ch); 3971 kfree(response); 3972 kfree(right_response); 3973 3974 return rv; 3975 } 3976 #endif 3977 3978 int drbdd_init(struct drbd_thread *thi) 3979 { 3980 struct drbd_conf *mdev = thi->mdev; 3981 unsigned int minor = mdev_to_minor(mdev); 3982 int h; 3983 3984 sprintf(current->comm, "drbd%d_receiver", minor); 3985 3986 dev_info(DEV, "receiver (re)started\n"); 3987 3988 do { 3989 h = drbd_connect(mdev); 3990 if (h == 0) { 3991 drbd_disconnect(mdev); 3992 __set_current_state(TASK_INTERRUPTIBLE); 3993 schedule_timeout(HZ); 3994 } 3995 if (h == -1) { 3996 dev_warn(DEV, "Discarding network configuration.\n"); 3997 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 3998 } 3999 } while (h == 0); 4000 4001 if (h > 0) { 4002 if (get_net_conf(mdev)) { 4003 drbdd(mdev); 4004 put_net_conf(mdev); 4005 } 4006 } 4007 4008 drbd_disconnect(mdev); 4009 4010 dev_info(DEV, "receiver terminated\n"); 4011 return 0; 4012 } 4013 4014 /* ********* acknowledge sender ******** */ 4015 4016 static int got_RqSReply(struct drbd_conf *mdev, struct p_header *h) 4017 { 4018 struct p_req_state_reply *p = (struct p_req_state_reply *)h; 4019 4020 int retcode = be32_to_cpu(p->retcode); 4021 4022 if (retcode >= SS_SUCCESS) { 4023 set_bit(CL_ST_CHG_SUCCESS, &mdev->flags); 4024 } else { 4025 set_bit(CL_ST_CHG_FAIL, &mdev->flags); 4026 dev_err(DEV, "Requested state change failed by peer: %s (%d)\n", 4027 drbd_set_st_err_str(retcode), retcode); 4028 } 4029 wake_up(&mdev->state_wait); 4030 4031 return TRUE; 4032 } 4033 4034 static int got_Ping(struct drbd_conf *mdev, struct p_header *h) 4035 { 4036 return drbd_send_ping_ack(mdev); 4037 4038 } 4039 4040 static int got_PingAck(struct drbd_conf *mdev, struct p_header *h) 4041 { 4042 /* restore idle timeout */ 4043 mdev->meta.socket->sk->sk_rcvtimeo = mdev->net_conf->ping_int*HZ; 4044 4045 return TRUE; 4046 } 4047 4048 static int got_IsInSync(struct drbd_conf *mdev, struct p_header *h) 4049 { 4050 struct p_block_ack *p = (struct p_block_ack *)h; 4051 sector_t sector = be64_to_cpu(p->sector); 4052 int blksize = be32_to_cpu(p->blksize); 4053 4054 D_ASSERT(mdev->agreed_pro_version >= 89); 4055 4056 update_peer_seq(mdev, be32_to_cpu(p->seq_num)); 4057 4058 drbd_rs_complete_io(mdev, sector); 4059 drbd_set_in_sync(mdev, sector, blksize); 4060 /* rs_same_csums is supposed to count in units of BM_BLOCK_SIZE */ 4061 mdev->rs_same_csum += (blksize >> BM_BLOCK_SHIFT); 4062 dec_rs_pending(mdev); 4063 4064 return TRUE; 4065 } 4066 4067 /* when we receive the ACK for a write request, 4068 * verify that we actually know about it */ 4069 static struct drbd_request *_ack_id_to_req(struct drbd_conf *mdev, 4070 u64 id, sector_t sector) 4071 { 4072 struct hlist_head *slot = tl_hash_slot(mdev, sector); 4073 struct hlist_node *n; 4074 struct drbd_request *req; 4075 4076 hlist_for_each_entry(req, n, slot, colision) { 4077 if ((unsigned long)req == (unsigned long)id) { 4078 if (req->sector != sector) { 4079 dev_err(DEV, "_ack_id_to_req: found req %p but it has " 4080 "wrong sector (%llus versus %llus)\n", req, 4081 (unsigned long long)req->sector, 4082 (unsigned long long)sector); 4083 break; 4084 } 4085 return req; 4086 } 4087 } 4088 dev_err(DEV, "_ack_id_to_req: failed to find req %p, sector %llus in list\n", 4089 (void *)(unsigned long)id, (unsigned long long)sector); 4090 return NULL; 4091 } 4092 4093 typedef struct drbd_request *(req_validator_fn) 4094 (struct drbd_conf *mdev, u64 id, sector_t sector); 4095 4096 static int validate_req_change_req_state(struct drbd_conf *mdev, 4097 u64 id, sector_t sector, req_validator_fn validator, 4098 const char *func, enum drbd_req_event what) 4099 { 4100 struct drbd_request *req; 4101 struct bio_and_error m; 4102 4103 spin_lock_irq(&mdev->req_lock); 4104 req = validator(mdev, id, sector); 4105 if (unlikely(!req)) { 4106 spin_unlock_irq(&mdev->req_lock); 4107 dev_err(DEV, "%s: got a corrupt block_id/sector pair\n", func); 4108 return FALSE; 4109 } 4110 __req_mod(req, what, &m); 4111 spin_unlock_irq(&mdev->req_lock); 4112 4113 if (m.bio) 4114 complete_master_bio(mdev, &m); 4115 return TRUE; 4116 } 4117 4118 static int got_BlockAck(struct drbd_conf *mdev, struct p_header *h) 4119 { 4120 struct p_block_ack *p = (struct p_block_ack *)h; 4121 sector_t sector = be64_to_cpu(p->sector); 4122 int blksize = be32_to_cpu(p->blksize); 4123 enum drbd_req_event what; 4124 4125 update_peer_seq(mdev, be32_to_cpu(p->seq_num)); 4126 4127 if (is_syncer_block_id(p->block_id)) { 4128 drbd_set_in_sync(mdev, sector, blksize); 4129 dec_rs_pending(mdev); 4130 return TRUE; 4131 } 4132 switch (be16_to_cpu(h->command)) { 4133 case P_RS_WRITE_ACK: 4134 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C); 4135 what = write_acked_by_peer_and_sis; 4136 break; 4137 case P_WRITE_ACK: 4138 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C); 4139 what = write_acked_by_peer; 4140 break; 4141 case P_RECV_ACK: 4142 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_B); 4143 what = recv_acked_by_peer; 4144 break; 4145 case P_DISCARD_ACK: 4146 D_ASSERT(mdev->net_conf->wire_protocol == DRBD_PROT_C); 4147 what = conflict_discarded_by_peer; 4148 break; 4149 default: 4150 D_ASSERT(0); 4151 return FALSE; 4152 } 4153 4154 return validate_req_change_req_state(mdev, p->block_id, sector, 4155 _ack_id_to_req, __func__ , what); 4156 } 4157 4158 static int got_NegAck(struct drbd_conf *mdev, struct p_header *h) 4159 { 4160 struct p_block_ack *p = (struct p_block_ack *)h; 4161 sector_t sector = be64_to_cpu(p->sector); 4162 4163 if (__ratelimit(&drbd_ratelimit_state)) 4164 dev_warn(DEV, "Got NegAck packet. Peer is in troubles?\n"); 4165 4166 update_peer_seq(mdev, be32_to_cpu(p->seq_num)); 4167 4168 if (is_syncer_block_id(p->block_id)) { 4169 int size = be32_to_cpu(p->blksize); 4170 dec_rs_pending(mdev); 4171 drbd_rs_failed_io(mdev, sector, size); 4172 return TRUE; 4173 } 4174 return validate_req_change_req_state(mdev, p->block_id, sector, 4175 _ack_id_to_req, __func__ , neg_acked); 4176 } 4177 4178 static int got_NegDReply(struct drbd_conf *mdev, struct p_header *h) 4179 { 4180 struct p_block_ack *p = (struct p_block_ack *)h; 4181 sector_t sector = be64_to_cpu(p->sector); 4182 4183 update_peer_seq(mdev, be32_to_cpu(p->seq_num)); 4184 dev_err(DEV, "Got NegDReply; Sector %llus, len %u; Fail original request.\n", 4185 (unsigned long long)sector, be32_to_cpu(p->blksize)); 4186 4187 return validate_req_change_req_state(mdev, p->block_id, sector, 4188 _ar_id_to_req, __func__ , neg_acked); 4189 } 4190 4191 static int got_NegRSDReply(struct drbd_conf *mdev, struct p_header *h) 4192 { 4193 sector_t sector; 4194 int size; 4195 struct p_block_ack *p = (struct p_block_ack *)h; 4196 4197 sector = be64_to_cpu(p->sector); 4198 size = be32_to_cpu(p->blksize); 4199 D_ASSERT(p->block_id == ID_SYNCER); 4200 4201 update_peer_seq(mdev, be32_to_cpu(p->seq_num)); 4202 4203 dec_rs_pending(mdev); 4204 4205 if (get_ldev_if_state(mdev, D_FAILED)) { 4206 drbd_rs_complete_io(mdev, sector); 4207 drbd_rs_failed_io(mdev, sector, size); 4208 put_ldev(mdev); 4209 } 4210 4211 return TRUE; 4212 } 4213 4214 static int got_BarrierAck(struct drbd_conf *mdev, struct p_header *h) 4215 { 4216 struct p_barrier_ack *p = (struct p_barrier_ack *)h; 4217 4218 tl_release(mdev, p->barrier, be32_to_cpu(p->set_size)); 4219 4220 return TRUE; 4221 } 4222 4223 static int got_OVResult(struct drbd_conf *mdev, struct p_header *h) 4224 { 4225 struct p_block_ack *p = (struct p_block_ack *)h; 4226 struct drbd_work *w; 4227 sector_t sector; 4228 int size; 4229 4230 sector = be64_to_cpu(p->sector); 4231 size = be32_to_cpu(p->blksize); 4232 4233 update_peer_seq(mdev, be32_to_cpu(p->seq_num)); 4234 4235 if (be64_to_cpu(p->block_id) == ID_OUT_OF_SYNC) 4236 drbd_ov_oos_found(mdev, sector, size); 4237 else 4238 ov_oos_print(mdev); 4239 4240 drbd_rs_complete_io(mdev, sector); 4241 dec_rs_pending(mdev); 4242 4243 if (--mdev->ov_left == 0) { 4244 w = kmalloc(sizeof(*w), GFP_NOIO); 4245 if (w) { 4246 w->cb = w_ov_finished; 4247 drbd_queue_work_front(&mdev->data.work, w); 4248 } else { 4249 dev_err(DEV, "kmalloc(w) failed."); 4250 ov_oos_print(mdev); 4251 drbd_resync_finished(mdev); 4252 } 4253 } 4254 return TRUE; 4255 } 4256 4257 struct asender_cmd { 4258 size_t pkt_size; 4259 int (*process)(struct drbd_conf *mdev, struct p_header *h); 4260 }; 4261 4262 static struct asender_cmd *get_asender_cmd(int cmd) 4263 { 4264 static struct asender_cmd asender_tbl[] = { 4265 /* anything missing from this table is in 4266 * the drbd_cmd_handler (drbd_default_handler) table, 4267 * see the beginning of drbdd() */ 4268 [P_PING] = { sizeof(struct p_header), got_Ping }, 4269 [P_PING_ACK] = { sizeof(struct p_header), got_PingAck }, 4270 [P_RECV_ACK] = { sizeof(struct p_block_ack), got_BlockAck }, 4271 [P_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck }, 4272 [P_RS_WRITE_ACK] = { sizeof(struct p_block_ack), got_BlockAck }, 4273 [P_DISCARD_ACK] = { sizeof(struct p_block_ack), got_BlockAck }, 4274 [P_NEG_ACK] = { sizeof(struct p_block_ack), got_NegAck }, 4275 [P_NEG_DREPLY] = { sizeof(struct p_block_ack), got_NegDReply }, 4276 [P_NEG_RS_DREPLY] = { sizeof(struct p_block_ack), got_NegRSDReply}, 4277 [P_OV_RESULT] = { sizeof(struct p_block_ack), got_OVResult }, 4278 [P_BARRIER_ACK] = { sizeof(struct p_barrier_ack), got_BarrierAck }, 4279 [P_STATE_CHG_REPLY] = { sizeof(struct p_req_state_reply), got_RqSReply }, 4280 [P_RS_IS_IN_SYNC] = { sizeof(struct p_block_ack), got_IsInSync }, 4281 [P_MAX_CMD] = { 0, NULL }, 4282 }; 4283 if (cmd > P_MAX_CMD || asender_tbl[cmd].process == NULL) 4284 return NULL; 4285 return &asender_tbl[cmd]; 4286 } 4287 4288 int drbd_asender(struct drbd_thread *thi) 4289 { 4290 struct drbd_conf *mdev = thi->mdev; 4291 struct p_header *h = &mdev->meta.rbuf.header; 4292 struct asender_cmd *cmd = NULL; 4293 4294 int rv, len; 4295 void *buf = h; 4296 int received = 0; 4297 int expect = sizeof(struct p_header); 4298 int empty; 4299 4300 sprintf(current->comm, "drbd%d_asender", mdev_to_minor(mdev)); 4301 4302 current->policy = SCHED_RR; /* Make this a realtime task! */ 4303 current->rt_priority = 2; /* more important than all other tasks */ 4304 4305 while (get_t_state(thi) == Running) { 4306 drbd_thread_current_set_cpu(mdev); 4307 if (test_and_clear_bit(SEND_PING, &mdev->flags)) { 4308 ERR_IF(!drbd_send_ping(mdev)) goto reconnect; 4309 mdev->meta.socket->sk->sk_rcvtimeo = 4310 mdev->net_conf->ping_timeo*HZ/10; 4311 } 4312 4313 /* conditionally cork; 4314 * it may hurt latency if we cork without much to send */ 4315 if (!mdev->net_conf->no_cork && 4316 3 < atomic_read(&mdev->unacked_cnt)) 4317 drbd_tcp_cork(mdev->meta.socket); 4318 while (1) { 4319 clear_bit(SIGNAL_ASENDER, &mdev->flags); 4320 flush_signals(current); 4321 if (!drbd_process_done_ee(mdev)) { 4322 dev_err(DEV, "process_done_ee() = NOT_OK\n"); 4323 goto reconnect; 4324 } 4325 /* to avoid race with newly queued ACKs */ 4326 set_bit(SIGNAL_ASENDER, &mdev->flags); 4327 spin_lock_irq(&mdev->req_lock); 4328 empty = list_empty(&mdev->done_ee); 4329 spin_unlock_irq(&mdev->req_lock); 4330 /* new ack may have been queued right here, 4331 * but then there is also a signal pending, 4332 * and we start over... */ 4333 if (empty) 4334 break; 4335 } 4336 /* but unconditionally uncork unless disabled */ 4337 if (!mdev->net_conf->no_cork) 4338 drbd_tcp_uncork(mdev->meta.socket); 4339 4340 /* short circuit, recv_msg would return EINTR anyways. */ 4341 if (signal_pending(current)) 4342 continue; 4343 4344 rv = drbd_recv_short(mdev, mdev->meta.socket, 4345 buf, expect-received, 0); 4346 clear_bit(SIGNAL_ASENDER, &mdev->flags); 4347 4348 flush_signals(current); 4349 4350 /* Note: 4351 * -EINTR (on meta) we got a signal 4352 * -EAGAIN (on meta) rcvtimeo expired 4353 * -ECONNRESET other side closed the connection 4354 * -ERESTARTSYS (on data) we got a signal 4355 * rv < 0 other than above: unexpected error! 4356 * rv == expected: full header or command 4357 * rv < expected: "woken" by signal during receive 4358 * rv == 0 : "connection shut down by peer" 4359 */ 4360 if (likely(rv > 0)) { 4361 received += rv; 4362 buf += rv; 4363 } else if (rv == 0) { 4364 dev_err(DEV, "meta connection shut down by peer.\n"); 4365 goto reconnect; 4366 } else if (rv == -EAGAIN) { 4367 if (mdev->meta.socket->sk->sk_rcvtimeo == 4368 mdev->net_conf->ping_timeo*HZ/10) { 4369 dev_err(DEV, "PingAck did not arrive in time.\n"); 4370 goto reconnect; 4371 } 4372 set_bit(SEND_PING, &mdev->flags); 4373 continue; 4374 } else if (rv == -EINTR) { 4375 continue; 4376 } else { 4377 dev_err(DEV, "sock_recvmsg returned %d\n", rv); 4378 goto reconnect; 4379 } 4380 4381 if (received == expect && cmd == NULL) { 4382 if (unlikely(h->magic != BE_DRBD_MAGIC)) { 4383 dev_err(DEV, "magic?? on meta m: 0x%lx c: %d l: %d\n", 4384 (long)be32_to_cpu(h->magic), 4385 h->command, h->length); 4386 goto reconnect; 4387 } 4388 cmd = get_asender_cmd(be16_to_cpu(h->command)); 4389 len = be16_to_cpu(h->length); 4390 if (unlikely(cmd == NULL)) { 4391 dev_err(DEV, "unknown command?? on meta m: 0x%lx c: %d l: %d\n", 4392 (long)be32_to_cpu(h->magic), 4393 h->command, h->length); 4394 goto disconnect; 4395 } 4396 expect = cmd->pkt_size; 4397 ERR_IF(len != expect-sizeof(struct p_header)) 4398 goto reconnect; 4399 } 4400 if (received == expect) { 4401 D_ASSERT(cmd != NULL); 4402 if (!cmd->process(mdev, h)) 4403 goto reconnect; 4404 4405 buf = h; 4406 received = 0; 4407 expect = sizeof(struct p_header); 4408 cmd = NULL; 4409 } 4410 } 4411 4412 if (0) { 4413 reconnect: 4414 drbd_force_state(mdev, NS(conn, C_NETWORK_FAILURE)); 4415 } 4416 if (0) { 4417 disconnect: 4418 drbd_force_state(mdev, NS(conn, C_DISCONNECTING)); 4419 } 4420 clear_bit(SIGNAL_ASENDER, &mdev->flags); 4421 4422 D_ASSERT(mdev->state.conn < C_CONNECTED); 4423 dev_info(DEV, "asender terminated\n"); 4424 4425 return 0; 4426 } 4427