From 471d6c97cce9aa1a903dedaf3b1f76b9588e9bba Mon Sep 17 00:00:00 2001 From: Kyotaro Horiguchi Date: Tue, 28 Jun 2016 17:23:16 +0900 Subject: [PATCH 4/7] Fix async execution framework. This commit changes some behavior of the framework and fixes some minor bugs. --- src/backend/executor/execAsync.c | 128 +++++++++++++++++++++++------------- src/backend/executor/execProcnode.c | 59 +++++++++++++++-- src/backend/executor/execScan.c | 33 +++++++--- src/backend/executor/nodeSeqscan.c | 7 +- src/include/executor/execAsync.h | 7 ++ src/include/executor/executor.h | 10 +++ src/include/nodes/execnodes.h | 21 ++++-- 7 files changed, 196 insertions(+), 69 deletions(-) diff --git a/src/backend/executor/execAsync.c b/src/backend/executor/execAsync.c index 20601fa..51902e6 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,15 @@ 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; + AsyncContext *async_cxt = estate->es_async_cxt; while (!planstate->result_ready) { - bool reinit = (estate->es_wait_event_set == NULL); + bool reinit = (async_cxt->wait_event_set == NULL); int n; int noccurred; + bool has_event = false; if (reinit) { @@ -53,18 +55,39 @@ 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); + async_cxt->max_events = async_cxt->total_events + 16; + async_cxt->wait_event_set = + CreateWaitEventSet(estate->es_query_cxt, async_cxt->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); + for (n = 0; n < async_cxt->num_waiting_nodes; ++n) + has_event |= + ExecAsyncConfigureWait(async_cxt->waiting_nodes[n], + reinit ? ASYNCCONF_TRY_ADD : ASYNCCONF_MODIFY); - /* Wait for at least one event to occur. */ - noccurred = WaitEventSetWait(estate->es_wait_event_set, -1, + if (!has_event) + { + /* + * No event to wait. This occurs when all the waiter shares the + * object for sync with nodes in other sync-subtree. Anyway we + * must have at least one event to wait. + */ + + for (n = 0; n < async_cxt->num_waiting_nodes; ++n) + { + if (ExecAsyncConfigureWait(async_cxt->waiting_nodes[n], + ASYNCCONF_FORCE_ADD)) + break; + } + + /* Too bad. We don't have anyone to wait. */ + if (n == async_cxt->num_waiting_nodes) + ereport(ERROR, + (errmsg("inconsistency in asynchronous execution"))); + } + + noccurred = WaitEventSetWait(async_cxt->wait_event_set, -1, occurred_event, EVENT_BUFFER_SIZE); Assert(noccurred > 0); @@ -115,9 +138,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,54 +167,69 @@ ExecAsyncWaitForNode(PlanState *planstate) void ExecAsyncNeedsWait(PlanState *planstate, int nevents, bool reinit) { - EState *estate = planstate->state; + EState *estate = planstate->state; + AsyncContext *async_cxt = estate->es_async_cxt; Assert(nevents > 0); /* otherwise, use ExecAsyncDoesNotNeedWait */ /* + * If no active async context found, make new one. + */ + if (async_cxt == NULL) + { + async_cxt = MemoryContextAlloc(estate->es_query_cxt, + sizeof(AsyncContext)); + memset(async_cxt, 0, sizeof(AsyncContext)); + + planstate->state->es_async_cxt = async_cxt; + } + + /* * If this node is not already present in the array of waiting nodes, * then add it. If that array hasn't been allocated or is full, this may * require (re)allocating it. */ if (planstate->n_async_events == 0) { - if (estate->es_max_waiting_nodes >= estate->es_num_waiting_nodes) + if (async_cxt->max_waiting_nodes <= async_cxt->num_waiting_nodes) { int newmax; - if (estate->es_max_waiting_nodes == 0) + if (async_cxt->max_waiting_nodes == 0) { newmax = 16; - estate->es_waiting_nodes = - MemoryContextAlloc(estate->es_query_cxt, newmax); + async_cxt->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 = async_cxt->max_waiting_nodes * 2; + async_cxt->waiting_nodes = + repalloc(async_cxt->waiting_nodes, newmax * sizeof(PlanState *)); } - estate->es_max_waiting_nodes = newmax; + async_cxt->max_waiting_nodes = newmax; } - estate->es_waiting_nodes[estate->es_num_waiting_nodes++] = planstate; + async_cxt->waiting_nodes[async_cxt->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. */ + async_cxt->total_events -= planstate->n_async_events; planstate->n_async_events = nevents; - estate->es_total_async_events += planstate->n_async_events; + async_cxt->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 (async_cxt->wait_event_set != NULL && (reinit || + async_cxt->total_events > async_cxt->max_events)) { - FreeWaitEventSet(estate->es_wait_event_set); - estate->es_wait_event_set = NULL; + FreeWaitEventSet(async_cxt->wait_event_set); + async_cxt->wait_event_set = NULL; } } @@ -202,7 +241,9 @@ void ExecAsyncDoesNotNeedWait(PlanState *planstate) { int n; - EState *estate = planstate->state; + AsyncContext *async_cxt = planstate->state->es_async_cxt; + + Assert(async_cxt); if (planstate->n_async_events <= 0) return; @@ -211,21 +252,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 < async_cxt->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 (async_cxt->waiting_nodes[n] == planstate) break; - } } /* We should always find ourselves in the array. */ - Assert(n < estate->es_num_waiting_nodes); + Assert(n < async_cxt->num_waiting_nodes); + + async_cxt->waiting_nodes[n] = + async_cxt->waiting_nodes[--async_cxt->num_waiting_nodes]; /* We no longer need any asynchronous events. */ - estate->es_total_async_events -= planstate->n_async_events; + async_cxt->total_events -= planstate->n_async_events; planstate->n_async_events = 0; /* @@ -234,18 +274,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 (async_cxt->wait_event_set != NULL) { - FreeWaitEventSet(estate->es_wait_event_set); - estate->es_wait_event_set = NULL; + FreeWaitEventSet(async_cxt->wait_event_set); + async_cxt->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..4f468c1 100644 --- a/src/backend/executor/execProcnode.c +++ b/src/backend/executor/execProcnode.c @@ -377,6 +377,9 @@ ExecInitNode(Plan *node, EState *estate, PlanState *parent, int eflags) void ExecDispatchNode(PlanState *node) { + if (node->result_ready) + return; + if (node->instrument) InstrStartNode(node->instrument); @@ -559,6 +562,8 @@ void ExecExecuteNode(PlanState *node) { node->result_ready = false; + if (node->chgParam != NULL) /* something changed */ + ExecReScan(node); /* let ReScan handle this */ ExecDispatchNode(node); } @@ -569,28 +574,44 @@ 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 */ + /* + * Stash an active async context to the parent context if any then + * activate that of the given node. + */ + if (node->parent) + node->parent->save_async_cxt = node->state->es_async_cxt; + node->state->es_async_cxt = node->save_async_cxt; + ExecDispatchNode(node); if (!node->result_ready) ExecAsyncWaitForNode(node); - /* Result should be a TupleTableSlot, unless it's NULL. */ - Assert(node->result == NULL || IsA(node->result, TupleTableSlot)); + /* + * Save the active async context to the given node and restore the + * parent's. The stored context may have running nodes. + */ + node->save_async_cxt = node->state->es_async_cxt; + if (node->parent) + node->state->es_async_cxt = node->parent->save_async_cxt; - return (TupleTableSlot *) node->result; + return ExecConsumeResult(node); } @@ -848,9 +869,22 @@ ExecEndNode(PlanState *node) bool ExecShutdownNode(PlanState *node) { + bool ret; + if (node == NULL) return false; + /* + * Maintain the active async context on executor state. Differnet from + * ExecProcNode, this should be done only when saved context exists. + */ + if (node->save_async_cxt) + { + if (node->parent) + node->parent->save_async_cxt = node->state->es_async_cxt; + node->state->es_async_cxt = node->save_async_cxt; + } + switch (nodeTag(node)) { case T_GatherState: @@ -860,5 +894,16 @@ ExecShutdownNode(PlanState *node) break; } - return planstate_tree_walker(node, ExecShutdownNode, NULL); + ret = planstate_tree_walker(node, ExecShutdownNode, NULL); + + /* + * Restore the async context of the upper subtree only if exists. + */ + if (node->parent && node->parent->save_async_cxt) + { + node->save_async_cxt = node->state->es_async_cxt; + node->state->es_async_cxt = node->parent->save_async_cxt; + } + + 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/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..ad19486 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -239,6 +239,16 @@ ExecReturnTuple(PlanState *node, TupleTableSlot *slot) node->result_ready = true; } +/* Convenience function to retrieve a node's result. */ +static inline TupleTableSlot * +ExecConsumeResult(PlanState *node) +{ + Assert(node->result_ready); + Assert(node->result == NULL || IsA(node->result, TupleTableSlot)); + node->result_ready = false; + return (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..9121537 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -345,6 +345,17 @@ typedef struct ResultRelInfo List *ri_onConflictSetWhere; } ResultRelInfo; +/* Asynchronous execution support */ +typedef struct AsyncContext +{ + 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; +} AsyncContext; + /* ---------------- * EState information * @@ -382,13 +393,7 @@ typedef struct EState ParamListInfo es_param_list_info; /* values of external params */ 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; + AsyncContext *es_async_cxt; /* Async context currently active */ /* Other working state: */ MemoryContext es_query_cxt; /* per-query context in which EState lives */ @@ -1060,6 +1065,8 @@ typedef struct PlanState * subselects) */ List *subPlan; /* SubPlanState nodes in my expressions */ + AsyncContext *save_async_cxt; /* Stash for async context */ + /* * State for management of parameter-change-driven rescanning */ -- 1.8.3.1