/*
 * Copyright 1998-1999, University of Notre Dame.
 * Authors: Brian W. Barrett, Arun F. Rodrigues, Jeffrey M. Squyres,
 * 	 and Andrew Lumsdaine
 *
 * This file is part of XMPI
 *
 * You should have received a copy of the License Agreement for XMPI 
 * along with the software; see the file LICENSE.  If not, contact 
 * Office of Research, University of Notre Dame, Notre Dame, IN 46556.
 *
 * Permission to modify the code and to distribute modified code is
 * granted, provided the text of this NOTICE is retained, a notice that
 * the code was modified is included with the above COPYRIGHT NOTICE and
 * with the COPYRIGHT NOTICE in the LICENSE file, and that the LICENSE
 * file is distributed with the modified code.
 *
 * LICENSOR MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR IMPLIED.
 * By way of example, but not limitation, Licensor MAKES NO
 * REPRESENTATIONS OR WARRANTIES OF MERCHANTABILITY OR FITNESS FOR ANY
 * PARTICULAR PURPOSE OR THAT THE USE OF THE LICENSED SOFTWARE COMPONENTS
 * OR DOCUMENTATION WILL NOT INFRINGE ANY PATENTS, COPYRIGHTS, TRADEMARKS
 * OR OTHER RIGHTS.
 *
 * Additional copyrights may follow.

 *
 *	$Id: xmpi_db_parse.c,v 1.5 1999/11/08 06:20:23 bbarrett Exp $
 * 
 *	Function:	- parse a trace file into the database and view it
 */

#include <Xm/Xm.h>

#include <errno.h>
#include <float.h>
#include <stdio.h>
#include <stdlib.h>

#include "all_list.h"
#include "lam.h"
#include "lamtrace.h"

#include "xmpi.h"
#include "xmpi_dbase.h"

/*
 * public functions
 */
void			xmpi_parse_tracefile();
int			xmpi_db_parse_segment();
void			xmpi_db_free();
void			xmpi_db_view();

/*
 * external functions
 */
extern void		xmpi_tr_select_segment();
extern void		lamtr_set_stream();
extern int4		stoi();

/*
 * private functions
 */
static int		parse_beginning();
static int		missing_traces();
static int		parse_comm();
static int		parse_dtype();
static int		parse_iof();
static int		parse_rtime();
static int		append_substrate();
static int		db_find_rank();
static int		db_init();
static void		free_cids();
static void		free_dtypes();
static int		dtype_cmp();
static int		tr_cmp();
static int		add_cid();
static int		cid_cmp();
static int		onoff_cmp();
static void		free_parse_structs();
static struct dbparse	*create_parse_structs();

/*
 * public variables
 */
struct dbparse   	*dbp = 0;		/* array parsing state */
int			xmpi_is_express = 0;	/* current appl. trace view? */
int			xmpi_do_express = 0;	/* creating current app view */

/*
 * external variables
 */
extern struct xmdb	dbase;			/* database */
extern FILE		*trace_stream;

/*
 * private variables
 */
static struct trsrc	src;			/* src trace */

/*
 *	xmpi_parse_tracefile
 *
 *	Function:	- parse a trace file into the database
 *	Accepts:	- open stream to trace file
 */
void
xmpi_parse_tracefile(file)

FILE			*file;

{
	int		maxsegs = 0;		/* maximum # of segments */
	int		nsegs;			/* number of segments */
	struct dbparse	*pp;			/* parsing state */
	int		i;

	lamtr_set_stream(file);
/*
 * Parse up to end of trace segment on/off traces.
 */
	xmpi_ctl_setinfo("Parsing trace file...");

	xmpi_busy();

	if (parse_beginning()) {
		xmpi_db_free();
		xmpi_ctl_resetinfo();
		xmpi_unbusy();
		return;
	}
/*
 * Check that we have at least one segment.
 */
	for (i = 0, pp = dbp; i < dbase.xdb_nprocs; i++, pp++) {
		if ((nsegs = al_count(pp->dbp_segs)) > maxsegs)
			maxsegs = nsegs;
	}

	if (maxsegs == 0) {
		xmpi_db_free();
		errno = 0;
		xmpi_error(0, "No segments found in trace file.");
		xmpi_ctl_resetinfo();
		xmpi_unbusy();
		return;
	}
/*
 * Select, parse and view a segment.
 */
	xmpi_tr_select_segment(maxsegs);
}

