← control_flow package

ParallelForEach

Parallel version of the For Each node.

Apply a function or loop body to each item in a collection or iterator. Unlike the (sequential) For Each loop, the loop always behaves in a "stateless" fashion, meaning that loop nodes do not carry over state across iterations. See For Each node for the basic principle of this type of node. The Parallel For Each node allows use of the Continue node to skip individual items, but does not support the Break node, since processing is not sequential. If the node receives an iterator, the node may attempt to exhaust the iterator before any processing can start, therefore it is best used with iterators that are able to deliver all the data without blocking (e.g., waiting for data). The node can run computations for multiple iterations in parallel, using the same options of other parallel nodes, if you specify num_procs as either None (default all cores) or some number >1. You may also experiment with the number of threads that each process may use (num_threads_per_proc), since numpy can easily cause excessive thread churn (100% utilization). If you have multiple GPUs and your pipeline uses one or more GPU backends (e.g., jax or torch), then the node can spread the iterations out over the GPUs, depending on num_procs_per_gpu and compute_backends. Note that pipelines will only use GPUs if you explicitly enable it in specific nodes. More Info... Version 1.0.0

Ports/Properties

body

Graph to apply (loop body).

• verbose name: Body
• default value: None
• port type: GraphPort
• value type: Graph

body__signature

The list of loop variables received by the loop body. In a "For Each" loop, the loop body depends on only a single loop variable, namely the current item from the wired-in iterable. Consequently, your loop body should contain a single Placeholder node whose slotname must match the name listed here (e.g., item, but you can also choose a different name). Then, anything downstream of (i.e., depending on the value of) that Placeholder will constitute your loop body. The final output node of your loop body is then wired into the "body" input port of the loop node. In graphical UIs, the edge will be drawn in dotted style to indicate that the preceding graph itself is given to the loop node as the loop body (which will then execute it zero or more times), which is in contrast to a regular (solid) edge that indices normal data flow wherein first the preceding node runs, and then its output is passed to the subsequent node. The For Each node will then return a collection (see collect setting) of the values received from the final node on each iteration.

• verbose name: Body [Item Parameter]
• default value: (item)
• port type: Port
• value type: object (can be None)

iterable

Data to iterate over.

• verbose name: Iterable
• default value: None
• port type: DataPort
• value type: object (can be None)
• data direction: IN

result

ForEach result.

• verbose name: Result
• default value: None
• port type: DataPort
• value type: object (can be None)
• data direction: OUT

collect

Collect results into a list, dictionary, graph, packet, array or flatten the results into a single container of the same type emitted by the function. For graphs, a mix of nodes and edges may be emitted, and the resulting graph will have only those edges for which both endpoint nodes were also emitted separately (i.e., emitting an edge will not implicitly add its endpoint nodes). If 'last', only the last result is returned (as in Repeat). When applied to numeric arrays, the 'flatten' option concatenates the arrays along their existing first axis while the 'array' option stacks them along a new first axis. Packet and Graph are analogous to dict and list options, in that they are comprehensions that allow the user to itemize the constituents of these data structures (for packets, each is a 2-element list of [chunkname, Chunk] and for Graph each is a nodes and/or an edges). The dict-of-lists option is convenient when the loop body returns a list of dictionaries, and you wish to collect the results as a dictionary of lists; this can be a useful pattern for loop bodies with multiple return values, where the return values are wired into a Create Dictionary node and that dictionary is returned as the loop body result. The loop itself will then return a dictionary that has a list of items for each key.

• verbose name: Collect
• default value: list
• port type: EnumPort
• value type: str (can be None)

num_procs

Number of processes to use for parallel computation. If None, the global setting NUM_PROC, which defaults to the number of CPUs on the system, will be used.

• verbose name: Max Parallel Processes
• default value: None
• port type: IntPort
• value type: int (can be None)

num_threads_per_proc

Number of threads to use for each process. This can be used to limit the number of threads by each process to mitigate potential churn.

• verbose name: Threads Per Process
• default value: 4
• port type: IntPort
• value type: int (can be None)

