Function get_shareinput_reference

src/backend/executor/nodeShareInputScan.c:800–856

* Get a reference to slot in shared memory for this shared scan. * * If the slot doesn't exist yet, it is created and initialized into * "not ready" state. * * The reference is tracked by the current ResourceOwner, and will be * automatically released on abort. */

 * automatically released on abort.
 */
static shareinput_Xslice_reference *
get_shareinput_reference(int share_id)
{
	shareinput_tag tag;
	shareinput_Xslice_state *xslice_state;
	bool		found;
	shareinput_Xslice_reference *ref;

	/* Register our resource owner callback to clean up on first call. */
	if (!shareinput_resowner_callback_registered)
	{
		RegisterResourceReleaseCallback(shareinput_release_callback, NULL);
		shareinput_resowner_callback_registered = true;
	}

	ref = MemoryContextAllocZero(TopMemoryContext,
								 sizeof(shareinput_Xslice_reference));

	LWLockAcquire(ShareInputScanLock, LW_EXCLUSIVE);

	tag.session_id = gp_session_id;
	tag.command_count = gp_command_count;
	tag.share_id = share_id;
	xslice_state = hash_search(shareinput_Xslice_hash,
							   &tag,
							   HASH_ENTER_NULL,
							   &found);
	if (!found)
	{
		if (xslice_state == NULL)
		{
			pfree(ref);
			ereport(ERROR,
					(errcode(ERRCODE_OUT_OF_MEMORY),
					 errmsg("out of cross-slice ShareInputScan slots")));
		}

		xslice_state->refcount = 0;
		pg_atomic_init_u32(&xslice_state->ready, 0);
		pg_atomic_init_u32(&xslice_state->ndone, 0);

		ConditionVariableInit(&xslice_state->ready_done_cv);
		elog((Debug_shareinput_xslice ? LOG : DEBUG1), "SISC (shareid=%d, slice=%d): initialized xslice state",
			 share_id, currentSliceId);
	}

	xslice_state->refcount++;

	ref->share_id = share_id;
	ref->xslice_state = xslice_state;
	ref->owner = CurrentResourceOwner;
	dlist_push_head(&shareinput_Xslice_refs, &ref->node);

	LWLockRelease(ShareInputScanLock);

	return ref;
}