/*
 *	parse_beginning
 *
 *	Function:	- parse beginning of trace file up to segment selection
 *	Returns:	- 0 or LAMERROR
 */
static int
parse_beginning()

{
	struct _gps	*world;			/* GPS array */
	int4		nprocs;			/* number of processes */
	int		rank;			/* rank of process */
	int		err;			/* file read error status */
/*
 * Read the size of the world and the process GPS array.
 */
	if (lamtr_parse_world(&nprocs, &world)) {
		if (errno == EINVAL) {
			errno = 0;
			xmpi_error(0, "Bad world trace in trace file.");
		} else {
			xmpi_error(0, "Parsing world trace");
		}
		return(LAMERROR);
	}
/*
 * Initialize the database with the world information.
 */
	if (db_init((int) nprocs, world)) {
		xmpi_error(0, "Initialiazing database");
		return(LAMERROR);
	}
	free((char *) world);
/*
 * Create the parsing state data structures.
 */
	if (create_parse_structs() == 0) {
		xmpi_error(0, "Creating parsing state data structures");
		return(LAMERROR);
	}
/*
 * Read and parse the init/on/off traces.
 */
	while ((err = lamtr_raw_src_m(src)) == 0) {
		mttoli4((int4 *) &src, sizeof(struct trsrc) / sizeof(int4));
		
		rank = db_find_rank(src.trs_node, src.trs_pid);
		if (rank < 0) {
			errno = 0;
			xmpi_error(0, "Invalid trace.");
			return(LAMERROR);
		}

		if (src.trs_listno == TRONOFF) {
			if (parse_iof(rank)) {
				return(LAMERROR);
			}
		} else if (src.trs_listno == TRCOMM
				|| src.trs_listno == TRDTYPE
				|| src.trs_listno == TRRUNTIME) {
/*
 * At end of init/on/off traces.
 */
			break;
		} else {
			errno = 0;
			xmpi_error(0, "Invalid trace.");
			return(LAMERROR);
		}
	}

	if (err == 1) {
		errno = 0;
		xmpi_error(0, "Unexpected EOF parsing trace file.");
	} else if (err < 0) {
		xmpi_error(0, "Parsing trace file");
	}

	return((err == 0) ? 0 : LAMERROR);
}

/*
 *	xmpi_db_view
 *
 *	Function:	- view the current database
 */
void
xmpi_db_view()

{
/*
 * Clear the old application or trace.
 */
	xmpi_dt_clear();
	xmpi_kv_popdown();
	xmpi_tr_destroy();
	if (!xmpi_do_express) {
		xmpi_mat_destroy();
		xmpi_vw_clear();
	}
	XmUpdateDisplay(xmpi_shell);
/*
 * View the current trace.
 */
	xmpi_tr_create();
	xmpi_vw_dbsetmode(1);
	xmpi_is_express = xmpi_do_express;

	if (xmpi_do_express && xmpi_vw_getprocs()) {
		xmpi_vw_update(1);
	} else {
		xmpi_vw_init();
	}

	xmpi_do_express = 0;
	xmpi_ctl_setsensitive(XMPI_BSNAPSHOT, False);
	xmpi_ctl_setsensitive(XMPI_BKIVIAT, True);
	xmpi_ctl_setsensitive(XMPI_BMATRIX, True);
	xmpi_ctl_resetinfo();
}

/*
 *	xmpi_db_parse_segment
 *
 *	Function:	- parse a trace segment
 *			- src header of first run-time trace has been read
 *	Accepts:	- segment to parse
 *	Returns:	- 0 or LAMERROR
 */
int
xmpi_db_parse_segment(seg)

int			seg;

