待機条件を使用したプロデューサーとコンシューマー

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

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

この例は、Producer とConsumer の 2 つのクラスで構成されています。どちらもQThread を継承しています。これら 2 つのクラス間の通信に使用される循環バッファと、それを保護する同期ツールは、グローバル変数です。

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

グローバル変数

まず、円形バッファとそれに関連する同期ツールについて確認しましょう。

constexpr int DataSize = 100000;
constexpr int BufferSize = 8192;

QMutex mutex; // protects the buffer and the counter
char buffer[BufferSize];
int numUsedBytes;

QWaitCondition bufferNotEmpty;
QWaitCondition bufferNotFull;

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

プロデューサーとコンシューマーを同期させるには、2つの待機コンディションと1つのミューテックスが必要です。「bufferNotEmpty 」コンディションは、プロデューサーがデータを生成した際にシグナルが送られ、コンシューマーに対して読み取りを開始してもよいことを伝えます。「bufferNotFull 」コンディションは、コンシューマーがデータを読み取った際にシグナルが送られ、プロデューサーに対してさらにデータを生成してもよいことを伝えます。numUsedBytes は、バッファ内のデータを含むバイト数を表します。

待機コンディション、ミューテックス、および `numUsedBytes ` カウンタが連携して、プロデューサーがコンシューマーより `BufferSize ` バイト以上先を行くことがないようにし、また、コンシューマーがプロデューサーがまだ生成していないデータを読み取らないようにします。

プロデューサークラス

Producer クラスのコードを確認してみましょう。

class Producer : public QThread
{
public:
    explicit Producer(QObject *parent = nullptr)
        : QThread(parent)
    {
    }

private:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            {
                const QMutexLocker locker(&mutex);
                while (numUsedBytes == BufferSize)
                    bufferNotFull.wait(&mutex);
            }

            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];

            {
                const QMutexLocker locker(&mutex);
                ++numUsedBytes;
                bufferNotEmpty.wakeAll();
            }
        }
    }
};

プロデューサーはDataSize バイト分のデータを生成します。循環バッファにバイトを書き込む前に、まずバッファが満杯かどうか（つまり、numUsedBytes がBufferSize に等しいかどうか）を確認する必要があります。バッファが満杯の場合、スレッドはbufferNotFull コンディションで待機します。

最後に、プロデューサーはnumUsedBytes をインクリメントし、numUsedBytes は必然的に0より大きいため、条件bufferNotEmpty が真であることをシグナルします。

numUsedBytes 変数へのすべてのアクセスは、ミューテックスで保護されています。さらに、QWaitCondition::wait() 関数は引数としてミューテックスを受け取ります。このミューテックスは、スレッドがスリープ状態に入る前にロックが解除され、スレッドが復帰した際にロックされます。また、競合状態の発生を防ぐため、ロック状態から待機状態への遷移はアトミックに行われます。

コンシューマクラス

次に、Consumer クラスを見てみましょう：

class Consumer : public QThread
{
public:
    explicit Consumer(QObject *parent = nullptr)
        : QThread(parent)
    {
    }

private:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            {
                const QMutexLocker locker(&mutex);
                while (numUsedBytes == 0)
                    bufferNotEmpty.wait(&mutex);
            }

            fprintf(stderr, "%c", buffer[i % BufferSize]);

            {
                const QMutexLocker locker(&mutex);
                --numUsedBytes;
                bufferNotFull.wakeAll();
            }
        }
        fprintf(stderr, "\n");
    }
};

このコードはプロデューサーのものと非常に似ています。バイトを読み取る前に、バッファが満杯かどうかではなく、空かどうか（numUsedBytes が 0 かどうか）を確認し、空の場合はbufferNotEmpty コンディションで待機します。 バイトを読み込んだ後、numUsedBytes をインクリメントする代わりにデクリメントし、bufferNotEmpty 条件の代わりにbufferNotFull 条件をシグナルします。

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;
}

では、プログラムを実行するとどうなるでしょうか？ 最初は、プロデューサースレッドだけが動作可能です。コンシューマースレッドは、bufferNotEmpty コンディションがシグナルされるのを待機してブロックされています（numUsedBytes は0です）。プロデューサーがバッファに1バイトを書き込むと、numUsedBytes は0より確実に大きくなり、bufferNotEmpty コンディションがシグナルされます。 その時点で、2つのケースが考えられます。コンシューマースレッドが処理を引き継いでそのバイトを読み取るか、あるいはプロデューサーが2バイト目を生成するかのいずれかです。

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

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

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