Parallel gescharte Daten -> cupynumeric-Ladevorgang

Warum es diese Funktion gibt. cupynumeric spiegelt die Array-API von NumPy wider, einschließlich cupynumeric.load für eine einzelne .npy-Datei. Darüber hinaus erfolgt das Laden von Dateien in Legate, nicht in cupynumeric:

Diese Funktion zeigt den kanonischen Weg, um die Lücke in der letzten Zeile zu füllen: Schreiben Sie eine Legate-Python-Aufgabe, die den für das Format erforderlichen Leser eines Drittanbieters (h5py, pyarrow, np.memmap, …) innerhalb des Aufgabenkörpers aufruft, und lassen Sie Legate die Lesevorgänge auf GPUs/Knoten verteilen. Bei Formaten mit integriertem Loader sollten Sie diesen bevorzugen, es sei denn, Sie benötigen einen benutzerdefinierten Loader im Task-Körper (mmap-basierter Loader, formatspezifischer Decoder, Sidecar-Metadaten, partielle/aufgeteilte Lesevorgänge).

Standardmuster: manuelle Partitionierung + manueller Task-Start, dimensioniert auf die Maschine, nicht auf die Dateien. Nur Achse 0 wird fragmentiert; nachfolgende Achsen werden innerhalb jeder Kachel mitgeführt. Die Zeilenanzahl pro Shard kann sich zwischen den Dateien unterscheiden (nur dtype und nachfolgende Achsen müssen übereinstimmen); der Start füllt jeden verfügbaren Prozessor, unabhängig davon, wie viele Dateien vorhanden sind.

.npy dient als Beispiel, da der Header Form und Datentyp auf der Festplatte enthält, aber das Grundgerüst gilt für jedes Format mit kostengünstigen Bereichs-/Slice-Lesevorgängen (Raw-Binär, HDF5, Parquet/Arrow – siehe „Andere Formate“ weiter unten). Referenzimplementierung: assets/examples/parallel_npy_load.py.

Annahme zum Datenlayout

Bei dieser Funktion geht es ausschließlich um das Laden – es wird davon ausgegangen, dass die Daten bereits auf einem gemeinsam genutzten Dateisystem in einer vorhersehbaren, indizierbaren Weise angeordnet sind. Die Erstellung dieser Dateien liegt außerhalb des Anwendungsbereichs (das Beispiel enthält der Einfachheit halber einen Schreib- Unterbefehl, aber echte Nutzer stellen ihre eigenen bereit).

Das Beispiel geht von einem bestimmten Layout aus:

• Ein Verzeichnis mit Dateien namens shard_0000.npy, shard_0001.npy, … in einer zusammenhängenden ganzzahligen Folge (mit Nullen aufgefüllt, Breite 4).
• Alle Shards haben denselben Datentyp und dieselben nachgestellten Achsen (shape[1:]); Achse 0 (Zeilen pro Shard) kann sich zwischen den Dateien unterscheiden – das Rezept erstellt eine kumulative Zeilen-Offset-Tabelle und liest den überlappenden Ausschnitt jeder Datei aus der Leaf-Task heraus.
• Das Verzeichnis ist für jeden Rank sichtbar (gemeinsames Dateisystem für Läufe mit mehreren Knoten).

Die Funktion `discover_layout()` des Beispiels gibt die Ergebnisse aus und bricht mit einer ausführlichen Fehlermeldung ab, wenn das Layout falsch ist (fehlendes Verzeichnis, keine Shards, nicht übereinstimmender Datentyp /nachfolgende Achsen oder eine Lücke in der zusammenhängenden `shard_NNNN.npy `-Sequenz).

Wenn Ihre Daten in einem anderen Layout vorliegen – als binäre Rohdaten mit festem Schritt, als HDF5-Datei mit einem Datensatz pro Shard, als Verzeichnisbaum usw. – ändern sich lediglich das Glob-Muster, der dateispezifische Reader (Schritt 4 unten) und die Metadaten-Ermittlung (Schritt 1 unten). Die Mechanismen zur Partitionierung und zum Start sind layoutunabhängig.

Wann ist diese Funktion zu verwenden?

Siehe die obige Formattabelle für die Routing-Entscheidung (integrierter Loader vs. diese Funktion). Darüber hinaus gibt es zwei weitere Anhaltspunkte dafür, dass diese Funktion die richtige Wahl ist:

• Ersetzen von sequenziellen `np.concatenate([read(f) for f in files]) ` durch parallele Lesevorgänge pro GPU.
• Demonstration, wie eine benutzerdefinierte Legate-Python-Aufgabe über einen manuellen Start in ein cupynumerisches Ausgabe-Array schreibt.

