Extend thread config to include affinity

The ThreadConfig structure should be extended to include an
isolcpus-style affinity mask, along with helper functions for parsing
and setting thread affinity.

Closes reactivemarkets#49

etc/cppcheck-suppressions.txt

@@ -1,4 +1,5 @@
 cppcheckError:*/toolbox/util/IntTypes.ut.cpp
+syntaxError:*/toolbox/sys/Thread.cpp
 preprocessorErrorDirective:*/toolbox/net/Error.cpp
 preprocessorErrorDirective:*/toolbox/sys/Error.cpp
 returnDanglingLifetime:*/toolbox/io/Timer.cpp

example/EchoClnt.cpp

@@ -188,8 +188,8 @@ int main(int argc, char* argv[])
 
         // Start service threads.
         pthread_setname_np(pthread_self(), "main");
-        ReactorRunner reactor_runner{reactor, ThreadConfig{"reactor"s}};
-        ResolverRunner resolver_runner{resolver, ThreadConfig{"resolver"s}};
+        ReactorRunner reactor_runner{reactor, "reactor"s};
+        ResolverRunner resolver_runner{resolver, "resolver"s};
 
         // Wait for termination.
         SigWait sig_wait;

example/EchoServ.cpp

@@ -149,7 +149,7 @@ int main(int argc, char* argv[])
 
         // Start service threads.
         pthread_setname_np(pthread_self(), "main");
-        ReactorRunner reactor_runner{reactor, ThreadConfig{"reactor"s}};
+        ReactorRunner reactor_runner{reactor, "reactor"s};
 
         // Wait for termination.
         SigWait sig_wait;

example/HttpServ.cpp

@@ -96,7 +96,7 @@ int main(int argc, char* argv[])
 
         // Start service threads.
         pthread_setname_np(pthread_self(), "main");
-        ReactorRunner reactor_runner{reactor, ThreadConfig{"reactor"s}};
+        ReactorRunner reactor_runner{reactor, "reactor"s};
 
         // Wait for termination.
         SigWait sig_wait;

toolbox/CMakeLists.txt

