From 505fb96f7ca0a3cc729311e68dbd010fdb098c27 Mon Sep 17 00:00:00 2001 From: Kyotaro Horiguchi Date: Thu, 23 Feb 2017 12:20:31 +0900 Subject: [PATCH 2/5] Asynchronous execution framework This is a framework for asynchronous execution based on Robert Haas's proposal. Any executor node can receive tuples from underlying nodes asynchronously by this. This is a different mechanism from parallel execution. While the parallel execution is analogous to threads, this frame work is analogous to select(2), which handles multiple input on single backend process. To avoid degradation of non-async execution, this framework uses completely different channel to convey tuples. You will see the deatil of the API at the end of src/backend/executor/README. --- src/backend/executor/Makefile | 2 +- src/backend/executor/README | 45 +++ src/backend/executor/execAmi.c | 5 + src/backend/executor/execAsync.c | 520 ++++++++++++++++++++++++++++++++ src/backend/executor/execProcnode.c | 1 + src/backend/executor/instrument.c | 2 +- src/backend/executor/nodeAppend.c | 169 ++++++++++- src/backend/executor/nodeForeignscan.c | 49 +++ src/backend/nodes/copyfuncs.c | 2 + src/backend/nodes/outfuncs.c | 2 + src/backend/nodes/readfuncs.c | 2 + src/backend/optimizer/plan/createplan.c | 69 ++++- src/backend/postmaster/pgstat.c | 2 + src/backend/utils/adt/ruleutils.c | 6 +- src/include/executor/execAsync.h | 30 ++ src/include/executor/nodeAppend.h | 3 + src/include/executor/nodeForeignscan.h | 7 + src/include/foreign/fdwapi.h | 17 ++ src/include/nodes/execnodes.h | 65 +++- src/include/nodes/plannodes.h | 2 + src/include/pgstat.h | 3 +- 21 files changed, 974 insertions(+), 29 deletions(-) create mode 100644 src/backend/executor/execAsync.c create mode 100644 src/include/executor/execAsync.h diff --git a/src/backend/executor/Makefile b/src/backend/executor/Makefile index 083b20f..21f5ad0 100644 --- a/src/backend/executor/Makefile +++ b/src/backend/executor/Makefile @@ -12,7 +12,7 @@ subdir = src/backend/executor top_builddir = ../../.. include $(top_builddir)/src/Makefile.global -OBJS = execAmi.o execCurrent.o execExpr.o execExprInterp.o \ +OBJS = execAmi.o execAsync.o execCurrent.o execExpr.o execExprInterp.o \ execGrouping.o execIndexing.o execJunk.o \ execMain.o execParallel.o execProcnode.o \ execReplication.o execScan.o execSRF.o execTuples.o \ diff --git a/src/backend/executor/README b/src/backend/executor/README index a004506..e6caeb7 100644 --- a/src/backend/executor/README +++ b/src/backend/executor/README @@ -349,3 +349,48 @@ query returning the same set of scan tuples multiple times. Likewise, SRFs are disallowed in an UPDATE's targetlist. There, they would have the effect of the same row being updated multiple times, which is not very useful --- and updates after the first would have no effect anyway. + +Asynchronous Execution +---------------------- + +In certain cases, it's desirable for a node to indicate that it cannot +return any tuple immediately but may be able to do at a later time. This +might either because the node is waiting on an event external to the +database system, such as a ForeignScan awaiting network I/O, or because +the node is waiting for an event internal to the database system - e.g. +one process involved in a parallel query may find that it cannot progress +a certain parallel operation until some other process reaches a certain +point in the computation. A process which discovers this type of situation +can always handle it simply by blocking, but this may waste time that could +be spent executing some other part of the plan where progress could be +made immediately. This is particularly likely to occur when the plan +contains an Append node. + +To use asynchronous execution, a node must first request a tuple from +an async-capable child node using ExecAsyncRequest. Next, when the +result is not available immediately, it must execute the asynchronous +event loop using ExecAsyncEventLoop; it can avoid giving up control +indefinitely by passing a timeout to this function, even passing -1 to +poll for events without blocking. Eventually, when a node to which an +asynchronous request has been made produces a tuple, the requesting +node will receive a callback from the event loop via +ExecAsyncResponse. Typically, the ExecAsyncResponse callback is the +only one required for nodes that wish to request tuples +asynchronously. + +On the other hand, nodes that wish to produce tuples asynchronously +generally need to implement three methods: + +1. When an asynchronous request is made, the node's ExecAsyncRequest callback +will be invoked; it should use ExecAsyncSetRequiredEvents to indicate the +number of file descriptor events for which it wishes to wait and whether it +wishes to receive a callback when the process latch is set. Alternatively, +it can instead use ExecAsyncRequestDone if a result is available immediately. + +2. When the event loop wishes to wait or poll for file descriptor events and +the process latch, the ExecAsyncConfigureWait callback is invoked to configure +the file descriptor wait events for which the node wishes to wait. This +callback isn't needed if the node only cares about the process latch. + +3. When file descriptors or the process latch become ready, the node's +ExecAsyncNotify callback is invoked. diff --git a/src/backend/executor/execAmi.c b/src/backend/executor/execAmi.c index 7e85c66..ddb6d64 100644 --- a/src/backend/executor/execAmi.c +++ b/src/backend/executor/execAmi.c @@ -478,11 +478,16 @@ ExecSupportsBackwardScan(Plan *node) { ListCell *l; + /* With async, tuples may be interleaved, so can't back up. */ + if (((Append *) node)->nasyncplans != 0) + return false; + foreach(l, ((Append *) node)->appendplans) { if (!ExecSupportsBackwardScan((Plan *) lfirst(l))) return false; } + /* need not check tlist because Append doesn't evaluate it */ return true; } diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c new file mode 100644 index 0000000..115b147 --- /dev/null +++ b/src/backend/executor/execAsync.c @@ -0,0 +1,520 @@ +/*------------------------------------------------------------------------- + * + * execAsync.c + * Support routines for asynchronous execution. + * + * Portions Copyright (c) 1996-2016, 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/nodeAppend.h" +#include "executor/nodeForeignscan.h" +#include "miscadmin.h" +#include "pgstat.h" +#include "storage/latch.h" +#include "utils/memutils.h" + +static bool ExecAsyncEventWait(EState *estate, long timeout); +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, + bool reinit); +static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq); +static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq); + +#define EVENT_BUFFER_SIZE 16 + +/* + * Asynchronously request a tuple from a designed async-aware node. + * + * requestor is the node that wants the tuple; requestee is the node from + * which it wants the tuple. request_index is an arbitrary integer specified + * by the requestor which will be available at the time the requestor receives + * the tuple. This is useful if the requestor has multiple children and + * needs an easy way to figure out which one is delivering a tuple. + */ +void +ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, + PlanState *requestee) +{ + PendingAsyncRequest *areq = NULL; + int nasync = estate->es_num_pending_async; + + if (requestee->instrument) + InstrStartNode(requestee->instrument); + + /* + * If the number of pending asynchronous nodes exceeds the number of + * available slots in the es_pending_async array, expand the array. + * We start with 16 slots, and thereafter double the array size each + * time we run out of slots. + */ + if (nasync >= estate->es_max_pending_async) + { + int newmax; + + newmax = estate->es_max_pending_async * 2; + if (estate->es_max_pending_async == 0) + { + newmax = 16; + estate->es_pending_async = + MemoryContextAllocZero(estate->es_query_cxt, + newmax * sizeof(PendingAsyncRequest *)); + } + else + { + int newentries = newmax - estate->es_max_pending_async; + + estate->es_pending_async = + repalloc(estate->es_pending_async, + newmax * sizeof(PendingAsyncRequest *)); + MemSet(&estate->es_pending_async[estate->es_max_pending_async], + 0, newentries * sizeof(PendingAsyncRequest *)); + } + estate->es_max_pending_async = newmax; + } + + /* + * To avoid unnecessary palloc traffic, we reuse a previously-allocated + * PendingAsyncRequest if there is one. If not, we must allocate a new + * one. + */ + if (estate->es_pending_async[nasync] == NULL) + { + areq = MemoryContextAllocZero(estate->es_query_cxt, + sizeof(PendingAsyncRequest)); + estate->es_pending_async[nasync] = areq; + } + else + { + areq = estate->es_pending_async[nasync]; + MemSet(areq, 0, sizeof(PendingAsyncRequest)); + } + areq->myindex = estate->es_num_pending_async; + + /* Initialize the new request. */ + areq->state = ASYNCREQ_IDLE; + areq->requestor = requestor; + areq->request_index = request_index; + areq->requestee = requestee; + + /* Give the requestee a chance to do whatever it wants. */ + switch (nodeTag(requestee)) + { + case T_ForeignScanState: + ExecAsyncForeignScanRequest(estate, areq); + break; + default: + /* If requestee doesn't support async, caller messed up. */ + elog(ERROR, "unrecognized node type: %d", + (int) nodeTag(requestee)); + } + + if (areq->requestee->instrument) + InstrStopNode(requestee->instrument, 0); + + /* No result available now, make this node pending */ + estate->es_num_pending_async++; + + return; +} + +/* + * Execute the main loop until the timeout expires or a result is delivered + * to the requestor. + * + * If the timeout is -1, there is no timeout; wait indefinitely until a + * result is ready for requestor. If the timeout is 0, do not block, but + * poll for events and fire callbacks for as long as we can do so without + * blocking. If timeout is greater than 0, block for at most the number + * of milliseconds indicated by the timeout. + * + * Returns true if a result was delivered to the requestor. A return value + * of false indicates that the timeout was reached without delivering a + * result to the requestor. + */ +bool +ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) +{ + instr_time start_time; + long cur_timeout = timeout; + bool requestor_done = false; + + Assert(requestor != NULL); + + /* + * If we plan to wait - but not indefinitely - we need to record the + * current time. + */ + if (timeout > 0) + INSTR_TIME_SET_CURRENT(start_time); + + /* Main event loop: poll for events, deliver notifications. */ + Assert(estate->es_async_callback_pending == 0); + for (;;) + { + int i; + bool any_node_done = false; + + CHECK_FOR_INTERRUPTS(); + + /* Check for events only if any node is async-not-ready. */ + if (estate->es_num_async_ready < estate->es_num_pending_async) + { + /* Don't block if any tuple available. */ + if (estate->es_async_callback_pending > 0) + ExecAsyncEventWait(estate, 0); + else if (!ExecAsyncEventWait(estate, cur_timeout)) + { /* Not fired */ + /* Exited before timeout. Calculate the remaining time. */ + instr_time cur_time; + long cur_timeout = -1; + + /* Wait forever */ + if (timeout < 0) + continue; + + INSTR_TIME_SET_CURRENT(cur_time); + INSTR_TIME_SUBTRACT(cur_time, start_time); + cur_timeout = + timeout - (long) INSTR_TIME_GET_MILLISEC(cur_time); + + if (cur_timeout > 0) + continue; + } + } + + /* Deliver notifications. */ + for (i = 0; i < estate->es_num_pending_async; ++i) + { + PendingAsyncRequest *areq = estate->es_pending_async[i]; + + if (areq->requestee->instrument) + InstrStartNode(areq->requestee->instrument); + + /* Notify if the requestee is ready */ + if (areq->state == ASYNCREQ_CALLBACK_PENDING) + ExecAsyncNotify(estate, areq); + + /* Deliver the acquired tuple to the requester */ + if (areq->state == ASYNCREQ_COMPLETE) + { + any_node_done = true; + if (requestor == areq->requestor) + requestor_done = true; + ExecAsyncResponse(estate, areq); + + if (areq->requestee->instrument) + InstrStopNode(areq->requestee->instrument, + TupIsNull((TupleTableSlot*)areq->result) ? + 0.0 : 1.0); + } + else if (areq->requestee->instrument) + InstrStopNode(areq->requestee->instrument, 0); + } + + /* If any node completed, compact the array. */ + if (any_node_done) + { + int hidx = 0, + tidx; + + /* + * Swap all non-yet-completed items to the start of the array. + * Keep them in the same order. + */ + for (tidx = 0; tidx < estate->es_num_pending_async; ++tidx) + { + PendingAsyncRequest *head; + PendingAsyncRequest *tail = estate->es_pending_async[tidx]; + + Assert(tail->state != ASYNCREQ_CALLBACK_PENDING); + + if (tail->state == ASYNCREQ_COMPLETE) + continue; + head = estate->es_pending_async[hidx]; + estate->es_pending_async[tidx] = head; + estate->es_pending_async[hidx] = tail; + ++hidx; + } + estate->es_num_pending_async = hidx; + } + + /* + * We only consider exiting the loop when no notifications are + * pending. Otherwise, each call to this function might advance + * the computation by only a very small amount; to the contrary, + * we want to push it forward as far as possible. + */ + if (estate->es_async_callback_pending == 0) + { + /* If requestor is ready, exit. */ + if (requestor_done) + return true; + /* If timeout was 0 or has expired, exit. */ + if (cur_timeout == 0) + return false; + } + } +} + +/* + * Wait or poll for events. As with ExecAsyncEventLoop, a timeout of -1 + * means wait forever, 0 means don't wait at all, and >0 means wait for the + * indicated number of milliseconds. + * + * Returns false if we timed out or true if anything found or there's no event + * to wait. + */ +static bool +ExecAsyncEventWait(EState *estate, long timeout) +{ + WaitEvent occurred_event[EVENT_BUFFER_SIZE]; + int noccurred; + int i; + int n; + bool reinit = false; + bool process_latch_set = false; + bool added = false; + bool fired = false; + + if (estate->es_wait_event_set == NULL) + { + /* + * Allow for a few extra events without reinitializing. It + * doesn't seem worth the complexity of doing anything very + * aggressive here, because plans that depend on massive numbers + * of external FDs are likely to run afoul of kernel limits anyway. + */ + estate->es_allocated_fd_events = estate->es_total_fd_events + 16; + + /* + * The wait event set created here should be live beyond ExecutorState + * context but released in case of error. + */ + estate->es_wait_event_set = + CreateWaitEventSet(TopTransactionContext, + TopTransactionResourceOwner, + estate->es_allocated_fd_events + 1); + + AddWaitEventToSet(estate->es_wait_event_set, + WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, NULL); + reinit = true; + } + + /* Give each waiting node a chance to add or modify events. */ + for (i = 0; i < estate->es_num_pending_async; ++i) + { + PendingAsyncRequest *areq = estate->es_pending_async[i]; + + if (areq->num_fd_events > 0 || areq->wants_process_latch) + added |= ExecAsyncConfigureWait(estate, areq, reinit); + } + + /* + * We may have no event to wait. This occurs when all nodes that + * is asynchronously executing have tuples immediately available. + */ + if (!added) + return true; + + /* Wait for at least one event to occur. */ + noccurred = WaitEventSetWait(estate->es_wait_event_set, timeout, + occurred_event, EVENT_BUFFER_SIZE, + WAIT_EVENT_ASYNC_WAIT); + + if (noccurred == 0) + return false; + + /* + * Loop over the occurred events and set the callback_pending flags + * for the appropriate requests. The waiting nodes should have + * registered their wait events with user_data pointing back to the + * PendingAsyncRequest, but the process latch needs special handling. + */ + for (n = 0; n < noccurred; ++n) + { + WaitEvent *w = &occurred_event[n]; + + if ((w->events & WL_LATCH_SET) != 0) + { + process_latch_set = true; + continue; + } + + if ((w->events & (WL_SOCKET_READABLE|WL_SOCKET_WRITEABLE)) != 0) + { + PendingAsyncRequest *areq = w->user_data; + + Assert(areq->state == ASYNCREQ_WAITING); + + areq->state = ASYNCREQ_CALLBACK_PENDING; + estate->es_async_callback_pending++; + fired = true; + } + } + + /* + * If the process latch got set, we must schedule a callback for every + * requestee that cares about it. + */ + if (process_latch_set) + { + for (i = 0; i < estate->es_num_pending_async; ++i) + { + PendingAsyncRequest *areq = estate->es_pending_async[i]; + + if (areq->wants_process_latch) + { + Assert(areq->state == ASYNCREQ_WAITING); + areq->state = ASYNCREQ_CALLBACK_PENDING; + estate->es_async_callback_pending++; + fired = true; + } + } + } + + return fired; +} + +/* + * Give the asynchronous node a chance to configure the file descriptor + * events for which it wishes to wait. We expect the node-type specific + * callback to make one or more calls of the following form: + * + * AddWaitEventToSet(es->es_wait_event_set, events, fd, NULL, areq); + * + * The events should include only WL_SOCKET_READABLE or WL_SOCKET_WRITEABLE, + * and the number of calls should not exceed areq->num_fd_events (as + * prevously set via ExecAsyncSetRequiredEvents). + * + * Individual requests can omit registering an event but it is a + * responsibility of the node driver to set at least one event per one + * requestor. + */ +static bool +ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, + bool reinit) +{ + switch (nodeTag(areq->requestee)) + { + case T_ForeignScanState: + return ExecAsyncForeignScanConfigureWait(estate, areq, reinit); + break; + default: + elog(ERROR, "unrecognized node type: %d", + (int) nodeTag(areq->requestee)); + } +} + +/* + * Call the asynchronous node back when a relevant event has occurred. + */ +static void +ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq) +{ + switch (nodeTag(areq->requestee)) + { + case T_ForeignScanState: + ExecAsyncForeignScanNotify(estate, areq); + break; + default: + elog(ERROR, "unrecognized node type: %d", + (int) nodeTag(areq->requestee)); + } + + estate->es_async_callback_pending--; +} + +/* + * Call the requestor back when an asynchronous node has produced a result. + */ +static void +ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq) +{ + switch (nodeTag(areq->requestor)) + { + case T_AppendState: + ExecAsyncAppendResponse(estate, areq); + break; + default: + elog(ERROR, "unrecognized node type: %d", + (int) nodeTag(areq->requestor)); + } + estate->es_num_async_ready--; +} + +/* + * An executor node should call this function to signal that it needs to wait + * on one or more file descriptor events that can be registered on a + * WaitEventSet, and possibly also on process latch. num_fd_events is the + * maximum number of file descriptor events that it will wish to register. + * force_reset should be true if the node can't reuse the WaitEventSet it most + * recently initialized, for example because it needs to drop a wait event + * from the set. + */ +void +ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq, + int num_fd_events, bool wants_process_latch, + bool force_reset) +{ + estate->es_total_fd_events += num_fd_events - areq->num_fd_events; + areq->num_fd_events = num_fd_events; + areq->wants_process_latch = wants_process_latch; + areq->state = ASYNCREQ_WAITING; + + if (force_reset && estate->es_wait_event_set != NULL) + ExecAsyncClearEvents(estate); +} + +/* + * An async-capable node should call this function to deliver the tuple to + * the node which requested it. The node can call this from its + * ExecAsyncRequest callback if the requested tuple is available immediately, + * or at a later time from its ExecAsyncNotify callback. + */ +void +ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result) +{ + /* + * Since the request is complete, the requestee is no longer allowed + * to wait for any events. Note that this forces a rebuild of + * es_wait_event_set every time a process that was previously waiting + * stops doing so. It might be possible to defer that decision until + * we actually wait again, because it's quite possible that a new + * request will be made of the same node before any wait actually + * happens. However, we have to balance the cost of rebuilding the + * WaitEventSet against the additional overhead of tracking which nodes + * need a callback to remove registered wait events. It's not clear + * that we would come out ahead, so use brute force for now. + */ + Assert(areq->state == ASYNCREQ_IDLE || + areq->state == ASYNCREQ_CALLBACK_PENDING); + + if (areq->num_fd_events > 0 || areq->wants_process_latch) + ExecAsyncSetRequiredEvents(estate, areq, 0, false, true); + + + /* Save result and mark request as complete. */ + areq->result = result; + areq->state = ASYNCREQ_COMPLETE; + estate->es_num_async_ready++; +} + + +/* Clear async events */ +void +ExecAsyncClearEvents(EState *estate) +{ + if (estate->es_wait_event_set == NULL) + return; + + FreeWaitEventSet(estate->es_wait_event_set); + estate->es_wait_event_set = NULL; +} diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c index 486ddf1..2f896ef 100644 --- a/src/backend/executor/execProcnode.c +++ b/src/backend/executor/execProcnode.c @@ -118,6 +118,7 @@ #include "executor/nodeValuesscan.h" #include "executor/nodeWindowAgg.h" #include "executor/nodeWorktablescan.h" +#include "foreign/fdwapi.h" #include "nodes/nodeFuncs.h" #include "miscadmin.h" diff --git a/src/backend/executor/instrument.c b/src/backend/executor/instrument.c index 6ec96ec..959ee90 100644 --- a/src/backend/executor/instrument.c +++ b/src/backend/executor/instrument.c @@ -102,7 +102,7 @@ InstrStopNode(Instrumentation *instr, double nTuples) &pgBufferUsage, &instr->bufusage_start); /* Is this the first tuple of this cycle? */ - if (!instr->running) + if (!instr->running && nTuples > 0) { instr->running = true; instr->firsttuple = INSTR_TIME_GET_DOUBLE(instr->counter); diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index a107545..d91e621 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -58,6 +58,7 @@ #include "postgres.h" #include "executor/execdebug.h" +#include "executor/execAsync.h" #include "executor/nodeAppend.h" static bool exec_append_initialize_next(AppendState *appendstate); @@ -79,16 +80,21 @@ exec_append_initialize_next(AppendState *appendstate) /* * get information from the append node */ - whichplan = appendstate->as_whichplan; + whichplan = appendstate->as_whichsyncplan; - if (whichplan < 0) + /* + * This routine is only responsible for setting up for nodes being scanned + * synchronously, so the first node we can scan is given by nasyncplans + * and the last is given by as_nplans - 1. + */ + if (whichplan < appendstate->as_nasyncplans) { /* * if scanning in reverse, we start at the last scan in the list and * then proceed back to the first.. in any case we inform ExecAppend * that we are at the end of the line by returning FALSE */ - appendstate->as_whichplan = 0; + appendstate->as_whichsyncplan = appendstate->as_nasyncplans; return FALSE; } else if (whichplan >= appendstate->as_nplans) @@ -96,7 +102,7 @@ exec_append_initialize_next(AppendState *appendstate) /* * as above, end the scan if we go beyond the last scan in our list.. */ - appendstate->as_whichplan = appendstate->as_nplans - 1; + appendstate->as_whichsyncplan = appendstate->as_nplans - 1; return FALSE; } else @@ -148,6 +154,15 @@ ExecInitAppend(Append *node, EState *estate, int eflags) appendstate->ps.state = estate; appendstate->appendplans = appendplanstates; appendstate->as_nplans = nplans; + appendstate->as_nasyncplans = node->nasyncplans; + appendstate->as_syncdone = (node->nasyncplans == nplans); + appendstate->as_asyncresult = (TupleTableSlot **) + palloc0(node->nasyncplans * sizeof(TupleTableSlot *)); + + /* initially, all async requests need a request */ + for (i = 0; i < appendstate->as_nasyncplans; ++i) + appendstate->as_needrequest = + bms_add_member(appendstate->as_needrequest, i); /* * Miscellaneous initialization @@ -182,9 +197,9 @@ ExecInitAppend(Append *node, EState *estate, int eflags) appendstate->ps.ps_ProjInfo = NULL; /* - * initialize to scan first subplan + * initialize to scan first synchronous subplan */ - appendstate->as_whichplan = 0; + appendstate->as_whichsyncplan = appendstate->as_nasyncplans; exec_append_initialize_next(appendstate); return appendstate; @@ -199,15 +214,85 @@ ExecInitAppend(Append *node, EState *estate, int eflags) TupleTableSlot * ExecAppend(AppendState *node) { + if (node->as_nasyncplans > 0) + { + EState *estate = node->ps.state; + int i; + + /* + * If there are any asynchronously-generated results that have + * not yet been returned, return one of them. + */ + if (node->as_nasyncresult > 0) + { + --node->as_nasyncresult; + return node->as_asyncresult[node->as_nasyncresult]; + } + + + /* + * XXXX: Always clear registered event. This seems a bit ineffecient + * but the events to wait are almost randomly altered for every + * calling. + */ + ExecAsyncClearEvents(estate); + + while ((i = bms_first_member(node->as_needrequest)) >= 0) + { + node->as_nasyncpending++; + ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); + } + + if (node->as_nasyncpending == 0 && node->as_syncdone) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); + } + for (;;) { PlanState *subnode; TupleTableSlot *result; /* - * figure out which subplan we are currently processing + * if we have async requests outstanding, run the event loop + */ + if (node->as_nasyncpending > 0) + { + long timeout = node->as_syncdone ? -1 : 0; + + while (node->as_nasyncpending > 0) + { + if (ExecAsyncEventLoop(node->ps.state, &node->ps, timeout) && + node->as_nasyncresult > 0) + { + /* Asynchronous subplan returned a tuple! */ + --node->as_nasyncresult; + return node->as_asyncresult[node->as_nasyncresult]; + } + + /* Timeout reached. Go through to sync nodes if exists */ + if (!node->as_syncdone) + break; + } + + /* + * If there is no asynchronous activity still pending and the + * synchronous activity is also complete, we're totally done + * scanning this node. Otherwise, we're done with the + * asynchronous stuff but must continue scanning the synchronous + * children. + */ + if (node->as_syncdone) + { + Assert(node->as_nasyncpending == 0); + return ExecClearTuple(node->ps.ps_ResultTupleSlot); + } + } + + /* + * figure out which synchronous subplan we are currently processing */ - subnode = node->appendplans[node->as_whichplan]; + Assert(!node->as_syncdone); + subnode = node->appendplans[node->as_whichsyncplan]; /* * get a tuple from the subplan @@ -227,14 +312,21 @@ ExecAppend(AppendState *node) /* * Go on to the "next" subplan in the appropriate direction. If no * more subplans, return the empty slot set up for us by - * ExecInitAppend. + * ExecInitAppend, unless there are async plans we have yet to finish. */ if (ScanDirectionIsForward(node->ps.state->es_direction)) - node->as_whichplan++; + node->as_whichsyncplan++; else - node->as_whichplan--; + node->as_whichsyncplan--; if (!exec_append_initialize_next(node)) - return ExecClearTuple(node->ps.ps_ResultTupleSlot); + { + node->as_syncdone = true; + if (node->as_nasyncpending == 0) + { + Assert(bms_is_empty(node->as_needrequest)); + return ExecClearTuple(node->ps.ps_ResultTupleSlot); + } + } /* Else loop back and try to get a tuple from the new subplan */ } @@ -273,6 +365,16 @@ ExecReScanAppend(AppendState *node) { int i; + /* + * XXX. Cancel outstanding asynchronous tuple requests here! (How?) + */ + + /* Reset async state. */ + for (i = 0; i < node->as_nasyncplans; ++i) + node->as_needrequest = bms_add_member(node->as_needrequest, i); + node->as_nasyncresult = 0; + node->as_syncdone = (node->as_nasyncplans == node->as_nplans); + for (i = 0; i < node->as_nplans; i++) { PlanState *subnode = node->appendplans[i]; @@ -291,6 +393,47 @@ ExecReScanAppend(AppendState *node) if (subnode->chgParam == NULL) ExecReScan(subnode); } - node->as_whichplan = 0; + node->as_whichsyncplan = node->as_nasyncplans; exec_append_initialize_next(node); } + +/* ---------------------------------------------------------------- + * ExecAsyncAppendResponse + * + * Receive a response from an asynchronous request we made. + * ---------------------------------------------------------------- + */ +void +ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq) +{ + AppendState *node = (AppendState *) areq->requestor; + TupleTableSlot *slot; + + /* We shouldn't be called until the request is complete. */ + Assert(areq->state == ASYNCREQ_COMPLETE); + + /* Our result slot shouldn't already be occupied. */ + Assert(TupIsNull(node->ps.ps_ResultTupleSlot)); + + /* Result should be a TupleTableSlot or NULL. */ + slot = (TupleTableSlot *) areq->result; + Assert(slot == NULL || IsA(slot, TupleTableSlot)); + + /* This is no longer pending */ + --node->as_nasyncpending; + + /* If the result is NULL or an empty slot, there's nothing more to do. */ + if (TupIsNull(slot)) + return; + + /* Save result so we can return it. */ + Assert(node->as_nasyncresult < node->as_nasyncplans); + node->as_asyncresult[node->as_nasyncresult++] = slot; + + /* + * Mark the node that returned a result as ready for a new request. We + * don't launch another one here immediately because it might compelte + */ + node->as_needrequest = + bms_add_member(node->as_needrequest, areq->request_index); +} diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 9ae1561..7db5c30 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -364,3 +364,52 @@ ExecShutdownForeignScan(ForeignScanState *node) if (fdwroutine->ShutdownForeignScan) fdwroutine->ShutdownForeignScan(node); } + +/* ---------------------------------------------------------------- + * ExecAsyncForeignScanRequest + * + * Initiate an asynchronous request + * ---------------------------------------------------------------- + */ +void +ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq) +{ + ForeignScanState *node = (ForeignScanState *) areq->requestee; + FdwRoutine *fdwroutine = node->fdwroutine; + + Assert(fdwroutine->ForeignAsyncRequest != NULL); + fdwroutine->ForeignAsyncRequest(estate, areq); +} + +/* ---------------------------------------------------------------- + * ExecAsyncForeignScanConfigureWait + * + * In async mode, configure for a wait + * ---------------------------------------------------------------- + */ +bool +ExecAsyncForeignScanConfigureWait(EState *estate, + PendingAsyncRequest *areq, bool reinit) +{ + ForeignScanState *node = (ForeignScanState *) areq->requestee; + FdwRoutine *fdwroutine = node->fdwroutine; + + Assert(fdwroutine->ForeignAsyncConfigureWait != NULL); + return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); +} + +/* ---------------------------------------------------------------- + * ExecAsyncForeignScanNotify + * + * Event loop callback + * ---------------------------------------------------------------- + */ +void +ExecAsyncForeignScanNotify(EState *estate, PendingAsyncRequest *areq) +{ + ForeignScanState *node = (ForeignScanState *) areq->requestee; + FdwRoutine *fdwroutine = node->fdwroutine; + + Assert(fdwroutine->ForeignAsyncNotify != NULL); + fdwroutine->ForeignAsyncNotify(estate, areq); +} diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 61bc502..9856dfb 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -239,6 +239,8 @@ _copyAppend(const Append *from) */ COPY_NODE_FIELD(partitioned_rels); COPY_NODE_FIELD(appendplans); + COPY_SCALAR_FIELD(nasyncplans); + COPY_SCALAR_FIELD(referent); return newnode; } diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index 83fb39f..f324b0c 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -372,6 +372,8 @@ _outAppend(StringInfo str, const Append *node) WRITE_NODE_FIELD(partitioned_rels); WRITE_NODE_FIELD(appendplans); + WRITE_INT_FIELD(nasyncplans); + WRITE_INT_FIELD(referent); } static void diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c index 766f2d8..8c57d81 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1575,6 +1575,8 @@ _readAppend(void) READ_NODE_FIELD(partitioned_rels); READ_NODE_FIELD(appendplans); + READ_INT_FIELD(nasyncplans); + READ_INT_FIELD(referent); READ_DONE(); } diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c index b121f40..c6825d2 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -203,7 +203,8 @@ static NamedTuplestoreScan *make_namedtuplestorescan(List *qptlist, List *qpqual Index scanrelid, char *enrname); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *appendplans, List *tlist, List *partitioned_rels); +static Append *make_append(List *asyncplans, int nasyncplans, + int referent, List *tlist, List *partitioned_rels); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -283,7 +284,7 @@ static ModifyTable *make_modifytable(PlannerInfo *root, List *rowMarks, OnConflictExpr *onconflict, int epqParam); static GatherMerge *create_gather_merge_plan(PlannerInfo *root, GatherMergePath *best_path); - +static bool is_async_capable_path(Path *path); /* * create_plan @@ -992,8 +993,12 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) { Append *plan; List *tlist = build_path_tlist(root, &best_path->path); - List *subplans = NIL; + List *asyncplans = NIL; + List *syncplans = NIL; ListCell *subpaths; + int nasyncplans = 0; + bool first = true; + bool referent_is_sync = true; /* * The subpaths list could be empty, if every child was proven empty by @@ -1019,7 +1024,14 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) return plan; } - /* Build the plan for each child */ + /* + * Build the plan for each child + + * The first child in an inheritance set is the representative in + * explaining tlist entries (see set_deparse_planstate). We should keep + * the first child in best_path->subpaths at the head of the subplan list + * for the reason. + */ foreach(subpaths, best_path->subpaths) { Path *subpath = (Path *) lfirst(subpaths); @@ -1028,7 +1040,18 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) /* Must insist that all children return the same tlist */ subplan = create_plan_recurse(root, subpath, CP_EXACT_TLIST); - subplans = lappend(subplans, subplan); + /* Classify as async-capable or not */ + if (is_async_capable_path(subpath)) + { + asyncplans = lappend(asyncplans, subplan); + ++nasyncplans; + if (first) + referent_is_sync = false; + } + else + syncplans = lappend(syncplans, subplan); + + first = false; } /* @@ -1038,7 +1061,9 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(subplans, tlist, best_path->partitioned_rels); + plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, + referent_is_sync ? nasyncplans : 0, tlist, + best_path->partitioned_rels); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -5245,17 +5270,23 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, List *tlist, List *partitioned_rels) +make_append(List *appendplans, int nasyncplans, int referent, + List *tlist, List *partitioned_rels) { Append *node = makeNode(Append); Plan *plan = &node->plan; + /* Currently async on partitioned tables is not available */ + Assert(nasyncplans == 0 || partitioned_rels == NIL); + plan->targetlist = tlist; plan->qual = NIL; plan->lefttree = NULL; plan->righttree = NULL; node->partitioned_rels = partitioned_rels; node->appendplans = appendplans; + node->nasyncplans = nasyncplans; + node->referent = referent; return node; } @@ -6578,3 +6609,27 @@ is_projection_capable_plan(Plan *plan) } return true; } + +/* + * is_projection_capable_path + * Check whether a given Path node is async-capable. + */ +static bool +is_async_capable_path(Path *path) +{ + switch (nodeTag(path)) + { + case T_ForeignPath: + { + FdwRoutine *fdwroutine = path->parent->fdwroutine; + + Assert(fdwroutine != NULL); + if (fdwroutine->IsForeignPathAsyncCapable != NULL && + fdwroutine->IsForeignPathAsyncCapable((ForeignPath *) path)) + return true; + } + default: + break; + } + return false; +} diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 56a8bf2..fbcdba6 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -3571,6 +3571,8 @@ pgstat_get_wait_ipc(WaitEventIPC w) break; case WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE: event_name = "LogicalSyncStateChange"; + case WAIT_EVENT_ASYNC_WAIT: + event_name = "AsyncExecWait"; break; /* no default case, so that compiler will warn */ } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 0c1a201..e224158 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -4335,7 +4335,11 @@ set_deparse_planstate(deparse_namespace *dpns, PlanState *ps) * lists containing references to non-target relations. */ if (IsA(ps, AppendState)) - dpns->outer_planstate = ((AppendState *) ps)->appendplans[0]; + { + int idx = ((Append*)(((AppendState *) ps)->ps.plan))->referent; + dpns->outer_planstate = + ((AppendState *) ps)->appendplans[idx]; + } else if (IsA(ps, MergeAppendState)) dpns->outer_planstate = ((MergeAppendState *) ps)->mergeplans[0]; else if (IsA(ps, ModifyTableState)) diff --git a/src/include/executor/execAsync.h b/src/include/executor/execAsync.h new file mode 100644 index 0000000..9e7845c --- /dev/null +++ b/src/include/executor/execAsync.h @@ -0,0 +1,30 @@ +/*-------------------------------------------------------------------- + * execAsync.c + * Support functions for asynchronous query execution + * + * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/backend/executor/execAsync.c + *-------------------------------------------------------------------- + */ + +#ifndef EXECASYNC_H +#define EXECASYNC_H + +#include "nodes/execnodes.h" + +extern void ExecAsyncRequest(EState *estate, PlanState *requestor, + int request_index, PlanState *requestee); +extern bool ExecAsyncEventLoop(EState *estate, PlanState *requestor, + long timeout); + +extern void ExecAsyncSetRequiredEvents(EState *estate, + PendingAsyncRequest *areq, int num_fd_events, + bool wants_process_latch, bool force_reset); +extern void ExecAsyncRequestDone(EState *estate, + PendingAsyncRequest *areq, Node *result); +extern void ExecAsyncClearEvents(EState *estate); + +#endif /* EXECASYNC_H */ diff --git a/src/include/executor/nodeAppend.h b/src/include/executor/nodeAppend.h index 6fb4662..3cbf9ff 100644 --- a/src/include/executor/nodeAppend.h +++ b/src/include/executor/nodeAppend.h @@ -21,4 +21,7 @@ extern TupleTableSlot *ExecAppend(AppendState *node); extern void ExecEndAppend(AppendState *node); extern void ExecReScanAppend(AppendState *node); +extern void ExecAsyncAppendResponse(EState *estate, + PendingAsyncRequest *areq); + #endif /* NODEAPPEND_H */ diff --git a/src/include/executor/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 1b167b8..e4ba4a9 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -30,4 +30,11 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node, shm_toc *toc); extern void ExecShutdownForeignScan(ForeignScanState *node); +extern void ExecAsyncForeignScanRequest(EState *estate, + PendingAsyncRequest *areq); +extern bool ExecAsyncForeignScanConfigureWait(EState *estate, + PendingAsyncRequest *areq, bool reinit); +extern void ExecAsyncForeignScanNotify(EState *estate, + PendingAsyncRequest *areq); + #endif /* NODEFOREIGNSCAN_H */ diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 6ca44f7..863ff0e 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -156,6 +156,16 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root, RelOptInfo *rel, RangeTblEntry *rte); +typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path); +typedef void (*ForeignAsyncRequest_function) (EState *estate, + PendingAsyncRequest *areq); +typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate, + PendingAsyncRequest *areq, + bool reinit); +typedef void (*ForeignAsyncNotify_function) (EState *estate, + PendingAsyncRequest *areq); +typedef void (*ShutdownForeignScan_function) (ForeignScanState *node); + /* * FdwRoutine is the struct returned by a foreign-data wrapper's handler * function. It provides pointers to the callback functions needed by the @@ -225,6 +235,13 @@ typedef struct FdwRoutine EstimateDSMForeignScan_function EstimateDSMForeignScan; InitializeDSMForeignScan_function InitializeDSMForeignScan; InitializeWorkerForeignScan_function InitializeWorkerForeignScan; + + /* Support functions for asynchronous execution */ + IsForeignPathAsyncCapable_function IsForeignPathAsyncCapable; + ForeignAsyncRequest_function ForeignAsyncRequest; + ForeignAsyncConfigureWait_function ForeignAsyncConfigureWait; + ForeignAsyncNotify_function ForeignAsyncNotify; + ShutdownForeignScan_function ShutdownForeignScan; } FdwRoutine; diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index fa99244..735a157 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -395,6 +395,32 @@ typedef struct ResultRelInfo } ResultRelInfo; /* ---------------- + * PendingAsyncRequest + * + * State for an asynchronous tuple request. + * ---------------- + */ +typedef enum AsyncRequestState +{ + ASYNCREQ_IDLE, /* Nothing is requested */ + ASYNCREQ_WAITING, /* Waiting for events */ + ASYNCREQ_CALLBACK_PENDING, /* Having events to be processed */ + ASYNCREQ_COMPLETE /* Result is available */ +} AsyncRequestState; + +typedef struct PendingAsyncRequest +{ + int myindex; /* Index in es_pending_async. */ + struct PlanState *requestor; /* Node that wants a tuple. */ + struct PlanState *requestee; /* Node from which a tuple is wanted. */ + int request_index; /* Scratch space for requestor. */ + int num_fd_events; /* Max number of FD events requestee needs. */ + bool wants_process_latch; /* Requestee cares about MyLatch. */ + AsyncRequestState state; + Node *result; /* Result (NULL if no more tuples). */ +} PendingAsyncRequest; + +/* ---------------- * EState information * * Master working state for an Executor invocation @@ -476,6 +502,32 @@ typedef struct EState /* The per-query shared memory area to use for parallel execution. */ struct dsa_area *es_query_dsa; + + /* + * Support for asynchronous execution. + * + * es_max_pending_async is the allocated size of es_pending_async, and + * es_num_pending_aync is the number of entries that are currently valid. + * (Entries after that may point to storage that can be reused.) + * es_async_ready is the number of PendingAsyncRequests that is ready to + * retrieve a tuple. + * + * es_total_fd_events is the total number of FD events needed by all + * pending async nodes, and es_allocated_fd_events is the number any + * current wait event set was allocated to handle. es_wait_event_set, if + * non-NULL, is a previously allocated event set that may be reusable by a + * future wait provided that nothing's been removed and not too many more + * events have been added. + */ + int es_num_pending_async; /* # of nodes to wait */ + int es_max_pending_async; /* max # of pending nodes */ + int es_async_callback_pending; /* # of nodes to callback */ + int es_num_async_ready; /* # of tuple-ready nodes */ + PendingAsyncRequest **es_pending_async; + + int es_total_fd_events; + int es_allocated_fd_events; + struct WaitEventSet *es_wait_event_set; } EState; @@ -939,17 +991,20 @@ typedef struct ModifyTableState /* ---------------- * AppendState information - * - * nplans how many plans are in the array - * whichplan which plan is being executed (0 .. n-1) * ---------------- */ typedef struct AppendState { PlanState ps; /* its first field is NodeTag */ PlanState **appendplans; /* array of PlanStates for my inputs */ - int as_nplans; - int as_whichplan; + int as_nplans; /* total # of children */ + int as_nasyncplans; /* # of async-capable children */ + int as_whichsyncplan; /* which sync plan is being executed */ + bool as_syncdone; /* all synchronous plans done? */ + Bitmapset *as_needrequest; /* async plans needing a new request */ + TupleTableSlot **as_asyncresult; /* unreturned results of async plans */ + int as_nasyncresult; /* # of valid entries in as_asyncresult */ + int as_nasyncpending; /* # of outstanding async requests */ } AppendState; /* ---------------- diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index a2dd26f..15f4de9 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -235,6 +235,8 @@ typedef struct Append /* RT indexes of non-leaf tables in a partition tree */ List *partitioned_rels; List *appendplans; + int nasyncplans; /* # of async plans, always at start of list */ + int referent; /* index of inheritance tree referent */ } Append; /* ---------------- diff --git a/src/include/pgstat.h b/src/include/pgstat.h index e29397f..8bcfcb2 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -811,7 +811,8 @@ typedef enum WAIT_EVENT_SAFE_SNAPSHOT, WAIT_EVENT_SYNC_REP, WAIT_EVENT_LOGICAL_SYNC_DATA, - WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE + WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE, + WAIT_EVENT_ASYNC_WAIT } WaitEventIPC; /* ---------- -- 2.9.2