src/hd/rholo2l.c

/* Copyright (c) 1997 Silicon Graphics, Inc. */

#ifndef lint
static char SCCSid[] = "$SunId$ SGI";
#endif

/*
 * Routines for local rtrace execution
 */

#include "rholo.h"
#include "random.h"
#include "selcall.h"
#include <signal.h>
#include <sys/time.h>

#ifndef MAXPROC
#define MAXPROC         16
#endif

static char     PFILE[] = "/usr/tmp/RHpersist"; /* persist file name */

static int      rtpd[MAXPROC][3];               /* process descriptors */
static float    *rtbuf = NULL;                  /* allocated i/o buffer */
static int      maxqlen = 0;                    /* maximum packets per queue */
static int      nprocs = 0;                     /* number of processes */

static PACKET   *pqueue[MAXPROC];               /* packet queues */
static int      pqlen[MAXPROC];                 /* packet queue lengths */


int
start_rtrace()                  /* start rtrace process */
{
        static char     buf1[8];
        int     rmaxpack = 0;
        int     psiz, npt, n;
                                        /* get number of processes */
        if ((npt = ncprocs) <= 0)
                return(0);
        if (npt > MAXPROC) {
                sprintf(errmsg,
                        "number of rtrace processes reduced from %d to %d",
                                npt, MAXPROC);
                error(WARNING, errmsg);
                npt = MAXPROC;
        }
                                        /* add compulsory options */
        rtargv[rtargc++] = "-i-";
        rtargv[rtargc++] = "-I-";
        rtargv[rtargc++] = "-h-";
        rtargv[rtargc++] = "-ld-";
        sprintf(buf1, "%d", RPACKSIZ);
        rtargv[rtargc++] = "-x"; rtargv[rtargc++] = buf1;
        rtargv[rtargc++] = "-y"; rtargv[rtargc++] = "0";
        rtargv[rtargc++] = "-fff";
        rtargv[rtargc++] = "-ovl";
        rtargv[rtargc++] = nowarn ? "-w-" : "-w+";
        if (npt > 1) {
                rtargv[rtargc++] = "-PP"; rtargv[rtargc++] = PFILE;
        }
        rtargv[rtargc++] = vval(OCTREE);
        rtargv[rtargc] = NULL;
        maxqlen = 0;
        for (nprocs = 0; nprocs < npt; nprocs++) {      /* spawn children */
                psiz = open_process(rtpd[nprocs], rtargv);
                if (psiz <= 0)
                        error(SYSTEM, "cannot start rtrace process");
                n = psiz/(RPACKSIZ*6*sizeof(float));
                if (maxqlen == 0) {
                        if (!(maxqlen = n))
                                error(INTERNAL,
                                        "bad pipe buffer size assumption");
                } else if (n != maxqlen)
                        error(INTERNAL, "varying pipe buffer size!");
                rmaxpack += n;
        }
        rtbuf = (float *)malloc(RPACKSIZ*6*sizeof(float)*maxqlen);
        if (rtbuf == NULL)
                error(SYSTEM, "malloc failure in start_rtrace");
        return(rmaxpack);
}


static int
bestout()                       /* get best process to process packet */
{
        int     cnt;
        register int    pn, i;

        pn = 0;                 /* find shortest queue */
        for (i = 1; i < nprocs; i++)
                if (pqlen[i] < pqlen[pn])
                        pn = i;
                                /* sanity check */
        if (pqlen[pn] == maxqlen)
                return(-1);
        cnt = 0;                /* count number of ties */
        for (i = pn; i < nprocs; i++)
                if (pqlen[i] == pqlen[pn])
                        cnt++;
                                /* break ties fairly */
        if ((cnt = random() % cnt))
                for (i = pn; i < nprocs; i++)
                        if (pqlen[i] == pqlen[pn] && !cnt--)
                                return(i);
        return(pn);
}


int
slots_avail()                   /* count packet slots available */
{
        register int    nslots = 0;
        register int    i;

        for (i = nprocs; i--; )
                nslots += maxqlen - pqlen[i];
        return(nslots);
}


