Skip to content

Script Development / Thread Pool DFF.THREAD

DFF.THREAD is used to execute IO-intensive functions concurrently within a single Task, such as batch HTTP requests. The thread pool is managed by DataFlux Func.

API

Method / Property Description
pool_size The explicitly configured thread pool size; None when not set yet
set_pool_size(pool_size) Sets the thread pool size; must be called before the first submit(...)
submit(fn, *args, **kwargs) Submits a function and returns a result Key
get_result(key, wait=True) Gets the specified result
get_all_results(wait=True) Gets all results
pop_result(wait=True) Pops a completed result; once popped, it will not be returned by other methods
is_all_finished Whether all executions are finished
wait_all_finished() Waits for all executions to finish

The result object type is DFFThreadResult, which contains the following properties:

Property Description
key The result Key returned by submit(...)
value The function return value; usually None when execution fails
error The exception thrown by the function; None when execution succeeds

set_pool_size(...) validates against the maximum value configured in the configuration file config.yaml; setting it again after the first submission will not take effect. When not explicitly set, the thread pool is created with the runtime default value.

Example

Batch HTTP Requests
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import requests

def fetch(url):
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return resp.text

@DFF.API('Batch Request')
def batch_fetch(urls):
    DFF.THREAD.set_pool_size(10)

    for url in urls:
        DFF.THREAD.submit(fetch, url)

    return [
        {'key': result.key, 'error': repr(result.error)}
        if result.error
        else {'key': result.key, 'value': result.value}
        for result in DFF.THREAD.get_all_results()
    ]

Result Retrieval Behavior

  • get_result(key, wait=False) and pop_result(wait=False) return None when there is no matching ready result.
  • get_all_results(...) can only be called after at least one function has been submitted.
  • get_all_results(...) returns results in the order of submission Keys, not in completion order.
  • When you need to process results as they complete, you can call pop_result(wait=not DFF.THREAD.is_all_finished).
  • pop_result(...) is suitable for independent tasks; get_all_results(...) is suitable for unified processing after all are complete.

Warning

Every result.error must be checked; otherwise, thread exceptions will be ignored. Even if results have not been collected yet, the Task will wait until all submitted thread work has stopped before ending. Therefore, every thread operation must have a clear timeout or other execution boundary.

The thread pool is suitable for IO-intensive work; CPU-intensive tasks usually do not see significant speedup.