Shared Data and Objects in C++ and Python

Dragon not only supports running Python code in a distributed multi-node environment, processes executing C++ compiled binaries are also supported. The code in this tutorial is from on an example program that is available to run in its entirety in our examples directory. The code is provided at https://github.com/DragonHPC/dragon/tree/main/examples/dragon_native/serializable_heat and in Heat Simulation Python Orchestration Code and Heat Simulation C++ Worker Code for your reference. The goal of this program is to show how data and Dragon objects can be shared between Python and C++.

Data and Dragon objects may be serialized, the serialized description can then passed to a new process (via some means) and then attached in the new process. This is a common means of sharing within Dragon. Dragon objects are designed to be shared between processes. In this example the main data sharing is done with an n-dimensional array representation that can be passed between processes and synced to a common location within a Dragon distributed dictionary (i.e. DDict).

The organization of most C++/Python hybrid programs run in the Dragon distributed multi-processing framework is to have a Python orchestration program and C++ worker programs. Typically there is some setup first in the orchestration program which creates objects to be shared and then starts the C++ worker programs. In the Python orchestrator, setup proceeds as follows.

Orchestration Setup

Orchestration Code
 1    # One dictionary holds the array data and everything the workers need to find.
 2    store = DDict(1, 1, 64 * 1024 * 1024)
 3    ser_store = store.serialize()
 4
 5    # The X picklers are what make the values readable as Serializables in C++.
 6    config = store.pickler(key_pickler=XPickler(), value_pickler=XPickler())
 7
 8    # An XNDArray keeps its data in the dictionary, so passing it to another
 9    # process only passes its meta data.
10    grid = XNDArray(initial_grid(ROWS, COLS), ser_store)
11
12    # The extra party is this process.
13    barrier = Barrier(parties=nworkers + 1)
14    ready = Semaphore(value=0)
15    results = Queue(maxsize=nworkers, pickler=XPickler())
16
17    config["cols"] = COLS
18    config["steps"] = steps
19    config["grid"] = grid
20    config["barrier"] = barrier
21    config["ready"] = ready
22    config["results"] = results

To demonstrate sharing a DDict between Python and C++ a DDict is created in the first few steps in the Orchestration Code snippet that will be used to share config data and will also be used as the central storage location for the 2D array used in this simulation.

The DDict is serialized in anticipation of passing the serialized descriptor to the C++ workers. The call to pickler on the store tells Python to create an alias for the store DDict called config that uses the XPickler for serializing keys and values stored within it. The XPickler is a Python pickler that supports pickling and unpickling between Python and C++. Anything stored or retrieved using the config alias is going to be readable/writable in both C++ and Python.

To demonstrate this cross language compatibility the 2D grid is stored in the DDict along with a barrier, semaphore, and a queue. All of these objects are supported in both Python and C++. Once stored, and with the C++ workers being given the serialized DDict when they are started, all processes will be able to access all the config data.

One thing to highlight is that the grid NDArray, when it is created, is given the ser_store serialized representation of the DDict object. This is required of NDArrays because they are stored within a DDict and when sent to another process, only metadata about the ndarray is communicated. The actual data, which may be large, is not copied between processes until absolutely necessary. In this way, the NDArray provides lazy, last minute loading of data.

Not shown here a few more details are stored in the DDict providing each worker with some additional process-specific code. Then each process is started using the Dragon native Popen code. This is not the standard Posix Popen. The Dragon native Popen will distributed the processes on all your nodes in a round-robin fashion by default. This could be further refined if the programmer were to provide policies to control process placement. Please refer to the Resource Placement and Affinity documentation for more information. In addition, for more control over groups of processes, there is a ProcessGroup within the Dragon Native API that provides more control and faster launch of large numbers of processes.

Worker Setup

Worker Code
 1    DDict<Serializable, Serializable> config(ser_config, &TIMEOUT);
 2
 3    // Every one of these values was written by Python.
 4    int cols = config[key("cols")];
 5    int steps = config[key("steps")];
 6    int start_row = config[key("start_", worker_id)];
 7    int end_row = config[key("end_", worker_id)];
 8
 9    // The whole grid. Its data lives in the DDict, so only the meta data was
10    // passed to us. Nothing is read until refresh is called.
11    SerializableDoubleNDArray grid = config[key("grid")];
12
13    // Just the rows this worker owns. This is where our results are published.
14    SerializableDoubleNDArray band = config[key("band_", worker_id)];
15
16    Barrier barrier = config[key("barrier")];
17    Semaphore ready = config[key("ready")];
18    Queue<Serializable> results = config[key("results")];