Beispiele

Die folgenden Pfade sind relativ zum Verzeichnis dieses Skills angegeben (das Skript befindet sich unter „assets/examples/parallel_npy_load.py“). Passen Sie das Präfix so an, dass es mit dem Installationsverzeichnis Ihres Skills übereinstimmt (z. B. skills/cupynumeric-parallel-data-load/assets/..., wenn sich der Skill im obersten Verzeichnis „skills/“ befindet).

# Ein Knoten, 4 GPUs.
legate --gpus 4 --fbmem 4000 --min-gpu-chunk 1 \
    assets/examples/parallel_npy_load.py \
    read --shard-dir /shared/scratch/demo

# Mehrere Knoten, 2 Knoten × 4 GPUs (SLURM), gemeinsames Dateisystem unter --shard-dir.
# Die Shards einmal auf Rank 0 generieren, dann `read` in beliebigem Umfang erneut ausführen.
legate --launcher srun --nodes 2 --cpus 1 \
    assets/examples/parallel_npy_load.py \
    write --shard-dir /shared/scratch/demo

legate --launcher srun --nodes 2 --ranks-per-node 4 \
    --gpus 4 --fbmem 4000 --min-gpu-chunk 1 \
    assets/examples/parallel_npy_load.py \
    read --shard-dir /shared/scratch/demo

Keine Layout-Flags – der Lese-Treiber durchläuft jeden .npy-Header, um die Zeilenanzahl pro Datei, die abschließende Form und den Datentyp zu ermitteln, und leitet dann tile_rows aus der verfügbaren Prozessoranzahl ab.

--min-gpu-chunk 1 ist nur erforderlich, wenn die Elementanzahl pro Kachel unter der von Legate festgelegten Standard-Mindest-Chunk-Größe für GPU-Starts liegt (z. B. die Standardeinstellungen des Beispiels – die Gesamtzahl der Zeilen verteilt sich auf 4 GPUs mit ~1 Mio. pro Kachel – den Schwellenwert unterschreiten und andernfalls auf eine einzige GPU zusammengefasst würden). Bei Datensätzen in Produktionsgröße (zig Millionen Elemente pro Kachel oder mehr) können Sie das Flag weglassen und Legate die Standardeinstellung verwenden lassen. Eine Anhebung auf einen moderaten Wert (z. B. --min-gpu-chunk 1024) ist sinnvoll, wenn jede Kachel groß genug ist, sodass der Overhead pro Aufgabe wichtiger ist als die Zuweisung einer Kachel pro GPU.

Anleitung

Fünf Schritte aus einem .npy -Beispiel; nur Schritt 1 (Parsen des Format-Headers) und Schritt 4 (der dateispezifische Reader innerhalb des Aufgaben-Körpers) sind formatspezifisch. Die anderen drei (Ziel zuweisen, partitionieren, Fence) werden formatübergreifend unverändert wiederverwendet – siehe „Andere Formate“ weiter unten für die Swap-Punkte.

1. Metadaten aus jedem Shard lesen

Durchsuche das Verzeichnis und werfe einen Blick auf jeden .npy-Header (mmap_mode="r" liest nur den Header). Der Header enthält die Shard-spezifische Form und den Datentyp, sodass der Treiber die Gesamtzahl der Zeilen, die abschließende Form und eine kumulative Zeilen-Offset-Tabelle ermitteln kann, ohne die Daten jemals zu laden:

paths = sorted(SHARD_DIR.glob("shard_*.npy"))

per_file_rows = []                       # Zeilen entlang der Achse 0 pro Datei
trailing_shape = None                    # shape[1:], muss dateiübergreifend übereinstimmen
dtype = None
for p in paths:
    hdr = np.load(p, mmap_mode="r")
    if trailing_shape is None:
        trailing_shape = tuple(hdr.shape[1:])
        dtype = hdr.dtype
    elif tuple(hdr.shape[1:]) != trailing_shape or hdr.dtype != dtype:
        raise RuntimeError(
            f"{p.name}: Abweichung bei Form und Datentyp "
            f"({hdr.shape[1:]}/{hdr.dtype} vs {trailing_shape}/{dtype})"
        )
    per_file_rows.append(int(hdr.shape[0]))

cum_rows = np.cumsum([0] + per_file_rows, dtype=np.int64)  # Länge N+1
total_rows = int(cum_rows[-1])

