Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * test_aio.c
4 : : * Helpers to write tests for AIO
5 : : *
6 : : * This module provides interface functions for C functionality to SQL, to
7 : : * make it possible to test AIO related behavior in a targeted way from SQL.
8 : : * It'd not generally be safe to export these functions to SQL, but for a test
9 : : * that's fine.
10 : : *
11 : : * Copyright (c) 2020-2026, PostgreSQL Global Development Group
12 : : *
13 : : * IDENTIFICATION
14 : : * src/test/modules/test_aio/test_aio.c
15 : : *
16 : : *-------------------------------------------------------------------------
17 : : */
18 : :
19 : : #include "postgres.h"
20 : :
21 : : #include "access/relation.h"
22 : : #include "catalog/pg_type.h"
23 : : #include "fmgr.h"
24 : : #include "funcapi.h"
25 : : #include "storage/aio.h"
26 : : #include "storage/aio_internal.h"
27 : : #include "storage/buf_internals.h"
28 : : #include "storage/bufmgr.h"
29 : : #include "storage/checksum.h"
30 : : #include "storage/condition_variable.h"
31 : : #include "storage/lwlock.h"
32 : : #include "storage/proc.h"
33 : : #include "storage/procnumber.h"
34 : : #include "storage/read_stream.h"
35 : : #include "utils/array.h"
36 : : #include "utils/builtins.h"
37 : : #include "utils/injection_point.h"
38 : : #include "utils/rel.h"
39 : : #include "utils/tuplestore.h"
40 : : #include "utils/wait_event.h"
41 : :
42 : :
43 : 7 : PG_MODULE_MAGIC;
44 : :
45 : :
46 : : /* In shared memory */
47 : : typedef struct InjIoErrorState
48 : : {
49 : : ConditionVariable cv;
50 : :
51 : : bool enabled_short_read;
52 : : bool enabled_reopen;
53 : :
54 : : bool enabled_completion_wait;
55 : : Oid completion_wait_relfilenode;
56 : : BlockNumber completion_wait_blockno;
57 : : pid_t completion_wait_pid;
58 : : uint32 completion_wait_event;
59 : :
60 : : bool short_read_result_set;
61 : : Oid short_read_relfilenode;
62 : : pid_t short_read_pid;
63 : : int short_read_result;
64 : : } InjIoErrorState;
65 : :
66 : : typedef struct BlocksReadStreamData
67 : : {
68 : : int nblocks;
69 : : int curblock;
70 : : uint32 *blocks;
71 : : } BlocksReadStreamData;
72 : :
73 : :
74 : : static InjIoErrorState *inj_io_error_state;
75 : :
76 : : /* Shared memory init callbacks */
77 : : static void test_aio_shmem_request(void *arg);
78 : : static void test_aio_shmem_init(void *arg);
79 : : static void test_aio_shmem_attach(void *arg);
80 : :
81 : : static const ShmemCallbacks inj_io_shmem_callbacks = {
82 : : .request_fn = test_aio_shmem_request,
83 : : .init_fn = test_aio_shmem_init,
84 : : .attach_fn = test_aio_shmem_attach,
85 : : };
86 : :
87 : :
88 : : static PgAioHandle *last_handle;
89 : :
90 : :
91 : :
92 : : static void
93 : 7 : test_aio_shmem_request(void *arg)
94 : : {
95 : 7 : ShmemRequestStruct(.name = "test_aio injection points",
96 : : .size = sizeof(InjIoErrorState),
97 : : .ptr = (void **) &inj_io_error_state,
98 : : );
99 : 7 : }
100 : :
101 : : static void
102 : 7 : test_aio_shmem_init(void *arg)
103 : : {
104 : : /* First time through, initialize */
105 : 7 : inj_io_error_state->enabled_short_read = false;
106 : 7 : inj_io_error_state->enabled_reopen = false;
107 : 7 : inj_io_error_state->enabled_completion_wait = false;
108 : :
109 : 7 : ConditionVariableInit(&inj_io_error_state->cv);
110 : 7 : inj_io_error_state->completion_wait_event = WaitEventInjectionPointNew("completion_wait");
111 : :
112 : : #ifdef USE_INJECTION_POINTS
113 : 7 : InjectionPointAttach("aio-process-completion-before-shared",
114 : : "test_aio",
115 : : "inj_io_completion_hook",
116 : : NULL,
117 : : 0);
118 : 7 : InjectionPointLoad("aio-process-completion-before-shared");
119 : :
120 : 7 : InjectionPointAttach("aio-worker-after-reopen",
121 : : "test_aio",
122 : : "inj_io_reopen",
123 : : NULL,
124 : : 0);
125 : 7 : InjectionPointLoad("aio-worker-after-reopen");
126 : :
127 : : #endif
128 : 7 : }
129 : :
130 : : static void
131 : 0 : test_aio_shmem_attach(void *arg)
132 : : {
133 : : /*
134 : : * Pre-load the injection points now, so we can call them in a critical
135 : : * section.
136 : : */
137 : : #ifdef USE_INJECTION_POINTS
138 : 0 : InjectionPointLoad("aio-process-completion-before-shared");
139 : 0 : InjectionPointLoad("aio-worker-after-reopen");
140 [ # # ]: 0 : elog(LOG, "injection point loaded");
141 : : #endif
142 : 0 : }
143 : :
144 : : void
145 : 7 : _PG_init(void)
146 : : {
147 [ - + ]: 7 : if (!process_shared_preload_libraries_in_progress)
148 : 0 : return;
149 : :
150 : 7 : RegisterShmemCallbacks(&inj_io_shmem_callbacks);
151 : : }
152 : :
153 : :
154 : 9 : PG_FUNCTION_INFO_V1(errno_from_string);
155 : : Datum
156 : 6 : errno_from_string(PG_FUNCTION_ARGS)
157 : : {
158 : 6 : const char *sym = text_to_cstring(PG_GETARG_TEXT_PP(0));
159 : :
160 [ + + ]: 6 : if (strcmp(sym, "EIO") == 0)
161 : 4 : PG_RETURN_INT32(EIO);
162 [ - + ]: 2 : else if (strcmp(sym, "EAGAIN") == 0)
163 : 0 : PG_RETURN_INT32(EAGAIN);
164 [ - + ]: 2 : else if (strcmp(sym, "EINTR") == 0)
165 : 0 : PG_RETURN_INT32(EINTR);
166 [ - + ]: 2 : else if (strcmp(sym, "ENOSPC") == 0)
167 : 0 : PG_RETURN_INT32(ENOSPC);
168 [ + - ]: 2 : else if (strcmp(sym, "EROFS") == 0)
169 : 2 : PG_RETURN_INT32(EROFS);
170 : :
171 [ # # ]: 0 : ereport(ERROR,
172 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
173 : : errmsg_internal("%s is not a supported errno value", sym));
174 : : PG_RETURN_INT32(0);
175 : : }
176 : :
177 : 9 : PG_FUNCTION_INFO_V1(grow_rel);
178 : : Datum
179 : 6 : grow_rel(PG_FUNCTION_ARGS)
180 : : {
181 : 6 : Oid relid = PG_GETARG_OID(0);
182 : 6 : uint32 nblocks = PG_GETARG_UINT32(1);
183 : : Relation rel;
184 : : #define MAX_BUFFERS_TO_EXTEND_BY 64
185 : : Buffer victim_buffers[MAX_BUFFERS_TO_EXTEND_BY];
186 : :
187 : 6 : rel = relation_open(relid, AccessExclusiveLock);
188 : :
189 [ + + ]: 12 : while (nblocks > 0)
190 : : {
191 : : uint32 extend_by_pages;
192 : :
193 : 6 : extend_by_pages = Min(nblocks, MAX_BUFFERS_TO_EXTEND_BY);
194 : :
195 : 6 : ExtendBufferedRelBy(BMR_REL(rel),
196 : : MAIN_FORKNUM,
197 : : NULL,
198 : : 0,
199 : : extend_by_pages,
200 : : victim_buffers,
201 : : &extend_by_pages);
202 : :
203 : 6 : nblocks -= extend_by_pages;
204 : :
205 [ + + ]: 78 : for (uint32 i = 0; i < extend_by_pages; i++)
206 : : {
207 : 72 : ReleaseBuffer(victim_buffers[i]);
208 : : }
209 : : }
210 : :
211 : 6 : relation_close(rel, NoLock);
212 : :
213 : 6 : PG_RETURN_VOID();
214 : : }
215 : :
216 : 19 : PG_FUNCTION_INFO_V1(modify_rel_block);
217 : : Datum
218 : 70 : modify_rel_block(PG_FUNCTION_ARGS)
219 : : {
220 : 70 : Oid relid = PG_GETARG_OID(0);
221 : 70 : BlockNumber blkno = PG_GETARG_UINT32(1);
222 : 70 : bool zero = PG_GETARG_BOOL(2);
223 : 70 : bool corrupt_header = PG_GETARG_BOOL(3);
224 : 70 : bool corrupt_checksum = PG_GETARG_BOOL(4);
225 : 70 : Page page = palloc_aligned(BLCKSZ, PG_IO_ALIGN_SIZE, 0);
226 : : bool flushed;
227 : : Relation rel;
228 : : Buffer buf;
229 : : PageHeader ph;
230 : :
231 : 70 : rel = relation_open(relid, AccessExclusiveLock);
232 : :
233 : 70 : buf = ReadBufferExtended(rel, MAIN_FORKNUM, blkno,
234 : : RBM_ZERO_ON_ERROR, NULL);
235 : :
236 : 70 : LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
237 : :
238 : : /*
239 : : * copy the page to local memory, seems nicer than to directly modify in
240 : : * the buffer pool.
241 : : */
242 : 70 : memcpy(page, BufferGetPage(buf), BLCKSZ);
243 : :
244 : 70 : UnlockReleaseBuffer(buf);
245 : :
246 : : /*
247 : : * Don't want to have a buffer in-memory that's marked valid where the
248 : : * on-disk contents are invalid. Particularly not if the in-memory buffer
249 : : * could be dirty...
250 : : *
251 : : * While we hold an AEL on the relation nobody else should be able to read
252 : : * the buffer in.
253 : : *
254 : : * NB: This is probably racy, better don't copy this to non-test code.
255 : : */
256 [ + + ]: 70 : if (BufferIsLocal(buf))
257 : 18 : InvalidateLocalBuffer(GetLocalBufferDescriptor(-buf - 1), true);
258 : : else
259 : 52 : EvictUnpinnedBuffer(buf, &flushed);
260 : :
261 : : /*
262 : : * Now modify the page as asked for by the caller.
263 : : */
264 [ + + ]: 70 : if (zero)
265 : 16 : memset(page, 0, BufferGetPageSize(buf));
266 : :
267 [ + + + + : 70 : if (PageIsEmpty(page) && (corrupt_header || corrupt_checksum))
+ + ]
268 : 16 : PageInit(page, BufferGetPageSize(buf), 0);
269 : :
270 : 70 : ph = (PageHeader) page;
271 : :
272 [ + + ]: 70 : if (corrupt_header)
273 : 32 : ph->pd_special = BLCKSZ + 1;
274 : :
275 [ + + ]: 70 : if (corrupt_checksum)
276 : : {
277 : 26 : bool successfully_corrupted = 0;
278 : :
279 : : /*
280 : : * Any single modification of the checksum could just end up being
281 : : * valid again, due to e.g. corrupt_header changing the data in a way
282 : : * that'd result in the "corrupted" checksum, or the checksum already
283 : : * being invalid. Retry in that, unlikely, case.
284 : : */
285 [ + - ]: 26 : for (int i = 0; i < 100; i++)
286 : : {
287 : : uint16 verify_checksum;
288 : : uint16 old_checksum;
289 : :
290 : 26 : old_checksum = ph->pd_checksum;
291 : 26 : ph->pd_checksum = old_checksum + 1;
292 : :
293 [ + - ]: 26 : elog(LOG, "corrupting checksum of blk %u from %u to %u",
294 : : blkno, old_checksum, ph->pd_checksum);
295 : :
296 : 26 : verify_checksum = pg_checksum_page(page, blkno);
297 [ + - ]: 26 : if (verify_checksum != ph->pd_checksum)
298 : : {
299 : 26 : successfully_corrupted = true;
300 : 26 : break;
301 : : }
302 : : }
303 : :
304 [ - + ]: 26 : if (!successfully_corrupted)
305 [ # # ]: 0 : elog(ERROR, "could not corrupt checksum, what's going on?");
306 : : }
307 : : else
308 : : {
309 : 44 : PageSetChecksum(page, blkno);
310 : : }
311 : :
312 : 70 : smgrwrite(RelationGetSmgr(rel),
313 : : MAIN_FORKNUM, blkno, page, true);
314 : :
315 : 70 : relation_close(rel, NoLock);
316 : :
317 : 70 : PG_RETURN_VOID();
318 : : }
319 : :
320 : : /*
321 : : * Ensures a buffer for rel & blkno is in shared buffers, without actually
322 : : * caring about the buffer contents. Used to set up test scenarios.
323 : : */
324 : : static Buffer
325 : 160 : create_toy_buffer(Relation rel, BlockNumber blkno)
326 : : {
327 : : Buffer buf;
328 : : BufferDesc *buf_hdr;
329 : : uint64 buf_state;
330 : 160 : bool was_pinned = false;
331 : 160 : uint64 unset_bits = 0;
332 : :
333 : : /* place buffer in shared buffers without erroring out */
334 : 160 : buf = ReadBufferExtended(rel, MAIN_FORKNUM, blkno, RBM_ZERO_AND_LOCK, NULL);
335 : 160 : LockBuffer(buf, BUFFER_LOCK_UNLOCK);
336 : :
337 [ + + ]: 160 : if (RelationUsesLocalBuffers(rel))
338 : : {
339 : 50 : buf_hdr = GetLocalBufferDescriptor(-buf - 1);
340 : 50 : buf_state = pg_atomic_read_u64(&buf_hdr->state);
341 : : }
342 : : else
343 : : {
344 : 110 : buf_hdr = GetBufferDescriptor(buf - 1);
345 : 110 : buf_state = LockBufHdr(buf_hdr);
346 : : }
347 : :
348 : : /*
349 : : * We should be the only backend accessing this buffer. This is just a
350 : : * small bit of belt-and-suspenders defense, none of this code should ever
351 : : * run in a cluster with real data.
352 : : */
353 [ - + ]: 160 : if (BUF_STATE_GET_REFCOUNT(buf_state) > 1)
354 : 0 : was_pinned = true;
355 : : else
356 : 160 : unset_bits |= BM_VALID | BM_DIRTY;
357 : :
358 [ + + ]: 160 : if (RelationUsesLocalBuffers(rel))
359 : : {
360 : 50 : buf_state &= ~unset_bits;
361 : 50 : pg_atomic_unlocked_write_u64(&buf_hdr->state, buf_state);
362 : : }
363 : : else
364 : : {
365 : 110 : UnlockBufHdrExt(buf_hdr, buf_state, 0, unset_bits, 0);
366 : : }
367 : :
368 [ - + ]: 160 : if (was_pinned)
369 [ # # ]: 0 : elog(ERROR, "toy buffer %d was already pinned",
370 : : buf);
371 : :
372 : 160 : return buf;
373 : : }
374 : :
375 : : /*
376 : : * A "low level" read. This does similar things to what
377 : : * StartReadBuffers()/WaitReadBuffers() do, but provides more control (and
378 : : * less sanity).
379 : : */
380 : 27 : PG_FUNCTION_INFO_V1(read_rel_block_ll);
381 : : Datum
382 : 94 : read_rel_block_ll(PG_FUNCTION_ARGS)
383 : : {
384 : 94 : Oid relid = PG_GETARG_OID(0);
385 : 94 : BlockNumber blkno = PG_GETARG_UINT32(1);
386 : 94 : int nblocks = PG_GETARG_INT32(2);
387 : 94 : bool wait_complete = PG_GETARG_BOOL(3);
388 : 94 : bool batchmode_enter = PG_GETARG_BOOL(4);
389 : 94 : bool call_smgrreleaseall = PG_GETARG_BOOL(5);
390 : 94 : bool batchmode_exit = PG_GETARG_BOOL(6);
391 : 94 : bool zero_on_error = PG_GETARG_BOOL(7);
392 : : Relation rel;
393 : : Buffer bufs[PG_IOV_MAX];
394 : : BufferDesc *buf_hdrs[PG_IOV_MAX];
395 : : Page pages[PG_IOV_MAX];
396 : 94 : uint8 srb_flags = 0;
397 : : PgAioReturn ior;
398 : : PgAioHandle *ioh;
399 : : PgAioWaitRef iow;
400 : : SMgrRelation smgr;
401 : :
402 [ + - - + ]: 94 : if (nblocks <= 0 || nblocks > PG_IOV_MAX)
403 [ # # ]: 0 : elog(ERROR, "nblocks is out of range");
404 : :
405 : 94 : rel = relation_open(relid, AccessShareLock);
406 : :
407 [ + + ]: 246 : for (int i = 0; i < nblocks; i++)
408 : : {
409 : 152 : bufs[i] = create_toy_buffer(rel, blkno + i);
410 : 152 : pages[i] = BufferGetBlock(bufs[i]);
411 : 152 : buf_hdrs[i] = BufferIsLocal(bufs[i]) ?
412 [ + + ]: 152 : GetLocalBufferDescriptor(-bufs[i] - 1) :
413 : 104 : GetBufferDescriptor(bufs[i] - 1);
414 : : }
415 : :
416 : 94 : smgr = RelationGetSmgr(rel);
417 : :
418 : 94 : pgstat_prepare_report_checksum_failure(smgr->smgr_rlocator.locator.dbOid);
419 : :
420 : 94 : ioh = pgaio_io_acquire(CurrentResourceOwner, &ior);
421 : 94 : pgaio_io_get_wref(ioh, &iow);
422 : :
423 [ + + ]: 94 : if (RelationUsesLocalBuffers(rel))
424 : : {
425 [ + + ]: 76 : for (int i = 0; i < nblocks; i++)
426 : 48 : StartLocalBufferIO(buf_hdrs[i], true, true, NULL);
427 : 28 : pgaio_io_set_flag(ioh, PGAIO_HF_REFERENCES_LOCAL);
428 : : }
429 : : else
430 : : {
431 [ + + ]: 170 : for (int i = 0; i < nblocks; i++)
432 : 104 : StartSharedBufferIO(buf_hdrs[i], true, true, NULL);
433 : : }
434 : :
435 : 94 : pgaio_io_set_handle_data_32(ioh, (uint32 *) bufs, nblocks);
436 : :
437 [ + + ]: 94 : if (zero_on_error | zero_damaged_pages)
438 : 22 : srb_flags |= READ_BUFFERS_ZERO_ON_ERROR;
439 [ + + ]: 94 : if (ignore_checksum_failure)
440 : 10 : srb_flags |= READ_BUFFERS_IGNORE_CHECKSUM_FAILURES;
441 : :
442 : 94 : pgaio_io_register_callbacks(ioh,
443 [ + + ]: 94 : RelationUsesLocalBuffers(rel) ?
444 : : PGAIO_HCB_LOCAL_BUFFER_READV :
445 : : PGAIO_HCB_SHARED_BUFFER_READV,
446 : : srb_flags);
447 : :
448 [ + + ]: 94 : if (batchmode_enter)
449 : 4 : pgaio_enter_batchmode();
450 : :
451 : 94 : smgrstartreadv(ioh, smgr, MAIN_FORKNUM, blkno,
452 : : (void *) pages, nblocks);
453 : :
454 [ + + ]: 94 : if (call_smgrreleaseall)
455 : 4 : smgrreleaseall();
456 : :
457 [ + + ]: 94 : if (batchmode_exit)
458 : 4 : pgaio_exit_batchmode();
459 : :
460 [ + + ]: 246 : for (int i = 0; i < nblocks; i++)
461 : 152 : ReleaseBuffer(bufs[i]);
462 : :
463 [ + + ]: 94 : if (wait_complete)
464 : : {
465 : 60 : pgaio_wref_wait(&iow);
466 : :
467 [ + + ]: 60 : if (ior.result.status != PGAIO_RS_OK)
468 : 46 : pgaio_result_report(ior.result,
469 : : &ior.target_data,
470 [ + + ]: 46 : ior.result.status == PGAIO_RS_ERROR ?
471 : : ERROR : WARNING);
472 : : }
473 : :
474 : 70 : relation_close(rel, NoLock);
475 : :
476 : 70 : PG_RETURN_VOID();
477 : : }
478 : :
479 : : /* helper for invalidate_rel_block() and evict_rel() */
480 : : static void
481 : 469 : invalidate_one_block(Relation rel, ForkNumber forknum, BlockNumber blkno)
482 : : {
483 : : PrefetchBufferResult pr;
484 : : Buffer buf;
485 : :
486 : : /*
487 : : * This is a gross hack, but there's no other API exposed that allows to
488 : : * get a buffer ID without actually reading the block in.
489 : : */
490 : 469 : pr = PrefetchBuffer(rel, forknum, blkno);
491 : 469 : buf = pr.recent_buffer;
492 : :
493 [ + + ]: 469 : if (BufferIsValid(buf))
494 : : {
495 : : /* if the buffer contents aren't valid, this'll return false */
496 [ + + ]: 230 : if (ReadRecentBuffer(rel->rd_locator, forknum, blkno, buf))
497 : : {
498 : 224 : BufferDesc *buf_hdr = BufferIsLocal(buf) ?
499 : 140 : GetLocalBufferDescriptor(-buf - 1)
500 [ + + ]: 224 : : GetBufferDescriptor(buf - 1);
501 : : bool flushed;
502 : :
503 : 224 : LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
504 : :
505 [ + + ]: 224 : if (pg_atomic_read_u64(&buf_hdr->state) & BM_DIRTY)
506 : : {
507 [ + + ]: 124 : if (BufferIsLocal(buf))
508 : 74 : FlushLocalBuffer(buf_hdr, NULL);
509 : : else
510 : 50 : FlushOneBuffer(buf);
511 : : }
512 : 224 : UnlockReleaseBuffer(buf);
513 : :
514 [ + + ]: 224 : if (BufferIsLocal(buf))
515 : 140 : InvalidateLocalBuffer(GetLocalBufferDescriptor(-buf - 1), true);
516 [ - + ]: 84 : else if (!EvictUnpinnedBuffer(buf, &flushed))
517 [ # # ]: 0 : elog(ERROR, "couldn't evict");
518 : : }
519 : : }
520 : :
521 : 469 : }
522 : :
523 : 16 : PG_FUNCTION_INFO_V1(invalidate_rel_block);
524 : : Datum
525 : 157 : invalidate_rel_block(PG_FUNCTION_ARGS)
526 : : {
527 : 157 : Oid relid = PG_GETARG_OID(0);
528 : 157 : BlockNumber blkno = PG_GETARG_UINT32(1);
529 : : Relation rel;
530 : :
531 : 157 : rel = relation_open(relid, AccessExclusiveLock);
532 : :
533 : 157 : invalidate_one_block(rel, MAIN_FORKNUM, blkno);
534 : :
535 : 157 : relation_close(rel, AccessExclusiveLock);
536 : :
537 : 157 : PG_RETURN_VOID();
538 : : }
539 : :
540 : 13 : PG_FUNCTION_INFO_V1(evict_rel);
541 : : Datum
542 : 40 : evict_rel(PG_FUNCTION_ARGS)
543 : : {
544 : 40 : Oid relid = PG_GETARG_OID(0);
545 : : Relation rel;
546 : :
547 : 40 : rel = relation_open(relid, AccessExclusiveLock);
548 : :
549 : : /*
550 : : * EvictRelUnpinnedBuffers() doesn't support temp tables, so for temp
551 : : * tables we have to do it the expensive way and evict every possible
552 : : * buffer.
553 : : */
554 [ + + ]: 40 : if (RelationUsesLocalBuffers(rel))
555 : : {
556 : 12 : SMgrRelation smgr = RelationGetSmgr(rel);
557 : :
558 [ + + ]: 60 : for (int forknum = MAIN_FORKNUM; forknum <= MAX_FORKNUM; forknum++)
559 : : {
560 : : BlockNumber nblocks;
561 : :
562 [ + + ]: 48 : if (!smgrexists(smgr, forknum))
563 : 24 : continue;
564 : :
565 : 24 : nblocks = smgrnblocks(smgr, forknum);
566 : :
567 [ + + ]: 336 : for (BlockNumber blkno = 0; blkno < nblocks; blkno++)
568 : : {
569 : 312 : invalidate_one_block(rel, forknum, blkno);
570 : : }
571 : : }
572 : : }
573 : : else
574 : : {
575 : : int32 buffers_evicted,
576 : : buffers_flushed,
577 : : buffers_skipped;
578 : :
579 : 28 : EvictRelUnpinnedBuffers(rel, &buffers_evicted, &buffers_flushed,
580 : : &buffers_skipped);
581 : : }
582 : :
583 : 40 : relation_close(rel, AccessExclusiveLock);
584 : :
585 : :
586 : 40 : PG_RETURN_VOID();
587 : : }
588 : :
589 : 8 : PG_FUNCTION_INFO_V1(buffer_create_toy);
590 : : Datum
591 : 8 : buffer_create_toy(PG_FUNCTION_ARGS)
592 : : {
593 : 8 : Oid relid = PG_GETARG_OID(0);
594 : 8 : BlockNumber blkno = PG_GETARG_UINT32(1);
595 : : Relation rel;
596 : : Buffer buf;
597 : :
598 : 8 : rel = relation_open(relid, AccessExclusiveLock);
599 : :
600 : 8 : buf = create_toy_buffer(rel, blkno);
601 : 8 : ReleaseBuffer(buf);
602 : :
603 : 8 : relation_close(rel, NoLock);
604 : :
605 : 8 : PG_RETURN_INT32(buf);
606 : : }
607 : :
608 : 10 : PG_FUNCTION_INFO_V1(buffer_call_start_io);
609 : : Datum
610 : 22 : buffer_call_start_io(PG_FUNCTION_ARGS)
611 : : {
612 : 22 : Buffer buf = PG_GETARG_INT32(0);
613 : 22 : bool for_input = PG_GETARG_BOOL(1);
614 : 22 : bool wait = PG_GETARG_BOOL(2);
615 : : StartBufferIOResult result;
616 : : bool can_start;
617 : :
618 [ + + ]: 22 : if (BufferIsLocal(buf))
619 : 8 : result = StartLocalBufferIO(GetLocalBufferDescriptor(-buf - 1),
620 : : for_input, wait, NULL);
621 : : else
622 : 14 : result = StartSharedBufferIO(GetBufferDescriptor(buf - 1),
623 : : for_input, wait, NULL);
624 : :
625 : 22 : can_start = result == BUFFER_IO_READY_FOR_IO;
626 : :
627 : : /*
628 : : * For tests we don't want the resowner release preventing us from
629 : : * orchestrating odd scenarios.
630 : : */
631 [ + + + + ]: 22 : if (can_start && !BufferIsLocal(buf))
632 : 8 : ResourceOwnerForgetBufferIO(CurrentResourceOwner,
633 : : buf);
634 : :
635 [ + - ]: 22 : ereport(LOG,
636 : : errmsg("buffer %d after StartBufferIO: %s",
637 : : buf, DebugPrintBufferRefcount(buf)),
638 : : errhidestmt(true), errhidecontext(true));
639 : :
640 : 22 : PG_RETURN_BOOL(can_start);
641 : : }
642 : :
643 : 10 : PG_FUNCTION_INFO_V1(buffer_call_terminate_io);
644 : : Datum
645 : 12 : buffer_call_terminate_io(PG_FUNCTION_ARGS)
646 : : {
647 : 12 : Buffer buf = PG_GETARG_INT32(0);
648 : 12 : bool for_input = PG_GETARG_BOOL(1);
649 : 12 : bool succeed = PG_GETARG_BOOL(2);
650 : 12 : bool io_error = PG_GETARG_BOOL(3);
651 : 12 : bool release_aio = PG_GETARG_BOOL(4);
652 : 12 : bool clear_dirty = false;
653 : 12 : uint64 set_flag_bits = 0;
654 : :
655 [ - + ]: 12 : if (io_error)
656 : 0 : set_flag_bits |= BM_IO_ERROR;
657 : :
658 [ + - ]: 12 : if (for_input)
659 : : {
660 : 12 : clear_dirty = false;
661 : :
662 [ + + ]: 12 : if (succeed)
663 : 5 : set_flag_bits |= BM_VALID;
664 : : }
665 : : else
666 : : {
667 [ # # ]: 0 : if (succeed)
668 : 0 : clear_dirty = true;
669 : : }
670 : :
671 [ + - ]: 12 : ereport(LOG,
672 : : errmsg("buffer %d before Terminate[Local]BufferIO: %s",
673 : : buf, DebugPrintBufferRefcount(buf)),
674 : : errhidestmt(true), errhidecontext(true));
675 : :
676 [ + + ]: 12 : if (BufferIsLocal(buf))
677 : 4 : TerminateLocalBufferIO(GetLocalBufferDescriptor(-buf - 1),
678 : : clear_dirty, set_flag_bits, release_aio);
679 : : else
680 : 8 : TerminateBufferIO(GetBufferDescriptor(buf - 1),
681 : : clear_dirty, set_flag_bits, false, release_aio);
682 : :
683 [ + - ]: 12 : ereport(LOG,
684 : : errmsg("buffer %d after Terminate[Local]BufferIO: %s",
685 : : buf, DebugPrintBufferRefcount(buf)),
686 : : errhidestmt(true), errhidecontext(true));
687 : :
688 : 12 : PG_RETURN_VOID();
689 : : }
690 : :
691 : 9 : PG_FUNCTION_INFO_V1(read_buffers);
692 : : /*
693 : : * Infrastructure to test StartReadBuffers()
694 : : */
695 : : Datum
696 : 74 : read_buffers(PG_FUNCTION_ARGS)
697 : : {
698 : 74 : Oid relid = PG_GETARG_OID(0);
699 : 74 : BlockNumber startblock = PG_GETARG_UINT32(1);
700 : 74 : int32 nblocks = PG_GETARG_INT32(2);
701 : 74 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
702 : : Relation rel;
703 : : SMgrRelation smgr;
704 : 74 : int nblocks_done = 0;
705 : 74 : int nblocks_disp = 0;
706 : 74 : int nios = 0;
707 : : ReadBuffersOperation *operations;
708 : : Buffer *buffers;
709 : : Datum *buffers_datum;
710 : : bool *io_reqds;
711 : : int *nblocks_per_io;
712 : :
713 : : Assert(nblocks > 0);
714 : :
715 : 74 : InitMaterializedSRF(fcinfo, 0);
716 : :
717 : : /* at worst each block gets its own IO */
718 : 74 : operations = palloc0(sizeof(ReadBuffersOperation) * nblocks);
719 : 74 : buffers = palloc0(sizeof(Buffer) * nblocks);
720 : 74 : buffers_datum = palloc0(sizeof(Datum) * nblocks);
721 : 74 : io_reqds = palloc0(sizeof(bool) * nblocks);
722 : 74 : nblocks_per_io = palloc0(sizeof(int) * nblocks);
723 : :
724 : 74 : rel = relation_open(relid, AccessShareLock);
725 : 74 : smgr = RelationGetSmgr(rel);
726 : :
727 : : /*
728 : : * Do StartReadBuffers() until IO for all the required blocks has been
729 : : * started (if required).
730 : : */
731 [ + + ]: 214 : while (nblocks_done < nblocks)
732 : : {
733 : 140 : ReadBuffersOperation *operation = &operations[nios];
734 : 140 : int nblocks_this_io =
735 : 140 : Min(nblocks - nblocks_done, io_combine_limit);
736 : :
737 : 140 : operation->rel = rel;
738 : 140 : operation->smgr = smgr;
739 : 140 : operation->persistence = rel->rd_rel->relpersistence;
740 : 140 : operation->strategy = NULL;
741 : 140 : operation->forknum = MAIN_FORKNUM;
742 : :
743 : 280 : io_reqds[nios] = StartReadBuffers(operation,
744 : 140 : &buffers[nblocks_done],
745 : : startblock + nblocks_done,
746 : : &nblocks_this_io,
747 : : 0);
748 : 140 : nblocks_per_io[nios] = nblocks_this_io;
749 : 140 : nios++;
750 : 140 : nblocks_done += nblocks_this_io;
751 : : }
752 : :
753 : : /*
754 : : * Now wait for all operations that required IO. This is done at the end,
755 : : * as otherwise waiting for IO in progress in other backends could
756 : : * influence the result for subsequent buffers / blocks.
757 : : */
758 [ + + ]: 214 : for (int nio = 0; nio < nios; nio++)
759 : : {
760 : 140 : ReadBuffersOperation *operation = &operations[nio];
761 : :
762 [ + + ]: 140 : if (io_reqds[nio])
763 : 75 : WaitReadBuffers(operation);
764 : : }
765 : :
766 : : /*
767 : : * Convert what has been done into SQL SRF return value.
768 : : */
769 [ + + ]: 214 : for (int nio = 0; nio < nios; nio++)
770 : : {
771 : 140 : ReadBuffersOperation *operation = &operations[nio];
772 : 140 : int nblocks_this_io = nblocks_per_io[nio];
773 : 140 : Datum values[6] = {0};
774 : 140 : bool nulls[6] = {0};
775 : : ArrayType *buffers_arr;
776 : :
777 : : /* convert buffer array to datum array */
778 [ + + ]: 330 : for (int i = 0; i < nblocks_this_io; i++)
779 : : {
780 : 190 : Buffer buf = buffers[nblocks_disp + i];
781 : :
782 : : Assert(BufferGetBlockNumber(buf) == startblock + nblocks_disp + i);
783 : :
784 : 190 : buffers_datum[nblocks_disp + i] = Int32GetDatum(buf);
785 : : }
786 : :
787 : 140 : buffers_arr = construct_array_builtin(&buffers_datum[nblocks_disp],
788 : : nblocks_this_io,
789 : : INT4OID);
790 : :
791 : : /* blockoff */
792 : 140 : values[0] = Int32GetDatum(nblocks_disp);
793 : 140 : nulls[0] = false;
794 : :
795 : : /* blocknum */
796 : 140 : values[1] = UInt32GetDatum(startblock + nblocks_disp);
797 : 140 : nulls[1] = false;
798 : :
799 : : /* io_reqd */
800 : 140 : values[2] = BoolGetDatum(io_reqds[nio]);
801 : 140 : nulls[2] = false;
802 : :
803 : : /* foreign IO - only valid when IO was required */
804 [ + + ]: 140 : values[3] = BoolGetDatum(io_reqds[nio] ? operation->foreign_io : false);
805 : 140 : nulls[3] = false;
806 : :
807 : : /* nblocks */
808 : 140 : values[4] = Int32GetDatum(nblocks_this_io);
809 : 140 : nulls[4] = false;
810 : :
811 : : /* array of buffers */
812 : 140 : values[5] = PointerGetDatum(buffers_arr);
813 : 140 : nulls[5] = false;
814 : :
815 : 140 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
816 : :
817 : 140 : nblocks_disp += nblocks_this_io;
818 : : }
819 : :
820 : : /* release pins on all the buffers */
821 [ + + ]: 264 : for (int i = 0; i < nblocks_done; i++)
822 : 190 : ReleaseBuffer(buffers[i]);
823 : :
824 : : /*
825 : : * Free explicitly, to have a chance to detect potential issues with too
826 : : * long lived references to the operation.
827 : : */
828 : 74 : pfree(operations);
829 : 74 : pfree(buffers);
830 : 74 : pfree(buffers_datum);
831 : 74 : pfree(io_reqds);
832 : 74 : pfree(nblocks_per_io);
833 : :
834 : 74 : relation_close(rel, NoLock);
835 : :
836 : 74 : return (Datum) 0;
837 : : }
838 : :
839 : :
840 : : static BlockNumber
841 : 104 : read_stream_for_blocks_cb(ReadStream *stream,
842 : : void *callback_private_data,
843 : : void *per_buffer_data)
844 : : {
845 : 104 : BlocksReadStreamData *stream_data = callback_private_data;
846 : :
847 [ + + ]: 104 : if (stream_data->curblock >= stream_data->nblocks)
848 : 16 : return InvalidBlockNumber;
849 : 88 : return stream_data->blocks[stream_data->curblock++];
850 : : }
851 : :
852 : 9 : PG_FUNCTION_INFO_V1(read_stream_for_blocks);
853 : : Datum
854 : 16 : read_stream_for_blocks(PG_FUNCTION_ARGS)
855 : : {
856 : 16 : Oid relid = PG_GETARG_OID(0);
857 : 16 : ArrayType *blocksarray = PG_GETARG_ARRAYTYPE_P(1);
858 : 16 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
859 : : Relation rel;
860 : : BlocksReadStreamData stream_data;
861 : : ReadStream *stream;
862 : :
863 : 16 : InitMaterializedSRF(fcinfo, 0);
864 : :
865 : : /*
866 : : * We expect the input to be an N-element int4 array; verify that. We
867 : : * don't need to use deconstruct_array() since the array data is just
868 : : * going to look like a C array of N int4 values.
869 : : */
870 [ + - ]: 16 : if (ARR_NDIM(blocksarray) != 1 ||
871 [ + - ]: 16 : ARR_HASNULL(blocksarray) ||
872 [ - + ]: 16 : ARR_ELEMTYPE(blocksarray) != INT4OID)
873 [ # # ]: 0 : elog(ERROR, "expected 1 dimensional int4 array");
874 : :
875 : 16 : stream_data.curblock = 0;
876 : 16 : stream_data.nblocks = ARR_DIMS(blocksarray)[0];
877 [ - + ]: 16 : stream_data.blocks = (uint32 *) ARR_DATA_PTR(blocksarray);
878 : :
879 : 16 : rel = relation_open(relid, AccessShareLock);
880 : :
881 : 16 : stream = read_stream_begin_relation(READ_STREAM_FULL,
882 : : NULL,
883 : : rel,
884 : : MAIN_FORKNUM,
885 : : read_stream_for_blocks_cb,
886 : : &stream_data,
887 : : 0);
888 : :
889 [ + + ]: 104 : for (int i = 0; i < stream_data.nblocks; i++)
890 : : {
891 : 88 : Buffer buf = read_stream_next_buffer(stream, NULL);
892 : 88 : Datum values[3] = {0};
893 : 88 : bool nulls[3] = {0};
894 : :
895 [ - + ]: 88 : if (!BufferIsValid(buf))
896 [ # # ]: 0 : elog(ERROR, "read_stream_next_buffer() call %d is unexpectedly invalid", i);
897 : :
898 : 88 : values[0] = Int32GetDatum(i);
899 : 88 : values[1] = UInt32GetDatum(stream_data.blocks[i]);
900 : 88 : values[2] = UInt32GetDatum(buf);
901 : :
902 : 88 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
903 : :
904 : 88 : ReleaseBuffer(buf);
905 : : }
906 : :
907 [ - + ]: 16 : if (read_stream_next_buffer(stream, NULL) != InvalidBuffer)
908 [ # # ]: 0 : elog(ERROR, "read_stream_next_buffer() call %d is unexpectedly valid",
909 : : stream_data.nblocks);
910 : :
911 : 16 : read_stream_end(stream);
912 : :
913 : 16 : relation_close(rel, NoLock);
914 : :
915 : 16 : return (Datum) 0;
916 : : }
917 : :
918 : :
919 : 7 : PG_FUNCTION_INFO_V1(handle_get);
920 : : Datum
921 : 12 : handle_get(PG_FUNCTION_ARGS)
922 : : {
923 : 12 : last_handle = pgaio_io_acquire(CurrentResourceOwner, NULL);
924 : :
925 : 12 : PG_RETURN_VOID();
926 : : }
927 : :
928 : 7 : PG_FUNCTION_INFO_V1(handle_release_last);
929 : : Datum
930 : 4 : handle_release_last(PG_FUNCTION_ARGS)
931 : : {
932 [ - + ]: 4 : if (!last_handle)
933 [ # # ]: 0 : elog(ERROR, "no handle");
934 : :
935 : 4 : pgaio_io_release(last_handle);
936 : :
937 : 2 : PG_RETURN_VOID();
938 : : }
939 : :
940 : 7 : PG_FUNCTION_INFO_V1(handle_get_and_error);
941 : : Datum
942 : 6 : handle_get_and_error(PG_FUNCTION_ARGS)
943 : : {
944 : 6 : pgaio_io_acquire(CurrentResourceOwner, NULL);
945 : :
946 [ + - ]: 6 : elog(ERROR, "as you command");
947 : : PG_RETURN_VOID();
948 : : }
949 : :
950 : 7 : PG_FUNCTION_INFO_V1(handle_get_twice);
951 : : Datum
952 : 2 : handle_get_twice(PG_FUNCTION_ARGS)
953 : : {
954 : 2 : pgaio_io_acquire(CurrentResourceOwner, NULL);
955 : 2 : pgaio_io_acquire(CurrentResourceOwner, NULL);
956 : :
957 : 0 : PG_RETURN_VOID();
958 : : }
959 : :
960 : 7 : PG_FUNCTION_INFO_V1(handle_get_release);
961 : : Datum
962 : 6 : handle_get_release(PG_FUNCTION_ARGS)
963 : : {
964 : : PgAioHandle *handle;
965 : :
966 : 6 : handle = pgaio_io_acquire(CurrentResourceOwner, NULL);
967 : 6 : pgaio_io_release(handle);
968 : :
969 : 6 : PG_RETURN_VOID();
970 : : }
971 : :
972 : 7 : PG_FUNCTION_INFO_V1(batch_start);
973 : : Datum
974 : 6 : batch_start(PG_FUNCTION_ARGS)
975 : : {
976 : 6 : pgaio_enter_batchmode();
977 : 6 : PG_RETURN_VOID();
978 : : }
979 : :
980 : 7 : PG_FUNCTION_INFO_V1(batch_end);
981 : : Datum
982 : 2 : batch_end(PG_FUNCTION_ARGS)
983 : : {
984 : 2 : pgaio_exit_batchmode();
985 : 2 : PG_RETURN_VOID();
986 : : }
987 : :
988 : : #ifdef USE_INJECTION_POINTS
989 : : extern PGDLLEXPORT void inj_io_completion_hook(const char *name,
990 : : const void *private_data,
991 : : void *arg);
992 : : extern PGDLLEXPORT void inj_io_reopen(const char *name,
993 : : const void *private_data,
994 : : void *arg);
995 : :
996 : : static bool
997 : 2155 : inj_io_short_read_matches(PgAioHandle *ioh)
998 : : {
999 : : PGPROC *io_proc;
1000 : : int32 io_pid;
1001 : : int32 inj_pid;
1002 : : PgAioTargetData *td;
1003 : :
1004 [ + + ]: 2155 : if (!inj_io_error_state->enabled_short_read)
1005 : 1990 : return false;
1006 : :
1007 [ - + ]: 165 : if (!inj_io_error_state->short_read_result_set)
1008 : 0 : return false;
1009 : :
1010 : 165 : io_proc = GetPGProcByNumber(pgaio_io_get_owner(ioh));
1011 : 165 : io_pid = io_proc->pid;
1012 : 165 : inj_pid = inj_io_error_state->short_read_pid;
1013 : :
1014 [ + + + + ]: 165 : if (inj_pid != InvalidPid && inj_pid != io_pid)
1015 : 8 : return false;
1016 : :
1017 : 157 : td = pgaio_io_get_target_data(ioh);
1018 : :
1019 [ + + ]: 157 : if (inj_io_error_state->short_read_relfilenode != InvalidOid &&
1020 [ - + ]: 2 : td->smgr.rlocator.relNumber != inj_io_error_state->short_read_relfilenode)
1021 : 0 : return false;
1022 : :
1023 : : /*
1024 : : * Only shorten reads that are actually longer than the target size,
1025 : : * otherwise we can trigger over-reads.
1026 : : */
1027 [ + + ]: 157 : if (inj_io_error_state->short_read_result >= ioh->result)
1028 : 36 : return false;
1029 : :
1030 : 121 : return true;
1031 : : }
1032 : :
1033 : : static bool
1034 : 2175 : inj_io_completion_wait_matches(PgAioHandle *ioh)
1035 : : {
1036 : : PGPROC *io_proc;
1037 : : int32 io_pid;
1038 : : PgAioTargetData *td;
1039 : : int32 inj_pid;
1040 : : BlockNumber io_blockno;
1041 : : BlockNumber inj_blockno;
1042 : : Oid inj_relfilenode;
1043 : :
1044 [ + + ]: 2175 : if (!inj_io_error_state->enabled_completion_wait)
1045 : 2038 : return false;
1046 : :
1047 : 137 : io_proc = GetPGProcByNumber(pgaio_io_get_owner(ioh));
1048 : 137 : io_pid = io_proc->pid;
1049 : 137 : inj_pid = inj_io_error_state->completion_wait_pid;
1050 : :
1051 [ + - + + ]: 137 : if (inj_pid != InvalidPid && inj_pid != io_pid)
1052 : 107 : return false;
1053 : :
1054 : 30 : td = pgaio_io_get_target_data(ioh);
1055 : :
1056 : 30 : inj_relfilenode = inj_io_error_state->completion_wait_relfilenode;
1057 [ + - ]: 30 : if (inj_relfilenode != InvalidOid &&
1058 [ + + ]: 30 : td->smgr.rlocator.relNumber != inj_relfilenode)
1059 : 10 : return false;
1060 : :
1061 : 20 : inj_blockno = inj_io_error_state->completion_wait_blockno;
1062 : 20 : io_blockno = td->smgr.blockNum;
1063 [ + + + - ]: 20 : if (inj_blockno != InvalidBlockNumber &&
1064 [ - + ]: 8 : !(inj_blockno >= io_blockno && inj_blockno < (io_blockno + td->smgr.nblocks)))
1065 : 0 : return false;
1066 : :
1067 : 20 : return true;
1068 : : }
1069 : :
1070 : : static void
1071 : 2155 : inj_io_completion_wait_hook(const char *name, const void *private_data, void *arg)
1072 : : {
1073 : 2155 : PgAioHandle *ioh = (PgAioHandle *) arg;
1074 : :
1075 [ + + ]: 2155 : if (!inj_io_completion_wait_matches(ioh))
1076 : 2145 : return;
1077 : :
1078 : 10 : ConditionVariablePrepareToSleep(&inj_io_error_state->cv);
1079 : :
1080 : : while (true)
1081 : : {
1082 [ + + ]: 20 : if (!inj_io_completion_wait_matches(ioh))
1083 : 10 : break;
1084 : :
1085 : 10 : ConditionVariableSleep(&inj_io_error_state->cv,
1086 : 10 : inj_io_error_state->completion_wait_event);
1087 : : }
1088 : :
1089 : 10 : ConditionVariableCancelSleep();
1090 : : }
1091 : :
1092 : : static void
1093 : 2155 : inj_io_short_read_hook(const char *name, const void *private_data, void *arg)
1094 : : {
1095 : 2155 : PgAioHandle *ioh = (PgAioHandle *) arg;
1096 : :
1097 [ + - ]: 2155 : ereport(LOG,
1098 : : errmsg("short read injection point called, is enabled: %d",
1099 : : inj_io_error_state->enabled_short_read),
1100 : : errhidestmt(true), errhidecontext(true));
1101 : :
1102 [ + + ]: 2155 : if (inj_io_short_read_matches(ioh))
1103 : : {
1104 : 121 : struct iovec *iov = &pgaio_ctl->iovecs[ioh->iovec_off];
1105 : 121 : int32 old_result = ioh->result;
1106 : 121 : int32 new_result = inj_io_error_state->short_read_result;
1107 : 121 : int32 processed = 0;
1108 : :
1109 [ + - ]: 121 : ereport(LOG,
1110 : : errmsg("short read inject point, changing result from %d to %d",
1111 : : old_result, new_result),
1112 : : errhidestmt(true), errhidecontext(true));
1113 : :
1114 : : /*
1115 : : * The underlying IO actually completed OK, and thus the "invalid"
1116 : : * portion of the IOV actually contains valid data. That can hide a
1117 : : * lot of problems, e.g. if we were to wrongly mark a buffer, that
1118 : : * wasn't read according to the shortened-read, IO as valid, the
1119 : : * contents would look valid and we might miss a bug.
1120 : : *
1121 : : * To avoid that, iterate through the IOV and zero out the "failed"
1122 : : * portion of the IO.
1123 : : */
1124 [ + + ]: 246 : for (int i = 0; i < ioh->op_data.read.iov_length; i++)
1125 : : {
1126 [ + + ]: 125 : if (processed + iov[i].iov_len <= new_result)
1127 : 12 : processed += iov[i].iov_len;
1128 [ + - ]: 113 : else if (processed <= new_result)
1129 : : {
1130 : 113 : uint32 ok_part = new_result - processed;
1131 : :
1132 : 113 : memset((char *) iov[i].iov_base + ok_part, 0, iov[i].iov_len - ok_part);
1133 : 113 : processed += iov[i].iov_len;
1134 : : }
1135 : : else
1136 : : {
1137 : 0 : memset((char *) iov[i].iov_base, 0, iov[i].iov_len);
1138 : : }
1139 : : }
1140 : :
1141 : 121 : ioh->result = new_result;
1142 : : }
1143 : 2155 : }
1144 : :
1145 : : void
1146 : 2155 : inj_io_completion_hook(const char *name, const void *private_data, void *arg)
1147 : : {
1148 : 2155 : inj_io_completion_wait_hook(name, private_data, arg);
1149 : 2155 : inj_io_short_read_hook(name, private_data, arg);
1150 : 2155 : }
1151 : :
1152 : : void
1153 : 504 : inj_io_reopen(const char *name, const void *private_data, void *arg)
1154 : : {
1155 [ + - ]: 504 : ereport(LOG,
1156 : : errmsg("reopen injection point called, is enabled: %d",
1157 : : inj_io_error_state->enabled_reopen),
1158 : : errhidestmt(true), errhidecontext(true));
1159 : :
1160 [ + + ]: 504 : if (inj_io_error_state->enabled_reopen)
1161 [ + - ]: 1 : elog(ERROR, "injection point triggering failure to reopen ");
1162 : 503 : }
1163 : : #endif
1164 : :
1165 : 9 : PG_FUNCTION_INFO_V1(inj_io_completion_wait);
1166 : : Datum
1167 : 10 : inj_io_completion_wait(PG_FUNCTION_ARGS)
1168 : : {
1169 : : #ifdef USE_INJECTION_POINTS
1170 : 10 : inj_io_error_state->enabled_completion_wait = true;
1171 : 10 : inj_io_error_state->completion_wait_pid =
1172 [ + - ]: 10 : PG_ARGISNULL(0) ? InvalidPid : PG_GETARG_INT32(0);
1173 : 10 : inj_io_error_state->completion_wait_relfilenode =
1174 [ + - ]: 10 : PG_ARGISNULL(1) ? InvalidOid : PG_GETARG_OID(1);
1175 : 10 : inj_io_error_state->completion_wait_blockno =
1176 [ + + ]: 10 : PG_ARGISNULL(2) ? InvalidBlockNumber : PG_GETARG_UINT32(2);
1177 : : #else
1178 : : elog(ERROR, "injection points not supported");
1179 : : #endif
1180 : :
1181 : 10 : PG_RETURN_VOID();
1182 : : }
1183 : :
1184 : 13 : PG_FUNCTION_INFO_V1(inj_io_completion_continue);
1185 : : Datum
1186 : 10 : inj_io_completion_continue(PG_FUNCTION_ARGS)
1187 : : {
1188 : : #ifdef USE_INJECTION_POINTS
1189 : 10 : inj_io_error_state->enabled_completion_wait = false;
1190 : 10 : inj_io_error_state->completion_wait_pid = InvalidPid;
1191 : 10 : inj_io_error_state->completion_wait_relfilenode = InvalidOid;
1192 : 10 : inj_io_error_state->completion_wait_blockno = InvalidBlockNumber;
1193 : 10 : ConditionVariableBroadcast(&inj_io_error_state->cv);
1194 : : #else
1195 : : elog(ERROR, "injection points not supported");
1196 : : #endif
1197 : :
1198 : 10 : PG_RETURN_VOID();
1199 : : }
1200 : :
1201 : 9 : PG_FUNCTION_INFO_V1(inj_io_short_read_attach);
1202 : : Datum
1203 : 16 : inj_io_short_read_attach(PG_FUNCTION_ARGS)
1204 : : {
1205 : : #ifdef USE_INJECTION_POINTS
1206 : 16 : inj_io_error_state->enabled_short_read = true;
1207 : 16 : inj_io_error_state->short_read_result_set = !PG_ARGISNULL(0);
1208 [ + - ]: 16 : if (inj_io_error_state->short_read_result_set)
1209 : 16 : inj_io_error_state->short_read_result = PG_GETARG_INT32(0);
1210 : 16 : inj_io_error_state->short_read_pid =
1211 [ + + ]: 16 : PG_ARGISNULL(1) ? InvalidPid : PG_GETARG_INT32(1);
1212 : 16 : inj_io_error_state->short_read_relfilenode =
1213 [ + + ]: 16 : PG_ARGISNULL(2) ? InvalidOid : PG_GETARG_OID(2);
1214 : : #else
1215 : : elog(ERROR, "injection points not supported");
1216 : : #endif
1217 : :
1218 : 16 : PG_RETURN_VOID();
1219 : : }
1220 : :
1221 : 9 : PG_FUNCTION_INFO_V1(inj_io_short_read_detach);
1222 : : Datum
1223 : 6 : inj_io_short_read_detach(PG_FUNCTION_ARGS)
1224 : : {
1225 : : #ifdef USE_INJECTION_POINTS
1226 : 6 : inj_io_error_state->enabled_short_read = false;
1227 : : #else
1228 : : elog(ERROR, "injection points not supported");
1229 : : #endif
1230 : 6 : PG_RETURN_VOID();
1231 : : }
1232 : :
1233 : 6 : PG_FUNCTION_INFO_V1(inj_io_reopen_attach);
1234 : : Datum
1235 : 1 : inj_io_reopen_attach(PG_FUNCTION_ARGS)
1236 : : {
1237 : : #ifdef USE_INJECTION_POINTS
1238 : 1 : inj_io_error_state->enabled_reopen = true;
1239 : : #else
1240 : : elog(ERROR, "injection points not supported");
1241 : : #endif
1242 : :
1243 : 1 : PG_RETURN_VOID();
1244 : : }
1245 : :
1246 : 6 : PG_FUNCTION_INFO_V1(inj_io_reopen_detach);
1247 : : Datum
1248 : 1 : inj_io_reopen_detach(PG_FUNCTION_ARGS)
1249 : : {
1250 : : #ifdef USE_INJECTION_POINTS
1251 : 1 : inj_io_error_state->enabled_reopen = false;
1252 : : #else
1253 : : elog(ERROR, "injection points not supported");
1254 : : #endif
1255 : 1 : PG_RETURN_VOID();
1256 : : }
|
