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 | 94 ++++++++++++++++++++++++++++++++-------
src/include/nodes/execnodes.h | 2 +
2 files changed, 80 insertions(+), 16 deletions(-)
diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c
index e0ce8c6..1c0d26e 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,49 +199,104 @@ ExecInitAppend(Append *node, EState *estate, int eflags)
void
ExecAppend(AppendState *node)
{
- for (;;)
+ int n_notready = 0;
+ PlanState *subnode;
+ int i, n;
+
+ n = node->as_whichplan;
+
+ for (i = 0 ; i < node->as_nplans ; i++)
{
- PlanState *subnode;
TupleTableSlot *result;
+ if (node->as_async)
+ {
+ if (n >= node->as_nplans)
+ n = 0;
+
+ if (node->as_finished[n])
+ {
+ n++;
+ continue;
+ }
+ }
+
/*
* figure out which subplan we are currently processing
*/
- subnode = node->appendplans[node->as_whichplan];
+ subnode = node->appendplans[n];
/*
- * get a tuple from the subplan
+ * execute the subplan to get a result if it is not ready yet
*/
- result = ExecProcNode(subnode);
+ if (!subnode->result_ready)
+ ExecExecuteNode(subnode);
+
+ /*
+ * This subnode is not ready yet when asynchrony is not allowed,
+ * immediately wait for this subnode.
+ */
+ if (!subnode->result_ready)
+ {
+ if (node->as_async)
+ {
+ n_notready++;
+ n++;
+ continue;
+ }
+ ExecAsyncWaitForNode(subnode);
+ }
+
+ Assert(subnode->result_ready);
+
+ result = ExecConsumeResult((PlanState *)subnode);
if (!TupIsNull(result))
{
/*
- * If the subplan gave us something then return it as-is. We do
- * NOT make use of the result slot that was set up in
- * ExecInitAppend; there's no need for it.
+ * If the subplan gave us something then return it
+ * as-is. We do NOT make use of the result slot that was
+ * set up in ExecInitAppend; there's no need for it.
*/
+ node->as_whichplan = n;
ExecReturnTuple(&node->ps, result);
return;
}
+ /* Tuple has been exhausted on this subnode */
+ if (node->as_async)
+ {
+ node->as_finished[n++] = true;
+ continue;
+ }
+
/*
* Go on to the "next" subplan in the appropriate direction. If no
* more subplans, return the empty slot set up for us by
* ExecInitAppend.
*/
if (ScanDirectionIsForward(node->ps.state->es_direction))
- node->as_whichplan++;
+ {
+ if (n >= node->as_nplans - 1)
+ break;
+ n++;
+ }
else
- node->as_whichplan--;
- if (!exec_append_initialize_next(node))
{
- ExecReturnTuple(&node->ps,
- ExecClearTuple(node->ps.ps_ResultTupleSlot));
- return;
+ if (n == 0)
+ break;
+ n--;
}
+ }
- /* Else loop back and try to get a tuple from the new subplan */
+ /*
+ * We are finished if reached here and no subnodes are not-ready
+ */
+ if (n_notready == 0)
+ {
+ node->as_whichplan = n;
+ ExecReturnTuple(&node->ps,
+ ExecClearTuple(node->ps.ps_ResultTupleSlot));
}
}
@@ -277,6 +337,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 9121537..e4f2bb6 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1170,6 +1170,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(Wed_Jul_06_16_29_14_2016_599)--
Content-Type: Text/Plain; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline; filename="gentbl.sql"
CREATE EXTENSION postgres_fdw;
CREATE TABLE pl (a int, b int);
CREATE TABLE cl1 (LIKE pl) INHERITS (pl);
CREATE TABLE cl2 (LIKE pl) INHERITS (pl);
CREATE TABLE cl3 (LIKE pl) INHERITS (pl);
CREATE TABLE cl4 (LIKE pl) INHERITS (pl);
INSERT INTO cl1 (SELECT a, a FROM generate_series(0000000, 0999999) a);
INSERT INTO cl2 (SELECT a, a FROM generate_series(1000000, 1999999) a);
INSERT INTO cl3 (SELECT a, a FROM generate_series(2000000, 2999999) a);
INSERT INTO cl4 (SELECT a, a FROM generate_series(3000000, 3999999) a);
CREATE TABLE t0 (LIKE pl);
INSERT INTO t0 (SELECT a, a FROM generate_series(0000000, 9999999) a);
CREATE SERVER sv0 FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '/tmp', dbname 'postgres');
CREATE SERVER sv1 FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '/tmp', dbname 'postgres');
CREATE SERVER sv2 FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '/tmp', dbname 'postgres');
CREATE SERVER sv3 FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '/tmp', dbname 'postgres');
CREATE SERVER sv4 FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '/tmp', dbname 'postgres');
CREATE USER MAPPING FOR public SERVER sv0;
CREATE USER MAPPING FOR public SERVER sv1;
CREATE USER MAPPING FOR public SERVER sv2;
CREATE USER MAPPING FOR public SERVER sv3;
CREATE USER MAPPING FOR public SERVER sv4;
CREATE FOREIGN TABLE ft10 (a int, b int) SERVER sv0 OPTIONS (table_name 'cl1');
CREATE FOREIGN TABLE ft20 (a int, b int) SERVER sv0 OPTIONS (table_name 'cl2');
CREATE FOREIGN TABLE ft30 (a int, b int) SERVER sv0 OPTIONS (table_name 'cl3');
CREATE FOREIGN TABLE ft40 (a int, b int) SERVER sv0 OPTIONS (table_name 'cl4');
CREATE FOREIGN TABLE ft11 (a int, b int) SERVER sv1 OPTIONS (table_name 'cl1');
CREATE FOREIGN TABLE ft22 (a int, b int) SERVER sv2 OPTIONS (table_name 'cl2');
CREATE FOREIGN TABLE ft33 (a int, b int) SERVER sv3 OPTIONS (table_name 'cl3');
CREATE FOREIGN TABLE ft44 (a int, b int) SERVER sv4 OPTIONS (table_name 'cl4');
CREATE TABLE pf0 (LIKE pl);
ALTER FOREIGN TABLE ft10 INHERIT pf0;
ALTER FOREIGN TABLE ft20 INHERIT pf0;
ALTER FOREIGN TABLE ft30 INHERIT pf0;
ALTER FOREIGN TABLE ft40 INHERIT pf0;
CREATE table pf1 (LIKE pl);
ALTER FOREIGN TABLE ft11 INHERIT pf1;
ALTER FOREIGN TABLE ft22 INHERIT pf1;
ALTER FOREIGN TABLE ft33 INHERIT pf1;
ALTER FOREIGN TABLE ft44 INHERIT pf1;
ANALYZE;
----Next_Part(Wed_Jul_06_16_29_14_2016_599)--
Content-Type: Text/Plain; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline; filename="testrun.sh"
#! /bin/bash
function do_test() {
echo $1
for i in $(seq 1 10); do psql postgres -c "set log_min_duration_statement = 0; set client_min_messages=log; select sum(a) from $1"; done | grep LOG
}
for i in "t0" "pl" "pf0" "pf1";
do
do_test $i
done
----Next_Part(Wed_Jul_06_16_29_14_2016_599)--
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(Wed_Jul_06_16_29_14_2016_599)----
view thread (57+ 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-632573@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