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