{
	struct dbparse	*pp;			/* parsing state */
	struct onoff	ooelem;			/* onoff search element */
	int		rank;			/* rank of process */
	int		err;			/* error status */
	int		i;

/*
 * Set the segment for each process and initialize.
 */
	ooelem.oo_segment = seg;
	for (i = 0, pp = dbp; i < dbase.xdb_nprocs; i++, pp++) {
		if ((pp->dbp_curseg = al_find(pp->dbp_segs, &ooelem))) {
			pp->dbp_curseg->oo_rton = 0;
			pp->dbp_curseg->oo_rtoff = 0;
			pp->dbp_end = pp->dbp_curseg->oo_ontime;
		} else {
			errno = 0;
			xmpi_error(0, "No segment found for process.");
			return(LAMERROR);
		}
	}
/*
 * Parse the communicator, datatype and runtime traces.
 */
	do {
		rank = db_find_rank(src.trs_node, src.trs_pid);

		if (rank < 0) {
			errno = 0;
			xmpi_error(0, "Invalid trace.");
			return(LAMERROR);
		}

		switch (src.trs_listno) {
		case TRCOMM:	
			if (parse_comm()) {
				return(LAMERROR);
			}
			break;
		case TRDTYPE:
			if (parse_dtype(rank)) {
				return(LAMERROR);
			}
			break;
		case TRRUNTIME:
			if (parse_rtime(rank)) {
				return(LAMERROR);
			}
			break;
		default:
			errno = 0;
			xmpi_error(0, "Invalid trace.");
			return(LAMERROR);
		}
/*
 * Read in the next trace src header.
 */
		err = lamtr_raw_src_m(src);
		if (err == 1) {
			break;			/* eof */
		} else if (err) {
			xmpi_error(0, "Parsing trace file");
			return(LAMERROR);
		}

		mttoli4((int4 *) &src, sizeof(struct trsrc) / sizeof(int4));

	} while (1);
/*
 * Handle cases of missing traces for each process.
 */
	for (i = 0, pp = dbp; i < dbase.xdb_nprocs; i++, pp++) {
		if (missing_traces(i, pp)) {
			xmpi_error(0, "Parsing trace file");
			return(LAMERROR);
		}
	}
/*
 * All traces have been read.  Set up the database internal structures.
 */
	if (xmpi_db_internals()) {
		xmpi_error(0, "Internal database error");
		return(LAMERROR);
	}
/*
 * Done with parsing state.
 */
	free_parse_structs();

	return(0);
}

/*
 *	missing_traces
 *
 *	Function:	- make sure database traces are consistent in case of
 *			  missing traces
 *	Accepts:	- process rank
 *			- parsing state
 */
static int
missing_traces(rank, pp)

int			rank;
struct dbparse		*pp;

{
	if (!pp->dbp_curseg->oo_inoff) {
		if (pp->dbp_curseg->oo_rtoff) {
/*
 * No initial off trace found but a run-time off trace found.
 * Report an error as this should not happen.
 */
			errno = 0;
			xmpi_error(0, "Invalid trace file.");
			return(LAMERROR);
		}
		else if (!al_top(dbase.xdb_procs[rank].xdbp_traces)) {
/*
 * There are traces at all.  Insert a short running trace.  Since there
 * was no initial off trace we cannot deduce that traces were dropped at
 * the end.  
 */
			pp->dbp_sub = XMPI_SRUN;
			if (append_substrate(rank,
					pp->dbp_curseg->oo_ontime + 1.0)) {
		    		xmpi_error(0, "While parsing");
				return(LAMERROR);
			}
			pp->dbp_end = pp->dbp_curseg->oo_ontime + 1.0;
		}
	} else if (! pp->dbp_curseg->oo_rtoff) {
/*
 * No runtime off trace was found hence runtime traces are missing at the end.
 */
		pp->dbp_sub = XMPI_SMISSING;
		if (append_substrate(rank, pp->dbp_curseg->oo_offtime)) {
			xmpi_error(0, "while parsing");
			return(LAMERROR);
		}
		pp->dbp_end = pp->dbp_curseg->oo_offtime;
	}

	return(0);
}

/*
 *	parse_comm
 *
 *	Function:	- parse a communicator trace
 *	Returns:	- 0 or LAMERROR
 */
static int
parse_comm()

{
	struct xmdbcid	xmcid;			/* cid database entry */
	struct _gps	*grps;			/* group GPS */
/*
 * Read the communicator trace.
 */
	if (lamtr_parse_comm(&xmcid.xdbc_cid, &xmcid.xdbc_nlprocs,
					&xmcid.xdbc_nrprocs, &grps)) {
		return(LAMERROR);
	}

	xmcid.xdbc_lgrp = grps;
	xmcid.xdbc_rgrp = grps + xmcid.xdbc_nlprocs;
/*
 * Add the cid to the database.
 */
	if (add_cid(&xmcid)) {
		free((char *) grps);
		return(LAMERROR);
	}

	return(0);
}

/*
 *	parse_dtype
 *
 *	Function:	- parse a datatype trace
 *	Accepts:	- rank of process that generated the trace
 *	Returns:	- 0 or LAMERROR
 */
static int
parse_dtype(rank)

int			rank;

