QThreadPool と QSemaphore を使ったマルチスレッドサンプル

この記事は、 私のソフトで利用したマルチスレッドの作り方を解説するものです。
もっとスマートな書き方もあると思いますし、 間違いを解説しているかも知れません。
ただ、 １０年以上マルチスレッドのソフトを仕事で開発して来て、 概ね間違っていないだろうと思います。

ネット上で全く日本語での解説を目にしないので、 マルチスレッド初心者の参考になればと書いています。
数回に分けて、 解説します。

QThreadPool の使い方

QThreadPool はスレッドをプールして管理してくれるクラスです。

• 最大スレッド数を設定する

  m_threadPool->setMaxThreadCount(最大スレッド数);
  この最大スレッド数で一番効率が良いとされているのは、 CPU コア数+1 です。
  
  自分でも大きくして試して見ましたが、 処理の終了時間は余り変わりませんでした。
  統計を取って msec 単位で比較すれば、 CPU コア数+1 が一番効率が良いのでしょう。
  +1 なのは、 スレッド終了判定用のスレッドが１つ必要だからです。
  このスレッドは各スレッドを起動させたら、 後は終了判定位なので CPU 使用を控えめにして、 他のスレッドはフルに CPU を利用するようにする。
  しかし必ず必要なので、 +1 が効率的なのですね。

threadcore.cpp
// addthread=1 CPU Core 数＋1
ThreadCore::ThreadCore(int addthread)
{
	// スレッドプールを作成
	m_threadPool = new QThreadPool(this);
	// CPU Core数
	int cpucore = QThread::idealThreadCount();
	// CPU Coreの数が１つの場合は＋2しないと動かない
	if(cpucore == 1 && addthread < 2)
		addthread = 2;

	m_maxThreadCnt = cpucore + addthread;
	// 最大スレッド数を設定
	m_threadPool->setMaxThreadCount(m_maxThreadCnt);
	m_abort = false;
}

• ThreadPoolTaskクラスを作って、これをスレッドプール内で走らせます。

threadcore.h
class ThreadPoolTask : public QObject, public QRunnable
{
	Q_OBJECT
signals:
	void start();

public:
	ThreadPoolTask(QThreadPool* tp,bool multiThread = true) 
	{
		threadPool = tp;
		m_multiThread = multiThread;
	}
	void SetMultiThread(bool multiThread) {
		QMutexLocker ml(&multiThreadMutex);
		m_multiThread = multiThread;
	}
	bool isMultiThread() {
		QMutexLocker ml(&multiThreadMutex);
		return m_multiThread;
	}

protected:
	void run() {
		if (isMultiThread())
			threadPool->tryStart(this);
		emit start();
	}

private:
	QMutex multiThreadMutex;
	QThreadPool* threadPool;
	bool m_multiThread;
};

ThreadPoolTask クラスを使って、 スレッド内で実行される関数を登録します。

threadcore.cpp
	// スレッドメイン処理 シングルスレッド
	dirwalker = new ThreadPoolTask(m_threadPool, false);
	connect(dirwalker,SIGNAL(start()),this,SLOT(doWalkDirs()),Qt::DirectConnection);
	dirwalker->setAutoDelete(true);
	// 開始
	m_threadPool->start(dirwalker);

	// サブディレクトリ単位で並列処理、マルチスレッド
	subdirwalker = new ThreadPoolTask(m_threadPool, true);
	connect(subdirwalker,SIGNAL(start()),this,SLOT(doWalkSubDirs()),Qt::DirectConnection);
	subdirwalker->setAutoDelete(true);
	// 開始
	m_threadPool->start(subdirwalker);

	// ファイル単位で並列処理、マルチスレッド
	loadtask = new ThreadPoolTask(m_threadPool, true);
	connect(loadtask,SIGNAL(start()),this,SLOT(doLoadFile()),Qt::DirectConnection);
	loadtask->setAutoDelete(true);
	// まだ開始しない

new ThreadPoolTask(m_threadPool, true);
最後のパラメータ true がマルチスレッドです。
false ではシングルスレッド指定になります。
dirwalker オブジェクトは、 最初のスレッド起動と終了判定なので、 シングルです。

loadtask はまだ start していないので、 threadpool には登録されていません。

• doWalkDirs 関数

  dirwalker オブジェクトで呼び出される関数です。
  スレッドメイン処理 シングルスレッドで動きます。

