Implement process and thread priority adjustments (#691)

Author: cgutman

src/audio.cpp

@@ -89,6 +89,9 @@ void encodeThread(sample_queue_t samples, config_t config, void *channel_data) {
   auto packets = mail::man->queue<packet_t>(mail::audio_packets);
   auto stream  = &stream_configs[map_stream(config.channels, config.flags[config_t::HIGH_QUALITY])];
 
+  // Encoding takes place on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::high);
+
   opus_t opus { opus_multistream_encoder_create(
     stream->sampleRate,
     stream->channelCount,
@@ -173,6 +176,9 @@ void capture(safe::mail_t mail, config_t config, void *channel_data) {
     }
   }
 
+  // Capture takes place on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::critical);
+
   auto samples = std::make_shared<sample_queue_t::element_type>(30);
   std::thread thread { encodeThread, samples, config, channel_data };
 

src/platform/common.h

@@ -306,6 +306,18 @@ std::vector<std::string> display_names(mem_type_e hwdevice_type);
 
 boost::process::child run_unprivileged(const std::string &cmd, boost::filesystem::path &working_dir, boost::process::environment &env, FILE *file, std::error_code &ec);
 
+enum class thread_priority_e : int {
+  low,
+  normal,
+  high,
+  critical
+};
+void adjust_thread_priority(thread_priority_e priority);
+
+// Allow OS-specific actions to be taken to prepare for streaming
+void streaming_will_start();
+void streaming_will_stop();
+
 input_t input();
 void move_mouse(input_t &input, int deltaX, int deltaY);
 void abs_mouse(input_t &input, const touch_port_t &touch_port, float x, float y);

src/platform/linux/misc.cpp

@@ -153,6 +153,18 @@ bp::child run_unprivileged(const std::string &cmd, boost::filesystem::path &work
   }
 }
 
+void adjust_thread_priority(thread_priority_e priority) {
+  // Unimplemented
+}
+
+void streaming_will_start() {
+  // Nothing to do
+}
+
+void streaming_will_stop() {
+  // Nothing to do
+}
+
 namespace source {
 enum source_e : std::size_t {
 #ifdef SUNSHINE_BUILD_CUDA

src/platform/macos/misc.cpp

@@ -131,6 +131,18 @@ bp::child run_unprivileged(const std::string &cmd, boost::filesystem::path &work
   }
 }
 
+void adjust_thread_priority(thread_priority_e priority) {
+  // Unimplemented
+}
+
+void streaming_will_start() {
+  // Nothing to do
+}
+
+void streaming_will_stop() {
+  // Nothing to do
+}
+
 } // namespace platf
 
 namespace dyn {

src/platform/windows/misc.cpp

@@ -13,9 +13,12 @@
 #include <winuser.h>
 #include <ws2tcpip.h>
 #include <userenv.h>
+#include <dwmapi.h>
+#include <timeapi.h>
 // clang-format on
 
 #include "src/main.h"
+#include "src/platform/common.h"
 #include "src/utility.h"
 
 namespace bp = boost::process;
@@ -470,4 +473,53 @@ bp::child run_unprivileged(const std::string &cmd, boost::filesystem::path &work
   }
 }
 
+void adjust_thread_priority(thread_priority_e priority) {
+  int win32_priority;
+
+  switch(priority) {
+  case thread_priority_e::low:
+    win32_priority = THREAD_PRIORITY_BELOW_NORMAL;
+    break;
+  case thread_priority_e::normal:
+    win32_priority = THREAD_PRIORITY_NORMAL;
+    break;
+  case thread_priority_e::high:
+    win32_priority = THREAD_PRIORITY_ABOVE_NORMAL;
+    break;
+  case thread_priority_e::critical:
+    win32_priority = THREAD_PRIORITY_HIGHEST;
+    break;
+  default:
+    BOOST_LOG(error) << "Unknown thread priority: "sv << (int)priority;
+    return;
+  }
+
+  if(!SetThreadPriority(GetCurrentThread(), win32_priority)) {
+    auto winerr = GetLastError();
+    BOOST_LOG(warning) << "Unable to set thread priority to "sv << win32_priority << ": "sv << winerr;
+  }
+}
+
+void streaming_will_start() {
+  // Enable MMCSS scheduling for DWM
+  DwmEnableMMCSS(true);
+
+  // Reduce timer period to 1ms
+  timeBeginPeriod(1);
+
+  // Promote ourselves to high priority class
+  SetPriorityClass(GetCurrentProcess(), HIGH_PRIORITY_CLASS);
+}
+
+void streaming_will_stop() {
+  // Demote ourselves back to normal priority class
+  SetPriorityClass(GetCurrentProcess(), NORMAL_PRIORITY_CLASS);
+
+  // End our 1ms timer request
+  timeEndPeriod(1);
+
+  // Disable MMCSS scheduling for DWM
+  DwmEnableMMCSS(false);
+}
+
 } // namespace platf

src/stream.cpp

@@ -740,6 +740,8 @@ void controlBroadcastThread(control_server_t *server) {
     input::passthrough(session->input, std::move(plaintext));
   });
 
