agora inbox for pgsql-hackers@postgresql.orghelp / color / mirror / Atom feed
[PATCH 3/7] Modify async execution infrastructure. 6+ messages / 2 participants [nested] [flat]
* [PATCH 3/7] Modify async execution infrastructure. @ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 0 siblings, 0 replies; 6+ messages in thread From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw) --- contrib/postgres_fdw/expected/postgres_fdw.out | 68 ++++++++-------- contrib/postgres_fdw/postgres_fdw.c | 5 +- src/backend/executor/execAsync.c | 105 ++++++++++++++----------- src/backend/executor/nodeAppend.c | 50 ++++++------ src/backend/executor/nodeForeignscan.c | 4 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/outfuncs.c | 1 + src/backend/nodes/readfuncs.c | 1 + src/backend/optimizer/plan/createplan.c | 24 +++++- src/backend/utils/adt/ruleutils.c | 6 +- src/include/executor/nodeForeignscan.h | 2 +- src/include/foreign/fdwapi.h | 2 +- src/include/nodes/execnodes.h | 10 ++- src/include/nodes/plannodes.h | 1 + 14 files changed, 167 insertions(+), 113 deletions(-) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 1b36579..a98e138 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -6321,13 +6321,13 @@ insert into bar2 values(4,44,44); insert into bar2 values(7,77,77); explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for update; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6335,10 +6335,10 @@ select * from bar where f1 in (select f1 from foo) for update; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6358,13 +6358,13 @@ select * from bar where f1 in (select f1 from foo) for update; explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for share; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6372,10 +6372,10 @@ select * from bar where f1 in (select f1 from foo) for share; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6396,22 +6396,22 @@ select * from bar where f1 in (select f1 from foo) for share; -- Check UPDATE with inherited target and an inherited source table explain (verbose, costs off) update bar set f2 = f2 + 100 where f1 in (select f1 from foo); - QUERY PLAN ---------------------------------------------------------------------------------------------------------- + QUERY PLAN +--------------------------------------------------------------------------------------------- Update on public.bar Update on public.bar Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar.f1 = foo2.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6419,16 +6419,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo); -> Seq Scan on public.foo Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> Hash Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar2.f1 = foo.f1) -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6462,8 +6462,8 @@ where bar.f1 = ss.f1; Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1)) - Hash Cond: (foo2.f1 = bar.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1)) + Hash Cond: (foo.f1 = bar.f1) -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 @@ -6480,8 +6480,8 @@ where bar.f1 = ss.f1; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Merge Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1)) - Merge Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1)) + Merge Cond: (bar2.f1 = foo.f1) -> Sort Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Sort Key: bar2.f1 @@ -6489,8 +6489,8 @@ where bar.f1 = ss.f1; Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Sort - Output: (ROW(foo2.f1)), foo2.f1 - Sort Key: foo2.f1 + Output: (ROW(foo.f1)), foo.f1 + Sort Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index e75f8a1..830212f 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root, static bool postgresIsForeignPathAsyncCapable(ForeignPath *path); static void postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq); -static void postgresForeignAsyncConfigureWait(EState *estate, +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void postgresForeignAsyncNotify(EState *estate, @@ -4479,11 +4479,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq) ExecAsyncRequestDone(estate, areq, (Node *) slot); } -static void +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { elog(ERROR, "postgresForeignAsyncConfigureWait"); + return false; } static void diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c index e070c26..33496a9 100644 --- a/src/backend/executor/execAsync.c +++ b/src/backend/executor/execAsync.c @@ -22,7 +22,7 @@ #include "storage/latch.h" static bool ExecAsyncEventWait(EState *estate, long timeout); -static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq); static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq); @@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, PlanState *requestee) { PendingAsyncRequest *areq = NULL; - int i = estate->es_num_pending_async; + int nasync = estate->es_num_pending_async; /* * If the number of pending asynchronous nodes exceeds the number of @@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * We start with 16 slots, and thereafter double the array size each * time we run out of slots. */ - if (i >= estate->es_max_pending_async) + if (nasync >= estate->es_max_pending_async) { int newmax; @@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * PendingAsyncRequest if there is one. If not, we must allocate a new * one. */ - if (estate->es_pending_async[i] == NULL) + if (estate->es_pending_async[nasync] == NULL) { areq = MemoryContextAllocZero(estate->es_query_cxt, sizeof(PendingAsyncRequest)); - estate->es_pending_async[i] = areq; + estate->es_pending_async[nasync] = areq; } else { - areq = estate->es_pending_async[i]; + areq = estate->es_pending_async[nasync]; MemSet(areq, 0, sizeof(PendingAsyncRequest)); } - areq->myindex = estate->es_num_pending_async++; + areq->myindex = estate->es_num_pending_async; /* Initialize the new request. */ areq->requestor = requestor; areq->request_index = request_index; areq->requestee = requestee; - /* Give the requestee a chance to do whatever it wants. */ + /* + * Give the requestee a chance to do whatever it wants. + * Requst functions return true if a result is immediately available. + */ switch (nodeTag(requestee)) { case T_ForeignScanState: @@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, elog(ERROR, "unrecognized node type: %d", (int) nodeTag(requestee)); } + + /* + * If a result is available, complete it immediately. + */ + if (areq->state == ASYNC_COMPLETE) + { + Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot)); + ExecAsyncResponse(estate, areq); + + return; + } + + /* No result available now, make this node pending */ + estate->es_num_pending_async++; } /* @@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) { PendingAsyncRequest *areq = estate->es_pending_async[i]; - /* Skip it if no callback is pending. */ - if (!areq->callback_pending) - continue; - - /* - * Mark it as no longer needing a callback. We must do this - * before dispatching the callback in case the callback resets - * the flag. - */ - areq->callback_pending = false; - estate->es_async_callback_pending--; - - /* Perform the actual callback; set request_done if appropraite. */ - if (!areq->request_complete) + /* Skip it if not pending. */ + if (areq->state == ASYNC_CALLBACK_PENDING) + { + /* + * Mark it as no longer needing a callback. We must do this + * before dispatching the callback in case the callback resets + * the flag. + */ + estate->es_async_callback_pending--; ExecAsyncNotify(estate, areq); - else + } + + if (areq->state == ASYNC_COMPLETE) { any_node_done = true; if (requestor == areq->requestor) @@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) PendingAsyncRequest *head; PendingAsyncRequest *tail = estate->es_pending_async[tidx]; - if (!tail->callback_pending && tail->request_complete) + if (tail->state == ASYNC_COMPLETE) continue; head = estate->es_pending_async[hidx]; estate->es_pending_async[tidx] = head; @@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) * means wait forever, 0 means don't wait at all, and >0 means wait for the * indicated number of milliseconds. * - * Returns true if we found some events and false if we timed out. + * Returns true if we found some events and false if we timed out or there's + * no event to wait. The latter is occur when the areq is processed during */ static bool ExecAsyncEventWait(EState *estate, long timeout) @@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout) int n; bool reinit = false; bool process_latch_set = false; + bool added = false; if (estate->es_wait_event_set == NULL) { @@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout) PendingAsyncRequest *areq = estate->es_pending_async[i]; if (areq->num_fd_events > 0) - ExecAsyncConfigureWait(estate, areq, reinit); + added |= ExecAsyncConfigureWait(estate, areq, reinit); } + Assert(added); + /* 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; @@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout) { PendingAsyncRequest *areq = w->user_data; - if (!areq->callback_pending) - { - Assert(!areq->request_complete); - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + Assert(areq->state == ASYNC_WAITING); + + areq->state = ASYNC_CALLBACK_PENDING; + estate->es_async_callback_pending++; } } @@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout) if (areq->wants_process_latch) { - Assert(!areq->request_complete); - areq->callback_pending = true; + Assert(areq->state == ASYNC_WAITING); + areq->state = ASYNC_CALLBACK_PENDING; } } } @@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout) * 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 void +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { switch (nodeTag(areq->requestee)) { case T_ForeignScanState: - ExecAsyncForeignScanConfigureWait(estate, areq, reinit); + return ExecAsyncForeignScanConfigureWait(estate, areq, reinit); break; default: elog(ERROR, "unrecognized node type: %d", @@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq, 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 = ASYNC_WAITING; if (force_reset && estate->es_wait_event_set != NULL) { @@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result) * 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 == ASYNC_IDLE || areq->state == ASYNC_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->request_complete = true; - - /* Make sure this request is flagged for a callback. */ - if (!areq->callback_pending) - { - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + areq->state = ASYNC_COMPLETE; } diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index bb06569..c234f1f 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -229,9 +229,15 @@ ExecAppend(AppendState *node) */ while ((i = bms_first_member(node->as_needrequest)) >= 0) { - ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); node->as_nasyncpending++; + + ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); + /* If this request immediately gives a result, take it. */ + if (node->as_nasyncresult > 0) + return node->as_asyncresult[--node->as_nasyncresult]; } + if (node->as_nasyncpending == 0 && node->as_syncdone) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); } for (;;) @@ -246,32 +252,32 @@ ExecAppend(AppendState *node) { long timeout = node->as_syncdone ? -1 : 0; - for (;;) + while (node->as_nasyncpending > 0) { - if (node->as_nasyncpending == 0) - { - /* - * 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) - return ExecClearTuple(node->ps.ps_ResultTupleSlot); - break; - } - if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout)) - { - /* Timeout reached. */ - break; - } - if (node->as_nasyncresult > 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); } } @@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq) TupleTableSlot *slot; /* We shouldn't be called until the request is complete. */ - Assert(areq->request_complete); + Assert(areq->state == ASYNC_COMPLETE); /* Our result slot shouldn't already be occupied. */ Assert(TupIsNull(node->ps.ps_ResultTupleSlot)); diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 85d436f..d3567bb 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq) * In async mode, configure for a wait * ---------------------------------------------------------------- */ -void +bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { @@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate, FdwRoutine *fdwroutine = node->fdwroutine; Assert(fdwroutine->ForeignAsyncConfigureWait != NULL); - fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); + return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); } /* ---------------------------------------------------------------- diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 23b4e18..72d8cd6 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -219,6 +219,7 @@ _copyAppend(const Append *from) */ 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 dc5b938..1ebdc48 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node) 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 69453b5..8443a62 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1520,6 +1520,7 @@ _readAppend(void) 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 7caa8d3..ff1d663 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual, Index scanrelid, int ctePlanId, int cteParam); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *asyncplans, int nasyncplans, List *tlist); +static Append *make_append(List *asyncplans, int nasyncplans, + int referent, List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) 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 @@ -985,7 +988,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); @@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) { asyncplans = lappend(asyncplans, subplan); ++nasyncplans; + if (first) + referent_is_sync = false; } else syncplans = lappend(syncplans, subplan); + + first = false; } /* @@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist); + plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, + referent_is_sync ? nasyncplans : 0, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -4951,7 +4966,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, int nasyncplans, List *tlist) +make_append(List *appendplans, int nasyncplans, int referent, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -4962,6 +4977,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist) plan->righttree = NULL; node->appendplans = appendplans; node->nasyncplans = nasyncplans; + node->referent = referent; return node; } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 8a81d7a..de0e96c 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -4056,7 +4056,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/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 3e69ab0..47a3920 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node, extern void ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq); -extern void ExecAsyncForeignScanConfigureWait(EState *estate, +extern bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); extern void ExecAsyncForeignScanNotify(EState *estate, PendingAsyncRequest *areq); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 88feb9a..65517fd 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root, typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path); typedef void (*ForeignAsyncRequest_function) (EState *estate, PendingAsyncRequest *areq); -typedef void (*ForeignAsyncConfigureWait_function) (EState *estate, +typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate, PendingAsyncRequest *areq, bool reinit); typedef void (*ForeignAsyncNotify_function) (EState *estate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index b50b41c..0c6af86 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -352,6 +352,13 @@ typedef struct ResultRelInfo * State for an asynchronous tuple request. * ---------------- */ +typedef enum AsyncRequestState +{ + ASYNC_IDLE, + ASYNC_WAITING, + ASYNC_CALLBACK_PENDING, + ASYNC_COMPLETE +} AsyncRequestState; typedef struct PendingAsyncRequest { int myindex; /* Index in es_pending_async. */ @@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest 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. */ - bool callback_pending; /* Callback is needed. */ - bool request_complete; /* Request complete, result valid. */ + AsyncRequestState state; Node *result; /* Result (NULL if no more tuples). */ } PendingAsyncRequest; diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index 327119b..1df6693 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -209,6 +209,7 @@ typedef struct Append Plan plan; List *appendplans; int nasyncplans; /* # of async plans, always at start of list */ + int referent; /* index of inheritance tree referent */ } Append; /* ---------------- -- 2.9.2 ----Next_Part(Mon_Oct_31_10_39_12_2016_712)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="0004-Make-postgres_fdw-async-capable.patch" ^ permalink raw reply [nested|flat] 6+ messages in thread
* [PATCH 03/13] Modify async execution infrastructure. @ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 0 siblings, 0 replies; 6+ messages in thread From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw) --- contrib/postgres_fdw/expected/postgres_fdw.out | 68 ++++++++-------- contrib/postgres_fdw/postgres_fdw.c | 5 +- src/backend/executor/execAsync.c | 105 ++++++++++++++----------- src/backend/executor/nodeAppend.c | 50 ++++++------ src/backend/executor/nodeForeignscan.c | 4 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/outfuncs.c | 1 + src/backend/nodes/readfuncs.c | 1 + src/backend/optimizer/plan/createplan.c | 24 +++++- src/backend/utils/adt/ruleutils.c | 6 +- src/include/executor/nodeForeignscan.h | 2 +- src/include/foreign/fdwapi.h | 2 +- src/include/nodes/execnodes.h | 10 ++- src/include/nodes/plannodes.h | 1 + 14 files changed, 167 insertions(+), 113 deletions(-) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index df22beb..9180afe 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -6402,13 +6402,13 @@ insert into bar2 values(4,44,44); insert into bar2 values(7,77,77); explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for update; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6416,10 +6416,10 @@ select * from bar where f1 in (select f1 from foo) for update; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6439,13 +6439,13 @@ select * from bar where f1 in (select f1 from foo) for update; explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for share; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6453,10 +6453,10 @@ select * from bar where f1 in (select f1 from foo) for share; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6477,22 +6477,22 @@ select * from bar where f1 in (select f1 from foo) for share; -- Check UPDATE with inherited target and an inherited source table explain (verbose, costs off) update bar set f2 = f2 + 100 where f1 in (select f1 from foo); - QUERY PLAN ---------------------------------------------------------------------------------------------------------- + QUERY PLAN +--------------------------------------------------------------------------------------------- Update on public.bar Update on public.bar Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar.f1 = foo2.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6500,16 +6500,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo); -> Seq Scan on public.foo Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> Hash Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar2.f1 = foo.f1) -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6543,8 +6543,8 @@ where bar.f1 = ss.f1; Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1)) - Hash Cond: (foo2.f1 = bar.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1)) + Hash Cond: (foo.f1 = bar.f1) -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 @@ -6561,8 +6561,8 @@ where bar.f1 = ss.f1; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Merge Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1)) - Merge Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1)) + Merge Cond: (bar2.f1 = foo.f1) -> Sort Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Sort Key: bar2.f1 @@ -6570,8 +6570,8 @@ where bar.f1 = ss.f1; Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Sort - Output: (ROW(foo2.f1)), foo2.f1 - Sort Key: foo2.f1 + Output: (ROW(foo.f1)), foo.f1 + Sort Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index f180838..abb256b 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root, static bool postgresIsForeignPathAsyncCapable(ForeignPath *path); static void postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq); -static void postgresForeignAsyncConfigureWait(EState *estate, +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void postgresForeignAsyncNotify(EState *estate, @@ -4477,11 +4477,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq) ExecAsyncRequestDone(estate, areq, (Node *) slot); } -static void +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { elog(ERROR, "postgresForeignAsyncConfigureWait"); + return false; } static void diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c index e070c26..33496a9 100644 --- a/src/backend/executor/execAsync.c +++ b/src/backend/executor/execAsync.c @@ -22,7 +22,7 @@ #include "storage/latch.h" static bool ExecAsyncEventWait(EState *estate, long timeout); -static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq); static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq); @@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, PlanState *requestee) { PendingAsyncRequest *areq = NULL; - int i = estate->es_num_pending_async; + int nasync = estate->es_num_pending_async; /* * If the number of pending asynchronous nodes exceeds the number of @@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * We start with 16 slots, and thereafter double the array size each * time we run out of slots. */ - if (i >= estate->es_max_pending_async) + if (nasync >= estate->es_max_pending_async) { int newmax; @@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * PendingAsyncRequest if there is one. If not, we must allocate a new * one. */ - if (estate->es_pending_async[i] == NULL) + if (estate->es_pending_async[nasync] == NULL) { areq = MemoryContextAllocZero(estate->es_query_cxt, sizeof(PendingAsyncRequest)); - estate->es_pending_async[i] = areq; + estate->es_pending_async[nasync] = areq; } else { - areq = estate->es_pending_async[i]; + areq = estate->es_pending_async[nasync]; MemSet(areq, 0, sizeof(PendingAsyncRequest)); } - areq->myindex = estate->es_num_pending_async++; + areq->myindex = estate->es_num_pending_async; /* Initialize the new request. */ areq->requestor = requestor; areq->request_index = request_index; areq->requestee = requestee; - /* Give the requestee a chance to do whatever it wants. */ + /* + * Give the requestee a chance to do whatever it wants. + * Requst functions return true if a result is immediately available. + */ switch (nodeTag(requestee)) { case T_ForeignScanState: @@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, elog(ERROR, "unrecognized node type: %d", (int) nodeTag(requestee)); } + + /* + * If a result is available, complete it immediately. + */ + if (areq->state == ASYNC_COMPLETE) + { + Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot)); + ExecAsyncResponse(estate, areq); + + return; + } + + /* No result available now, make this node pending */ + estate->es_num_pending_async++; } /* @@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) { PendingAsyncRequest *areq = estate->es_pending_async[i]; - /* Skip it if no callback is pending. */ - if (!areq->callback_pending) - continue; - - /* - * Mark it as no longer needing a callback. We must do this - * before dispatching the callback in case the callback resets - * the flag. - */ - areq->callback_pending = false; - estate->es_async_callback_pending--; - - /* Perform the actual callback; set request_done if appropraite. */ - if (!areq->request_complete) + /* Skip it if not pending. */ + if (areq->state == ASYNC_CALLBACK_PENDING) + { + /* + * Mark it as no longer needing a callback. We must do this + * before dispatching the callback in case the callback resets + * the flag. + */ + estate->es_async_callback_pending--; ExecAsyncNotify(estate, areq); - else + } + + if (areq->state == ASYNC_COMPLETE) { any_node_done = true; if (requestor == areq->requestor) @@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) PendingAsyncRequest *head; PendingAsyncRequest *tail = estate->es_pending_async[tidx]; - if (!tail->callback_pending && tail->request_complete) + if (tail->state == ASYNC_COMPLETE) continue; head = estate->es_pending_async[hidx]; estate->es_pending_async[tidx] = head; @@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) * means wait forever, 0 means don't wait at all, and >0 means wait for the * indicated number of milliseconds. * - * Returns true if we found some events and false if we timed out. + * Returns true if we found some events and false if we timed out or there's + * no event to wait. The latter is occur when the areq is processed during */ static bool ExecAsyncEventWait(EState *estate, long timeout) @@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout) int n; bool reinit = false; bool process_latch_set = false; + bool added = false; if (estate->es_wait_event_set == NULL) { @@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout) PendingAsyncRequest *areq = estate->es_pending_async[i]; if (areq->num_fd_events > 0) - ExecAsyncConfigureWait(estate, areq, reinit); + added |= ExecAsyncConfigureWait(estate, areq, reinit); } + Assert(added); + /* 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; @@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout) { PendingAsyncRequest *areq = w->user_data; - if (!areq->callback_pending) - { - Assert(!areq->request_complete); - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + Assert(areq->state == ASYNC_WAITING); + + areq->state = ASYNC_CALLBACK_PENDING; + estate->es_async_callback_pending++; } } @@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout) if (areq->wants_process_latch) { - Assert(!areq->request_complete); - areq->callback_pending = true; + Assert(areq->state == ASYNC_WAITING); + areq->state = ASYNC_CALLBACK_PENDING; } } } @@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout) * 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 void +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { switch (nodeTag(areq->requestee)) { case T_ForeignScanState: - ExecAsyncForeignScanConfigureWait(estate, areq, reinit); + return ExecAsyncForeignScanConfigureWait(estate, areq, reinit); break; default: elog(ERROR, "unrecognized node type: %d", @@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq, 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 = ASYNC_WAITING; if (force_reset && estate->es_wait_event_set != NULL) { @@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result) * 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 == ASYNC_IDLE || areq->state == ASYNC_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->request_complete = true; - - /* Make sure this request is flagged for a callback. */ - if (!areq->callback_pending) - { - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + areq->state = ASYNC_COMPLETE; } diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index e61218a..568fa25 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -229,9 +229,15 @@ ExecAppend(AppendState *node) */ while ((i = bms_first_member(node->as_needrequest)) >= 0) { - ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); node->as_nasyncpending++; + + ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); + /* If this request immediately gives a result, take it. */ + if (node->as_nasyncresult > 0) + return node->as_asyncresult[--node->as_nasyncresult]; } + if (node->as_nasyncpending == 0 && node->as_syncdone) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); } for (;;) @@ -246,32 +252,32 @@ ExecAppend(AppendState *node) { long timeout = node->as_syncdone ? -1 : 0; - for (;;) + while (node->as_nasyncpending > 0) { - if (node->as_nasyncpending == 0) - { - /* - * 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) - return ExecClearTuple(node->ps.ps_ResultTupleSlot); - break; - } - if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout)) - { - /* Timeout reached. */ - break; - } - if (node->as_nasyncresult > 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); } } @@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq) TupleTableSlot *slot; /* We shouldn't be called until the request is complete. */ - Assert(areq->request_complete); + Assert(areq->state == ASYNC_COMPLETE); /* Our result slot shouldn't already be occupied. */ Assert(TupIsNull(node->ps.ps_ResultTupleSlot)); diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 61899d1..85dad79 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -376,7 +376,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq) * In async mode, configure for a wait * ---------------------------------------------------------------- */ -void +bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { @@ -384,7 +384,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate, FdwRoutine *fdwroutine = node->fdwroutine; Assert(fdwroutine->ForeignAsyncConfigureWait != NULL); - fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); + return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); } /* ---------------------------------------------------------------- diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index a8cabdf..c62aaf2 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -237,6 +237,7 @@ _copyAppend(const Append *from) */ 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 a894a9d..c2e34a8 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -370,6 +370,7 @@ _outAppend(StringInfo str, const Append *node) 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 67439ec..9837eff 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1540,6 +1540,7 @@ _readAppend(void) 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 2140094..0575541 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -194,7 +194,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual, Index scanrelid, int ctePlanId, int cteParam); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *asyncplans, int nasyncplans, List *tlist); +static Append *make_append(List *asyncplans, int nasyncplans, + int referent, List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -966,6 +967,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) 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 @@ -991,7 +994,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); @@ -1005,9 +1015,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) { asyncplans = lappend(asyncplans, subplan); ++nasyncplans; + if (first) + referent_is_sync = false; } else syncplans = lappend(syncplans, subplan); + + first = false; } /* @@ -1017,7 +1031,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist); + plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, + referent_is_sync ? nasyncplans : 0, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -5019,7 +5034,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, int nasyncplans, List *tlist) +make_append(List *appendplans, int nasyncplans, int referent, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -5030,6 +5045,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist) plan->righttree = NULL; node->appendplans = appendplans; node->nasyncplans = nasyncplans; + node->referent = referent; return node; } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index f355954..76dd07a 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -4242,7 +4242,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/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 5a61306..2d9a62b 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node, extern void ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq); -extern void ExecAsyncForeignScanConfigureWait(EState *estate, +extern bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); extern void ExecAsyncForeignScanNotify(EState *estate, PendingAsyncRequest *areq); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 4c50f1e..41fc76f 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root, typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path); typedef void (*ForeignAsyncRequest_function) (EState *estate, PendingAsyncRequest *areq); -typedef void (*ForeignAsyncConfigureWait_function) (EState *estate, +typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate, PendingAsyncRequest *areq, bool reinit); typedef void (*ForeignAsyncNotify_function) (EState *estate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 81e997e..5afcd34 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -361,6 +361,13 @@ typedef struct ResultRelInfo * State for an asynchronous tuple request. * ---------------- */ +typedef enum AsyncRequestState +{ + ASYNC_IDLE, + ASYNC_WAITING, + ASYNC_CALLBACK_PENDING, + ASYNC_COMPLETE +} AsyncRequestState; typedef struct PendingAsyncRequest { int myindex; /* Index in es_pending_async. */ @@ -369,8 +376,7 @@ typedef struct PendingAsyncRequest 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. */ - bool callback_pending; /* Callback is needed. */ - bool request_complete; /* Request complete, result valid. */ + AsyncRequestState state; Node *result; /* Result (NULL if no more tuples). */ } PendingAsyncRequest; diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index f0daada..ebbc78d 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -229,6 +229,7 @@ typedef struct Append Plan plan; List *appendplans; int nasyncplans; /* # of async plans, always at start of list */ + int referent; /* index of inheritance tree referent */ } Append; /* ---------------- -- 2.9.2 ----Next_Part(Wed_Feb_22_17_39_45_2017_090)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="0002-Fix-some-bugs.patch" ^ permalink raw reply [nested|flat] 6+ messages in thread
* [PATCH 3/7] Modify async execution infrastructure. @ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 0 siblings, 0 replies; 6+ messages in thread From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw) --- contrib/postgres_fdw/expected/postgres_fdw.out | 68 ++++++++-------- contrib/postgres_fdw/postgres_fdw.c | 5 +- src/backend/executor/execAsync.c | 105 ++++++++++++++----------- src/backend/executor/nodeAppend.c | 50 ++++++------ src/backend/executor/nodeForeignscan.c | 4 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/outfuncs.c | 1 + src/backend/nodes/readfuncs.c | 1 + src/backend/optimizer/plan/createplan.c | 24 +++++- src/backend/utils/adt/ruleutils.c | 6 +- src/include/executor/nodeForeignscan.h | 2 +- src/include/foreign/fdwapi.h | 2 +- src/include/nodes/execnodes.h | 10 ++- src/include/nodes/plannodes.h | 1 + 14 files changed, 167 insertions(+), 113 deletions(-) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 457cfdb..083d947 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -6321,13 +6321,13 @@ insert into bar2 values(4,44,44); insert into bar2 values(7,77,77); explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for update; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6335,10 +6335,10 @@ select * from bar where f1 in (select f1 from foo) for update; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6358,13 +6358,13 @@ select * from bar where f1 in (select f1 from foo) for update; explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for share; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6372,10 +6372,10 @@ select * from bar where f1 in (select f1 from foo) for share; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6396,22 +6396,22 @@ select * from bar where f1 in (select f1 from foo) for share; -- Check UPDATE with inherited target and an inherited source table explain (verbose, costs off) update bar set f2 = f2 + 100 where f1 in (select f1 from foo); - QUERY PLAN ---------------------------------------------------------------------------------------------------------- + QUERY PLAN +--------------------------------------------------------------------------------------------- Update on public.bar Update on public.bar Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar.f1 = foo2.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6419,16 +6419,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo); -> Seq Scan on public.foo Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> Hash Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar2.f1 = foo.f1) -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6462,8 +6462,8 @@ where bar.f1 = ss.f1; Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1)) - Hash Cond: (foo2.f1 = bar.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1)) + Hash Cond: (foo.f1 = bar.f1) -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 @@ -6480,8 +6480,8 @@ where bar.f1 = ss.f1; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Merge Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1)) - Merge Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1)) + Merge Cond: (bar2.f1 = foo.f1) -> Sort Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Sort Key: bar2.f1 @@ -6489,8 +6489,8 @@ where bar.f1 = ss.f1; Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Sort - Output: (ROW(foo2.f1)), foo2.f1 - Sort Key: foo2.f1 + Output: (ROW(foo.f1)), foo.f1 + Sort Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index c64ae41..b92b279 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root, static bool postgresIsForeignPathAsyncCapable(ForeignPath *path); static void postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq); -static void postgresForeignAsyncConfigureWait(EState *estate, +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void postgresForeignAsyncNotify(EState *estate, @@ -4479,11 +4479,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq) ExecAsyncRequestDone(estate, areq, (Node *) slot); } -static void +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { elog(ERROR, "postgresForeignAsyncConfigureWait"); + return false; } static void diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c index e070c26..33496a9 100644 --- a/src/backend/executor/execAsync.c +++ b/src/backend/executor/execAsync.c @@ -22,7 +22,7 @@ #include "storage/latch.h" static bool ExecAsyncEventWait(EState *estate, long timeout); -static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq); static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq); @@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, PlanState *requestee) { PendingAsyncRequest *areq = NULL; - int i = estate->es_num_pending_async; + int nasync = estate->es_num_pending_async; /* * If the number of pending asynchronous nodes exceeds the number of @@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * We start with 16 slots, and thereafter double the array size each * time we run out of slots. */ - if (i >= estate->es_max_pending_async) + if (nasync >= estate->es_max_pending_async) { int newmax; @@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * PendingAsyncRequest if there is one. If not, we must allocate a new * one. */ - if (estate->es_pending_async[i] == NULL) + if (estate->es_pending_async[nasync] == NULL) { areq = MemoryContextAllocZero(estate->es_query_cxt, sizeof(PendingAsyncRequest)); - estate->es_pending_async[i] = areq; + estate->es_pending_async[nasync] = areq; } else { - areq = estate->es_pending_async[i]; + areq = estate->es_pending_async[nasync]; MemSet(areq, 0, sizeof(PendingAsyncRequest)); } - areq->myindex = estate->es_num_pending_async++; + areq->myindex = estate->es_num_pending_async; /* Initialize the new request. */ areq->requestor = requestor; areq->request_index = request_index; areq->requestee = requestee; - /* Give the requestee a chance to do whatever it wants. */ + /* + * Give the requestee a chance to do whatever it wants. + * Requst functions return true if a result is immediately available. + */ switch (nodeTag(requestee)) { case T_ForeignScanState: @@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, elog(ERROR, "unrecognized node type: %d", (int) nodeTag(requestee)); } + + /* + * If a result is available, complete it immediately. + */ + if (areq->state == ASYNC_COMPLETE) + { + Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot)); + ExecAsyncResponse(estate, areq); + + return; + } + + /* No result available now, make this node pending */ + estate->es_num_pending_async++; } /* @@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) { PendingAsyncRequest *areq = estate->es_pending_async[i]; - /* Skip it if no callback is pending. */ - if (!areq->callback_pending) - continue; - - /* - * Mark it as no longer needing a callback. We must do this - * before dispatching the callback in case the callback resets - * the flag. - */ - areq->callback_pending = false; - estate->es_async_callback_pending--; - - /* Perform the actual callback; set request_done if appropraite. */ - if (!areq->request_complete) + /* Skip it if not pending. */ + if (areq->state == ASYNC_CALLBACK_PENDING) + { + /* + * Mark it as no longer needing a callback. We must do this + * before dispatching the callback in case the callback resets + * the flag. + */ + estate->es_async_callback_pending--; ExecAsyncNotify(estate, areq); - else + } + + if (areq->state == ASYNC_COMPLETE) { any_node_done = true; if (requestor == areq->requestor) @@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) PendingAsyncRequest *head; PendingAsyncRequest *tail = estate->es_pending_async[tidx]; - if (!tail->callback_pending && tail->request_complete) + if (tail->state == ASYNC_COMPLETE) continue; head = estate->es_pending_async[hidx]; estate->es_pending_async[tidx] = head; @@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) * means wait forever, 0 means don't wait at all, and >0 means wait for the * indicated number of milliseconds. * - * Returns true if we found some events and false if we timed out. + * Returns true if we found some events and false if we timed out or there's + * no event to wait. The latter is occur when the areq is processed during */ static bool ExecAsyncEventWait(EState *estate, long timeout) @@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout) int n; bool reinit = false; bool process_latch_set = false; + bool added = false; if (estate->es_wait_event_set == NULL) { @@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout) PendingAsyncRequest *areq = estate->es_pending_async[i]; if (areq->num_fd_events > 0) - ExecAsyncConfigureWait(estate, areq, reinit); + added |= ExecAsyncConfigureWait(estate, areq, reinit); } + Assert(added); + /* 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; @@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout) { PendingAsyncRequest *areq = w->user_data; - if (!areq->callback_pending) - { - Assert(!areq->request_complete); - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + Assert(areq->state == ASYNC_WAITING); + + areq->state = ASYNC_CALLBACK_PENDING; + estate->es_async_callback_pending++; } } @@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout) if (areq->wants_process_latch) { - Assert(!areq->request_complete); - areq->callback_pending = true; + Assert(areq->state == ASYNC_WAITING); + areq->state = ASYNC_CALLBACK_PENDING; } } } @@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout) * 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 void +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { switch (nodeTag(areq->requestee)) { case T_ForeignScanState: - ExecAsyncForeignScanConfigureWait(estate, areq, reinit); + return ExecAsyncForeignScanConfigureWait(estate, areq, reinit); break; default: elog(ERROR, "unrecognized node type: %d", @@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq, 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 = ASYNC_WAITING; if (force_reset && estate->es_wait_event_set != NULL) { @@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result) * 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 == ASYNC_IDLE || areq->state == ASYNC_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->request_complete = true; - - /* Make sure this request is flagged for a callback. */ - if (!areq->callback_pending) - { - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + areq->state = ASYNC_COMPLETE; } diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index bb06569..c234f1f 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -229,9 +229,15 @@ ExecAppend(AppendState *node) */ while ((i = bms_first_member(node->as_needrequest)) >= 0) { - ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); node->as_nasyncpending++; + + ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); + /* If this request immediately gives a result, take it. */ + if (node->as_nasyncresult > 0) + return node->as_asyncresult[--node->as_nasyncresult]; } + if (node->as_nasyncpending == 0 && node->as_syncdone) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); } for (;;) @@ -246,32 +252,32 @@ ExecAppend(AppendState *node) { long timeout = node->as_syncdone ? -1 : 0; - for (;;) + while (node->as_nasyncpending > 0) { - if (node->as_nasyncpending == 0) - { - /* - * 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) - return ExecClearTuple(node->ps.ps_ResultTupleSlot); - break; - } - if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout)) - { - /* Timeout reached. */ - break; - } - if (node->as_nasyncresult > 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); } } @@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq) TupleTableSlot *slot; /* We shouldn't be called until the request is complete. */ - Assert(areq->request_complete); + Assert(areq->state == ASYNC_COMPLETE); /* Our result slot shouldn't already be occupied. */ Assert(TupIsNull(node->ps.ps_ResultTupleSlot)); diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 85d436f..d3567bb 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq) * In async mode, configure for a wait * ---------------------------------------------------------------- */ -void +bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { @@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate, FdwRoutine *fdwroutine = node->fdwroutine; Assert(fdwroutine->ForeignAsyncConfigureWait != NULL); - fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); + return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); } /* ---------------------------------------------------------------- diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index e4a103f..27ccf9d 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -219,6 +219,7 @@ _copyAppend(const Append *from) */ 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 1566e0d..c8b9f31 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node) 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 69453b5..8443a62 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1520,6 +1520,7 @@ _readAppend(void) 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 7caa8d3..ff1d663 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual, Index scanrelid, int ctePlanId, int cteParam); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *asyncplans, int nasyncplans, List *tlist); +static Append *make_append(List *asyncplans, int nasyncplans, + int referent, List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) 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 @@ -985,7 +988,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); @@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) { asyncplans = lappend(asyncplans, subplan); ++nasyncplans; + if (first) + referent_is_sync = false; } else syncplans = lappend(syncplans, subplan); + + first = false; } /* @@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist); + plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, + referent_is_sync ? nasyncplans : 0, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -4951,7 +4966,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, int nasyncplans, List *tlist) +make_append(List *appendplans, int nasyncplans, int referent, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -4962,6 +4977,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist) plan->righttree = NULL; node->appendplans = appendplans; node->nasyncplans = nasyncplans; + node->referent = referent; return node; } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index a3a4174..9a2ee83 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -4079,7 +4079,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/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 3e69ab0..47a3920 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node, extern void ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq); -extern void ExecAsyncForeignScanConfigureWait(EState *estate, +extern bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); extern void ExecAsyncForeignScanNotify(EState *estate, PendingAsyncRequest *areq); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 88feb9a..65517fd 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root, typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path); typedef void (*ForeignAsyncRequest_function) (EState *estate, PendingAsyncRequest *areq); -typedef void (*ForeignAsyncConfigureWait_function) (EState *estate, +typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate, PendingAsyncRequest *areq, bool reinit); typedef void (*ForeignAsyncNotify_function) (EState *estate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index b50b41c..0c6af86 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -352,6 +352,13 @@ typedef struct ResultRelInfo * State for an asynchronous tuple request. * ---------------- */ +typedef enum AsyncRequestState +{ + ASYNC_IDLE, + ASYNC_WAITING, + ASYNC_CALLBACK_PENDING, + ASYNC_COMPLETE +} AsyncRequestState; typedef struct PendingAsyncRequest { int myindex; /* Index in es_pending_async. */ @@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest 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. */ - bool callback_pending; /* Callback is needed. */ - bool request_complete; /* Request complete, result valid. */ + AsyncRequestState state; Node *result; /* Result (NULL if no more tuples). */ } PendingAsyncRequest; diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index 327119b..1df6693 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -209,6 +209,7 @@ typedef struct Append Plan plan; List *appendplans; int nasyncplans; /* # of async plans, always at start of list */ + int referent; /* index of inheritance tree referent */ } Append; /* ---------------- -- 2.9.2 ----Next_Part(Tue_Nov_15_20_25_13_2016_668)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="0004-Make-postgres_fdw-async-capable.patch" ^ permalink raw reply [nested|flat] 6+ messages in thread
* [PATCH 3/4] Modify async execution infrastructure. @ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 0 siblings, 0 replies; 6+ messages in thread From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw) --- contrib/postgres_fdw/expected/postgres_fdw.out | 68 ++++++++-------- contrib/postgres_fdw/postgres_fdw.c | 5 +- src/backend/executor/execAsync.c | 105 ++++++++++++++----------- src/backend/executor/nodeAppend.c | 50 ++++++------ src/backend/executor/nodeForeignscan.c | 4 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/outfuncs.c | 1 + src/backend/nodes/readfuncs.c | 1 + src/backend/optimizer/plan/createplan.c | 24 +++++- src/backend/utils/adt/ruleutils.c | 6 +- src/include/executor/nodeForeignscan.h | 2 +- src/include/foreign/fdwapi.h | 2 +- src/include/nodes/execnodes.h | 10 ++- src/include/nodes/plannodes.h | 1 + 14 files changed, 167 insertions(+), 113 deletions(-) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 6677bc4..d429790 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -5230,13 +5230,13 @@ insert into bar2 values(4,44,44); insert into bar2 values(7,77,77); explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for update; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -5244,10 +5244,10 @@ select * from bar where f1 in (select f1 from foo) for update; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -5267,13 +5267,13 @@ select * from bar where f1 in (select f1 from foo) for update; explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for share; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -5281,10 +5281,10 @@ select * from bar where f1 in (select f1 from foo) for share; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -5305,22 +5305,22 @@ select * from bar where f1 in (select f1 from foo) for share; -- Check UPDATE with inherited target and an inherited source table explain (verbose, costs off) update bar set f2 = f2 + 100 where f1 in (select f1 from foo); - QUERY PLAN ---------------------------------------------------------------------------------------------------------- + QUERY PLAN +--------------------------------------------------------------------------------------------- Update on public.bar Update on public.bar Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar.f1 = foo2.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -5328,16 +5328,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo); -> Seq Scan on public.foo Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> Hash Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar2.f1 = foo.f1) -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -5371,8 +5371,8 @@ where bar.f1 = ss.f1; Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1)) - Hash Cond: (foo2.f1 = bar.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1)) + Hash Cond: (foo.f1 = bar.f1) -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 @@ -5389,8 +5389,8 @@ where bar.f1 = ss.f1; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Merge Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1)) - Merge Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1)) + Merge Cond: (bar2.f1 = foo.f1) -> Sort Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Sort Key: bar2.f1 @@ -5398,8 +5398,8 @@ where bar.f1 = ss.f1; Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Sort - Output: (ROW(foo2.f1)), foo2.f1 - Sort Key: foo2.f1 + Output: (ROW(foo.f1)), foo.f1 + Sort Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 6da5843..997bd6c 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -348,7 +348,7 @@ static bool postgresRecheckForeignScan(ForeignScanState *node, static bool postgresIsForeignPathAsyncCapable(ForeignPath *path); static void postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq); -static void postgresForeignAsyncConfigureWait(EState *estate, +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void postgresForeignAsyncNotify(EState *estate, @@ -4379,11 +4379,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq) ExecAsyncRequestDone(estate, areq, (Node *) slot); } -static void +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { elog(ERROR, "postgresForeignAsyncConfigureWait"); + return false; } static void diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c index e070c26..33496a9 100644 --- a/src/backend/executor/execAsync.c +++ b/src/backend/executor/execAsync.c @@ -22,7 +22,7 @@ #include "storage/latch.h" static bool ExecAsyncEventWait(EState *estate, long timeout); -static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq); static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq); @@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, PlanState *requestee) { PendingAsyncRequest *areq = NULL; - int i = estate->es_num_pending_async; + int nasync = estate->es_num_pending_async; /* * If the number of pending asynchronous nodes exceeds the number of @@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * We start with 16 slots, and thereafter double the array size each * time we run out of slots. */ - if (i >= estate->es_max_pending_async) + if (nasync >= estate->es_max_pending_async) { int newmax; @@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * PendingAsyncRequest if there is one. If not, we must allocate a new * one. */ - if (estate->es_pending_async[i] == NULL) + if (estate->es_pending_async[nasync] == NULL) { areq = MemoryContextAllocZero(estate->es_query_cxt, sizeof(PendingAsyncRequest)); - estate->es_pending_async[i] = areq; + estate->es_pending_async[nasync] = areq; } else { - areq = estate->es_pending_async[i]; + areq = estate->es_pending_async[nasync]; MemSet(areq, 0, sizeof(PendingAsyncRequest)); } - areq->myindex = estate->es_num_pending_async++; + areq->myindex = estate->es_num_pending_async; /* Initialize the new request. */ areq->requestor = requestor; areq->request_index = request_index; areq->requestee = requestee; - /* Give the requestee a chance to do whatever it wants. */ + /* + * Give the requestee a chance to do whatever it wants. + * Requst functions return true if a result is immediately available. + */ switch (nodeTag(requestee)) { case T_ForeignScanState: @@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, elog(ERROR, "unrecognized node type: %d", (int) nodeTag(requestee)); } + + /* + * If a result is available, complete it immediately. + */ + if (areq->state == ASYNC_COMPLETE) + { + Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot)); + ExecAsyncResponse(estate, areq); + + return; + } + + /* No result available now, make this node pending */ + estate->es_num_pending_async++; } /* @@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) { PendingAsyncRequest *areq = estate->es_pending_async[i]; - /* Skip it if no callback is pending. */ - if (!areq->callback_pending) - continue; - - /* - * Mark it as no longer needing a callback. We must do this - * before dispatching the callback in case the callback resets - * the flag. - */ - areq->callback_pending = false; - estate->es_async_callback_pending--; - - /* Perform the actual callback; set request_done if appropraite. */ - if (!areq->request_complete) + /* Skip it if not pending. */ + if (areq->state == ASYNC_CALLBACK_PENDING) + { + /* + * Mark it as no longer needing a callback. We must do this + * before dispatching the callback in case the callback resets + * the flag. + */ + estate->es_async_callback_pending--; ExecAsyncNotify(estate, areq); - else + } + + if (areq->state == ASYNC_COMPLETE) { any_node_done = true; if (requestor == areq->requestor) @@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) PendingAsyncRequest *head; PendingAsyncRequest *tail = estate->es_pending_async[tidx]; - if (!tail->callback_pending && tail->request_complete) + if (tail->state == ASYNC_COMPLETE) continue; head = estate->es_pending_async[hidx]; estate->es_pending_async[tidx] = head; @@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) * means wait forever, 0 means don't wait at all, and >0 means wait for the * indicated number of milliseconds. * - * Returns true if we found some events and false if we timed out. + * Returns true if we found some events and false if we timed out or there's + * no event to wait. The latter is occur when the areq is processed during */ static bool ExecAsyncEventWait(EState *estate, long timeout) @@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout) int n; bool reinit = false; bool process_latch_set = false; + bool added = false; if (estate->es_wait_event_set == NULL) { @@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout) PendingAsyncRequest *areq = estate->es_pending_async[i]; if (areq->num_fd_events > 0) - ExecAsyncConfigureWait(estate, areq, reinit); + added |= ExecAsyncConfigureWait(estate, areq, reinit); } + Assert(added); + /* 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; @@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout) { PendingAsyncRequest *areq = w->user_data; - if (!areq->callback_pending) - { - Assert(!areq->request_complete); - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + Assert(areq->state == ASYNC_WAITING); + + areq->state = ASYNC_CALLBACK_PENDING; + estate->es_async_callback_pending++; } } @@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout) if (areq->wants_process_latch) { - Assert(!areq->request_complete); - areq->callback_pending = true; + Assert(areq->state == ASYNC_WAITING); + areq->state = ASYNC_CALLBACK_PENDING; } } } @@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout) * 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 void +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { switch (nodeTag(areq->requestee)) { case T_ForeignScanState: - ExecAsyncForeignScanConfigureWait(estate, areq, reinit); + return ExecAsyncForeignScanConfigureWait(estate, areq, reinit); break; default: elog(ERROR, "unrecognized node type: %d", @@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq, 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 = ASYNC_WAITING; if (force_reset && estate->es_wait_event_set != NULL) { @@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result) * 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 == ASYNC_IDLE || areq->state == ASYNC_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->request_complete = true; - - /* Make sure this request is flagged for a callback. */ - if (!areq->callback_pending) - { - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + areq->state = ASYNC_COMPLETE; } diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index bb06569..c234f1f 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -229,9 +229,15 @@ ExecAppend(AppendState *node) */ while ((i = bms_first_member(node->as_needrequest)) >= 0) { - ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); node->as_nasyncpending++; + + ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); + /* If this request immediately gives a result, take it. */ + if (node->as_nasyncresult > 0) + return node->as_asyncresult[--node->as_nasyncresult]; } + if (node->as_nasyncpending == 0 && node->as_syncdone) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); } for (;;) @@ -246,32 +252,32 @@ ExecAppend(AppendState *node) { long timeout = node->as_syncdone ? -1 : 0; - for (;;) + while (node->as_nasyncpending > 0) { - if (node->as_nasyncpending == 0) - { - /* - * 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) - return ExecClearTuple(node->ps.ps_ResultTupleSlot); - break; - } - if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout)) - { - /* Timeout reached. */ - break; - } - if (node->as_nasyncresult > 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); } } @@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq) TupleTableSlot *slot; /* We shouldn't be called until the request is complete. */ - Assert(areq->request_complete); + Assert(areq->state == ASYNC_COMPLETE); /* Our result slot shouldn't already be occupied. */ Assert(TupIsNull(node->ps.ps_ResultTupleSlot)); diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 85d436f..d3567bb 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq) * In async mode, configure for a wait * ---------------------------------------------------------------- */ -void +bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { @@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate, FdwRoutine *fdwroutine = node->fdwroutine; Assert(fdwroutine->ForeignAsyncConfigureWait != NULL); - fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); + return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); } /* ---------------------------------------------------------------- diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 23b4e18..72d8cd6 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -219,6 +219,7 @@ _copyAppend(const Append *from) */ 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 dc5b938..1ebdc48 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node) 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 69453b5..8443a62 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1520,6 +1520,7 @@ _readAppend(void) 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 e7e55c0..c73bbb3 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual, Index scanrelid, int ctePlanId, int cteParam); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *asyncplans, int nasyncplans, List *tlist); +static Append *make_append(List *asyncplans, int nasyncplans, + int referent, List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) 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 @@ -985,7 +988,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); @@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) { asyncplans = lappend(asyncplans, subplan); ++nasyncplans; + if (first) + referent_is_sync = false; } else syncplans = lappend(syncplans, subplan); + + first = false; } /* @@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist); + plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, + referent_is_sync ? nasyncplans : 0, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -4944,7 +4959,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, int nasyncplans, List *tlist) +make_append(List *appendplans, int nasyncplans, int referent, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -4955,6 +4970,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist) plan->righttree = NULL; node->appendplans = appendplans; node->nasyncplans = nasyncplans; + node->referent = referent; return node; } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 8a81d7a..de0e96c 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -4056,7 +4056,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/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 3e69ab0..47a3920 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node, extern void ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq); -extern void ExecAsyncForeignScanConfigureWait(EState *estate, +extern bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); extern void ExecAsyncForeignScanNotify(EState *estate, PendingAsyncRequest *areq); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 88feb9a..65517fd 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root, typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path); typedef void (*ForeignAsyncRequest_function) (EState *estate, PendingAsyncRequest *areq); -typedef void (*ForeignAsyncConfigureWait_function) (EState *estate, +typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate, PendingAsyncRequest *areq, bool reinit); typedef void (*ForeignAsyncNotify_function) (EState *estate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index b50b41c..0c6af86 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -352,6 +352,13 @@ typedef struct ResultRelInfo * State for an asynchronous tuple request. * ---------------- */ +typedef enum AsyncRequestState +{ + ASYNC_IDLE, + ASYNC_WAITING, + ASYNC_CALLBACK_PENDING, + ASYNC_COMPLETE +} AsyncRequestState; typedef struct PendingAsyncRequest { int myindex; /* Index in es_pending_async. */ @@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest 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. */ - bool callback_pending; /* Callback is needed. */ - bool request_complete; /* Request complete, result valid. */ + AsyncRequestState state; Node *result; /* Result (NULL if no more tuples). */ } PendingAsyncRequest; diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index 327119b..1df6693 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -209,6 +209,7 @@ typedef struct Append Plan plan; List *appendplans; int nasyncplans; /* # of async plans, always at start of list */ + int referent; /* index of inheritance tree referent */ } Append; /* ---------------- -- 2.9.2 ----Next_Part(Tue_Oct_18_10_30_51_2016_252)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="0004-Make-postgres_fdw-async-capable.patch" ^ permalink raw reply [nested|flat] 6+ messages in thread
* [PATCH 3/6] Modify async execution infrastructure. @ 2016-10-17 06:54 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 0 siblings, 0 replies; 6+ messages in thread From: Kyotaro Horiguchi @ 2016-10-17 06:54 UTC (permalink / raw) --- contrib/postgres_fdw/expected/postgres_fdw.out | 68 ++++++++-------- contrib/postgres_fdw/postgres_fdw.c | 5 +- src/backend/executor/execAsync.c | 105 ++++++++++++++----------- src/backend/executor/nodeAppend.c | 50 ++++++------ src/backend/executor/nodeForeignscan.c | 4 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/outfuncs.c | 1 + src/backend/nodes/readfuncs.c | 1 + src/backend/optimizer/plan/createplan.c | 24 +++++- src/backend/utils/adt/ruleutils.c | 6 +- src/include/executor/nodeForeignscan.h | 2 +- src/include/foreign/fdwapi.h | 2 +- src/include/nodes/execnodes.h | 10 ++- src/include/nodes/plannodes.h | 1 + 14 files changed, 167 insertions(+), 113 deletions(-) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index f9fd172..4b76e41 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -6329,13 +6329,13 @@ insert into bar2 values(4,44,44); insert into bar2 values(7,77,77); explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for update; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6343,10 +6343,10 @@ select * from bar where f1 in (select f1 from foo) for update; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6366,13 +6366,13 @@ select * from bar where f1 in (select f1 from foo) for update; explain (verbose, costs off) select * from bar where f1 in (select f1 from foo) for share; - QUERY PLAN ------------------------------------------------------------------------------------------------------------------------- + QUERY PLAN +---------------------------------------------------------------------------------------------- LockRows - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid -> Hash Join - Output: bar2.f1, bar2.f2, bar2.ctid, ((bar2.*)::bar), bar2.tableoid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Append -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.ctid, bar2.*, bar2.tableoid @@ -6380,10 +6380,10 @@ select * from bar where f1 in (select f1 from foo) for share; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid, bar.*, bar.tableoid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6404,22 +6404,22 @@ select * from bar where f1 in (select f1 from foo) for share; -- Check UPDATE with inherited target and an inherited source table explain (verbose, costs off) update bar set f2 = f2 + 100 where f1 in (select f1 from foo); - QUERY PLAN ---------------------------------------------------------------------------------------------------------- + QUERY PLAN +--------------------------------------------------------------------------------------------- Update on public.bar Update on public.bar Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar.f1 = foo2.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar.f1 = foo.f1) -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6427,16 +6427,16 @@ update bar set f2 = f2 + 100 where f1 in (select f1 from foo); -> Seq Scan on public.foo Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> Hash Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo2.ctid, ((foo2.*)::foo), foo2.tableoid - Hash Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, foo.ctid, foo.*, foo.tableoid + Hash Cond: (bar2.f1 = foo.f1) -> Foreign Scan on public.bar2 Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Hash - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 -> HashAggregate - Output: foo2.ctid, ((foo2.*)::foo), foo2.tableoid, foo2.f1 - Group Key: foo2.f1 + Output: foo.ctid, foo.*, foo.tableoid, foo.f1 + Group Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: foo2.ctid, foo2.*, foo2.tableoid, foo2.f1 @@ -6470,8 +6470,8 @@ where bar.f1 = ss.f1; Foreign Update on public.bar2 Remote SQL: UPDATE public.loct2 SET f2 = $2 WHERE ctid = $1 -> Hash Join - Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo2.f1)) - Hash Cond: (foo2.f1 = bar.f1) + Output: bar.f1, (bar.f2 + 100), bar.ctid, (ROW(foo.f1)) + Hash Cond: (foo.f1 = bar.f1) -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 @@ -6488,8 +6488,8 @@ where bar.f1 = ss.f1; -> Seq Scan on public.bar Output: bar.f1, bar.f2, bar.ctid -> Merge Join - Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo2.f1)) - Merge Cond: (bar2.f1 = foo2.f1) + Output: bar2.f1, (bar2.f2 + 100), bar2.f3, bar2.ctid, (ROW(foo.f1)) + Merge Cond: (bar2.f1 = foo.f1) -> Sort Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Sort Key: bar2.f1 @@ -6497,8 +6497,8 @@ where bar.f1 = ss.f1; Output: bar2.f1, bar2.f2, bar2.f3, bar2.ctid Remote SQL: SELECT f1, f2, f3, ctid FROM public.loct2 FOR UPDATE -> Sort - Output: (ROW(foo2.f1)), foo2.f1 - Sort Key: foo2.f1 + Output: (ROW(foo.f1)), foo.f1 + Sort Key: foo.f1 -> Append -> Foreign Scan on public.foo2 Output: ROW(foo2.f1), foo2.f1 diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index e75f8a1..830212f 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -354,7 +354,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root, static bool postgresIsForeignPathAsyncCapable(ForeignPath *path); static void postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq); -static void postgresForeignAsyncConfigureWait(EState *estate, +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void postgresForeignAsyncNotify(EState *estate, @@ -4479,11 +4479,12 @@ postgresForeignAsyncRequest(EState *estate, PendingAsyncRequest *areq) ExecAsyncRequestDone(estate, areq, (Node *) slot); } -static void +static bool postgresForeignAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { elog(ERROR, "postgresForeignAsyncConfigureWait"); + return false; } static void diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c index e070c26..33496a9 100644 --- a/src/backend/executor/execAsync.c +++ b/src/backend/executor/execAsync.c @@ -22,7 +22,7 @@ #include "storage/latch.h" static bool ExecAsyncEventWait(EState *estate, long timeout); -static void ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); static void ExecAsyncNotify(EState *estate, PendingAsyncRequest *areq); static void ExecAsyncResponse(EState *estate, PendingAsyncRequest *areq); @@ -43,7 +43,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, PlanState *requestee) { PendingAsyncRequest *areq = NULL; - int i = estate->es_num_pending_async; + int nasync = estate->es_num_pending_async; /* * If the number of pending asynchronous nodes exceeds the number of @@ -51,7 +51,7 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * We start with 16 slots, and thereafter double the array size each * time we run out of slots. */ - if (i >= estate->es_max_pending_async) + if (nasync >= estate->es_max_pending_async) { int newmax; @@ -81,25 +81,28 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, * PendingAsyncRequest if there is one. If not, we must allocate a new * one. */ - if (estate->es_pending_async[i] == NULL) + if (estate->es_pending_async[nasync] == NULL) { areq = MemoryContextAllocZero(estate->es_query_cxt, sizeof(PendingAsyncRequest)); - estate->es_pending_async[i] = areq; + estate->es_pending_async[nasync] = areq; } else { - areq = estate->es_pending_async[i]; + areq = estate->es_pending_async[nasync]; MemSet(areq, 0, sizeof(PendingAsyncRequest)); } - areq->myindex = estate->es_num_pending_async++; + areq->myindex = estate->es_num_pending_async; /* Initialize the new request. */ areq->requestor = requestor; areq->request_index = request_index; areq->requestee = requestee; - /* Give the requestee a chance to do whatever it wants. */ + /* + * Give the requestee a chance to do whatever it wants. + * Requst functions return true if a result is immediately available. + */ switch (nodeTag(requestee)) { case T_ForeignScanState: @@ -110,6 +113,20 @@ ExecAsyncRequest(EState *estate, PlanState *requestor, int request_index, elog(ERROR, "unrecognized node type: %d", (int) nodeTag(requestee)); } + + /* + * If a result is available, complete it immediately. + */ + if (areq->state == ASYNC_COMPLETE) + { + Assert(areq->result == NULL || IsA(areq->result, TupleTableSlot)); + ExecAsyncResponse(estate, areq); + + return; + } + + /* No result available now, make this node pending */ + estate->es_num_pending_async++; } /* @@ -175,22 +192,19 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) { PendingAsyncRequest *areq = estate->es_pending_async[i]; - /* Skip it if no callback is pending. */ - if (!areq->callback_pending) - continue; - - /* - * Mark it as no longer needing a callback. We must do this - * before dispatching the callback in case the callback resets - * the flag. - */ - areq->callback_pending = false; - estate->es_async_callback_pending--; - - /* Perform the actual callback; set request_done if appropraite. */ - if (!areq->request_complete) + /* Skip it if not pending. */ + if (areq->state == ASYNC_CALLBACK_PENDING) + { + /* + * Mark it as no longer needing a callback. We must do this + * before dispatching the callback in case the callback resets + * the flag. + */ + estate->es_async_callback_pending--; ExecAsyncNotify(estate, areq); - else + } + + if (areq->state == ASYNC_COMPLETE) { any_node_done = true; if (requestor == areq->requestor) @@ -214,7 +228,7 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) PendingAsyncRequest *head; PendingAsyncRequest *tail = estate->es_pending_async[tidx]; - if (!tail->callback_pending && tail->request_complete) + if (tail->state == ASYNC_COMPLETE) continue; head = estate->es_pending_async[hidx]; estate->es_pending_async[tidx] = head; @@ -247,7 +261,8 @@ ExecAsyncEventLoop(EState *estate, PlanState *requestor, long timeout) * means wait forever, 0 means don't wait at all, and >0 means wait for the * indicated number of milliseconds. * - * Returns true if we found some events and false if we timed out. + * Returns true if we found some events and false if we timed out or there's + * no event to wait. The latter is occur when the areq is processed during */ static bool ExecAsyncEventWait(EState *estate, long timeout) @@ -258,6 +273,7 @@ ExecAsyncEventWait(EState *estate, long timeout) int n; bool reinit = false; bool process_latch_set = false; + bool added = false; if (estate->es_wait_event_set == NULL) { @@ -282,13 +298,16 @@ ExecAsyncEventWait(EState *estate, long timeout) PendingAsyncRequest *areq = estate->es_pending_async[i]; if (areq->num_fd_events > 0) - ExecAsyncConfigureWait(estate, areq, reinit); + added |= ExecAsyncConfigureWait(estate, areq, reinit); } + Assert(added); + /* 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; @@ -312,12 +331,10 @@ ExecAsyncEventWait(EState *estate, long timeout) { PendingAsyncRequest *areq = w->user_data; - if (!areq->callback_pending) - { - Assert(!areq->request_complete); - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + Assert(areq->state == ASYNC_WAITING); + + areq->state = ASYNC_CALLBACK_PENDING; + estate->es_async_callback_pending++; } } @@ -333,8 +350,8 @@ ExecAsyncEventWait(EState *estate, long timeout) if (areq->wants_process_latch) { - Assert(!areq->request_complete); - areq->callback_pending = true; + Assert(areq->state == ASYNC_WAITING); + areq->state = ASYNC_CALLBACK_PENDING; } } } @@ -352,15 +369,19 @@ ExecAsyncEventWait(EState *estate, long timeout) * 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 void +static bool ExecAsyncConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { switch (nodeTag(areq->requestee)) { case T_ForeignScanState: - ExecAsyncForeignScanConfigureWait(estate, areq, reinit); + return ExecAsyncForeignScanConfigureWait(estate, areq, reinit); break; default: elog(ERROR, "unrecognized node type: %d", @@ -419,6 +440,7 @@ ExecAsyncSetRequiredEvents(EState *estate, PendingAsyncRequest *areq, 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 = ASYNC_WAITING; if (force_reset && estate->es_wait_event_set != NULL) { @@ -448,17 +470,12 @@ ExecAsyncRequestDone(EState *estate, PendingAsyncRequest *areq, Node *result) * 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 == ASYNC_IDLE || areq->state == ASYNC_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->request_complete = true; - - /* Make sure this request is flagged for a callback. */ - if (!areq->callback_pending) - { - areq->callback_pending = true; - estate->es_async_callback_pending++; - } + areq->state = ASYNC_COMPLETE; } diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index bb06569..c234f1f 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -229,9 +229,15 @@ ExecAppend(AppendState *node) */ while ((i = bms_first_member(node->as_needrequest)) >= 0) { - ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); node->as_nasyncpending++; + + ExecAsyncRequest(estate, &node->ps, i, node->appendplans[i]); + /* If this request immediately gives a result, take it. */ + if (node->as_nasyncresult > 0) + return node->as_asyncresult[--node->as_nasyncresult]; } + if (node->as_nasyncpending == 0 && node->as_syncdone) + return ExecClearTuple(node->ps.ps_ResultTupleSlot); } for (;;) @@ -246,32 +252,32 @@ ExecAppend(AppendState *node) { long timeout = node->as_syncdone ? -1 : 0; - for (;;) + while (node->as_nasyncpending > 0) { - if (node->as_nasyncpending == 0) - { - /* - * 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) - return ExecClearTuple(node->ps.ps_ResultTupleSlot); - break; - } - if (!ExecAsyncEventLoop(node->ps.state, &node->ps, timeout)) - { - /* Timeout reached. */ - break; - } - if (node->as_nasyncresult > 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); } } @@ -397,7 +403,7 @@ ExecAsyncAppendResponse(EState *estate, PendingAsyncRequest *areq) TupleTableSlot *slot; /* We shouldn't be called until the request is complete. */ - Assert(areq->request_complete); + Assert(areq->state == ASYNC_COMPLETE); /* Our result slot shouldn't already be occupied. */ Assert(TupIsNull(node->ps.ps_ResultTupleSlot)); diff --git a/src/backend/executor/nodeForeignscan.c b/src/backend/executor/nodeForeignscan.c index 85d436f..d3567bb 100644 --- a/src/backend/executor/nodeForeignscan.c +++ b/src/backend/executor/nodeForeignscan.c @@ -378,7 +378,7 @@ ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq) * In async mode, configure for a wait * ---------------------------------------------------------------- */ -void +bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit) { @@ -386,7 +386,7 @@ ExecAsyncForeignScanConfigureWait(EState *estate, FdwRoutine *fdwroutine = node->fdwroutine; Assert(fdwroutine->ForeignAsyncConfigureWait != NULL); - fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); + return fdwroutine->ForeignAsyncConfigureWait(estate, areq, reinit); } /* ---------------------------------------------------------------- diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 23b4e18..72d8cd6 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -219,6 +219,7 @@ _copyAppend(const Append *from) */ 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 dc5b938..1ebdc48 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -360,6 +360,7 @@ _outAppend(StringInfo str, const Append *node) 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 69453b5..8443a62 100644 --- a/src/backend/nodes/readfuncs.c +++ b/src/backend/nodes/readfuncs.c @@ -1520,6 +1520,7 @@ _readAppend(void) 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 7caa8d3..ff1d663 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -193,7 +193,8 @@ static CteScan *make_ctescan(List *qptlist, List *qpqual, Index scanrelid, int ctePlanId, int cteParam); static WorkTableScan *make_worktablescan(List *qptlist, List *qpqual, Index scanrelid, int wtParam); -static Append *make_append(List *asyncplans, int nasyncplans, List *tlist); +static Append *make_append(List *asyncplans, int nasyncplans, + int referent, List *tlist); static RecursiveUnion *make_recursive_union(List *tlist, Plan *lefttree, Plan *righttree, @@ -960,6 +961,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) 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 @@ -985,7 +988,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); @@ -999,9 +1009,13 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) { asyncplans = lappend(asyncplans, subplan); ++nasyncplans; + if (first) + referent_is_sync = false; } else syncplans = lappend(syncplans, subplan); + + first = false; } /* @@ -1011,7 +1025,8 @@ create_append_plan(PlannerInfo *root, AppendPath *best_path) * parent-rel Vars it'll be asked to emit. */ - plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, tlist); + plan = make_append(list_concat(asyncplans, syncplans), nasyncplans, + referent_is_sync ? nasyncplans : 0, tlist); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -4951,7 +4966,7 @@ make_foreignscan(List *qptlist, } static Append * -make_append(List *appendplans, int nasyncplans, List *tlist) +make_append(List *appendplans, int nasyncplans, int referent, List *tlist) { Append *node = makeNode(Append); Plan *plan = &node->plan; @@ -4962,6 +4977,7 @@ make_append(List *appendplans, int nasyncplans, List *tlist) plan->righttree = NULL; node->appendplans = appendplans; node->nasyncplans = nasyncplans; + node->referent = referent; return node; } diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 8a81d7a..de0e96c 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -4056,7 +4056,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/nodeForeignscan.h b/src/include/executor/nodeForeignscan.h index 3e69ab0..47a3920 100644 --- a/src/include/executor/nodeForeignscan.h +++ b/src/include/executor/nodeForeignscan.h @@ -31,7 +31,7 @@ extern void ExecForeignScanInitializeWorker(ForeignScanState *node, extern void ExecAsyncForeignScanRequest(EState *estate, PendingAsyncRequest *areq); -extern void ExecAsyncForeignScanConfigureWait(EState *estate, +extern bool ExecAsyncForeignScanConfigureWait(EState *estate, PendingAsyncRequest *areq, bool reinit); extern void ExecAsyncForeignScanNotify(EState *estate, PendingAsyncRequest *areq); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 88feb9a..65517fd 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -158,7 +158,7 @@ typedef bool (*IsForeignScanParallelSafe_function) (PlannerInfo *root, typedef bool (*IsForeignPathAsyncCapable_function) (ForeignPath *path); typedef void (*ForeignAsyncRequest_function) (EState *estate, PendingAsyncRequest *areq); -typedef void (*ForeignAsyncConfigureWait_function) (EState *estate, +typedef bool (*ForeignAsyncConfigureWait_function) (EState *estate, PendingAsyncRequest *areq, bool reinit); typedef void (*ForeignAsyncNotify_function) (EState *estate, diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index b50b41c..0c6af86 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -352,6 +352,13 @@ typedef struct ResultRelInfo * State for an asynchronous tuple request. * ---------------- */ +typedef enum AsyncRequestState +{ + ASYNC_IDLE, + ASYNC_WAITING, + ASYNC_CALLBACK_PENDING, + ASYNC_COMPLETE +} AsyncRequestState; typedef struct PendingAsyncRequest { int myindex; /* Index in es_pending_async. */ @@ -360,8 +367,7 @@ typedef struct PendingAsyncRequest 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. */ - bool callback_pending; /* Callback is needed. */ - bool request_complete; /* Request complete, result valid. */ + AsyncRequestState state; Node *result; /* Result (NULL if no more tuples). */ } PendingAsyncRequest; diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index 327119b..1df6693 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -209,6 +209,7 @@ typedef struct Append Plan plan; List *appendplans; int nasyncplans; /* # of async plans, always at start of list */ + int referent; /* index of inheritance tree referent */ } Append; /* ---------------- -- 2.9.2 ----Next_Part(Tue_Oct_25_18_21_50_2016_530)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="0004-Make-postgres_fdw-async-capable.patch" ^ permalink raw reply [nested|flat] 6+ messages in thread
* [PATCH v2] Fix exported snapshot xmin handoff race @ 2026-07-30 04:00 chee.wooson <chee.wooson@gmail.com> 0 siblings, 0 replies; 6+ messages in thread From: chee.wooson @ 2026-07-30 04:00 UTC (permalink / raw) ProcArrayInstallImportedXmin() verifies that the source transaction is still running and then installs the imported xmin. Both steps have to be serialized with concurrent proc-array horizon computations; otherwise VACUUM can scan the importer before the xmin is installed while the source transaction is still allowed to clear its xmin concurrently. Take ProcArrayLock in exclusive mode while importing an exported snapshot. That makes the xmin handoff atomic with respect to ComputeXidHorizons(), while keeping the no-importer transaction end path unchanged. Add an injection-point TAP test that pauses VACUUM inside ComputeXidHorizons() while it holds ProcArrayLock shared, starts SET TRANSACTION SNAPSHOT concurrently, and verifies that the importer waits for ProcArrayLock before the imported snapshot is allowed to protect the deleted tuple. --- src/backend/storage/ipc/procarray.c | 13 +- src/test/modules/test_misc/meson.build | 1 + .../test_misc/t/015_export_snapshot.pl | 261 ++++++++++++++++++ 3 files changed, 274 insertions(+), 1 deletion(-) create mode 100644 src/test/modules/test_misc/t/015_export_snapshot.pl diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 60336b31803..f5ed9c1c7a1 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -1736,6 +1736,15 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) TransactionId xid; TransactionId xmin; +#ifdef USE_INJECTION_POINTS + { + char ip_name[64]; + + snprintf(ip_name, sizeof(ip_name), + "compute-xid-horizons-at-%d", index); + InjectionPointRun(ip_name, NULL); + } +#endif /* Fetch xid just once - see GetNewTransactionId */ xid = UINT32_ACCESS_ONCE(other_xids[index]); xmin = UINT32_ACCESS_ONCE(proc->xmin); @@ -1945,6 +1954,8 @@ GetOldestNonRemovableTransactionId(Relation rel) { ComputeXidHorizonsResult horizons; + INJECTION_POINT("get-oldest-nonremovable-txid", NULL); + ComputeXidHorizons(&horizons); switch (GlobalVisHorizonKindForRel(rel)) @@ -2488,7 +2499,7 @@ ProcArrayInstallImportedXmin(TransactionId xmin, return false; /* Get lock so source xact can't end while we're doing this */ - LWLockAcquire(ProcArrayLock, LW_SHARED); + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); /* * Find the PGPROC entry of the source transaction. (This could use diff --git a/src/test/modules/test_misc/meson.build b/src/test/modules/test_misc/meson.build index ee290698b31..eb48ee35d1d 100644 --- a/src/test/modules/test_misc/meson.build +++ b/src/test/modules/test_misc/meson.build @@ -23,6 +23,7 @@ tests += { 't/012_ddlutils.pl', 't/013_temp_obj_multisession.pl', 't/014_log_statement_max_length.pl', + 't/015_export_snapshot.pl', ], # The injection points are cluster-wide, so disable installcheck 'runningcheck': false, diff --git a/src/test/modules/test_misc/t/015_export_snapshot.pl b/src/test/modules/test_misc/t/015_export_snapshot.pl new file mode 100644 index 00000000000..1cc9b482965 --- /dev/null +++ b/src/test/modules/test_misc/t/015_export_snapshot.pl @@ -0,0 +1,261 @@ +# Copyright (c) 2024-2026, PostgreSQL Global Development Group +# +# Test: reproduce the exported-snapshot xmin handoff race. +# +# Strategy (two-injection-point approach, for the exclusive-lock fix): +# 1. VACUUM attaches "get-oldest-nonremovable-txid" with injection_wait +# (PID-filtered via injection_points_set_local). +# 2. Start VACUUM. The first ComputeXidHorizons call (from on-access +# catalog pruning via GlobalVisUpdate) passes through unblocked. +# 3. VACUUM reaches GetOldestNonRemovableTransactionId -> blocked at +# "get-oldest-nonremovable-txid". +# 4. Coordinator detects the block -> GLOBALLY attaches +# "compute-xid-horizons-at-1" with injection_wait (no PID filter). +# 5. Coordinator wakes VACUUM from "get-oldest-nonremovable-txid". +# 6. VACUUM enters ComputeXidHorizons (holding ProcArrayLock shared) +# -> blocked at "compute-xid-horizons-at-1" (after scanning importer +# at index 0 with xmin=Invalid). +# 7. Coordinator detects the second block, detaches the injection point so +# later snapshots do not hit it, and starts SET TRANSACTION SNAPSHOT +# asynchronously. The fix makes ProcArrayInstallImportedXmin wait for +# ProcArrayLock exclusive instead of installing xmin while VACUUM is in +# the middle of a proc-array scan. +# 8. Coordinator verifies that the importer is waiting, then wakes +# "compute-xid-horizons-at-1". +# 9. VACUUM continues while the source transaction is still open, so its +# horizon remains low enough to preserve the deleted tuple. Afterwards +# the importer installs the snapshot. +# 10. Importer queries -> 1 row. +# +# Without the fix, SET TRANSACTION SNAPSHOT completes under a shared +# ProcArrayLock while VACUUM is paused, so the lock-wait assertion fails. +# +# Depends only on: injection_points (built-in test module) + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; +use Time::HiRes qw(usleep); + +if ($ENV{enable_injection_points} ne 'yes') +{ + plan skip_all => 'Injection points not supported by this build'; +} + +my $node = PostgreSQL::Test::Cluster->new('export_race_node'); +$node->init; +$node->append_conf('postgresql.conf', + "shared_preload_libraries = 'injection_points'"); +$node->start; + +$node->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + +$node->safe_psql('postgres', qq{ + CREATE TABLE race_test (id int, data text); + INSERT INTO race_test VALUES (1, 'should_be_visible'); +}); + +# ComputeXidHorizons iterates over pgprocnos[], which contains only +# client backends that called ProcArrayAdd(). The test relies on the importer +# preceding the source transaction, with at least one entry between them. +my $imp = $node->background_psql('postgres'); # index 0 +my $fill = $node->background_psql('postgres'); # index 1 (gap) +my $src = $node->background_psql('postgres'); # index 2 (source) +my $vac = $node->background_psql('postgres'); # index 3 (VACUUM) +my $del = $node->background_psql('postgres'); # index 4 (deleter) +my $coord = $node->background_psql('postgres'); + +my $imp_pid = $imp->query("SELECT pg_backend_pid()"); +$imp_pid =~ s/\s+//g; + +$src->query("BEGIN ISOLATION LEVEL REPEATABLE READ"); +$src->query("SELECT * FROM race_test"); +my $token = $src->query("SELECT pg_export_snapshot()"); +$token =~ s/\s+//g; +diag("exported snapshot token: $token"); + +$del->query("DELETE FROM race_test WHERE id = 1"); +$del->query("COMMIT"); +diag("deleter committed"); + +# Advance the XID counter so that the horizon (latestCompletedXid + 1 when +# no backend has a valid xmin) is strictly greater than the deleter's XID. +for (my $i = 0; $i < 100; $i++) +{ + $node->safe_psql('postgres', "SELECT txid_current()"); +} +diag("100 filler XIDs consumed"); + +# No query after BEGIN: importer xmin stays Invalid until SET TRANSACTION +# SNAPSHOT, which is essential for this race. +$imp->query("BEGIN ISOLATION LEVEL REPEATABLE READ"); + +my $vac_pid = $vac->query("SELECT pg_backend_pid()"); +$vac_pid =~ s/\s+//g; +diag("VACUUM PID: $vac_pid"); + +$vac->query("SELECT injection_points_set_local()"); +$vac->query( + "SELECT injection_points_attach('get-oldest-nonremovable-txid', 'wait')"); +diag("VACUUM attached get-oldest-nonremovable-txid"); + +$vac->query_until(qr/vac_started/, + "\\echo vac_started\nVACUUM race_test;\n"); +diag("VACUUM started"); + +{ + my $blocked = 0; + for (my $i = 0; $i < 1800; $i++) + { + my $result = $coord->query( + "SELECT count(*) = 1 FROM pg_stat_activity" + . " WHERE pid = $vac_pid" + . " AND wait_event_type = 'InjectionPoint'"); + if ($result =~ /t/) + { + $blocked = 1; + last; + } + usleep(100_000); + } + die "VACUUM did not reach first injection point within 180s" + unless $blocked; +} +diag("VACUUM blocked at get-oldest-nonremovable-txid"); + +$coord->query( + "SELECT injection_points_attach('compute-xid-horizons-at-1', 'wait')"); +diag("coordinator attached compute-xid-horizons-at-1"); + +$coord->query( + "SELECT injection_points_wakeup('get-oldest-nonremovable-txid')"); +diag("woke VACUUM from get-oldest-nonremovable-txid"); + +{ + my $blocked = 0; + for (my $i = 0; $i < 1800; $i++) + { + my $result = $coord->query( + "SELECT count(*) = 1 FROM pg_stat_activity" + . " WHERE pid = $vac_pid" + . " AND wait_event_type = 'InjectionPoint'"); + if ($result =~ /t/) + { + $blocked = 1; + last; + } + usleep(100_000); + } + die "VACUUM did not reach second injection point within 180s" + unless $blocked; +} +diag("VACUUM blocked at compute-xid-horizons-at-1"); + +$coord->query( + "SELECT injection_points_detach('compute-xid-horizons-at-1')"); +diag("detached compute-xid-horizons-at-1 while VACUUM remains blocked"); + +# An exported snapshot that is never imported must retain the normal no-XID +# commit path: it should not wait for ProcArrayLock. +$fill->query("BEGIN ISOLATION LEVEL REPEATABLE READ"); +my $unused_token = $fill->query("SELECT pg_export_snapshot()"); +$unused_token =~ s/\s+//g; + +my $fill_pid = $fill->query("SELECT pg_backend_pid()"); +$fill_pid =~ s/\s+//g; +$fill->query_until(qr/unreferenced_commit_started/, + "\\echo unreferenced_commit_started\nCOMMIT;\n" + . "\\echo unreferenced_commit_done\n"); + +my $unreferenced_finished = 0; +for (my $i = 0; $i < 100; $i++) +{ + my $result = $coord->query( + "SELECT CASE" + . " WHEN state = 'idle' THEN 'done'" + . " WHEN wait_event_type = 'LWLock'" + . " AND wait_event = 'ProcArray' THEN 'blocked'" + . " ELSE 'running' END" + . " FROM pg_stat_activity WHERE pid = $fill_pid"); + if ($result =~ /done/) + { + $unreferenced_finished = 1; + last; + } + last if $result =~ /blocked/; + usleep(100_000); +} +diag("unreferenced export started COMMIT while VACUUM holds ProcArrayLock"); + +$imp->query_until(qr/import_started/, + "\\echo import_started\nSET TRANSACTION SNAPSHOT '$token';\n" + . "\\echo import_done\n"); +diag("importer started SET TRANSACTION SNAPSHOT"); + +my $importer_waiting = 0; +for (my $i = 0; $i < 100; $i++) +{ + my $result = $coord->query( + "SELECT state || '|' || COALESCE(wait_event_type, '') || '|' ||" + . " COALESCE(wait_event, '')" + . " FROM pg_stat_activity" + . " WHERE pid = $imp_pid"); + if ($result =~ /active\|LWLock\|ProcArray/) + { + $importer_waiting = 1; + last; + } + last if $result eq ''; + usleep(100_000); +} + +$coord->query( + "SELECT injection_points_detach('get-oldest-nonremovable-txid')"); +$coord->query( + "SELECT injection_points_wakeup('compute-xid-horizons-at-1')"); +diag("detached remaining injection point and woke VACUUM"); + +$imp->query_until(qr/import_done/, ""); +$src->query("COMMIT"); +$fill->query_until(qr/unreferenced_commit_done/, ""); + +ok($unreferenced_finished, + "unreferenced export clears xmin without waiting for ProcArrayLock"); +ok($importer_waiting, + "importing transaction waits for ProcArrayLock while VACUUM computes horizons"); + +{ + my $done = 0; + for (my $i = 0; $i < 1800; $i++) + { + my $result = $coord->query( + "SELECT count(*) = 1 FROM pg_stat_activity" + . " WHERE pid = $vac_pid" + . " AND state = 'idle'"); + if ($result =~ /t/) + { + $done = 1; + last; + } + usleep(100_000); + } + die "VACUUM did not finish within 180s" unless $done; +} +diag("VACUUM finished"); + +my $count = $imp->query("SELECT count(*) FROM race_test"); +$count =~ s/\s+//g; +diag("importer sees $count row(s)"); + +is($count, 1, + "imported snapshot still sees the row after concurrent VACUUM") + or diag("BUG DETECTED: export-snapshot xmin race caused " + . "premature tuple removal (expected 1 row, got $count)"); + +$imp->query("COMMIT"); + +$node->stop; +done_testing(); -- 2.43.0 --PGPATCH-v2-export-xmin-- ^ permalink raw reply [nested|flat] 6+ messages in thread
end of thread, other threads:[~2026-07-30 04:00 UTC | newest] Thread overview: 6+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2016-10-17 06:54 [PATCH 03/13] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 2016-10-17 06:54 [PATCH 3/7] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 2016-10-17 06:54 [PATCH 3/4] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 2016-10-17 06:54 [PATCH 3/6] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 2016-10-17 06:54 [PATCH 3/7] Modify async execution infrastructure. Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp> 2026-07-30 04:00 [PATCH v2] Fix exported snapshot xmin handoff race chee.wooson <chee.wooson@gmail.com>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox