Merge pull request #53 from HDFGroup/chogan/limit_name_size

Set upper limit to name lengths
HDFGroup · Dec 14, 2020 · 232a1a5 · 232a1a5
2 parents 4f18fb1 + 329f6e9
commit 232a1a5
Show file tree

Hide file tree

Showing 6 changed files with 84 additions and 7 deletions.
diff --git a/src/api/bucket.cc b/src/api/bucket.cc
@@ -15,7 +15,12 @@ Bucket::Bucket(const std::string &initial_name,
                const std::shared_ptr<Hermes> &h, Context ctx)
     : name_(initial_name), hermes_(h) {
   (void)ctx;
-  id_ = GetOrCreateBucketId(&hermes_->context_, &hermes_->rpc_, name_);
+
+  if (IsNameTooLong(name_)) {
+    id_.as_int = 0;
+  } else {
+    id_ = GetOrCreateBucketId(&hermes_->context_, &hermes_->rpc_, name_);
+  }
 }
 
 bool Bucket::IsValid() const {
@@ -28,7 +33,12 @@ Status Bucket::Put(const std::string &name, const u8 *data, size_t size,
                    Context &ctx) {
   Status ret = 0;
 
-  if (IsValid()) {
+  if (IsNameTooLong(name)) {
+    // TODO(chogan): @errorhandling
+    ret = 1;
+  }
+
+  if (IsValid() && ret == 0) {
     std::vector<size_t> sizes(1, size);
     std::vector<PlacementSchema> schemas;
     HERMES_BEGIN_TIMED_BLOCK("CalculatePlacement");
@@ -154,8 +164,13 @@ Status Bucket::RenameBlob(const std::string &old_name,
   (void)ctx;
   Status ret = 0;
 
-  LOG(INFO) << "Renaming Blob " << old_name << " to " << new_name << '\n';
-  hermes::RenameBlob(&hermes_->context_, &hermes_->rpc_, old_name, new_name);
+  if (IsNameTooLong(new_name)) {
+    ret = 1;
+    // TODO(chogan): @errorhandling
+  } else {
+    LOG(INFO) << "Renaming Blob " << old_name << " to " << new_name << '\n';
+    hermes::RenameBlob(&hermes_->context_, &hermes_->rpc_, old_name, new_name);
+  }
 
   return ret;
 }
@@ -198,8 +213,13 @@ Status Bucket::Rename(const std::string &new_name, Context &ctx) {
   (void)ctx;
   Status ret = 0;
 
-  LOG(INFO) << "Renaming a bucket to" << new_name << '\n';
-  RenameBucket(&hermes_->context_, &hermes_->rpc_, id_, name_, new_name);
+  if (IsNameTooLong(new_name)) {
+    ret = 1;
+    // TODO(chogan): @errorhandling
+  } else {
+    LOG(INFO) << "Renaming a bucket to" << new_name << '\n';
+    RenameBucket(&hermes_->context_, &hermes_->rpc_, id_, name_, new_name);
+  }
 
   return ret;
 }

diff --git a/src/api/bucket.h b/src/api/bucket.h
@@ -165,7 +165,15 @@ Status Bucket::Put(std::vector<std::string> &names,
                    std::vector<std::vector<T>> &blobs, Context &ctx) {
   Status ret = 0;
 
-  if (IsValid()) {
+  for (auto &name : names) {
+    if (IsNameTooLong(name)) {
+      // TODO(chogan): @errorhandling
+      ret = 1;
+      break;
+    }
+  }
+
+  if (IsValid() && ret == 0) {
     size_t num_blobs = blobs.size();
     std::vector<size_t> sizes_in_bytes(num_blobs);
     for (size_t i = 0; i < num_blobs; ++i) {

diff --git a/src/hermes_types.h b/src/hermes_types.h
@@ -39,6 +39,7 @@ static constexpr int kMaxBufferPoolSlabs = 8;
 constexpr int kMaxPathLength = 256;
 constexpr int kMaxBufferPoolShmemNameLength = 64;
 constexpr int kMaxDevices = 8;
+constexpr int kMaxNameSize = 128;
 
 constexpr char kPlaceInHierarchy[] = "PlaceInHierarchy";
 

diff --git a/src/metadata_management.cc b/src/metadata_management.cc
@@ -12,6 +12,17 @@
 
 namespace hermes {
 
+bool IsNameTooLong(const std::string &name) {
+  bool result = false;
+  if (name.size() + 1 >= kMaxNameSize) {
+    LOG(WARNING) << "Name '" << name << "' exceeds the maximum name size of "
+                 << kMaxNameSize << " bytes." << std::endl;
+    result = true;
+  }
+
+  return result;
+}
+
 bool IsNullBucketId(BucketID id) {
   bool result = id.as_int == 0;
 

diff --git a/src/metadata_management.h b/src/metadata_management.h
@@ -228,6 +228,11 @@ SwapBlob VecToSwapBlob(std::vector<BufferID> &vec);
  */
 SwapBlob IdArrayToSwapBlob(BufferIdArray ids);
 
+/**
+ *
+ */
+bool IsNameTooLong(const std::string &name);
+
 }  // namespace hermes
 
 #endif  // HERMES_METADATA_MANAGEMENT_H_
diff --git a/test/mdm_test.cc b/test/mdm_test.cc
@@ -146,6 +146,37 @@ static void TestBucketRefCounting(HermesPtr hermes) {
   Assert(!bucket1.IsValid());
 }
 
+static void TestMaxNameLength(HermesPtr hermes) {
+  // Bucket with a name that's too large is invalid.
+  hapi::Context ctx;
+  std::string long_name(kMaxNameSize + 1, 'x');
+  hapi::Bucket invalid_bucket(long_name, hermes, ctx);
+  Assert(!invalid_bucket.IsValid());
+
+  // Put fails when a blob name is too long
+  std::string name = "b1";
+  hapi::Bucket bucket(name, hermes, ctx);
+  hapi::Blob blob('x');
+  Status status = bucket.Put(long_name, blob, ctx);
+  Assert(status != 0);
+  Assert(!bucket.ContainsBlob(long_name));
+
+  // Vector Put fails if one name is too long
+  std::string a = "a";
+  std::string b = "b";
+  std::string c = "c";
+  std::vector<std::string> blob_names = {a, b, long_name, c};
+  std::vector<hapi::Blob> blobs = {blob, blob, blob, blob};
+  status = bucket.Put(blob_names, blobs, ctx);
+  Assert(status != 0);
+  Assert(!bucket.ContainsBlob(long_name));
+  Assert(!bucket.ContainsBlob(a));
+  Assert(!bucket.ContainsBlob(b));
+  Assert(!bucket.ContainsBlob(c));
+
+  bucket.Destroy(ctx);
+}
+
 int main(int argc, char **argv) {
   int mpi_threads_provided;
   MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &mpi_threads_provided);
@@ -163,6 +194,7 @@ int main(int argc, char **argv) {
   TestRenameBlob(hermes);
   TestRenameBucket(hermes);
   TestBucketRefCounting(hermes);
+  TestMaxNameLength(hermes);
 
   hermes->Finalize(true);