diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index cd21e0e..a8c5ec5 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -16,14 +16,20 @@ #include "access/gin_private.h" #include "access/xloginsert.h" +#include "access/parallel.h" +#include "access/xact.h" #include "catalog/index.h" #include "miscadmin.h" #include "storage/bufmgr.h" #include "storage/smgr.h" #include "storage/indexfsm.h" +#include "storage/spin.h" +#include "utils/datum.h" #include "utils/memutils.h" #include "utils/rel.h" +/* GUC parameter */ +int gin_parallel_workers = 0; typedef struct { @@ -265,6 +271,141 @@ ginHeapTupleBulkInsert(GinBuildState *buildstate, OffsetNumber attnum, MemoryContextReset(buildstate->funcCtx); } +#define KEY_TASK 42 +#define GIN_MAX_WORKERS 24 +#define GIN_BLOCKS_PER_WORKER 4 +#define GIN_RESULT_LEN 1024 +#define GIN_RESULT_KEYLEN 1024 + +typedef struct { + bool ready; + bool finished; + + Datum key; + OffsetNumber attnum; + GinNullCategory category; + + char keycoded[GIN_RESULT_KEYLEN]; + int keylen; + + ItemPointerData list[GIN_RESULT_LEN]; + int nlist; + + Latch blatch; + Latch wlatch; +} WorkerResult; + +/* + * This structure describes the GIN build task for the parallel workers. We use + * OIDs here because workers are separate processes and pointers may become + * meaningless for them. The "lock" is used to protect the "scanned" and + * "reltuples" fields as the workers modify them. + */ +typedef struct { + int to_scan; + int scanned; + slock_t lock; + Oid heap_oid; + Oid index_oid; + double reltuples; + WorkerResult results[GIN_MAX_WORKERS]; +} PGinBuildTask; + +static volatile PGinBuildTask *task; + +static void waitBool(volatile bool *actual, bool wanted, volatile Latch *l) +{ + if (*actual == wanted) return; + + while (*actual != wanted) + WaitLatch(l, WL_LATCH_SET, 0); + ResetLatch(l); +} + +static void setBool(volatile bool *actual, bool newvalue, volatile Latch *l) +{ + *actual = newvalue; + SetLatch(l); +} + +static void ginDumpEntry(GinState *ginstate, + volatile WorkerResult *r, + OffsetNumber attnum, + Datum key, + GinNullCategory category, + ItemPointerData *list, + int nlist) +{ + volatile char *addr; + bool isnull; + Form_pg_attribute att; + + Assert(nlist > 0); + waitBool(&r->ready, false, &r->wlatch); + + Assert(r->keylen == 0); + addr = r->keycoded; + isnull = category == GIN_CAT_NULL_KEY; + att = ginstate->origTupdesc->attrs[attnum - 1]; + + r->attnum = attnum; + r->category = category; + r->keylen = datumEstimateSpace(key, isnull, att->attbyval, att->attlen); + Assert(r->keylen > 0); + Assert(r->keylen <= GIN_RESULT_KEYLEN); + + datumSerialize(key, isnull, att->attbyval, att->attlen, (char**)&addr); + + while (nlist > 0) + { + if (nlist > GIN_RESULT_LEN) + r->nlist = GIN_RESULT_LEN; + else + r->nlist = nlist; + nlist -= r->nlist; + + memcpy((void*)r->list, list, r->nlist * sizeof(ItemPointerData)); + setBool(&r->ready, true, &r->blatch); + waitBool(&r->ready, false, &r->wlatch); + } +} + +static void ginDumpAccumulator(GinBuildState *buildstate) +{ + ItemPointerData *list; + Datum key; + GinNullCategory category; + uint32 nlist; + OffsetNumber attnum; + MemoryContext oldCtx; + + oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx); + + ginBeginBAScan(&buildstate->accum); + while ((list = ginGetBAEntry(&buildstate->accum, + &attnum, &key, &category, &nlist)) != NULL) + { + /* there could be many entries, so be willing to abort here */ + CHECK_FOR_INTERRUPTS(); + + if (IsParallelWorker()) + { + volatile WorkerResult *r = &task->results[ParallelWorkerNumber]; + ginDumpEntry(&buildstate->ginstate, r, attnum, key, category, list, nlist); + } + else + ginEntryInsert(&buildstate->ginstate, + attnum, key, category, + list, nlist, + &buildstate->buildStats); + } + + MemoryContextReset(buildstate->tmpCtx); + ginInitBA(&buildstate->accum); + + MemoryContextSwitchTo(oldCtx); +} + static void ginBuildCallback(Relation index, HeapTuple htup, Datum *values, bool *isnull, bool tupleIsAlive, void *state) @@ -283,52 +424,307 @@ ginBuildCallback(Relation index, HeapTuple htup, Datum *values, /* If we've maxed out our available memory, dump everything to the index */ if (buildstate->accum.allocatedMemory >= (Size)maintenance_work_mem * 1024L) { - ItemPointerData *list; - Datum key; - GinNullCategory category; - uint32 nlist; - OffsetNumber attnum; - - ginBeginBAScan(&buildstate->accum); - while ((list = ginGetBAEntry(&buildstate->accum, - &attnum, &key, &category, &nlist)) != NULL) - { - /* there could be many entries, so be willing to abort here */ - CHECK_FOR_INTERRUPTS(); - ginEntryInsert(&buildstate->ginstate, attnum, key, category, - list, nlist, &buildstate->buildStats); - } - - MemoryContextReset(buildstate->tmpCtx); - ginInitBA(&buildstate->accum); + ginDumpAccumulator(buildstate); } MemoryContextSwitchTo(oldCtx); } +/* + * Get the next key from the specified worker. Wait until it is available or + * the result is exhausted. Return true if got a key, false if the result is + * exhausted. Fill everything in, except "list". + */ +static bool getKeyFromWorker(volatile WorkerResult *result) +{ + if (result->finished) return false; + + if (result->keylen) + { + bool isnull; + volatile char *addr = result->keycoded; + result->key = datumRestore((char**)&addr, &isnull); + if (isnull) + Assert(result->category == GIN_CAT_NULL_KEY); + else + Assert(result->category == GIN_CAT_NORM_KEY); + Assert(result->nlist > 0); + result->keylen = 0; + } + return true; +} + +/* + * Claim "max_blocks" or less blocks. Return the actual number of claimed + * blocks and set "first" to point to the first block of the claimed range. + * 0 return value means the task has been finished. + */ +static int claimSomeBlocks(volatile PGinBuildTask *task, int max_blocks, int *first) +{ + int blocks = 0; + + SpinLockAcquire(&task->lock); + + if (task->scanned >= task->to_scan) + { + SpinLockRelease(&task->lock); + return 0; + } + + *first = task->scanned; + blocks = max_blocks; + if (blocks > task->to_scan - task->scanned) + blocks = task->to_scan - task->scanned; + task->scanned += blocks; + + SpinLockRelease(&task->lock); + return blocks; +} + +static void reportReltuples(volatile PGinBuildTask *task, double reltuples) +{ + SpinLockAcquire(&task->lock); + task->reltuples += reltuples; + SpinLockRelease(&task->lock); +} + +static double +ginbuildCommon(GinBuildState *buildstate, Relation heap, Relation index, IndexInfo *indexInfo) +{ + double reltuples = 0; + + /* + * create a temporary memory context that is used to hold data not yet + * dumped out to the index + */ + buildstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext, + "Gin build temporary context", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + + /* + * create a temporary memory context that is used for calling + * ginExtractEntries(), and can be reset after each tuple + */ + buildstate->funcCtx = AllocSetContextCreate(CurrentMemoryContext, + "Gin build temporary context for user-defined function", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + + buildstate->accum.ginstate = &buildstate->ginstate; + ginInitBA(&buildstate->accum); + + /* + * Do the heap scan. We disallow sync scan here because dataPlaceToPage + * prefers to receive tuples in TID order. + */ + if (IsParallelWorker()) + { + while (true) + { + double subtuples; + int first, blocks; + + blocks = claimSomeBlocks(task, GIN_BLOCKS_PER_WORKER, &first); + if (blocks == 0) + break; + + subtuples = IndexBuildHeapRangeScan(heap, index, indexInfo, false, false, + first, blocks, + ginBuildCallback, (void *)buildstate); + reltuples += subtuples; + } + } + else + { + reltuples = IndexBuildHeapScan(heap, index, indexInfo, false, + ginBuildCallback, (void *)buildstate); + } + + /* dump remaining entries to the index */ + ginDumpAccumulator(buildstate); + + MemoryContextDelete(buildstate->funcCtx); + MemoryContextDelete(buildstate->tmpCtx); + + /* + * Update metapage stats + */ + buildstate->buildStats.nTotalPages = RelationGetNumberOfBlocks(index); + ginUpdateStats(index, &buildstate->buildStats); + + return reltuples; +} + +/* + * The idea behind parallel GIN build is letting the parallel workers scan the + * relation and build rbtree in parallel. Each block is scanned by one of the + * workers. + */ +static void ginbuildWorker(dsm_segment *seg, shm_toc *toc) +{ + GinBuildState buildstate; + + Relation heap; + Relation index; + IndexInfo *indexInfo; + + volatile WorkerResult *r; + double reltuples; + + task = (PGinBuildTask*)shm_toc_lookup(toc, KEY_TASK); + r = &task->results[ParallelWorkerNumber]; + r->finished = false; + + OwnLatch(&r->wlatch); + + heap = heap_open(task->heap_oid, NoLock); + index = index_open(task->index_oid, NoLock); + indexInfo = BuildIndexInfo(index); + + initGinState(&buildstate.ginstate, index); + buildstate.indtuples = 0; + memset(&buildstate.buildStats, 0, sizeof(GinStatsData)); + + reltuples = ginbuildCommon(&buildstate, heap, index, indexInfo); + + index_close(index, NoLock); + heap_close(heap, NoLock); + + reportReltuples(task, reltuples); + + waitBool(&r->ready, false, &r->wlatch); + r->finished = true; + setBool(&r->ready, true, &r->blatch); +} + +typedef struct GinEntryStack { + struct GinEntryStack *next; + Datum key; + GinNullCategory category; + OffsetNumber attnum; + ItemPointerData *list; + int nlist; +} GinEntryStack; + +static GinEntryStack *pushEntry(GinEntryStack *stack) +{ + GinEntryStack *head = palloc(sizeof(GinEntryStack)); + head->next = stack; + head->list = palloc(sizeof(ItemPointerData)); /* make ginMergeItemPointers happy */ + head->nlist = 0; + return head; +} + +static GinEntryStack *popEntry(GinEntryStack *stack) +{ + GinEntryStack *head = stack; + Assert(stack != NULL); + stack = stack->next; + pfree(head->list); + pfree(head); + return stack; +} + +static void mergeResults(GinBuildState *buildstate, ParallelContext *pcxt, volatile PGinBuildTask *task) +{ + GinEntryStack *entry = NULL; + + while (true) + { + bool merged = false; + int i; + + for (i = 0; i < pcxt->nworkers; ++i) + { + ItemPointerData *oldlist; + bool have_key; + int cmp; + volatile WorkerResult *r = &task->results[i]; + if (pcxt->worker[i].error_mqh == NULL) continue; + waitBool(&r->ready, true, &r->blatch); + if (r->finished) continue; + + /* The worker has something for us. */ + + have_key = getKeyFromWorker(r); + Assert(have_key); + + cmp = -1; + if (entry != NULL) + { + cmp = ginCompareAttEntries(&buildstate->ginstate, + r->attnum, r->key, r->category, + entry->attnum, entry->key, entry->category); + } + + if (cmp > 0) + { + /* The key is greater, skip the worker. */ + continue; + } + + if (cmp < 0) + { + /* + * The key is less than what we have on the stack. + * Push a new entry onto the stack. + */ + entry = pushEntry(entry); + entry->key = r->key; + entry->category = r->category; + entry->attnum = r->attnum; + } + + /* + * The key is less than or equal. Merge the item pointers. + * FIXME: Should we first copy the list and let the worker continue + * before merging? + */ + oldlist = entry->list; + entry->list = ginMergeItemPointers(entry->list, entry->nlist, + (ItemPointerData*)r->list, r->nlist, + &entry->nlist); + pfree(oldlist); + setBool(&r->ready, false, &r->wlatch); + merged = true; + } + + if (!merged) + { + /* Nothing merged. Insert the entry into the index and pop the stack. */ + if (entry == NULL) + { + /* Also nothing to dump - we have finished. */ + break; + } + + ginEntryInsert(&buildstate->ginstate, + entry->attnum, entry->key, entry->category, + entry->list, entry->nlist, + &buildstate->buildStats); + entry = popEntry(entry); + } + } +} + IndexBuildResult * ginbuild(Relation heap, Relation index, IndexInfo *indexInfo) { + GinBuildState buildstate; + IndexBuildResult *result; - double reltuples; - GinBuildState buildstate; Buffer RootBuffer, MetaBuffer; - ItemPointerData *list; - Datum key; - GinNullCategory category; - uint32 nlist; - MemoryContext oldCtx; - OffsetNumber attnum; + double reltuples = 0; + bool parallel_workers_helped = false; if (RelationGetNumberOfBlocks(index) != 0) elog(ERROR, "index \"%s\" already contains data", RelationGetRelationName(index)); - initGinState(&buildstate.ginstate, index); - buildstate.indtuples = 0; - memset(&buildstate.buildStats, 0, sizeof(GinStatsData)); - /* initialize the meta page */ MetaBuffer = GinNewBuffer(index); @@ -363,60 +759,76 @@ ginbuild(Relation heap, Relation index, IndexInfo *indexInfo) UnlockReleaseBuffer(RootBuffer); END_CRIT_SECTION(); + initGinState(&buildstate.ginstate, index); + buildstate.indtuples = 0; + memset(&buildstate.buildStats, 0, sizeof(GinStatsData)); + /* count the root as first entry page */ buildstate.buildStats.nEntryPages++; - /* - * create a temporary memory context that is used to hold data not yet - * dumped out to the index - */ - buildstate.tmpCtx = AllocSetContextCreate(CurrentMemoryContext, - "Gin build temporary context", - ALLOCSET_DEFAULT_MINSIZE, - ALLOCSET_DEFAULT_INITSIZE, - ALLOCSET_DEFAULT_MAXSIZE); - - /* - * create a temporary memory context that is used for calling - * ginExtractEntries(), and can be reset after each tuple - */ - buildstate.funcCtx = AllocSetContextCreate(CurrentMemoryContext, - "Gin build temporary context for user-defined function", - ALLOCSET_DEFAULT_MINSIZE, - ALLOCSET_DEFAULT_INITSIZE, - ALLOCSET_DEFAULT_MAXSIZE); - - buildstate.accum.ginstate = &buildstate.ginstate; - ginInitBA(&buildstate.accum); - - /* - * Do the heap scan. We disallow sync scan here because dataPlaceToPage - * prefers to receive tuples in TID order. - */ - reltuples = IndexBuildHeapScan(heap, index, indexInfo, false, - ginBuildCallback, (void *) &buildstate); - - /* dump remaining entries to the index */ - oldCtx = MemoryContextSwitchTo(buildstate.tmpCtx); - ginBeginBAScan(&buildstate.accum); - while ((list = ginGetBAEntry(&buildstate.accum, - &attnum, &key, &category, &nlist)) != NULL) + if (gin_parallel_workers > GIN_MAX_WORKERS) + gin_parallel_workers = GIN_MAX_WORKERS; + + if (gin_parallel_workers > 0) { - /* there could be many entries, so be willing to abort here */ - CHECK_FOR_INTERRUPTS(); - ginEntryInsert(&buildstate.ginstate, attnum, key, category, - list, nlist, &buildstate.buildStats); - } - MemoryContextSwitchTo(oldCtx); + if (RelationUsesLocalBuffers(heap)) + { + fprintf(stderr, "not using parallel GIN build on temporary table %s\n", NameStr(heap->rd_rel->relname)); + } + else + { + EnterParallelMode(); + { + int i; + int size = sizeof(PGinBuildTask); + int keys = 1; + PGinBuildTask *task; + ParallelContext *pcxt; + + pcxt = CreateParallelContext(ginbuildWorker, gin_parallel_workers); + + shm_toc_estimate_chunk(&pcxt->estimator, size); + shm_toc_estimate_keys(&pcxt->estimator, keys); + InitializeParallelDSM(pcxt); + task = (PGinBuildTask*)shm_toc_allocate(pcxt->toc, size); + shm_toc_insert(pcxt->toc, KEY_TASK, (void*)task); + SpinLockInit(&task->lock); + for (i = 0; i < pcxt->nworkers; i++) + { + volatile WorkerResult *r = &task->results[i]; + + InitSharedLatch(&r->blatch); + OwnLatch(&r->blatch); + InitSharedLatch(&r->wlatch); - MemoryContextDelete(buildstate.funcCtx); - MemoryContextDelete(buildstate.tmpCtx); + r->keylen = 0; + r->ready = false; + } + task->reltuples = 0; + task->to_scan = RelationGetNumberOfBlocks(heap); + task->heap_oid = heap->rd_id; + task->index_oid = index->rd_id; + LaunchParallelWorkers(pcxt); + if (pcxt->nworkers_launched > 0) + { + mergeResults(&buildstate, pcxt, task); - /* - * Update metapage stats - */ - buildstate.buildStats.nTotalPages = RelationGetNumberOfBlocks(index); - ginUpdateStats(index, &buildstate.buildStats); + WaitForParallelWorkersToFinish(pcxt); + reltuples = task->reltuples; + + parallel_workers_helped = true; + } + DestroyParallelContext(pcxt); + } + ExitParallelMode(); + } + } + + if (!parallel_workers_helped) + { + /* Do everything myself */ + reltuples = ginbuildCommon(&buildstate, heap, index, indexInfo); + } /* * Return statistics diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 31a69ca..b26e362 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -2697,6 +2697,19 @@ static struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"gin_parallel_workers", + PGC_USERSET, + RESOURCES_ASYNCHRONOUS, + gettext_noop("Maximum number of parallel workers for GIN buiding."), + NULL, + }, + &gin_parallel_workers, + 0, 0, MAX_BACKENDS, + NULL, NULL, NULL + }, + + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 09b2003..70cb37a 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -537,6 +537,7 @@ #xmloption = 'content' #gin_fuzzy_search_limit = 0 #gin_pending_list_limit = 4MB +#gin_parallel_workers = 0 # 0 disables parallel gin build # - Locale and Formatting - diff --git a/src/include/access/gin.h b/src/include/access/gin.h index e5b2e10..c487677 100644 --- a/src/include/access/gin.h +++ b/src/include/access/gin.h @@ -68,6 +68,7 @@ typedef char GinTernaryValue; /* GUC parameters */ extern PGDLLIMPORT int GinFuzzySearchLimit; extern int gin_pending_list_limit; +extern int gin_parallel_workers; /* ginutil.c */ extern void ginGetStats(Relation index, GinStatsData *stats);