Effiziente metrische Sammlung in Pytorch: Vermeiden Sie die Leistungsfälle von Torchmetrics

Die Metriksammlung ist ein wesentlicher Bestandteil jedes maschinellen Lernprojekts, mit dem wir die Modellleistung verfolgen und den Trainingschritt überwachen können. Ideally suited, Metriken sollte gesammelt und berechnet werden, ohne zusätzlichen Aufwand in den Schulungsprozess einzuführen. Genau wie andere Komponenten der Trainingsschleife kann ineffiziente metrische Berechnung unnötige Gemeinkosten einführen, die Trainingsstufe erhöhen und die Schulungskosten aufblasen.

Dieser Beitrag ist der siebte in unserer Serie auf Leistungsprofilierung und Optimierung in Pytorch. Die Serie zielte darauf ab, die kritische Rolle der Leistungsanalyse zu betonen und Optimierung in der Entwicklung des maschinellen Lernens. Jeder Beitrag hat sich auf verschiedene Phasen der Trainingspipeline konzentriert und praktische Instruments und Techniken zur Analyse und Steigerung der Ressourcenauslastung und zur Laufzeiteffizienz demonstriert.

In dieser Charge konzentrieren wir uns auf die metrische Sammlung. Wir werden zeigen, wie sich eine naive Implementierung der metrischen Sammlung negativ auf die Laufzeitleistung auswirken und Instruments und Techniken für ihre Analyse und Optimierung untersuchen kann.

Um unsere metrische Sammlung zu implementieren, werden wir verwenden Torchmetrics Eine beliebte Bibliothek, die zur Vereinfachung und Standardisierung der metrischen Berechnung in der Berechnung in der Berechnung von Pytorch. Unsere Ziele werden sein,:

1. Demonstrieren Sie den Laufzeitaufwand verursacht durch eine naive Implementierung der metrischen Sammlung.
2. Verwenden Sie den Pytorch -Profiler Um Leistungs Engpässe zu bestimmen, die durch metrische Berechnung eingeführt wurden.
3. Demonstrieren Sie Optimierungstechniken Um die metrische Sammlung zu reduzieren.

Um unsere Diskussion zu erleichtern, werden wir ein Spielzeug -Pytorch -Modell definieren und beurteilen, wie sich die metrische Sammlung auf die Laufzeitleistung auswirken kann. Wir werden unsere Experimente auf einer Nvidia A40 -GPU mit a durchführen Pytorch 2.5.1 Docker Bild und TorchMetrics 1.6.1.

Es ist wichtig zu beachten, dass das metrische Sammlungsverhalten je nach {Hardware}, Laufzeitumgebung und Modellarchitektur stark variieren kann. Die in diesem Beitrag bereitgestellten Codeausschnitte dienen nur für Demonstrationszwecke. Bitte interpretieren Sie unsere Erwähnung eines Werkzeugs oder einer Technik nicht als Bestätigung für die Verwendung.

Toy Resnet -Modell

Im folgenden Codeblock definieren wir ein einfaches Bildklassifizierungsmodell mit a Resnet-18 Rückgrat.

import time
import torch
import torchvision

machine = "cuda"

mannequin = torchvision.fashions.resnet18().to(machine)
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(mannequin.parameters())

Wir definieren einen synthetischen Datensatz, mit dem wir unser Spielzeugmodell trainieren werden.

from torch.utils.information import Dataset, DataLoader

# A dataset with random pictures and labels
class FakeDataset(Dataset):
    def __len__(self):
        return 100000000

    def __getitem__(self, index):
        rand_image = torch.randn((3, 224, 224), dtype=torch.float32)
        label = torch.tensor(information=index % 1000, dtype=torch.int64)
        return rand_image, label

train_set = FakeDataset()

batch_size = 128
num_workers = 12

train_loader = DataLoader(
    dataset=train_set,
    batch_size=batch_size,
    num_workers=num_workers,
    pin_memory=True
)

Wir definieren eine Sammlung von Standardmetriken aus TorchMetrics sowie ein Kontrollflag, um die metrische Berechnung zu aktivieren oder zu deaktivieren.

from torchmetrics import (
    MeanMetric,
    Accuracy,
    Precision,
    Recall,
    F1Score,
)

