Use dedicated call and progress event for progress reporting in status

reporter interface.

Refs codership/wsrep-lib#174

dbsim/db_server.cpp

@@ -163,5 +163,5 @@ void db::server::log_state_change(enum wsrep::server_state::state from,
                                   enum wsrep::server_state::state to)
 {
     wsrep::log_info() << "State changed " << from << " -> " << to;
-    reporter_.report_state(to, 0);
+    reporter_.report_state(to);
 }

include/wsrep/reporter.hpp

@@ -43,11 +43,19 @@ namespace wsrep
 
         virtual ~reporter();
 
-        // indefinite progress value
+        void report_state(enum server_state::state state);
-        static float constexpr indefinite = -1.0f;
 
-        void report_state(enum server_state::state state,
+        /**
-                          float progress = indefinite);
+         * Report progres in the form of a JSON string (all values integers):
+         * {
+         *     "from":       FROM,       // from wsrep API state number
+         *     "to":         TO,         // to wsrep API state number
+         *     "total":      TOTAL,      // total work to do
+         *     "done":       DONE,       // work already done
+         *     "indefinite": INDEFINITE  // indefinite value of work constant
+         * }
+         */
+        void report_progress(const std::string& json);
 
         enum log_level
         {
@@ -80,9 +88,9 @@ namespace wsrep
 
         wsrep::mutex&       mutex_;
         std::string const   file_name_;
+        std::string         progress_;
         char*               template_;
         substates           state_;
-        float               progress_;
         bool                initialized_;
 
         typedef struct {

src/reporter.cpp

@@ -33,6 +33,26 @@
 
 static std::string const TEMP_EXTENSION(".XXXXXX");
 
+static std::string make_progress_string(int const from, int const to,
+                                        int const total,int const done,
+                                        int const indefinite)
+{
+    std::ostringstream os;
+
+    os << "{ \"from\": "     << from       << ", "
+       << "\"to\": "         << to         << ", "
+       << "\"total\": "      << total      << ", "
+       << "\"done\": "       << done       << ", "
+       << "\"indefinite\": " << indefinite << " }";
+
+    return os.str();
+}
+
+static std::string const indefinite_progress
+    (make_progress_string(-1, -1, -1, -1, -1));
+static std::string const steady_state
+    (make_progress_string(-1, -1, 0, 0, -1));
+
 static inline double
 timestamp()
 {
@@ -46,9 +66,9 @@ wsrep::reporter::reporter(wsrep::mutex&      mutex,
                           size_t const       max_msg)
     : mutex_(mutex)
     , file_name_(file_name)
+    , progress_(indefinite_progress)
     , template_(new char [file_name_.length() + TEMP_EXTENSION.length() + 1])
     , state_(wsrep::reporter::s_disconnected_disconnected)
-    , progress_(indefinite)
     , initialized_(false)
     , err_msg_()
     , warn_msg_()
@@ -117,48 +137,6 @@ wsrep::reporter::substate_map(enum wsrep::server_state::state const state)
     }
 }
 
-static float const SST_SHARE  = 0.5f; // SST share of JOINING progress
-static float const INIT_SHARE = 0.1f; // initialization share of JOINING progress
-static float const IST_SHARE  = (1.0f - SST_SHARE - INIT_SHARE); // IST share
-
-float
-wsrep::reporter::progress_map(float const progress) const
-{
-    assert(progress >= 0.0f);
-    assert(progress <= 1.0f);
-
-    switch (state_)
-    {
-    case s_disconnected_disconnected:
-        return indefinite;
-    case s_disconnected_initializing:
-        return indefinite;
-    case s_disconnected_initialized:
-        return indefinite;
-    case s_connected_waiting:
-        return indefinite;
-    case s_joining_initialized:
-        return progress;
-    case s_joining_sst:
-        return progress * SST_SHARE;
-    case s_joining_initializing:
-        return SST_SHARE + progress * INIT_SHARE;
-    case s_joining_ist:
-        return SST_SHARE + INIT_SHARE + progress * IST_SHARE;
-    case s_joined_syncing:
-        return progress;
-    case s_synced_running:
-        return 1.0;
-    case s_donor_sending:
-        return progress;
-    case s_disconnecting_disconnecting:
-        return indefinite;
-    default:
-        assert(0);
-        return progress;
-    }
-}
-
 void
 wsrep::reporter::write_log_msg(std::ostream&  os,
                                const log_msg& msg)
@@ -250,8 +228,7 @@ wsrep::reporter::write_file(double const tstamp)
     os << "\t\"status\": {\n";
     os << "\t\t\"state\": \"" << strings[state_].state << "\",\n";
     os << "\t\t\"comment\": \"" << strings[state_].comment << "\",\n";
-    os << "\t\t\"progress\": " << std::showpoint << std::setprecision(6)
+    os << "\t\t\"progress\": " << progress_ << "\n";
-       << progress_ << "\n";
     os << "\t}\n";
     os << "}\n";
 
@@ -270,13 +247,8 @@ wsrep::reporter::write_file(double const tstamp)
 }
 
 void
-wsrep::reporter::report_state(enum server_state::state const s, float const p)
+wsrep::reporter::report_state(enum server_state::state const s)
 {
-    assert(p >= -1);
-    assert(p <= 1);
-
-    bool flush(false);
-
     wsrep::unique_lock<wsrep::mutex> lock(mutex_);
 
     substates const state(substate_map(s));
@@ -284,25 +256,32 @@ wsrep::reporter::report_state(enum server_state::state const s, float const p)
     if (state != state_)
     {
         state_ = state;
-        flush = true;
-    }
 
-    float const progress(progress_map(p));
+        if (state_ == s_synced_running)
-    assert(progress >= -1);
+            progress_ = steady_state;
-    assert(progress <= 1);
+        else
+            progress_ = indefinite_progress;
 
-    if (progress != progress_)
-    {
-        progress_ = progress;
-        flush = true;
-    }
-
-    if (flush)
-    {
         write_file(timestamp());
     }
 }
 
+void
+wsrep::reporter::report_progress(const std::string& json)
+{
+    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+
+    if (json != progress_)
+    {
+        if (state_ != s_synced_running)
+        {
+            // ignore any progress in SYNCED state
+            progress_ = json;
+            write_file(timestamp());
+        }
+    }
+}
+
 void
 wsrep::reporter::report_log_msg(log_level const    lvl,
                                 const std::string& msg,

test/reporter_test.cpp

@@ -40,6 +40,22 @@ timestamp()
     return (double(time.tv_sec) + double(time.tv_nsec)*1.0e-9);
 }
 
+static std::string make_progress_string(int const from, int const to,
+                                        int const total,int const done,
+                                        int const indefinite)
+{
+    std::ostringstream os;
+
+    os << "{ \"from\": "     << from       << ", "
+       << "\"to\": "         << to         << ", "
+       << "\"total\": "      << total      << ", "
+       << "\"done\": "       << done       << ", "
+       << "\"indefinite\": " << indefinite << " }";
+
+    return os.str();
+}
+
+
 static json::value
 read_file(const char* const filename)
 {
@@ -71,6 +87,52 @@ struct logs
     std::deque<std::string> msg_;
 };
 
+struct progress
+{
+    int from_;
+    int to_;
+    int total_;
+    int done_;
+    int indefinite_;
+
+    bool indefinite() const { return total_ == indefinite_; }
+    bool steady()     const { return total_ == 0; }
+    progress(int const from, int const to, int const total, int const done,
+        int const indefinite)
+        : from_(from)
+        , to_(to)
+        , total_(total)
+        , done_(done)
+        , indefinite_(indefinite)
+    {}
+};
+
+static bool
+operator==(const progress& left, const progress& right)
+{
+    if (left.indefinite() && right.indefinite()) return true;
+    if (left.steady() && right.steady()) return true;
+
+    return (left.from_       == right.from_ &&
+            left.to_         == right.to_ &&
+            left.total_      == right.total_ &&
+            left.done_       == right.done_ &&
+            left.indefinite_ == right.indefinite_);
+}
+
+static bool
+operator!=(const progress& left, const progress& right)
+{
+    return !(left == right);
+}
+
+static std::ostream&
+operator<<(std::ostream& os, const progress& p)
+{
+    os << make_progress_string(p.from_, p.to_, p.total_, p.done_,p.indefinite_);
+    return os;
+}
+
 struct result
 {
     logs errors_;
@@ -79,7 +141,7 @@ struct result
     {
         std::string state_;
         std::string comment_;
-        float       progress_;
+        progress    progress_;
     } status_;
 };
 
@@ -137,15 +199,31 @@ parse_result(const json::value& value, struct result& res,
         return;
     }
     case json::kind::uint64:
-        return;
+        WSREP_FALLTHROUGH;
     case json::kind::int64:
+        if (path == ".status.progress.from")
+        {
+            res.status_.progress_.from_ = int(value.get_int64());
+        }
+        else if (path == ".status.progress.to")
+        {
+            res.status_.progress_.to_ = int(value.get_int64());
+        }
+        else if (path == ".status.progress.total")
+        {
+            res.status_.progress_.total_ = int(value.get_int64());
+        }
+        else if (path == ".status.progress.done")
+        {
+            res.status_.progress_.done_ = int(value.get_int64());
+        }
+        else if (path == ".status.progress.indefinite")
+        {
+            res.status_.progress_.indefinite_ = int(value.get_int64());
+        }
         return;
     case json::kind::double_:
-        if (path == ".status.progress")
+        if (path == ".errors.[].timestamp")
-        {
-            res.status_.progress_ = float(value.get_double());
-        }
-        else if (path == ".errors.[].timestamp")
         {
             res.errors_.tstamp_.push_back(value.get_double());
         }
@@ -270,8 +348,11 @@ print(const result& left, const result& right, size_t it)
 static const char*
 const REPORT = "report.json";
 
-static auto
+static progress
-const indefinite(wsrep::reporter::indefinite);
+const indefinite(-1, -1, -1, -1, -1);
+
+static progress
+const steady(-1, -1, 0, 0, -1);
 
 static struct logs
 const LOGS_INIT = { std::deque<double>(), std::deque<std::string>() };
@@ -391,7 +472,7 @@ BOOST_AUTO_TEST_CASE(state_test)
           { "CONNECTED",     "Waiting",         indefinite }}},
         {{ server_state::s_joiner,        0 },
          { ERRS_INIT, LOGS_INIT,
-          { "JOINING",       "Receiving state", 0.0f       }}},
+          { "JOINING",       "Receiving state", indefinite }}},
         {{ server_state::s_disconnecting, 0 },
          { ERRS_INIT, LOGS_INIT,
           { "DISCONNECTING", "Disconnecting",   indefinite }}},
@@ -403,28 +484,28 @@ BOOST_AUTO_TEST_CASE(state_test)
           { "CONNECTED",     "Waiting",         indefinite }}},
         {{ server_state::s_joiner,        0 },
          { ERRS_INIT, LOGS_INIT,
-          { "JOINING",       "Receiving SST",   0.0f       }}},
+          { "JOINING",       "Receiving SST",   indefinite }}},
         {{ server_state::s_initializing,  0 },
          { ERRS_INIT, LOGS_INIT,
-          { "JOINING",       "Initializing",    0.5f       }}},
+          { "JOINING",       "Initializing",    indefinite }}},
         {{ server_state::s_initialized,   0 },
          { ERRS_INIT, LOGS_INIT,
-          { "JOINING",       "Receiving IST",   0.6f       }}},
+          { "JOINING",       "Receiving IST",   indefinite }}},
         {{ server_state::s_joined,        0 },
          { ERRS_INIT, LOGS_INIT,
-          { "JOINED",        "Syncing",         0.0f       }}},
+          { "JOINED",        "Syncing",         indefinite }}},
         {{ server_state::s_synced,        0 },
          { ERRS_INIT, LOGS_INIT,
-          { "SYNCED",        "Operational",     1.0f       }}},
+          { "SYNCED",        "Operational",     steady     }}},
         {{ server_state::s_donor,         0 },
          { ERRS_INIT, LOGS_INIT,
-          { "DONOR",         "Donating SST",    0.0f       }}},
+          { "DONOR",         "Donating SST",    indefinite }}},
         {{ server_state::s_joined,        0 },
          { ERRS_INIT, LOGS_INIT,
-          { "JOINED",        "Syncing",         0.0f       }}},
+          { "JOINED",        "Syncing",         indefinite }}},
         {{ server_state::s_synced,        0 },
          { ERRS_INIT, LOGS_INIT,
-          { "SYNCED",        "Operational",     1.0f       }}},
+          { "SYNCED",        "Operational",     steady     }}},
         {{ server_state::s_disconnecting, 0 },
          { ERRS_INIT, LOGS_INIT,
           { "DISCONNECTING", "Disconnecting",   indefinite }}},