queue_packet(p)                 /* queue up a beam packet */
register PACKET *p;
{
        int     pn, n;
                                /* determine process to write to */
        if ((pn = bestout()) < 0)
                error(INTERNAL, "rtrace input queues are full!");
                                /* write out the packet */
        packrays(rtbuf, p);
        if ((n = p->nr) < RPACKSIZ)     /* add flush block? */
                bzero((char *)(rtbuf+6*n++), 6*sizeof(float));
        if (writebuf(rtpd[pn][1], (char *)rtbuf, 6*sizeof(float)*n) < 0)
                error(SYSTEM, "write error in queue_packet");
        p->next = NULL;
        if (!pqlen[pn]++)       /* add it to the end of the queue */
                pqueue[pn] = p;
        else {
                register PACKET *rpl = pqueue[pn];
                while (rpl->next != NULL)
                        rpl = rpl->next;
                rpl->next = p;
        }
}


PACKET *
get_packets(poll)               /* read packets from rtrace processes */
int     poll;
{
        static struct timeval   tpoll;  /* zero timeval struct */
        fd_set  readset, errset;
        PACKET  *pldone = NULL, *plend;
        register PACKET *p;
        int     n, nr;
        register int    pn;
        float   *bp;
                                        /* prepare select call */
        FD_ZERO(&readset); FD_ZERO(&errset); n = 0;
        for (pn = nprocs; pn--; ) {
                if (pqlen[pn])
                        FD_SET(rtpd[pn][0], &readset);
                FD_SET(rtpd[pn][0], &errset);
                if (rtpd[pn][0] >= n)
                        n = rtpd[pn][0] + 1;
        }
                                        /* make the call */
        n = select(n, &readset, (fd_set *)NULL, &errset,
                        poll ? &tpoll : (struct timeval *)NULL);
        if (n < 0) {
                if (errno == EINTR)     /* interrupted select call */
                        return(NULL);
                error(SYSTEM, "select call failure in get_packets");
        }
        if (n == 0)                     /* is nothing ready? */
                return(NULL);
                                        /* make read call(s) */
        for (pn = 0; pn < nprocs; pn++) {
                if (!FD_ISSET(rtpd[pn][0], &readset) &&
                                !FD_ISSET(rtpd[pn][0], &errset))
                        continue;
        reread:
                n = read(rtpd[pn][0], (char *)rtbuf,
                                4*sizeof(float)*RPACKSIZ*pqlen[pn]);
                if (n < 0) {
                        if (errno == EINTR | errno == EAGAIN)
                                goto reread;
                        error(SYSTEM, "read error in get_packets");
                }
                if (n == 0)
                        goto eoferr;
                bp = rtbuf;                             /* finish processing */
                for (p = pqueue[pn]; n && p != NULL; p = p->next) {
                        if ((nr = p->nr) < RPACKSIZ)
                                nr++;                   /* add flush block */
                        n -= 4*sizeof(float)*nr;
                        if (n < 0) {                    /* get remainder */
                                n += readbuf(rtpd[pn][0],
                                                (char *)(bp+4*nr)+n, -n);
                                if (n)
                                        goto eoferr;
                        }
                        donerays(p, bp);
                        bp += 4*nr;
                        pqlen[pn]--;
                }
                if (n)                                  /* read past end? */
                        error(INTERNAL, "packet sync error in get_packets");
                                                        /* take from queue */
                if (pldone == NULL)
                        pldone = plend = pqueue[pn];
                else
                        plend->next = pqueue[pn];
                while (plend->next != p)
                        plend = plend->next;
                plend->next = NULL;
                pqueue[pn] = p;
        }
        return(pldone);                         /* return finished packets */
eoferr:
        error(USER, "rtrace process died");
}


PACKET *
do_packets(pl)                  /* queue a packet list, return finished */
register PACKET *pl;
{
        register PACKET *p;
                                        /* consistency check */
        if (nprocs < 1)
                error(CONSISTENCY, "do_packets called with no active process");
                                        /* queue each new packet */
        while (pl != NULL) {
                p = pl; pl = p->next; p->next = NULL;
                queue_packet(p);
        }
        return(get_packets(slots_avail()));     /* return processed packets */
}