# toggle to allow/disable metric assortment
capture_metrics = False

if capture_metrics:
        metrics = {
        "avg_loss": MeanMetric(),
        "accuracy": Accuracy(job="multiclass", num_classes=1000),
        "precision": Precision(job="multiclass", num_classes=1000),
        "recall": Recall(job="multiclass", num_classes=1000),
        "f1_score": F1Score(job="multiclass", num_classes=1000),
    }

    # Transfer all metrics to the machine
    metrics = {title: metric.to(machine) for title, metric in metrics.objects()}

Als nächstes definieren wir a Pytorch -Profiler Instanz, zusammen mit einem Kontrollflag, das es uns ermöglicht, das Profiling zu aktivieren oder zu deaktivieren. Ein detailliertes Tutorial zur Verwendung von Pytorch -Profiler finden Sie in der Erster Beitrag In dieser Serie.

from torch import profiler

# toggle to allow/disable profiling
enable_profiler = True

if enable_profiler:
    prof = profiler.profile(
        schedule=profiler.schedule(wait=10, warmup=2, lively=3, repeat=1),
        on_trace_ready=profiler.tensorboard_trace_handler("./logs/"),
        profile_memory=True,
        with_stack=True
    )
    prof.begin()

Zuletzt definieren wir einen Commonplace -Trainingsschritt:

mannequin.practice()

t0 = time.perf_counter()
total_time = 0
depend = 0

for idx, (information, goal) in enumerate(train_loader):
    information = information.to(machine, non_blocking=True)
    goal = goal.to(machine, non_blocking=True)
    optimizer.zero_grad()
    output = mannequin(information)
    loss = criterion(output, goal)
    loss.backward()
    optimizer.step()

    if capture_metrics:
        # replace metrics
        metrics("avg_loss").replace(loss)
        for title, metric in metrics.objects():
            if title != "avg_loss":
                metric.replace(output, goal)

        if (idx + 1) % 100 == 0:
            # compute metrics
            metric_results = {
                title: metric.compute().merchandise() 
                    for title, metric in metrics.objects()
            }
            # print metrics
            print(f"Step {idx + 1}: {metric_results}")
            # reset metrics
            for metric in metrics.values():
                metric.reset()

    elif (idx + 1) % 100 == 0:
        # print final loss worth
        print(f"Step {idx + 1}: Loss = {loss.merchandise():.4f}")

    batch_time = time.perf_counter() - t0
    t0 = time.perf_counter()
    if idx > 10:  # skip first steps
        total_time += batch_time
        depend += 1

    if enable_profiler:
        prof.step()

    if idx > 200:
        break

if enable_profiler:
    prof.cease()

avg_time = total_time/depend
print(f'Common step time: {avg_time}')
print(f'Throughput: {batch_size/avg_time:.2f} pictures/sec')

Metrische Sammlung Overhead

Um die Auswirkungen der metrischen Sammlung auf die Trainingsschrittzeit zu messen, haben wir unser Trainingsskript sowohl mit als auch ohne metrische Berechnung durchgeführt. Die Ergebnisse sind in der folgenden Tabelle zusammengefasst.

Unsere naive metrische Sammlung führte zu einem Rückgang der Laufzeitleistung von quick 10% !! Während die metrische Sammlung für die Entwicklung des maschinellen Lernens unerlässlich ist, beinhaltet sie normalerweise relativ einfache mathematische Operationen und rechtfertigt kaum einen so erheblichen Aufwand. Was ist los?!!

Identifizierung von Leistungsproblemen mit Pytorch -Profiler

Um die Quelle der Leistungsverschlechterung besser zu verstehen, können wir das Trainingsskript mit dem aktivierten Pytorch -Profiler umgeben. Die resultierende Spur ist unten dargestellt:

