Google Git
Sign in
apache / hawq / bd6196f487c8028f3eca0f56509fe9d718ee31c6 / . / src / backend / cdb / dispatcher.c
blob: 6322871deeecb591d11809e30add1044bb11b889 [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
/*-------------------------------------------------------------------------
*
* dispatcher.c
* Dispatcher is a coordinator between optimizer, wokermgr, tx mgr, and so on.
* Dispatcher like the logical dispatcher doing plan stuff and workermgr is like
* physical part of dispatcher. But workermgr should still be general to other
* module too.
*
* Dispatcher needs to compute the necessary information that QE needs and
* delivery them to the workermgr.
*
* TODO: QueryContext is not cool that its leve is not clear. ALTER use it
* outside of dispatcher and queryDesc use it inside. Have to redesign it.
*/
#include "postgres.h"
#include <pthread.h>
#include <limits.h>
#include "cdb/dispatcher.h"
#include "cdb/dispatcher_mgt.h"
#include "cdb/workermgr.h"
#include "cdb/executormgr.h"
#include "postmaster/identity.h"
#include "cdb/cdbdisp.h" /* TODO: should remove */
#include "cdb/cdbgang.h" /* SliceTable & Gang */
#include "cdb/cdbvars.h" /* gp_max_plan_size */
#include "executor/executor.h" /* RootSliceIndex */
#include "cdb/cdbrelsize.h" /* clear_relsize_cache */
#include "utils/memutils.h" /* GetMemoryChunkContext */
#include "cdb/cdbsrlz.h" /* serializeNode */
#include "utils/datum.h" /* datumGetSize */
#include "utils/faultinjector.h"
#include "utils/lsyscache.h" /* get_typlenbyval */
#include "miscadmin.h" /* CHECK_FOR_INTERRUPTS */
#include "tcop/tcopprot.h" /* ResetUsage */
#include "portability/instr_time.h" /* Monitor the dispatcher performance */
#include "cdb/cdbdispatchresult.h" /* cdbdisp_makeDispatchResults */
#include "lib/stringinfo.h" /* StringInfoData */
#include "tcop/pquery.h" /* PortalGetResource */
#include "resourcemanager/communication/rmcomm_QD2RM.h"
#include "storage/ipc.h"
/* Define and structure */
typedef struct DispatchTask
{
ProcessIdentity id;
/*
* Information about executor which one is assigned this task.
* NULL means match any segment except entry db!
*/
Segment *segment;
bool entrydb; /*whether this task should be dispatched to entrydb*/
} DispatchTask;
typedef struct DispatchSlice
{
int tasks_num;
DispatchTask *tasks;
} DispatchSlice;
typedef struct DispatchMetaData {
/* Share Information */
int all_slices_num; /* EXPLAIN needs all slices num to store ANALYZE stats */
int used_slices_num;
DispatchSlice *slices;
} DispatchMetaData;
/*
* DispatchState
* Change and check state should in a wise way.
*/
typedef enum DispatchState {
DS_INIT, /* Not running */
DS_RUN, /* Running */
DS_OK, /* Running without error */
DS_ERROR, /* Running encounter error */
DS_DONE, /* Cleanup and Error consumed */
} DispatchState;
#define dispatcher_set_state_init(data) ((data)->state = DS_INIT)
#define dispatcher_set_state_run(data) ((data)->state = DS_RUN)
#define dispatcher_set_state_ok(data) ((data)->state = DS_OK)
#define dispatcher_set_state_error(data) ((data)->state = DS_ERROR)
#define dispatcher_set_state_done(data) ((data)->state = DS_DONE)
#define dispatcher_is_state_init(data) ((data)->state == DS_INIT)
#define dispatcher_is_state_run(data) ((data)->state == DS_RUN)
#define dispatcher_is_state_ok(data) ((data)->state == DS_OK)
#define dispatcher_is_state_error(data) ((data)->state == DS_ERROR)
#define dispatcher_is_state_done(data) ((data)->state == DS_DONE)
#define ERROR_HOST_INIT_SIZE 5
typedef struct DispatchData
{
DispatchState state;
/* Original Information */
struct QueryDesc *queryDesc;
int localSliceId;
/* Payload */
DispatchCommandQueryParms *pQueryParms;
struct SliceTable *sliceTable;
/* Resource that we can use. */
bool resource_is_mine;
QueryResource *resource;
bool dispatch_to_all_cached_executors; /* sync all of executors */
/* Readonly for worker manager */
DispatchMetaData job;
/* Communication between dispatcher and worker manager. */
struct QueryExecutorTeam *query_executor_team;
struct CdbDispatchResults *results; /* TODO: libpq is ugly, should invent a new connection manager. */
/* Maintained by worker manager. */
void *worker_mgr_state;
/* Statistics */
int segment_num; /* the number of segments for this query */
int segment_num_on_entrydb;
int num_of_cached_executors;
int num_of_new_connected_executors;
/*
* This is the start time of dispatcher, use last dispatched executor
* finished time substract this is one is the total time.
*/
instr_time time_begin;
instr_time time_total; /* time_last_dispatch_end - time_begin */
instr_time time_connect; /* time_last_connect_end - time_first_connect_begin */
instr_time time_dispatch; /* time_last_dispatch_end - time_first_dispatch_begin */
/* Performance Counter: dispatcher side */
instr_time time_first_connect_begin;
instr_time time_last_connect_end;
instr_time time_first_dispatch_begin;
instr_time time_last_dispatch_end;
/* Performance Counter: executor side */
instr_time time_max_dispatched;
instr_time time_min_dispatched;
instr_time time_total_dispatched;
instr_time time_max_consumed;
instr_time time_min_consumed;
instr_time time_total_consumed;
instr_time time_max_free;
instr_time time_min_free;
instr_time time_total_free;
int num_of_dispatched;
} DispatchData;
/* The following two structures used to generailize dispatch statement */
typedef enum DispatchStatementType {
DST_STRING,
DST_NODE,
} DispatchStatementType;
typedef struct DispatchStatement {
DispatchStatementType type;
/* String type: change this to union is better */
const char *string;
const char *serializeQuerytree;
int serializeLenQuerytree;
/* Node type */
struct Node *node;
struct QueryContextInfo *ctxt;
/* Control tag */
bool dispatch_to_all_cached_executor;
} DispatchStatement;
/* Global states: DO NOT add global state unless you have to. */
static MemoryContext DispatchDataContext;
static int DispatchInitCount = -1;
/* Static function */
static DispatchCommandQueryParms *dispatcher_fill_query_param(const char *strCommand,
char *serializeQuerytree,
int serializeLenQuerytree,
char *serializePlantree,
int serializeLenPlantree,
char *serializeParams,
int serializeLenParams,
char *serializeSliceInfo,
int serializeLenSliceInfo,
int rootIdx,
QueryResource *resource);
static void dispatcher_split_logical_tasks_for_query_desc(DispatchData *data);
static void dispatcher_split_logical_tasks_for_statemet_or_node(DispatchData *data);
static bool dispatcher_init_works(DispatchData *data);
static bool dispatcher_compute_threads_num(DispatchData *data);
static bool dispatch_collect_executors_error(DispatchData * data);
static void dispatch_throw_error(DispatchData *data);
static void dispatch_init_env(void);
static int dispatcher_get_total_query_executors_num(DispatchData *data);
static void
remove_subquery_in_RTEs(Node *node)
{
if (node == NULL)
{
return;
}
if (IsA(node, RangeTblEntry))
{
RangeTblEntry *rte = (RangeTblEntry *)node;
if (RTE_SUBQUERY == rte->rtekind && NULL != rte->subquery)
{
/*
* replace subquery with a dummy subquery
*/
rte->subquery = makeNode(Query);
}
return;
}
if (IsA(node, List))
{
List *list = (List *) node;
ListCell *lc = NULL;
foreach(lc, list)
{
remove_subquery_in_RTEs((Node *) lfirst(lc));
}
}
}
/* should remove */
typedef struct {
int sliceIndex;
int children;
Slice *slice;
} sliceVec;
static int
compare_slice_order(const void *aa, const void *bb)
{
sliceVec *a = (sliceVec *)aa;
sliceVec *b = (sliceVec *)bb;
if (a->slice == NULL)
return 1;
if (b->slice == NULL)
return -1;
/* sort the writer gang slice first, because he sets the shared snapshot */
if (a->slice->is_writer && !b->slice->is_writer)
return -1;
else if (b->slice->is_writer && !a->slice->is_writer)
return 1;
if (a->children == b->children)
return 0;
else if (a->children > b->children)
return 1;
else
return -1;
}
static void
mark_bit(char *bits, int nth)
{
int nthbyte = nth >> 3;
char nthbit = 1 << (nth & 7);
bits[nthbyte] |= nthbit;
}
static void
or_bits(char* dest, char* src, int n)
{
int i;
for(i=0; i<n; i++)
dest[i] |= src[i];
}
static int
count_bits(char* bits, int nbyte)
{
int i;
int nbit = 0;
int bitcount[] = {
0, 1, 1, 2, 1, 2, 2, 3, 1, 2, 2, 3, 2, 3, 3, 4
};
for(i=0; i<nbyte; i++)
{
nbit += bitcount[bits[i] & 0x0F];
nbit += bitcount[(bits[i] >> 4) & 0x0F];
}
return nbit;
}
static int markbit_dep_children(SliceTable *sliceTable, int sliceIdx, sliceVec *sliceVec, int bitmasklen, char* bits)
{
ListCell *sublist;
Slice *slice = (Slice *) list_nth(sliceTable->slices, sliceIdx);
foreach(sublist, slice->children)
{
int childIndex = lfirst_int(sublist);
char *newbits = palloc0(bitmasklen);
markbit_dep_children(sliceTable, childIndex, sliceVec, bitmasklen, newbits);
or_bits(bits, newbits, bitmasklen);
mark_bit(bits, childIndex);
pfree(newbits);
}
sliceVec[sliceIdx].sliceIndex = sliceIdx;
sliceVec[sliceIdx].children = count_bits(bits, bitmasklen);
sliceVec[sliceIdx].slice = slice;
return sliceVec[sliceIdx].children;
}
static int
count_dependent_children(SliceTable * sliceTable, int sliceIndex, sliceVec *sliceVector, int len)
{
int ret = 0;
int bitmasklen = (len+7) >> 3;
char *bitmask = palloc0(bitmasklen);
ret = markbit_dep_children(sliceTable, sliceIndex, sliceVector, bitmasklen, bitmask);
pfree(bitmask);
return ret;
}
static int
fillSliceVector(SliceTable *sliceTbl, int rootIdx, sliceVec *sliceVector, int sliceLim)
{
int top_count;
/* count doesn't include top slice add 1 */
top_count = 1 + count_dependent_children(sliceTbl, rootIdx, sliceVector, sliceLim);
qsort(sliceVector, sliceLim, sizeof(sliceVec), compare_slice_order);
return top_count;
}
static void
aggregateQueryResource(QueryResource *queryRes)
{
MemoryContext old;
old = MemoryContextSwitchTo(DispatchDataContext);
if (queryRes &&
queryRes->segments &&
(list_length(queryRes->segments) > 0))
{
ListCell *lc1 = NULL;
ListCell *lc2 = NULL;
Segment *seg1 = NULL;
Segment *seg2 = NULL;
int nseg = 0;
int i = 0;
queryRes->numSegments = list_length(queryRes->segments);
queryRes->segment_vcore_agg = NULL;
queryRes->segment_vcore_agg = palloc0(sizeof(int) * queryRes->numSegments);
queryRes->segment_vcore_writer = palloc0(sizeof(int) * queryRes->numSegments);
foreach (lc1, queryRes->segments)
{
int j = 0;
nseg = 0;
seg1 = (Segment *) lfirst(lc1);
queryRes->segment_vcore_writer[i] = i;
foreach (lc2, queryRes->segments)
{
seg2 = (Segment *) lfirst(lc2);
if (strcmp(seg1->hostname, seg2->hostname) == 0)
{
nseg++;
if (j < queryRes->segment_vcore_writer[i])
{
queryRes->segment_vcore_writer[i] = j;
}
}
j++;
}
queryRes->segment_vcore_agg[i++] = nseg;
}
}
MemoryContextSwitchTo(old);
}
static void
dispatch_init_env(void)
{
++DispatchInitCount;
if (DispatchDataContext) {
if (DispatchInitCount == 0)
MemoryContextResetAndDeleteChildren(DispatchDataContext);
return;
}
DispatchDataContext = AllocSetContextCreate(TopMemoryContext,
"Dispatch Data Context",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
executormgr_setup_env(TopMemoryContext);
}
/* Implementation TODO: Need Modify resource parameter */
DispatchData *
initialize_dispatch_data(QueryResource *resource, bool dispatch_to_all_cached_executors)
{
DispatchData *dispatch_data;
MemoryContext old;
dispatch_init_env();
old = MemoryContextSwitchTo(DispatchDataContext);
dispatch_data = palloc0(sizeof(DispatchData));
MemoryContextSwitchTo(old);
dispatcher_set_state_init(dispatch_data);
dispatch_data->dispatch_to_all_cached_executors = dispatch_to_all_cached_executors;
if (dispatch_to_all_cached_executors)
{
/*
* If one statement needs to be dispatched to all cached
* executors, resource is useless.
*/
Assert(resource == NULL);
dispatch_data->segment_num = executormgr_get_cached_executor_num();
dispatch_data->segment_num_on_entrydb = executormgr_get_cached_executor_num_onentrydb();
}
else
{
Assert(resource);
dispatch_data->resource_is_mine = false;
dispatch_data->resource = resource;
dispatch_data->segment_num = list_length(resource->segments);
}
INSTR_TIME_SET_ZERO(dispatch_data->time_max_dispatched);
INSTR_TIME_SET_ZERO(dispatch_data->time_min_dispatched);
INSTR_TIME_SET_ZERO(dispatch_data->time_total_dispatched);
INSTR_TIME_SET_ZERO(dispatch_data->time_max_consumed);
INSTR_TIME_SET_ZERO(dispatch_data->time_min_consumed);
INSTR_TIME_SET_ZERO(dispatch_data->time_total_consumed);
INSTR_TIME_SET_ZERO(dispatch_data->time_max_free);
INSTR_TIME_SET_ZERO(dispatch_data->time_min_free);
INSTR_TIME_SET_ZERO(dispatch_data->time_total_free);
return dispatch_data;
}
/*
* prepare_dispatch_query_desc
* Have to generate logical splits of the plan.
*/
void
prepare_dispatch_query_desc(DispatchData *data,
struct QueryDesc *queryDesc)
{
char *splan,
*sparams;
int splan_len,
splan_len_uncompressed,
sparams_len;
SliceTable *sliceTbl;
int rootIdx;
PlannedStmt *stmt;
bool is_SRI;
data->queryDesc = queryDesc;
Assert(queryDesc != NULL && queryDesc->estate != NULL);
/*
* Later we'll need to operate with the slice table provided via the
* EState structure in the argument QueryDesc. Cache this information
* locally and assert our expectations about it.
*/
sliceTbl = queryDesc->estate->es_sliceTable;
rootIdx = RootSliceIndex(queryDesc->estate);
Assert(sliceTbl != NULL);
Assert(rootIdx == 0 ||
(rootIdx > sliceTbl->nMotions && rootIdx <= sliceTbl->nMotions + sliceTbl->nInitPlans));
/*
* Keep old value so we can restore it. We use this field as a parameter.
*/
data->localSliceId = sliceTbl->localSlice;
/*
* This function is called only for planned statements.
*/
stmt = queryDesc->plannedstmt;
Assert(stmt);
/*
* Let's evaluate STABLE functions now, so we get consistent values on the QEs
*
* Also, if this is a single-row INSERT statement, let's evaluate
* nextval() and currval() now, so that we get the QD's values, and a
* consistent value for everyone
*
*/
is_SRI = false;
if (queryDesc->operation == CMD_INSERT)
{
Assert(stmt->commandType == CMD_INSERT);
/* We might look for constant input relation (instead of SRI), but I'm afraid
* that wouldn't scale.
*/
is_SRI = IsA(stmt->planTree, Result) && stmt->planTree->lefttree == NULL;
}
if (!is_SRI)
clear_relsize_cache();
if (queryDesc->operation == CMD_INSERT ||
queryDesc->operation == CMD_SELECT ||
queryDesc->operation == CMD_UPDATE ||
queryDesc->operation == CMD_DELETE)
{
MemoryContext oldContext;
oldContext = CurrentMemoryContext;
if ( stmt->qdContext ) /* Temporary! See comment in PlannedStmt. */
{
oldContext = MemoryContextSwitchTo(stmt->qdContext);
}
else /* MPP-8382: memory context of plan tree should not change */
{
MemoryContext mc = GetMemoryChunkContext(stmt->planTree);
oldContext = MemoryContextSwitchTo(mc);
}
stmt->planTree = (Plan *) exec_make_plan_constant(stmt, is_SRI);
MemoryContextSwitchTo(oldContext);
}
/*
* MPP-20785:
* remove subquery field from RTE's since it is not needed during query
* execution,
* this is an optimization to reduce size of serialized plan before dispatching
*/
remove_subquery_in_RTEs((Node *) (queryDesc->plannedstmt->rtable));
/*
* serialized plan tree. Note that we're called for a single
* slice tree (corresponding to an initPlan or the main plan), so the
* parameters are fixed and we can include them in the prefix.
*/
splan = serializeNode((Node *) queryDesc->plannedstmt, &splan_len, &splan_len_uncompressed);
/* compute the total uncompressed size of the query plan for all slices */
int num_slices = queryDesc->plannedstmt->planTree->nMotionNodes + 1;
uint64 plan_size_in_kb = ((uint64)splan_len_uncompressed * (uint64)num_slices) / (uint64)1024;
elog(DEBUG1, "Query plan size to dispatch: " UINT64_FORMAT "KB", plan_size_in_kb);
if (0 < gp_max_plan_size && plan_size_in_kb > gp_max_plan_size)
{
ereport(ERROR,
(errcode(ERRCODE_STATEMENT_TOO_COMPLEX),
(errmsg("Query plan size limit exceeded, current size: " UINT64_FORMAT "KB, max allowed size: %dKB", plan_size_in_kb, gp_max_plan_size),
errhint("Size controlled by gp_max_plan_size"))));
}
Assert(splan != NULL && splan_len > 0 && splan_len_uncompressed > 0);
if (queryDesc->params != NULL && queryDesc->params->numParams > 0)
{
ParamListInfoData *pli;
ParamExternData *pxd;
StringInfoData parambuf;
Size length;
int plioff;
int32 iparam;
/* Allocate buffer for params */
initStringInfo(&parambuf);
/* Copy ParamListInfoData header and ParamExternData array */
pli = queryDesc->params;
length = (char *)&pli->params[pli->numParams] - (char *)pli;
plioff = parambuf.len;
Assert(plioff == MAXALIGN(plioff));
appendBinaryStringInfo(&parambuf, pli, length);
/* Copy pass-by-reference param values. */
for (iparam = 0; iparam < queryDesc->params->numParams; iparam++)
{
int16 typlen;
bool typbyval;
/* Recompute pli each time in case parambuf.data is repalloc'ed */
pli = (ParamListInfoData *)(parambuf.data + plioff);
pxd = &pli->params[iparam];
/* Does pxd->value contain the value itself, or a pointer? */
get_typlenbyval(pxd->ptype, &typlen, &typbyval);
if (!typbyval)
{
char *s = DatumGetPointer(pxd->value);
if (pxd->isnull ||
!PointerIsValid(s))
{
pxd->isnull = true;
pxd->value = 0;
}
else
{
length = datumGetSize(pxd->value, typbyval, typlen);
/* MPP-1637: we *must* set this before we
* append. Appending may realloc, which will
* invalidate our pxd ptr. (obviously we could
* append first if we recalculate pxd from the new
* base address) */
pxd->value = Int32GetDatum(length);
appendBinaryStringInfo(&parambuf, &iparam, sizeof(iparam));
appendBinaryStringInfo(&parambuf, s, length);
}
}
}
sparams = parambuf.data;
sparams_len = parambuf.len;
}
else
{
sparams = NULL;
sparams_len = 0;
}
data->pQueryParms = dispatcher_fill_query_param(queryDesc->sourceText,
NULL, 0,
splan, splan_len,
sparams, sparams_len,
NULL, 0,
rootIdx,
queryDesc->resource);
Assert(sliceTbl);
Assert(sliceTbl->slices != NIL);
data->sliceTable = sliceTbl;
dispatcher_split_logical_tasks_for_query_desc(data);
}
static void
prepare_dispatch_statement_string(DispatchData *data,
const char *statement,
const char *serializeQuerytree,
int serializeLenQuerytree)
{
data->pQueryParms = dispatcher_fill_query_param(statement,
(char *) serializeQuerytree, serializeLenQuerytree,
NULL, 0,
NULL, 0,
NULL, 0,
0,
data->resource);
dispatcher_split_logical_tasks_for_statemet_or_node(data);
}
static void
prepare_dispatch_statement_node(struct DispatchData *data,
struct Node *node,
struct QueryContextInfo *ctxt)
{
char *serializedQuerytree;
int serializedQuerytree_len;
Query *q = makeNode(Query);
/* Construct an utility query node. */
q->commandType = CMD_UTILITY;
q->utilityStmt = node;
q->contextdisp = ctxt;
q->querySource = QSRC_ORIGINAL;
/*
* We must set q->canSetTag = true. False would be used to hide a command
* introduced by rule expansion which is not allowed to return its
* completion status in the command tag (PQcmdStatus/PQcmdTuples). For
* example, if the original unexpanded command was SELECT, the status
* should come back as "SELECT n" and should not reflect other commands
* inserted by rewrite rules. True means we want the status.
*/
q->canSetTag = true;
serializedQuerytree = serializeNode((Node *) q, &serializedQuerytree_len, NULL /*uncompressed_size*/);
data->pQueryParms = dispatcher_fill_query_param(debug_query_string,
serializedQuerytree, serializedQuerytree_len,
NULL, 0,
NULL, 0,
NULL, 0,
0,
data->resource);
dispatcher_split_logical_tasks_for_statemet_or_node(data);
}
/*
* dispatcher_split_logical_tasks_for_query_desc
* Have to generate a parallel plan to execute.
*/
static void
dispatcher_split_logical_tasks_for_query_desc(DispatchData *data)
{
MemoryContext old;
old = MemoryContextSwitchTo(DispatchDataContext);
sliceVec *sliceVector = NULL;
int i;
int slice_num;
/*
* Dispatch order is top-bottom, this will reduce the useless pkts sends
* from scan nodes.
*/
sliceVector = palloc0(list_length(data->sliceTable->slices) * sizeof(*sliceVector));
slice_num = fillSliceVector(data->sliceTable,
data->pQueryParms->rootIdx,
sliceVector,
list_length(data->sliceTable->slices));
data->job.used_slices_num = slice_num;
data->job.all_slices_num = list_length(data->sliceTable->slices);
data->job.slices = palloc0(data->job.used_slices_num * sizeof(DispatchSlice));
for (i = 0; i < data->job.used_slices_num; i++)
{
data->job.slices[i].tasks_num = list_length(data->resource->segments);
data->job.slices[i].tasks = palloc0(data->job.slices[i].tasks_num * sizeof(DispatchTask));
}
for (i = 0; i < slice_num; i++)
{
Slice *slice = sliceVector[i].slice;
bool is_writer = false;
is_writer = (i == 0 && !data->queryDesc->extended_query);
/* Not all of slices need to dispatch, such as root slice! */
if (slice->gangType == GANGTYPE_UNALLOCATED)
{
data->job.used_slices_num--;
continue;
}
/* Direct dispatch may not useful in Hawq 2.0. */
if (slice->directDispatch.isDirectDispatch)
{
DispatchTask *task = &data->job.slices[i].tasks[0];
data->job.slices[i].tasks_num = 1;
task->id.id_in_slice = linitial_int(slice->directDispatch.contentIds);
task->id.slice_id = slice->sliceIndex;
task->id.gang_member_num = list_length(data->resource->segments);
task->id.command_count = gp_command_count;
task->id.is_writer = is_writer;
task->segment = list_nth(data->resource->segments, task->id.id_in_slice);
task->id.init = true;
}
else if (slice->gangSize == 1)
{
DispatchTask *task = &data->job.slices[i].tasks[0];
/* Run some single executor case. Should merge with direct dispatch! */
data->job.slices[i].tasks_num = 1;
/* TODO: I really want to remove the MASTER_CONTENT_ID. */
if (slice->gangType == GANGTYPE_ENTRYDB_READER)
{
task->id.id_in_slice = MASTER_CONTENT_ID;
task->segment = data->resource->master;
}
else
{
task->id.id_in_slice = 0;
task->segment = linitial(data->resource->segments);
}
task->id.slice_id = slice->sliceIndex;
task->id.gang_member_num = list_length(data->resource->segments);
task->id.command_count = gp_command_count;
task->id.is_writer = is_writer;
task->id.init = true;
}
else
{
int j;
/* Assign the id from 0. */
for (j = 0; j < data->job.slices[i].tasks_num; j++)
{
DispatchTask *task = &data->job.slices[i].tasks[j];
task->id.id_in_slice = j;
task->id.slice_id = slice->sliceIndex;
task->id.gang_member_num = list_length(data->resource->segments);
task->id.command_count = gp_command_count;
task->id.is_writer = is_writer;
task->segment = list_nth(data->resource->segments, task->id.id_in_slice);
task->id.init = true;
}
}
}
pfree(sliceVector);
MemoryContextSwitchTo(old);
}
/*
* dispatcher_split_logical_tasks_for_statemet_or_node
*/
static void
dispatcher_split_logical_tasks_for_statemet_or_node(DispatchData *data)
{
MemoryContext old;
int i;
int segment_num, segment_num_entrydb;
segment_num = data->segment_num;
segment_num_entrydb = data->segment_num_on_entrydb;
if ((segment_num == 0) && (segment_num_entrydb == 0))
{
return;
}
old = MemoryContextSwitchTo(DispatchDataContext);
data->job.used_slices_num = segment_num_entrydb + 1;
data->job.all_slices_num = segment_num_entrydb + 1;
data->job.slices = palloc0(data->job.used_slices_num * sizeof(DispatchSlice));
for (i = 0; i < segment_num_entrydb; i++)
{
/* TODO: where should I store the runtime context ?? */
data->job.slices[i].tasks_num = 1;
data->job.slices[i].tasks = palloc0(
data->job.slices[i].tasks_num * sizeof(DispatchTask));
DispatchTask *task = &data->job.slices[i].tasks[0];
task->id.slice_id = i;
task->id.id_in_slice = MASTER_CONTENT_ID;
task->id.gang_member_num = 1;
task->id.command_count = gp_command_count;
task->id.is_writer = true;
/* NULL means ANY segment */
task->segment = NULL; // ?? How to get the master resource
task->entrydb = true;
task->id.init = true;
}
for (i = segment_num_entrydb; i < data->job.used_slices_num; i++)
{
/* TODO: where should I store the runtime context ?? */
data->job.slices[i].tasks_num = segment_num;
data->job.slices[i].tasks = palloc0(data->job.slices[i].tasks_num * sizeof(DispatchTask));
}
/* Set tasks id for segment. We have only one slice! */
for (i = 0; i < data->job.slices[segment_num_entrydb].tasks_num; i++)
{
DispatchTask *task = &data->job.slices[segment_num_entrydb].tasks[i];
task->id.slice_id = segment_num_entrydb;
task->id.id_in_slice = i;
task->id.gang_member_num = segment_num;
task->id.command_count = gp_command_count;
task->id.is_writer = true;
/* NULL means ANY segment */
task->segment = data->dispatch_to_all_cached_executors ? NULL : list_nth(data->resource->segments, i);
task->id.init = true;
}
data->pQueryParms->primary_gang_id = -1;
MemoryContextSwitchTo(old);
}
/*
* dispatcher_bind_executor
* Bind the dispatcher smallest dispatch data to executor.
*/
static bool
dispatcher_bind_executor(DispatchData *data)
{
int i, j;
struct QueryExecutor *executor;
QueryExecutorIterator iterator;
List *concurrent_connect_executors = NIL;
ListCell *lc;
bool ret = true;
bool bindExecutorError = false;
dispmgt_init_query_executor_iterator(data->query_executor_team, &iterator);
for (i = 0; i < data->job.used_slices_num; i++)
{
DispatchSlice *slice = &data->job.slices[i];
if (bindExecutorError)
break;
for (j = 0; j < slice->tasks_num; j++)
{
DispatchTask *task = &slice->tasks[j];
struct SegmentDatabaseDescriptor *desc;
executor = dispmgt_get_query_executor_iterator(&iterator, false);
desc = executormgr_allocate_executor(task->segment, task->id.is_writer, task->entrydb);
if (!desc)
{
// may have problem for tasks assigned to entrydb for dispatch_statement_node
// and dispatch_statement_string
if (task->segment != NULL)
{
struct ConcurrentConnectExecutorInfo *info;
info = dispmgt_build_preconnect_info(task->segment,
task->id.is_writer, executor,
data, slice, task);
concurrent_connect_executors = lappend(concurrent_connect_executors,
info);
data->num_of_new_connected_executors++;
if (!executormgr_bind_executor_task(data, executor, info->desc, task, slice))
{
bindExecutorError = true;
break;
}
}
continue;
}
#ifdef FAULT_INJECTOR
FaultInjector_InjectFaultIfSet(
ConnectionFailAfterGangCreation,
DDLNotSpecified,
"", // databaseName
""); // tableName
#endif
if (!executormgr_bind_executor_task(data, executor, desc, task, slice)) {
bindExecutorError = true;
break;
}
data->num_of_cached_executors++;
}
}
if (data->dispatch_to_all_cached_executors)
elog(DEBUG1, "dispatch to all cached executors: expected: %d actual: %d new connected(should be 0): %d",
dispatcher_get_total_query_executors_num(data),
data->num_of_cached_executors,
data->num_of_new_connected_executors);
if (concurrent_connect_executors == NIL)
return true;
if (!bindExecutorError) {
ret = dispmgt_concurrent_connect(concurrent_connect_executors,
gp_connections_per_thread);
} else {
ret = false;
}
foreach(lc, concurrent_connect_executors)
dispmgt_free_preconnect_info(lfirst(lc));
list_free(concurrent_connect_executors);
return ret;
}
/*
* dispatcher_serialize_state
*/
static void
dispatcher_serialize_state(DispatchData *data)
{
int i,
j;
struct QueryExecutor *executor;
QueryExecutorIterator iterator;
char *sliceInfo;
int sliceInfoLen;
SliceTable *sliceTable = data->queryDesc ? data->queryDesc->estate->es_sliceTable : NULL;
/* Firstly, serialize every executor state. */
dispmgt_init_query_executor_iterator(data->query_executor_team, &iterator);
for (i = 0; i < data->job.used_slices_num; i++)
{
DispatchSlice *slice = &data->job.slices[i];
for (j = 0; j < slice->tasks_num; j++)
{
DispatchTask *task = &slice->tasks[j];
executor = dispmgt_get_query_executor_iterator(&iterator, true);
if (executor == NULL)
continue;
executormgr_serialize_executor_state(data, executor, task, slice);
if (sliceTable)
{
Slice *plan_slice = list_nth(sliceTable->slices, task->id.slice_id);
CdbProcess *proc = makeNode(CdbProcess);//list_nth(plan_slice->primaryProcesses, get_task_id_in_slice(slice, task));
/* Update the real executors information for interconnect. */
executormgr_get_executor_connection_info(executor, &proc->listenerAddr, &proc->listenerPort, &proc->pid);
proc->contentid = task->id.id_in_slice;
plan_slice->primaryProcesses = lappend(plan_slice->primaryProcesses, proc);
}
}
}
/* Ok, we can serialize the global state. */
if (sliceTable)
{
sliceInfo = serializeNode((Node *) data->queryDesc->estate->es_sliceTable, &sliceInfoLen, NULL /*uncompressed_size*/);
data->pQueryParms->serializedSliceInfo = sliceInfo;
data->pQueryParms->serializedSliceInfolen = sliceInfoLen;
}
}
/*
* dispatcher_serialize_query_resource
*/
static void
dispatcher_serialize_query_resource(DispatchData *data)
{
char *serializedQueryResource = NULL;
int serializedQueryResourcelen = 0;
QueryResource * queryResource = data->resource;
if (queryResource)
{
aggregateQueryResource(queryResource);
serializedQueryResource = serializeNode(( Node *) queryResource, &serializedQueryResourcelen, NULL);
data->pQueryParms->serializedQueryResource = serializedQueryResource;
data->pQueryParms->serializedQueryResourcelen = serializedQueryResourcelen;
if (queryResource->segment_vcore_agg)
{
pfree(queryResource->segment_vcore_agg);
}
if (queryResource->segment_vcore_writer)
{
pfree(queryResource->segment_vcore_writer);
}
}
}
/*
* dispatcher_unbind_executor
*/
static void
dispatcher_unbind_executor(DispatchData *data)
{
int i,
j;
struct QueryExecutor *executor;
QueryExecutorIterator iterator;
dispmgt_init_query_executor_iterator(data->query_executor_team, &iterator);
for (i = 0; i < data->job.used_slices_num; i++)
{
DispatchSlice *slice = &data->job.slices[i];
for (j = 0; j < slice->tasks_num; j++)
{
DispatchTask *task = &slice->tasks[j];
executor = dispmgt_get_query_executor_iterator(&iterator, true);
if (executor == NULL)
continue;
executormgr_unbind_executor_task(data, executor, task, slice);
}
}
}
static void
dispatcher_update_statistics_per_executor(DispatchData *data, struct QueryExecutor *executor)
{
instr_time time_connect_begin;
instr_time time_connect_end;
instr_time time_connect_diff;
instr_time time_dispatch_begin;
instr_time time_dispatch_end;
instr_time time_dispatch_diff;
instr_time time_consume_begin;
instr_time time_consume_end;
instr_time time_consume_diff;
instr_time time_free_begin;
instr_time time_free_end;
instr_time time_free_diff;
bool first_time = (data->num_of_dispatched == 0);
executormgr_get_statistics(executor, &time_connect_begin, &time_connect_end,
&time_dispatch_begin, &time_dispatch_end,
&time_consume_begin, &time_consume_end,
&time_free_begin, &time_free_end);
INSTR_TIME_SET_ZERO(time_connect_diff);
INSTR_TIME_SET_ZERO(time_dispatch_diff);
INSTR_TIME_SET_ZERO(time_consume_diff);
INSTR_TIME_SET_ZERO(time_free_diff);
INSTR_TIME_ACCUM_DIFF(time_connect_diff, time_connect_end, time_connect_begin);
INSTR_TIME_ACCUM_DIFF(time_dispatch_diff, time_dispatch_end, time_dispatch_begin);
INSTR_TIME_ACCUM_DIFF(time_consume_diff, time_consume_end, time_consume_begin);
INSTR_TIME_ACCUM_DIFF(time_free_diff, time_free_end, time_free_begin);
data->num_of_dispatched++;
if (first_time)
{
/* dispatcher side */
INSTR_TIME_ASSIGN(data->time_first_connect_begin, time_connect_begin);
INSTR_TIME_ASSIGN(data->time_last_connect_end, time_connect_end);
INSTR_TIME_ASSIGN(data->time_first_dispatch_begin, time_dispatch_begin);
INSTR_TIME_ASSIGN(data->time_last_dispatch_end, time_dispatch_end);
/* executor side */
INSTR_TIME_ACCUM_DIFF(data->time_total_dispatched, time_dispatch_end, time_dispatch_begin);
INSTR_TIME_ACCUM_DIFF(data->time_max_dispatched, time_dispatch_end, time_dispatch_begin);
INSTR_TIME_ACCUM_DIFF(data->time_min_dispatched, time_dispatch_end, time_dispatch_begin);
INSTR_TIME_ACCUM_DIFF(data->time_total_consumed, time_consume_end, time_consume_begin);
INSTR_TIME_ACCUM_DIFF(data->time_max_consumed, time_consume_end, time_consume_begin);
INSTR_TIME_ACCUM_DIFF(data->time_min_consumed, time_consume_end, time_consume_begin);
INSTR_TIME_ACCUM_DIFF(data->time_total_free, time_free_end, time_free_begin);
INSTR_TIME_ACCUM_DIFF(data->time_max_free, time_free_end, time_free_begin);
INSTR_TIME_ACCUM_DIFF(data->time_min_free, time_free_end, time_free_begin);
return;
}
/* dispatch side */
if (data->num_of_new_connected_executors > 0)
{
if (INSTR_TIME_GREATER_THAN(data->time_first_connect_begin, time_connect_begin))
INSTR_TIME_ASSIGN(data->time_first_connect_begin, time_connect_begin);
if (INSTR_TIME_LESS_THAN(data->time_last_connect_end, time_connect_end))
INSTR_TIME_ASSIGN(data->time_last_connect_end, time_connect_end);
}
if (INSTR_TIME_GREATER_THAN(data->time_first_dispatch_begin, time_dispatch_begin))
INSTR_TIME_ASSIGN(data->time_first_dispatch_begin, time_dispatch_begin);
if (INSTR_TIME_LESS_THAN(data->time_last_dispatch_end, time_dispatch_end))
INSTR_TIME_ASSIGN(data->time_last_dispatch_end, time_dispatch_end);
/* executor side */
if (INSTR_TIME_LESS_THAN(data->time_max_dispatched, time_dispatch_diff))
INSTR_TIME_ASSIGN(data->time_max_dispatched, time_dispatch_diff);
if (INSTR_TIME_GREATER_THAN(data->time_min_dispatched, time_dispatch_diff))
INSTR_TIME_ASSIGN(data->time_min_dispatched, time_dispatch_diff);
if (INSTR_TIME_LESS_THAN(data->time_max_consumed, time_consume_diff))
INSTR_TIME_ASSIGN(data->time_max_consumed, time_consume_diff);
if (INSTR_TIME_GREATER_THAN(data->time_min_consumed, time_consume_diff))
INSTR_TIME_ASSIGN(data->time_min_consumed, time_consume_diff);
if (INSTR_TIME_LESS_THAN(data->time_max_free, time_free_diff))
INSTR_TIME_ASSIGN(data->time_max_free, time_free_diff);
if (INSTR_TIME_GREATER_THAN(data->time_min_free, time_free_diff))
INSTR_TIME_ASSIGN(data->time_min_free, time_free_diff);
INSTR_TIME_ADD(data->time_total_dispatched, time_dispatch_diff);
INSTR_TIME_ADD(data->time_total_consumed, time_consume_diff);
INSTR_TIME_ADD(data->time_total_free, time_free_diff);
}
static void
dispatcher_update_statistics(DispatchData *data)
{
int i,
j;
struct QueryExecutor *executor;
QueryExecutorIterator iterator;
dispmgt_init_query_executor_iterator(data->query_executor_team, &iterator);
for (i = 0; i < data->job.used_slices_num; i++)
{
DispatchSlice *slice = &data->job.slices[i];
for (j = 0; j < slice->tasks_num; j++)
{
executor = dispmgt_get_query_executor_iterator(&iterator, true);
if (executor == NULL)
continue;
dispatcher_update_statistics_per_executor(data, executor);
}
}
INSTR_TIME_ACCUM_DIFF(data->time_total, data->time_last_dispatch_end, data->time_begin);
INSTR_TIME_ACCUM_DIFF(data->time_connect, data->time_last_connect_end, data->time_first_connect_begin);
INSTR_TIME_ACCUM_DIFF(data->time_dispatch, data->time_last_dispatch_end, data->time_first_dispatch_begin);
}
/*
* dispatch_plan
* Create corresponding dispatching data structure. It requires caller splits
* plan. This function will compute the necessary threads number and dispatch
* plan to each executor.
*/
void
dispatch_run(DispatchData *data)
{
if (Debug_print_execution_detail) {
instr_time time;
INSTR_TIME_SET_CURRENT(time);
elog(DEBUG1,"The time on entering dispatch_run: %.3f ms",
1000.0 * INSTR_TIME_GET_DOUBLE(time));
}
if ((data->segment_num == 0) && (data->segment_num_on_entrydb == 0))
{
data->state = DS_DONE;
return;
}
dispatcher_init_works(data);
INSTR_TIME_SET_CURRENT(data->time_begin);
if (!dispatcher_bind_executor(data))
goto error;
#ifdef FAULT_INJECTOR
FaultInjector_InjectFaultIfSet(
FailQeAfterConnection,
DDLNotSpecified,
"", // databaseName
""); // tableName
#endif
/*
* Only after we have the executors, we can serialize the state. Or we
* don't know the executor listening address.
*/
CHECK_FOR_INTERRUPTS();
dispatcher_serialize_state(data);
dispatcher_serialize_query_resource(data);
dispatcher_set_state_run(data);
dispmgt_dispatch_and_run(data->worker_mgr_state, data->query_executor_team);
if (Debug_print_execution_detail) {
instr_time time;
INSTR_TIME_SET_CURRENT(time);
elog(DEBUG1,"The time after dispatching all the results: %.3f ms",
1000.0 * INSTR_TIME_GET_DOUBLE(time));
}
return;
error:
dispatcher_set_state_error(data);
elog(ERROR, "dispatcher encounter error");
return;
}
/*
* dispatch_wait
* Wait the workermgr return and setup error informaton returned from executors.
*/
void
dispatch_wait(DispatchData *data)
{
if (Debug_print_execution_detail) {
instr_time time;
INSTR_TIME_SET_CURRENT(time);
elog(DEBUG1,"The time on entering dispatch_wait: %.3f ms",
1000.0 * INSTR_TIME_GET_DOUBLE(time));
}
if (dispatcher_is_state_ok(data) ||
dispatcher_is_state_error(data) ||
dispatcher_is_state_done(data))
return;
workermgr_wait_job(data->worker_mgr_state);
CHECK_FOR_INTERRUPTS();
/* Check executors state before return. */
if (dispatch_collect_executors_error(data))
dispatcher_set_state_error(data);
else
dispatcher_set_state_ok(data);
if (Debug_print_execution_detail) {
instr_time time;
INSTR_TIME_SET_CURRENT(time);
elog(DEBUG1,"The time after collecting all the executor results: %.3f ms",
1000.0 * INSTR_TIME_GET_DOUBLE(time));
}
}
/*
* free_dispatch_data
* free the DispatchData allocated resource
*/
void free_dispatch_data(struct DispatchData *data) {
if (data->pQueryParms) {
pfree(data->pQueryParms);
data->pQueryParms = NULL;
}
if (data->job.slices) {
for (int i = 0; i < data->job.used_slices_num; ++i) {
if (data->job.slices[i].tasks) {
pfree(data->job.slices[i].tasks);
data->job.slices[i].tasks = NULL;
}
}
pfree(data->job.slices);
data->job.slices = NULL;
}
if (data->query_executor_team) {
QueryExecutorGroup *qeGrp = data->query_executor_team->query_executor_groups;
if (qeGrp) {
if (qeGrp->fds) {
pfree(qeGrp->fds);
qeGrp->fds = NULL;
}
for (int j = 0; j < qeGrp->query_executor_num; ++j) {
pfree(qeGrp->query_executors[j]);
qeGrp->query_executors[j] = NULL;
}
}
pfree(data->query_executor_team);
data->query_executor_team = NULL;
}
pfree(data);
}
/*
* dispatch_end_env
* accompanied with dispatch_init_env
*/
void dispatch_end_env(struct DispatchData *data) {
Assert(data != NULL);
--DispatchInitCount;
cdbdisp_destroyDispatchResults(data->results);
data->results = NULL;
dispatcher_set_state_done(data);
PG_TRY();
{
dispatcher_unbind_executor(data);
if (data->resource_is_mine)
{
FreeResource(data->resource);
data->resource_is_mine = false;
}
}
PG_CATCH();
{
free_dispatch_data(data);
PG_RE_THROW();
}
PG_END_TRY();
free_dispatch_data(data);
}
/*
* dispatch_cleanup
* Cleanup the workermgr. Report error if something happended on executors.
*/
void
dispatch_cleanup(DispatchData *data)
{
/* to avoid duplicate dispatch cleanup */
if (data == NULL) return;
if (dispatcher_is_state_done(data)) return;
/*
* We should not get here when dispatcher hit an exception. But
* executors may have some troubles.
*/
if (dispatcher_is_state_error(data))
{
/* We cannot unbind executors until we retrieve error message! */
if (!proc_exit_inprogress) {
dispatch_throw_error(data);
Assert(false);
return; /* should not hit */
}
}
dispatch_end_env(data);
}
/*
* dispatch_catch_error
* Cancel the workermgr work and report error based on bug on dispatcher or
* executors.
*/
void
dispatch_catch_error(DispatchData *data)
{
int qderrcode = elog_geterrcode();
bool useQeError = false;
/* Init state means no thread and data is not correct! */
if (!dispatcher_is_state_init(data))
{
workermgr_cancel_job(data->worker_mgr_state);
/* Have to figure out exception source */
dispatch_collect_executors_error(data);
}
/*
* When a QE stops executing a command due to an error, as a
* consequence there can be a cascade of interconnect errors
* (usually "sender closed connection prematurely") thrown in
* downstream processes (QEs and QD). So if we are handling
* an interconnect error, and a QE hit a more interesting error,
* we'll let the QE's error report take precedence.
*/
if (qderrcode == ERRCODE_GP_INTERCONNECTION_ERROR)
{
bool qd_lost_flag = false;
char *qderrtext = elog_message();
if (qderrtext && strcmp(qderrtext, CDB_MOTION_LOST_CONTACT_STRING) == 0)
qd_lost_flag = true;
if (data->results && data->results->errcode)
{
if (qd_lost_flag && data->results->errcode == ERRCODE_GP_INTERCONNECTION_ERROR)
useQeError = true;
else if (data->results->errcode != ERRCODE_GP_INTERCONNECTION_ERROR)
useQeError = true;
}
}
if (useQeError)
{
/*
* Throw the QE's error, catch it, and fall thru to return
* normally so caller can finish cleaning up. Afterwards
* caller must exit via PG_RE_THROW().
*/
PG_TRY();
{
dispatch_throw_error(data);
}
PG_CATCH();
{} /* nop; fall thru */
PG_END_TRY();
}
dispatch_end_env(data);
}
void
cleanup_dispatch_data(DispatchData *data)
{
if (data->queryDesc)
data->queryDesc->estate->es_sliceTable->localSlice = data->localSliceId;
}
static void
dispatch_statement(DispatchStatement *stmt, QueryResource *resource, DispatchDataResult *result)
{
volatile DispatchData *data;
data = initialize_dispatch_data(resource, stmt->dispatch_to_all_cached_executor);
if (stmt->type == DST_STRING)
prepare_dispatch_statement_string((DispatchData *) data,
stmt->string,
stmt->serializeQuerytree,
stmt->serializeLenQuerytree);
else
prepare_dispatch_statement_node((DispatchData *) data,
stmt->node,
stmt->ctxt);
PG_TRY();
{
dispatch_run((DispatchData *) data);
dispatch_wait((DispatchData *) data);
if (dispatcher_is_state_ok(data) && result)
{
int segment_num = data->segment_num + data->segment_num_on_entrydb;
if (segment_num > 0)
{
initStringInfo(&result->errbuf);
result->result = cdbdisp_returnResults(segment_num, data->results, &result->errbuf, &result->numresults);
/* cdbdisp_returnResults free the pointer ! */
data->results = NULL;
/* Takeover the list of segment connections. */
result->segment_conns = dispmgt_takeover_segment_conns(data->query_executor_team);
}
}
dispatch_cleanup((DispatchData *) data);
}
PG_CATCH();
{
dispatch_wait((DispatchData *) data);
dispatch_cleanup((DispatchData *) data);
PG_RE_THROW();
}
PG_END_TRY();
}
void
dispatch_statement_string(const char *string,
const char *serializeQuerytree,
int serializeLenQuerytree,
QueryResource *resource,
DispatchDataResult *result,
bool sync_on_all_executors)
{
DispatchStatement stmt;
MemSet(&stmt, 0, sizeof(stmt));
stmt.type = DST_STRING;
stmt.string = string;
stmt.serializeQuerytree = serializeQuerytree;
stmt.serializeLenQuerytree = serializeLenQuerytree;
stmt.dispatch_to_all_cached_executor = sync_on_all_executors;
dispatch_statement(&stmt, resource, result);
}
/*
* dispatch_statement_node
*/
void
dispatch_statement_node(struct Node *node,
struct QueryContextInfo *ctxt,
QueryResource *resource,
DispatchDataResult *result)
{
DispatchStatement stmt;
Assert(resource);
MemSet(&stmt, 0, sizeof(stmt));
stmt.type = DST_NODE;
stmt.node = node;
stmt.ctxt = ctxt;
stmt.dispatch_to_all_cached_executor = false;
dispatch_statement(&stmt, resource, result);
}
void
dispatch_free_result(DispatchDataResult *result)
{
if (!result)
return;
if(result->errbuf.data)
{
pfree(result->errbuf.data);
result->errbuf.data = NULL;
}
if (result->result)
{
for (int i = 0; i <= result->numresults; i++){
PQclear(result->result[i]);
result->result[i] = NULL;
}
free(result->result);
}
dispmgt_free_takeoved_segment_conns(result->segment_conns);
}
struct CdbDispatchResults *
dispatch_get_results(struct DispatchData *data)
{
return data->results;
}
int
dispatch_get_segment_num(struct DispatchData *data)
{
return list_length(data->resource->segments);
}
/*
*
*/
static DispatchCommandQueryParms *
dispatcher_fill_query_param(const char *strCommand,
char *serializeQuerytree,
int serializeLenQuerytree,
char *serializePlantree,
int serializeLenPlantree,
char *serializeParams,
int serializeLenParams,
char *serializeSliceInfo,
int serializeLenSliceInfo,
int rootIdx,
QueryResource *resource)
{
DispatchCommandQueryParms *queryParms;
Segment *master = NULL;
MemoryContext old;
if (resource)
{
master = resource->master;
}
old = MemoryContextSwitchTo(DispatchDataContext);
queryParms = palloc0(sizeof(*queryParms));
MemoryContextSwitchTo(old);
queryParms->strCommand = strCommand;
queryParms->strCommandlen = strCommand ? strlen(strCommand) + 1 : 0;
queryParms->serializedQuerytree = serializeQuerytree;
queryParms->serializedQuerytreelen = serializeLenQuerytree;
queryParms->serializedPlantree = serializePlantree;
queryParms->serializedPlantreelen = serializeLenPlantree;
queryParms->serializedParams = serializeParams;
queryParms->serializedParamslen = serializeLenParams;
queryParms->serializedSliceInfo = serializeSliceInfo;
queryParms->serializedSliceInfolen= serializeLenSliceInfo;
queryParms->rootIdx = rootIdx;
/* sequence server info */
if (master)
{
queryParms->seqServerHost = pstrdup(master->hostip);
queryParms->seqServerHostlen = strlen(master->hostip) + 1;
}
queryParms->seqServerPort = seqServerCtl->seqServerPort;
queryParms->primary_gang_id = 0; /* We are relying on the slice table to provide gang ids */
/*
* in hawq, there is no distributed transaction
*/
queryParms->serializedDtxContextInfo = NULL;
queryParms->serializedDtxContextInfolen = 0;
return queryParms;
}
/*
* dispatcher_init_works
* This function is create all of data structures.
*/
static bool
dispatcher_init_works(DispatchData *data)
{
MemoryContext old;
old = MemoryContextSwitchTo(DispatchDataContext);
/* Compute the threads number & */
dispatcher_compute_threads_num(data);
/* Create the threads data strucutre. */
data->worker_mgr_state = workermgr_create_workermgr_state(dispmgt_get_group_num(data->query_executor_team));
MemoryContextSwitchTo(old);
return true;
}
static bool
dispatcher_compute_threads_num(DispatchData *data)
{
int threads_num = 1;
int query_executors_num;
int executors_num_per_thread = gp_connections_per_thread;
query_executors_num = dispatcher_get_total_query_executors_num(data);
data->results = cdbdisp_makeDispatchResults(query_executors_num, data->job.all_slices_num, false);
/* TODO: decide the groups(threads) number. */
if (query_executors_num == 0)
threads_num = 1;
if (executors_num_per_thread > query_executors_num)
threads_num = 1;
else
threads_num = (query_executors_num / executors_num_per_thread) + ((query_executors_num % executors_num_per_thread == 0) ? 0 : 1);
data->query_executor_team = dispmgt_create_dispmgt_state(data,
threads_num,
query_executors_num,
executors_num_per_thread);
return true;
}
DispatchCommandQueryParms *
dispatcher_get_QueryParms(DispatchData *data)
{
return data->pQueryParms;
}
static int
dispatcher_get_total_query_executors_num(DispatchData *data)
{
int num = 0;
int i;
for (i = 0; i < data->job.used_slices_num; i++)
num += data->job.slices[i].tasks_num;
return num;
}
ProcessIdentity *
dispatch_get_task_identity(DispatchTask *task)
{
return &task->id;
}
static bool
dispatch_collect_executors_error(DispatchData *data)
{
MemoryContext old;
old = MemoryContextSwitchTo(DispatchDataContext);
QueryExecutorIterator iterator;
struct QueryExecutor *executor;
bool ret = false;
// the host list of error executors
int errHostNum = 0;
int errHostSize = ERROR_HOST_INIT_SIZE;
char **errHostName = (char**) palloc0(errHostSize * sizeof(char *));
dispmgt_init_query_executor_iterator(data->query_executor_team, &iterator);
while ((executor = dispmgt_get_query_executor_iterator(&iterator, true)) != NULL)
{
executormgr_merge_error_for_dispatcher(executor, &errHostSize,
&errHostNum, &errHostName);
}
if (dispatcher_has_error(data)) {
if (errHostNum > 0)
sendFailedNodeToResourceManager(errHostNum, errHostName);
ret = true;
}
for (int i = 0; i < errHostNum; i++) {
pfree(errHostName[i]);
}
pfree(errHostName);
MemoryContextSwitchTo(old);
return ret;
}
static void
dispatch_throw_error(DispatchData *data)
{
StringInfoData buf;
int errorcode = data->results->errcode;
/* Format error messages from the primary gang. */
initStringInfo(&buf);
cdbdisp_dumpDispatchResults(data->results, &buf, false);
/* Too bad, our gang got an error. */
PG_TRY();
{
ereport(ERROR, (errcode(errorcode),
errOmitLocation(true),
errmsg("%s", buf.data)));
}
PG_CATCH();
{
pfree(buf.data);
PG_RE_THROW();
}
PG_END_TRY();
}
bool
dispatcher_has_error(DispatchData *data)
{
return data->results && data->results->errcode;
}
void
dispatcher_print_statistics(StringInfo buf, DispatchData *data)
{
if (!data)
return;
dispatcher_update_statistics(data);
if (data->num_of_dispatched == 0)
return;
appendStringInfo(buf, "Dispatcher statistics:\n");
appendStringInfo(buf,
" executors used(total/cached/new connection): (%d/%d/%d);",
data->num_of_dispatched, data->num_of_cached_executors,
data->num_of_new_connected_executors);
appendStringInfo(buf,
" dispatcher time(total/connection/dispatch data): (%.3f ms/%.3f ms/%.3f ms).\n",
INSTR_TIME_GET_MILLISEC(data->time_total),
INSTR_TIME_GET_MILLISEC(data->time_connect),
INSTR_TIME_GET_MILLISEC(data->time_dispatch));
appendStringInfo(buf,
" dispatch data time(max/min/avg): (%.3f ms/%.3f ms/%.3f ms);",
INSTR_TIME_GET_MILLISEC(data->time_max_dispatched),
INSTR_TIME_GET_MILLISEC(data->time_min_dispatched),
INSTR_TIME_GET_MILLISEC(data->time_total_dispatched)
/ data->num_of_dispatched);
appendStringInfo(buf,
" consume executor data time(max/min/avg): (%.3f ms/%.3f ms/%.3f ms);",
INSTR_TIME_GET_MILLISEC(data->time_max_consumed),
INSTR_TIME_GET_MILLISEC(data->time_min_consumed),
INSTR_TIME_GET_MILLISEC(data->time_total_consumed)
/ data->num_of_dispatched);
appendStringInfo(buf,
" free executor time(max/min/avg): (%.3f ms/%.3f ms/%.3f ms).\n",
INSTR_TIME_GET_MILLISEC(data->time_max_free),
INSTR_TIME_GET_MILLISEC(data->time_min_free),
INSTR_TIME_GET_MILLISEC(data->time_total_free) / data->num_of_dispatched);
}
/*
* Check the connection from the dispatcher to verify that it is still there.
* Return true if the dispatcher connection is still alive.
*/
bool dispatch_validate_conn(pgsocket sock)
{
ssize_t ret;
char buf;
if (sock < 0)
return false;
#ifndef WIN32
ret = recv(sock, &buf, 1, MSG_PEEK|MSG_DONTWAIT);
#else
ret = recv(sock, &buf, 1, MSG_PEEK|MSG_PARTIAL);
#endif
if (ret == 0) /* socket has been closed. EOF */
return false;
if (ret > 0) /* data waiting on socket */
{
if (buf == 'E') /* waiting data indicates error */
return false;
else
return true;
}
if (ret == -1) /* error, or would be block. */
{
if (errno == EAGAIN || errno == EINPROGRESS || errno == EWOULDBLOCK)
return true; /* connection intact, no data available */
else
return false;
}
/* not reached */
return true;
}