Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * tqueue.c
4 : : * Use shm_mq to send & receive tuples between parallel backends
5 : : *
6 : : * A DestReceiver of type DestTupleQueue, which is a TQueueDestReceiver
7 : : * under the hood, writes tuples from the executor to a shm_mq.
8 : : *
9 : : * A TupleQueueReader reads tuples from a shm_mq and returns the tuples.
10 : : *
11 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
12 : : * Portions Copyright (c) 1994, Regents of the University of California
13 : : *
14 : : * IDENTIFICATION
15 : : * src/backend/executor/tqueue.c
16 : : *
17 : : *-------------------------------------------------------------------------
18 : : */
19 : :
20 : : #include "postgres.h"
21 : :
22 : : #include "access/htup_details.h"
23 : : #include "executor/tqueue.h"
24 : :
25 : : /*
26 : : * DestReceiver object's private contents
27 : : *
28 : : * queue is a pointer to data supplied by DestReceiver's caller.
29 : : */
30 : : typedef struct TQueueDestReceiver
31 : : {
32 : : DestReceiver pub; /* public fields */
33 : : shm_mq_handle *queue; /* shm_mq to send to */
34 : : } TQueueDestReceiver;
35 : :
36 : : /*
37 : : * TupleQueueReader object's private contents
38 : : *
39 : : * queue is a pointer to data supplied by reader's caller.
40 : : *
41 : : * "typedef struct TupleQueueReader TupleQueueReader" is in tqueue.h
42 : : */
43 : : struct TupleQueueReader
44 : : {
45 : : shm_mq_handle *queue; /* shm_mq to receive from */
46 : : };
47 : :
48 : : /*
49 : : * Receive a tuple from a query, and send it to the designated shm_mq.
50 : : *
51 : : * Returns true if successful, false if shm_mq has been detached.
52 : : */
53 : : static bool
54 : 249981 : tqueueReceiveSlot(TupleTableSlot *slot, DestReceiver *self)
55 : : {
56 : 249981 : TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
57 : 249981 : MinimalTuple tuple;
58 : 249981 : shm_mq_result result;
59 : 249981 : bool should_free;
60 : :
61 : : /* Send the tuple itself. */
62 : 249981 : tuple = ExecFetchSlotMinimalTuple(slot, &should_free);
63 : 249981 : result = shm_mq_send(tqueue->queue, tuple->t_len, tuple, false, false);
64 : :
65 [ + + ]: 249981 : if (should_free)
66 : 249938 : pfree(tuple);
67 : :
68 : : /* Check for failure. */
69 [ - + ]: 249981 : if (result == SHM_MQ_DETACHED)
70 : 0 : return false;
71 [ + - ]: 249981 : else if (result != SHM_MQ_SUCCESS)
72 [ # # # # ]: 0 : ereport(ERROR,
73 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
74 : : errmsg("could not send tuple to shared-memory queue")));
75 : :
76 : 249981 : return true;
77 : 249981 : }
78 : :
79 : : /*
80 : : * Prepare to receive tuples from executor.
81 : : */
82 : : static void
83 : 436 : tqueueStartupReceiver(DestReceiver *self, int operation, TupleDesc typeinfo)
84 : : {
85 : : /* do nothing */
86 : 436 : }
87 : :
88 : : /*
89 : : * Clean up at end of an executor run
90 : : */
91 : : static void
92 : 434 : tqueueShutdownReceiver(DestReceiver *self)
93 : : {
94 : 434 : TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
95 : :
96 [ - + ]: 434 : if (tqueue->queue != NULL)
97 : 434 : shm_mq_detach(tqueue->queue);
98 : 434 : tqueue->queue = NULL;
99 : 434 : }
100 : :
101 : : /*
102 : : * Destroy receiver when done with it
103 : : */
104 : : static void
105 : 434 : tqueueDestroyReceiver(DestReceiver *self)
106 : : {
107 : 434 : TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
108 : :
109 : : /* We probably already detached from queue, but let's be sure */
110 [ + - ]: 434 : if (tqueue->queue != NULL)
111 : 0 : shm_mq_detach(tqueue->queue);
112 : 434 : pfree(self);
113 : 434 : }
114 : :
115 : : /*
116 : : * Create a DestReceiver that writes tuples to a tuple queue.
117 : : */
118 : : DestReceiver *
119 : 436 : CreateTupleQueueDestReceiver(shm_mq_handle *handle)
120 : : {
121 : 436 : TQueueDestReceiver *self;
122 : :
123 : 436 : self = palloc0_object(TQueueDestReceiver);
124 : :
125 : 436 : self->pub.receiveSlot = tqueueReceiveSlot;
126 : 436 : self->pub.rStartup = tqueueStartupReceiver;
127 : 436 : self->pub.rShutdown = tqueueShutdownReceiver;
128 : 436 : self->pub.rDestroy = tqueueDestroyReceiver;
129 : 436 : self->pub.mydest = DestTupleQueue;
130 : 436 : self->queue = handle;
131 : :
132 : 872 : return (DestReceiver *) self;
133 : 436 : }
134 : :
135 : : /*
136 : : * Create a tuple queue reader.
137 : : */
138 : : TupleQueueReader *
139 : 436 : CreateTupleQueueReader(shm_mq_handle *handle)
140 : : {
141 : 436 : TupleQueueReader *reader = palloc0_object(TupleQueueReader);
142 : :
143 : 436 : reader->queue = handle;
144 : :
145 : 872 : return reader;
146 : 436 : }
147 : :
148 : : /*
149 : : * Destroy a tuple queue reader.
150 : : *
151 : : * Note: cleaning up the underlying shm_mq is the caller's responsibility.
152 : : * We won't access it here, as it may be detached already.
153 : : */
154 : : void
155 : 434 : DestroyTupleQueueReader(TupleQueueReader *reader)
156 : : {
157 : 434 : pfree(reader);
158 : 434 : }
159 : :
160 : : /*
161 : : * Fetch a tuple from a tuple queue reader.
162 : : *
163 : : * The return value is NULL if there are no remaining tuples or if
164 : : * nowait = true and no tuple is ready to return. *done, if not NULL,
165 : : * is set to true when there are no remaining tuples and otherwise to false.
166 : : *
167 : : * The returned tuple, if any, is either in shared memory or a private buffer
168 : : * and should not be freed. The pointer is invalid after the next call to
169 : : * TupleQueueReaderNext().
170 : : *
171 : : * Even when shm_mq_receive() returns SHM_MQ_WOULD_BLOCK, this can still
172 : : * accumulate bytes from a partially-read message, so it's useful to call
173 : : * this with nowait = true even if nothing is returned.
174 : : */
175 : : MinimalTuple
176 : 1103922 : TupleQueueReaderNext(TupleQueueReader *reader, bool nowait, bool *done)
177 : : {
178 : 1103922 : MinimalTuple tuple;
179 : 1103922 : shm_mq_result result;
180 : 1103922 : Size nbytes;
181 : 1103922 : void *data;
182 : :
183 [ - + ]: 1103922 : if (done != NULL)
184 : 1103922 : *done = false;
185 : :
186 : : /* Attempt to read a message. */
187 : 1103922 : result = shm_mq_receive(reader->queue, &nbytes, &data, nowait);
188 : :
189 : : /* If queue is detached, set *done and return NULL. */
190 [ + + ]: 1103922 : if (result == SHM_MQ_DETACHED)
191 : : {
192 [ - + ]: 434 : if (done != NULL)
193 : 434 : *done = true;
194 : 434 : return NULL;
195 : : }
196 : :
197 : : /* In non-blocking mode, bail out if no message ready yet. */
198 [ + + ]: 1103488 : if (result == SHM_MQ_WOULD_BLOCK)
199 : 853507 : return NULL;
200 [ + - ]: 249981 : Assert(result == SHM_MQ_SUCCESS);
201 : :
202 : : /*
203 : : * Return a pointer to the queue memory directly (which had better be
204 : : * sufficiently aligned).
205 : : */
206 : 249981 : tuple = (MinimalTuple) data;
207 [ + - ]: 249981 : Assert(tuple->t_len == nbytes);
208 : :
209 : 249981 : return tuple;
210 : 1103922 : }
|