PACKET *
flush_queue()                   /* empty all rtrace queues */
{
        PACKET  *rpdone = NULL;
        register PACKET *rpl;
        float   *bp;
        register PACKET *p;
        int     i, n, nr;

        for (i = 0; i < nprocs; i++)
                if (pqlen[i]) {
                        if (rpdone == NULL) {           /* tack on queue */
                                rpdone = rpl = pqueue[i];
                                if ((nr = rpl->nr) < RPACKSIZ) nr++;
                        } else {
                                rpl->next = pqueue[i];
                                nr = 0;
                        }
                        while (rpl->next != NULL) {
                                nr += (rpl = rpl->next)->nr;
                                if (rpl->nr < RPACKSIZ)
                                        nr++;           /* add flush block */
                        }
                        n = readbuf(rtpd[i][0], (char *)rtbuf,
                                        4*sizeof(float)*nr);
                        if (n < 0)
                                error(SYSTEM, "read failure in flush_queue");
                        bp = rtbuf;                     /* process packets */
                        for (p = pqueue[i]; p != NULL; p = p->next) {
                                if ((nr = p->nr) < RPACKSIZ)
                                        nr++;           /* add flush block */
                                n -= 4*sizeof(float)*nr;
                                if (n >= 0)
                                        donerays(p, bp);
                                else
                                        p->nr = 0;      /* short data error */
                                bp += 4*nr;
                        }
                        pqueue[i] = NULL;               /* zero this queue */
                        pqlen[i] = 0;
                }
        return(rpdone);         /* return all packets completed */
}


static
killpersist()                   /* kill persistent process */
{
        FILE    *fp;
        int     pid;

        if ((fp = fopen(PFILE, "r")) == NULL)
                return;
        if (fscanf(fp, "%*s %d", &pid) != 1 || kill(pid, SIGALRM) < 0)
                unlink(PFILE);
        fclose(fp);
}


