ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/radiance/ray/src/hd/rholo2l.c
Revision: 3.16
Committed: Thu Jan 1 11:21:55 2004 UTC (20 years, 3 months ago) by schorsch
Content type: text/plain
Branch: MAIN
CVS Tags: rad4R2P2, rad5R0, rad3R7P2, rad3R7P1, rad4R2, rad4R1, rad4R0, rad3R6, rad3R6P1, rad3R8, rad3R9, rad4R2P1
Changes since 3.15: +32 -21 lines
Log Message:
Ansification and prototypes.

File Contents

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