catalog_traversal.h

Go to the documentation of this file.

#ifndef CVMFS_CATALOG_TRAVERSAL_H_
#define CVMFS_CATALOG_TRAVERSAL_H_

#include <cassert>
#include <limits>
#include <set>
#include <stack>
#include <string>
#include <vector>

#include "catalog.h"
#include "compression/compression.h"
#include "crypto/signature.h"
#include "history_sqlite.h"
#include "manifest.h"
#include "object_fetcher.h"
#include "util/concurrency.h"
#include "util/logging.h"

namespace catalog {
class Catalog;
class WritableCatalog;
}

namespace swissknife {

template <class CatalogT>
struct CatalogTraversalData {
  CatalogTraversalData(const CatalogT     *catalog,
                       const shash::Any   &catalog_hash,
                       const unsigned      tree_level,
                       const size_t        file_size,
                       const uint64_t      history_depth)
  : catalog(catalog)
  , catalog_hash(catalog_hash)
  , tree_level(tree_level)
  , file_size(file_size)
  , history_depth(history_depth) {}

  const CatalogT     *catalog;
  const shash::Any    catalog_hash;
  const unsigned int  tree_level;
  const size_t        file_size;
  const uint64_t      history_depth;
};


template <class CatalogT>
class CatalogTraversalInfoShim {
 public:
  virtual ~CatalogTraversalInfoShim() { }

