catalog_mgr_rw.h

Go to the documentation of this file.

#ifndef CVMFS_CATALOG_MGR_RW_H_
#define CVMFS_CATALOG_MGR_RW_H_

#include <pthread.h>
#include <stdint.h>

#include <list>
#include <map>
#include <set>
#include <string>
#include <unordered_set>
#include <unordered_map>

#include "catalog_mgr_ro.h"
#include "catalog_rw.h"
#include "file_chunk.h"
#include "upload_spooler_result.h"
#include "util/future.h"
#include "xattr.h"
#include "catalog_downloader.h"
#include "ingestion/pipeline.h"

class XattrList;
namespace upload {
class Spooler;
}

namespace download {
class DownloadManager;
}

namespace manifest {
class Manifest;
}

namespace perf {
class Statistics;
}

namespace catalog {
template<class CatalogMgrT>
class CatalogBalancer;
}

namespace catalog {

class WritableCatalogManager : public SimpleCatalogManager {
  friend class CatalogBalancer<WritableCatalogManager>;
  // TODO(jblomer): only needed to get Spooler's hash algorithm.  Remove me
  // after refactoring of the swissknife utility.
  friend class VirtualCatalog;

 public:
  WritableCatalogManager(const shash::Any &base_hash,
                         const std::string &stratum0,
                         const std::string &dir_temp,
                         upload::Spooler *spooler,
                         download::DownloadManager *download_manager,
                         bool enforce_limits,
                         const unsigned nested_kcatalog_limit,
                         const unsigned root_kcatalog_limit,
                         const unsigned file_mbyte_limit,
                         perf::Statistics *statistics,
                         bool is_balanceable,
                         unsigned max_weight,
                         unsigned min_weight,
                         const std::string &dir_cache = "");
  ~WritableCatalogManager();
  static manifest::Manifest *CreateRepository(const std::string &dir_temp,
                                              const bool volatile_content,
                                              const std::string &voms_authz,
                                              upload::Spooler *spooler);

  // DirectoryEntry handling
  void AddFile(const DirectoryEntryBase &entry,
               const XattrList &xattrs,
               const std::string &parent_directory) {
    AddFile(DirectoryEntry(entry), xattrs, parent_directory);
  }
  void AddChunkedFile(const DirectoryEntryBase &entry,
                      const XattrList &xattrs,
                      const std::string &parent_directory,
                      const FileChunkList &file_chunks);
  void RemoveFile(const std::string &file_path);

  void AddDirectory(const DirectoryEntryBase &entry,
                    const XattrList &xattrs,
                    const std::string &parent_directory);
  void TouchDirectory(const DirectoryEntryBase &entry,
                      const XattrList &xattrs,
                      const std::string &directory_path);
  void RemoveDirectory(const std::string &directory_path);

  void Clone(const std::string from, const std::string to);
  void CloneTree(const std::string &from_dir, const std::string &to_dir);

  // Hardlink group handling
  void AddHardlinkGroup(const DirectoryEntryBaseList &entries,
                        const XattrList &xattrs,
                        const std::string &parent_directory,
                        const FileChunkList &file_chunks);
  void ShrinkHardlinkGroup(const std::string &remove_path);

  // Nested catalog handling
  void CreateNestedCatalog(const std::string &mountpoint);
  void RemoveNestedCatalog(const std::string &mountpoint,
                           const bool merge = true);
  void SwapNestedCatalog(const string &mountpoint, const shash::Any &new_hash,
                         const uint64_t new_size);
  void GraftNestedCatalog(const string &mountpoint, const shash::Any &new_hash,
                          const uint64_t new_size);
  bool IsTransitionPoint(const std::string &mountpoint);
  WritableCatalog *GetHostingCatalog(const std::string &path);

  inline bool IsBalanceable() const { return is_balanceable_; }
  void PrecalculateListings();

  void SetTTL(const uint64_t new_ttl);
  bool SetVOMSAuthz(const std::string &voms_authz);
  bool Commit(const bool stop_for_tweaks,
              const uint64_t manual_revision,
              manifest::Manifest *manifest);

  void Balance() {
    if (IsBalanceable()) {
      DoBalance();
    } else {
      LogCvmfs(kLogCatalog, kLogVerboseMsg,
               "Not balancing the catalog "
               "manager because it is not balanceable");
    }
  }