{
	struct xmdbdt	xmdt;			/* datatype database entry */
/*
 * Read the datatype trace.
 */
	if (lamtr_parse_dtype(&xmdt.xdbd_dtype, &xmdt.xdbd_dtbuf)) {
		return(LAMERROR);
	}
/*
 * Add the datatype to the database.
 */
	if (al_append(dbase.xdb_procs[rank].xdbp_dtypes, &xmdt) == 0) {
		return(LAMERROR);
	}

	return(0);
}

/*
 *	parse_iof
 *
 *	Function:	- parse a init/on/off trace
 *	Accepts:	- rank of process that generated the trace
 *	Returns:	- 0 or LAMERROR
 */
static int
parse_iof(rank)

int			rank;

{
	struct trrthdr	rthdr;			/* run-time header */
	struct tronoff	onoff;			/* onoff trace data */
	struct trinit	init;			/* init data */
	struct onoff	ooelem;			/* onoff list element */
	struct onoff	*pelem;			/* matched onoff list element */
	struct dbparse  *pp;			/* parsing state */
	float8		trtime;			/* trace time */
	float8		skew;			/* cloack skew */
	int4		type;			/* trace type */
	int4		segment;		/* trace segment number */
	char		*name;			/* program name */
/*
 * Read the run-time header.
 */
	if (lamtr_raw_rthdr_m(rthdr)) {
		return(LAMERROR);
	}
	lamtr_xt_rt_time_m(&rthdr, &trtime);
	lamtr_xt_rt_type_m(&rthdr, &type);
/*
 * Parse according to the type of trace.
 */
	pp = dbp + rank;

	switch (type) {
	case TRTINIT:
		if (lamtr_raw_init_m(init)) {
			return(LAMERROR);
		}
		lamtr_xt_init_skew_m(&init, &skew);
		lamtr_xt_init_prog_m(&init, name);
		strncpy(dbase.xdb_procs[rank].xdbp_prog, name, TRDPROGMAX);
		pp->dbp_skew = skew;
		pp->dbp_start = trtime + skew;
		break;
	case TRTRON:
	case TRTROFF:
		if (lamtr_raw_onoff_m(onoff)) {
			return(LAMERROR);
		}
		lamtr_xt_onoff_num_m(&onoff, &segment);
		memset(&ooelem,0,sizeof(ooelem));
		ooelem.oo_segment = segment;
		
		if (type == TRTRON) {
/*
 * Add a new segment description to the segment list for the current process.
 */
			ooelem.oo_inoff = 0;
			ooelem.oo_ontime = trtime + pp->dbp_skew;
			if (al_insert(pp->dbp_segs, &ooelem) == 0) {
				return(LAMERROR);
			}
		} else {
/*
 * It's a tracing off trace.  Find the segment description and update it.
 */
			if ((pelem = al_find(pp->dbp_segs, &ooelem))) {
				pelem->oo_inoff = 1;
				pelem->oo_offtime = trtime + pp->dbp_skew;
			} else {
				errno = 0;
				xmpi_error(0, "On/off trace mismatch.");
				return(LAMERROR);
			}
		}
	}

	return(0);
}

/*
 *	parse_rtime
 *
 *	Function:	- parse a run-time trace
 *	Accepts:	- rank of process that generated the trace
 *	Returns:	- 0 or LAMERROR
 */
static int
parse_rtime(rank)

int			rank;

