ディープラーニングモデルの安全な並列推論とパフォーマンス最適化

ディープラーニングモデルの安全な並列推論とパフォーマンス最適化
Photo by Amy Chen / Unsplash

こんにちは!

今日は、よく聞かれる質問の1つである「単一のモデルインスタンスで安全に並列推論を行えるか?」に関する内容です!

evalモードでの並列推論の安全性

PyTorchモデルがmodel.eval()を使用してevalモードに設定されている場合、一般的に並列推論に対して安全になります。

(ここでいう「並列」はマルチスレッドによる処理ととらえてください。バッチ推論については後述します。)

その理由は、

  1. パラメータの不変性
    evalモードでは、順伝播(forward pass)中にモデルのパラメータが更新されません。
  2. 学習特有レイヤーの非活性化
    BatchNormなどのレイヤーは、バッチ統計の計算ではなく、実行時統計(running statistics)を使用するモードに切り替わります。
  3. 入力データの独立性
    各スレッドやプロセスは独自の入力データで動作し、それぞれ別のメモリ領域に存在します。

以下は、evalモードでの安全な並列推論の基本的な例です:

import torch
import threading

def safe_inference(model, data):
    with torch.no_grad():
        return model(data)

model = YourModel()
model.eval()  # 重要: evalモードに設定

# 複数スレッドで推論を実行
threads = []
for i in range(10):
    t = threading.Thread(target=safe_inference, args=(model, your_data[i]))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

注意が必要な場合

しかし、以下のような状況では注意が必要です:

  1. カスタムレイヤーの存在
    独自に実装したレイヤーがある場合、その並列実行時の挙動を慎重に確認する必要があります。
