[webkitcorepy] Ensure TaskPool queue full before reading

https://bugs.webkit.org/show_bug.cgi?id=245144
<rdar://problem/99873350>

Reviewed by Stephanie Lewis.

* Tools/Scripts/libraries/webkitcorepy/setup.py: Bump version.
* Tools/Scripts/libraries/webkitcorepy/webkitcorepy/__init__.py: Ditto.
* Tools/Scripts/libraries/webkitcorepy/webkitcorepy/task_pool.py:
(_Result.__call__): Decrement pending task counter.
(TaskPool): Set pending task counter to 0.
(TaskPool.do): Increment pending task counter, only consume from readers
if enough tasks are available for workers.

Canonical link: https://commits.webkit.org/254456@main

Tools/Scripts/libraries/webkitcorepy/setup.py

@@ -30,7 +30,7 @@ def readme():
 
 setup(
     name='webkitcorepy',
-    version='0.13.12',
+    version='0.13.13',
     description='Library containing various Python support classes and functions.',
     long_description=readme(),
     classifiers=[

Tools/Scripts/libraries/webkitcorepy/webkitcorepy/__init__.py

@@ -44,7 +44,7 @@
 from webkitcorepy.editor import Editor
 from webkitcorepy.file_lock import FileLock
 
-version = Version(0, 13, 12)
+version = Version(0, 13, 13)
 
 from webkitcorepy.autoinstall import Package, AutoInstall
 if sys.version_info > (3, 0):

Tools/Scripts/libraries/webkitcorepy/webkitcorepy/task_pool.py

@@ -64,6 +64,7 @@ def __init__(self, value, id):
 
     def __call__(self, caller):
         if caller:
+            caller.pending_count -= 1
             caller.callbacks.pop(self.id, lambda value: value)(self.value)
         return self.value
 
@@ -360,6 +361,7 @@ def __init__(
 
         self.callbacks = {}
         self._id_count = 0
+        self.pending_count = 0
         self.grace_period = grace_period
         self.block_size = block_size
         self.force_fork = force_fork
@@ -407,12 +409,13 @@ def do(self, function, *args, **kwargs):
 
         if callback:
             self.callbacks[self._id_count] = callback
+        self.pending_count += 1
         self.queue.send(self.Task(function, self._id_count, *args, **kwargs))
         self._id_count += 1
 
         # For every block of tasks passed to our workers, we need consume messages so we don't get deadlocked
         if not self._id_count % self.block_size:
-            while True:
+            while self.pending_count > 2 * self._num_workers:
                 try:
                     self.queue.receive(blocking=False)(self)
                 except Queue.Empty: