diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index cd21e0e..ff267b3 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -16,14 +16,21 @@ #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; +int gin_shared_mem = 0; typedef struct { @@ -35,7 +42,6 @@ typedef struct BuildAccumulator accum; } GinBuildState; - /* * Adds array of item pointers to tuple's posting list, or * creates posting tree and tuple pointing to tree in case @@ -265,6 +271,169 @@ ginHeapTupleBulkInsert(GinBuildState *buildstate, OffsetNumber attnum, MemoryContextReset(buildstate->funcCtx); } +#define KEY_TASK 42 +#define KEY_SHM_ORIGIN 43 +#define KEY_SHM_PER_WORKER 44 +#define GIN_MAX_WORKERS 24 +#define GIN_BLOCKS_PER_WORKER 4 + +/* + * The shmem message structure: + * Entry, Key, List + */ + +typedef struct { + GinNullCategory category; + OffsetNumber attnum; + int nlist; +} GinShmemEntry; + +typedef struct { + void *mq; + shm_mq_handle *mqh; + bool end_of_tree; + bool end_of_forest; + + void *msg_body; + Size msg_len; + Datum key; + int skipkey; +} 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 shm_mq *mq; +static shm_mq_handle *mqh; + +static void ginDumpEntry(GinState *ginstate, + volatile WorkerResult *r, + OffsetNumber attnum, + Datum key, + GinNullCategory category, + ItemPointerData *list, + int nlist) +{ + int keylen, listlen; + + bool isnull; + Form_pg_attribute att; + GinShmemEntry e; + + // The message consists of 2 or 3 parts. iovec allows us to send them as + // one message though the parts are located at unrelated addresses. + shm_mq_iovec iov[3]; + int iovlen = 0; + + char *buf = NULL; + + e.category = category; + e.attnum = attnum; + e.nlist = nlist; + + Assert(nlist > 0); + + isnull = category == GIN_CAT_NULL_KEY; + att = ginstate->origTupdesc->attrs[attnum - 1]; + + if (e.category == GIN_CAT_NORM_KEY) + { + keylen = datumEstimateSpace(key, isnull, att->attbyval, att->attlen); + Assert(keylen > 0); + listlen = e.nlist * sizeof(ItemPointerData); + } + else + keylen = 0; + + listlen = e.nlist * sizeof(ItemPointerData); + + iov[iovlen].data = (char *)&e; + iov[iovlen++].len = sizeof(e); + + if (keylen > 0) + { + char *cursor; + buf = palloc(keylen); + cursor = buf; + datumSerialize(key, isnull, att->attbyval, att->attlen, &cursor); + iov[iovlen].data = buf; + iov[iovlen++].len = keylen; + } + + iov[iovlen].data = (char *)list; + iov[iovlen++].len = listlen; + + shm_mq_sendv(mqh, iov, iovlen, false); + + if (buf) + pfree(buf); +} + +static void ginDumpAccumulator(GinBuildState *buildstate) +{ + ItemPointerData *list; + Datum key; + GinNullCategory category; + uint32 nlist; + OffsetNumber attnum; + MemoryContext oldCtx; + volatile WorkerResult *r = NULL; + + if (IsParallelWorker()) + { + r = &task->results[ParallelWorkerNumber]; + } + 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 (r) + 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); + + if (IsParallelWorker()) + { + // send empty message as an "end-of-tree" marker + shm_mq_result r = shm_mq_send(mqh, 0, NULL, false); + if (r != SHM_MQ_SUCCESS) + { + elog(ERROR, "failed to send the results from worker to backend"); + } + } + + MemoryContextSwitchTo(oldCtx); +} + static void ginBuildCallback(Relation index, HeapTuple htup, Datum *values, bool *isnull, bool tupleIsAlive, void *state) @@ -283,52 +452,350 @@ 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); } +/* + * 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; + + double reltuples; + + char *shm_origin; + int mqsize; + + task = (PGinBuildTask*)shm_toc_lookup(toc, KEY_TASK); + + shm_origin = (char *)shm_toc_lookup(toc, KEY_SHM_ORIGIN); + mqsize = *(int*)shm_toc_lookup(toc, KEY_SHM_PER_WORKER); + mq = (shm_mq *)(shm_origin + ParallelWorkerNumber * mqsize); + shm_mq_set_sender(mq, MyProc); + mqh = shm_mq_attach(mq, seg, NULL); + shm_mq_wait_for_attach(mqh); + + 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); + + shm_mq_detach(mq); +} + +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; +} + +/* + * Wait until all unfinished workers start dumping their trees. + * Return the number of trees to merge. + */ +static int waitNextTree(ParallelContext *pcxt, volatile PGinBuildTask *task) +{ + int i; + int trees = 0; + + for (i = 0; i < pcxt->nworkers; ++i) + { + volatile WorkerResult *r = &task->results[i]; + + if (shm_mq_receive(r->mqh, (Size *)&r->msg_len, (void *)&r->msg_body, false) == SHM_MQ_SUCCESS) + { + r->end_of_tree = false; + trees++; + } + else + { + r->end_of_forest = true; + } + } + return trees; +} + +/* Merge the results from all ready (but unfinished) workers. */ +static void mergeReadyAndUnfinished(GinBuildState *buildstate, ParallelContext *pcxt, volatile PGinBuildTask *task) +{ + GinEntryStack *entry = NULL; + while (true) + { + int i; + bool merged = false; + + for (i = 0; i < pcxt->nworkers; ++i) + { + GinShmemEntry *shentry; + ItemPointerData *oldlist; + int cmp; + char *cursor; + volatile WorkerResult *r = &task->results[i]; + + if (r->end_of_forest || r->end_of_tree) + { + continue; + } + + if (r->msg_len == 0) /* end-of-tree */ + { + r->end_of_tree = true; + continue; + } + + cursor = r->msg_body; + Assert(cursor != NULL); + shentry = (GinShmemEntry*)cursor; + cursor += sizeof(shentry); + + if (r->skipkey) + cursor += r->skipkey; + else + { + r->key = 0; + if (shentry->category == GIN_CAT_NORM_KEY) + { + bool isnull; + char *oldcursor = cursor; + r->key = datumRestore(&cursor, &isnull); // TODO: check if this leaks memory in a long-living context + r->skipkey = cursor - oldcursor; + Assert(!isnull); + Assert(r->skipkey); + } + } + + cmp = -1; + if (entry != NULL) + { + cmp = ginCompareAttEntries(&buildstate->ginstate, + shentry->attnum, r->key, shentry->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 = shentry->category; + entry->attnum = shentry->attnum; + } + + /* The key is less than or equal. Merge the item pointers. */ + { + ItemPointerData *list = (ItemPointerData*)cursor; + oldlist = entry->list; + entry->list = ginMergeItemPointers(entry->list, entry->nlist, + list, shentry->nlist, + &entry->nlist); + + pfree(oldlist); + } + + /* Message consumed. Receive the next one. */ + r->skipkey = 0; + if (shm_mq_receive(r->mqh, (Size *)&r->msg_len, (void *)&r->msg_body, false) != SHM_MQ_SUCCESS) + r->end_of_forest = true; + 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); + } + } +} + +static void mergeResults(GinBuildState *buildstate, ParallelContext *pcxt, volatile PGinBuildTask *task) +{ + int trees; + while ((trees = waitNextTree(pcxt, task)) > 0) + { + mergeReadyAndUnfinished(buildstate, pcxt, task); + } +} + 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 +830,97 @@ 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)) + { + elog(WARNING, "not using parallel GIN build on temporary table %s\n", NameStr(heap->rd_rel->relname)); + } + else + { + EnterParallelMode(); + { + int i; + + PGinBuildTask *task; + ParallelContext *pcxt; + void *shm; + void *ptr; + + int *mqsize; + + int size = 0, keys = 0; + keys++; size += sizeof(PGinBuildTask); + keys++; size += gin_shared_mem * 1024; + keys++; size += sizeof(int); + + pcxt = CreateParallelContext(ginbuildWorker, gin_parallel_workers); + + shm_toc_estimate_chunk(&pcxt->estimator, size); + shm_toc_estimate_keys(&pcxt->estimator, keys); + InitializeParallelDSM(pcxt); + + ptr = shm_toc_allocate(pcxt->toc, sizeof(PGinBuildTask)); + shm_toc_insert(pcxt->toc, KEY_TASK, ptr); + task = (PGinBuildTask*)ptr; - MemoryContextDelete(buildstate.funcCtx); - MemoryContextDelete(buildstate.tmpCtx); + ptr = shm_toc_allocate(pcxt->toc, gin_shared_mem * 1024); + shm_toc_insert(pcxt->toc, KEY_SHM_ORIGIN, ptr); + shm = ptr; - /* - * Update metapage stats - */ - buildstate.buildStats.nTotalPages = RelationGetNumberOfBlocks(index); - ginUpdateStats(index, &buildstate.buildStats); + mqsize = (int *)shm_toc_allocate(pcxt->toc, sizeof(int)); + *mqsize = gin_shared_mem * 1024 / pcxt->nworkers; + shm_toc_insert(pcxt->toc, KEY_SHM_PER_WORKER, mqsize); + + SpinLockInit(&task->lock); + + for (i = 0; i < pcxt->nworkers; i++) + { + volatile WorkerResult *r = &task->results[i]; + r->mq = shm_mq_create((char *)shm + i * (*mqsize), *mqsize); + shm_mq_set_receiver(r->mq, MyProc); + r->mqh = shm_mq_attach(r->mq, pcxt->seg, NULL); + r->end_of_tree = false; + r->end_of_forest = false; + r->msg_body = NULL; + r->msg_len = 0; + r->skipkey = 0; + } + 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); + + 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 edcafce..e53cc93 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -2744,6 +2744,31 @@ 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 + }, + + { + {"gin_shared_mem", + PGC_USERSET, + RESOURCES_ASYNCHRONOUS, + gettext_noop("The size of shared memory segment for parallel GIN buiding."), + NULL, + GUC_UNIT_KB + }, + &gin_shared_mem, + 16 * 1024, 1024, MAX_KILOBYTES, + 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 ee3d378..a756218 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -538,6 +538,8 @@ #xmloption = 'content' #gin_fuzzy_search_limit = 0 #gin_pending_list_limit = 4MB +#gin_parallel_workers = 0 # 0 disables parallel gin build +#gin_shared_mem = 16MB # - Locale and Formatting - diff --git a/src/include/access/gin.h b/src/include/access/gin.h index e5b2e10..91e5b27 100644 --- a/src/include/access/gin.h +++ b/src/include/access/gin.h @@ -68,6 +68,8 @@ typedef char GinTernaryValue; /* GUC parameters */ extern PGDLLIMPORT int GinFuzzySearchLimit; extern int gin_pending_list_limit; +extern int gin_parallel_workers; +extern int gin_shared_mem; /* ginutil.c */ extern void ginGetStats(Relation index, GinStatsData *stats);