ceph/src/mds/CDentry.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15
  16
  17 #include "CDentry.h"
  18 #include "CInode.h"
  19 #include "CDir.h"
  20
  21 #include "MDSRank.h"
  22 #include "MDCache.h"
  23 #include "Locker.h"
  24 #include "LogSegment.h"
  25
  26 #include "messages/MLock.h"
  27
  28 #define dout_context g_ceph_context
  29 #define dout_subsys ceph_subsys_mds
  30 #undef dout_prefix
  31 #define dout_prefix *_dout << "mds." << dir->cache->mds->get_nodeid() << ".cache.den(" << dir->dirfrag() << " " << name << ") "
  32
  33
  34 ostream& CDentry::print_db_line_prefix(ostream& out)
  35 {
  36   return out << ceph_clock_now() << " mds." << dir->cache->mds->get_nodeid() << ".cache.den(" << dir->ino() << " " << name << ") ";
  37 }
  38
  39 LockType CDentry::lock_type(CEPH_LOCK_DN);
  40 LockType CDentry::versionlock_type(CEPH_LOCK_DVERSION);
  41
  42
  43 // CDentry
  44
  45 ostream& operator<<(ostream& out, const CDentry& dn)
  46 {
  47   filepath path;
  48   dn.make_path(path);
  49
  50   out << "[dentry " << path;
  51
  52   if (true || dn.first != 0 || dn.last != CEPH_NOSNAP) {
  53     out << " [" << dn.first << ",";
  54     if (dn.last == CEPH_NOSNAP)
  55       out << "head";
  56     else
  57       out << dn.last;
  58     out << ']';
  59   }
  60
  61   if (dn.is_auth()) {
  62     out << " auth";
  63     if (dn.is_replicated())
  64       out << dn.get_replicas();
  65   } else {
  66     out << " rep@" << dn.authority();
  67     out << "." << dn.get_replica_nonce();
  68   }
  69
  70   if (dn.get_linkage()->is_null()) out << " NULL";
  71   if (dn.get_linkage()->is_remote()) {
  72     out << " REMOTE(";
  73     out << dn.get_linkage()->get_remote_d_type_string();
  74     out << ")";
  75   }
  76
  77   if (!dn.lock.is_sync_and_unlocked())
  78     out << " " << dn.lock;
  79   if (!dn.versionlock.is_sync_and_unlocked())
  80     out << " " << dn.versionlock;
  81
  82   if (dn.get_projected_version() != dn.get_version())
  83     out << " pv=" << dn.get_projected_version();
  84   out << " v=" << dn.get_version();
  85
  86   if (dn.is_auth_pinned())
  87     out << " ap=" << dn.get_num_auth_pins() << "+" << dn.get_num_nested_auth_pins();
  88
  89   out << " inode=" << dn.get_linkage()->get_inode();
  90
  91   if (dn.is_new()) out << " state=new";
  92
  93   if (dn.get_num_ref()) {
  94     out << " |";
  95     dn.print_pin_set(out);
  96   }
  97
  98   out << " " << &dn;
  99   out << "]";
 100   return out;
 101 }
 102
 103
 104 bool operator<(const CDentry& l, const CDentry& r)
 105 {
 106   if ((l.get_dir()->ino() < r.get_dir()->ino()) ||
 107       (l.get_dir()->ino() == r.get_dir()->ino() &&
 108        (l.get_name() < r.get_name() ||
 109         (l.get_name() == r.get_name() && l.last < r.last))))
 110     return true;
 111   return false;
 112 }
 113
 114
 115 void CDentry::print(ostream& out)
 116 {
 117   out << *this;
 118 }
 119
 120
 121 /*
 122 inodeno_t CDentry::get_ino()
 123 {
 124   if (get_inode())
 125     return get_inode()->ino();
 126   return inodeno_t();
 127 }
 128 */
 129
 130 mds_authority_t CDentry::authority() const
 131 {
 132   return dir->authority();
 133 }
 134
 135
 136 void CDentry::add_waiter(uint64_t tag, MDSInternalContextBase *c)
 137 {
 138   // wait on the directory?
 139   if (tag & (WAIT_UNFREEZE|WAIT_SINGLEAUTH)) {
 140     dir->add_waiter(tag, c);
 141     return;
 142   }
 143   MDSCacheObject::add_waiter(tag, c);
 144 }
 145
 146
 147 version_t CDentry::pre_dirty(version_t min)
 148 {
 149   projected_version = dir->pre_dirty(min);
 150   dout(10) << " pre_dirty " << *this << dendl;
 151   return projected_version;
 152 }
 153
 154
 155 void CDentry::_mark_dirty(LogSegment *ls)
 156 {
 157   // state+pin
 158   if (!state_test(STATE_DIRTY)) {
 159     state_set(STATE_DIRTY);
 160     dir->inc_num_dirty();
 161     get(PIN_DIRTY);
 162     assert(ls);
 163   }
 164   if (ls)
 165     ls->dirty_dentries.push_back(&item_dirty);
 166 }
 167
 168 void CDentry::mark_dirty(version_t pv, LogSegment *ls)
 169 {
 170   dout(10) << " mark_dirty " << *this << dendl;
 171
 172   // i now live in this new dir version
 173   assert(pv <= projected_version);
 174   version = pv;
 175   _mark_dirty(ls);
 176
 177   // mark dir too
 178   dir->mark_dirty(pv, ls);
 179 }
 180
 181
 182 void CDentry::mark_clean()
 183 {
 184   dout(10) << " mark_clean " << *this << dendl;
 185   assert(is_dirty());
 186
 187   // not always true for recalc_auth_bits during resolve finish
 188   //assert(dir->get_version() == 0 || version <= dir->get_version());  // hmm?
 189
 190   // state+pin
 191   state_clear(STATE_DIRTY);
 192   dir->dec_num_dirty();
 193   put(PIN_DIRTY);
 194
 195   item_dirty.remove_myself();
 196
 197   clear_new();
 198 }
 199
 200 void CDentry::mark_new()
 201 {
 202   dout(10) << " mark_new " << *this << dendl;
 203   state_set(STATE_NEW);
 204 }
 205
 206 void CDentry::make_path_string(string& s, bool projected) const
 207 {
 208   if (dir) {
 209     dir->inode->make_path_string(s, projected);
 210   } else {
 211     s = "???";
 212   }
 213   s += "/";
 214   s.append(name.data(), name.length());
 215 }
 216
 217 void CDentry::make_path(filepath& fp, bool projected) const
 218 {
 219   assert(dir);
 220   dir->inode->make_path(fp, projected);
 221   fp.push_dentry(name);
 222 }
 223
 224 /*
 225  * we only add ourselves to remote_parents when the linkage is
 226  * active (no longer projected).  if the passed dnl is projected,
 227  * don't link in, and do that work later in pop_projected_linkage().
 228  */
 229 void CDentry::link_remote(CDentry::linkage_t *dnl, CInode *in)
 230 {
 231   assert(dnl->is_remote());
 232   assert(in->ino() == dnl->get_remote_ino());
 233   dnl->inode = in;
 234
 235   if (dnl == &linkage)
 236     in->add_remote_parent(this);
 237 }
 238
 239 void CDentry::unlink_remote(CDentry::linkage_t *dnl)
 240 {
 241   assert(dnl->is_remote());
 242   assert(dnl->inode);
 243
 244   if (dnl == &linkage)
 245     dnl->inode->remove_remote_parent(this);
 246
 247   dnl->inode = 0;
 248 }
 249
 250 void CDentry::push_projected_linkage()
 251 {
 252   _project_linkage();
 253
 254   if (is_auth()) {
 255     CInode *diri = dir->inode;
 256     if (diri->is_stray())
 257       diri->mdcache->notify_stray_removed();
 258   }
 259 }
 260
 261
 262 void CDentry::push_projected_linkage(CInode *inode)
 263 {
 264   // dirty rstat tracking is in the projected plane
 265   bool dirty_rstat = inode->is_dirty_rstat();
 266   if (dirty_rstat)
 267     inode->clear_dirty_rstat();
 268
 269   _project_linkage()->inode = inode;
 270   inode->push_projected_parent(this);
 271
 272   if (dirty_rstat)
 273     inode->mark_dirty_rstat();
 274
 275   if (is_auth()) {
 276     CInode *diri = dir->inode;
 277     if (diri->is_stray())
 278       diri->mdcache->notify_stray_created();
 279   }
 280 }
 281
 282 CDentry::linkage_t *CDentry::pop_projected_linkage()
 283 {
 284   assert(projected.size());
 285
 286   linkage_t& n = projected.front();
 287
 288   /*
 289    * the idea here is that the link_remote_inode(), link_primary_inode(),
 290    * etc. calls should make linkage identical to &n (and we assert as
 291    * much).
 292    */
 293
 294   if (n.remote_ino) {
 295     dir->link_remote_inode(this, n.remote_ino, n.remote_d_type);
 296     if (n.inode) {
 297       linkage.inode = n.inode;
 298       linkage.inode->add_remote_parent(this);
 299     }
 300   } else if (n.inode) {
 301     dir->link_primary_inode(this, n.inode);
 302     n.inode->pop_projected_parent();
 303   }
 304
 305   assert(n.inode == linkage.inode);
 306   assert(n.remote_ino == linkage.remote_ino);
 307   assert(n.remote_d_type == linkage.remote_d_type);
 308
 309   projected.pop_front();
 310
 311   return &linkage;
 312 }
 313
 314
 315
 316 // ----------------------------
 317 // auth pins
 318
 319 int CDentry::get_num_dir_auth_pins() const
 320 {
 321   assert(!is_projected());
 322   if (get_linkage()->is_primary())
 323     return auth_pins + get_linkage()->get_inode()->get_num_auth_pins();
 324   return auth_pins;
 325 }
 326
 327 bool CDentry::can_auth_pin() const
 328 {
 329   assert(dir);
 330   return dir->can_auth_pin();
 331 }
 332
 333 void CDentry::auth_pin(void *by)
 334 {
 335   if (auth_pins == 0)
 336     get(PIN_AUTHPIN);
 337   auth_pins++;
 338
 339 #ifdef MDS_AUTHPIN_SET
 340   auth_pin_set.insert(by);
 341 #endif
 342
 343   dout(10) << "auth_pin by " << by << " on " << *this
 344            << " now " << auth_pins << "+" << nested_auth_pins
 345            << dendl;
 346
 347   dir->adjust_nested_auth_pins(1, 1, by);
 348 }
 349
 350 void CDentry::auth_unpin(void *by)
 351 {
 352   auth_pins--;
 353
 354 #ifdef MDS_AUTHPIN_SET
 355   assert(auth_pin_set.count(by));
 356   auth_pin_set.erase(auth_pin_set.find(by));
 357 #endif
 358
 359   if (auth_pins == 0)
 360     put(PIN_AUTHPIN);
 361
 362   dout(10) << "auth_unpin by " << by << " on " << *this
 363            << " now " << auth_pins << "+" << nested_auth_pins
 364            << dendl;
 365   assert(auth_pins >= 0);
 366
 367   dir->adjust_nested_auth_pins(-1, -1, by);
 368 }
 369
 370 void CDentry::adjust_nested_auth_pins(int adjustment, int diradj, void *by)
 371 {
 372   nested_auth_pins += adjustment;
 373
 374   dout(35) << "adjust_nested_auth_pins by " << by
 375            << ", change " << adjustment << " yields "
 376            << auth_pins << "+" << nested_auth_pins
 377            << dendl;
 378   assert(nested_auth_pins >= 0);
 379
 380   dir->adjust_nested_auth_pins(adjustment, diradj, by);
 381 }
 382
 383 bool CDentry::is_frozen() const
 384 {
 385   return dir->is_frozen();
 386 }
 387
 388 bool CDentry::is_freezing() const
 389 {
 390   return dir->is_freezing();
 391 }
 392
 393 void CDentry::decode_replica(bufferlist::iterator& p, bool is_new)
 394 {
 395   __u32 nonce;
 396   ::decode(nonce, p);
 397   replica_nonce = nonce;
 398
 399   ::decode(first, p);
 400
 401   inodeno_t rino;
 402   unsigned char rdtype;
 403   __s32 ls;
 404   ::decode(rino, p);
 405   ::decode(rdtype, p);
 406   ::decode(ls, p);
 407
 408   if (is_new) {
 409     if (rino)
 410       dir->link_remote_inode(this, rino, rdtype);
 411     lock.set_state(ls);
 412   }
 413 }
 414
 415 // ----------------------------
 416 // locking
 417
 418 void CDentry::set_object_info(MDSCacheObjectInfo &info)
 419 {
 420   info.dirfrag = dir->dirfrag();
 421   info.dname = name;
 422   info.snapid = last;
 423 }
 424
 425 void CDentry::encode_lock_state(int type, bufferlist& bl)
 426 {
 427   ::encode(first, bl);
 428
 429   // null, ino, or remote_ino?
 430   char c;
 431   if (linkage.is_primary()) {
 432     c = 1;
 433     ::encode(c, bl);
 434     ::encode(linkage.get_inode()->inode.ino, bl);
 435   }
 436   else if (linkage.is_remote()) {
 437     c = 2;
 438     ::encode(c, bl);
 439     ::encode(linkage.get_remote_ino(), bl);
 440   }
 441   else if (linkage.is_null()) {
 442     // encode nothing.
 443   }
 444   else ceph_abort();
 445 }
 446
 447 void CDentry::decode_lock_state(int type, bufferlist& bl)
 448 {
 449   bufferlist::iterator p = bl.begin();
 450
 451   snapid_t newfirst;
 452   ::decode(newfirst, p);
 453
 454   if (!is_auth() && newfirst != first) {
 455     dout(10) << "decode_lock_state first " << first << " -> " << newfirst << dendl;
 456     assert(newfirst > first);
 457     first = newfirst;
 458   }
 459
 460   if (p.end()) {
 461     // null
 462     assert(linkage.is_null());
 463     return;
 464   }
 465
 466   char c;
 467   inodeno_t ino;
 468   ::decode(c, p);
 469
 470   switch (c) {
 471   case 1:
 472   case 2:
 473     ::decode(ino, p);
 474     // newly linked?
 475     if (linkage.is_null() && !is_auth()) {
 476       // force trim from cache!
 477       dout(10) << "decode_lock_state replica dentry null -> non-null, must trim" << dendl;
 478       //assert(get_num_ref() == 0);
 479     } else {
 480       // verify?
 481
 482     }
 483     break;
 484   default:
 485     ceph_abort();
 486   }
 487 }
 488
 489
 490 ClientLease *CDentry::add_client_lease(client_t c, Session *session)
 491 {
 492   ClientLease *l;
 493   if (client_lease_map.count(c))
 494     l = client_lease_map[c];
 495   else {
 496     dout(20) << "add_client_lease client." << c << " on " << lock << dendl;
 497     if (client_lease_map.empty())
 498       get(PIN_CLIENTLEASE);
 499     l = client_lease_map[c] = new ClientLease(c, this);
 500     l->seq = ++session->lease_seq;
 501
 502     lock.get_client_lease();
 503   }
 504
 505   return l;
 506 }
 507
 508 void CDentry::remove_client_lease(ClientLease *l, Locker *locker)
 509 {
 510   assert(l->parent == this);
 511
 512   bool gather = false;
 513
 514   dout(20) << "remove_client_lease client." << l->client << " on " << lock << dendl;
 515   lock.put_client_lease();
 516   if (lock.get_num_client_lease() == 0 && !lock.is_stable())
 517     gather = true;
 518
 519   client_lease_map.erase(l->client);
 520   l->item_lease.remove_myself();
 521   l->item_session_lease.remove_myself();
 522   delete l;
 523
 524   if (client_lease_map.empty())
 525     put(PIN_CLIENTLEASE);
 526
 527   if (gather)
 528     locker->eval_gather(&lock);
 529 }
 530
 531 void CDentry::remove_client_leases(Locker *locker)
 532 {
 533   while (!client_lease_map.empty())
 534     remove_client_lease(client_lease_map.begin()->second, locker);
 535 }
 536
 537 void CDentry::_put()
 538 {
 539   if (get_num_ref() <= ((int)is_dirty() + 1)) {
 540     CDentry::linkage_t *dnl = get_projected_linkage();
 541     if (dnl->is_primary()) {
 542       CInode *in = dnl->get_inode();
 543       if (get_num_ref() == (int)is_dirty() + !!in->get_num_ref())
 544         in->mdcache->maybe_eval_stray(in, true);
 545     }
 546   }
 547 }
 548
 549 void CDentry::dump(Formatter *f) const
 550 {
 551   assert(f != NULL);
 552
 553   filepath path;
 554   make_path(path);
 555
 556   f->dump_string("path", path.get_path());
 557   f->dump_unsigned("path_ino", path.get_ino().val);
 558   f->dump_unsigned("snap_first", first);
 559   f->dump_unsigned("snap_last", last);
 560
 561   f->dump_bool("is_primary", get_linkage()->is_primary());
 562   f->dump_bool("is_remote", get_linkage()->is_remote());
 563   f->dump_bool("is_null", get_linkage()->is_null());
 564   f->dump_bool("is_new", is_new());
 565   if (get_linkage()->get_inode()) {
 566     f->dump_unsigned("inode", get_linkage()->get_inode()->ino());
 567   } else {
 568     f->dump_unsigned("inode", 0);
 569   }
 570
 571   if (linkage.is_remote()) {
 572     f->dump_string("remote_type", linkage.get_remote_d_type_string());
 573   } else {
 574     f->dump_string("remote_type", "");
 575   }
 576
 577   f->dump_unsigned("version", get_version());
 578   f->dump_unsigned("projected_version", get_projected_version());
 579
 580   f->dump_int("auth_pins", auth_pins);
 581   f->dump_int("nested_auth_pins", nested_auth_pins);
 582
 583   MDSCacheObject::dump(f);
 584
 585   f->open_object_section("lock");
 586   lock.dump(f);
 587   f->close_section();
 588
 589   f->open_object_section("versionlock");
 590   versionlock.dump(f);
 591   f->close_section();
 592
 593   f->open_array_section("states");
 594   MDSCacheObject::dump_states(f);
 595   if (state_test(STATE_NEW))
 596     f->dump_string("state", "new");
 597   if (state_test(STATE_FRAGMENTING))
 598     f->dump_string("state", "fragmenting");
 599   if (state_test(STATE_PURGING))
 600     f->dump_string("state", "purging");
 601   if (state_test(STATE_BADREMOTEINO))
 602     f->dump_string("state", "badremoteino");
 603   if (state_test(STATE_STRAY))
 604     f->dump_string("state", "stray");
 605   f->close_section();
 606 }
 607
 608 std::string CDentry::linkage_t::get_remote_d_type_string() const
 609 {
 610   switch (DTTOIF(remote_d_type)) {
 611     case S_IFSOCK: return "sock";
 612     case S_IFLNK: return "lnk";
 613     case S_IFREG: return "reg";
 614     case S_IFBLK: return "blk";
 615     case S_IFDIR: return "dir";
 616     case S_IFCHR: return "chr";
 617     case S_IFIFO: return "fifo";
 618     default: ceph_abort(); return "";
 619   }
 620 }