ustd: put_subbuf: handle case where ustcomm_send_request returns 0
[ust.git] / ustd / ustd.c
1 /* Copyright (C) 2009 Pierre-Marc Fournier
2 *
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.
7 *
8 * This library is distributed in the hope that it will be useful,
9 * but WITHOUT ANY WARRANTY; without even the implied warranty of
10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
11 * Lesser General Public License for more details.
12 *
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
16 */
17
18 #define _GNU_SOURCE
19
20 #include <sys/types.h>
21 #include <sys/stat.h>
22 #include <sys/shm.h>
23 #include <fcntl.h>
24 #include <unistd.h>
25 #include <pthread.h>
26 #include <signal.h>
27
28 #include <stdlib.h>
29 #include <stdio.h>
30 #include <string.h>
31 #include <errno.h>
32 #include <assert.h>
33 #include <getopt.h>
34
35 #include "ustd.h"
36 #include "usterr.h"
37 #include "ustcomm.h"
38
39 /* return value: 0 = subbuffer is finished, it won't produce data anymore
40 * 1 = got subbuffer successfully
41 * <0 = error
42 */
43
44 #define GET_SUBBUF_OK 1
45 #define GET_SUBBUF_DONE 0
46 #define GET_SUBBUF_DIED 2
47
48 #define PUT_SUBBUF_OK 1
49 #define PUT_SUBBUF_DIED 0
50 #define PUT_SUBBUF_PUSHED 2
51 #define PUT_SUBBUF_DONE 3
52
53 char *sock_path=NULL;
54 char *trace_path=NULL;
55 int daemon_mode = 0;
56 char *pidfile = NULL;
57
58 /* Number of active buffers and the mutex to protect it. */
59 int active_buffers = 0;
60 pthread_mutex_t active_buffers_mutex = PTHREAD_MUTEX_INITIALIZER;
61 /* Whether a request to end the program was received. */
62 sig_atomic_t terminate_req = 0;
63
64 int get_subbuffer(struct buffer_info *buf)
65 {
66 char *send_msg=NULL;
67 char *received_msg=NULL;
68 char *rep_code=NULL;
69 int retval;
70 int result;
71
72 asprintf(&send_msg, "get_subbuffer %s", buf->name);
73 result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
74 if((result == -1 && errno == EPIPE) || result == 0) {
75 DBG("app died while being traced");
76 retval = GET_SUBBUF_DIED;
77 goto end;
78 }
79 else if(result < 0) {
80 ERR("get_subbuffer: ustcomm_send_request failed");
81 retval = -1;
82 goto end;
83 }
84
85 result = sscanf(received_msg, "%as %ld", &rep_code, &buf->consumed_old);
86 if(result != 2 && result != 1) {
87 ERR("unable to parse response to get_subbuffer");
88 retval = -1;
89 goto end_rep;
90 }
91
92 DBG("received msg is %s", received_msg);
93
94 if(!strcmp(rep_code, "OK")) {
95 DBG("got subbuffer %s", buf->name);
96 retval = GET_SUBBUF_OK;
97 }
98 else if(nth_token_is(received_msg, "END", 0) == 1) {
99 retval = GET_SUBBUF_DONE;
100 goto end_rep;
101 }
102 else if(!strcmp(received_msg, "NOTFOUND")) {
103 WARN("For buffer %s, the trace was not found. This likely means it was destroyed by the user.", buf->name);
104 retval = GET_SUBBUF_DONE;
105 goto end_rep;
106 }
107 else {
108 DBG("error getting subbuffer %s", buf->name);
109 retval = -1;
110 }
111
112 /* FIMXE: free correctly the stuff */
113 end_rep:
114 if(rep_code)
115 free(rep_code);
116 end:
117 if(send_msg)
118 free(send_msg);
119 if(received_msg)
120 free(received_msg);
121
122 return retval;
123 }
124
125 int put_subbuffer(struct buffer_info *buf)
126 {
127 char *send_msg=NULL;
128 char *received_msg=NULL;
129 char *rep_code=NULL;
130 int retval;
131 int result;
132
133 asprintf(&send_msg, "put_subbuffer %s %ld", buf->name, buf->consumed_old);
134 result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
135 if(result < 0 && errno == ECONNRESET) {
136 retval = PUT_SUBBUF_DIED;
137 goto end;
138 }
139 else if(result < 0) {
140 ERR("put_subbuffer: send_message failed");
141 retval = -1;
142 goto end;
143 }
144 else if(result == 0) {
145 /* Program seems finished. However this might not be
146 * the last subbuffer that has to be collected.
147 */
148 retval = PUT_SUBBUF_DIED;
149 goto end;
150 }
151
152 result = sscanf(received_msg, "%as", &rep_code);
153 if(result != 1) {
154 ERR("unable to parse response to put_subbuffer");
155 retval = -1;
156 goto end_rep;
157 }
158
159 if(!strcmp(rep_code, "OK")) {
160 DBG("subbuffer put %s", buf->name);
161 retval = PUT_SUBBUF_OK;
162 }
163 else {
164 DBG("put_subbuffer: received error, we were pushed");
165 retval = PUT_SUBBUF_PUSHED;
166 goto end_rep;
167 }
168
169 end_rep:
170 if(rep_code)
171 free(rep_code);
172
173 end:
174 if(send_msg)
175 free(send_msg);
176 if(received_msg)
177 free(received_msg);
178
179 return retval;
180 }
181
182 void decrement_active_buffers(void *arg)
183 {
184 pthread_mutex_lock(&active_buffers_mutex);
185 active_buffers--;
186 pthread_mutex_unlock(&active_buffers_mutex);
187 }
188
189 int create_dir_if_needed(char *dir)
190 {
191 int result;
192 result = mkdir(dir, 0777);
193 if(result == -1) {
194 if(errno != EEXIST) {
195 PERROR("mkdir");
196 return -1;
197 }
198 }
199
200 return 0;
201 }
202
203 int is_directory(const char *dir)
204 {
205 int result;
206 struct stat st;
207
208 result = stat(dir, &st);
209 if(result == -1) {
210 PERROR("stat");
211 return 0;
212 }
213
214 if(!S_ISDIR(st.st_mode)) {
215 return 0;
216 }
217
218 return 1;
219 }
220
221 struct buffer_info *connect_buffer(pid_t pid, const char *bufname)
222 {
223 struct buffer_info *buf;
224 char *send_msg;
225 char *received_msg;
226 int result;
227 char *tmp;
228 int fd;
229 struct shmid_ds shmds;
230
231 buf = (struct buffer_info *) malloc(sizeof(struct buffer_info));
232 if(buf == NULL) {
233 ERR("add_buffer: insufficient memory");
234 return NULL;
235 }
236
237 buf->name = bufname;
238 buf->pid = pid;
239
240 /* connect to app */
241 result = ustcomm_connect_app(buf->pid, &buf->conn);
242 if(result) {
243 WARN("unable to connect to process, it probably died before we were able to connect");
244 return NULL;
245 }
246
247 /* get pidunique */
248 asprintf(&send_msg, "get_pidunique");
249 result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
250 free(send_msg);
251 if(result == -1) {
252 ERR("problem in ustcomm_send_request(get_pidunique)");
253 return NULL;
254 }
255
256 result = sscanf(received_msg, "%lld", &buf->pidunique);
257 if(result != 1) {
258 ERR("unable to parse response to get_pidunique");
259 return NULL;
260 }
261 free(received_msg);
262 DBG("got pidunique %lld", buf->pidunique);
263
264 /* get shmid */
265 asprintf(&send_msg, "get_shmid %s", buf->name);
266 result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
267 free(send_msg);
268 if(result == -1) {
269 ERR("problem in ustcomm_send_request(get_shmid)");
270 return NULL;
271 }
272
273 result = sscanf(received_msg, "%d %d", &buf->shmid, &buf->bufstruct_shmid);
274 if(result != 2) {
275 ERR("unable to parse response to get_shmid (\"%s\")", received_msg);
276 return NULL;
277 }
278 free(received_msg);
279 DBG("got shmids %d %d", buf->shmid, buf->bufstruct_shmid);
280
281 /* get n_subbufs */
282 asprintf(&send_msg, "get_n_subbufs %s", buf->name);
283 result = ustcomm_send_request(&buf->conn, send_msg, &received_msg);
284 free(send_msg);
285 if(result == -1) {
286 ERR("problem in ustcomm_send_request(g_n_subbufs)");
287 return NULL;
288 }
289
290 result = sscanf(received_msg, "%d", &buf->n_subbufs);
291 if(result != 1) {
292 ERR("unable to parse response to get_n_subbufs");
293 return NULL;
294 }
295 free(received_msg);
296 DBG("got n_subbufs %d", buf->n_subbufs);
297
298 /* get subbuf size */
299 asprintf(&send_msg, "get_subbuf_size %s", buf->name);
300 ustcomm_send_request(&buf->conn, send_msg, &received_msg);
301 free(send_msg);
302
303 result = sscanf(received_msg, "%d", &buf->subbuf_size);
304 if(result != 1) {
305 ERR("unable to parse response to get_subbuf_size");
306 return NULL;
307 }
308 free(received_msg);
309 DBG("got subbuf_size %d", buf->subbuf_size);
310
311 /* attach memory */
312 buf->mem = shmat(buf->shmid, NULL, 0);
313 if(buf->mem == (void *) 0) {
314 PERROR("shmat");
315 return NULL;
316 }
317 DBG("successfully attached buffer memory");
318
319 buf->bufstruct_mem = shmat(buf->bufstruct_shmid, NULL, 0);
320 if(buf->bufstruct_mem == (void *) 0) {
321 PERROR("shmat");
322 return NULL;
323 }
324 DBG("successfully attached buffer bufstruct memory");
325
326 /* obtain info on the memory segment */
327 result = shmctl(buf->shmid, IPC_STAT, &shmds);
328 if(result == -1) {
329 PERROR("shmctl");
330 return NULL;
331 }
332 buf->memlen = shmds.shm_segsz;
333
334 /* open file for output */
335 if(!trace_path) {
336 /* Only create the directory if using the default path, because
337 * of the risk of typo when using trace path override. We don't
338 * want to risk creating plenty of useless directories in that case.
339 */
340 result = create_dir_if_needed(USTD_DEFAULT_TRACE_PATH);
341 if(result == -1) {
342 ERR("could not create directory %s", USTD_DEFAULT_TRACE_PATH);
343 return NULL;
344 }
345
346 trace_path = USTD_DEFAULT_TRACE_PATH;
347 }
348
349 asprintf(&tmp, "%s/%u_%lld", trace_path, buf->pid, buf->pidunique);
350 result = create_dir_if_needed(tmp);
351 if(result == -1) {
352 ERR("could not create directory %s", tmp);
353 free(tmp);
354 return NULL;
355 }
356 free(tmp);
357
358 asprintf(&tmp, "%s/%u_%lld/%s", trace_path, buf->pid, buf->pidunique, buf->name);
359 result = fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC | O_EXCL, 00600);
360 if(result == -1) {
361 PERROR("open");
362 ERR("failed opening trace file %s", tmp);
363 return NULL;
364 }
365 buf->file_fd = fd;
366 free(tmp);
367
368 pthread_mutex_lock(&active_buffers_mutex);
369 active_buffers++;
370 pthread_mutex_unlock(&active_buffers_mutex);
371
372 return buf;
373 }
374
375 int write_current_subbuffer(struct buffer_info *buf)
376 {
377 int result;
378
379 void *subbuf_mem = buf->mem + (buf->consumed_old & (buf->n_subbufs * buf->subbuf_size-1));
380
381 size_t cur_sb_size = subbuffer_data_size(subbuf_mem);
382
383 result = patient_write(buf->file_fd, subbuf_mem, cur_sb_size);
384 if(result == -1) {
385 PERROR("write");
386 /* FIXME: maybe drop this trace */
387 return 0;
388 }
389
390 return 0;
391 }
392
393 int consumer_loop(struct buffer_info *buf)
394 {
395 int result;
396
397 pthread_cleanup_push(decrement_active_buffers, NULL);
398
399 for(;;) {
400 /* get the subbuffer */
401 result = get_subbuffer(buf);
402 if(result == -1) {
403 ERR("error getting subbuffer");
404 continue;
405 }
406 else if(result == GET_SUBBUF_DONE) {
407 /* this is done */
408 break;
409 }
410 else if(result == GET_SUBBUF_DIED) {
411 finish_consuming_dead_subbuffer(buf);
412 break;
413 }
414
415 /* write data to file */
416 write_current_subbuffer(buf);
417 /* FIXME: handle return value? */
418
419 /* put the subbuffer */
420 /* FIXME: we actually should unput the buffer before consuming... */
421 result = put_subbuffer(buf);
422 if(result == -1) {
423 ERR("unknown error putting subbuffer (channel=%s)", buf->name);
424 break;
425 }
426 else if(result == PUT_SUBBUF_PUSHED) {
427 ERR("Buffer overflow (channel=%s), reader pushed. This channel will not be usable passed this point.", buf->name);
428 break;
429 }
430 else if(result == PUT_SUBBUF_DIED) {
431 WARN("application died while putting subbuffer");
432 /* FIXME: probably need to skip the first subbuffer in finish_consuming_dead_subbuffer */
433 finish_consuming_dead_subbuffer(buf);
434 break;
435 }
436 else if(result == PUT_SUBBUF_DONE) {
437 /* Done with this subbuffer */
438 /* FIXME: add a case where this branch is used? Upon
439 * normal trace termination, at put_subbuf time, a
440 * special last-subbuffer code could be returned by
441 * the listener.
442 */
443 break;
444 }
445 else if(result == PUT_SUBBUF_OK) {
446 }
447 }
448
449 DBG("thread for buffer %s is stopping", buf->name);
450
451 /* FIXME: destroy, unalloc... */
452
453 pthread_cleanup_pop(1);
454
455 return 0;
456 }
457
458 void free_buffer(struct buffer_info *buf)
459 {
460 }
461
462 struct consumer_thread_args {
463 pid_t pid;
464 const char *bufname;
465 };
466
467 void *consumer_thread(void *arg)
468 {
469 struct buffer_info *buf = (struct buffer_info *) arg;
470 struct consumer_thread_args *args = (struct consumer_thread_args *) arg;
471
472 DBG("GOT ARGS: pid %d bufname %s", args->pid, args->bufname);
473
474 buf = connect_buffer(args->pid, args->bufname);
475 if(buf == NULL) {
476 ERR("failed to connect to buffer");
477 goto end;
478 }
479
480 consumer_loop(buf);
481
482 free_buffer(buf);
483
484 end:
485 /* bufname is free'd in free_buffer() */
486 free(args);
487 return NULL;
488 }
489
490 int start_consuming_buffer(pid_t pid, const char *bufname)
491 {
492 pthread_t thr;
493 struct consumer_thread_args *args;
494
495 DBG("beginning of start_consuming_buffer: args: pid %d bufname %s", pid, bufname);
496
497 args = (struct consumer_thread_args *) malloc(sizeof(struct consumer_thread_args));
498
499 args->pid = pid;
500 args->bufname = strdup(bufname);
501 DBG("beginning2 of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname);
502
503 pthread_create(&thr, NULL, consumer_thread, args);
504 DBG("end of start_consuming_buffer: args: pid %d bufname %s", args->pid, args->bufname);
505
506 return 0;
507 }
508
509 void usage(void)
510 {
511 fprintf(stderr, "Usage:\nustd OPTIONS\n\nOptions:\n"
512 "\t-h\t\tDisplay this usage.\n"
513 "\t-o DIR\t\tSpecify the directory where to output the traces.\n"
514 "\t-s PATH\t\tSpecify the path to use for the daemon socket.\n"
515 "\t-d\t\tStart as a daemon.\n"
516 "\t--pidfile FILE\tWrite the PID in this file (when using -d).\n");
517 }
518
519 int parse_args(int argc, char **argv)
520 {
521 int c;
522
523 while (1) {
524 int option_index = 0;
525 static struct option long_options[] = {
526 {"pidfile", 1, 0, 'p'},
527 {"help", 0, 0, 'h'},
528 {"version", 0, 0, 'V'},
529 {0, 0, 0, 0}
530 };
531
532 c = getopt_long(argc, argv, "hs:o:d", long_options, &option_index);
533 if (c == -1)
534 break;
535
536 switch (c) {
537 case 0:
538 printf("option %s", long_options[option_index].name);
539 if (optarg)
540 printf(" with arg %s", optarg);
541 printf("\n");
542 break;
543 case 's':
544 sock_path = optarg;
545 break;
546 case 'o':
547 trace_path = optarg;
548 if(!is_directory(trace_path)) {
549 ERR("Not a valid directory. (%s)", trace_path);
550 return -1;
551 }
552 break;
553 case 'd':
554 daemon_mode = 1;
555 break;
556 case 'p':
557 pidfile = strdup(optarg);
558 break;
559 case 'h':
560 usage();
561 exit(0);
562 case 'V':
563 printf("Version 0.0\n");
564 break;
565
566 default:
567 /* unknown option or other error; error is
568 printed by getopt, just return */
569 return -1;
570 }
571 }
572
573 return 0;
574 }
575
576 void sigterm_handler(int sig)
577 {
578 terminate_req = 1;
579 }
580
581 static int write_pidfile(const char *file_name, pid_t pid)
582 {
583 FILE *pidfp;
584
585 pidfp = fopen(file_name, "w");
586 if(!pidfp) {
587 PERROR("fopen (%s)", pidfile);
588 WARN("killing child process");
589 return -1;
590 }
591
592 fprintf(pidfp, "%d\n", pid);
593
594 fclose(pidfp);
595
596 return 0;
597 }
598
599 int start_ustd(int fd)
600 {
601 struct ustcomm_ustd ustd;
602 int result;
603 sigset_t sigset;
604 struct sigaction sa;
605
606 result = sigemptyset(&sigset);
607 if(result == -1) {
608 PERROR("sigemptyset");
609 return 1;
610 }
611 sa.sa_handler = sigterm_handler;
612 sa.sa_mask = sigset;
613 sa.sa_flags = SA_RESTART;
614 result = sigaction(SIGTERM, &sa, NULL);
615 if(result == -1) {
616 PERROR("sigaction");
617 return 1;
618 }
619
620 result = ustcomm_init_ustd(&ustd, sock_path);
621 if(result == -1) {
622 ERR("failed to initialize socket");
623 return 1;
624 }
625
626 /* setup handler for SIGPIPE */
627 result = sigemptyset(&sigset);
628 if(result == -1) {
629 PERROR("sigemptyset");
630 return 1;
631 }
632 result = sigaddset(&sigset, SIGPIPE);
633 if(result == -1) {
634 PERROR("sigaddset");
635 return 1;
636 }
637 result = sigprocmask(SIG_BLOCK, &sigset, NULL);
638 if(result == -1) {
639 PERROR("sigprocmask");
640 return 1;
641 }
642
643 /* Write pidfile */
644 if(pidfile) {
645 result = write_pidfile(pidfile, getpid());
646 if(result == -1) {
647 ERR("failed to write pidfile");
648 return 1;
649 }
650 }
651
652 /* Notify parent that we are successfully started. */
653 if(fd != -1) {
654 /* write any one character */
655 result = write(fd, "!", 1);
656 if(result == -1) {
657 PERROR("write");
658 return -1;
659 }
660 if(result != 1) {
661 ERR("Problem sending confirmation of daemon start to parent");
662 return -1;
663 }
664 result = close(fd);
665 if(result == -1) {
666 PERROR("close");
667 }
668 }
669
670 /* app loop */
671 for(;;) {
672 char *recvbuf;
673
674 /* check for requests on our public socket */
675 result = ustcomm_ustd_recv_message(&ustd, &recvbuf, NULL, 100);
676 if(result == -1) {
677 ERR("error in ustcomm_ustd_recv_message");
678 goto loop_end;
679 }
680 if(result > 0) {
681 if(!strncmp(recvbuf, "collect", 7)) {
682 pid_t pid;
683 char *bufname;
684 int result;
685
686 result = sscanf(recvbuf, "%*s %d %50as", &pid, &bufname);
687 if(result != 2) {
688 ERR("parsing error: %s", recvbuf);
689 goto free_bufname;
690 }
691
692 result = start_consuming_buffer(pid, bufname);
693 if(result < 0) {
694 ERR("error in add_buffer");
695 goto free_bufname;
696 }
697
698 free_bufname:
699 free(bufname);
700 }
701
702 free(recvbuf);
703 }
704
705 loop_end:
706
707 if(terminate_req) {
708 pthread_mutex_lock(&active_buffers_mutex);
709 if(active_buffers == 0) {
710 pthread_mutex_unlock(&active_buffers_mutex);
711 break;
712 }
713 pthread_mutex_unlock(&active_buffers_mutex);
714 }
715 }
716
717 return 0;
718 }
719
720 int start_ustd_daemon()
721 {
722 int result;
723 int fd[2];
724 pid_t child_pid;
725
726 result = pipe(fd);
727
728 result = child_pid = fork();
729 if(result == -1) {
730 PERROR("fork");
731 return -1;
732 }
733 else if(result == 0) {
734 return start_ustd(fd[1]);
735 }
736 else {
737 char buf;
738
739 result = read(fd[0], &buf, 1);
740 if(result == -1) {
741 PERROR("read");
742 return -1;
743 }
744 if(result != 1) {
745 ERR("did not receive valid confirmation that the daemon is started");
746 return -1;
747 }
748
749 result = close(fd[0]);
750 if(result == -1) {
751 PERROR("close");
752 }
753
754 DBG("The daemon is now successfully started");
755 }
756
757 /* Wait for confirmation that the server is ready. */
758
759
760 return 0;
761 }
762
763 int main(int argc, char **argv)
764 {
765 int result;
766
767 result = parse_args(argc, argv);
768 if(result == -1) {
769 exit(1);
770 }
771
772 if(daemon_mode) {
773 result = start_ustd_daemon();
774 }
775 else {
776 result = start_ustd(-1);
777 }
778
779 return result;
780 }
This page took 0.044271 seconds and 5 git commands to generate.