Back out "Revert D32606547: torch/monitor: add C++ events and handler…

…s" (pytorch#69923)

Summary:
Pull Request resolved: pytorch#69923

Original commit changeset: fbaf2cc06ad4

Original Phabricator Diff: D32606547 (pytorch@e61fc1c)

This is the same thing as the original diff but just using a normal std::mutex instead of std::shared_timed_mutex which is not available on OSX 10.11. The performance difference should be negligible and easy to change down the line if it does become a bottleneck.

Old failing build: https://github.com/pytorch/pytorch/runs/4495465412?check_suite_focus=true

Pull Request resolved: pytorch#68783

Test Plan:
buck test //caffe2/test/cpp/monitor:monitor

will add ciflow tags to ensure mac builds are fine

Reviewed By: aivanou

Differential Revision: D33102715

fbshipit-source-id: 3816ff01c578d8e844d303d881a63cf5c3817bdb

test/cpp/monitor/test_counters.cpp

@@ -1,130 +1,131 @@
 #include <gtest/gtest.h>
 
+#include <thread>
+
 #include <torch/csrc/monitor/counters.h>
+#include <torch/csrc/monitor/events.h>
 
 using namespace torch::monitor;
 
 TEST(MonitorTest, CounterDouble) {
-  Stat<double> a{
+  FixedCountStat<double> a{
       "a",
       {MEAN, COUNT},
+      2,
   };
   a.add(5.0);
   ASSERT_EQ(a.count(), 1);
   a.add(6.0);
-  ASSERT_EQ(a.count(), 2);
-  a.closeWindow();
-  auto stats = a.get();
   ASSERT_EQ(a.count(), 0);
 
-  std::vector<std::pair<Aggregation, double>> want = {
+  auto stats = a.get();
+  std::unordered_map<Aggregation, double> want = {
       {MEAN, 5.5},
       {COUNT, 2.0},
   };
   ASSERT_EQ(stats, want);
 }
 
 TEST(MonitorTest, CounterInt64Sum) {
-  Stat<int64_t> a{
+  FixedCountStat<int64_t> a{
       "a",
       {SUM},
+      2,
   };
   a.add(5);
   a.add(6);
-  a.closeWindow();
   auto stats = a.get();
-  std::vector<std::pair<Aggregation, int64_t>> want = {
+  std::unordered_map<Aggregation, int64_t> want = {
       {SUM, 11},
   };
   ASSERT_EQ(stats, want);
 }
 
 TEST(MonitorTest, CounterInt64Value) {
-  Stat<int64_t> a{
+  FixedCountStat<int64_t> a{
       "a",
       {VALUE},
+      2,
   };
   a.add(5);
   a.add(6);
-  a.closeWindow();
   auto stats = a.get();
-  std::vector<std::pair<Aggregation, int64_t>> want = {
+  std::unordered_map<Aggregation, int64_t> want = {
       {VALUE, 6},
   };
   ASSERT_EQ(stats, want);
 }
 
 TEST(MonitorTest, CounterInt64Mean) {
-  Stat<int64_t> a{
+  FixedCountStat<int64_t> a{
       "a",
       {MEAN},
+      2,
   };
-  a.add(0);
-  a.add(10);
-
   {
-    a.closeWindow();
+    // zero samples case
     auto stats = a.get();
-    std::vector<std::pair<Aggregation, int64_t>> want = {
-        {MEAN, 5},
+    std::unordered_map<Aggregation, int64_t> want = {
+        {MEAN, 0},
     };
     ASSERT_EQ(stats, want);
   }
 
+  a.add(0);
+  a.add(10);
+
   {
-    // zero samples case
-    a.closeWindow();
     auto stats = a.get();
-    std::vector<std::pair<Aggregation, int64_t>> want = {
-        {MEAN, 0},
+    std::unordered_map<Aggregation, int64_t> want = {
+        {MEAN, 5},
     };
     ASSERT_EQ(stats, want);
   }
 }
 
 TEST(MonitorTest, CounterInt64Count) {
-  Stat<int64_t> a{
+  FixedCountStat<int64_t> a{
       "a",
       {COUNT},
+      2,
   };
   ASSERT_EQ(a.count(), 0);
   a.add(0);
   ASSERT_EQ(a.count(), 1);
   a.add(10);
-  ASSERT_EQ(a.count(), 2);
-  a.closeWindow();
-  auto stats = a.get();
   ASSERT_EQ(a.count(), 0);
-  std::vector<std::pair<Aggregation, int64_t>> want = {
+
+  auto stats = a.get();
+  std::unordered_map<Aggregation, int64_t> want = {
       {COUNT, 2},
   };
   ASSERT_EQ(stats, want);
 }
 
 TEST(MonitorTest, CounterInt64MinMax) {
-  Stat<int64_t> a{
+  FixedCountStat<int64_t> a{
       "a",
       {MIN, MAX},
+      6,
   };
   {
-    a.closeWindow();
     auto stats = a.get();
-    std::vector<std::pair<Aggregation, int64_t>> want = {
+    std::unordered_map<Aggregation, int64_t> want = {
         {MAX, 0},
         {MIN, 0},
     };
     ASSERT_EQ(stats, want);
   }
+
   a.add(0);
   a.add(5);
   a.add(-5);
   a.add(-6);
   a.add(9);
   a.add(2);
   {
-    a.closeWindow();
     auto stats = a.get();
-    std::vector<std::pair<Aggregation, int64_t>> want = {
+    std::unordered_map<Aggregation, int64_t> want = {
         {MAX, 9},
         {MIN, -6},
     };
@@ -133,7 +134,7 @@ TEST(MonitorTest, CounterInt64MinMax) {
 }
 
 TEST(MonitorTest, CounterInt64WindowSize) {
-  Stat<int64_t> a{
+  FixedCountStat<int64_t> a{
       "a",
       {COUNT, SUM},
       /*windowSize=*/3,
@@ -144,54 +145,187 @@ TEST(MonitorTest, CounterInt64WindowSize) {
   a.add(3);
   ASSERT_EQ(a.count(), 0);
 
-  a.closeWindow();
+  a.add(4);
+  ASSERT_EQ(a.count(), 1);
+
   auto stats = a.get();
-  std::vector<std::pair<Aggregation, int64_t>> want = {
+  std::unordered_map<Aggregation, int64_t> want = {
       {COUNT, 3},
       {SUM, 6},
   };
   ASSERT_EQ(stats, want);
-  a.closeWindow();
-  ASSERT_EQ(stats, a.get());
 }
 
-TEST(MonitorTest, CloseAndGetStats) {
-  Stat<int64_t> a{
+template <typename T>
+struct TestIntervalStat : public IntervalStat<T> {
+  uint64_t mockWindowId{0};
+
+  TestIntervalStat(
+      std::string name,
+      std::initializer_list<Aggregation> aggregations,
+      std::chrono::milliseconds windowSize)
+      : IntervalStat<T>(name, aggregations, windowSize) {}
+
+  uint64_t currentWindowId() const override {
+    return mockWindowId;
+  }
+};
+
+struct AggregatingEventHandler : public EventHandler {
+  std::vector<Event> events;
+
+  void handle(const Event& e) override {
+    events.emplace_back(e);
+  }
+};
+
+template <typename T>
+struct HandlerGuard {
+  std::shared_ptr<T> handler;
+
+  HandlerGuard() : handler(std::make_shared<T>()) {
+    registerEventHandler(handler);
+  }
+
+  ~HandlerGuard() {
+    unregisterEventHandler(handler);
+  }
+};
+
+TEST(MonitorTest, IntervalStat) {
+  HandlerGuard<AggregatingEventHandler> guard;
+
+  IntervalStat<int64_t> a{
       "a",
       {COUNT, SUM},
-      /*windowSize=*/3,
+      std::chrono::milliseconds(1),
   };
-  Stat<double> b{
-      "b",
-      {MIN, MAX},
-      2,
+  ASSERT_EQ(guard.handler->events.size(), 0);
+
+  a.add(1);
+  ASSERT_LE(a.count(), 1);
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(2));
+  a.add(2);
+  ASSERT_LE(a.count(), 1);
+
+  ASSERT_GE(guard.handler->events.size(), 1);
+  ASSERT_LE(guard.handler->events.size(), 2);
+}
+
+TEST(MonitorTest, IntervalStatEvent) {
+  HandlerGuard<AggregatingEventHandler> guard;
+
+  TestIntervalStat<int64_t> a{
+      "a",
+      {COUNT, SUM},
+      std::chrono::milliseconds(1),
   };
+  ASSERT_EQ(guard.handler->events.size(), 0);
 
   a.add(1);
-  b.add(1);
+  ASSERT_EQ(a.count(), 1);
+  a.add(2);
+  ASSERT_EQ(a.count(), 2);
+  ASSERT_EQ(guard.handler->events.size(), 0);
+
+  a.mockWindowId = 100;
+
+  a.add(3);
+  ASSERT_LE(a.count(), 1);
+
+  ASSERT_EQ(guard.handler->events.size(), 1);
+  Event e = guard.handler->events.at(0);
+  ASSERT_EQ(e.type, "torch.monitor.Stat");
+  ASSERT_EQ(e.message, "a");
+  ASSERT_NE(e.timestamp, std::chrono::system_clock::time_point{});
+  std::unordered_map<std::string, metadata_value_t> metadata{
+      {"a.sum", 3L},
+      {"a.count", 2L},
+  };
+  ASSERT_EQ(e.metadata, metadata);
+}
+
+TEST(MonitorTest, IntervalStatEventDestruction) {
+  HandlerGuard<AggregatingEventHandler> guard;
 
   {
-    auto out = closeAndGetStats();
-    std::pair<
-        std::unordered_map<std::string, double>,
-        std::unordered_map<std::string, int64_t>>
-        want = {
-            {{"a.count", 1}, {"a.sum", 1}},
-            {{"b.min", 0}, {"b.max", 0}},
-        };
+    TestIntervalStat<int64_t> a{
+        "a",
+        {COUNT, SUM},
+        std::chrono::hours(10),
+    };
+    a.add(1);
+    ASSERT_EQ(a.count(), 1);
+    ASSERT_EQ(guard.handler->events.size(), 0);
   }
+  ASSERT_EQ(guard.handler->events.size(), 1);
 
+  Event e = guard.handler->events.at(0);
+  ASSERT_EQ(e.type, "torch.monitor.Stat");
+  ASSERT_EQ(e.message, "a");
+  ASSERT_NE(e.timestamp, std::chrono::system_clock::time_point{});
+  std::unordered_map<std::string, metadata_value_t> metadata{
+      {"a.sum", 1L},
+      {"a.count", 1L},
+  };
+  ASSERT_EQ(e.metadata, metadata);
+}
+
+TEST(MonitorTest, FixedCountStatEvent) {
+  HandlerGuard<AggregatingEventHandler> guard;
+
+  FixedCountStat<int64_t> a{
+      "a",
+      {COUNT, SUM},
+      3,
+  };
+  ASSERT_EQ(guard.handler->events.size(), 0);
+
+  a.add(1);
+  ASSERT_EQ(a.count(), 1);
   a.add(2);
-  b.add(2);
+  ASSERT_EQ(a.count(), 2);
+  ASSERT_EQ(guard.handler->events.size(), 0);
+
+  a.add(1);
+  ASSERT_EQ(a.count(), 0);
+  ASSERT_EQ(guard.handler->events.size(), 1);
+
+  Event e = guard.handler->events.at(0);
+  ASSERT_EQ(e.type, "torch.monitor.Stat");
+  ASSERT_EQ(e.message, "a");
+  ASSERT_NE(e.timestamp, std::chrono::system_clock::time_point{});
+  std::unordered_map<std::string, metadata_value_t> metadata{
+      {"a.sum", 4L},
+      {"a.count", 3L},
+  };
+  ASSERT_EQ(e.metadata, metadata);
+}
+
+TEST(MonitorTest, FixedCountStatEventDestruction) {
+  HandlerGuard<AggregatingEventHandler> guard;
 
   {
-    auto out = closeAndGetStats();
-    std::pair<
-        std::unordered_map<std::string, double>,
-        std::unordered_map<std::string, int64_t>>
-        want = {
-            {{"a.count", 1}, {"a.sum", 2}},
-            {{"b.min", 1}, {"b.max", 2}},
-        };
+    FixedCountStat<int64_t> a{
+        "a",
+        {COUNT, SUM},
+        3,
+    };
+    ASSERT_EQ(guard.handler->events.size(), 0);
+    a.add(1);
+    ASSERT_EQ(a.count(), 1);
+    ASSERT_EQ(guard.handler->events.size(), 0);
   }
+  ASSERT_EQ(guard.handler->events.size(), 1);
+
+  Event e = guard.handler->events.at(0);
+  ASSERT_EQ(e.type, "torch.monitor.Stat");
+  ASSERT_EQ(e.message, "a");
+  ASSERT_NE(e.timestamp, std::chrono::system_clock::time_point{});
+  std::unordered_map<std::string, metadata_value_t> metadata{
+      {"a.sum", 1L},
+      {"a.count", 1L},
+  };
+  ASSERT_EQ(e.metadata, metadata);
 }