The workers connect to the shared config by creating a DDict using the ser_config passed to the process when it starts as shown in the Worker Code snippet. Once attached to the DDict, the workers can retrieve their config data from the DDict in much the same way it was stored in it in the first place. This just works because of a base class of Serializable objects and a collection of Serializable subclasses. What’s nice is that much of the conversion to serializable objects is done automatically for the programmer. There are serializables for strings, integers, doubles, Dragon objects, and the n-dimensional array that is attachd when grid is initilialized. The barrier, ready semaphore, and the results queue are all attachable right from there DDict values: deceptively simple that makes it very simple to share objects and data between C++ and Python. See the C++ Reference for more information on all these classes of objects.

One thing to note is that the C++ workers and the Python orchestrator all fully participate in the Barrier in the program. One barrier waiter is the Python orchestrator, the others are the C++ workers. Dragon’s barrier implementation, which is also a part of multiprocessing, works in both C++ and Python.

The n-dimensional array support deserves a bit more description. In Python, an ndarray is a numpy n-dimensional array and so all ndarray operations work on it. The numpy object is wrapped in a Dragon native NDArray object (which inherits from ndarray - so it is still an ndarray when wrapped). When provided to C++, it becomes a Dragon C++ SerializableNDArray object as shown above where band is initialized. When initialized like this a copy of the n-dimensional array data is lazily cached as soon as it is indexed.

The NDArray, when it is serialized and passed to C++, or (not present in this example) passed between C++ processes through a Queue or a DDict, contains all the meta-information about the n-dimensional array. NDArray and SerializableNDArray are designed to support n-dimensional arrays of any given dimension and remember their dimensionality as they are passed between processes.

Slicing
1            SerializableDoubleNDArray above = grid[i-1];
2            SerializableDoubleNDArray row = grid[i];
3            SerializableDoubleNDArray below = grid[i+1];
4            SerializableDoubleNDArray out = band[i - start_row];

The Slicing snippet provides code that is used to slice the grid and the band given as work to each of the workers. A slice on an NDArray in C++ shares the same underlying cached data, so later, when the band has sync called on it, it is then copied back to the backing DDict that holds the shared copy of the data. Back in the orchestrator process, refresh is called at the appropriate time to refresh the band and then the various bands are copied back into the shared grid data by the orchestrator program as shown in the heat4 snippet.

Full Program

The full program comes in two parts, the Python orchestration code and the C++ worker code.

Python Orchestration Code

Heat Simulation Python Orchestration Code
  1"""
  2A 2D heat diffusion simulation split between Python and C++.
  3
  4Python owns the simulation: it builds the grid, decomposes it into row bands,
  5starts a C++ worker per band and stitches the results back together after every
  6step. The C++ workers do the stencil arithmetic.
  7
  8Nothing is passed to the workers on the command line except a serialized
  9Distributed Dictionary. The grid, the per worker bands, the Barrier used to
 10synchronize each step, the Semaphore used for the startup handshake and the
 11Queue used to report results are all stored in that dictionary and travel to
 12C++ as Serializables.
 13
 14Run it with, for example:
 15
 16    make
 17    dragon heat_simulation.py 4
 18
 19The number of workers and the number of steps may both be given on the command
 20line, so `dragon heat_simulation.py 4 200` runs 200 steps across 4 workers.
 21"""
 22
 23import multiprocessing as mp
 24import os
 25import pathlib
 26import sys
 27import time
 28
 29import numpy as np
 30
 31import dragon
 32from dragon.data.ddict.ddict import DDict
 33from dragon.infrastructure.facts import DRAGON_LIB_DIR
 34from dragon.native.barrier import Barrier
 35from dragon.native.process import Popen
 36from dragon.native.queue import Queue
 37from dragon.native.semaphore import Semaphore
 38from dragon.utils import XNDArray, XPickler
 39
 40HERE = pathlib.Path(__file__).resolve().parent
 41WORKER = HERE / "heat_worker"
 42
 43ROWS = 34
 44COLS = 34
 45STEPS = 500
 46HOT = 100.0
 47
 48ENV = dict(os.environ)
 49ENV["LD_LIBRARY_PATH"] = str(DRAGON_LIB_DIR) + ":" + str(ENV.get("LD_LIBRARY_PATH", ""))
 50ENV["DYLD_FALLBACK_LIBRARY_PATH"] = str(DRAGON_LIB_DIR) + ":" + str(ENV.get("DYLD_FALLBACK_LIBRARY_PATH", ""))
 51
 52
 53def row_bands(rows: int, nworkers: int) -> list:
 54    """Split the interior rows as evenly as possible across the workers."""
 55
 56    interior = rows - 2
 57    base, extra = divmod(interior, nworkers)
 58
 59    bands = []
 60    start = 1
 61
 62    for worker in range(nworkers):
 63        count = base + (1 if worker < extra else 0)
 64        bands.append((start, start + count))
 65        start += count
 66
 67    return bands
 68
 69
 70def initial_grid(rows: int, cols: int) -> np.ndarray:
 71    """A cold plate with a hot top edge."""
 72
 73    grid = np.zeros((rows, cols))
 74    grid[0, :] = HOT
 75
 76    return grid
 77
 78
 79def render(grid: np.ndarray, size: int = 16) -> str:
 80    """Draw a coarse picture of the plate."""
 81
 82    ramp = " .:-=+*#%@"
 83    rows = np.linspace(0, grid.shape[0] - 1, size).astype(int)
 84    cols = np.linspace(0, grid.shape[1] - 1, size).astype(int)
 85
 86    lines = []
 87    for i in rows:
 88        line = ""
 89        for j in cols:
 90            shade = int(grid[i, j] / HOT * (len(ramp) - 1))
 91            line += ramp[min(shade, len(ramp) - 1)] * 2
 92        lines.append("    " + line)
 93
 94    return "\n".join(lines)
 95
 96
 97def main():
 98    nworkers = int(sys.argv[1]) if len(sys.argv) > 1 else 4
 99    steps = int(sys.argv[2]) if len(sys.argv) > 2 else STEPS
