Merge pull request #1044 from leapmotion/feat-thread-sched-policy

Add support for adjusting the thread scheduling policy

src/autowiring/BasicThread.cpp

@@ -14,17 +14,40 @@
 
 using namespace autowiring;
 
+std::atomic<SchedulingPolicy> BasicThread::s_schedulingPolicy{ SchedulingPolicy::StandardRoundRobin };
+
 static auto mainTID = std::this_thread::get_id();
 
 BasicThread::BasicThread(const char* pName):
   ContextMember(pName),
-  m_state(std::make_shared<BasicThreadStateBlock>())
+  m_state(std::make_shared<BasicThreadStateBlock>(ThreadPriority::Default, s_schedulingPolicy))
+{}
+
+BasicThread::BasicThread(ThreadPriority threadPriority, const char* pName):
+  ContextMember(pName),
+  m_state(std::make_shared<BasicThreadStateBlock>(threadPriority, s_schedulingPolicy))
+{}
+
+BasicThread::BasicThread(ThreadPriority threadPriority, SchedulingPolicy schedPolicy, const char* pName):
+ ContextMember(pName),
+ m_state(std::make_shared<BasicThreadStateBlock>(threadPriority, schedPolicy))
 {}
 
 BasicThread::~BasicThread(void) {
   NotifyTeardownListeners();
 }
 
+SchedulingPolicy BasicThread::SetDefaultSchedulingPolicy(SchedulingPolicy schedPolicy) {
+  SchedulingPolicy oldSchedPolicy = s_schedulingPolicy;
+  while (!s_schedulingPolicy.compare_exchange_strong(oldSchedPolicy, schedPolicy))
+    ; // empty body
+  return oldSchedPolicy;
+}
+
+SchedulingPolicy BasicThread::GetDefaultSchedulingPolicy(void) {
+  return s_schedulingPolicy;
+}
+
 std::mutex& BasicThread::GetLock(void) const {
   return m_state->m_lock;
 }
