catalog.cc

Go to the documentation of this file.

#include "catalog.h"

#include <alloca.h>
#include <errno.h>

#include <algorithm>
#include <cassert>

#include "catalog_mgr.h"
#include "util/concurrency.h"
#include "util/logging.h"
#include "util/platform.h"
#include "util/smalloc.h"

using namespace std;  // NOLINT

namespace catalog {

const shash::Md5 Catalog::kMd5PathEmpty("", 0);


Catalog* Catalog::AttachFreely(const string     &imaginary_mountpoint,
                               const string     &file,
                               const shash::Any &catalog_hash,
                                     Catalog    *parent,
                               const bool        is_nested) {
  Catalog *catalog =
    new Catalog(PathString(imaginary_mountpoint.data(),
                           imaginary_mountpoint.length()),
                catalog_hash,
                parent,
                is_nested);
  const bool successful_init = catalog->InitStandalone(file);
  if (!successful_init) {
    delete catalog;
    return NULL;
  }
  return catalog;
}


Catalog::Catalog(const PathString &mountpoint,
                 const shash::Any &catalog_hash,
                       Catalog    *parent,
                 const bool        is_nested) :
  catalog_hash_(catalog_hash),
  mountpoint_(mountpoint),
  is_regular_mountpoint_(mountpoint_ == root_prefix_),
  volatile_flag_(false),
  is_root_(parent == NULL && !is_nested),
  managed_database_(false),
  parent_(parent),
  nested_catalog_cache_dirty_(true),
  voms_authz_status_(kVomsUnknown),
  initialized_(false)
{
  max_row_id_ = 0;
  inode_annotation_ = NULL;
  lock_ = reinterpret_cast<pthread_mutex_t *>(smalloc(sizeof(pthread_mutex_t)));
  int retval = pthread_mutex_init(lock_, NULL);
  assert(retval == 0);

  database_ = NULL;
  uid_map_ = NULL;
  gid_map_ = NULL;
  sql_listing_ = NULL;
  sql_lookup_md5path_ = NULL;
  sql_lookup_nested_ = NULL;
  sql_list_nested_ = NULL;
  sql_own_list_nested_ = NULL;
  sql_all_chunks_ = NULL;
  sql_chunks_listing_ = NULL;
  sql_lookup_xattrs_ = NULL;
}


Catalog::~Catalog() {
  pthread_mutex_destroy(lock_);
  free(lock_);
  FinalizePreparedStatements();
  delete database_;
}


void Catalog::InitPreparedStatements() {
  sql_listing_          = new SqlListing(database());
  sql_lookup_md5path_   = new SqlLookupPathHash(database());
  sql_lookup_nested_    = new SqlNestedCatalogLookup(database());
  sql_list_nested_      = new SqlNestedCatalogListing(database());
  sql_own_list_nested_  = new SqlOwnNestedCatalogListing(database());
  sql_all_chunks_       = new SqlAllChunks(database());
  sql_chunks_listing_   = new SqlChunksListing(database());
  sql_lookup_xattrs_    = new SqlLookupXattrs(database());
}


void Catalog::FinalizePreparedStatements() {
  delete sql_lookup_xattrs_;
  delete sql_chunks_listing_;
  delete sql_all_chunks_;
  delete sql_listing_;
  delete sql_lookup_md5path_;
  delete sql_lookup_nested_;
  delete sql_list_nested_;
  delete sql_own_list_nested_;
}


bool Catalog::InitStandalone(const std::string &database_file) {
  bool retval = OpenDatabase(database_file);
  if (!retval) {
    return false;
  }

  InodeRange inode_range;
  inode_range.MakeDummy();
  set_inode_range(inode_range);
  return true;
}


bool Catalog::ReadCatalogCounters() {
  assert(database_ != NULL);
  bool statistics_loaded;
  if (database().schema_version() <
      CatalogDatabase::kLatestSupportedSchema - CatalogDatabase::kSchemaEpsilon)
  {
    statistics_loaded =
      counters_.ReadFromDatabase(database(), LegacyMode::kLegacy);
  } else if (database().schema_revision() < 2) {
    statistics_loaded =
      counters_.ReadFromDatabase(database(), LegacyMode::kNoXattrs);
  } else if (database().schema_revision() < 3) {
    statistics_loaded =
      counters_.ReadFromDatabase(database(), LegacyMode::kNoExternals);
  } else if (database().schema_revision() < 5) {
    statistics_loaded =
      counters_.ReadFromDatabase(database(), LegacyMode::kNoSpecials);
  } else {
    statistics_loaded = counters_.ReadFromDatabase(database());
  }
  return statistics_loaded;
}


bool Catalog::OpenDatabase(const string &db_path) {
  database_ = CatalogDatabase::Open(db_path, DatabaseOpenMode());
  if (NULL == database_) {
    return false;
  }

  if (database_->IsEqualSchema(database_->schema_version(), 1.0)) {
    // Possible fix-up for database layout lacking the content hash of
    // nested catalogs
    SqlCatalog sql_has_nested_sha1(database(),
      "SELECT count(*) FROM sqlite_master "
      "WHERE type='table' AND name='nested_catalogs' AND sql LIKE '%sha1%';");
    bool retval = sql_has_nested_sha1.FetchRow();
    assert(retval == true);
    bool has_nested_sha1 = sql_has_nested_sha1.RetrieveInt64(0);
    if (!has_nested_sha1) {
      database_->EnforceSchema(0.9, 0);
    }
  }

  InitPreparedStatements();

  // Set the database file ownership if requested
  if (managed_database_) {
    database_->TakeFileOwnership();
  }

  // Find out the maximum row id of this database file
  SqlCatalog sql_max_row_id(database(), "SELECT MAX(rowid) FROM catalog;");
  if (!sql_max_row_id.FetchRow()) {
    LogCvmfs(kLogCatalog, kLogDebug,
             "Cannot retrieve maximal row id for database file %s "
             "(SqliteErrorcode: %d)",
             db_path.c_str(), sql_max_row_id.GetLastError());
    return false;
  }
  max_row_id_ = sql_max_row_id.RetrieveInt64(0);

  // Get root prefix
  if (database_->HasProperty("root_prefix")) {
    const std::string root_prefix =
                           database_->GetProperty<std::string>("root_prefix");
    root_prefix_.Assign(root_prefix.data(), root_prefix.size());
    LogCvmfs(kLogCatalog, kLogDebug,
             "found root prefix %s in root catalog file %s",
             root_prefix_.c_str(), db_path.c_str());
    is_regular_mountpoint_ = (root_prefix_ == mountpoint_);
  } else {
    LogCvmfs(kLogCatalog, kLogDebug,
             "no root prefix for root catalog file %s", db_path.c_str());
  }

  // Get volatile content flag
  volatile_flag_ = database_->GetPropertyDefault<bool>("volatile",
                                                       volatile_flag_);

  // Read Catalog Counter Statistics
  if (!ReadCatalogCounters()) {
    LogCvmfs(kLogCatalog, kLogStderr,
             "failed to load statistics counters for catalog %s (file %s)",
             mountpoint_.c_str(), db_path.c_str());
    return false;
  }

  if (HasParent()) {
    parent_->AddChild(this);
  }

  initialized_ = true;
  return true;
}


shash::Md5 Catalog::NormalizePath(const PathString &path) const {
  if (is_regular_mountpoint_)
    return shash::Md5(path.GetChars(), path.GetLength());

  assert(path.GetLength() >= mountpoint_.GetLength());
  // Piecewise hash calculation: root_prefix plus tail of path
  shash::Any result(shash::kMd5);
  shash::ContextPtr ctx(shash::kMd5);
  ctx.buffer = alloca(ctx.size);
  shash::Init(ctx);
  shash::Update(
    reinterpret_cast<const unsigned char *>(root_prefix_.GetChars()),
    root_prefix_.GetLength(),
    ctx);
  shash::Update(
    reinterpret_cast<const unsigned char *>(path.GetChars()) +
      mountpoint_.GetLength(),
    path.GetLength() - mountpoint_.GetLength(),
    ctx);
  shash::Final(ctx, &result);
  return result.CastToMd5();
}


PathString Catalog::NormalizePath2(const PathString &path) const {
  if (is_regular_mountpoint_)
    return path;

  assert(path.GetLength() >= mountpoint_.GetLength());
  PathString result = root_prefix_;
  PathString suffix = path.Suffix(mountpoint_.GetLength());
  result.Append(suffix.GetChars(), suffix.GetLength());
  return result;
}


PathString Catalog::PlantPath(const PathString &path) const {
  if (is_regular_mountpoint_)
    return path;

  assert(path.GetLength() >= root_prefix_.GetLength());
  PathString result = mountpoint_;
  PathString suffix = path.Suffix(root_prefix_.GetLength());
  result.Append(suffix.GetChars(), suffix.GetLength());
  return result;
}


bool Catalog::LookupEntry(const shash::Md5 &md5path, const bool expand_symlink,
                          DirectoryEntry *dirent) const
{
  assert(IsInitialized());

  MutexLockGuard m(lock_);
  sql_lookup_md5path_->BindPathHash(md5path);
  bool found = sql_lookup_md5path_->FetchRow();
  if (found && (dirent != NULL)) {
    *dirent = sql_lookup_md5path_->GetDirent(this, expand_symlink);
    FixTransitionPoint(md5path, dirent);
  }
  sql_lookup_md5path_->Reset();

  return found;
}


bool Catalog::LookupMd5Path(const shash::Md5 &md5path,
                            DirectoryEntry *dirent) const
{
  return LookupEntry(md5path, true, dirent);
}


bool Catalog::LookupRawSymlink(const PathString &path,
                               LinkString *raw_symlink) const
{
  DirectoryEntry dirent;
  bool result = (LookupEntry(NormalizePath(path), false, &dirent));
  if (result)
    raw_symlink->Assign(dirent.symlink());
  return result;
}


bool Catalog::LookupXattrsMd5Path(
  const shash::Md5 &md5path,
  XattrList *xattrs) const
{
  assert(IsInitialized());

  MutexLockGuard m(lock_);
  sql_lookup_xattrs_->BindPathHash(md5path);
  bool found = sql_lookup_xattrs_->FetchRow();
  if (found && (xattrs != NULL)) {
    *xattrs = sql_lookup_xattrs_->GetXattrs();
  }
  sql_lookup_xattrs_->Reset();

  return found;
}


bool Catalog::ListingMd5PathStat(const shash::Md5 &md5path,
                                 StatEntryList *listing) const
{
  assert(IsInitialized());

  DirectoryEntry dirent;
  StatEntry entry;

  MutexLockGuard m(lock_);
  sql_listing_->BindPathHash(md5path);
  while (sql_listing_->FetchRow()) {
    dirent = sql_listing_->GetDirent(this);
    if (dirent.IsHidden())
      continue;
    FixTransitionPoint(md5path, &dirent);
    entry.name = dirent.name();
    entry.info = dirent.GetStatStructure();
    listing->PushBack(entry);
  }
  sql_listing_->Reset();

  return true;
}


bool Catalog::ListingMd5Path(const shash::Md5 &md5path,
                             DirectoryEntryList *listing,
                             const bool expand_symlink) const
{
  assert(IsInitialized());

  MutexLockGuard m(lock_);

  sql_listing_->BindPathHash(md5path);
  while (sql_listing_->FetchRow()) {
    DirectoryEntry dirent = sql_listing_->GetDirent(this, expand_symlink);
    FixTransitionPoint(md5path, &dirent);
    listing->push_back(dirent);
  }
  sql_listing_->Reset();

  return true;
}


bool Catalog::AllChunksBegin() {
  return sql_all_chunks_->Open();
}


bool Catalog::AllChunksNext(shash::Any *hash, zlib::Algorithms *compression_alg)
{
  return sql_all_chunks_->Next(hash, compression_alg);
}


bool Catalog::AllChunksEnd() {
  return sql_all_chunks_->Close();
}


bool Catalog::ListMd5PathChunks(const shash::Md5  &md5path,
                                const shash::Algorithms interpret_hashes_as,
                                FileChunkList    *chunks) const
{
  assert(IsInitialized() && chunks->IsEmpty());

  MutexLockGuard m(lock_);

  sql_chunks_listing_->BindPathHash(md5path);
  while (sql_chunks_listing_->FetchRow()) {
    chunks->PushBack(sql_chunks_listing_->GetFileChunk(interpret_hashes_as));
  }
  sql_chunks_listing_->Reset();

  return true;
}


const Catalog::HashVector& Catalog::GetReferencedObjects() const {
  if (!referenced_hashes_.empty()) {
    return referenced_hashes_;
  }

  // retrieve all referenced content hashes of both files and file chunks
  SqlListContentHashes list_content_hashes(database());
  while (list_content_hashes.FetchRow()) {
    referenced_hashes_.push_back(list_content_hashes.GetHash());
  }

  return referenced_hashes_;
}


void Catalog::TakeDatabaseFileOwnership() {
  managed_database_ = true;
  if (NULL != database_) {
    database_->TakeFileOwnership();
  }
}


void Catalog::DropDatabaseFileOwnership() {
  managed_database_ = false;
  if (NULL != database_) {
    database_->DropFileOwnership();
  }
}


uint64_t Catalog::GetTTL() const {
  MutexLockGuard m(lock_);
  return database().GetPropertyDefault<uint64_t>("TTL", kDefaultTTL);
}


bool Catalog::HasExplicitTTL() const {
  MutexLockGuard m(lock_);
  return database().HasProperty("TTL");
}


bool Catalog::GetVOMSAuthz(string *authz) const {
  bool result;
  MutexLockGuard m(lock_);
  if (voms_authz_status_ == kVomsPresent) {
    if (authz) {*authz = voms_authz_;}
    result = true;
  } else if (voms_authz_status_ == kVomsNone) {
    result = false;
  } else {
    if (database().HasProperty("voms_authz")) {
      voms_authz_ = database().GetProperty<string>("voms_authz");
      if (authz) {*authz = voms_authz_;}
      voms_authz_status_ = kVomsPresent;
    } else {
      voms_authz_status_ = kVomsNone;
    }
    result = (voms_authz_status_ == kVomsPresent);
  }
  return result;
}

uint64_t Catalog::GetRevision() const {
  MutexLockGuard m(lock_);
  return database().GetPropertyDefault<uint64_t>("revision", 0);
}

uint64_t Catalog::GetLastModified() const {
  const std::string prop_name = "last_modified";
  return (database().HasProperty(prop_name))
    ? database().GetProperty<int>(prop_name)
    : 0u;
}


uint64_t Catalog::GetNumChunks() const {
  return counters_.Get("self_regular") + counters_.Get("self_chunks");
}


uint64_t Catalog::GetNumEntries() const {
  const string sql = "SELECT count(*) FROM catalog;";

  MutexLockGuard m(lock_);
  SqlCatalog stmt(database(), sql);
  return (stmt.FetchRow()) ? stmt.RetrieveInt64(0) : 0;
}


shash::Any Catalog::GetPreviousRevision() const {
  MutexLockGuard m(lock_);
  const std::string hash_string =
    database().GetPropertyDefault<std::string>("previous_revision", "");
  return (!hash_string.empty())
    ? shash::MkFromHexPtr(shash::HexPtr(hash_string), shash::kSuffixCatalog)
    : shash::Any();
}


string Catalog::PrintMemStatistics() const {
  sqlite::MemStatistics stats;
  {
    MutexLockGuard m(lock_);
    database().GetMemStatistics(&stats);
  }
  return string(mountpoint().GetChars(), mountpoint().GetLength()) + ": " +
    StringifyInt(stats.lookaside_slots_used) + " / " +
      StringifyInt(stats.lookaside_slots_max) + " slots -- " +
      StringifyInt(stats.lookaside_hit) + " hits, " +
      StringifyInt(stats.lookaside_miss_size) + " misses-size, " +
      StringifyInt(stats.lookaside_miss_full) + " misses-full -- " +
    StringifyInt(stats.page_cache_used / 1024) + " kB pages -- " +
      StringifyInt(stats.page_cache_hit) + " hits, " +
      StringifyInt(stats.page_cache_miss) + " misses -- " +
    StringifyInt(stats.schema_used / 1024) + " kB schema -- " +
    StringifyInt(stats.stmt_used / 1024) + " kB statements";
}


inode_t Catalog::GetMangledInode(const uint64_t row_id,
                                 const uint64_t hardlink_group) const {
  assert(IsInitialized());

  if (inode_range_.IsDummy()) {
    return DirectoryEntry::kInvalidInode;
  }

  inode_t inode = row_id + inode_range_.offset;

  // Hardlinks are encoded in catalog-wide unique hard link group ids.
  // These ids must be resolved to actual inode relationships at runtime.
  if (hardlink_group > 0) {
    HardlinkGroupMap::const_iterator inode_iter =
      hardlink_groups_.find(hardlink_group);

    // Use cached entry if possible
    if (inode_iter == hardlink_groups_.end()) {
      hardlink_groups_[hardlink_group] = inode;
    } else {
      inode = inode_iter->second;
    }
  }

  if (inode_annotation_) {
    inode = inode_annotation_->Annotate(inode);
  }

  return inode;
}


const Catalog::NestedCatalogList& Catalog::ListNestedCatalogs() const {
  MutexLockGuard m(lock_);

  if (nested_catalog_cache_dirty_) {
    LogCvmfs(kLogCatalog, kLogDebug, "refreshing nested catalog cache of '%s'",
             mountpoint().c_str());
    while (sql_list_nested_->FetchRow()) {
      NestedCatalog nested;
      nested.mountpoint = PlantPath(sql_list_nested_->GetPath());
      nested.hash = sql_list_nested_->GetContentHash();
      nested.size = sql_list_nested_->GetSize();
      nested_catalog_cache_.push_back(nested);
    }
    sql_list_nested_->Reset();
    nested_catalog_cache_dirty_ = false;
  }

  return nested_catalog_cache_;
}


const Catalog::NestedCatalogList Catalog::ListOwnNestedCatalogs() const {
  NestedCatalogList result;

  MutexLockGuard m(lock_);

  while (sql_own_list_nested_->FetchRow()) {
    NestedCatalog nested;
    nested.mountpoint = PlantPath(sql_own_list_nested_->GetPath());
    nested.hash = sql_own_list_nested_->GetContentHash();
    nested.size = sql_own_list_nested_->GetSize();
    result.push_back(nested);
  }
  sql_own_list_nested_->Reset();

  return result;
}


void Catalog::ResetNestedCatalogCacheUnprotected() {
  nested_catalog_cache_.clear();
  nested_catalog_cache_dirty_ = true;
}


bool Catalog::FindNested(const PathString &mountpoint,
                         shash::Any *hash, uint64_t *size) const
{
  MutexLockGuard m(lock_);
  PathString normalized_mountpoint = NormalizePath2(mountpoint);
  sql_lookup_nested_->BindSearchPath(normalized_mountpoint);
  bool found = sql_lookup_nested_->FetchRow();
  if (found && (hash != NULL)) {
    *hash = sql_lookup_nested_->GetContentHash();
    *size = sql_lookup_nested_->GetSize();
  }
  sql_lookup_nested_->Reset();

  return found;
}


void Catalog::SetInodeAnnotation(InodeAnnotation *new_annotation) {
  MutexLockGuard m(lock_);
  // Since annotated inodes could come back to the catalog in order to
  // get stripped, exchanging the annotation is not allowed
  assert((inode_annotation_ == NULL) || (inode_annotation_ == new_annotation));
  inode_annotation_ = new_annotation;
}


void Catalog::SetOwnerMaps(const OwnerMap *uid_map, const OwnerMap *gid_map) {
  uid_map_ = (uid_map && uid_map->HasEffect()) ? uid_map : NULL;
  gid_map_ = (gid_map && gid_map->HasEffect()) ? gid_map : NULL;
}


void Catalog::AddChild(Catalog *child) {
  assert(NULL == FindChild(child->mountpoint()));

  MutexLockGuard m(lock_);
  children_[child->mountpoint()] = child;
  child->set_parent(this);
}


void Catalog::RemoveChild(Catalog *child) {
  assert(NULL != FindChild(child->mountpoint()));

  MutexLockGuard m(lock_);
  child->set_parent(NULL);
  children_.erase(child->mountpoint());
}


CatalogList Catalog::GetChildren() const {
  CatalogList result;

  MutexLockGuard m(lock_);
  for (NestedCatalogMap::const_iterator i = children_.begin(),
       iEnd = children_.end(); i != iEnd; ++i)
  {
    result.push_back(i->second);
  }

  return result;
}


Catalog* Catalog::FindSubtree(const PathString &path) const {
  // Check if this catalog fits the beginning of the path.
  if (!path.StartsWith(mountpoint_))
    return NULL;

  PathString remaining(path.Suffix(mountpoint_.GetLength()));
  remaining.Append("/", 1);

  // now we recombine the path elements successively
  // in order to find a child which serves a part of the path
  PathString path_prefix(mountpoint_);
  Catalog *result = NULL;
  // Skip the first '/'
  path_prefix.Append("/", 1);
  const char *c = remaining.GetChars() + 1;
  for (unsigned i = 1; i < remaining.GetLength(); ++i, ++c) {
    if (*c == '/') {
      result = FindChild(path_prefix);

      // If we found a child serving a part of the path we can stop searching.
      // Remaining sub path elements are possibly served by a grand child.
      if (result != NULL)
        break;
    }
    path_prefix.Append(c, 1);
  }

  return result;
}


Catalog* Catalog::FindChild(const PathString &mountpoint) const {
  NestedCatalogMap::const_iterator nested_iter;

  MutexLockGuard m(lock_);
  nested_iter = children_.find(mountpoint);
  Catalog* result =
    (nested_iter == children_.end()) ? NULL : nested_iter->second;

  return result;
}


void Catalog::FixTransitionPoint(const shash::Md5 &md5path,
                                 DirectoryEntry *dirent) const
{
  if (!HasParent())
    return;

  if (dirent->IsNestedCatalogRoot()) {
    // Normal nested catalog
    DirectoryEntry parent_dirent;
    const bool retval = parent_->LookupMd5Path(md5path, &parent_dirent);
    assert(retval);
    dirent->set_inode(parent_dirent.inode());
  } else if (md5path == kMd5PathEmpty) {
    // Bind mountpoint
    DirectoryEntry parent_dirent;
    const bool retval = parent_->LookupPath(mountpoint_, &parent_dirent);
    assert(retval);
    dirent->set_inode(parent_dirent.inode());
  }
}

}  // namespace catalog