head 1.34; access; symbols release_4_2:1.34 aix_ok:1.33 Version_2_1:1.13; locks; strict; comment @ * @; 1.34 date 94.02.02.00.22.27; author marc; state Exp; branches; next 1.33; 1.33 date 92.07.04.04.38.27; author mer; state Exp; branches; next 1.32; 1.32 date 92.06.28.03.48.22; author mao; state Exp; branches; next 1.31; 1.31 date 92.06.18.06.04.44; author hong; state Exp; branches; next 1.30; 1.30 date 92.06.16.21.29.18; author hong; state Exp; branches; next 1.29; 1.29 date 92.03.31.23.12.23; author mer; state Exp; branches; next 1.28; 1.28 date 92.02.07.10.50.35; author hong; state Exp; branches; next 1.27; 1.27 date 92.01.21.16.43.58; author hong; state Exp; branches; next 1.26; 1.26 date 91.11.17.21.06.36; author mer; state Exp; branches; next 1.25; 1.25 date 91.11.11.22.59.33; author hong; state Exp; branches; next 1.24; 1.24 date 91.11.09.09.51.00; author hong; state Exp; branches; next 1.23; 1.23 date 91.10.09.14.53.29; author hong; state Exp; branches; next 1.22; 1.22 date 91.10.08.01.13.48; author hong; state Exp; branches; next 1.21; 1.21 date 91.08.25.13.35.03; author hong; state Exp; branches; next 1.20; 1.20 date 91.08.18.12.33.44; author hong; state Exp; branches; next 1.19; 1.19 date 91.08.08.15.39.11; author hong; state Exp; branches; next 1.18; 1.18 date 91.08.02.19.00.46; author hong; state Exp; branches; next 1.17; 1.17 date 91.07.24.16.13.40; author hong; state Exp; branches; next 1.16; 1.16 date 91.07.17.23.45.03; author hong; state Exp; branches; next 1.15; 1.15 date 91.06.18.23.28.36; author cimarron; state Exp; branches; next 1.14; 1.14 date 91.04.18.11.38.42; author hong; state Exp; branches; next 1.13; 1.13 date 91.02.27.16.04.17; author cimarron; state Exp; branches; next 1.12; 1.12 date 91.02.27.14.32.03; author hong; state Exp; branches; next 1.11; 1.11 date 90.11.12.08.16.58; author hong; state Exp; branches; next 1.10; 1.10 date 90.11.10.12.29.23; author hong; state Exp; branches; next 1.9; 1.9 date 90.11.07.12.06.50; author hong; state Exp; branches; next 1.8; 1.8 date 90.10.02.14.38.43; author hong; state Exp; branches; next 1.7; 1.7 date 90.10.01.08.44.45; author cimarron; state Exp; branches; next 1.6; 1.6 date 90.09.18.22.03.44; author hong; state Exp; branches; next 1.5; 1.5 date 90.09.10.20.43.41; author hong; state Exp; branches; next 1.4; 1.4 date 90.09.06.16.57.50; author hong; state Exp; branches; next 1.3; 1.3 date 90.09.06.14.44.04; author hong; state Exp; branches; next 1.2; 1.2 date 90.08.22.11.14.54; author hong; state Exp; branches; next 1.1; 1.1 date 90.08.17.13.28.20; author hong; state Exp; branches; next ; desc @routines for parallel query processing originally in pquery.c @ 1.34 log @Fixes to make sure we don't expect NameData variables to be NULL terminated and that we don't write mroe than NAMEDATALEN chars into them @ text @/* ---------------------------------------------------------------- * FILE * pfrag.c * * DESCRIPTION * POSTGRES process query command code * * INTERFACE ROUTINES * ProcessQuery * * NOTES * * IDENTIFICATION * $Header: /usr/local/devel/postgres.test/src/backend/tcop/RCS/pfrag.c,v 1.33 1992/07/04 04:38:27 mer Exp marc $ * ---------------------------------------------------------------- */ #include #include #include "tmp/postgres.h" #include "tcop/tcopdebug.h" #include "tcop/slaves.h" #include "nodes/pg_lisp.h" #include "nodes/plannodes.h" #include "nodes/plannodes.a.h" #include "nodes/execnodes.h" #include "nodes/execnodes.a.h" #include "executor/execmisc.h" #include "executor/x_execinit.h" #include "executor/x_hash.h" #include "tcop/dest.h" #include "tmp/portal.h" #include "commands/command.h" #include "parser/parsetree.h" #include "storage/lmgr.h" #include "catalog/pg_relation.h" #include "utils/lsyscache.h" #include "utils/log.h" #include "utils/palloc.h" #include "nodes/relation.h" #include "lib/copyfuncs.h" #include "lib/catalog.h" #include "tcop/pquery.h" RcsId("$Header: /usr/local/devel/postgres.test/src/backend/tcop/RCS/pfrag.c,v 1.33 1992/07/04 04:38:27 mer Exp marc $"); int AdjustParallelismEnabled = 1; extern int NStriping; static MasterSchedulingInfoData MasterSchedulingInfoD = {-1, NULL, -1, NULL}; extern ParallelismModes ParallelismMode; /* ------------------------------------ * FingFragments * * find all the plan fragments under plan node, mark the fragments starting * with fragmentNo * plan fragments are obtained by decomposing the plan tree across all * blocking edges, i.e., edges out of Hash nodes and Sort nodes * ------------------------------------ */ static List FindFragments(parsetree, node, fragmentNo) List parsetree; Plan node; int fragmentNo; { List subFragments = LispNil; List newFragments; Fragment fragment; set_fragment(node, fragmentNo); if (get_lefttree(node) != NULL) if (ExecIsHash(get_lefttree(node))||ExecIsMergeJoin(get_lefttree(node))) { /* ----------------------------- * detected a blocking edge, fragment boundary * ----------------------------- */ fragment = RMakeFragment(); set_frag_root(fragment, (Plan)get_lefttree(node)); set_frag_parent_op(fragment, node); set_frag_parallel(fragment, 1); set_frag_subtrees(fragment, LispNil); set_frag_parsetree(fragment, parsetree); set_frag_is_inprocess(fragment, false); set_frag_iorate(fragment, 0.0); subFragments = nappend1(subFragments, (LispValue)fragment); } else { subFragments = FindFragments(parsetree,get_lefttree(node),fragmentNo); } if (get_righttree(node) != NULL) if (ExecIsHash(get_righttree(node)) || ExecIsSort(get_righttree(node))) { /* ----------------------------- * detected a blocking edge, fragment boundary * ----------------------------- */ fragment = RMakeFragment(); set_frag_root(fragment, (Plan)get_righttree(node)); set_frag_parent_op(fragment, node); set_frag_parallel(fragment, 1); set_frag_subtrees(fragment, LispNil); set_frag_parsetree(fragment, parsetree); set_frag_is_inprocess(fragment, false); set_frag_iorate(fragment, 0.0); subFragments = nappend1(subFragments, (LispValue)fragment); } else { newFragments = FindFragments(parsetree,get_righttree(node),fragmentNo); subFragments = nconc(subFragments, newFragments); } return subFragments; } #define SEQPERTUPTIME 2e-3 /* second */ #define INDPERTUPTIME 0.1 #define DEFPERTUPTIME 4e-3 static float get_pertuptime(plan) Plan plan; { switch (NodeType(plan)) { case classTag(SeqScan): return SEQPERTUPTIME; case classTag(IndexScan): return INDPERTUPTIME; } return DEFPERTUPTIME; } #define AVGINDTUPS 5 static float compute_frag_iorate(fragment) Fragment fragment; { Plan plan; int fragmentno; float pertupletime; Plan p; float iorate; int tupsize; plan = get_frag_root(fragment); fragmentno = get_fragment(plan); pertupletime = 0.0; for (;;) { pertupletime += get_pertuptime(plan); p = get_outerPlan(plan); /* walk down the outer path only for now */ if (p == NULL || get_fragment(p) != fragmentno) break; else plan = p; } if (ExecIsSeqScan(plan) || ExecIsScanTemps(plan)) { iorate = 1.0/(pertupletime * get_plan_tupperpage(plan)); } else if (ExecIsIndexScan(plan)) { iorate = 1.0/(pertupletime * AVGINDTUPS); } return iorate; } /* ------------------------------- * SetIoRate * * compute and set the io rate of each fragment * ------------------------------- */ static void SetIoRate(fragment) Fragment fragment; { LispValue x; Fragment f; set_frag_iorate(fragment, compute_frag_iorate(fragment)); foreach (x, get_frag_subtrees(fragment)) { f = (Fragment)CAR(x); SetIoRate(f); } } /* -------------------------------- * InitialPlanFragments * * calls FindFragments() recursively to obtain the initial set of * plan fragments -- the largest possible, further decomposition * might be necessary in DecomposeFragments(). * -------------------------------- */ Fragment InitialPlanFragments(parsetree, plan) Plan plan; List parsetree; { Plan node; LispValue x, y; List fragmentlist; List subFragments; List newFragmentList; int fragmentNo = 0; Fragment rootFragment; Fragment fragment, frag; rootFragment = RMakeFragment(); set_frag_root(rootFragment, plan); set_frag_parent_op(rootFragment, NULL); set_frag_parallel(rootFragment, 1); set_frag_subtrees(rootFragment, LispNil); set_frag_parent_frag(rootFragment, NULL); set_frag_parsetree(rootFragment, parsetree); set_frag_is_inprocess(rootFragment, false); set_frag_iorate(rootFragment, 0.0); fragmentlist = lispCons((LispValue)rootFragment, LispNil); while (!lispNullp(fragmentlist)) { newFragmentList = LispNil; foreach (x, fragmentlist) { fragment = (Fragment)CAR(x); node = get_frag_root(fragment); subFragments = FindFragments(parsetree, node, fragmentNo++); set_frag_subtrees(fragment, subFragments); foreach (y, subFragments) { frag = (Fragment)CAR(y); set_frag_parent_frag(frag, (FragmentPtr)fragment); } newFragmentList = nconc(newFragmentList, subFragments); } fragmentlist = newFragmentList; } SetIoRate(rootFragment); return rootFragment; } extern int NBuffers; /* ----------------------------- * GetCurrentMemSize * * get the current amount of available memory * ----------------------------- */ static int GetCurrentMemSize() { return NBuffers; /* YYY functionalities to be added later */ } /* ----------------------------- * GetCurrentLoadAverage * * get the current load average of the system * ----------------------------- */ static float GetCurrentLoadAverage() { return 0.0; /* YYY functionalities to be added later */ } /* ------------------------------ * GetReadyFragments * * get the set of fragments that are ready to fire, i.e., those that * have no children * ------------------------------ */ static List GetReadyFragments(fragments) Fragment fragments; { LispValue x; Fragment frag; List readyFragments = LispNil; List readyFrags; if (lispNullp(get_frag_subtrees(fragments)) && !get_frag_is_inprocess(fragments)) return lispCons((LispValue)fragments, LispNil); foreach (x, get_frag_subtrees(fragments)) { frag = (Fragment)CAR(x); readyFrags = GetReadyFragments(frag); readyFragments = nconc(readyFragments, readyFrags); } return readyFragments; } /* --------------------------------- * GetFitFragments * * get the set of fragments that can fit in the current available memory * --------------------------------- */ static List GetFitFragments(fragmentlist, memsize) List fragmentlist; int memsize; { return fragmentlist; /* YYY functionalities to be added later */ } /* ---------------------------------- * DecomposeFragments * * decompose fragments into smaller fragments to fit in memsize amount * of memory * ---------------------------------- */ static List DecomposeFragments(fragmentlist, memsize) List fragmentlist; int memsize; { return fragmentlist; /* YYY functionalities to be added later */ } /* ----------------------------------- * ChooseToFire * * choose among all the ready-to-fire fragments which * to execute in parallel * ----------------------------------- */ static List ChooseToFire(fragmentlist, memsize) List fragmentlist; int memsize; { return lispCons(CAR(fragmentlist), LispNil); /* YYY functionalities to be added later */ } /* ---------------------------------- * ChooseFragments * * choose the fragments to execute in parallel * ----------------------------------- */ static List ChooseFragments(fragments, memsize) Fragment fragments; int memsize; { List readyFragments; List fitFragments; List fireFragments; readyFragments = GetReadyFragments(fragments); if (lispNullp(readyFragments)) return LispNil; fitFragments = GetFitFragments(readyFragments, memsize); if (lispNullp(fitFragments)) { fitFragments = DecomposeFragments(fitFragments, memsize); if (lispNullp(fitFragments)) elog(WARN, "memory below hashjoin threshold."); } fireFragments = ChooseToFire(fitFragments, memsize); return fireFragments; } /* ------------------------------ * SetParallelDegree * * set the degree of parallelism for fragments in fragmentlist * ------------------------------ */ static void SetParallelDegree(fragmentlist, nfreeslaves) List fragmentlist; int nfreeslaves; { LispValue x; Fragment fragment; Plan plan; int fragmentno; /* YYY more functionalities to be added later */ foreach (x,fragmentlist) { fragment = (Fragment)CAR(x); plan = get_frag_root(fragment); fragmentno = get_fragment(plan); if (fragmentno < 0) { set_frag_parallel(fragment, 1); /* YYY */ } else { set_frag_parallel(fragment, nfreeslaves); /* YYY */ } } } /* --------------------------------- * plan_is_parallelizable * * returns true if plan is parallelizable, false otherwise * --------------------------------- */ static bool plan_is_parallelizable(plan) Plan plan; { if (plan == NULL) return true; if (ExecIsMergeJoin(plan)) return false; if (ExecIsSort(plan)) return false; if (ExecIsIndexScan(plan)) { List indxqual; LispValue x; NameData opname; Oper op; indxqual = get_indxqual((IndexScan)plan); if (length(CAR(indxqual)) < 2) return false; foreach (x, CAR(indxqual)) { op = (Oper)CAR(CAR(x)); opname = get_opname(get_opno(op)); if (namestrcmp(&opname, "=") == 0) return false; } } if (plan_is_parallelizable(get_outerPlan(plan))) return true; return false; } /* ---------------------------------------- * nappend1iobound * * insert an io-bound fragment into a list in * descending io rate order. * ---------------------------------------- */ static List nappend1iobound(ioboundlist, frag) List ioboundlist; Fragment frag; { LispValue x; Fragment f; if (lispNullp(ioboundlist)) return lispCons((LispValue)frag, LispNil); f = (Fragment)CAR(ioboundlist); if (get_frag_iorate(frag) > get_frag_iorate(f)) { return(nconc(lispCons((LispValue)frag, LispNil), ioboundlist)); } else { return(nconc(lispCons((LispValue)f, LispNil), nappend1iobound(CDR(ioboundlist), frag))); } } /* ---------------------------------------- * nappend1cpubound * * insert a cpu-bound fragment into a list in * ascending io rate order. * ---------------------------------------- */ static List nappend1cpubound(cpuboundlist, frag) List cpuboundlist; Fragment frag; { LispValue x; Fragment f; if (lispNullp(cpuboundlist)) return lispCons((LispValue)frag, LispNil); f = (Fragment)CAR(cpuboundlist); if (get_frag_iorate(frag) < get_frag_iorate(f)) { return(nconc(lispCons((LispValue)frag, LispNil), cpuboundlist)); } else { return(nconc(lispCons((LispValue)f, LispNil), nappend1cpubound(CDR(cpuboundlist), frag))); } } #define DISKBANDWIDTH 60 /* IO per second */ /* ------------------------------------- * ClassifyFragments * * classify fragments into io-bound, cpu-bound, unparallelizable or * parallelism-preset. * ------------------------------------- */ static void ClassifyFragments(fraglist, ioboundlist, cpuboundlist, unparallelizablelist, presetlist) List fraglist; List *ioboundlist, *cpuboundlist, *unparallelizablelist, *presetlist; { LispValue x; Fragment f; Plan p; float iorate; float diagonal; *ioboundlist = LispNil; *cpuboundlist = LispNil; *unparallelizablelist = LispNil; *presetlist = LispNil; diagonal = (float)NStriping * DISKBANDWIDTH/(float)GetNumberSlaveBackends(); foreach (x, fraglist) { f = (Fragment)CAR(x); p = get_frag_root(f); if (!plan_is_parallelizable(p)) { *unparallelizablelist = nappend1(*unparallelizablelist, (LispValue)f); } else if (!lispNullp(parse_parallel(get_frag_parsetree(f)))) { *presetlist = nappend1(*presetlist, (LispValue)f); } else { iorate = get_frag_iorate(f); if (iorate > diagonal) { *ioboundlist = nappend1iobound(*ioboundlist, f); } else { *cpuboundlist = nappend1cpubound(*cpuboundlist, f); } } } } /* --------------------------------- * ComputeIoCpuBalancePoint * * compute io/cpu balance point of two fragments: * f1, io-bound * f2, cpu-bound * --------------------------------- */ static void ComputeIoCpuBalancePoint(f1, f2, x1, x2) Fragment f1, f2; int *x1, *x2; { float bandwidth; float iorate1, iorate2; int nfreeslaves; nfreeslaves = GetNumberSlaveBackends(); bandwidth = NStriping * DISKBANDWIDTH; iorate1 = get_frag_iorate(f1); iorate2 = get_frag_iorate(f2); *x1=(int)floor((double)((bandwidth-iorate2*nfreeslaves)/(iorate1-iorate2))); *x2=(int)ceil((double)((iorate1*nfreeslaves-bandwidth)/(iorate1-iorate2))); } /* -------------------------------------- * MaxFragParallelism * * return the maximum parallelism for a fragment * within the limit of number of free processors and disk bandwidth * ------------------------------------- */ static int MaxFragParallelism(frag) Fragment frag; { float ioRate; int par; if (!plan_is_parallelizable(get_frag_root(frag))) return 1; if (!lispNullp(parse_parallel(get_frag_parsetree(frag)))) return CInteger(parse_parallel(get_frag_parsetree(frag))); ioRate = (float)get_frag_iorate(frag); par = MIN((int)floor((double)NStriping*DISKBANDWIDTH/ioRate), NumberOfFreeSlaves); return par; } /* -------------------------------------- * CurMaxFragParallelism * * return the maximum parallelism for a fragment * within the limit of current number of free processors and * available disk bandwidth * ------------------------------------- */ static int CurMaxFragParallelism(frag, curbandwidth, nfreeslaves) Fragment frag; float curbandwidth; int nfreeslaves; { float ioRate; int par; if (!plan_is_parallelizable(get_frag_root(frag))) return 1; if (!lispNullp(parse_parallel(get_frag_parsetree(frag)))) return CInteger(parse_parallel(get_frag_parsetree(frag))); ioRate = (float)get_frag_iorate(frag); par = MIN((int)floor((double)curbandwidth/ioRate), nfreeslaves); return par; } /* ------------------------ * AdjustParallelism * * dynamically adjust degrees of parallelism of the fragments that * are already in process to take advantage of the extra processors * ------------------------ */ static void AdjustParallelism(pardelta, groupid) int pardelta; int groupid; { int j; int slave; int max_curpage; int size; int oldnproc; Assert(pardelta != 0); SLAVE_elog(DEBUG, "master trying to adjust degrees of parallelism"); SLAVE1_elog(DEBUG, "master sending signal to process group %d", groupid); signalProcGroup(groupid, SIGPARADJ); max_curpage = getProcGroupMaxPage(groupid); SLAVE1_elog(DEBUG, "master gets maxpage = %d", max_curpage); oldnproc = ProcGroupInfoP[groupid].nprocess; if (max_curpage == NOPARADJ) { /* -------------------------- * forget about adjustment to parallelism * in this case -- the fragment is almost finished * --------------------------- */ SLAVE_elog(DEBUG, "master changes mind on adjusting parallelism"); ProcGroupInfoP[groupid].paradjpage = NOPARADJ; OneSignalM(&(ProcGroupInfoP[groupid].m1lock), oldnproc); return; } ProcGroupInfoP[groupid].paradjpage = max_curpage + 1; /* page on which to adjust par. */ if (pardelta > 0) { ProcGroupInfoP[groupid].nprocess += pardelta; ProcGroupInfoP[groupid].scounter.count = ProcGroupInfoP[groupid].nprocess; ProcGroupInfoP[groupid].newparallel = ProcGroupInfoP[groupid].nprocess; SLAVE2_elog(DEBUG, "master signals waiting slaves with adjpage=%d,newpar=%d", ProcGroupInfoP[groupid].paradjpage, ProcGroupInfoP[groupid].newparallel); OneSignalM(&(ProcGroupInfoP[groupid].m1lock), oldnproc); set_frag_parallel(ProcGroupLocalInfoP[groupid].fragment, get_frag_parallel(ProcGroupLocalInfoP[groupid].fragment)+ pardelta); ProcGroupSMBeginAlloc(groupid); size = sizeofTmpRelDesc(QdGetPlan(ProcGroupInfoP[groupid].queryDesc)); for (j=0; j= 0) { SetParallelDegree(fireFragmentList, x2); } else { SetParallelDegree(fireFragmentList, NumberOfFreeSlaves); } } MasterSchedulingInfoD.cpuBoundFrag = f2; } else if (cpu_running) { curpar = get_frag_parallel(f2); if (ParallelismMode == INTER_WO_ADJ) { int newpar; float curband; curband = NStriping*DISKBANDWIDTH - curpar*get_frag_iorate(f2); newpar = CurMaxFragParallelism(f1, curband, NumberOfFreeSlaves); if (newpar == 0) return LispNil; fireFragmentList = lispCons((LispValue)f1, LispNil); SetParallelDegree(fireFragmentList, newpar); } else { pardelta = x2 - curpar; if (pardelta != 0) { SLAVE_elog(DEBUG, "adjusting parallelism of cpu-bound task."); AdjustParallelism(pardelta, MasterSchedulingInfoD.cpuBoundGroupId); } fireFragmentList = lispCons((LispValue)f1, LispNil); if (pardelta >= 0) { SetParallelDegree(fireFragmentList, x1); } else { SetParallelDegree(fireFragmentList, NumberOfFreeSlaves); } } MasterSchedulingInfoD.ioBoundFrag = f1; } else { SetParallelDegree((y=lispCons((LispValue)f1, LispNil)), x1); fireFragmentList = nconc(fireFragmentList, y); SetParallelDegree((y=lispCons((LispValue)f2, LispNil)), x2); fireFragmentList = nconc(fireFragmentList, y); MasterSchedulingInfoD.ioBoundFrag = f1; MasterSchedulingInfoD.cpuBoundFrag = f2; } } else if (f1 == NULL && f2 != NULL) { /* ----------------------------- * out of io-bound fragments, use intra-operation parallelism only * for cpu-bound fragments. * ------------------------------ */ fireFragmentList = lispCons((LispValue)f2, LispNil); SetParallelDegree(fireFragmentList, nfreeslaves); MasterSchedulingInfoD.cpuBoundFrag = f2; SLAVE1_elog(DEBUG, "out of io-bound tasks, running cpu-bound task with parallelism %d", nfreeslaves); } else if (f1 != NULL && f2 == NULL) { nfreeslaves = MaxFragParallelism(f1); fireFragmentList = lispCons((LispValue)f1, LispNil); SetParallelDegree(fireFragmentList, nfreeslaves); fireFragmentList = nconc(fireFragmentList, y); MasterSchedulingInfoD.ioBoundFrag = f1; SLAVE1_elog(DEBUG, "out of cpu-bound tasks, running io-bound task with parallelism %d", nfreeslaves); } return fireFragmentList; } #define MINHASHTABLEMEMORYKEY 1000 static IpcMemoryKey nextHashTableMemoryKey = 0; /* ------------------------- * getNextHashTableMemoryKey * * get the next hash table key * ------------------------- */ static IpcMemoryKey getNextHashTableMemoryKey() { extern int MasterPid; return (nextHashTableMemoryKey++ + MINHASHTABLEMEMORYKEY + MasterPid); } /* ------------------------ * sizeofTmpRelDesc * * calculate the size of reldesc of the temporary relation of plan * ------------------------ */ static int sizeofTmpRelDesc(plan) Plan plan; { List targetList; int natts; int size; targetList = get_qptargetlist(plan); natts = ExecTargetListLength(targetList); /* ------------------ * see CopyRelDescUsing() in lib/C/copyfuncs.c if you want to know * how size if derived. * ------------------ */ size = sizeof(RelationData) + (natts - 1) * sizeof(TupleDescriptorData) + sizeof(RuleLock) + sizeof(RelationTupleFormD) + sizeof(LockInfoData) + natts * (sizeof(AttributeTupleFormData) + sizeof(RuleLock)) + 48; /* some extra for possible LONGALIGN() */ return size; } /* ---------------------------------------------------------------- * OptimizeAndExecuteFragments * * Optimize plan fragments to explore both intra-fragment * and inter-fragment parallelism and execute the "optimal" * parallel plan * * ---------------------------------------------------------------- */ void OptimizeAndExecuteFragments(fragmentlist, destination) List fragmentlist; CommandDest destination; { LispValue x; int i; List currentFragmentList; Fragment fragment; int nparallel; List finalResultRelation; Plan plan; List parsetree; Plan parentPlan; Fragment parentFragment; int groupid; ProcessNode *p; List subtrees; List fragQueryDesc; HashJoinTable hashtable; int hashTableMemorySize; IpcMemoryKey hashTableMemoryKey; IpcMemoryKey getNextHashTableMemoryKey(); ScanTemps scantempNode; Relation tempRelationDesc; List tempRelationDescList; Relation shmTempRelationDesc; List fraglist; CommandDest dest; int size; fraglist = fragmentlist; while (!lispNullp(fraglist)) { /* ------------ * choose the set of fragments to execute and parallelism * for each fragment. * ------------ */ currentFragmentList = ParallelOptimize(fraglist); foreach (x, currentFragmentList) { fragment = (Fragment)CAR(x); nparallel = get_frag_parallel(fragment); plan = get_frag_root(fragment); parsetree = get_frag_parsetree(fragment); parentFragment = (Fragment)get_frag_parent_frag(fragment); finalResultRelation = parse_tree_result_relation(parsetree); dest = destination; dest = None; /* WWW */ if (ExecIsHash(plan)) { /* ------------ * if it is hashjoin, create the hash table * so that the slaves can share it * ------------ */ hashTableMemoryKey = getNextHashTableMemoryKey(); set_hashtablekey((Hash)plan, hashTableMemoryKey); hashtable = ExecHashTableCreate(plan); set_hashtable((Hash)plan, hashtable); hashTableMemorySize = get_hashtablesize((Hash)plan); parse_tree_result_relation(parsetree) = LispNil; } else if (get_fragment(plan) >= 0) { /* ------------ * this means that this an intermediate fragment, so * the result should be kept in some temporary relation * ------------ */ /* WWW parse_tree_result_relation(parsetree) = lispCons(lispAtom("intotemp"), LispNil); */ dest = None; } /* --------------- * create query descriptor for the fragment * --------------- */ fragQueryDesc = CreateQueryDesc(parsetree, plan, (char *) NULL, (ObjectId *) NULL, 0, dest); /* --------------- * assign a process group to work on the fragment * --------------- */ groupid = getFreeProcGroup(nparallel); if (fragment == MasterSchedulingInfoD.ioBoundFrag) MasterSchedulingInfoD.ioBoundGroupId = groupid; else if (fragment == MasterSchedulingInfoD.cpuBoundFrag) MasterSchedulingInfoD.cpuBoundGroupId = groupid; ProcGroupLocalInfoP[groupid].fragment = fragment; ProcGroupInfoP[groupid].status = WORKING; ProcGroupSMBeginAlloc(groupid); ProcGroupInfoP[groupid].queryDesc = (List) CopyObjectUsing((Node)fragQueryDesc, ProcGroupSMAlloc); size = sizeofTmpRelDesc(plan); for (p = ProcGroupLocalInfoP[groupid].memberProc; p != NULL; p = p->next) { SlaveInfoP[p->pid].resultTmpRelDesc = (Relation)ProcGroupSMAlloc(size); } ProcGroupSMEndAlloc(); ProcGroupInfoP[groupid].scounter.count = nparallel; ProcGroupInfoP[groupid].nprocess = nparallel; #ifdef TCOP_SLAVESYNCDEBUG { char procs[100]; p = ProcGroupLocalInfoP[groupid].memberProc; sprintf(procs, "%d", p->pid); for (p = p->next; p != NULL; p = p->next) sprintf(procs+strlen(procs), ",%d", p->pid); SLAVE2_elog(DEBUG, "master to wake up procgroup %d {%s} for", groupid, procs); set_query_range_table(parsetree); pplan(plan); fprintf(stderr, "\n"); } #endif wakeupProcGroup(groupid); set_frag_is_inprocess(fragment, true); /* --------------- * restore the original result relation descriptor * --------------- */ parse_tree_result_relation(parsetree) = finalResultRelation; } /* ------------ * if there are extra processors lying around, * dynamically adjust degrees of parallelism of * fragments that are already in process. if (NumberOfFreeSlaves > 0 && AdjustParallelismEnabled) { AdjustParallelism(NumberOfFreeSlaves, -1); } * ------------ */ /* ---------------- * wait for some process group to complete execution * ---------------- */ MasterWait: P_Finished(); /* -------------- * some process group has finished processing a fragment, * find that group * -------------- */ groupid = getFinishedProcGroup(); if (ProcGroupInfoP[groupid].status == PARADJPENDING) { /* ----------------------------- * master decided earlier than process group, groupid's parallelism * was going to be reduced. now the adjustment point is reached. * master is ready to collect those slaves spared from the * group. * ----------------------------- */ ProcessNode *p, *prev, *nextp; int newparallel = ProcGroupInfoP[groupid].newparallel; List tmpreldesclist = LispNil; int nfreeslave; SLAVE1_elog(DEBUG, "master woken up by paradjpending process group %d", groupid); ProcGroupInfoP[groupid].status = WORKING; tempRelationDescList = ProcGroupLocalInfoP[groupid].resultTmpRelDescList; prev = NULL; for (p = ProcGroupLocalInfoP[groupid].memberProc; p != NULL; p = nextp) { nextp = p->next; if (SlaveInfoP[p->pid].groupPid >= newparallel) { /* ------------------------ * before freeing this slave, we have to save its * resultTmpRelDesc somewhere. * ------------------------- */ #ifndef PALLOC_DEBUG tempRelationDesc = CopyRelDescUsing( SlaveInfoP[p->pid].resultTmpRelDesc, palloc); #else tempRelationDesc = CopyRelDescUsing( SlaveInfoP[p->pid].resultTmpRelDesc, palloc_debug); #endif tempRelationDescList = nappend1(tempRelationDescList, (LispValue)tempRelationDesc); if (prev == NULL) { ProcGroupLocalInfoP[groupid].memberProc = nextp; } else { prev->next = nextp; } freeSlave(p->pid); } } ProcGroupLocalInfoP[groupid].resultTmpRelDescList = tempRelationDescList; /* -------------------------- * now we have to adjust nprocess and countdown of group groupid * -------------------------- */ nfreeslave = ProcGroupInfoP[groupid].nprocess - ProcGroupInfoP[groupid].newparallel; ProcGroupInfoP[groupid].nprocess = ProcGroupInfoP[groupid].newparallel; ProcGroupInfoP[groupid].countdown -= nfreeslave; /* --------------------------- * adjust parallelism with the freed slaves * --------------------------- */ if (MasterSchedulingInfoD.ioBoundGroupId == groupid) AdjustParallelism(nfreeslave, MasterSchedulingInfoD.cpuBoundGroupId); else AdjustParallelism(nfreeslave, MasterSchedulingInfoD.ioBoundGroupId); if (ProcGroupInfoP[groupid].countdown == 0) { /* ----------------------- * this means that this group has actually finished * go down to process the group * ----------------------- */ } else { /* ------------------------ * otherwise, go to P_Finished() * ------------------------ */ goto MasterWait; } } SLAVE1_elog(DEBUG, "master woken up by finished process group %d", groupid); fragment = ProcGroupLocalInfoP[groupid].fragment; if (fragment == MasterSchedulingInfoD.ioBoundFrag) { MasterSchedulingInfoD.ioBoundFrag = NULL; MasterSchedulingInfoD.ioBoundGroupId = -1; } else if (fragment == MasterSchedulingInfoD.cpuBoundFrag) { MasterSchedulingInfoD.cpuBoundFrag = NULL; MasterSchedulingInfoD.cpuBoundGroupId = -1; } nparallel = get_frag_parallel(fragment); plan = get_frag_root(fragment); parentPlan = get_frag_parent_op(fragment); parentFragment = (Fragment)get_frag_parent_frag(fragment); /* --------------- * delete the finished fragment from the subtree list of its * parent fragment * --------------- */ if (parentFragment == NULL) subtrees = LispNil; else { subtrees = get_frag_subtrees(parentFragment); set_frag_subtrees(parentFragment, nLispRemove(subtrees, (LispValue)fragment)); } /* ---------------- * let the parent fragment know where the result is * ---------------- */ if (ExecIsHash(plan)) { /* ---------------- * if it is hashjoin, let the parent know where the hash table is * ---------------- */ set_hashjointable((HashJoin)parentPlan, hashtable); set_hashjointablekey((HashJoin)parentPlan, hashTableMemoryKey); set_hashjointablesize((HashJoin)parentPlan, hashTableMemorySize); set_hashdone((HashJoin)parentPlan, true); } else { List unionplans = LispNil; /* ------------------ * if it is ScanTemps node, clean up the temporary relations * they are not needed any more * ------------------ */ if (ExecIsScanTemps(plan)) { Relation tempreldesc; List tempRelDescs; LispValue y; tempRelDescs = get_temprelDescs((ScanTemps)plan); foreach (y, tempRelDescs) { tempreldesc = (Relation)CAR(y); ReleaseTmpRelBuffers(tempreldesc); if (FileNameUnlink( relpath((char*)&(tempreldesc->rd_rel->relname))) < 0) elog(WARN, "ExecEndScanTemp: unlink: %m"); } } if (parentPlan == NULL /* WWW && nparallel == 1 */) /* in this case the whole plan has been finished */ fraglist = nLispRemove(fraglist, (LispValue)fragment); else { /* ----------------- * make a ScanTemps node to let the parent collect the tuples * from a set of temporary relations * ----------------- */ tempRelationDescList = ProcGroupLocalInfoP[groupid].resultTmpRelDescList; p = ProcGroupLocalInfoP[groupid].memberProc; for (p = ProcGroupLocalInfoP[groupid].memberProc; p != NULL; p = p->next) { shmTempRelationDesc = SlaveInfoP[p->pid].resultTmpRelDesc; #ifndef PALLOC_DEBUG tempRelationDesc = CopyRelDescUsing(shmTempRelationDesc, palloc); #else tempRelationDesc = CopyRelDescUsing(shmTempRelationDesc, palloc_debug); #endif PALLOC_DEBUG tempRelationDescList = nappend1(tempRelationDescList, (LispValue)tempRelationDesc); } scantempNode = RMakeScanTemps(); set_qptargetlist((Plan)scantempNode, get_qptargetlist(plan)); set_temprelDescs(scantempNode, tempRelationDescList); if (parentPlan == NULL) { set_frag_root(fragment, (Plan)scantempNode); set_frag_subtrees(fragment, LispNil); set_fragment((Plan)scantempNode,-1); /*means end of parallelism */ set_frag_is_inprocess(fragment, false); set_frag_iorate(fragment, 0.0); } else { if (plan == (Plan)get_lefttree(parentPlan)) { set_lefttree(parentPlan, (PlanPtr)scantempNode); } else { set_righttree(parentPlan, (PlanPtr)scantempNode); } set_fragment((Plan)scantempNode, get_fragment(parentPlan)); } } } /* ----------------- * free shared memory * free the finished processed group * ----------------- */ ProcGroupSMClean(groupid); freeProcGroup(groupid); } } @ 1.33 log @compute target list correctly @ text @d14 1 a14 1 * $Header: /private/mer/pg/src/tcop/RCS/pfrag.c,v 1.32 1992/06/28 03:48:22 mao Exp mer $ d45 1 a45 1 RcsId("$Header: /private/mer/pg/src/tcop/RCS/pfrag.c,v 1.32 1992/06/28 03:48:22 mao Exp mer $"); d420 1 a420 1 if (strcmp(&opname, "=") == 0) @ 1.32 log @rearrange parse, plan to support postquel function invocations @ text @d14 1 a14 1 * $Header: /private/mao/postgres/src/tcop/RCS/pfrag.c,v 1.31 1992/06/18 06:04:44 hong Exp mao $ d45 1 a45 1 RcsId("$Header: /private/mao/postgres/src/tcop/RCS/pfrag.c,v 1.31 1992/06/18 06:04:44 hong Exp mao $"); d925 1 a925 1 natts = length(targetList); @ 1.31 log @take care of a compiler warning @ text @d14 1 a14 1 * $Header: /mnt/hong/postgres/src/tcop/RCS/pfrag.c,v 1.30 1992/06/16 21:29:18 hong Exp hong $ d45 1 a45 1 RcsId("$Header: /mnt/hong/postgres/src/tcop/RCS/pfrag.c,v 1.30 1992/06/16 21:29:18 hong Exp hong $"); d1025 4 a1028 1 fragQueryDesc = CreateQueryDesc(parsetree, plan, dest); @ 1.30 log @to get rid of a compiler complaint @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.29 92/03/31 23:12:23 mer Exp $ d45 1 a45 1 RcsId("$Header: RCS/pfrag.c,v 1.29 92/03/31 23:12:23 mer Exp $"); a127 2 default: return DEFPERTUPTIME; d129 1 a129 1 return 0.0; @ 1.29 log @change accessor functions into macros @ text @d14 1 a14 1 * $Header: /users/mer/pg/src/tcop/RCS/pfrag.c,v 1.28 1992/02/07 10:50:35 hong Exp mer $ d45 1 a45 1 RcsId("$Header: /users/mer/pg/src/tcop/RCS/pfrag.c,v 1.28 1992/02/07 10:50:35 hong Exp mer $"); d131 1 @ 1.28 log @fixed prototyping warnings @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.27 92/01/21 16:43:58 hong Exp $ d45 1 a45 1 RcsId("$Header: RCS/pfrag.c,v 1.27 92/01/21 16:43:58 hong Exp $"); d79 1 a79 1 set_frag_root(fragment, get_lefttree(node)); d98 1 a98 1 set_frag_root(fragment, get_righttree(node)); d229 1 a229 1 set_frag_parent_frag(frag, fragment); d993 1 a993 1 parentFragment = get_frag_parent_frag(fragment); d1196 1 a1196 1 parentFragment = get_frag_parent_frag(fragment); d1282 2 a1283 2 if (plan == get_lefttree(parentPlan)) { set_lefttree(parentPlan, (Plan)scantempNode); d1286 1 a1286 1 set_righttree(parentPlan, (Plan)scantempNode); @ 1.27 log @changes for inter-fragment parallelism @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.26 91/11/17 21:06:36 mer Exp Locker: hong $ d45 1 a45 1 RcsId("$Header: RCS/pfrag.c,v 1.26 91/11/17 21:06:36 mer Exp Locker: hong $"); a130 1 return 0.0; d446 1 a446 1 return lispCons(frag, LispNil); d449 1 a449 1 return(nconc(lispCons(frag, LispNil), ioboundlist)); d452 1 a452 1 return(nconc(lispCons(f, LispNil), d473 1 a473 1 return lispCons(frag, LispNil); d476 1 a476 1 return(nconc(lispCons(frag, LispNil), cpuboundlist)); d479 1 a479 1 return(nconc(lispCons(f, LispNil), d513 2 a514 1 *unparallelizablelist = nappend1(*unparallelizablelist, f); d517 1 a517 1 *presetlist = nappend1(*presetlist, f); d735 1 a735 1 fireFragmentList = lispCons(f, LispNil); d766 1 a766 1 SetParallelDegree((y=lispCons(f, LispNil)), parallel); d812 1 a812 1 fireFragmentList = lispCons(f2, LispNil); d822 1 a822 1 fireFragmentList = lispCons(f2, LispNil); d840 1 a840 1 fireFragmentList = lispCons(f1, LispNil); d850 1 a850 1 fireFragmentList = lispCons(f1, LispNil); d861 1 a861 1 SetParallelDegree((y=lispCons(f1, LispNil)), x1); d863 1 a863 1 SetParallelDegree((y=lispCons(f2, LispNil)), x2); d875 1 a875 1 fireFragmentList = lispCons(f2, LispNil); d884 1 a884 1 fireFragmentList = lispCons(f1, LispNil); @ 1.26 log @prototyping @ text @d14 1 a14 1 * $Header: /users/mer/postgres/src/tcop/RCS/pfrag.c,v 1.25 1991/11/11 22:59:33 hong Exp $ d19 1 d45 1 a45 1 RcsId("$Header: /users/mer/postgres/src/tcop/RCS/pfrag.c,v 1.25 1991/11/11 22:59:33 hong Exp $"); d48 3 d85 1 d104 1 d115 72 d217 1 d236 1 d395 7 a401 1 bool d431 6 a436 8 /* ---------------------------------------------------------------- * ParallelOptimize * * this analyzes the plan in the query descriptor and determines * which fragments to execute based on available virtual * memory resources... * * ---------------------------------------------------------------- d439 3 a441 2 ParallelOptimize(fragmentlist) List fragmentlist; d444 1 a444 10 Fragment fragment; int memAvail; float loadAvg; List fireFragments; List fireFragmentList; int nfreeslaves; List parsetree; LispValue k; int parallel; Plan plan; d446 69 a514 8 fireFragmentList = LispNil; nfreeslaves = NumberOfFreeSlaves; foreach (x, fragmentlist) { fragment = (Fragment)CAR(x); plan = get_frag_root(fragment); if (!plan_is_parallelizable(plan)) { parallel = 1; elog(NOTICE, "nonparallelizable fragment, running sequentially\n"); d516 3 d520 3 a522 7 parsetree = get_frag_parsetree(fragment); k = parse_parallel(parsetree); if (lispNullp(k)) { if (lispNullp(CDR(x))) parallel = nfreeslaves; else parallel = 1; d525 1 a525 3 parallel = CInteger(k); if (parallel > nfreeslaves || parallel == 0) parallel = nfreeslaves; a527 7 memAvail = GetCurrentMemSize(); loadAvg = GetCurrentLoadAverage(); fireFragments = ChooseFragments(fragment, memAvail); SetParallelDegree(fireFragments, parallel); nfreeslaves -= parallel; if (parallel > 0) fireFragmentList = nconc(fireFragmentList, fireFragments); a528 1 return fireFragmentList; d531 16 a546 2 #define MINHASHTABLEMEMORYKEY 1000 static IpcMemoryKey nextHashTableMemoryKey = 0; d548 10 a557 2 /* ------------------------- * getNextHashTableMemoryKey d559 3 a561 2 * get the next hash table key * ------------------------- d563 3 a565 2 static IpcMemoryKey getNextHashTableMemoryKey() d567 11 a577 2 extern int MasterPid; return (nextHashTableMemoryKey++ + MINHASHTABLEMEMORYKEY + MasterPid); d580 2 a581 2 /* ------------------------ * sizeofTmpRelDesc d583 4 a586 2 * calculate the size of reldesc of the temporary relation of plan * ------------------------ d589 4 a592 2 sizeofTmpRelDesc(plan) Plan plan; d594 2 a595 3 List targetList; int natts; int size; d597 8 a604 13 targetList = get_qptargetlist(plan); natts = length(targetList); /* ------------------ * see CopyRelDescUsing() in lib/C/copyfuncs.c if you want to know * how size if derived. * ------------------ */ size = sizeof(RelationData) + (natts - 1) * sizeof(TupleDescriptorData) + sizeof(RuleLock) + sizeof(RelationTupleFormD) + sizeof(LockInfoData) + natts * (sizeof(AttributeTupleFormData) + sizeof(RuleLock)) + 48; /* some extra for possible LONGALIGN() */ return size; d615 1 a615 1 AdjustParallelism(pardelta, notgroupid) d617 1 a617 1 int notgroupid; /* do not adjust this proc group */ d619 1 a619 1 int i, j; d627 3 a629 19 for (i=0; i= 0) break; }; if (i == GetNumberSlaveBackends()) { SLAVE_elog(DEBUG, "master finds no fragment to adjust parallelism to"); return; } SLAVE1_elog(DEBUG, "master sending signal to process group %d", i); signalProcGroup(i, SIGPARADJ); max_curpage = getProcGroupMaxPage(i); d631 1 a631 1 oldnproc = ProcGroupInfoP[i].nprocess; d639 2 a640 2 ProcGroupInfoP[i].paradjpage = NOPARADJ; OneSignalM(&(ProcGroupInfoP[i].m1lock), oldnproc); d643 1 a643 1 ProcGroupInfoP[i].paradjpage = max_curpage + 1; d646 4 a649 3 ProcGroupInfoP[i].nprocess += pardelta; ProcGroupInfoP[i].scounter.count = ProcGroupInfoP[i].nprocess; ProcGroupInfoP[i].newparallel = ProcGroupInfoP[i].nprocess; d652 8 a659 7 ProcGroupInfoP[i].paradjpage,ProcGroupInfoP[i].newparallel); OneSignalM(&(ProcGroupInfoP[i].m1lock), oldnproc); set_frag_parallel(ProcGroupLocalInfoP[i].fragment, get_frag_parallel(ProcGroupLocalInfoP[i].fragment)+ pardelta); ProcGroupSMBeginAlloc(i); size = sizeofTmpRelDesc(QdGetPlan(ProcGroupInfoP[i].queryDesc)); d667 1 a667 1 slave, i); d670 1 a670 1 addSlaveToProcGroup(slave, i, oldnproc+j); d676 2 a677 1 ProcGroupInfoP[i].newparallel = ProcGroupInfoP[i].nprocess + pardelta; d680 4 a683 3 ProcGroupInfoP[i].paradjpage,ProcGroupInfoP[i].newparallel); ProcGroupInfoP[i].dropoutcounter.count = -pardelta; OneSignalM(&(ProcGroupInfoP[i].m1lock), oldnproc); d688 253 d996 1 d1016 1 d1019 1 d1033 4 a1078 2 * ------------ */ d1082 2 d1113 1 d1161 6 a1166 1 AdjustParallelism(nfreeslave, groupid); d1185 8 d1244 1 a1244 1 if (parentPlan == NULL && nparallel == 1) d1279 1 @ 1.25 log @for prototyping @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.24 91/11/09 09:51:00 hong Exp $ d44 1 a44 1 RcsId("$Header: RCS/pfrag.c,v 1.24 91/11/09 09:51:00 hong Exp $"); d81 1 a81 1 subFragments = nappend1(subFragments, fragment); d99 1 a99 1 subFragments = nappend1(subFragments, fragment); d140 1 a140 1 fragmentlist = lispCons(rootFragment, LispNil); d204 1 a204 1 return lispCons(fragments, LispNil); d730 1 a730 1 tempRelationDesc); d787 2 a788 1 set_frag_subtrees(parentFragment, nLispRemove(subtrees, fragment)); d827 1 a827 1 fraglist = nLispRemove(fraglist, fragment); d849 1 a849 1 tempRelationDesc); @ 1.24 log @added mechanism for reducing parallelism of a fragment @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.23 91/10/09 14:53:29 hong Exp $ d18 2 a19 1 #include "tcop/tcop.h" a22 1 #include "nodes/execnodes.h" d25 1 a26 1 #include "nodes/plannodes.a.h" d31 2 d37 6 d44 1 a44 1 RcsId("$Header: RCS/pfrag.c,v 1.23 91/10/09 14:53:29 hong Exp $"); a45 4 extern ScanTemps RMakeScanTemps(); extern Fragment RMakeFragment(); extern Relation CopyRelDescUsing(); d57 1 a57 1 List d168 1 a168 1 int d180 1 a180 1 float d193 1 a193 1 List d219 1 a219 1 List d234 1 a234 1 List d249 1 a249 1 List d264 1 a264 1 List d291 1 a291 1 void d330 1 a330 1 indxqual = get_indxqual(plan); d354 1 a354 1 List d414 1 a414 1 IpcMemoryKey d457 1 a457 1 void d530 1 a530 1 ProcGroupSMEndAlloc(i); d605 1 a605 1 set_hashtablekey(plan, hashTableMemoryKey); d607 2 a608 2 set_hashtable(plan, hashtable); hashTableMemorySize = get_hashtablesize(plan); d636 1 a636 1 CopyObjectUsing(fragQueryDesc, ProcGroupSMAlloc); d798 4 a801 4 set_hashjointable(parentPlan, hashtable); set_hashjointablekey(parentPlan, hashTableMemoryKey); set_hashjointablesize(parentPlan, hashTableMemorySize); set_hashdone(parentPlan, true); d815 1 a815 1 tempRelDescs = get_temprelDescs(plan); d820 1 a820 1 relpath(&(tempreldesc->rd_rel->relname))) < 0) d851 1 a851 1 set_qptargetlist(scantempNode, get_qptargetlist(plan)); d854 1 a854 1 set_frag_root(fragment, scantempNode); d856 2 a857 1 set_fragment(scantempNode,-1);/*means end of parallelism */ d861 2 a862 2 if (plan == (Plan)get_lefttree(parentPlan)) { set_lefttree(parentPlan, scantempNode); d865 1 a865 1 set_righttree(parentPlan, scantempNode); d867 1 a867 1 set_fragment(scantempNode, get_fragment(parentPlan)); @ 1.23 log @fixed a bug in parallelism adjustment @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.22 91/10/08 01:13:48 hong Exp $ d36 1 a36 1 RcsId("$Header: RCS/pfrag.c,v 1.22 91/10/08 01:13:48 hong Exp $"); d454 3 a456 1 AdjustParallelism() a462 2 int nfree; extern MasterCommunicationData *MasterDataP; d464 1 d473 2 a474 1 if (ProcGroupInfoP[i].status == WORKING && d494 2 a495 2 MasterDataP->data[0] = NOPARADJ; OneSignalM(&(MasterDataP->m1lock), oldnproc); d498 29 a526 19 MasterDataP->data[0] = max_curpage + 1; /* page on which to adjust par. */ ProcGroupInfoP[i].nprocess += NumberOfFreeSlaves; ProcGroupInfoP[i].countdown = ProcGroupInfoP[i].nprocess; MasterDataP->data[1] = ProcGroupInfoP[i].nprocess; SLAVE2_elog(DEBUG,"master signals waiting slaves with adjpage=%d,newpar=%d", MasterDataP->data[0], MasterDataP->data[1]); OneSignalM(&(MasterDataP->m1lock), oldnproc); set_frag_parallel(ProcGroupLocalInfoP[i].fragment, get_frag_parallel(ProcGroupLocalInfoP[i].fragment)+ NumberOfFreeSlaves); ProcGroupSMBeginAlloc(i); size = sizeofTmpRelDesc(QdGetPlan(ProcGroupInfoP[i].queryDesc)); nfree = NumberOfFreeSlaves; for (j=0; jm1lock), m); d495 4 a498 1 for (i=0; i 0 && AdjustParallelismEnabled) AdjustParallelism(); d577 1 a577 1 else if (parentFragment != NULL || nparallel > 1) { d613 14 d636 10 d658 1 @ 1.20 log @fixed a stupid bug @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.19 91/08/08 15:39:11 hong Exp Locker: hong $ d34 1 d36 1 a36 1 RcsId("$Header: RCS/pfrag.c,v 1.19 91/08/08 15:39:11 hong Exp Locker: hong $"); d42 2 d65 1 a65 1 if (ExecIsHash(get_lefttree(node)) || ExecIsSort(get_lefttree(node))) { a280 29 /* ----------------------------- * set_plan_parallel * * set the degree of parallelism for each node in plan * ----------------------------- */ void set_plan_parallel(plan, nparallel) Plan plan; int nparallel; { if (plan == NULL) return; if (ExecIsNestLoop(plan)) { /* -------------------- * inner path of nestloop join plan should have parallelism 1 * -------------------- */ set_parallel(plan, nparallel); set_plan_parallel(get_outerPlan(plan), nparallel); set_plan_parallel(get_innerPlan(plan), 1); } else { set_parallel(plan, nparallel); set_plan_parallel(get_lefttree(plan), nparallel); set_plan_parallel(get_righttree(plan), nparallel); } } a303 1 set_plan_parallel(plan, 1); a306 1 set_plan_parallel(plan, nfreeslaves); d311 29 d364 1 d370 4 a373 7 parsetree = get_frag_parsetree(fragment); k = parse_parallel(parsetree); if (lispNullp(k)) { if (lispNullp(CDR(x))) parallel = nfreeslaves; else parallel = 1; d376 13 a388 3 parallel = CInteger(k); if (parallel > nfreeslaves || parallel == 0) parallel = nfreeslaves; d442 1 a442 1 12; /* some extra for possible LONGALIGN() */ d446 44 d528 1 d538 9 d594 1 d599 1 a599 1 (Relation)ProcGroupSMAlloc(sizeofTmpRelDesc(plan)); d603 1 @ 1.19 log @more support for inter-fragment inter-query parallelism @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.18 91/08/02 19:00:46 hong Exp Locker: kemnitz $ d35 1 a35 1 RcsId("$Header: RCS/pfrag.c,v 1.18 91/08/02 19:00:46 hong Exp Locker: kemnitz $"); d643 1 @ 1.18 log @minor bug fixes and added some more comments @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.17 91/07/24 16:13:40 hong Exp Locker: hong $ d35 1 a35 1 RcsId("$Header: RCS/pfrag.c,v 1.17 91/07/24 16:13:40 hong Exp Locker: hong $"); d73 1 d91 1 d131 1 d195 2 a196 1 if (lispNullp(get_frag_subtrees(fragments))) d267 1 d359 4 d365 1 d368 13 d384 4 a387 4 SetParallelDegree(fireFragments, NumberOfFreeSlaves); fireFragmentList = fireFragments; break; fireFragmentList = nconc(fireFragmentList, fireFragments); d540 1 @ 1.17 log @changes to use the new shared memory manager for parallel executor @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.16 91/07/17 23:45:03 hong Exp Locker: hong $ d32 1 a32 1 #include "utils/lmgr.h" d35 1 a35 1 RcsId("$Header: RCS/pfrag.c,v 1.16 91/07/17 23:45:03 hong Exp Locker: hong $"); d41 9 d63 4 d76 1 a76 1 subFragments = FindFragments(get_lefttree(node), fragmentNo); d80 4 d93 1 a93 1 newFragments = FindFragments(get_righttree(node), fragmentNo); d100 8 d111 1 d152 6 d164 6 d176 7 d202 6 d216 7 d231 7 d247 6 d273 6 d287 4 d302 6 d362 2 d372 6 d385 6 @ 1.16 log @changes to support inter-fragment and inter-query parallelism in parallel executor, non-parallel processing remain the same as before @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.15 91/06/18 23:28:36 cimarron Exp $ d32 2 d35 1 a35 1 RcsId("$Header: RCS/pfrag.c,v 1.15 91/06/18 23:28:36 cimarron Exp $"); d290 23 d404 1 d406 8 a413 1 CopyObjectUsing(fragQueryDesc, ExecSMAlloc); d531 1 d535 1 a537 6 /* ---------------- * Clean Shared Memory used during the query * ---------------- */ ExecSMClean(); @ 1.15 log @reorganized executor to use tuple table properly for nested dot stuff @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.14 91/04/18 11:38:42 hong Exp Locker: cimarron $ d20 1 d24 1 d33 1 a33 1 RcsId("$Header: RCS/pfrag.c,v 1.14 91/04/18 11:38:42 hong Exp Locker: cimarron $"); a34 11 /* ---------------------------------------------------------------- * ExecuteFragments * * Read the comments for ProcessQuery() below... * * Since we do no parallism, planFragments is totally * ignored for now.. -cim 9/18/89 * ---------------------------------------------------------------- */ extern Pointer *SlaveQueryDescsP; extern Pointer *SlaveRetStateP; a38 204 Fragment ExecuteFragments(queryDesc, fragmentlist, rootFragment) List queryDesc; List fragmentlist; Fragment rootFragment; { int nparallel; int i; int nproc; int nfrag; LispValue x; Fragment fragment; Plan plan; Plan parentPlan; List parseTree; HashJoinTable hashtable; List subtrees; Fragment parentFragment; List fragQueryDesc; ScanTemps scantempNode; Relation tempRelationDesc; List tempRelationDescList; Relation shmTempRelationDesc; List finalResultRelation; int hashTableMemorySize; IpcMemoryKey hashTableMemoryKey; IpcMemoryKey getNextHashTableMemoryKey(); CommandDest dest; /* ---------------- * execute the query appropriately if we are running one or * several backend processes. * ---------------- */ if (! ParallelExecutorEnabled()) { /* ---------------- * single-backend case. just execute the query * and return NULL. * ---------------- */ ProcessQueryDesc(queryDesc); return NULL; } else { /* ---------------- * parallel-backend case. place each plan fragment * in shared memory and signal slave-backend execution. * When slave execution is completed, form a new plan * representing the query with some of the work materialized * and return this to ProcessQuery(). * * ---------------- */ nproc = 0; nfrag = 0; foreach (x, fragmentlist) { fragment = (Fragment)CAR(x); nparallel = get_frag_parallel(fragment); plan = get_frag_root(fragment); parseTree = QdGetParseTree(queryDesc); dest = QdGetDest(queryDesc); finalResultRelation = parse_tree_result_relation(parseTree); if (ExecIsHash(plan)) { hashTableMemoryKey = getNextHashTableMemoryKey(); set_hashtablekey(plan, hashTableMemoryKey); hashtable = ExecHashTableCreate(plan); set_hashtable(plan, hashtable); hashTableMemorySize = get_hashtablesize(plan); parse_tree_result_relation(parseTree) = LispNil; } else if (fragment != rootFragment || nparallel > 1) { parse_tree_result_relation(parseTree) = lispCons(lispAtom("intotemp"), LispNil); dest = None; } fragQueryDesc = CreateQueryDesc(parseTree, plan, dest); /* ---------------- * place fragments in shared memory here. * each fragment only put one copy of the its * querydesc in the shared memory. each slave process * that executes this fragment will have to * make a local copy of the querydesc to work on. * ---------------- */ SlaveQueryDescsP[nfrag] = (Pointer) CopyObjectUsing(fragQueryDesc, ExecSMAlloc); nproc += nparallel; nfrag++; } parse_tree_result_relation(parseTree) = finalResultRelation; /* ---------------- * signal slave execution start * ---------------- */ for (i=1; ird_rel->relname))) < 0) elog(WARN, "ExecEndScanTemp: unlink: %m"); } } if (parentPlan == NULL && nparallel == 1) rootFragment = NULL; else { tempRelationDescList = LispNil; for (i=0; i 1) { parse_tree_result_relation(parseTree) = lispCons(lispAtom("intotemp"), LispNil); } fragQueryDesc = CreateQueryDesc(parseTree, plan); d125 1 a125 10 /* ---------------- * place fragments in shared memory here. * ---------------- */ for (i=0; i 1) { parse_tree_result_relation(parseTree) = lispCons(lispAtom("intotemp"), LispNil); } fragQueryDesc = CreateQueryDesc(parseTree, plan); d106 27 a132 10 /* ---------------- * place fragments in shared memory here. * ---------------- */ for (i=0; ird_rel->relname))) < 0) @ 1.4 log @turned off debugging messages @ text @d14 1 a14 1 * $Header: RCS/pfrag.c,v 1.3 90/09/06 14:44:04 hong Exp $ d29 1 a29 1 RcsId("$Header: RCS/pfrag.c,v 1.3 90/09/06 14:44:04 hong Exp $"); d128 1 a128 1 for (i=0; i