In ensure_num_threads, if we have to call executor->schedule more than twice, create a task that does that instead.

This avoids blocking the calling thread too long on repeated calls to `executor->schedule`, which can be expensive.

Ideally, the `pthreadpool_executor` should provide a `schedule_n` function to schedule a group of tasks, but that would require a larger API change.

PiperOrigin-RevId: 811255624

Author: gonnet

src/pthreads.c

@@ -73,6 +73,12 @@
 #include <cpuinfo.h>
 #endif
 
+/* Alloca */
+#if defined(_MSC_VER)
+#include <malloc.h>
+#define alloca _alloca
+#endif  // defined(_MSC_VER)
+
 /* Public library header */
 #include <pthreadpool.h>
 
@@ -593,6 +599,23 @@ struct pthreadpool* pthreadpool_create_v2(struct pthreadpool_executor* executor,
   return threadpool;
 }
 
+static void wake_up_threads(void** contexts) {
+  struct pthreadpool* threadpool = (struct pthreadpool*)contexts[0];
+  struct pthreadpool_executor* executor = &threadpool->executor;
+  for (uint32_t k = 1; contexts[k] != NULL; k++) {
+    if (contexts[k + 1] != NULL) {
+      /* Fly, my pretties! Fly, fly, fly! */
+      executor->schedule(threadpool->executor_context, contexts[k],
+                         (void (*)(void*))thread_main);
+    } else {
+      void* context = contexts[k];
+      free(contexts);
+      thread_main(context);
+      return;
+    }
+  }
+}
+
 static void ensure_num_threads(struct pthreadpool* threadpool,
                                uint32_t num_threads) {
   assert(num_threads >= 1);
@@ -604,21 +627,37 @@ static void ensure_num_threads(struct pthreadpool* threadpool,
     return;
   }
 
+  void** thread_contexts = alloca(sizeof(void*) * num_threads);
+  int32_t num_threads_to_wake = 0;
+
   /* Create any missing threads for this threadpool. */
   for (uint32_t tid = 1;
        tid < num_threads &&
        pthreadpool_load_consume_int32_t(&threadpool->num_active_threads) > 0;
        tid++) {
     struct thread_info* thread = &threadpool->threads[tid];
 
-    // Check whether this thread was active, and if not, start it up.
+    // Check whether this thread was active, and if not, add it to the list of
+    // threads that need starting.
     if (!pthreadpool_exchange_sequentially_consistent_uint32_t(
             &thread->is_active, 1)) {
       pthreadpool_register_threads(threadpool, 1);
+      thread_contexts[num_threads_to_wake++] = thread;
+    }
+  }
 
+  if (num_threads_to_wake > 1) {
+    void** contexts = malloc(sizeof(void*) * (num_threads_to_wake + 2));
+    contexts[0] = threadpool;
+    memcpy(contexts + 1, thread_contexts, sizeof(void*) * num_threads_to_wake);
+    contexts[num_threads_to_wake + 1] = NULL;
+    /* Fly, my pretties! Fly, fly, fly! */
+    executor->schedule(threadpool->executor_context, contexts,
+                       (void (*)(void*))wake_up_threads);
+  } else if (num_threads_to_wake == 1) {
+    for (int k = 0; k < num_threads_to_wake; k++) {
       /* Fly, my pretties! Fly, fly, fly! */
-      pthreadpool_log_debug("starting thread %u (arg=%p).", tid, thread);
-      executor->schedule(threadpool->executor_context, thread,
+      executor->schedule(threadpool->executor_context, thread_contexts[k],
                          (void (*)(void*))thread_main);
     }
   }