Merge branch 'rusage'

main.cc

@@ -32,15 +32,11 @@ class MyHandler : public Net::Http::Handler {
                     .add<Header::Server>("lys")
                     .add<Header::ContentType>(MIME(Text, Plain));
 
-#if 0
                 auto stream = response.stream(Net::Http::Code::Ok);
                 stream << "PO";
-                stream << flush;
                 stream << "NG";
                 stream << ends;
-#endif
 
-                response.send(Code::Ok, "PONG");
             }
         }
         else if (req.resource() == "/echo") {
@@ -63,6 +59,50 @@ class MyHandler : public Net::Http::Handler {
     }
 };
 
+struct LoadMonitor {
+    LoadMonitor(const std::shared_ptr<Net::Http::Endpoint>& endpoint)
+        : endpoint_(endpoint)
+        , interval(std::chrono::seconds(1))
+    { }
+
+    void setInterval(std::chrono::seconds secs) {
+        interval = secs;
+    }
+
+    void start() {
+        thread.reset(new std::thread(std::bind(&LoadMonitor::run, this)));
+    }
+
+    ~LoadMonitor() {
+        thread->join();
+    }
+
+private:
+    std::shared_ptr<Net::Http::Endpoint> endpoint_;
+    std::unique_ptr<std::thread> thread;
+    std::chrono::seconds interval;
+
+    void run() {
+        Net::Tcp::Listener::Load old;
+        while (endpoint_->isBound()) {
+            endpoint_->requestLoad(old).then([&](const Net::Tcp::Listener::Load& load) {
+                old = load;
+
+                double global = load.global;
+                if (global > 100) global = 100;
+
+                if (global > 1)
+                    std::cout << "Global load is " << global << "%" << std::endl;
+                else
+                    std::cout << "Global load is 0%" << std::endl;
+            },
+            Async::NoExcept);
+
+            std::this_thread::sleep_for(std::chrono::seconds(interval));
+        }
+    }
+};
+
 int main(int argc, char *argv[]) {
     Net::Port port(9080);
 
@@ -81,12 +121,17 @@ int main(int argc, char *argv[]) {
     cout << "Cores = " << hardware_concurrency() << endl;
     cout << "Using " << thr << " threads" << endl;
 
-    Net::Http::Endpoint server(addr);
+    auto server = std::make_shared<Net::Http::Endpoint>(addr);
+
     auto opts = Net::Http::Endpoint::options()
         .threads(thr)
         .flags(Net::Tcp::Options::InstallSignalHandler);
-    server.init(opts);
-    server.setHandler(std::make_shared<MyHandler>());
+    server->init(opts);
+    server->setHandler(std::make_shared<MyHandler>());
+
+    LoadMonitor monitor(server);
+    monitor.setInterval(std::chrono::seconds(5));
+    monitor.start();
 
-    server.serve();
+    server->serve();
 }

src/http.cc

@@ -586,6 +586,11 @@ Endpoint::serve()
     }
 }
 
