diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 0c3e2b0..1bbaca9 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -58,6 +58,7 @@ #include "catalog/namespace.h" #include "miscadmin.h" #include "pgstat.h" +#include "port/atomics.h" #include "storage/bufmgr.h" #include "storage/freespace.h" #include "storage/lmgr.h" @@ -89,6 +90,7 @@ static HeapScanDesc heap_beginscan_internal(Relation relation, bool is_bitmapscan, bool is_samplescan, bool temp_snap); +static void heap_parallelscan_startblock_init(HeapScanDesc scan); static BlockNumber heap_parallelscan_nextpage(HeapScanDesc scan); static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid, CommandId cid, int options); @@ -510,6 +512,8 @@ heapgettup(HeapScanDesc scan, } if (scan->rs_parallel != NULL) { + heap_parallelscan_startblock_init(scan); + page = heap_parallelscan_nextpage(scan); /* Other processes might have already finished the scan. */ @@ -812,6 +816,8 @@ heapgettup_pagemode(HeapScanDesc scan, } if (scan->rs_parallel != NULL) { + heap_parallelscan_startblock_init(scan); + page = heap_parallelscan_nextpage(scan); /* Other processes might have already finished the scan. */ @@ -1535,14 +1541,10 @@ heap_rescan(HeapScanDesc scan, /* * Caller is responsible for making sure that all workers have - * finished the scan before calling this, so it really shouldn't be - * necessary to acquire the mutex at all. We acquire it anyway, just - * to be tidy. + * finished the scan before calling this. */ parallel_scan = scan->rs_parallel; - SpinLockAcquire(¶llel_scan->phs_mutex); - parallel_scan->phs_cblock = parallel_scan->phs_startblock; - SpinLockRelease(¶llel_scan->phs_mutex); + pg_atomic_write_u64(¶llel_scan->phs_blockcounter, 0); } } @@ -1635,7 +1637,7 @@ heap_parallelscan_initialize(ParallelHeapScanDesc target, Relation relation, !RelationUsesLocalBuffers(relation) && target->phs_nblocks > NBuffers / 4; SpinLockInit(&target->phs_mutex); - target->phs_cblock = InvalidBlockNumber; + pg_atomic_write_u64(&target->phs_blockcounter, 0); target->phs_startblock = InvalidBlockNumber; SerializeSnapshot(snapshot, target->phs_snapshot_data); } @@ -1660,20 +1662,17 @@ heap_beginscan_parallel(Relation relation, ParallelHeapScanDesc parallel_scan) } /* ---------------- - * heap_parallelscan_nextpage - get the next page to scan + * heap_parallelscan_startblock_init - find and set the scan's startblock * - * Get the next page to scan. Even if there are no pages left to scan, - * another backend could have grabbed a page to scan and not yet finished - * looking at it, so it doesn't follow that the scan is done when the - * first backend gets an InvalidBlockNumber return. + * Determine where the parallel seq scan should start. This function may + * be called many times, once by each parallel worker. We must be careful + * only to set the startblock once. * ---------------- */ -static BlockNumber -heap_parallelscan_nextpage(HeapScanDesc scan) +static void +heap_parallelscan_startblock_init(HeapScanDesc scan) { - BlockNumber page = InvalidBlockNumber; BlockNumber sync_startpage = InvalidBlockNumber; - BlockNumber report_page = InvalidBlockNumber; ParallelHeapScanDesc parallel_scan; Assert(scan->rs_parallel); @@ -1705,46 +1704,64 @@ retry: sync_startpage = ss_get_location(scan->rs_rd, scan->rs_nblocks); goto retry; } - parallel_scan->phs_cblock = parallel_scan->phs_startblock; } + SpinLockRelease(¶llel_scan->phs_mutex); +} + +/* ---------------- + * heap_parallelscan_nextpage - get the next page to scan + * + * Get the next page to scan. Even if there are no pages left to scan, + * another backend could have grabbed a page to scan and not yet finished + * looking at it, so it doesn't follow that the scan is done when the + * first backend gets an InvalidBlockNumber return. + * ---------------- + */ +static BlockNumber +heap_parallelscan_nextpage(HeapScanDesc scan) +{ + BlockNumber page; + ParallelHeapScanDesc parallel_scan; + uint64 counter; + + Assert(scan->rs_parallel); + parallel_scan = scan->rs_parallel; /* - * The current block number is the next one that needs to be scanned, - * unless it's InvalidBlockNumber already, in which case there are no more - * blocks to scan. After remembering the current value, we must advance - * it so that the next call to this function returns the next block to be - * scanned. + * We make use of phs_blockcounter to track how many pages have been + * allocated to workers already. We're able to easily determine if we've + * allocated all blocks out already by testing if counter has reached or + * exceeded rs_nblocks. Even after the first worker has found counter to + * have exceeded rs_nblocks, subsequent workers might report in here + * looking for another page, which will further increment + * phs_blockcounter. For this reason the block counter must be 64bit. We + * don't want to risk an overflow when scanning very large relations. + * + * The actual page to return must be calculated using the counter. We + * simply add on the startblock number and modulo rs_nblocks to handle + * the scanning of any blocks earlier than phs_startblock. */ - page = parallel_scan->phs_cblock; - if (page != InvalidBlockNumber) - { - parallel_scan->phs_cblock++; - if (parallel_scan->phs_cblock >= scan->rs_nblocks) - parallel_scan->phs_cblock = 0; - if (parallel_scan->phs_cblock == parallel_scan->phs_startblock) - { - parallel_scan->phs_cblock = InvalidBlockNumber; - report_page = parallel_scan->phs_startblock; - } - } + counter = pg_atomic_fetch_add_u64(¶llel_scan->phs_blockcounter, 1); - /* Release the lock. */ - SpinLockRelease(¶llel_scan->phs_mutex); + /* Have we already allocated all blocks of this relation? */ + if (counter >= scan->rs_nblocks) + page = InvalidBlockNumber; + else + page = (counter + parallel_scan->phs_startblock) % scan->rs_nblocks; /* * Report scan location. Normally, we report the current page number. * When we reach the end of the scan, though, we report the starting page, * not the ending page, just so the starting positions for later scans * doesn't slew backwards. We only report the position at the end of the - * scan once, though: subsequent callers will have report nothing, since - * they will have page == InvalidBlockNumber. + * scan once, though: subsequent callers will report nothing. */ if (scan->rs_syncscan) { - if (report_page == InvalidBlockNumber) - report_page = page; - if (report_page != InvalidBlockNumber) - ss_report_location(scan->rs_rd, report_page); + if (page != InvalidBlockNumber) + ss_report_location(scan->rs_rd, page); + else if (counter == scan->rs_nblocks) + ss_report_location(scan->rs_rd, parallel_scan->phs_startblock); } return page; diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h index 3fc726d..bf19b88 100644 --- a/src/include/access/relscan.h +++ b/src/include/access/relscan.h @@ -35,9 +35,10 @@ typedef struct ParallelHeapScanDescData Oid phs_relid; /* OID of relation to scan */ bool phs_syncscan; /* report location to syncscan logic? */ BlockNumber phs_nblocks; /* # blocks in relation at start of scan */ - slock_t phs_mutex; /* mutual exclusion for block number fields */ + slock_t phs_mutex; /* mutual exclusion for setting startblock */ BlockNumber phs_startblock; /* starting block number */ - BlockNumber phs_cblock; /* current block number */ + pg_atomic_uint64 phs_blockcounter; /* number of blocks allocated to + * workers so far. */ char phs_snapshot_data[FLEXIBLE_ARRAY_MEMBER]; } ParallelHeapScanDescData;