quota_posix.h

Go to the documentation of this file.

#ifndef CVMFS_QUOTA_POSIX_H_
#define CVMFS_QUOTA_POSIX_H_

#include <pthread.h>
#include <stdint.h>
#include <sys/types.h>
#include <unistd.h>

#include <map>
#include <string>
#include <vector>

#include "crypto/hash.h"
#include "duplex_sqlite3.h"
#include "gtest/gtest_prod.h"
#include "quota.h"
#include "statistics.h"
#include "util/single_copy.h"
#include "util/string.h"

namespace perf {
class Recorder;
}

class PosixQuotaManager : public QuotaManager {
  FRIEND_TEST(T_QuotaManager, BindReturnPipe);
  FRIEND_TEST(T_QuotaManager, Cleanup);
  FRIEND_TEST(T_QuotaManager, Contains);
  FRIEND_TEST(T_QuotaManager, InitDatabase);
  FRIEND_TEST(T_QuotaManager, MakeReturnPipe);

 public:
  static PosixQuotaManager *Create(const std::string &cache_workspace,
    const uint64_t limit, const uint64_t cleanup_threshold,
    const bool rebuild_database);
  static PosixQuotaManager *CreateShared(
    const std::string &exe_path,
    const std::string &cache_workspace,
    const uint64_t limit,
    const uint64_t cleanup_threshold,
    bool foreground);
  static int MainCacheManager(int argc, char **argv);

  virtual ~PosixQuotaManager();
  virtual bool HasCapability(Capabilities capability) { return true; }

  virtual void Insert(const shash::Any &hash, const uint64_t size,
                      const std::string &description);
  virtual void InsertVolatile(const shash::Any &hash, const uint64_t size,
                              const std::string &description);
  virtual bool Pin(const shash::Any &hash, const uint64_t size,
                   const std::string &description, const bool is_catalog);
  virtual void Unpin(const shash::Any &hash);
  virtual void Touch(const shash::Any &hash);
  virtual void Remove(const shash::Any &file);
  virtual bool Cleanup(const uint64_t leave_size);

  virtual void RegisterBackChannel(int back_channel[2],
                                   const std::string &channel_id);
  virtual void UnregisterBackChannel(int back_channel[2],
                                     const std::string &channel_id);

  virtual std::vector<std::string> List();
  virtual std::vector<std::string> ListPinned();
  virtual std::vector<std::string> ListCatalogs();
  virtual std::vector<std::string> ListVolatile();
  virtual uint64_t GetMaxFileSize();
  virtual uint64_t GetCapacity();
  virtual uint64_t GetSize();
  virtual uint64_t GetSizePinned();
  virtual bool     SetLimit(uint64_t limit);
  virtual uint64_t GetCleanupRate(uint64_t period_s);

  virtual void Spawn();
  virtual pid_t GetPid();
  virtual uint32_t GetProtocolRevision();

  void ManagedReadHalfPipe(int fd, void *buf, size_t nbyte);
  void SetCacheMgrPid(pid_t pid_) { cachemgr_pid_ = pid_;};


 private:
  enum FileTypes {
    kFileRegular = 0,
    kFileCatalog,
  };

  enum CommandType {
    kTouch = 0,
    kInsert,
    kReserve,
    kPin,
    kUnpin,
    kRemove,
    kCleanup,
    kList,
    kListPinned,
    kListCatalogs,
    kStatus,
    kLimits,
    kPid,
    kPinRegular,
    kRegisterBackChannel,
    kUnregisterBackChannel,
    kGetProtocolRevision,
    kInsertVolatile,
    // as of protocol revision 2
    kListVolatile,
    kCleanupRate,
    kSetLimit,
  };

  struct LruCommand {
    CommandType command_type;
    uint64_t size;    
    int return_pipe;  
    unsigned char digest[shash::kMaxDigestSize];
    uint16_t desc_length;

    LruCommand()
      : command_type(static_cast<CommandType>(0))
      , size(0)
      , return_pipe(-1)
      , desc_length(0)
    {
      memset(digest, 0, shash::kMaxDigestSize);
    }

    void SetSize(const uint64_t new_size) {
      uint64_t mask = 7;
      mask = ~(mask << (64-3));
      size = (new_size & mask) | size;
    }