Die Spur zeigt wiederkehrende „Cudastreamsynchronize“ -Operationen, die mit merklichen Tropfen der GPU -Nutzung übereinstimmen. Diese Arten von „CPU-GPU-Synchronisationsereignis Teil zwei unserer Serie. In einem typischen Trainingsschritt arbeitet die CPU und die GPU parallel: Die CPU verwaltet Aufgaben wie Datenübertragungen in die GPU- und Kernelbelastung, und die GPU führt das Modell für die Eingabedaten aus und aktualisiert ihre Gewichte. Im Idealfall möchten wir die Synchronisationspunkte zwischen CPU und GPU minimieren, um die Leistung zu maximieren. Hier können wir jedoch sehen, dass die metrische Sammlung ein Synchronisierungsereignis ausgelöst hat, indem eine CPU an die GPU -Datenkopie durchgeführt wird. Dies erfordert, dass die CPU ihre Verarbeitung aushält, bis die GPU auffasst, was wiederum dazu führt, dass die GPU darauf wartet, dass die CPU die nachfolgenden Kernelvorgänge aufnimmt. Das Fazit ist, dass diese Synchronisationspunkte zu einer ineffizienten Nutzung sowohl der CPU als auch der GPU führen. Unsere Implementation für die metrische Sammlung fügt jedem Trainingsschritt acht solcher Synchronisationsereignisse hinzu.

Eine genauere Untersuchung der Spur zeigt, dass die SYNC -Ereignisse von der stammen aktualisieren Ruf der Gemein Torchmetrik. Für den erfahrenen Profilerstellungsexperten kann dies ausreichen, um die Grundursache zu identifizieren, aber wir werden einen Schritt weiter gehen und die verwenden fackel.profiler.record_function Dienstprogramm zur Identifizierung der genauen beleidigenden Codezeile.

Profilerstellung mit record_function

Um die genaue Quelle des Synchronisationsereignisses zu bestimmen, haben wir die erweitert Gemein Klasse und überschreiben die aktualisieren Methode verwendet record_function Kontextblöcke. Dieser Ansatz ermöglicht es uns, individuelle Vorgänge innerhalb der Methode zu profilieren und Leistungsgpässe zu identifizieren.

class ProfileMeanMetric(MeanMetric):
    def replace(self, worth, weight = 1.0):
        # broadcast weight to worth form
        with profiler.record_function("course of worth"):
            if not isinstance(worth, torch.Tensor):
                worth = torch.as_tensor(worth, dtype=self.dtype,
                                        machine=self.machine)
        with profiler.record_function("course of weight"):
            if weight shouldn't be None and never isinstance(weight, torch.Tensor):
                weight = torch.as_tensor(weight, dtype=self.dtype,
                                         machine=self.machine)
        with profiler.record_function("broadcast weight"):
            weight = torch.broadcast_to(weight, worth.form)
        with profiler.record_function("cast_and_nan_check"):
            worth, weight = self._cast_and_nan_check_input(worth, weight)

        if worth.numel() == 0:
            return

        with profiler.record_function("replace worth"):
            self.mean_value += (worth * weight).sum()
        with profiler.record_function("replace weight"):
            self.weight += weight.sum()

Anschließend haben wir unsere AVG_Loss -Metrik aktualisiert, um das neu erstellte ProfileMeanMetric und das Trainingskript erneut zu verwenden.

Die aktualisierte Hint zeigt, dass das Sync -Ereignis aus der folgenden Zeile stammt:

weight = torch.as_tensor(weight, dtype=self.dtype, machine=self.machine)

Dieser Vorgang wandelt den Commonplace -Skalarwert um weight=1.0 in einen Pytorch -Tensor und legt ihn auf die GPU. Das Synchronisierungsereignis tritt auf, da diese Aktion eine CPU-to-GPU-Datenkopie auslöst, wodurch die CPU darauf wartet, dass die GPU den kopierten Wert verarbeitet.

Optimierung 1: Gewichtswert angeben

Nachdem wir die Quelle des Issues gefunden haben, können wir sie leicht überwinden, indem wir a angeben Gewicht Wert in unserem aktualisieren Anruf. Dies verhindert, dass die Laufzeit den Commonplace -Skalar konvertieren weight=1.0 in einen Tensor auf der GPU und vermeiden Sie das Synchronisierungsereignis:

# replace metrics
 if capture_metric:
     metrics("avg_loss").replace(loss, weight=torch.ones_like(loss))

Nach dem Drehbuch nach der Anwendung dieser Änderung ergibt _cast_and_nan_check_input Funktion:

Profilerstellung mit record_function – Teil 2

Um unser neues Synchronisationsereignis zu untersuchen, haben wir unsere benutzerdefinierte Metrik mit zusätzlichen Profile -Sonden erweitert und unser Skript erneut übertragen.

