Add GOMA support

1 files changed, 339 insertions, 0 deletions

diff --git a/src/goma.cc b/src/goma.cc
new file mode 100644
index 0000000..717d7b1
--- /dev/null
+++ b/src/goma.cc
@@ -0,0 +1,339 @@
+#include "common.hh"
+
+#include <condition_variable>
+#include <curl/curl.h>
+#include <functional>
+#include <mutex>
+#include <rapidjson/document.h>
+#include <rapidjson/error/en.h>
+#include <string>
+#include <thread>
+#include <unordered_set>
+
+#include "args.hh"
+#include "cairo.hh"
+#include "goma.hh"
+#include "monmon.hh"
+#include "pango.hh"
+
+namespace {
+
+struct CurlDelete {
+  void operator()(CURL* curl) const {
+    curl_easy_cleanup(curl);
+  }
+};
+
+struct State {
+  bool error_{true};
+  std::string msg_;
+  uint64_t running_{0};
+  uint64_t failed_{0};
+};
+
+class GomaPoll {
+public:
+  GomaPoll(std::shared_ptr<PollLooper> const& looper, std::string addr,
+           std::function<void()> update)
+    : looper_(looper), addr_(std::move(addr)), update_(update) {
+    pipe_.open();
+    looper_->add(pipe_.read(), Looper::EV_READ,
+                 std::bind(&GomaPoll::pipe, this, std::placeholders::_1,
+                           std::placeholders::_2, std::placeholders::_3));
+    thread_ = std::thread(&GomaPoll::loop, this);
+  }
+
+  ~GomaPoll() {
+    looper_->remove(pipe_.read());
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      quit_ = true;
+    }
+    cond_.notify_all();
+    thread_.join();
+  }
+
+  State const& state() const {
+    return state_;
+  }
+
+private:
+  void loop() {
+    std::unique_ptr<CURL, CurlDelete> curl(curl_easy_init());
+
+    uint64_t after = 0;
+    std::unordered_set<uint64_t> active;
+    std::unordered_set<uint64_t> failed;
+
+    curl_easy_setopt(curl.get(), CURLOPT_NOPROGRESS, 1L);
+    curl_easy_setopt(curl.get(), CURLOPT_WRITEFUNCTION,
+                     &GomaPoll::write_callback);
+    curl_easy_setopt(curl.get(), CURLOPT_WRITEDATA, this);
+    curl_easy_setopt(curl.get(), CURLOPT_POSTFIELDS, "");
+    curl_easy_setopt(curl.get(), CURLOPT_TIMEOUT, 10L);
+
+    while (true) {
+      State state;
+      content_.clear();
+      std::string url = addr_;
+      if (after > 0)
+        url += "?after=" + std::to_string(after);
+      curl_easy_setopt(curl.get(), CURLOPT_URL, url.c_str());
+      auto ret = curl_easy_perform(curl.get());
+      if (ret == CURLE_OK) {
+        parse(content_, &state, &active, &failed, &after);
+      } else {
+        state.error_ = true;
+        state.msg_ = curl_easy_strerror(ret);
+      }
+
+      {
+        std::unique_lock<std::mutex> lock(mutex_);
+        state_ = state;
+        io::write(pipe_, "s", 1);
+        cond_.wait_for(lock, std::chrono::seconds(1));
+        if (quit_)
+          break;
+      }
+    }
+  }
+
+  static void parse(std::string const& content, State* state,
+                    std::unordered_set<uint64_t>* active,
+                    std::unordered_set<uint64_t>* failed,
+                    uint64_t* after) {
+    *after = 0;
+    state->error_ = false;
+
+    rapidjson::Document doc;
+    doc.Parse(content.data(), content.size());
+    if (doc.HasParseError() || !doc.IsObject()) {
+      state->error_ = true;
+      state->msg_ = rapidjson::GetParseError_En(doc.GetParseError());
+      return;
+    }
+    if (doc.HasMember("last_update_ms")) {
+      auto& member = doc["last_update_ms"];
+      if (member.IsUint64()) {
+        *after = member.GetUint64();
+      }
+    }
+    if (doc.HasMember("active")) {
+      auto& member = doc["active"];
+      if (member.IsArray()) {
+        for (auto& task : member.GetArray()) {
+          if (task.HasMember("id")) {
+            auto& id = task["id"];
+            if (id.IsUint64()) {
+              active->insert(id.GetUint64());
+            }
+          }
+        }
+      }
+    }
+    if (doc.HasMember("failed")) {
+      auto& member = doc["failed"];
+      if (member.IsArray()) {
+        for (auto& task : member.GetArray()) {
+          if (task.HasMember("id")) {
+            auto& id = task["id"];
+            if (id.IsUint64()) {
+              failed->insert(id.GetUint64());
+            }
+          }
+        }
+      }
+    }
+    if (doc.HasMember("finished")) {
+      auto& member = doc["finished"];
+      if (member.IsArray()) {
+        for (auto& task : member.GetArray()) {
+          if (task.HasMember("id")) {
+            auto& id = task["id"];
+            if (id.IsUint64()) {
+              active->erase(id.GetUint64());
+            }
+          }
+        }
+      }
+    }
+
+    state->running_ = active->size();
+    state->failed_ = failed->size();
+  }
+
+  static size_t write_callback(char* ptr, size_t size, size_t nmemb,
+                             void* userdata) {
+    return reinterpret_cast<GomaPoll*>(userdata)->write(ptr, size * nmemb);
+  }
+
+  size_t write(char* ptr, size_t size) {
+    content_.append(ptr, size);
+    return size;
+  }
+
+  void pipe(Looper*, int, uint8_t event) {
+    if (event & Looper::EV_READ) {
+      char tmp;
+      if (io::read(pipe_, &tmp, 1)) {
+        assert(tmp == 's');
+        update_();
+      }
+    }
+  }
+
+  std::shared_ptr<PollLooper> looper_;
+  io::pipe pipe_;
+  std::string const addr_;
+  std::thread thread_;
+  std::function<void()> const update_;
+
+  std::mutex mutex_;
+  std::condition_variable cond_;
+  bool quit_{false};
+  State state_;
+
+  std::string content_;
+};
+
+class GomaMonMon : public MonMon {
+public:
+  GomaMonMon(std::shared_ptr<PollLooper> const& looper)
+    : MonMon(looper) {
+    curl_global_init(CURL_GLOBAL_ALL);
+  }
+
+  ~GomaMonMon() {
+    curl_global_cleanup();
+  }
+
+  bool connect(Args const* args) override {
+    std::string addr(args->arg("goma_addr", "http://127.0.0.1:8088/"));
+    if (addr.empty() || addr.back() != '/')
+      addr += '/';
+    addr += "api/taskz";
+
+    poll_ = std::make_unique<GomaPoll>(
+        looper_, std::move(addr),
+        std::bind(&GomaMonMon::draw, this));
+
+    return true;
+  }
+
+private:
+  std::string title() const override {
+    return "MonMon (goma)";
+  }
+
+  void width_changed() override {
+    progress_pattern_.reset();
+  }
+
+  void internal_quit() override {
+    poll_.reset();
+  }
+
+  void draw_content(cairo_t* cairo, PangoLayout* layout,
+                    uint16_t w, uint16_t h) override {
+    if (!poll_) return;
+    auto state = poll_->state();
+
+    auto const pad_x = std::min(9.0, w / 20.0), pad_y = pad_x;
+    auto const margin_x = std::min(7.5, w / 20.0), margin_y = margin_x;
+    auto const box_w = w - pad_x * 2;
+    if (!progress_pattern_) {
+      progress_pattern_.reset(
+          cairo_pattern_create_linear(0.0, 0.0, box_w, 0.0));
+      cairo_pattern_add_color_stop_rgb(progress_pattern_.get(), 0.000000,
+                                       0.000000, 0.000000, 0.000000);
+      cairo_pattern_add_color_stop_rgb(progress_pattern_.get(), 0.594324,
+                                       0.729412, 0.000000, 0.000000);
+      cairo_pattern_add_color_stop_rgb(progress_pattern_.get(), 0.809683,
+                                       1.000000, 0.545098, 0.196078);
+      cairo_pattern_add_color_stop_rgb(progress_pattern_.get(), 0.899833,
+                                       0.972549, 0.937255, 0.074510);
+      cairo_pattern_add_color_stop_rgb(progress_pattern_.get(), 1.000000,
+                                       0.976471, 0.968627, 0.831373);
+    }
+    pango_layout_set_width(layout, (box_w - margin_x * 2) * PANGO_SCALE);
+
+    auto x = pad_x;
+    auto y = pad_y;
+
+    if (state.error_) {
+      draw_box(cairo, layout, x, y, box_w, margin_y, 1.0, 0.0, 0.0, state.msg_,
+               0.0);
+    } else {
+      float progress;
+      if (state.running_ > max_running_) {
+        max_running_ = state.running_;
+        progress = 1.0f;
+      } else {
+        progress = static_cast<float>(state.running_) / max_running_;
+      }
+      std::string msg = "Active: " + std::to_string(state.running_);
+      y = draw_box(cairo, layout, x, y, box_w, margin_y, 1.0, 1.0, 1.0, msg,
+                   progress)
+        + pad_y;
+      if (state.failed_ > 0 && y < h) {
+        msg = "Failed: " + std::to_string(state.failed_);
+        draw_box(cairo, layout, x, y, box_w, margin_y, 1.0, 0, 0, msg, 0);
+      }
+    }
+  }
+
+  uint16_t draw_box(cairo_t* cairo, PangoLayout* layout,
+                    uint16_t x, uint16_t y,
+                    uint16_t w, uint16_t margin_y,
+                    float r, float g, float b,
+                    std::string const& text,
+                    float progress) {
+    pango_layout_set_text(layout, text.data(), text.size());
+    int layout_height;
+    pango_layout_get_size(layout, nullptr, &layout_height);
+    uint16_t height = (layout_height + PANGO_SCALE - 1) / PANGO_SCALE;
+    rounded_path(cairo, x, y, w, height + margin_y * 2);
+    cairo_set_source_rgba(cairo, 0.1, 0.1, 0.1, 0.7);
+
+    if (progress > 0.0) {
+      cairo_fill_preserve(cairo);
+      auto old = cairo::unique_path(cairo_copy_path(cairo));
+      cairo_new_path(cairo);
+      cairo_rectangle(cairo, x, y, progress * w, height + margin_y * 2);
+      cairo_clip(cairo);
+      cairo_append_path(cairo, old.get());
+      cairo_matrix_t matrix;
+      cairo_matrix_init_translate(&matrix, -x, 0);
+      cairo_pattern_set_matrix(progress_pattern_.get(), &matrix);
+      cairo_set_source(cairo, progress_pattern_.get());
+      cairo_fill(cairo);
+      cairo_reset_clip(cairo);
+    } else {
+      cairo_fill(cairo);
+    }
+
+    cairo_move_to(cairo, x, y + margin_y);
+    pango_cairo_layout_path(cairo, layout);
+    cairo_set_source_rgb(cairo, 0.0, 0.0, 0.0);
+    cairo_set_line_cap(cairo, CAIRO_LINE_CAP_ROUND);
+    cairo_set_line_width(cairo, 2.0);
+    cairo_stroke(cairo);
+    cairo_set_source_rgb(cairo, r, g, b);
+    cairo_move_to(cairo, x, y + margin_y);
+    pango_cairo_show_layout(cairo, layout);
+
+    return y + margin_y * 2 + height;
+  }
+
+  uint64_t max_running_{10};
+  cairo::unique_pattern progress_pattern_;
+  std::unique_ptr<GomaPoll> poll_;
+};
+
+}  // namespace
+
+std::unique_ptr<MonMon> create_goma_monmon(
+    std::shared_ptr<PollLooper> const& looper) {
+  return std::make_unique<GomaMonMon>(looper);
+}
+