pack.h

Go to the documentation of this file.

#ifndef CVMFS_PACK_H_
#define CVMFS_PACK_H_

#include <inttypes.h>
#include <pthread.h>

#include <cstdio>
#include <set>
#include <string>
#include <vector>

#include "crypto/hash.h"
#include "gtest/gtest_prod.h"
#include "util/concurrency.h"
#include "util/single_copy.h"

class ObjectPack : SingleCopy {
  FRIEND_TEST(T_Pack, Bucket);
  FRIEND_TEST(T_Pack, ObjectPack);
  FRIEND_TEST(T_Pack, ObjectPackTransfer);

 private:
  struct Bucket;

 public:
  typedef Bucket *BucketHandle;

  enum BucketContentType { kEmpty, kNamed, kCas };

  static const uint64_t kDefaultLimit = 200 * 1024 * 1024;  // 200MB

  static const uint64_t kMaxObjects = 100000;

  explicit ObjectPack(const uint64_t limit = kDefaultLimit);
  ~ObjectPack();

  static void AddToBucket(const void *buf, const uint64_t size,
                          const BucketHandle handle);

  BucketHandle NewBucket();

  bool CommitBucket(const BucketContentType type, const shash::Any &id,
                    const BucketHandle handle, const std::string &name = "");

  void DiscardBucket(const BucketHandle handle);
  void TransferBucket(const BucketHandle handle, ObjectPack *other);

  unsigned char *BucketContent(size_t idx) const;
  uint64_t BucketSize(size_t idx) const;
  const shash::Any &BucketId(size_t idx) const;

  uint64_t size() const { return size_; }

  // This returns the number of objects in the pack (equal to the number of
  // committed buckets)
  size_t GetNoObjects() const { return buckets_.size(); }

 private:
  struct Bucket : SingleCopy {
    static const unsigned kInitialSize = 128;

    Bucket();
    ~Bucket();
    void Add(const void *buf, const uint64_t buf_size);

    unsigned char *content;
    uint64_t size;
    uint64_t capacity;
    shash::Any id;
    BucketContentType content_type;
    std::string name;
  };

  void InitLock();

  pthread_mutex_t *lock_;

  uint64_t limit_;
  uint64_t size_;
  std::set<BucketHandle> open_buckets_;
  std::vector<BucketHandle> buckets_;
};

namespace ObjectPackBuild {
struct Event {
  Event(const shash::Any &id, uint64_t size, unsigned buf_size, const void *buf,
        ObjectPack::BucketContentType type, const std::string &name)
      : id(id),
        size(size),
        buf_size(buf_size),
        buf(buf),
        object_type(type),
        object_name(name) {}

  shash::Any id;
  uint64_t size;
  unsigned buf_size;
  const void *buf;
  ObjectPack::BucketContentType object_type;
  std::string object_name;
};

enum State {
  kStateContinue = 0,
  kStateDone,
  kStateCorrupt,
  kStateBadFormat,
  kStateHeaderTooBig,
  kStateTrailingBytes,
};
}  // namespace ObjectPackBuild

class ObjectPackProducer {
 public:
  explicit ObjectPackProducer(ObjectPack *pack);
  ObjectPackProducer(const shash::Any &id, FILE *big_file,
                     const std::string &file_name = "");
  unsigned ProduceNext(const unsigned buf_size, unsigned char *buf);
  void GetDigest(shash::Any *hash);
  unsigned GetHeaderSize() { return header_.size(); }

 private:
  ObjectPack *pack_;

  FILE *big_file_;

  uint64_t pos_;

  size_t idx_;

  size_t pos_in_bucket_;

  std::string header_;
};

class ObjectPackConsumer : public Observable<ObjectPackBuild::Event> {
 public:
  explicit ObjectPackConsumer(const shash::Any &expected_digest,
                              const unsigned expected_header_size);
  ObjectPackBuild::State ConsumeNext(const unsigned buf_size,
                                     const unsigned char *buf);

 private:
  static const unsigned kAccuSize = 128 * 1024;

  struct IndexEntry {
    IndexEntry() : id(), size(), entry_type(), entry_name() {}
    IndexEntry(const shash::Any &id, const uint64_t size,
               ObjectPack::BucketContentType type, const std::string &name)
        : id(id), size(size), entry_type(type), entry_name(name) {}
    shash::Any id;
    uint64_t size;
    ObjectPack::BucketContentType entry_type;
    std::string entry_name;
  };

  bool ParseHeader();
  bool ParseItem(const std::string &line, IndexEntry *entry,
                 uint64_t *sum_size);

  ObjectPackBuild::State ConsumePayload(const unsigned buf_size,
                                        const unsigned char *buf);

  shash::Any expected_digest_;
  unsigned expected_header_size_;

  uint64_t pos_;

  unsigned idx_;

  unsigned pos_in_object_;

  unsigned char accumulator_[kAccuSize];

  unsigned pos_in_accu_;

  ObjectPackBuild::State state_;

  std::string raw_header_;

  uint64_t size_;

  std::vector<IndexEntry> index_;
};

#endif  // CVMFS_PACK_H_