[Jobs]: Follow-up on "Expose worker count to Job users"

Addressing gab comments on
https://chromium-review.googlesource.com/c/chromium/src/+/2304972

Change-Id: I4ac6d2378bdf509c3c606d0bafab965212f07638
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2390860
Commit-Queue: Etienne Pierre-Doray <etiennep@chromium.org>
Reviewed-by: Gabriel Charette <gab@chromium.org>
Cr-Commit-Position: refs/heads/master@{#806199}

base/task/post_job.cc

@@ -126,7 +126,7 @@ void JobHandle::Detach() {
 JobHandle PostJob(const Location& from_here,
                   const TaskTraits& traits,
                   RepeatingCallback<void(JobDelegate*)> worker_task,
-                  RepeatingCallback<size_t(size_t)> max_concurrency_callback) {
+                  MaxConcurrencyCallback max_concurrency_callback) {
   DCHECK(ThreadPoolInstance::Get())
       << "Ref. Prerequisite section of post_task.h.\n\n"
          "Hint: if this is in a unit test, you're likely merely missing a "

base/task/post_job.h

@@ -125,6 +125,15 @@ class BASE_EXPORT JobHandle {
   DISALLOW_COPY_AND_ASSIGN(JobHandle);
 };
 
+// Callback used in PostJob() to control the maximum number of threads calling
+// the worker task concurrently.
+
+// Returns the maximum number of threads which may call a job's worker task
+// concurrently. |worker_count| is the number of threads currently assigned to
+// this job which some callers may need to determine their return value.
+using MaxConcurrencyCallback =
+    RepeatingCallback<size_t(size_t /*worker_count*/)>;
+
 // Posts a repeating |worker_task| with specific |traits| to run in parallel on
 // base::ThreadPool.
 // Returns a JobHandle associated with the Job, which can be joined, canceled or
@@ -154,9 +163,8 @@ class BASE_EXPORT JobHandle {
 //   }
 //
 // |max_concurrency_callback| controls the maximum number of threads calling
-// |worker_task| concurrently, given the number of threads currently assigned to
-// this job. |worker_task| is only invoked if the number of threads previously
-// running |worker_task| was less than the value returned by
+// |worker_task| concurrently. |worker_task| is only invoked if the number of
+// threads previously running |worker_task| was less than the value returned by
 // |max_concurrency_callback|. In general, |max_concurrency_callback| should
 // return the latest number of incomplete work items (smallest unit of work)
 // left to processed. JobHandle/JobDelegate::NotifyConcurrencyIncrease() *must*
@@ -171,11 +179,10 @@ class BASE_EXPORT JobHandle {
 // |traits| requirements:
 // - base::ThreadPolicy must be specified if the priority of the task runner
 //   will ever be increased from BEST_EFFORT.
-JobHandle BASE_EXPORT
-PostJob(const Location& from_here,
-        const TaskTraits& traits,
-        RepeatingCallback<void(JobDelegate*)> worker_task,
-        RepeatingCallback<size_t(size_t)> max_concurrency_callback);
+JobHandle BASE_EXPORT PostJob(const Location& from_here,
+                              const TaskTraits& traits,
+                              RepeatingCallback<void(JobDelegate*)> worker_task,
+                              MaxConcurrencyCallback max_concurrency_callback);
 
 }  // namespace base
 

base/task/thread_pool/job_task_source.cc

@@ -78,12 +78,11 @@ bool JobTaskSource::JoinFlag::ShouldWorkerSignal() {
   return value_.exchange(kNotWaiting, std::memory_order_relaxed) != kNotWaiting;
 }
 
-JobTaskSource::JobTaskSource(
-    const Location& from_here,
-    const TaskTraits& traits,
-    RepeatingCallback<void(JobDelegate*)> worker_task,
-    RepeatingCallback<size_t(size_t)> max_concurrency_callback,
-    PooledTaskRunnerDelegate* delegate)
+JobTaskSource::JobTaskSource(const Location& from_here,
+                             const TaskTraits& traits,
+                             RepeatingCallback<void(JobDelegate*)> worker_task,
+                             MaxConcurrencyCallback max_concurrency_callback,
+                             PooledTaskRunnerDelegate* delegate)
     : TaskSource(traits, nullptr, TaskSourceExecutionMode::kJob),
       from_here_(from_here),
       max_concurrency_callback_(std::move(max_concurrency_callback)),

base/task/thread_pool/job_task_source.h

@@ -35,7 +35,7 @@ class BASE_EXPORT JobTaskSource : public TaskSource {
   JobTaskSource(const Location& from_here,
                 const TaskTraits& traits,
                 RepeatingCallback<void(JobDelegate*)> worker_task,
-                RepeatingCallback<size_t(size_t)> max_concurrency_callback,
+                MaxConcurrencyCallback max_concurrency_callback,
                 PooledTaskRunnerDelegate* delegate);
 
   static JobHandle CreateJobHandle(

docs/threading_and_tasks.md

@@ -655,7 +655,13 @@ void WorkerTask(base::JobDelegate* job_delegate) {
 }
 
 // Returns the latest thread-safe number of incomplete work items.
-void NumIncompleteWorkItems(size_t worker_count);
+void NumIncompleteWorkItems(size_t worker_count) {
+  // NumIncompleteWorkItems() may use |worker_count| if it needs to account for
+  // local work lists, which is easier than doing its own accounting, keeping in
+  // mind that the actual number of items may be racily overestimated and thus
+  // WorkerTask() may be called when there's no available work.
+  return GlobalQueueSize() + worker_count;
+}
 
 base::PostJob(FROM_HERE, {},
               base::BindRepeating(&WorkerTask),