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