ltt-usertrace/ltt-usertrace-fast.c

   1 /* LTTng user-space "fast" library
   2  *
   3  * This daemon is spawned by each traced thread (to share the mmap).
   4  *
   5  * Its job is to dump periodically this buffer to disk (when it receives a
   6  * SIGUSR1 from its parent).
   7  *
   8  * It uses the control information in the shared memory area (producer/consumer
   9  * count).
  10  *
  11  * When the parent thread dies (yes, those thing may happen) ;) , this daemon
  12  * will flush the last buffer and write it to disk.
  13  *
  14  * Supplement note for streaming : the daemon is responsible for flushing
  15  * periodically the buffer if it is streaming data.
  16  *
  17  *
  18  * Notes :
  19  * shm memory is typically limited to 4096 units (system wide limit SHMMNI in
  20  * /proc/sys/kernel/shmmni). As it requires computation time upon creation, we
  21  * do not use it : we will use a shared mmap() instead which is passed through
  22  * the fork().
  23  * MAP_SHARED mmap segment. Updated when msync or munmap are called.
  24  * MAP_ANONYMOUS.
  25  * Memory  mapped  by  mmap()  is  preserved across fork(2), with the same
  26  *   attributes.
  27  *
  28  * Eventually, there will be two mode :
  29  * * Slow thread spawn : a fork() is done for each new thread. If the process
  30  *   dies, the data is not lost.
  31  * * Fast thread spawn : a pthread_create() is done by the application for each
  32  *   new thread.
  33  *
  34  * We use a timer to check periodically if the parent died. I think it is less
  35  * intrusive than a ptrace() on the parent, which would get every signal. The
  36  * side effect of this is that we won't be notified if the parent does an
  37  * exec(). In this case, we will just sit there until the parent exits.
  38  *
  39  *
  40  * Copyright 2006 Mathieu Desnoyers
  41  *
  42  */
  43
  44 #define inline inline __attribute__((always_inline))
  45
  46 #define _GNU_SOURCE
  47 #define LTT_TRACE
  48 #define LTT_TRACE_FAST
  49 #include <unistd.h>
  50 #include <stdlib.h>
  51 #include <stdio.h>
  52 #include <signal.h>
  53 #include <syscall.h>
  54 #include <features.h>
  55 #include <pthread.h>
  56 #include <malloc.h>
  57 #include <string.h>
  58 #include <signal.h>
  59 #include <fcntl.h>
  60 #include <stdlib.h>
  61 #include <errno.h>
  62 #include <sys/param.h>
  63 #include <sys/time.h>
  64 #include <sys/types.h>
  65 #include <sys/wait.h>
  66 #include <sys/stat.h>
  67 #include <sys/mman.h>
  68 #include <unistd.h>
  69 #include <sys/syscall.h>
  70
  71 // included with hack for powerpc in ltt-usertrace.h #include <asm/atomic.h>
  72 #include <asm/timex.h>  //for get_cycles()
  73
  74 #include <ltt/ltt-usertrace.h>
  75
  76 #define gettid() syscall(__NR_gettid)
  77
  78 #ifdef LTT_SHOW_DEBUG
  79 #define dbg_printf(...) printf(__VA_ARGS__)
  80 #else
  81 #define dbg_printf(...)
  82 #endif //LTT_SHOW_DEBUG
  83
  84
  85 enum force_switch_mode { FORCE_ACTIVE, FORCE_FLUSH };
  86
  87 /* Writer (the traced application) */
  88
  89 __thread struct ltt_trace_info *thread_trace_info = NULL;
  90
  91 void ltt_usertrace_fast_buffer_switch(void)
  92 {
  93         struct ltt_trace_info *tmp = thread_trace_info;
  94         if(tmp)
  95                 kill(tmp->daemon_id, SIGUSR1);
  96 }
  97
  98 /* The cleanup should never be called from a signal handler */
  99 static void ltt_usertrace_fast_cleanup(void *arg)
 100 {
 101         struct ltt_trace_info *tmp = thread_trace_info;
 102         if(tmp) {
 103                 thread_trace_info = NULL;
 104                 kill(tmp->daemon_id, SIGUSR2);
 105                 munmap(tmp, sizeof(*tmp));
 106         }
 107 }
 108
 109 /* Reader (the disk dumper daemon) */
 110
 111 static pid_t traced_pid = 0;
 112 static pid_t traced_tid = 0;
 113 static int parent_exited = 0;
 114 static int fd_process = -1;
 115 static char outfile_name[PATH_MAX];
 116 static char identifier_name[PATH_MAX];
 117
 118 /* signal handling */
 119 static void handler_sigusr1(int signo)
 120 {
 121         dbg_printf("LTT Signal %d received : parent buffer switch.\n", signo);
 122 }
 123
 124 static void handler_sigusr2(int signo)
 125 {
 126         dbg_printf("LTT Signal %d received : parent exited.\n", signo);
 127         parent_exited = 1;
 128 }
 129
 130 static void handler_sigalarm(int signo)
 131 {
 132         dbg_printf("LTT Signal %d received\n", signo);
 133
 134         if(getppid() != traced_pid) {
 135                 /* Parent died */
 136                 dbg_printf("LTT Parent %lu died, cleaning up\n", traced_pid);
 137                 traced_pid = 0;
 138         }
 139         alarm(3);
 140 }
 141
 142 /* Do a buffer switch. Don't switch if buffer is completely empty */
 143 static void flush_buffer(struct ltt_buf *ltt_buf, enum force_switch_mode mode)
 144 {
 145         uint64_t tsc;
 146         int offset_begin, offset_end, offset_old;
 147         int reserve_commit_diff;
 148         int consumed_old, consumed_new;
 149         int commit_count, reserve_count;
 150         int end_switch_old;
 151
 152         do {
 153                 offset_old = atomic_read(&ltt_buf->offset);
 154                 offset_begin = offset_old;
 155                 end_switch_old = 0;
 156                 tsc = ltt_get_timestamp();
 157                 if(tsc == 0) {
 158                         /* Error in getting the timestamp : should not happen : it would
 159                          * mean we are called from an NMI during a write seqlock on xtime. */
 160                         return;
 161                 }
 162
 163                 if(SUBBUF_OFFSET(offset_begin, ltt_buf) != 0) {
 164                         offset_begin = SUBBUF_ALIGN(offset_begin, ltt_buf);
 165                         end_switch_old = 1;
 166                 } else {
 167       /* we do not have to switch : buffer is empty */
 168       return;
 169     }
 170                 if(mode == FORCE_ACTIVE)
 171                         offset_begin += ltt_subbuf_header_len(ltt_buf);
 172                 /* Always begin_switch in FORCE_ACTIVE mode */
 173
 174                 /* Test new buffer integrity */
 175                 reserve_commit_diff =
 176                         atomic_read(
 177                                 &ltt_buf->reserve_count[SUBBUF_INDEX(offset_begin, ltt_buf)])
 178                         - atomic_read(
 179                                         &ltt_buf->commit_count[SUBBUF_INDEX(offset_begin, ltt_buf)]);
 180                 if(reserve_commit_diff == 0) {
 181       /* Next buffer not corrupted. */
 182       if(mode == FORCE_ACTIVE
 183                                 && (offset_begin-atomic_read(&ltt_buf->consumed))
 184                                                                                         >= ltt_buf->alloc_size) {
 185         /* We do not overwrite non consumed buffers and we are full : ignore
 186                  switch while tracing is active. */
 187         return;
 188       }
 189     } else {
 190       /* Next subbuffer corrupted. Force pushing reader even in normal mode */
 191     }
 192
 193                 offset_end = offset_begin;
 194         } while(atomic_cmpxchg(&ltt_buf->offset, offset_old, offset_end)
 195                                                         != offset_old);
 196
 197
 198         if(mode == FORCE_ACTIVE) {
 199                 /* Push the reader if necessary */
 200                 do {
 201                         consumed_old = atomic_read(&ltt_buf->consumed);
 202                         /* If buffer is in overwrite mode, push the reader consumed count if
 203                                  the write position has reached it and we are not at the first
 204                                  iteration (don't push the reader farther than the writer).
 205                                  This operation can be done concurrently by many writers in the
 206                                  same buffer, the writer being at the fartest write position sub-buffer
 207                                  index in the buffer being the one which will win this loop. */
 208                         /* If the buffer is not in overwrite mode, pushing the reader only
 209                                  happen if a sub-buffer is corrupted */
 210                         if((SUBBUF_TRUNC(offset_end-1, ltt_buf)
 211                                         - SUBBUF_TRUNC(consumed_old, ltt_buf))
 212                                                         >= ltt_buf->alloc_size)
 213                                 consumed_new = SUBBUF_ALIGN(consumed_old, ltt_buf);
 214                         else {
 215                                 consumed_new = consumed_old;
 216                                 break;
 217                         }
 218                 } while(atomic_cmpxchg(&ltt_buf->consumed, consumed_old, consumed_new)
 219                                 != consumed_old);
 220
 221                 if(consumed_old != consumed_new) {
 222                         /* Reader pushed : we are the winner of the push, we can therefore
 223                                  reequilibrate reserve and commit. Atomic increment of the commit
 224                                  count permits other writers to play around with this variable
 225                                  before us. We keep track of corrupted_subbuffers even in overwrite
 226                                  mode :
 227                                  we never want to write over a non completely committed sub-buffer :
 228                                  possible causes : the buffer size is too low compared to the unordered
 229                                  data input, or there is a writer who died between the reserve and the
 230                                  commit. */
 231                         if(reserve_commit_diff) {
 232                                 /* We have to alter the sub-buffer commit count : a sub-buffer is
 233                                          corrupted */
 234                                 atomic_add(reserve_commit_diff,
 235                                                                 &ltt_buf->commit_count[SUBBUF_INDEX(offset_begin, ltt_buf)]);
 236                                 atomic_inc(&ltt_buf->corrupted_subbuffers);
 237                         }
 238                 }
 239         }
 240
 241         /* Always switch */
 242
 243         if(end_switch_old) {
 244                 /* old subbuffer */
 245                 /* Concurrency safe because we are the last and only thread to alter this
 246                          sub-buffer. As long as it is not delivered and read, no other thread can
 247                          alter the offset, alter the reserve_count or call the
 248                          client_buffer_end_callback on this sub-buffer.
 249                          The only remaining threads could be the ones with pending commits. They
 250                          will have to do the deliver themself.
 251                          Not concurrency safe in overwrite mode. We detect corrupted subbuffers with
 252                          commit and reserve counts. We keep a corrupted sub-buffers count and push
 253                          the readers across these sub-buffers.
 254                          Not concurrency safe if a writer is stalled in a subbuffer and
 255                          another writer switches in, finding out it's corrupted. The result will be
 256                          than the old (uncommited) subbuffer will be declared corrupted, and that
 257                          the new subbuffer will be declared corrupted too because of the commit
 258                          count adjustment.
 259                          Offset old should never be 0. */
 260                 ltt_buffer_end_callback(ltt_buf, tsc, offset_old,
 261                                 SUBBUF_INDEX((offset_old), ltt_buf));
 262                 /* Setting this reserve_count will allow the sub-buffer to be delivered by
 263                          the last committer. */
 264                 reserve_count = atomic_add_return((SUBBUF_OFFSET((offset_old-1),
 265                                                       ltt_buf) + 1),
 266                                                                                 &ltt_buf->reserve_count[SUBBUF_INDEX((offset_old),
 267                                                           ltt_buf)]);
 268                 if(reserve_count == atomic_read(
 269                                 &ltt_buf->commit_count[SUBBUF_INDEX((offset_old), ltt_buf)])) {
 270                         ltt_deliver_callback(ltt_buf, SUBBUF_INDEX((offset_old), ltt_buf), NULL);
 271                 }
 272         }
 273
 274         if(mode == FORCE_ACTIVE) {
 275                 /* New sub-buffer */
 276                 /* This code can be executed unordered : writers may already have written
 277                          to the sub-buffer before this code gets executed, caution. */
 278                 /* The commit makes sure that this code is executed before the deliver
 279                          of this sub-buffer */
 280                 ltt_buffer_begin_callback(ltt_buf, tsc, SUBBUF_INDEX(offset_begin, ltt_buf));
 281                 commit_count = atomic_add_return(ltt_subbuf_header_len(ltt_buf),
 282                                                                  &ltt_buf->commit_count[SUBBUF_INDEX(offset_begin, ltt_buf)]);
 283                 /* Check if the written buffer has to be delivered */
 284                 if(commit_count == atomic_read(
 285                                         &ltt_buf->reserve_count[SUBBUF_INDEX(offset_begin, ltt_buf)])) {
 286                         ltt_deliver_callback(ltt_buf, SUBBUF_INDEX(offset_begin, ltt_buf), NULL);
 287                 }
 288         }
 289
 290 }
 291
 292
 293 static int open_output_files(void)
 294 {
 295         int ret;
 296         int fd;
 297         /* Open output files */
 298         umask(00000);
 299         ret = mkdir(LTT_USERTRACE_ROOT, 0777);
 300         if(ret < 0 && errno != EEXIST) {
 301                 perror("LTT Error in creating output (mkdir)");
 302                 exit(-1);
 303         }
 304         ret = chdir(LTT_USERTRACE_ROOT);
 305         if(ret < 0) {
 306                 perror("LTT Error in creating output (chdir)");
 307                 exit(-1);
 308         }
 309         snprintf(identifier_name, PATH_MAX-1,   "%lu.%lu.%llu",
 310                         traced_tid, traced_pid, get_cycles());
 311         snprintf(outfile_name, PATH_MAX-1,      "process-%s", identifier_name);
 312
 313 #ifndef LTT_NULL_OUTPUT_TEST
 314         fd = creat(outfile_name, 0644);
 315 #else
 316         /* NULL test */
 317         ret = symlink("/dev/null", outfile_name);
 318         if(ret < 0) {
 319                 perror("error in symlink");
 320                 exit(-1);
 321         }
 322         fd = open(outfile_name, O_WRONLY);
 323         if(fd_process < 0) {
 324                 perror("Error in open");
 325                 exit(-1);
 326         }
 327 #endif //LTT_NULL_OUTPUT_TEST
 328         return fd;
 329 }
 330
 331 static inline int ltt_buffer_get(struct ltt_buf *ltt_buf,
 332                 unsigned int *offset)
 333 {
 334         unsigned int consumed_old, consumed_idx;
 335         consumed_old = atomic_read(&ltt_buf->consumed);
 336         consumed_idx = SUBBUF_INDEX(consumed_old, ltt_buf);
 337
 338         if(atomic_read(&ltt_buf->commit_count[consumed_idx])
 339                 != atomic_read(&ltt_buf->reserve_count[consumed_idx])) {
 340                 return -EAGAIN;
 341         }
 342         if((SUBBUF_TRUNC(atomic_read(&ltt_buf->offset), ltt_buf)
 343                                 -SUBBUF_TRUNC(consumed_old, ltt_buf)) == 0) {
 344                 return -EAGAIN;
 345         }
 346
 347         *offset = consumed_old;
 348
 349         return 0;
 350 }
 351
 352 static inline int ltt_buffer_put(struct ltt_buf *ltt_buf,
 353                 unsigned int offset)
 354 {
 355         unsigned int consumed_old, consumed_new;
 356         int ret;
 357
 358         consumed_old = offset;
 359         consumed_new = SUBBUF_ALIGN(consumed_old, ltt_buf);
 360         if(atomic_cmpxchg(&ltt_buf->consumed, consumed_old, consumed_new)
 361                         != consumed_old) {
 362                 /* We have been pushed by the writer : the last buffer read _is_
 363                  * corrupted!
 364                  * It can also happen if this is a buffer we never got. */
 365                 return -EIO;
 366         } else {
 367                 if(traced_pid == 0 || parent_exited) return 0;
 368
 369                 ret = sem_post(&ltt_buf->writer_sem);
 370                 if(ret < 0) {
 371                         printf("error in sem_post");
 372                 }
 373         }
 374 }
 375
 376 static int read_subbuffer(struct ltt_buf *ltt_buf, int fd)
 377 {
 378         unsigned int consumed_old;
 379         int err;
 380         dbg_printf("LTT read buffer\n");
 381
 382
 383         err = ltt_buffer_get(ltt_buf, &consumed_old);
 384         if(err != 0) {
 385                 if(err != -EAGAIN) dbg_printf("LTT Reserving sub buffer failed\n");
 386                 goto get_error;
 387         }
 388         if(fd_process == -1) {
 389                 fd_process = fd = open_output_files();
 390         }
 391
 392         err = TEMP_FAILURE_RETRY(write(fd,
 393                                 ltt_buf->start
 394                                         + (consumed_old & ((ltt_buf->alloc_size)-1)),
 395                                 ltt_buf->subbuf_size));
 396
 397         if(err < 0) {
 398                 perror("Error in writing to file");
 399                 goto write_error;
 400         }
 401 #if 0
 402         err = fsync(pair->trace);
 403         if(err < 0) {
 404                 ret = errno;
 405                 perror("Error in writing to file");
 406                 goto write_error;
 407         }
 408 #endif //0
 409 write_error:
 410         err = ltt_buffer_put(ltt_buf, consumed_old);
 411
 412         if(err != 0) {
 413                 if(err == -EIO) {
 414                         dbg_printf("Reader has been pushed by the writer, last subbuffer corrupted.\n");
 415                         /* FIXME : we may delete the last written buffer if we wish. */
 416                 }
 417                 goto get_error;
 418         }
 419
 420 get_error:
 421         return err;
 422 }
 423
 424 /* This function is called by ltt_rw_init which has signals blocked */
 425 static void ltt_usertrace_fast_daemon(struct ltt_trace_info *shared_trace_info,
 426                 sigset_t oldset, pid_t l_traced_pid, pthread_t l_traced_tid)
 427 {
 428         struct sigaction act;
 429         int ret;
 430
 431         traced_pid = l_traced_pid;
 432         traced_tid = l_traced_tid;
 433
 434         dbg_printf("LTT ltt_usertrace_fast_daemon : init is %d, pid is %lu, traced_pid is %lu, traced_tid is %lu\n",
 435                         shared_trace_info->init, getpid(), traced_pid, traced_tid);
 436
 437         act.sa_handler = handler_sigusr1;
 438         act.sa_flags = 0;
 439         sigemptyset(&(act.sa_mask));
 440         sigaddset(&(act.sa_mask), SIGUSR1);
 441         sigaction(SIGUSR1, &act, NULL);
 442
 443         act.sa_handler = handler_sigusr2;
 444         act.sa_flags = 0;
 445         sigemptyset(&(act.sa_mask));
 446         sigaddset(&(act.sa_mask), SIGUSR2);
 447         sigaction(SIGUSR2, &act, NULL);
 448
 449         act.sa_handler = handler_sigalarm;
 450         act.sa_flags = 0;
 451         sigemptyset(&(act.sa_mask));
 452         sigaddset(&(act.sa_mask), SIGALRM);
 453         sigaction(SIGALRM, &act, NULL);
 454
 455         alarm(3);
 456
 457         while(1) {
 458                 ret = sigsuspend(&oldset);
 459                 if(ret != -1) {
 460                         perror("LTT Error in sigsuspend\n");
 461                 }
 462                 if(traced_pid == 0) break; /* parent died */
 463                 if(parent_exited) break;
 464                 dbg_printf("LTT Doing a buffer switch read. pid is : %lu\n", getpid());
 465
 466                 do {
 467                         ret = read_subbuffer(&shared_trace_info->channel.process, fd_process);
 468                 } while(ret == 0);
 469         }
 470         /* The parent thread is dead and we have finished with the buffer */
 471
 472         /* Buffer force switch (flush). Using FLUSH instead of ACTIVE because we know
 473          * there is no writer. */
 474         flush_buffer(&shared_trace_info->channel.process, FORCE_FLUSH);
 475         do {
 476                 ret = read_subbuffer(&shared_trace_info->channel.process, fd_process);
 477         } while(ret == 0);
 478
 479         if(fd_process != -1)
 480                 close(fd_process);
 481
 482         ret = sem_destroy(&shared_trace_info->channel.process.writer_sem);
 483         if(ret < 0) {
 484                 perror("error in sem_destroy");
 485         }
 486         munmap(shared_trace_info, sizeof(*shared_trace_info));
 487
 488         exit(0);
 489 }
 490
 491
 492 /* Reader-writer initialization */
 493
 494 static enum ltt_process_role { LTT_ROLE_WRITER, LTT_ROLE_READER }
 495         role = LTT_ROLE_WRITER;
 496
 497
 498 void ltt_rw_init(void)
 499 {
 500         pid_t pid;
 501         struct ltt_trace_info *shared_trace_info;
 502         int ret;
 503         sigset_t set, oldset;
 504         pid_t l_traced_pid = getpid();
 505         pid_t l_traced_tid = gettid();
 506
 507         /* parent : create the shared memory map */
 508         shared_trace_info = mmap(0, sizeof(*thread_trace_info),
 509                         PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, 0, 0);
 510         shared_trace_info->init=0;
 511         shared_trace_info->filter=0;
 512         shared_trace_info->daemon_id=0;
 513         shared_trace_info->nesting=0;
 514         memset(&shared_trace_info->channel.process, 0,
 515                         sizeof(shared_trace_info->channel.process));
 516         //Need NPTL!
 517         ret = sem_init(&shared_trace_info->channel.process.writer_sem, 1,
 518                                                                         LTT_N_SUBBUFS);
 519         if(ret < 0) {
 520                 perror("error in sem_init");
 521         }
 522         shared_trace_info->channel.process.alloc_size = LTT_BUF_SIZE_PROCESS;
 523         shared_trace_info->channel.process.subbuf_size = LTT_SUBBUF_SIZE_PROCESS;
 524         shared_trace_info->channel.process.start =
 525                                                 shared_trace_info->channel.process_buf;
 526         ltt_buffer_begin_callback(&shared_trace_info->channel.process,
 527                         ltt_get_timestamp(), 0);
 528
 529         shared_trace_info->init = 1;
 530
 531         /* Disable signals */
 532   ret = sigfillset(&set);
 533   if(ret) {
 534     dbg_printf("LTT Error in sigfillset\n");
 535   }
 536
 537   ret = pthread_sigmask(SIG_BLOCK, &set, &oldset);
 538   if(ret) {
 539     dbg_printf("LTT Error in pthread_sigmask\n");
 540   }
 541
 542         pid = fork();
 543         if(pid > 0) {
 544                 /* Parent */
 545                 shared_trace_info->daemon_id = pid;
 546                 thread_trace_info = shared_trace_info;
 547
 548                 /* Enable signals */
 549                 ret = pthread_sigmask(SIG_SETMASK, &oldset, NULL);
 550                 if(ret) {
 551                         dbg_printf("LTT Error in pthread_sigmask\n");
 552                 }
 553         } else if(pid == 0) {
 554                 pid_t sid;
 555                 /* Child */
 556                 role = LTT_ROLE_READER;
 557                 sid = setsid();
 558                 //Not a good idea to renice, unless futex wait eventually implement
 559                 //priority inheritence.
 560                 //ret = nice(1);
 561                 //if(ret < 0) {
 562                 //      perror("Error in nice");
 563                 //}
 564                 if(sid < 0) {
 565                         perror("Error setting sid");
 566                 }
 567                 ltt_usertrace_fast_daemon(shared_trace_info, oldset, l_traced_pid,
 568                                         l_traced_tid);
 569                 /* Should never return */
 570                 exit(-1);
 571         } else if(pid < 0) {
 572                 /* fork error */
 573                 perror("LTT Error in forking ltt-usertrace-fast");
 574         }
 575 }
 576
 577 static __thread struct _pthread_cleanup_buffer cleanup_buffer;
 578
 579 void ltt_thread_init(void)
 580 {
 581         _pthread_cleanup_push(&cleanup_buffer, ltt_usertrace_fast_cleanup, NULL);
 582         ltt_rw_init();
 583 }
 584
 585 void __attribute__((constructor)) __ltt_usertrace_fast_init(void)
 586 {
 587   dbg_printf("LTT usertrace-fast init\n");
 588
 589         ltt_rw_init();
 590 }
 591
 592 void __attribute__((destructor)) __ltt_usertrace_fast_fini(void)
 593 {
 594         if(role == LTT_ROLE_WRITER) {
 595           dbg_printf("LTT usertrace-fast fini\n");
 596                 ltt_usertrace_fast_cleanup(NULL);
 597         }
 598 }
 599