  void LoadCatalogs(const std::string &base_path, const std::unordered_set<std::string> &dirs);
  void SetupSingleCatalogUploadCallback();
  void RemoveSingleCatalogUploadCallback();
  void AddCatalogToQueue(const std::string &path);
  void ScheduleReadyCatalogs();
  bool LookupDirEntry(const std::string &path,
                      const LookupOptions options,
                      DirectoryEntry *dirent);



 protected:
  void EnforceSqliteMemLimit() { }

  Catalog *CreateCatalog(const PathString &mountpoint,
                         const shash::Any &catalog_hash,
                         Catalog *parent_catalog);
  void ActivateCatalog(Catalog *catalog);

  void AddFile(const DirectoryEntry &entry,
               const XattrList &xattrs,
               const std::string &parent_directory);

 private:
  bool FindCatalog(const std::string &path,
                   WritableCatalog **result,
                   DirectoryEntry *dirent = NULL);
  void DoBalance();
  void FixWeight(WritableCatalog *catalog);

  void CloneTreeImpl(const PathString &source_dir,
                     const std::string &dest_parent_dir,
                     const NameString &dest_name);

  struct CatalogInfo {
    uint64_t ttl;
    size_t size;
    shash::Any content_hash;
    uint64_t revision;
  };

  struct CatalogUploadContext {
    Future<CatalogInfo> *root_catalog_info;
    bool stop_for_tweaks;
  };

  struct CatalogDownloadContext {
    const std::unordered_set<std::string> * dirs;
  };

  void CatalogDownloadCallback(const CatalogDownloadResult &result,
                               const CatalogDownloadContext context);
  void SingleCatalogUploadCallback(const upload::SpoolerResult &result);



  CatalogInfo SnapshotCatalogs(const bool stop_for_tweaks);
  void FinalizeCatalog(WritableCatalog *catalog, const bool stop_for_tweaks);
  void ScheduleCatalogProcessing(WritableCatalog *catalog);

  void GetModifiedCatalogLeafs(WritableCatalogList *result) const {
    const bool dirty = GetModifiedCatalogLeafsRecursively(GetRootCatalog(),
                                                          result);
    assert(dirty);
  }
  bool GetModifiedCatalogLeafsRecursively(Catalog *catalog,
                                          WritableCatalogList *result) const;

  void CatalogUploadCallback(const upload::SpoolerResult &result,
                             const CatalogUploadContext clg_upload_context);

  bool CopyCatalogToLocalCache(const upload::SpoolerResult &result);

 private:
  inline void SyncLock() { pthread_mutex_lock(sync_lock_); }
  inline void SyncUnlock() { pthread_mutex_unlock(sync_lock_); }

  //****************************************************************************
  // Workaround -- Serialized Catalog Committing
  void GetModifiedCatalogs(WritableCatalogList *result) const {
    const unsigned int
        number_of_dirty_catalogs = GetModifiedCatalogsRecursively(
            GetRootCatalog(), result);
    assert(number_of_dirty_catalogs <= result->size());
  }
  int GetModifiedCatalogsRecursively(const Catalog *catalog,
                                     WritableCatalogList *result) const;
  void CatalogUploadSerializedCallback(const upload::SpoolerResult &result,
                                       const CatalogUploadContext unused);
  CatalogInfo SnapshotCatalogsSerialized(const bool stop_for_tweaks);
    void CatalogHashSerializedCallback(
    const CompressHashResult &result);
  //****************************************************************************

  // defined in catalog_mgr_rw.cc
  static const std::string kCatalogFilename;

  // private lock of WritableCatalogManager
  pthread_mutex_t *sync_lock_;
  upload::Spooler *spooler_;

  pthread_mutex_t *catalog_processing_lock_;
  std::map<std::string, WritableCatalog *> catalog_processing_map_;

  // ingestsql
  std::list<WritableCatalog*>              pending_catalogs_;
  CatalogDownloadPipeline                  *catalog_download_pipeline_;
  pthread_mutex_t                          *catalog_download_lock_;
  std::unordered_map<std::string, Catalog*> catalog_download_map_;
  pthread_mutex_t                   *catalog_hash_lock_; 
  std::map<std::string, shash::Any>  catalog_hash_map_;

  // TODO(jblomer): catalog limits should become its own struct
  bool enforce_limits_;
  unsigned nested_kcatalog_limit_;
  unsigned root_kcatalog_limit_;
  unsigned file_mbyte_limit_;

  XattrList empty_xattrs;

  const bool is_balanceable_;

  const unsigned max_weight_;

  const unsigned min_weight_;

  const unsigned balance_weight_;
};  // class WritableCatalogManager

}  // namespace catalog

#endif  // CVMFS_CATALOG_MGR_RW_H_