Bug fixes and improvements related to Hermes VPIC proxy

adapter/src/hermes/adapter/constants.h

@@ -58,4 +58,13 @@ const char kPathDelimiter = ',';
 const char* kAdapterDefaultMode = "DEFAULT";
 const char* kAdapterBypassMode = "BYPASS";
 const char* kAdapterScratchMode = "SCRATCH";
+
+/**
+ * If the \c HERMES_STOP_DAEMON environment variable is unset or has a non-zero
+ * value, the adapter client will kill the running Hermes daemon when it
+ * finishes execution.
+ *
+ * Default value: \c 1
+ */
+const char* kStopDaemon = "HERMES_STOP_DAEMON";
 #endif  // HERMES_ADAPTER_CONSTANTS_H

adapter/src/hermes/adapter/mpiio/metadata_manager.h

@@ -83,6 +83,8 @@ class MetadataManager {
       char* hermes_config = getenv(kHermesConf);
       MPI_Comm_rank(MPI_COMM_WORLD, &rank);
       MPI_Comm_size(MPI_COMM_WORLD, &comm_size);
+      // TODO(chogan): Need a better way to distinguish between client and
+      // daemon. https://github.com/HDFGroup/hermes/issues/206
       if (comm_size > 1) {
         hermes = hermes::InitHermesClient(hermes_config);
       } else {
@@ -98,15 +100,7 @@ class MetadataManager {
    */
   void FinalizeHermes() {
     if (ref == 1) {
-      if (this->comm_size > 1) {
-        MPI_Barrier(MPI_COMM_WORLD);
-        if (this->rank == 0) {
-          hermes->RemoteFinalize();
-        }
-        hermes->Finalize();
-      } else {
-        hermes->Finalize(true);
-      }
+      hermes->FinalizeClient();
     }
     ref--;
   }

adapter/src/hermes/adapter/mpiio/mpiio.cc

@@ -70,7 +70,6 @@ int simple_open(MPI_Comm &comm, const char *path, int &amode, MPI_Info &info,
     }
     stat.info = info;
     stat.comm = comm;
-    mdm->InitializeHermes();
     hapi::Context ctx;
     stat.st_bkid = std::make_shared<hapi::Bucket>(path, mdm->GetHermes(), ctx);
     mdm->Create(fh, stat);
@@ -612,7 +611,6 @@ int HERMES_DECL(MPI_File_close)(MPI_File *fh) {
           INTERCEPTOR_LIST->hermes_flush_exclusion.erase(filename);
         }
         existing.first.st_bkid->Destroy(ctx);
-        mdm->FinalizeHermes();
         if (existing.first.a_mode & MPI_MODE_DELETE_ON_CLOSE) {
           fs::remove(filename);
         }

adapter/src/hermes/adapter/stdio/metadata_manager.h

@@ -87,6 +87,8 @@ class MetadataManager {
       if (this->is_mpi) {
         MPI_Comm_rank(MPI_COMM_WORLD, &rank);
         MPI_Comm_size(MPI_COMM_WORLD, &comm_size);
+        // TODO(chogan): Need a better way to distinguish between client and
+        // daemon. https://github.com/HDFGroup/hermes/issues/206
         if (comm_size > 1) {
           hermes = hermes::InitHermesClient(hermes_config);
         } else {
@@ -108,10 +110,17 @@ class MetadataManager {
     if (ref == 1) {
       if (this->is_mpi) {
         MPI_Barrier(MPI_COMM_WORLD);
-        if (this->rank == 0) {
-          hermes->RemoteFinalize();
+        char *stop_daemon = getenv(kStopDaemon);
+        bool shutdown_daemon = true;
+
+        if (stop_daemon && stop_daemon[0] == '0') {
+          HERMES_NOT_IMPLEMENTED_YET;
+          // TODO(chogan): The Hermes core needs a few tweaks before it can
+          // support this feature. https://github.com/HDFGroup/hermes/issues/181
+          shutdown_daemon = false;
         }
-        hermes->Finalize();
+
+        hermes->FinalizeClient(shutdown_daemon);
       } else {
         hermes->Finalize(true);
       }

adapter/src/hermes/adapter/stdio/stdio.cc

@@ -188,7 +188,7 @@ size_t write_internal(std::pair<AdapterStat, bool> &existing, const void *ptr,
       FileStruct(mdm->Convert(fp), existing.first.st_ptr, total_size));
   size_t data_offset = 0;
   auto filename = existing.first.st_bkid->GetName();
-  LOG(INFO) << "Mapping for read has " << mapping.size() << " mapping."
+  LOG(INFO) << "Mapping for write has " << mapping.size() << " mapping."
             << std::endl;
   for (const auto &item : mapping) {
     hapi::Context ctx;

adapter/test/hermes_daemon.cc

@@ -19,6 +19,6 @@ int main(int argc, char* argv[]) {
     hermes_config = getenv(kHermesConf);
   }
   auto hermes = hermes::InitHermesDaemon(hermes_config);
-  hermes->Finalize();
+  hermes->RunDaemon();
   MPI_Finalize();
 }

benchmarks/CMakeLists.txt

@@ -1,15 +1,14 @@
 include_directories(
   ${PROJECT_SOURCE_DIR}/src/api
+  ${PROJECT_SOURCE_DIR}/test
 )
 
-add_executable(mdm_bench mdm_bench.cc)
-target_link_libraries(mdm_bench hermes MPI::MPI_CXX
-  $<$<BOOL:${HERMES_RPC_THALLIUM}>:thallium>)
-target_compile_definitions(mdm_bench
-  PRIVATE $<$<BOOL:${HERMES_RPC_THALLIUM}>:HERMES_RPC_THALLIUM>)
+set(BENCHMARKS mdm_bench dpe_bench)
 
-add_executable(dpe_bench dpe_bench.cc)
-target_link_libraries(dpe_bench hermes MPI::MPI_CXX
-  $<$<BOOL:${HERMES_RPC_THALLIUM}>:thallium>)
-target_compile_definitions(dpe_bench
-  PRIVATE $<$<BOOL:${HERMES_RPC_THALLIUM}>:HERMES_RPC_THALLIUM>)
+foreach(benchmark ${BENCHMARKS})
+  add_executable(${benchmark} ${benchmark}.cc)
+  target_link_libraries(${benchmark} hermes MPI::MPI_CXX
+    $<$<BOOL:${HERMES_RPC_THALLIUM}>:thallium>)
+  target_compile_definitions(${benchmark}
+    PRIVATE $<$<BOOL:${HERMES_RPC_THALLIUM}>:HERMES_RPC_THALLIUM>)
+endforeach()

src/api/bucket.h

@@ -224,6 +224,10 @@ Status Bucket::PlaceBlobs(std::vector<PlacementSchema> &schemas,
               << "'" << 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;

src/api/hermes.cc

@@ -25,6 +25,8 @@
 
 namespace hermes {
 
+std::vector<DeviceID> RoundRobinState::devices_;
+
 namespace api {
 
 int Context::default_buffer_organizer_retries;
@@ -112,10 +114,19 @@ void Hermes::Finalize(bool force_rpc_shutdown) {
   is_initialized = false;
 }
 
+void Hermes::FinalizeClient(bool stop_daemon) {
+  hermes::FinalizeClient(&context_, &rpc_, &comm_, &trans_arena_, stop_daemon);
+}
+
 void Hermes::RemoteFinalize() {
   hermes::RpcCall<void>(&rpc_, rpc_.node_id, "RemoteFinalize");
 }
 
+void Hermes::RunDaemon() {
+  hermes::RunDaemon(&context_, &rpc_, &comm_, &trans_arena_,
+                    shmem_name_.c_str());
+}
+
 }  // namespace api
 
 ArenaInfo GetArenaInfo(Config *config) {
@@ -257,7 +268,8 @@ std::shared_ptr<api::Hermes> InitHermes(Config *config, bool is_daemon,
   }
   bool create_shared_files = (comm.proc_kind == ProcessKind::kHermes &&
                               comm.first_on_node);
-  InitFilesForBuffering(&context, create_shared_files);
+  InitFilesForBuffering(&context, create_shared_files, comm.node_id,
+                        comm.first_on_node);
 
   WorldBarrier(&comm);
 
@@ -301,6 +313,13 @@ std::shared_ptr<api::Hermes> InitHermes(Config *config, bool is_daemon,
     config->num_buffer_organizer_retries;
   api::Context::default_placement_policy = config->default_placement_policy;
 
+  RoundRobinState::devices_.reserve(config->num_devices);
+  for (DeviceID id = 0; id < config->num_devices; ++id) {
+    if (GetNumBuffersAvailable(&result->context_, id)) {
+      RoundRobinState::devices_.push_back(id);
+    }
+  }
+
   InitRpcClients(&result->rpc_);
 
   // NOTE(chogan): Can only initialize the neighborhood Targets once the RPC

src/api/hermes.h

@@ -71,7 +71,9 @@ class Hermes {
   int GetNumProcesses();
   void *GetAppCommunicator();
   void Finalize(bool force_rpc_shutdown = false);
+  void FinalizeClient(bool stop_daemon = true);
   void RemoteFinalize();
+  void RunDaemon();
 
   bool BucketContainsBlob(const std::string &bucket_name,
                           const std::string &blob_name);