class ProfileMeanMetric(MeanMetric):
    def replace(self, worth, weight = 1.0):
        # broadcast weight to worth form
        with profiler.record_function("course of worth"):
            if not isinstance(worth, torch.Tensor):
                worth = torch.as_tensor(worth, dtype=self.dtype,
                                        machine=self.machine)
        with profiler.record_function("course of weight"):
            if weight shouldn't be None and never isinstance(weight, torch.Tensor):
                weight = torch.as_tensor(weight, dtype=self.dtype,
                                         machine=self.machine)
        with profiler.record_function("broadcast weight"):
            weight = torch.broadcast_to(weight, worth.form)
        with profiler.record_function("cast_and_nan_check"):
            worth, weight = self._cast_and_nan_check_input(worth, weight)

        if worth.numel() == 0:
            return

        with profiler.record_function("replace worth"):
            self.mean_value += (worth * weight).sum()
        with profiler.record_function("replace weight"):
            self.weight += weight.sum()

    def _cast_and_nan_check_input(self, x, weight = None):
        """Convert enter ``x`` to a tensor and test for Nans."""
        with profiler.record_function("course of x"):
            if not isinstance(x, torch.Tensor):
                x = torch.as_tensor(x, dtype=self.dtype,
                                    machine=self.machine)
        with profiler.record_function("course of weight"):
            if weight shouldn't be None and never isinstance(weight, torch.Tensor):
                weight = torch.as_tensor(weight, dtype=self.dtype,
                                         machine=self.machine)
            nans = torch.isnan(x)
            if weight shouldn't be None:
                nans_weight = torch.isnan(weight)
            else:
                nans_weight = torch.zeros_like(nans).bool()
                weight = torch.ones_like(x)

        with profiler.record_function("any nans"):
            anynans = nans.any() or nans_weight.any()

        with profiler.record_function("course of nans"):
            if anynans:
                if self.nan_strategy == "error":
                    elevate RuntimeError("Encountered `nan` values in tensor")
                if self.nan_strategy in ("ignore", "warn"):
                    if self.nan_strategy == "warn":
                        print("Encountered `nan` values in tensor."
                              " Might be eliminated.")
                    x = x(~(nans | nans_weight))
                    weight = weight(~(nans | nans_weight))
                else:
                    if not isinstance(self.nan_strategy, float):
                        elevate ValueError(f"`nan_strategy` shall be float"
                                         f" however you move {self.nan_strategy}")
                    x(nans | nans_weight) = self.nan_strategy
                    weight(nans | nans_weight) = self.nan_strategy

        with profiler.record_function("return worth"):
            retval = x.to(self.dtype), weight.to(self.dtype)
        return retval

Die resultierende Spur wird unten erfasst:

Die Ablaufverfolgung zeigt direkt in die beleidigende Linie:

anynans = nans.any() or nans_weight.any()

Dieser Betrieb überprüft nach NaN Werte in den Eingangstensoren, jedoch ein kostspieliges CPU-GPU-Synchronisationsereignis einführt, da der Betrieb Daten von der GPU an die CPU kopiert.

Bei einer genaueren Prüfung des Torchmetric Baseaggregator Klasse finden wir mehrere Optionen für die Bearbeitung von NAN -Wert -Updates, die alle durch die beleidigende Codezeile geleitet werden. Für unseren Anwendungsfall – die Berechnung der durchschnittlichen Verlustmetrik – ist dieser Scheck jedoch unnötig und rechtfertigt die Laufzeitleistung nicht.

Optimierung 2: Deaktivieren Sie NAN -Wertprüfungen

Um den Overhead zu beseitigen, schlagen wir vor, die zu deaktivieren NaN Wertprüfungen durch Überschreiben der _cast_and_nan_check_input Funktion. Anstelle einer statischen Übersteuerung haben wir eine dynamische Lösung implementiert, die flexibel auf alle Nachkommen der angewendet werden kann BaseaggregatorKlasse.

from torchmetrics.aggregation import BaseAggregator

def suppress_nan_check(MetricClass):
    assert issubclass(MetricClass, BaseAggregator), MetricClass
    class DisableNanCheck(MetricClass):
        def _cast_and_nan_check_input(self, x, weight=None):
            if not isinstance(x, torch.Tensor):
                x = torch.as_tensor(x, dtype=self.dtype, 
                                    machine=self.machine)
            if weight shouldn't be None and never isinstance(weight, torch.Tensor):
                weight = torch.as_tensor(weight, dtype=self.dtype,
                                         machine=self.machine)
            if weight is None:
                weight = torch.ones_like(x)
            return x.to(self.dtype), weight.to(self.dtype)
    return DisableNanCheck

NoNanMeanMetric = suppress_nan_check(MeanMetric)

metrics("avg_loss") = NoNanMeanMetric().to(machine)

Nachoptimierungsergebnisse: Erfolg

Nach der Implementierung der beiden Optimierungen – Angabe des Gewichtswerts und Deaktivieren der NaN Überprüfungen – Wir finden die Schrittzeitleistung und die GPU -Auslastung, die denen unseres Baseline -Experiments entspricht. Darüber hinaus zeigt die resultierende Pytorch -Profiler -Hint, dass alle zusätzlichen Ereignisse „Cudastreamsynchronize“, die mit der metrischen Sammlung verbunden waren, eliminiert wurden. Mit ein paar kleinen Änderungen haben wir die Schulungskosten um ~ 10% gesenkt, ohne dass sich das Verhalten der Metriksammlung ändert.

Im nächsten Abschnitt werden wir eine zusätzliche Optimierung der metrischen Sammlung untersuchen.

Beispiel 2: Optimierung der Metrik -Geräteplatzierung

Im vorherigen Abschnitt befanden sich die Metrikwerte auf der GPU, sodass sie logisch die Metriken auf der GPU speichern und berechnen können. In Szenarien, in denen sich die Werte, die wir aggregieren möchten, auf der CPU liegen, kann es jedoch vorzuziehen sein, die Metriken auf der CPU zu speichern, um unnötige Geräteübertragungen zu vermeiden.

Im folgenden Codeblock ändern wir unser Skript, um die durchschnittliche Schrittzeit mit a zu berechnen Gemein auf der CPU. Diese Änderung hat keinen Einfluss auf die Laufzeitleistung unseres Trainingsschritts:

avg_time = NoNanMeanMetric()
t0 = time.perf_counter()

for idx, (information, goal) in enumerate(train_loader):
    # transfer information to machine
    information = information.to(machine, non_blocking=True)
    goal = goal.to(machine, non_blocking=True)

    optimizer.zero_grad()
    output = mannequin(information)
    loss = criterion(output, goal)
    loss.backward()
    optimizer.step()

    if capture_metrics:
        metrics("avg_loss").replace(loss)
        for title, metric in metrics.objects():
            if title != "avg_loss":
                metric.replace(output, goal)

        if (idx + 1) % 100 == 0:
            # compute metrics
            metric_results = {
                title: metric.compute().merchandise()
                    for title, metric in metrics.objects()
            }
            # print metrics
            print(f"Step {idx + 1}: {metric_results}")
            # reset metrics
            for metric in metrics.values():
                metric.reset()

    elif (idx + 1) % 100 == 0:
        # print final loss worth
        print(f"Step {idx + 1}: Loss = {loss.merchandise():.4f}")

    batch_time = time.perf_counter() - t0
    t0 = time.perf_counter()
    if idx > 10:  # skip first steps
        avg_time.replace(batch_time)

    if enable_profiler:
        prof.step()

    if idx > 200:
        break

if enable_profiler:
    prof.cease()

avg_time = avg_time.compute().merchandise()
print(f'Common step time: {avg_time}')
print(f'Throughput: {batch_size/avg_time:.2f} pictures/sec')

Das Downside tritt auf, wenn wir versuchen, unser Skript zu erweitern, um verteilte Schulungen zu unterstützen. Um das Downside zu demonstrieren, haben wir unsere Modelldefinition geändert, um sie zu verwenden DistributedDataparallel (DDP) :

# toggle to allow/disable ddp
use_ddp = True

if use_ddp:
    import os
    import torch.distributed as dist
    from torch.nn.parallel import DistributedDataParallel as DDP
    os.environ("MASTER_ADDR") = "127.0.0.1"
    os.environ("MASTER_PORT") = "29500"
    dist.init_process_group("nccl", rank=0, world_size=1)
    torch.cuda.set_device(0)
    mannequin = DDP(torchvision.fashions.resnet18().to(machine))
