1 files changed, 62 insertions, 11 deletions

diff --git a/src/file_collection.hh b/src/file_collection.hh
index 926f8f1..9ff260d 100644
--- a/src/file_collection.hh
+++ b/src/file_collection.hh
@@ -39,12 +39,14 @@
 #include <string>
 #include <utility>
 
+#include <sys/resource.h>
+
 #include "archive_manager.hh"
+#include "base/auto_pid.hh"
 #include "base/future_util.hh"
 #include "file_format.hh"
 #include "logfile_fwd.hh"
 #include "safe/safe.h"
-#include "tailer/tailer.looper.hh"
 
 struct tailer_progress {
     std::string tp_message;
@@ -53,6 +55,11 @@ struct tailer_progress {
 struct scan_progress {
     std::list<archive_manager::extract_progress> sp_extractions;
     std::map<std::string, tailer_progress> sp_tailers;
+
+    bool empty() const
+    {
+        return this->sp_extractions.empty() && this->sp_tailers.empty();
+    }
 };
 
 using safe_scan_progress = safe::Safe<scan_progress>;
@@ -73,6 +80,8 @@ struct file_error_info {
     const std::string fei_description;
 };
 
+using safe_name_to_errors = safe::Safe<std::map<std::string, file_error_info>>;
+
 struct file_collection;
 
 enum class child_poll_result_t {
@@ -83,16 +92,18 @@ enum class child_poll_result_t {
 class child_poller {
 public:
     explicit child_poller(
+        nonstd::optional<std::string> filename,
         auto_pid<process_state::running> child,
         std::function<void(file_collection&,
                            auto_pid<process_state::finished>&)> finalizer)
-        : cp_child(std::move(child)), cp_finalizer(std::move(finalizer))
+        : cp_filename(filename), cp_child(std::move(child)),
+          cp_finalizer(std::move(finalizer))
     {
         ensure(this->cp_finalizer);
     }
 
     child_poller(child_poller&& other) noexcept
-        : cp_child(std::move(other.cp_child)),
+        : cp_filename(other.cp_filename), cp_child(std::move(other.cp_child)),
           cp_finalizer(std::move(other.cp_finalizer))
     {
         ensure(this->cp_finalizer);
@@ -102,6 +113,7 @@ public:
     {
         require(other.cp_finalizer);
 
+        this->cp_filename = other.cp_filename;
         this->cp_child = std::move(other.cp_child);
         this->cp_finalizer = std::move(other.cp_finalizer);
 
@@ -114,9 +126,17 @@ public:
 
     child_poller& operator=(const child_poller&) = delete;
 
+    const nonstd::optional<std::string>& get_filename() const
+    {
+        return this->cp_filename;
+    }
+
+    void send_sigint();
+
     child_poll_result_t poll(file_collection& fc);
 
 private:
+    nonstd::optional<std::string> cp_filename;
     nonstd::optional<auto_pid<process_state::running>> cp_child;
     std::function<void(file_collection&, auto_pid<process_state::finished>&)>
         cp_finalizer;
@@ -128,7 +148,8 @@ struct file_collection {
     bool fc_recursive{false};
     bool fc_rotated{false};
 
-    std::map<std::string, file_error_info> fc_name_to_errors;
+    std::shared_ptr<safe_name_to_errors> fc_name_to_errors{
+        std::make_shared<safe_name_to_errors>()};
     std::map<std::string, logfile_open_options> fc_file_names;
     std::vector<std::shared_ptr<logfile>> fc_files;
     int fc_files_generation{0};
@@ -137,19 +158,39 @@ struct file_collection {
     std::set<std::string> fc_closed_files;
     std::map<std::string, other_file_descriptor> fc_other_files;
     std::set<std::string> fc_synced_files;
-    std::shared_ptr<safe_scan_progress> fc_progress;
+    std::shared_ptr<safe_scan_progress> fc_progress{
+        std::make_shared<safe_scan_progress>()};
     std::vector<struct stat> fc_new_stats;
     std::list<child_poller> fc_child_pollers;
     size_t fc_largest_path_length{0};
 
-    file_collection()
-        : fc_progress(std::make_shared<safe::Safe<scan_progress>>())
+    struct limits_t {
+        limits_t();
+
+        rlim_t l_fds;
+        rlim_t l_open_files;
+    };
+
+    static const limits_t& get_limits();
+
+    file_collection() = default;
+    file_collection(const file_collection&) = delete;
+    file_collection& operator=(const file_collection&) = delete;
+    file_collection(file_collection&&) = default;
+
+    file_collection copy();
+
+    bool empty() const
     {
+        return this->fc_name_to_errors->readAccess()->empty()
+            && this->fc_file_names.empty() && this->fc_files.empty()
+            && this->fc_progress->readAccess()->empty()
+            && this->fc_other_files.empty();
     }
 
     void clear()
     {
-        this->fc_name_to_errors.clear();
+        this->fc_name_to_errors->writeAccess()->clear();
         this->fc_file_names.clear();
         this->fc_files.clear();
         this->fc_closed_files.clear();
@@ -157,6 +198,13 @@ struct file_collection {
         this->fc_new_stats.clear();
     }
 
+    bool is_below_open_file_limit() const
+    {
+        return this->fc_files.size() < get_limits().l_open_files;
+    }
+
+    size_t other_file_format_count(file_format_t ff) const;
+
     file_collection rescan_files(bool required = false);
 
     void expand_filename(lnav::futures::future_queue<file_collection>& fq,
@@ -164,9 +212,8 @@ struct file_collection {
                          logfile_open_options& loo,
                          bool required);
 
-    std::future<file_collection> watch_logfile(const std::string& filename,
-                                               logfile_open_options& loo,
-                                               bool required);
+    nonstd::optional<std::future<file_collection>> watch_logfile(
+        const std::string& filename, logfile_open_options& loo, bool required);
 
     void merge(file_collection& other);
 
@@ -175,6 +222,10 @@ struct file_collection {
     void close_files(const std::vector<std::shared_ptr<logfile>>& files);
 
     void regenerate_unique_file_names();
+
+    size_t active_pipers() const;
+
+    size_t finished_pipers();
 };
 
 #endif