Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * execAsync.c
4 : : * Support routines for asynchronous execution
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/executor/execAsync.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres.h"
16 : :
17 : : #include "executor/execAsync.h"
18 : : #include "executor/executor.h"
19 : : #include "executor/instrument.h"
20 : : #include "executor/nodeAppend.h"
21 : : #include "executor/nodeForeignscan.h"
22 : :
23 : : /*
24 : : * Asynchronously request a tuple from a designed async-capable node.
25 : : */
26 : : void
27 : 6175 : ExecAsyncRequest(AsyncRequest *areq)
28 : : {
29 [ + + ]: 6175 : if (areq->requestee->chgParam != NULL) /* something changed? */
30 : 2 : ExecReScan(areq->requestee); /* let ReScan handle this */
31 : :
32 : : /* must provide our own instrumentation support */
33 [ + + ]: 6175 : if (areq->requestee->instrument)
34 : 810 : InstrStartNode(areq->requestee->instrument);
35 : :
36 [ + - ]: 6175 : switch (nodeTag(areq->requestee))
37 : : {
38 : 6175 : case T_ForeignScanState:
39 : 6175 : ExecAsyncForeignScanRequest(areq);
40 : 6175 : break;
41 : 0 : default:
42 : : /* If the node doesn't support async, caller messed up. */
43 [ # # ]: 0 : elog(ERROR, "unrecognized node type: %d",
44 : : (int) nodeTag(areq->requestee));
45 : : }
46 : :
47 : 6175 : ExecAsyncResponse(areq);
48 : :
49 : : /* must provide our own instrumentation support */
50 [ + + ]: 6175 : if (areq->requestee->instrument)
51 : 810 : InstrStopNode(areq->requestee->instrument,
52 [ + + - + ]: 810 : TupIsNull(areq->result) ? 0.0 : 1.0);
53 : 6175 : }
54 : :
55 : : /*
56 : : * Give the asynchronous node a chance to configure the file descriptor event
57 : : * for which it wishes to wait. We expect the node-type specific callback to
58 : : * make a single call of the following form:
59 : : *
60 : : * AddWaitEventToSet(set, WL_SOCKET_READABLE, fd, NULL, areq);
61 : : */
62 : : void
63 : 248 : ExecAsyncConfigureWait(AsyncRequest *areq)
64 : : {
65 : : /* must provide our own instrumentation support */
66 [ + + ]: 248 : if (areq->requestee->instrument)
67 : 21 : InstrStartNode(areq->requestee->instrument);
68 : :
69 [ + - ]: 248 : switch (nodeTag(areq->requestee))
70 : : {
71 : 248 : case T_ForeignScanState:
72 : 248 : ExecAsyncForeignScanConfigureWait(areq);
73 : 247 : break;
74 : 0 : default:
75 : : /* If the node doesn't support async, caller messed up. */
76 [ # # ]: 0 : elog(ERROR, "unrecognized node type: %d",
77 : : (int) nodeTag(areq->requestee));
78 : : }
79 : :
80 : : /* must provide our own instrumentation support */
81 [ + + ]: 247 : if (areq->requestee->instrument)
82 : 21 : InstrStopNode(areq->requestee->instrument, 0.0);
83 : 247 : }
84 : :
85 : : /*
86 : : * Call the asynchronous node back when a relevant event has occurred.
87 : : */
88 : : void
89 : 149 : ExecAsyncNotify(AsyncRequest *areq)
90 : : {
91 : : /* must provide our own instrumentation support */
92 [ + + ]: 149 : if (areq->requestee->instrument)
93 : 14 : InstrStartNode(areq->requestee->instrument);
94 : :
95 [ + - ]: 149 : switch (nodeTag(areq->requestee))
96 : : {
97 : 149 : case T_ForeignScanState:
98 : 149 : ExecAsyncForeignScanNotify(areq);
99 : 149 : break;
100 : 0 : default:
101 : : /* If the node doesn't support async, caller messed up. */
102 [ # # ]: 0 : elog(ERROR, "unrecognized node type: %d",
103 : : (int) nodeTag(areq->requestee));
104 : : }
105 : :
106 : 149 : ExecAsyncResponse(areq);
107 : :
108 : : /* must provide our own instrumentation support */
109 [ + + ]: 149 : if (areq->requestee->instrument)
110 : 14 : InstrStopNode(areq->requestee->instrument,
111 [ + + - + ]: 14 : TupIsNull(areq->result) ? 0.0 : 1.0);
112 : 149 : }
113 : :
114 : : /*
115 : : * Call the requestor back when an asynchronous node has produced a result.
116 : : */
117 : : void
118 : 6329 : ExecAsyncResponse(AsyncRequest *areq)
119 : : {
120 [ + - ]: 6329 : switch (nodeTag(areq->requestor))
121 : : {
122 : 6329 : case T_AppendState:
123 : 6329 : ExecAsyncAppendResponse(areq);
124 : 6329 : break;
125 : 0 : default:
126 : : /* If the node doesn't support async, caller messed up. */
127 [ # # ]: 0 : elog(ERROR, "unrecognized node type: %d",
128 : : (int) nodeTag(areq->requestor));
129 : : }
130 : 6329 : }
131 : :
132 : : /*
133 : : * A requestee node should call this function to deliver the tuple to its
134 : : * requestor node. The requestee node can call this from its ExecAsyncRequest
135 : : * or ExecAsyncNotify callback.
136 : : */
137 : : void
138 : 6168 : ExecAsyncRequestDone(AsyncRequest *areq, TupleTableSlot *result)
139 : : {
140 : 6168 : areq->request_complete = true;
141 : 6168 : areq->result = result;
142 : 6168 : }
143 : :
144 : : /*
145 : : * A requestee node should call this function to indicate that it is pending
146 : : * for a callback. The requestee node can call this from its ExecAsyncRequest
147 : : * or ExecAsyncNotify callback.
148 : : */
149 : : void
150 : 161 : ExecAsyncRequestPending(AsyncRequest *areq)
151 : : {
152 : 161 : areq->callback_pending = true;
153 : 161 : areq->request_complete = false;
154 : 161 : areq->result = NULL;
155 : 161 : }
|
