PostgreSQL/開発

バックグラウンドワーカプロセス

PostgreSQLでは、クライアントコネクション接続を処理するバックエンドプロセスとは異なるプロセスを実行することができる。このバックグラウンドワーカプロセスは、postmasterプロセスによって起動(fork)され、ワーカプロセスの状態はpostmasterの起動・停止・監視とリンクしている。このバックグラウンドワーカの仕組みは、パラレルワーカでも使われており、バックエンドプロセスから動的にバックグラウンドワーカプロセスを登録(※)することも可能である。

バックグラウンドワーカは、バージョン9.3から機能追加されている。
9.4からは動的にバックグラウンドワーカを起動できるように拡張されている。

background-worker-overview.png

補足※ 登録と記述しているのは、実際にプロセス生成を行なうのがpostmasterプロセスであるためである。バックグラウンドワーカの起動は、呼び出し側がposmasterにシグナル送信(SIGUSR1)し、postmasterがforkすることで実現されている。

バックグラウンドワーカプロセスのフロー

バックグラウンドワーカは、Postmaster起動時及び起動後のタイミングで登録することができる(便宜上、ここではPostmaster起動時のワーカを静的、Posmaster起動後のワーカを動的と呼ぶこととする)。静的、動的では以下の違いがある。

静的・動的でバックグラウンドワーカを登録する関数は異なるが、バックグラウンドワーカ自体は、同じ構造体変数で表現され、共有メモリ上のバックグラウンドワーカ配列で管理される。

background-worker-process-image.png

BackgroundWorker構造体