class CustomLayer(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.counter = 0  # 潜在的な問題源

    def forward(self, x):
        self.counter += 1  # スレッドセーフではない
        return x + self.counter

# このようなカスタムレイヤーは並列実行時に問題を引き起こす可能性があります
  1. GPUメモリの制約
    複数スレッドが同時に大量のデータを処理する場合、GPUメモリ不足が発生する可能性があります。
  2. 複雑なモデル構造
    特定のタイプのAttentionメカニズムなど、一部の複雑なモデル構造では、並列実行時に予期せぬ挙動を示す可能性があります。

プールの使用

上記のような注意が必要な場合、モデルインスタンスのプールを使用することで問題を回避できる場合があります。
以下は簡単なモデルプールの実装例です

import torch
from queue import Queue

class ModelPool:
    def __init__(self, model_class, num_instances):
        self.pool = Queue()
        for _ in range(num_instances):
            model = model_class().to('cuda')
            model.eval()
            self.pool.put(model)

    def get_model(self):
        return self.pool.get()

    def return_model(self, model):
        self.pool.put(model)

def safe_pooled_inference(pool, data):
    model = pool.get_model()
    try:
        with torch.no_grad():
            result = model(data)
        return result
    finally:
        pool.return_model(model)

# 使用例
pool = ModelPool(YourModel, num_instances=3)
results = [safe_pooled_inference(pool, data) for data in your_data_list]

このアプローチでは、各推論タスクが独立したモデルインスタンスを使用するため、並列実行時の問題を回避できます。

パフォーマンスの最適化の基本はバッチ

並列推論は柔軟性を提供しますが、オーバーヘッドによりパフォーマンスが低下する可能性があります。ここでは、パフォーマンスを向上させるための重要なヒントを紹介します。

バッチ処理の活用

個別の並列推論よりも、バッチ処理を活用することで大幅なパフォーマンス向上が見込めます。GPUは大量のデータを同時に処理するのに適しているため、バッチ処理はGPUの能力を最大限に活用できます。

1. 静的バッチ処理

最も単純な方法は、固定サイズのバッチを使用することです:

def batch_inference(model, data_list, batch_size=32):
    results = []
    for i in range(0, len(data_list), batch_size):
        batch = torch.stack(data_list[i:i+batch_size])
        with torch.no_grad():
            batch_results = model(batch)
        results.extend(batch_results)
    return results

# 使用例
results = batch_inference(model, your_data_list)

ただし、都合よくバッチのタイミングでアクセスは来ない

Webサービスなのでオンデマンドな推論サービスをつくってるときには、GPUの単純な並列推論だけでは対処しきれません。

なぜなら、都合よく、同じタイミングでユーザーがアクセスしてこないからです。
むしろうまくバッチにのせられるタイミングのほうがマレです。

2. ダイナミックバッチング

リアルタイムで到着するデータを効率的に処理するために、ダイナミックバッチングを使用できます

import time
from collections import deque

class DynamicBatcher:
    def __init__(self, model, max_batch_size=32, max_wait_time=0.1):
        self.model = model
        self.max_batch_size = max_batch_size
        self.max_wait_time = max_wait_time
        self.queue = deque()
        self.results = {}

    def add_item(self, item_id, data):
        self.queue.append((item_id, data))
        if len(self.queue) >= self.max_batch_size:
            self.process_batch()

    def process_batch(self):
        batch_ids, batch_data = zip(*[self.queue.popleft() for _ in range(len(self.queue))])
        batch_tensor = torch.stack(batch_data)
        with torch.no_grad():
            batch_results = self.model(batch_tensor)
        for item_id, result in zip(batch_ids, batch_results):
            self.results[item_id] = result

    def get_result(self, item_id):
        start_time = time.time()
        while item_id not in self.results:
            if time.time() - start_time > self.max_wait_time:
                self.process_batch()
            time.sleep(0.01)
        return self.results.pop(item_id)

# 使用例
batcher = DynamicBatcher(model)

def process_item(item_id, data):
    batcher.add_item(item_id, data)
    return batcher.get_result(item_id)

# 複数スレッドからprocess_itemを呼び出す

このアプローチでは、データが到着次第バッチに追加され、バッチサイズが最大に達するか、最大待機時間を超えた場合に処理が実行されます。

3. 連続バッチ処理

また、連続的にデータが生成される場合、以下のような連続バッチ処理が効果的です

import torch
from torch.utils.data import DataLoader, IterableDataset

class ContinuousDataset(IterableDataset):
    def __iter__(self):
        while True:
            yield self.get_next_item()  # データ生成ロジックを実装

    def get_next_item(self):
        # 実際のデータ生成ロジックをここに実装
        pass

def continuous_batch_inference(model, dataset, batch_size=32):
    dataloader = DataLoader(dataset, batch_size=batch_size)
    for batch in dataloader:
        with torch.no_grad():
            yield model(batch)

# 使用例
dataset = ContinuousDataset()
for batch_results in continuous_batch_inference(model, dataset):
    process_results(batch_results)  # 結果の処理

この方法では、データが連続的に生成される場合でも、効率的にバッチ処理を行うことができます。

まとめ

今回は、とくに1台のGPUにおける並列化とパフォーマンスについて解説しました。
evalモードでの並列推論は多くの場合安全ですが、パフォーマンスを最大化するためにはバッチ処理が必須ですね。またディープラーニング、LLM系のサービスの推論シーンは多くの場合でダイナミックバッチング、連続バッチ処理などの技術が重要となります。当社でも、ダイナミックバッチ、連続バッチを当初から研究しており、LLMや動画生成、AIキャラクター応答にも応用しています。
これらのテクニックを適切に選択し、実装することで、推論のスループットを大幅に向上させることができます。

並列化とは別観点ではありますが、モデルの量子化、TorchScript の使用、GPU 最適化など、追加の手法を組み合わせることで、さらなるパフォーマンス向上が期待できます。

GPUはとても高額な機器なので、1台のGPUを「使い切る」という視点は非常に重要で当社Qualitegでも日々技術を磨いています。

さらに大規模なアクセスには「GPUクラスター」の導入を考えましょう

一方、大量の同時アクセスが想定されるシーンでは複数台のGPUを使用した負荷分散が必須となります。そちらのテクニックについてもまた別途ブログにて投稿させていただこうとおもいますが、以下の動画に LLM におけるGPUクラスターの構成方法について解説していますので、こちらもよろしければご覧くださいませ。

それでは、また次回お会いしましょう!

Read more

Pythonの落とし穴:__len__メソッドを実装したらオブジェクトの真偽値判定が変わってしまった話

Pythonの落とし穴:__len__メソッドを実装したらオブジェクトの真偽値判定が変わってしまった話

こんにちは! Pythonでカスタムクラスを作成していて、 「オブジェクトは存在するのにif文でFalseと判定される」 という不可解な現象に遭遇したことはありませんか? この記事では、__len__メソッドを実装することで生じる、予期しない真偽値判定の挙動について解説いたします! 実際に遭遇したバグ ユーザーの投稿を管理するクラスを実装していたときのことです class PostManager: """ブログ投稿を管理するクラス""" def __init__(self, user_id): self.user_id = user_id self._posts = [] self._cache = {} def __len__(self): """投稿数を返す""" return len(self._posts) def add_post(

By Qualiteg プロダクト開発部
CEATEC 2025に出展します!フォトリアルAIアバター「MotionVox🄬」の最新版を実体験いただけます

CEATEC 2025に出展します!フォトリアルAIアバター「MotionVox🄬」の最新版を実体験いただけます

株式会社Qualitegは、2025年10月14日(火)~17日(金)に幕張メッセで開催される「CEATEC 2025」に出展いたします。今回の出展では、当社が開発したフォトリアリスティックAIアバター技術「MotionVox🄬」をはじめ、最新のAI技術とビジネスイノベーションソリューションをご紹介いたします。 出展概要 * 会期:2025年10月14日(火)~10月17日(金) * 会場:幕張メッセ * 出展エリア:ネクストジェネレーションパーク * ブース番号:ホール6 6H207 * CEATEC内特設サイト:https://www.ceatec.com/nj/exhibitor_detail_ja?id=1915 見どころ:最先端AI技術を体感できる特別展示 1. フォトリアルAIアバター「MotionVox🄬」 テキスト入力だけで、まるで本物の人間のような動画を生成できる革新的なAIアバターシステムです。 MotionVox🄬は自社開発している「Expression Aware🄬」技術により日本人の演者データを基に開発された、

By Qualiteg ニュース
その処理、GPUじゃなくて勝手にCPUで実行されてるかも  ~ONNX RuntimeのcuDNN 警告と対策~

その処理、GPUじゃなくて勝手にCPUで実行されてるかも ~ONNX RuntimeのcuDNN 警告と対策~

こんにちは! 本日は、ONNX RuntimeでGPU推論時の「libcudnn.so.9: cannot open shared object file」エラーの解決方法についての内容となります。 ONNX Runtimeを使用してGPU推論を行う際、CUDAプロバイダの初期化エラーに遭遇することがありますので、このエラーの原因と解決方法を解説いたします。 エラーメッセージの詳細 [E:onnxruntime:Default, provider_bridge_ort.cc:2195 TryGetProviderInfo_CUDA] /onnxruntime_src/onnxruntime/core/session/provider_bridge_ort.cc:1778 onnxruntime::Provider& onnxruntime::ProviderLibrary::Get() [ONNXRuntimeError] : 1 : FAIL : Failed to load

By Qualiteg プロダクト開発部
大企業のAIセキュリティを支える基盤技術 - 今こそ理解するActive Directory 第3回 クライアントとサーバーのドメイン参加

大企業のAIセキュリティを支える基盤技術 - 今こそ理解するActive Directory 第3回 クライアントとサーバーのドメイン参加

こんにちは、今回はシリーズ第3回クライアントとサーバーのドメイン参加について解説いたします! はじめに こんにちは!シリーズ第3回「クライアントとサーバーのドメイン参加」へようこそ。 前回(第2回)では、Active Directoryドメイン環境の構築手順について、ドメインコントローラーのセットアップからDNS設定まで詳しく解説しました。ドメイン環境の「土台」が整ったところで、今回はいよいよ実際にコンピューターをドメインに参加させる手順に進みます。 「ドメインユーザーアカウントを作ったのに、なぜかログインできない」「新しいPCを追加したけど、ドメイン認証が使えない」といった経験はありませんか?実は、Active Directoryの世界では、ユーザーアカウントを作成しただけでは不十分で、そのユーザーが使用するコンピューター自体もドメインに「参加」させる必要があるのです。 本記事では、このドメイン参加について、単なる手順の説明にとどまらず、「なぜドメイン参加が必要なのか」「裏側で何が起きているのか」という本質的な仕組みまで、初心者の方にも分かりやすく解説していきます。Win

By Qualiteg コンサルティング