セマフォを使用したプロデューサーとコンシューマー

プロデューサーは、バッファの末尾に到達するまでデータを書き込み、その時点で先頭から書き直し、既存のデータを上書きします。コンシューマースレッドは、生成されたデータを読み取り、標準エラー出力に書き込みます。

セマフォを使用することで、ミューテックスよりも高いレベルの並行性を実現できます。もしバッファへのアクセスが `QMutex` によって保護されていた場合、コンシューマースレッドはプロデューサースレッドと同時にバッファにアクセスすることはできません。しかし、両方のスレッドがバッファの異なる部分を同時に処理しても問題はありません。

この例には、Producer とConsumer の2つのクラスが含まれています。両クラスともQThread を継承しています。これら2つのクラス間の通信に使用される循環バッファと、それを保護するセマフォはグローバル変数です。

QSemaphore を使用してプロデューサー・コンシューマー問題を解決する代わりに、QWaitCondition およびQMutex を使用する方法もあります。これは、「待機条件を使用したプロデューサーとコンシューマー」の例で実施されている方法です。

グローバル変数

まず、循環バッファとそれに関連するセマフォについて確認しましょう：

constexpr int DataSize = 100000;

constexpr int BufferSize = 8192;
char buffer[BufferSize];

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

DataSize は、プロデューサーが生成するデータの量です。例をできるだけ単純にするため、これを定数とします。BufferSize は循環バッファのサイズです。これはDataSize よりも小さいため、ある時点でプロデューサーはバッファの末尾に到達し、先頭から再開始することになります。

プロデューサーとコンシューマーを同期させるには、2つのセマフォが必要です。「freeBytes 」セマフォは、バッファの「空き」領域（プロデューサーがまだデータを書き込んでいない領域、またはコンシューマーがすでに読み取った領域）を制御します。usedBytes セマフォは、バッファの「使用済み」領域（プロデューサーがデータを書き込んだが、コンシューマーがまだ読み取っていない領域）を制御します。

これら2つのセマフォが連携することで、プロデューサーがコンシューマーよりBufferSize バイト以上先を行くことがなく、またコンシューマーがプロデューサーがまだ生成していないデータを読み取ることがないよう保証されます。

freeBytes セマフォはBufferSize で初期化されます。これは、初期状態ではバッファ全体が空であるためです。usedBytes セマフォは 0（指定がない場合のデフォルト値）で初期化されます。

プロデューサークラス

Producer クラスのコードを確認しましょう：

class Producer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            freeBytes.acquire();
            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];
            usedBytes.release();
        }
    }
};

プロデューサーはDataSize バイト分のデータを生成します。循環バッファにバイトを書き込む前に、freeBytes セマフォを使用して「空き」バイトを取得する必要があります。コンシューマーがプロデューサーのペースに追いついていない場合、QSemaphore::acquire()の呼び出しがブロックされる可能性があります。

最後に、プロデューサーはusedBytes セマフォを使用して1バイトを解放します。これにより、「空き」バイトは「使用済み」バイトへと正常に変換され、コンシューマーが読み取る準備が整います。

コンシューマークラス

それでは、Consumer クラスについて見ていきましょう。

class Consumer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            usedBytes.acquire();
            fprintf(stderr, "%c", buffer[i % BufferSize]);
            freeBytes.release();
        }
        fprintf(stderr, "\n");
    }
};

このコードはプロデューサーのコードと非常に似ていますが、今回は逆の操作、つまり「使用済み」バイトを取得し、「空き」バイトを解放するという点が異なります。

main() 関数

main() では、2つのスレッドを作成し、QThread::wait() を呼び出して、終了する前に両方のスレッドが処理を完了する時間を確保します：

int main(int argc, char *argv[])
{
    QCoreApplication app(argc, argv);
    Producer producer;
    Consumer consumer;
    producer.start();
    consumer.start();
    producer.wait();
    consumer.wait();
    return 0;
}

では、プログラムを実行すると何が起こるのでしょうか？ 最初は、プロデューサースレッドだけが動作可能です。コンシューマースレッドは、usedBytes セマフォが解放されるのを待ってブロックされています（その初期のavailable() のカウントは 0 です）。 プロデューサーがバッファに 1 バイトを書き込んだ時点で、freeBytes.available() はBufferSize - 1となり、usedBytes.available() は1になります。この時点で、2つのケースが考えられます。コンシューマースレッドが処理を引き継いでそのバイトを読み取るか、あるいはプロデューサースレッドが 2 バイト目を書き込むかのいずれかです。

この例で示したプロデューサー・コンシューマーモデルにより、高度に並行性のあるマルチスレッドアプリケーションを作成することが可能になります。マルチプロセッサマシン上では、2つのスレッドがバッファの異なる部分で同時にアクティブになることができるため、このプログラムは、同等のミューテックスベースのプログラムに比べて最大で2倍の速度が出る可能性があります。

ただし、これらの利点が常に実現されるとは限らない点に注意してください。QSemaphore の取得と解放にはコストがかかります。実際には、バッファをチャンクに分割し、個々のバイトではなくチャンク単位で処理を行う方が有益である可能性が高いでしょう。また、バッファサイズも、実験に基づいて慎重に選択しなければならないパラメータです。

サンプルプロジェクト @ code.qt.io