else:
    mannequin = torchvision.fashions.resnet18().to(machine)

# insert coaching loop

# append to finish of the script:
if use_ddp:
    # destroy the method group
    dist.destroy_process_group()

Die DDP -Modifikation führt zum folgenden Fehler:

RuntimeError: No backend sort related to machine sort cpu

Standardmäßig werden Metriken im verteilten Coaching so programmiert, dass alle im verwendeten Geräte über alle Geräte synchronisiert werden. Das von DDP verwendete Synchronisations -Backend unterstützt jedoch keine auf der CPU gespeicherten Metriken.

Eine Möglichkeit, dies zu lösen, besteht darin, die Cross-System-Metriksynchronisation zu deaktivieren:

avg_time = NoNanMeanMetric(sync_on_compute=False)

In unserem Fall ist diese Lösung akzeptabel, wenn wir die durchschnittliche Zeit messen. In einigen Fällen ist die metrische Synchronisation jedoch unerlässlich, und wir haben möglicherweise keine andere Wahl, als die Metrik auf die GPU zu bewegen:

avg_time = NoNanMeanMetric().to(machine)

Leider führt diese Scenario zu einem neuen CPU-GPU-Synchronisationsereignis aus dem aktualisieren Funktion.

Dieses Synchronisierungsereignis sollte kaum überraschen – nach allem aktualisieren wir eine GPU -Metrik mit einem Wert, der sich auf der CPU befindet, was eine Speicherkopie erfordern sollte. Bei einer skalaren Metrik kann diese Datenübertragung jedoch mit einer einfachen Optimierung vollständig vermieden werden.

Optimierung 3: Führen Sie metrische Aktualisierungen mit Tensoren anstelle von Skalaren durch

Die Lösung ist unkompliziert: Anstatt die Metrik mit einem Float -Wert zu aktualisieren replace.

batch_time = torch.as_tensor(batch_time)
avg_time.replace(batch_time, torch.ones_like(batch_time))

Diese geringfügige Änderung umgeht die problematische Codezeile, beseitigt das SYNC -Ereignis und stellt die Schrittzeit für die Basisleistung wieder her.

Auf den ersten Blick magazine dieses Ergebnis überraschend erscheinen: Wir würden erwarten, dass die Aktualisierung einer GPU -Metrik mit einem CPU -Tensor noch eine Speicherkopie erfordern sollte. Pytorch optimiert jedoch den Vorgang auf skalaren Tensoren, indem ein dedizierter Kernel verwendet wird, der die Zugabe ohne explizite Datenübertragung durchführt. Dies vermeidet das teure Synchronisierungsereignis, das sonst auftreten würde.

Zusammenfassung

In diesem Beitrag haben wir untersucht, wie ein naiver Ansatz für TorchMetrics CPU-GPU-Synchronisierungsereignisse einführen und die Pytorch-Trainingsleistung erheblich beeinträchtigen kann. Mit dem Pytorch -Profiler haben wir die Codezeilen identifiziert, die für diese Synchronisierungsereignisse verantwortlich sind, und gezielte Optimierungen angewendet haben, um sie zu beseitigen:

• Geben Sie beim Aufrufen des MeanMetric.replace Funktion anstatt sich auf den Standardwert zu verlassen.
• Deaktivieren Sie NAN -Überprüfungen in der Foundation Aggregator Klasse oder ersetzen Sie sie durch eine effizientere Various.
• Verwalten Sie die Geräteplatzierung jeder Metrik sorgfältig, um unnötige Übertragungen zu minimieren.
• Deaktivieren Sie die metrische Synchronisation mit Cross-System-Synchronisation, wenn nicht erforderlich.
• Wenn die Metrik auf einer GPU liegt replace Funktion, um implizite Synchronisation zu vermeiden.

Wir haben eine engagierte geschaffen Anfrage ziehenauf der Torchmetrics GithubSeiten, die einige der in diesem Beitrag diskutierten Optimierungen abdecken. Bitte zögern Sie nicht, Ihre eigenen Verbesserungen und Optimierungen beizutragen!