bucket.h

Go to the documentation of this file.

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
 * Distributed under BSD 3-Clause license.                                   *
 * Copyright by The HDF Group.                                               *
 * Copyright by the Illinois Institute of Technology.                        *
 * All rights reserved.                                                      *
 *                                                                           *
 * This file is part of Hermes. The full Hermes copyright notice, including  *
 * terms governing use, modification, and redistribution, is contained in    *
 * the COPYING file, which can be found at the top directory. If you do not  *
 * have access to the file, you may request a copy from help@hdfgroup.org.   *
 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
 
#ifndef BUCKET_H_
#define BUCKET_H_
 
#include <glog/logging.h>
 
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>
 
#include "dpe/round_robin.h"
#include "hermes.h"
#include "metadata_management.h"
#include "utils.h"
 
namespace hermes {
 
namespace api {
 
class Bucket {
 private:
  std::string name_;
  hermes::BucketID id_;
 
 public:
  std::shared_ptr<Hermes> hermes_;
  Context ctx_;
 
  // TODO(chogan): Think about copy/move constructor/assignment operators
 
  Bucket() : name_(""), id_{0, 0}, hermes_(nullptr) {
    LOG(INFO) << "Create NULL Bucket " << std::endl;
  }
 
  Bucket(const std::string &initial_name, std::shared_ptr<Hermes> const &h,
         Context ctx = Context());
 
  ~Bucket();
 
  std::string GetName() const;
 
  u64 GetId() const;
 
  bool IsValid() const;
 
  size_t GetTotalBlobSize();
 
  template <typename T>
  Status Put(const std::string &name, const std::vector<T> &data);
 
  template <typename T>
  Status Put(const std::string &name, const std::vector<T> &data, Context &ctx);
 
  Status Put(const std::string &name, const u8 *data, size_t size);
 
  Status Put(const std::string &name, const u8 *data, size_t size,
             const Context &ctx);
 
  template <typename T>
  Status Put(const std::vector<std::string> &names,
             const std::vector<std::vector<T>> &blobs);
 
  template <typename T>
  Status Put(const std::vector<std::string> &names,
             const std::vector<std::vector<T>> &blobs, const Context &ctx);
 
  size_t GetBlobSize(const std::string &name, const Context &ctx);
 
  size_t GetBlobSize(Arena *arena, const std::string &name, const Context &ctx);
 
  size_t Get(const std::string &name, Blob &user_blob);
 
  size_t Get(const std::string &name, Blob &user_blob, const Context &ctx);
 
  std::vector<size_t> Get(const std::vector<std::string> &names,
                          std::vector<Blob> &blobs, const Context &ctx);
 
  size_t Get(const std::string &name, void *user_blob, size_t blob_size,
             const Context &ctx);
 
  size_t GetNext(u64 blob_index, Blob &user_blob);
 
  size_t GetNext(u64 blob_index, Blob &user_blob, const Context &ctx);
 
  size_t GetNext(u64 blob_index, void *user_blob, size_t blob_size,
                 const Context &ctx);
 
  // TODO(chogan):
  std::vector<size_t> GetNext(u64 blob_index, u64 count,
                              std::vector<Blob> &blobs, const Context &ctx);
 
  template <class Predicate>
  Status GetV(void *user_blob, Predicate pred, Context &ctx);
 
  Status DeleteBlob(const std::string &name);
 
  Status DeleteBlob(const std::string &name, const Context &ctx);
 
  Status RenameBlob(const std::string &old_name, const std::string &new_name);
 
  Status RenameBlob(const std::string &old_name, const std::string &new_name,
                    const Context &ctx);
 
  bool ContainsBlob(const std::string &name);
 
  bool BlobIsInSwap(const std::string &name);
 
  template <class Predicate>
  std::vector<std::string> GetBlobNames(Predicate pred, Context &ctx);
 
  Status Rename(const std::string &new_name);
 
  Status Rename(const std::string &new_name, const Context &ctx);
 
  Status Persist(const std::string &file_name);
 
  Status Persist(const std::string &file_name, const Context &ctx);
 
  void OrganizeBlob(const std::string &blob_name, f32 epsilon,
                    f32 custom_importance = -1.f);
 
  Status Release();
 
  Status Release(const Context &ctx);
 
  Status Destroy();
 
  Status Destroy(const Context &ctx);
 
