1 #include <linux/ceph/ceph_debug.h> 2 3 #include <linux/fs.h> 4 #include <linux/kernel.h> 5 #include <linux/sched.h> 6 #include <linux/slab.h> 7 #include <linux/vmalloc.h> 8 #include <linux/wait.h> 9 #include <linux/writeback.h> 10 11 #include "super.h" 12 #include "mds_client.h" 13 #include <linux/ceph/decode.h> 14 #include <linux/ceph/messenger.h> 15 16 /* 17 * Capability management 18 * 19 * The Ceph metadata servers control client access to inode metadata 20 * and file data by issuing capabilities, granting clients permission 21 * to read and/or write both inode field and file data to OSDs 22 * (storage nodes). Each capability consists of a set of bits 23 * indicating which operations are allowed. 24 * 25 * If the client holds a *_SHARED cap, the client has a coherent value 26 * that can be safely read from the cached inode. 27 * 28 * In the case of a *_EXCL (exclusive) or FILE_WR capabilities, the 29 * client is allowed to change inode attributes (e.g., file size, 30 * mtime), note its dirty state in the ceph_cap, and asynchronously 31 * flush that metadata change to the MDS. 32 * 33 * In the event of a conflicting operation (perhaps by another 34 * client), the MDS will revoke the conflicting client capabilities. 35 * 36 * In order for a client to cache an inode, it must hold a capability 37 * with at least one MDS server. When inodes are released, release 38 * notifications are batched and periodically sent en masse to the MDS 39 * cluster to release server state. 40 */ 41 42 43 /* 44 * Generate readable cap strings for debugging output. 45 */ 46 #define MAX_CAP_STR 20 47 static char cap_str[MAX_CAP_STR][40]; 48 static DEFINE_SPINLOCK(cap_str_lock); 49 static int last_cap_str; 50 51 static char *gcap_string(char *s, int c) 52 { 53 if (c & CEPH_CAP_GSHARED) 54 *s++ = 's'; 55 if (c & CEPH_CAP_GEXCL) 56 *s++ = 'x'; 57 if (c & CEPH_CAP_GCACHE) 58 *s++ = 'c'; 59 if (c & CEPH_CAP_GRD) 60 *s++ = 'r'; 61 if (c & CEPH_CAP_GWR) 62 *s++ = 'w'; 63 if (c & CEPH_CAP_GBUFFER) 64 *s++ = 'b'; 65 if (c & CEPH_CAP_GLAZYIO) 66 *s++ = 'l'; 67 return s; 68 } 69 70 const char *ceph_cap_string(int caps) 71 { 72 int i; 73 char *s; 74 int c; 75 76 spin_lock(&cap_str_lock); 77 i = last_cap_str++; 78 if (last_cap_str == MAX_CAP_STR) 79 last_cap_str = 0; 80 spin_unlock(&cap_str_lock); 81 82 s = cap_str[i]; 83 84 if (caps & CEPH_CAP_PIN) 85 *s++ = 'p'; 86 87 c = (caps >> CEPH_CAP_SAUTH) & 3; 88 if (c) { 89 *s++ = 'A'; 90 s = gcap_string(s, c); 91 } 92 93 c = (caps >> CEPH_CAP_SLINK) & 3; 94 if (c) { 95 *s++ = 'L'; 96 s = gcap_string(s, c); 97 } 98 99 c = (caps >> CEPH_CAP_SXATTR) & 3; 100 if (c) { 101 *s++ = 'X'; 102 s = gcap_string(s, c); 103 } 104 105 c = caps >> CEPH_CAP_SFILE; 106 if (c) { 107 *s++ = 'F'; 108 s = gcap_string(s, c); 109 } 110 111 if (s == cap_str[i]) 112 *s++ = '-'; 113 *s = 0; 114 return cap_str[i]; 115 } 116 117 void ceph_caps_init(struct ceph_mds_client *mdsc) 118 { 119 INIT_LIST_HEAD(&mdsc->caps_list); 120 spin_lock_init(&mdsc->caps_list_lock); 121 } 122 123 void ceph_caps_finalize(struct ceph_mds_client *mdsc) 124 { 125 struct ceph_cap *cap; 126 127 spin_lock(&mdsc->caps_list_lock); 128 while (!list_empty(&mdsc->caps_list)) { 129 cap = list_first_entry(&mdsc->caps_list, 130 struct ceph_cap, caps_item); 131 list_del(&cap->caps_item); 132 kmem_cache_free(ceph_cap_cachep, cap); 133 } 134 mdsc->caps_total_count = 0; 135 mdsc->caps_avail_count = 0; 136 mdsc->caps_use_count = 0; 137 mdsc->caps_reserve_count = 0; 138 mdsc->caps_min_count = 0; 139 spin_unlock(&mdsc->caps_list_lock); 140 } 141 142 void ceph_adjust_min_caps(struct ceph_mds_client *mdsc, int delta) 143 { 144 spin_lock(&mdsc->caps_list_lock); 145 mdsc->caps_min_count += delta; 146 BUG_ON(mdsc->caps_min_count < 0); 147 spin_unlock(&mdsc->caps_list_lock); 148 } 149 150 int ceph_reserve_caps(struct ceph_mds_client *mdsc, 151 struct ceph_cap_reservation *ctx, int need) 152 { 153 int i; 154 struct ceph_cap *cap; 155 int have; 156 int alloc = 0; 157 LIST_HEAD(newcaps); 158 int ret = 0; 159 160 dout("reserve caps ctx=%p need=%d\n", ctx, need); 161 162 /* first reserve any caps that are already allocated */ 163 spin_lock(&mdsc->caps_list_lock); 164 if (mdsc->caps_avail_count >= need) 165 have = need; 166 else 167 have = mdsc->caps_avail_count; 168 mdsc->caps_avail_count -= have; 169 mdsc->caps_reserve_count += have; 170 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count + 171 mdsc->caps_reserve_count + 172 mdsc->caps_avail_count); 173 spin_unlock(&mdsc->caps_list_lock); 174 175 for (i = have; i < need; i++) { 176 cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS); 177 if (!cap) { 178 ret = -ENOMEM; 179 goto out_alloc_count; 180 } 181 list_add(&cap->caps_item, &newcaps); 182 alloc++; 183 } 184 BUG_ON(have + alloc != need); 185 186 spin_lock(&mdsc->caps_list_lock); 187 mdsc->caps_total_count += alloc; 188 mdsc->caps_reserve_count += alloc; 189 list_splice(&newcaps, &mdsc->caps_list); 190 191 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count + 192 mdsc->caps_reserve_count + 193 mdsc->caps_avail_count); 194 spin_unlock(&mdsc->caps_list_lock); 195 196 ctx->count = need; 197 dout("reserve caps ctx=%p %d = %d used + %d resv + %d avail\n", 198 ctx, mdsc->caps_total_count, mdsc->caps_use_count, 199 mdsc->caps_reserve_count, mdsc->caps_avail_count); 200 return 0; 201 202 out_alloc_count: 203 /* we didn't manage to reserve as much as we needed */ 204 pr_warning("reserve caps ctx=%p ENOMEM need=%d got=%d\n", 205 ctx, need, have); 206 return ret; 207 } 208 209 int ceph_unreserve_caps(struct ceph_mds_client *mdsc, 210 struct ceph_cap_reservation *ctx) 211 { 212 dout("unreserve caps ctx=%p count=%d\n", ctx, ctx->count); 213 if (ctx->count) { 214 spin_lock(&mdsc->caps_list_lock); 215 BUG_ON(mdsc->caps_reserve_count < ctx->count); 216 mdsc->caps_reserve_count -= ctx->count; 217 mdsc->caps_avail_count += ctx->count; 218 ctx->count = 0; 219 dout("unreserve caps %d = %d used + %d resv + %d avail\n", 220 mdsc->caps_total_count, mdsc->caps_use_count, 221 mdsc->caps_reserve_count, mdsc->caps_avail_count); 222 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count + 223 mdsc->caps_reserve_count + 224 mdsc->caps_avail_count); 225 spin_unlock(&mdsc->caps_list_lock); 226 } 227 return 0; 228 } 229 230 static struct ceph_cap *get_cap(struct ceph_mds_client *mdsc, 231 struct ceph_cap_reservation *ctx) 232 { 233 struct ceph_cap *cap = NULL; 234 235 /* temporary, until we do something about cap import/export */ 236 if (!ctx) { 237 cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS); 238 if (cap) { 239 spin_lock(&mdsc->caps_list_lock); 240 mdsc->caps_use_count++; 241 mdsc->caps_total_count++; 242 spin_unlock(&mdsc->caps_list_lock); 243 } 244 return cap; 245 } 246 247 spin_lock(&mdsc->caps_list_lock); 248 dout("get_cap ctx=%p (%d) %d = %d used + %d resv + %d avail\n", 249 ctx, ctx->count, mdsc->caps_total_count, mdsc->caps_use_count, 250 mdsc->caps_reserve_count, mdsc->caps_avail_count); 251 BUG_ON(!ctx->count); 252 BUG_ON(ctx->count > mdsc->caps_reserve_count); 253 BUG_ON(list_empty(&mdsc->caps_list)); 254 255 ctx->count--; 256 mdsc->caps_reserve_count--; 257 mdsc->caps_use_count++; 258 259 cap = list_first_entry(&mdsc->caps_list, struct ceph_cap, caps_item); 260 list_del(&cap->caps_item); 261 262 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count + 263 mdsc->caps_reserve_count + mdsc->caps_avail_count); 264 spin_unlock(&mdsc->caps_list_lock); 265 return cap; 266 } 267 268 void ceph_put_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap) 269 { 270 spin_lock(&mdsc->caps_list_lock); 271 dout("put_cap %p %d = %d used + %d resv + %d avail\n", 272 cap, mdsc->caps_total_count, mdsc->caps_use_count, 273 mdsc->caps_reserve_count, mdsc->caps_avail_count); 274 mdsc->caps_use_count--; 275 /* 276 * Keep some preallocated caps around (ceph_min_count), to 277 * avoid lots of free/alloc churn. 278 */ 279 if (mdsc->caps_avail_count >= mdsc->caps_reserve_count + 280 mdsc->caps_min_count) { 281 mdsc->caps_total_count--; 282 kmem_cache_free(ceph_cap_cachep, cap); 283 } else { 284 mdsc->caps_avail_count++; 285 list_add(&cap->caps_item, &mdsc->caps_list); 286 } 287 288 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count + 289 mdsc->caps_reserve_count + mdsc->caps_avail_count); 290 spin_unlock(&mdsc->caps_list_lock); 291 } 292 293 void ceph_reservation_status(struct ceph_fs_client *fsc, 294 int *total, int *avail, int *used, int *reserved, 295 int *min) 296 { 297 struct ceph_mds_client *mdsc = fsc->mdsc; 298 299 if (total) 300 *total = mdsc->caps_total_count; 301 if (avail) 302 *avail = mdsc->caps_avail_count; 303 if (used) 304 *used = mdsc->caps_use_count; 305 if (reserved) 306 *reserved = mdsc->caps_reserve_count; 307 if (min) 308 *min = mdsc->caps_min_count; 309 } 310 311 /* 312 * Find ceph_cap for given mds, if any. 313 * 314 * Called with i_ceph_lock held. 315 */ 316 static struct ceph_cap *__get_cap_for_mds(struct ceph_inode_info *ci, int mds) 317 { 318 struct ceph_cap *cap; 319 struct rb_node *n = ci->i_caps.rb_node; 320 321 while (n) { 322 cap = rb_entry(n, struct ceph_cap, ci_node); 323 if (mds < cap->mds) 324 n = n->rb_left; 325 else if (mds > cap->mds) 326 n = n->rb_right; 327 else 328 return cap; 329 } 330 return NULL; 331 } 332 333 struct ceph_cap *ceph_get_cap_for_mds(struct ceph_inode_info *ci, int mds) 334 { 335 struct ceph_cap *cap; 336 337 spin_lock(&ci->i_ceph_lock); 338 cap = __get_cap_for_mds(ci, mds); 339 spin_unlock(&ci->i_ceph_lock); 340 return cap; 341 } 342 343 /* 344 * Return id of any MDS with a cap, preferably FILE_WR|BUFFER|EXCL, else -1. 345 */ 346 static int __ceph_get_cap_mds(struct ceph_inode_info *ci) 347 { 348 struct ceph_cap *cap; 349 int mds = -1; 350 struct rb_node *p; 351 352 /* prefer mds with WR|BUFFER|EXCL caps */ 353 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 354 cap = rb_entry(p, struct ceph_cap, ci_node); 355 mds = cap->mds; 356 if (cap->issued & (CEPH_CAP_FILE_WR | 357 CEPH_CAP_FILE_BUFFER | 358 CEPH_CAP_FILE_EXCL)) 359 break; 360 } 361 return mds; 362 } 363 364 int ceph_get_cap_mds(struct inode *inode) 365 { 366 struct ceph_inode_info *ci = ceph_inode(inode); 367 int mds; 368 spin_lock(&ci->i_ceph_lock); 369 mds = __ceph_get_cap_mds(ceph_inode(inode)); 370 spin_unlock(&ci->i_ceph_lock); 371 return mds; 372 } 373 374 /* 375 * Called under i_ceph_lock. 376 */ 377 static void __insert_cap_node(struct ceph_inode_info *ci, 378 struct ceph_cap *new) 379 { 380 struct rb_node **p = &ci->i_caps.rb_node; 381 struct rb_node *parent = NULL; 382 struct ceph_cap *cap = NULL; 383 384 while (*p) { 385 parent = *p; 386 cap = rb_entry(parent, struct ceph_cap, ci_node); 387 if (new->mds < cap->mds) 388 p = &(*p)->rb_left; 389 else if (new->mds > cap->mds) 390 p = &(*p)->rb_right; 391 else 392 BUG(); 393 } 394 395 rb_link_node(&new->ci_node, parent, p); 396 rb_insert_color(&new->ci_node, &ci->i_caps); 397 } 398 399 /* 400 * (re)set cap hold timeouts, which control the delayed release 401 * of unused caps back to the MDS. Should be called on cap use. 402 */ 403 static void __cap_set_timeouts(struct ceph_mds_client *mdsc, 404 struct ceph_inode_info *ci) 405 { 406 struct ceph_mount_options *ma = mdsc->fsc->mount_options; 407 408 ci->i_hold_caps_min = round_jiffies(jiffies + 409 ma->caps_wanted_delay_min * HZ); 410 ci->i_hold_caps_max = round_jiffies(jiffies + 411 ma->caps_wanted_delay_max * HZ); 412 dout("__cap_set_timeouts %p min %lu max %lu\n", &ci->vfs_inode, 413 ci->i_hold_caps_min - jiffies, ci->i_hold_caps_max - jiffies); 414 } 415 416 /* 417 * (Re)queue cap at the end of the delayed cap release list. 418 * 419 * If I_FLUSH is set, leave the inode at the front of the list. 420 * 421 * Caller holds i_ceph_lock 422 * -> we take mdsc->cap_delay_lock 423 */ 424 static void __cap_delay_requeue(struct ceph_mds_client *mdsc, 425 struct ceph_inode_info *ci) 426 { 427 __cap_set_timeouts(mdsc, ci); 428 dout("__cap_delay_requeue %p flags %d at %lu\n", &ci->vfs_inode, 429 ci->i_ceph_flags, ci->i_hold_caps_max); 430 if (!mdsc->stopping) { 431 spin_lock(&mdsc->cap_delay_lock); 432 if (!list_empty(&ci->i_cap_delay_list)) { 433 if (ci->i_ceph_flags & CEPH_I_FLUSH) 434 goto no_change; 435 list_del_init(&ci->i_cap_delay_list); 436 } 437 list_add_tail(&ci->i_cap_delay_list, &mdsc->cap_delay_list); 438 no_change: 439 spin_unlock(&mdsc->cap_delay_lock); 440 } 441 } 442 443 /* 444 * Queue an inode for immediate writeback. Mark inode with I_FLUSH, 445 * indicating we should send a cap message to flush dirty metadata 446 * asap, and move to the front of the delayed cap list. 447 */ 448 static void __cap_delay_requeue_front(struct ceph_mds_client *mdsc, 449 struct ceph_inode_info *ci) 450 { 451 dout("__cap_delay_requeue_front %p\n", &ci->vfs_inode); 452 spin_lock(&mdsc->cap_delay_lock); 453 ci->i_ceph_flags |= CEPH_I_FLUSH; 454 if (!list_empty(&ci->i_cap_delay_list)) 455 list_del_init(&ci->i_cap_delay_list); 456 list_add(&ci->i_cap_delay_list, &mdsc->cap_delay_list); 457 spin_unlock(&mdsc->cap_delay_lock); 458 } 459 460 /* 461 * Cancel delayed work on cap. 462 * 463 * Caller must hold i_ceph_lock. 464 */ 465 static void __cap_delay_cancel(struct ceph_mds_client *mdsc, 466 struct ceph_inode_info *ci) 467 { 468 dout("__cap_delay_cancel %p\n", &ci->vfs_inode); 469 if (list_empty(&ci->i_cap_delay_list)) 470 return; 471 spin_lock(&mdsc->cap_delay_lock); 472 list_del_init(&ci->i_cap_delay_list); 473 spin_unlock(&mdsc->cap_delay_lock); 474 } 475 476 /* 477 * Common issue checks for add_cap, handle_cap_grant. 478 */ 479 static void __check_cap_issue(struct ceph_inode_info *ci, struct ceph_cap *cap, 480 unsigned issued) 481 { 482 unsigned had = __ceph_caps_issued(ci, NULL); 483 484 /* 485 * Each time we receive FILE_CACHE anew, we increment 486 * i_rdcache_gen. 487 */ 488 if ((issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) && 489 (had & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0) 490 ci->i_rdcache_gen++; 491 492 /* 493 * if we are newly issued FILE_SHARED, clear D_COMPLETE; we 494 * don't know what happened to this directory while we didn't 495 * have the cap. 496 */ 497 if ((issued & CEPH_CAP_FILE_SHARED) && 498 (had & CEPH_CAP_FILE_SHARED) == 0) { 499 ci->i_shared_gen++; 500 if (S_ISDIR(ci->vfs_inode.i_mode)) 501 ceph_dir_clear_complete(&ci->vfs_inode); 502 } 503 } 504 505 /* 506 * Add a capability under the given MDS session. 507 * 508 * Caller should hold session snap_rwsem (read) and s_mutex. 509 * 510 * @fmode is the open file mode, if we are opening a file, otherwise 511 * it is < 0. (This is so we can atomically add the cap and add an 512 * open file reference to it.) 513 */ 514 int ceph_add_cap(struct inode *inode, 515 struct ceph_mds_session *session, u64 cap_id, 516 int fmode, unsigned issued, unsigned wanted, 517 unsigned seq, unsigned mseq, u64 realmino, int flags, 518 struct ceph_cap_reservation *caps_reservation) 519 { 520 struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc; 521 struct ceph_inode_info *ci = ceph_inode(inode); 522 struct ceph_cap *new_cap = NULL; 523 struct ceph_cap *cap; 524 int mds = session->s_mds; 525 int actual_wanted; 526 527 dout("add_cap %p mds%d cap %llx %s seq %d\n", inode, 528 session->s_mds, cap_id, ceph_cap_string(issued), seq); 529 530 /* 531 * If we are opening the file, include file mode wanted bits 532 * in wanted. 533 */ 534 if (fmode >= 0) 535 wanted |= ceph_caps_for_mode(fmode); 536 537 retry: 538 spin_lock(&ci->i_ceph_lock); 539 cap = __get_cap_for_mds(ci, mds); 540 if (!cap) { 541 if (new_cap) { 542 cap = new_cap; 543 new_cap = NULL; 544 } else { 545 spin_unlock(&ci->i_ceph_lock); 546 new_cap = get_cap(mdsc, caps_reservation); 547 if (new_cap == NULL) 548 return -ENOMEM; 549 goto retry; 550 } 551 552 cap->issued = 0; 553 cap->implemented = 0; 554 cap->mds = mds; 555 cap->mds_wanted = 0; 556 557 cap->ci = ci; 558 __insert_cap_node(ci, cap); 559 560 /* clear out old exporting info? (i.e. on cap import) */ 561 if (ci->i_cap_exporting_mds == mds) { 562 ci->i_cap_exporting_issued = 0; 563 ci->i_cap_exporting_mseq = 0; 564 ci->i_cap_exporting_mds = -1; 565 } 566 567 /* add to session cap list */ 568 cap->session = session; 569 spin_lock(&session->s_cap_lock); 570 list_add_tail(&cap->session_caps, &session->s_caps); 571 session->s_nr_caps++; 572 spin_unlock(&session->s_cap_lock); 573 } else if (new_cap) 574 ceph_put_cap(mdsc, new_cap); 575 576 if (!ci->i_snap_realm) { 577 /* 578 * add this inode to the appropriate snap realm 579 */ 580 struct ceph_snap_realm *realm = ceph_lookup_snap_realm(mdsc, 581 realmino); 582 if (realm) { 583 ceph_get_snap_realm(mdsc, realm); 584 spin_lock(&realm->inodes_with_caps_lock); 585 ci->i_snap_realm = realm; 586 list_add(&ci->i_snap_realm_item, 587 &realm->inodes_with_caps); 588 spin_unlock(&realm->inodes_with_caps_lock); 589 } else { 590 pr_err("ceph_add_cap: couldn't find snap realm %llx\n", 591 realmino); 592 WARN_ON(!realm); 593 } 594 } 595 596 __check_cap_issue(ci, cap, issued); 597 598 /* 599 * If we are issued caps we don't want, or the mds' wanted 600 * value appears to be off, queue a check so we'll release 601 * later and/or update the mds wanted value. 602 */ 603 actual_wanted = __ceph_caps_wanted(ci); 604 if ((wanted & ~actual_wanted) || 605 (issued & ~actual_wanted & CEPH_CAP_ANY_WR)) { 606 dout(" issued %s, mds wanted %s, actual %s, queueing\n", 607 ceph_cap_string(issued), ceph_cap_string(wanted), 608 ceph_cap_string(actual_wanted)); 609 __cap_delay_requeue(mdsc, ci); 610 } 611 612 if (flags & CEPH_CAP_FLAG_AUTH) 613 ci->i_auth_cap = cap; 614 else if (ci->i_auth_cap == cap) 615 ci->i_auth_cap = NULL; 616 617 dout("add_cap inode %p (%llx.%llx) cap %p %s now %s seq %d mds%d\n", 618 inode, ceph_vinop(inode), cap, ceph_cap_string(issued), 619 ceph_cap_string(issued|cap->issued), seq, mds); 620 cap->cap_id = cap_id; 621 cap->issued = issued; 622 cap->implemented |= issued; 623 cap->mds_wanted |= wanted; 624 cap->seq = seq; 625 cap->issue_seq = seq; 626 cap->mseq = mseq; 627 cap->cap_gen = session->s_cap_gen; 628 629 if (fmode >= 0) 630 __ceph_get_fmode(ci, fmode); 631 spin_unlock(&ci->i_ceph_lock); 632 wake_up_all(&ci->i_cap_wq); 633 return 0; 634 } 635 636 /* 637 * Return true if cap has not timed out and belongs to the current 638 * generation of the MDS session (i.e. has not gone 'stale' due to 639 * us losing touch with the mds). 640 */ 641 static int __cap_is_valid(struct ceph_cap *cap) 642 { 643 unsigned long ttl; 644 u32 gen; 645 646 spin_lock(&cap->session->s_gen_ttl_lock); 647 gen = cap->session->s_cap_gen; 648 ttl = cap->session->s_cap_ttl; 649 spin_unlock(&cap->session->s_gen_ttl_lock); 650 651 if (cap->cap_gen < gen || time_after_eq(jiffies, ttl)) { 652 dout("__cap_is_valid %p cap %p issued %s " 653 "but STALE (gen %u vs %u)\n", &cap->ci->vfs_inode, 654 cap, ceph_cap_string(cap->issued), cap->cap_gen, gen); 655 return 0; 656 } 657 658 return 1; 659 } 660 661 /* 662 * Return set of valid cap bits issued to us. Note that caps time 663 * out, and may be invalidated in bulk if the client session times out 664 * and session->s_cap_gen is bumped. 665 */ 666 int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented) 667 { 668 int have = ci->i_snap_caps | ci->i_cap_exporting_issued; 669 struct ceph_cap *cap; 670 struct rb_node *p; 671 672 if (implemented) 673 *implemented = 0; 674 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 675 cap = rb_entry(p, struct ceph_cap, ci_node); 676 if (!__cap_is_valid(cap)) 677 continue; 678 dout("__ceph_caps_issued %p cap %p issued %s\n", 679 &ci->vfs_inode, cap, ceph_cap_string(cap->issued)); 680 have |= cap->issued; 681 if (implemented) 682 *implemented |= cap->implemented; 683 } 684 return have; 685 } 686 687 /* 688 * Get cap bits issued by caps other than @ocap 689 */ 690 int __ceph_caps_issued_other(struct ceph_inode_info *ci, struct ceph_cap *ocap) 691 { 692 int have = ci->i_snap_caps; 693 struct ceph_cap *cap; 694 struct rb_node *p; 695 696 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 697 cap = rb_entry(p, struct ceph_cap, ci_node); 698 if (cap == ocap) 699 continue; 700 if (!__cap_is_valid(cap)) 701 continue; 702 have |= cap->issued; 703 } 704 return have; 705 } 706 707 /* 708 * Move a cap to the end of the LRU (oldest caps at list head, newest 709 * at list tail). 710 */ 711 static void __touch_cap(struct ceph_cap *cap) 712 { 713 struct ceph_mds_session *s = cap->session; 714 715 spin_lock(&s->s_cap_lock); 716 if (s->s_cap_iterator == NULL) { 717 dout("__touch_cap %p cap %p mds%d\n", &cap->ci->vfs_inode, cap, 718 s->s_mds); 719 list_move_tail(&cap->session_caps, &s->s_caps); 720 } else { 721 dout("__touch_cap %p cap %p mds%d NOP, iterating over caps\n", 722 &cap->ci->vfs_inode, cap, s->s_mds); 723 } 724 spin_unlock(&s->s_cap_lock); 725 } 726 727 /* 728 * Check if we hold the given mask. If so, move the cap(s) to the 729 * front of their respective LRUs. (This is the preferred way for 730 * callers to check for caps they want.) 731 */ 732 int __ceph_caps_issued_mask(struct ceph_inode_info *ci, int mask, int touch) 733 { 734 struct ceph_cap *cap; 735 struct rb_node *p; 736 int have = ci->i_snap_caps; 737 738 if ((have & mask) == mask) { 739 dout("__ceph_caps_issued_mask %p snap issued %s" 740 " (mask %s)\n", &ci->vfs_inode, 741 ceph_cap_string(have), 742 ceph_cap_string(mask)); 743 return 1; 744 } 745 746 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 747 cap = rb_entry(p, struct ceph_cap, ci_node); 748 if (!__cap_is_valid(cap)) 749 continue; 750 if ((cap->issued & mask) == mask) { 751 dout("__ceph_caps_issued_mask %p cap %p issued %s" 752 " (mask %s)\n", &ci->vfs_inode, cap, 753 ceph_cap_string(cap->issued), 754 ceph_cap_string(mask)); 755 if (touch) 756 __touch_cap(cap); 757 return 1; 758 } 759 760 /* does a combination of caps satisfy mask? */ 761 have |= cap->issued; 762 if ((have & mask) == mask) { 763 dout("__ceph_caps_issued_mask %p combo issued %s" 764 " (mask %s)\n", &ci->vfs_inode, 765 ceph_cap_string(cap->issued), 766 ceph_cap_string(mask)); 767 if (touch) { 768 struct rb_node *q; 769 770 /* touch this + preceding caps */ 771 __touch_cap(cap); 772 for (q = rb_first(&ci->i_caps); q != p; 773 q = rb_next(q)) { 774 cap = rb_entry(q, struct ceph_cap, 775 ci_node); 776 if (!__cap_is_valid(cap)) 777 continue; 778 __touch_cap(cap); 779 } 780 } 781 return 1; 782 } 783 } 784 785 return 0; 786 } 787 788 /* 789 * Return true if mask caps are currently being revoked by an MDS. 790 */ 791 int ceph_caps_revoking(struct ceph_inode_info *ci, int mask) 792 { 793 struct inode *inode = &ci->vfs_inode; 794 struct ceph_cap *cap; 795 struct rb_node *p; 796 int ret = 0; 797 798 spin_lock(&ci->i_ceph_lock); 799 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 800 cap = rb_entry(p, struct ceph_cap, ci_node); 801 if (__cap_is_valid(cap) && 802 (cap->implemented & ~cap->issued & mask)) { 803 ret = 1; 804 break; 805 } 806 } 807 spin_unlock(&ci->i_ceph_lock); 808 dout("ceph_caps_revoking %p %s = %d\n", inode, 809 ceph_cap_string(mask), ret); 810 return ret; 811 } 812 813 int __ceph_caps_used(struct ceph_inode_info *ci) 814 { 815 int used = 0; 816 if (ci->i_pin_ref) 817 used |= CEPH_CAP_PIN; 818 if (ci->i_rd_ref) 819 used |= CEPH_CAP_FILE_RD; 820 if (ci->i_rdcache_ref || ci->vfs_inode.i_data.nrpages) 821 used |= CEPH_CAP_FILE_CACHE; 822 if (ci->i_wr_ref) 823 used |= CEPH_CAP_FILE_WR; 824 if (ci->i_wb_ref || ci->i_wrbuffer_ref) 825 used |= CEPH_CAP_FILE_BUFFER; 826 return used; 827 } 828 829 /* 830 * wanted, by virtue of open file modes 831 */ 832 int __ceph_caps_file_wanted(struct ceph_inode_info *ci) 833 { 834 int want = 0; 835 int mode; 836 for (mode = 0; mode < CEPH_FILE_MODE_NUM; mode++) 837 if (ci->i_nr_by_mode[mode]) 838 want |= ceph_caps_for_mode(mode); 839 return want; 840 } 841 842 /* 843 * Return caps we have registered with the MDS(s) as 'wanted'. 844 */ 845 int __ceph_caps_mds_wanted(struct ceph_inode_info *ci) 846 { 847 struct ceph_cap *cap; 848 struct rb_node *p; 849 int mds_wanted = 0; 850 851 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 852 cap = rb_entry(p, struct ceph_cap, ci_node); 853 if (!__cap_is_valid(cap)) 854 continue; 855 mds_wanted |= cap->mds_wanted; 856 } 857 return mds_wanted; 858 } 859 860 /* 861 * called under i_ceph_lock 862 */ 863 static int __ceph_is_any_caps(struct ceph_inode_info *ci) 864 { 865 return !RB_EMPTY_ROOT(&ci->i_caps) || ci->i_cap_exporting_mds >= 0; 866 } 867 868 /* 869 * Remove a cap. Take steps to deal with a racing iterate_session_caps. 870 * 871 * caller should hold i_ceph_lock. 872 * caller will not hold session s_mutex if called from destroy_inode. 873 */ 874 void __ceph_remove_cap(struct ceph_cap *cap) 875 { 876 struct ceph_mds_session *session = cap->session; 877 struct ceph_inode_info *ci = cap->ci; 878 struct ceph_mds_client *mdsc = 879 ceph_sb_to_client(ci->vfs_inode.i_sb)->mdsc; 880 int removed = 0; 881 882 dout("__ceph_remove_cap %p from %p\n", cap, &ci->vfs_inode); 883 884 /* remove from session list */ 885 spin_lock(&session->s_cap_lock); 886 if (session->s_cap_iterator == cap) { 887 /* not yet, we are iterating over this very cap */ 888 dout("__ceph_remove_cap delaying %p removal from session %p\n", 889 cap, cap->session); 890 } else { 891 list_del_init(&cap->session_caps); 892 session->s_nr_caps--; 893 cap->session = NULL; 894 removed = 1; 895 } 896 /* protect backpointer with s_cap_lock: see iterate_session_caps */ 897 cap->ci = NULL; 898 spin_unlock(&session->s_cap_lock); 899 900 /* remove from inode list */ 901 rb_erase(&cap->ci_node, &ci->i_caps); 902 if (ci->i_auth_cap == cap) 903 ci->i_auth_cap = NULL; 904 905 if (removed) 906 ceph_put_cap(mdsc, cap); 907 908 if (!__ceph_is_any_caps(ci) && ci->i_snap_realm) { 909 struct ceph_snap_realm *realm = ci->i_snap_realm; 910 spin_lock(&realm->inodes_with_caps_lock); 911 list_del_init(&ci->i_snap_realm_item); 912 ci->i_snap_realm_counter++; 913 ci->i_snap_realm = NULL; 914 spin_unlock(&realm->inodes_with_caps_lock); 915 ceph_put_snap_realm(mdsc, realm); 916 } 917 if (!__ceph_is_any_real_caps(ci)) 918 __cap_delay_cancel(mdsc, ci); 919 } 920 921 /* 922 * Build and send a cap message to the given MDS. 923 * 924 * Caller should be holding s_mutex. 925 */ 926 static int send_cap_msg(struct ceph_mds_session *session, 927 u64 ino, u64 cid, int op, 928 int caps, int wanted, int dirty, 929 u32 seq, u64 flush_tid, u32 issue_seq, u32 mseq, 930 u64 size, u64 max_size, 931 struct timespec *mtime, struct timespec *atime, 932 u64 time_warp_seq, 933 uid_t uid, gid_t gid, umode_t mode, 934 u64 xattr_version, 935 struct ceph_buffer *xattrs_buf, 936 u64 follows) 937 { 938 struct ceph_mds_caps *fc; 939 struct ceph_msg *msg; 940 941 dout("send_cap_msg %s %llx %llx caps %s wanted %s dirty %s" 942 " seq %u/%u mseq %u follows %lld size %llu/%llu" 943 " xattr_ver %llu xattr_len %d\n", ceph_cap_op_name(op), 944 cid, ino, ceph_cap_string(caps), ceph_cap_string(wanted), 945 ceph_cap_string(dirty), 946 seq, issue_seq, mseq, follows, size, max_size, 947 xattr_version, xattrs_buf ? (int)xattrs_buf->vec.iov_len : 0); 948 949 msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, sizeof(*fc), GFP_NOFS, false); 950 if (!msg) 951 return -ENOMEM; 952 953 msg->hdr.tid = cpu_to_le64(flush_tid); 954 955 fc = msg->front.iov_base; 956 memset(fc, 0, sizeof(*fc)); 957 958 fc->cap_id = cpu_to_le64(cid); 959 fc->op = cpu_to_le32(op); 960 fc->seq = cpu_to_le32(seq); 961 fc->issue_seq = cpu_to_le32(issue_seq); 962 fc->migrate_seq = cpu_to_le32(mseq); 963 fc->caps = cpu_to_le32(caps); 964 fc->wanted = cpu_to_le32(wanted); 965 fc->dirty = cpu_to_le32(dirty); 966 fc->ino = cpu_to_le64(ino); 967 fc->snap_follows = cpu_to_le64(follows); 968 969 fc->size = cpu_to_le64(size); 970 fc->max_size = cpu_to_le64(max_size); 971 if (mtime) 972 ceph_encode_timespec(&fc->mtime, mtime); 973 if (atime) 974 ceph_encode_timespec(&fc->atime, atime); 975 fc->time_warp_seq = cpu_to_le32(time_warp_seq); 976 977 fc->uid = cpu_to_le32(uid); 978 fc->gid = cpu_to_le32(gid); 979 fc->mode = cpu_to_le32(mode); 980 981 fc->xattr_version = cpu_to_le64(xattr_version); 982 if (xattrs_buf) { 983 msg->middle = ceph_buffer_get(xattrs_buf); 984 fc->xattr_len = cpu_to_le32(xattrs_buf->vec.iov_len); 985 msg->hdr.middle_len = cpu_to_le32(xattrs_buf->vec.iov_len); 986 } 987 988 ceph_con_send(&session->s_con, msg); 989 return 0; 990 } 991 992 static void __queue_cap_release(struct ceph_mds_session *session, 993 u64 ino, u64 cap_id, u32 migrate_seq, 994 u32 issue_seq) 995 { 996 struct ceph_msg *msg; 997 struct ceph_mds_cap_release *head; 998 struct ceph_mds_cap_item *item; 999 1000 spin_lock(&session->s_cap_lock); 1001 BUG_ON(!session->s_num_cap_releases); 1002 msg = list_first_entry(&session->s_cap_releases, 1003 struct ceph_msg, list_head); 1004 1005 dout(" adding %llx release to mds%d msg %p (%d left)\n", 1006 ino, session->s_mds, msg, session->s_num_cap_releases); 1007 1008 BUG_ON(msg->front.iov_len + sizeof(*item) > PAGE_CACHE_SIZE); 1009 head = msg->front.iov_base; 1010 le32_add_cpu(&head->num, 1); 1011 item = msg->front.iov_base + msg->front.iov_len; 1012 item->ino = cpu_to_le64(ino); 1013 item->cap_id = cpu_to_le64(cap_id); 1014 item->migrate_seq = cpu_to_le32(migrate_seq); 1015 item->seq = cpu_to_le32(issue_seq); 1016 1017 session->s_num_cap_releases--; 1018 1019 msg->front.iov_len += sizeof(*item); 1020 if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) { 1021 dout(" release msg %p full\n", msg); 1022 list_move_tail(&msg->list_head, &session->s_cap_releases_done); 1023 } else { 1024 dout(" release msg %p at %d/%d (%d)\n", msg, 1025 (int)le32_to_cpu(head->num), 1026 (int)CEPH_CAPS_PER_RELEASE, 1027 (int)msg->front.iov_len); 1028 } 1029 spin_unlock(&session->s_cap_lock); 1030 } 1031 1032 /* 1033 * Queue cap releases when an inode is dropped from our cache. Since 1034 * inode is about to be destroyed, there is no need for i_ceph_lock. 1035 */ 1036 void ceph_queue_caps_release(struct inode *inode) 1037 { 1038 struct ceph_inode_info *ci = ceph_inode(inode); 1039 struct rb_node *p; 1040 1041 p = rb_first(&ci->i_caps); 1042 while (p) { 1043 struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node); 1044 struct ceph_mds_session *session = cap->session; 1045 1046 __queue_cap_release(session, ceph_ino(inode), cap->cap_id, 1047 cap->mseq, cap->issue_seq); 1048 p = rb_next(p); 1049 __ceph_remove_cap(cap); 1050 } 1051 } 1052 1053 /* 1054 * Send a cap msg on the given inode. Update our caps state, then 1055 * drop i_ceph_lock and send the message. 1056 * 1057 * Make note of max_size reported/requested from mds, revoked caps 1058 * that have now been implemented. 1059 * 1060 * Make half-hearted attempt ot to invalidate page cache if we are 1061 * dropping RDCACHE. Note that this will leave behind locked pages 1062 * that we'll then need to deal with elsewhere. 1063 * 1064 * Return non-zero if delayed release, or we experienced an error 1065 * such that the caller should requeue + retry later. 1066 * 1067 * called with i_ceph_lock, then drops it. 1068 * caller should hold snap_rwsem (read), s_mutex. 1069 */ 1070 static int __send_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap, 1071 int op, int used, int want, int retain, int flushing, 1072 unsigned *pflush_tid) 1073 __releases(cap->ci->i_ceph_lock) 1074 { 1075 struct ceph_inode_info *ci = cap->ci; 1076 struct inode *inode = &ci->vfs_inode; 1077 u64 cap_id = cap->cap_id; 1078 int held, revoking, dropping, keep; 1079 u64 seq, issue_seq, mseq, time_warp_seq, follows; 1080 u64 size, max_size; 1081 struct timespec mtime, atime; 1082 int wake = 0; 1083 umode_t mode; 1084 uid_t uid; 1085 gid_t gid; 1086 struct ceph_mds_session *session; 1087 u64 xattr_version = 0; 1088 struct ceph_buffer *xattr_blob = NULL; 1089 int delayed = 0; 1090 u64 flush_tid = 0; 1091 int i; 1092 int ret; 1093 1094 held = cap->issued | cap->implemented; 1095 revoking = cap->implemented & ~cap->issued; 1096 retain &= ~revoking; 1097 dropping = cap->issued & ~retain; 1098 1099 dout("__send_cap %p cap %p session %p %s -> %s (revoking %s)\n", 1100 inode, cap, cap->session, 1101 ceph_cap_string(held), ceph_cap_string(held & retain), 1102 ceph_cap_string(revoking)); 1103 BUG_ON((retain & CEPH_CAP_PIN) == 0); 1104 1105 session = cap->session; 1106 1107 /* don't release wanted unless we've waited a bit. */ 1108 if ((ci->i_ceph_flags & CEPH_I_NODELAY) == 0 && 1109 time_before(jiffies, ci->i_hold_caps_min)) { 1110 dout(" delaying issued %s -> %s, wanted %s -> %s on send\n", 1111 ceph_cap_string(cap->issued), 1112 ceph_cap_string(cap->issued & retain), 1113 ceph_cap_string(cap->mds_wanted), 1114 ceph_cap_string(want)); 1115 want |= cap->mds_wanted; 1116 retain |= cap->issued; 1117 delayed = 1; 1118 } 1119 ci->i_ceph_flags &= ~(CEPH_I_NODELAY | CEPH_I_FLUSH); 1120 1121 cap->issued &= retain; /* drop bits we don't want */ 1122 if (cap->implemented & ~cap->issued) { 1123 /* 1124 * Wake up any waiters on wanted -> needed transition. 1125 * This is due to the weird transition from buffered 1126 * to sync IO... we need to flush dirty pages _before_ 1127 * allowing sync writes to avoid reordering. 1128 */ 1129 wake = 1; 1130 } 1131 cap->implemented &= cap->issued | used; 1132 cap->mds_wanted = want; 1133 1134 if (flushing) { 1135 /* 1136 * assign a tid for flush operations so we can avoid 1137 * flush1 -> dirty1 -> flush2 -> flushack1 -> mark 1138 * clean type races. track latest tid for every bit 1139 * so we can handle flush AxFw, flush Fw, and have the 1140 * first ack clean Ax. 1141 */ 1142 flush_tid = ++ci->i_cap_flush_last_tid; 1143 if (pflush_tid) 1144 *pflush_tid = flush_tid; 1145 dout(" cap_flush_tid %d\n", (int)flush_tid); 1146 for (i = 0; i < CEPH_CAP_BITS; i++) 1147 if (flushing & (1 << i)) 1148 ci->i_cap_flush_tid[i] = flush_tid; 1149 1150 follows = ci->i_head_snapc->seq; 1151 } else { 1152 follows = 0; 1153 } 1154 1155 keep = cap->implemented; 1156 seq = cap->seq; 1157 issue_seq = cap->issue_seq; 1158 mseq = cap->mseq; 1159 size = inode->i_size; 1160 ci->i_reported_size = size; 1161 max_size = ci->i_wanted_max_size; 1162 ci->i_requested_max_size = max_size; 1163 mtime = inode->i_mtime; 1164 atime = inode->i_atime; 1165 time_warp_seq = ci->i_time_warp_seq; 1166 uid = inode->i_uid; 1167 gid = inode->i_gid; 1168 mode = inode->i_mode; 1169 1170 if (flushing & CEPH_CAP_XATTR_EXCL) { 1171 __ceph_build_xattrs_blob(ci); 1172 xattr_blob = ci->i_xattrs.blob; 1173 xattr_version = ci->i_xattrs.version; 1174 } 1175 1176 spin_unlock(&ci->i_ceph_lock); 1177 1178 ret = send_cap_msg(session, ceph_vino(inode).ino, cap_id, 1179 op, keep, want, flushing, seq, flush_tid, issue_seq, mseq, 1180 size, max_size, &mtime, &atime, time_warp_seq, 1181 uid, gid, mode, xattr_version, xattr_blob, 1182 follows); 1183 if (ret < 0) { 1184 dout("error sending cap msg, must requeue %p\n", inode); 1185 delayed = 1; 1186 } 1187 1188 if (wake) 1189 wake_up_all(&ci->i_cap_wq); 1190 1191 return delayed; 1192 } 1193 1194 /* 1195 * When a snapshot is taken, clients accumulate dirty metadata on 1196 * inodes with capabilities in ceph_cap_snaps to describe the file 1197 * state at the time the snapshot was taken. This must be flushed 1198 * asynchronously back to the MDS once sync writes complete and dirty 1199 * data is written out. 1200 * 1201 * Unless @again is true, skip cap_snaps that were already sent to 1202 * the MDS (i.e., during this session). 1203 * 1204 * Called under i_ceph_lock. Takes s_mutex as needed. 1205 */ 1206 void __ceph_flush_snaps(struct ceph_inode_info *ci, 1207 struct ceph_mds_session **psession, 1208 int again) 1209 __releases(ci->i_ceph_lock) 1210 __acquires(ci->i_ceph_lock) 1211 { 1212 struct inode *inode = &ci->vfs_inode; 1213 int mds; 1214 struct ceph_cap_snap *capsnap; 1215 u32 mseq; 1216 struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc; 1217 struct ceph_mds_session *session = NULL; /* if session != NULL, we hold 1218 session->s_mutex */ 1219 u64 next_follows = 0; /* keep track of how far we've gotten through the 1220 i_cap_snaps list, and skip these entries next time 1221 around to avoid an infinite loop */ 1222 1223 if (psession) 1224 session = *psession; 1225 1226 dout("__flush_snaps %p\n", inode); 1227 retry: 1228 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) { 1229 /* avoid an infiniute loop after retry */ 1230 if (capsnap->follows < next_follows) 1231 continue; 1232 /* 1233 * we need to wait for sync writes to complete and for dirty 1234 * pages to be written out. 1235 */ 1236 if (capsnap->dirty_pages || capsnap->writing) 1237 break; 1238 1239 /* 1240 * if cap writeback already occurred, we should have dropped 1241 * the capsnap in ceph_put_wrbuffer_cap_refs. 1242 */ 1243 BUG_ON(capsnap->dirty == 0); 1244 1245 /* pick mds, take s_mutex */ 1246 if (ci->i_auth_cap == NULL) { 1247 dout("no auth cap (migrating?), doing nothing\n"); 1248 goto out; 1249 } 1250 1251 /* only flush each capsnap once */ 1252 if (!again && !list_empty(&capsnap->flushing_item)) { 1253 dout("already flushed %p, skipping\n", capsnap); 1254 continue; 1255 } 1256 1257 mds = ci->i_auth_cap->session->s_mds; 1258 mseq = ci->i_auth_cap->mseq; 1259 1260 if (session && session->s_mds != mds) { 1261 dout("oops, wrong session %p mutex\n", session); 1262 mutex_unlock(&session->s_mutex); 1263 ceph_put_mds_session(session); 1264 session = NULL; 1265 } 1266 if (!session) { 1267 spin_unlock(&ci->i_ceph_lock); 1268 mutex_lock(&mdsc->mutex); 1269 session = __ceph_lookup_mds_session(mdsc, mds); 1270 mutex_unlock(&mdsc->mutex); 1271 if (session) { 1272 dout("inverting session/ino locks on %p\n", 1273 session); 1274 mutex_lock(&session->s_mutex); 1275 } 1276 /* 1277 * if session == NULL, we raced against a cap 1278 * deletion or migration. retry, and we'll 1279 * get a better @mds value next time. 1280 */ 1281 spin_lock(&ci->i_ceph_lock); 1282 goto retry; 1283 } 1284 1285 capsnap->flush_tid = ++ci->i_cap_flush_last_tid; 1286 atomic_inc(&capsnap->nref); 1287 if (!list_empty(&capsnap->flushing_item)) 1288 list_del_init(&capsnap->flushing_item); 1289 list_add_tail(&capsnap->flushing_item, 1290 &session->s_cap_snaps_flushing); 1291 spin_unlock(&ci->i_ceph_lock); 1292 1293 dout("flush_snaps %p cap_snap %p follows %lld tid %llu\n", 1294 inode, capsnap, capsnap->follows, capsnap->flush_tid); 1295 send_cap_msg(session, ceph_vino(inode).ino, 0, 1296 CEPH_CAP_OP_FLUSHSNAP, capsnap->issued, 0, 1297 capsnap->dirty, 0, capsnap->flush_tid, 0, mseq, 1298 capsnap->size, 0, 1299 &capsnap->mtime, &capsnap->atime, 1300 capsnap->time_warp_seq, 1301 capsnap->uid, capsnap->gid, capsnap->mode, 1302 capsnap->xattr_version, capsnap->xattr_blob, 1303 capsnap->follows); 1304 1305 next_follows = capsnap->follows + 1; 1306 ceph_put_cap_snap(capsnap); 1307 1308 spin_lock(&ci->i_ceph_lock); 1309 goto retry; 1310 } 1311 1312 /* we flushed them all; remove this inode from the queue */ 1313 spin_lock(&mdsc->snap_flush_lock); 1314 list_del_init(&ci->i_snap_flush_item); 1315 spin_unlock(&mdsc->snap_flush_lock); 1316 1317 out: 1318 if (psession) 1319 *psession = session; 1320 else if (session) { 1321 mutex_unlock(&session->s_mutex); 1322 ceph_put_mds_session(session); 1323 } 1324 } 1325 1326 static void ceph_flush_snaps(struct ceph_inode_info *ci) 1327 { 1328 spin_lock(&ci->i_ceph_lock); 1329 __ceph_flush_snaps(ci, NULL, 0); 1330 spin_unlock(&ci->i_ceph_lock); 1331 } 1332 1333 /* 1334 * Mark caps dirty. If inode is newly dirty, return the dirty flags. 1335 * Caller is then responsible for calling __mark_inode_dirty with the 1336 * returned flags value. 1337 */ 1338 int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask) 1339 { 1340 struct ceph_mds_client *mdsc = 1341 ceph_sb_to_client(ci->vfs_inode.i_sb)->mdsc; 1342 struct inode *inode = &ci->vfs_inode; 1343 int was = ci->i_dirty_caps; 1344 int dirty = 0; 1345 1346 dout("__mark_dirty_caps %p %s dirty %s -> %s\n", &ci->vfs_inode, 1347 ceph_cap_string(mask), ceph_cap_string(was), 1348 ceph_cap_string(was | mask)); 1349 ci->i_dirty_caps |= mask; 1350 if (was == 0) { 1351 if (!ci->i_head_snapc) 1352 ci->i_head_snapc = ceph_get_snap_context( 1353 ci->i_snap_realm->cached_context); 1354 dout(" inode %p now dirty snapc %p auth cap %p\n", 1355 &ci->vfs_inode, ci->i_head_snapc, ci->i_auth_cap); 1356 BUG_ON(!list_empty(&ci->i_dirty_item)); 1357 spin_lock(&mdsc->cap_dirty_lock); 1358 if (ci->i_auth_cap) 1359 list_add(&ci->i_dirty_item, &mdsc->cap_dirty); 1360 else 1361 list_add(&ci->i_dirty_item, 1362 &mdsc->cap_dirty_migrating); 1363 spin_unlock(&mdsc->cap_dirty_lock); 1364 if (ci->i_flushing_caps == 0) { 1365 ihold(inode); 1366 dirty |= I_DIRTY_SYNC; 1367 } 1368 } 1369 BUG_ON(list_empty(&ci->i_dirty_item)); 1370 if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) && 1371 (mask & CEPH_CAP_FILE_BUFFER)) 1372 dirty |= I_DIRTY_DATASYNC; 1373 __cap_delay_requeue(mdsc, ci); 1374 return dirty; 1375 } 1376 1377 /* 1378 * Add dirty inode to the flushing list. Assigned a seq number so we 1379 * can wait for caps to flush without starving. 1380 * 1381 * Called under i_ceph_lock. 1382 */ 1383 static int __mark_caps_flushing(struct inode *inode, 1384 struct ceph_mds_session *session) 1385 { 1386 struct ceph_mds_client *mdsc = ceph_sb_to_client(inode->i_sb)->mdsc; 1387 struct ceph_inode_info *ci = ceph_inode(inode); 1388 int flushing; 1389 1390 BUG_ON(ci->i_dirty_caps == 0); 1391 BUG_ON(list_empty(&ci->i_dirty_item)); 1392 1393 flushing = ci->i_dirty_caps; 1394 dout("__mark_caps_flushing flushing %s, flushing_caps %s -> %s\n", 1395 ceph_cap_string(flushing), 1396 ceph_cap_string(ci->i_flushing_caps), 1397 ceph_cap_string(ci->i_flushing_caps | flushing)); 1398 ci->i_flushing_caps |= flushing; 1399 ci->i_dirty_caps = 0; 1400 dout(" inode %p now !dirty\n", inode); 1401 1402 spin_lock(&mdsc->cap_dirty_lock); 1403 list_del_init(&ci->i_dirty_item); 1404 1405 ci->i_cap_flush_seq = ++mdsc->cap_flush_seq; 1406 if (list_empty(&ci->i_flushing_item)) { 1407 list_add_tail(&ci->i_flushing_item, &session->s_cap_flushing); 1408 mdsc->num_cap_flushing++; 1409 dout(" inode %p now flushing seq %lld\n", inode, 1410 ci->i_cap_flush_seq); 1411 } else { 1412 list_move_tail(&ci->i_flushing_item, &session->s_cap_flushing); 1413 dout(" inode %p now flushing (more) seq %lld\n", inode, 1414 ci->i_cap_flush_seq); 1415 } 1416 spin_unlock(&mdsc->cap_dirty_lock); 1417 1418 return flushing; 1419 } 1420 1421 /* 1422 * try to invalidate mapping pages without blocking. 1423 */ 1424 static int try_nonblocking_invalidate(struct inode *inode) 1425 { 1426 struct ceph_inode_info *ci = ceph_inode(inode); 1427 u32 invalidating_gen = ci->i_rdcache_gen; 1428 1429 spin_unlock(&ci->i_ceph_lock); 1430 invalidate_mapping_pages(&inode->i_data, 0, -1); 1431 spin_lock(&ci->i_ceph_lock); 1432 1433 if (inode->i_data.nrpages == 0 && 1434 invalidating_gen == ci->i_rdcache_gen) { 1435 /* success. */ 1436 dout("try_nonblocking_invalidate %p success\n", inode); 1437 /* save any racing async invalidate some trouble */ 1438 ci->i_rdcache_revoking = ci->i_rdcache_gen - 1; 1439 return 0; 1440 } 1441 dout("try_nonblocking_invalidate %p failed\n", inode); 1442 return -1; 1443 } 1444 1445 /* 1446 * Swiss army knife function to examine currently used and wanted 1447 * versus held caps. Release, flush, ack revoked caps to mds as 1448 * appropriate. 1449 * 1450 * CHECK_CAPS_NODELAY - caller is delayed work and we should not delay 1451 * cap release further. 1452 * CHECK_CAPS_AUTHONLY - we should only check the auth cap 1453 * CHECK_CAPS_FLUSH - we should flush any dirty caps immediately, without 1454 * further delay. 1455 */ 1456 void ceph_check_caps(struct ceph_inode_info *ci, int flags, 1457 struct ceph_mds_session *session) 1458 { 1459 struct ceph_fs_client *fsc = ceph_inode_to_client(&ci->vfs_inode); 1460 struct ceph_mds_client *mdsc = fsc->mdsc; 1461 struct inode *inode = &ci->vfs_inode; 1462 struct ceph_cap *cap; 1463 int file_wanted, used; 1464 int took_snap_rwsem = 0; /* true if mdsc->snap_rwsem held */ 1465 int issued, implemented, want, retain, revoking, flushing = 0; 1466 int mds = -1; /* keep track of how far we've gone through i_caps list 1467 to avoid an infinite loop on retry */ 1468 struct rb_node *p; 1469 int tried_invalidate = 0; 1470 int delayed = 0, sent = 0, force_requeue = 0, num; 1471 int queue_invalidate = 0; 1472 int is_delayed = flags & CHECK_CAPS_NODELAY; 1473 1474 /* if we are unmounting, flush any unused caps immediately. */ 1475 if (mdsc->stopping) 1476 is_delayed = 1; 1477 1478 spin_lock(&ci->i_ceph_lock); 1479 1480 if (ci->i_ceph_flags & CEPH_I_FLUSH) 1481 flags |= CHECK_CAPS_FLUSH; 1482 1483 /* flush snaps first time around only */ 1484 if (!list_empty(&ci->i_cap_snaps)) 1485 __ceph_flush_snaps(ci, &session, 0); 1486 goto retry_locked; 1487 retry: 1488 spin_lock(&ci->i_ceph_lock); 1489 retry_locked: 1490 file_wanted = __ceph_caps_file_wanted(ci); 1491 used = __ceph_caps_used(ci); 1492 want = file_wanted | used; 1493 issued = __ceph_caps_issued(ci, &implemented); 1494 revoking = implemented & ~issued; 1495 1496 retain = want | CEPH_CAP_PIN; 1497 if (!mdsc->stopping && inode->i_nlink > 0) { 1498 if (want) { 1499 retain |= CEPH_CAP_ANY; /* be greedy */ 1500 } else { 1501 retain |= CEPH_CAP_ANY_SHARED; 1502 /* 1503 * keep RD only if we didn't have the file open RW, 1504 * because then the mds would revoke it anyway to 1505 * journal max_size=0. 1506 */ 1507 if (ci->i_max_size == 0) 1508 retain |= CEPH_CAP_ANY_RD; 1509 } 1510 } 1511 1512 dout("check_caps %p file_want %s used %s dirty %s flushing %s" 1513 " issued %s revoking %s retain %s %s%s%s\n", inode, 1514 ceph_cap_string(file_wanted), 1515 ceph_cap_string(used), ceph_cap_string(ci->i_dirty_caps), 1516 ceph_cap_string(ci->i_flushing_caps), 1517 ceph_cap_string(issued), ceph_cap_string(revoking), 1518 ceph_cap_string(retain), 1519 (flags & CHECK_CAPS_AUTHONLY) ? " AUTHONLY" : "", 1520 (flags & CHECK_CAPS_NODELAY) ? " NODELAY" : "", 1521 (flags & CHECK_CAPS_FLUSH) ? " FLUSH" : ""); 1522 1523 /* 1524 * If we no longer need to hold onto old our caps, and we may 1525 * have cached pages, but don't want them, then try to invalidate. 1526 * If we fail, it's because pages are locked.... try again later. 1527 */ 1528 if ((!is_delayed || mdsc->stopping) && 1529 ci->i_wrbuffer_ref == 0 && /* no dirty pages... */ 1530 inode->i_data.nrpages && /* have cached pages */ 1531 (file_wanted == 0 || /* no open files */ 1532 (revoking & (CEPH_CAP_FILE_CACHE| 1533 CEPH_CAP_FILE_LAZYIO))) && /* or revoking cache */ 1534 !tried_invalidate) { 1535 dout("check_caps trying to invalidate on %p\n", inode); 1536 if (try_nonblocking_invalidate(inode) < 0) { 1537 if (revoking & (CEPH_CAP_FILE_CACHE| 1538 CEPH_CAP_FILE_LAZYIO)) { 1539 dout("check_caps queuing invalidate\n"); 1540 queue_invalidate = 1; 1541 ci->i_rdcache_revoking = ci->i_rdcache_gen; 1542 } else { 1543 dout("check_caps failed to invalidate pages\n"); 1544 /* we failed to invalidate pages. check these 1545 caps again later. */ 1546 force_requeue = 1; 1547 __cap_set_timeouts(mdsc, ci); 1548 } 1549 } 1550 tried_invalidate = 1; 1551 goto retry_locked; 1552 } 1553 1554 num = 0; 1555 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 1556 cap = rb_entry(p, struct ceph_cap, ci_node); 1557 num++; 1558 1559 /* avoid looping forever */ 1560 if (mds >= cap->mds || 1561 ((flags & CHECK_CAPS_AUTHONLY) && cap != ci->i_auth_cap)) 1562 continue; 1563 1564 /* NOTE: no side-effects allowed, until we take s_mutex */ 1565 1566 revoking = cap->implemented & ~cap->issued; 1567 dout(" mds%d cap %p issued %s implemented %s revoking %s\n", 1568 cap->mds, cap, ceph_cap_string(cap->issued), 1569 ceph_cap_string(cap->implemented), 1570 ceph_cap_string(revoking)); 1571 1572 if (cap == ci->i_auth_cap && 1573 (cap->issued & CEPH_CAP_FILE_WR)) { 1574 /* request larger max_size from MDS? */ 1575 if (ci->i_wanted_max_size > ci->i_max_size && 1576 ci->i_wanted_max_size > ci->i_requested_max_size) { 1577 dout("requesting new max_size\n"); 1578 goto ack; 1579 } 1580 1581 /* approaching file_max? */ 1582 if ((inode->i_size << 1) >= ci->i_max_size && 1583 (ci->i_reported_size << 1) < ci->i_max_size) { 1584 dout("i_size approaching max_size\n"); 1585 goto ack; 1586 } 1587 } 1588 /* flush anything dirty? */ 1589 if (cap == ci->i_auth_cap && (flags & CHECK_CAPS_FLUSH) && 1590 ci->i_dirty_caps) { 1591 dout("flushing dirty caps\n"); 1592 goto ack; 1593 } 1594 1595 /* completed revocation? going down and there are no caps? */ 1596 if (revoking && (revoking & used) == 0) { 1597 dout("completed revocation of %s\n", 1598 ceph_cap_string(cap->implemented & ~cap->issued)); 1599 goto ack; 1600 } 1601 1602 /* want more caps from mds? */ 1603 if (want & ~(cap->mds_wanted | cap->issued)) 1604 goto ack; 1605 1606 /* things we might delay */ 1607 if ((cap->issued & ~retain) == 0 && 1608 cap->mds_wanted == want) 1609 continue; /* nope, all good */ 1610 1611 if (is_delayed) 1612 goto ack; 1613 1614 /* delay? */ 1615 if ((ci->i_ceph_flags & CEPH_I_NODELAY) == 0 && 1616 time_before(jiffies, ci->i_hold_caps_max)) { 1617 dout(" delaying issued %s -> %s, wanted %s -> %s\n", 1618 ceph_cap_string(cap->issued), 1619 ceph_cap_string(cap->issued & retain), 1620 ceph_cap_string(cap->mds_wanted), 1621 ceph_cap_string(want)); 1622 delayed++; 1623 continue; 1624 } 1625 1626 ack: 1627 if (ci->i_ceph_flags & CEPH_I_NOFLUSH) { 1628 dout(" skipping %p I_NOFLUSH set\n", inode); 1629 continue; 1630 } 1631 1632 if (session && session != cap->session) { 1633 dout("oops, wrong session %p mutex\n", session); 1634 mutex_unlock(&session->s_mutex); 1635 session = NULL; 1636 } 1637 if (!session) { 1638 session = cap->session; 1639 if (mutex_trylock(&session->s_mutex) == 0) { 1640 dout("inverting session/ino locks on %p\n", 1641 session); 1642 spin_unlock(&ci->i_ceph_lock); 1643 if (took_snap_rwsem) { 1644 up_read(&mdsc->snap_rwsem); 1645 took_snap_rwsem = 0; 1646 } 1647 mutex_lock(&session->s_mutex); 1648 goto retry; 1649 } 1650 } 1651 /* take snap_rwsem after session mutex */ 1652 if (!took_snap_rwsem) { 1653 if (down_read_trylock(&mdsc->snap_rwsem) == 0) { 1654 dout("inverting snap/in locks on %p\n", 1655 inode); 1656 spin_unlock(&ci->i_ceph_lock); 1657 down_read(&mdsc->snap_rwsem); 1658 took_snap_rwsem = 1; 1659 goto retry; 1660 } 1661 took_snap_rwsem = 1; 1662 } 1663 1664 if (cap == ci->i_auth_cap && ci->i_dirty_caps) 1665 flushing = __mark_caps_flushing(inode, session); 1666 else 1667 flushing = 0; 1668 1669 mds = cap->mds; /* remember mds, so we don't repeat */ 1670 sent++; 1671 1672 /* __send_cap drops i_ceph_lock */ 1673 delayed += __send_cap(mdsc, cap, CEPH_CAP_OP_UPDATE, used, want, 1674 retain, flushing, NULL); 1675 goto retry; /* retake i_ceph_lock and restart our cap scan. */ 1676 } 1677 1678 /* 1679 * Reschedule delayed caps release if we delayed anything, 1680 * otherwise cancel. 1681 */ 1682 if (delayed && is_delayed) 1683 force_requeue = 1; /* __send_cap delayed release; requeue */ 1684 if (!delayed && !is_delayed) 1685 __cap_delay_cancel(mdsc, ci); 1686 else if (!is_delayed || force_requeue) 1687 __cap_delay_requeue(mdsc, ci); 1688 1689 spin_unlock(&ci->i_ceph_lock); 1690 1691 if (queue_invalidate) 1692 ceph_queue_invalidate(inode); 1693 1694 if (session) 1695 mutex_unlock(&session->s_mutex); 1696 if (took_snap_rwsem) 1697 up_read(&mdsc->snap_rwsem); 1698 } 1699 1700 /* 1701 * Try to flush dirty caps back to the auth mds. 1702 */ 1703 static int try_flush_caps(struct inode *inode, struct ceph_mds_session *session, 1704 unsigned *flush_tid) 1705 { 1706 struct ceph_mds_client *mdsc = ceph_sb_to_client(inode->i_sb)->mdsc; 1707 struct ceph_inode_info *ci = ceph_inode(inode); 1708 int unlock_session = session ? 0 : 1; 1709 int flushing = 0; 1710 1711 retry: 1712 spin_lock(&ci->i_ceph_lock); 1713 if (ci->i_ceph_flags & CEPH_I_NOFLUSH) { 1714 dout("try_flush_caps skipping %p I_NOFLUSH set\n", inode); 1715 goto out; 1716 } 1717 if (ci->i_dirty_caps && ci->i_auth_cap) { 1718 struct ceph_cap *cap = ci->i_auth_cap; 1719 int used = __ceph_caps_used(ci); 1720 int want = __ceph_caps_wanted(ci); 1721 int delayed; 1722 1723 if (!session) { 1724 spin_unlock(&ci->i_ceph_lock); 1725 session = cap->session; 1726 mutex_lock(&session->s_mutex); 1727 goto retry; 1728 } 1729 BUG_ON(session != cap->session); 1730 if (cap->session->s_state < CEPH_MDS_SESSION_OPEN) 1731 goto out; 1732 1733 flushing = __mark_caps_flushing(inode, session); 1734 1735 /* __send_cap drops i_ceph_lock */ 1736 delayed = __send_cap(mdsc, cap, CEPH_CAP_OP_FLUSH, used, want, 1737 cap->issued | cap->implemented, flushing, 1738 flush_tid); 1739 if (!delayed) 1740 goto out_unlocked; 1741 1742 spin_lock(&ci->i_ceph_lock); 1743 __cap_delay_requeue(mdsc, ci); 1744 } 1745 out: 1746 spin_unlock(&ci->i_ceph_lock); 1747 out_unlocked: 1748 if (session && unlock_session) 1749 mutex_unlock(&session->s_mutex); 1750 return flushing; 1751 } 1752 1753 /* 1754 * Return true if we've flushed caps through the given flush_tid. 1755 */ 1756 static int caps_are_flushed(struct inode *inode, unsigned tid) 1757 { 1758 struct ceph_inode_info *ci = ceph_inode(inode); 1759 int i, ret = 1; 1760 1761 spin_lock(&ci->i_ceph_lock); 1762 for (i = 0; i < CEPH_CAP_BITS; i++) 1763 if ((ci->i_flushing_caps & (1 << i)) && 1764 ci->i_cap_flush_tid[i] <= tid) { 1765 /* still flushing this bit */ 1766 ret = 0; 1767 break; 1768 } 1769 spin_unlock(&ci->i_ceph_lock); 1770 return ret; 1771 } 1772 1773 /* 1774 * Wait on any unsafe replies for the given inode. First wait on the 1775 * newest request, and make that the upper bound. Then, if there are 1776 * more requests, keep waiting on the oldest as long as it is still older 1777 * than the original request. 1778 */ 1779 static void sync_write_wait(struct inode *inode) 1780 { 1781 struct ceph_inode_info *ci = ceph_inode(inode); 1782 struct list_head *head = &ci->i_unsafe_writes; 1783 struct ceph_osd_request *req; 1784 u64 last_tid; 1785 1786 spin_lock(&ci->i_unsafe_lock); 1787 if (list_empty(head)) 1788 goto out; 1789 1790 /* set upper bound as _last_ entry in chain */ 1791 req = list_entry(head->prev, struct ceph_osd_request, 1792 r_unsafe_item); 1793 last_tid = req->r_tid; 1794 1795 do { 1796 ceph_osdc_get_request(req); 1797 spin_unlock(&ci->i_unsafe_lock); 1798 dout("sync_write_wait on tid %llu (until %llu)\n", 1799 req->r_tid, last_tid); 1800 wait_for_completion(&req->r_safe_completion); 1801 spin_lock(&ci->i_unsafe_lock); 1802 ceph_osdc_put_request(req); 1803 1804 /* 1805 * from here on look at first entry in chain, since we 1806 * only want to wait for anything older than last_tid 1807 */ 1808 if (list_empty(head)) 1809 break; 1810 req = list_entry(head->next, struct ceph_osd_request, 1811 r_unsafe_item); 1812 } while (req->r_tid < last_tid); 1813 out: 1814 spin_unlock(&ci->i_unsafe_lock); 1815 } 1816 1817 int ceph_fsync(struct file *file, loff_t start, loff_t end, int datasync) 1818 { 1819 struct inode *inode = file->f_mapping->host; 1820 struct ceph_inode_info *ci = ceph_inode(inode); 1821 unsigned flush_tid; 1822 int ret; 1823 int dirty; 1824 1825 dout("fsync %p%s\n", inode, datasync ? " datasync" : ""); 1826 sync_write_wait(inode); 1827 1828 ret = filemap_write_and_wait_range(inode->i_mapping, start, end); 1829 if (ret < 0) 1830 return ret; 1831 mutex_lock(&inode->i_mutex); 1832 1833 dirty = try_flush_caps(inode, NULL, &flush_tid); 1834 dout("fsync dirty caps are %s\n", ceph_cap_string(dirty)); 1835 1836 /* 1837 * only wait on non-file metadata writeback (the mds 1838 * can recover size and mtime, so we don't need to 1839 * wait for that) 1840 */ 1841 if (!datasync && (dirty & ~CEPH_CAP_ANY_FILE_WR)) { 1842 dout("fsync waiting for flush_tid %u\n", flush_tid); 1843 ret = wait_event_interruptible(ci->i_cap_wq, 1844 caps_are_flushed(inode, flush_tid)); 1845 } 1846 1847 dout("fsync %p%s done\n", inode, datasync ? " datasync" : ""); 1848 mutex_unlock(&inode->i_mutex); 1849 return ret; 1850 } 1851 1852 /* 1853 * Flush any dirty caps back to the mds. If we aren't asked to wait, 1854 * queue inode for flush but don't do so immediately, because we can 1855 * get by with fewer MDS messages if we wait for data writeback to 1856 * complete first. 1857 */ 1858 int ceph_write_inode(struct inode *inode, struct writeback_control *wbc) 1859 { 1860 struct ceph_inode_info *ci = ceph_inode(inode); 1861 unsigned flush_tid; 1862 int err = 0; 1863 int dirty; 1864 int wait = wbc->sync_mode == WB_SYNC_ALL; 1865 1866 dout("write_inode %p wait=%d\n", inode, wait); 1867 if (wait) { 1868 dirty = try_flush_caps(inode, NULL, &flush_tid); 1869 if (dirty) 1870 err = wait_event_interruptible(ci->i_cap_wq, 1871 caps_are_flushed(inode, flush_tid)); 1872 } else { 1873 struct ceph_mds_client *mdsc = 1874 ceph_sb_to_client(inode->i_sb)->mdsc; 1875 1876 spin_lock(&ci->i_ceph_lock); 1877 if (__ceph_caps_dirty(ci)) 1878 __cap_delay_requeue_front(mdsc, ci); 1879 spin_unlock(&ci->i_ceph_lock); 1880 } 1881 return err; 1882 } 1883 1884 /* 1885 * After a recovering MDS goes active, we need to resend any caps 1886 * we were flushing. 1887 * 1888 * Caller holds session->s_mutex. 1889 */ 1890 static void kick_flushing_capsnaps(struct ceph_mds_client *mdsc, 1891 struct ceph_mds_session *session) 1892 { 1893 struct ceph_cap_snap *capsnap; 1894 1895 dout("kick_flushing_capsnaps mds%d\n", session->s_mds); 1896 list_for_each_entry(capsnap, &session->s_cap_snaps_flushing, 1897 flushing_item) { 1898 struct ceph_inode_info *ci = capsnap->ci; 1899 struct inode *inode = &ci->vfs_inode; 1900 struct ceph_cap *cap; 1901 1902 spin_lock(&ci->i_ceph_lock); 1903 cap = ci->i_auth_cap; 1904 if (cap && cap->session == session) { 1905 dout("kick_flushing_caps %p cap %p capsnap %p\n", inode, 1906 cap, capsnap); 1907 __ceph_flush_snaps(ci, &session, 1); 1908 } else { 1909 pr_err("%p auth cap %p not mds%d ???\n", inode, 1910 cap, session->s_mds); 1911 } 1912 spin_unlock(&ci->i_ceph_lock); 1913 } 1914 } 1915 1916 void ceph_kick_flushing_caps(struct ceph_mds_client *mdsc, 1917 struct ceph_mds_session *session) 1918 { 1919 struct ceph_inode_info *ci; 1920 1921 kick_flushing_capsnaps(mdsc, session); 1922 1923 dout("kick_flushing_caps mds%d\n", session->s_mds); 1924 list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) { 1925 struct inode *inode = &ci->vfs_inode; 1926 struct ceph_cap *cap; 1927 int delayed = 0; 1928 1929 spin_lock(&ci->i_ceph_lock); 1930 cap = ci->i_auth_cap; 1931 if (cap && cap->session == session) { 1932 dout("kick_flushing_caps %p cap %p %s\n", inode, 1933 cap, ceph_cap_string(ci->i_flushing_caps)); 1934 delayed = __send_cap(mdsc, cap, CEPH_CAP_OP_FLUSH, 1935 __ceph_caps_used(ci), 1936 __ceph_caps_wanted(ci), 1937 cap->issued | cap->implemented, 1938 ci->i_flushing_caps, NULL); 1939 if (delayed) { 1940 spin_lock(&ci->i_ceph_lock); 1941 __cap_delay_requeue(mdsc, ci); 1942 spin_unlock(&ci->i_ceph_lock); 1943 } 1944 } else { 1945 pr_err("%p auth cap %p not mds%d ???\n", inode, 1946 cap, session->s_mds); 1947 spin_unlock(&ci->i_ceph_lock); 1948 } 1949 } 1950 } 1951 1952 static void kick_flushing_inode_caps(struct ceph_mds_client *mdsc, 1953 struct ceph_mds_session *session, 1954 struct inode *inode) 1955 { 1956 struct ceph_inode_info *ci = ceph_inode(inode); 1957 struct ceph_cap *cap; 1958 int delayed = 0; 1959 1960 spin_lock(&ci->i_ceph_lock); 1961 cap = ci->i_auth_cap; 1962 dout("kick_flushing_inode_caps %p flushing %s flush_seq %lld\n", inode, 1963 ceph_cap_string(ci->i_flushing_caps), ci->i_cap_flush_seq); 1964 __ceph_flush_snaps(ci, &session, 1); 1965 if (ci->i_flushing_caps) { 1966 delayed = __send_cap(mdsc, cap, CEPH_CAP_OP_FLUSH, 1967 __ceph_caps_used(ci), 1968 __ceph_caps_wanted(ci), 1969 cap->issued | cap->implemented, 1970 ci->i_flushing_caps, NULL); 1971 if (delayed) { 1972 spin_lock(&ci->i_ceph_lock); 1973 __cap_delay_requeue(mdsc, ci); 1974 spin_unlock(&ci->i_ceph_lock); 1975 } 1976 } else { 1977 spin_unlock(&ci->i_ceph_lock); 1978 } 1979 } 1980 1981 1982 /* 1983 * Take references to capabilities we hold, so that we don't release 1984 * them to the MDS prematurely. 1985 * 1986 * Protected by i_ceph_lock. 1987 */ 1988 static void __take_cap_refs(struct ceph_inode_info *ci, int got) 1989 { 1990 if (got & CEPH_CAP_PIN) 1991 ci->i_pin_ref++; 1992 if (got & CEPH_CAP_FILE_RD) 1993 ci->i_rd_ref++; 1994 if (got & CEPH_CAP_FILE_CACHE) 1995 ci->i_rdcache_ref++; 1996 if (got & CEPH_CAP_FILE_WR) 1997 ci->i_wr_ref++; 1998 if (got & CEPH_CAP_FILE_BUFFER) { 1999 if (ci->i_wb_ref == 0) 2000 ihold(&ci->vfs_inode); 2001 ci->i_wb_ref++; 2002 dout("__take_cap_refs %p wb %d -> %d (?)\n", 2003 &ci->vfs_inode, ci->i_wb_ref-1, ci->i_wb_ref); 2004 } 2005 } 2006 2007 /* 2008 * Try to grab cap references. Specify those refs we @want, and the 2009 * minimal set we @need. Also include the larger offset we are writing 2010 * to (when applicable), and check against max_size here as well. 2011 * Note that caller is responsible for ensuring max_size increases are 2012 * requested from the MDS. 2013 */ 2014 static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want, 2015 int *got, loff_t endoff, int *check_max, int *err) 2016 { 2017 struct inode *inode = &ci->vfs_inode; 2018 int ret = 0; 2019 int have, implemented; 2020 int file_wanted; 2021 2022 dout("get_cap_refs %p need %s want %s\n", inode, 2023 ceph_cap_string(need), ceph_cap_string(want)); 2024 spin_lock(&ci->i_ceph_lock); 2025 2026 /* make sure file is actually open */ 2027 file_wanted = __ceph_caps_file_wanted(ci); 2028 if ((file_wanted & need) == 0) { 2029 dout("try_get_cap_refs need %s file_wanted %s, EBADF\n", 2030 ceph_cap_string(need), ceph_cap_string(file_wanted)); 2031 *err = -EBADF; 2032 ret = 1; 2033 goto out; 2034 } 2035 2036 if (need & CEPH_CAP_FILE_WR) { 2037 if (endoff >= 0 && endoff > (loff_t)ci->i_max_size) { 2038 dout("get_cap_refs %p endoff %llu > maxsize %llu\n", 2039 inode, endoff, ci->i_max_size); 2040 if (endoff > ci->i_wanted_max_size) { 2041 *check_max = 1; 2042 ret = 1; 2043 } 2044 goto out; 2045 } 2046 /* 2047 * If a sync write is in progress, we must wait, so that we 2048 * can get a final snapshot value for size+mtime. 2049 */ 2050 if (__ceph_have_pending_cap_snap(ci)) { 2051 dout("get_cap_refs %p cap_snap_pending\n", inode); 2052 goto out; 2053 } 2054 } 2055 have = __ceph_caps_issued(ci, &implemented); 2056 2057 /* 2058 * disallow writes while a truncate is pending 2059 */ 2060 if (ci->i_truncate_pending) 2061 have &= ~CEPH_CAP_FILE_WR; 2062 2063 if ((have & need) == need) { 2064 /* 2065 * Look at (implemented & ~have & not) so that we keep waiting 2066 * on transition from wanted -> needed caps. This is needed 2067 * for WRBUFFER|WR -> WR to avoid a new WR sync write from 2068 * going before a prior buffered writeback happens. 2069 */ 2070 int not = want & ~(have & need); 2071 int revoking = implemented & ~have; 2072 dout("get_cap_refs %p have %s but not %s (revoking %s)\n", 2073 inode, ceph_cap_string(have), ceph_cap_string(not), 2074 ceph_cap_string(revoking)); 2075 if ((revoking & not) == 0) { 2076 *got = need | (have & want); 2077 __take_cap_refs(ci, *got); 2078 ret = 1; 2079 } 2080 } else { 2081 dout("get_cap_refs %p have %s needed %s\n", inode, 2082 ceph_cap_string(have), ceph_cap_string(need)); 2083 } 2084 out: 2085 spin_unlock(&ci->i_ceph_lock); 2086 dout("get_cap_refs %p ret %d got %s\n", inode, 2087 ret, ceph_cap_string(*got)); 2088 return ret; 2089 } 2090 2091 /* 2092 * Check the offset we are writing up to against our current 2093 * max_size. If necessary, tell the MDS we want to write to 2094 * a larger offset. 2095 */ 2096 static void check_max_size(struct inode *inode, loff_t endoff) 2097 { 2098 struct ceph_inode_info *ci = ceph_inode(inode); 2099 int check = 0; 2100 2101 /* do we need to explicitly request a larger max_size? */ 2102 spin_lock(&ci->i_ceph_lock); 2103 if ((endoff >= ci->i_max_size || 2104 endoff > (inode->i_size << 1)) && 2105 endoff > ci->i_wanted_max_size) { 2106 dout("write %p at large endoff %llu, req max_size\n", 2107 inode, endoff); 2108 ci->i_wanted_max_size = endoff; 2109 check = 1; 2110 } 2111 spin_unlock(&ci->i_ceph_lock); 2112 if (check) 2113 ceph_check_caps(ci, CHECK_CAPS_AUTHONLY, NULL); 2114 } 2115 2116 /* 2117 * Wait for caps, and take cap references. If we can't get a WR cap 2118 * due to a small max_size, make sure we check_max_size (and possibly 2119 * ask the mds) so we don't get hung up indefinitely. 2120 */ 2121 int ceph_get_caps(struct ceph_inode_info *ci, int need, int want, int *got, 2122 loff_t endoff) 2123 { 2124 int check_max, ret, err; 2125 2126 retry: 2127 if (endoff > 0) 2128 check_max_size(&ci->vfs_inode, endoff); 2129 check_max = 0; 2130 err = 0; 2131 ret = wait_event_interruptible(ci->i_cap_wq, 2132 try_get_cap_refs(ci, need, want, 2133 got, endoff, 2134 &check_max, &err)); 2135 if (err) 2136 ret = err; 2137 if (check_max) 2138 goto retry; 2139 return ret; 2140 } 2141 2142 /* 2143 * Take cap refs. Caller must already know we hold at least one ref 2144 * on the caps in question or we don't know this is safe. 2145 */ 2146 void ceph_get_cap_refs(struct ceph_inode_info *ci, int caps) 2147 { 2148 spin_lock(&ci->i_ceph_lock); 2149 __take_cap_refs(ci, caps); 2150 spin_unlock(&ci->i_ceph_lock); 2151 } 2152 2153 /* 2154 * Release cap refs. 2155 * 2156 * If we released the last ref on any given cap, call ceph_check_caps 2157 * to release (or schedule a release). 2158 * 2159 * If we are releasing a WR cap (from a sync write), finalize any affected 2160 * cap_snap, and wake up any waiters. 2161 */ 2162 void ceph_put_cap_refs(struct ceph_inode_info *ci, int had) 2163 { 2164 struct inode *inode = &ci->vfs_inode; 2165 int last = 0, put = 0, flushsnaps = 0, wake = 0; 2166 struct ceph_cap_snap *capsnap; 2167 2168 spin_lock(&ci->i_ceph_lock); 2169 if (had & CEPH_CAP_PIN) 2170 --ci->i_pin_ref; 2171 if (had & CEPH_CAP_FILE_RD) 2172 if (--ci->i_rd_ref == 0) 2173 last++; 2174 if (had & CEPH_CAP_FILE_CACHE) 2175 if (--ci->i_rdcache_ref == 0) 2176 last++; 2177 if (had & CEPH_CAP_FILE_BUFFER) { 2178 if (--ci->i_wb_ref == 0) { 2179 last++; 2180 put++; 2181 } 2182 dout("put_cap_refs %p wb %d -> %d (?)\n", 2183 inode, ci->i_wb_ref+1, ci->i_wb_ref); 2184 } 2185 if (had & CEPH_CAP_FILE_WR) 2186 if (--ci->i_wr_ref == 0) { 2187 last++; 2188 if (!list_empty(&ci->i_cap_snaps)) { 2189 capsnap = list_first_entry(&ci->i_cap_snaps, 2190 struct ceph_cap_snap, 2191 ci_item); 2192 if (capsnap->writing) { 2193 capsnap->writing = 0; 2194 flushsnaps = 2195 __ceph_finish_cap_snap(ci, 2196 capsnap); 2197 wake = 1; 2198 } 2199 } 2200 } 2201 spin_unlock(&ci->i_ceph_lock); 2202 2203 dout("put_cap_refs %p had %s%s%s\n", inode, ceph_cap_string(had), 2204 last ? " last" : "", put ? " put" : ""); 2205 2206 if (last && !flushsnaps) 2207 ceph_check_caps(ci, 0, NULL); 2208 else if (flushsnaps) 2209 ceph_flush_snaps(ci); 2210 if (wake) 2211 wake_up_all(&ci->i_cap_wq); 2212 if (put) 2213 iput(inode); 2214 } 2215 2216 /* 2217 * Release @nr WRBUFFER refs on dirty pages for the given @snapc snap 2218 * context. Adjust per-snap dirty page accounting as appropriate. 2219 * Once all dirty data for a cap_snap is flushed, flush snapped file 2220 * metadata back to the MDS. If we dropped the last ref, call 2221 * ceph_check_caps. 2222 */ 2223 void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr, 2224 struct ceph_snap_context *snapc) 2225 { 2226 struct inode *inode = &ci->vfs_inode; 2227 int last = 0; 2228 int complete_capsnap = 0; 2229 int drop_capsnap = 0; 2230 int found = 0; 2231 struct ceph_cap_snap *capsnap = NULL; 2232 2233 spin_lock(&ci->i_ceph_lock); 2234 ci->i_wrbuffer_ref -= nr; 2235 last = !ci->i_wrbuffer_ref; 2236 2237 if (ci->i_head_snapc == snapc) { 2238 ci->i_wrbuffer_ref_head -= nr; 2239 if (ci->i_wrbuffer_ref_head == 0 && 2240 ci->i_dirty_caps == 0 && ci->i_flushing_caps == 0) { 2241 BUG_ON(!ci->i_head_snapc); 2242 ceph_put_snap_context(ci->i_head_snapc); 2243 ci->i_head_snapc = NULL; 2244 } 2245 dout("put_wrbuffer_cap_refs on %p head %d/%d -> %d/%d %s\n", 2246 inode, 2247 ci->i_wrbuffer_ref+nr, ci->i_wrbuffer_ref_head+nr, 2248 ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head, 2249 last ? " LAST" : ""); 2250 } else { 2251 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) { 2252 if (capsnap->context == snapc) { 2253 found = 1; 2254 break; 2255 } 2256 } 2257 BUG_ON(!found); 2258 capsnap->dirty_pages -= nr; 2259 if (capsnap->dirty_pages == 0) { 2260 complete_capsnap = 1; 2261 if (capsnap->dirty == 0) 2262 /* cap writeback completed before we created 2263 * the cap_snap; no FLUSHSNAP is needed */ 2264 drop_capsnap = 1; 2265 } 2266 dout("put_wrbuffer_cap_refs on %p cap_snap %p " 2267 " snap %lld %d/%d -> %d/%d %s%s%s\n", 2268 inode, capsnap, capsnap->context->seq, 2269 ci->i_wrbuffer_ref+nr, capsnap->dirty_pages + nr, 2270 ci->i_wrbuffer_ref, capsnap->dirty_pages, 2271 last ? " (wrbuffer last)" : "", 2272 complete_capsnap ? " (complete capsnap)" : "", 2273 drop_capsnap ? " (drop capsnap)" : ""); 2274 if (drop_capsnap) { 2275 ceph_put_snap_context(capsnap->context); 2276 list_del(&capsnap->ci_item); 2277 list_del(&capsnap->flushing_item); 2278 ceph_put_cap_snap(capsnap); 2279 } 2280 } 2281 2282 spin_unlock(&ci->i_ceph_lock); 2283 2284 if (last) { 2285 ceph_check_caps(ci, CHECK_CAPS_AUTHONLY, NULL); 2286 iput(inode); 2287 } else if (complete_capsnap) { 2288 ceph_flush_snaps(ci); 2289 wake_up_all(&ci->i_cap_wq); 2290 } 2291 if (drop_capsnap) 2292 iput(inode); 2293 } 2294 2295 /* 2296 * Handle a cap GRANT message from the MDS. (Note that a GRANT may 2297 * actually be a revocation if it specifies a smaller cap set.) 2298 * 2299 * caller holds s_mutex and i_ceph_lock, we drop both. 2300 * 2301 * return value: 2302 * 0 - ok 2303 * 1 - check_caps on auth cap only (writeback) 2304 * 2 - check_caps (ack revoke) 2305 */ 2306 static void handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant, 2307 struct ceph_mds_session *session, 2308 struct ceph_cap *cap, 2309 struct ceph_buffer *xattr_buf) 2310 __releases(ci->i_ceph_lock) 2311 { 2312 struct ceph_inode_info *ci = ceph_inode(inode); 2313 int mds = session->s_mds; 2314 int seq = le32_to_cpu(grant->seq); 2315 int newcaps = le32_to_cpu(grant->caps); 2316 int issued, implemented, used, wanted, dirty; 2317 u64 size = le64_to_cpu(grant->size); 2318 u64 max_size = le64_to_cpu(grant->max_size); 2319 struct timespec mtime, atime, ctime; 2320 int check_caps = 0; 2321 int wake = 0; 2322 int writeback = 0; 2323 int revoked_rdcache = 0; 2324 int queue_invalidate = 0; 2325 2326 dout("handle_cap_grant inode %p cap %p mds%d seq %d %s\n", 2327 inode, cap, mds, seq, ceph_cap_string(newcaps)); 2328 dout(" size %llu max_size %llu, i_size %llu\n", size, max_size, 2329 inode->i_size); 2330 2331 /* 2332 * If CACHE is being revoked, and we have no dirty buffers, 2333 * try to invalidate (once). (If there are dirty buffers, we 2334 * will invalidate _after_ writeback.) 2335 */ 2336 if (((cap->issued & ~newcaps) & CEPH_CAP_FILE_CACHE) && 2337 (newcaps & CEPH_CAP_FILE_LAZYIO) == 0 && 2338 !ci->i_wrbuffer_ref) { 2339 if (try_nonblocking_invalidate(inode) == 0) { 2340 revoked_rdcache = 1; 2341 } else { 2342 /* there were locked pages.. invalidate later 2343 in a separate thread. */ 2344 if (ci->i_rdcache_revoking != ci->i_rdcache_gen) { 2345 queue_invalidate = 1; 2346 ci->i_rdcache_revoking = ci->i_rdcache_gen; 2347 } 2348 } 2349 } 2350 2351 /* side effects now are allowed */ 2352 2353 issued = __ceph_caps_issued(ci, &implemented); 2354 issued |= implemented | __ceph_caps_dirty(ci); 2355 2356 cap->cap_gen = session->s_cap_gen; 2357 2358 __check_cap_issue(ci, cap, newcaps); 2359 2360 if ((issued & CEPH_CAP_AUTH_EXCL) == 0) { 2361 inode->i_mode = le32_to_cpu(grant->mode); 2362 inode->i_uid = le32_to_cpu(grant->uid); 2363 inode->i_gid = le32_to_cpu(grant->gid); 2364 dout("%p mode 0%o uid.gid %d.%d\n", inode, inode->i_mode, 2365 inode->i_uid, inode->i_gid); 2366 } 2367 2368 if ((issued & CEPH_CAP_LINK_EXCL) == 0) 2369 set_nlink(inode, le32_to_cpu(grant->nlink)); 2370 2371 if ((issued & CEPH_CAP_XATTR_EXCL) == 0 && grant->xattr_len) { 2372 int len = le32_to_cpu(grant->xattr_len); 2373 u64 version = le64_to_cpu(grant->xattr_version); 2374 2375 if (version > ci->i_xattrs.version) { 2376 dout(" got new xattrs v%llu on %p len %d\n", 2377 version, inode, len); 2378 if (ci->i_xattrs.blob) 2379 ceph_buffer_put(ci->i_xattrs.blob); 2380 ci->i_xattrs.blob = ceph_buffer_get(xattr_buf); 2381 ci->i_xattrs.version = version; 2382 } 2383 } 2384 2385 /* size/ctime/mtime/atime? */ 2386 ceph_fill_file_size(inode, issued, 2387 le32_to_cpu(grant->truncate_seq), 2388 le64_to_cpu(grant->truncate_size), size); 2389 ceph_decode_timespec(&mtime, &grant->mtime); 2390 ceph_decode_timespec(&atime, &grant->atime); 2391 ceph_decode_timespec(&ctime, &grant->ctime); 2392 ceph_fill_file_time(inode, issued, 2393 le32_to_cpu(grant->time_warp_seq), &ctime, &mtime, 2394 &atime); 2395 2396 /* max size increase? */ 2397 if (ci->i_auth_cap == cap && max_size != ci->i_max_size) { 2398 dout("max_size %lld -> %llu\n", ci->i_max_size, max_size); 2399 ci->i_max_size = max_size; 2400 if (max_size >= ci->i_wanted_max_size) { 2401 ci->i_wanted_max_size = 0; /* reset */ 2402 ci->i_requested_max_size = 0; 2403 } 2404 wake = 1; 2405 } 2406 2407 /* check cap bits */ 2408 wanted = __ceph_caps_wanted(ci); 2409 used = __ceph_caps_used(ci); 2410 dirty = __ceph_caps_dirty(ci); 2411 dout(" my wanted = %s, used = %s, dirty %s\n", 2412 ceph_cap_string(wanted), 2413 ceph_cap_string(used), 2414 ceph_cap_string(dirty)); 2415 if (wanted != le32_to_cpu(grant->wanted)) { 2416 dout("mds wanted %s -> %s\n", 2417 ceph_cap_string(le32_to_cpu(grant->wanted)), 2418 ceph_cap_string(wanted)); 2419 grant->wanted = cpu_to_le32(wanted); 2420 } 2421 2422 cap->seq = seq; 2423 2424 /* file layout may have changed */ 2425 ci->i_layout = grant->layout; 2426 2427 /* revocation, grant, or no-op? */ 2428 if (cap->issued & ~newcaps) { 2429 int revoking = cap->issued & ~newcaps; 2430 2431 dout("revocation: %s -> %s (revoking %s)\n", 2432 ceph_cap_string(cap->issued), 2433 ceph_cap_string(newcaps), 2434 ceph_cap_string(revoking)); 2435 if (revoking & used & CEPH_CAP_FILE_BUFFER) 2436 writeback = 1; /* initiate writeback; will delay ack */ 2437 else if (revoking == CEPH_CAP_FILE_CACHE && 2438 (newcaps & CEPH_CAP_FILE_LAZYIO) == 0 && 2439 queue_invalidate) 2440 ; /* do nothing yet, invalidation will be queued */ 2441 else if (cap == ci->i_auth_cap) 2442 check_caps = 1; /* check auth cap only */ 2443 else 2444 check_caps = 2; /* check all caps */ 2445 cap->issued = newcaps; 2446 cap->implemented |= newcaps; 2447 } else if (cap->issued == newcaps) { 2448 dout("caps unchanged: %s -> %s\n", 2449 ceph_cap_string(cap->issued), ceph_cap_string(newcaps)); 2450 } else { 2451 dout("grant: %s -> %s\n", ceph_cap_string(cap->issued), 2452 ceph_cap_string(newcaps)); 2453 cap->issued = newcaps; 2454 cap->implemented |= newcaps; /* add bits only, to 2455 * avoid stepping on a 2456 * pending revocation */ 2457 wake = 1; 2458 } 2459 BUG_ON(cap->issued & ~cap->implemented); 2460 2461 spin_unlock(&ci->i_ceph_lock); 2462 if (writeback) 2463 /* 2464 * queue inode for writeback: we can't actually call 2465 * filemap_write_and_wait, etc. from message handler 2466 * context. 2467 */ 2468 ceph_queue_writeback(inode); 2469 if (queue_invalidate) 2470 ceph_queue_invalidate(inode); 2471 if (wake) 2472 wake_up_all(&ci->i_cap_wq); 2473 2474 if (check_caps == 1) 2475 ceph_check_caps(ci, CHECK_CAPS_NODELAY|CHECK_CAPS_AUTHONLY, 2476 session); 2477 else if (check_caps == 2) 2478 ceph_check_caps(ci, CHECK_CAPS_NODELAY, session); 2479 else 2480 mutex_unlock(&session->s_mutex); 2481 } 2482 2483 /* 2484 * Handle FLUSH_ACK from MDS, indicating that metadata we sent to the 2485 * MDS has been safely committed. 2486 */ 2487 static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid, 2488 struct ceph_mds_caps *m, 2489 struct ceph_mds_session *session, 2490 struct ceph_cap *cap) 2491 __releases(ci->i_ceph_lock) 2492 { 2493 struct ceph_inode_info *ci = ceph_inode(inode); 2494 struct ceph_mds_client *mdsc = ceph_sb_to_client(inode->i_sb)->mdsc; 2495 unsigned seq = le32_to_cpu(m->seq); 2496 int dirty = le32_to_cpu(m->dirty); 2497 int cleaned = 0; 2498 int drop = 0; 2499 int i; 2500 2501 for (i = 0; i < CEPH_CAP_BITS; i++) 2502 if ((dirty & (1 << i)) && 2503 flush_tid == ci->i_cap_flush_tid[i]) 2504 cleaned |= 1 << i; 2505 2506 dout("handle_cap_flush_ack inode %p mds%d seq %d on %s cleaned %s," 2507 " flushing %s -> %s\n", 2508 inode, session->s_mds, seq, ceph_cap_string(dirty), 2509 ceph_cap_string(cleaned), ceph_cap_string(ci->i_flushing_caps), 2510 ceph_cap_string(ci->i_flushing_caps & ~cleaned)); 2511 2512 if (ci->i_flushing_caps == (ci->i_flushing_caps & ~cleaned)) 2513 goto out; 2514 2515 ci->i_flushing_caps &= ~cleaned; 2516 2517 spin_lock(&mdsc->cap_dirty_lock); 2518 if (ci->i_flushing_caps == 0) { 2519 list_del_init(&ci->i_flushing_item); 2520 if (!list_empty(&session->s_cap_flushing)) 2521 dout(" mds%d still flushing cap on %p\n", 2522 session->s_mds, 2523 &list_entry(session->s_cap_flushing.next, 2524 struct ceph_inode_info, 2525 i_flushing_item)->vfs_inode); 2526 mdsc->num_cap_flushing--; 2527 wake_up_all(&mdsc->cap_flushing_wq); 2528 dout(" inode %p now !flushing\n", inode); 2529 2530 if (ci->i_dirty_caps == 0) { 2531 dout(" inode %p now clean\n", inode); 2532 BUG_ON(!list_empty(&ci->i_dirty_item)); 2533 drop = 1; 2534 if (ci->i_wrbuffer_ref_head == 0) { 2535 BUG_ON(!ci->i_head_snapc); 2536 ceph_put_snap_context(ci->i_head_snapc); 2537 ci->i_head_snapc = NULL; 2538 } 2539 } else { 2540 BUG_ON(list_empty(&ci->i_dirty_item)); 2541 } 2542 } 2543 spin_unlock(&mdsc->cap_dirty_lock); 2544 wake_up_all(&ci->i_cap_wq); 2545 2546 out: 2547 spin_unlock(&ci->i_ceph_lock); 2548 if (drop) 2549 iput(inode); 2550 } 2551 2552 /* 2553 * Handle FLUSHSNAP_ACK. MDS has flushed snap data to disk and we can 2554 * throw away our cap_snap. 2555 * 2556 * Caller hold s_mutex. 2557 */ 2558 static void handle_cap_flushsnap_ack(struct inode *inode, u64 flush_tid, 2559 struct ceph_mds_caps *m, 2560 struct ceph_mds_session *session) 2561 { 2562 struct ceph_inode_info *ci = ceph_inode(inode); 2563 u64 follows = le64_to_cpu(m->snap_follows); 2564 struct ceph_cap_snap *capsnap; 2565 int drop = 0; 2566 2567 dout("handle_cap_flushsnap_ack inode %p ci %p mds%d follows %lld\n", 2568 inode, ci, session->s_mds, follows); 2569 2570 spin_lock(&ci->i_ceph_lock); 2571 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) { 2572 if (capsnap->follows == follows) { 2573 if (capsnap->flush_tid != flush_tid) { 2574 dout(" cap_snap %p follows %lld tid %lld !=" 2575 " %lld\n", capsnap, follows, 2576 flush_tid, capsnap->flush_tid); 2577 break; 2578 } 2579 WARN_ON(capsnap->dirty_pages || capsnap->writing); 2580 dout(" removing %p cap_snap %p follows %lld\n", 2581 inode, capsnap, follows); 2582 ceph_put_snap_context(capsnap->context); 2583 list_del(&capsnap->ci_item); 2584 list_del(&capsnap->flushing_item); 2585 ceph_put_cap_snap(capsnap); 2586 drop = 1; 2587 break; 2588 } else { 2589 dout(" skipping cap_snap %p follows %lld\n", 2590 capsnap, capsnap->follows); 2591 } 2592 } 2593 spin_unlock(&ci->i_ceph_lock); 2594 if (drop) 2595 iput(inode); 2596 } 2597 2598 /* 2599 * Handle TRUNC from MDS, indicating file truncation. 2600 * 2601 * caller hold s_mutex. 2602 */ 2603 static void handle_cap_trunc(struct inode *inode, 2604 struct ceph_mds_caps *trunc, 2605 struct ceph_mds_session *session) 2606 __releases(ci->i_ceph_lock) 2607 { 2608 struct ceph_inode_info *ci = ceph_inode(inode); 2609 int mds = session->s_mds; 2610 int seq = le32_to_cpu(trunc->seq); 2611 u32 truncate_seq = le32_to_cpu(trunc->truncate_seq); 2612 u64 truncate_size = le64_to_cpu(trunc->truncate_size); 2613 u64 size = le64_to_cpu(trunc->size); 2614 int implemented = 0; 2615 int dirty = __ceph_caps_dirty(ci); 2616 int issued = __ceph_caps_issued(ceph_inode(inode), &implemented); 2617 int queue_trunc = 0; 2618 2619 issued |= implemented | dirty; 2620 2621 dout("handle_cap_trunc inode %p mds%d seq %d to %lld seq %d\n", 2622 inode, mds, seq, truncate_size, truncate_seq); 2623 queue_trunc = ceph_fill_file_size(inode, issued, 2624 truncate_seq, truncate_size, size); 2625 spin_unlock(&ci->i_ceph_lock); 2626 2627 if (queue_trunc) 2628 ceph_queue_vmtruncate(inode); 2629 } 2630 2631 /* 2632 * Handle EXPORT from MDS. Cap is being migrated _from_ this mds to a 2633 * different one. If we are the most recent migration we've seen (as 2634 * indicated by mseq), make note of the migrating cap bits for the 2635 * duration (until we see the corresponding IMPORT). 2636 * 2637 * caller holds s_mutex 2638 */ 2639 static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex, 2640 struct ceph_mds_session *session, 2641 int *open_target_sessions) 2642 { 2643 struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc; 2644 struct ceph_inode_info *ci = ceph_inode(inode); 2645 int mds = session->s_mds; 2646 unsigned mseq = le32_to_cpu(ex->migrate_seq); 2647 struct ceph_cap *cap = NULL, *t; 2648 struct rb_node *p; 2649 int remember = 1; 2650 2651 dout("handle_cap_export inode %p ci %p mds%d mseq %d\n", 2652 inode, ci, mds, mseq); 2653 2654 spin_lock(&ci->i_ceph_lock); 2655 2656 /* make sure we haven't seen a higher mseq */ 2657 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) { 2658 t = rb_entry(p, struct ceph_cap, ci_node); 2659 if (ceph_seq_cmp(t->mseq, mseq) > 0) { 2660 dout(" higher mseq on cap from mds%d\n", 2661 t->session->s_mds); 2662 remember = 0; 2663 } 2664 if (t->session->s_mds == mds) 2665 cap = t; 2666 } 2667 2668 if (cap) { 2669 if (remember) { 2670 /* make note */ 2671 ci->i_cap_exporting_mds = mds; 2672 ci->i_cap_exporting_mseq = mseq; 2673 ci->i_cap_exporting_issued = cap->issued; 2674 2675 /* 2676 * make sure we have open sessions with all possible 2677 * export targets, so that we get the matching IMPORT 2678 */ 2679 *open_target_sessions = 1; 2680 2681 /* 2682 * we can't flush dirty caps that we've seen the 2683 * EXPORT but no IMPORT for 2684 */ 2685 spin_lock(&mdsc->cap_dirty_lock); 2686 if (!list_empty(&ci->i_dirty_item)) { 2687 dout(" moving %p to cap_dirty_migrating\n", 2688 inode); 2689 list_move(&ci->i_dirty_item, 2690 &mdsc->cap_dirty_migrating); 2691 } 2692 spin_unlock(&mdsc->cap_dirty_lock); 2693 } 2694 __ceph_remove_cap(cap); 2695 } 2696 /* else, we already released it */ 2697 2698 spin_unlock(&ci->i_ceph_lock); 2699 } 2700 2701 /* 2702 * Handle cap IMPORT. If there are temp bits from an older EXPORT, 2703 * clean them up. 2704 * 2705 * caller holds s_mutex. 2706 */ 2707 static void handle_cap_import(struct ceph_mds_client *mdsc, 2708 struct inode *inode, struct ceph_mds_caps *im, 2709 struct ceph_mds_session *session, 2710 void *snaptrace, int snaptrace_len) 2711 { 2712 struct ceph_inode_info *ci = ceph_inode(inode); 2713 int mds = session->s_mds; 2714 unsigned issued = le32_to_cpu(im->caps); 2715 unsigned wanted = le32_to_cpu(im->wanted); 2716 unsigned seq = le32_to_cpu(im->seq); 2717 unsigned mseq = le32_to_cpu(im->migrate_seq); 2718 u64 realmino = le64_to_cpu(im->realm); 2719 u64 cap_id = le64_to_cpu(im->cap_id); 2720 2721 if (ci->i_cap_exporting_mds >= 0 && 2722 ceph_seq_cmp(ci->i_cap_exporting_mseq, mseq) < 0) { 2723 dout("handle_cap_import inode %p ci %p mds%d mseq %d" 2724 " - cleared exporting from mds%d\n", 2725 inode, ci, mds, mseq, 2726 ci->i_cap_exporting_mds); 2727 ci->i_cap_exporting_issued = 0; 2728 ci->i_cap_exporting_mseq = 0; 2729 ci->i_cap_exporting_mds = -1; 2730 2731 spin_lock(&mdsc->cap_dirty_lock); 2732 if (!list_empty(&ci->i_dirty_item)) { 2733 dout(" moving %p back to cap_dirty\n", inode); 2734 list_move(&ci->i_dirty_item, &mdsc->cap_dirty); 2735 } 2736 spin_unlock(&mdsc->cap_dirty_lock); 2737 } else { 2738 dout("handle_cap_import inode %p ci %p mds%d mseq %d\n", 2739 inode, ci, mds, mseq); 2740 } 2741 2742 down_write(&mdsc->snap_rwsem); 2743 ceph_update_snap_trace(mdsc, snaptrace, snaptrace+snaptrace_len, 2744 false); 2745 downgrade_write(&mdsc->snap_rwsem); 2746 ceph_add_cap(inode, session, cap_id, -1, 2747 issued, wanted, seq, mseq, realmino, CEPH_CAP_FLAG_AUTH, 2748 NULL /* no caps context */); 2749 kick_flushing_inode_caps(mdsc, session, inode); 2750 up_read(&mdsc->snap_rwsem); 2751 2752 /* make sure we re-request max_size, if necessary */ 2753 spin_lock(&ci->i_ceph_lock); 2754 ci->i_wanted_max_size = 0; /* reset */ 2755 ci->i_requested_max_size = 0; 2756 spin_unlock(&ci->i_ceph_lock); 2757 } 2758 2759 /* 2760 * Handle a caps message from the MDS. 2761 * 2762 * Identify the appropriate session, inode, and call the right handler 2763 * based on the cap op. 2764 */ 2765 void ceph_handle_caps(struct ceph_mds_session *session, 2766 struct ceph_msg *msg) 2767 { 2768 struct ceph_mds_client *mdsc = session->s_mdsc; 2769 struct super_block *sb = mdsc->fsc->sb; 2770 struct inode *inode; 2771 struct ceph_inode_info *ci; 2772 struct ceph_cap *cap; 2773 struct ceph_mds_caps *h; 2774 int mds = session->s_mds; 2775 int op; 2776 u32 seq, mseq; 2777 struct ceph_vino vino; 2778 u64 cap_id; 2779 u64 size, max_size; 2780 u64 tid; 2781 void *snaptrace; 2782 size_t snaptrace_len; 2783 void *flock; 2784 u32 flock_len; 2785 int open_target_sessions = 0; 2786 2787 dout("handle_caps from mds%d\n", mds); 2788 2789 /* decode */ 2790 tid = le64_to_cpu(msg->hdr.tid); 2791 if (msg->front.iov_len < sizeof(*h)) 2792 goto bad; 2793 h = msg->front.iov_base; 2794 op = le32_to_cpu(h->op); 2795 vino.ino = le64_to_cpu(h->ino); 2796 vino.snap = CEPH_NOSNAP; 2797 cap_id = le64_to_cpu(h->cap_id); 2798 seq = le32_to_cpu(h->seq); 2799 mseq = le32_to_cpu(h->migrate_seq); 2800 size = le64_to_cpu(h->size); 2801 max_size = le64_to_cpu(h->max_size); 2802 2803 snaptrace = h + 1; 2804 snaptrace_len = le32_to_cpu(h->snap_trace_len); 2805 2806 if (le16_to_cpu(msg->hdr.version) >= 2) { 2807 void *p, *end; 2808 2809 p = snaptrace + snaptrace_len; 2810 end = msg->front.iov_base + msg->front.iov_len; 2811 ceph_decode_32_safe(&p, end, flock_len, bad); 2812 flock = p; 2813 } else { 2814 flock = NULL; 2815 flock_len = 0; 2816 } 2817 2818 mutex_lock(&session->s_mutex); 2819 session->s_seq++; 2820 dout(" mds%d seq %lld cap seq %u\n", session->s_mds, session->s_seq, 2821 (unsigned)seq); 2822 2823 /* lookup ino */ 2824 inode = ceph_find_inode(sb, vino); 2825 ci = ceph_inode(inode); 2826 dout(" op %s ino %llx.%llx inode %p\n", ceph_cap_op_name(op), vino.ino, 2827 vino.snap, inode); 2828 if (!inode) { 2829 dout(" i don't have ino %llx\n", vino.ino); 2830 2831 if (op == CEPH_CAP_OP_IMPORT) 2832 __queue_cap_release(session, vino.ino, cap_id, 2833 mseq, seq); 2834 goto flush_cap_releases; 2835 } 2836 2837 /* these will work even if we don't have a cap yet */ 2838 switch (op) { 2839 case CEPH_CAP_OP_FLUSHSNAP_ACK: 2840 handle_cap_flushsnap_ack(inode, tid, h, session); 2841 goto done; 2842 2843 case CEPH_CAP_OP_EXPORT: 2844 handle_cap_export(inode, h, session, &open_target_sessions); 2845 goto done; 2846 2847 case CEPH_CAP_OP_IMPORT: 2848 handle_cap_import(mdsc, inode, h, session, 2849 snaptrace, snaptrace_len); 2850 } 2851 2852 /* the rest require a cap */ 2853 spin_lock(&ci->i_ceph_lock); 2854 cap = __get_cap_for_mds(ceph_inode(inode), mds); 2855 if (!cap) { 2856 dout(" no cap on %p ino %llx.%llx from mds%d\n", 2857 inode, ceph_ino(inode), ceph_snap(inode), mds); 2858 spin_unlock(&ci->i_ceph_lock); 2859 goto flush_cap_releases; 2860 } 2861 2862 /* note that each of these drops i_ceph_lock for us */ 2863 switch (op) { 2864 case CEPH_CAP_OP_REVOKE: 2865 case CEPH_CAP_OP_GRANT: 2866 case CEPH_CAP_OP_IMPORT: 2867 handle_cap_grant(inode, h, session, cap, msg->middle); 2868 goto done_unlocked; 2869 2870 case CEPH_CAP_OP_FLUSH_ACK: 2871 handle_cap_flush_ack(inode, tid, h, session, cap); 2872 break; 2873 2874 case CEPH_CAP_OP_TRUNC: 2875 handle_cap_trunc(inode, h, session); 2876 break; 2877 2878 default: 2879 spin_unlock(&ci->i_ceph_lock); 2880 pr_err("ceph_handle_caps: unknown cap op %d %s\n", op, 2881 ceph_cap_op_name(op)); 2882 } 2883 2884 goto done; 2885 2886 flush_cap_releases: 2887 /* 2888 * send any full release message to try to move things 2889 * along for the mds (who clearly thinks we still have this 2890 * cap). 2891 */ 2892 ceph_add_cap_releases(mdsc, session); 2893 ceph_send_cap_releases(mdsc, session); 2894 2895 done: 2896 mutex_unlock(&session->s_mutex); 2897 done_unlocked: 2898 if (inode) 2899 iput(inode); 2900 if (open_target_sessions) 2901 ceph_mdsc_open_export_target_sessions(mdsc, session); 2902 return; 2903 2904 bad: 2905 pr_err("ceph_handle_caps: corrupt message\n"); 2906 ceph_msg_dump(msg); 2907 return; 2908 } 2909 2910 /* 2911 * Delayed work handler to process end of delayed cap release LRU list. 2912 */ 2913 void ceph_check_delayed_caps(struct ceph_mds_client *mdsc) 2914 { 2915 struct ceph_inode_info *ci; 2916 int flags = CHECK_CAPS_NODELAY; 2917 2918 dout("check_delayed_caps\n"); 2919 while (1) { 2920 spin_lock(&mdsc->cap_delay_lock); 2921 if (list_empty(&mdsc->cap_delay_list)) 2922 break; 2923 ci = list_first_entry(&mdsc->cap_delay_list, 2924 struct ceph_inode_info, 2925 i_cap_delay_list); 2926 if ((ci->i_ceph_flags & CEPH_I_FLUSH) == 0 && 2927 time_before(jiffies, ci->i_hold_caps_max)) 2928 break; 2929 list_del_init(&ci->i_cap_delay_list); 2930 spin_unlock(&mdsc->cap_delay_lock); 2931 dout("check_delayed_caps on %p\n", &ci->vfs_inode); 2932 ceph_check_caps(ci, flags, NULL); 2933 } 2934 spin_unlock(&mdsc->cap_delay_lock); 2935 } 2936 2937 /* 2938 * Flush all dirty caps to the mds 2939 */ 2940 void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc) 2941 { 2942 struct ceph_inode_info *ci; 2943 struct inode *inode; 2944 2945 dout("flush_dirty_caps\n"); 2946 spin_lock(&mdsc->cap_dirty_lock); 2947 while (!list_empty(&mdsc->cap_dirty)) { 2948 ci = list_first_entry(&mdsc->cap_dirty, struct ceph_inode_info, 2949 i_dirty_item); 2950 inode = &ci->vfs_inode; 2951 ihold(inode); 2952 dout("flush_dirty_caps %p\n", inode); 2953 spin_unlock(&mdsc->cap_dirty_lock); 2954 ceph_check_caps(ci, CHECK_CAPS_NODELAY|CHECK_CAPS_FLUSH, NULL); 2955 iput(inode); 2956 spin_lock(&mdsc->cap_dirty_lock); 2957 } 2958 spin_unlock(&mdsc->cap_dirty_lock); 2959 dout("flush_dirty_caps done\n"); 2960 } 2961 2962 /* 2963 * Drop open file reference. If we were the last open file, 2964 * we may need to release capabilities to the MDS (or schedule 2965 * their delayed release). 2966 */ 2967 void ceph_put_fmode(struct ceph_inode_info *ci, int fmode) 2968 { 2969 struct inode *inode = &ci->vfs_inode; 2970 int last = 0; 2971 2972 spin_lock(&ci->i_ceph_lock); 2973 dout("put_fmode %p fmode %d %d -> %d\n", inode, fmode, 2974 ci->i_nr_by_mode[fmode], ci->i_nr_by_mode[fmode]-1); 2975 BUG_ON(ci->i_nr_by_mode[fmode] == 0); 2976 if (--ci->i_nr_by_mode[fmode] == 0) 2977 last++; 2978 spin_unlock(&ci->i_ceph_lock); 2979 2980 if (last && ci->i_vino.snap == CEPH_NOSNAP) 2981 ceph_check_caps(ci, 0, NULL); 2982 } 2983 2984 /* 2985 * Helpers for embedding cap and dentry lease releases into mds 2986 * requests. 2987 * 2988 * @force is used by dentry_release (below) to force inclusion of a 2989 * record for the directory inode, even when there aren't any caps to 2990 * drop. 2991 */ 2992 int ceph_encode_inode_release(void **p, struct inode *inode, 2993 int mds, int drop, int unless, int force) 2994 { 2995 struct ceph_inode_info *ci = ceph_inode(inode); 2996 struct ceph_cap *cap; 2997 struct ceph_mds_request_release *rel = *p; 2998 int used, dirty; 2999 int ret = 0; 3000 3001 spin_lock(&ci->i_ceph_lock); 3002 used = __ceph_caps_used(ci); 3003 dirty = __ceph_caps_dirty(ci); 3004 3005 dout("encode_inode_release %p mds%d used|dirty %s drop %s unless %s\n", 3006 inode, mds, ceph_cap_string(used|dirty), ceph_cap_string(drop), 3007 ceph_cap_string(unless)); 3008 3009 /* only drop unused, clean caps */ 3010 drop &= ~(used | dirty); 3011 3012 cap = __get_cap_for_mds(ci, mds); 3013 if (cap && __cap_is_valid(cap)) { 3014 if (force || 3015 ((cap->issued & drop) && 3016 (cap->issued & unless) == 0)) { 3017 if ((cap->issued & drop) && 3018 (cap->issued & unless) == 0) { 3019 dout("encode_inode_release %p cap %p %s -> " 3020 "%s\n", inode, cap, 3021 ceph_cap_string(cap->issued), 3022 ceph_cap_string(cap->issued & ~drop)); 3023 cap->issued &= ~drop; 3024 cap->implemented &= ~drop; 3025 if (ci->i_ceph_flags & CEPH_I_NODELAY) { 3026 int wanted = __ceph_caps_wanted(ci); 3027 dout(" wanted %s -> %s (act %s)\n", 3028 ceph_cap_string(cap->mds_wanted), 3029 ceph_cap_string(cap->mds_wanted & 3030 ~wanted), 3031 ceph_cap_string(wanted)); 3032 cap->mds_wanted &= wanted; 3033 } 3034 } else { 3035 dout("encode_inode_release %p cap %p %s" 3036 " (force)\n", inode, cap, 3037 ceph_cap_string(cap->issued)); 3038 } 3039 3040 rel->ino = cpu_to_le64(ceph_ino(inode)); 3041 rel->cap_id = cpu_to_le64(cap->cap_id); 3042 rel->seq = cpu_to_le32(cap->seq); 3043 rel->issue_seq = cpu_to_le32(cap->issue_seq), 3044 rel->mseq = cpu_to_le32(cap->mseq); 3045 rel->caps = cpu_to_le32(cap->issued); 3046 rel->wanted = cpu_to_le32(cap->mds_wanted); 3047 rel->dname_len = 0; 3048 rel->dname_seq = 0; 3049 *p += sizeof(*rel); 3050 ret = 1; 3051 } else { 3052 dout("encode_inode_release %p cap %p %s\n", 3053 inode, cap, ceph_cap_string(cap->issued)); 3054 } 3055 } 3056 spin_unlock(&ci->i_ceph_lock); 3057 return ret; 3058 } 3059 3060 int ceph_encode_dentry_release(void **p, struct dentry *dentry, 3061 int mds, int drop, int unless) 3062 { 3063 struct inode *dir = dentry->d_parent->d_inode; 3064 struct ceph_mds_request_release *rel = *p; 3065 struct ceph_dentry_info *di = ceph_dentry(dentry); 3066 int force = 0; 3067 int ret; 3068 3069 /* 3070 * force an record for the directory caps if we have a dentry lease. 3071 * this is racy (can't take i_ceph_lock and d_lock together), but it 3072 * doesn't have to be perfect; the mds will revoke anything we don't 3073 * release. 3074 */ 3075 spin_lock(&dentry->d_lock); 3076 if (di->lease_session && di->lease_session->s_mds == mds) 3077 force = 1; 3078 spin_unlock(&dentry->d_lock); 3079 3080 ret = ceph_encode_inode_release(p, dir, mds, drop, unless, force); 3081 3082 spin_lock(&dentry->d_lock); 3083 if (ret && di->lease_session && di->lease_session->s_mds == mds) { 3084 dout("encode_dentry_release %p mds%d seq %d\n", 3085 dentry, mds, (int)di->lease_seq); 3086 rel->dname_len = cpu_to_le32(dentry->d_name.len); 3087 memcpy(*p, dentry->d_name.name, dentry->d_name.len); 3088 *p += dentry->d_name.len; 3089 rel->dname_seq = cpu_to_le32(di->lease_seq); 3090 __ceph_mdsc_drop_dentry_lease(dentry); 3091 } 3092 spin_unlock(&dentry->d_lock); 3093 return ret; 3094 } 3095