Merge profiling, icommunicate, version, and installation changes

CMakeLists.txt

@@ -8,6 +8,8 @@ option                      (log                "Build DIY with logging"
 option                      (profile            "Build DIY with profiling"                                  OFF)
 option                      (mpi                "Build DIY with mpi"                                        ON)
 option                      (wrapped_mpi        "MPI compiler wrapper requires no further MPI libraries"    OFF)
+option                      (build_examples     "Build DIY examples"                                        ON)
+option                      (build_tests        "Build DIY tests"                                           ON)
 
 # Default to Release
 if                          (NOT CMAKE_BUILD_TYPE)
@@ -69,5 +71,13 @@ endif                       (mpi)
 enable_testing              ()
 include                     (CTest)
 
-add_subdirectory            (examples)
-add_subdirectory            (tests)
+if                          (build_examples)
+    add_subdirectory        (examples)
+endif                       (build_examples)
+
+if                          (build_tests)
+    add_subdirectory        (tests)
+endif                       (build_tests)
+
+install                     (DIRECTORY   ${CMAKE_SOURCE_DIR}/include/diy
+                             DESTINATION ${CMAKE_INSTALL_PREFIX}/include)

include/diy/io/shared.hpp

@@ -0,0 +1,49 @@
+#ifndef DIY_IO_SHARED_HPP
+#define DIY_IO_SHARED_HPP
+
+#include <sstream>
+#include <fstream>
+#include "../mpi.hpp"
+
+namespace diy
+{
+namespace io
+{
+
+class SharedOutFile: public std::ostringstream
+{
+    public:
+                SharedOutFile(std::string filename, diy::mpi::communicator world, int root = 0):
+                    filename_(filename),
+                    world_(world),
+                    root_(root)                     {}
+
+                ~SharedOutFile()                    { close(); }
+
+        void    close()
+        {
+            auto str = this->str();
+            std::vector<char> contents(str.begin(), str.end());
+            if (world_.rank() == root_)
+            {
+                std::vector<std::vector<char>> all_contents;
+                diy::mpi::gather(world_, contents, all_contents, root_);
+
+                // write the file serially
+                std::ofstream out(filename_);
+                for (auto& contents : all_contents)
+                    out.write(contents.data(), contents.size());
+            } else
+                diy::mpi::gather(world_, contents, root_);
+        }
+
+    private:
+        std::string             filename_;
+        diy::mpi::communicator  world_;
+        int                     root_;
+};
+
+}
+}
+
+#endif

include/diy/master.hpp

@@ -670,6 +670,8 @@ void
 diy::Master::
 iexchange_(const ICallback<Block>& f)
 {
+    auto scoped = prof.scoped("iexchange");
+
     // prepare for next round
     incoming_.erase(exchange_round_);
     ++exchange_round_;
@@ -684,10 +686,13 @@ iexchange_(const ICallback<Block>& f)
     {
         for (size_t i = 0; i < size(); i++)     // for all blocks
         {
+
             icommunicate(&iexchange);            // TODO: separate comm thread std::thread t(icommunicate);
             ProxyWithLink cp = proxy(i, &iexchange);
 
+            prof << "callback";
             bool done = f(block<Block>(i), cp);
+            prof >> "callback";
 
             int nundeq_after = 0;
             int nunenq_after = 0;
@@ -731,6 +736,7 @@ void
 diy::Master::
 comm_exchange(GidSendOrder& gid_order, IExchangeInfo* iexchange)
 {
+    auto scoped = prof.scoped("comm-exchange");
     send_outgoing_queues(gid_order, false, iexchange);
     while(nudge());                   // kick requests
     check_incoming_queues(iexchange);
@@ -832,6 +838,8 @@ void
 diy::Master::
 icommunicate(IExchangeInfo* iexchange)
 {
+    auto scoped = prof.scoped("icommunicate");
+
     log->debug("Entering icommunicate()");
 
     // lock out other threads
@@ -861,6 +869,8 @@ send_outgoing_queues(GidSendOrder&   gid_order,
                      bool            remote,                     // TODO: are remote and iexchange mutually exclusive? If so, use single enum?
                      IExchangeInfo*  iexchange)
 {
+    auto scoped = prof.scoped("send-outgoing-queues");
+
     while (inflight_sends().size() < gid_order.limit && !gid_order.empty())
     {
         int from = gid_order.pop();
@@ -1071,6 +1081,8 @@ void
 diy::Master::
 check_incoming_queues(IExchangeInfo* iexchange)
 {
+    auto scoped = prof.scoped("check-incoming-queues");
+
     mpi::optional<mpi::status> ostatus = comm_.iprobe(mpi::any_source, mpi::any_tag);
     while (ostatus)
     {

include/diy/mpi/io.hpp

@@ -74,7 +74,9 @@ file(const communicator& comm__, const std::string& filename, int mode)
 : comm_(comm__)
 {
 #ifndef DIY_NO_MPI
-  MPI_File_open(comm__, const_cast<char*>(filename.c_str()), mode, MPI_INFO_NULL, &fh);
+  int ret = MPI_File_open(comm__, const_cast<char*>(filename.c_str()), mode, MPI_INFO_NULL, &fh);
+  if (ret)
+      throw std::runtime_error("DIY cannot open file: " + filename);
 #else
   DIY_UNUSED(comm__);
   DIY_UNUSED(filename);

include/diy/proxy.hpp

@@ -167,6 +167,51 @@ namespace diy
       void*             block_;
       Link*             link_;
       IExchangeInfo*    iexchange_;         // not used for iexchange presently, but later could trigger some special behavior
+
+    public:
+      template<class T>
+      void enqueue(const BlockID&     to,
+              const T&                x,
+              void (*save)(BinaryBuffer&, const T&) = &::diy::save<T>) const
+      {
+          diy::Master::Proxy::enqueue(to, x, save);
+          if (iexchange_)
+              master()->icommunicate(iexchange_);
+      }
+
+      template<class T>
+      void enqueue(const BlockID&     to,
+              const T*                x,
+              size_t                  n,
+              void (*save)(BinaryBuffer&, const T&) = &::diy::save<T>) const
+      {
+          diy::Master::Proxy::enqueue(to, x, n, save);
+          if (iexchange_)
+              master()->icommunicate(iexchange_);
+      }
+
+      template<class T>
+      void dequeue(int                from,
+              T&                      x,
+              void (*load)(BinaryBuffer&, T&) = &::diy::load<T>) const
+      {
+          // TODO: uncomment if necessary, try first without icommunicating on dequeue
+//           if (iexchange_)
+//               master()->icommunicate(iexchange_);
+          diy::Master::Proxy::dequeue(from, x, load);
+      }
+
+      template<class T>
+      void dequeue(int                from,
+              T*                      x,
+              size_t                  n,
+              void (*load)(BinaryBuffer&, T&) = &::diy::load<T>) const
+      {
+          // TODO: uncomment if necessary, try first without icommunicating on dequeue
+//           if (iexchange_)
+//               master()->icommunicate(iexchange_);
+          diy::Master::Proxy::dequeue(from, x, n, load);
+      }
   };
 }                                           // diy namespace
 

include/diy/stats.hpp

@@ -65,14 +65,18 @@ struct Profiler
     void    enter(std::string name)                     { events.push_back(Event(name, true)); }
     void    exit(std::string name)                      { events.push_back(Event(name, false)); }
 
-    void    output(std::ostream& out)
+    void    output(std::ostream& out, std::string prefix = "")
     {
+        if (!prefix.empty())
+            prefix += " ";
+
         for (size_t i = 0; i < events.size(); ++i)
         {
             const Event& e = events[i];
             auto time = std::chrono::duration_cast<std::chrono::microseconds>(e.stamp - start).count();
 
-            fmt::print(out, "{:02d}:{:02d}:{:02d}.{:06d} {}{}\n",
+            fmt::print(out, "{}{:02d}:{:02d}:{:02d}.{:06d} {}{}\n",
+                            prefix,
                             time/1000000/60/60,
                             time/1000000/60 % 60,
                             time/1000000 % 60,
@@ -103,7 +107,7 @@ struct Profiler
     void    enter(const std::string&)                   {}
     void    exit(const std::string&)                    {}
 
-    void    output(std::ostream&)                       {}
+    void    output(std::ostream&, std::string = "")     {}
     void    clear()                                     {}
 
     Scoped  scoped(std::string)                         { return Scoped(); }

include/diy/version.hpp

@@ -0,0 +1,8 @@
+#ifndef DIY_VERSION_HPP
+#define DIY_VERSION_HPP
+
+#define DIY_VERSION_MAJOR 3
+#define DIY_VERSION_MINOR 5
+#define DIY_VERSION_PATCH dev0
+
+#endif

tests/CMakeLists.txt

@@ -46,6 +46,9 @@ target_link_libraries       (serialization-test ${libraries})
 add_executable              (two-masters         two-masters.cpp)
 target_link_libraries       (two-masters         ${libraries})
 
+add_executable              (shared-output         shared-output.cpp)
+target_link_libraries       (shared-output         ${libraries})
+
 foreach                     (b 2 4 8 64)
     add_test                (NAME kd-tree-test-b${b}
                              COMMAND kd-tree-test -b ${b}

tests/io.cpp

@@ -47,6 +47,17 @@ TEST_CASE("Test BOV", "[io]")
     }
 }
 
+TEST_CASE("Test MPI-IO", "[io]")
+{
+    diy::mpi::communicator world;
+
+    SECTION("file doesn't exist")
+    {
+        REQUIRE_THROWS_AS  (diy::mpi::io::file(world, "non-existent-file", diy::mpi::io::file::rdonly), std::runtime_error);
+        REQUIRE_THROWS_WITH(diy::mpi::io::file(world, "non-existent-file", diy::mpi::io::file::rdonly), Catch::Contains("DIY cannot open file"));
+    }
+}
+
 int main()
 {
   diy::mpi::environment env;

tests/shared-output.cpp

@@ -0,0 +1,12 @@
+#include <diy/mpi.hpp>
+#include <diy/io/shared.hpp>
+
+int main(int argc, char** argv)
+{
+    diy::mpi::environment   env(argc, argv);
+    diy::mpi::communicator  world;
+
+    diy::io::SharedOutFile out("shared-output.txt", world);
+
+    out << world.rank() << " out of " << world.size() << std::endl;
+}