catalog_mgr_impl.h

Go to the documentation of this file.

#ifndef CVMFS_CATALOG_MGR_IMPL_H_
#define CVMFS_CATALOG_MGR_IMPL_H_

#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS
#endif

#include "cvmfs_config.h"

#include <cassert>
#include <string>
#include <vector>

#include "shortstring.h"
#include "statistics.h"
#include "util/logging.h"
#include "xattr.h"

using namespace std;  // NOLINT

namespace catalog {

template <class CatalogT>
AbstractCatalogManager<CatalogT>::AbstractCatalogManager(
    perf::Statistics *statistics) :
  statistics_(statistics) {
  inode_watermark_status_ = 0;
  inode_gauge_ = AbstractCatalogManager<CatalogT>::kInodeOffset;
  revision_cache_ = 0;
  timestamp_cache_ = 0;
  catalog_watermark_ = 0;
  volatile_flag_ = false;
  has_authz_cache_ = false;
  inode_annotation_ = NULL;
  incarnation_ = 0;
  rwlock_ =
    reinterpret_cast<pthread_rwlock_t *>(smalloc(sizeof(pthread_rwlock_t)));
  int retval = pthread_rwlock_init(rwlock_, NULL);
  assert(retval == 0);
  retval = pthread_key_create(&pkey_sqlitemem_, NULL);
  assert(retval == 0);
}

template <class CatalogT>
AbstractCatalogManager<CatalogT>::~AbstractCatalogManager() {
  DetachAll();
  pthread_key_delete(pkey_sqlitemem_);
  pthread_rwlock_destroy(rwlock_);
  free(rwlock_);
}

template <class CatalogT>
void AbstractCatalogManager<CatalogT>::SetInodeAnnotation(
    InodeAnnotation *new_annotation)
{
  assert(catalogs_.empty() || (new_annotation == inode_annotation_));
  inode_annotation_ = new_annotation;
}

template <class CatalogT>
void AbstractCatalogManager<CatalogT>::SetOwnerMaps(const OwnerMap &uid_map,
                                          const OwnerMap &gid_map)
{
  uid_map_ = uid_map;
  gid_map_ = gid_map;
}

template <class CatalogT>
void AbstractCatalogManager<CatalogT>::SetCatalogWatermark(unsigned limit) {
  catalog_watermark_ = limit;
}

template <class CatalogT>
void AbstractCatalogManager<CatalogT>::CheckInodeWatermark() {
  if (inode_watermark_status_ > 0)
    return;

  uint64_t highest_inode = inode_gauge_;
  if (inode_annotation_)
    highest_inode += inode_annotation_->GetGeneration();
  uint64_t uint32_border = 1;
  uint32_border = uint32_border << 32;
  if (highest_inode >= uint32_border) {
    LogCvmfs(kLogCatalog, kLogDebug | kLogSyslogWarn, "inodes exceed 32bit");
    inode_watermark_status_++;
  }
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::Init() {
  LogCvmfs(kLogCatalog, kLogDebug, "Initialize catalog");
  WriteLock();
  bool attached = MountCatalog(PathString("", 0), shash::Any(), NULL);
  Unlock();

  if (!attached) {
    LogCvmfs(kLogCatalog, kLogDebug, "failed to initialize root catalog");
  }

  return attached;
}


template <class CatalogT>
LoadReturn AbstractCatalogManager<CatalogT>::RemountDryrun() {
  LogCvmfs(kLogCatalog, kLogDebug,
           "dryrun remounting repositories");
  CatalogContext ctlg_context;
  return GetNewRootCatalogContext(&ctlg_context);
}

template <class CatalogT>
LoadReturn AbstractCatalogManager<CatalogT>::Remount() {
  LogCvmfs(kLogCatalog, kLogDebug, "remounting repositories");
  CatalogContext ctlg_context;

  if (GetNewRootCatalogContext(&ctlg_context) != kLoadNew
      && GetNewRootCatalogContext(&ctlg_context) != kLoadUp2Date) {
    LogCvmfs(kLogCatalog, kLogDebug, "remounting repositories: "
                                "Did not find any valid root catalog to mount");
    return kLoadFail;
  }

  WriteLock();

  const LoadReturn load_error = LoadCatalogByHash(&ctlg_context);

  if (load_error == kLoadNew) {
    inode_t old_inode_gauge = inode_gauge_;
    DetachAll();
    inode_gauge_ = AbstractCatalogManager<CatalogT>::kInodeOffset;

    CatalogT *new_root = CreateCatalog(ctlg_context.mountpoint(),
                                       ctlg_context.hash(), NULL);
    assert(new_root);
    bool retval = AttachCatalog(ctlg_context.sqlite_path(), new_root);
    assert(retval);

    if (inode_annotation_) {
      inode_annotation_->IncGeneration(old_inode_gauge);
    }
  }
  CheckInodeWatermark();
  Unlock();

  return load_error;
}

template <class CatalogT>
LoadReturn AbstractCatalogManager<CatalogT>::ChangeRoot(
  const shash::Any &root_hash)
{
  assert(!root_hash.IsNull());
  LogCvmfs(kLogCatalog, kLogDebug,
           "switching to root hash %s", root_hash.ToString().c_str());

  WriteLock();

  CatalogContext ctlg_context(root_hash, PathString("", 0),
                                                         kCtlgNoLocationNeeded);
  // we do not need to set revision as LoadCatalogByHash
  // needs only mountpoint, hash

  const LoadReturn load_error = LoadCatalogByHash(&ctlg_context);

  if (load_error == kLoadNew) {
    inode_t old_inode_gauge = inode_gauge_;
    DetachAll();
    inode_gauge_ = AbstractCatalogManager<CatalogT>::kInodeOffset;

    CatalogT *new_root =
                    CreateCatalog(PathString("", 0), ctlg_context.hash(), NULL);
    assert(new_root);
    bool retval = AttachCatalog(ctlg_context.sqlite_path(), new_root);
    assert(retval);

    if (inode_annotation_) {
      inode_annotation_->IncGeneration(old_inode_gauge);
    }
  }
  CheckInodeWatermark();
  Unlock();

  return load_error;
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::DetachNested() {
  WriteLock();
  if (catalogs_.empty()) {
    Unlock();
    return;
  }

  typename CatalogList::const_iterator i;
  typename CatalogList::const_iterator iend;
  CatalogList catalogs_to_detach = GetRootCatalog()->GetChildren();
  for (i = catalogs_to_detach.begin(), iend = catalogs_to_detach.end();
       i != iend; ++i)
  {
    DetachSubtree(*i);
  }

  Unlock();
}


template <class CatalogT>
shash::Any AbstractCatalogManager<CatalogT>::GetNestedCatalogHash(
  const PathString &mountpoint)
{
  assert(!mountpoint.IsEmpty());
  CatalogT *catalog = FindCatalog(mountpoint);
  assert(catalog != NULL);
  if (catalog->mountpoint() == mountpoint) {
    catalog = catalog->parent();
    assert(catalog != NULL);
  }
  shash::Any result;
  uint64_t size;
  catalog->FindNested(mountpoint, &result, &size);
  return result;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::LookupPath(const PathString &path,
                                        const LookupOptions options,
                                        DirectoryEntry *dirent)
{
  // initialize as non-negative
  assert(dirent);
  *dirent = DirectoryEntry();

  // create a dummy negative directory entry
  const DirectoryEntry dirent_negative =
    DirectoryEntry(catalog::kDirentNegative);

  EnforceSqliteMemLimit();
  ReadLock();

  CatalogT *best_fit = FindCatalog(path);
  assert(best_fit != NULL);

  perf::Inc(statistics_.n_lookup_path);
  LogCvmfs(kLogCatalog, kLogDebug, "looking up '%s' in catalog: '%s'",
           path.c_str(), best_fit->mountpoint().c_str());
  bool found = best_fit->LookupPath(path, dirent);

  // Possibly in a nested catalog
  if (!found && MountSubtree(path, best_fit, false /* is_listable */, NULL)) {
    LogCvmfs(kLogCatalog, kLogDebug, "looking up '%s' in a nested catalog",
             path.c_str());
    StageNestedCatalogAndUnlock(path, best_fit, false /* is_listable */);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(path);
    assert(best_fit != NULL);
    perf::Inc(statistics_.n_lookup_path);
    found = best_fit->LookupPath(path, dirent);

    if (!found) {
      LogCvmfs(kLogCatalog, kLogDebug,
               "entry not found, we may have to load nested catalogs");

      CatalogT *nested_catalog;
      found =
        MountSubtree(path, best_fit, false /* is_listable */, &nested_catalog);

      if (!found) {
        LogCvmfs(kLogCatalog, kLogDebug,
                 "failed to load nested catalog for '%s'", path.c_str());
        goto lookup_path_notfound;
      }

      if (nested_catalog != best_fit) {
        perf::Inc(statistics_.n_lookup_path);
        found = nested_catalog->LookupPath(path, dirent);
        if (!found) {
          LogCvmfs(kLogCatalog, kLogDebug,
                   "nested catalogs loaded but entry '%s' was still not found",
                   path.c_str());
          if (dirent != NULL) *dirent = dirent_negative;
          goto lookup_path_notfound;
        } else {
          best_fit = nested_catalog;
        }
      } else {
        LogCvmfs(kLogCatalog, kLogDebug, "no nested catalog fits");
        if (dirent != NULL) *dirent = dirent_negative;
        goto lookup_path_notfound;
      }
    }
    assert(found);
  }
  // Not in a nested catalog (because no nested cataog fits), ENOENT
  if (!found) {
    LogCvmfs(kLogCatalog, kLogDebug, "ENOENT: '%s'", path.c_str());
    if (dirent != NULL) *dirent = dirent_negative;
    goto lookup_path_notfound;
  }

  LogCvmfs(kLogCatalog, kLogDebug, "found entry '%s' in catalog '%s'",
           path.c_str(), best_fit->mountpoint().c_str());

  if ((options & kLookupRawSymlink) == kLookupRawSymlink) {
    LinkString raw_symlink;
    bool retval = best_fit->LookupRawSymlink(path, &raw_symlink);
    assert(retval);  // Must be true, we have just found the entry
    dirent->set_symlink(raw_symlink);
  }

  Unlock();
  return true;

 lookup_path_notfound:
  Unlock();
  // Includes both: ENOENT and not found due to I/O error
  perf::Inc(statistics_.n_lookup_path_negative);
  return false;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::LookupNested(
  const PathString &path,
  PathString *mountpoint,
  shash::Any *hash,
  uint64_t *size
) {
  EnforceSqliteMemLimit();
  bool result = false;
  ReadLock();

  // Look past current path to mount up to intended location
  PathString catalog_path(path);
  catalog_path.Append("/.cvmfscatalog", 14);

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(catalog_path);
  CatalogT *catalog = best_fit;
  if (MountSubtree(catalog_path, best_fit, false /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, false);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(catalog_path);
    result =
      MountSubtree(catalog_path, best_fit, false /* is_listable */, &catalog);
    // Result is false if an available catalog failed to load (error happened)
    if (!result) {
      Unlock();
      return false;
    }
  }

  // If the found catalog is the Root there is no parent to lookup
  if (catalog->HasParent()) {
    result = catalog->parent()->FindNested(catalog->root_prefix(), hash, size);
  }

  // Mountpoint now points to the found catalog
  mountpoint->Assign(catalog->root_prefix());

  // If the result is false, it means that no nested catalog was found for
  // this path. As the root catalog does not have a Nested Catalog of
  // itself, we manually set the values and leave the size as 0.
  // TODO(nhazekam) Allow for Root Catalog to be returned
  if (!result) {
    *hash =  GetRootCatalog()->hash();
    *size = 0;
    result = true;
  }

  Unlock();
  return result;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::ListCatalogSkein(
  const PathString &path,
  std::vector<PathString> *result_list
) {
  EnforceSqliteMemLimit();
  bool result;
  ReadLock();

  // Look past current path to mount up to intended location
  PathString test(path);
  test.Append("/.cvmfscatalog", 14);

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(test);
  CatalogT *catalog = best_fit;
  // True if there is an available nested catalog
  if (MountSubtree(test, best_fit, false /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, false);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(test);
    result = MountSubtree(test, best_fit, false /* is_listable */, &catalog);
    // result is false if an available catalog failed to load
    if (!result) {
      Unlock();
      return false;
    }
  }

  // Build listing
  CatalogT *cur_parent = catalog->parent();
  if (cur_parent) {
    // Walk up parent tree to find base
    std::vector<catalog::Catalog*> parents;
    while (cur_parent->HasParent()) {
      parents.push_back(cur_parent);
      cur_parent = cur_parent->parent();
    }
    parents.push_back(cur_parent);
    while (!parents.empty()) {
      // Add to list in order starting at root
      result_list->push_back(parents.back()->root_prefix());
      parents.pop_back();
    }
  }
  // Add the current catalog
  result_list->push_back(catalog->root_prefix());

  Catalog::NestedCatalogList children = catalog->ListOwnNestedCatalogs();

  // Add all children nested catalogs
  for (unsigned i = 0; i < children.size(); i++) {
    result_list->push_back(children.at(i).mountpoint);
  }

  Unlock();
  return true;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::LookupXattrs(
  const PathString &path,
  XattrList *xattrs)
{
  EnforceSqliteMemLimit();
  bool result;
  ReadLock();

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(path);
  CatalogT *catalog = best_fit;
  if (MountSubtree(path, best_fit, false /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, false);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(path);
    result = MountSubtree(path, best_fit, false /* is_listable */, &catalog);
    if (!result) {
      Unlock();
      return false;
    }
  }

  perf::Inc(statistics_.n_lookup_xattrs);
  result = catalog->LookupXattrsPath(path, xattrs);

  Unlock();
  return result;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::Listing(const PathString &path,
                                     DirectoryEntryList *listing,
                                     const bool expand_symlink)
{
  EnforceSqliteMemLimit();
  bool result;
  ReadLock();

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(path);
  CatalogT *catalog = best_fit;
  if (MountSubtree(path, best_fit, true /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, true /* is_listable */);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(path);
    result = MountSubtree(path, best_fit, true /* is_listable */, &catalog);
    if (!result) {
      Unlock();
      return false;
    }
  }

  perf::Inc(statistics_.n_listing);
  result = catalog->ListingPath(path, listing, expand_symlink);

  Unlock();
  return result;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::ListingStat(const PathString &path,
                                        StatEntryList *listing)
{
  EnforceSqliteMemLimit();
  bool result;
  ReadLock();

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(path);
  CatalogT *catalog = best_fit;
  if (MountSubtree(path, best_fit, true /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, true /* is_listable */);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(path);
    result = MountSubtree(path, best_fit, true /* is_listable */, &catalog);
    if (!result) {
      Unlock();
      return false;
    }
  }

  perf::Inc(statistics_.n_listing);
  result = catalog->ListingPathStat(path, listing);

  Unlock();
  return result;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::ListFileChunks(
  const PathString &path,
  const shash::Algorithms interpret_hashes_as,
  FileChunkList *chunks)
{
  EnforceSqliteMemLimit();
  bool result;
  ReadLock();

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(path);
  CatalogT *catalog = best_fit;
  if (MountSubtree(path, best_fit, false /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, false);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(path);
    result = MountSubtree(path, best_fit, false /* is_listable */, &catalog);
    if (!result) {
      Unlock();
      return false;
    }
  }

  result = catalog->ListPathChunks(path, interpret_hashes_as, chunks);

  Unlock();
  return result;
}

template <class CatalogT>
catalog::Counters AbstractCatalogManager<CatalogT>::LookupCounters(
  const PathString &path,
  std::string *subcatalog_path,
  shash::Any *hash
  )
{
  EnforceSqliteMemLimit();
  bool result;
  ReadLock();

  // Look past current path to mount up to intended location
  PathString catalog_path(path);
  catalog_path.Append("/.cvmfscatalog", 14);

  // Find catalog, possibly load nested
  CatalogT *best_fit = FindCatalog(catalog_path);
  CatalogT *catalog = best_fit;
  if (MountSubtree(catalog_path, best_fit, false /* is_listable */, NULL)) {
    StageNestedCatalogAndUnlock(path, best_fit, false /* is_listable */);
    WriteLock();
    // Check again to avoid race
    best_fit = FindCatalog(catalog_path);
    result =
      MountSubtree(catalog_path, best_fit, false /* is_listable */, &catalog);
    // Result is false if an available catalog failed to load (error happened)
    if (!result) {
      Unlock();
      *subcatalog_path = "error: failed to load catalog!";
      *hash = shash::Any();
      return catalog::Counters();
    }
  }

  *hash = catalog->hash();
  *subcatalog_path = catalog->mountpoint().ToString();
  catalog::Counters counters = catalog->GetCounters();
  Unlock();
  return counters;
}


template <class CatalogT>
uint64_t AbstractCatalogManager<CatalogT>::GetRevision() const {
  ReadLock();
  const uint64_t revision = GetRevisionNoLock();
  Unlock();

  return revision;
}

template <class CatalogT>
uint64_t AbstractCatalogManager<CatalogT>::GetRevisionNoLock() const {
  return revision_cache_;
}

template <class CatalogT>
uint64_t AbstractCatalogManager<CatalogT>::GetTimestamp() const {
  ReadLock();
  const uint64_t timestamp = GetTimestampNoLock();
  Unlock();

  return timestamp;
}

template <class CatalogT>
uint64_t AbstractCatalogManager<CatalogT>::GetTimestampNoLock() const {
  return timestamp_cache_;
}

template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::GetVOMSAuthz(std::string *authz) const {
  ReadLock();
  const bool has_authz = has_authz_cache_;
  if (has_authz && authz)
    *authz = authz_cache_;
  Unlock();
  return has_authz;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::HasExplicitTTL() const {
  ReadLock();
  const bool result = GetRootCatalog()->HasExplicitTTL();
  Unlock();
  return result;
}


template <class CatalogT>
uint64_t AbstractCatalogManager<CatalogT>::GetTTL() const {
  ReadLock();
  const uint64_t ttl = GetRootCatalog()->GetTTL();
  Unlock();
  return ttl;
}


template <class CatalogT>
int AbstractCatalogManager<CatalogT>::GetNumCatalogs() const {
  ReadLock();
  int result = catalogs_.size();
  Unlock();
  return result;
}


template <class CatalogT>
string AbstractCatalogManager<CatalogT>::PrintHierarchy() const {
  ReadLock();
  string output = PrintHierarchyRecursively(GetRootCatalog(), 0);
  Unlock();
  return output;
}


template <class CatalogT>
InodeRange AbstractCatalogManager<CatalogT>::AcquireInodes(uint64_t size) {
  InodeRange result;
  result.offset = inode_gauge_;
  result.size = size;

  inode_gauge_ += size;
  LogCvmfs(kLogCatalog, kLogDebug, "allocating inodes from %lu to %lu.",
           result.offset + 1, inode_gauge_);

  return result;
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::ReleaseInodes(const InodeRange chunk) {
  // TODO(jblomer) currently inodes are only released on remount
}


template <class CatalogT>
CatalogT* AbstractCatalogManager<CatalogT>::FindCatalog(
    const PathString &path) const {
  assert(catalogs_.size() > 0);

  // Start at the root catalog and successively go down the catalog tree
  CatalogT *best_fit = GetRootCatalog();
  CatalogT *next_fit = NULL;
  while (best_fit->mountpoint() != path) {
    next_fit = best_fit->FindSubtree(path);
    if (next_fit == NULL)
      break;
    best_fit = next_fit;
  }

  return best_fit;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::IsAttached(const PathString &root_path,
                                        CatalogT **attached_catalog) const
{
  if (catalogs_.size() == 0)
    return false;

  CatalogT *best_fit = FindCatalog(root_path);
  if (best_fit->mountpoint() != root_path)
    return false;

  if (attached_catalog != NULL) *attached_catalog = best_fit;
  return true;
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::StageNestedCatalogAndUnlock(
  const PathString &path,
  const CatalogT *parent,
  bool is_listable)
{
  assert(parent);
  const unsigned path_len = path.GetLength();

  perf::Inc(statistics_.n_nested_listing);
  typedef typename CatalogT::NestedCatalogList NestedCatalogList;
  const NestedCatalogList& nested_catalogs = parent->ListNestedCatalogs();

  for (typename NestedCatalogList::const_iterator i = nested_catalogs.begin(),
       iEnd = nested_catalogs.end(); i != iEnd; ++i)
  {
    if (!path.StartsWith(i->mountpoint))
      continue;

    // in this case the path doesn't start with
    // the mountpoint in a file path sense
    // (e.g. path is /a/bc and mountpoint is /a/b), and will be ignored
    const unsigned mountpoint_len = i->mountpoint.GetLength();
    if (path_len > mountpoint_len && path.GetChars()[mountpoint_len] != '/')
      continue;

    // Found a nested catalog transition point
    if (!is_listable && (path_len == mountpoint_len))
      break;

    Unlock();
    LogCvmfs(kLogCatalog, kLogDebug, "staging nested catalog at %s (%s)",
             i->mountpoint.c_str(), i->hash.ToString().c_str());
    StageNestedCatalogByHash(i->hash, i->mountpoint);
    return;
  }
  Unlock();
}

template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::MountSubtree(
  const PathString &path,
  const CatalogT *entry_point,
  bool is_listable,
  CatalogT **leaf_catalog)
{
  bool result = true;
  CatalogT *parent = (entry_point == NULL) ?
                     GetRootCatalog() : const_cast<CatalogT *>(entry_point);
  assert(path.StartsWith(parent->mountpoint()));

  unsigned path_len = path.GetLength();

  // Try to find path as a super string of nested catalog mount points
  perf::Inc(statistics_.n_nested_listing);
  typedef typename CatalogT::NestedCatalogList NestedCatalogList;
  const NestedCatalogList& nested_catalogs =
    parent->ListNestedCatalogs();
  for (typename NestedCatalogList::const_iterator i = nested_catalogs.begin(),
       iEnd = nested_catalogs.end(); i != iEnd; ++i)
  {
    // Next nesting level
    if (path.StartsWith(i->mountpoint)) {
      // in this case the path doesn't start with
      // the mountpoint in a file path sense
      // (e.g. path is /a/bc and mountpoint is /a/b), and will be ignored
      unsigned mountpoint_len = i->mountpoint.GetLength();
      if (path_len > mountpoint_len && path.GetChars()[mountpoint_len] != '/')
        continue;

      // Found a nested catalog transition point
      if (!is_listable && (path_len == mountpoint_len))
        break;

      if (leaf_catalog == NULL)
        return true;
      CatalogT *new_nested;
      LogCvmfs(kLogCatalog, kLogDebug, "load nested catalog at %s",
               i->mountpoint.c_str());
      // prevent endless recursion with corrupted catalogs
      // (due to reloading root)
      if (i->hash.IsNull())
        return false;
      new_nested = MountCatalog(i->mountpoint, i->hash, parent);
      if (!new_nested)
        return false;

      result = MountSubtree(path, new_nested, is_listable, &parent);
      break;
    }
  }

  if (leaf_catalog == NULL)
    return false;
  *leaf_catalog = parent;
  return result;
}


template <class CatalogT>
CatalogT *AbstractCatalogManager<CatalogT>::MountCatalog(
                                              const PathString &mountpoint,
                                              const shash::Any &hash,
                                              CatalogT *parent_catalog)
{
  CatalogT *attached_catalog = NULL;
  if (IsAttached(mountpoint, &attached_catalog)) {
    return attached_catalog;
  }

  CatalogContext ctlg_context(hash, mountpoint, kCtlgLocationMounted);

  if (ctlg_context.IsRootCatalog() && hash.IsNull()) {
    if (GetNewRootCatalogContext(&ctlg_context) == kLoadFail) {
      LogCvmfs(kLogCatalog, kLogDebug,
                                   "failed to retrieve valid root catalog '%s'",
                                   mountpoint.c_str());
      return NULL;
    }
  }

  const LoadReturn retval = LoadCatalogByHash(&ctlg_context);
  if ((retval == kLoadFail) || (retval == kLoadNoSpace)) {
    LogCvmfs(kLogCatalog, kLogDebug, "failed to load catalog '%s' (%d - %s)",
             mountpoint.c_str(), retval, Code2Ascii(retval));
    return NULL;
  }

  attached_catalog = CreateCatalog(ctlg_context.mountpoint(),
                                   ctlg_context.hash(),
                                   parent_catalog);

  // Attach loaded catalog
  if (!AttachCatalog(ctlg_context.sqlite_path(), attached_catalog)) {
    LogCvmfs(kLogCatalog, kLogDebug, "failed to attach catalog '%s'",
             mountpoint.c_str());
    UnloadCatalog(attached_catalog);
    return NULL;
  }

  if ((catalog_watermark_ > 0) && (catalogs_.size() >= catalog_watermark_)) {
    DetachSiblings(mountpoint);
  }

  return attached_catalog;
}


template <class CatalogT>
CatalogT *AbstractCatalogManager<CatalogT>::LoadFreeCatalog(
                                            const PathString     &mountpoint,
                                            const shash::Any     &hash)
{
  assert(!hash.IsNull());
  CatalogContext ctlg_context(hash, mountpoint, kCtlgNoLocationNeeded);

  const LoadReturn load_ret = LoadCatalogByHash(&ctlg_context);

  if (load_ret != kLoadNew) {
    return NULL;
  }

  CatalogT *catalog = CatalogT::AttachFreely(mountpoint.ToString(),
                                             ctlg_context.sqlite_path(),
                                             ctlg_context.hash());
  catalog->TakeDatabaseFileOwnership();
  return catalog;
}


template <class CatalogT>
bool AbstractCatalogManager<CatalogT>::AttachCatalog(const string &db_path,
                                           CatalogT *new_catalog)
{
  LogCvmfs(kLogCatalog, kLogDebug, "attaching catalog file %s",
           db_path.c_str());

  // Initialize the new catalog
  if (!new_catalog->OpenDatabase(db_path)) {
    LogCvmfs(kLogCatalog, kLogDebug, "initialization of catalog %s failed",
             db_path.c_str());
    return false;
  }

  // Determine the inode offset of this catalog
  uint64_t inode_chunk_size = new_catalog->max_row_id();
  InodeRange range = AcquireInodes(inode_chunk_size);
  new_catalog->set_inode_range(range);
  new_catalog->SetInodeAnnotation(inode_annotation_);
  new_catalog->SetOwnerMaps(&uid_map_, &gid_map_);

  // Add catalog to the manager
  if (!new_catalog->IsInitialized()) {
    LogCvmfs(kLogCatalog, kLogDebug,
             "catalog initialization failed (obscure data)");
    inode_gauge_ -= inode_chunk_size;
    return false;
  }
  CheckInodeWatermark();

  // The revision of the catalog tree is given by the root catalog revision
  if (catalogs_.empty()) {
    revision_cache_ = new_catalog->GetRevision();
    timestamp_cache_ = new_catalog->GetLastModified();
    statistics_.catalog_revision->Set(revision_cache_);
    has_authz_cache_ = new_catalog->GetVOMSAuthz(&authz_cache_);
    volatile_flag_ = new_catalog->volatile_flag();
  }

  catalogs_.push_back(new_catalog);
  ActivateCatalog(new_catalog);
  return true;
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::DetachCatalog(CatalogT *catalog) {
  if (catalog->HasParent())
    catalog->parent()->RemoveChild(catalog);

  ReleaseInodes(catalog->inode_range());
  UnloadCatalog(catalog);

  // Delete catalog from internal lists
  typename CatalogList::iterator i;
  typename CatalogList::const_iterator iend;
  for (i = catalogs_.begin(), iend = catalogs_.end(); i != iend; ++i) {
    if (*i == catalog) {
      catalogs_.erase(i);
      delete catalog;
      return;
    }
  }

  assert(false);
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::DetachSubtree(CatalogT *catalog) {
  // Detach all child catalogs recursively
  typename CatalogList::const_iterator i;
  typename CatalogList::const_iterator iend;
  CatalogList catalogs_to_detach = catalog->GetChildren();
  for (i = catalogs_to_detach.begin(), iend = catalogs_to_detach.end();
       i != iend; ++i)
  {
    DetachSubtree(*i);
  }

  DetachCatalog(catalog);
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::DetachSiblings(
  const PathString &current_tree)
{
  bool again;
  do {
    again = false;
    unsigned N = catalogs_.size();
    for (unsigned i = 0; i < N; ++i) {
      if (!HasPrefix(current_tree.ToString(),
                     catalogs_[i]->mountpoint().ToString(),
                     false /* ignore_case */))
      {
        DetachSubtree(catalogs_[i]);
        again = true;
        break;
      }
    }
  } while (again);
  perf::Inc(statistics_.n_detach_siblings);
}


template <class CatalogT>
string AbstractCatalogManager<CatalogT>::PrintHierarchyRecursively(
                                                      const CatalogT *catalog,
                                                      const int level) const
{
  string output;

  // Indent according to level
  for (int i = 0; i < level; ++i)
    output += "    ";

  output += "-> " + string(catalog->mountpoint().GetChars(),
                           catalog->mountpoint().GetLength())
            + "\n";

  CatalogList children = catalog->GetChildren();
  typename CatalogList::const_iterator i = children.begin();
  typename CatalogList::const_iterator iend = children.end();
  for (; i != iend; ++i) {
    output += PrintHierarchyRecursively(*i, level + 1);
  }

  return output;
}


template <class CatalogT>
std::string AbstractCatalogManager<CatalogT>::PrintMemStatsRecursively(
  const CatalogT *catalog) const
{
  string result = catalog->PrintMemStatistics() + "\n";

  CatalogList children = catalog->GetChildren();
  typename CatalogList::const_iterator i = children.begin();
  typename CatalogList::const_iterator iend = children.end();
  for (; i != iend; ++i) {
    result += PrintMemStatsRecursively(*i);
  }
  return result;
}


template <class CatalogT>
std::string AbstractCatalogManager<CatalogT>::PrintAllMemStatistics() const {
  string result;
  ReadLock();
  result = PrintMemStatsRecursively(GetRootCatalog());
  Unlock();
  return result;
}


template <class CatalogT>
void AbstractCatalogManager<CatalogT>::EnforceSqliteMemLimit() {
  char *mem_enforced =
    static_cast<char *>(pthread_getspecific(pkey_sqlitemem_));
  if (mem_enforced == NULL) {
    sqlite3_soft_heap_limit(kSqliteMemPerThread);
    pthread_setspecific(pkey_sqlitemem_, this);
  }
}

}  // namespace catalog




#endif  // CVMFS_CATALOG_MGR_IMPL_H_