+  // This thread handles latency-sensitive control messages
+  platf::adjust_thread_priority(platf::thread_priority_e::critical);
 
   auto shutdown_event = mail::man->event<bool>(mail::broadcast_shutdown);
   while(!shutdown_event->peek()) {
@@ -905,6 +907,9 @@ void videoBroadcastThread(udp::socket &sock) {
   auto packets        = mail::man->queue<video::packet_t>(mail::video_packets);
   auto timebase       = boost::posix_time::microsec_clock::universal_time();
 
+  // Video traffic is sent on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::high);
+
   while(auto packet = packets->pop()) {
     if(shutdown_event->peek()) {
       break;
@@ -1079,6 +1084,9 @@ void audioBroadcastThread(udp::socket &sock) {
   audio_packet->rtp.packetType = 97;
   audio_packet->rtp.ssrc       = 0;
 
+  // Audio traffic is sent on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::high);
+
   while(auto packet = packets->pop()) {
     if(shutdown_event->peek()) {
       break;
@@ -1333,6 +1341,8 @@ void audioThread(session_t *session) {
 }
 
 namespace session {
+std::atomic_uint running_sessions;
+
 state_e state(session_t &session) {
   return session.state.load(std::memory_order_relaxed);
 }
@@ -1394,6 +1404,11 @@ void join(session_t &session) {
     }
   }
 
+  // If this is the last session, invoke the platform callbacks
+  if(--running_sessions == 0) {
+    platf::streaming_will_stop();
+  }
+
   BOOST_LOG(debug) << "Session ended"sv;
 }
 
@@ -1432,6 +1447,11 @@ int start(session_t &session, const std::string &addr_string) {
 
   session.state.store(state_e::RUNNING, std::memory_order_relaxed);
 
+  // If this is the first session, invoke the platform callbacks
+  if(++running_sessions == 1) {
+    platf::streaming_will_start();
+  }
+
   return 0;
 }
 

src/video.cpp

@@ -673,6 +673,9 @@ void captureThread(
     }
   }
 
+  // Capture takes place on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::critical);
+
   while(capture_ctx_queue->running()) {
     bool artificial_reinit = false;
 
@@ -703,7 +706,10 @@ void captureThread(
       }
 
       auto &next_img = *round_robin++;
-      while(next_img.use_count() > 1) {}
+      while(next_img.use_count() > 1) {
+        // Sleep a bit to avoid starving the encoder threads
+        std::this_thread::sleep_for(2ms);
+      }
 
       return next_img;
     },
@@ -1335,6 +1341,9 @@ void captureThreadSync() {
     }
   });
 
+  // Encoding and capture takes place on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::high);
+
   while(encode_run_sync(synced_session_ctxs, ctx) == encode_e::reinit) {}
 }
 
@@ -1367,6 +1376,9 @@ void capture_async(
 
   auto touch_port_event = mail->event<input::touch_port_t>(mail::touch_port);
 
+  // Encoding takes place on this thread
+  platf::adjust_thread_priority(platf::thread_priority_e::high);
+
   while(!shutdown_event->peek() && images->running()) {
     // Wait for the main capture event when the display is being reinitialized
     if(ref->reinit_event.peek()) {

tools/sunshinesvc.cpp

@@ -271,7 +271,7 @@ VOID WINAPI ServiceMain(DWORD dwArgc, LPTSTR *lpszArgv) {
          NULL,
          NULL,
          TRUE,
-         ABOVE_NORMAL_PRIORITY_CLASS | CREATE_UNICODE_ENVIRONMENT | CREATE_NO_WINDOW | EXTENDED_STARTUPINFO_PRESENT,
+         CREATE_UNICODE_ENVIRONMENT | CREATE_NO_WINDOW | EXTENDED_STARTUPINFO_PRESENT,
          NULL,
          NULL,
          (LPSTARTUPINFOW)&startup_info,