ceph/src/mds/SnapRealm.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 #include "SnapRealm.h"
  16 #include "MDCache.h"
  17 #include "MDSRank.h"
  18 #include "SnapClient.h"
  19
  20 #include <string_view>
  21
  22
  23 /*
  24  * SnapRealm
  25  */
  26
  27 #define dout_context g_ceph_context
  28 #define dout_subsys ceph_subsys_mds
  29 #undef dout_prefix
  30 #define dout_prefix _prefix(_dout, mdcache->mds->get_nodeid(), inode, srnode.seq, this)
  31 static ostream& _prefix(std::ostream *_dout, int whoami, const CInode *inode,
  32                         uint64_t seq, const SnapRealm *realm) {
  33   return *_dout << " mds." << whoami
  34                 << ".cache.snaprealm(" << inode->ino()
  35                 << " seq " << seq << " " << realm << ") ";
  36 }
  37
  38 ostream& operator<<(ostream& out, const SnapRealm& realm)
  39 {
  40   out << "snaprealm(" << realm.inode->ino()
  41       << " seq " << realm.srnode.seq
  42       << " lc " << realm.srnode.last_created
  43       << " cr " << realm.srnode.created;
  44   if (realm.srnode.created != realm.srnode.current_parent_since)
  45     out << " cps " << realm.srnode.current_parent_since;
  46   out << " snaps=" << realm.srnode.snaps;
  47   out << " past_parent_snaps=" << realm.srnode.past_parent_snaps;
  48
  49   if (realm.srnode.past_parents.size()) {
  50     out << " past_parents=(";
  51     for (map<snapid_t, snaplink_t>::const_iterator p = realm.srnode.past_parents.begin();
  52          p != realm.srnode.past_parents.end();
  53          ++p) {
  54       if (p != realm.srnode.past_parents.begin()) out << ",";
  55       out << p->second.first << "-" << p->first
  56           << "=" << p->second.ino;
  57     }
  58     out << ")";
  59   }
  60
  61   if (realm.srnode.is_parent_global())
  62     out << " global ";
  63   out << " " << &realm << ")";
  64   return out;
  65 }
  66
  67 SnapRealm::SnapRealm(MDCache *c, CInode *in) :
  68     mdcache(c), inode(in)
  69 {
  70   global = (inode->ino() == MDS_INO_GLOBAL_SNAPREALM);
  71 }
  72
  73 void SnapRealm::add_open_past_parent(SnapRealm *parent, snapid_t last)
  74 {
  75   auto p = open_past_parents.find(parent->inode->ino());
  76   if (p != open_past_parents.end()) {
  77     ceph_assert(p->second.second.count(last) == 0);
  78     p->second.second.insert(last);
  79   } else {
  80     open_past_parents[parent->inode->ino()].first = parent;
  81     open_past_parents[parent->inode->ino()].second.insert(last);
  82     parent->open_past_children.insert(this);
  83     parent->inode->get(CInode::PIN_PASTSNAPPARENT);
  84   }
  85   ++num_open_past_parents;
  86 }
  87
  88 void SnapRealm::remove_open_past_parent(inodeno_t ino, snapid_t last)
  89 {
  90   auto p = open_past_parents.find(ino);
  91   ceph_assert(p != open_past_parents.end());
  92   auto q = p->second.second.find(last);
  93   ceph_assert(q != p->second.second.end());
  94   p->second.second.erase(q);
  95   --num_open_past_parents;
  96   if (p->second.second.empty()) {
  97     SnapRealm *parent = p->second.first;
  98     open_past_parents.erase(p);
  99     parent->open_past_children.erase(this);
 100     parent->inode->put(CInode::PIN_PASTSNAPPARENT);
 101   }
 102 }
 103
 104 struct C_SR_RetryOpenParents : public MDSContext {
 105   SnapRealm *sr;
 106   snapid_t first, last, parent_last;
 107   inodeno_t parent;
 108   MDSContext* fin;
 109   C_SR_RetryOpenParents(SnapRealm *s, snapid_t f, snapid_t l, snapid_t pl,
 110                         inodeno_t p, MDSContext *c) :
 111     sr(s), first(f), last(l), parent_last(pl),  parent(p), fin(c) {
 112     sr->inode->get(CInode::PIN_OPENINGSNAPPARENTS);
 113   }
 114   MDSRank *get_mds() override { return sr->mdcache->mds; }
 115   void finish(int r) override {
 116     if (r < 0)
 117       sr->_remove_missing_parent(parent_last, parent, r);
 118     if (sr->_open_parents(fin, first, last)) {
 119       if (fin)
 120         fin->complete(0);
 121     }
 122     sr->inode->put(CInode::PIN_OPENINGSNAPPARENTS);
 123   }
 124 };
 125
 126 void SnapRealm::_remove_missing_parent(snapid_t snapid, inodeno_t parent, int err)
 127 {
 128   map<snapid_t, snaplink_t>::iterator p = srnode.past_parents.find(snapid);
 129   if (p != srnode.past_parents.end()) {
 130     dout(10) << __func__ << " " << parent << " [" << p->second.first << ","
 131              << p->first << "]  errno " << err << dendl;
 132     srnode.past_parents.erase(p);
 133     past_parents_dirty = true;
 134   } else {
 135     dout(10) << __func__ << " " << parent << " not found" << dendl;
 136   }
 137 }
 138
 139 bool SnapRealm::_open_parents(MDSContext *finish, snapid_t first, snapid_t last)
 140 {
 141   dout(10) << "open_parents [" << first << "," << last << "]" << dendl;
 142   if (open)
 143     return true;
 144
 145   // make sure my current parents' parents are open...
 146   if (parent) {
 147     dout(10) << " current parent [" << srnode.current_parent_since << ",head] is " << *parent
 148              << " on " << *parent->inode << dendl;
 149     if (last >= srnode.current_parent_since &&
 150         !parent->_open_parents(finish, std::max(first, srnode.current_parent_since), last))
 151       return false;
 152   }
 153
 154   if (!srnode.past_parent_snaps.empty())
 155     ceph_assert(mdcache->mds->snapclient->get_cached_version() > 0);
 156
 157   if (!srnode.past_parents.empty() &&
 158       mdcache->mds->allows_multimds_snaps()) {
 159     dout(10) << " skip non-empty past_parents since multimds_snaps is allowed" << dendl;
 160     open = true;
 161     return true;
 162   }
 163
 164   // and my past parents too!
 165   ceph_assert(srnode.past_parents.size() >= num_open_past_parents);
 166   if (srnode.past_parents.size() > num_open_past_parents) {
 167     for (map<snapid_t, snaplink_t>::iterator p = srnode.past_parents.begin();
 168          p != srnode.past_parents.end(); ) {
 169       dout(10) << " past_parent [" << p->second.first << "," << p->first << "] is "
 170                << p->second.ino << dendl;
 171       CInode *parent = mdcache->get_inode(p->second.ino);
 172       if (!parent) {
 173         C_SR_RetryOpenParents *fin = new C_SR_RetryOpenParents(this, first, last, p->first,
 174                                                                p->second.ino, finish);
 175         mdcache->open_ino(p->second.ino, mdcache->mds->mdsmap->get_metadata_pool(), fin);
 176         return false;
 177       }
 178       if (parent->state_test(CInode::STATE_PURGING)) {
 179         dout(10) << " skip purging past_parent " << *parent << dendl;
 180         srnode.past_parents.erase(p++);
 181         past_parents_dirty = true;
 182         continue;
 183       }
 184       ceph_assert(parent->snaprealm);  // hmm!
 185       if (!parent->snaprealm->_open_parents(finish, p->second.first, p->first))
 186         return false;
 187       auto q = open_past_parents.find(p->second.ino);
 188       if (q == open_past_parents.end() ||
 189           q->second.second.count(p->first) == 0) {
 190         add_open_past_parent(parent->snaprealm, p->first);
 191       }
 192       ++p;
 193     }
 194   }
 195
 196   open = true;
 197   return true;
 198 }
 199
 200 bool SnapRealm::open_parents(MDSContext *retryorfinish) {
 201   if (!_open_parents(retryorfinish))
 202     return false;
 203   delete retryorfinish;
 204   return true;
 205 }
 206
 207 bool SnapRealm::have_past_parents_open(snapid_t first, snapid_t last) const
 208 {
 209   dout(10) << "have_past_parents_open [" << first << "," << last << "]" << dendl;
 210   if (open)
 211     return true;
 212
 213   if (!srnode.past_parent_snaps.empty())
 214     ceph_assert(mdcache->mds->snapclient->get_cached_version() > 0);
 215
 216   if (!srnode.past_parents.empty() &&
 217       mdcache->mds->allows_multimds_snaps()) {
 218     dout(10) << " skip non-empty past_parents since multimds_snaps is allowed" << dendl;
 219     open = true;
 220     return true;
 221   }
 222
 223   for (auto p = srnode.past_parents.lower_bound(first);
 224        p != srnode.past_parents.end();
 225        ++p) {
 226     if (p->second.first > last)
 227       break;
 228     dout(10) << " past parent [" << p->second.first << "," << p->first << "] was "
 229              << p->second.ino << dendl;
 230     auto q = open_past_parents.find(p->second.ino);
 231     if (q == open_past_parents.end()) {
 232       dout(10) << " past parent " << p->second.ino << " is not open" << dendl;
 233       return false;
 234     }
 235     SnapRealm *parent_realm = q->second.first;
 236     if (!parent_realm->have_past_parents_open(std::max(first, p->second.first),
 237                                               std::min(last, p->first)))
 238       return false;
 239   }
 240
 241   open = true;
 242   return true;
 243 }
 244
 245 void SnapRealm::close_parents()
 246 {
 247   for (auto p = open_past_parents.begin(); p != open_past_parents.end(); ++p) {
 248     num_open_past_parents -= p->second.second.size();
 249     p->second.first->inode->put(CInode::PIN_PASTSNAPPARENT);
 250     p->second.first->open_past_children.erase(this);
 251   }
 252   open_past_parents.clear();
 253 }
 254
 255
 256 /*
 257  * get list of snaps for this realm.  we must include parents' snaps
 258  * for the intervals during which they were our parent.
 259  */
 260 void SnapRealm::build_snap_set() const
 261 {
 262   dout(10) << "build_snap_set on " << *this << dendl;
 263
 264   cached_snaps.clear();
 265
 266   if (global) {
 267     mdcache->mds->snapclient->get_snaps(cached_snaps);
 268     return;
 269   }
 270
 271   // include my snaps
 272   for (const auto& p : srnode.snaps)
 273     cached_snaps.insert(p.first);
 274
 275   if (!srnode.past_parent_snaps.empty()) {
 276     set<snapid_t> snaps = mdcache->mds->snapclient->filter(srnode.past_parent_snaps);
 277     if (!snaps.empty()) {
 278       snapid_t last = *snaps.rbegin();
 279       cached_seq = std::max(cached_seq, last);
 280       cached_last_created = std::max(cached_last_created, last);
 281     }
 282     cached_snaps.insert(snaps.begin(), snaps.end());
 283   } else {
 284     // include snaps for parents
 285     for (const auto& p : srnode.past_parents) {
 286       const CInode *oldparent = mdcache->get_inode(p.second.ino);
 287       ceph_assert(oldparent);  // call open_parents first!
 288       ceph_assert(oldparent->snaprealm);
 289
 290       const set<snapid_t>& snaps = oldparent->snaprealm->get_snaps();
 291       snapid_t last = 0;
 292       for (auto q = snaps.lower_bound(p.second.first);
 293           q != snaps.end() && *q <= p.first;
 294           q++) {
 295         cached_snaps.insert(*q);
 296         last = *q;
 297       }
 298       cached_seq = std::max(cached_seq, last);
 299       cached_last_created = std::max(cached_last_created, last);
 300     }
 301   }
 302
 303   snapid_t parent_seq = parent ? parent->get_newest_seq() : snapid_t(0);
 304   if (parent_seq >= srnode.current_parent_since) {
 305     auto& snaps = parent->get_snaps();
 306     auto p = snaps.lower_bound(srnode.current_parent_since);
 307     cached_snaps.insert(p, snaps.end());
 308     cached_seq = std::max(cached_seq, parent_seq);
 309     cached_last_created = std::max(cached_last_created, parent->get_last_created());
 310   }
 311 }
 312
 313 void SnapRealm::check_cache() const
 314 {
 315   ceph_assert(have_past_parents_open());
 316   snapid_t seq;
 317   snapid_t last_created;
 318   snapid_t last_destroyed = mdcache->mds->snapclient->get_last_destroyed();
 319   if (global || srnode.is_parent_global()) {
 320     last_created = mdcache->mds->snapclient->get_last_created();
 321     seq = std::max(last_created, last_destroyed);
 322   } else {
 323     last_created = srnode.last_created;
 324     seq = srnode.seq;
 325   }
 326   if (cached_seq >= seq &&
 327       cached_last_destroyed == last_destroyed)
 328     return;
 329
 330   cached_snap_context.clear();
 331
 332   cached_seq = seq;
 333   cached_last_created = last_created;
 334   cached_last_destroyed = last_destroyed;
 335   build_snap_set();
 336
 337   build_snap_trace();
 338
 339   dout(10) << "check_cache rebuilt " << cached_snaps
 340            << " seq " << seq
 341            << " cached_seq " << cached_seq
 342            << " cached_last_created " << cached_last_created
 343            << " cached_last_destroyed " << cached_last_destroyed
 344            << ")" << dendl;
 345 }
 346
 347 const set<snapid_t>& SnapRealm::get_snaps() const
 348 {
 349   check_cache();
 350   dout(10) << "get_snaps " << cached_snaps
 351            << " (seq " << srnode.seq << " cached_seq " << cached_seq << ")"
 352            << dendl;
 353   return cached_snaps;
 354 }
 355
 356 /*
 357  * build vector in reverse sorted order
 358  */
 359 const SnapContext& SnapRealm::get_snap_context() const
 360 {
 361   check_cache();
 362
 363   if (!cached_snap_context.seq) {
 364     cached_snap_context.seq = cached_seq;
 365     cached_snap_context.snaps.resize(cached_snaps.size());
 366     unsigned i = 0;
 367     for (set<snapid_t>::reverse_iterator p = cached_snaps.rbegin();
 368          p != cached_snaps.rend();
 369          ++p)
 370       cached_snap_context.snaps[i++] = *p;
 371   }
 372
 373   return cached_snap_context;
 374 }
 375
 376 void SnapRealm::get_snap_info(map<snapid_t, const SnapInfo*>& infomap, snapid_t first, snapid_t last)
 377 {
 378   const set<snapid_t>& snaps = get_snaps();
 379   dout(10) << "get_snap_info snaps " << snaps << dendl;
 380
 381   // include my snaps within interval [first,last]
 382   for (auto p = srnode.snaps.lower_bound(first); // first element >= first
 383        p != srnode.snaps.end() && p->first <= last;
 384        ++p)
 385     infomap[p->first] = &p->second;
 386
 387   if (!srnode.past_parent_snaps.empty()) {
 388     set<snapid_t> snaps;
 389     for (auto p = srnode.past_parent_snaps.lower_bound(first); // first element >= first
 390         p != srnode.past_parent_snaps.end() && *p <= last;
 391         ++p) {
 392       snaps.insert(*p);
 393     }
 394
 395     map<snapid_t, const SnapInfo*> _infomap;
 396     mdcache->mds->snapclient->get_snap_infos(_infomap, snaps);
 397     infomap.insert(_infomap.begin(), _infomap.end());
 398   } else {
 399     // include snaps for parents during intervals that intersect [first,last]
 400     for (map<snapid_t, snaplink_t>::iterator p = srnode.past_parents.lower_bound(first);
 401         p != srnode.past_parents.end() && p->first >= first && p->second.first <= last;
 402         ++p) {
 403       CInode *oldparent = mdcache->get_inode(p->second.ino);
 404       ceph_assert(oldparent);  // call open_parents first!
 405       ceph_assert(oldparent->snaprealm);
 406       oldparent->snaprealm->get_snap_info(infomap,
 407                                           std::max(first, p->second.first),
 408                                           std::min(last, p->first));
 409     }
 410   }
 411
 412   if (srnode.current_parent_since <= last && parent)
 413     parent->get_snap_info(infomap, std::max(first, srnode.current_parent_since), last);
 414 }
 415
 416 std::string_view SnapRealm::get_snapname(snapid_t snapid, inodeno_t atino)
 417 {
 418   auto srnode_snaps_entry = srnode.snaps.find(snapid);
 419   if (srnode_snaps_entry != srnode.snaps.end()) {
 420     if (atino == inode->ino())
 421       return srnode_snaps_entry->second.name;
 422     else
 423       return srnode_snaps_entry->second.get_long_name();
 424   }
 425
 426   if (!srnode.past_parent_snaps.empty()) {
 427     if (srnode.past_parent_snaps.count(snapid)) {
 428       const SnapInfo *sinfo = mdcache->mds->snapclient->get_snap_info(snapid);
 429       if (sinfo) {
 430         if (atino == sinfo->ino)
 431           return sinfo->name;
 432         else
 433           return sinfo->get_long_name();
 434       }
 435     }
 436   } else {
 437     map<snapid_t,snaplink_t>::iterator p = srnode.past_parents.lower_bound(snapid);
 438     if (p != srnode.past_parents.end() && p->second.first <= snapid) {
 439       CInode *oldparent = mdcache->get_inode(p->second.ino);
 440       ceph_assert(oldparent);  // call open_parents first!
 441       ceph_assert(oldparent->snaprealm);
 442       return oldparent->snaprealm->get_snapname(snapid, atino);
 443     }
 444   }
 445
 446   ceph_assert(srnode.current_parent_since <= snapid);
 447   ceph_assert(parent);
 448   return parent->get_snapname(snapid, atino);
 449 }
 450
 451 snapid_t SnapRealm::resolve_snapname(std::string_view n, inodeno_t atino, snapid_t first, snapid_t last)
 452 {
 453   // first try me
 454   dout(10) << "resolve_snapname '" << n << "' in [" << first << "," << last << "]" << dendl;
 455
 456   bool actual = (atino == inode->ino());
 457   string pname;
 458   inodeno_t pino;
 459   if (n.length() && n[0] == '_') {
 460     size_t next_ = n.find_last_of('_');
 461     if (next_ > 1 && next_ + 1 < n.length()) {
 462       pname = n.substr(1, next_ - 1);
 463       pino = atoll(n.data() + next_ + 1);
 464       dout(10) << " " << n << " parses to name '" << pname << "' dirino " << pino << dendl;
 465     }
 466   }
 467
 468   for (auto p = srnode.snaps.lower_bound(first); // first element >= first
 469        p != srnode.snaps.end() && p->first <= last;
 470        ++p) {
 471     dout(15) << " ? " << p->second << dendl;
 472     //if (num && p->second.snapid == num)
 473     //return p->first;
 474     if (actual && p->second.name == n)
 475         return p->first;
 476     if (!actual && p->second.name == pname && p->second.ino == pino)
 477       return p->first;
 478   }
 479
 480   if (!srnode.past_parent_snaps.empty()) {
 481     set<snapid_t> snaps;
 482     for (auto p = srnode.past_parent_snaps.lower_bound(first); // first element >= first
 483          p != srnode.past_parent_snaps.end() && *p <= last;
 484          ++p)
 485       snaps.insert(*p);
 486
 487     map<snapid_t, const SnapInfo*> _infomap;
 488     mdcache->mds->snapclient->get_snap_infos(_infomap, snaps);
 489
 490     for (auto& it : _infomap) {
 491       dout(15) << " ? " << *it.second << dendl;
 492       actual = (it.second->ino == atino);
 493       if (actual && it.second->name == n)
 494         return it.first;
 495       if (!actual && it.second->name == pname && it.second->ino == pino)
 496         return it.first;
 497     }
 498   } else {
 499     // include snaps for parents during intervals that intersect [first,last]
 500     for (map<snapid_t, snaplink_t>::iterator p = srnode.past_parents.lower_bound(first);
 501          p != srnode.past_parents.end() && p->first >= first && p->second.first <= last;
 502          ++p) {
 503       CInode *oldparent = mdcache->get_inode(p->second.ino);
 504       ceph_assert(oldparent);  // call open_parents first!
 505       ceph_assert(oldparent->snaprealm);
 506       snapid_t r = oldparent->snaprealm->resolve_snapname(n, atino,
 507                                                           std::max(first, p->second.first),
 508                                                           std::min(last, p->first));
 509       if (r)
 510         return r;
 511     }
 512   }
 513
 514   if (parent && srnode.current_parent_since <= last)
 515     return parent->resolve_snapname(n, atino, std::max(first, srnode.current_parent_since), last);
 516   return 0;
 517 }
 518
 519
 520 void SnapRealm::adjust_parent()
 521 {
 522   SnapRealm *newparent;
 523   if (srnode.is_parent_global()) {
 524     newparent = mdcache->get_global_snaprealm();
 525   } else {
 526     CDentry *pdn = inode->get_parent_dn();
 527     newparent = pdn ? pdn->get_dir()->get_inode()->find_snaprealm() : NULL;
 528   }
 529   if (newparent != parent) {
 530     dout(10) << "adjust_parent " << parent << " -> " << newparent << dendl;
 531     if (parent)
 532       parent->open_children.erase(this);
 533     parent = newparent;
 534     if (parent)
 535       parent->open_children.insert(this);
 536
 537     invalidate_cached_snaps();
 538   }
 539 }
 540
 541 void SnapRealm::split_at(SnapRealm *child)
 542 {
 543   dout(10) << "split_at " << *child
 544            << " on " << *child->inode << dendl;
 545
 546   if (inode->is_mdsdir() || !child->inode->is_dir()) {
 547     // it's not a dir.
 548     if (child->inode->containing_realm) {
 549       //  - no open children.
 550       //  - only need to move this child's inode's caps.
 551       child->inode->move_to_realm(child);
 552     } else {
 553       // no caps, nothing to move/split.
 554       dout(20) << " split no-op, no caps to move on file " << *child->inode << dendl;
 555       ceph_assert(!child->inode->is_any_caps());
 556     }
 557     return;
 558   }
 559
 560   // it's a dir.
 561
 562   // split open_children
 563   dout(10) << " open_children are " << open_children << dendl;
 564   for (set<SnapRealm*>::iterator p = open_children.begin();
 565        p != open_children.end(); ) {
 566     SnapRealm *realm = *p;
 567     if (realm != child &&
 568         child->inode->is_ancestor_of(realm->inode)) {
 569       dout(20) << " child gets child realm " << *realm << " on " << *realm->inode << dendl;
 570       realm->parent = child;
 571       child->open_children.insert(realm);
 572       open_children.erase(p++);
 573     } else {
 574       dout(20) << "    keeping child realm " << *realm << " on " << *realm->inode << dendl;
 575       ++p;
 576     }
 577   }
 578
 579   // split inodes_with_caps
 580   for (elist<CInode*>::iterator p = inodes_with_caps.begin(member_offset(CInode, item_caps));
 581        !p.end(); ) {
 582     CInode *in = *p;
 583     ++p;
 584     // does inode fall within the child realm?
 585     if (child->inode->is_ancestor_of(in)) {
 586       dout(20) << " child gets " << *in << dendl;
 587       in->move_to_realm(child);
 588     } else {
 589       dout(20) << "    keeping " << *in << dendl;
 590     }
 591   }
 592 }
 593
 594 void SnapRealm::merge_to(SnapRealm *newparent)
 595 {
 596   if (!newparent)
 597     newparent = parent;
 598   dout(10) << "merge to " << *newparent << " on " << *newparent->inode << dendl;
 599
 600   ceph_assert(open_past_children.empty());
 601
 602   dout(10) << " open_children are " << open_children << dendl;
 603   for (auto realm : open_children) {
 604     dout(20) << " child realm " << *realm << " on " << *realm->inode << dendl;
 605     newparent->open_children.insert(realm);
 606     realm->parent = newparent;
 607   }
 608   open_children.clear();
 609
 610   elist<CInode*>::iterator p = inodes_with_caps.begin(member_offset(CInode, item_caps));
 611   while (!p.end()) {
 612     CInode *in = *p;
 613     ++p;
 614     in->move_to_realm(newparent);
 615   }
 616   ceph_assert(inodes_with_caps.empty());
 617
 618   // delete this
 619   inode->close_snaprealm();
 620 }
 621
 622 const bufferlist& SnapRealm::get_snap_trace() const
 623 {
 624   check_cache();
 625   return cached_snap_trace;
 626 }
 627
 628 void SnapRealm::build_snap_trace() const
 629 {
 630   cached_snap_trace.clear();
 631
 632   if (global) {
 633     SnapRealmInfo info(inode->ino(), 0, cached_seq, 0);
 634     info.my_snaps.reserve(cached_snaps.size());
 635     for (auto p = cached_snaps.rbegin(); p != cached_snaps.rend(); ++p)
 636       info.my_snaps.push_back(*p);
 637
 638     dout(10) << "build_snap_trace my_snaps " << info.my_snaps << dendl;
 639     encode(info, cached_snap_trace);
 640     return;
 641   }
 642
 643   SnapRealmInfo info(inode->ino(), srnode.created, srnode.seq, srnode.current_parent_since);
 644   if (parent) {
 645     info.h.parent = parent->inode->ino();
 646
 647     set<snapid_t> past;
 648     if (!srnode.past_parent_snaps.empty()) {
 649       past = mdcache->mds->snapclient->filter(srnode.past_parent_snaps);
 650       if (srnode.is_parent_global()) {
 651         auto p = past.lower_bound(srnode.current_parent_since);
 652         past.erase(p, past.end());
 653       }
 654     } else if (!srnode.past_parents.empty()) {
 655       const set<snapid_t>& snaps = get_snaps();
 656       for (const auto& p : srnode.past_parents) {
 657         for (auto q = snaps.lower_bound(p.second.first);
 658              q != snaps.end() && *q <= p.first;
 659              q++) {
 660           if (srnode.snaps.count(*q))
 661             continue;
 662           past.insert(*q);
 663         }
 664       }
 665     }
 666
 667     if (!past.empty()) {
 668       info.prior_parent_snaps.reserve(past.size());
 669       for (set<snapid_t>::reverse_iterator p = past.rbegin(); p != past.rend(); ++p)
 670         info.prior_parent_snaps.push_back(*p);
 671       dout(10) << "build_snap_trace prior_parent_snaps from [1," << *past.rbegin() << "] "
 672                << info.prior_parent_snaps << dendl;
 673     }
 674   }
 675
 676   info.my_snaps.reserve(srnode.snaps.size());
 677   for (auto p = srnode.snaps.rbegin();
 678        p != srnode.snaps.rend();
 679        ++p)
 680     info.my_snaps.push_back(p->first);
 681   dout(10) << "build_snap_trace my_snaps " << info.my_snaps << dendl;
 682
 683   encode(info, cached_snap_trace);
 684
 685   if (parent)
 686     cached_snap_trace.append(parent->get_snap_trace());
 687 }
 688
 689 void SnapRealm::prune_past_parents()
 690 {
 691   dout(10) << "prune_past_parents" << dendl;
 692   check_cache();
 693
 694   // convert past_parents to past_parent_snaps
 695   if (!srnode.past_parents.empty()) {
 696     for (auto p = cached_snaps.begin();
 697          p != cached_snaps.end() && *p < srnode.current_parent_since;
 698          ++p) {
 699       if (!srnode.snaps.count(*p))
 700         srnode.past_parent_snaps.insert(*p);
 701     }
 702     srnode.past_parents.clear();
 703     past_parents_dirty = true;
 704   }
 705
 706   for (auto p = srnode.past_parent_snaps.begin();
 707        p != srnode.past_parent_snaps.end(); ) {
 708     auto q = cached_snaps.find(*p);
 709     if (q == cached_snaps.end()) {
 710       dout(10) << "prune_past_parents pruning " << *p << dendl;
 711       srnode.past_parent_snaps.erase(p++);
 712     } else {
 713       dout(10) << "prune_past_parents keeping " << *p << dendl;
 714       ++p;
 715     }
 716   }
 717 }
 718