    uint64_t GetSize() const {
      uint64_t mask = 7;
      mask = ~(mask << (64-3));
      return size & mask;
    }

    void StoreHash(const shash::Any &hash) {
      memcpy(digest, hash.digest, hash.GetDigestSize());
      // Exclude MD5
      uint64_t algo_flags = hash.algorithm - 1;
      algo_flags = algo_flags << (64-3);
      size |= algo_flags;
    }

    shash::Any RetrieveHash() const {
      uint64_t algo_flags = size >> (64-3);
      shash::Any result(static_cast<shash::Algorithms>(algo_flags+1));
      memcpy(result.digest, digest, result.GetDigestSize());
      return result;
    }
  };

  struct EvictCandidate {
    uint64_t size;
    uint64_t acseq;
    shash::Any hash;
    EvictCandidate(const shash::Any &h, uint64_t s, uint64_t a)
      : size(s), acseq(a), hash(h) {}
  };

  static const unsigned kLockFileMagicNumber = 142857;

  static const unsigned kSqliteMemPerThread = 2*1024*1024;

  static const unsigned kCommandBufferSize = 32;

  static const unsigned kEvictBatchSize = 1000;

  static const unsigned kMaxDescription = 512-sizeof(LruCommand);

  static const unsigned kHighPinWatermark = 75;

  static const uint64_t kVolatileFlag = 1ULL << 63;

  bool InitDatabase(const bool rebuild_database);
  bool RebuildDatabase();
  void CloseDatabase();
  bool Contains(const std::string &hash_str);
  bool DoCleanup(const uint64_t leave_size);
  bool EmptyTrash(const std::vector<std::string> &trash);

  void MakeReturnPipe(int pipe[2]);
  int BindReturnPipe(int pipe_wronly);
  void UnbindReturnPipe(int pipe_wronly);
  void UnlinkReturnPipe(int pipe_wronly);
  void CloseReturnPipe(int pipe[2]);
  void CleanupPipes();

  void CheckFreeSpace();
  void CheckHighPinWatermark();
  void ProcessCommandBunch(const unsigned num,
                           const LruCommand *commands,
                           const char *descriptions);
  static void *MainCommandServer(void *data);

  void DoInsert(const shash::Any &hash, const uint64_t size,
                const std::string &description, const CommandType command_type);
  std::vector<std::string> DoList(const CommandType list_command);
  void GetSharedStatus(uint64_t *gauge, uint64_t *pinned);
  bool SetSharedLimit(uint64_t limit);
  void GetLimits(uint64_t *limit, uint64_t *cleanup_threshold);

  static void ParseDirectories(const std::string cache_workspace,
                               std::string *cache_dir,
                               std::string *workspace_dir);
  PosixQuotaManager(const uint64_t limit, const uint64_t cleanup_threshold,
                    const std::string &cache_workspace);

  bool shared_;

  bool spawned_;

  uint64_t limit_;

  uint64_t cleanup_threshold_;

  uint64_t gauge_;

  uint64_t pinned_;

  uint64_t seq_;

  std::string cache_dir_;

  std::string workspace_dir_;

  std::map<shash::Any, uint64_t> pinned_chunks_;

  int pipe_lru_[2];

  pthread_t thread_lru_;

  int fd_lock_cachedb_;

  bool async_delete_;


  pid_t cachemgr_pid_;

  perf::MultiRecorder cleanup_recorder_;

  sqlite3 *database_;
  sqlite3_stmt *stmt_touch_;
  sqlite3_stmt *stmt_unpin_;
  sqlite3_stmt *stmt_block_;
  sqlite3_stmt *stmt_unblock_;
  sqlite3_stmt *stmt_new_;
  sqlite3_stmt *stmt_lru_;
  sqlite3_stmt *stmt_size_;
  sqlite3_stmt *stmt_rm_;
  sqlite3_stmt *stmt_rm_batch_;
  sqlite3_stmt *stmt_list_;
  sqlite3_stmt *stmt_list_pinned_;  
  sqlite3_stmt *stmt_list_catalogs_;
  sqlite3_stmt *stmt_list_volatile_;

  bool initialized_;
};  // class PosixQuotaManager

#endif  // CVMFS_QUOTA_POSIX_H_