src/hd/rholo2l.c

#ifndef lint
static const char       RCSid[] = "$Id: rholo2l.c,v 3.21 2024/08/02 18:46:27 greg Exp $";
#endif
/*
 * Routines for local rtrace execution
 */

#include <signal.h>
#include <sys/time.h>
#include <string.h>

#include "rholo.h"
#include "random.h"
#include "paths.h"
#include "selcall.h"
#include "rtprocess.h"

#ifndef MAXPROC
#define MAXPROC         64
#endif

int     nprocs = 0;                             /* running process count */

static char     pfile[] = TEMPLATE;             /* persist file name */

static SUBPROC  rtpd[MAXPROC];                  /* process descriptors */
static float    *rtbuf = NULL;                  /* allocated i/o buffer */
static int      maxqlen;                        /* maximum packets per queue */

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

static int bestout(void);
static int slots_avail(void);
static void queue_packet(PACKET *p);
static PACKET * get_packets(int poll);
static void killpersist(void);


int
start_rtrace(void)                      /* start rtrace process */
{
        static char     buf1[8];
        int     rmaxpack = 0;
        int     psiz, n;
                                        /* get number of processes */
        if (ncprocs <= 0 || nprocs > 0)
                return(0);
        if (ncprocs > MAXPROC) {
                sprintf(errmsg,
                        "number of rtrace processes reduced from %d to %d",
                                ncprocs, MAXPROC);
                error(WARNING, errmsg);
                ncprocs = MAXPROC;
        }
        if (rtargv[rtargc-1] != vval(OCTREE)) {
                                                /* add compulsory options */
                rtargv[rtargc++] = "-i-";
                rtargv[rtargc++] = "-I-";
                rtargv[rtargc++] = "-h-";
                rtargv[rtargc++] = "-ld-";
                rtargv[rtargc++] = "-co-";
                sprintf(buf1, "%d", RPACKSIZ);
                rtargv[rtargc++] = "-x"; rtargv[rtargc++] = buf1;
                rtargv[rtargc++] = "-y"; rtargv[rtargc++] = "0";
                rtargv[rtargc++] = "-fff";
                rtargv[rtargc++] = vbool(VDIST) ? "-ovl" : "-ovL";
                if (nowarn)
                        rtargv[rtargc++] = "-w-";
                if (ncprocs > 1) {
                        mktemp(pfile);
                        rtargv[rtargc++] = "-PP"; rtargv[rtargc++] = pfile;
                }
                rtargv[rtargc++] = vval(OCTREE);
                rtargv[rtargc] = NULL;
        }
        maxqlen = 0;
        for (nprocs = 0; nprocs < ncprocs; nprocs++) {  /* spawn children */
                psiz = open_process(&rtpd[nprocs], rtargv);
                if (psiz <= 0)
                        error(SYSTEM, "cannot start rtrace process");
                n = psiz/(RPACKSIZ*6*sizeof(float)) + 1;
                if (!maxqlen) {
                        if (!(maxqlen = n))
                                error(INTERNAL,
                                        "bad pipe buffer size assumption");
                        sleep(2);
                } 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(void)                   /* get best process to process packet */
{
        int     cnt;
        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 = irandom(cnt)))
                for (i = pn; i < nprocs; i++)
                        if (pqlen[i] == pqlen[pn] && !cnt--)
                                return(i);
        return(pn);
}


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

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


static void
queue_packet(                   /* queue up a beam packet */
        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? */
                memset((char *)(rtbuf+6*n++), '\0', 6*sizeof(float));
        if (writebuf(rtpd[pn].w, 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 {
                PACKET  *rpl = pqueue[pn];
                while (rpl->next != NULL)
                        rpl = rpl->next;
                rpl->next = p;
        }
}


static PACKET *
get_packets(            /* read packets from rtrace processes */
        int     poll
)
{
        static struct timeval   tpoll;  /* zero timeval struct */
        fd_set  readset, errset;
        PACKET  *pldone = NULL, *plend;
        PACKET  *p;
        int     n, nr;
        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].r, &readset);
                FD_SET(rtpd[pn].r, &errset);
                if (rtpd[pn].r >= n)
                        n = rtpd[pn].r + 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].r, &readset) &&
                                !FD_ISSET(rtpd[pn].r, &errset))
                        continue;
        reread:
                n = read(rtpd[pn].r, (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].r,
                                                (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");
        return NULL; /* pro forma return */
}


PACKET *
do_packets(                     /* queue a packet list, return finished */
        PACKET  *pl
)
{
        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(void)                       /* empty all rtrace queues */
{
        PACKET  *rpdone = NULL;
        PACKET  *rpl = NULL;
        float   *bp;
        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].r, 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 void
killpersist(void)                       /* 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(void)                        /* close rtrace process(es) */
{
        int     status = 0, rv;

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