PostgreSQL（简称PG）从9.6【1】开始支持并行查询，并行查询后面离不开PG的Parallel Worker编程模型，按照PG的多进程模型写一个多进程并行的任务非常容易，笔者团队在PG 11.3的基础上开发的向量检索，支持多进程加速构建向量索引，但是PG的多进程编程与我们熟知的多线程编程还是有较大的差别，并且网络上缺少这方便的资料，笔者在这个上面踩了一些坑，所以这里首先与大家简单分享PG的多进程编程模型，然后以此为入口分析PG的多进程通信，最后总结一下笔者踩的几个坑。

1 PG的多进程编程模型

1) 首先定义一个多进程间共享的数据结构

typedef struct ParallelXXXShared
{
	/* immutable arguments */
	Oid 		oid;
	int 		nworkers;
	int			launched_workers;

	/* mutable arguments */
	int 		data;
	slock_t		mutex;
} ParallelXXXShared;

这个数据结构分成2个部分，一个是工作进程不能修改的，另外一个是工作进程需要修改的部分，其中mutex是为了在多进程对data做修改的时候做加锁保护。

2）定义工作进程的工作函数

函数原型如下：

void 
ParallelJobMain(dsm_segment *seg, shm_toc *toc)
{
}

其中dsm_segment和toc都是和共享内存相关的数据，在后面详细讲解，这里我们只需要知道主进程和工作进程共享这块数据。

3）初始化并行任务，启动，并等待任务结束

        /* prohibit unsafe state changes */
	EnterParallelMode();		
	
        /* 在这里设定工程进程的工作函数ParallelJobMain */  
	pcxt = CreateParallelContext("fastann", "ParallelJobMain", nworkers, true);
	
        /* Estimate space for fixed-stimator, 1);
        shm_toc_estimate_chunk(&pcxt->estimator,
				sizeof(ParallelXXXShared));
	shm_toc_estimate_keys(&pcxt->estimator, 1);

        /* create DSM and copy state to it */
	InitializeParallelDSM(pcxt);	

	/* Store the data for which we reserved space. */
	pshared = shm_toc_allocate(pcxt->toc, sizeof(ParallelXXXShared));
	pshared->oid = oid;
	pshared->nworkers = nworkers;
	pshared->launched_workers = 0;
	pshared->data = 0;
	shm_toc_insert(pcxt->toc, 1, pshared);

	LaunchParallelWorkers(pcxt);

	/* do parallel stuff */

	WaitForParallelWorkersToFinish(pcxt);

	/* read any final results from dynamic shared memory */

	DestroyParallelContext(pcxt);
	ExitParallelMode();

InitializeParallelDSM中会初始化一段共享内存，用来保存进程共享的数据，这块共享内存并不是从shared_memory中分配而来，是额外申请的，多进程任务结束就会释放。但是它与shared_memory中的内存也一样也是固定大小的。所以这里有一个estimator用来计算每个部分所需要的内存大小，然后再分配。参考InitilizeParallelDSM中的代码：

         /*
	 * Create DSM and initialize with new table of contents.  But if the user
	 * didn't request any workers, then don't bother creating a dynamic shared
	 * memory segment; instead, just use backend-private memory.
	 *
	 * Also, if we can't create a dynamic shared memory segment because the
	 * maximum number of segments have already been created, then fall back to
	 * backend-private memory, and plan not to use any workers.  We hope this
	 * won't happen very often, but it's better to abandon the use of
	 * parallelism than to fail outright.
	 */
	segsize = shm_toc_estimate(&pcxt->estimator);
	if (pcxt->nworkers > 0)
		pcxt->seg = dsm_create(segsize, DSM_CREATE_NULL_IF_MAXSEGMENTS);
	if (pcxt->seg != NULL)
		pcxt->toc = shm_toc_create(PARALLEL_MAGIC,
								   dsm_segment_address(pcxt->seg),
								   segsize);

shm_segment初始化的时候，会同时初始化一个shm_toc，shm_toc也是一块连续的内存，里面保存的数据，以是key-value形式保存，key是uint64，value是内存内存偏移。

struct shm_toc
{
	uint64		toc_magic;		/* Magic number identifying this TOC */
	slock_t		toc_mutex;		/* Spinlock for mutual exclusion */
	Size		toc_total_bytes;	/* Bytes managed by this TOC */
	Size		toc_allocated_bytes;	/* Bytes allocated of those managed */
	uint32		toc_nentry;		/* Number of entries in TOC */
	shm_toc_entry toc_entry[FLEXIBLE_ARRAY_MEMBER];
};

在shm_toc初始化的时候，toc_total_bytes就被设定好了。

shm_toc管理的内存结构如下：

shm_toc一开始会申请一块固定大小的内存，shm_toc数据结构的末尾是一个shm_entry的动态数组，当往shm_toc插入数据的时候，会从内存的末尾开始插入，上图的toc entry data，并在shm_entry数组后面增一个entry，把key-value，即uint64的key和内存偏移写入 toc_entry。