Der obige Codeausschnitt stellt sicher, dass dtype und trailing_shape (d. h. shape[1:]) über alle Dateien hinweg übereinstimmen. Die Zeilenanzahl pro Shard kann variieren – die Tabelle „cum-rows“ berücksichtigt dies. Produktionscode sollte außerdem überprüfen, ob die Namen eine zusammenhängende Sequenz von shard_0000.npy … shard_NNNN.npy bilden (im Codeausschnitt der Kürze halber weggelassen; siehe discover_layout() im Beispiel). Die Erkennung stützt sich ausschließlich auf das, was das Format auf der Festplatte selbst offenlegt (hier der .npy-Header, .shape / .dtype für HDF5 usw.); etwaige Begleitdateien (Manifest, Inhalts-Hashes) sind ein separater Überprüfungsschritt darüber hinaus.

2. Erstellen Sie den cupynumeric-Ausgabespeicher aus den Metadaten

Das Gesamtarray erstreckt sich über total_rows entlang der Achse 0; nachfolgende Achsen stammen unverändert aus trailing_shape. Verwenden Sie cn.empty – die Aufgabe überschreibt jede Zelle, eine Initialisierung auf Null wäre verschwendet.

import cupynumeric as cn

total_shape = (total_rows,) + trailing_shape
out = cn.empty(total_shape, dtype=dtype)

3. Aufteilung des Speicherbereichs nach Prozessoranzahl

Die Startform wird an die verfügbaren Prozessoren angepasst, nicht an die Anzahl der Dateien. Wähle `tile_rows = ceil(total_rows / num_processors) ` und partitioniere die Achse 0 nach dieser Kachelgröße. Nachfolgende Achsen werden nicht partitioniert (die Kachel erstreckt sich dort über den gesamten Umfang). Die letzte Kachel darf kurz sein – genau das unterstützt „partition_by_tiling“ –, sodass das Rezept keine Teilbarkeitsbeschränkung benötigt.

from legate.core import TaskTarget, get_legate_runtime
from legate.core.data_interface import as_logical_array

runtime = get_legate_runtime()
machine = runtime.get_machine()
num_processors = max(
    machine.count(TaskTarget.GPU),
    machine.count(TaskTarget.OMP),
    machine.count(TaskTarget.CPU),
    1,
)

