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