diff options
| 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/parallel_dummy | |
| parent | 08fbad0afd62690cc82990c0504529ef238ac24d (diff) | |
Thunk.parallel_count
Diffstat (limited to 'contrib/parallel_dummy')
| -rw-r--r-- | contrib/parallel_dummy/Makefile | 19 | ||||
| -rw-r--r-- | contrib/parallel_dummy/parallel_dummy--1.0.sql | 7 | ||||
| -rw-r--r-- | contrib/parallel_dummy/parallel_dummy.c | 137 | ||||
| -rw-r--r-- | contrib/parallel_dummy/parallel_dummy.control | 4 |
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 |