番号データ型フィールド説明
1charbgw_name[BGW_MAXLEN]バックグラウンドワーカの名前。プロセスリスト、ログで使用される。
2intbgw_flagsバックグラウンドワーカが要求する機能。
BGWORKER_SHMEM_ACCESS
共有メモリへのアクセスをする。
BGWORKER_BACKEND_DATABASE_CONNECTION
データベース接続をする。注意データベース接続をする場合は、BGWORKER_SHMEM_ACCESSも指定する必要がある。
3BgWorkerStartTimebgw_start_timeバックグラウンドワーカの起動タイミング。参考bgworker_should_start_now() - https://git.postgresql.org/gitweb/
BgWorkerStart_PostmasterStart
postmasterの起動時でServerLoop()の前(pmState = PM_STARTUP、StartupStatus = STARTUP_RUNNING)。
リカバリ開始のタイミング 参考Postmasterプロセス
BgWorkerStart_ConsistentState
ホットスタンバイモードで参照クエリの受け付けが可能になったタイミング(pmState = PM_HOT_STANDBY、次のログが確認されたタイミング:database system is ready to accept read only connections
BgWorkerStart_RecoveryFinished
リカバリが完了し、システムが通常の参照/更新クエリを実行できるようになった時(pmState = PM_RUN、StartupStatus = STARTUP_NOT_RUNNING
参考reaper() - https://git.postgresql.org/gitweb/
4intbgw_restart_timeバックグラウンドワーカがクラッシュ(終了コード0以外)してから再起動するまでの時間間隔(秒)。
BGW_NEVER_RESTARTが指定された場合は再起動しない。
5charbgw_library_name[BGW_MAXLEN]バックグラウンドワーカモジュールのライブラリ名。ワーカのコードを含む拡張機能モジュールのファイル名。モジュールの動的ロードに必要。
6charbgw_function_name[BGW_MAXLEN]バックグラウンドワーカのエントリーポイント。entrypoint(Datum main_arg)という呼び出しとなる。
7Datumbgw_main_argバックグラウンドワーカのエントリーポイントに渡される引数。
8charbgw_extra[BGW_EXTRALEN]引数以外にデータの受け渡しが必要な場合に使用可能な領域。
9pid_tbgw_notify_pidバックグラウンドワーカの終了を通知するプロセスID(SIGUSR1シグナルが送信される)。通常は、呼び出しプロセスのPIDであるMyProcPidを指定する。

参考struct BackgroundWorker - https://git.postgresql.org/gitweb/

サンプル 静的バックグラウンドワーカの登録

バックグラウンドワーカの登録は、BackgroundWorker構造体を介して行なう。以下、静的バックグラウンドワーカを登録する例である。

src/test/modules/worker_spiにあるサンプルが大変参考になる。

void
_PG_init(void)
{
	BackgroundWorker worker;

	memset(&worker, 0, sizeof(worker));

        // ワーカは、データベース接続を必要とする
	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |
		BGWORKER_BACKEND_DATABASE_CONNECTION;

        /* リカバリ完了しクエリの受け付け可能になったタイミングで起動 */
	worker.bgw_start_time = BgWorkerStart_RecoveryFinished;
        /* クラッシュしても再起動しない */
	worker.bgw_restart_time = BGW_NEVER_RESTART;
        /* モジュール名 */
	sprintf(worker.bgw_library_name, "myworker");
        /* エントリーポイント */
	sprintf(worker.bgw_function_name, "myworker_main");
        /* 静的バックグラウンドワーカの場合は、0以外は無効(ワーカが登録されない) */
	worker.bgw_notify_pid = 0;
        /* プロセスリストやログに表示される名前 */
	snprintf(worker.bgw_name, BGW_MAXLEN, "myworker");

	/* ワーカスロットに登録 */
	RegisterBackgroundWorker(&worker);
}


/*
 * ワーカのエントリポイント
 */
void myworker_main(Datum main_arg)
{
    /* ワーカの処理 */
}

サンプル 動的バックグラウンドワーカの登録

*handleには、pallocBackgroundWorkerHandle構造体のメモリ領域が確保される。メモリコンテキストは、特に指定されていないので、CurrentMemoryContextから割り当てられるのに注意。

static void
RegisterDynamicWorker()
{
	BackgroundWorker worker;

	/* 
	 * Handle for worker.
	 * You can wait for the worker to shutdown with this handle.
	 */
	BackgroundWorkerHandle *handle;

	memset(&worker, 0, sizeof(worker));
	worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
	worker.bgw_start_time = BgWorkerStart_ConsistentState;
	worker.bgw_restart_time = BGW_NEVER_RESTART;
	sprintf(worker.bgw_library_name, "myworker");
	sprintf(worker.bgw_function_name, "myworker_main");
	snprintf(worker.bgw_name, BGW_MAXLEN, "myworker");
	worker.bgw_main_arg = Int32GetDatum(MyProcPid);
	/* Send SIGUSR1 at end */
	worker.bgw_notify_pid = MyProcPid;

	/*
	 * Register a worker.
	 * The worker is boot via postmaster.
	 */
	if (!RegisterDynamicBackgroundWorker(&worker, &handle))
		ereport(WARNING,
				(errmsg("could not register background worker")));
}

バックグラウンドワーカの終了

静的バックグラウンドワーカ

postmasterの停止でワーカも停止する。
ワーカのみ停止したい場合は、シグナルを送ることで可能(ブロックや停止拒否していない場合)。

動的バックグラウンドワーカ

以下説明のため、バックグラウンドワーカ(ワーカ)を登録するプロセスを便宜上オーナと呼ぶ。

オーナが終了

ワーカはオーナの終了を知ることはない。BackgroundWorker 構造体のメンバ変数bgw_notify_pid0以外を指定していた場合、オーナが終了するとBackgroundWorkerStopNotifications関数が呼ばれ、bgw_notify_pid0に設定される。従って、ワーカでこの変数をチェックすればオーナの終了を知ることができる。

参考BackgroundWorkerStopNotifications() - https://git.postgresql.org/gitweb/

また、別の方法としてon_proc_exitによる終了ハンドラを利用するという方法も考えられる。通常、ワーカはループ内でLatchによるWaitを使って何らかの処理を繰り返し実行するような実装になると思うが、オーナありきのワーカの場合、オーナの終了をできるだけ早期に検知し、なおかつ自身も停止した方がワーカスロットの無駄な占有もしなくて済むはず。

例えば以下のような感じである。

void
RegisterDynamicWorker(void)
{
	/* Register a background worker */

	/* Set a cleanup handler for the worker */
	on_proc_exit(worker_cleanup, 0);
}

static void
worker_cleanup(int code, Datum arg)
{
	pid_t pid;
	BgwHandleStatus status;

	if (worker_handle)
	{
		/*
		 * If the worker has started, kill it.
		 */
		status = GetBackgroundWorkerPid(worker_handle, &pid);
		if (status != BGWH_STOPPED)
		{
			kill(pid, SIGTERM);
			return;
		}

		/*
		 * We wait for the worker to startup.
		 */
		status = WaitForBackgroundWorkerStartup(worker_handle, &pid);
		if (status == BGWH_STARTED)
			kill(pid, SIGTERM); /* Other approach: TerminateBackgroundWorker(worker_handle) */
	}

	return;
}

ワーカの終了を待つ

バッググラウンドワーカが何らかの処理を行ない、呼び出し側でその終了を待つ場合は以下のようになる。

BgwHandleStatus status;
BackgroundWorker worker;
BackgroundWorkerHandle *handle;

/* Setup struct BackgroundWorker */

/*
 * Register a worker.
 * The worker is boot via postmaster.
 */
if (!RegisterDynamicBackgroundWorker(&worker, &handle))
{
	ereport(WARNING,
			(errmsg("could not register background worker")));
	return;
}
status = WaitForBackgroundWorkerShutdown(handle);
if (status == BGWH_POSTMASTER_DIED)
	proc_exit(1);

/* Do something */

ワーカの開始・終了通知(SIGUSR1)を受けとる

バックグラウンドワーカの登録時、ワーカを登録するプロセス(ここでは、オーナと呼ぶ)は、BackgroundWorker構造体変数のbgw_notify_pidに自身のプロセスIDを指定すると、ワーカ開始・終了の通知をシグナルSIGUSR1で受け取ることができる。

サンプル オーナでワーカの開始・終了通知を受け取る

static volatile sig_atomic_t got_sigusr1 = false;
static volatile sig_atomic_t got_sigterm = false;

static void
myworker_sigusr1(SIGNAL_ARGS)
{
	int	save_errno = errno;

	got_sigusr1 = true;

	procsignal_sigusr1_handler(postgres_signal_arg);

	errno = save_errno;
}

static void
myworker_sigterm(SIGNAL_ARGS)
{
	int	save_errno = errno;

	got_sigterm = true;

	SetLatch(MyLatch);

	errno = save_errno;
}

void myworker_main(Datum main_arg)
{
	pqsignal(SIGUSR1, myworker_sigusr1);
        pqsignal(SIGTERM, myworker_sigterm);

        /* Now we can receive signals */
	BackgroundWorkerUnblockSignals();

	while (!git_sigterm)
	{
		/* Here Wait Latch */

		if (got_sigusr1)
		{
			got_sigusr1 = false;
			/* Do something */
		}

		/*
		 * Do something
		 */
	}

	proc_exit(0);
}

参考リンク


トップ   一覧 単語検索 最終更新   ヘルプ   最終更新のRSS
目次
ダブルクリックで閉じるTOP | 閉じる
GO TO TOP