head	1.24;
access;
symbols
	Version_2_1:1.6
	Version_2:1.3;
locks; strict;
comment	@ * @;


1.24
date	92.07.06.05.03.08;	author mao;	state Exp;
branches;
next	1.23;

1.23
date	92.03.31.23.12.15;	author mer;	state Exp;
branches;
next	1.22;

1.22
date	92.01.21.16.44.25;	author hong;	state Exp;
branches;
next	1.21;

1.21
date	91.11.11.23.00.24;	author hong;	state Exp;
branches;
next	1.20;

1.20
date	91.11.09.09.50.34;	author hong;	state Exp;
branches;
next	1.19;

1.19
date	91.11.02.22.02.26;	author hong;	state Exp;
branches;
next	1.18;

1.18
date	91.10.25.16.22.05;	author hong;	state Exp;
branches;
next	1.17;

1.17
date	91.10.09.14.52.07;	author hong;	state Exp;
branches;
next	1.16;

1.16
date	91.10.08.01.15.07;	author hong;	state Exp;
branches;
next	1.15;

1.15
date	91.08.27.14.19.11;	author hong;	state Exp;
branches;
next	1.14;

1.14
date	91.08.25.13.35.20;	author hong;	state Exp;
branches;
next	1.13;

1.13
date	91.07.31.12.37.57;	author hong;	state Exp;
branches;
next	1.12;

1.12
date	91.07.24.16.14.49;	author hong;	state Exp;
branches;
next	1.11;

1.11
date	91.07.19.15.12.21;	author hong;	state Exp;
branches;
next	1.10;

1.10
date	91.07.17.23.37.44;	author hong;	state Exp;
branches;
next	1.9;

1.9
date	91.05.31.10.55.02;	author hong;	state Exp;
branches;
next	1.8;

1.8
date	91.04.18.11.39.44;	author hong;	state Exp;
branches;
next	1.7;

1.7
date	91.04.04.13.51.16;	author hong;	state Exp;
branches;
next	1.6;

1.6
date	90.09.10.20.45.02;	author hong;	state Exp;
branches;
next	1.5;

1.5
date	90.08.18.14.16.20;	author cimarron;	state Exp;
branches;
next	1.4;

1.4
date	90.08.14.16.20.21;	author cimarron;	state Exp;
branches;
next	1.3;

1.3
date	90.05.25.10.36.35;	author cimarron;	state Version_2;
branches;
next	1.2;

1.2
date	90.03.31.19.01.51;	author cimarron;	state Exp;
branches;
next	1.1;

1.1
date	90.03.12.16.27.18;	author cimarron;	state Exp;
branches;
next	;


desc
@Integration Checkin
@


1.24
log
@support for 'returntype = setof typename' in function definition
@
text
@/* ----------------------------------------------------------------
 *   FILE
 *	slaves.c
 *	
 *   DESCRIPTION
 *	slave backend management routines
 *
 *   INTERFACE ROUTINES
 *	SlaveMain()
 *	SlaveBackendsInit()
 *	SlaveBackendsAbort()
 *
 *   NOTES
 *
 *   IDENTIFICATION
 *	$Header: /private/mao/postgres/src/tcop/RCS/slaves.c,v 1.23 1992/03/31 23:12:15 mer Exp $
 * ----------------------------------------------------------------
 */
#include <signal.h>
#include <setjmp.h>

#include "tmp/postgres.h"

 RcsId("$Header: /private/mao/postgres/src/tcop/RCS/slaves.c,v 1.23 1992/03/31 23:12:15 mer Exp $");

/* ----------------
 *	FILE INCLUDE ORDER GUIDELINES
 *
 *	1) tcopdebug.h
 *	2) various support files ("everything else")
 *	3) node files
 *	4) catalog/ files
 *	5) execdefs.h and execmisc.h, if necessary.
 *	6) extern files come last.
 * ----------------
 */
#include "tcop/tcopdebug.h"
#include "nodes/plannodes.h"
#include "nodes/plannodes.a.h"
#include "nodes/execnodes.h"
#include "executor/execdesc.h"
#include "tcop/dest.h"
#include "tcop/pquery.h"
#include "tcop/slaves.h"

#include "access/xact.h"
#include "utils/log.h"
#include "catalog/syscache.h"

/* ----------------
 *	parallel state variables
 * ----------------
 */
/*
 *	local data structures
 */
extern int MyPid; /* int representing the process id, defined in execipc.c */
static ProcessNode *SlaveArray, *FreeSlaveP;
int NumberOfFreeSlaves;
ProcGroupLocalInfo	ProcGroupLocalInfoP; /* process group local info */
static ProcGroupLocalInfo FreeProcGroupP;
extern SlaveLocalInfoData SlaveLocalInfoD;  /* defined in execipc.c */
extern int AdjustParallelismEnabled;
FILE *StatFp;
static bool RestartForParAdj = false;  /* indicating that the longjmp to
					  SlaveRestartPoint is for paradj */
static List QueryDesc;

/*
 *	shared data structures
 */
int 	*MasterProcessIdP;	/* master backend process id */
int	*SlaveAbortFlagP;	/* flag set during a transaction abort */
extern SlaveInfo	SlaveInfoP;	/* slave backend info */
extern ProcGroupInfo  ProcGroupInfoP; /* process group info */
                                      /* defined in execipc.c to make
                                         postmaster happy */

TransactionState SharedTransactionState; /* current transaction info */

#define CONDITION_NORMAL	0
#define CONDITION_ABORT		1

/* --------------------------------
 *	SendAbortSignals
 *
 *	This sends a SIGHUP to every other backend in order
 *	to cause them to preform their abort processing when
 *	we discover a reason to abort the current transaction.
 * --------------------------------
 */
void
SendAbortSignals()    
{
    int nslaves;		/* number of slaves */
    int i;			/* counter */
    int p;			/* process id */

#ifdef TCOP_SLAVESYNCDEBUG    
    if (IsMaster)
	elog(DEBUG, "Master Backend sending abort signals");
    else
	elog(DEBUG, "Slave Backend %d sending abort signals", MyPid);
#endif TCOP_SLAVESYNCDEBUG    

    nslaves = GetNumberSlaveBackends();
    for (i=0; i<nslaves; i++)
	if (i != MyPid) {
	    p = SlaveInfoP[i].unixPid;
	    if (kill(p, SIGHUP) != 0) {
		fprintf(stderr, "signaling slave %d (pid %d): ", i, p);
		perror("kill");
	    }
	}
    
    if (! IsMaster) {
	p = (*MasterProcessIdP);
	if (kill(p, SIGHUP) != 0) {
	    fprintf(stderr, "signaling master (pid %d): ", p);
	    perror("kill");
	}
    }
}

/* --------------------------------
 *	SlaveRestart
 *
 *	This is the signal / exception handler for the slave
 *	backends.  It causes processing to resume at the top
 *	of SlaveMain(), right before SlaveBackendsAbort().
 * --------------------------------
 */