compute_backends

GPU compute backends that may be used by the pipeline. If you include GPU compute backends here, workloads using those backends will be farmed out across multiple GPUs (if available) when running cross-validation folds in parallel. The 'auto' mode will attempt to auto-detect any backend settings in the given pipeline's nodes, but note that this will only catch nodes where this is explicit in the node's properties, and GPU workloads missed in this fashion will run by default on GPU 0.

• verbose name: Compute Backends
• default value: ['auto']
• port type: SubsetPort
• value type: list (can be None)

num_procs_per_gpu

Number of processes to use per GPU. This is only relevant if you have GPU compute backends enabled. If your GPU(s) are under-utilized during cross-validation, you can increase this to run this many loop iterations on each GPU.

• verbose name: Processes Per Gpu
• default value: 1
• port type: IntPort
• value type: int (can be None)

multiprocess_backend

Backend to use for farming out computation across multiple (CPU) processes. Multiprocessing is the simple Python default, which is not a bad start. Nestable is a version of multiprocessing that allows your pipeline to itself use parallel computation. If you are getting an error that "daemonic processes cannot have children", the most likely cause is that you have two nested parallel loops, and at least the outer loop needs to be set to nestable. Loky is a fast and fairly stable backend, but it does not support nested parallelism and has different limitations than multiprocessing. It can be helpful to try either if you are running into an issue trying to run something in parallel. Serial means to not run things in parallel but instead in series (even if num_procs is >1), which can help with debugging. Threading uses Python threads in the same process, but this is not recommended for most use cases due to what is known as GIL contention.

• verbose name: Multiprocess Backend
• default value: loky
• port type: EnumPort
• value type: str (can be None)

serial_if_debugger

If True, then if the Python debugger is detected, the node will run in serial mode, even if multiprocess_backend is set to something else. This is useful for debugging, since the debugger does not work well with parallel processes. This can be disabled if certain steps should nevertheless run in parallel (e.g., to reach a breakpoint more quickly).

• verbose name: Serial If Debugger
• default value: True
• port type: BoolPort
• value type: bool (can be None)

materialize

Materialize parallel instances of the loop body such that they can retain prior state. This is equivalent to creating one parallel copy of the loop body for each of the inputs in the iterable, which therefore persist across successive (re-)entries into the loop. Regardless of the values in the iterable, the k'th value will then be processed by the k'th materialized instance of the loop body. Note that this is currently not compatible with true parallel computation, and requires the multiprocess backend to be set to 'serial'.

• verbose name: Materialize
• default value: False
• port type: BoolPort
• value type: bool (can be None)

conserve_memory

The auto option defaults to 'on' if the backend is set to multiprocessing or nestable, and 'off' otherwise. If enabled, then the memory will be cleared periodically across iterations. This is useful if you are running out of memory during a run, but it will slow down the computation somewhat. The aggressive mode will force memory to be cleared after each iteration, which has the highest overhead, but guarantees a minimal memory footprint. However, the multiprocessing backend is known to occasionally hang in this mode so this may be reverted to on if you experience hangs. The off mode will leave memory reclamation to the individual nodes and scheduler behavior within the loop body, which may be further tuned using e.g. the default_conserve_memory config-file setting or per-node settings.

• verbose name: Conserve Memory
• default value: auto
• port type: EnumPort
• value type: str (can be None)

verbose

Whether to print verbose output.

• verbose name: Verbose
• default value: False
• port type: BoolPort
• value type: bool (can be None)

log_errors

If True, log exceptions occurring in the loop body.

• verbose name: Log Errors
• default value: True
• port type: BoolPort
• value type: bool (can be None)

set_breakpoint

Set a breakpoint on this node. If this is enabled, your debugger (if one is attached) will trigger a breakpoint.

• verbose name: Set Breakpoint (Debug Only)
• default value: False
• port type: BoolPort
• value type: bool (can be None)

metadata

User-definable meta-data associated with the node. Usually reserved for technical purposes.

• verbose name: Metadata
• default value: {}
• port type: DictPort
• value type: dict (can be None)