libpressio version 0.55.0

Major Changes

+ launch modules are now user extendable with the
  libpressio_ext/launch/external_launch.h header this will allow
  users to provide their own process launching mechanisms where
  traditional methods are blocked/disallowed.
+ Support for remote metrics servers which use a JSON protocol was
  added.  This enables servers that use MPI to be nested beneath MPI and
  allows for metrics servers that have long startup times to remain
  running and respond to multiple metrics requests.  The protocol format
  is still unstable and may change.
+ New meta-IO module called `copy_template` which copies a template from
  another file before dispatching to another IO module for writing.
  This enables users to copy most of an HDF5 file and overwrite only a
  portion of it when using it with external metrics.
+ POSSIBLE BREAKING CHANGE, `external:command` is now a write only
  setting.  If the module supports it, the corresponding option is now
  external:commands which is a std::vector<std::string>.  The old name
  sets this new parameter by splitting on spaces.
+ the external metrics module gained the ability to choose to intercept
  pressio_compressor_compress_many or pressio_compressor_compress via
  `external:use_many`
+ New higher level python interface `libpressio.py`  It's interface is
  also unstable and may change frequently.

Minor Changes

+ Support for newer versions of mgard which use mgard::mgard_library as
  the cmake target

Bug Fixes

+ fixed some static analysis warnings in the test cases.

Author: robertu94

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.13 FATAL_ERROR)
-project(libpressio VERSION "0.54.1" LANGUAGES CXX C)
+project(libpressio VERSION "0.55.0" LANGUAGES CXX C)
 
 #correct was to set a default build type
 # https://blog.kitware.com/cmake-and-the-default-build-type/
@@ -53,11 +53,11 @@ add_library(libpressio
   ./src/plugins/compressors/sampling.cc
   ./src/plugins/compressors/resize.cc
   ./src/plugins/compressors/transpose.cc
+  ./src/plugins/launch/external_forkexec.cc
   ./src/plugins/metrics/ks_test.cc
   ./src/plugins/metrics/kth_error.cc
   ./src/plugins/metrics/composite.cc
   ./src/plugins/metrics/external.cc
-  ./src/plugins/metrics/external_forkexec.cc
   ./src/plugins/metrics/metrics_base.cc
   ./src/plugins/metrics/size.cc
   ./src/plugins/metrics/time.cc
@@ -67,6 +67,7 @@ add_library(libpressio
   ./src/plugins/metrics/printer.cc
   ./src/plugins/metrics/region_of_interest.cc
   ./src/plugins/metrics/noop.cc
+  ./src/plugins/io/copy_template.cc
   ./src/plugins/io/posix.cc
   ./src/plugins/io/noop.cc
   ./src/plugins/io/csv.cc
@@ -177,7 +178,11 @@ if(LIBPRESSIO_HAS_MGARD)
     PRIVATE
     ${CMAKE_CURRENT_SOURCE_DIR}/src/plugins/compressors/mgard_plugin.cc
     )
+  if(TARGET mgard::mgard)
   target_link_libraries(libpressio PRIVATE mgard::mgard)
+  elseif(TARGET mgard::mgard_library)
+    target_link_libraries(libpressio PRIVATE mgard::mgard_library)
+  endif()
 endif()
 
 option(LIBPRESSIO_HAS_ZFP "build the ZFP plugin" OFF)
@@ -281,7 +286,7 @@ if(LIBPRESSIO_HAS_MPI)
   find_package(MPI COMPONENTS CXX)
   target_sources(libpressio
     PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/src/plugins/metrics/external_mpispawn.cc
+    ${CMAKE_CURRENT_SOURCE_DIR}/src/plugins/launch/external_mpispawn.cc
     )
   target_link_libraries(libpressio PRIVATE MPI::MPI_CXX)
 endif()
@@ -341,6 +346,19 @@ if(LIBPRESSIO_HAS_LIBDISTRIBUTED)
     )
 endif()
 
