PostgreSQL Source Code git master
Loading...
Searching...
No Matches
read_stream.h File Reference
#include "storage/bufmgr.h"
#include "storage/smgr.h"
Include dependency graph for read_stream.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  BlockRangeReadStreamPrivate
 

Macros

#define READ_STREAM_DEFAULT   0x00
 
#define READ_STREAM_MAINTENANCE   0x01
 
#define READ_STREAM_SEQUENTIAL   0x02
 
#define READ_STREAM_FULL   0x04
 
#define READ_STREAM_USE_BATCHING   0x08
 

Typedefs

typedef struct ReadStream ReadStream
 
typedef struct BlockRangeReadStreamPrivate BlockRangeReadStreamPrivate
 
typedef BlockNumber(* ReadStreamBlockNumberCB) (ReadStream *stream, void *callback_private_data, void *per_buffer_data)
 

Functions

BlockNumber block_range_read_stream_cb (ReadStream *stream, void *callback_private_data, void *per_buffer_data)
 
ReadStreamread_stream_begin_relation (int flags, BufferAccessStrategy strategy, Relation rel, ForkNumber forknum, ReadStreamBlockNumberCB callback, void *callback_private_data, size_t per_buffer_data_size)
 
Buffer read_stream_next_buffer (ReadStream *stream, void **per_buffer_data)
 
BlockNumber read_stream_next_block (ReadStream *stream, BufferAccessStrategy *strategy)
 
ReadStreamread_stream_begin_smgr_relation (int flags, BufferAccessStrategy strategy, SMgrRelation smgr, char smgr_persistence, ForkNumber forknum, ReadStreamBlockNumberCB callback, void *callback_private_data, size_t per_buffer_data_size)
 
BlockNumber read_stream_pause (ReadStream *stream)
 
void read_stream_resume (ReadStream *stream)
 
void read_stream_reset (ReadStream *stream)
 
void read_stream_end (ReadStream *stream)
 
void read_stream_enable_stats (ReadStream *stream, struct IOStats *stats)
 

Macro Definition Documentation

◆ READ_STREAM_DEFAULT

#define READ_STREAM_DEFAULT   0x00

Definition at line 21 of file read_stream.h.

◆ READ_STREAM_FULL

#define READ_STREAM_FULL   0x04

Definition at line 43 of file read_stream.h.

◆ READ_STREAM_MAINTENANCE

#define READ_STREAM_MAINTENANCE   0x01

Definition at line 28 of file read_stream.h.

◆ READ_STREAM_SEQUENTIAL

#define READ_STREAM_SEQUENTIAL   0x02

Definition at line 36 of file read_stream.h.

◆ READ_STREAM_USE_BATCHING

#define READ_STREAM_USE_BATCHING   0x08

Definition at line 64 of file read_stream.h.

Typedef Documentation

◆ BlockRangeReadStreamPrivate

◆ ReadStream

Definition at line 67 of file read_stream.h.

◆ ReadStreamBlockNumberCB

typedef BlockNumber(* ReadStreamBlockNumberCB) (ReadStream *stream, void *callback_private_data, void *per_buffer_data)

Definition at line 78 of file read_stream.h.

Function Documentation

◆ block_range_read_stream_cb()

◆ read_stream_begin_relation()

ReadStream * read_stream_begin_relation ( int  flags,
BufferAccessStrategy  strategy,
Relation  rel,
ForkNumber  forknum,
ReadStreamBlockNumberCB  callback,
void callback_private_data,
size_t  per_buffer_data_size 
)
extern

Definition at line 966 of file read_stream.c.

