@@ -109,6 +109,7 @@ typedef struct InProgressIO
109
109
struct ReadStream
110
110
{
111
111
int16 max_ios ;
112
+ int16 io_combine_limit ;
112
113
int16 ios_in_progress ;
113
114
int16 queue_size ;
114
115
int16 max_pinned_buffers ;
@@ -236,7 +237,7 @@ read_stream_start_pending_read(ReadStream *stream, bool suppress_advice)
236
237
237
238
/* This should only be called with a pending read. */
238
239
Assert (stream -> pending_read_nblocks > 0 );
239
- Assert (stream -> pending_read_nblocks <= io_combine_limit );
240
+ Assert (stream -> pending_read_nblocks <= stream -> io_combine_limit );
240
241
241
242
/* We had better not exceed the pin limit by starting this read. */
242
243
Assert (stream -> pinned_buffers + stream -> pending_read_nblocks <=
@@ -324,7 +325,7 @@ read_stream_look_ahead(ReadStream *stream, bool suppress_advice)
324
325
int16 buffer_index ;
325
326
void * per_buffer_data ;
326
327
327
- if (stream -> pending_read_nblocks == io_combine_limit )
328
+ if (stream -> pending_read_nblocks == stream -> io_combine_limit )
328
329
{
329
330
read_stream_start_pending_read (stream , suppress_advice );
330
331
suppress_advice = false;
@@ -384,7 +385,7 @@ read_stream_look_ahead(ReadStream *stream, bool suppress_advice)
384
385
* signaled end-of-stream, we start the read immediately.
385
386
*/
386
387
if (stream -> pending_read_nblocks > 0 &&
387
- (stream -> pending_read_nblocks == io_combine_limit ||
388
+ (stream -> pending_read_nblocks == stream -> io_combine_limit ||
388
389
(stream -> pending_read_nblocks == stream -> distance &&
389
390
stream -> pinned_buffers == 0 ) ||
390
391
stream -> distance == 0 ) &&
@@ -415,6 +416,7 @@ read_stream_begin_impl(int flags,
415
416
ReadStream * stream ;
416
417
size_t size ;
417
418
int16 queue_size ;
419
+ int16 queue_overflow ;
418
420
int max_ios ;
419
421
int strategy_pin_limit ;
420
422
uint32 max_pinned_buffers ;
@@ -445,6 +447,14 @@ read_stream_begin_impl(int flags,
445
447
/* Cap to INT16_MAX to avoid overflowing below */
446
448
max_ios = Min (max_ios , PG_INT16_MAX );
447
449
450
+ /*
451
+ * If starting a multi-block I/O near the end of the queue, we might
452
+ * temporarily need extra space for overflowing buffers before they are
453
+ * moved to regular circular position. This is the maximum extra space we
454
+ * could need.
455
+ */
456
+ queue_overflow = io_combine_limit - 1 ;
457
+
448
458
/*
449
459
* Choose the maximum number of buffers we're prepared to pin. We try to
450
460
* pin fewer if we can, though. We add one so that we can make progress
@@ -459,7 +469,7 @@ read_stream_begin_impl(int flags,
459
469
*/
460
470
max_pinned_buffers = (max_ios + 1 ) * io_combine_limit ;
461
471
max_pinned_buffers = Min (max_pinned_buffers ,
462
- PG_INT16_MAX - io_combine_limit - 1 );
472
+ PG_INT16_MAX - queue_overflow - 1 );
463
473
464
474
/* Give the strategy a chance to limit the number of buffers we pin. */
465
475
strategy_pin_limit = GetAccessStrategyPinLimit (strategy );
@@ -485,18 +495,17 @@ read_stream_begin_impl(int flags,
485
495
* one big chunk. Though we have queue_size buffers, we want to be able
486
496
* to assume that all the buffers for a single read are contiguous (i.e.
487
497
* don't wrap around halfway through), so we allow temporary overflows of
488
- * up to the maximum possible read size by allocating an extra
489
- * io_combine_limit - 1 elements.
498
+ * up to the maximum possible overflow size.
490
499
*/
491
500
size = offsetof(ReadStream , buffers );
492
- size += sizeof (Buffer ) * (queue_size + io_combine_limit - 1 );
501
+ size += sizeof (Buffer ) * (queue_size + queue_overflow );
493
502
size += sizeof (InProgressIO ) * Max (1 , max_ios );
494
503
size += per_buffer_data_size * queue_size ;
495
504
size += MAXIMUM_ALIGNOF * 2 ;
496
505
stream = (ReadStream * ) palloc (size );
497
506
memset (stream , 0 , offsetof(ReadStream , buffers ));
498
507
stream -> ios = (InProgressIO * )
499
- MAXALIGN (& stream -> buffers [queue_size + io_combine_limit - 1 ]);
508
+ MAXALIGN (& stream -> buffers [queue_size + queue_overflow ]);
500
509
if (per_buffer_data_size > 0 )
501
510
stream -> per_buffer_data = (void * )
502
511
MAXALIGN (& stream -> ios [Max (1 , max_ios )]);
@@ -523,7 +532,14 @@ read_stream_begin_impl(int flags,
523
532
if (max_ios == 0 )
524
533
max_ios = 1 ;
525
534
535
+ /*
536
+ * Capture stable values for these two GUC-derived numbers for the
537
+ * lifetime of this stream, so we don't have to worry about the GUCs
538
+ * changing underneath us beyond this point.
539
+ */
526
540
stream -> max_ios = max_ios ;
541
+ stream -> io_combine_limit = io_combine_limit ;
542
+
527
543
stream -> per_buffer_data_size = per_buffer_data_size ;
528
544
stream -> max_pinned_buffers = max_pinned_buffers ;
529
545
stream -> queue_size = queue_size ;
@@ -537,7 +553,7 @@ read_stream_begin_impl(int flags,
537
553
* doing full io_combine_limit sized reads (behavior B).
538
554
*/
539
555
if (flags & READ_STREAM_FULL )
540
- stream -> distance = Min (max_pinned_buffers , io_combine_limit );
556
+ stream -> distance = Min (max_pinned_buffers , stream -> io_combine_limit );
541
557
else
542
558
stream -> distance = 1 ;
543
559
@@ -752,14 +768,14 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data)
752
768
else
753
769
{
754
770
/* No advice; move towards io_combine_limit (behavior B). */
755
- if (stream -> distance > io_combine_limit )
771
+ if (stream -> distance > stream -> io_combine_limit )
756
772
{
757
773
stream -> distance -- ;
758
774
}
759
775
else
760
776
{
761
777
distance = stream -> distance * 2 ;
762
- distance = Min (distance , io_combine_limit );
778
+ distance = Min (distance , stream -> io_combine_limit );
763
779
distance = Min (distance , stream -> max_pinned_buffers );
764
780
stream -> distance = distance ;
765
781
}
0 commit comments