ProgressTracker: reduce memory usage with long-running work items

ProgressTracker was only freeing memory for work items when they reach
the head of the work tracking queue.  If the first work item takes
hours to complete, and thousands of items are processed every second,
this leads to millions of completed items tracked in memory at a time,
wasting gigabytes of system RAM.

Rewrite ProgressHolderState methods to keep only incomplete work items
in memory, regardless of the order in which they are added or removed.

Also fix the unit tests which were relying on the memory leak to work,
and add test cases for code coverage.

Signed-off-by: Zygo Blaxell <bees@furryterror.org>

diff --git a/include/crucible/progress.h b/include/crucible/progress.h
index b16b7b65e6d73f1f1fca469d2a7cc8c23aa96850..a9fd003f9efb90fe6be8d534f760f3e11a2d0ab2 100644 (file)
--- a/include/crucible/progress.h
+++ b/include/crucible/progress.h
@@ -4,13 +4,20 @@
 #include "crucible/error.h"
 
 #include <functional>
-#include <map>
 #include <memory>
 #include <mutex>
+#include <set>
+
+#include <cassert>
 
 namespace crucible {
        using namespace std;
 
+       /// A class to track progress of multiple workers using only two points:
+       /// the first and last incomplete state.  The first incomplete
+       /// state can be recorded as a checkpoint to resume later on.
+       /// The last completed state is the starting point for workers that
+       /// need something to do.
        template <class T>
        class ProgressTracker {
                struct ProgressTrackerState;
@@ -19,8 +26,16 @@ namespace crucible {
                using value_type = T;
                using ProgressHolder = shared_ptr<ProgressHolderState>;
 
+               /// Create ProgressTracker with initial begin and end state 'v'.
                ProgressTracker(const value_type &v);
+
+               /// The first incomplete state.  This is not "sticky",
+               /// it will revert to the end state if there are no
+               /// items in progress.
                value_type begin() const;
+
+               /// The last incomplete state.  This is "sticky",
+               /// it can only increase and never decrease.
                value_type end() const;
 
                ProgressHolder hold(const value_type &v);
@@ -31,7 +46,7 @@ namespace crucible {
                struct ProgressTrackerState {
                        using key_type = pair<value_type, ProgressHolderState *>;
                        mutex                   m_mutex;
-                       map<key_type, bool>     m_in_progress;
+                       set<key_type>           m_in_progress;
                        value_type              m_begin;
                        value_type              m_end;
                };
@@ -39,6 +54,7 @@ namespace crucible {
                class ProgressHolderState {
                        shared_ptr<ProgressTrackerState>        m_state;
                        const value_type                        m_value;
+                       using key_type = typename ProgressTrackerState::key_type;
                public:
                        ProgressHolderState(shared_ptr<ProgressTrackerState> state, const value_type &v);
                        ~ProgressHolderState();
@@ -86,7 +102,11 @@ namespace crucible {
                m_value(v)
        {
                unique_lock<mutex> lock(m_state->m_mutex);
-               m_state->m_in_progress[make_pair(m_value, this)] = true;
+               const auto rv = m_state->m_in_progress.insert(key_type(m_value, this));
+               THROW_CHECK1(runtime_error, m_value, rv.second);
+               // Set the beginning to the first existing in-progress item
+               m_state->m_begin = m_state->m_in_progress.begin()->first;
+               // If this value is past the end, move the end, but don't go backwards
                if (m_state->m_end < m_value) {
                        m_state->m_end = m_value;
                }
@@ -96,17 +116,15 @@ namespace crucible {
        ProgressTracker<T>::ProgressHolderState::~ProgressHolderState()
        {
                unique_lock<mutex> lock(m_state->m_mutex);
-               m_state->m_in_progress[make_pair(m_value, this)] = false;
-               auto p = m_state->m_in_progress.begin();
-               while (p != m_state->m_in_progress.end()) {
-                       if (p->second) {
-                               break;
-                       }
-                       if (m_state->m_begin < p->first.first) {
-                               m_state->m_begin = p->first.first;
-                       }
-                       m_state->m_in_progress.erase(p);
-                       p = m_state->m_in_progress.begin();
+               const auto rv = m_state->m_in_progress.erase(key_type(m_value, this));
+               // THROW_CHECK2(runtime_error, m_value, rv, rv == 1);
+               assert(rv == 1);
+               if (m_state->m_in_progress.empty()) {
+                       // If we made the list empty, then m_begin == m_end
+                       m_state->m_begin = m_state->m_end;
+               } else {
+                       // If we deleted the first element, then m_begin = current first element
+                       m_state->m_begin = m_state->m_in_progress.begin()->first;
                }
        }
 

diff --git a/test/progress.cc b/test/progress.cc
index 51f1efc0b12e6436d342f6a058ea3b6a364c672c..c3d8967aceb01bf795cb0930d706548c116a9049 100644 (file)
--- a/test/progress.cc
+++ b/test/progress.cc
@@ -12,23 +12,49 @@ using namespace std;
 void
 test_progress()
 {
+       // On create, begin == end == constructor argument
        ProgressTracker<uint64_t> pt(123);
-       auto hold = pt.hold(234);
-       auto hold2 = pt.hold(345);
        assert(pt.begin() == 123);
+       assert(pt.end() == 123);
+
+       // Holding a position past the end increases the end (and moves begin to match)
+       auto hold345 = pt.hold(345);
+       assert(pt.begin() == 345);
        assert(pt.end() == 345);
-       auto hold3 = pt.hold(456);
-       assert(pt.begin() == 123);
+
+       // Holding a position before begin reduces begin, without changing end
+       auto hold234 = pt.hold(234);
+       assert(pt.begin() == 234);
+       assert(pt.end() == 345);
+
+       // Holding a position past the end increases the end, without affecting begin
+       auto hold456 = pt.hold(456);
+       assert(pt.begin() == 234);
        assert(pt.end() == 456);
-       hold2.reset();
-       assert(pt.begin() == 123);
+
+       // Releasing a position in the middle affects neither begin nor end
+       hold345.reset();
+       assert(pt.begin() == 234);
        assert(pt.end() == 456);
-       hold.reset();
-       assert(pt.begin() == 345);
+
+       // Hold another position in the middle to test begin moving forward
+       auto hold400 = pt.hold(400);
+
+       // Releasing a position at the beginning moves begin forward
+       hold234.reset();
+       assert(pt.begin() == 400);
        assert(pt.end() == 456);
-       hold3.reset();
+
+       // Releasing a position at the end doesn't move end backward
+       hold456.reset();
+       assert(pt.begin() == 400);
+       assert(pt.end() == 456);
+
+       // Releasing a position in the middle doesn't move end backward but does move begin forward
+       hold400.reset();
        assert(pt.begin() == 456);
        assert(pt.end() == 456);
+
 }
 
 int