Function GpInsertParallelDSMHash

src/backend/access/transam/parallel.c:1747–1890

}

GpParallelDSMEntry *
GpInsertParallelDSMHash(PlanState *planstate)
{
	GpParallelDSMEntry *entry;
	bool found = false;
	static bool init = false;

	int localSliceId = LocallyExecutingSliceIndex(planstate->state);
	int parallel_workers = 0;

	if (planstate->state->es_plannedstmt && planstate->state->es_plannedstmt->slices)
	{
		parallel_workers = planstate->state->es_plannedstmt->slices[localSliceId].parallel_workers;
	}

	if (parallel_workers <= 1)
		return NULL;

	ParallelEntryTag tag;
	INIT_PARALLELENTRYTAG(tag, gp_command_count, localSliceId, gp_session_id);

	LWLockAcquire(GpParallelDSMHashLock, LW_EXCLUSIVE);

	entry = (GpParallelDSMEntry *)
		hash_search(GpParallelDSMHash,
					&tag,
					HASH_ENTER_NULL,
					&found);

	if (!entry)
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("out of shared memory"),
				 errhint("out of cross-slice SHARED_PARALLEL_DSM_TABLE_SIZE slots.")));

	if (!found)
	{
		shm_toc_estimator *estimator = NULL;
		ParallelContext context = {
			.nworkers = parallel_workers,
		};

		Size		dsa_minsize = dsa_minimum_size();
		estimator = &context.estimator;
		shm_toc_initialize_estimator(estimator);

		/* Estimate space for parallel DSA area. */
		shm_toc_estimate_chunk(estimator, dsa_minsize);
		shm_toc_estimate_keys(estimator, 1);

		EstimateGpParallelDSMEntrySize(planstate, &context);

		Size segsize = shm_toc_estimate(estimator);

		shm_toc    *toc;
		dsm_segment* seg = dsm_create(segsize, DSM_CREATE_NULL_IF_MAXSEGMENTS);

		if (seg != NULL)