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