Thunk.parallel_count

author: Robert Haas 2015-10-15 21:45:22 +0000
committer: Robert Haas 2015-10-15 21:45:22 +0000
commit: 91945e758095f496642464652bedb22905647501 (patch)
tree: ab42ee07af16bf9a7c984c9560b46f96c68b81ae /contrib
parent: 08fbad0afd62690cc82990c0504529ef238ac24d (diff)
4 files changed, 167 insertions, 0 deletions
diff --git a/contrib/parallel_dummy/Makefile b/contrib/parallel_dummy/Makefile
new file mode 100644
index 0000000000..de00f50381
--- /dev/null
+++ b/contrib/parallel_dummy/Makefile
@@ -0,0 +1,19 @@
+MODULE_big = parallel_dummy
+OBJS = parallel_dummy.o $(WIN32RES)
+PGFILEDESC = "parallel_dummy - dummy use of parallel infrastructure"
+
+EXTENSION = parallel_dummy
+DATA = parallel_dummy--1.0.sql
+
+REGRESS = parallel_dummy
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/parallel_dummy
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/parallel_dummy/parallel_dummy--1.0.sql b/contrib/parallel_dummy/parallel_dummy--1.0.sql
new file mode 100644
index 0000000000..d49bd0fab0
--- /dev/null
+++ b/contrib/parallel_dummy/parallel_dummy--1.0.sql
@@ -0,0 +1,7 @@
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION parallel_dummy" to load this file. \quit
+
+CREATE FUNCTION parallel_count(rel pg_catalog.regclass,
+							  nworkers pg_catalog.int4)
+    RETURNS pg_catalog.int8 STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
diff --git a/contrib/parallel_dummy/parallel_dummy.c b/contrib/parallel_dummy/parallel_dummy.c
new file mode 100644
index 0000000000..84c863cae8
--- /dev/null
+++ b/contrib/parallel_dummy/parallel_dummy.c
@@ -0,0 +1,137 @@
+/*--------------------------------------------------------------------------
+ *
+ * parallel_dummy.c
+ *		Test harness code for parallel mode code.
+ *
+ * Copyright (C) 2013-2014, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/parallel_dummy/parallel_dummy.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/parallel.h"
+#include "access/relscan.h"
+#include "access/xact.h"
+#include "fmgr.h"
+#include "miscadmin.h"
+#include "storage/spin.h"
+#include "utils/builtins.h"
+#include "utils/guc.h"
+#include "utils/snapmgr.h"
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(parallel_count);
+
+#define		TOC_SCAN_KEY			1
+#define		TOC_RESULT_KEY			2
+
+void		_PG_init(void);
+void		count_worker_main(dsm_segment *seg, shm_toc *toc);
+
+static void count_helper(Relation rel, ParallelHeapScanDesc pscan,
+						 int64 *resultp);
+
+Datum
+parallel_count(PG_FUNCTION_ARGS)
+{
+	Oid			relid = PG_GETARG_OID(0);
+	int32		nworkers = PG_GETARG_INT32(1);
+	int64	   *resultp;
+	int64		result;
+	ParallelContext *pcxt;
+	ParallelHeapScanDesc pscan;
+	Relation	rel;
+	Size		sz;
+
+	if (nworkers < 0)
+		ereport(ERROR,
+				(errmsg("number of parallel workers must be non-negative")));
+
+	rel = relation_open(relid, AccessShareLock);
+
+	EnterParallelMode();
+
+	pcxt = CreateParallelContextForExternalFunction("parallel_dummy",
+											 "count_worker_main",
+											 nworkers);
+	sz = heap_parallelscan_estimate(GetActiveSnapshot());
+	shm_toc_estimate_chunk(&pcxt->estimator, sz);
+	shm_toc_estimate_chunk(&pcxt->estimator, sizeof(int64));
+	shm_toc_estimate_keys(&pcxt->estimator, 2);
+	InitializeParallelDSM(pcxt);
+	pscan = shm_toc_allocate(pcxt->toc, sz);
+	heap_parallelscan_initialize(pscan, rel, GetActiveSnapshot());
+	shm_toc_insert(pcxt->toc, TOC_SCAN_KEY, pscan);
+	resultp = shm_toc_allocate(pcxt->toc, sizeof(int64));
+	shm_toc_insert(pcxt->toc, TOC_RESULT_KEY, resultp);
+
+	LaunchParallelWorkers(pcxt);
+
+	/* here's where we do the "real work" ... */
+	count_helper(rel, pscan, resultp);
+
+	WaitForParallelWorkersToFinish(pcxt);
+
+	result = *resultp;
+
+	DestroyParallelContext(pcxt);
+
+	relation_close(rel, AccessShareLock);
+
+	ExitParallelMode();
+
+	PG_RETURN_INT64(result);
+}
+
+void
+count_worker_main(dsm_segment *seg, shm_toc *toc)
+{
+	ParallelHeapScanDesc	pscan;
+	Relation	rel;
+	int64	   *resultp;
+
+	pscan = shm_toc_lookup(toc, TOC_SCAN_KEY);
+	resultp = shm_toc_lookup(toc, TOC_RESULT_KEY);
+	Assert(pscan != NULL && resultp != NULL);
+
+	rel = relation_open(pscan->phs_relid, AccessShareLock);
+	count_helper(rel, pscan, resultp);
+	relation_close(rel, AccessShareLock);
+}
+
+static void
+count_helper(Relation rel, ParallelHeapScanDesc pscan, int64 *resultp)
+{
+	int64		mytuples = 0;
+	HeapScanDesc	scan;
+	BlockNumber	firstblock = InvalidBlockNumber;
+
+	scan = heap_beginscan_parallel(rel, pscan);
+
+	for (;;)
+	{
+		HeapTuple	tup = heap_getnext(scan, ForwardScanDirection);
+
+		if (!HeapTupleIsValid(tup))
+			break;
+		if (firstblock == InvalidBlockNumber)
+			firstblock = scan->rs_cblock;
+
+		++mytuples;
+	}
+
+	heap_endscan(scan);
+
+	SpinLockAcquire(&pscan->phs_mutex); /* dirty hack */
+	*resultp += mytuples;
+	SpinLockRelease(&pscan->phs_mutex);
+
+	elog(NOTICE, "PID %d counted " INT64_FORMAT " tuples starting at block %u",
+		MyProcPid, mytuples, firstblock);
+}
diff --git a/contrib/parallel_dummy/parallel_dummy.control b/contrib/parallel_dummy/parallel_dummy.control
new file mode 100644
index 0000000000..90bae3f6e0
--- /dev/null
+++ b/contrib/parallel_dummy/parallel_dummy.control
@@ -0,0 +1,4 @@
+comment = 'Dummy parallel code'
+default_version = '1.0'
+module_pathname = '$libdir/parallel_dummy'
+relocatable = true
author	Robert Haas	2015-10-15 21:45:22 +0000
committer	Robert Haas	2015-10-15 21:45:22 +0000
commit	91945e758095f496642464652bedb22905647501 (patch)
tree	ab42ee07af16bf9a7c984c9560b46f96c68b81ae /contrib
parent	08fbad0afd62690cc82990c0504529ef238ac24d (diff)