+Async::Promise<Tcp::Listener::Load>
+Endpoint::requestLoad(const Tcp::Listener::Load& old) {
+    return listener.requestLoad(old);
+}
+
 Endpoint::Options
 Endpoint::options() {
     return Options();

src/http.h

@@ -530,6 +530,12 @@ public:
     void setHandler(const std::shared_ptr<Handler>& handler);
     void serve();
 
+    bool isBound() const {
+        return listener.isBound();
+    }
+
+    Async::Promise<Tcp::Listener::Load> requestLoad(const Tcp::Listener::Load& old);
+
     static Options options();
 
 private:

src/io.cc

@@ -231,6 +231,9 @@ IoWorker::run() {
 
     mailbox.bind(poller);
 
+    auto n = notifier.tag();
+    poller.addFd(n.value(), NotifyOn::Read, n, Polling::Mode::Edge);
+
     std::chrono::milliseconds timeout(-1);
 
     for (;;) {
@@ -250,7 +253,11 @@ IoWorker::run() {
                     if (msg->type() == Message::Type::Shutdown) {
                         return;
                     }
-                } else {
+                }
+                else if (event.tag == notifier.tag()) {
+                    handleNotify();
+                }
+                else {
                     if (event.flags.hasFlag(NotifyOn::Read)) {
                         auto fd = event.tag.value();
                         if (fd == timerFd) {
@@ -314,6 +321,23 @@ IoWorker::handleTimeout() {
     });
 }
 
+void
+IoWorker::handleNotify() {
+    optionally_do(load, [&](const Load& entry) {
+        while (this->notifier.tryRead()) ;
+
+        rusage now;
+
+        auto res = getrusage(RUSAGE_THREAD, &now);
+        if (res == -1)
+            entry.reject(std::runtime_error("Could not compute usage"));
+
+        entry.resolve(now);
+    });
+
+    load = None();
+}
+
 Async::Promise<ssize_t>
 IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
     return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {

src/io.h

@@ -16,6 +16,8 @@
 #include <mutex>
 #include <memory>
 #include <unordered_map>
+#include <sys/time.h>
+#include <sys/resource.h>
 
 namespace Net {
 
@@ -49,6 +51,13 @@ public:
 
     void disarmTimer();
 
+    Async::Promise<rusage> getLoad() {
+        return Async::Promise<rusage>([=](Async::Resolver& resolve, Async::Rejection& reject) {
+            this->load = Some(Load(std::move(resolve), std::move(reject)));
+            this->notifier.notify();
+        });
+    }
+
 private:
     struct OnHoldWrite {
         OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
@@ -84,6 +93,17 @@ private:
         Async::Rejection reject;
     };
 
+    struct Load {
+        Load(Async::Resolver resolve,
+             Async::Rejection reject)
+            : resolve(std::move(resolve))
+            , reject(std::move(reject))
+        { }
+
+        Async::Resolver resolve;
+        Async::Rejection reject;
+    };
+
     Polling::Epoll poller;
     std::unique_ptr<std::thread> thread;
     mutable std::mutex peersMutex;
@@ -91,8 +111,11 @@ private:
     std::unordered_map<Fd, OnHoldWrite> toWrite;
 
     Optional<Timer> timer;
+    Optional<Load> load;
     Fd timerFd;
 
+    NotifyFd notifier;
+
     std::shared_ptr<Handler> handler_;
     Flags<Options> options_;
 
@@ -107,6 +130,7 @@ private:
 
     void handleIncoming(const std::shared_ptr<Peer>& peer);
     void handleTimeout();
+    void handleNotify();
     void run();
 
 };

src/listener.cc

@@ -175,10 +175,15 @@ Listener::bind(const Address& address) {
     for (auto& io: ioGroup) {
         io->start(handler_, options_);
     }
-    
+
     return true;
 }
 
+bool
+Listener::isBound() const {
+    return g_listen_fd != -1;
+}
+
 void
 Listener::run() {
     for (;;) {
@@ -212,6 +217,55 @@ Listener::shutdown() {
     }
 }
 
+Async::Promise<Listener::Load>
+Listener::requestLoad(const Listener::Load& old) {
+    std::vector<Async::Promise<rusage>> loads;
+    loads.reserve(ioGroup.size());
+
+    for (const auto& io: ioGroup) {
+        loads.push_back(io->getLoad());
+    }
+
+    return Async::whenAll(std::begin(loads), std::end(loads)).then([=](const std::vector<rusage>& usages) {
+
+        Load res;
+        res.raw = usages;
+
+        if (old.raw.empty()) {
+            res.global = 0.0;
+            for (size_t i = 0; i < ioGroup.size(); ++i) res.workers.push_back(0.0);
+        } else {
+
+            auto totalElapsed = [](rusage usage) {
+                return (usage.ru_stime.tv_sec * 1e6 + usage.ru_stime.tv_usec)
+                     + (usage.ru_utime.tv_sec * 1e6 + usage.ru_utime.tv_usec);
+            };
+
+            auto now = std::chrono::system_clock::now();
+            std::chrono::microseconds tick = now - old.tick;
+            res.tick = now;
+
+            for (size_t i = 0; i < usages.size(); ++i) {
+                auto last = old.raw[i];
+                const auto& usage = usages[i];
+
+                auto nowElapsed = totalElapsed(usage);
+                auto timeElapsed = nowElapsed - totalElapsed(last);
+
+                auto loadPct = (timeElapsed * 100.0) / tick.count();
+                res.workers.push_back(loadPct);
+                res.global += loadPct;
+
+            }
+
+            res.global /= usages.size();
+        }
+
+        return Async::Promise<Load>::resolved(std::move(res));
+
+     }, Async::NoExcept);
+}
+
 Address
 Listener::address() const {
     return addr_;

src/listener.h

@@ -26,6 +26,15 @@ void setSocketOptions(Fd fd, Flags<Options> options);
 class Listener {
 public:
 
+    struct Load {
+        typedef std::chrono::system_clock::time_point TimePoint;
+        double global;
+        std::vector<double> workers;
+
+        std::vector<rusage> raw;
+        TimePoint tick;
+    };
+
     Listener();
 
     Listener(const Address& address);
@@ -37,9 +46,13 @@ public:
     bool bind();
     bool bind(const Address& adress);
 
+    bool isBound() const;
+
     void run();
     void shutdown();
 
+    Async::Promise<Load> requestLoad(const Load& old);
+
     Options options() const;
     Address address() const;
 
@@ -54,6 +67,7 @@ private:
     std::shared_ptr<Handler> handler_;
 
     void dispatchPeer(const std::shared_ptr<Peer>& peer);
+    void runLoadThread();
 };
 
 } // namespace Tcp

src/os.cc

@@ -11,6 +11,7 @@
 #include <iterator>
 #include <algorithm>
 #include <sys/epoll.h>
+#include <sys/eventfd.h>
 
 using namespace std;
 
@@ -236,3 +237,37 @@ namespace Polling {
     }
 
 } // namespace Poller
+
+NotifyFd::NotifyFd() {
+    event_fd = TRY_RET(eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC));
+}
+
+Polling::Tag
+NotifyFd::tag() const {
+    return Polling::Tag(event_fd);
+}
+
+void
+NotifyFd::notify() const {
+    eventfd_t val = 1;
+    TRY(eventfd_write(event_fd, val));
+}
+
+void
+NotifyFd::read() const {
+    eventfd_t val;
+    TRY(eventfd_read(event_fd, &val));
+}
+
+bool
+NotifyFd::tryRead() const {
+    eventfd_t val;
+    int res = eventfd_read(event_fd, &val);
+    if (res == -1) {
+        if (errno == EAGAIN || errno == EWOULDBLOCK)
+            return false;
+        throw std::runtime_error("Failed to read eventfd");
+    }
+
+    return true;
+}

src/os.h

@@ -112,3 +112,19 @@ private:
 };
 
 } // namespace Polling
+
+class NotifyFd {
+public:
+    NotifyFd();
+
+    Polling::Tag tag() const;
+
+    void notify() const;
+
+    void read() const;
+    bool tryRead() const;
+
+private:
+    int event_fd;
+};
+