dstandish commented on code in PR #26639:
URL: https://github.com/apache/airflow/pull/26639#discussion_r1036353099
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -62,6 +64,50 @@
KubernetesWatchType = Tuple[str, str, Optional[str], Dict[str, str], str]
+def multi_threads_queue_process(
+ queue_size: int,
+ queue_type: str,
+ process_method: Callable,
+ max_threads: int,
+ log: Logger,
+ batch_size: Optional[int] = None,
+) -> None:
+ """
+ Helper method to enable multi-threads for processing queues used with
kubernetes executor
+ :param queue_size: the size of the queue getting processed
+ :param queue_type: the type of the queue
+ :param process_method: the real method processing the queue
+ :param max_threads: the max num of threads to be used
+ :param log: log
+ :param batch_size: the max num of items we want to process in this round.
+ If it's not set, the current queue size will be used.
+ """
+ if queue_size == 0:
+ log.info(f'There is no item to process in the {queue_type} queue.')
+ return
+
+ start_time = time.time()
+ log.info(f'Start processing {queue_type} queue with at most {max_threads}
threads.')
+
+ batch_size = min(batch_size or queue_size, queue_size)
+ max_threads = min(max_threads, queue_size)
+
+ threads = []
+ quotient, remainder = divmod(batch_size, max_threads)
+ for i in range(max_threads):
+ sub_batch_size = quotient + 1 if i < remainder else quotient
+ t = Thread(target=process_method, args=[sub_batch_size])
+ threads.append(t)
+ t.start()
+ for t in threads:
+ t.join()
Review Comment:
ok i guess we can do something like this to see what failed and handle
appropriately:
```
import time
from concurrent.futures import ThreadPoolExecutor
def print_and_sleep(x):
print(x)
if x % 50 == 0:
raise ValueError("hi")
time.sleep(0.25)
max_threads = 4
with ThreadPoolExecutor(max_threads) as pool:
futures = [(x, pool.submit(print_and_sleep, x)) for x in range(123)]
failures = []
for arg, f in futures:
print(arg)
if f.exception():
failures.append((arg, f))
for arg, f in failures:
print(f"Func print_and_sleep failed for arg {arg} with exception
{f.exception()}")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
