WIP --- Temporal reduction profile

alg/teca_temporal_reduction.cxx

@@ -14,6 +14,7 @@
 #include <string>
 #include <cmath>
 #include <typeinfo>
+#include <chrono>
 
 #if defined(TECA_HAS_BOOST)
 #include <boost/program_options.hpp>
@@ -28,14 +29,15 @@
 using namespace teca_variant_array_util;
 using allocator = teca_variant_array::allocator;
 
+using microseconds_t = std::chrono::duration<double, std::chrono::microseconds::period>;
 
 
 // PIMPL idiom hides internals
 // defines the API for reduction operators
 class teca_cpp_temporal_reduction::internals_t
 {
 public:
-    internals_t() {}
+    internals_t() : m_runtime(0) {}
     ~internals_t() {}
 
     /** check if the passed array contains integer data, if so deep-copy to
@@ -68,6 +70,7 @@ class teca_cpp_temporal_reduction::internals_t
     teca_metadata metadata;
     std::vector<time_interval> indices;
     std::map<std::thread::id, std::map<std::string, p_reduction_operator>> operation;
+    std::atomic<long> m_runtime;
 };
 
 // --------------------------------------------------------------------------
@@ -2448,6 +2451,9 @@ const_p_teca_dataset teca_cpp_temporal_reduction::execute(
 {
     (void)port;
 
+    std::chrono::high_resolution_clock::time_point t0, t1;
+    t0 = std::chrono::high_resolution_clock::now();
+
     // get the requested ind
     unsigned long req_id[2] = {0};
     std::string request_key;
@@ -2463,17 +2469,6 @@ const_p_teca_dataset teca_cpp_temporal_reduction::execute(
     int device_id = -1;
     req_in.get("device_id", device_id);
 
-    if (this->get_verbose())
-    {
-        std::cerr << teca_parallel_id()
-            << "teca_cpp_temporal_reduction::execute request "
-            << req_id[0] << " device " << device_id
-            << " (" << this->internals->indices[req_id[0]].start_index
-            << " - " << this->internals->indices[req_id[0]].end_index
-            << "), reducing " << data_in.size() << ", "
-            << streaming << " remain" << std::endl;
-    }
-
 #if defined(TECA_HAS_CUDA)
     if (device_id >= 0)
     {
@@ -2600,5 +2595,31 @@ const_p_teca_dataset teca_cpp_temporal_reduction::execute(
         mesh_out->set_time(this->internals->indices[req_id[0]].time);
     }
 
+    t1 = std::chrono::high_resolution_clock::now();
+    microseconds_t dt(t1 - t0);
+
+    this->internals->m_runtime.fetch_add( dt.count(), std::memory_order_relaxed);
+
+    if (this->get_verbose())
+    {
+        std::ostringstream oss;
+        oss << teca_parallel_id()
+            << "teca_cpp_temporal_reduction::execute request "
+            << req_id[0] << " device " << device_id
+            << " (" << this->internals->indices[req_id[0]].start_index
+            << " - " << this->internals->indices[req_id[0]].end_index
+            << "), reducing " << data_in.size() << ", "
+            << streaming << " remain (" << ( dt.count() / 1e6 ) << "s)";
+
+        if (!streaming)
+        {
+            oss << " step " << req_id[0] << " runtime: "
+                << this->internals->m_runtime.load(std::memory_order_relaxed) / 1e6 << "s";
+
+            this->internals->m_runtime.store(0, std::memory_order_relaxed);
+        }
+        std::cerr << oss.str() << std::endl;
+    }
+
     return mesh_out;
 }

io/teca_cf_reader.cxx

@@ -28,6 +28,7 @@
 #include <utility>
 #include <memory>
 #include <iomanip>
+#include <chrono>
 
 #if defined(TECA_HAS_MPI)
 #include <mpi.h>
@@ -51,6 +52,8 @@ using std::cerr;
 using namespace teca_variant_array_util;
 using allocator = teca_variant_array::allocator;
 
+using microseconds_t = std::chrono::duration<double, std::chrono::microseconds::period>;
+
 // internals for the cf reader
 class teca_cf_reader_internals
 {
@@ -1040,6 +1043,9 @@ const_p_teca_dataset teca_cf_reader::execute(unsigned int port,
     (void)port;
     (void)input_data;
 
+    std::chrono::high_resolution_clock::time_point t0, t1;
+    t0 = std::chrono::high_resolution_clock::now();
+
     int rank = 0;
 #if defined(TECA_HAS_MPI)
     MPI_Comm comm = this->get_communicator();
@@ -1349,8 +1355,8 @@ const_p_teca_dataset teca_cf_reader::execute(unsigned int port,
     // get the requested target device
     allocator alloc = allocator::malloc;
     allocator tmp_alloc = allocator::malloc;
-#if defined(TECA_HAS_CUDA)
     int device_id = -1;
+#if defined(TECA_HAS_CUDA)
     request.get("device_id", device_id);
     if (device_id >= 0)
     {
@@ -1695,5 +1701,16 @@ const_p_teca_dataset teca_cf_reader::execute(unsigned int port,
         }
     }
 
+    t1 = std::chrono::high_resolution_clock::now();
+    microseconds_t dt(t1 - t0);
+
+    if (this->get_verbose())
+    {
+        std::cerr << teca_parallel_id()
+            << "teca_cf_reader::execute device_id " << device_id << " steps ["
+            << temporal_extent << "] extent [" << extent << "] " << ( dt.count() / 1e6 )
+            << "s" << std::endl;
+    }
+
     return mesh;
 }

io/teca_cf_writer.cxx

@@ -22,21 +22,23 @@
 #include <string>
 #include <unordered_map>
 #include <set>
+#include <chrono>
 
 #if defined(TECA_HAS_BOOST)
 #include <boost/program_options.hpp>
 #endif
 
+using microseconds_t = std::chrono::duration<double, std::chrono::microseconds::period>;
 
 
 class teca_cf_writer::internals_t
 {
 public:
-    internals_t() : mapper(), layout_defined(0)
-    {}
+    internals_t() : mapper(), layout_defined(0), m_runtime(0) {}
 
     p_teca_cf_time_step_mapper mapper;
     int layout_defined;
+    std::atomic<long> m_runtime;
 };
 
 
@@ -681,6 +683,9 @@ const_p_teca_dataset teca_cf_writer::execute(unsigned int port,
     (void)port;
     (void)request;
 
+    std::chrono::high_resolution_clock::time_point t0, t1;
+    t0 = std::chrono::high_resolution_clock::now();
+
     int rank = 0;
 #if defined(TECA_HAS_MPI)
     int n_ranks = 1;
@@ -696,6 +701,7 @@ const_p_teca_dataset teca_cf_writer::execute(unsigned int port,
 
     // get the number of datasets in hand. these will be written to one of
     // the files, depending on its time step
+    long n_steps_total = 0;
     long n_indices = input_data.size();
     for (long i = 0; i < n_indices; ++i)
     {
@@ -736,6 +742,8 @@ const_p_teca_dataset teca_cf_writer::execute(unsigned int port,
             return nullptr;
         }
 
+        n_steps_total += temporal_extent[1] - temporal_extent[0] + 1;
+
         // get the layout managers needed to write this extent
         std::vector<p_teca_cf_layout_manager> managers;
         if (this->internals->mapper->get_layout_manager(temporal_extent, managers))
@@ -783,5 +791,38 @@ const_p_teca_dataset teca_cf_writer::execute(unsigned int port,
         }
     }
 
+    // get the requested ind
+    unsigned long req_id[2] = {0};
+    std::string request_key;
+
+    if (request.get("index_request_key", request_key) ||
+        request.get(request_key, req_id))
+    {
+        TECA_FATAL_ERROR("metadata issue. failed to get the requested indices")
+        return nullptr;
+    }
+
+    t1 = std::chrono::high_resolution_clock::now();
+    microseconds_t dt(t1 - t0);
+
+    this->internals->m_runtime.fetch_add( dt.count(), std::memory_order_relaxed);
+
+    if (this->get_verbose())
+    {
+        std::ostringstream oss;
+        oss << teca_parallel_id()
+            << "teca_cf_writer::execute request " << req_id[0] << " wrote "
+            << n_steps_total << " steps " << streaming << " remain ("
+            << ( dt.count() / 1e6 ) << "s)";
+        if (!streaming)
+        {
+            oss << " runtime: "
+                << this->internals->m_runtime.load(std::memory_order_relaxed) / 1e6 << "s";
+
+            this->internals->m_runtime.store(0, std::memory_order_relaxed);
+        }
+        std::cerr << oss.str() << std::endl;
+    }
+
     return nullptr;
 }

test/test_cpp_temporal_reduction_io.cpp

@@ -92,6 +92,7 @@ int main(int argc, char **argv)
     cf_reader->set_z_axis_variable(z_axis_var == "." ? std::string() : z_axis_var);
     cf_reader->set_t_axis_variable(t_axis_var);
     cf_reader->set_files_regex(files_regex);
+    cf_reader->set_verbose(1);
 
     // temporal reduction
     auto reduc = teca_cpp_temporal_reduction::New();