Added workers CPU pinning

src/http.cc

@@ -454,7 +454,14 @@ Endpoint::serve()
     if (!handler_)
         throw std::runtime_error("Must call setHandler() prior to serve()");
 
-    listener.init(24, Tcp::Options::InstallSignalHandler);
+    listener.init(4, Tcp::Options::InstallSignalHandler);
+#if 1
+    listener.pinWorker(0, CpuSet { 0 } );
+    listener.pinWorker(1, CpuSet { 1 } );
+    listener.pinWorker(2, CpuSet { 2 } );
+    listener.pinWorker(3, CpuSet { 3 } );
+#endif
+
     listener.setHandler(handler_);
 
     if (listener.bind()) { 

src/listener.cc

@@ -12,6 +12,7 @@
 #include <arpa/inet.h>
 #include <netdb.h>
 #include <sys/epoll.h>
+#include <pthread.h>
 #include <signal.h>
 #include <cassert>
 #include <cstring>
@@ -98,6 +99,23 @@ IoWorker::start(const std::shared_ptr<Handler>& handler, Flags<Options> options)
     thread.reset(new std::thread([this]() {
         this->run();
     }));
+
+    if (pins.count() > 0) {
+        auto cpuset = pins.toPosix();
+        auto handle = thread->native_handle();
+        pthread_setaffinity_np(handle, sizeof (cpuset), &cpuset);
+    }
+}
+
+void
+IoWorker::pin(const CpuSet& set) {
+    pins = set;
+
+    if (thread) {
+        auto cpuset = set.toPosix();
+        auto handle = thread->native_handle();
+        pthread_setaffinity_np(handle, sizeof (cpuset), &cpuset);
+    }
 }
 
 std::shared_ptr<Peer>
@@ -182,6 +200,9 @@ IoWorker::handleNewPeer(const std::shared_ptr<Peer>& peer)
 void
 IoWorker::run() {
 
+    if (pins.count() > 0) {
+    }
+
     mailbox.bind(poller);
 
     std::chrono::milliseconds timeout(-1);
@@ -271,6 +292,20 @@ Listener::setHandler(const std::shared_ptr<Handler>& handler)
     handler_ = handler;
 }
 
+void
+Listener::pinWorker(size_t worker, const CpuSet& set)
+{
+    if (ioGroup.empty()) {
+        throw std::domain_error("Invalid operation, did you call init() before ?");
+    }
+    if (worker > ioGroup.size()) {
+        throw std::invalid_argument("Trying to pin invalid worker");
+    }
+
+    auto &wrk = ioGroup[worker];
+    wrk->pin(set);
+}
+
 bool
 Listener::bind() {
     return bind(addr_);

src/listener.h

@@ -49,6 +49,8 @@ public:
     void start(const std::shared_ptr<Handler> &handler, Flags<Options> options);
     void handleNewPeer(const std::shared_ptr<Peer>& peer);
 
+    void pin(const CpuSet& set);
+
 private:
     Polling::Epoll poller;
     std::unique_ptr<std::thread> thread;
@@ -57,6 +59,8 @@ private:
     std::shared_ptr<Handler> handler_;
     Flags<Options> options_;
 
+    CpuSet pins;
+
     std::shared_ptr<Peer> getPeer(Fd fd) const;
     std::shared_ptr<Peer> getPeer(Polling::Tag tag) const;
 
@@ -87,6 +91,8 @@ public:
     Options options() const;
     Address address() const;
 
+    void pinWorker(size_t worker, const CpuSet& set);
+
 private: 
     Address addr_; 
     int listen_fd;

src/os.cc

@@ -12,6 +12,8 @@
 #include <algorithm>
 #include <sys/epoll.h>
 
+using namespace std;
+
 int hardware_concurrency() {
     std::ifstream cpuinfo("/proc/cpuinfo");
     if (cpuinfo) {
@@ -36,6 +38,104 @@ bool make_non_blocking(int sfd)
     return true;
 }
 
+CpuSet::CpuSet() {
+    bits.reset();
+}
+
+CpuSet::CpuSet(std::initializer_list<size_t> cpus) {
+    set(cpus);
+}
+
+void
+CpuSet::clear() {
+    bits.reset();
+}
+
+CpuSet&
+CpuSet::set(size_t cpu) {
+    if (cpu >= Size) {
+        throw std::invalid_argument("Trying to set invalid cpu number");
+    }
+
+    bits.set(cpu);
+    return *this;
+}
+
+CpuSet&
+CpuSet::unset(size_t cpu) {
+    if (cpu >= Size) {
+        throw std::invalid_argument("Trying to unset invalid cpu number");
+    }
+
+    bits.set(cpu, false);
+    return *this;
+}
+
+CpuSet&
+CpuSet::set(std::initializer_list<size_t> cpus) {
+    for (auto cpu: cpus) set(cpu);
+    return *this;
+}
+
+CpuSet&
+CpuSet::unset(std::initializer_list<size_t> cpus) {
+    for (auto cpu: cpus) unset(cpu);
+    return *this;
+}
+
+CpuSet&
+CpuSet::setRange(size_t begin, size_t end) {
+    if (begin > end) {
+        throw std::range_error("Invalid range, begin > end");
+    }
+
+    for (size_t cpu = begin; cpu < end; ++cpu) {
+        set(cpu);
+    }
+
+    return *this;
+}
+
+CpuSet&
+CpuSet::unsetRange(size_t begin, size_t end) {
+    if (begin > end) {
+        throw std::range_error("Invalid range, begin > end");
+    }
+
+    for (size_t cpu = begin; cpu < end; ++cpu) {
+        unset(cpu);
+    }
+
+    return *this;
+}
+
+bool
+CpuSet::isset(size_t cpu) const {
+    if (cpu >= Size) {
+        throw std::invalid_argument("Trying to test invalid cpu number");
+    }
+
+    return bits.test(cpu);
+}
+
+size_t
+CpuSet::count() const {
+    return bits.count();
+}
+
+cpu_set_t
+CpuSet::toPosix() const {
+    cpu_set_t cpu_set;
+    CPU_ZERO(&cpu_set);
+
+    for (size_t cpu = 0; cpu < Size; ++cpu) {
+        if (bits.test(cpu))
+            CPU_SET(cpu, &cpu_set);
+    }
+
+    return cpu_set;
+};
+
 namespace Polling {
 
     Epoll::Epoll(size_t max) {

src/os.h

@@ -8,6 +8,8 @@
 
 #include <chrono>
 #include <vector>
+#include <bitset>
+#include <sched.h>
 #include "flags.h"
 #include "common.h"
 
@@ -16,6 +18,32 @@ typedef int Fd;
 int hardware_concurrency();
 bool make_non_blocking(int fd);
 
+class CpuSet {
+public:
+    static constexpr size_t Size = 1024;
+
+    CpuSet();
+    explicit CpuSet(std::initializer_list<size_t> cpus);
+
+    void clear();
+    CpuSet& set(size_t cpu);
+    CpuSet& unset(size_t cpu);
+
+    CpuSet& set(std::initializer_list<size_t> cpus);
+    CpuSet& unset(std::initializer_list<size_t> cpus);
+
+    CpuSet& setRange(size_t begin, size_t end);
+    CpuSet& unsetRange(size_t begin, size_t end);
+
+    bool isset(size_t cpu) const;
+    size_t count() const;
+
+    cpu_set_t toPosix() const;
+
+private:
+    std::bitset<Size> bits;
+};
+
 namespace Polling {
 
 enum class Mode {