@@ -212,41 +235,55 @@ bool BasicThread::IsMainThread(void) {
 
 ThreadPriority BasicThread::GetThreadPriority(void) {
   std::lock_guard<std::mutex> lk(m_state->m_threadLock);
-  return m_state->m_priority;
+  return m_state->m_threadPriority;
+}
+
+SchedulingPolicy BasicThread::GetSchedulingPolicy(void) {
+  std::lock_guard<std::mutex> lk(m_state->m_threadLock);
+  return m_state->m_schedPolicy;
 }
 
 void BasicThread::UpdateThreadPriority(std::unique_lock<std::mutex>&& lock) {
   if (m_state->m_thisThread.get_id() == std::thread::id())
     return;
-  SetThreadPriority(m_state->m_thisThread.native_handle(), m_state->m_priority);
+  SetThreadPriority(m_state->m_thisThread.native_handle(), m_state->m_threadPriority, m_state->m_schedPolicy);
 }
 
 ThreadPriority BasicThread::SetThreadPriority(ThreadPriority threadPriority) {
   std::unique_lock<std::mutex> lk(m_state->m_threadLock);
-  ThreadPriority prevThreadPriority = m_state->m_priority;
-  m_state->m_priority = threadPriority;
+  ThreadPriority prevThreadPriority = m_state->m_threadPriority;
+  m_state->m_threadPriority = threadPriority;
+  UpdateThreadPriority(std::move(lk));
+  return prevThreadPriority;
+}
+
+ThreadPriority BasicThread::SetThreadPriority(ThreadPriority threadPriority, SchedulingPolicy schedPolicy) {
+  std::unique_lock<std::mutex> lk(m_state->m_threadLock);
+  ThreadPriority prevThreadPriority = m_state->m_threadPriority;
+  m_state->m_threadPriority = threadPriority;
+  m_state->m_schedPolicy = schedPolicy;
   UpdateThreadPriority(std::move(lk));
   return prevThreadPriority;
 }
 
 ThreadPriority BasicThread::ElevateThreadPriority(ThreadPriority threadPriority) {
   std::unique_lock<std::mutex> lk(m_state->m_threadLock);
-  ThreadPriority prevThreadPriority = m_state->m_priority;
-  if (threadPriority < m_state->m_priority) {
+  ThreadPriority prevThreadPriority = m_state->m_threadPriority;
+  if (threadPriority < m_state->m_threadPriority) {
     return prevThreadPriority;
   }
-  m_state->m_priority = threadPriority;
+  m_state->m_threadPriority = threadPriority;
   UpdateThreadPriority(std::move(lk));
   return prevThreadPriority;
 }
 
 ThreadPriority BasicThread::DeelevateThreadPriority(ThreadPriority threadPriority) {
   std::unique_lock<std::mutex> lk(m_state->m_threadLock);
-  ThreadPriority prevThreadPriority = m_state->m_priority;
-  if (threadPriority >= m_state->m_priority) {
+  ThreadPriority prevThreadPriority = m_state->m_threadPriority;
+  if (threadPriority >= m_state->m_threadPriority) {
     return prevThreadPriority;
   }
-  m_state->m_priority = threadPriority;
+  m_state->m_threadPriority = threadPriority;
   UpdateThreadPriority(std::move(lk));
   return prevThreadPriority;
 }

src/autowiring/BasicThread.h

@@ -7,6 +7,7 @@
 #include MEMORY_HEADER
 #include MUTEX_HEADER
 #include THREAD_HEADER
+#include <atomic>
 
 class BasicThread;
 class CoreContext;
@@ -19,8 +20,9 @@ namespace autowiring {
 /// Thread priority classifications from low to high.
 /// </summary>
 /// <remarks>
-/// Use the ThreadPriority enumeration with the BasicThread::ElevatePriority class
-/// to raise the priority of a thread when needed.
+/// Use the ThreadPriority enumeration with BasicThread::BasicThread,
+/// BasicThread::SetThreadPriority, or the BasicThread::ElevatePriority class
+/// to adjust the priority of a thread when needed.
 /// </remarks>
 enum class ThreadPriority {
   /// This is the default thread priority. It is treated as a value lower than any of the
@@ -43,6 +45,25 @@ enum class ThreadPriority {
   Multimedia
 };
 
+/// <summary>
+/// Scheduling policies.
+/// </summary>
+/// <remarks>
+/// Use the SchedulingPolicy enumeration with BasicThread::BasicThread,
+/// BasicThread::SetThreadPriority, BasicThread::SetSchedulingPolicy, or
+/// BasicThread::SetDefaultSchedulingPolicy. Not all platforms support the scheduling policy.
+/// </remarks>
+enum class SchedulingPolicy {
+  /// This is the default scheduling policy used by all new threads. It is a standard
+  /// round-robin policy. It will also run with a lower priority than any of the realtime
+  /// policies.
+  StandardRoundRobin,
+
+  /// For time-critical applications, realtime policies may be used:
+  RealtimeFIFO, ///< A realtime first-in, first-out policy
+  RealtimeRoundRobin ///< A realtime round-robin policy
+};
+
 /// <summary>
 /// An abstract class for creating a thread with a single Run method.
 /// </summary>
@@ -71,8 +92,27 @@ class BasicThread:
   /// Creates a BasicThread object.
   /// <param name="pName">An optional name for this thread. The name is visible in some debuggers.</param>
   BasicThread(const char* pName = nullptr);
+  /// Creates a BasicThread object with the specified thread priority and name.
+  /// <param name="threadPriority">A priority for this thread.</param>
+  /// <param name="pName">An optional name for this thread. The name is visible in some debuggers.</param>
+  BasicThread(ThreadPriority threadPriority, const char* pName = nullptr);
+  /// Creates a BasicThread object with the specified thread priority, scheduling policy, and name.
+  /// <param name="threadPriority">A priority for this thread.</param>
+  /// <param name="schedPolicy">A scheduling policy for this thread.</param>
+  /// <param name="pName">An optional name for this thread. The name is visible in some debuggers.</param>
+  BasicThread(ThreadPriority threadPriority, SchedulingPolicy schedPolicy, const char* pName = nullptr);
   virtual ~BasicThread(void);
 
+  /// <summary>
+  /// Set the default scheduling policy to use for newly created threads
+  /// </summary>
+  static SchedulingPolicy SetDefaultSchedulingPolicy(SchedulingPolicy schedPolicy);
+
+  /// <summary>
+  /// Get the current default scheduling policy
+  /// </summary>
+  static SchedulingPolicy GetDefaultSchedulingPolicy(void);
+
   /// \internal Only implemented on Windows (as of 0.4.1).
   /// <summary>
   /// Boosts thread priority while an instance of this type exists.
@@ -93,10 +133,10 @@ class BasicThread:
     /// Elevates the priority of a BasicThread instance if the specified priority is higher
     /// then the current thread priority. Destroy this ElevatePriority instance
     /// to restore the normal thread priority.
-    ElevatePriority(BasicThread& thread, ThreadPriority priority) :
+    ElevatePriority(BasicThread& thread, ThreadPriority threadPriority) :
       m_thread(thread),
       // Elevate if the new level is greater than the current level:
-      m_oldPriority(thread.ElevateThreadPriority(priority))
+      m_oldPriority(thread.ElevateThreadPriority(threadPriority))
     {
     }
 
@@ -113,6 +153,9 @@ class BasicThread:
   };
 
 protected:
+  /// The default scheduling policy used when creating threads (defaults to StandardRoundRobin)
+  static std::atomic<SchedulingPolicy> s_schedulingPolicy;
+
   // Internally held thread status block.  This has to be a shared pointer because we need to signal
   // the held state condition after releasing all shared pointers to ourselves, and this could mean
   // we're actually signalling this event after we free ourselves.
@@ -150,6 +193,32 @@ class BasicThread:
   /// </returns>
   ThreadPriority SetThreadPriority(ThreadPriority threadPriority);
 
+  /// <summary>
+  /// Sets the thread priority and scheduling policy of this thread
+  /// </summary>
+  /// <remarks>
+  /// This method may be called while the thread is running, or before it starts to run.  If it is
+  /// invoked before the thread starts to run, the thread will take on the specified priority and
+  /// scheduling policy when it is started.
+  /// </remarks>
+  /// <returns>
+  /// The previous thread priority
+  /// </returns>
+  ThreadPriority SetThreadPriority(ThreadPriority threadPriority, SchedulingPolicy schedPolicy);
+
+  /// <summary>
+  /// Sets the scheduling policy of this thread
+  /// </summary>
+  /// <remarks>
+  /// This method may be called while the thread is running, or before it starts to run.  If it is
+  /// invoked before the thread starts to run, the thread will take on the scheduling policy when
+  /// it is started.
+  /// </remarks>
+  /// <returns>
+  /// The previous thread priority
+  /// </returns>
+  SchedulingPolicy SetSchedulingPolicy(SchedulingPolicy schedPolicy);
+
   /// <summary>
   /// Sets the thread priority of this thread only if it elevates the priority
   /// </summary>
@@ -176,10 +245,10 @@ class BasicThread:
   /// </returns>
   ThreadPriority DeelevateThreadPriority(ThreadPriority threadPriority);
 
-  /// <summary<>
-  /// Low-level function to set the thread priority
+  /// <summary>
+  /// Low-level function to set the thread priority and scheduling policy
   /// </summary>
-  static void SetThreadPriority(const std::thread::native_handle_type& handle, ThreadPriority threadPriority);
+  static void SetThreadPriority(const std::thread::native_handle_type& handle, ThreadPriority threadPriority, SchedulingPolicy schedPolicy);
 
   /// <summary>
   /// Recovers a general lock used to synchronize entities in this thread internally.
@@ -248,6 +317,11 @@ class BasicThread:
   /// </returns>
   ThreadPriority GetThreadPriority(void);
 
+  /// <returns>
+  /// The current scheduling policy
+  /// </returns>
+  SchedulingPolicy GetSchedulingPolicy(void);
+
   /// <returns>
   /// True if this thread has transitioned to a completed state
   /// </returns>

src/autowiring/BasicThreadStateBlock.cpp

@@ -5,8 +5,9 @@
 
 using namespace autowiring;
 
-BasicThreadStateBlock::BasicThreadStateBlock(void) :
-  m_priority{ ThreadPriority::Default }
+BasicThreadStateBlock::BasicThreadStateBlock(ThreadPriority threadPriority, SchedulingPolicy schedPolicy) :
+  m_threadPriority{ threadPriority },
+  m_schedPolicy{ schedPolicy }
 {}
 
 BasicThreadStateBlock::~BasicThreadStateBlock(void)

src/autowiring/BasicThreadStateBlock.h

@@ -1,17 +1,13 @@
 // Copyright (C) 2012-2018 Leap Motion, Inc. All rights reserved.
 #pragma once
-#include MEMORY_HEADER
-#include MUTEX_HEADER
-#include THREAD_HEADER
-
-enum class ThreadPriority;
+#include "BasicThread.h"
 
 namespace autowiring {
 
 struct BasicThreadStateBlock:
   std::enable_shared_from_this<BasicThreadStateBlock>
 {
-  BasicThreadStateBlock(void);
+  BasicThreadStateBlock(ThreadPriority threadPriority, SchedulingPolicy schedPolicy);
   ~BasicThreadStateBlock(void);
 
   // Lock used to protect the actual thread
@@ -24,7 +20,8 @@ struct BasicThreadStateBlock:
   // The current thread, if running
   std::thread m_thisThread;
 
-  ThreadPriority m_priority;
+  ThreadPriority m_threadPriority;
+  SchedulingPolicy m_schedPolicy;
 
   // Completion condition, true when this thread is no longer running and has run at least once
   bool m_completed = false;

src/autowiring/CoreThread.cpp

@@ -8,6 +8,14 @@ CoreThread::CoreThread(const char* pName):
   BasicThread(pName)
 {}
 
+CoreThread::CoreThread(ThreadPriority threadPriority, const char* pName):
+  BasicThread(threadPriority, pName)
+{}
+
+CoreThread::CoreThread(ThreadPriority threadPriority, SchedulingPolicy schedPolicy, const char* pName):
+  BasicThread(threadPriority, schedPolicy, pName)
+{}
+
 CoreThread::~CoreThread(void){}
 
 void CoreThread::DoRunLoopCleanup(std::shared_ptr<CoreContext>&& ctxt, std::shared_ptr<CoreObject>&& refTracker) {

src/autowiring/CoreThread.h

@@ -34,6 +34,20 @@ class CoreThread:
   /// The name assigned to a thread is visible in some debuggers.
   /// <param name="pName">An optional name for this thread.</param>
   CoreThread(const char* pName = nullptr);
+  /// Constructs a core thread object with priority and a name.
+  ///
+  /// The name assigned to a thread is visible in some debuggers.
+  /// <param name="threadPriority">A priority for this thread.</param>
+  /// <param name="pName">An optional name for this thread.</param>
+  CoreThread(ThreadPriority threadPriority, const char* pName = nullptr);
+  /// Constructs a core thread object with priority, scheduling policy,
+  /// and a name.
+  ///
+  /// The name assigned to a thread is visible in some debuggers.
+  /// <param name="threadPriority">A priority for this thread.</param>
+  /// <param name="schedPolicy">A scheduling policy for this thread.</param>
+  /// <param name="pName">An optional name for this thread.</param>
+  CoreThread(ThreadPriority threadPriority, SchedulingPolicy schedPolicy, const char* pName = nullptr);
   virtual ~CoreThread(void);
 
 protected:

src/autowiring/CoreThreadLinux.cpp

@@ -28,11 +28,25 @@ void BasicThread::GetThreadTimes(std::chrono::milliseconds& kernelTime, std::chr
   userTime = std::chrono::duration_cast<milliseconds>(seconds(usage.ru_utime.tv_sec) + microseconds(usage.ru_utime.tv_usec));
 }
 
-void BasicThread::SetThreadPriority(const std::thread::native_handle_type& handle, ThreadPriority threadPriority) {
+void BasicThread::SetThreadPriority(const std::thread::native_handle_type& handle, ThreadPriority threadPriority, SchedulingPolicy schedPolicy) {
   struct sched_param param = { 0 };
   int policy = SCHED_OTHER;
   int percent = 0;
-  int min_priority;
+
+  switch (schedPolicy) {
+  case SchedulingPolicy::StandardRoundRobin:
+    policy = SCHED_OTHER;
+    break;
+  case SchedulingPolicy::RealtimeFIFO:
+    policy = SCHED_FIFO;
+    break;
+  case SchedulingPolicy::RealtimeRoundRobin:
+    policy = SCHED_RR;
+    break;
+  default:
+    throw std::invalid_argument("Attempted to assign an unrecognized scheduling policy");
+    break;
+  }
 
   switch (threadPriority) {
   case ThreadPriority::Idle:
@@ -59,14 +73,18 @@ void BasicThread::SetThreadPriority(const std::thread::native_handle_type& handl
     percent = 83;
     break;
   case ThreadPriority::TimeCritical:
+    percent = 99;
+    break;
   case ThreadPriority::Multimedia:
     percent = 100;
     break;
   default:
     throw std::invalid_argument("Attempted to assign an unrecognized thread priority");
   }
-  min_priority = sched_get_priority_min(policy);
-  pthread_getschedparam(handle, &policy, &param);
-  param.sched_priority = min_priority + (percent * (sched_get_priority_max(policy) - min_priority) + 50) / 100;
+  int min_priority = sched_get_priority_min(policy);
+  int max_priority = sched_get_priority_max(policy);
+  int prev_policy;
+  pthread_getschedparam(handle, &prev_policy, &param);
+  param.sched_priority = min_priority + (percent * (max_priority - min_priority) + 50) / 100;
   pthread_setschedparam(handle, policy, &param);
 }