{
    struct dbparse  	*pp;			/* ptr parsing state */
    struct onoff	*oo;			/* current seg. onoff data */
    struct trrthdr	rthdr;			/* run-time header */
    struct trmsg	msg;			/* message trace data */
    struct tronoff	onoff;			/* onoff trace data */
    struct trxchg	xchg;			/* substrate change data */
    struct xmdbtr   	xtrace;			/* new database trace */
    struct xmdbtr   	*dbtr;			/* ptr to trace in database */
    struct xmdbtr   	*sendtr;		/* send trace */
    int             	dest;                   /* destination process */
    float8		trtime;			/* trace time */
    int4		type;			/* trace type */
    int4		segment;		/* trace segment number */

    memset(&xtrace,0,sizeof(xtrace));

    pp = dbp + rank;
    oo = pp->dbp_curseg;
/*
 * Read the run-time header and adjust the trace time.
 */
    if (lamtr_raw_rthdr_m(rthdr)) {
		return(LAMERROR);
    }
    lamtr_xt_rt_time_m(&rthdr, &trtime);
    lamtr_xt_rt_type_m(&rthdr, &type);

    trtime += pp->dbp_skew;
/*
 * Parse according to the type of run-time trace.
 */
    switch (type) {

    case TRTRON:
    case TRTROFF:
	if (lamtr_raw_onoff_m(onoff)) {
	    return(LAMERROR);
	}
	lamtr_xt_onoff_num_m(&onoff, &segment);

	if (segment == oo->oo_segment) {
	    if (type == TRTRON) {
		pp->dbp_sub = XMPI_SRUN;
		oo->oo_rton = 1;
	    }
	    else {
		if (append_substrate(rank, trtime)) {
		    return(LAMERROR);
		}
		pp->dbp_end = trtime;
		oo->oo_rtoff = 1;
	    }
	}
	break;
		
    case TRTSUBCHG:
	if (lamtr_raw_xchg_m(xchg)) {
	    return(LAMERROR);
	}
	if ((trtime < oo->oo_ontime)
		|| (oo->oo_inoff && (trtime > oo->oo_offtime))) {
	    return(0);
	}
	if (append_substrate(rank, trtime)) {
	    return(LAMERROR);
	}
/*
 * Save the information for the new substrate.
 */
	pp->dbp_end = trtime;
	lamtr_xt_xchg_m(&xchg, &pp->dbp_top, &pp->dbp_sub);
	if (pp->dbp_sub == XMPI_SRUN) {
	    pp->dbp_top = 0;
	}
	break;

    case TRTINPUT:
    case TRTOUTPUT:
    case TRTNOIO:
	if (lamtr_raw_msg_m(msg)) {
	    return(LAMERROR);
	}
	if ((trtime < oo->oo_ontime)
		|| (oo->oo_inoff && (trtime > oo->oo_offtime))) {
	    return(0);
	}
/*
 * Add a substrate trace from the end of the last trace up until the start
 * of this one.
 */
	if (append_substrate(rank, trtime)) {
	    return(LAMERROR);
	}
/*
 * Create and append a system time trace.
 */
	mttoli4((int4 *) &msg, sizeof(struct trmsg) / sizeof(int4));

	if (msg.trm_syst < 0 || msg.trm_blkt < 0) {
		errno = 0;
		xmpi_error(0, "Invalid trace: duration < 0.");
		return(LAMERROR);
	}

	if (msg.trm_syst == 0 && msg.trm_blkt == 0) {
	    msg.trm_syst = 1;
	}

	xtrace.xdbt_envelop.xdbe_func = msg.trm_topfunc;
	xtrace.xdbt_envelop.xdbe_wfunc = msg.trm_wrapfunc;
	xtrace.xdbt_envelop.xdbe_seqnum = msg.trm_seqnum;
	xtrace.xdbt_envelop.xdbe_lpeer = msg.trm_peer;
	xtrace.xdbt_envelop.xdbe_mlpeer = msg.trm_mrank;
	xtrace.xdbt_envelop.xdbe_tag = msg.trm_tag;
	xtrace.xdbt_envelop.xdbe_cid = msg.trm_cid;
	xtrace.xdbt_envelop.xdbe_dtype = msg.trm_dtype;
	xtrace.xdbt_envelop.xdbe_count = msg.trm_count;
	xtrace.xdbt_envelop.xdbe_mtag = msg.trm_mtag;

	xtrace.xdbt_grank = rank;     
	xtrace.xdbt_senders = 0;
	xtrace.xdbt_arrow = 0;

	if (msg.trm_syst > 0) {

	    xtrace.xdbt_state = XMPI_SSYSTEM;     
	    xtrace.xdbt_time = trtime;
	    xtrace.xdbt_lapse = (double) msg.trm_syst / 1000000.0;
	    xtrace.xdbt_blktotal = pp->dbp_blktot;
	    xtrace.xdbt_systotal = pp->dbp_systot;

	    if (type == TRTNOIO) {
		xtrace.xdbt_arrowdir = XMPI_DBNA;
	    }
	    else if (type == TRTOUTPUT) {
		xtrace.xdbt_arrowdir = XMPI_DBOUT;
		xtrace.xdbt_arrowloc = XMPI_DBSTART;
	    }
	    else if (msg.trm_blkt == 0) {
/*
 * It is an input and no blocking occurred so the input arrow must
 * go to the end of the system trace as there will be no blocking trace.
 */
		xtrace.xdbt_arrowdir =  XMPI_DBIN;
		xtrace.xdbt_arrowloc = XMPI_DBEND;
	    }
	    else {
		xtrace.xdbt_arrowdir =  XMPI_DBNA;
	    }

	    dbtr = al_append(dbase.xdb_procs[rank].xdbp_traces, &xtrace);
	    if (dbtr == 0) return(LAMERROR);

	    if (xtrace.xdbt_arrowdir == XMPI_DBOUT) {
		sendtr = dbtr;
	    }

	    pp->dbp_systot += xtrace.xdbt_lapse;
	    pp->dbp_end = xtrace.xdbt_time + xtrace.xdbt_lapse;
	}

	if (msg.trm_blkt > 0) {
/*
 * Add a blocking trace since some blocking occurred.
 */
	    xtrace.xdbt_state = XMPI_SBLOCK;
	    xtrace.xdbt_time = (msg.trm_syst == 0) ? trtime : pp->dbp_end;
	    xtrace.xdbt_lapse = (double) msg.trm_blkt / 1000000.0;
	    xtrace.xdbt_blktotal = pp->dbp_blktot;
	    xtrace.xdbt_systotal = pp->dbp_systot;

	    if (type == TRTNOIO) {
		xtrace.xdbt_arrowdir = XMPI_DBNA;
	    }
	    else if (type == TRTINPUT) {
		xtrace.xdbt_arrowloc = XMPI_DBEND;
		xtrace.xdbt_arrowdir = XMPI_DBIN;
	    }
	    else if (msg.trm_syst == 0) {
/*
 * It is an output and there was no system trace.
 */
		xtrace.xdbt_arrowloc = XMPI_DBSTART;
		xtrace.xdbt_arrowdir = XMPI_DBOUT;
	    }
	    else {
		xtrace.xdbt_arrowdir = XMPI_DBNA;
	    }

	    dbtr = al_append(dbase.xdb_procs[rank].xdbp_traces, &xtrace);
	    if (dbtr == 0) return(LAMERROR);

	    if (xtrace.xdbt_arrowdir == XMPI_DBOUT) {
		sendtr = dbtr;
	    }
	    
	    pp->dbp_blktot += xtrace.xdbt_lapse;
	    pp->dbp_end += xtrace.xdbt_lapse;
	}

	if (type == TRTOUTPUT) {
/*
 * This is a send so append to the list of senders to the destination process.
 */
	    dest = xmpi_db_getgpeer(msg.trm_cid, rank, msg.trm_peer);
	    if (dest >= 0) {
		if (al_append(dbp[dest].dbp_senders, &sendtr) == 0) {
		    return(LAMERROR);
		}
	    }
	}
	break;

    default:
	errno = 0;
	xmpi_error(0, "Invalid trace.");
	return(LAMERROR);
    }

    return(0);
}