973{
974 return read_stream_begin_impl(flags,
975 strategy,
976 rel,
977 RelationGetSmgr(rel),
978 rel->rd_rel->relpersistence,
979 forknum,
980 callback,
981 callback_private_data,
982 per_buffer_data_size);
983}
static ReadStream * read_stream_begin_impl(int flags, BufferAccessStrategy strategy, Relation rel, SMgrRelation smgr, char persistence, ForkNumber forknum, ReadStreamBlockNumberCB callback, void *callback_private_data, size_t per_buffer_data_size)
static SMgrRelation RelationGetSmgr(Relation rel)
Definition rel.h:578
Form_pg_class rd_rel
Definition rel.h:111
static void callback(struct sockaddr *addr, struct sockaddr *mask, void *unused)

References callback(), RelationData::rd_rel, read_stream_begin_impl(), and RelationGetSmgr().

Referenced by acquire_sample_rows(), autoprewarm_database_main(), blbulkdelete(), blgetbitmap(), blvacuumcleanup(), brin_vacuum_scan(), btvacuumscan(), collect_corrupt_items(), collect_visibility_data(), ginvacuumcleanup(), gistvacuumscan(), hashbulkdelete(), heap_beginscan(), lazy_scan_heap(), lazy_vacuum_heap_rel(), pg_prewarm(), pgstathashindex(), pgstatindex_impl(), read_stream_for_blocks(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

◆ read_stream_begin_smgr_relation()

ReadStream * read_stream_begin_smgr_relation ( int  flags,
BufferAccessStrategy  strategy,
SMgrRelation  smgr,
char  smgr_persistence,
ForkNumber  forknum,
ReadStreamBlockNumberCB  callback,
void callback_private_data,
size_t  per_buffer_data_size 
)
extern

Definition at line 990 of file read_stream.c.

998{
999 return read_stream_begin_impl(flags,
1000 strategy,
1001 NULL,
1002 smgr,
1004 forknum,
1005 callback,
1006 callback_private_data,
1007 per_buffer_data_size);
1008}
static int fb(int x)

References callback(), fb(), and read_stream_begin_impl().

Referenced by RelationCopyStorageUsingBuffer().

◆ read_stream_enable_stats()

void read_stream_enable_stats ( ReadStream stream,
struct IOStats stats 
)
extern

Definition at line 255 of file read_stream.c.

256{
257 stream->stats = stats;
258 if (stream->stats)
259 stream->stats->distance_capacity = stream->max_pinned_buffers;
260}
int16 distance_capacity
int16 max_pinned_buffers
IOStats * stats

References IOStats::distance_capacity, ReadStream::max_pinned_buffers, and ReadStream::stats.

Referenced by heap_beginscan().

◆ read_stream_end()

◆ read_stream_next_block()

BlockNumber read_stream_next_block ( ReadStream stream,
BufferAccessStrategy strategy 
)
extern

Definition at line 1367 of file read_stream.c.

1368{
1369 *strategy = stream->ios[0].op.strategy;
1370 return read_stream_get_block(stream, NULL);
1371}
static BlockNumber read_stream_get_block(ReadStream *stream, void *per_buffer_data)
ReadBuffersOperation op
Definition read_stream.c:89
BufferAccessStrategy strategy
Definition bufmgr.h:138
InProgressIO * ios

References fb(), ReadStream::ios, InProgressIO::op, read_stream_get_block(), and ReadBuffersOperation::strategy.

◆ read_stream_next_buffer()

Buffer read_stream_next_buffer ( ReadStream stream,
void **  per_buffer_data 
)
extern

Definition at line 1020 of file read_stream.c.

1021{
1022 Buffer buffer;
1023 int16 oldest_buffer_index;
1024
1025#ifndef READ_STREAM_DISABLE_FAST_PATH
1026
1027 /*
1028 * A fast path for all-cached scans. This is the same as the usual
1029 * algorithm, but it is specialized for no I/O and no per-buffer data, so
1030 * we can skip the queue management code, stay in the same buffer slot and
1031 * use singular StartReadBuffer().
1032 */
1033 if (likely(stream->fast_path))
1034 {
1036
1037 /* Fast path assumptions. */
1038 Assert(stream->ios_in_progress == 0);
1039 Assert(stream->forwarded_buffers == 0);
1040 Assert(stream->pinned_buffers == 1);
1041 Assert(stream->readahead_distance == 1);
1042 Assert(stream->combine_distance == 1);
1043 Assert(stream->pending_read_nblocks == 0);
1044 Assert(stream->per_buffer_data_size == 0);
1046
1047 /* We're going to return the buffer we pinned last time. */
1048 oldest_buffer_index = stream->oldest_buffer_index;
1049 Assert((oldest_buffer_index + 1) % stream->queue_size ==
1050 stream->next_buffer_index);
1051 buffer = stream->buffers[oldest_buffer_index];
1052 Assert(buffer != InvalidBuffer);
1053
1054 /* Choose the next block to pin. */
1056
1058 {
1059 int flags = stream->read_buffers_flags;
1060
1061 if (stream->advice_enabled)
1063
1064 /*
1065 * While in fast-path, execute any IO that we might encounter
1066 * synchronously. Because we are, right now, only looking one
1067 * block ahead, dispatching any occasional IO to workers would
1068 * have the overhead of dispatching to workers, without any
1069 * realistic chance of the IO completing before we need it. We
1070 * will switch to non-synchronous IO after this.
1071 *
1072 * Arguably we should do so only for worker, as there's far less
1073 * dispatch overhead with io_uring. However, tests so far have not
1074 * shown a clear downside and additional io_method awareness here
1075 * seems not great from an abstraction POV.
1076 */
1078
1079 /*
1080 * Pin a buffer for the next call. Same buffer entry, and
1081 * arbitrary I/O entry (they're all free). We don't have to
1082 * adjust pinned_buffers because we're transferring one to caller
1083 * but pinning one more.
1084 *
1085 * In the fast path we don't need to check the pin limit. We're
1086 * always allowed at least one pin so that progress can be made,
1087 * and that's all we need here. Although two pins are momentarily
1088 * held at the same time, the model used here is that the stream
1089 * holds only one, and the other now belongs to the caller.
1090 */
1091 if (likely(!StartReadBuffer(&stream->ios[0].op,
1092 &stream->buffers[oldest_buffer_index],
1094 flags)))
1095 {
1096 /* Fast return. */
1098 return buffer;
1099 }
1100
1101 /* Next call must wait for I/O for the newly pinned buffer. */
1102 stream->oldest_io_index = 0;
1103 stream->next_io_index = stream->max_ios > 1 ? 1 : 0;
1104 stream->ios_in_progress = 1;
1105 stream->ios[0].buffer_index = oldest_buffer_index;
1106 stream->seq_blocknum = next_blocknum + 1;
1107
1108 /*
1109 * XXX: It might be worth triggering additional read-ahead here,
1110 * to avoid having to effectively do another synchronous IO for
1111 * the next block (if it were also a miss).
1112 */
1113
1114 /* update I/O stats */
1115 read_stream_count_io(stream, 1, stream->ios_in_progress);
1116
1117 /* update prefetch distance */
1119 }
1120 else
1121 {
1122 /* No more blocks, end of stream. */
1123 stream->readahead_distance = 0;
1124 stream->combine_distance = 0;
1125 stream->oldest_buffer_index = stream->next_buffer_index;
1126 stream->pinned_buffers = 0;
1127 stream->buffers[oldest_buffer_index] = InvalidBuffer;
1128 }
1129
1130 stream->fast_path = false;
1131 return buffer;
1132 }
1133#endif
1134
1135 if (unlikely(stream->pinned_buffers == 0))
1136 {
1137 Assert(stream->oldest_buffer_index == stream->next_buffer_index);
1138
1139 /* End of stream reached? */
1140 if (stream->readahead_distance == 0)
1141 return InvalidBuffer;
1142
1143 /*
1144 * The usual order of operations is that we look ahead at the bottom
1145 * of this function after potentially finishing an I/O and making
1146 * space for more, but if we're just starting up we'll need to crank
1147 * the handle to get started.
1148 */
1149 read_stream_look_ahead(stream);
1150
1151 /* End of stream reached? */
1152 if (stream->pinned_buffers == 0)
1153 {
1154 Assert(stream->readahead_distance == 0);
1155 return InvalidBuffer;
1156 }
1157 }
1158
1159 /* Grab the oldest pinned buffer and associated per-buffer data. */
1160 Assert(stream->pinned_buffers > 0);
1161 oldest_buffer_index = stream->oldest_buffer_index;
1162 Assert(oldest_buffer_index >= 0 &&
1164 buffer = stream->buffers[oldest_buffer_index];
1165 if (per_buffer_data)
1166 *per_buffer_data = get_per_buffer_data(stream, oldest_buffer_index);
1167
1168 Assert(BufferIsValid(buffer));
1169
1170 /* Do we have to wait for an associated I/O first? */
1171 if (stream->ios_in_progress > 0 &&
1172 stream->ios[stream->oldest_io_index].buffer_index == oldest_buffer_index)
1173 {
1174 int16 io_index = stream->oldest_io_index;
1175 bool needed_wait;
1176
1177 /* Sanity check that we still agree on the buffers. */
1178 Assert(stream->ios[io_index].op.buffers ==
1179 &stream->buffers[oldest_buffer_index]);
1180
1182
1183 Assert(stream->ios_in_progress > 0);
1184 stream->ios_in_progress--;
1185 if (++stream->oldest_io_index == stream->max_ios)
1186 stream->oldest_io_index = 0;
1187
1188 /*
1189 * If the IO was executed synchronously, we will never see
1190 * WaitReadBuffers() block. Treat it as if it did block. This is
1191 * particularly crucial when effective_io_concurrency=0 is used, as
1192 * all IO will be synchronous. Without treating synchronous IO as
1193 * having waited, we'd never allow the distance to get large enough to
1194 * allow for IO combining, resulting in bad performance.
1195 */
1197 needed_wait = true;
1198
1199 /* Count it as a wait if we need to wait for IO */
1200 if (needed_wait)
1201 read_stream_count_wait(stream);
1202
1203 /*
1204 * Have the read-ahead distance ramp up rapidly after we needed to
1205 * wait for IO. We only increase the read-ahead-distance when we
1206 * needed to wait, to avoid increasing the distance further than
1207 * necessary, as looking ahead too far can be costly, both due to the
1208 * cost of unnecessarily pinning many buffers and due to doing IOs
1209 * that may never be consumed if the stream is ended/reset before
1210 * completion.
1211 *
1212 * If we did not need to wait, the current distance was evidently
1213 * sufficient.
1214 *
1215 * NB: Must not increase the distance if we already reached the end of
1216 * the stream, as stream->readahead_distance == 0 is used to keep
1217 * track of having reached the end.
1218 */
1219 if (stream->readahead_distance > 0 && needed_wait)
1220 {
1221 /* wider temporary value, due to overflow risk */
1222 int32 readahead_distance;
1223
1224 readahead_distance = stream->readahead_distance * 2;
1225 readahead_distance = Min(readahead_distance, stream->max_pinned_buffers);
1226 stream->readahead_distance = readahead_distance;
1227 }
1228
1229 /*
1230 * As we needed IO, prevent distances from being reduced within our
1231 * maximum look-ahead window. This avoids collapsing distances too
1232 * quickly in workloads where most of the required blocks are cached,
1233 * but where the remaining IOs are a sufficient enough factor to cause
1234 * a substantial slowdown if executed synchronously.
1235 *
1236 * There are valid arguments for preventing decay for max_ios or for
1237 * max_pinned_buffers. But the argument for max_pinned_buffers seems
1238 * clearer - if we can't see any misses within the maximum look-ahead
1239 * distance, we can't do any useful read-ahead.
1240 */
1242
1243 /*
1244 * Whether we needed to wait or not, allow for more IO combining if we
1245 * needed to do IO. The reason to do so independent of needing to wait
1246 * is that when the data is resident in the kernel page cache, IO
1247 * combining reduces the syscall / dispatch overhead, making it
1248 * worthwhile regardless of needing to wait.
1249 *
1250 * It is also important with io_uring as it will never signal the need
1251 * to wait for reads if all the data is in the page cache. There are
1252 * heuristics to deal with that in method_io_uring.c, but they only
1253 * work when the IO gets large enough.
1254 */
1255 if (stream->combine_distance > 0 &&
1256 stream->combine_distance < stream->io_combine_limit)
1257 {
1258 /* wider temporary value, due to overflow risk */
1259 int32 combine_distance;
1260
1261 combine_distance = stream->combine_distance * 2;
1262 combine_distance = Min(combine_distance, stream->io_combine_limit);
1263 combine_distance = Min(combine_distance, stream->max_pinned_buffers);
1264 stream->combine_distance = combine_distance;
1265 }
1266
1267 /*
1268 * If we've reached the first block of a sequential region we're
1269 * issuing advice for, cancel that until the next jump. The kernel
1270 * will see the sequential preadv() pattern starting here.
1271 */
1272 if (stream->advice_enabled &&
1273 stream->ios[io_index].op.blocknum == stream->seq_until_processed)
1275 }
1276
1277 /*
1278 * We must zap this queue entry, or else it would appear as a forwarded
1279 * buffer. If it's potentially in the overflow zone (ie from a
1280 * multi-block I/O that wrapped around the queue), also zap the copy.
1281 */
1282 stream->buffers[oldest_buffer_index] = InvalidBuffer;
1284 stream->buffers[stream->queue_size + oldest_buffer_index] =
1286
1287#if defined(CLOBBER_FREED_MEMORY) || defined(USE_VALGRIND)
1288
1289 /*
1290 * The caller will get access to the per-buffer data, until the next call.
1291 * We wipe the one before, which is never occupied because queue_size
1292 * allowed one extra element. This will hopefully trip up client code
1293 * that is holding a dangling pointer to it.
1294 */
1295 if (stream->per_buffer_data)
1296 {
1297 void *per_buffer_data;
1298
1299 per_buffer_data = get_per_buffer_data(stream,
1300 oldest_buffer_index == 0 ?
1301 stream->queue_size - 1 :
1302 oldest_buffer_index - 1);
1303
1304#if defined(CLOBBER_FREED_MEMORY)
1305 /* This also tells Valgrind the memory is "noaccess". */
1306 wipe_mem(per_buffer_data, stream->per_buffer_data_size);
1307#elif defined(USE_VALGRIND)
1308 /* Tell it ourselves. */
1309 VALGRIND_MAKE_MEM_NOACCESS(per_buffer_data,
1310 stream->per_buffer_data_size);
1311#endif
1312 }
1313#endif
1314
1316
1317 /* Pin transferred to caller. */
1318 Assert(stream->pinned_buffers > 0);
1319 stream->pinned_buffers--;
1320
1321 /* Advance oldest buffer, with wrap-around. */
1322 stream->oldest_buffer_index++;
1323 if (stream->oldest_buffer_index == stream->queue_size)
1324 stream->oldest_buffer_index = 0;
1325
1326 /* Prepare for the next call. */
1327 read_stream_look_ahead(stream);
1328
1329#ifndef READ_STREAM_DISABLE_FAST_PATH
1330 /* See if we can take the fast path for all-cached scans next time. */
1331 if (stream->ios_in_progress == 0 &&
1332 stream->forwarded_buffers == 0 &&
1333 stream->pinned_buffers == 1 &&
1334 stream->readahead_distance == 1 &&
1335 stream->combine_distance == 1 &&
1336 stream->pending_read_nblocks == 0 &&
1337 stream->per_buffer_data_size == 0)
1338 {
1339 /*
1340 * The fast path spins on one buffer entry repeatedly instead of
1341 * rotating through the whole queue and clearing the entries behind
1342 * it. If the buffer it starts with happened to be forwarded between
1343 * StartReadBuffers() calls and also wrapped around the circular queue
1344 * partway through, then a copy also exists in the overflow zone, and
1345 * it won't clear it out as the regular path would. Do that now, so
1346 * it doesn't need code for that.
1347 */
1348 if (stream->oldest_buffer_index < stream->io_combine_limit - 1)
1349 stream->buffers[stream->queue_size + stream->oldest_buffer_index] =
1351
1352 stream->fast_path = true;
1353 }
1354#endif
1355
1356 return buffer;
1357}
uint32 BlockNumber
Definition block.h:31
int Buffer
Definition buf.h:23
#define InvalidBuffer
Definition buf.h:25
bool WaitReadBuffers(ReadBuffersOperation *operation)
Definition bufmgr.c:1750
bool StartReadBuffer(ReadBuffersOperation *operation, Buffer *buffer, BlockNumber blocknum, int flags)
Definition bufmgr.c:1628
int io_combine_limit
Definition bufmgr.c:215
#define READ_BUFFERS_ISSUE_ADVICE
Definition bufmgr.h:124
#define READ_BUFFERS_SYNCHRONOUSLY
Definition bufmgr.h:128
static bool BufferIsValid(Buffer bufnum)
Definition bufmgr.h:419
#define Min(x, y)
Definition c.h:1091
#define likely(x)
Definition c.h:437
#define Assert(condition)
Definition c.h:943
int16_t int16
Definition c.h:619
int32_t int32
Definition c.h:620
#define unlikely(x)
Definition c.h:438
#define VALGRIND_MAKE_MEM_NOACCESS(addr, size)
Definition memdebug.h:27
static void * get_per_buffer_data(ReadStream *stream, int16 buffer_index)
static void read_stream_look_ahead(ReadStream *stream)
static void read_stream_count_wait(ReadStream *stream)
static void read_stream_count_io(ReadStream *stream, int nblocks, int in_progress)
static void read_stream_count_prefetch(ReadStream *stream)
int16 buffer_index
Definition read_stream.c:88
BlockNumber blocknum
Definition bufmgr.h:146
int16 io_combine_limit
Definition read_stream.c:98
uint16 distance_decay_holdoff
int16 ios_in_progress
Definition read_stream.c:99
void * per_buffer_data
BlockNumber seq_until_processed
int16 pinned_buffers
int16 max_ios
Definition read_stream.c:97
int16 oldest_buffer_index
BlockNumber seq_blocknum
bool advice_enabled
int16 oldest_io_index
int16 combine_distance
int16 readahead_distance
int read_buffers_flags
int16 queue_size
int16 next_buffer_index
int16 initialized_buffers
size_t per_buffer_data_size
int16 forwarded_buffers
int16 next_io_index
int16 pending_read_nblocks
Buffer buffers[FLEXIBLE_ARRAY_MEMBER]

References ReadStream::advice_enabled, Assert, ReadBuffersOperation::blocknum, InProgressIO::buffer_index, BufferIsValid(), ReadStream::buffers, ReadBuffersOperation::buffers, ReadStream::combine_distance, ReadStream::distance_decay_holdoff, ReadStream::fast_path, fb(), ReadBuffersOperation::flags, ReadStream::forwarded_buffers, get_per_buffer_data(), ReadStream::initialized_buffers, InvalidBlockNumber, InvalidBuffer, ReadStream::io_combine_limit, io_combine_limit, ReadStream::ios, ReadStream::ios_in_progress, likely, ReadStream::max_ios, ReadStream::max_pinned_buffers, Min, ReadStream::next_buffer_index, ReadStream::next_io_index, ReadStream::oldest_buffer_index, ReadStream::oldest_io_index, InProgressIO::op, ReadStream::pending_read_nblocks, ReadStream::per_buffer_data, ReadStream::per_buffer_data_size, ReadStream::pinned_buffers, ReadStream::queue_size, ReadStream::read_buffers_flags, READ_BUFFERS_ISSUE_ADVICE, READ_BUFFERS_SYNCHRONOUSLY, read_stream_count_io(), read_stream_count_prefetch(), read_stream_count_wait(), read_stream_get_block(), read_stream_look_ahead(), ReadStream::readahead_distance, ReadStream::seq_blocknum, ReadStream::seq_until_processed, StartReadBuffer(), unlikely, VALGRIND_MAKE_MEM_NOACCESS, and WaitReadBuffers().

Referenced by autoprewarm_database_main(), BitmapHeapScanNextBlock(), blbulkdelete(), blgetbitmap(), blvacuumcleanup(), brin_vacuum_scan(), btvacuumscan(), collect_corrupt_items(), collect_visibility_data(), ginvacuumcleanup(), gistvacuumscan(), hashbulkdelete(), heap_fetch_next_buffer(), heapam_scan_analyze_next_block(), lazy_scan_heap(), lazy_vacuum_heap_rel(), pg_prewarm(), pgstathashindex(), pgstatindex_impl(), read_stream_for_blocks(), read_stream_reset(), RelationCopyStorageUsingBuffer(), spgvacuumscan(), statapprox_heap(), and verify_heapam().

◆ read_stream_pause()

BlockNumber read_stream_pause ( ReadStream stream)
extern

◆ read_stream_reset()

void read_stream_reset ( ReadStream stream)
extern

Definition at line 1407 of file read_stream.c.

1408{
1409 int16 index;
1410 Buffer buffer;
1411
1412 /* Stop looking ahead. */
1413 stream->readahead_distance = 0;
1414 stream->combine_distance = 0;
1415
1416 /* Forget buffered block number and fast path state. */
1418 stream->fast_path = false;
1419
1420 /* Unpin anything that wasn't consumed. */
1421 while ((buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
1422 ReleaseBuffer(buffer);
1423
1424 /* Unpin any unused forwarded buffers. */
1425 index = stream->next_buffer_index;
1426 while (index < stream->initialized_buffers &&
1427 (buffer = stream->buffers[index]) != InvalidBuffer)
1428 {
1429 Assert(stream->forwarded_buffers > 0);
1430 stream->forwarded_buffers--;
1431 ReleaseBuffer(buffer);
1432
1433 stream->buffers[index] = InvalidBuffer;
1435 stream->buffers[stream->queue_size + index] = InvalidBuffer;
1436
1437 if (++index == stream->queue_size)
1438 index = 0;
1439 }
1440
1441 Assert(stream->forwarded_buffers == 0);
1442 Assert(stream->pinned_buffers == 0);
1443 Assert(stream->ios_in_progress == 0);
1444
1445 /* Start off assuming data is cached. */
1446 stream->readahead_distance = 1;
1447 stream->combine_distance = 1;
1449 stream->resume_combine_distance = stream->combine_distance;
1450 stream->distance_decay_holdoff = 0;
1451}
void ReleaseBuffer(Buffer buffer)
Definition bufmgr.c:5586
Buffer read_stream_next_buffer(ReadStream *stream, void **per_buffer_data)
BlockNumber buffered_blocknum
Definition type.h:96

References Assert, ReadStream::buffered_blocknum, ReadStream::buffers, ReadStream::combine_distance, ReadStream::distance_decay_holdoff, ReadStream::fast_path, fb(), ReadStream::forwarded_buffers, InvalidBlockNumber, InvalidBuffer, io_combine_limit, ReadStream::ios_in_progress, ReadStream::next_buffer_index, ReadStream::pinned_buffers, ReadStream::queue_size, read_stream_next_buffer(), ReadStream::readahead_distance, ReleaseBuffer(), ReadStream::resume_combine_distance, and ReadStream::resume_readahead_distance.

Referenced by btvacuumscan(), gistvacuumscan(), hashbulkdelete(), heap_fetch_next_buffer(), heap_rescan(), read_stream_end(), and spgvacuumscan().

◆ read_stream_resume()