@@ -220,6 +220,7 @@ set(test_SOURCES
   net/Socket.ut.cpp
   sys/Date.ut.cpp
   sys/Log.ut.cpp
+  sys/Thread.ut.cpp
   sys/Time.ut.cpp
   util/Argv.ut.cpp
   util/Array.ut.cpp

toolbox/io/Runner.cpp

@@ -27,9 +27,9 @@ namespace {
 void run_reactor(Reactor& r, ThreadConfig config, const std::atomic<bool>& stop)
 {
     sig_block_all();
-    pthread_setname_np(pthread_self(), config.name.c_str());
-    TOOLBOX_NOTICE << "started " << config.name << " thread";
     try {
+        set_thread_attrs(config);
+        TOOLBOX_NOTICE << "started " << config.name << " thread";
         while (!stop.load(std::memory_order_acquire)) {
             r.poll(CyclTime::now());
         }

toolbox/net/Runner.cpp

@@ -29,9 +29,9 @@ namespace {
 void run_resolver(Resolver& r, ThreadConfig config)
 {
     sig_block_all();
-    pthread_setname_np(pthread_self(), config.name.c_str());
-    TOOLBOX_NOTICE << "started " << config.name << " thread";
     try {
+        set_thread_attrs(config);
+        TOOLBOX_NOTICE << "started " << config.name << " thread";
         // The run() function returns -1 when resolver is closed.
         while (r.run() >= 0)
             ;

toolbox/net/Runner.ut.cpp

@@ -31,7 +31,7 @@ BOOST_AUTO_TEST_CASE(ResolverRunnerCase)
     Resolver res;
 
     pthread_setname_np(pthread_self(), "main");
-    ResolverRunner resolver_runner{res, ThreadConfig{"resolver"s}};
+    ResolverRunner resolver_runner{res, "resolver"s};
 
     const auto uri1 = "tcp4://192.168.1.3:443"s;
     const auto uri2 = "tcp6://[fe80::c8bf:7d86:cbdc:bda9]:443"s;

toolbox/sys/Thread.cpp

@@ -15,3 +15,53 @@
 // limitations under the License.
 
 #include "Thread.hpp"
+
+#include <toolbox/sys/Error.hpp>
+#include <toolbox/util/Tokeniser.hpp>
+
+namespace toolbox {
+inline namespace sys {
+using namespace std;
+namespace {
+pair<int, int> split_range(string_view s) noexcept
+{
+    auto [first, last] = split_pair(s, '-');
+    const auto i = ston<int>(first);
+    return {i, last.empty() ? i : ston<int>(last)};
+}
+} // namespace
+
+cpu_set_t parse_cpu_set(string_view s) noexcept
+{
+    cpu_set_t bs;
+    CPU_ZERO(&bs);
+
+    Tokeniser toks{s, ","sv};
+    while (!toks.empty()) {
+        auto [i, j] = split_range(toks.top());
+        for (; i <= j; ++i) {
+            CPU_SET(i, &bs);
+        }
+        toks.pop();
+    }
+    return bs;
+}
+
+void set_thread_attrs(const ThreadConfig& config)
+{
+    const auto tid = pthread_self();
+    if (!config.name.empty()) {
+        if (const auto err = pthread_setname_np(tid, config.name.c_str()); err != 0) {
+            throw system_error{make_sys_error(err), "pthread_setname_np"};
+        }
+    }
+    if (!config.affinity.empty()) {
+        const auto bs = parse_cpu_set(config.affinity);
+        if (const auto err = pthread_setaffinity_np(tid, sizeof(bs), &bs); err != 0) {
+            throw system_error{make_sys_error(err), "pthread_setaffinity_np"};
+        }
+    }
+}
+
+} // namespace sys
+} // namespace toolbox

toolbox/sys/Thread.hpp

@@ -24,10 +24,39 @@
 namespace toolbox {
 inline namespace sys {
 
+/// ThreadConfig holds the thread attributes.
 struct ThreadConfig {
+    ThreadConfig(std::string name, std::string affinity) noexcept
+    : name{std::move(name)}
+    , affinity{std::move(affinity)}
+    {
+    }
+    ThreadConfig(std::string name) noexcept
+    : name{std::move(name)}
+    {
+    }
+    ThreadConfig() noexcept = default;
+
+    /// The thread's name.
     std::string name;
+    /// The thread's affinity.
+    std::string affinity;
 };
 
+/// Parse an isolcpus-style set of CPUs.
+///
+/// The CPU set is either a list: "<cpu>,...,<cpu>", or a range: "<cpu>-<cpu>", or a combination or
+/// both: "<cpu>,...,<cpu>-<cpu>", where "<cpu>" begins at 0 and the maximum value is "number of
+/// CPUs in system - 1". For example: "0,1,10-11,22-23".
+///
+/// \param s An isolcpus-style set of CPUs.
+/// \return a copy of the CPU set.
+TOOLBOX_API cpu_set_t parse_cpu_set(std::string_view s) noexcept;
+
+/// Set attributes for current thread from config.
+/// \param config The configuration.
+TOOLBOX_API void set_thread_attrs(const ThreadConfig& config);
+
 } // namespace sys
 } // namespace toolbox
 

toolbox/sys/Thread.ut.cpp

@@ -0,0 +1,49 @@
+// The Reactive C++ Toolbox.
+// Copyright (C) 2013-2019 Swirly Cloud Limited
+// Copyright (C) 2019 Reactive Markets Limited
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "Thread.hpp"
+
+#include <boost/test/unit_test.hpp>
+
+using namespace std;
+using namespace toolbox;
+
+namespace {
+uint32_t to_bitset(const cpu_set_t& cpuset) noexcept
+{
+    uint32_t bs{0};
+    for (int i{0}; i < 32; ++i) {
+        if (CPU_ISSET(i, &cpuset)) {
+            bs |= (1 << i);
+        }
+    }
+    return bs;
+}
+} // namespace
+
+BOOST_AUTO_TEST_SUITE(ThreadSuite)
+
+BOOST_AUTO_TEST_CASE(ThreadParseCpuSetCase)
+{
+    BOOST_TEST(to_bitset(parse_cpu_set(""sv)) == 0b0);
+    BOOST_TEST(to_bitset(parse_cpu_set("0"sv)) == 0b1);
+    BOOST_TEST(to_bitset(parse_cpu_set("1,2"sv)) == 0b110);
+    BOOST_TEST(to_bitset(parse_cpu_set("0-3"sv)) == 0b1111);
+    BOOST_TEST(to_bitset(parse_cpu_set("1,2,4-7"sv)) == 0b11110110);
+    BOOST_TEST(to_bitset(parse_cpu_set("0-3,5,6,8-11"sv)) == 0b111101101111);
+}
+
+BOOST_AUTO_TEST_SUITE_END()