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