threadcore.cpp
void ThreadCore::doWalkDirs()
{
	qDebug() << "###### Start #####"<< getCurrentThreadId() ;
	// パラメータでセットされたパス分ループする
	for(int i=0; i < m_pathlist.size(); ++i)
	{
		QString path = m_pathlist.at(i);
		if(path.isEmpty())
			continue;
		qDebug() << path;
		// 処理中のディレクトリ名をセット
		setProcessName(path);
		// ディレクトリ待ち行列に登録
		appendDirName(path);
		// ディレクトリ以下の検索開始
		m_dirSemaphore.release();	// セマフォを１つ確保
	}
	// 以下スレッド終了判定
		: 中略
}

// 処理中のファイル名をセット
void ThreadCore::setProcessName(const QString &name)
{
	QMutexLocker ml(&m_nameMutex);
	m_procname = name;
}

// ディレクトリ待ち行列の最後に登録
void ThreadCore::appendDirName(const QString &path)
{
	// ディレクトリ以下の検索登録
	QMutexLocker ml(&m_dirNameMutex);
	// ディレクトリ待ち行列に登録
	m_dirName.append(path);
}

setProcessName(path)
親画面で現在処理しているディレクトリ名 ・ ファイル名をステータスバーに表示する時に使う変数にセットします。
appendDirName(path)
ディレクトリ待ち行列の最後に登録します。
doWalkSubDirs 関数で待ち行列から取り出す時は先頭です。

• 変数の排他

  ２つ共、 QMutexLocker ml(&ミューテックス変数) となっています。
  これは複数のスレッドから呼び出されるため、 排他しないと読み出しと書込みでプログラムが落ちてしまうのでミューテックスで排他しています。
  この排他は必ずマルチスレッドプログラミングでは行わなければなりません。

   m_dirNmaeMutex.lock();
   m_dirName.append(path);
   m_dirNmaeMutex.unlock();

通常はこうしますが、 Qt では QMutexLocker という便利なクラスがあります。
変数のスコープが有効な間だけ lock してくれます。
unlock が自動という事です。

• doWalkSubDirs 関数

  マルチスレッドで動かす関数です。
  ディレクトリ単位で実行します。
  セマフォでマルチスレッド実行を同期制御します。

threadcore.cpp
// ファイル名を全て読み取る
// セマフォ処理あり
void ThreadCore::doWalkSubDirs()
{
	qDebug() << "> doWalkSubDir Thread start "<< getCurrentThreadId();
	int foldercnt = 0;
	while( !isAbort() )
	{
		// １つセマフォを取得
		// ディレクトリ待ち行列が無い場合はここで停止して許可されるのを待ちます
		m_dirSemaphore.acquire();
		int dncnt = getDirNameCount();
		if( dncnt == 0) break;
		QString path = takeDirNameFirst();
//		qDebug() <<  "subdir" << m_dirName.count() << path;

• セマフォ

  m_dirSemaphore.acquire();
  セマフォが解放されていない場合は、 この行で停止します。
  m_threadPool->start(subdirwalker); で実行された、 doWalkSubDirs 関数は全てこの行で停止しています。
  マルチスレッドの同期制御は、 Mutex 変数のみでは無理なのです。
  OS のタイムスライスは、 順不同でどのスレッドが実行されるかは未定なので偏りが出てしまいます。
  
  どのスレッドが終了するのが早いか分かないから、 排他だけでは無理なのです。
  そこでセマフォ変数で、 排他と同期を制御します。
  m_dirSemaphore.release();
  １つリリースすると、 acquire() が動きます。
  たったこれだけです。とてもシンプルですが、 数で管理が出来ます。
  ただ、 終了させる時に、 空 release()が必要になります。

• 待ち行列とセマフォの連動

  待ち行列とセマフォは１対で同じタイミングで追加します。
  そして、 release() すると、 スレッドで、 acquire()で停止していた行がセマフォ変数の１つを消費して、 動きます。

		// セマフォ待ち
		m_dirSemaphore.acquire();
		// セマフォ１つを消費して、動き始める
		int dncnt = getDirNameCount();
		if( dncnt == 0) break;
		// 待ち行列の先頭を取り出して
		QString path = takeDirNameFirst();
		// 処理を開始
			 : 中略

• マルチスレッド動作

  次々に待ち行列とセマフォを追加すると、 スレッドは自動で割り当てられ実行されます。
  この部分がセマフォを使った、 同期制御のメリットです。
  サンプルプログラムを実行すると、 スレッドＩＤに対して処理数がちゃんと分散されて
  実行されているのが分かると思います。

• サンプルプログラム

  
  サンプルプログラムは、 Windows でも linux でも動作確認をしましたので、
  ダウンロードして Qt5 以降の処理系でコンパイル実行して見てください。

マルチスレッドサンプルプログラム
ソースダウンロード