/*
 *	append_substrate
 *
 *	Function:	- insert substrate trace before last trace and next
 *			  trace if required
 *	Accepts:	- process rank
 *			- start time of next trace
 *	Returns:	- 0 or LAMERROR
 */
static int
append_substrate(rank, nexttime)

int			rank;
double			nexttime;

{
	struct xmdbtr   xtr;			/* database runtime trace */
	struct dbparse  *pp;			/* parsing state */

	pp = dbp + rank;
/*
 * If start time of new trace is greater than the end time of the last trace
 * then add a substrate trace that goes between them.
 */
	memset(&xtr,0,sizeof(xtr));
	if (nexttime > pp->dbp_end) {
		xtr.xdbt_envelop.xdbe_func = pp->dbp_top;
		xtr.xdbt_envelop.xdbe_wfunc = 0;
		xtr.xdbt_envelop.xdbe_lpeer = -1;
		xtr.xdbt_state = pp->dbp_sub;
		xtr.xdbt_arrow = 0;
		xtr.xdbt_arrowdir = XMPI_DBNA;
		xtr.xdbt_senders = 0;
		xtr.xdbt_time = pp->dbp_end;
		xtr.xdbt_lapse = nexttime - pp->dbp_end;
		xtr.xdbt_grank = rank;
		xtr.xdbt_systotal = pp->dbp_systot;
		xtr.xdbt_blktotal = pp->dbp_blktot;

		if (xtr.xdbt_state == XMPI_SSYSTEM) {
			pp->dbp_systot += xtr.xdbt_lapse;
		}
		else if (xtr.xdbt_state == XMPI_SBLOCK) {
			pp->dbp_blktot += xtr.xdbt_lapse;
		}

		if (al_append(dbase.xdb_procs[rank].xdbp_traces, &xtr) == 0) {
			return(LAMERROR);
		}
	} 

  	return(0);
}