jmp_buf	SlaveRestartPoint;
int	SlaveWarnings = 0;

void
SlaveRestart()
{
    SLAVE1_elog(DEBUG, "Slave Backend %d *** SlaveRestart ***", MyPid);

    longjmp(SlaveRestartPoint, 1);
}

/* --------------------------------
 *	SlaveBackendsAbort
 *
 *	this is called in each process when trouble arises.
 * --------------------------------
 */
void
SlaveBackendsAbort()
{
    int nslaves;		/* number of slaves */
    int i;			/* counter */
    
    /* ----------------
     *	if the abort flag is not set, then it means the problem
     *  occurred within our process.  So set the flag and tell
     *  all the other backends to abort.
     *
     *  only the aborting process does this:
     * ----------------
     */
    if ((*SlaveAbortFlagP) != CONDITION_ABORT) {
	(*SlaveAbortFlagP) = CONDITION_ABORT;
	
	SendAbortSignals();
    }

    /* ----------------
     *	all processes do this:
     * ----------------
     */
    if (IsMaster) {
	/* ----------------
	 *  in the master:
	 *
	 *	+ we record the aborted transaction,
	 *  	+ reinitialize the synchronization semaphores
	 *  	+ reinitialize the executor shared memory
	 *  	+ signal the slave backends to cleanup.
	 *
	 *  We then wait for all the slaves to finish doing their
	 *  cleanup and then clear the abort flag.
	 * ----------------
	 */
	SLAVE_elog(DEBUG, "Master Backend aborting current transaction");
	AbortCurrentTransaction();

	SLAVE_elog(DEBUG, "Master Backend reinitializing semaphores");
	nslaves = GetNumberSlaveBackends();
	for (i=0; i<nslaves; i++) {
	    I_Start(i);
	}
	I_Finished();
	
	SLAVE_elog(DEBUG, "Master Backend reinitializing shared memory");
	ExecSMInit();
	
	SLAVE_elog(DEBUG, "Master Backend signaling slaves abort");
	V_Abort();

	SLAVE_elog(DEBUG, "Master Backend waiting for slave aborts");
	P_FinishedAbort();

	SLAVE_elog(DEBUG, "Master Backend reinitializing abort semaphore");
	I_Abort();

	SLAVE_elog(DEBUG, "Master Backend abort processing finished");
	(*SlaveAbortFlagP) = CONDITION_NORMAL;

    } else {
	/* ----------------
	 *  the slave backends preform their cleanup processing
	 *  after the master backend records the abort.  This is
	 *  guaranteed because the slaves P on the semaphore which
	 *  isn't V'ed by the master until after its done aborting.
	 * ----------------
	 */
	SLAVE1_elog(DEBUG, "Slave Backend %d waiting to abort..",
		    MyPid);
	P_Abort();

	SLAVE1_elog(DEBUG, "Slave Backend %d processing abort..",
		    MyPid);
	SlaveAbortTransaction();

	SLAVE1_elog(DEBUG, "Slave Backend %d abort finished, signaling..",
		    MyPid);
	
	V_Abort();
	V_FinishedAbort();
    }
}

/* --------------------------------
 *	SlaveMain is the main loop executed by the slave backends
 * --------------------------------
 */
void
SlaveMain()
{
    int i;
    int groupid;
    extern int ShowExecutorStats;

    /* -----------------
     * set the flag to false in case the SIGPARADJ is received
     * -----------------
     */
    SlaveLocalInfoD.isworking = false;
    /* ------------------
     * set stat file pointer if required
     * ------------------
     */
    if (ShowExecutorStats) {
	char fname[30];
	sprintf(fname, "/usr/tmp/hong/slave%d.stat", MyPid);
        StatFp = fopen(fname, "w");
      }
    /* ----------------
     *  before we begin processing we have to register a SIGHUP
     *  handler so that if a problem occurs in one slave backend,
     *  all the backends resynchronize.
     * ----------------
     */
    if (setjmp(SlaveRestartPoint) != 0) {
	if (RestartForParAdj) {
	    /* restart for parallelism adjustment */
	    RestartForParAdj = false;
	  }
	else {
	    /* restart for transaction abort */
	    SlaveWarnings++;
	    SLAVE1_elog(DEBUG, "Slave Backend %d SlaveBackendsAbort()",
		        MyPid);
	
	    SlaveBackendsAbort();
	    SLAVE1_elog(DEBUG, "Slave Backend %d SlaveBackendsAbort() done",
		        MyPid);
	  }
      }	

    signal(SIGHUP, SlaveRestart);

    /* ----------------
     *	POSTGRES slave processing loop begins here
     * ----------------
     */
    SLAVE1_elog(DEBUG, "Slave Backend %d entering slave loop...", MyPid);
    
    for(;;) {
	/* ----------------
	 *  setup caches, lock tables and memory stuff just
	 *  as if we were running within a transaction.
	 * ----------------
	 */
	SlaveStartTransaction();
	
	/* ----------------
	 *  The master V's on the slave start semaphores after
	 *  placing the plan fragment in shared memory.
	 *  Meanwhile each slave waits on its start semaphore
	 *  until signaled by the master.
	 * ----------------
	 */
	SLAVE1_elog(DEBUG, "Slave Backend %d waiting...", MyPid);
	
	P_Start(MyPid);
	
	SLAVE1_elog(DEBUG, "Slave Backend %d starting task.", MyPid);

	if (ShowExecutorStats)
	    ResetUsage();
	/* ------------------
	 * initialize slave local info
	 * ------------------
	 */
	groupid = SlaveInfoP[MyPid].groupId;
	SlaveLocalInfoD.startpage = SlaveInfoP[MyPid].groupPid +
				    (SlaveInfoP[MyPid].isAddOnSlave ? 
				     ProcGroupInfoP[groupid].paradjpage : 0);
	SlaveLocalInfoD.nparallel = ProcGroupInfoP[groupid].nprocess;
	SlaveLocalInfoD.paradjpending = false;
	SlaveLocalInfoD.paradjpage = -1;
	SlaveLocalInfoD.newparallel = -1;
	SlaveLocalInfoD.heapscandesc = NULL;
	SlaveLocalInfoD.indexscandesc = NULL;
	SlaveLocalInfoD.isworking = true;

	/* ----------------
	 *  get the query descriptor to execute.
	 * ----------------
	 */
	QueryDesc = (List)CopyObject((List)ProcGroupInfoP[groupid].queryDesc);

	/* ----------------
	 *  process the query descriptor
	 * ----------------
	 */
	if (QueryDesc != NULL)
	    ProcessQueryDesc(QueryDesc);

	/* ---------------
	 * it is important to set isDone to true before
	 * SlaveCommitTransaction(), because it may
	 * free the memory of SlaveInfoP[MyPid].heapscandesc
	 * ---------------
	 */
	SlaveInfoP[MyPid].isDone = true;

	if (ShowExecutorStats)
	    ShowUsage();
	/* ----------------
	 *  clean caches, lock tables and memory stuff just
	 *  as if we were running within a transaction.
	 * ----------------
	 */
	SlaveCommitTransaction();
	    
	/* ----------------
	 *  when the slave finishes, it signals the master
	 *  backend by V'ing on the master's finished semaphore.
	 *
	 *  Since the master started by placing nslaves locks
	 *  on the semaphore, the semaphore will be decremented
	 *  each time a slave finishes.  the last slave to finish
	 *  should bring the finished count to 1;
	 * ----------------
	 */
	SLAVE1_elog(DEBUG, "Slave Backend %d task complete.", MyPid);
	SlaveLocalInfoD.isworking = false;
	V_Finished(groupid, &(ProcGroupInfoP[groupid].scounter), FINISHED);
    }
}