 private:
  template <typename T>
  Status PutInternal(const std::vector<std::string> &names,
                     const std::vector<size_t> &sizes,
                     const std::vector<std::vector<T>> &blobs,
                     const Context &ctx);
  template <typename T>
  Status PlaceBlobs(std::vector<PlacementSchema> &schemas,
                    const std::vector<std::vector<T>> &blobs,
                    const std::vector<std::string> &names, const Context &ctx);
};
 
template <typename T>
Status Bucket::Put(const std::string &name, const std::vector<T> &data,
                   Context &ctx) {
  Status result = Put(name, (u8 *)data.data(), data.size() * sizeof(T), ctx);
 
  return result;
}
 
template <typename T>
Status Bucket::Put(const std::string &name, const std::vector<T> &data) {
  Status result = Put(name, data, ctx_);
 
  return result;
}
 
template <typename T>
Status Bucket::PlaceBlobs(std::vector<PlacementSchema> &schemas,
                          const std::vector<std::vector<T>> &blobs,
                          const std::vector<std::string> &names,
                          const Context &ctx) {
  Status result;
 
  for (size_t i = 0; i < schemas.size(); ++i) {
    PlacementSchema &schema = schemas[i];
    hermes::Blob blob = {};
    blob.data = (u8 *)blobs[i].data();
    blob.size = blobs[i].size() * sizeof(T);
    LOG(INFO) << "Attaching blob '" << names[i] << "' to Bucket '" << name_
              << "'" << std::endl;
    result = PlaceBlob(&hermes_->context_, &hermes_->rpc_, schema, blob,
                       names[i], id_, ctx);
    if (result.Failed()) {
      // TODO(chogan): Need to return a std::vector<Status>
      break;
    }
  }
 
  return result;
}
 
template <typename T>
Status Bucket::Put(const std::vector<std::string> &names,
                   const std::vector<std::vector<T>> &blobs) {
  Status result = Put(names, blobs, ctx_);
 
  return result;
}
 
template <typename T>
Status Bucket::PutInternal(const std::vector<std::string> &names,
                           const std::vector<size_t> &sizes,
                           const std::vector<std::vector<T>> &blobs,
                           const Context &ctx) {
  std::vector<PlacementSchema> schemas;
  HERMES_BEGIN_TIMED_BLOCK("CalculatePlacement");
  Status result = CalculatePlacement(&hermes_->context_, &hermes_->rpc_, sizes,
                                     schemas, ctx);
  HERMES_END_TIMED_BLOCK();
 
  if (result.Succeeded()) {
    result = PlaceBlobs(schemas, blobs, names, ctx);
  } else {
    LOG(ERROR) << result.Msg();
  }
 
  return result;
}
 
template <typename T>
Status Bucket::Put(const std::vector<std::string> &names,
                   const std::vector<std::vector<T>> &blobs,
                   const Context &ctx) {
  Status ret;
 
  for (auto &name : names) {
    if (IsBlobNameTooLong(name)) {
      ret = BLOB_NAME_TOO_LONG;
      LOG(ERROR) << ret.Msg();
      return ret;
    }
  }
 
  if (blobs.size() == 0) {
    ret = INVALID_BLOB;
    LOG(ERROR) << ret.Msg();
    return ret;
  }
 
  if (IsValid()) {
    size_t num_blobs = blobs.size();
    std::vector<size_t> sizes_in_bytes(num_blobs);
    for (size_t i = 0; i < num_blobs; ++i) {
      sizes_in_bytes[i] = blobs[i].size() * sizeof(T);
    }
 
    if (ctx.rr_retry) {
      int num_devices =
          GetLocalSystemViewState(&hermes_->context_)->num_devices;
 
      for (int i = 0; i < num_devices; ++i) {
        ret = PutInternal(names, sizes_in_bytes, blobs, ctx);
 
        if (ret.Failed()) {
          RoundRobin rr_state;
          int current = rr_state.GetCurrentDeviceIndex();
          rr_state.SetCurrentDeviceIndex((current + 1) % num_devices);
        } else {
          break;
        }
      }
    } else {
      ret = PutInternal(names, sizes_in_bytes, blobs, ctx);
    }
  } else {
    ret = INVALID_BUCKET;
    LOG(ERROR) << ret.Msg();
    return ret;
  }
 
  return ret;
}
 
}  // namespace api
}  // namespace hermes
 
#endif  // BUCKET_H_