Commit e52871cb authored by Mathieu Stefani's avatar Mathieu Stefani

Now providing a much more convenient API for rusage

parent 145f472c
...@@ -32,15 +32,11 @@ class MyHandler : public Net::Http::Handler { ...@@ -32,15 +32,11 @@ class MyHandler : public Net::Http::Handler {
.add<Header::Server>("lys") .add<Header::Server>("lys")
.add<Header::ContentType>(MIME(Text, Plain)); .add<Header::ContentType>(MIME(Text, Plain));
#if 0
auto stream = response.stream(Net::Http::Code::Ok); auto stream = response.stream(Net::Http::Code::Ok);
stream << "PO"; stream << "PO";
stream << flush;
stream << "NG"; stream << "NG";
stream << ends; stream << ends;
#endif
response.send(Code::Ok, "PONG");
} }
} }
else if (req.resource() == "/echo") { else if (req.resource() == "/echo") {
...@@ -63,6 +59,50 @@ class MyHandler : public Net::Http::Handler { ...@@ -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[]) { int main(int argc, char *argv[]) {
Net::Port port(9080); Net::Port port(9080);
...@@ -81,12 +121,17 @@ int main(int argc, char *argv[]) { ...@@ -81,12 +121,17 @@ int main(int argc, char *argv[]) {
cout << "Cores = " << hardware_concurrency() << endl; cout << "Cores = " << hardware_concurrency() << endl;
cout << "Using " << thr << " threads" << 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() auto opts = Net::Http::Endpoint::options()
.threads(thr) .threads(thr)
.flags(Net::Tcp::Options::InstallSignalHandler); .flags(Net::Tcp::Options::InstallSignalHandler);
server.init(opts); server->init(opts);
server.setHandler(std::make_shared<MyHandler>()); server->setHandler(std::make_shared<MyHandler>());
LoadMonitor monitor(server);
monitor.setInterval(std::chrono::seconds(5));
monitor.start();
server.serve(); server->serve();
} }
...@@ -586,6 +586,11 @@ Endpoint::serve() ...@@ -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
Endpoint::options() { Endpoint::options() {
return Options(); return Options();
......
...@@ -530,6 +530,12 @@ public: ...@@ -530,6 +530,12 @@ public:
void setHandler(const std::shared_ptr<Handler>& handler); void setHandler(const std::shared_ptr<Handler>& handler);
void serve(); void serve();
bool isBound() const {
return listener.isBound();
}
Async::Promise<Tcp::Listener::Load> requestLoad(const Tcp::Listener::Load& old);
static Options options(); static Options options();
private: private:
......
...@@ -176,14 +176,14 @@ Listener::bind(const Address& address) { ...@@ -176,14 +176,14 @@ Listener::bind(const Address& address) {
io->start(handler_, options_); io->start(handler_, options_);
} }
sh = false;
loadThread.reset(new std::thread([=]() {
this->runLoadThread();
}));
return true; return true;
} }
bool
Listener::isBound() const {
return g_listen_fd != -1;
}
void void
Listener::run() { Listener::run() {
for (;;) { for (;;) {
...@@ -211,52 +211,59 @@ Listener::run() { ...@@ -211,52 +211,59 @@ Listener::run() {
} }
void void
Listener::runLoadThread() { Listener::shutdown() {
std::vector<rusage> lastUsages; for (auto &worker: ioGroup) {
worker->shutdown();
}
}
while (!sh) { Async::Promise<Listener::Load>
std::vector<Async::Promise<rusage>> loads; Listener::requestLoad(const Listener::Load& old) {
loads.reserve(ioGroup.size()); std::vector<Async::Promise<rusage>> loads;
loads.reserve(ioGroup.size());
for (const auto& io: ioGroup) { for (const auto& io: ioGroup) {
loads.push_back(io->getLoad()); loads.push_back(io->getLoad());
} }
Async::whenAll(std::begin(loads), std::end(loads)) return Async::whenAll(std::begin(loads), std::end(loads)).then([=](const std::vector<rusage>& usages) {
.then([&](const std::vector<rusage>& usages) {
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);
};
if (lastUsages.empty()) lastUsages = usages; Load res;
else { res.raw = usages;
for (size_t i = 0; i < usages.size(); ++i) {
auto last = lastUsages[i];
const auto& usage = usages[i];
auto now = totalElapsed(usage); if (old.raw.empty()) {
auto time = now - totalElapsed(last); res.global = 0.0;
for (size_t i = 0; i < ioGroup.size(); ++i) res.workers.push_back(0.0);
} else {
auto load = (time * 100.0) / 1e6; 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);
};
//printf("Total load for I/O thread %lu = %.3lf%%\n", i, load); 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) {
lastUsages = usages; auto last = old.raw[i];
} const auto& usage = usages[i];
}, Async::NoExcept);
std::this_thread::sleep_for(std::chrono::seconds(1)); auto nowElapsed = totalElapsed(usage);
} auto timeElapsed = nowElapsed - totalElapsed(last);
}
void auto loadPct = (timeElapsed * 100.0) / tick.count();
Listener::shutdown() { res.workers.push_back(loadPct);
for (auto &worker: ioGroup) { res.global += loadPct;
worker->shutdown();
} }
sh = true;
res.global /= usages.size();
}
return Async::Promise<Load>::resolved(std::move(res));
}, Async::NoExcept);
} }
Address Address
......
...@@ -26,10 +26,16 @@ void setSocketOptions(Fd fd, Flags<Options> options); ...@@ -26,10 +26,16 @@ void setSocketOptions(Fd fd, Flags<Options> options);
class Listener { class Listener {
public: 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();
~Listener() {
loadThread->join();
}
Listener(const Address& address); Listener(const Address& address);
void init( void init(
...@@ -40,9 +46,13 @@ public: ...@@ -40,9 +46,13 @@ public:
bool bind(); bool bind();
bool bind(const Address& adress); bool bind(const Address& adress);
bool isBound() const;
void run(); void run();
void shutdown(); void shutdown();
Async::Promise<Load> requestLoad(const Load& old);
Options options() const; Options options() const;
Address address() const; Address address() const;
...@@ -56,10 +66,6 @@ private: ...@@ -56,10 +66,6 @@ private:
Flags<Options> options_; Flags<Options> options_;
std::shared_ptr<Handler> handler_; std::shared_ptr<Handler> handler_;
std::atomic<bool> sh;
std::unique_ptr<std::thread> loadThread;
void dispatchPeer(const std::shared_ptr<Peer>& peer); void dispatchPeer(const std::shared_ptr<Peer>& peer);
void runLoadThread(); void runLoadThread();
}; };
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment