Parallelization

Even for large cases foxes calculations are fast, thanks to

  • Vectorization: The states (and also the points, in the case of data calculation at evaluation points) are split into so-called chunks, which are sub-arrays of the large original data.

  • Parallelization: These chunks are being sent to individual processes for calculation. Those calculations can be carried out simultaneously, i.e., in parallel.

Vectorization and parallelization are managed by so-called engines in foxes. If you do not explicitly specify the engine, a default will be chosen. This means that even if you do not know or care about foxes engines, your calculations will be vectorized and parallelized.

Available engines

These are the currently available engines, where each can be addressed by the short name or the full class name:

Short name

Class name

Base package

Description

threads

ThreadsEngine

concurrent.futures

Runs on a workstation/laptop, sends chunks to threads

process

ProcessEngine

concurrent.futures

Runs on a workstation/laptop, sends chunks to parallel processes

multiprocess

MultiprocessEngine

multiprocess

Runs on a workstation/laptop, sends chunks to parallel processes

ray

RayEngine

ray

Runs on a workstation/laptop, sends chunks to parallel processes

dask

DaskEngine

dask

Runs on a workstation/laptop, using processes or threads

local_cluster

LocalClusterEngine

distributed

Runs on a workstation/laptop, creates a virtual local cluster

slurm_cluster

SlurmClusterEngine

dask_jobqueue

Runs on a multi-node HPC cluster which is using SLURM

mpi

MPIEngine

mpi4py

Runs on laptop/workstation/cluster, also supports multi-node runs

numpy

NumpyEngine

numpy

Runs a loop over chunks, without parallelization

single

SingleChunkEngine

numpy

Runs all in a single chunk, without parallelization

default

DefaultEngine

numpy, concurrent.futures

Runs either the single or the process engine, depending on the case size

Note that the external packages are not installed by default. You can install them manually on demand, or use the option pip install foxes[eng] for the complete installation of all requirements of the complete list of engines.

Furthermore, scripts that use the mpi engine have to be started in a special way. For example, when running a script named run.py on 12 processors, the terminal command is

mpiexec -n 12 python -m mpi4py.futures run.py

Default engine

Let’s start by importing foxes and other required packages:

import matplotlib.pyplot as plt

import foxes
import foxes.variables as FV
/home/runner/work/foxes/foxes/foxes/core/engine.py:6: TqdmExperimentalWarning: Using `tqdm.autonotebook.tqdm` in notebook mode. Use `tqdm.tqdm` instead to force console mode (e.g. in jupyter console)
  from tqdm.autonotebook import tqdm

Next, we create a random wind farm and a random time series:

n_times = 1000
n_turbines = 50
seed = 42

sdata = foxes.input.states.create.random_timseries_data(
    n_times,
    seed=seed,
)
states = foxes.input.states.Timeseries(
    data_source=sdata,
    output_vars=[FV.WS, FV.WD, FV.TI, FV.RHO],
    fixed_vars={FV.RHO: 1.225, FV.TI: 0.02},
)

farm = foxes.WindFarm()
foxes.input.farm_layout.add_random(
    farm, n_turbines, min_dist=500, turbine_models=["DTU10MW"], seed=seed, verbosity=0
)
sdata
WS WD
Time
2000-01-01 00:00:00 11.236204 66.647854
2000-01-01 01:00:00 28.521429 195.084341
2000-01-01 02:00:00 21.959818 314.260501
2000-01-01 03:00:00 17.959755 263.600959
2000-01-01 04:00:00 4.680559 290.362013
... ... ...
2000-02-11 11:00:00 2.747462 236.503856
2000-02-11 12:00:00 27.519407 344.381264
2000-02-11 13:00:00 4.104559 24.824886
2000-02-11 14:00:00 28.507121 20.539700
2000-02-11 15:00:00 13.380173 101.587347

1000 rows × 2 columns

foxes.output.FarmLayoutOutput(farm).get_figure(figsize=(6, 6))
plt.show()
../_images/3b8d0367769c3f257366f5c53fc6307093d96aeac23088dfd4c1886eee019d17.png

You can run the wind farm calculations by simply creating an algorithm and calling farm_calc:

algo = foxes.algorithms.Downwind(
    farm,
    states,
    wake_models=["Bastankhah2014"],
    verbosity=1,
)
farm_results = algo.calc_farm()
farm_results.to_dataframe()
Initializing model 'Timeseries'

Initializing algorithm 'Downwind'

------------------------------------------------------------
  Algorithm: Downwind
  Running Downwind: calc_farm
------------------------------------------------------------
  n_states : 1000
  n_turbines: 50
------------------------------------------------------------
  states    : Timeseries()
  rotor     : CentreRotor()
  controller: BasicFarmController()
  deflection: NoDeflection()
  wake frame: RotorWD()
  wake lngth: None
------------------------------------------------------------
  wakes:
    0) Bastankhah2014: Bastankhah2014(ws_linear, induction=Madsen, k=0.04)
------------------------------------------------------------
  partial wakes:
    0) Bastankhah2014: gaussian_lookup, PartialGaussianLookup(lookup_data=None, bounds_policy=clip)
------------------------------------------------------------
  turbine models:
    0) DTU10MW: PCtFile(D=178.3, H=119.0, P_nominal=10000.0, P_unit=kW, rho=1.225, var_ws_ct=REWS, var_ws_P=REWS)
------------------------------------------------------------


--------------------------------------------------
  Model oder
--------------------------------------------------
00) basic_ctrl
01) InitFarmData
02) centre
03) basic_ctrl
  03.0) Post-rotor: DTU10MW
04) SetAmbFarmResults
05) FarmWakesCalculation
06) ReorderFarmOutput
--------------------------------------------------


Input data:

 <xarray.Dataset> Size: 25kB
Dimensions:          (state: 1000, Timeseries_vars: 2, DTU10MW_WS: 22,
                      DTU10MW_vars: 2, tmodels: 1)
Coordinates:
  * state            (state) datetime64[ns] 8kB 2000-01-01 ... 2000-02-11T15:...
  * Timeseries_vars  (Timeseries_vars) <U2 16B 'WD' 'WS'
  * DTU10MW_WS       (DTU10MW_WS) float64 176B 4.0 5.0 6.0 ... 23.0 24.0 25.0
  * DTU10MW_vars     (DTU10MW_vars) <U2 16B 'P' 'CT'
  * tmodels          (tmodels) <U7 28B 'DTU10MW'
Data variables:
    Timeseries_data  (state, Timeseries_vars) float64 16kB 66.65 11.24 ... 13.38
    DTU10MW_data     (DTU10MW_WS, DTU10MW_vars) float64 352B 280.2 ... 0.059 

Extra data:
  PartialGaussianLookup_weights: Dataset

Farm variables: AMB_CT, AMB_P, AMB_REWS, AMB_RHO, AMB_TI, AMB_WD, AMB_YAW, CT, D, H, P, REWS, RHO, TI, WD, X, Y, YAW, order, order_inv, order_ssel, weight

Output variables: AMB_CT, AMB_P, AMB_REWS, AMB_RHO, AMB_TI, AMB_WD, AMB_YAW, CT, D, H, P, REWS, RHO, TI, WD, X, Y, YAW, order, order_inv, order_ssel, weight
DefaultEngine: Selecting engine 'process'
ProcessEngine: Calculating 1000 states for 50 turbines
ProcessEngine: Starting calculation using 3 workers, for 3 states chunks.
ProcessEngine: Completed all 3 chunks
AMB_CT AMB_P AMB_REWS AMB_RHO AMB_TI AMB_WD AMB_YAW CT D H ... TI WD X Y YAW order order_inv order_ssel weight tname
state turbine
2000-01-01 00:00:00 0 0.758020 9920.520314 11.236204 1.225 0.02 66.647854 66.647854 0.814000 178.3 119.0 ... 0.02 66.647854 -913.371126 -2258.451177 66.647854 7 3 0 0.001 T0
1 0.758020 9920.520314 11.236204 1.225 0.02 66.647854 66.647854 0.814000 178.3 119.0 ... 0.02 66.647854 1856.218011 12.379281 66.647854 19 24 0 0.001 T1
2 0.758020 9920.520314 11.236204 1.225 0.02 66.647854 66.647854 0.794072 178.3 119.0 ... 0.02 66.647854 509.875003 -3495.898269 66.647854 45 39 0 0.001 T2
3 0.758020 9920.520314 11.236204 1.225 0.02 66.647854 66.647854 0.758020 178.3 119.0 ... 0.02 66.647854 2790.732598 -568.566969 66.647854 35 5 0 0.001 T3
4 0.758020 9920.520314 11.236204 1.225 0.02 66.647854 66.647854 0.814000 178.3 119.0 ... 0.02 66.647854 -2908.657106 -741.116557 66.647854 36 21 0 0.001 T4
... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ...
2000-02-11 15:00:00 45 0.382503 10645.002407 13.380173 1.225 0.02 101.587347 101.587347 0.498773 178.3 119.0 ... 0.02 101.587347 3535.885533 2425.322506 101.587347 16 0 333 0.001 T45
46 0.382503 10645.002407 13.380173 1.225 0.02 101.587347 101.587347 0.413694 178.3 119.0 ... 0.02 101.587347 -36.924282 3446.533432 101.587347 10 27 333 0.001 T46
47 0.382503 10645.002407 13.380173 1.225 0.02 101.587347 101.587347 0.382503 178.3 119.0 ... 0.02 101.587347 897.194248 -2375.686163 101.587347 15 45 333 0.001 T47
48 0.382503 10645.002407 13.380173 1.225 0.02 101.587347 101.587347 0.583145 178.3 119.0 ... 0.02 101.587347 -3378.240508 2245.962624 101.587347 48 48 333 0.001 T48
49 0.382503 10645.002407 13.380173 1.225 0.02 101.587347 101.587347 0.507540 178.3 119.0 ... 0.02 101.587347 -1022.301466 -3003.666094 101.587347 25 3 333 0.001 T49

50000 rows × 23 columns

In summary, if you simply run algo.calc_farm() (or algo.calc_points()or any other function that calls these functions), the DefaultEngine will be created and launched in the background.

As stated in the calc_farm printout, the DefaultEngine selected to run the ProcessEngine for this size of problem. The criteria are:

  • If n_states >= sqrt(n_procs) * (500/n_turbines)**1.5: Run engine ProcessEngine,

  • Else if algo.calc_points() has been called and n_states*n_points > 10000: Run engine ProcessEngine,

  • Else: Run engine SingleChunkEngine.

The above selection is based on test runs on a Ubuntu workstation with 64 physical cores and might not be the optimal choice for your system. Be aware of this whenever relying on the default engine for smallish cases - if in doubt, better explicitly specify the engine. This will be explained in the following section.

Engine selection through a with-block

Engines other than the DefaultEngine are selected by using a Python context manager, i.e., a with block, on an engine objecz. This ensures the proper launch and shutdown of the engine’s machinery for parallelization.

The syntax is straight forward. Note that the engine object is not required as a parameter for the algorithm, since it is set as a globally accessible object when entering the with block:

algo = foxes.algorithms.Downwind(
    farm,
    states,
    wake_models=["Bastankhah2014"],
    verbosity=0,
)

engine = foxes.Engine.new(
    "process",
    n_procs=4,
    chunk_size_states=200,
    chunk_size_points=5000,
)

with engine:
    farm_results = algo.calc_farm()

    o = foxes.output.FlowPlots2D(algo, farm_results)
    plot_data = o.get_states_data_xy(FV.WS, resolution=30, states_isel=[0])

g = o.gen_states_fig_xy(plot_data, figsize=(6, 6))
next(g)
plt.show()
ProcessEngine: Calculating 1000 states for 50 turbines
ProcessEngine: Starting calculation using 3 workers, for 5 states chunks.
ProcessEngine: Completed all 5 chunks

ProcessEngine: Calculating data at 91728 points for 1 states
ProcessEngine: Starting calculation using 3 workers, for 1 states chunks and 19 targets chunks.
ProcessEngine: Completed all 19 chunks
../_images/2af039db3d67de89c072e65e53f37d5f252f641d8f2189de7fddd7bc4067fae5.png

Remarks & recommendations

  • Take the time to think about your engine choice, and its parameters. Your choice might matter a lot for the performance of your run.

  • In general, all engines accept the parameters n_procs, chunk_size_states, chunk_size_points (the single engine ignores them, though).

  • If n_procs is not set, the maximal number of processes is applied, according to os.cpu_count() for Python version < 3.13 and os.process_cpu_count() for Python version >= 3.13. The number of workers used for chunk calculations is n_procs - 1.

  • If chunk_size_states is not set, the number of states is divided by n_procs. This might be non-optimal for small cases.

  • If chunk_size_points is not set and there is more than one states chunk, the full number of points is selected such that there is only one point chunk for each state chunk. If there is only one states chunk, the default points chunk size is the number of points divided by n_procs.

  • In general, for not too small cases, the default process engine is a good choice for runs on a linux based laptop or a workstation computer, or within Windows WSL.

  • For runs on native Windows, i.e., without WSL, the best engine choices have not been tested. Make sure you try different ones, e.g. process, multiprocess, dask, numpy, ray, and also vary the parameters.

  • The mpi engine requires the installation of MPI on the system, for example OpenMPI. Don’t forget to run this as for example for 12 cores by mpiexec -n 12 python -m mpi4py.futures run.py, with engine = foxes.Engine.new("mpi", n_procs=12, ...).

  • If you run into memory problems, the best options are to either reduce the number of processes or the chunk sizes. Alternatively, use shared memory.

  • Shared-memory switch: all pool-based engines (including process, ray, dask, and mpi) inherit supports_shared_data and min_shared_array_bytes. Set supports_shared_data=False to switch shared-data handling off. Set min_shared_array_bytes=0 to share all eligible arrays, or increase it to share only larger arrays.

  • The dask engine has additional options, accessible through the dask_pars dictionary parameter, for example the scheduler choice. See API and dask documentation for syntax and more info.

  • The local_cluster is not always faster than the process, multiprocess or dask engines, but offers a more detailed setup. For example, the memory and the number of threads per worker can be modified, if needed.

  • The numpy and single engines are intended for testing and small cases, and also for sequential runs without large point evaluations.