/*
 *	db_find_rank
 *
 *	Function:	- finds global rank corresponding to node and pid
 *	Accepts:	- node
 *			- pid
 *	Returns:	- global rank corresponding to node and pid, else (-1)
 */
static int
db_find_rank(node, pid)

int			node;
int			pid;

{
	struct xmdbproc *proc;			/* process */
	int		i;

	proc = dbase.xdb_procs;
	for (i = 0; i < dbase.xdb_nprocs; i++, proc++) {
		if (proc->xdbp_node == node && proc->xdbp_pid == pid) {
			return(i);
		}
	}
			
	return(-1);
}

/*
 *	db_init
 *
 *	Function:	- initialize the database with process info
 *	Accepts:	- number of processes
 *			- array of GPS
 *	Returns:	- 0 or LAMERROR
 */
static int
db_init(nprocs, pgps)

int			nprocs;
struct _gps		*pgps;

{
	struct xmdbproc *proc;			/* process in database */
	int		i;
/*
 * Free current database if any.
 */
	if (dbase.xdb_nprocs != 0) {
		xmpi_db_free();
	}

	dbase.xdb_procs = (struct xmdbproc *)
				malloc(nprocs * sizeof(struct xmdbproc));
	if (dbase.xdb_procs == 0) {
		return(LAMERROR);
	}

	dbase.xdb_cids = al_init(sizeof(struct xmdbcid), cid_cmp);
	if (dbase.xdb_cids == 0) {
		free((char *) dbase.xdb_procs);
		return(LAMERROR);
	}
	
	dbase.xdb_nprocs = nprocs;
	dbase.xdb_mintime = 0.0;
	dbase.xdb_maxtime = 0.0;
	dbase.xdb_curtime = 0.0;
	dbase.xdb_minlapse = DBL_MAX;

	for (i = 0, proc = dbase.xdb_procs; i < nprocs; i++, proc++, pgps++) {
		proc->xdbp_node = pgps->gps_node;
		proc->xdbp_pid = pgps->gps_pid;
		proc->xdbp_prog[0] = 0;
		proc->xdbp_traces = 0;
		proc->xdbp_curtrace = 0;
		proc->xdbp_msgsnd = 0;
		proc->xdbp_dtypes = 0;
	}

	for (i = 0, proc = dbase.xdb_procs; i < nprocs; i++, proc++, pgps++) {
		proc->xdbp_traces = al_init(sizeof(struct xmdbtr), tr_cmp);
		proc->xdbp_dtypes = al_init(sizeof(struct xmdbdt), dtype_cmp);
		proc->xdbp_msgsnd = al_init(sizeof(struct xmdbtr *), 0);
		if (proc->xdbp_traces == 0 || proc->xdbp_dtypes == 0
				|| proc->xdbp_msgsnd == 0) {
			xmpi_db_free();
			return(LAMERROR);
		}
	}

	return(0);
}

/*
 *	xmpi_db_free
 *
 *	Function:	- frees the database
 */
void
xmpi_db_free()

{
	struct xmdbproc *proc;			/* process in database */
	int		i;

	free_parse_structs();

	if (dbase.xdb_nprocs == 0) {
		return;
	}
	
	proc = dbase.xdb_procs;
	for (i = 0; i < dbase.xdb_nprocs; i++, proc++) {
		if (proc->xdbp_dtypes) {
			free_dtypes(proc->xdbp_dtypes);
		}
		if (proc->xdbp_traces) {
			al_free(proc->xdbp_traces);
		}
		if (proc->xdbp_msgsnd) {
			al_free(proc->xdbp_msgsnd);
		}
	}

	free_cids(dbase.xdb_cids);
	free((char *) dbase.xdb_procs);

	dbase.xdb_nprocs = 0;
}

/*
 *	free_cids
 *
 *	Function:	- frees a context ID list
 *	Accepts:	- context ID list
 */
static void
free_cids(cids)

LIST			*cids;

{
	struct xmdbcid	*cid;			/* ptr context ID entry */

	cid = al_top(cids);
	while (cid) {
		free((char *) cid->xdbc_lgrp);
		cid = al_next(cids, cid);
	}
	al_free(cids);
}

/*
 *	free_dtypes
 *
 *	Function:	- frees a datatype list
 *	Accepts:	- datatype list
 */
static void
free_dtypes(dtypes)

LIST			*dtypes;

{
	struct xmdbdt	*dt;			/* ptr datatype entry */

	dt = al_top(dtypes);
	while (dt) {
		free(dt->xdbd_dtbuf);
		dt = al_next(dtypes, dt);
	}
	al_free(dtypes);
}

