1 #include <linux/ceph/ceph_debug.h> 2 3 #include <linux/sort.h> 4 #include <linux/slab.h> 5 6 #include "super.h" 7 #include "mds_client.h" 8 9 #include <linux/ceph/decode.h> 10 11 /* 12 * Snapshots in ceph are driven in large part by cooperation from the 13 * client. In contrast to local file systems or file servers that 14 * implement snapshots at a single point in the system, ceph's 15 * distributed access to storage requires clients to help decide 16 * whether a write logically occurs before or after a recently created 17 * snapshot. 18 * 19 * This provides a perfect instantanous client-wide snapshot. Between 20 * clients, however, snapshots may appear to be applied at slightly 21 * different points in time, depending on delays in delivering the 22 * snapshot notification. 23 * 24 * Snapshots are _not_ file system-wide. Instead, each snapshot 25 * applies to the subdirectory nested beneath some directory. This 26 * effectively divides the hierarchy into multiple "realms," where all 27 * of the files contained by each realm share the same set of 28 * snapshots. An individual realm's snap set contains snapshots 29 * explicitly created on that realm, as well as any snaps in its 30 * parent's snap set _after_ the point at which the parent became it's 31 * parent (due to, say, a rename). Similarly, snaps from prior parents 32 * during the time intervals during which they were the parent are included. 33 * 34 * The client is spared most of this detail, fortunately... it must only 35 * maintains a hierarchy of realms reflecting the current parent/child 36 * realm relationship, and for each realm has an explicit list of snaps 37 * inherited from prior parents. 38 * 39 * A snap_realm struct is maintained for realms containing every inode 40 * with an open cap in the system. (The needed snap realm information is 41 * provided by the MDS whenever a cap is issued, i.e., on open.) A 'seq' 42 * version number is used to ensure that as realm parameters change (new 43 * snapshot, new parent, etc.) the client's realm hierarchy is updated. 44 * 45 * The realm hierarchy drives the generation of a 'snap context' for each 46 * realm, which simply lists the resulting set of snaps for the realm. This 47 * is attached to any writes sent to OSDs. 48 */ 49 /* 50 * Unfortunately error handling is a bit mixed here. If we get a snap 51 * update, but don't have enough memory to update our realm hierarchy, 52 * it's not clear what we can do about it (besides complaining to the 53 * console). 54 */ 55 56 57 /* 58 * increase ref count for the realm 59 * 60 * caller must hold snap_rwsem for write. 61 */ 62 void ceph_get_snap_realm(struct ceph_mds_client *mdsc, 63 struct ceph_snap_realm *realm) 64 { 65 dout("get_realm %p %d -> %d\n", realm, 66 atomic_read(&realm->nref), atomic_read(&realm->nref)+1); 67 /* 68 * since we _only_ increment realm refs or empty the empty 69 * list with snap_rwsem held, adjusting the empty list here is 70 * safe. we do need to protect against concurrent empty list 71 * additions, however. 72 */ 73 if (atomic_inc_return(&realm->nref) == 1) { 74 spin_lock(&mdsc->snap_empty_lock); 75 list_del_init(&realm->empty_item); 76 spin_unlock(&mdsc->snap_empty_lock); 77 } 78 } 79 80 static void __insert_snap_realm(struct rb_root *root, 81 struct ceph_snap_realm *new) 82 { 83 struct rb_node **p = &root->rb_node; 84 struct rb_node *parent = NULL; 85 struct ceph_snap_realm *r = NULL; 86 87 while (*p) { 88 parent = *p; 89 r = rb_entry(parent, struct ceph_snap_realm, node); 90 if (new->ino < r->ino) 91 p = &(*p)->rb_left; 92 else if (new->ino > r->ino) 93 p = &(*p)->rb_right; 94 else 95 BUG(); 96 } 97 98 rb_link_node(&new->node, parent, p); 99 rb_insert_color(&new->node, root); 100 } 101 102 /* 103 * create and get the realm rooted at @ino and bump its ref count. 104 * 105 * caller must hold snap_rwsem for write. 106 */ 107 static struct ceph_snap_realm *ceph_create_snap_realm( 108 struct ceph_mds_client *mdsc, 109 u64 ino) 110 { 111 struct ceph_snap_realm *realm; 112 113 realm = kzalloc(sizeof(*realm), GFP_NOFS); 114 if (!realm) 115 return ERR_PTR(-ENOMEM); 116 117 atomic_set(&realm->nref, 1); /* for caller */ 118 realm->ino = ino; 119 INIT_LIST_HEAD(&realm->children); 120 INIT_LIST_HEAD(&realm->child_item); 121 INIT_LIST_HEAD(&realm->empty_item); 122 INIT_LIST_HEAD(&realm->dirty_item); 123 INIT_LIST_HEAD(&realm->inodes_with_caps); 124 spin_lock_init(&realm->inodes_with_caps_lock); 125 __insert_snap_realm(&mdsc->snap_realms, realm); 126 dout("create_snap_realm %llx %p\n", realm->ino, realm); 127 return realm; 128 } 129 130 /* 131 * lookup the realm rooted at @ino. 132 * 133 * caller must hold snap_rwsem for write. 134 */ 135 static struct ceph_snap_realm *__lookup_snap_realm(struct ceph_mds_client *mdsc, 136 u64 ino) 137 { 138 struct rb_node *n = mdsc->snap_realms.rb_node; 139 struct ceph_snap_realm *r; 140 141 while (n) { 142 r = rb_entry(n, struct ceph_snap_realm, node); 143 if (ino < r->ino) 144 n = n->rb_left; 145 else if (ino > r->ino) 146 n = n->rb_right; 147 else { 148 dout("lookup_snap_realm %llx %p\n", r->ino, r); 149 return r; 150 } 151 } 152 return NULL; 153 } 154 155 struct ceph_snap_realm *ceph_lookup_snap_realm(struct ceph_mds_client *mdsc, 156 u64 ino) 157 { 158 struct ceph_snap_realm *r; 159 r = __lookup_snap_realm(mdsc, ino); 160 if (r) 161 ceph_get_snap_realm(mdsc, r); 162 return r; 163 } 164 165 static void __put_snap_realm(struct ceph_mds_client *mdsc, 166 struct ceph_snap_realm *realm); 167 168 /* 169 * called with snap_rwsem (write) 170 */ 171 static void __destroy_snap_realm(struct ceph_mds_client *mdsc, 172 struct ceph_snap_realm *realm) 173 { 174 dout("__destroy_snap_realm %p %llx\n", realm, realm->ino); 175 176 rb_erase(&realm->node, &mdsc->snap_realms); 177 178 if (realm->parent) { 179 list_del_init(&realm->child_item); 180 __put_snap_realm(mdsc, realm->parent); 181 } 182 183 kfree(realm->prior_parent_snaps); 184 kfree(realm->snaps); 185 ceph_put_snap_context(realm->cached_context); 186 kfree(realm); 187 } 188 189 /* 190 * caller holds snap_rwsem (write) 191 */ 192 static void __put_snap_realm(struct ceph_mds_client *mdsc, 193 struct ceph_snap_realm *realm) 194 { 195 dout("__put_snap_realm %llx %p %d -> %d\n", realm->ino, realm, 196 atomic_read(&realm->nref), atomic_read(&realm->nref)-1); 197 if (atomic_dec_and_test(&realm->nref)) 198 __destroy_snap_realm(mdsc, realm); 199 } 200 201 /* 202 * caller needn't hold any locks 203 */ 204 void ceph_put_snap_realm(struct ceph_mds_client *mdsc, 205 struct ceph_snap_realm *realm) 206 { 207 dout("put_snap_realm %llx %p %d -> %d\n", realm->ino, realm, 208 atomic_read(&realm->nref), atomic_read(&realm->nref)-1); 209 if (!atomic_dec_and_test(&realm->nref)) 210 return; 211 212 if (down_write_trylock(&mdsc->snap_rwsem)) { 213 __destroy_snap_realm(mdsc, realm); 214 up_write(&mdsc->snap_rwsem); 215 } else { 216 spin_lock(&mdsc->snap_empty_lock); 217 list_add(&realm->empty_item, &mdsc->snap_empty); 218 spin_unlock(&mdsc->snap_empty_lock); 219 } 220 } 221 222 /* 223 * Clean up any realms whose ref counts have dropped to zero. Note 224 * that this does not include realms who were created but not yet 225 * used. 226 * 227 * Called under snap_rwsem (write) 228 */ 229 static void __cleanup_empty_realms(struct ceph_mds_client *mdsc) 230 { 231 struct ceph_snap_realm *realm; 232 233 spin_lock(&mdsc->snap_empty_lock); 234 while (!list_empty(&mdsc->snap_empty)) { 235 realm = list_first_entry(&mdsc->snap_empty, 236 struct ceph_snap_realm, empty_item); 237 list_del(&realm->empty_item); 238 spin_unlock(&mdsc->snap_empty_lock); 239 __destroy_snap_realm(mdsc, realm); 240 spin_lock(&mdsc->snap_empty_lock); 241 } 242 spin_unlock(&mdsc->snap_empty_lock); 243 } 244 245 void ceph_cleanup_empty_realms(struct ceph_mds_client *mdsc) 246 { 247 down_write(&mdsc->snap_rwsem); 248 __cleanup_empty_realms(mdsc); 249 up_write(&mdsc->snap_rwsem); 250 } 251 252 /* 253 * adjust the parent realm of a given @realm. adjust child list, and parent 254 * pointers, and ref counts appropriately. 255 * 256 * return true if parent was changed, 0 if unchanged, <0 on error. 257 * 258 * caller must hold snap_rwsem for write. 259 */ 260 static int adjust_snap_realm_parent(struct ceph_mds_client *mdsc, 261 struct ceph_snap_realm *realm, 262 u64 parentino) 263 { 264 struct ceph_snap_realm *parent; 265 266 if (realm->parent_ino == parentino) 267 return 0; 268 269 parent = ceph_lookup_snap_realm(mdsc, parentino); 270 if (!parent) { 271 parent = ceph_create_snap_realm(mdsc, parentino); 272 if (IS_ERR(parent)) 273 return PTR_ERR(parent); 274 } 275 dout("adjust_snap_realm_parent %llx %p: %llx %p -> %llx %p\n", 276 realm->ino, realm, realm->parent_ino, realm->parent, 277 parentino, parent); 278 if (realm->parent) { 279 list_del_init(&realm->child_item); 280 ceph_put_snap_realm(mdsc, realm->parent); 281 } 282 realm->parent_ino = parentino; 283 realm->parent = parent; 284 list_add(&realm->child_item, &parent->children); 285 return 1; 286 } 287 288 289 static int cmpu64_rev(const void *a, const void *b) 290 { 291 if (*(u64 *)a < *(u64 *)b) 292 return 1; 293 if (*(u64 *)a > *(u64 *)b) 294 return -1; 295 return 0; 296 } 297 298 299 /* 300 * build the snap context for a given realm. 301 */ 302 static int build_snap_context(struct ceph_snap_realm *realm, 303 struct list_head* dirty_realms) 304 { 305 struct ceph_snap_realm *parent = realm->parent; 306 struct ceph_snap_context *snapc; 307 int err = 0; 308 u32 num = realm->num_prior_parent_snaps + realm->num_snaps; 309 310 /* 311 * build parent context, if it hasn't been built. 312 * conservatively estimate that all parent snaps might be 313 * included by us. 314 */ 315 if (parent) { 316 if (!parent->cached_context) { 317 err = build_snap_context(parent, dirty_realms); 318 if (err) 319 goto fail; 320 } 321 num += parent->cached_context->num_snaps; 322 } 323 324 /* do i actually need to update? not if my context seq 325 matches realm seq, and my parents' does to. (this works 326 because we rebuild_snap_realms() works _downward_ in 327 hierarchy after each update.) */ 328 if (realm->cached_context && 329 realm->cached_context->seq == realm->seq && 330 (!parent || 331 realm->cached_context->seq >= parent->cached_context->seq)) { 332 dout("build_snap_context %llx %p: %p seq %lld (%u snaps)" 333 " (unchanged)\n", 334 realm->ino, realm, realm->cached_context, 335 realm->cached_context->seq, 336 (unsigned int)realm->cached_context->num_snaps); 337 return 0; 338 } 339 340 /* alloc new snap context */ 341 err = -ENOMEM; 342 if (num > (SIZE_MAX - sizeof(*snapc)) / sizeof(u64)) 343 goto fail; 344 snapc = ceph_create_snap_context(num, GFP_NOFS); 345 if (!snapc) 346 goto fail; 347 348 /* build (reverse sorted) snap vector */ 349 num = 0; 350 snapc->seq = realm->seq; 351 if (parent) { 352 u32 i; 353 354 /* include any of parent's snaps occurring _after_ my 355 parent became my parent */ 356 for (i = 0; i < parent->cached_context->num_snaps; i++) 357 if (parent->cached_context->snaps[i] >= 358 realm->parent_since) 359 snapc->snaps[num++] = 360 parent->cached_context->snaps[i]; 361 if (parent->cached_context->seq > snapc->seq) 362 snapc->seq = parent->cached_context->seq; 363 } 364 memcpy(snapc->snaps + num, realm->snaps, 365 sizeof(u64)*realm->num_snaps); 366 num += realm->num_snaps; 367 memcpy(snapc->snaps + num, realm->prior_parent_snaps, 368 sizeof(u64)*realm->num_prior_parent_snaps); 369 num += realm->num_prior_parent_snaps; 370 371 sort(snapc->snaps, num, sizeof(u64), cmpu64_rev, NULL); 372 snapc->num_snaps = num; 373 dout("build_snap_context %llx %p: %p seq %lld (%u snaps)\n", 374 realm->ino, realm, snapc, snapc->seq, 375 (unsigned int) snapc->num_snaps); 376 377 if (realm->cached_context) { 378 ceph_put_snap_context(realm->cached_context); 379 /* queue realm for cap_snap creation */ 380 list_add_tail(&realm->dirty_item, dirty_realms); 381 } 382 realm->cached_context = snapc; 383 return 0; 384 385 fail: 386 /* 387 * if we fail, clear old (incorrect) cached_context... hopefully 388 * we'll have better luck building it later 389 */ 390 if (realm->cached_context) { 391 ceph_put_snap_context(realm->cached_context); 392 realm->cached_context = NULL; 393 } 394 pr_err("build_snap_context %llx %p fail %d\n", realm->ino, 395 realm, err); 396 return err; 397 } 398 399 /* 400 * rebuild snap context for the given realm and all of its children. 401 */ 402 static void rebuild_snap_realms(struct ceph_snap_realm *realm, 403 struct list_head *dirty_realms) 404 { 405 struct ceph_snap_realm *child; 406 407 dout("rebuild_snap_realms %llx %p\n", realm->ino, realm); 408 build_snap_context(realm, dirty_realms); 409 410 list_for_each_entry(child, &realm->children, child_item) 411 rebuild_snap_realms(child, dirty_realms); 412 } 413 414 415 /* 416 * helper to allocate and decode an array of snapids. free prior 417 * instance, if any. 418 */ 419 static int dup_array(u64 **dst, __le64 *src, u32 num) 420 { 421 u32 i; 422 423 kfree(*dst); 424 if (num) { 425 *dst = kcalloc(num, sizeof(u64), GFP_NOFS); 426 if (!*dst) 427 return -ENOMEM; 428 for (i = 0; i < num; i++) 429 (*dst)[i] = get_unaligned_le64(src + i); 430 } else { 431 *dst = NULL; 432 } 433 return 0; 434 } 435 436 static bool has_new_snaps(struct ceph_snap_context *o, 437 struct ceph_snap_context *n) 438 { 439 if (n->num_snaps == 0) 440 return false; 441 /* snaps are in descending order */ 442 return n->snaps[0] > o->seq; 443 } 444 445 /* 446 * When a snapshot is applied, the size/mtime inode metadata is queued 447 * in a ceph_cap_snap (one for each snapshot) until writeback 448 * completes and the metadata can be flushed back to the MDS. 449 * 450 * However, if a (sync) write is currently in-progress when we apply 451 * the snapshot, we have to wait until the write succeeds or fails 452 * (and a final size/mtime is known). In this case the 453 * cap_snap->writing = 1, and is said to be "pending." When the write 454 * finishes, we __ceph_finish_cap_snap(). 455 * 456 * Caller must hold snap_rwsem for read (i.e., the realm topology won't 457 * change). 458 */ 459 void ceph_queue_cap_snap(struct ceph_inode_info *ci) 460 { 461 struct inode *inode = &ci->vfs_inode; 462 struct ceph_cap_snap *capsnap; 463 struct ceph_snap_context *old_snapc, *new_snapc; 464 int used, dirty; 465 466 capsnap = kzalloc(sizeof(*capsnap), GFP_NOFS); 467 if (!capsnap) { 468 pr_err("ENOMEM allocating ceph_cap_snap on %p\n", inode); 469 return; 470 } 471 472 spin_lock(&ci->i_ceph_lock); 473 used = __ceph_caps_used(ci); 474 dirty = __ceph_caps_dirty(ci); 475 476 old_snapc = ci->i_head_snapc; 477 new_snapc = ci->i_snap_realm->cached_context; 478 479 /* 480 * If there is a write in progress, treat that as a dirty Fw, 481 * even though it hasn't completed yet; by the time we finish 482 * up this capsnap it will be. 483 */ 484 if (used & CEPH_CAP_FILE_WR) 485 dirty |= CEPH_CAP_FILE_WR; 486 487 if (__ceph_have_pending_cap_snap(ci)) { 488 /* there is no point in queuing multiple "pending" cap_snaps, 489 as no new writes are allowed to start when pending, so any 490 writes in progress now were started before the previous 491 cap_snap. lucky us. */ 492 dout("queue_cap_snap %p already pending\n", inode); 493 goto update_snapc; 494 } 495 if (ci->i_wrbuffer_ref_head == 0 && 496 !(dirty & (CEPH_CAP_ANY_EXCL|CEPH_CAP_FILE_WR))) { 497 dout("queue_cap_snap %p nothing dirty|writing\n", inode); 498 goto update_snapc; 499 } 500 501 BUG_ON(!old_snapc); 502 503 /* 504 * There is no need to send FLUSHSNAP message to MDS if there is 505 * no new snapshot. But when there is dirty pages or on-going 506 * writes, we still need to create cap_snap. cap_snap is needed 507 * by the write path and page writeback path. 508 * 509 * also see ceph_try_drop_cap_snap() 510 */ 511 if (has_new_snaps(old_snapc, new_snapc)) { 512 if (dirty & (CEPH_CAP_ANY_EXCL|CEPH_CAP_FILE_WR)) 513 capsnap->need_flush = true; 514 } else { 515 if (!(used & CEPH_CAP_FILE_WR) && 516 ci->i_wrbuffer_ref_head == 0) { 517 dout("queue_cap_snap %p " 518 "no new_snap|dirty_page|writing\n", inode); 519 goto update_snapc; 520 } 521 } 522 523 dout("queue_cap_snap %p cap_snap %p queuing under %p %s %s\n", 524 inode, capsnap, old_snapc, ceph_cap_string(dirty), 525 capsnap->need_flush ? "" : "no_flush"); 526 ihold(inode); 527 528 refcount_set(&capsnap->nref, 1); 529 INIT_LIST_HEAD(&capsnap->ci_item); 530 531 capsnap->follows = old_snapc->seq; 532 capsnap->issued = __ceph_caps_issued(ci, NULL); 533 capsnap->dirty = dirty; 534 535 capsnap->mode = inode->i_mode; 536 capsnap->uid = inode->i_uid; 537 capsnap->gid = inode->i_gid; 538 539 if (dirty & CEPH_CAP_XATTR_EXCL) { 540 __ceph_build_xattrs_blob(ci); 541 capsnap->xattr_blob = 542 ceph_buffer_get(ci->i_xattrs.blob); 543 capsnap->xattr_version = ci->i_xattrs.version; 544 } else { 545 capsnap->xattr_blob = NULL; 546 capsnap->xattr_version = 0; 547 } 548 549 capsnap->inline_data = ci->i_inline_version != CEPH_INLINE_NONE; 550 551 /* dirty page count moved from _head to this cap_snap; 552 all subsequent writes page dirties occur _after_ this 553 snapshot. */ 554 capsnap->dirty_pages = ci->i_wrbuffer_ref_head; 555 ci->i_wrbuffer_ref_head = 0; 556 capsnap->context = old_snapc; 557 list_add_tail(&capsnap->ci_item, &ci->i_cap_snaps); 558 559 if (used & CEPH_CAP_FILE_WR) { 560 dout("queue_cap_snap %p cap_snap %p snapc %p" 561 " seq %llu used WR, now pending\n", inode, 562 capsnap, old_snapc, old_snapc->seq); 563 capsnap->writing = 1; 564 } else { 565 /* note mtime, size NOW. */ 566 __ceph_finish_cap_snap(ci, capsnap); 567 } 568 capsnap = NULL; 569 old_snapc = NULL; 570 571 update_snapc: 572 if (ci->i_head_snapc) { 573 ci->i_head_snapc = ceph_get_snap_context(new_snapc); 574 dout(" new snapc is %p\n", new_snapc); 575 } 576 spin_unlock(&ci->i_ceph_lock); 577 578 kfree(capsnap); 579 ceph_put_snap_context(old_snapc); 580 } 581 582 /* 583 * Finalize the size, mtime for a cap_snap.. that is, settle on final values 584 * to be used for the snapshot, to be flushed back to the mds. 585 * 586 * If capsnap can now be flushed, add to snap_flush list, and return 1. 587 * 588 * Caller must hold i_ceph_lock. 589 */ 590 int __ceph_finish_cap_snap(struct ceph_inode_info *ci, 591 struct ceph_cap_snap *capsnap) 592 { 593 struct inode *inode = &ci->vfs_inode; 594 struct ceph_mds_client *mdsc = ceph_sb_to_client(inode->i_sb)->mdsc; 595 596 BUG_ON(capsnap->writing); 597 capsnap->size = inode->i_size; 598 capsnap->mtime = inode->i_mtime; 599 capsnap->atime = inode->i_atime; 600 capsnap->ctime = inode->i_ctime; 601 capsnap->time_warp_seq = ci->i_time_warp_seq; 602 capsnap->truncate_size = ci->i_truncate_size; 603 capsnap->truncate_seq = ci->i_truncate_seq; 604 if (capsnap->dirty_pages) { 605 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu " 606 "still has %d dirty pages\n", inode, capsnap, 607 capsnap->context, capsnap->context->seq, 608 ceph_cap_string(capsnap->dirty), capsnap->size, 609 capsnap->dirty_pages); 610 return 0; 611 } 612 613 ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS; 614 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu\n", 615 inode, capsnap, capsnap->context, 616 capsnap->context->seq, ceph_cap_string(capsnap->dirty), 617 capsnap->size); 618 619 spin_lock(&mdsc->snap_flush_lock); 620 list_add_tail(&ci->i_snap_flush_item, &mdsc->snap_flush_list); 621 spin_unlock(&mdsc->snap_flush_lock); 622 return 1; /* caller may want to ceph_flush_snaps */ 623 } 624 625 /* 626 * Queue cap_snaps for snap writeback for this realm and its children. 627 * Called under snap_rwsem, so realm topology won't change. 628 */ 629 static void queue_realm_cap_snaps(struct ceph_snap_realm *realm) 630 { 631 struct ceph_inode_info *ci; 632 struct inode *lastinode = NULL; 633 634 dout("queue_realm_cap_snaps %p %llx inodes\n", realm, realm->ino); 635 636 spin_lock(&realm->inodes_with_caps_lock); 637 list_for_each_entry(ci, &realm->inodes_with_caps, i_snap_realm_item) { 638 struct inode *inode = igrab(&ci->vfs_inode); 639 if (!inode) 640 continue; 641 spin_unlock(&realm->inodes_with_caps_lock); 642 iput(lastinode); 643 lastinode = inode; 644 ceph_queue_cap_snap(ci); 645 spin_lock(&realm->inodes_with_caps_lock); 646 } 647 spin_unlock(&realm->inodes_with_caps_lock); 648 iput(lastinode); 649 650 dout("queue_realm_cap_snaps %p %llx done\n", realm, realm->ino); 651 } 652 653 /* 654 * Parse and apply a snapblob "snap trace" from the MDS. This specifies 655 * the snap realm parameters from a given realm and all of its ancestors, 656 * up to the root. 657 * 658 * Caller must hold snap_rwsem for write. 659 */ 660 int ceph_update_snap_trace(struct ceph_mds_client *mdsc, 661 void *p, void *e, bool deletion, 662 struct ceph_snap_realm **realm_ret) 663 { 664 struct ceph_mds_snap_realm *ri; /* encoded */ 665 __le64 *snaps; /* encoded */ 666 __le64 *prior_parent_snaps; /* encoded */ 667 struct ceph_snap_realm *realm = NULL; 668 struct ceph_snap_realm *first_realm = NULL; 669 int invalidate = 0; 670 int err = -ENOMEM; 671 LIST_HEAD(dirty_realms); 672 673 dout("update_snap_trace deletion=%d\n", deletion); 674 more: 675 ceph_decode_need(&p, e, sizeof(*ri), bad); 676 ri = p; 677 p += sizeof(*ri); 678 ceph_decode_need(&p, e, sizeof(u64)*(le32_to_cpu(ri->num_snaps) + 679 le32_to_cpu(ri->num_prior_parent_snaps)), bad); 680 snaps = p; 681 p += sizeof(u64) * le32_to_cpu(ri->num_snaps); 682 prior_parent_snaps = p; 683 p += sizeof(u64) * le32_to_cpu(ri->num_prior_parent_snaps); 684 685 realm = ceph_lookup_snap_realm(mdsc, le64_to_cpu(ri->ino)); 686 if (!realm) { 687 realm = ceph_create_snap_realm(mdsc, le64_to_cpu(ri->ino)); 688 if (IS_ERR(realm)) { 689 err = PTR_ERR(realm); 690 goto fail; 691 } 692 } 693 694 /* ensure the parent is correct */ 695 err = adjust_snap_realm_parent(mdsc, realm, le64_to_cpu(ri->parent)); 696 if (err < 0) 697 goto fail; 698 invalidate += err; 699 700 if (le64_to_cpu(ri->seq) > realm->seq) { 701 dout("update_snap_trace updating %llx %p %lld -> %lld\n", 702 realm->ino, realm, realm->seq, le64_to_cpu(ri->seq)); 703 /* update realm parameters, snap lists */ 704 realm->seq = le64_to_cpu(ri->seq); 705 realm->created = le64_to_cpu(ri->created); 706 realm->parent_since = le64_to_cpu(ri->parent_since); 707 708 realm->num_snaps = le32_to_cpu(ri->num_snaps); 709 err = dup_array(&realm->snaps, snaps, realm->num_snaps); 710 if (err < 0) 711 goto fail; 712 713 realm->num_prior_parent_snaps = 714 le32_to_cpu(ri->num_prior_parent_snaps); 715 err = dup_array(&realm->prior_parent_snaps, prior_parent_snaps, 716 realm->num_prior_parent_snaps); 717 if (err < 0) 718 goto fail; 719 720 if (realm->seq > mdsc->last_snap_seq) 721 mdsc->last_snap_seq = realm->seq; 722 723 invalidate = 1; 724 } else if (!realm->cached_context) { 725 dout("update_snap_trace %llx %p seq %lld new\n", 726 realm->ino, realm, realm->seq); 727 invalidate = 1; 728 } else { 729 dout("update_snap_trace %llx %p seq %lld unchanged\n", 730 realm->ino, realm, realm->seq); 731 } 732 733 dout("done with %llx %p, invalidated=%d, %p %p\n", realm->ino, 734 realm, invalidate, p, e); 735 736 /* invalidate when we reach the _end_ (root) of the trace */ 737 if (invalidate && p >= e) 738 rebuild_snap_realms(realm, &dirty_realms); 739 740 if (!first_realm) 741 first_realm = realm; 742 else 743 ceph_put_snap_realm(mdsc, realm); 744 745 if (p < e) 746 goto more; 747 748 /* 749 * queue cap snaps _after_ we've built the new snap contexts, 750 * so that i_head_snapc can be set appropriately. 751 */ 752 while (!list_empty(&dirty_realms)) { 753 realm = list_first_entry(&dirty_realms, struct ceph_snap_realm, 754 dirty_item); 755 list_del_init(&realm->dirty_item); 756 queue_realm_cap_snaps(realm); 757 } 758 759 if (realm_ret) 760 *realm_ret = first_realm; 761 else 762 ceph_put_snap_realm(mdsc, first_realm); 763 764 __cleanup_empty_realms(mdsc); 765 return 0; 766 767 bad: 768 err = -EINVAL; 769 fail: 770 if (realm && !IS_ERR(realm)) 771 ceph_put_snap_realm(mdsc, realm); 772 if (first_realm) 773 ceph_put_snap_realm(mdsc, first_realm); 774 pr_err("update_snap_trace error %d\n", err); 775 return err; 776 } 777 778 779 /* 780 * Send any cap_snaps that are queued for flush. Try to carry 781 * s_mutex across multiple snap flushes to avoid locking overhead. 782 * 783 * Caller holds no locks. 784 */ 785 static void flush_snaps(struct ceph_mds_client *mdsc) 786 { 787 struct ceph_inode_info *ci; 788 struct inode *inode; 789 struct ceph_mds_session *session = NULL; 790 791 dout("flush_snaps\n"); 792 spin_lock(&mdsc->snap_flush_lock); 793 while (!list_empty(&mdsc->snap_flush_list)) { 794 ci = list_first_entry(&mdsc->snap_flush_list, 795 struct ceph_inode_info, i_snap_flush_item); 796 inode = &ci->vfs_inode; 797 ihold(inode); 798 spin_unlock(&mdsc->snap_flush_lock); 799 ceph_flush_snaps(ci, &session); 800 iput(inode); 801 spin_lock(&mdsc->snap_flush_lock); 802 } 803 spin_unlock(&mdsc->snap_flush_lock); 804 805 if (session) { 806 mutex_unlock(&session->s_mutex); 807 ceph_put_mds_session(session); 808 } 809 dout("flush_snaps done\n"); 810 } 811 812 813 /* 814 * Handle a snap notification from the MDS. 815 * 816 * This can take two basic forms: the simplest is just a snap creation 817 * or deletion notification on an existing realm. This should update the 818 * realm and its children. 819 * 820 * The more difficult case is realm creation, due to snap creation at a 821 * new point in the file hierarchy, or due to a rename that moves a file or 822 * directory into another realm. 823 */ 824 void ceph_handle_snap(struct ceph_mds_client *mdsc, 825 struct ceph_mds_session *session, 826 struct ceph_msg *msg) 827 { 828 struct super_block *sb = mdsc->fsc->sb; 829 int mds = session->s_mds; 830 u64 split; 831 int op; 832 int trace_len; 833 struct ceph_snap_realm *realm = NULL; 834 void *p = msg->front.iov_base; 835 void *e = p + msg->front.iov_len; 836 struct ceph_mds_snap_head *h; 837 int num_split_inos, num_split_realms; 838 __le64 *split_inos = NULL, *split_realms = NULL; 839 int i; 840 int locked_rwsem = 0; 841 842 /* decode */ 843 if (msg->front.iov_len < sizeof(*h)) 844 goto bad; 845 h = p; 846 op = le32_to_cpu(h->op); 847 split = le64_to_cpu(h->split); /* non-zero if we are splitting an 848 * existing realm */ 849 num_split_inos = le32_to_cpu(h->num_split_inos); 850 num_split_realms = le32_to_cpu(h->num_split_realms); 851 trace_len = le32_to_cpu(h->trace_len); 852 p += sizeof(*h); 853 854 dout("handle_snap from mds%d op %s split %llx tracelen %d\n", mds, 855 ceph_snap_op_name(op), split, trace_len); 856 857 mutex_lock(&session->s_mutex); 858 session->s_seq++; 859 mutex_unlock(&session->s_mutex); 860 861 down_write(&mdsc->snap_rwsem); 862 locked_rwsem = 1; 863 864 if (op == CEPH_SNAP_OP_SPLIT) { 865 struct ceph_mds_snap_realm *ri; 866 867 /* 868 * A "split" breaks part of an existing realm off into 869 * a new realm. The MDS provides a list of inodes 870 * (with caps) and child realms that belong to the new 871 * child. 872 */ 873 split_inos = p; 874 p += sizeof(u64) * num_split_inos; 875 split_realms = p; 876 p += sizeof(u64) * num_split_realms; 877 ceph_decode_need(&p, e, sizeof(*ri), bad); 878 /* we will peek at realm info here, but will _not_ 879 * advance p, as the realm update will occur below in 880 * ceph_update_snap_trace. */ 881 ri = p; 882 883 realm = ceph_lookup_snap_realm(mdsc, split); 884 if (!realm) { 885 realm = ceph_create_snap_realm(mdsc, split); 886 if (IS_ERR(realm)) 887 goto out; 888 } 889 890 dout("splitting snap_realm %llx %p\n", realm->ino, realm); 891 for (i = 0; i < num_split_inos; i++) { 892 struct ceph_vino vino = { 893 .ino = le64_to_cpu(split_inos[i]), 894 .snap = CEPH_NOSNAP, 895 }; 896 struct inode *inode = ceph_find_inode(sb, vino); 897 struct ceph_inode_info *ci; 898 struct ceph_snap_realm *oldrealm; 899 900 if (!inode) 901 continue; 902 ci = ceph_inode(inode); 903 904 spin_lock(&ci->i_ceph_lock); 905 if (!ci->i_snap_realm) 906 goto skip_inode; 907 /* 908 * If this inode belongs to a realm that was 909 * created after our new realm, we experienced 910 * a race (due to another split notifications 911 * arriving from a different MDS). So skip 912 * this inode. 913 */ 914 if (ci->i_snap_realm->created > 915 le64_to_cpu(ri->created)) { 916 dout(" leaving %p in newer realm %llx %p\n", 917 inode, ci->i_snap_realm->ino, 918 ci->i_snap_realm); 919 goto skip_inode; 920 } 921 dout(" will move %p to split realm %llx %p\n", 922 inode, realm->ino, realm); 923 /* 924 * Move the inode to the new realm 925 */ 926 spin_lock(&realm->inodes_with_caps_lock); 927 list_del_init(&ci->i_snap_realm_item); 928 list_add(&ci->i_snap_realm_item, 929 &realm->inodes_with_caps); 930 oldrealm = ci->i_snap_realm; 931 ci->i_snap_realm = realm; 932 spin_unlock(&realm->inodes_with_caps_lock); 933 spin_unlock(&ci->i_ceph_lock); 934 935 ceph_get_snap_realm(mdsc, realm); 936 ceph_put_snap_realm(mdsc, oldrealm); 937 938 iput(inode); 939 continue; 940 941 skip_inode: 942 spin_unlock(&ci->i_ceph_lock); 943 iput(inode); 944 } 945 946 /* we may have taken some of the old realm's children. */ 947 for (i = 0; i < num_split_realms; i++) { 948 struct ceph_snap_realm *child = 949 __lookup_snap_realm(mdsc, 950 le64_to_cpu(split_realms[i])); 951 if (!child) 952 continue; 953 adjust_snap_realm_parent(mdsc, child, realm->ino); 954 } 955 } 956 957 /* 958 * update using the provided snap trace. if we are deleting a 959 * snap, we can avoid queueing cap_snaps. 960 */ 961 ceph_update_snap_trace(mdsc, p, e, 962 op == CEPH_SNAP_OP_DESTROY, NULL); 963 964 if (op == CEPH_SNAP_OP_SPLIT) 965 /* we took a reference when we created the realm, above */ 966 ceph_put_snap_realm(mdsc, realm); 967 968 __cleanup_empty_realms(mdsc); 969 970 up_write(&mdsc->snap_rwsem); 971 972 flush_snaps(mdsc); 973 return; 974 975 bad: 976 pr_err("corrupt snap message from mds%d\n", mds); 977 ceph_msg_dump(msg); 978 out: 979 if (locked_rwsem) 980 up_write(&mdsc->snap_rwsem); 981 return; 982 } 983