100
101    if not WORKER.exists():
102        raise SystemExit(f"{WORKER} does not exist. Build it by running make in {HERE}.")
103
104    if ROWS - 2 < nworkers:
105        raise SystemExit(f"{nworkers} workers is too many for a grid with {ROWS - 2} interior rows.")
106
107    print(f"Diffusing heat across a {ROWS}x{COLS} plate for {steps} steps using {nworkers} C++ workers.\n")
108
109    # [begin-heat-orc-setup]
110    # One dictionary holds the array data and everything the workers need to find.
111    store = DDict(1, 1, 64 * 1024 * 1024)
112    ser_store = store.serialize()
113
114    # The X picklers are what make the values readable as Serializables in C++.
115    config = store.pickler(key_pickler=XPickler(), value_pickler=XPickler())
116
117    # An XNDArray keeps its data in the dictionary, so passing it to another
118    # process only passes its meta data.
119    grid = XNDArray(initial_grid(ROWS, COLS), ser_store)
120
121    # The extra party is this process.
122    barrier = Barrier(parties=nworkers + 1)
123    ready = Semaphore(value=0)
124    results = Queue(maxsize=nworkers, pickler=XPickler())
125
126    config["cols"] = COLS
127    config["steps"] = steps
128    config["grid"] = grid
129    config["barrier"] = barrier
130    config["ready"] = ready
131    config["results"] = results
132    # [end-heat-orc-setup]
133
134    bands = []
135
136    for worker, (start, end) in enumerate(row_bands(ROWS, nworkers)):
137        band = XNDArray(np.zeros((end - start, COLS)), ser_store)
138
139        config[f"band_{worker}"] = band
140        config[f"start_{worker}"] = start
141        config[f"end_{worker}"] = end
142
143        bands.append((start, end, band))
144
145    procs = [
146        Popen(executable=str(WORKER), args=[ser_store, str(worker)], env=ENV)
147        for worker in range(nworkers)
148    ]
149
150    # Wait for every worker to attach before timing the run.
151    for _ in range(nworkers):
152        ready.acquire()
153
154    began = time.monotonic()
155
156    # [begin-grd-copyback]
157    for _ in range(steps):
158        # Every worker has published its band.
159        barrier.wait()
160
161        for start, end, band in bands:
162            band.refresh()
163            grid[start:end, :] = band
164
165        grid.sync()
166
167        # The new grid is published, so the workers may start the next step.
168        barrier.wait()
169    # [end-grid-copyback]
170
171    elapsed = time.monotonic() - began
172
173    deltas = [results.get() for _ in range(nworkers)]
174
175    for proc in procs:
176        proc.wait()
177
178    failed = [worker for worker, proc in enumerate(procs) if proc.returncode != 0]
179    if failed:
180        raise SystemExit(f"Workers {failed} exited with an error.")
181
182    grid.refresh()
183
184    print(render(grid))
185    print(f"\n    center temperature  {grid[ROWS // 2, COLS // 2]:.4f}")
186    print(f"    largest last change {max(deltas):.3e}")
187    print(f"    {steps} steps in {elapsed:.2f} seconds\n")
188
189    for _, _, band in bands:
190        band.destroy()
191
192    grid.destroy()
193    results.destroy()
194    store.destroy()
195
196
197if __name__ == "__main__":
198    mp.set_start_method("dragon", force=True)
199    main()

C++ Worker Code