tile_rows = max(1, (total_rows + num_processors - 1) // num_processors)
tile_shape = (tile_rows,) + trailing_shape
partition = as_logical_array(out).data.partition_by_tiling(tile_shape)

num_tasks = (total_rows + tile_rows - 1) // tile_rows  # Anpassung an die Anzahl der Kacheln in der Partition

4. Definieren Sie die Blatt-Aufgabe und starten Sie sie manuell

PATHS und CUM_ROWS (die Dateipfade und die Tabelle mit den kumulativen Zeilen-Offsets aus Schritt 1) sowie TILE_ROWS werden vor dem Start vom Treiber als globale Modulvariablen gesetzt; die Replikationssteuerung führt den Treiber auf jedem Rank aus, sodass jeder Worker identische Werte sieht.

Jede Aufgabe erstellt zunächst ihre Consumer-Ansicht (cupy auf der GPU, numpy auf der CPU/OMP) und liest die tatsächliche Zeilenanzahl des Tiles aus ` view.shape[0]` – PhysicalStore selbst verfügt über kein .shape-Attribut, daher ist der Zugriff über die Ansicht erforderlich. Anschließend berechnet sie ihren globalen Zeilenbereich anhand ihrer Startkoordinate und dieser Zeilenanzahl, teilt cum_rows für die überlappenden Dateien in zwei Hälften und kopiert jeden überlappenden Dateischnitt in den entsprechenden Zielschnitt. Registriere CPU-, OMP- und GPU-Varianten, damit derselbe Start überall unverändert ausgeführt wird; die Verteilung über ctx.get_variant_kind() wählt den Consumer aus, der dem Speicherort des OutputStore ansässig ist (cp.from_dlpack(dst) für FBMEM, np.asarray(dst) für SYSMEM). cupy wird nur innerhalb des GPU-Zweigs importiert, sodass der Task-Körper auf Rechnern ohne cupy geladen wird.

import bisect
from legate.core import TaskContext, VariantCode
from legate.core.task import OutputStore, task

@task(variants=(VariantCode.CPU, VariantCode.OMP, VariantCode.GPU))
def load_tile(ctx: TaskContext, dst: OutputStore) -> None:
    t = ctx.task_index[0]                              # Kachelindex 0..num_tasks-1

    variant = ctx.get_variant_kind()
    if variant == VariantCode.GPU:
        import cupy as cp                              # verzögert: nur auf der GPU
        view = cp.from_dlpack(dst)
    else:
        view = np.asarray(dst)                         # Zero-Copy-Numpy-Ansicht

    tile_rows_actual = view.shape[0]                   # zu kurz auf der letzten Kachel
    row_start = t * TILE_ROWS                          # globaler Start der Achse 0
    row_end = row_start + tile_rows_actual

    # Ermitteln des halboffenen Bereichs der Dateiindizes, die sich mit [row_start, row_end) überschneiden.
    first_file = bisect.bisect_right(CUM_ROWS, row_start) - 1
    last_file = bisect.bisect_right(CUM_ROWS, row_end - 1) - 1

    for f in range(first_file, last_file + 1):
        # Schnittmenge von Kachel [row_start, row_end) mit Datei [cum[f], cum[f+1]).
        lo = max(row_start, int(CUM_ROWS[f]))
        hi = min(row_end, int(CUM_ROWS[f + 1]))
        file_lo = lo - int(CUM_ROWS[f])
        file_hi = hi - int(CUM_ROWS[f])
        dst_lo = lo - row_start
        dst_hi = hi - row_start
        chunk = np.ascontiguousarray(
            np.load(PATHS[f], mmap_mode="r")[file_lo:file_hi]
        )
        if variant == VariantCode.GPU:
            view[dst_lo:dst_hi].set(chunk)             # cudaMemcpyAsync H2D
        else:
            view[dst_lo:dst_hi] = chunk                # Zero-Copy-Numpy-Schreiben

manual_task = runtime.create_manual_task(
    load_tile.library,
    load_tile.task_id,
    (num_tasks,),                                      # Startdomäne == Anzahl der Kacheln
)
manual_task.add_output(partition)
manual_task.execute()

Beide Verbraucher nutzen die nativen Produzenten von PhysicalStore (__dlpack__ für cupy, __array_interface__ für np.asarray) – Zero-Copy-Ansichten der lokalen Kachel. Der Bisect-Aufwand beträgt O(log num_shards) und die innere Schleife wird typischerweise 1–2 Mal durchlaufen (Kacheln überlappen sich höchstens über ein paar Dateien).

5. Abgrenzen und Verifizieren

get_legate_runtime().issue_execution_fence(block=True)

Harte Einschränkungen

1. Alle Shards müssen denselben Datentyp und dieselben nachlaufenden Achsen (shape[1:]) aufweisen. Das Rezept stapelt die Shards entlang der Achse 0; die nachlaufenden Achsen des Ziels stammen aus trailing_shape, das im Erkennungsschritt auf den Wert der ersten Datei festgelegt wird. Die Zeilenanzahl pro Shard (shape[0]) darf frei variieren – die Tabelle mit kumulativen Offsets berücksichtigt dies. Das Beispiel lehnt jeden Shard, dessen Datentyp oder nachfolgende Form vom ersten abweicht, mit einer aussagekräftigen Fehlermeldung ab.

2. Wählen Sie den Consumer aus, der zur Variante passt. cp.from_dlpack lehnt SYSMEM-residente Speicher ab; np.asarray gibt stillschweigend eine Host-Ansicht eines FBMEM-residenten Speichers zurück, in den Sie tatsächlich nicht schreiben können. Führen Sie eine Verzweigung anhand von ` ctx.get_variant_kind() ` durch, damit jede Variante ihren eigenen Consumer verwendet – siehe Schritt 4.

3. mmap-Ansichten sind nicht immer C-kontigu — umschließe jeden dateispezifischen Ausschnitt mit `np.ascontiguousarray(arr[file_lo:file_hi])`, bevor du `.set() ` oder den Numpy-In-Place-Schreibvorgang ausführst.

4. Mehrknoten: SHARD_DIR muss sich auf einem gemeinsam genutzten Dateisystem befinden. Jeder Worker (auf jedem Rank) öffnet Shards anhand des Pfads; knotenlokale /tmp-Pfade funktionieren nur bei Ein-Knoten-Demos.

Varianten

Schneller Pfad mit einheitlichen Shards (eine Aufgabe pro Datei)

Wenn jeder Shard bereits dieselbe Form und denselben Datentyp hat und Sie zufällig genau so viele Prozessoren wie num_shards zur Verfügung haben, entsteht durch die „cum-rows“- und „bisect“-Mechanismen ein Overhead. Setze `tile_rows` = `shard_shape[0] ` und `num_tasks` = `num_shards`; die Partition hat dann eine Kachel pro Datei und jede Aufgabe liest genau eine Datei von Anfang bis Ende (kein Bisect, keine innere Schleife). Der Schalter auf der Treiberseite ist eine Einzeiler-Anweisung:

if all(r == per_file_rows[0] for r in per_file_rows) and num_shards == num_processors:
    tile_rows = per_file_rows[0]
else:
    tile_rows = max(1, (total_rows + num_processors - 1) // num_processors)

Der gleiche „load_tile“-Aufgaben körper funktioniert in beiden Modi – die innere Schleife wird lediglich genau einmal pro Aufgabe durchlaufen. Es ist kein separater Aufgabenkörper für den Fast-Path erforderlich.

Überdekomponierung für eine bessere Lastverteilung

Die Standardeinstellung `tile_rows = ceil(total_rows / num_processors) ` ergibt eine Kachel pro Prozessor. Um um den Faktor K zu überdekomponieren (kleinere Kacheln, mehr Punkt-Aufgaben, feinkörnigere Warteschlangen), dividiere stattdessen durch `K * num_processors`:

tile_rows = max(1, (total_rows + K * num_processors - 1) // (K * num_processors))

num_tasks = ceil(total_rows / tile_rows) ergibt dann ungefähr K * num_processors. Der gleiche Aufgabentext funktioniert weiterhin – die Bisektionsmethode führt lediglich zu mehr Aufgaben pro Datei.

Andere Formate

Nur der dateispezifische Reader innerhalb von `load_tile` ändert sich. Die Aufgabe des Readers: Bei Angabe eines Dateipfads und eines halb-offenen Zeilenbereichs `[file_lo, file_hi)` entlang der Achse 0 soll ein NumPy-Array der Form (file_hi - file_lo,) + trailing_shape, das C-kontiguös gemacht werden kann. Es sind kostengünstige Bereichs-/Slice-Lesevorgänge erforderlich – Formate, die nur „das Lesen der gesamten Datei“ unterstützen, machen den Fall der teilweisen Überlappung zunichte (ein Kachelbereich, der nur einen Teil einer Datei abdeckt).

(Informationen zu den integrierten Ein-Aufruf-Ladern pro Format finden Sie in der Tabelle „Warum es diese Funktion gibt“ am Anfang dieser Datei.)

Im Erkennungsschritt (Schritt 1) werden die Metadaten jedes Formats analysiert: .npy / HDF5 / Parquet enthalten auf der Festplatte die Zeilenanzahl pro Datei sowie den Datentyp. Bei rohen Binärdateien ist dies nicht der Fall – hier wird der Wert aus der Dateigröße abgeleitet oder in einer separaten Datei gespeichert.

Häufige Fallstricke

cn.asarray(dst) ist in einer Blatt-Aufgabe unzulässig

Innerhalb eines @task- Körpers löst jede cupynumeric-Operation, die die oberste Laufzeitebene berührt – cn.asarray(store), Slice-Zuweisung cn_dst[s] = host_np – „create_index_space“ aus dem falschen Kontext aus, und Legion bricht ab:

LEGION API USAGE EXCEPTION: Ungültiger Task-Kontext an den Laufzeitaufruf
create_index_space übergeben

Behebung: Verwenden Sie die DLPack-Kapsel mit einer Drittanbieter-Bibliothek (cupy / torch / numpy) innerhalb von Leaf-Tasks. cn.asarray ist im Treiber zulässig, jedoch nicht in Leaf-Tasks. Siehe examples/dlpack/leaf_task_interop.py für die „Torch“-basierte Problemumgehung.

Ein „assert“ innerhalb einer Aufgabe bricht die Laufzeit ab

Legate behandelt nicht ausgelöste Ausnahmen in einem @task als Vertragsverletzung und bricht den Vorgang ab, sofern die Aufgabe nicht mit throws_exception() registriert wurde. Führen Sie vor dem Start eine Plausibilitätsprüfung auf dem Host durch.

Die Startdomäne muss mit der Anzahl der Partitionskacheln übereinstimmen

`create_manual_task(launch_shape=...) ` und ` partition_by_tiling(...)` sind voneinander unabhängig – die Laufzeit erkennt eine Nichtübereinstimmung nicht. Größere Startdomäne → Kacheln außerhalb des Bereichs; kleinere → nicht beschriebene Kacheln. Leiten Sie immer beide aus denselben (total_rows, tile_rows) über zwei separate Ceil- Divisionen ab (die direkte Anpassung der Startdomäne an num_processors würde zu einem übermäßigen Start führen, wenn num_processors > total_rows):

tile_rows = max(1, (total_rows + num_processors - 1) // num_processors)
num_tasks = (total_rows + tile_rows - 1) // tile_rows
partition = ...partition_by_tiling((tile_rows,) + trailing_shape)
runtime.create_manual_task(load_tile.library, load_tile.task_id, (num_tasks,))