+option(LIBPRESSIO_HAS_REMOTELAUNCH "build the remote external launch plugin" OFF)
+if(LIBPRESSIO_HAS_REMOTELAUNCH)
+  set(LIBPRESSIO_FEATURES "${LIBPRESSIO_FEATURES} remotelaunch")
+  find_package(nlohmann_json REQUIRED)
+  find_package(CURL REQUIRED)
+  target_link_libraries(libpressio PRIVATE nlohmann_json::nlohmann_json CURL::libcurl)
+  target_sources(libpressio
+    PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/src/plugins/launch/external_remotelaunch.cc
+    )
+
+endif()
+
 configure_file(
   ${CMAKE_CURRENT_SOURCE_DIR}/src/pressio_version.h.in
   ${CMAKE_CURRENT_BINARY_DIR}/include/pressio_version.h

src/plugins/metrics/external_launch.h renamed to include/libpressio_ext/launch/external_launch.h

@@ -27,7 +27,7 @@ struct extern_proc_results {
 
 struct libpressio_launch_plugin: public pressio_configurable, public pressio_errorable {
   virtual ~libpressio_launch_plugin()=default;
-  virtual extern_proc_results launch(std::string const&, std::string const&) const =0;
+  virtual extern_proc_results launch(std::vector<std::string> const&) const =0;
   virtual std::unique_ptr<libpressio_launch_plugin> clone() const = 0;
 };
 

src/plugins/io/copy_template.cc

@@ -0,0 +1,75 @@
+#include <iostream>
+#include <fstream>
+#include "pressio_compressor.h"
+#include "libpressio_ext/io/posix.h"
+#include "libpressio_ext/cpp/pressio.h"
+#include "libpressio_ext/cpp/options.h"
+#include "libpressio_ext/cpp/data.h"
+#include "libpressio_ext/cpp/io.h"
+#include "std_compat/memory.h"
+
+struct copy_template_io : public libpressio_io_plugin {
+  virtual struct pressio_data* read_impl(struct pressio_data* data) override {
+    return impl->read(data);
+  }
+  virtual int write_impl(struct pressio_data const* data) override{
+    if(not template_path.empty()) {
+      if(copy_template()) return error_code();
+    }
+    return impl->write(data);
+  }
+  virtual int set_options_impl(struct pressio_options const& options) override{
+    get_meta(options, "copy_template:io", io_plugins(), impl_id, impl);
+    get(options, "copy_template:template_path", &template_path);
+    get(options, "io:path", &path);
+    return 0;
+  }
+  virtual struct pressio_options get_options_impl() const override{
+    pressio_options opts;
+    set_meta(opts, "copy_template:io", impl_id, impl);
+    set(opts, "copy_template:template_path", template_path);
+    set(opts, "io:path", path);
+    return opts;
+  }
+  virtual struct pressio_options get_configuration_impl() const override{
+    pressio_options opts;
+    set(opts, "pressio:thread_safe",  static_cast<int32_t>(pressio_thread_safety_single));
+    return opts;
+  }
+
+
+  int patch_version() const override{ 
+    return 1;
+  }
+  virtual const char* version() const override{
+    return "0.0.1";
+  }
+  const char* prefix() const override {
+    return "copy_template";
+  }
+
+  std::shared_ptr<libpressio_io_plugin> clone() override {
+    return compat::make_unique<copy_template_io>(*this);
+  }
+
+  void set_name_impl(std::string const& new_name) override {
+    impl->set_name(new_name + "/" + impl->prefix());
+  }
+
+  private:
+  int copy_template() {
+    std::ifstream template_file(template_path, std::ios::binary);
+    std::ofstream output_file(path, std::ios::binary);
+    if(!template_file) return set_error(1, "template_file does not exist");
+    if(!output_file) return set_error(2, "output_file does not exist");
+    output_file << template_file.rdbuf();
+    return 0;
+  }
+
+  std::string path;
+  std::string template_path;
+  std::string impl_id = "posix";
+  pressio_io impl = io_plugins().build("posix");
+};
+
+static pressio_register io_copy_template_plugin(io_plugins(), "copy_template", [](){ return compat::make_unique<copy_template_io>(); });

src/plugins/metrics/external_forkexec.cc renamed to src/plugins/launch/external_forkexec.cc

@@ -1,4 +1,4 @@
-#include "external_launch.h"
+#include "libpressio_ext/launch/external_launch.h"
 #include <memory>
 #include <sstream>
 #include <unistd.h>
@@ -7,7 +7,7 @@
 #include "std_compat/memory.h"
 
 struct external_forkexec: public libpressio_launch_plugin {
-extern_proc_results launch(std::string const& full_command, std::string const& workdir) const override {
+extern_proc_results launch(std::vector<std::string> const& full_command) const override {
       extern_proc_results results;
 
       //create the pipe for stdout
@@ -47,12 +47,11 @@ extern_proc_results launch(std::string const& full_command, std::string const& w
             exit(-2);
           }
 
-          std::istringstream command_stream(full_command);
-          std::vector<std::string> args_mem(
-              std::istream_iterator<std::string>{command_stream},
-              std::istream_iterator<std::string>());
           std::vector<char*> args;
-          std::transform(std::begin(args_mem), std::end(args_mem),
+          for(auto const& command: commands) {
+            args.push_back(const_cast<char*>(command.c_str()));
+          }
+          std::transform(std::begin(full_command), std::end(full_command),
               std::back_inserter(args), [](std::string const& s){return const_cast<char*>(s.c_str());});
           args.push_back(nullptr);
           if(args.front() != nullptr) {
@@ -107,10 +106,26 @@ extern_proc_results launch(std::string const& full_command, std::string const& w
     return "forkexec";
   }
 
+  int set_options(pressio_options const& options) override {
+    get(options, "external:workdir", &workdir);
+    get(options, "external:commands", &commands);
+    return 0;
+  }
+
+  pressio_options get_options() const override {
+    pressio_options options;
+    set(options, "external:workdir", workdir);
+    set(options, "external:commands", commands);
+    return options;
+  }
+
 
   std::unique_ptr<libpressio_launch_plugin> clone() const override {
     return compat::make_unique<external_forkexec>(*this);
   }
+
+  std::string workdir=".";
+  std::vector<std::string> commands;
 };
 
 static pressio_register launch_forkexec_plugin(launch_plugins(), "forkexec", [](){ return compat::make_unique<external_forkexec>();});

src/plugins/metrics/external_mpispawn.cc renamed to src/plugins/launch/external_mpispawn.cc

@@ -1,20 +1,19 @@
-#include "external_launch.h"
+#include "libpressio_ext/launch/external_launch.h"
 #include <iterator>
 #include <memory>
 #include <sstream>
 #include <mpi.h>
 #include "std_compat/memory.h"
 
 struct external_mpispawn: public libpressio_launch_plugin {
-extern_proc_results launch(std::string const& full_command, std::string const& workdir) const override {
+extern_proc_results launch(std::vector<std::string> const& full_command) const override {
       extern_proc_results results;
 
-      std::istringstream command_stream(full_command);
-      std::vector<std::string> args_mem(
-          std::istream_iterator<std::string>{command_stream},
-          std::istream_iterator<std::string>());
       std::vector<char*> args;
-      std::transform(std::begin(args_mem), std::end(args_mem),
+      for (auto const& command: commands) {
+        args.push_back(const_cast<char*>(command.c_str()));
+      }
+      std::transform(std::begin(full_command), std::end(full_command),
           std::back_inserter(args), [](std::string const& s){return const_cast<char*>(s.c_str());});
       args.push_back(nullptr);
 
@@ -49,10 +48,25 @@ extern_proc_results launch(std::string const& full_command, std::string const& w
     return "mpispawn";
   }
 
+  int set_options(pressio_options const& options) override {
+    get(options, "external:workdir", &workdir);
+    get(options, "external:commands", &commands);
+    return 0;
+  }
+
+  pressio_options get_options() const override {
+    pressio_options options;
+    set(options, "external:workdir", workdir);
+    set(options, "external:commands", commands);
+    return options;
+  }
+
   std::unique_ptr<libpressio_launch_plugin> clone() const override {
     return compat::make_unique<external_mpispawn>(*this);
   }
 
+  std::string workdir=".";
+  std::vector<std::string> commands;
 };
 
 static pressio_register launch_spawn_plugin(launch_plugins(), "mpispawn", [](){ return compat::make_unique<external_mpispawn>();});

src/plugins/launch/external_remotelaunch.cc

@@ -0,0 +1,127 @@
+#include "libpressio_ext/launch/external_launch.h"
+#include <mutex>
+#include "std_compat/memory.h"
+#include <nlohmann/json.hpp>
+#include <curl/curl.h>
+#include <curl/easy.h>
+
+static size_t write_to_std_string(char* txt, size_t size, size_t nelms, void* user_data) {
+  std::string* user_str = reinterpret_cast<std::string*>(user_data);
+  user_str->append(txt, size*nelms);
+  return size*nelms;
+}
+
+static std::mutex libpressio_curl_init_lock;
+struct libpressio_external_curl_manager {
+  libpressio_external_curl_manager() {
+    curl_global_init(CURL_GLOBAL_ALL);
+  }
+  ~libpressio_external_curl_manager() {
+    curl_global_cleanup();
+  }
+
+  static std::shared_ptr<libpressio_external_curl_manager> get_library() {
+    std::lock_guard<std::mutex> guard(libpressio_curl_init_lock);
+    static std::weak_ptr<libpressio_external_curl_manager> weak{};
+    if(auto observed = weak.lock())
+    {
+      return observed;
+    } else {
+      auto library = std::make_shared<libpressio_external_curl_manager>();
+      weak = library;
+      return library;
+    }
+  }
+};
+
+
+
+struct external_remote: public libpressio_launch_plugin {
+  external_remote(std::shared_ptr<libpressio_external_curl_manager>&& curl_singleton):
+    curl_singleton(curl_singleton) {}
+
+extern_proc_results launch(std::vector<std::string> const& full_command) const override {
+      extern_proc_results results;
+      nlohmann::json request;
+      request["args"] = full_command;
+      std::string request_str = request.dump();
+      std::string response_str;
+      char errbuf[CURL_ERROR_SIZE] = {0};
+
+      CURLcode ret;
+      CURL *hnd;
+      curl_slist* headers = nullptr;
+      headers = curl_slist_append(headers, "Content-Type: application/json");
+      hnd = curl_easy_init();
+      curl_easy_setopt(hnd, CURLOPT_BUFFERSIZE, 102400L);
+      curl_easy_setopt(hnd, CURLOPT_URL, connection_string.c_str());
+      curl_easy_setopt(hnd, CURLOPT_NOPROGRESS, 1L);
+      curl_easy_setopt(hnd, CURLOPT_USERAGENT, "curl/7.72.0");
+      curl_easy_setopt(hnd, CURLOPT_MAXREDIRS, 50L);
+      curl_easy_setopt(hnd, CURLOPT_HTTP_VERSION, (long)CURL_HTTP_VERSION_2TLS);
+      curl_easy_setopt(hnd, CURLOPT_TCP_KEEPALIVE, 1L);
+      curl_easy_setopt(hnd, CURLOPT_POST, 1L);
+      curl_easy_setopt(hnd, CURLOPT_POSTFIELDS, request_str.c_str());
+      curl_easy_setopt(hnd, CURLOPT_POSTFIELDSIZE, request_str.size());
+      curl_easy_setopt(hnd, CURLOPT_WRITEFUNCTION, &write_to_std_string);
+      curl_easy_setopt(hnd, CURLOPT_WRITEDATA, &response_str);
+      curl_easy_setopt(hnd, CURLOPT_HTTPHEADER, headers);
+      curl_easy_setopt(hnd, CURLOPT_ERRORBUFFER, errbuf);
+
+      ret = curl_easy_perform(hnd);
+
+      if(ret != CURLE_OK) {
+        results.error_code = ret;
+        if(strlen(errbuf)) {
+          results.proc_stderr = std::string(errbuf);
+        } else {
+          results.proc_stderr = curl_easy_strerror(ret);
+        }
+      } else {
+        try {
+          nlohmann::json response = nlohmann::json::parse(response_str);
+          results.proc_stdout = response["stdout"].get<std::string>();
+          results.proc_stderr = response["stderr"].get<std::string>();
+          results.return_code = response["return_code"].get<int>();
+        } catch(nlohmann::json::exception const& e) {
+          results.proc_stdout = "";
+          results.proc_stderr = response_str + "\n\n" +  e.what();
+          results.error_code = -1;
+        }
+      }
+
+      curl_easy_cleanup(hnd);
+      curl_slist_free_all(headers);
+      hnd = NULL;
+
+
+      return results;
+    }
+  const char* prefix() const override {
+    return "remote";
+  }
+
+  int set_options(pressio_options const& options) override {
+    get(options, "external:connection_string", &connection_string);
+    return 0;
+  }
+
+  pressio_options get_options() const override {
+    pressio_options options;
+    set(options, "external:connection_string", connection_string);
+    return options;
+  }
+
+  std::unique_ptr<libpressio_launch_plugin> clone() const override {
+    return compat::make_unique<external_remote>(*this);
+  }
+
+  std::string connection_string;
+  std::shared_ptr<libpressio_external_curl_manager> curl_singleton;
+};
+
+static pressio_register launch_spawn_plugin(launch_plugins(), "remote", [](){
+    return compat::make_unique<external_remote>(
+        libpressio_external_curl_manager::get_library()
+        );
+});