Implement cached thread pool

native/jni/Android.mk

@@ -25,6 +25,7 @@ LOCAL_SRC_FILES := \
     core/restorecon.cpp \
     core/module.cpp \
     core/logging.cpp \
+    core/thread.cpp \
     magiskhide/magiskhide.cpp \
     magiskhide/hide_utils.cpp \
     magiskhide/hide_policy.cpp \

native/jni/core/core.hpp

@@ -2,6 +2,7 @@
 
 #include <string>
 #include <vector>
+#include <functional>
 
 extern bool RECOVERY_MODE;
 extern int DAEMON_STATE;
@@ -12,6 +13,9 @@ void start_log_daemon();
 void setup_logfile(bool reset);
 void magisk_logging();
 
+// Thread pool
+void exec_task(std::function<void()> &&task);
+
 // Module stuffs
 void handle_modules();
 void magic_mount();

native/jni/core/daemon.cpp

@@ -144,8 +144,8 @@ static void handle_request(int client) {
         goto done;
     }
 
-    // Create new thread to handle complex requests
-    new_daemon_thread([=] { handle_request_async(client, code, cred); });
+    // Handle complex requests in another thread
+    exec_task([=] { handle_request_async(client, code, cred); });
     return;
 
 done:

native/jni/core/logging.cpp

@@ -185,6 +185,6 @@ void start_log_daemon() {
     int fds[2];
     if (pipe2(fds, O_CLOEXEC) == 0) {
         logd_fd = fds[1];
-        new_daemon_thread([fd = fds[0]] { logfile_writer(fd); });
+        exec_task([fd = fds[0]] { logfile_writer(fd); });
     }
 }

native/jni/core/thread.cpp

@@ -0,0 +1,82 @@
+// Cached thread pool implementation
+
+#include <utils.hpp>
+
+using namespace std;
+
+#define THREAD_IDLE_MAX_SEC 60
+#define MAX_THREAD_BLOCK_MS 5
+#define CORE_POOL_SIZE 3
+
+static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t send_task = PTHREAD_COND_INITIALIZER;
+static pthread_cond_t recv_task = PTHREAD_COND_INITIALIZER;
+
+// The following variables should be guarded by lock
+static int available_threads = 0;
+static int active_threads = 0;
+static function<void()> pending_task;
+
+static void operator+=(timeval &a, const timeval &b) {
+    a.tv_sec += b.tv_sec;
+    a.tv_usec += b.tv_usec;
+    if (a.tv_usec >= 1000000) {
+        a.tv_sec++;
+        a.tv_usec -= 1000000;
+    }
+}
+
+static timespec to_ts(const timeval &tv) { return { tv.tv_sec, tv.tv_usec * 1000 };  }
+
+static void *thread_pool_loop(void * const is_core_pool) {
+    // Block all signals
+    sigset_t mask;
+    sigfillset(&mask);
+
+    for (;;) {
+        // Restore sigmask
+        pthread_sigmask(SIG_SETMASK, &mask, nullptr);
+        function<void()> local_task;
+        {
+            mutex_guard g(lock);
+            ++available_threads;
+            if (!pending_task) {
+                if (is_core_pool) {
+                    pthread_cond_wait(&send_task, &lock);
+                } else {
+                    timeval tv;
+                    gettimeofday(&tv, nullptr);
+                    tv += { THREAD_IDLE_MAX_SEC, 0 };
+                    auto ts = to_ts(tv);
+                    if (pthread_cond_timedwait(&send_task, &lock, &ts) == ETIMEDOUT) {
+                        // Terminate thread after max idle time
+                        --available_threads;
+                        --active_threads;
+                        return nullptr;
+                    }
+                }
+            }
+            local_task.swap(pending_task);
+            pthread_cond_signal(&recv_task);
+            --available_threads;
+        }
+        local_task();
+    }
+}
+
+void exec_task(function<void()> &&task) {
+    mutex_guard g(lock);
+    pending_task.swap(task);
+    if (available_threads == 0) {
+        ++active_threads;
+        new_daemon_thread(thread_pool_loop, active_threads > CORE_POOL_SIZE ? nullptr : (void*)(1));
+    } else {
+        pthread_cond_signal(&send_task);
+    }
+    timeval tv;
+    gettimeofday(&tv, nullptr);
+    // Wait for task consumption
+    tv += { 0, MAX_THREAD_BLOCK_MS * 1000 };
+    auto ts = to_ts(tv);
+    pthread_cond_timedwait(&recv_task, &lock, &ts);
+}

native/jni/utils/misc.cpp

@@ -121,24 +121,6 @@ int new_daemon_thread(thread_entry entry, void *arg) {
     return xpthread_create(&thread, &attr, entry, arg);
 }
 
-int new_daemon_thread(void(*entry)()) {
-    thread_entry proxy = [](void *entry) -> void * {
-        reinterpret_cast<void(*)()>(entry)();
-        return nullptr;
-    };
-    return new_daemon_thread(proxy, (void *) entry);
-}
-
-int new_daemon_thread(std::function<void()> &&entry) {
-    thread_entry proxy = [](void *fp) -> void * {
-        auto fn = reinterpret_cast<std::function<void()>*>(fp);
-        (*fn)();
-        delete fn;
-        return nullptr;
-    };
-    return new_daemon_thread(proxy, new std::function<void()>(std::move(entry)));
-}
-
 static char *argv0;
 static size_t name_len;
 void init_argv0(int argc, char **argv) {

native/jni/utils/misc.hpp

@@ -82,8 +82,6 @@ static inline int parse_int(std::string_view s) { return parse_int(s.data()); }
 
 using thread_entry = void *(*)(void *);
 int new_daemon_thread(thread_entry entry, void *arg = nullptr);
-int new_daemon_thread(void(*entry)());
-int new_daemon_thread(std::function<void()> &&entry);
 
 static inline bool str_contains(std::string_view s, std::string_view ss) {
     return s.find(ss) != std::string::npos;