需要注意的是，在PG的多进程并行编程里面，初始化的时候除了用户自定义的ParallelXXXShared，还有guc，snapshot等，这些系统已经占用的key，用户不能够使用：

		/* Serialize shared libraries we have loaded. */
		libraryspace = shm_toc_allocate(pcxt->toc, library_len);
		SerializeLibraryState(library_len, libraryspace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_LIBRARY, libraryspace);

		/* Serialize GUC settings. */
		gucspace = shm_toc_allocate(pcxt->toc, guc_len);
		SerializeGUCState(guc_len, gucspace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_GUC, gucspace);

		/* Serialize combo CID state. */
		combocidspace = shm_toc_allocate(pcxt->toc, combocidlen);
		SerializeComboCIDState(combocidlen, combocidspace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_COMBO_CID, combocidspace);

		/* Serialize transaction snapshot and active snapshot. */
		tsnapspace = shm_toc_allocate(pcxt->toc, tsnaplen);
		SerializeSnapshot(transaction_snapshot, tsnapspace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT,
					   tsnapspace);
		asnapspace = shm_toc_allocate(pcxt->toc, asnaplen);
		SerializeSnapshot(active_snapshot, asnapspace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, asnapspace);

		/* Provide the handle for per-session segment. */
		session_dsm_handle_space = shm_toc_allocate(pcxt->toc,
													sizeof(dsm_handle));
		*(dsm_handle *) session_dsm_handle_space = session_dsm_handle;
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_SESSION_DSM,
					   session_dsm_handle_space);

		/* Serialize transaction state. */
		tstatespace = shm_toc_allocate(pcxt->toc, tstatelen);
		SerializeTransactionState(tstatelen, tstatespace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_STATE, tstatespace);

		/* Serialize reindex state. */
		reindexspace = shm_toc_allocate(pcxt->toc, reindexlen);
		SerializeReindexState(reindexlen, reindexspace);
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_REINDEX_STATE, reindexspace);


		for (i = 0; i < pcxt->nworkers; ++i)
		{
			char	   *start;
			shm_mq	   *mq;

			start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
			mq = shm_mq_create(start, PARALLEL_ERROR_QUEUE_SIZE);
			shm_mq_set_receiver(mq, MyProc);
			pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
		}
		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, error_queue_space);

上述PARALLEL_KEY_ERROR_QUEUE对应的toc_entry是用来实现多进程通信，每个进程一个shm_mq。

2 PG的多进程通信

PG的多进程通信模型如下：

父进程：或者叫启动进程，由它启动工作进程，并管理工作进程。每个进程在初始化的时候，都会初始化一个MyLatch的数据结构，当本进程依赖其他进程的状态需要等待的时候，可以通过MyLatch进入等待状态，当其他进程完成对应的工作的时候，再把自己唤醒。MyLatch唤醒是通过向进程发送SIGUSR1信号的方式，所以父进程在初始化工作进程的时候，会把自己的进程号传递给工作进程。

工作进程：即运行并行任务的进程，会有1到多个。它在初始化的时候，会得到父进程的pid，并且会得到自己与父进程通信时的共享内存shn_mq。

dsm_segment：父进程和工作进程共享的内存。

shm_toc: shm_toc保存父进程与工作进程通信用到的数据，每个工作进程会有一个shm_mq用来保存出错信息。

子进程与父进程通信的流程：

父进程启动子进程后，等待子进程结束
子进程结束，把需要传递的消息写入自己对应的shm_mq
子进程向父进程发送SIGUSR1信号
父进程进程收到SIGUSR1信号，被唤醒
父进程读取shm_mq里面的消息

3 PG多进程编程容易踩的2个坑

启动的进程个数并不等同于设定的nworker

查看LaunchParallelWorkers代码，可以看到下列注释：

	/*
	 * Start workers.
	 *
	 * The caller must be able to tolerate ending up with fewer workers than
	 * expected, so there is no need to throw an error here if registration
	 * fails.  It wouldn't help much anyway, because registering the worker in
	 * no way guarantees that it will start up and initialize successfully.
	 */

由于进程通信使用信号，lwlock会block信号，当使用lwlock对父进程启动子进程加以保护的时候，会导致消息传递失败，进而导致父进程不知道子进程已经结束，父进程hang住。

查看函数调用栈：

LWLockAcquire(LWLock *lock, LWLockMode mode) ->HOLD_INTERRUPTS()

HOLD_INTERRUPTS的实现是把InterruptHoldoffCount加1.

而进程处理信号的代码ProcessInterrupts

void
ProcessInterrupts(void)
{
	/* OK to accept any interrupts now? */
	if (InterruptHoldoffCount != 0 || CritSectionCount != 0)
		return;
	InterruptPending = false;

当InterruptHoldoffCount不为0的时候，直接返回，导致没有调用HandleParallelMessages。

ITPub博客

PostgreSQL Parallel Woker编程模型

1 PG的多进程编程模型

2 PG的多进程通信

3 PG多进程编程容易踩的2个坑

4 相关资料

最新文章