public inbox for [email protected]
help / color / mirror / Atom feedFrom: Kyotaro Horiguchi <[email protected]>
Subject: [PATCH 4/7] Fix async execution framework.
Date: Tue, 28 Jun 2016 17:23:16 +0900
This commit changes some behavior of the framework and fixes some
minor bugs.
---
src/backend/executor/execAsync.c | 141 +++++++++++++++++++++++++-----------
src/backend/executor/execProcnode.c | 33 ++++++---
src/backend/executor/execScan.c | 33 +++++++--
src/backend/executor/execUtils.c | 8 ++
src/backend/executor/nodeSeqscan.c | 7 +-
src/include/executor/execAsync.h | 7 ++
src/include/executor/executor.h | 20 +++--
src/include/nodes/execnodes.h | 26 +++++--
8 files changed, 199 insertions(+), 76 deletions(-)
diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c
index 20601fa..6da7ef2 100644
--- a/src/backend/executor/execAsync.c
+++ b/src/backend/executor/execAsync.c
@@ -29,7 +29,7 @@
#define EVENT_BUFFER_SIZE 16
-static void ExecAsyncConfigureWait(PlanState *planstate, bool reinit);
+static bool ExecAsyncConfigureWait(PlanState *planstate, AsyncConfigMode mode);
void
ExecAsyncWaitForNode(PlanState *planstate)
@@ -37,13 +37,14 @@ ExecAsyncWaitForNode(PlanState *planstate)
WaitEvent occurred_event[EVENT_BUFFER_SIZE];
PlanState *callbacks[EVENT_BUFFER_SIZE];
int ncallbacks = 0;
- EState *estate = planstate->state;
+ EState *estate = planstate->state;
while (!planstate->result_ready)
{
- bool reinit = (estate->es_wait_event_set == NULL);
+ bool reinit = (estate->wait_event_set == NULL);
int n;
int noccurred;
+ bool has_event = false;
if (reinit)
{
@@ -53,18 +54,68 @@ ExecAsyncWaitForNode(PlanState *planstate)
* aggressive here, because plans that depend on massive numbers
* of external FDs are likely to run afoul of kernel limits anyway.
*/
- estate->es_max_async_events = estate->es_total_async_events + 16;
- estate->es_wait_event_set =
- CreateWaitEventSet(estate->es_query_cxt,
- estate->es_max_async_events);
+ estate->max_events = estate->total_events + 16;
+ estate->wait_event_set =
+ CreateWaitEventSet(estate->es_query_cxt, estate->max_events);
}
- /* Give each waiting node a chance to add or modify events. */
- for (n = 0; n < estate->es_num_waiting_nodes; ++n)
- ExecAsyncConfigureWait(estate->es_waiting_nodes[n], reinit);
+ /*
+ * Give each waiting node a chance to add or modify events to the
+ * descendants of this planstate.
+ */
+ for (n = 0; n < estate->num_waiting_nodes; ++n)
+ {
+ PlanState *node = estate->waiting_nodes[n];
+
+ /*
+ * We assume that few nodes are async-aware and async-unaware
+ * nodes cannot be revserse-dispatched from lower nodes that is
+ * async-aware. Firing of an async node that is not a descendant
+ * of the planstate will cause such reverse-diaptching to
+ * async-aware nodes, which is unexpected behavior for them.
+ *
+ * For instance, consider an async-unaware Hashjoin(OUTER, INNER)
+ * where the OUTER is running asynchronously but the Hashjoin is
+ * waiting on the async INNER during inner-hash creation. If the
+ * OUTER fires for the case, since anyone is waiting on it,
+ * ExecAsyncWaitForNode finally dispatches to the Hashjoin which
+ * is now in the middle of thing its work.
+ */
+ if (!IsParent(planstate, node))
+ continue;
+
+ has_event |=
+ ExecAsyncConfigureWait(node,
+ reinit ? ASYNCCONF_TRY_ADD : ASYNCCONF_MODIFY);
+ }
+
+ if (!has_event)
+ {
+ /*
+ * No event to wait. This occurs when all of the waiters share the
+ * same object for sync with the nodes in other
+ * sync-subtree. Anyway we must have at least one event to wait.
+ */
+
+ for (n = 0; n < estate->num_waiting_nodes; ++n)
+ {
+ PlanState *node = estate->waiting_nodes[n];
- /* Wait for at least one event to occur. */
- noccurred = WaitEventSetWait(estate->es_wait_event_set, -1,
+ /* Skip if this node is not a descendant of planstate */
+ if (!IsParent(planstate, node))
+ continue;
+
+ if (ExecAsyncConfigureWait(node, ASYNCCONF_FORCE_ADD))
+ break;
+ }
+
+ /* Too bad. We don't have anyone to wait. Something wrong. */
+ if (n == estate->num_waiting_nodes)
+ ereport(ERROR,
+ (errmsg("inconsistency in asynchronous execution")));
+ }
+
+ noccurred = WaitEventSetWait(estate->wait_event_set, -1,
occurred_event, EVENT_BUFFER_SIZE);
Assert(noccurred > 0);
@@ -115,9 +166,10 @@ ExecAsyncWaitForNode(PlanState *planstate)
/*
* If there's now a tuple ready, we must dispatch to the
- * parent node; otherwise, there's nothing more to do.
+ * parent node up to the waiting root; otherwise, there's
+ * nothing more to do.
*/
- if (callbacks[i]->result_ready)
+ if (callbacks[i]->result_ready && callbacks[i] != planstate)
callbacks[i] = callbacks[i]->parent;
else
callbacks[i] = NULL;
@@ -143,7 +195,7 @@ ExecAsyncWaitForNode(PlanState *planstate)
void
ExecAsyncNeedsWait(PlanState *planstate, int nevents, bool reinit)
{
- EState *estate = planstate->state;
+ EState *estate = planstate->state;
Assert(nevents > 0); /* otherwise, use ExecAsyncDoesNotNeedWait */
@@ -154,43 +206,45 @@ ExecAsyncNeedsWait(PlanState *planstate, int nevents, bool reinit)
*/
if (planstate->n_async_events == 0)
{
- if (estate->es_max_waiting_nodes >= estate->es_num_waiting_nodes)
+ if (estate->max_waiting_nodes <= estate->num_waiting_nodes)
{
int newmax;
- if (estate->es_max_waiting_nodes == 0)
+ if (estate->max_waiting_nodes == 0)
{
newmax = 16;
- estate->es_waiting_nodes =
- MemoryContextAlloc(estate->es_query_cxt, newmax);
+ estate->waiting_nodes =
+ MemoryContextAlloc(estate->es_query_cxt,
+ newmax * sizeof(PlanState *));
}
else
{
- newmax = estate->es_max_waiting_nodes * 2;
- estate->es_waiting_nodes =
- repalloc(estate->es_waiting_nodes,
+ newmax = estate->max_waiting_nodes * 2;
+ estate->waiting_nodes =
+ repalloc(estate->waiting_nodes,
newmax * sizeof(PlanState *));
}
- estate->es_max_waiting_nodes = newmax;
+ estate->max_waiting_nodes = newmax;
}
- estate->es_waiting_nodes[estate->es_num_waiting_nodes++] = planstate;
+ estate->waiting_nodes[estate->num_waiting_nodes++] =
+ planstate;
}
- /* Adjust per-node and per-estate totals. */
- estate->es_total_async_events -= planstate->n_async_events;
+ /* Adjust per-node and per-asstate totals. */
+ estate->total_events -= planstate->n_async_events;
planstate->n_async_events = nevents;
- estate->es_total_async_events += planstate->n_async_events;
+ estate->total_events += planstate->n_async_events;
/*
* If a WaitEventSet has already been created, we need to discard it and
* start again if the user passed reinit = true, or if the total number of
* required events exceeds the supported number.
*/
- if (estate->es_wait_event_set != NULL && (reinit ||
- estate->es_total_async_events > estate->es_max_async_events))
+ if (estate->wait_event_set != NULL && (reinit ||
+ estate->total_events > estate->max_events))
{
- FreeWaitEventSet(estate->es_wait_event_set);
- estate->es_wait_event_set = NULL;
+ FreeWaitEventSet(estate->wait_event_set);
+ estate->wait_event_set = NULL;
}
}
@@ -211,21 +265,20 @@ ExecAsyncDoesNotNeedWait(PlanState *planstate)
* Remove the node from the list of waiting nodes. (Is a linear search
* going to be a problem here? I think probably not.)
*/
- for (n = 0; n < estate->es_num_waiting_nodes; ++n)
+ for (n = 0; n < estate->num_waiting_nodes; ++n)
{
- if (estate->es_waiting_nodes[n] == planstate)
- {
- estate->es_waiting_nodes[n] =
- estate->es_waiting_nodes[--estate->es_num_waiting_nodes];
+ if (estate->waiting_nodes[n] == planstate)
break;
- }
}
/* We should always find ourselves in the array. */
- Assert(n < estate->es_num_waiting_nodes);
+ Assert(n < estate->num_waiting_nodes);
+
+ estate->waiting_nodes[n] =
+ estate->waiting_nodes[--estate->num_waiting_nodes];
/* We no longer need any asynchronous events. */
- estate->es_total_async_events -= planstate->n_async_events;
+ estate->total_events -= planstate->n_async_events;
planstate->n_async_events = 0;
/*
@@ -234,18 +287,18 @@ ExecAsyncDoesNotNeedWait(PlanState *planstate)
* assumes we actually did register some events at one point, because we
* needed to wait at some point and we don't any more.
*/
- if (estate->es_wait_event_set != NULL)
+ if (estate->wait_event_set != NULL)
{
- FreeWaitEventSet(estate->es_wait_event_set);
- estate->es_wait_event_set = NULL;
+ FreeWaitEventSet(estate->wait_event_set);
+ estate->wait_event_set = NULL;
}
}
/*
* Give per-nodetype function a chance to register wait events.
*/
-static void
-ExecAsyncConfigureWait(PlanState *planstate, bool reinit)
+static bool
+ExecAsyncConfigureWait(PlanState *planstate, AsyncConfigMode config_mode)
{
switch (nodeTag(planstate))
{
diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c
index b7ac08e..3590ab1 100644
--- a/src/backend/executor/execProcnode.c
+++ b/src/backend/executor/execProcnode.c
@@ -139,6 +139,7 @@ ExecInitNode(Plan *node, EState *estate, PlanState *parent, int eflags)
PlanState *result;
List *subps;
ListCell *l;
+ int this_node_id = estate->next_node_id++;
/*
* do nothing when we get to the end of a leaf on tree.
@@ -344,6 +345,10 @@ ExecInitNode(Plan *node, EState *estate, PlanState *parent, int eflags)
/* Set parent pointer. */
result->parent = parent;
+ /* Set this node id and that of the right sibling */
+ result->node_id = this_node_id;
+ result->right_node_id = estate->next_node_id;
+
/*
* Initialize any initPlans present in this node. The planner put them in
* a separate list for us.
@@ -374,9 +379,13 @@ ExecInitNode(Plan *node, EState *estate, PlanState *parent, int eflags)
* Invoke the given node's dispatch function.
* ----------------------------------------------------------------
*/
-void
+
+inline void
ExecDispatchNode(PlanState *node)
{
+ if (node->result_ready)
+ return;
+
if (node->instrument)
InstrStartNode(node->instrument);
@@ -559,6 +568,8 @@ void
ExecExecuteNode(PlanState *node)
{
node->result_ready = false;
+ if (node->chgParam != NULL) /* something changed */
+ ExecReScan(node); /* let ReScan handle this */
ExecDispatchNode(node);
}
@@ -569,15 +580,18 @@ ExecExecuteNode(PlanState *node)
* Get the next tuple from the given node. If the node is
* asynchronous, wait for a tuple to be ready before
* returning.
- * ----------------------------------------------------------------
+ * The given node works as the termination node of an asynchronous
+ * execution subtree and every subtree should have an individual context.
+ * ----------------------------------------------------------------
*/
TupleTableSlot *
ExecProcNode(PlanState *node)
{
CHECK_FOR_INTERRUPTS();
- /* mark any previous result as having been consumed */
- node->result_ready = false;
+ /* Return unconsumed result if any */
+ if (node->result_ready)
+ return ExecConsumeResult(node);
if (node->chgParam != NULL) /* something changed */
ExecReScan(node); /* let ReScan handle this */
@@ -587,10 +601,7 @@ ExecProcNode(PlanState *node)
if (!node->result_ready)
ExecAsyncWaitForNode(node);
- /* Result should be a TupleTableSlot, unless it's NULL. */
- Assert(node->result == NULL || IsA(node->result, TupleTableSlot));
-
- return (TupleTableSlot *) node->result;
+ return ExecConsumeResult(node);
}
@@ -848,6 +859,8 @@ ExecEndNode(PlanState *node)
bool
ExecShutdownNode(PlanState *node)
{
+ bool ret;
+
if (node == NULL)
return false;
@@ -860,5 +873,7 @@ ExecShutdownNode(PlanState *node)
break;
}
- return planstate_tree_walker(node, ExecShutdownNode, NULL);
+ ret = planstate_tree_walker(node, ExecShutdownNode, NULL);
+
+ return ret;
}
diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 095d40b..69d616b 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -128,6 +128,9 @@ ExecScan(ScanState *node,
ExprDoneCond isDone;
TupleTableSlot *resultSlot;
+ if (node->ps.result_ready)
+ return;
+
/*
* Fetch data from node
*/
@@ -136,14 +139,25 @@ ExecScan(ScanState *node,
econtext = node->ps.ps_ExprContext;
/*
+ * The underlying nodes don't use ExecReturnTuple. Set this flag here so
+ * that the async-unaware/incapable children don't need to touch it
+ * explicitly. Async-aware/capable nodes will unset it instead if needed.
+ */
+ node->ps.result_ready = true;
+
+ /*
* If we have neither a qual to check nor a projection to do, just skip
* all the overhead and produce the raw scan tuple.
*/
if (!qual && !projInfo)
{
+ TupleTableSlot *slot;
+
ResetExprContext(econtext);
- ExecReturnTuple(&node->ps,
- ExecScanFetch(node, accessMtd, recheckMtd));
+ slot = ExecScanFetch(node, accessMtd, recheckMtd);
+ if (node->ps.result_ready)
+ node->ps.result = (Node *) slot;
+
return;
}
@@ -158,7 +172,7 @@ ExecScan(ScanState *node,
resultSlot = ExecProject(projInfo, &isDone);
if (isDone == ExprMultipleResult)
{
- ExecReturnTuple(&node->ps, resultSlot);
+ node->ps.result = (Node *) resultSlot;
return;
}
/* Done with that source tuple... */
@@ -184,6 +198,9 @@ ExecScan(ScanState *node,
slot = ExecScanFetch(node, accessMtd, recheckMtd);
+ if (!node->ps.result_ready)
+ return;
+
/*
* if the slot returned by the accessMtd contains NULL, then it means
* there is nothing more to scan so we just return an empty slot,
@@ -193,9 +210,9 @@ ExecScan(ScanState *node,
if (TupIsNull(slot))
{
if (projInfo)
- ExecReturnTuple(&node->ps, ExecClearTuple(projInfo->pi_slot));
- else
- ExecReturnTuple(&node->ps, slot);
+ slot = ExecClearTuple(projInfo->pi_slot);
+
+ node->ps.result = (Node *) slot;
return;
}
@@ -227,7 +244,7 @@ ExecScan(ScanState *node,
if (isDone != ExprEndResult)
{
node->ps.ps_TupFromTlist = (isDone == ExprMultipleResult);
- ExecReturnTuple(&node->ps, resultSlot);
+ node->ps.result = (Node *) resultSlot;
return;
}
}
@@ -236,7 +253,7 @@ ExecScan(ScanState *node,
/*
* Here, we aren't projecting, so just return scan tuple.
*/
- ExecReturnTuple(&node->ps, slot);
+ node->ps.result = (Node *) slot;
return;
}
}
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index e937cf8..bb90844 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -117,6 +117,14 @@ CreateExecutorState(void)
estate->es_param_list_info = NULL;
estate->es_param_exec_vals = NULL;
+ estate->waiting_nodes = NULL;
+ estate->num_waiting_nodes = 0;
+ estate->max_waiting_nodes = 0;
+ estate->total_events = 0;
+ estate->max_events = 0;
+ estate->wait_event_set = NULL;
+ estate->next_node_id = 1;
+
estate->es_query_cxt = qcontext;
estate->es_tupleTable = NIL;
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index 0ca86d9..ef1ce9c 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -124,9 +124,10 @@ SeqRecheck(SeqScanState *node, TupleTableSlot *slot)
void
ExecSeqScan(SeqScanState *node)
{
- return ExecScan((ScanState *) node,
- (ExecScanAccessMtd) SeqNext,
- (ExecScanRecheckMtd) SeqRecheck);
+ ExecScan((ScanState *) node,
+ (ExecScanAccessMtd) SeqNext,
+ (ExecScanRecheckMtd) SeqRecheck);
+
}
/* ----------------------------------------------------------------
diff --git a/src/include/executor/execAsync.h b/src/include/executor/execAsync.h
index 38b37a1..f1c748b 100644
--- a/src/include/executor/execAsync.h
+++ b/src/include/executor/execAsync.h
@@ -15,6 +15,13 @@
#include "nodes/execnodes.h"
+typedef enum AsyncConfigMode
+{
+ ASYNCCONF_MODIFY,
+ ASYNCCONF_TRY_ADD,
+ ASYNCCONF_FORCE_ADD
+} AsyncConfigMode;
+
extern void ExecAsyncWaitForNode(PlanState *planstate);
extern void ExecAsyncNeedsWait(PlanState *planstate, int nevents,
bool reinit);
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 7abc361..c1ef2ab 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -231,14 +231,22 @@ extern void ExecEndNode(PlanState *node);
extern bool ExecShutdownNode(PlanState *node);
/* Convenience function to set a node's result to a TupleTableSlot. */
-static inline void
-ExecReturnTuple(PlanState *node, TupleTableSlot *slot)
-{
- Assert(!node->result_ready);
- node->result = (Node *) slot;
- node->result_ready = true;
+#define ExecReturnTuple(node, slot) \
+{ \
+ Assert(!(node)->result_ready); \
+ (node)->result = (Node *) (slot); \
+ (node)->result_ready = true; \
}
+/* Convenience function to retrieve a node's result. */
+#define ExecConsumeResult(node) \
+( \
+ Assert((node)->result_ready), \
+ Assert((node)->result == NULL || IsA((node)->result, TupleTableSlot)), \
+ (node)->result_ready = false, \
+ (TupleTableSlot *) node->result)
+
+
/*
* prototypes from functions in execQual.c
*/
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 76e36a2..b72decc 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -383,12 +383,14 @@ typedef struct EState
ParamExecData *es_param_exec_vals; /* values of internal params */
/* Asynchronous execution support */
- struct PlanState **es_waiting_nodes; /* array of waiting nodes */
- int es_num_waiting_nodes; /* # of waiters in array */
- int es_max_waiting_nodes; /* # of allocated entries */
- int es_total_async_events; /* total of per-node n_async_events */
- int es_max_async_events; /* # supported by event set */
- struct WaitEventSet *es_wait_event_set;
+ struct PlanState **waiting_nodes; /* array of waiting nodes */
+ int num_waiting_nodes; /* # of waiters in array */
+ int max_waiting_nodes; /* # of allocated entries */
+ int total_events; /* total of per-node n_async_events */
+ int max_events; /* # supported by event set */
+ struct WaitEventSet *wait_event_set;
+
+ int next_node_id; /* node id for the next plan state */
/* Other working state: */
MemoryContext es_query_cxt; /* per-query context in which EState lives */
@@ -1038,6 +1040,15 @@ typedef struct PlanState
* nodes point to one EState for the whole
* top-level plan */
+ /*
+ * node_id and right_node_id represents ancestor-descendant relationship
+ * by nested set model. The ids are in depth-first order and that of all
+ * the descendants of a node are between node_id and right_node_id - 1 of
+ * that node.
+ */
+ int node_id; /* node id according to nested set model */
+ int right_node_id; /* node id of the right sibling */
+
struct PlanState *parent; /* node which will receive tuples from us */
bool result_ready; /* true if result is ready */
Node *result; /* result, most often TupleTableSlot */
@@ -1075,6 +1086,9 @@ typedef struct PlanState
* functions in targetlist */
} PlanState;
+/* Macros applied on PlanStates */
+#define IsParent(p, d) ((p)->node_id <= (d)->node_id && (d)->node_id < (p)->right_node_id)
+
/* ----------------
* these are defined to avoid confusion problems with "left"
* and "right" and "inner" and "outer". The convention is that
--
1.8.3.1
----Next_Part(Thu_Jul_21_18_50_07_2016_597)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="0005-Add-new-fdwroutine-AsyncConfigureWait-and-ShutdownFo.patch"
view thread (69+ 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 4/7] Fix async execution framework.
In-Reply-To: <no-message-id-510215@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