Release BlobInfo lock if entry doesn't exist

HDFGroup · Jan 26, 2022 · b4076d4 · b4076d4
1 parent c128541
commit b4076d4
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 5 deletions.
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
@@ -12,7 +12,7 @@ if(ORTOOLS_FOUND)
   set(HERMES_EXT_LIB_DEPENDENCIES
       ${ORTOOLS_LIBRARIES} ${HERMES_EXT_LIB_DEPENDENCIES})
 else()
-  message(STATUS, "ortools not found")
+  message(STATUS "ortools not found")
 endif()
 
 #------------------------------------------------------------------------------

diff --git a/src/buffer_organizer.cc b/src/buffer_organizer.cc
@@ -488,6 +488,8 @@ bool LocalEnqueueFlushingTask(SharedMemoryContext *context, RpcContext *rpc,
     pool->run(std::bind(FlushBlob, context, rpc, blob_id, filename, offset,
                         async));
     result = true;
+  } else {
+    HERMES_NOT_IMPLEMENTED_YET;
   }
 
   return result;
@@ -519,17 +521,15 @@ void LocalAdjustFlushCount(SharedMemoryContext *context,
                            const std::string &vbkt_name, int adjustment) {
   MetadataManager *mdm = GetMetadataManagerFromContext(context);
   VBucketID id = LocalGetVBucketId(context, vbkt_name.c_str());
+  BeginTicketMutex(&mdm->vbucket_mutex);
   VBucketInfo *info = LocalGetVBucketInfoById(mdm, id);
   if (info) {
     int flush_count = info->async_flush_count.fetch_add(adjustment);
     VLOG(1) << "Flush count on VBucket " << vbkt_name
             << (adjustment > 0 ? "incremented" : "decremented") << " to "
             << flush_count + adjustment << "\n";
-  } else {
-    // TEMP(chogan):
-    LOG(FATAL) << "No VBucketInfo for " << vbkt_name
-               << " when attempting to adjust flush_count\n";
   }
+  EndTicketMutex(&mdm->vbucket_mutex);
 }
 
 void LocalIncrementFlushCount(SharedMemoryContext *context,

diff --git a/src/metadata_management.cc b/src/metadata_management.cc
@@ -1375,11 +1375,13 @@ f32 ScoringFunction(MetadataManager *mdm, Stats *stats) {
 int LocalGetNumOutstandingFlushingTasks(SharedMemoryContext *context,
                                         VBucketID id) {
   MetadataManager *mdm = GetMetadataManagerFromContext(context);
+  BeginTicketMutex(&mdm->vbucket_mutex);
   VBucketInfo *info = LocalGetVBucketInfoById(mdm, id);
   int result = 0;
   if (info) {
     result = info->async_flush_count;
   }
+  EndTicketMutex(&mdm->vbucket_mutex);
 
   return result;
 }
@@ -1412,6 +1414,10 @@ bool LocalLockBlob(SharedMemoryContext *context, BlobID blob_id) {
       if (!ticket) {
         blob_info->last = t.ticket;
       }
+    } else {
+      result = false;
+      ReleaseBlobInfoPtr(mdm);
+      break;
     }
     if (!t.acquired) {
       ReleaseBlobInfoPtr(mdm);