  // Default implementation: use the catalog's timestamp
  virtual uint64_t GetLastModified(const CatalogT *catalog) {
    return catalog->GetLastModified();
  }
};


template <class ObjectFetcherT>
class CatalogTraversalBase
  : public Observable<CatalogTraversalData<typename ObjectFetcherT::CatalogTN> >
{
 public:
  typedef ObjectFetcherT                      ObjectFetcherTN;
  typedef typename ObjectFetcherT::CatalogTN  CatalogTN;
  typedef typename ObjectFetcherT::HistoryTN  HistoryTN;
  typedef CatalogTraversalData<CatalogTN>     CallbackDataTN;

  struct Parameters {
    Parameters()
      : object_fetcher(NULL)
      , history(kNoHistory)
      , timestamp(kNoTimestampThreshold)
      , no_repeat_history(false)
      , no_close(false)
      , ignore_load_failure(false)
      , quiet(false)
      , num_threads(8)
      , serialize_callbacks(true) {}

    static const uint64_t     kFullHistory;
    static const uint64_t     kNoHistory;
    static const time_t       kNoTimestampThreshold;

    ObjectFetcherT *object_fetcher;

    uint64_t        history;
    time_t          timestamp;
    bool            no_repeat_history;
    bool            no_close;
    bool            ignore_load_failure;
    bool            quiet;
    // These parameters only apply for CatalogTraversalParallel
    unsigned int    num_threads;
    bool            serialize_callbacks;
  };

  enum TraversalType {
    kBreadthFirst,
    kDepthFirst
  };

  explicit CatalogTraversalBase(const Parameters &params)
    : object_fetcher_(params.object_fetcher)
    , catalog_info_shim_(&catalog_info_default_shim_)
    , default_history_depth_(params.history)
    , default_timestamp_threshold_(params.timestamp)
    , no_close_(params.no_close)
    , ignore_load_failure_(params.ignore_load_failure)
    , no_repeat_history_(params.no_repeat_history)
    , error_sink_((params.quiet) ? kLogDebug : kLogStderr)
  {
    assert(object_fetcher_ != NULL);
  }

  virtual bool Traverse(const TraversalType type = kBreadthFirst) = 0;

  virtual bool Traverse(const shash::Any &root_catalog_hash,
                        const TraversalType type = kBreadthFirst) = 0;

  virtual bool TraverseList(const std::vector<shash::Any> &catalog_list,
    const TraversalType type = kBreadthFirst) = 0;

  virtual bool TraverseRevision(const shash::Any     &root_catalog_hash,
                        const TraversalType type = kBreadthFirst) = 0;

  virtual bool TraverseNamedSnapshots(
    const TraversalType type = kBreadthFirst)
  {
    typedef std::vector<shash::Any> HashList;

    UniquePtr<HistoryTN> tag_db;
    const typename ObjectFetcherT::Failures retval =
                                         object_fetcher_->FetchHistory(&tag_db);
    switch (retval) {
      case ObjectFetcherT::kFailOk:
        break;

      case ObjectFetcherT::kFailNotFound:
        LogCvmfs(kLogCatalogTraversal, kLogDebug,
                 "didn't find a history database to traverse");
        return true;

      default:
        LogCvmfs(kLogCatalogTraversal, kLogStderr,
                 "failed to download history database (%d - %s)",
                 retval, Code2Ascii(retval));
        return false;
    }

    HashList root_hashes;
    bool success = tag_db->GetHashes(&root_hashes);
    assert(success);
    return TraverseList(root_hashes, type);
  }

  void SetCatalogInfoShim(CatalogTraversalInfoShim<CatalogTN> *shim) {
    catalog_info_shim_ = shim;
  }

 protected:
  typedef std::set<shash::Any> HashSet;

  struct CatalogJob {
    CatalogJob(const std::string  &path,
               const shash::Any   &hash,
               const unsigned      tree_level,
               const uint64_t      history_depth,
                     CatalogTN    *parent = NULL) :
      path(path),
      hash(hash),
      tree_level(tree_level),
      history_depth(history_depth),
      parent(parent),
      catalog_file_size(0),
      ignore(false),
      catalog(NULL),
      referenced_catalogs(0),
      postponed(false) {}

    bool IsRootCatalog() const { return tree_level == 0; }

    CallbackDataTN GetCallbackData() const {
      return CallbackDataTN(catalog, hash, tree_level,
                            catalog_file_size, history_depth);
    }

    // initial state description
    const std::string   path;
    const shash::Any    hash;
    const unsigned      tree_level;
    const uint64_t      history_depth;
          CatalogTN    *parent;

    // dynamic processing state (used internally)
    std::string   catalog_file_path;
    size_t        catalog_file_size;
    bool          ignore;
    CatalogTN    *catalog;
    uint64_t      referenced_catalogs;
    bool          postponed;
  };

  bool PrepareCatalog(CatalogJob *job) {
    const typename ObjectFetcherT::Failures retval =
      object_fetcher_->FetchCatalog(job->hash,
                                    job->path,
                                    &job->catalog,
                                    !job->IsRootCatalog(),
                                    job->parent);
    switch (retval) {
      case ObjectFetcherT::kFailOk:
        break;

      case ObjectFetcherT::kFailNotFound:
        if (ignore_load_failure_) {
          LogCvmfs(kLogCatalogTraversal, kLogDebug, "ignoring missing catalog "
                                                    "%s (swept before?)",
                   job->hash.ToString().c_str());
          job->ignore = true;
          return true;
        }

      default:
        LogCvmfs(kLogCatalogTraversal, error_sink_, "failed to load catalog %s "
                                                    "(%d - %s)",
                 job->hash.ToStringWithSuffix().c_str(),
                 retval, Code2Ascii(retval));
        return false;
    }

    // catalogs returned by ObjectFetcher<> are managing their database files by
    // default... we need to manage this file manually here
    job->catalog->DropDatabaseFileOwnership();

    job->catalog_file_path = job->catalog->database_path();
    job->catalog_file_size = GetFileSize(job->catalog->database_path());

    return true;
  }


  bool ReopenCatalog(CatalogJob *job) {
    assert(!job->ignore);
    assert(job->catalog == NULL);

    job->catalog = CatalogTN::AttachFreely(job->path,
                                           job->catalog_file_path,
                                           job->hash,
                                           job->parent,
                                           !job->IsRootCatalog());

    if (job->catalog == NULL) {
      LogCvmfs(kLogCatalogTraversal, error_sink_,
               "failed to re-open catalog %s", job->hash.ToString().c_str());
      return false;
    }

    return true;
  }


  bool CloseCatalog(const bool unlink_db, CatalogJob *job) {
    delete job->catalog;
    job->catalog = NULL;
    if (!job->catalog_file_path.empty() && unlink_db) {
      const int retval = unlink(job->catalog_file_path.c_str());
      if (retval != 0) {
        LogCvmfs(kLogCatalogTraversal, error_sink_, "Failed to unlink %s - %d",
                 job->catalog_file_path.c_str(), errno);
        return false;
      }
    }

    return true;
  }

  shash::Any GetRepositoryRootCatalogHash() {
    // get the manifest of the repository to learn about the entry point or the
    // root catalog of the repository to be traversed
    UniquePtr<manifest::Manifest> manifest;
    const typename ObjectFetcherT::Failures retval =
                                      object_fetcher_->FetchManifest(&manifest);
    if (retval != ObjectFetcherT::kFailOk) {
      LogCvmfs(kLogCatalogTraversal, kLogStderr, "failed to load manifest "
                                                 "(%d - %s)",
                                                 retval, Code2Ascii(retval));
      return shash::Any();
    }

    assert(manifest.IsValid());
    return manifest->catalog_hash();
  }

  bool IsBelowPruningThresholds(
    const CatalogJob &job,
    const uint64_t history_depth,
    const time_t timestamp_threshold
  ) {
    assert(job.IsRootCatalog());
    assert(job.catalog != NULL);

    const bool h = job.history_depth >= history_depth;
    assert(timestamp_threshold >= 0);
    const bool t =
      catalog_info_shim_->GetLastModified(job.catalog) <
      unsigned(timestamp_threshold);

    return t || h;
  }

  ObjectFetcherT         *object_fetcher_;
  CatalogTraversalInfoShim<CatalogTN> catalog_info_default_shim_;
  CatalogTraversalInfoShim<CatalogTN> *catalog_info_shim_;
  const uint64_t          default_history_depth_;
  const time_t            default_timestamp_threshold_;
  const bool              no_close_;
  const bool              ignore_load_failure_;
  const bool              no_repeat_history_;
  LogFacilities           error_sink_;
};

template<class ObjectFetcherT>
class CatalogTraversal
  : public CatalogTraversalBase<ObjectFetcherT>
{
 public:
  typedef CatalogTraversalBase<ObjectFetcherT> Base;
  typedef ObjectFetcherT                      ObjectFetcherTN;
  typedef typename ObjectFetcherT::CatalogTN  CatalogTN;
  typedef typename ObjectFetcherT::HistoryTN  HistoryTN;
  typedef CatalogTraversalData<CatalogTN>     CallbackDataTN;
  typedef typename Base::Parameters           Parameters;
  typedef typename Base::TraversalType        TraversalType;
  typedef typename Base::CatalogJob           CatalogJob;

 protected:
  typedef std::set<shash::Any> HashSet;
  typedef std::stack<CatalogJob> CatalogJobStack;

  struct TraversalContext {
    TraversalContext(const uint64_t       history_depth,
                     const time_t         timestamp_threshold,
                     const TraversalType  traversal_type) :
      history_depth(history_depth),
      timestamp_threshold(timestamp_threshold),
      traversal_type(traversal_type) {}

    const uint64_t       history_depth;
    const time_t         timestamp_threshold;
    const TraversalType  traversal_type;
    CatalogJobStack      catalog_stack;
    CatalogJobStack      callback_stack;
  };

 public:
  explicit CatalogTraversal(const Parameters &params)
    : CatalogTraversalBase<ObjectFetcherT>(params)
  { }

  bool Traverse(const TraversalType type = Base::kBreadthFirst) {
    const shash::Any root_catalog_hash = this->GetRepositoryRootCatalogHash();
    if (root_catalog_hash.IsNull()) {
      return false;
    }
    return Traverse(root_catalog_hash, type);
  }

  bool Traverse(const shash::Any     &root_catalog_hash,
                const TraversalType   type = Base::kBreadthFirst) {
    // add the root catalog of the repository as the first element on the job
    // stack
    TraversalContext ctx(this->default_history_depth_,
                         this->default_timestamp_threshold_,
                         type);
    Push(root_catalog_hash, &ctx);
    return DoTraverse(&ctx);
  }

  bool TraverseList(const std::vector<shash::Any> &catalog_list,
                    const TraversalType type = Base::kBreadthFirst) {
    std::vector<shash::Any>::const_iterator i = catalog_list.begin();
    const std::vector<shash::Any>::const_iterator iend = catalog_list.end();
    bool success = true;
    for (; i != iend; ++i) {
      success = success && TraverseRevision(*i, type);
    }
    return success;
  }

  bool TraverseRevision(
    const shash::Any     &root_catalog_hash,
    const TraversalType type = Base::kBreadthFirst)
  {
    // add the given root catalog as the first element on the job stack
    TraversalContext ctx(Parameters::kNoHistory,
                         Parameters::kNoTimestampThreshold,
                         type);
    Push(root_catalog_hash, &ctx);
    return DoTraverse(&ctx);
  }

 protected:
  bool DoTraverse(TraversalContext *ctx) {
    assert(ctx->callback_stack.empty());

    while (!ctx->catalog_stack.empty()) {
      // Get the top most catalog for the next processing step
      CatalogJob job = Pop(ctx);

      if (ShouldBeSkipped(job)) {
        job.ignore = true;
      // download and open the catalog for processing
      } else if (!this->PrepareCatalog(&job)) {
        return false;
      }

      // ignored catalogs don't need to be processed anymore but they might
      // release postponed yields
      if (job.ignore) {
        if (!HandlePostponedYields(job, ctx)) {
          return false;
        }
        continue;
      }

      // push catalogs referenced by the current catalog (onto stack)
      this->MarkAsVisited(job);
      PushReferencedCatalogs(&job, ctx);

      // notify listeners
      if (!YieldToListeners(&job, ctx)) {
        return false;
      }
    }

    // invariant: after the traversal finished, there should be no more catalogs
    //            to traverse or to yield!
    assert(ctx->catalog_stack.empty());
    assert(ctx->callback_stack.empty());
    return true;
  }

  void PushReferencedCatalogs(CatalogJob *job, TraversalContext *ctx) {
    assert(!job->ignore);
    assert(job->catalog != NULL);
    assert(ctx->traversal_type == Base::kBreadthFirst ||
           ctx->traversal_type == Base::kDepthFirst);

    // this differs, depending on the traversal strategy.
    //
    // Breadths First Traversal
    //   Catalogs are traversed from top (root catalog) to bottom (leaf
    //   catalogs) and from more recent (HEAD revision) to older (historic
    //   revisions)
    //
    // Depth First Traversal
    //   Catalogs are traversed from oldest revision (depends on the configured
    //   maximal history depth) to the HEAD revision and from bottom (leafs) to
    //   top (root catalogs)
    job->referenced_catalogs =
      (ctx->traversal_type == Base::kBreadthFirst)
      ? PushPreviousRevision(*job, ctx) + PushNestedCatalogs(*job, ctx)
      : PushNestedCatalogs(*job, ctx) + PushPreviousRevision(*job, ctx);
  }

  unsigned int PushPreviousRevision(
    const CatalogJob &job,
    TraversalContext *ctx
  ) {
    // only root catalogs are used for entering a previous revision (graph)
    if (!job.catalog->IsRoot()) {
      return 0;
    }

    const shash::Any previous_revision = job.catalog->GetPreviousRevision();
    if (previous_revision.IsNull()) {
      return 0;
    }

    // check if the next deeper history level is actually requested
    // Note: if the current catalog is below the timestamp threshold it will be
    //       traversed and only its ancestor revision will not be pushed anymore
    if (this->IsBelowPruningThresholds(job, ctx->history_depth,
                                       ctx->timestamp_threshold)) {
      return 0;
    }

    Push(CatalogJob("", previous_revision, 0, job.history_depth + 1, NULL),
         ctx);
    return 1;
  }

  unsigned int PushNestedCatalogs(
    const CatalogJob &job,
    TraversalContext *ctx
  ) {
    typedef typename CatalogTN::NestedCatalogList NestedCatalogList;
    const NestedCatalogList nested = job.catalog->ListOwnNestedCatalogs();
    typename NestedCatalogList::const_iterator i    = nested.begin();
    typename NestedCatalogList::const_iterator iend = nested.end();
    for (; i != iend; ++i) {
      CatalogTN* parent = (this->no_close_) ? job.catalog : NULL;
      const CatalogJob new_job(i->mountpoint.ToString(),
                               i->hash,
                               job.tree_level + 1,
                               job.history_depth,
                               parent);
      Push(new_job, ctx);
    }

    return nested.size();
  }

  void Push(const shash::Any &root_catalog_hash, TraversalContext *ctx) {
    Push(CatalogJob("", root_catalog_hash, 0, 0), ctx);
  }

  bool YieldToListeners(CatalogJob *job, TraversalContext *ctx) {
    assert(!job->ignore);
    assert(job->catalog != NULL);
    assert(ctx->traversal_type == Base::kBreadthFirst ||
           ctx->traversal_type == Base::kDepthFirst);

    // in breadth first search mode, every catalog is simply handed out once
    // it is visited. No extra magic required...
    if (ctx->traversal_type == Base::kBreadthFirst) {
      return Yield(job);
    }

    // in depth first search mode, catalogs might need to wait until all of
    // their referenced catalogs are yielded (ctx.callback_stack)...
    assert(ctx->traversal_type == Base::kDepthFirst);
    if (job->referenced_catalogs > 0) {
      PostponeYield(job, ctx);
      return true;
    }

    // this catalog can be yielded
    return Yield(job) && HandlePostponedYields(*job, ctx);
  }

  bool ShouldBeSkipped(const CatalogJob &job) {
    return this->no_repeat_history_ && (visited_catalogs_.count(job.hash) > 0);
  }

  void MarkAsVisited(const CatalogJob &job) {
    if (this->no_repeat_history_) {
      visited_catalogs_.insert(job.hash);
    }
  }

 private:
  bool Yield(CatalogJob *job) {
    assert(!job->ignore);
    assert(job->catalog != NULL || job->postponed);

    // catalog was pushed on ctx.callback_stack before, it might need to be re-
    // opened. If CatalogTraversal<> is configured with no_close, it was not
    // closed before, hence does not need a re-open.
    if (job->postponed && !this->no_close_ && !this->ReopenCatalog(job)) {
      return false;
    }

    // hand the catalog out to the user code (see Observable<>)
    assert(job->catalog != NULL);
    this->NotifyListeners(job->GetCallbackData());

    // skip the catalog closing procedure if asked for
    // Note: In this case it is the user's responsibility to both delete the
    //       yielded catalog object and the underlying database temp file
    if (this->no_close_) {
      return true;
    }

    // we can close the catalog here and delete the temporary file
    const bool unlink_db = true;
    return this->CloseCatalog(unlink_db, job);
  }


  void PostponeYield(CatalogJob *job, TraversalContext *ctx) {
    assert(job->referenced_catalogs > 0);

    job->postponed = true;
    if (!this->no_close_) {
      const bool unlink_db = false;  // will reopened just before yielding
      this->CloseCatalog(unlink_db, job);
    }
    ctx->callback_stack.push(*job);
  }

  bool HandlePostponedYields(const CatalogJob &job, TraversalContext *ctx) {
    if (ctx->traversal_type == Base::kBreadthFirst) {
      return true;
    }

    assert(ctx->traversal_type == Base::kDepthFirst);
    assert(job.referenced_catalogs == 0);

    // walk through the callback_stack and yield all catalogs that have no un-
    // yielded referenced_catalogs anymore. Every time a CatalogJob in the
    // callback_stack gets yielded it decrements the referenced_catalogs of the
    // next top of the stack (it's parent CatalogJob waiting for yielding)
    CatalogJobStack &clbs = ctx->callback_stack;
    while (!clbs.empty()) {
      CatalogJob &postponed_job = clbs.top();
      if (--postponed_job.referenced_catalogs > 0) {
        break;
      }

      if (!Yield(&postponed_job)) {
        return false;
      }
      clbs.pop();
    }

    return true;
  }

  void Push(const CatalogJob &job, TraversalContext *ctx) {
    ctx->catalog_stack.push(job);
  }

  CatalogJob Pop(TraversalContext *ctx) {
    CatalogJob job = ctx->catalog_stack.top();
    ctx->catalog_stack.pop();
    return job;
  }

 protected:
  HashSet                 visited_catalogs_;
};

template <class ObjectFetcherT>
const uint64_t CatalogTraversalBase<ObjectFetcherT>::Parameters::kFullHistory =
    std::numeric_limits<uint64_t>::max();

template <class ObjectFetcherT>
const uint64_t CatalogTraversalBase<ObjectFetcherT>::Parameters::kNoHistory = 0;

template <class ObjectFetcherT>
const time_t
  CatalogTraversalBase<ObjectFetcherT>::Parameters::kNoTimestampThreshold = 0;

}  // namespace swissknife

#endif  // CVMFS_CATALOG_TRAVERSAL_H_