int
end_rtrace()                    /* close rtrace process(es) */
{
        int     status = 0, rv;

        if (nprocs > 1)
                killpersist();
        while (nprocs > 0) {
                rv = close_process(rtpd[--nprocs]);
                if (rv > 0)
                        status = rv;
        }
        free((char *)rtbuf);
        rtbuf = NULL;
        maxqlen = 0;
        return(status);
}
Revision:	3.4
Committed:	Mon Nov 3 18:33:13 1997 UTC (26 years, 5 months ago) by gregl
Content type:	text/plain
Branch:	MAIN
Changes since 3.3:	+1 -1 lines
Log Message:	added more routines for display support
#	User	Rev	Content
1	gregl	3.1	/* Copyright (c) 1997 Silicon Graphics, Inc. */
2
3			#ifndef lint
4			static char SCCSid[] = "$SunId$ SGI";
5			#endif
6
7			/*
8			* Routines for local rtrace execution
9			*/
10
11			#include "rholo.h"
12			#include "random.h"
13	gregl	3.3	#include "selcall.h"
14	gregl	3.1	#include <signal.h>
15			#include <sys/time.h>
16
17			#ifndef MAXPROC
18			#define MAXPROC 16
19			#endif
20
21			static char PFILE[] = "/usr/tmp/RHpersist"; /* persist file name */
22
23			static int rtpd[MAXPROC][3]; /* process descriptors */
24			static float rtbuf = NULL; / allocated i/o buffer */
25			static int maxqlen = 0; /* maximum packets per queue */
26			static int nprocs = 0; /* number of processes */
27
28			static PACKET pqueue[MAXPROC]; / packet queues */
29			static int pqlen[MAXPROC]; /* packet queue lengths */
30
31
32			int
33			start_rtrace() /* start rtrace process */
34			{
35			static char buf1[8];
36			int rmaxpack = 0;
37			int psiz, npt, n;
38			/* get number of processes */
39			if ((npt = ncprocs) <= 0)
40			return(0);
41			if (npt > MAXPROC) {
42			sprintf(errmsg,
43			"number of rtrace processes reduced from %d to %d",
44			npt, MAXPROC);
45			error(WARNING, errmsg);
46			npt = MAXPROC;
47			}
48			/* add compulsory options */
49			rtargv[rtargc++] = "-i-";
50			rtargv[rtargc++] = "-I-";
51			rtargv[rtargc++] = "-h-";
52			rtargv[rtargc++] = "-ld-";
53			sprintf(buf1, "%d", RPACKSIZ);
54			rtargv[rtargc++] = "-x"; rtargv[rtargc++] = buf1;
55			rtargv[rtargc++] = "-y"; rtargv[rtargc++] = "0";
56			rtargv[rtargc++] = "-fff";
57			rtargv[rtargc++] = "-ovl";
58			rtargv[rtargc++] = nowarn ? "-w-" : "-w+";
59			if (npt > 1) {
60			rtargv[rtargc++] = "-PP"; rtargv[rtargc++] = PFILE;
61			}
62			rtargv[rtargc++] = vval(OCTREE);
63			rtargv[rtargc] = NULL;
64			maxqlen = 0;
65			for (nprocs = 0; nprocs < npt; nprocs++) { /* spawn children */
66			psiz = open_process(rtpd[nprocs], rtargv);
67			if (psiz <= 0)
68			error(SYSTEM, "cannot start rtrace process");
69			n = psiz/(RPACKSIZ6sizeof(float));
70			if (maxqlen == 0) {
71			if (!(maxqlen = n))
72			error(INTERNAL,
73			"bad pipe buffer size assumption");
74			} else if (n != maxqlen)
75			error(INTERNAL, "varying pipe buffer size!");
76			rmaxpack += n;
77			}
78			rtbuf = (float )malloc(RPACKSIZ6sizeof(float)maxqlen);
79			if (rtbuf == NULL)
80			error(SYSTEM, "malloc failure in start_rtrace");
81			return(rmaxpack);
82			}
83
84
85			static int
86			bestout() /* get best process to process packet */
87			{
88			int cnt;
89			register int pn, i;
90
91			pn = 0; /* find shortest queue */
92			for (i = 1; i < nprocs; i++)
93			if (pqlen[i] < pqlen[pn])
94			pn = i;
95			/* sanity check */
96			if (pqlen[pn] == maxqlen)
97			return(-1);
98			cnt = 0; /* count number of ties */
99			for (i = pn; i < nprocs; i++)
100			if (pqlen[i] == pqlen[pn])
101			cnt++;
102			/* break ties fairly */
103			if ((cnt = random() % cnt))
104			for (i = pn; i < nprocs; i++)
105			if (pqlen[i] == pqlen[pn] && !cnt--)
106			return(i);
107			return(pn);
108			}
109
110
111			int
112			slots_avail() /* count packet slots available */
113			{
114			register int nslots = 0;
115			register int i;
116
117			for (i = nprocs; i--; )
118			nslots += maxqlen - pqlen[i];
119			return(nslots);
120			}
121
122
123			queue_packet(p) /* queue up a beam packet */
124			register PACKET *p;
125			{
126			int pn, n;
127			/* determine process to write to */
128			if ((pn = bestout()) < 0)
129			error(INTERNAL, "rtrace input queues are full!");
130			/* write out the packet */
131			packrays(rtbuf, p);
132			if ((n = p->nr) < RPACKSIZ) /* add flush block? */
133			bzero((char )(rtbuf+6n++), 6*sizeof(float));
134			if (writebuf(rtpd[pn][1], (char )rtbuf, 6sizeof(float)*n) < 0)
135			error(SYSTEM, "write error in queue_packet");
136			p->next = NULL;
137			if (!pqlen[pn]++) /* add it to the end of the queue */
138			pqueue[pn] = p;
139			else {
140			register PACKET *rpl = pqueue[pn];
141			while (rpl->next != NULL)
142			rpl = rpl->next;
143			rpl->next = p;
144			}
145			}
146
147
148			PACKET *
149			get_packets(poll) /* read packets from rtrace processes */
150			int poll;
151			{
152			static struct timeval tpoll; /* zero timeval struct */
153			fd_set readset, errset;
154			PACKET pldone = NULL, plend;
155			register PACKET *p;
156			int n, nr;
157			register int pn;
158			float *bp;
159			/* prepare select call */
160			FD_ZERO(&readset); FD_ZERO(&errset); n = 0;
161			for (pn = nprocs; pn--; ) {
162			if (pqlen[pn])
163			FD_SET(rtpd[pn][0], &readset);
164			FD_SET(rtpd[pn][0], &errset);
165			if (rtpd[pn][0] >= n)
166			n = rtpd[pn][0] + 1;
167			}
168			/* make the call */
169			n = select(n, &readset, (fd_set *)NULL, &errset,
170			poll ? &tpoll : (struct timeval *)NULL);
171			if (n < 0) {
172			if (errno == EINTR) /* interrupted select call */
173			return(NULL);
174			error(SYSTEM, "select call failure in get_packets");
175			}
176			if (n == 0) /* is nothing ready? */
177			return(NULL);
178			/* make read call(s) */
179			for (pn = 0; pn < nprocs; pn++) {
180			if (!FD_ISSET(rtpd[pn][0], &readset) &&
181			!FD_ISSET(rtpd[pn][0], &errset))
182			continue;
183			reread:
184			n = read(rtpd[pn][0], (char *)rtbuf,
185			4sizeof(float)RPACKSIZ*pqlen[pn]);
186			if (n < 0) {
187			if (errno == EINTR \| errno == EAGAIN)
188			goto reread;
189			error(SYSTEM, "read error in get_packets");
190			}
191			if (n == 0)
192			goto eoferr;
193			bp = rtbuf; /* finish processing */
194			for (p = pqueue[pn]; n && p != NULL; p = p->next) {
195			if ((nr = p->nr) < RPACKSIZ)
196			nr++; /* add flush block */
197			n -= 4sizeof(float)nr;
198			if (n < 0) { /* get remainder */
199			n += readbuf(rtpd[pn][0],
200			(char )(bp+4nr)+n, -n);
201			if (n)
202			goto eoferr;
203			}
204			donerays(p, bp);
205			bp += 4*nr;
206			pqlen[pn]--;
207			}
208			if (n) /* read past end? */
209			error(INTERNAL, "packet sync error in get_packets");
210			/* take from queue */
211	gregl	3.2	if (pldone == NULL)
212	gregl	3.1	pldone = plend = pqueue[pn];
213			else
214			plend->next = pqueue[pn];
215			while (plend->next != p)
216			plend = plend->next;
217			plend->next = NULL;
218			pqueue[pn] = p;
219			}
220			return(pldone); /* return finished packets */
221			eoferr:
222	gregl	3.3	error(USER, "rtrace process died");
223	gregl	3.1	}
224
225
226			PACKET *
227			do_packets(pl) /* queue a packet list, return finished */
228			register PACKET *pl;
229			{
230			register PACKET *p;
231			/* consistency check */
232			if (nprocs < 1)
233			error(CONSISTENCY, "do_packets called with no active process");
234			/* queue each new packet */
235			while (pl != NULL) {
236			p = pl; pl = p->next; p->next = NULL;
237			queue_packet(p);
238			}
239			return(get_packets(slots_avail())); /* return processed packets */
240			}
241
242
243			PACKET *
244			flush_queue() /* empty all rtrace queues */
245			{
246			PACKET *rpdone = NULL;
247			register PACKET *rpl;
248			float *bp;
249			register PACKET *p;
250			int i, n, nr;
251
252			for (i = 0; i < nprocs; i++)
253			if (pqlen[i]) {
254			if (rpdone == NULL) { /* tack on queue */
255			rpdone = rpl = pqueue[i];
256	gregl	3.4	if ((nr = rpl->nr) < RPACKSIZ) nr++;
257	gregl	3.1	} else {
258			rpl->next = pqueue[i];
259	gregl	3.2	nr = 0;
260	gregl	3.1	}
261	gregl	3.2	while (rpl->next != NULL) {
262			nr += (rpl = rpl->next)->nr;
263			if (rpl->nr < RPACKSIZ)
264			nr++; /* add flush block */
265			}
266	gregl	3.1	n = readbuf(rtpd[i][0], (char *)rtbuf,
267	gregl	3.2	4sizeof(float)nr);
268	gregl	3.1	if (n < 0)
269			error(SYSTEM, "read failure in flush_queue");
270			bp = rtbuf; /* process packets */
271			for (p = pqueue[i]; p != NULL; p = p->next) {
272			if ((nr = p->nr) < RPACKSIZ)
273			nr++; /* add flush block */
274			n -= 4sizeof(float)nr;
275			if (n >= 0)
276			donerays(p, bp);
277			else
278			p->nr = 0; /* short data error */
279			bp += 4*nr;
280			}
281			pqueue[i] = NULL; /* zero this queue */
282			pqlen[i] = 0;
283			}
284			return(rpdone); /* return all packets completed */
285			}
286
287
288			static
289			killpersist() /* kill persistent process */
290			{
291			FILE *fp;
292			int pid;
293
294			if ((fp = fopen(PFILE, "r")) == NULL)
295			return;
296			if (fscanf(fp, "%*s %d", &pid) != 1 \|\| kill(pid, SIGALRM) < 0)
297			unlink(PFILE);
298			fclose(fp);
299			}
300
301
302			int
303			end_rtrace() /* close rtrace process(es) */
304			{
305			int status = 0, rv;
306
307			if (nprocs > 1)
308			killpersist();
309			while (nprocs > 0) {
310			rv = close_process(rtpd[--nprocs]);
311			if (rv > 0)
312			status = rv;
313			}
314			free((char *)rtbuf);
315			rtbuf = NULL;
316			maxqlen = 0;
317			return(status);
318			}