/*
 *	dtype_cmp
 *
 *	Function:	- compare two datatype entries
 *	Accepts:	- ptr to two entries
 *	Returns:	- -1/0/+1 comparison
 */
static int
dtype_cmp(dt1, dt2)

struct xmdbdt		*dt1, *dt2;

{
	return( (dt1->xdbd_dtype < dt2->xdbd_dtype) ? -1 :
		((dt1->xdbd_dtype > dt2->xdbd_dtype) ? 1 : 0) );
}

/*
 *	tr_cmp
 *
 *	Function:	- compare two traces
 *	Accepts:	- ptr to two entries
 *	Returns:	- -1/0/+1 comparison
 */
static int
tr_cmp(t1, t2)

struct xmdbtr		*t1, *t2;

{
	return( (t1->xdbt_time < t2->xdbt_time) ? -1 :
		((t1->xdbt_time > t2->xdbt_time) ? 1 : 0) );
}

/*
 *	add_cid
 *
 *	Function:	- add a new cid to the database if not already present
 *	Accepts:	- ptr to cid entry
 */
static int
add_cid(pcid)

struct xmdbcid		*pcid;

{
	if (al_find(dbase.xdb_cids, pcid)) {
	  free ((char*) pcid->xdbc_lgrp);
	  return(0);
	}
	return ((al_insert(dbase.xdb_cids, pcid)) ? 0 : LAMERROR);
}

/*
 *	cid_cmp
 *
 *	Function:	- compare two cids
 *			- cids are ordered by context ID and then by
 *			  the global rank of the first process in local group
 *	Accepts:	- ptr to two entries
 *	Returns:	- -1/0/1 comparison
 */
static int
cid_cmp(a, b)

struct xmdbcid		*a, *b;

{
	if (a->xdbc_cid < b->xdbc_cid)
		return(-1);
	else if (a->xdbc_cid > b->xdbc_cid)
		return(1);
	else if (a->xdbc_lgrp->gps_grank < b->xdbc_lgrp->gps_grank)
		return(-1);
	else if (a->xdbc_lgrp->gps_grank == b->xdbc_lgrp->gps_grank)
		return(0);
	else
		return(-1);
}

/*
 *	onoff_cmp
 *
 *	Function:	- compare two onoff list elements
 *			- they are ordered by segment number
 *	Accepts:	- ptr to two entries
 *	Returns:	- -1/0/1 comparison
 */
static int
onoff_cmp(a, b)

struct	onoff		*a, *b;

{
	return( (a->oo_segment < b->oo_segment) ? -1 :
		((a->oo_segment > b->oo_segment) ? 1 : 0) );
}

/*
 *	create_parse_structs
 *
 *	Function:	- create new initial parsing state
 *	Returns:	- initial parsing state or 0 (error)
 */
static struct dbparse *
create_parse_structs()

{
	int		i;

	dbp = (struct dbparse *)
			malloc(dbase.xdb_nprocs * sizeof(struct dbparse));
	if (dbp == 0) {
		return(0);
	}

	for (i = 0; i < dbase.xdb_nprocs; i++) {
		dbp[i].dbp_sub = XMPI_SMISSING;
		dbp[i].dbp_top = 0;
		dbp[i].dbp_curseg = 0;
		dbp[i].dbp_systot = dbp[i].dbp_blktot = 0.0;
		dbp[i].dbp_segs = dbp[i].dbp_senders = 0;
	}

	for (i = 0; i < dbase.xdb_nprocs; i++) {
		dbp[i].dbp_segs = al_init(sizeof(struct onoff), onoff_cmp);
		dbp[i].dbp_senders = al_init(sizeof(struct xmdbtr *), 0);
		if (dbp[i].dbp_segs == 0 || dbp[i].dbp_senders == 0) {
			free_parse_structs();
			return(0);
		}
	}
	
	return(dbp);
}

/*
 *	free_parse_structs
 *
 *	Function:	- free the current parsing state
 */
static void
free_parse_structs()

{
	int		i;

	if (dbp == 0) {
		return;
	}

	for (i = 0; i < dbase.xdb_nprocs; i++) {
		if (dbp[i].dbp_segs) {
			al_free(dbp[i].dbp_segs);
		}
		if (dbp[i].dbp_senders) {
			al_free(dbp[i].dbp_senders);
		}
	}

	free((char *) dbp);
	dbp = 0;
}