ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/radiance/ray/src/hd/rholo2l.c
Revision: 3.15
Committed: Sun Jul 27 22:12:02 2003 UTC (20 years, 8 months ago) by schorsch
Content type: text/plain
Branch: MAIN
Changes since 3.14: +2 -2 lines
Log Message:
Added grouping parens to reduce ambiguity warnings.

File Contents

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