Commit | Line | Data |
---|---|---|
c39c72ee | 1 | /* Copyright (C) 2009 Pierre-Marc Fournier |
1f8b0dff | 2 | * |
c39c72ee PMF |
3 | * This library is free software; you can redistribute it and/or |
4 | * modify it under the terms of the GNU Lesser General Public | |
5 | * License as published by the Free Software Foundation; either | |
6 | * version 2.1 of the License, or (at your option) any later version. | |
1f8b0dff | 7 | * |
c39c72ee | 8 | * This library is distributed in the hope that it will be useful, |
1f8b0dff | 9 | * but WITHOUT ANY WARRANTY; without even the implied warranty of |
c39c72ee PMF |
10 | * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
11 | * Lesser General Public License for more details. | |
1f8b0dff | 12 | * |
c39c72ee PMF |
13 | * You should have received a copy of the GNU Lesser General Public |
14 | * License along with this library; if not, write to the Free Software | |
15 | * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA | |
1f8b0dff PMF |
16 | */ |
17 | ||
3796af9b PMF |
18 | #define _GNU_SOURCE |
19 | ||
20 | #include <sys/types.h> | |
21 | #include <sys/shm.h> | |
688760ef PMF |
22 | #include <fcntl.h> |
23 | #include <unistd.h> | |
3a7b90de | 24 | #include <pthread.h> |
a3cdd4a7 | 25 | #include <signal.h> |
3796af9b PMF |
26 | |
27 | #include <stdlib.h> | |
28 | #include <stdio.h> | |
29 | #include <string.h> | |
a3cdd4a7 PMF |
30 | #include <errno.h> |
31 | #include <assert.h> | |
3796af9b | 32 | |
0b0cd937 | 33 | #include "ustd.h" |
3796af9b PMF |
34 | #include "localerr.h" |
35 | #include "ustcomm.h" | |
36 | ||
3a7b90de PMF |
37 | /* return value: 0 = subbuffer is finished, it won't produce data anymore |
38 | * 1 = got subbuffer successfully | |
39 | * <0 = error | |
40 | */ | |
3796af9b | 41 | |
8cefc145 PMF |
42 | #define GET_SUBBUF_OK 1 |
43 | #define GET_SUBBUF_DONE 0 | |
44 | #define GET_SUBBUF_DIED 2 | |
45 | ||
a3cdd4a7 PMF |
46 | #define PUT_SUBBUF_OK 1 |
47 | #define PUT_SUBBUF_DIED 0 | |
48 | #define PUT_SUBBUF_PUSHED 2 | |
49 | ||
50 | int test_sigpipe(void) | |
51 | { | |
52 | sigset_t sigset; | |
53 | int result; | |
54 | ||
55 | result = sigemptyset(&sigset); | |
56 | if(result == -1) { | |
57 | perror("sigemptyset"); | |
58 | return -1; | |
59 | } | |
60 | result = sigaddset(&sigset, SIGPIPE); | |
61 | if(result == -1) { | |
62 | perror("sigaddset"); | |
63 | return -1; | |
64 | } | |
65 | ||
66 | result = sigtimedwait(&sigset, NULL, &(struct timespec){0,0}); | |
67 | if(result == -1 && errno == EAGAIN) { | |
68 | /* no signal received */ | |
69 | return 0; | |
70 | } | |
71 | else if(result == -1) { | |
72 | perror("sigtimedwait"); | |
73 | return -1; | |
74 | } | |
75 | else if(result == SIGPIPE) { | |
76 | /* received sigpipe */ | |
77 | return 1; | |
78 | } | |
79 | else { | |
80 | assert(0); | |
81 | } | |
82 | } | |
83 | ||
688760ef PMF |
84 | int get_subbuffer(struct buffer_info *buf) |
85 | { | |
86 | char *send_msg; | |
87 | char *received_msg; | |
88 | char *rep_code; | |
89 | int retval; | |
90 | int result; | |
91 | ||
92 | asprintf(&send_msg, "get_subbuffer %s", buf->name); | |
3bb56863 | 93 | result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); |
8cefc145 | 94 | free(send_msg); |
a3cdd4a7 PMF |
95 | if(test_sigpipe()) { |
96 | WARN("process %d destroyed before we could connect to it", buf->pid); | |
97 | return GET_SUBBUF_DONE; | |
98 | } | |
99 | else if(result < 0) { | |
3bb56863 | 100 | ERR("get_subbuffer: ustcomm_send_request failed"); |
688760ef PMF |
101 | return -1; |
102 | } | |
8cefc145 PMF |
103 | else if(result == 0) { |
104 | DBG("app died while being traced"); | |
105 | return GET_SUBBUF_DIED; | |
106 | } | |
688760ef PMF |
107 | |
108 | result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old); | |
3a7b90de | 109 | if(result != 2 && result != 1) { |
688760ef PMF |
110 | ERR("unable to parse response to get_subbuffer"); |
111 | return -1; | |
112 | } | |
3a7b90de PMF |
113 | |
114 | DBG("received msg is %s", received_msg); | |
688760ef PMF |
115 | |
116 | if(!strcmp(rep_code, "OK")) { | |
117 | DBG("got subbuffer %s", buf->name); | |
8cefc145 | 118 | retval = GET_SUBBUF_OK; |
688760ef | 119 | } |
3a7b90de | 120 | else if(nth_token_is(received_msg, "END", 0) == 1) { |
8cefc145 | 121 | return GET_SUBBUF_DONE; |
3a7b90de | 122 | } |
688760ef | 123 | else { |
3a7b90de PMF |
124 | DBG("error getting subbuffer %s", buf->name); |
125 | retval = -1; | |
688760ef PMF |
126 | } |
127 | ||
3a7b90de PMF |
128 | /* FIMXE: free correctly the stuff */ |
129 | free(received_msg); | |
688760ef PMF |
130 | free(rep_code); |
131 | return retval; | |
132 | } | |
133 | ||
134 | int put_subbuffer(struct buffer_info *buf) | |
135 | { | |
136 | char *send_msg; | |
137 | char *received_msg; | |
138 | char *rep_code; | |
139 | int retval; | |
140 | int result; | |
141 | ||
142 | asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old); | |
3bb56863 | 143 | result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); |
688760ef PMF |
144 | if(result < 0) { |
145 | ERR("put_subbuffer: send_message failed"); | |
146 | return -1; | |
147 | } | |
148 | free(send_msg); | |
149 | ||
150 | result = sscanf(received_msg, "%as", &rep_code); | |
151 | if(result != 1) { | |
152 | ERR("unable to parse response to put_subbuffer"); | |
153 | return -1; | |
154 | } | |
155 | free(received_msg); | |
156 | ||
157 | if(!strcmp(rep_code, "OK")) { | |
158 | DBG("subbuffer put %s", buf->name); | |
a3cdd4a7 | 159 | retval = PUT_SUBBUF_OK; |
688760ef PMF |
160 | } |
161 | else { | |
a3cdd4a7 PMF |
162 | DBG("put_subbuffer: received error, we were pushed"); |
163 | return PUT_SUBBUF_PUSHED; | |
688760ef PMF |
164 | } |
165 | ||
166 | free(rep_code); | |
167 | return retval; | |
168 | } | |
169 | ||
a3cdd4a7 PMF |
170 | /* This write is patient because it restarts if it was incomplete. |
171 | */ | |
172 | ||
688760ef PMF |
173 | ssize_t patient_write(int fd, const void *buf, size_t count) |
174 | { | |
175 | const char *bufc = (const char *) buf; | |
176 | int result; | |
177 | ||
178 | for(;;) { | |
179 | result = write(fd, bufc, count); | |
180 | if(result <= 0) { | |
181 | return result; | |
182 | } | |
183 | count -= result; | |
184 | bufc += result; | |
185 | ||
186 | if(count == 0) { | |
187 | break; | |
188 | } | |
189 | } | |
190 | ||
191 | return bufc-(const char *)buf; | |
192 | } | |
193 | ||
3a7b90de PMF |
194 | void *consumer_thread(void *arg) |
195 | { | |
196 | struct buffer_info *buf = (struct buffer_info *) arg; | |
197 | int result; | |
198 | ||
199 | for(;;) { | |
8cefc145 | 200 | /* get the subbuffer */ |
0b0cd937 PMF |
201 | result = get_subbuffer(buf); |
202 | if(result == -1) { | |
203 | ERR("error getting subbuffer"); | |
204 | continue; | |
3a7b90de | 205 | } |
0b0cd937 PMF |
206 | else if(result == GET_SUBBUF_DONE) { |
207 | /* this is done */ | |
208 | break; | |
209 | } | |
210 | else if(result == GET_SUBBUF_DIED) { | |
211 | finish_consuming_dead_subbuffer(buf); | |
212 | break; | |
3a7b90de PMF |
213 | } |
214 | ||
215 | /* write data to file */ | |
216 | result = patient_write(buf->file_fd, buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1)), buf->subbuf_size); | |
217 | if(result == -1) { | |
218 | PERROR("write"); | |
219 | /* FIXME: maybe drop this trace */ | |
220 | } | |
221 | ||
8cefc145 | 222 | /* put the subbuffer */ |
0b0cd937 PMF |
223 | result = put_subbuffer(buf); |
224 | if(result == -1) { | |
a3cdd4a7 PMF |
225 | ERR("unknown error putting subbuffer (channel=%s)", buf->name); |
226 | break; | |
227 | } | |
228 | else if(result == PUT_SUBBUF_PUSHED) { | |
229 | ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name); | |
0b0cd937 | 230 | break; |
3a7b90de | 231 | } |
a3cdd4a7 PMF |
232 | else if(result == PUT_SUBBUF_DIED) { |
233 | WARN("application died while putting subbuffer"); | |
234 | /* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */ | |
235 | finish_consuming_dead_subbuffer(buf); | |
236 | } | |
237 | else if(result == PUT_SUBBUF_OK) { | |
238 | } | |
3a7b90de PMF |
239 | } |
240 | ||
241 | DBG("thread for buffer %s is stopping", buf->name); | |
242 | ||
8cefc145 PMF |
243 | /* FIXME: destroy, unalloc... */ |
244 | ||
3a7b90de PMF |
245 | return NULL; |
246 | } | |
247 | ||
72ebd39a PMF |
248 | int create_dir_if_needed(char *dir) |
249 | { | |
250 | int result; | |
251 | result = mkdir(dir, 0777); | |
252 | if(result == -1) { | |
253 | if(errno != EEXIST) { | |
254 | perror("mkdir"); | |
255 | return -1; | |
256 | } | |
257 | } | |
258 | ||
259 | return 0; | |
260 | } | |
261 | ||
3a7b90de PMF |
262 | int add_buffer(pid_t pid, char *bufname) |
263 | { | |
264 | struct buffer_info *buf; | |
265 | char *send_msg; | |
266 | char *received_msg; | |
267 | int result; | |
268 | char *tmp; | |
269 | int fd; | |
270 | pthread_t thr; | |
a3cdd4a7 | 271 | struct shmid_ds shmds; |
3a7b90de PMF |
272 | |
273 | buf = (struct buffer_info *) malloc(sizeof(struct buffer_info)); | |
274 | if(buf == NULL) { | |
275 | ERR("add_buffer: insufficient memory"); | |
276 | return -1; | |
277 | } | |
278 | ||
279 | buf->name = bufname; | |
280 | buf->pid = pid; | |
281 | ||
4e2a8808 PMF |
282 | /* connect to app */ |
283 | result = ustcomm_connect_app(buf->pid, &buf->conn); | |
284 | if(result) { | |
a3cdd4a7 | 285 | WARN("unable to connect to process, it probably died before we were able to connect"); |
4e2a8808 PMF |
286 | return -1; |
287 | } | |
288 | ||
3a7b90de PMF |
289 | /* get shmid */ |
290 | asprintf(&send_msg, "get_shmid %s", buf->name); | |
a3cdd4a7 | 291 | result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); |
3a7b90de | 292 | free(send_msg); |
a3cdd4a7 PMF |
293 | if(result == -1) { |
294 | ERR("problem in ustcomm_send_request(get_shmid)"); | |
295 | return -1; | |
296 | } | |
3a7b90de | 297 | |
8cefc145 PMF |
298 | result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid); |
299 | if(result != 2) { | |
3a7b90de PMF |
300 | ERR("unable to parse response to get_shmid"); |
301 | return -1; | |
302 | } | |
303 | free(received_msg); | |
8cefc145 | 304 | DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid); |
3a7b90de PMF |
305 | |
306 | /* get n_subbufs */ | |
307 | asprintf(&send_msg, "get_n_subbufs %s", buf->name); | |
a3cdd4a7 | 308 | result = ustcomm_send_request(&buf->conn, send_msg, &received_msg); |
3a7b90de | 309 | free(send_msg); |
a3cdd4a7 PMF |
310 | if(result == -1) { |
311 | ERR("problem in ustcomm_send_request(g_n_subbufs)"); | |
312 | return -1; | |
313 | } | |
3a7b90de PMF |
314 | |
315 | result = sscanf(received_msg, "%d", &buf->n_subbufs); | |
316 | if(result != 1) { | |
317 | ERR("unable to parse response to get_n_subbufs"); | |
318 | return -1; | |
319 | } | |
320 | free(received_msg); | |
321 | DBG("got n_subbufs %d", buf->n_subbufs); | |
322 | ||
323 | /* get subbuf size */ | |
324 | asprintf(&send_msg, "get_subbuf_size %s", buf->name); | |
4e2a8808 | 325 | ustcomm_send_request(&buf->conn, send_msg, &received_msg); |
3a7b90de PMF |
326 | free(send_msg); |
327 | ||
328 | result = sscanf(received_msg, "%d", &buf->subbuf_size); | |
329 | if(result != 1) { | |
330 | ERR("unable to parse response to get_subbuf_size"); | |
331 | return -1; | |
332 | } | |
333 | free(received_msg); | |
334 | DBG("got subbuf_size %d", buf->subbuf_size); | |
335 | ||
336 | /* attach memory */ | |
337 | buf->mem = shmat(buf->shmid, NULL, 0); | |
338 | if(buf->mem == (void *) 0) { | |
339 | perror("shmat"); | |
340 | return -1; | |
341 | } | |
8cefc145 PMF |
342 | DBG("successfully attached buffer memory"); |
343 | ||
344 | buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0); | |
345 | if(buf->bufstruct_mem == (void *) 0) { | |
346 | perror("shmat"); | |
347 | return -1; | |
348 | } | |
349 | DBG("successfully attached buffer bufstruct memory"); | |
3a7b90de | 350 | |
a3cdd4a7 PMF |
351 | /* obtain info on the memory segment */ |
352 | result = shmctl(buf->shmid, IPC_STAT, &shmds); | |
353 | if(result == -1) { | |
354 | perror("shmctl"); | |
355 | return -1; | |
356 | } | |
357 | buf->memlen = shmds.shm_segsz; | |
358 | ||
3a7b90de | 359 | /* open file for output */ |
72ebd39a PMF |
360 | result = create_dir_if_needed(USTD_DEFAULT_TRACE_PATH); |
361 | if(result == -1) { | |
362 | ERR("could not create directory %s", USTD_DEFAULT_TRACE_PATH); | |
363 | return -1; | |
364 | } | |
365 | ||
366 | asprintf(&tmp, "%s/%u", USTD_DEFAULT_TRACE_PATH, buf->pid); | |
367 | result = create_dir_if_needed(tmp); | |
368 | if(result == -1) { | |
369 | ERR("could not create directory %s", tmp); | |
370 | free(tmp); | |
371 | return -1; | |
372 | } | |
373 | free(tmp); | |
374 | ||
375 | asprintf(&tmp, "%s/%u/%s_0", USTD_DEFAULT_TRACE_PATH, buf->pid, buf->name); | |
3a7b90de PMF |
376 | result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00600); |
377 | if(result == -1) { | |
378 | PERROR("open"); | |
6cb88bc0 | 379 | ERR("failed opening trace file %s", tmp); |
3a7b90de PMF |
380 | return -1; |
381 | } | |
382 | buf->file_fd = fd; | |
383 | free(tmp); | |
384 | ||
3a7b90de PMF |
385 | pthread_create(&thr, NULL, consumer_thread, buf); |
386 | ||
387 | return 0; | |
388 | } | |
389 | ||
3796af9b PMF |
390 | int main(int argc, char **argv) |
391 | { | |
392 | struct ustcomm_ustd ustd; | |
393 | int result; | |
a3cdd4a7 | 394 | sigset_t sigset; |
3796af9b PMF |
395 | |
396 | result = ustcomm_init_ustd(&ustd); | |
397 | if(result == -1) { | |
398 | ERR("failed to initialize socket"); | |
399 | return 1; | |
400 | } | |
401 | ||
a3cdd4a7 PMF |
402 | result = sigemptyset(&sigset); |
403 | if(result == -1) { | |
404 | perror("sigemptyset"); | |
405 | return 1; | |
406 | } | |
407 | result = sigaddset(&sigset, SIGPIPE); | |
408 | if(result == -1) { | |
409 | perror("sigaddset"); | |
410 | return 1; | |
411 | } | |
412 | result = sigprocmask(SIG_BLOCK, &sigset, NULL); | |
413 | if(result == -1) { | |
414 | perror("sigprocmask"); | |
415 | return 1; | |
416 | } | |
417 | ||
688760ef | 418 | /* app loop */ |
3796af9b PMF |
419 | for(;;) { |
420 | char *recvbuf; | |
421 | ||
3a7b90de | 422 | /* check for requests on our public socket */ |
688760ef PMF |
423 | result = ustcomm_ustd_recv_message(&ustd, &recvbuf, NULL, 100); |
424 | if(result == -1) { | |
425 | ERR("error in ustcomm_ustd_recv_message"); | |
426 | continue; | |
427 | } | |
428 | if(result > 0) { | |
429 | if(!strncmp(recvbuf, "collect", 7)) { | |
430 | pid_t pid; | |
431 | char *bufname; | |
432 | int result; | |
3796af9b | 433 | |
688760ef PMF |
434 | result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname); |
435 | if(result != 2) { | |
436 | fprintf(stderr, "parsing error: %s\n", recvbuf); | |
437 | } | |
3796af9b | 438 | |
688760ef PMF |
439 | result = add_buffer(pid, bufname); |
440 | if(result < 0) { | |
441 | ERR("error in add_buffer"); | |
442 | continue; | |
443 | } | |
3796af9b PMF |
444 | } |
445 | ||
688760ef | 446 | free(recvbuf); |
3796af9b | 447 | } |
3796af9b PMF |
448 | } |
449 | ||
450 | return 0; | |
451 | } |