[19435] Apply thread settings

include/fastdds/rtps/attributes/ThreadSettings.hpp

@@ -59,16 +59,16 @@ struct RTPS_DllAPI ThreadSettings
     int32_t priority = 0;
 
     /**
-     * @brief The thread's core affinity.
+     * @brief The thread's affinity.
      *
-     * cpu_mask is a bit mask for setting the threads affinity to each core individually.
+     * On some systems, this is a bit mask for setting the threads affinity to each core individually.
      * A value of 0 indicates no particular affinity.
      *
      * This value is platform specific and it is used as-is to configure the specific platform thread.
      * Setting this value to something other than the default one may require different privileges
      * on different platforms.
      */
-    uint32_t cpu_mask = 0;
+    uint64_t affinity = 0;
 
     /**
      * @brief The thread's stack size in bytes.

src/cpp/rtps/attributes/ThreadSettings.cpp

@@ -21,7 +21,7 @@ bool ThreadSettings::operator ==(
 {
     return (scheduling_policy == rhs.scheduling_policy &&
            priority == rhs.priority &&
-           cpu_mask == rhs.cpu_mask &&
+           affinity == rhs.affinity &&
            stack_size == rhs.stack_size);
 }
 

src/cpp/utils/threading/threading_osx.ipp

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <limits>
+
 #include <pthread.h>
 #include <string.h>
 #include <stdio.h>
@@ -103,18 +105,21 @@ static void configure_current_thread_scheduler(
 }
 
 static void configure_current_thread_affinity(
-        uint32_t affinity_mask)
+        uint64_t affinity)
 {
-    thread_affinity_policy_data_t policy = { m_affinityMask };
-    pthread_t self_tid = pthread_self();
-    thread_policy_set(pthread_mach_thread_np(self_tid), THREAD_AFFINITY_POLICY, (thread_policy_t)&policy, 1);
+    if (affinity <= static_cast<uint64_t>(std::numeric_limits<integer_t>::max()))
+    {
+        thread_affinity_policy_data_t policy = { static_cast<integer_t>(affinity) };
+        pthread_t self_tid = pthread_self();
+        thread_policy_set(pthread_mach_thread_np(self_tid), THREAD_AFFINITY_POLICY, (thread_policy_t)&policy, 1);
+    }
 }
 
 void apply_thread_settings_to_current_thread(
         const fastdds::rtps::ThreadSettings& settings)
 {
     configure_current_thread_scheduler(settings.scheduling_policy, settings.priority);
-    configure_current_thread_affinity(settings.cpu_mask);
+    configure_current_thread_affinity(settings.affinity);
 }
 
 }  // namespace eprosima

src/cpp/utils/threading/threading_pthread.ipp

@@ -108,12 +108,12 @@ static void configure_current_thread_scheduler(
 }
 
 static void configure_current_thread_affinity(
-        uint32_t affinity_mask)
-{    
+        uint64_t affinity_mask)
+{
     int a;
     int result;
     int cpu_count;
-    cpu_set_t  cpu_set;
+    cpu_set_t cpu_set;
     pthread_t self_tid = pthread_self();
 
     result = 0;
@@ -129,16 +129,21 @@ static void configure_current_thread_affinity(
     // We only consider up to the total number of CPU's the
     // system has.
     //
-
     cpu_count = get_nprocs_conf();
-    
+
     for(a = 0; a < cpu_count; a++)
     {
-        if(affinity_mask & (1 << a))
-        {            
+        if(0 != (affinity_mask & 1))
+        {
             CPU_SET(a, &cpu_set);
             result++;
         }
+        affinity_mask >>= 1;
+    }
+
+    if (affinity_mask > 0)
+    {
+        EPROSIMA_LOG_ERROR(SYSTEM, "Affinity mask has more processors than the ones present in the system");
     }
 
     if(result > 0)
@@ -156,7 +161,7 @@ void apply_thread_settings_to_current_thread(
         const fastdds::rtps::ThreadSettings& settings)
 {
     configure_current_thread_scheduler(settings.scheduling_policy, settings.priority);
-    configure_current_thread_affinity(settings.cpu_mask);
+    configure_current_thread_affinity(settings.affinity);
 }
 
 }  // namespace eprosima

test/unittest/dds/participant/ParticipantTests.cpp

@@ -3002,23 +3002,23 @@ TEST(ParticipantTests, UpdatableDomainParticipantQos)
 
     // Check that the builtin_controllers_sender_thread can not be changed in an enabled participant
     participant->get_qos(pqos);
-    pqos.builtin_controllers_sender_thread().cpu_mask = 1;
+    pqos.builtin_controllers_sender_thread().affinity = 1;
     ASSERT_EQ(participant->set_qos(pqos), ReturnCode_t::RETCODE_IMMUTABLE_POLICY);
 
     // Check that the timed_events_thread can not be changed in an enabled participant
     participant->get_qos(pqos);
-    pqos.timed_events_thread().cpu_mask = 1;
+    pqos.timed_events_thread().affinity = 1;
     ASSERT_EQ(participant->set_qos(pqos), ReturnCode_t::RETCODE_IMMUTABLE_POLICY);
 
     // Check that the discovery_server_thread can not be changed in an enabled participant
     participant->get_qos(pqos);
-    pqos.discovery_server_thread().cpu_mask = 1;
+    pqos.discovery_server_thread().affinity = 1;
     ASSERT_EQ(participant->set_qos(pqos), ReturnCode_t::RETCODE_IMMUTABLE_POLICY);
 
 #if HAVE_SECURITY
     // Check that the security_log_thread can not be changed in an enabled participant
     participant->get_qos(pqos);
-    pqos.security_log_thread().cpu_mask = 1;
+    pqos.security_log_thread().affinity = 1;
     ASSERT_EQ(participant->set_qos(pqos), ReturnCode_t::RETCODE_IMMUTABLE_POLICY);
 
     ASSERT_EQ(DomainParticipantFactory::get_instance()->delete_participant(participant), ReturnCode_t::RETCODE_OK);

test/unittest/logging/LogTests.cpp

@@ -673,7 +673,7 @@ TEST_F(LogTests, thread_config)
     // Set thread settings
     eprosima::fastdds::rtps::ThreadSettings thr_settings{};
 #if defined(_POSIX_SOURCE)
-    thr_settings.cpu_mask = 3;
+    thr_settings.affinity = 3;
     thr_settings.scheduling_policy = SCHED_OTHER;
     thr_settings.priority = 1;
 #endif // if defined(_POSIX_SOURCE)

test/unittest/rtps/attributes/ThreadSettingsTests.cpp

@@ -27,116 +27,116 @@ TEST(ThreadSettingsTests, EqualOperators)
     // Fixed scheduling_policy cases
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     // Fixed priority cases (not already covered)
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
-    // Fixed cpu_mask cases (not already covered)
+    // Fixed affinity cases (not already covered)
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     // Fixed stack_size cases (not already covered)
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask;
+    settings_2.affinity = settings_1.affinity;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);
 
     // All different
     settings_2.scheduling_policy = settings_1.scheduling_policy + 1;
     settings_2.priority = settings_1.priority + 1;
-    settings_2.cpu_mask = settings_1.cpu_mask + 1;
+    settings_2.affinity = settings_1.affinity + 1;
     settings_2.stack_size = settings_1.stack_size + 1;
     ASSERT_FALSE(settings_1 == settings_2);
     ASSERT_TRUE(settings_1 != settings_2);

test/unittest/transport/PortBasedTransportDescriptorTests.cpp

@@ -65,7 +65,7 @@ TEST_F(PortBasedTransportDescriptorTests, get_thread_config_for_port)
     PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings;
     set_settings[1234].scheduling_policy = 33;
     set_settings[1234].priority = 33;
-    set_settings[1234].cpu_mask = 33;
+    set_settings[1234].affinity = 33;
     set_settings[1234].stack_size = 33;
 
     ASSERT_TRUE(reception_threads(set_settings));
@@ -86,7 +86,7 @@ TEST_F(PortBasedTransportDescriptorTests, set_thread_config_for_port)
     PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings;
     set_settings[1234].scheduling_policy = 33;
     set_settings[1234].priority = 33;
-    set_settings[1234].cpu_mask = 33;
+    set_settings[1234].affinity = 33;
     set_settings[1234].stack_size = 33;
     ASSERT_TRUE(reception_threads(set_settings));
 
@@ -116,7 +116,7 @@ TEST_F(PortBasedTransportDescriptorTests, set_default_reception_threads)
     ThreadSettings set_settings;
     set_settings.scheduling_policy = 33;
     set_settings.priority = 33;
-    set_settings.cpu_mask = 33;
+    set_settings.affinity = 33;
     set_settings.stack_size = 33;
 
     ASSERT_NE(initial_settings, set_settings);
@@ -137,7 +137,7 @@ TEST_F(PortBasedTransportDescriptorTests, set_reception_threads)
     PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings;
     set_settings[1234].scheduling_policy = 33;
     set_settings[1234].priority = 33;
-    set_settings[1234].cpu_mask = 33;
+    set_settings[1234].affinity = 33;
     set_settings[1234].stack_size = 33;
 
     ASSERT_NE(initial_settings, set_settings);
@@ -170,7 +170,7 @@ TEST_F(PortBasedTransportDescriptorTests, equal_operator)
         ThreadSettings set_settings;
         set_settings.scheduling_policy = 33;
         set_settings.priority = 33;
-        set_settings.cpu_mask = 33;
+        set_settings.affinity = 33;
         set_settings.stack_size = 33;
         other.default_reception_threads(set_settings);
         ASSERT_FALSE(*this == other);
@@ -185,7 +185,7 @@ TEST_F(PortBasedTransportDescriptorTests, equal_operator)
         PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings_map;
         set_settings_map[1234].scheduling_policy = 33;
         set_settings_map[1234].priority = 33;
-        set_settings_map[1234].cpu_mask = 33;
+        set_settings_map[1234].affinity = 33;
         set_settings_map[1234].stack_size = 33;
         ASSERT_TRUE(other.reception_threads(set_settings_map));
         ASSERT_FALSE(*this == other);
@@ -202,7 +202,7 @@ TEST_F(PortBasedTransportDescriptorTests, equal_operator)
         ThreadSettings set_settings;
         set_settings.scheduling_policy = 33;
         set_settings.priority = 33;
-        set_settings.cpu_mask = 33;
+        set_settings.affinity = 33;
         set_settings.stack_size = 33;
         other.default_reception_threads(set_settings);
 
@@ -220,7 +220,7 @@ TEST_F(PortBasedTransportDescriptorTests, equal_operator)
         PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings_map;
         set_settings_map[1234].scheduling_policy = 33;
         set_settings_map[1234].priority = 33;
-        set_settings_map[1234].cpu_mask = 33;
+        set_settings_map[1234].affinity = 33;
         set_settings_map[1234].stack_size = 33;
         ASSERT_TRUE(other.reception_threads(set_settings_map));
 
@@ -236,14 +236,14 @@ TEST_F(PortBasedTransportDescriptorTests, equal_operator)
         ThreadSettings set_settings;
         set_settings.scheduling_policy = 33;
         set_settings.priority = 33;
-        set_settings.cpu_mask = 33;
+        set_settings.affinity = 33;
         set_settings.stack_size = 33;
         other.default_reception_threads(set_settings);
 
         PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings_map;
         set_settings_map[1234].scheduling_policy = 33;
         set_settings_map[1234].priority = 33;
-        set_settings_map[1234].cpu_mask = 33;
+        set_settings_map[1234].affinity = 33;
         set_settings_map[1234].stack_size = 33;
         ASSERT_TRUE(other.reception_threads(set_settings_map));
 
@@ -261,14 +261,14 @@ TEST_F(PortBasedTransportDescriptorTests, equal_operator)
         ThreadSettings set_settings;
         set_settings.scheduling_policy = 33;
         set_settings.priority = 33;
-        set_settings.cpu_mask = 33;
+        set_settings.affinity = 33;
         set_settings.stack_size = 33;
         other.default_reception_threads(set_settings);
 
         PortBasedTransportDescriptor::ReceptionThreadsConfigMap set_settings_map;
         set_settings_map[1234].scheduling_policy = 33;
         set_settings_map[1234].priority = 33;
-        set_settings_map[1234].cpu_mask = 33;
+        set_settings_map[1234].affinity = 33;
         set_settings_map[1234].stack_size = 33;
         ASSERT_TRUE(other.reception_threads(set_settings_map));