Heat Simulation C++ Worker Code
  1/**
  2 * A C++ worker for the heat diffusion example. See heat_simulation.py for the
  3 * Python side of this program and README.md for a description of the example.
  4 *
  5 * The worker is given nothing but a serialized Distributed Dictionary and its
  6 * worker id. Everything else it needs, including the shared grid, the Barrier
  7 * it synchronizes on and the Queue it reports results to, is stored in that
  8 * dictionary as a Serializable. This is the point of the example: any Dragon
  9 * object that can be serialized can be handed from Python to C++ this way.
 10 */
 11
 12#include <iostream>
 13#include <string>
 14#include <cmath>
 15
 16#include <dragon/serializable.hpp>
 17#include <dragon/dictionary.hpp>
 18#include <dragon/queue.hpp>
 19#include <dragon/barrier.hpp>
 20#include <dragon/semaphore.hpp>
 21
 22using namespace dragon;
 23
 24static timespec_t TIMEOUT = {30, 0};
 25
 26static SerializableString key(const std::string& name) {
 27    return SerializableString(name);
 28}
 29
 30static SerializableString key(const std::string& name, int worker_id) {
 31    return SerializableString(name + std::to_string(worker_id));
 32}
 33
 34int simulate(const char* ser_config, int worker_id) {
 35    //! [start-attaching-cpp]
 36    DDict<Serializable, Serializable> config(ser_config, &TIMEOUT);
 37
 38    // Every one of these values was written by Python.
 39    int cols = config[key("cols")];
 40    int steps = config[key("steps")];
 41    int start_row = config[key("start_", worker_id)];
 42    int end_row = config[key("end_", worker_id)];
 43
 44    // The whole grid. Its data lives in the DDict, so only the meta data was
 45    // passed to us. Nothing is read until refresh is called.
 46    SerializableDoubleNDArray grid = config[key("grid")];
 47
 48    // Just the rows this worker owns. This is where our results are published.
 49    SerializableDoubleNDArray band = config[key("band_", worker_id)];
 50
 51    Barrier barrier = config[key("barrier")];
 52    Semaphore ready = config[key("ready")];
 53    Queue<Serializable> results = config[key("results")];
 54
 55    //! [end-attaching-cpp]
 56    // Let Python know this worker has attached to everything.
 57    ready.release();
 58
 59    double max_delta = 0.0;
 60
 61    for (int step = 0; step < steps; step++) {
 62        // Pull the grid Python published at the end of the previous step.
 63        grid.refresh();
 64
 65        max_delta = 0.0;
 66
 67        for (int i = start_row; i < end_row; i++) {
 68            // Indexing a row once per row rather than once per column keeps the
 69            // temporaries out of the inner loop.
 70            //! [slice-start]
 71            SerializableDoubleNDArray above = grid[i-1];
 72            SerializableDoubleNDArray row = grid[i];
 73            SerializableDoubleNDArray below = grid[i+1];
 74            SerializableDoubleNDArray out = band[i - start_row];
 75            //! [slice-end]
 76
 77            for (int j = 0; j < cols; j++) {
 78                double current = row[j];
 79                double updated;
 80
 81                if (j == 0 || j == cols - 1) {
 82                    // The left and right edges are held at a fixed temperature.
 83                    updated = current;
 84                } else {
 85                    updated = 0.25 * ((double)above[j] + (double)below[j] +
 86                                      (double)row[j-1] + (double)row[j+1]);
 87                }
 88
 89                double delta = std::fabs(updated - current);
 90                if (delta > max_delta)
 91                    max_delta = delta;
 92
 93                out[j] = updated;
 94            }
 95        }
 96
 97        // Publish this worker's rows, then wait for every other worker to publish
 98        // theirs before Python stitches them back into the grid.
 99        band.sync();
100        barrier.wait(&TIMEOUT);
101
102        // Wait again so nobody starts the next step until the new grid is published.
103        barrier.wait(&TIMEOUT);
104    }
105
106    results.put(max_delta);
107
108    return 0;
109}
110
111int main(int argc, char* argv[]) {
112    if (argc < 3) {
113        std::cerr << "Usage: " << argv[0] << " <serialized_ddict> <worker_id>" << std::endl;
114        return 1;
115    }
116
117    const char* ser_config = argv[1];
118    int worker_id = std::stoi(argv[2]);
119
120    try {
121        return simulate(ser_config, worker_id);
122    } catch (const TimeoutError& e) {
123        std::cerr << "worker " << worker_id << ": timed out: " << e.err_str() << std::endl;
124    } catch (const DragonError& e) {
125        std::cerr << "worker " << worker_id << ": dragon error: " << e.err_str() << std::endl;
126    } catch (const std::exception& e) {
127        std::cerr << "worker " << worker_id << ": exception: " << e.what() << std::endl;
128    } catch (...) {
129        std::cerr << "worker " << worker_id << ": unknown exception" << std::endl;
130    }
131
132    return 1;
133}