Hf prewrite pause

cpp/skyweaver/MultiFileWriter.cuh

@@ -26,8 +26,7 @@ struct MultiFileWriterConfig{
   std::string prefix;
   std::string extension;
   std::string output_basename;
-
-
+  PreWriteConfig pre_write;
 
   MultiFileWriterConfig() : header_size(4096), max_file_size(2147483647), stokes_mode("I"), output_dir("default/"), prefix(""), extension(""){};
   MultiFileWriterConfig(std::size_t header_size, std::size_t max_file_size, std::string stokes_mode, std::string output_dir, std::string prefix, std::string extension) : header_size(header_size), max_file_size(max_file_size), stokes_mode(stokes_mode), output_dir(output_dir), prefix(prefix), extension(extension), output_basename(""){ };
@@ -59,13 +58,13 @@ template <typename VectorType>
 class MultiFileWriter
 {
 public:
-
   using CreateStreamCallBackType = std::function<std::unique_ptr<FileOutputStream>(MultiFileWriterConfig const&,
                                   ObservationHeader const&,
                                   VectorType const&,
                                   std::size_t)>;
 
   public:
+     using PreWriteCallback = std::function<void(std::size_t, MultiFileWriterConfig const&)>;
     /**
      * @brief Construct a new Multi File Writer object
      *
@@ -75,7 +74,9 @@ public:
      */
     // MultiFileWriter(PipelineConfig const& config, std::string tag = "");
     MultiFileWriter(PipelineConfig const& config, std::string tag, CreateStreamCallBackType create_stream_callback);
+    MultiFileWriter(PipelineConfig const& config, std::string tag, CreateStreamCallBackType create_stream_callback, PreWriteCallback pre_write_callback);
     MultiFileWriter(MultiFileWriterConfig config, std::string tag, CreateStreamCallBackType create_stream_callback);
+    MultiFileWriter(MultiFileWriterConfig config, std::string tag, CreateStreamCallBackType create_stream_callback, PreWriteCallback pre_write_callback);
     MultiFileWriter(MultiFileWriter const&) = delete;
 
     /**
@@ -124,6 +125,7 @@ public:
     std::string get_extension(VectorType const& stream_data);
     CreateStreamCallBackType _create_stream_callback;
     MultiFileWriterConfig _config;
+    PreWriteCallback _pre_write_callback;
     std::string _tag;
     ObservationHeader _header;
     std::map<std::size_t, std::unique_ptr<FileOutputStream>> _file_streams;
@@ -136,4 +138,4 @@ public:
 #include "skyweaver/detail/MultiFileWriter.cu"
 #include "skyweaver/detail/file_writer_callbacks.cpp"
 
-#endif // SKYWEAVER_MULTIFILEWRITER_CUH
+#endif // SKYWEAVER_MULTIFILEWRITER_CUH

cpp/skyweaver/PipelineConfig.hpp

@@ -9,6 +9,19 @@
 
 namespace skyweaver
 {
+  struct WaitConfig
+  {
+    int iterations;
+    int sleep_time;
+    std::size_t min_free_space;
+  };
+
+  struct PreWriteConfig
+  {
+    bool is_enabled;
+    WaitConfig wait;
+  };
+
 
 /**
  * @brief      Class for wrapping the skyweaver pipeline configuration.
@@ -153,6 +166,11 @@ class PipelineConfig
      */
     DedispersionPlan const& ddplan() const;
 
+    /**
+     * @brief      configures wait for filesystem space
+     */
+    void configure_wait(std::string argument);
+
     /**
      * @brief      Enable/disable incoherent dedispersion based fscrunch after
      * beamforming
@@ -209,6 +227,11 @@ class PipelineConfig
         return SKYWEAVER_CB_NSAMPLES_PER_BLOCK;
     }
 
+    PreWriteConfig pre_write_config() const
+    {
+        return _pre_write_config;
+    }
+
     /**
      * @brief Return the total number of samples to read from file in each gulp.
      *
@@ -323,6 +346,7 @@ class PipelineConfig
     }
 
   private:
+    std::size_t convertMemorySize(const std::string& str) const;
     void calculate_channel_frequencies() const;
     void update_power_offsets_and_scalings();
 
@@ -350,6 +374,7 @@ class PipelineConfig
     float _output_level;
     DedispersionPlan _ddplan;
     mutable std::vector<double> _channel_frequencies;
+    PreWriteConfig _pre_write_config;
 };
 
 } // namespace skyweaver

cpp/skyweaver/detail/MultiFileWriter.cu

@@ -52,6 +52,24 @@ MultiFileWriter<VectorType>::MultiFileWriter(PipelineConfig const& config,
     writer_config.base_output_dir = config.output_dir();
 
     _config = writer_config;
+    _pre_write_callback = nullptr;
+}
+
+ template <typename VectorType>
+MultiFileWriter<VectorType>::MultiFileWriter(PipelineConfig const& config,
+                                             std::string tag,
+                                             CreateStreamCallBackType create_stream_callback,
+                                             PreWriteCallback pre_write_callback)
+    : _tag(tag), _create_stream_callback(create_stream_callback), _pre_write_callback(pre_write_callback)
+{
+    MultiFileWriterConfig writer_config;
+    writer_config.header_size = config.dada_header_size();
+    writer_config.max_file_size = config.max_output_filesize();
+    writer_config.stokes_mode = config.stokes_mode();
+    writer_config.output_dir = config.output_dir();
+    writer_config.pre_write = config.pre_write_config();
+    _config = writer_config;
+    _config.pre_write = writer_config.pre_write;
 }
 
 template <typename VectorType>
@@ -60,8 +78,19 @@ MultiFileWriter<VectorType>::MultiFileWriter(MultiFileWriterConfig config,
                                              CreateStreamCallBackType create_stream_callback)
     : _config(config), _tag(tag), _create_stream_callback(create_stream_callback)
 {
+   _pre_write_callback = nullptr;
 }
 
+template <typename VectorType>
+MultiFileWriter<VectorType>::MultiFileWriter(MultiFileWriterConfig config,
+                                             std::string tag,
+                                             CreateStreamCallBackType create_stream_callback,
+                                             PreWriteCallback pre_write_callback)
+    : _config(config), _tag(tag), _create_stream_callback(create_stream_callback), _pre_write_callback(pre_write_callback)
+{
+}
+
+
 
 
 template <typename VectorType>
@@ -160,6 +189,11 @@ template <typename VectorType>
 bool MultiFileWriter<VectorType>::operator()(VectorType const& stream_data,
                                              std::size_t stream_idx)
 {
+    std::size_t const data_size = stream_data.size() * sizeof(typename VectorType::value_type);
+    if (_pre_write_callback != nullptr && _config.pre_write.is_enabled)
+    {
+      _pre_write_callback(data_size, _config);
+    }
     if(!has_stream(stream_idx)) {
         create_stream(stream_data, stream_idx);
     }
@@ -174,8 +208,7 @@ bool MultiFileWriter<VectorType>::operator()(VectorType const& stream_data,
         _file_streams.at(stream_idx)
             ->write(reinterpret_cast<char const*>(
                         thrust::raw_pointer_cast(stream_data.data())),
-                    stream_data.size() *
-                        sizeof(typename VectorType::value_type));
+                    data_size);
     }
     return false;
 }

cpp/skyweaver/src/PipelineConfig.cpp

@@ -15,7 +15,7 @@ PipelineConfig::PipelineConfig()
       _bw(13375000.0), _channel_frequencies_stale(true),
       _gulp_length_samps(4096), _start_time(0.0f),
       _duration(std::numeric_limits<float>::infinity()), _total_nchans(4096),
-      _stokes_mode("I"), _output_level(24.0f)
+      _stokes_mode("I"), _output_level(24.0f), _pre_write_config({0, {false, 0, 0}})
 {
 }
 
@@ -157,6 +157,64 @@ DedispersionPlan const& PipelineConfig::ddplan() const
     return _ddplan;
 }
 
+std::size_t PipelineConfig::convertMemorySize(const std::string& str) const {
+    std::size_t lastCharPos = str.find_last_not_of("0123456789");
+    std::string numberPart = str.substr(0, lastCharPos);
+    std::string unitPart = str.substr(lastCharPos);
+
+    std::size_t number = std::stoull(numberPart);
+
+    if (unitPart.empty())
+        return number;
+    else if (unitPart == "K" || unitPart == "k")
+        return number * 1024;
+    else if (unitPart == "M" || unitPart == "m")
+        return number * 1024 * 1024;
+    else if (unitPart == "G" || unitPart == "g")
+        return number * 1024 * 1024 * 1024;
+    else
+        throw std::runtime_error("Invalid memory unit!");
+}
+
+void PipelineConfig::configure_wait(std::string argument)
+{
+    std::vector<std::string> tokens;
+    std::string token;
+    std::istringstream tokenStream(argument);
+    int indx = 0;
+    _pre_write_config.is_enabled = true;
+    while (std::getline(tokenStream, token, ':')) {
+        if(indx == 0)
+        {
+            errno = 0;
+            _pre_write_config.wait.iterations = std::stoi(token);
+            if (errno == ERANGE) {
+              throw std::runtime_error("Wait iteration number out of range!");
+            }
+            if (_pre_write_config.wait.iterations < 0) _pre_write_config.wait.iterations = 0;
+        } else if(indx == 1) {
+            errno = 0;
+            _pre_write_config.wait.sleep_time = std::stoi(token);
+            if (errno == ERANGE) {
+              throw std::runtime_error("Sleep time out of range!");
+            }
+            if (_pre_write_config.wait.sleep_time < 1) _pre_write_config.wait.sleep_time = 1;
+        } else if(indx == 2) {
+            if (!token.empty() && std::all_of(token.begin(), token.end(), ::isdigit))
+            {
+                _pre_write_config.wait.min_free_space = std::stoull(token);
+            } else {
+                try {
+                    _pre_write_config.wait.min_free_space = convertMemorySize(token);
+                } catch (std::runtime_error& e) {
+                    std::cout << "Memory conversion error: " << e.what() << std::endl;
+                    throw;
+                }
+            }
+        }
+        indx++;
+    }
+}
 
 
 void PipelineConfig::enable_incoherent_dedispersion(bool enable)