agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Kyotaro Horiguchi <[email protected]>
Subject: [PATCH 7/7] Make Append node async-aware.
Date: Tue, 28 Jun 2016 18:52:37 +0900
Change append node to be capable to handle asynchronous children
properly. As soon as it receives !async_ready from a child, it moves
to the next child and if no child is ready, it sleeps until at least
one of them become ready.
---
src/backend/executor/nodeAppend.c | 67 +++++++++++++++++++++++++++++++++++++--
src/include/nodes/execnodes.h | 2 ++
2 files changed, 67 insertions(+), 2 deletions(-)
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index e0ce8c6..9a4063a 100644
--- a/src/backend/executor/nodeAppend.c
+++ b/src/backend/executor/nodeAppend.c
@@ -58,6 +58,7 @@
#include "postgres.h"
#include "executor/execdebug.h"
+#include "executor/execAsync.h"
#include "executor/nodeAppend.h"
static bool exec_append_initialize_next(AppendState *appendstate);
@@ -121,6 +122,7 @@ ExecInitAppend(Append *node, EState *estate, int eflags)
{
AppendState *appendstate = makeNode(AppendState);
PlanState **appendplanstates;
+ bool *finished;
int nplans;
int i;
ListCell *lc;
@@ -134,14 +136,17 @@ ExecInitAppend(Append *node, EState *estate, int eflags)
nplans = list_length(node->appendplans);
appendplanstates = (PlanState **) palloc0(nplans * sizeof(PlanState *));
-
+ finished = (bool *) palloc0(nplans * sizeof(bool));
+
/*
* create new AppendState for our append node
*/
appendstate->ps.plan = (Plan *) node;
appendstate->ps.state = estate;
appendstate->appendplans = appendplanstates;
+ appendstate->as_finished = finished;
appendstate->as_nplans = nplans;
+ appendstate->as_async = ((eflags & EXEC_FLAG_BACKWARD) == 0);
/*
* Miscellaneous initialization
@@ -194,6 +199,8 @@ ExecInitAppend(Append *node, EState *estate, int eflags)
void
ExecAppend(AppendState *node)
{
+ int stopplan = node->as_whichplan;
+
for (;;)
{
PlanState *subnode;
@@ -207,7 +214,36 @@ ExecAppend(AppendState *node)
/*
* get a tuple from the subplan
*/
- result = ExecProcNode(subnode);
+ Assert(!node->as_finished[node->as_whichplan]);
+
+ if (!subnode->result_ready)
+ ExecExecuteNode(subnode);
+
+ if (!subnode->result_ready)
+ {
+ if (node->as_async)
+ {
+ /* Move to the next living node */
+ do
+ {
+ node->as_whichplan =
+ (node->as_whichplan + 1) % node->as_nplans;
+ } while (node->as_whichplan != stopplan &&
+ node->as_finished[node->as_whichplan]);
+
+ /* No node is ready yet, return as not-ready */
+ if (node->as_whichplan == stopplan)
+ return;
+
+ /* Try the next node */
+ continue;
+ }
+
+ /* If not async, immediately wait for this subnode */
+ ExecAsyncWaitForNode(subnode);
+ }
+
+ result = ExecConsumeResult((PlanState *) subnode);
if (!TupIsNull(result))
{
@@ -220,6 +256,31 @@ ExecAppend(AppendState *node)
return;
}
+ if (node->as_async)
+ {
+ node->as_finished[node->as_whichplan] = true;
+ stopplan = node->as_whichplan;
+
+ /* Find the next living subnode */
+ do
+ {
+ node->as_whichplan =
+ (node->as_whichplan + 1) % node->as_nplans;
+ } while (node->as_whichplan != stopplan &&
+ node->as_finished[node->as_whichplan]);
+
+ if (node->as_whichplan != stopplan)
+ {
+ stopplan = node->as_whichplan;
+ continue;
+ }
+
+ /* All subnodes are exhausted. Finish this node. */
+ ExecReturnTuple(&node->ps,
+ ExecClearTuple(node->ps.ps_ResultTupleSlot));
+ return;
+ }
+
/*
* Go on to the "next" subplan in the appropriate direction. If no
* more subplans, return the empty slot set up for us by
@@ -277,6 +338,8 @@ ExecReScanAppend(AppendState *node)
{
PlanState *subnode = node->appendplans[i];
+ node->as_finished[i] = false;
+
/*
* ExecReScan doesn't know about my subplans, so I have to do
* changed-parameter signaling myself.
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b72decc..b0a86c5 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1177,6 +1177,8 @@ typedef struct AppendState
{
PlanState ps; /* its first field is NodeTag */
PlanState **appendplans; /* array of PlanStates for my inputs */
+ bool as_async; /* true to allow async execution */
+ bool *as_finished; /* array of the running state of subplans */
int as_nplans;
int as_whichplan;
} AppendState;
--
1.8.3.1
----Next_Part(Thu_Jul_21_18_50_07_2016_597)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0
--
Sent via pgsql-hackers mailing list ([email protected])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
----Next_Part(Thu_Jul_21_18_50_07_2016_597)----
view thread (52+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH 7/7] Make Append node async-aware.
In-Reply-To: <no-message-id-399590@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox