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