Use atomic ops to hand out pages to scan in parallel scan.

With a lot of CPUs, the spinlock that protects the current scan location
in a parallel scan can become a bottleneck. Use an atomic fetch-and-add
instruction instead.

David Rowley

Discussion: https://www.postgresql.org/message-id/CAKJS1f9tgsPhqBcoPjv9_KUPZvTLCZ4jy%3DB%3DbhqgaKn7cYzm-w@mail.gmail.com

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index e283fe5b1f51ac8dbc2712b16f6774efda96ee87..7dea8472c17243dfd99d7db0ab77a8531d384d55 100644 (file)
--- 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(&parallel_scan->phs_mutex);
-               parallel_scan->phs_cblock = parallel_scan->phs_startblock;
-               SpinLockRelease(&parallel_scan->phs_mutex);
+               pg_atomic_write_u64(&parallel_scan->phs_nallocated, 0);
        }
 }
 
@@ -1635,8 +1637,8 @@ heap_parallelscan_initialize(ParallelHeapScanDesc target, Relation relation,
                !RelationUsesLocalBuffers(relation) &&
                target->phs_nblocks > NBuffers / 4;
        SpinLockInit(&target->phs_mutex);
-       target->phs_cblock = InvalidBlockNumber;
        target->phs_startblock = InvalidBlockNumber;
+       pg_atomic_write_u64(&target->phs_nallocated, 0);
        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,63 @@ retry:
                        sync_startpage = ss_get_location(scan->rs_rd, scan->rs_nblocks);
                        goto retry;
                }
-               parallel_scan->phs_cblock = parallel_scan->phs_startblock;
        }
+       SpinLockRelease(&parallel_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          nallocated;
+
+       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.
+        * phs_nallocated tracks how many pages have been allocated to workers
+        * already.  When phs_nallocated >= rs_nblocks, all blocks have been
+        * allocated.
+        *
+        * Because we use an atomic fetch-and-add to fetch the current value, the
+        * phs_nallocated counter will exceed rs_nblocks, because workers will
+        * still increment the value, when they try to allocate the next block but
+        * all blocks have been allocated already. The counter must be 64 bits
+        * wide because of that, to avoid wrapping around when rs_nblocks is close
+        * to 2^32.
+        *
+        * The actual page to return is calculated by adding the counter to the
+        * starting block number, modulo nblocks.
         */
-       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;
-               }
-       }
-
-       /* Release the lock. */
-       SpinLockRelease(&parallel_scan->phs_mutex);
+       nallocated = pg_atomic_fetch_add_u64(&parallel_scan->phs_nallocated, 1);
+       if (nallocated >= scan->rs_nblocks)
+               page = InvalidBlockNumber;      /* all blocks have been allocated */
+       else
+               page = (nallocated + 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 (nallocated == 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 a20646b2b706191b442e98f72420238015488238..147f862a2b85fcf19ab3992b3c5d2eda6ecc6b5b 100644 (file)
--- 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_nallocated;        /* number of blocks allocated to
+                                                                                * workers so far. */
        char            phs_snapshot_data[FLEXIBLE_ARRAY_MEMBER];
 }                      ParallelHeapScanDescData;