b00142639098c859b17e22d1c8f94c2db8f39312
[ltt-control.git] / tags / ltt-control-0.51-12082008 / lttd / lttd.c
1 /* lttd
2 *
3 * Linux Trace Toolkit Daemon
4 *
5 * This is a simple daemon that reads a few relay+debugfs channels and save
6 * them in a trace.
7 *
8 * CPU hot-plugging is supported using inotify.
9 *
10 * Copyright 2005 -
11 * Mathieu Desnoyers <mathieu.desnoyers@polymtl.ca>
12 */
13
14 #ifdef HAVE_CONFIG_H
15 #include <config.h>
16 #endif
17
18 #define _REENTRANT
19 #define _GNU_SOURCE
20 #include <features.h>
21 #include <stdio.h>
22 #include <unistd.h>
23 #include <errno.h>
24 #include <sys/types.h>
25 #include <sys/stat.h>
26 #include <stdlib.h>
27 #include <dirent.h>
28 #include <string.h>
29 #include <fcntl.h>
30 #include <sys/poll.h>
31 #include <sys/mman.h>
32 #include <signal.h>
33 #include <pthread.h>
34 #include <sys/syscall.h>
35 #include <unistd.h>
36 #include <asm/ioctls.h>
37
38 #include <linux/version.h>
39
40 /* Relayfs IOCTL */
41 #include <asm/ioctl.h>
42 #include <asm/types.h>
43
44 /* Get the next sub buffer that can be read. */
45 #define RELAY_GET_SUBBUF _IOR(0xF5, 0x00,__u32)
46 /* Release the oldest reserved (by "get") sub buffer. */
47 #define RELAY_PUT_SUBBUF _IOW(0xF5, 0x01,__u32)
48 /* returns the number of sub buffers in the per cpu channel. */
49 #define RELAY_GET_N_SUBBUFS _IOR(0xF5, 0x02,__u32)
50 /* returns the size of the sub buffers. */
51 #define RELAY_GET_SUBBUF_SIZE _IOR(0xF5, 0x03,__u32)
52
53 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,14)
54 #include <linux/inotify.h>
55 /* From the inotify-tools 2.6 package */
56 static inline int inotify_init (void)
57 {
58 return syscall (__NR_inotify_init);
59 }
60
61 static inline int inotify_add_watch (int fd, const char *name, __u32 mask)
62 {
63 return syscall (__NR_inotify_add_watch, fd, name, mask);
64 }
65
66 static inline int inotify_rm_watch (int fd, __u32 wd)
67 {
68 return syscall (__NR_inotify_rm_watch, fd, wd);
69 }
70 #define HAS_INOTIFY
71 #else
72 static inline int inotify_init (void)
73 {
74 return -1;
75 }
76
77 static inline int inotify_add_watch (int fd, const char *name, __u32 mask)
78 {
79 return 0;
80 }
81
82 static inline int inotify_rm_watch (int fd, __u32 wd)
83 {
84 return 0;
85 }
86 #undef HAS_INOTIFY
87 #endif
88
89 enum {
90 GET_SUBBUF,
91 PUT_SUBBUF,
92 GET_N_BUBBUFS,
93 GET_SUBBUF_SIZE
94 };
95
96 struct fd_pair {
97 int channel;
98 int trace;
99 unsigned int n_subbufs;
100 unsigned int subbuf_size;
101 void *mmap;
102 pthread_mutex_t mutex;
103 };
104
105 struct channel_trace_fd {
106 struct fd_pair *pair;
107 int num_pairs;
108 };
109
110 struct inotify_watch {
111 int wd;
112 char path_channel[PATH_MAX];
113 char path_trace[PATH_MAX];
114 };
115
116 struct inotify_watch_array {
117 struct inotify_watch *elem;
118 int num;
119 };
120
121
122
123 struct channel_trace_fd fd_pairs = { NULL, 0 };
124 int inotify_fd = -1;
125 struct inotify_watch_array inotify_watch_array = { NULL, 0 };
126
127 /* protects fd_pairs and inotify_watch_array */
128 pthread_rwlock_t fd_pairs_lock = PTHREAD_RWLOCK_INITIALIZER;
129
130
131 static char *trace_name = NULL;
132 static char *channel_name = NULL;
133 static int daemon_mode = 0;
134 static int append_mode = 0;
135 static unsigned long num_threads = 1;
136 volatile static int quit_program = 0; /* For signal handler */
137 static int dump_flight_only = 0;
138 static int dump_normal_only = 0;
139
140 /* Args :
141 *
142 * -t directory Directory name of the trace to write to. Will be created.
143 * -c directory Root directory of the debugfs trace channels.
144 * -d Run in background (daemon).
145 * -a Trace append mode.
146 * -s Send SIGUSR1 to parent when ready for IO.
147 */
148 void show_arguments(void)
149 {
150 printf("Please use the following arguments :\n");
151 printf("\n");
152 printf("-t directory Directory name of the trace to write to.\n"
153 " It will be created.\n");
154 printf("-c directory Root directory of the debugfs trace channels.\n");
155 printf("-d Run in background (daemon).\n");
156 printf("-a Append to an possibly existing trace.\n");
157 printf("-N Number of threads to start.\n");
158 printf("-f Dump only flight recorder channels.\n");
159 printf("-n Dump only normal channels.\n");
160 printf("\n");
161 }
162
163
164 /* parse_arguments
165 *
166 * Parses the command line arguments.
167 *
168 * Returns 1 if the arguments were correct, but doesn't ask for program
169 * continuation. Returns -1 if the arguments are incorrect, or 0 if OK.
170 */
171 int parse_arguments(int argc, char **argv)
172 {
173 int ret = 0;
174 int argn = 1;
175
176 if(argc == 2) {
177 if(strcmp(argv[1], "-h") == 0) {
178 return 1;
179 }
180 }
181
182 while(argn < argc) {
183
184 switch(argv[argn][0]) {
185 case '-':
186 switch(argv[argn][1]) {
187 case 't':
188 if(argn+1 < argc) {
189 trace_name = argv[argn+1];
190 argn++;
191 }
192 break;
193 case 'c':
194 if(argn+1 < argc) {
195 channel_name = argv[argn+1];
196 argn++;
197 }
198 break;
199 case 'd':
200 daemon_mode = 1;
201 break;
202 case 'a':
203 append_mode = 1;
204 break;
205 case 'N':
206 if(argn+1 < argc) {
207 num_threads = strtoul(argv[argn+1], NULL, 0);
208 argn++;
209 }
210 break;
211 case 'f':
212 dump_flight_only = 1;
213 break;
214 case 'n':
215 dump_normal_only = 1;
216 break;
217 default:
218 printf("Invalid argument '%s'.\n", argv[argn]);
219 printf("\n");
220 ret = -1;
221 }
222 break;
223 default:
224 printf("Invalid argument '%s'.\n", argv[argn]);
225 printf("\n");
226 ret = -1;
227 }
228 argn++;
229 }
230
231 if(trace_name == NULL) {
232 printf("Please specify a trace name.\n");
233 printf("\n");
234 ret = -1;
235 }
236
237 if(channel_name == NULL) {
238 printf("Please specify a channel name.\n");
239 printf("\n");
240 ret = -1;
241 }
242
243 return ret;
244 }
245
246 void show_info(void)
247 {
248 printf("Linux Trace Toolkit Trace Daemon " VERSION "\n");
249 printf("\n");
250 printf("Reading from debugfs directory : %s\n", channel_name);
251 printf("Writing to trace directory : %s\n", trace_name);
252 printf("\n");
253 }
254
255
256 /* signal handling */
257
258 static void handler(int signo)
259 {
260 printf("Signal %d received : exiting cleanly\n", signo);
261 quit_program = 1;
262 }
263
264
265 int open_buffer_file(char *filename, char *path_channel, char *path_trace,
266 struct channel_trace_fd *fd_pairs)
267 {
268 int open_ret = 0;
269 int ret = 0;
270 struct stat stat_buf;
271
272 if(strncmp(filename, "flight-", sizeof("flight-")-1) != 0) {
273 if(dump_flight_only) {
274 printf("Skipping normal channel %s\n", path_channel);
275 return 0;
276 }
277 } else {
278 if(dump_normal_only) {
279 printf("Skipping flight channel %s\n", path_channel);
280 return 0;
281 }
282 }
283 printf("Opening file.\n");
284
285 fd_pairs->pair = realloc(fd_pairs->pair,
286 ++fd_pairs->num_pairs * sizeof(struct fd_pair));
287
288 /* Open the channel in read mode */
289 fd_pairs->pair[fd_pairs->num_pairs-1].channel =
290 open(path_channel, O_RDONLY | O_NONBLOCK);
291 if(fd_pairs->pair[fd_pairs->num_pairs-1].channel == -1) {
292 perror(path_channel);
293 fd_pairs->num_pairs--;
294 return 0; /* continue */
295 }
296 /* Open the trace in write mode, only append if append_mode */
297 ret = stat(path_trace, &stat_buf);
298 if(ret == 0) {
299 if(append_mode) {
300 printf("Appending to file %s as requested\n", path_trace);
301
302 fd_pairs->pair[fd_pairs->num_pairs-1].trace =
303 open(path_trace, O_WRONLY|O_APPEND,
304 S_IRWXU|S_IRWXG|S_IRWXO);
305
306 if(fd_pairs->pair[fd_pairs->num_pairs-1].trace == -1) {
307 perror(path_trace);
308 }
309 } else {
310 printf("File %s exists, cannot open. Try append mode.\n", path_trace);
311 open_ret = -1;
312 goto end;
313 }
314 } else {
315 if(errno == ENOENT) {
316 fd_pairs->pair[fd_pairs->num_pairs-1].trace =
317 open(path_trace, O_WRONLY|O_CREAT|O_EXCL,
318 S_IRWXU|S_IRWXG|S_IRWXO);
319 if(fd_pairs->pair[fd_pairs->num_pairs-1].trace == -1) {
320 perror(path_trace);
321 }
322 }
323 }
324 end:
325 return open_ret;
326 }
327
328 int open_channel_trace_pairs(char *subchannel_name, char *subtrace_name,
329 struct channel_trace_fd *fd_pairs, int *inotify_fd,
330 struct inotify_watch_array *iwatch_array)
331 {
332 DIR *channel_dir = opendir(subchannel_name);
333 struct dirent *entry;
334 struct stat stat_buf;
335 int ret;
336 char path_channel[PATH_MAX];
337 int path_channel_len;
338 char *path_channel_ptr;
339 char path_trace[PATH_MAX];
340 int path_trace_len;
341 char *path_trace_ptr;
342 int open_ret = 0;
343
344 if(channel_dir == NULL) {
345 perror(subchannel_name);
346 open_ret = ENOENT;
347 goto end;
348 }
349
350 printf("Creating trace subdirectory %s\n", subtrace_name);
351 ret = mkdir(subtrace_name, S_IRWXU|S_IRWXG|S_IRWXO);
352 if(ret == -1) {
353 if(errno != EEXIST) {
354 perror(subtrace_name);
355 open_ret = -1;
356 goto end;
357 }
358 }
359
360 strncpy(path_channel, subchannel_name, PATH_MAX-1);
361 path_channel_len = strlen(path_channel);
362 path_channel[path_channel_len] = '/';
363 path_channel_len++;
364 path_channel_ptr = path_channel + path_channel_len;
365
366 strncpy(path_trace, subtrace_name, PATH_MAX-1);
367 path_trace_len = strlen(path_trace);
368 path_trace[path_trace_len] = '/';
369 path_trace_len++;
370 path_trace_ptr = path_trace + path_trace_len;
371
372 #ifdef HAS_INOTIFY
373 iwatch_array->elem = realloc(iwatch_array->elem,
374 ++iwatch_array->num * sizeof(struct inotify_watch));
375
376 printf("Adding inotify for channel %s\n", path_channel);
377 iwatch_array->elem[iwatch_array->num-1].wd = inotify_add_watch(*inotify_fd, path_channel, IN_CREATE);
378 strcpy(iwatch_array->elem[iwatch_array->num-1].path_channel, path_channel);
379 strcpy(iwatch_array->elem[iwatch_array->num-1].path_trace, path_trace);
380 printf("Added inotify for channel %s, wd %u\n", iwatch_array->elem[iwatch_array->num-1].path_channel,
381 iwatch_array->elem[iwatch_array->num-1].wd);
382 #endif
383
384 while((entry = readdir(channel_dir)) != NULL) {
385
386 if(entry->d_name[0] == '.') continue;
387
388 strncpy(path_channel_ptr, entry->d_name, PATH_MAX - path_channel_len);
389 strncpy(path_trace_ptr, entry->d_name, PATH_MAX - path_trace_len);
390
391 ret = stat(path_channel, &stat_buf);
392 if(ret == -1) {
393 perror(path_channel);
394 continue;
395 }
396
397 printf("Channel file : %s\n", path_channel);
398
399 if(S_ISDIR(stat_buf.st_mode)) {
400
401 printf("Entering channel subdirectory...\n");
402 ret = open_channel_trace_pairs(path_channel, path_trace, fd_pairs,
403 inotify_fd, iwatch_array);
404 if(ret < 0) continue;
405 } else if(S_ISREG(stat_buf.st_mode)) {
406 open_ret = open_buffer_file(entry->d_name, path_channel, path_trace,
407 fd_pairs);
408 if(open_ret)
409 goto end;
410 }
411 }
412
413 end:
414 closedir(channel_dir);
415
416 return open_ret;
417 }
418
419
420 int read_subbuffer(struct fd_pair *pair)
421 {
422 unsigned int consumed_old;
423 int err, ret=0;
424
425
426 err = ioctl(pair->channel, RELAY_GET_SUBBUF,
427 &consumed_old);
428 printf("cookie : %u\n", consumed_old);
429 if(err != 0) {
430 ret = errno;
431 perror("Reserving sub buffer failed (everything is normal, it is due to concurrency)");
432 goto get_error;
433 }
434
435 err = TEMP_FAILURE_RETRY(write(pair->trace,
436 pair->mmap
437 + (consumed_old & ((pair->n_subbufs * pair->subbuf_size)-1)),
438 pair->subbuf_size));
439
440 if(err < 0) {
441 ret = errno;
442 perror("Error in writing to file");
443 goto write_error;
444 }
445 #if 0
446 err = fsync(pair->trace);
447 if(err < 0) {
448 ret = errno;
449 perror("Error in writing to file");
450 goto write_error;
451 }
452 #endif //0
453 write_error:
454 err = ioctl(pair->channel, RELAY_PUT_SUBBUF, &consumed_old);
455 if(err != 0) {
456 ret = errno;
457 if(errno == EFAULT) {
458 perror("Error in unreserving sub buffer\n");
459 } else if(errno == EIO) {
460 perror("Reader has been pushed by the writer, last subbuffer corrupted.");
461 /* FIXME : we may delete the last written buffer if we wish. */
462 }
463 goto get_error;
464 }
465
466 get_error:
467 return ret;
468 }
469
470
471
472 int map_channels(struct channel_trace_fd *fd_pairs,
473 int idx_begin, int idx_end)
474 {
475 int i,j;
476 int ret=0;
477
478 if(fd_pairs->num_pairs <= 0) {
479 printf("No channel to read\n");
480 goto end;
481 }
482
483 /* Get the subbuf sizes and number */
484
485 for(i=idx_begin;i<idx_end;i++) {
486 struct fd_pair *pair = &fd_pairs->pair[i];
487
488 ret = ioctl(pair->channel, RELAY_GET_N_SUBBUFS,
489 &pair->n_subbufs);
490 if(ret != 0) {
491 perror("Error in getting the number of subbuffers");
492 goto end;
493 }
494 ret = ioctl(pair->channel, RELAY_GET_SUBBUF_SIZE,
495 &pair->subbuf_size);
496 if(ret != 0) {
497 perror("Error in getting the size of the subbuffers");
498 goto end;
499 }
500 ret = pthread_mutex_init(&pair->mutex, NULL); /* Fast mutex */
501 if(ret != 0) {
502 perror("Error in mutex init");
503 goto end;
504 }
505 }
506
507 /* Mmap each FD */
508 for(i=idx_begin;i<idx_end;i++) {
509 struct fd_pair *pair = &fd_pairs->pair[i];
510
511 pair->mmap = mmap(0, pair->subbuf_size * pair->n_subbufs, PROT_READ,
512 MAP_SHARED, pair->channel, 0);
513 if(pair->mmap == MAP_FAILED) {
514 perror("Mmap error");
515 goto munmap;
516 }
517 }
518
519 goto end; /* success */
520
521 /* Error handling */
522 /* munmap only the successfully mmapped indexes */
523 munmap:
524 /* Munmap each FD */
525 for(j=idx_begin;j<i;j++) {
526 struct fd_pair *pair = &fd_pairs->pair[j];
527 int err_ret;
528
529 err_ret = munmap(pair->mmap, pair->subbuf_size * pair->n_subbufs);
530 if(err_ret != 0) {
531 perror("Error in munmap");
532 }
533 ret |= err_ret;
534 }
535
536 end:
537 return ret;
538
539
540 }
541
542
543 int unmap_channels(struct channel_trace_fd *fd_pairs)
544 {
545 int j;
546 int ret=0;
547
548 /* Munmap each FD */
549 for(j=0;j<fd_pairs->num_pairs;j++) {
550 struct fd_pair *pair = &fd_pairs->pair[j];
551 int err_ret;
552
553 err_ret = munmap(pair->mmap, pair->subbuf_size * pair->n_subbufs);
554 if(err_ret != 0) {
555 perror("Error in munmap");
556 }
557 ret |= err_ret;
558 err_ret = pthread_mutex_destroy(&pair->mutex);
559 if(err_ret != 0) {
560 perror("Error in mutex destroy");
561 }
562 ret |= err_ret;
563 }
564
565 return ret;
566 }
567
568 #ifdef HAS_INOTIFY
569 /* Inotify event arrived.
570 *
571 * Only support add file for now.
572 */
573
574 int read_inotify(int inotify_fd,
575 struct channel_trace_fd *fd_pairs,
576 struct inotify_watch_array *iwatch_array)
577 {
578 char buf[sizeof(struct inotify_event) + PATH_MAX];
579 char path_channel[PATH_MAX];
580 char path_trace[PATH_MAX];
581 ssize_t len;
582 struct inotify_event *ievent;
583 size_t offset;
584 unsigned int i;
585 int ret;
586 int old_num;
587
588 offset = 0;
589 len = read(inotify_fd, buf, sizeof(struct inotify_event) + PATH_MAX);
590 if(len < 0) {
591
592 if(errno == EAGAIN)
593 return 0; /* another thread got the data before us */
594
595 printf("Error in read from inotify FD %s.\n", strerror(len));
596 return -1;
597 }
598 while(offset < len) {
599 ievent = (struct inotify_event *)&(buf[offset]);
600 for(i=0; i<iwatch_array->num; i++) {
601 if(iwatch_array->elem[i].wd == ievent->wd &&
602 ievent->mask == IN_CREATE) {
603 printf("inotify wd %u event mask : %u for %s%s\n",
604 ievent->wd, ievent->mask,
605 iwatch_array->elem[i].path_channel, ievent->name);
606 old_num = fd_pairs->num_pairs;
607 strcpy(path_channel, iwatch_array->elem[i].path_channel);
608 strcat(path_channel, ievent->name);
609 strcpy(path_trace, iwatch_array->elem[i].path_trace);
610 strcat(path_trace, ievent->name);
611 if(ret = open_buffer_file(ievent->name, path_channel,
612 path_trace, fd_pairs)) {
613 printf("Error opening buffer file\n");
614 return -1;
615 }
616 if(ret = map_channels(fd_pairs, old_num, fd_pairs->num_pairs)) {
617 printf("Error mapping channel\n");
618 return -1;
619 }
620
621 }
622 }
623 offset += sizeof(*ievent) + ievent->len;
624 }
625 }
626 #endif //HAS_INOTIFY
627
628 /* read_channels
629 *
630 * Thread worker.
631 *
632 * Read the debugfs channels and write them in the paired tracefiles.
633 *
634 * @fd_pairs : paired channels and trace files.
635 *
636 * returns 0 on success, -1 on error.
637 *
638 * Note that the high priority polled channels are consumed first. We then poll
639 * again to see if these channels are still in priority. Only when no
640 * high priority channel is left, we start reading low priority channels.
641 *
642 * Note that a channel is considered high priority when the buffer is almost
643 * full.
644 */
645
646 int read_channels(unsigned long thread_num, struct channel_trace_fd *fd_pairs,
647 int inotify_fd, struct inotify_watch_array *iwatch_array)
648 {
649 struct pollfd *pollfd = NULL;
650 int num_pollfd;
651 int i,j;
652 int num_rdy, num_hup;
653 int high_prio;
654 int ret = 0;
655 int inotify_fds;
656 unsigned int old_num;
657
658 #ifdef HAS_INOTIFY
659 inotify_fds = 1;
660 #else
661 inotify_fds = 0;
662 #endif
663
664 pthread_rwlock_rdlock(&fd_pairs_lock);
665
666 /* Start polling the FD. Keep one fd for inotify */
667 pollfd = malloc((inotify_fds + fd_pairs->num_pairs) * sizeof(struct pollfd));
668
669 #ifdef HAS_INOTIFY
670 pollfd[0].fd = inotify_fd;
671 pollfd[0].events = POLLIN|POLLPRI;
672 #endif
673
674 for(i=0;i<fd_pairs->num_pairs;i++) {
675 pollfd[inotify_fds+i].fd = fd_pairs->pair[i].channel;
676 pollfd[inotify_fds+i].events = POLLIN|POLLPRI;
677 }
678 num_pollfd = inotify_fds + fd_pairs->num_pairs;
679
680
681 pthread_rwlock_unlock(&fd_pairs_lock);
682
683 while(1) {
684 high_prio = 0;
685 num_hup = 0;
686 #ifdef DEBUG
687 printf("Press a key for next poll...\n");
688 char buf[1];
689 read(STDIN_FILENO, &buf, 1);
690 printf("Next poll (polling %d fd) :\n", num_pollfd);
691 #endif //DEBUG
692
693 /* Have we received a signal ? */
694 if(quit_program) break;
695
696 num_rdy = poll(pollfd, num_pollfd, -1);
697
698 if(num_rdy == -1) {
699 perror("Poll error");
700 goto free_fd;
701 }
702
703 printf("Data received\n");
704 #ifdef HAS_INOTIFY
705 switch(pollfd[0].revents) {
706 case POLLERR:
707 printf("Error returned in polling inotify fd %d.\n", pollfd[0].fd);
708 break;
709 case POLLHUP:
710 printf("Polling inotify fd %d tells it has hung up.\n", pollfd[0].fd);
711 break;
712 case POLLNVAL:
713 printf("Polling inotify fd %d tells fd is not open.\n", pollfd[0].fd);
714 break;
715 case POLLPRI:
716 case POLLIN:
717
718 printf("Polling inotify fd %d : data ready.\n", pollfd[0].fd);
719
720 pthread_rwlock_wrlock(&fd_pairs_lock);
721 read_inotify(inotify_fd, fd_pairs, iwatch_array);
722 pthread_rwlock_unlock(&fd_pairs_lock);
723
724 break;
725 }
726 #endif
727
728 for(i=inotify_fds;i<num_pollfd;i++) {
729 switch(pollfd[i].revents) {
730 case POLLERR:
731 printf("Error returned in polling fd %d.\n", pollfd[i].fd);
732 num_hup++;
733 break;
734 case POLLHUP:
735 printf("Polling fd %d tells it has hung up.\n", pollfd[i].fd);
736 num_hup++;
737 break;
738 case POLLNVAL:
739 printf("Polling fd %d tells fd is not open.\n", pollfd[i].fd);
740 num_hup++;
741 break;
742 case POLLPRI:
743 pthread_rwlock_rdlock(&fd_pairs_lock);
744 if(pthread_mutex_trylock(&fd_pairs->pair[i-inotify_fds].mutex) == 0) {
745 printf("Urgent read on fd %d\n", pollfd[i].fd);
746 /* Take care of high priority channels first. */
747 high_prio = 1;
748 /* it's ok to have an unavailable subbuffer */
749 ret = read_subbuffer(&fd_pairs->pair[i-inotify_fds]);
750 if(ret == EAGAIN) ret = 0;
751
752 ret = pthread_mutex_unlock(&fd_pairs->pair[i-inotify_fds].mutex);
753 if(ret)
754 printf("Error in mutex unlock : %s\n", strerror(ret));
755 }
756 pthread_rwlock_unlock(&fd_pairs_lock);
757 break;
758 }
759 }
760 /* If every buffer FD has hung up, we end the read loop here */
761 if(num_hup == num_pollfd - inotify_fds) break;
762
763 if(!high_prio) {
764 for(i=inotify_fds;i<num_pollfd;i++) {
765 switch(pollfd[i].revents) {
766 case POLLIN:
767 pthread_rwlock_rdlock(&fd_pairs_lock);
768 if(pthread_mutex_trylock(&fd_pairs->pair[i-inotify_fds].mutex) == 0) {
769 /* Take care of low priority channels. */
770 printf("Normal read on fd %d\n", pollfd[i].fd);
771 /* it's ok to have an unavailable subbuffer */
772 ret = read_subbuffer(&fd_pairs->pair[i-inotify_fds]);
773 if(ret == EAGAIN) ret = 0;
774
775 ret = pthread_mutex_unlock(&fd_pairs->pair[i-inotify_fds].mutex);
776 if(ret)
777 printf("Error in mutex unlock : %s\n", strerror(ret));
778 }
779 pthread_rwlock_unlock(&fd_pairs_lock);
780 break;
781 }
782 }
783 }
784
785 /* Update pollfd array if an entry was added to fd_pairs */
786 pthread_rwlock_rdlock(&fd_pairs_lock);
787 if((inotify_fds + fd_pairs->num_pairs) != num_pollfd) {
788 pollfd = realloc(pollfd,
789 (inotify_fds + fd_pairs->num_pairs) * sizeof(struct pollfd));
790 for(i=num_pollfd-inotify_fds;i<fd_pairs->num_pairs;i++) {
791 pollfd[inotify_fds+i].fd = fd_pairs->pair[i].channel;
792 pollfd[inotify_fds+i].events = POLLIN|POLLPRI;
793 }
794 num_pollfd = fd_pairs->num_pairs + inotify_fds;
795 }
796 pthread_rwlock_unlock(&fd_pairs_lock);
797
798 /* NB: If the fd_pairs structure is updated by another thread from this
799 * point forward, the current thread will wait in the poll without
800 * monitoring the new channel. However, this thread will add the
801 * new channel on next poll (and this should not take too much time
802 * on a loaded system).
803 *
804 * This event is quite unlikely and can only occur if a CPU is
805 * hot-plugged while multple lttd threads are running.
806 */
807 }
808
809 free_fd:
810 free(pollfd);
811
812 end:
813 return ret;
814 }
815
816
817 void close_channel_trace_pairs(struct channel_trace_fd *fd_pairs, int inotify_fd,
818 struct inotify_watch_array *iwatch_array)
819 {
820 int i;
821 int ret;
822
823 for(i=0;i<fd_pairs->num_pairs;i++) {
824 ret = close(fd_pairs->pair[i].channel);
825 if(ret == -1) perror("Close error on channel");
826 ret = close(fd_pairs->pair[i].trace);
827 if(ret == -1) perror("Close error on trace");
828 }
829 free(fd_pairs->pair);
830 free(iwatch_array->elem);
831 }
832
833 /* Thread worker */
834 void * thread_main(void *arg)
835 {
836 long ret;
837 unsigned long thread_num = (unsigned long)arg;
838
839 ret = read_channels(thread_num, &fd_pairs, inotify_fd, &inotify_watch_array);
840
841 return (void*)ret;
842 }
843
844
845 int channels_init()
846 {
847 int ret = 0;
848
849 inotify_fd = inotify_init();
850 fcntl(inotify_fd, F_SETFL, O_NONBLOCK);
851
852 if(ret = open_channel_trace_pairs(channel_name, trace_name, &fd_pairs,
853 &inotify_fd, &inotify_watch_array))
854 goto close_channel;
855
856 if(ret = map_channels(&fd_pairs, 0, fd_pairs.num_pairs))
857 goto close_channel;
858
859 return 0;
860
861 close_channel:
862 close_channel_trace_pairs(&fd_pairs, inotify_fd, &inotify_watch_array);
863 if(inotify_fd >= 0)
864 close(inotify_fd);
865 return ret;
866 }
867
868
869 int main(int argc, char ** argv)
870 {
871 int ret = 0;
872 struct sigaction act;
873 pthread_t *tids;
874 unsigned long i;
875 void *tret;
876
877 ret = parse_arguments(argc, argv);
878
879 if(ret != 0) show_arguments();
880 if(ret < 0) return EINVAL;
881 if(ret > 0) return 0;
882
883 show_info();
884
885 if(daemon_mode) {
886 ret = daemon(0, 0);
887
888 if(ret == -1) {
889 perror("An error occured while daemonizing.");
890 exit(-1);
891 }
892 }
893
894 /* Connect the signal handlers */
895 act.sa_handler = handler;
896 act.sa_flags = 0;
897 sigemptyset(&(act.sa_mask));
898 sigaddset(&(act.sa_mask), SIGTERM);
899 sigaddset(&(act.sa_mask), SIGQUIT);
900 sigaddset(&(act.sa_mask), SIGINT);
901 sigaction(SIGTERM, &act, NULL);
902 sigaction(SIGQUIT, &act, NULL);
903 sigaction(SIGINT, &act, NULL);
904
905 if(ret = channels_init())
906 return ret;
907
908 tids = malloc(sizeof(pthread_t) * num_threads);
909 for(i=0; i<num_threads; i++) {
910
911 ret = pthread_create(&tids[i], NULL, thread_main, (void*)i);
912 if(ret) {
913 perror("Error creating thread");
914 break;
915 }
916 }
917
918 for(i=0; i<num_threads; i++) {
919 ret = pthread_join(tids[i], &tret);
920 if(ret) {
921 perror("Error joining thread");
922 break;
923 }
924 if((long)tret != 0) {
925 printf("Error %s occured in thread %u\n",
926 strerror((long)tret), i);
927 }
928 }
929
930 free(tids);
931
932 ret = unmap_channels(&fd_pairs);
933 close_channel_trace_pairs(&fd_pairs, inotify_fd, &inotify_watch_array);
934 if(inotify_fd >= 0)
935 close(inotify_fd);
936
937 return ret;
938 }
This page took 0.04716 seconds and 3 git commands to generate.