/* --------------------------------
 *	MoveTransactionState copies the transaction system's
 *	CurrentTransactionState information into permanent shared
 *	memory and then sets the transaction system's state
 *	pointer to the shared memory copy.  This is necessary to
 *	have the transaction system use the shared state in each
 *	of the parallel backends.
 * --------------------------------
 */
void
MoveTransactionState()
{
    int nbytes = sizeof(TransactionStateData);
    extern TransactionState CurrentTransactionState;
	
    bcopy(CurrentTransactionState, SharedTransactionState, nbytes);
    CurrentTransactionState = SharedTransactionState;
}
	
/* --------------------------------
 *	SlaveBackendsInit
 *
 *	SlaveBackendsInit initializes the communication structures,
 *	forks several "worker" backends and returns.
 *
 *	Note: this function only returns in the master
 *	      backend.  The worker backends run forever in
 *	      a separate execution loop.
 * --------------------------------
 */

void
SlaveBackendsInit()
{
    int nslaves;		/* number of slaves */
    int i;			/* counter */
    int p;			/* process id returned by fork() */
    extern int ProcessAffinityOn; /* process affinity on flag */
    
    /* ----------------
     *  first initialize shared memory and get the number of
     *  slave backends.
     * ----------------
     */
    nslaves = GetNumberSlaveBackends();

    /* -----------------
     * the following calls to SearchSysCacheTuple() are total hacks
     * what they do is to pre-initialize all the caches so that
     * performance numbers will look better.  they can be removed at
     * any time.
     * -----------------
     */
    SearchSysCacheTuple(AMOPOPID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(AMOPSTRATEGY, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(ATTNAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(ATTNUM, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(INDEXRELID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(LANNAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(OPRNAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(OPROID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(PRONAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(PROOID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(RELNAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(RELOID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(TYPNAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(TYPOID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(AMNAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(CLANAME, "", NULL, NULL, NULL);
    SearchSysCacheTuple(INDRELIDKEY, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(INHRELID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(PRS2PLANCODE, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(RULOID, NULL, NULL, NULL, NULL);
    SearchSysCacheTuple(PRS2STUB, NULL, NULL, NULL, NULL);
    
    /* --------------------
     * set signal for dynamically adjusting degrees of parallelism
     * --------------------
     */
    if (AdjustParallelismEnabled)
        signal(SIGPARADJ, paradj_handler);

    /* ----------------
     *	initialize Start, Finished, and Abort semaphores
     * ----------------
     */
    for (i=0; i<nslaves; i++)
	I_Start(i);
    I_Finished();
    I_Abort();
    
    /* ----------------
     *	allocate area in shared memory for process ids and
     *  communication mechanisms.  All backends share these pointers.
     * ----------------
     */
    MasterProcessIdP = (int*)ExecSMReserve(sizeof(int));
    SlaveAbortFlagP = (int*)ExecSMReserve(sizeof(int));
    SlaveInfoP = (SlaveInfo)ExecSMReserve(nslaves * sizeof(SlaveInfoData));
    ProcGroupInfoP = (ProcGroupInfo)ExecSMReserve(nslaves *
						    sizeof(ProcGroupInfoData));
    SharedTransactionState = (TransactionState)ExecSMReserve(
						  sizeof(TransactionStateData));

    /* ----------------
     *	move the transaction system state data into shared memory
     *  before we fork so that all backends share the transaction state.
     * ----------------
     */
    MoveTransactionState();
    
    /* ----------------
     *	initialize executor shared memory
     * ----------------
     */
    ExecSMInit();
    /* ----------------
     *	initialize shared memory variables
     * ----------------
     */
    (*MasterProcessIdP) = getpid();
    (*SlaveAbortFlagP) = CONDITION_NORMAL;
    /* ----------------
     *	fork several slave processes and save the process id's
     * ----------------
     */
    MyPid = -1;
    
    for (i=0; i<nslaves; i++)
	if (IsMaster) {
	    if ((p = fork()) != 0) {
		if (p < 0) {
		    perror("fork");
		    exitpg(1);
		  }
		/* initialize shared data structures */
		SlaveInfoP[i].unixPid = p;
		SlaveInfoP[i].groupId = -1;
		SlaveInfoP[i].groupPid = -1;
		SlaveInfoP[i].resultTmpRelDesc = NULL;
#ifdef sequent
		S_INIT_LOCK(&(SlaveInfoP[i].lock));
		S_LOCK(&(SlaveInfoP[i].lock));
#endif
		ProcGroupInfoP[i].status = IDLE;
		ProcGroupInfoP[i].queryDesc = NULL;
		ProcGroupInfoP[i].scounter.count = 0;
		ProcGroupInfoP[i].dropoutcounter.count = 0;
#ifdef sequent
		S_INIT_LOCK(&(ProcGroupInfoP[i].scounter.exlock));
		S_INIT_LOCK(&(ProcGroupInfoP[i].dropoutcounter.exlock));
#endif
		ProcGroupInfoP[i].paradjpage = NULLPAGE;
		InitMWaitOneLock(&(ProcGroupInfoP[i].m1lock));
	    } else {
		MyPid = i;
#ifdef sequent
		if (ProcessAffinityOn && GetNumberSlaveBackends() <= 12) {
		    int prev;
		    /* ----------------------
		     * bind slave to a specific processor
		     * ----------------------
		     */
		    prev = tmp_affinity(MyPid);
		    if (prev == -1) {
			fprintf(stderr, "slave %d ", MyPid);
			perror("tmp_affinity");
		      }
		  }
#endif
		SLAVE1_elog(DEBUG, "Slave Backend %d alive!", MyPid);
	    }
	}
    
    /* ----------------
     *	now ExecIsMaster is true in the master backend and is false
     *  in each of the slaves.  In addition, each slave is branded
     *  with a slave id to identify it.  So, if we're a slave, we now
     *  get sent off to the labor camp, never to return..
     * ----------------
     */
    if (IsMaster) {
	/* initialize local data structures of the master */
	SlaveArray = (ProcessNode*)malloc(nslaves * sizeof(ProcessNode));
	FreeSlaveP = SlaveArray;
	NumberOfFreeSlaves = nslaves;
	for (i=0; i<nslaves; i++) {
	    SlaveArray[i].pid = i;
	    SlaveArray[i].next = SlaveArray + i + 1;
	  }
	SlaveArray[nslaves-1].next = NULL;
	ProcGroupLocalInfoP = (ProcGroupLocalInfo)malloc(nslaves *
						sizeof(ProcGroupLocalInfoData));
	FreeProcGroupP = ProcGroupLocalInfoP;
	for (i=0; i<nslaves; i++) {
	    ProcGroupLocalInfoP[i].id = i;
	    ProcGroupLocalInfoP[i].fragment = NULL;
	    ProcGroupLocalInfoP[i].memberProc = NULL;
	    ProcGroupLocalInfoP[i].nextfree = ProcGroupLocalInfoP + i + 1;
	    ProcGroupLocalInfoP[i].resultTmpRelDescList = LispNil;
	  }
	ProcGroupLocalInfoP[nslaves - 1].nextfree = NULL;
      }
    else {
	SlaveMain();
      }

    /* ----------------
     *	here we're in the master and we've completed initialization
     *  so we return to the read..parse..plan..execute loop.
     * ----------------
     */
    return;
}

/* ----------------------
 *	getFreeSlave
 *
 *	get a free slave backend from the FreeSlaveP queue
 *	also decrements NumberOfFreeSlaves
 * ----------------------
 */
int
getFreeSlave()
{
    ProcessNode *p;

    p = FreeSlaveP;
    FreeSlaveP = p->next;
    NumberOfFreeSlaves--;

    return p->pid;
}

/* ------------------------
 *	freeSlave
 *
 *	frees a slave to FreeSlaveP queue
 *	increments NumberOfFreeSlaves
 * ------------------------
 */
void
freeSlave(i)
int i;
{
    SlaveArray[i].next = FreeSlaveP;
    FreeSlaveP = SlaveArray + i;
    SlaveInfoP[i].groupId = -1;
    SlaveInfoP[i].groupPid = -1;
    NumberOfFreeSlaves++;
}

/* ------------------------
 *	getFreeProcGroup
 *
 *	get a free process group with nproc free slave processes
 * ------------------------
 */
int
getFreeProcGroup(nproc)
int nproc;
{
    ProcGroupLocalInfo p;
    ProcessNode *slavep;
    int i;
    int pid;

    p = FreeProcGroupP;
    FreeProcGroupP = p->nextfree;
    pid = getFreeSlave();
    SlaveInfoP[pid].groupId = p->id;
    SlaveInfoP[pid].groupPid = 0;
    SlaveInfoP[pid].isAddOnSlave = false;
    SlaveInfoP[pid].isDone = false;
    p->memberProc = SlaveArray + pid;
    slavep = p->memberProc;
    for (i=1; i<nproc; i++) {
	pid = getFreeSlave();
	SlaveInfoP[pid].groupId = p->id;
	SlaveInfoP[pid].groupPid = i;
        SlaveInfoP[pid].isAddOnSlave = false;
        SlaveInfoP[pid].isDone = false;
	slavep->next = SlaveArray + pid;
	slavep = slavep->next;
      }
    slavep->next = NULL;
    return p->id;
}

/* --------------------------
 *	addSlaveToProcGroup
 *
 *	add a free slave to an existing process group
 * --------------------------
 */
void
addSlaveToProcGroup(slave, group, groupid)
int slave;
int group;
int groupid;
{
    SlaveInfoP[slave].groupId = group;
    SlaveInfoP[slave].groupPid = groupid;
    SlaveInfoP[slave].isAddOnSlave = true;
    SlaveArray[slave].next = ProcGroupLocalInfoP[group].memberProc;
    ProcGroupLocalInfoP[group].memberProc = SlaveArray + slave;
}

/* -------------------------
 *	freeProcGroup
 *
 *	frees a process group and all the slaves in the group
 * -------------------------
 */
void
freeProcGroup(gid)
int gid;
{
    ProcessNode *p, *nextp;

    p=ProcGroupLocalInfoP[gid].memberProc;
    while (p != NULL) {
	nextp = p->next;
	freeSlave(p->pid);
	p = nextp;
      }
    ProcGroupInfoP[gid].status = IDLE;
    ProcGroupLocalInfoP[gid].fragment = NULL;
    ProcGroupLocalInfoP[gid].nextfree = FreeProcGroupP;
    ProcGroupLocalInfoP[gid].resultTmpRelDescList = LispNil;
    FreeProcGroupP = ProcGroupLocalInfoP + gid;
}

/* ----------------------------
 *	getFinishedProcGroup
 *
 *	walks the array of processes group and find the first
 *	process group with status = FINISHED or PARADJPENDING
 * -----------------------------
 */
int
getFinishedProcGroup()
{
    int i;

    for (i=0; i<GetNumberSlaveBackends(); i++) {
	if (ProcGroupInfoP[i].status == FINISHED ||
	    ProcGroupInfoP[i].status == PARADJPENDING)
	    return i;
      }
    return -1;
}

/* -------------------------------
 *	wakeupProcGroup
 *
 *	wake up the processes in process group
 * -------------------------------
 */
void
wakeupProcGroup(groupid)
int groupid;
{
    ProcessNode *p;

    for (p = ProcGroupLocalInfoP[groupid].memberProc;
	 p != NULL;
	 p = p->next) {
      V_Start(p->pid);
     }
}

/* ---------------------------------
 *	signalProcGroup
 *
 *	send a signal to a process group
 * ---------------------------------
 */
void
signalProcGroup(groupid, sig)
int groupid;
int sig;
{
    ProcessNode *p;

    for (p = ProcGroupLocalInfoP[groupid].memberProc;
	 p != NULL;
	 p = p->next) {
	kill(SlaveInfoP[p->pid].unixPid, sig);
      }
}

/* ------------------------------
 * the following routines are a specialized memory allocator for 
 * the process groups.  they only supposed to be called by the master
 * backend.  no mutex is done.
 * ------------------------------
 */
static char *CurrentSMSegmentStart;
static char *CurrentSMSegmentEnd;
static int CurrentSMGroupid;
static char *CurrentSMSegmentPointer;

/* --------------------------------
 *	ProcGroupSMBeginAlloc
 *
 *	begins shared memory allocation for process group
 * --------------------------------
 */
void
ProcGroupSMBeginAlloc(groupid)
int groupid;
{
    MemoryHeader mp;

    mp = ExecGetSMSegment();
    ProcGroupLocalInfoP[groupid].groupSMQueue = mp;
    mp->next = NULL;
    CurrentSMSegmentStart = mp->beginaddr;
    CurrentSMSegmentEnd =  CurrentSMSegmentStart + mp->size;
    CurrentSMGroupid = groupid;
    CurrentSMSegmentPointer = CurrentSMSegmentStart;
}

/* -------------------------------
 *	ProcGroupSMEndAlloc
 *
 *	ends shared memory allocation for process group
 *	frees the leftover memory from current memory segment
 * -------------------------------
 */
void
ProcGroupSMEndAlloc()
{
    int usedsize;
    MemoryHeader mp;

    usedsize = CurrentSMSegmentPointer - CurrentSMSegmentStart;
    mp = ProcGroupLocalInfoP[CurrentSMGroupid].groupSMQueue;
    CurrentSMSegmentStart=CurrentSMSegmentPointer=CurrentSMSegmentEnd = NULL;
    ExecSMSegmentFreeUnused(mp, usedsize);
}

/* --------------------------------
 *	ProcGroupSMAlloc
 *
 *	allocate shared memory within a process group
 *	if the current memory segment runs out, allocate a new segment
 * ---------------------------------
 */
char *
ProcGroupSMAlloc(size)
int size;
{
    MemoryHeader mp;
    char *retP;

    while (CurrentSMSegmentPointer + size > CurrentSMSegmentEnd) {
	mp = ExecGetSMSegment();
	if (mp == NULL)
	    elog(WARN, "out of executor shared memory, got to die.");
	mp->next = ProcGroupLocalInfoP[CurrentSMGroupid].groupSMQueue;
	ProcGroupLocalInfoP[CurrentSMGroupid].groupSMQueue = mp;
	CurrentSMSegmentStart = mp->beginaddr;
	CurrentSMSegmentEnd =  CurrentSMSegmentStart + mp->size;
	CurrentSMSegmentPointer = CurrentSMSegmentStart;
      }
    retP = CurrentSMSegmentPointer;
    CurrentSMSegmentPointer = (char*)LONGALIGN(CurrentSMSegmentPointer + size);
    return retP;
}
    
/* -------------------------------
 *	ProcGroupSMClean
 *
 *	frees the shared memory allocated for a process group
 * -------------------------------
 */
void
ProcGroupSMClean(groupid)
int groupid;
{
    MemoryHeader mp, nextp;

    mp = ProcGroupLocalInfoP[groupid].groupSMQueue;
    while (mp != NULL) {
	nextp = mp->next;
	ExecSMSegmentFree(mp);
	mp = nextp;
      }
}

/* -----------------------------
 * the following routines are special functions for copying reldescs to
 * shared memory
 * -----------------------------
 */
static char *SlaveTmpRelDescMemoryP;

/* -------------------------------
 *	SlaveTmpRelDescInit
 *
 *	initialize shared memory preallocated for temporary relation descriptors
 * -------------------------------
 */
void
SlaveTmpRelDescInit()
{
    SlaveTmpRelDescMemoryP = (char*)SlaveInfoP[MyPid].resultTmpRelDesc;
}

/* -------------------------------
 *	SlaveTmpRelDescAlloc
 *
 *	memory allocation for reldesc copying
 *	Note: there is no boundary checking, so had better pre-allocate
 *	enough memory!
 * -------------------------------
 */
char *
SlaveTmpRelDescAlloc(size)
int size;
{
    char *retP;

    retP = SlaveTmpRelDescMemoryP;
    SlaveTmpRelDescMemoryP = (char*)LONGALIGN(SlaveTmpRelDescMemoryP + size);

    return retP;
}

/* ---------------------------------
 *	getProcGroupMaxPage
 *
 *	find out the largest page number the slaves are scanning
 *	used only after SIGPARADJ signal has been sent to the
 *	process group.
 * ---------------------------------
 */
int
getProcGroupMaxPage(groupid)
int groupid;
{
    ProcessNode *p;
    int maxpage = NULLPAGE;
    int page;

    for (p = ProcGroupLocalInfoP[groupid].memberProc;
	 p != NULL;
	 p = p->next) {
#ifdef HAS_TEST_AND_SET
	S_LOCK(&(SlaveInfoP[p->pid].lock));
#endif
	page = SlaveInfoP[p->pid].curpage;
	if (page == NOPARADJ)
	    maxpage = NOPARADJ;
	if (maxpage < page && maxpage != NOPARADJ)
	    maxpage = page;
      }
    return maxpage;
}

/* ---------------------------------------
 *	paradj_handler
 *
 *	signal handler for dynamically adjusting degrees of parallelism
 *	XXX only handle heap scan now.
 * ---------------------------------------
 */
int
paradj_handler()
{
    BlockNumber curpage;
    HeapTuple curtuple;
    ItemPointer tid;
    int groupid;

    SLAVE1_elog(DEBUG, "slave %d got SIGPARADJ", MyPid);
    if (SlaveInfoP[MyPid].isDone) {
	/* -----------------------
	 * this means that the whole job is almost done
	 * no adjustment to parallelism should be made
	 * ------------------------
	 */
	SlaveInfoP[MyPid].curpage = NOPARADJ;
	curpage = NOPARADJ;
      }
    else
    if (!SlaveLocalInfoD.isworking || SlaveLocalInfoD.heapscandesc == NULL) {
	if (SlaveInfoP[MyPid].isAddOnSlave) {
	    SlaveInfoP[MyPid].curpage = SlaveLocalInfoD.startpage;
	    curpage = SlaveLocalInfoD.startpage;
	  }
	else {
	    SlaveInfoP[MyPid].curpage = NULLPAGE;
	    curpage = NULLPAGE;
	  }
      }
    else {
        curtuple = SlaveLocalInfoD.heapscandesc->rs_ctup;
        tid = &(curtuple->t_ctid);
	if (ItemPointerIsValid(tid)) {
            curpage = ItemPointerGetBlockNumber(tid);
	  }
	else {
	    curpage = NULLPAGE;
	  }
        SlaveInfoP[MyPid].curpage = curpage;
      }
#ifdef HAS_TEST_AND_SET
    S_UNLOCK(&(SlaveInfoP[MyPid].lock));
#endif
    SLAVE2_elog(DEBUG, "slave %d sending back curpage = %d", MyPid, curpage);
    groupid = SlaveInfoP[MyPid].groupId;
    MWaitOne(&(ProcGroupInfoP[groupid].m1lock));
    SLAVE1_elog(DEBUG, "slave %d complete handshaking with master", MyPid);
    if (ProcGroupInfoP[groupid].paradjpage == NOPARADJ) {
	/* ----------------------
	 * this means that the master changed his/her mind
	 * no adjustment to parallelism will be done
	 * ----------------------
	 */
	return;
      }
    SlaveLocalInfoD.paradjpending = true;
    SlaveLocalInfoD.paradjpage = ProcGroupInfoP[groupid].paradjpage;
    SlaveLocalInfoD.newparallel = ProcGroupInfoP[groupid].newparallel;
    return;
}

/* ------------------------------------
 *	paradj_nextpage
 *
 *	check if parallelism adjustment point is reached, if so
 *	figure out and return the next page to scan.
 *	XXX only works for heapscan right now.
 * -------------------------------------
 */
int
paradj_nextpage(page, dir)
int page;
int dir;
{
    if (SlaveLocalInfoD.paradjpending) {
        if (page >= SlaveLocalInfoD.paradjpage) {
            SLAVE2_elog(DEBUG, "slave %d adjusting page skip to %d",
                        MyPid, SlaveLocalInfoD.newparallel);
            if (SlaveLocalInfoD.newparallel >= SlaveLocalInfoD.nparallel ||
                SlaveInfoP[MyPid].groupPid < SlaveLocalInfoD.newparallel) {
                SlaveLocalInfoD.nparallel = SlaveLocalInfoD.newparallel;
                SlaveLocalInfoD.paradjpending = false;
		if (dir < 0)
                    return 
			SlaveLocalInfoD.paradjpage-SlaveInfoP[MyPid].groupPid;
		else
                    return
			SlaveLocalInfoD.paradjpage+SlaveInfoP[MyPid].groupPid;
              }
            else {
                int groupid = SlaveInfoP[MyPid].groupId;
		Plan plan = QdGetPlan(QueryDesc);
		EState estate = (EState)get_state(plan);
		EndPlan(plan, estate);
                V_Finished(groupid, &(ProcGroupInfoP[groupid].dropoutcounter),
                           PARADJPENDING);
		RestartForParAdj = true;
		SlaveRestart();
              }
          }
      }
    else
	return NULLPAGE;
}
@


1.23
log
@change accessor functions into macros
@
text
@d16 1
a16 1
 *	$Header: /users/mer/pg/src/tcop/RCS/slaves.c,v 1.22 1992/01/21 16:44:25 hong Exp mer $
d24 1
a24 1
 RcsId("$Header: /users/mer/pg/src/tcop/RCS/slaves.c,v 1.22 1992/01/21 16:44:25 hong Exp mer $");
d432 1
a432 1
    SearchSysCacheTuple(ATTNAME, "");
d435 2
a436 2
    SearchSysCacheTuple(LANNAME, "");
    SearchSysCacheTuple(OPRNAME, "");
d438 1
a438 1
    SearchSysCacheTuple(PRONAME, "");
d440 1
a440 1
    SearchSysCacheTuple(RELNAME, "");
d442 1
a442 1
    SearchSysCacheTuple(TYPNAME, "");
d444 2
a445 2
    SearchSysCacheTuple(AMNAME, "");
    SearchSysCacheTuple(CLANAME, "");
@


1.22
log
@changes for inter-fragment parallelism
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.21 91/11/11 23:00:24 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.21 91/11/11 23:00:24 hong Exp Locker: hong $");
d1038 1
a1038 1
		EState estate = get_state(plan);
@


1.21
log
@for prototyping
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.20 91/11/09 09:50:34 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.20 91/11/09 09:50:34 hong Exp Locker: hong $");
d978 6
a983 1
        curpage = ItemPointerGetBlockNumber(tid);
@


1.20
log
@added mechanism for reducing parallelism of a fragment
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.18 91/10/25 16:22:05 hong Exp $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.18 91/10/25 16:22:05 hong Exp $");
d38 6
a48 4
#include "executor/execdesc.h"
#include "nodes/plannodes.h"
#include "nodes/plannodes.a.h"
#include "nodes/execnodes.h"
a413 1
    int paradj_handler(); 	/* intr handler for adjusting parallelism */
@


1.19
log
@default stat file to stderr
@
text
@d44 3
d62 4
a65 1
FILE *StatFp = stderr;
a70 2
MasterCommunicationData *MasterDataP; /* communication data area for 
					master backend */
a240 1
    List queryDesc;
d242 1
d256 1
a256 1
	sprintf(fname, "/usr/tmp/slave%d.stat", MyPid);
d266 9
a274 3
	SlaveWarnings++;
	SLAVE1_elog(DEBUG, "Slave Backend %d SlaveBackendsAbort()",
		    MyPid);
d276 5
a280 5
	SlaveBackendsAbort();
	
	SLAVE1_elog(DEBUG, "Slave Backend %d SlaveBackendsAbort() done",
		    MyPid);
    }	
d317 1
d320 2
a321 3
				     MasterDataP->data[0] : 0);
	SlaveLocalInfoD.nparallel = 
			    ProcGroupInfoP[SlaveInfoP[MyPid].groupId].nprocess;
d333 1
a333 2
	queryDesc = (List)CopyObject(
		     (List)ProcGroupInfoP[SlaveInfoP[MyPid].groupId].queryDesc);
d339 2
a340 2
	if (queryDesc != NULL)
	    ProcessQueryDesc(queryDesc);
d371 1
a371 1
	V_Finished(SlaveInfoP[MyPid].groupId);
a472 2
    MasterDataP = 
       (MasterCommunicationData*)ExecSMReserve(sizeof(MasterCommunicationData));
a497 1
    InitMWaitOneLock(&(MasterDataP->m1lock));
d517 2
a518 2
		S_INIT_LOCK(&(SlaveInfoP[i].comdata.lock));
		S_LOCK(&(SlaveInfoP[i].comdata.lock));
d522 2
a523 1
		ProcGroupInfoP[i].countdown = 0;
d525 2
a526 1
		S_INIT_LOCK(&(ProcGroupInfoP[i].lock));
d528 2
d575 1
d617 1
a617 1
static void
d705 1
d713 1
a713 1
 *	process group with status = FINISHED
d722 2
a723 1
	if (ProcGroupInfoP[i].status == FINISHED)
d920 1
a920 1
    int maxpage = -1;
d927 1
a927 1
	S_LOCK(&(SlaveInfoP[p->pid].comdata.lock));
d929 1
a929 1
	page = SlaveInfoP[p->pid].comdata.data;
d951 1
d960 1
a960 1
	SlaveInfoP[MyPid].comdata.data = NOPARADJ;
d966 1
a966 1
	    SlaveInfoP[MyPid].comdata.data = SlaveLocalInfoD.startpage;
d970 1
a970 1
	    SlaveInfoP[MyPid].comdata.data = NULLPAGE;
d978 1
a978 1
        SlaveInfoP[MyPid].comdata.data = curpage;
d981 1
a981 1
    S_UNLOCK(&(SlaveInfoP[MyPid].comdata.lock));
d984 2
a985 1
    MWaitOne(&(MasterDataP->m1lock));
d987 1
a987 1
    if (MasterDataP->data[0] == NOPARADJ) {
d996 2
a997 2
    SlaveLocalInfoD.paradjpage = MasterDataP->data[0];
    SlaveLocalInfoD.newparallel = MasterDataP->data[1];
d999 44
@


1.18
log
@added process affinity to slave backends
print slave stats to stat file
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.17 91/10/09 14:52:07 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.17 91/10/09 14:52:07 hong Exp Locker: hong $");
d59 1
a59 1
FILE *StatFp;
@


1.17
log
@fixed a bug in parallelism adjustment
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.16 91/10/08 01:15:07 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.16 91/10/08 01:15:07 hong Exp Locker: hong $");
d59 1
d239 1
d246 9
d301 2
d333 10
a361 1
	SlaveInfoP[MyPid].isDone = true;
d404 1
d522 14
a535 1
		
d568 1
a568 1
    else
d570 1
@


1.16
log
@bug fixing for dynamic adjustment of parallelism
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.15 91/08/27 14:19:11 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.15 91/08/27 14:19:11 hong Exp Locker: hong $");
d339 1
d601 1
d609 1
d880 3
a882 1
	if (maxpage < page)
d903 10
d919 2
a920 2
	    SlaveInfoP[MyPid].comdata.data = -1;
	    curpage = -1;
d935 8
@


1.15
log
@stupid fixes to make postmaster link
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.14 91/08/25 13:35:20 hong Exp $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.14 91/08/25 13:35:20 hong Exp $");
d239 5
d302 3
d338 1
d477 4
d621 1
a621 1
addSlaveToProcGroup(slave, group)
d624 1
d627 1
a627 1
    SlaveInfoP[slave].groupPid = ++(ProcGroupInfoP[group].nprocess);
d887 1
d897 17
a913 4
    curtuple = SlaveLocalInfoD.heapscandesc->rs_ctup;
    tid = &(curtuple->t_ctid);
    curpage = ItemPointerGetBlockNumber(tid);
    SlaveInfoP[MyPid].comdata.data = curpage;
d917 1
d919 1
@


1.14
log
@added supports for dynamic parallelism adjustment
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.13 91/07/31 12:37:57 hong Exp $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.13 91/07/31 12:37:57 hong Exp $");
d52 1
a52 1
int MyPid = -1;	/* int representing the process id */
d57 1
a57 1
SlaveLocalInfoData SlaveLocalInfoD;
d67 1
a67 1
SlaveInfo	SlaveInfoP;	/* slave backend info */
@


1.13
log
@minor bug fix
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.12 91/07/24 16:14:49 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.12 91/07/24 16:14:49 hong Exp Locker: hong $");
d57 2
d64 2
d285 13
d371 1
d409 7
d431 2
d458 1
a458 1
    
d473 4
a513 1
	    ProcGroupLocalInfoP[i].nmembers = 0;
d536 1
a536 1
static int
d586 1
d593 1
a597 1
    p->nmembers = nproc;
d601 18
a638 1
    ProcGroupLocalInfoP[gid].nmembers = 0;
d681 20
d838 56
@


1.12
log
@added new routines for shared memory management of parallel process groups
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.10 91/07/17 23:37:44 hong Exp $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.10 91/07/17 23:37:44 hong Exp $");
d52 1
a52 1
int MyPid;	/* int representing the process id */
d580 1
a580 1
    ProcessNode *p;
d582 3
a584 1
    for (p=ProcGroupLocalInfoP[gid].memberProc; p!=NULL; p=p->next)
d586 2
@


1.11
log
@stupid fixes to make postmaster happy to link.
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.10 91/07/17 23:37:44 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.10 91/07/17 23:37:44 hong Exp Locker: hong $");
d64 3
a66 3
extern ProcGroupInfo	ProcGroupInfoP; /* process group info */
					/* defined in execipc.c to make
					   postmaster happy */
d187 1
a187 2
	I_SharedMemoryMutex();
	ExecSMClean();
a331 2
    SharedTransactionState = (TransactionState) ExecSMHighAlloc(nbytes);
	
d405 4
a408 4
    MasterProcessIdP = (int*)ExecSMHighAlloc(sizeof(int));
    SlaveAbortFlagP = (int*)ExecSMHighAlloc(sizeof(int));
    SlaveInfoP = (SlaveInfo)ExecSMHighAlloc(nslaves * sizeof(SlaveInfoData));
    ProcGroupInfoP = (ProcGroupInfo)ExecSMHighAlloc(nslaves *
d410 2
d421 5
d611 1
a611 1
 *	WakeupProcGroup
d627 139
@


1.10
log
@changes to support inter-fragment and inter-query parallelism in parallel
executor, non-parallel processing remain the same as before
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.9 91/05/31 10:55:02 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.9 91/05/31 10:55:02 hong Exp Locker: hong $");
d64 3
a66 1
ProcGroupInfo	ProcGroupInfoP; /* process group info */
@


1.9
log
@fixed a bug, P_Finished() should be called with a parameter
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.8 91/04/18 11:39:44 hong Exp $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.8 91/04/18 11:39:44 hong Exp $");
d49 8
a56 1
int     MyPid;			/* int representing the process id */
d58 3
d62 3
a64 4
int 	*SlaveProcessIdsP;	/* array of slave backend process id's */
Pointer *SlaveQueryDescsP;	/* array of pointers to slave query descs */
Pointer *SlaveRetStateP;	/* array of pointers to slave return states */
int     *SlaveAbortFlagP;	/* flag set during a transaction abort */
d68 2
a69 2
#define CONDITION_NORMAL 0
#define CONDITION_ABORT  1
d96 1
a96 1
	    p = SlaveProcessIdsP[i];
d192 1
a192 1
	P_Finished(nslaves);
d220 1
a220 1
	V_Finished();
a232 1
    int nproc;
d284 2
a285 9
	nproc = 0;
	for (i=0; i<GetNumberSlaveBackends(); i++) {
	    nproc += get_parallel(QdGetPlan((List) SlaveQueryDescsP[i]));
	    if (MyPid < nproc) {
	        queryDesc = (List)CopyObject((List)SlaveQueryDescsP[i]);
		break;
	      }
	  }
	Assert(i < GetNumberSlaveBackends());
a286 1
#if 0
a287 25
	 *	for now, just test synchronization.
	 * ----------------
	 */
	srandom(MyPid);
	{
	    int sleeptime = 5 + (random() & 017);
	    
	    elog(DEBUG, "Slave Backend %d executing (ETA:%d)",
		 MyPid, sleeptime);
	    
	    sleep(sleeptime);
	    
	    elog(DEBUG, "Slave Backend %d finished", MyPid);
	}
#endif	
#if 0
	/* ----------------
	 *  If the query desc is NULL, then we assume the master
	 *  backend has finished execution.
	 * ----------------
	 */
	if (queryDesc == NULL)
	    exitpg();
#endif	
	/* ----------------
d312 1
a312 1
	V_Finished();
d406 5
a410 5
    MasterProcessIdP =  (int *)     ExecSMHighAlloc(sizeof(int));
    SlaveProcessIdsP =  (int *)     ExecSMHighAlloc(nslaves * sizeof(int));
    SlaveAbortFlagP  =  (int *)     ExecSMHighAlloc(sizeof(int));
    SlaveQueryDescsP =  (Pointer *) ExecSMHighAlloc(nslaves * sizeof(Pointer));
    SlaveRetStateP =  (Pointer *) ExecSMHighAlloc(nslaves * sizeof(Pointer));
d430 1
a430 1
    MyPid = 0;
d432 1
a432 1
    for (i=1; i<nslaves; i++)
d435 11
a445 3
		SlaveProcessIdsP[i] = p;
		SlaveQueryDescsP[i] = NULL;
		SlaveRetStateP[i] = NULL;
d460 23
a482 1
    if (! IsMaster)
d491 130
@


1.8
log
@changes to save plan copying in communication between master backend
and slave backend
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.7 91/04/04 13:51:16 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.7 91/04/04 13:51:16 hong Exp Locker: hong $");
d183 1
a183 1
	P_Finished();
@


1.7
log
@added a hack to initialize syscaches beforehand so performance
numbers will look better.
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.7 91/04/04 11:56:15 hong Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.7 91/04/04 11:56:15 hong Exp Locker: hong $");
d43 1
d54 1
d223 2
d276 9
a284 1
	queryDesc = (List) SlaveQueryDescsP[MyPid];
d435 1
d462 1
@


1.6
log
@changed some debugging prints.
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.5 90/08/18 14:16:20 cimarron Exp Locker: hong $
d24 1
a24 1
 RcsId("$Header: RCS/slaves.c,v 1.5 90/08/18 14:16:20 cimarron Exp Locker: hong $");
d42 1
d375 29
@


1.5
log
@rearranged #includes to eliminate tcop/tcop.h and executor/executor.h
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.4 90/08/14 16:20:21 cimarron Exp Locker: cimarron $
d24 1
a24 1
 RcsId("$Header$");
d38 1
d47 1
a47 2
bool    IsMaster;		/* true if this process is the master */
int     SlaveId;		/* int representing the slave id */
d78 1
a78 1
	elog(DEBUG, "Slave Backend %d sending abort signals", SlaveId);
d83 1
a83 1
	if (i != SlaveId) {
d114 1
a114 1
    SLAVE1_elog(DEBUG, "Slave Backend %d *** SlaveRestart ***", SlaveId);
d197 1
a197 1
		    SlaveId);
d201 1
a201 1
		    SlaveId);
d205 1
a205 1
		    SlaveId);
d230 1
a230 1
		    SlaveId);
d235 1
a235 1
		    SlaveId);
d244 1
a244 1
    SLAVE1_elog(DEBUG, "Slave Backend %d entering slave loop...", SlaveId);
d261 1
a261 1
	SLAVE1_elog(DEBUG, "Slave Backend %d waiting...", SlaveId);
d263 1
a263 1
	P_Start(SlaveId);
d265 1
a265 1
	SLAVE1_elog(DEBUG, "Slave Backend %d starting task.", SlaveId);
d271 1
a271 1
	queryDesc = (List) SlaveQueryDescsP[SlaveId];
d278 1
a278 1
	srandom(SlaveId);
d283 1
a283 1
		 SlaveId, sleeptime);
d287 1
a287 1
	    elog(DEBUG, "Slave Backend %d finished", SlaveId);
d323 1
a323 1
	SLAVE1_elog(DEBUG, "Slave Backend %d task complete.", SlaveId);
d412 1
a412 2
    IsMaster = true;
    SlaveId = -1;
d414 1
a414 1
    for (i=0; i<nslaves; i++)
d420 1
a420 2
		IsMaster = false;
		SlaveId = i;
d422 1
a422 1
		SLAVE1_elog(DEBUG, "Slave Backend %d alive!", SlaveId);
@


1.4
log
@added pathnames to header includes
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.3 90/05/25 10:36:35 cimarron Version_2 Locker: cimarron $
d19 2
d22 1
a22 1
#include "tcop/tcop.h"
d24 17
a40 1
 RcsId("$Header: RCS/slaves.c,v 1.3 90/05/25 10:36:35 cimarron Version_2 Locker: cimarron $");
@


1.3
log
@checked in slaves changes for wei..
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.2 90/03/31 19:01:51 cimarron Exp $
d20 3
a22 2
#include "tcop.h"
 RcsId("$Header: RCS/slaves.c,v 1.2 90/03/31 19:01:51 cimarron Exp $");
@


1.2
log
@*** empty log message ***
@
text
@d16 1
a16 1
 *	$Header: RCS/slaves.c,v 1.1 90/03/12 16:27:18 cimarron Exp $
d21 1
a21 1
 RcsId("$Header: RCS/slaves.c,v 1.1 90/03/12 16:27:18 cimarron Exp $");
d75 1
a75 1
	    fprintf(stderr, "signaling master (pid %d): ", i, p);
d254 1
d270 1
d279 1
a279 1

d284 2
a285 2
	ProcessQueryDesc(queryDesc);
#endif	
@


1.1
log
@Initial revision
@
text
@a13 1
 *	The semaphore operations should be moved elsewhere
d16 1
a16 1
 *	$Header: RCS/initslaves.c,v 1.1 90/03/01 15:40:24 cimarron Exp $
d20 2
a21 2
#include "executor.h"
 RcsId("$Header$");
d33 1
d35 5
d41 1
a41 1
 *	SlaveMain is the main loop executed by the slave backends
d43 3
a45 2
 *	Note: have to add code to communicate transaction information
 *	      in slaves and synchronize aborts and stuff.
d49 149
d201 1
a201 1
    
d203 3
a205 2
     *	XXX have to initialize the transaction exception handlers
     *  here so that aborts and signals are processed correctly..
d208 18
d229 8
a236 1
	 *  The master V's on the start semaphore after
d238 2
a239 2
	 *  Meanwhile slaves wait on their start semaphore
	 *  until the master signals them.
d242 2
d245 2
d250 21
a274 1
	queryDesc = (List) SlaveQueryDescsP[SlaveId];
d283 2
a284 1
	
d286 7
d299 1
a299 1
	 *  should bring the finished count to -1;
d302 1
d306 15
d322 6
d355 1
a355 1
     *	initialize Start[] and Finished semaphores to 0
d361 1
d370 1
d372 7
d381 1
a381 1
     *	save master backend's process id.
d385 1
d391 14
a404 8
    for (i=0; i<nslaves; i++) {
	if ((p = fork()) != 0) {
	    IsMaster = true;
	    SlaveProcessIdsP[i] = p;
	    SlaveQueryDescsP[i] = NULL;
	} else {
	    SlaveId = i;
	    IsMaster = false;
d406 1
a406 2
    }

d408 1
a408 1
     *	now IsMaster is true in the master backend and is false
a423 13


/* --------------------------------
 *	SlaveBackendsAbort tells the slave backends to abandon
 *	their task and wait for new instructions.
 * --------------------------------
 */
void
SlaveBackendsAbort()
{
    return;
}

@