@@ -434,7 +515,7 @@ BOOST_AUTO_TEST_CASE(state_test)
 
     for (auto i(tests.begin()); i != tests.end(); ++i)
     {
-        rep.report_state(i->input.state, i->input.progress);
+        rep.report_state(i->input.state);
         auto value = read_file(REPORT);
         result res(RES_INIT);
         parse_result(value, res);
@@ -454,12 +535,17 @@ BOOST_AUTO_TEST_CASE(progress_test)
     double const      warn_tstamp(timestamp());
     std::string const warn_msg("Warn!");
 
+    static progress const progress5_0(-1, -1, 5, 0, -1);
+    static progress const progress5_2(-1, -1, 5, 2, -1);
+    static progress const progress5_5(-1, -1, 5, 5, -1);
+
     struct test
     {
         struct
         {
             enum wsrep::server_state::state state;
-            float progress;
+            int total;
+            int done;
         } input;
         struct result output;
     };
@@ -468,70 +554,77 @@ BOOST_AUTO_TEST_CASE(progress_test)
 
     std::vector<test> tests =
     {
-        {{ server_state::s_initialized,   0 },
+        {{ server_state::s_initialized,  -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "DISCONNECTED",  "Connecting",      indefinite }}},
+          { "DISCONNECTED", "Connecting",      indefinite  }}},
-        {{ server_state::s_connected,     0 },
+        {{ server_state::s_connected,    -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "CONNECTED",     "Waiting",         indefinite }}},
+          { "CONNECTED",    "Waiting",         indefinite  }}},
-        {{ server_state::s_joiner,        0 },
+        {{ server_state::s_joiner,        5,  0 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Receiving state", 0.0f       }}},
+          { "JOINING",      "Receiving state", progress5_0 }}},
-        {{ server_state::s_joiner,        0.5 },
+        {{ server_state::s_joiner,        5,  2 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Receiving state", 0.5f       }}},
+          { "JOINING",      "Receiving state", progress5_2 }}},
-        {{ server_state::s_disconnected,  0 },
+        {{ server_state::s_disconnected, -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "DISCONNECTED",  "Disconnected",    indefinite }}},
+          { "DISCONNECTED", "Disconnected",    indefinite  }}},
-        {{ server_state::s_connected,     0 },
+        {{ server_state::s_connected,    -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "CONNECTED",     "Waiting",         indefinite }}},
+          { "CONNECTED",    "Waiting",         indefinite  }}},
-        {{ server_state::s_joiner,        0 },
+        {{ server_state::s_joiner,        5,  0 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Receiving SST",   0.0f       }}},
+          { "JOINING",      "Receiving SST",   progress5_0 }}},
-        {{ server_state::s_joiner,        0.5 },
+        {{ server_state::s_joiner,        5,  2 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Receiving SST",   0.25f      }}},
+          { "JOINING",      "Receiving SST",   progress5_2 }}},
-        {{ server_state::s_initializing,  0 },
+        {{ server_state::s_joiner,        5,  5 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Initializing",    0.5f       }}},
+          { "JOINING",      "Receiving SST",   progress5_5 }}},
-        {{ server_state::s_initializing,  0.5 },
+        {{ server_state::s_initializing, -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Initializing",    0.55f      }}},
+          { "JOINING",      "Initializing",    indefinite  }}},
-        {{ server_state::s_initialized,   0 },
+        {{ server_state::s_initializing,  5,  2 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Receiving IST",   0.6f       }}},
+          { "JOINING",      "Initializing",    progress5_2 }}},
-        {{ server_state::s_initialized,   0.5 },
+        {{ server_state::s_initialized,   5,  0 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINING",       "Receiving IST",   0.8f       }}},
+          { "JOINING",      "Receiving IST",   progress5_0 }}},
-        {{ server_state::s_joined,        0 },
+        {{ server_state::s_initialized,   5, 5 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINED",        "Syncing",         0.0f       }}},
+          { "JOINING",      "Receiving IST",   progress5_5 }}},
-        {{ server_state::s_joined,        0.5 },
+        {{ server_state::s_joined,       -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINED",        "Syncing",         0.5f       }}},
+          { "JOINED",       "Syncing",         indefinite  }}},
-        {{ server_state::s_synced,        0 },
+        {{ server_state::s_joined,        5, 2 },
          { LOGS_INIT, WARN_INIT,
-          { "SYNCED",        "Operational",     1.0f       }}},
+          { "JOINED",       "Syncing",         progress5_2 }}},
-        {{ server_state::s_donor,         0 },
+        {{ server_state::s_synced,       -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "DONOR",         "Donating SST",    0.0f       }}},
+          { "SYNCED",       "Operational",     steady      }}},
-        {{ server_state::s_donor,         0.5 },
+        {{ server_state::s_donor,        -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "DONOR",         "Donating SST",    0.5f       }}},
+          { "DONOR",        "Donating SST",    indefinite  }}},
-        {{ server_state::s_joined,        0 },
+        {{ server_state::s_donor,         5,  0 },
          { LOGS_INIT, WARN_INIT,
-          { "JOINED",        "Syncing",         0.0f       }}},
+          { "DONOR",        "Donating SST",    progress5_0 }}},
-        {{ server_state::s_synced,        0 },
+        {{ server_state::s_joined,       -1, -1 },
          { LOGS_INIT, WARN_INIT,
-          { "SYNCED",        "Operational",     1.0f       }}},
+          { "JOINED",       "Syncing",         indefinite }}},
+        {{ server_state::s_synced,       -1, -1 },
+         { LOGS_INIT, WARN_INIT,
+          { "SYNCED",       "Operational",     steady     }}},
     };
 
     rep.report_log_msg(wsrep::reporter::warning, warn_msg, warn_tstamp);
 
     for (auto i(tests.begin()); i != tests.end(); ++i)
     {
-        rep.report_state(i->input.state, i->input.progress);
+        rep.report_state(i->input.state);
+        if (i->input.total >= 0)
+            rep.report_progress(make_progress_string(-1, -1,
+                                                 i->input.total, i->input.done,
+                                                 -1));
         auto value = read_file(REPORT);
         result res(RES_INIT);
         parse_result(value, res);