libust/buffers.c

   1 /*
   2  * buffers.c
   3  * LTTng userspace tracer buffering system
   4  *
   5  * Copyright (C) 2009 - Pierre-Marc Fournier (pierre-marc dot fournier at polymtl dot ca)
   6  * Copyright (C) 2008 - Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2.1 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Lesser General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Lesser General Public
  19  * License along with this library; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
  21  */
  22
  23 #include <unistd.h>
  24 #include <sys/mman.h>
  25 #include <sys/ipc.h>
  26 #include <sys/shm.h>
  27 #include <fcntl.h>
  28 #include <ust/kernelcompat.h>
  29 #include <kcompat/kref.h>
  30 #include "buffers.h"
  31 #include "channels.h"
  32 #include "tracer.h"
  33 #include "tracercore.h"
  34 #include "usterr.h"
  35
  36 struct ltt_reserve_switch_offsets {
  37         long begin, end, old;
  38         long begin_switch, end_switch_current, end_switch_old;
  39         size_t before_hdr_pad, size;
  40 };
  41
  42
  43 static DEFINE_MUTEX(ust_buffers_channels_mutex);
  44 static LIST_HEAD(ust_buffers_channels);
  45
  46 static int get_n_cpus(void)
  47 {
  48         int result;
  49         static int n_cpus = 0;
  50
  51         if(n_cpus) {
  52                 return n_cpus;
  53         }
  54
  55         /* On Linux, when some processors are offline
  56          * _SC_NPROCESSORS_CONF counts the offline
  57          * processors, whereas _SC_NPROCESSORS_ONLN
  58          * does not. If we used _SC_NPROCESSORS_ONLN,
  59          * getcpu() could return a value greater than
  60          * this sysconf, in which case the arrays
  61          * indexed by processor would overflow.
  62          */
  63         result = sysconf(_SC_NPROCESSORS_CONF);
  64         if(result == -1) {
  65                 return -1;
  66         }
  67
  68         n_cpus = result;
  69
  70         return result;
  71 }
  72
  73 /* _ust_buffers_write()
  74  *
  75  * @buf: destination buffer
  76  * @offset: offset in destination
  77  * @src: source buffer
  78  * @len: length of source
  79  * @cpy: already copied
  80  */
  81
  82 void _ust_buffers_write(struct ust_buffer *buf, size_t offset,
  83         const void *src, size_t len, ssize_t cpy)
  84 {
  85         do {
  86                 len -= cpy;
  87                 src += cpy;
  88                 offset += cpy;
  89
  90                 WARN_ON(offset >= buf->buf_size);
  91
  92                 cpy = min_t(size_t, len, buf->buf_size - offset);
  93                 ust_buffers_do_copy(buf->buf_data + offset, src, cpy);
  94         } while (unlikely(len != cpy));
  95 }
  96
  97 static int ust_buffers_init_buffer(struct ust_trace *trace,
  98                 struct ust_channel *ltt_chan,
  99                 struct ust_buffer *buf,
 100                 unsigned int n_subbufs);
 101
 102 static int ust_buffers_alloc_buf(struct ust_buffer *buf, size_t *size)
 103 {
 104         void *ptr;
 105         int result;
 106
 107         *size = PAGE_ALIGN(*size);
 108
 109         result = buf->shmid = shmget(getpid(), *size, IPC_CREAT | IPC_EXCL | 0700);
 110         if(result == -1 && errno == EINVAL) {
 111                 ERR("shmget() returned EINVAL; maybe /proc/sys/kernel/shmmax should be increased.");
 112                 return -1;
 113         }
 114         else if(result == -1) {
 115                 PERROR("shmget");
 116                 return -1;
 117         }
 118
 119         /* FIXME: should have matching call to shmdt */
 120         ptr = shmat(buf->shmid, NULL, 0);
 121         if(ptr == (void *) -1) {
 122                 perror("shmat");
 123                 goto destroy_shmem;
 124         }
 125
 126         /* Already mark the shared memory for destruction. This will occur only
 127          * when all users have detached.
 128          */
 129         result = shmctl(buf->shmid, IPC_RMID, NULL);
 130         if(result == -1) {
 131                 perror("shmctl");
 132                 return -1;
 133         }
 134
 135         buf->buf_data = ptr;
 136         buf->buf_size = *size;
 137
 138         return 0;
 139
 140         destroy_shmem:
 141         result = shmctl(buf->shmid, IPC_RMID, NULL);
 142         if(result == -1) {
 143                 perror("shmctl");
 144         }
 145
 146         return -1;
 147 }
 148
 149 int ust_buffers_create_buf(struct ust_channel *channel, int cpu)
 150 {
 151         int result;
 152         struct ust_buffer *buf = channel->buf[cpu];
 153
 154         buf->cpu = cpu;
 155         result = ust_buffers_alloc_buf(buf, &channel->alloc_size);
 156         if(result)
 157                 return -1;
 158
 159         buf->chan = channel;
 160         kref_get(&channel->kref);
 161         return 0;
 162 }
 163
 164 static void ust_buffers_destroy_channel(struct kref *kref)
 165 {
 166         struct ust_channel *chan = container_of(kref, struct ust_channel, kref);
 167         free(chan);
 168 }
 169
 170 static void ust_buffers_destroy_buf(struct ust_buffer *buf)
 171 {
 172         struct ust_channel *chan = buf->chan;
 173         int result;
 174
 175         result = munmap(buf->buf_data, buf->buf_size);
 176         if(result == -1) {
 177                 PERROR("munmap");
 178         }
 179
 180 //ust// chan->buf[buf->cpu] = NULL;
 181         free(buf);
 182         kref_put(&chan->kref, ust_buffers_destroy_channel);
 183 }
 184
 185 /* called from kref_put */
 186 static void ust_buffers_remove_buf(struct kref *kref)
 187 {
 188         struct ust_buffer *buf = container_of(kref, struct ust_buffer, kref);
 189         ust_buffers_destroy_buf(buf);
 190 }
 191
 192 int ust_buffers_open_buf(struct ust_channel *chan, int cpu)
 193 {
 194         int result;
 195
 196         result = ust_buffers_create_buf(chan, cpu);
 197         if (result == -1)
 198                 return -1;
 199
 200         kref_init(&chan->buf[cpu]->kref);
 201
 202         result = ust_buffers_init_buffer(chan->trace, chan, chan->buf[cpu], chan->subbuf_cnt);
 203         if(result == -1)
 204                 return -1;
 205
 206         return 0;
 207
 208         /* FIXME: decrementally destroy on error? */
 209 }
 210
 211 /**
 212  *      ust_buffers_close_buf - close a channel buffer
 213  *      @buf: buffer
 214  */
 215 static void ust_buffers_close_buf(struct ust_buffer *buf)
 216 {
 217         kref_put(&buf->kref, ust_buffers_remove_buf);
 218 }
 219
 220 int ust_buffers_channel_open(struct ust_channel *chan, size_t subbuf_size, size_t subbuf_cnt)
 221 {
 222         int i;
 223         int result;
 224
 225         if(subbuf_size == 0 || subbuf_cnt == 0)
 226                 return -1;
 227
 228         /* Check that the subbuffer size is larger than a page. */
 229         WARN_ON_ONCE(subbuf_size < PAGE_SIZE);
 230
 231         /*
 232          * Make sure the number of subbuffers and subbuffer size are power of 2.
 233          */
 234         WARN_ON_ONCE(hweight32(subbuf_size) != 1);
 235         WARN_ON(hweight32(subbuf_cnt) != 1);
 236
 237         chan->version = UST_CHANNEL_VERSION;
 238         chan->subbuf_cnt = subbuf_cnt;
 239         chan->subbuf_size = subbuf_size;
 240         chan->subbuf_size_order = get_count_order(subbuf_size);
 241         chan->alloc_size = subbuf_size * subbuf_cnt;
 242
 243         kref_init(&chan->kref);
 244
 245         mutex_lock(&ust_buffers_channels_mutex);
 246         for(i=0; i<chan->n_cpus; i++) {
 247                 result = ust_buffers_open_buf(chan, i);
 248                 if (result == -1)
 249                         goto error;
 250         }
 251         list_add(&chan->list, &ust_buffers_channels);
 252         mutex_unlock(&ust_buffers_channels_mutex);
 253
 254         return 0;
 255
 256         /* Jump directly inside the loop to close the buffers that were already
 257          * opened. */
 258         for(; i>=0; i--) {
 259                 ust_buffers_close_buf(chan->buf[i]);
 260 error:
 261                 do {} while(0);
 262         }
 263
 264         kref_put(&chan->kref, ust_buffers_destroy_channel);
 265         mutex_unlock(&ust_buffers_channels_mutex);
 266         return -1;
 267 }
 268
 269 void ust_buffers_channel_close(struct ust_channel *chan)
 270 {
 271         int i;
 272         if(!chan)
 273                 return;
 274
 275         mutex_lock(&ust_buffers_channels_mutex);
 276         for(i=0; i<chan->n_cpus; i++) {
 277         /* FIXME: if we make it here, then all buffers were necessarily allocated. Moreover, we don't
 278          * initialize to NULL so we cannot use this check. Should we? */
 279 //ust//         if (chan->buf[i])
 280                         ust_buffers_close_buf(chan->buf[i]);
 281         }
 282
 283         list_del(&chan->list);
 284         kref_put(&chan->kref, ust_buffers_destroy_channel);
 285         mutex_unlock(&ust_buffers_channels_mutex);
 286 }
 287
 288 /*
 289  * -------
 290  */
 291
 292 static void ust_buffers_destroy_buffer(struct ust_channel *ltt_chan, int cpu);
 293
 294 static void ltt_force_switch(struct ust_buffer *buf,
 295                 enum force_switch_mode mode);
 296
 297 /*
 298  * Trace callbacks
 299  */
 300 static void ltt_buffer_begin(struct ust_buffer *buf,
 301                         u64 tsc, unsigned int subbuf_idx)
 302 {
 303         struct ust_channel *channel = buf->chan;
 304         struct ltt_subbuffer_header *header =
 305                 (struct ltt_subbuffer_header *)
 306                         ust_buffers_offset_address(buf,
 307                                 subbuf_idx * buf->chan->subbuf_size);
 308
 309         header->cycle_count_begin = tsc;
 310         header->lost_size = 0xFFFFFFFF; /* for debugging */
 311         header->buf_size = buf->chan->subbuf_size;
 312         ltt_write_trace_header(channel->trace, header);
 313 }
 314
 315 /*
 316  * offset is assumed to never be 0 here : never deliver a completely empty
 317  * subbuffer. The lost size is between 0 and subbuf_size-1.
 318  */
 319 static notrace void ltt_buffer_end(struct ust_buffer *buf,
 320                 u64 tsc, unsigned int offset, unsigned int subbuf_idx)
 321 {
 322         struct ltt_subbuffer_header *header =
 323                 (struct ltt_subbuffer_header *)
 324                         ust_buffers_offset_address(buf,
 325                                 subbuf_idx * buf->chan->subbuf_size);
 326
 327         header->lost_size = SUBBUF_OFFSET((buf->chan->subbuf_size - offset),
 328                                 buf->chan);
 329         header->cycle_count_end = tsc;
 330         header->events_lost = local_read(&buf->events_lost);
 331         header->subbuf_corrupt = local_read(&buf->corrupted_subbuffers);
 332 }
 333
 334 /*
 335  * This function should not be called from NMI interrupt context
 336  */
 337 static notrace void ltt_buf_unfull(struct ust_buffer *buf,
 338                 unsigned int subbuf_idx,
 339                 long offset)
 340 {
 341 }
 342
 343 /*
 344  * Promote compiler barrier to a smp_mb().
 345  * For the specific LTTng case, this IPI call should be removed if the
 346  * architecture does not reorder writes.  This should eventually be provided by
 347  * a separate architecture-specific infrastructure.
 348  */
 349 //ust// static void remote_mb(void *info)
 350 //ust// {
 351 //ust//         smp_mb();
 352 //ust// }
 353
 354 int ust_buffers_get_subbuf(struct ust_buffer *buf, long *consumed)
 355 {
 356         struct ust_channel *channel = buf->chan;
 357         long consumed_old, consumed_idx, commit_count, write_offset;
 358 //ust// int retval;
 359
 360         consumed_old = atomic_long_read(&buf->consumed);
 361         consumed_idx = SUBBUF_INDEX(consumed_old, buf->chan);
 362         commit_count = local_read(&buf->commit_count[consumed_idx].cc_sb);
 363         /*
 364          * Make sure we read the commit count before reading the buffer
 365          * data and the write offset. Correct consumed offset ordering
 366          * wrt commit count is insured by the use of cmpxchg to update
 367          * the consumed offset.
 368          * smp_call_function_single can fail if the remote CPU is offline,
 369          * this is OK because then there is no wmb to execute there.
 370          * If our thread is executing on the same CPU as the on the buffers
 371          * belongs to, we don't have to synchronize it at all. If we are
 372          * migrated, the scheduler will take care of the memory barriers.
 373          * Normally, smp_call_function_single() should ensure program order when
 374          * executing the remote function, which implies that it surrounds the
 375          * function execution with :
 376          * smp_mb()
 377          * send IPI
 378          * csd_lock_wait
 379          *                recv IPI
 380          *                smp_mb()
 381          *                exec. function
 382          *                smp_mb()
 383          *                csd unlock
 384          * smp_mb()
 385          *
 386          * However, smp_call_function_single() does not seem to clearly execute
 387          * such barriers. It depends on spinlock semantic to provide the barrier
 388          * before executing the IPI and, when busy-looping, csd_lock_wait only
 389          * executes smp_mb() when it has to wait for the other CPU.
 390          *
 391          * I don't trust this code. Therefore, let's add the smp_mb() sequence
 392          * required ourself, even if duplicated. It has no performance impact
 393          * anyway.
 394          *
 395          * smp_mb() is needed because smp_rmb() and smp_wmb() only order read vs
 396          * read and write vs write. They do not ensure core synchronization. We
 397          * really have to ensure total order between the 3 barriers running on
 398          * the 2 CPUs.
 399          */
 400 //ust// #ifdef LTT_NO_IPI_BARRIER
 401         /*
 402          * Local rmb to match the remote wmb to read the commit count before the
 403          * buffer data and the write offset.
 404          */
 405         smp_rmb();
 406 //ust// #else
 407 //ust//         if (raw_smp_processor_id() != buf->cpu) {
 408 //ust//                 smp_mb();       /* Total order with IPI handler smp_mb() */
 409 //ust//                 smp_call_function_single(buf->cpu, remote_mb, NULL, 1);
 410 //ust//                 smp_mb();       /* Total order with IPI handler smp_mb() */
 411 //ust//         }
 412 //ust// #endif
 413
 414         write_offset = local_read(&buf->offset);
 415         /*
 416          * Check that the subbuffer we are trying to consume has been
 417          * already fully committed.
 418          */
 419         if (((commit_count - buf->chan->subbuf_size)
 420              & channel->commit_count_mask)
 421             - (BUFFER_TRUNC(consumed_old, buf->chan)
 422                >> channel->n_subbufs_order)
 423             != 0) {
 424                 return -EAGAIN;
 425         }
 426         /*
 427          * Check that we are not about to read the same subbuffer in
 428          * which the writer head is.
 429          */
 430         if ((SUBBUF_TRUNC(write_offset, buf->chan)
 431            - SUBBUF_TRUNC(consumed_old, buf->chan))
 432            == 0) {
 433                 return -EAGAIN;
 434         }
 435
 436         /* FIXME: is this ok to disable the reading feature? */
 437 //ust// retval = update_read_sb_index(buf, consumed_idx);
 438 //ust// if (retval)
 439 //ust//         return retval;
 440
 441         *consumed = consumed_old;
 442
 443         return 0;
 444 }
 445
 446 int ust_buffers_put_subbuf(struct ust_buffer *buf, unsigned long uconsumed_old)
 447 {
 448         long consumed_new, consumed_old;
 449
 450         consumed_old = atomic_long_read(&buf->consumed);
 451         consumed_old = consumed_old & (~0xFFFFFFFFL);
 452         consumed_old = consumed_old | uconsumed_old;
 453         consumed_new = SUBBUF_ALIGN(consumed_old, buf->chan);
 454
 455 //ust// spin_lock(&ltt_buf->full_lock);
 456         if (atomic_long_cmpxchg(&buf->consumed, consumed_old,
 457                                 consumed_new)
 458             != consumed_old) {
 459                 /* We have been pushed by the writer : the last
 460                  * buffer read _is_ corrupted! It can also
 461                  * happen if this is a buffer we never got. */
 462 //ust//         spin_unlock(&ltt_buf->full_lock);
 463                 return -EIO;
 464         } else {
 465                 /* tell the client that buffer is now unfull */
 466                 int index;
 467                 long data;
 468                 index = SUBBUF_INDEX(consumed_old, buf->chan);
 469                 data = BUFFER_OFFSET(consumed_old, buf->chan);
 470                 ltt_buf_unfull(buf, index, data);
 471 //ust//         spin_unlock(&ltt_buf->full_lock);
 472         }
 473         return 0;
 474 }
 475
 476 //ust// static void switch_buffer(unsigned long data)
 477 //ust// {
 478 //ust//         struct ltt_channel_buf_struct *ltt_buf =
 479 //ust//                 (struct ltt_channel_buf_struct *)data;
 480 //ust//         struct rchan_buf *buf = ltt_buf->rbuf;
 481 //ust//
 482 //ust//         if (buf)
 483 //ust//                 ltt_force_switch(buf, FORCE_ACTIVE);
 484 //ust//
 485 //ust//         ltt_buf->switch_timer.expires += ltt_buf->switch_timer_interval;
 486 //ust//         add_timer_on(&ltt_buf->switch_timer, smp_processor_id());
 487 //ust// }
 488 //ust//
 489 //ust// static void start_switch_timer(struct ltt_channel_struct *ltt_channel)
 490 //ust// {
 491 //ust//         struct rchan *rchan = ltt_channel->trans_channel_data;
 492 //ust//         int cpu;
 493 //ust//
 494 //ust//         if (!ltt_channel->switch_timer_interval)
 495 //ust//                 return;
 496 //ust//
 497 //ust//         // TODO : hotplug
 498 //ust//         for_each_online_cpu(cpu) {
 499 //ust//                 struct ltt_channel_buf_struct *ltt_buf;
 500 //ust//                 struct rchan_buf *buf;
 501 //ust//
 502 //ust//                 buf = rchan->buf[cpu];
 503 //ust//                 ltt_buf = buf->chan_private;
 504 //ust//                 buf->random_access = 1;
 505 //ust//                 ltt_buf->switch_timer_interval =
 506 //ust//                         ltt_channel->switch_timer_interval;
 507 //ust//                 init_timer(&ltt_buf->switch_timer);
 508 //ust//                 ltt_buf->switch_timer.function = switch_buffer;
 509 //ust//                 ltt_buf->switch_timer.expires = jiffies +
 510 //ust//                                         ltt_buf->switch_timer_interval;
 511 //ust//                 ltt_buf->switch_timer.data = (unsigned long)ltt_buf;
 512 //ust//                 add_timer_on(&ltt_buf->switch_timer, cpu);
 513 //ust//         }
 514 //ust// }
 515 //ust//
 516 //ust// /*
 517 //ust//  * Cannot use del_timer_sync with add_timer_on, so use an IPI to locally
 518 //ust//  * delete the timer.
 519 //ust//  */
 520 //ust// static void stop_switch_timer_ipi(void *info)
 521 //ust// {
 522 //ust//         struct ltt_channel_buf_struct *ltt_buf =
 523 //ust//                 (struct ltt_channel_buf_struct *)info;
 524 //ust//
 525 //ust//         del_timer(&ltt_buf->switch_timer);
 526 //ust// }
 527 //ust//
 528 //ust// static void stop_switch_timer(struct ltt_channel_struct *ltt_channel)
 529 //ust// {
 530 //ust//         struct rchan *rchan = ltt_channel->trans_channel_data;
 531 //ust//         int cpu;
 532 //ust//
 533 //ust//         if (!ltt_channel->switch_timer_interval)
 534 //ust//                 return;
 535 //ust//
 536 //ust//         // TODO : hotplug
 537 //ust//         for_each_online_cpu(cpu) {
 538 //ust//                 struct ltt_channel_buf_struct *ltt_buf;
 539 //ust//                 struct rchan_buf *buf;
 540 //ust//
 541 //ust//                 buf = rchan->buf[cpu];
 542 //ust//                 ltt_buf = buf->chan_private;
 543 //ust//                 smp_call_function(stop_switch_timer_ipi, ltt_buf, 1);
 544 //ust//                 buf->random_access = 0;
 545 //ust//         }
 546 //ust// }
 547
 548 //ust// static void ust_buffers_print_written(struct ust_channel *chan,
 549 //ust//                 long cons_off, unsigned int cpu)
 550 //ust// {
 551 //ust//         struct ust_buffer *buf = chan->buf[cpu];
 552 //ust//         long cons_idx, events_count;
 553 //ust//
 554 //ust//         cons_idx = SUBBUF_INDEX(cons_off, chan);
 555 //ust//         events_count = local_read(&buf->commit_count[cons_idx].events);
 556 //ust//
 557 //ust//         if (events_count)
 558 //ust//                 printk(KERN_INFO
 559 //ust//                         "channel %s: %lu events written (cpu %u, index %lu)\n",
 560 //ust//                         chan->channel_name, events_count, cpu, cons_idx);
 561 //ust// }
 562
 563 static void ltt_relay_print_subbuffer_errors(
 564                 struct ust_channel *channel,
 565                 long cons_off, int cpu)
 566 {
 567         struct ust_buffer *ltt_buf = channel->buf[cpu];
 568         long cons_idx, commit_count, commit_count_sb, write_offset;
 569
 570         cons_idx = SUBBUF_INDEX(cons_off, channel);
 571         commit_count = local_read(&ltt_buf->commit_count[cons_idx].cc);
 572         commit_count_sb = local_read(&ltt_buf->commit_count[cons_idx].cc_sb);
 573
 574         /*
 575          * No need to order commit_count and write_offset reads because we
 576          * execute after trace is stopped when there are no readers left.
 577          */
 578         write_offset = local_read(&ltt_buf->offset);
 579         WARN( "LTT : unread channel %s offset is %ld "
 580                 "and cons_off : %ld (cpu %d)\n",
 581                 channel->channel_name, write_offset, cons_off, cpu);
 582         /* Check each sub-buffer for non filled commit count */
 583         if (((commit_count - channel->subbuf_size) & channel->commit_count_mask)
 584             - (BUFFER_TRUNC(cons_off, channel) >> channel->n_subbufs_order) != 0) {
 585                 ERR("LTT : %s : subbuffer %lu has non filled "
 586                         "commit count [cc, cc_sb] [%lu,%lu].\n",
 587                         channel->channel_name, cons_idx, commit_count, commit_count_sb);
 588         }
 589         ERR("LTT : %s : commit count : %lu, subbuf size %zd\n",
 590                         channel->channel_name, commit_count,
 591                         channel->subbuf_size);
 592 }
 593
 594 static void ltt_relay_print_errors(struct ust_trace *trace,
 595                 struct ust_channel *channel, int cpu)
 596 {
 597         struct ust_buffer *ltt_buf = channel->buf[cpu];
 598         long cons_off;
 599
 600         /*
 601          * Can be called in the error path of allocation when
 602          * trans_channel_data is not yet set.
 603          */
 604         if (!channel)
 605                 return;
 606
 607 //ust// for (cons_off = 0; cons_off < rchan->alloc_size;
 608 //ust//      cons_off = SUBBUF_ALIGN(cons_off, rchan))
 609 //ust//         ust_buffers_print_written(ltt_chan, cons_off, cpu);
 610         for (cons_off = atomic_long_read(&ltt_buf->consumed);
 611                         (SUBBUF_TRUNC(local_read(&ltt_buf->offset),
 612                                       channel)
 613                          - cons_off) > 0;
 614                         cons_off = SUBBUF_ALIGN(cons_off, channel))
 615                 ltt_relay_print_subbuffer_errors(channel, cons_off, cpu);
 616 }
 617
 618 static void ltt_relay_print_buffer_errors(struct ust_channel *channel, int cpu)
 619 {
 620         struct ust_trace *trace = channel->trace;
 621         struct ust_buffer *ltt_buf = channel->buf[cpu];
 622
 623         if (local_read(&ltt_buf->events_lost))
 624                 ERR("channel %s: %ld events lost (cpu %d)",
 625                         channel->channel_name,
 626                         local_read(&ltt_buf->events_lost), cpu);
 627         if (local_read(&ltt_buf->corrupted_subbuffers))
 628                 ERR("channel %s : %ld corrupted subbuffers (cpu %d)",
 629                         channel->channel_name,
 630                         local_read(&ltt_buf->corrupted_subbuffers), cpu);
 631
 632         ltt_relay_print_errors(trace, channel, cpu);
 633 }
 634
 635 static void ltt_relay_release_channel(struct kref *kref)
 636 {
 637         struct ust_channel *ltt_chan = container_of(kref,
 638                         struct ust_channel, kref);
 639         free(ltt_chan->buf);
 640 }
 641
 642 /*
 643  * Create ltt buffer.
 644  */
 645 //ust// static int ltt_relay_create_buffer(struct ust_trace *trace,
 646 //ust//                 struct ltt_channel_struct *ltt_chan, struct rchan_buf *buf,
 647 //ust//                 unsigned int cpu, unsigned int n_subbufs)
 648 //ust// {
 649 //ust//         struct ltt_channel_buf_struct *ltt_buf =
 650 //ust//                 percpu_ptr(ltt_chan->buf, cpu);
 651 //ust//         unsigned int j;
 652 //ust//
 653 //ust//         ltt_buf->commit_count =
 654 //ust//                 kzalloc_node(sizeof(ltt_buf->commit_count) * n_subbufs,
 655 //ust//                         GFP_KERNEL, cpu_to_node(cpu));
 656 //ust//         if (!ltt_buf->commit_count)
 657 //ust//                 return -ENOMEM;
 658 //ust//         kref_get(&trace->kref);
 659 //ust//         kref_get(&trace->ltt_transport_kref);
 660 //ust//         kref_get(&ltt_chan->kref);
 661 //ust//         local_set(&ltt_buf->offset, ltt_subbuffer_header_size());
 662 //ust//         atomic_long_set(&ltt_buf->consumed, 0);
 663 //ust//         atomic_long_set(&ltt_buf->active_readers, 0);
 664 //ust//         for (j = 0; j < n_subbufs; j++)
 665 //ust//                 local_set(&ltt_buf->commit_count[j], 0);
 666 //ust//         init_waitqueue_head(&ltt_buf->write_wait);
 667 //ust//         atomic_set(&ltt_buf->wakeup_readers, 0);
 668 //ust//         spin_lock_init(&ltt_buf->full_lock);
 669 //ust//
 670 //ust//         ltt_buffer_begin_callback(buf, trace->start_tsc, 0);
 671 //ust//         /* atomic_add made on local variable on data that belongs to
 672 //ust//          * various CPUs : ok because tracing not started (for this cpu). */
 673 //ust//         local_add(ltt_subbuffer_header_size(), &ltt_buf->commit_count[0]);
 674 //ust//
 675 //ust//         local_set(&ltt_buf->events_lost, 0);
 676 //ust//         local_set(&ltt_buf->corrupted_subbuffers, 0);
 677 //ust//
 678 //ust//         return 0;
 679 //ust// }
 680
 681 static int ust_buffers_init_buffer(struct ust_trace *trace,
 682                 struct ust_channel *ltt_chan, struct ust_buffer *buf,
 683                 unsigned int n_subbufs)
 684 {
 685         unsigned int j;
 686         int fds[2];
 687         int result;
 688
 689         buf->commit_count =
 690                 zmalloc(sizeof(*buf->commit_count) * n_subbufs);
 691         if (!buf->commit_count)
 692                 return -ENOMEM;
 693         kref_get(&trace->kref);
 694         kref_get(&trace->ltt_transport_kref);
 695         kref_get(&ltt_chan->kref);
 696         local_set(&buf->offset, ltt_subbuffer_header_size());
 697         atomic_long_set(&buf->consumed, 0);
 698         atomic_long_set(&buf->active_readers, 0);
 699         for (j = 0; j < n_subbufs; j++) {
 700                 local_set(&buf->commit_count[j].cc, 0);
 701                 local_set(&buf->commit_count[j].cc_sb, 0);
 702         }
 703 //ust// init_waitqueue_head(&buf->write_wait);
 704 //ust// atomic_set(&buf->wakeup_readers, 0);
 705 //ust// spin_lock_init(&buf->full_lock);
 706
 707         ltt_buffer_begin(buf, trace->start_tsc, 0);
 708
 709         local_add(ltt_subbuffer_header_size(), &buf->commit_count[0].cc);
 710
 711         local_set(&buf->events_lost, 0);
 712         local_set(&buf->corrupted_subbuffers, 0);
 713
 714         result = pipe(fds);
 715         if(result == -1) {
 716                 PERROR("pipe");
 717                 return -1;
 718         }
 719         buf->data_ready_fd_read = fds[0];
 720         buf->data_ready_fd_write = fds[1];
 721
 722         /* FIXME: do we actually need this? */
 723         result = fcntl(fds[0], F_SETFL, O_NONBLOCK);
 724         if(result == -1) {
 725                 PERROR("fcntl");
 726         }
 727
 728 //ust// buf->commit_seq = malloc(sizeof(buf->commit_seq) * n_subbufs);
 729 //ust// if(!ltt_buf->commit_seq) {
 730 //ust//         return -1;
 731 //ust// }
 732         memset(buf->commit_seq, 0, sizeof(buf->commit_seq[0]) * n_subbufs);
 733
 734         /* FIXME: decrementally destroy on error */
 735
 736         return 0;
 737 }
 738
 739 /* FIXME: use this function */
 740 static void ust_buffers_destroy_buffer(struct ust_channel *ltt_chan, int cpu)
 741 {
 742         struct ust_trace *trace = ltt_chan->trace;
 743         struct ust_buffer *ltt_buf = ltt_chan->buf[cpu];
 744
 745         kref_put(&ltt_chan->trace->ltt_transport_kref,
 746                 ltt_release_transport);
 747         ltt_relay_print_buffer_errors(ltt_chan, cpu);
 748 //ust// free(ltt_buf->commit_seq);
 749         kfree(ltt_buf->commit_count);
 750         ltt_buf->commit_count = NULL;
 751         kref_put(&ltt_chan->kref, ltt_relay_release_channel);
 752         kref_put(&trace->kref, ltt_release_trace);
 753 //ust// wake_up_interruptible(&trace->kref_wq);
 754 }
 755
 756 static int ust_buffers_alloc_channel_buf_structs(struct ust_channel *chan)
 757 {
 758         void *ptr;
 759         int result;
 760         size_t size;
 761         int i;
 762
 763         size = PAGE_ALIGN(1);
 764
 765         for(i=0; i<chan->n_cpus; i++) {
 766
 767                 result = chan->buf_struct_shmids[i] = shmget(getpid(), size, IPC_CREAT | IPC_EXCL | 0700);
 768                 if(result == -1) {
 769                         PERROR("shmget");
 770                         goto destroy_previous;
 771                 }
 772
 773                 /* FIXME: should have matching call to shmdt */
 774                 ptr = shmat(chan->buf_struct_shmids[i], NULL, 0);
 775                 if(ptr == (void *) -1) {
 776                         perror("shmat");
 777                         goto destroy_shm;
 778                 }
 779
 780                 /* Already mark the shared memory for destruction. This will occur only
 781                  * when all users have detached.
 782                  */
 783                 result = shmctl(chan->buf_struct_shmids[i], IPC_RMID, NULL);
 784                 if(result == -1) {
 785                         perror("shmctl");
 786                         goto destroy_previous;
 787                 }
 788
 789                 chan->buf[i] = ptr;
 790         }
 791
 792         return 0;
 793
 794         /* Jumping inside this loop occurs from within the other loop above with i as
 795          * counter, so it unallocates the structures for the cpu = current_i down to
 796          * zero. */
 797         for(; i>=0; i--) {
 798                 destroy_shm:
 799                 result = shmctl(chan->buf_struct_shmids[i], IPC_RMID, NULL);
 800                 if(result == -1) {
 801                         perror("shmctl");
 802                 }
 803
 804                 destroy_previous:
 805                 continue;
 806         }
 807
 808         return -1;
 809 }
 810
 811 /*
 812  * Create channel.
 813  */
 814 static int ust_buffers_create_channel(const char *trace_name, struct ust_trace *trace,
 815         const char *channel_name, struct ust_channel *ltt_chan,
 816         unsigned int subbuf_size, unsigned int n_subbufs, int overwrite)
 817 {
 818         int result;
 819
 820         kref_init(&ltt_chan->kref);
 821
 822         ltt_chan->trace = trace;
 823         ltt_chan->overwrite = overwrite;
 824         ltt_chan->n_subbufs_order = get_count_order(n_subbufs);
 825         ltt_chan->commit_count_mask = (~0UL >> ltt_chan->n_subbufs_order);
 826         ltt_chan->n_cpus = get_n_cpus();
 827 //ust// ltt_chan->buf = percpu_alloc_mask(sizeof(struct ltt_channel_buf_struct), GFP_KERNEL, cpu_possible_map);
 828         ltt_chan->buf = (void *) malloc(ltt_chan->n_cpus * sizeof(void *));
 829         if(ltt_chan->buf == NULL) {
 830                 goto error;
 831         }
 832         ltt_chan->buf_struct_shmids = (int *) malloc(ltt_chan->n_cpus * sizeof(int));
 833         if(ltt_chan->buf_struct_shmids == NULL)
 834                 goto free_buf;
 835
 836         result = ust_buffers_alloc_channel_buf_structs(ltt_chan);
 837         if(result != 0) {
 838                 goto free_buf_struct_shmids;
 839         }
 840
 841         result = ust_buffers_channel_open(ltt_chan, subbuf_size, n_subbufs);
 842         if (result != 0) {
 843                 ERR("Cannot open channel for trace %s", trace_name);
 844                 goto unalloc_buf_structs;
 845         }
 846
 847         return 0;
 848
 849 unalloc_buf_structs:
 850         /* FIXME: put a call here to unalloc the buf structs! */
 851
 852 free_buf_struct_shmids:
 853         free(ltt_chan->buf_struct_shmids);
 854
 855 free_buf:
 856         free(ltt_chan->buf);
 857
 858 error:
 859         return -1;
 860 }
 861
 862 /*
 863  * LTTng channel flush function.
 864  *
 865  * Must be called when no tracing is active in the channel, because of
 866  * accesses across CPUs.
 867  */
 868 static notrace void ltt_relay_buffer_flush(struct ust_buffer *buf)
 869 {
 870         int result;
 871
 872 //ust// buf->finalized = 1;
 873         ltt_force_switch(buf, FORCE_FLUSH);
 874
 875         result = write(buf->data_ready_fd_write, "1", 1);
 876         if(result == -1) {
 877                 PERROR("write (in ltt_relay_buffer_flush)");
 878                 ERR("this should never happen!");
 879         }
 880 }
 881
 882 static void ltt_relay_async_wakeup_chan(struct ust_channel *ltt_channel)
 883 {
 884 //ust// unsigned int i;
 885 //ust// struct rchan *rchan = ltt_channel->trans_channel_data;
 886 //ust//
 887 //ust// for_each_possible_cpu(i) {
 888 //ust//         struct ltt_channel_buf_struct *ltt_buf =
 889 //ust//                 percpu_ptr(ltt_channel->buf, i);
 890 //ust//
 891 //ust//         if (atomic_read(&ltt_buf->wakeup_readers) == 1) {
 892 //ust//                 atomic_set(&ltt_buf->wakeup_readers, 0);
 893 //ust//                 wake_up_interruptible(&rchan->buf[i]->read_wait);
 894 //ust//         }
 895 //ust// }
 896 }
 897
 898 static void ltt_relay_finish_buffer(struct ust_channel *channel, unsigned int cpu)
 899 {
 900 //      int result;
 901
 902         if (channel->buf[cpu]) {
 903                 struct ust_buffer *buf = channel->buf[cpu];
 904                 ltt_relay_buffer_flush(buf);
 905 //ust//         ltt_relay_wake_writers(ltt_buf);
 906                 /* closing the pipe tells the consumer the buffer is finished */
 907
 908                 //result = write(ltt_buf->data_ready_fd_write, "D", 1);
 909                 //if(result == -1) {
 910                 //      PERROR("write (in ltt_relay_finish_buffer)");
 911                 //      ERR("this should never happen!");
 912                 //}
 913                 close(buf->data_ready_fd_write);
 914         }
 915 }
 916
 917
 918 static void ltt_relay_finish_channel(struct ust_channel *channel)
 919 {
 920         unsigned int i;
 921
 922         for(i=0; i<channel->n_cpus; i++) {
 923                 ltt_relay_finish_buffer(channel, i);
 924         }
 925 }
 926
 927 static void ltt_relay_remove_channel(struct ust_channel *channel)
 928 {
 929         ust_buffers_channel_close(channel);
 930         kref_put(&channel->kref, ltt_relay_release_channel);
 931 }
 932
 933 //ust// /*
 934 //ust//  * Returns :
 935 //ust//  * 0 if ok
 936 //ust//  * !0 if execution must be aborted.
 937 //ust//  */
 938 //ust// static inline int ltt_relay_try_reserve(
 939 //ust//                 struct ust_channel *channel, struct ust_buffer *buf,
 940 //ust//                 struct ltt_reserve_switch_offsets *offsets, size_t data_size,
 941 //ust//                 u64 *tsc, unsigned int *rflags, int largest_align)
 942 //ust// {
 943 //ust//         offsets->begin = local_read(&buf->offset);
 944 //ust//         offsets->old = offsets->begin;
 945 //ust//         offsets->begin_switch = 0;
 946 //ust//         offsets->end_switch_current = 0;
 947 //ust//         offsets->end_switch_old = 0;
 948 //ust//
 949 //ust//         *tsc = trace_clock_read64();
 950 //ust//         if (last_tsc_overflow(buf, *tsc))
 951 //ust//                 *rflags = LTT_RFLAG_ID_SIZE_TSC;
 952 //ust//
 953 //ust//         if (SUBBUF_OFFSET(offsets->begin, buf->chan) == 0) {
 954 //ust//                 offsets->begin_switch = 1;              /* For offsets->begin */
 955 //ust//         } else {
 956 //ust//                 offsets->size = ust_get_header_size(channel,
 957 //ust//                                         offsets->begin, data_size,
 958 //ust//                                         &offsets->before_hdr_pad, *rflags);
 959 //ust//                 offsets->size += ltt_align(offsets->begin + offsets->size,
 960 //ust//                                            largest_align)
 961 //ust//                                  + data_size;
 962 //ust//                 if ((SUBBUF_OFFSET(offsets->begin, buf->chan) + offsets->size)
 963 //ust//                                 > buf->chan->subbuf_size) {
 964 //ust//                         offsets->end_switch_old = 1;    /* For offsets->old */
 965 //ust//                         offsets->begin_switch = 1;      /* For offsets->begin */
 966 //ust//                 }
 967 //ust//         }
 968 //ust//         if (offsets->begin_switch) {
 969 //ust//                 long subbuf_index;
 970 //ust//
 971 //ust//                 if (offsets->end_switch_old)
 972 //ust//                         offsets->begin = SUBBUF_ALIGN(offsets->begin,
 973 //ust//                                                       buf->chan);
 974 //ust//                 offsets->begin = offsets->begin + ltt_subbuffer_header_size();
 975 //ust//                 /* Test new buffer integrity */
 976 //ust//                 subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
 977 //ust//                 offsets->reserve_commit_diff =
 978 //ust//                         (BUFFER_TRUNC(offsets->begin, buf->chan)
 979 //ust//                          >> channel->n_subbufs_order)
 980 //ust//                         - (local_read(&buf->commit_count[subbuf_index])
 981 //ust//                                 & channel->commit_count_mask);
 982 //ust//                 if (offsets->reserve_commit_diff == 0) {
 983 //ust//                         long consumed;
 984 //ust//
 985 //ust//                         consumed = atomic_long_read(&buf->consumed);
 986 //ust//
 987 //ust//                         /* Next buffer not corrupted. */
 988 //ust//                         if (!channel->overwrite &&
 989 //ust//                                 (SUBBUF_TRUNC(offsets->begin, buf->chan)
 990 //ust//                                  - SUBBUF_TRUNC(consumed, buf->chan))
 991 //ust//                                 >= channel->alloc_size) {
 992 //ust//
 993 //ust//                                 long consumed_idx = SUBBUF_INDEX(consumed, buf->chan);
 994 //ust//                                 long commit_count = local_read(&buf->commit_count[consumed_idx]);
 995 //ust//                                 if(((commit_count - buf->chan->subbuf_size) & channel->commit_count_mask) - (BUFFER_TRUNC(consumed, buf->chan) >> channel->n_subbufs_order) != 0) {
 996 //ust//                                         WARN("Event dropped. Caused by non-committed event.");
 997 //ust//                                 }
 998 //ust//                                 else {
 999 //ust//                                         WARN("Event dropped. Caused by non-consumed buffer.");
1000 //ust//                                 }
1001 //ust//                                 /*
1002 //ust//                                  * We do not overwrite non consumed buffers
1003 //ust//                                  * and we are full : event is lost.
1004 //ust//                                  */
1005 //ust//                                 local_inc(&buf->events_lost);
1006 //ust//                                 return -1;
1007 //ust//                         } else {
1008 //ust//                                 /*
1009 //ust//                                  * next buffer not corrupted, we are either in
1010 //ust//                                  * overwrite mode or the buffer is not full.
1011 //ust//                                  * It's safe to write in this new subbuffer.
1012 //ust//                                  */
1013 //ust//                         }
1014 //ust//                 } else {
1015 //ust//                         /*
1016 //ust//                          * Next subbuffer corrupted. Force pushing reader even
1017 //ust//                          * in normal mode. It's safe to write in this new
1018 //ust//                          * subbuffer.
1019 //ust//                          */
1020 //ust//                 }
1021 //ust//                 offsets->size = ust_get_header_size(channel,
1022 //ust//                                         offsets->begin, data_size,
1023 //ust//                                         &offsets->before_hdr_pad, *rflags);
1024 //ust//                 offsets->size += ltt_align(offsets->begin + offsets->size,
1025 //ust//                                            largest_align)
1026 //ust//                                  + data_size;
1027 //ust//                 if ((SUBBUF_OFFSET(offsets->begin, buf->chan) + offsets->size)
1028 //ust//                                 > buf->chan->subbuf_size) {
1029 //ust//                         /*
1030 //ust//                          * Event too big for subbuffers, report error, don't
1031 //ust//                          * complete the sub-buffer switch.
1032 //ust//                          */
1033 //ust//                         local_inc(&buf->events_lost);
1034 //ust//                         return -1;
1035 //ust//                 } else {
1036 //ust//                         /*
1037 //ust//                          * We just made a successful buffer switch and the event
1038 //ust//                          * fits in the new subbuffer. Let's write.
1039 //ust//                          */
1040 //ust//                 }
1041 //ust//         } else {
1042 //ust//                 /*
1043 //ust//                  * Event fits in the current buffer and we are not on a switch
1044 //ust//                  * boundary. It's safe to write.
1045 //ust//                  */
1046 //ust//         }
1047 //ust//         offsets->end = offsets->begin + offsets->size;
1048 //ust//
1049 //ust//         if ((SUBBUF_OFFSET(offsets->end, buf->chan)) == 0) {
1050 //ust//                 /*
1051 //ust//                  * The offset_end will fall at the very beginning of the next
1052 //ust//                  * subbuffer.
1053 //ust//                  */
1054 //ust//                 offsets->end_switch_current = 1;        /* For offsets->begin */
1055 //ust//         }
1056 //ust//         return 0;
1057 //ust// }
1058 //ust//
1059 //ust// /*
1060 //ust//  * Returns :
1061 //ust//  * 0 if ok
1062 //ust//  * !0 if execution must be aborted.
1063 //ust//  */
1064 //ust// static inline int ltt_relay_try_switch(
1065 //ust//                 enum force_switch_mode mode,
1066 //ust//                 struct ust_channel *channel,
1067 //ust//                 struct ust_buffer *buf,
1068 //ust//                 struct ltt_reserve_switch_offsets *offsets,
1069 //ust//                 u64 *tsc)
1070 //ust// {
1071 //ust//         long subbuf_index;
1072 //ust//
1073 //ust//         offsets->begin = local_read(&buf->offset);
1074 //ust//         offsets->old = offsets->begin;
1075 //ust//         offsets->begin_switch = 0;
1076 //ust//         offsets->end_switch_old = 0;
1077 //ust//
1078 //ust//         *tsc = trace_clock_read64();
1079 //ust//
1080 //ust//         if (SUBBUF_OFFSET(offsets->begin, buf->chan) != 0) {
1081 //ust//                 offsets->begin = SUBBUF_ALIGN(offsets->begin, buf->chan);
1082 //ust//                 offsets->end_switch_old = 1;
1083 //ust//         } else {
1084 //ust//                 /* we do not have to switch : buffer is empty */
1085 //ust//                 return -1;
1086 //ust//         }
1087 //ust//         if (mode == FORCE_ACTIVE)
1088 //ust//                 offsets->begin += ltt_subbuffer_header_size();
1089 //ust//         /*
1090 //ust//          * Always begin_switch in FORCE_ACTIVE mode.
1091 //ust//          * Test new buffer integrity
1092 //ust//          */
1093 //ust//         subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
1094 //ust//         offsets->reserve_commit_diff =
1095 //ust//                 (BUFFER_TRUNC(offsets->begin, buf->chan)
1096 //ust//                  >> channel->n_subbufs_order)
1097 //ust//                 - (local_read(&buf->commit_count[subbuf_index])
1098 //ust//                         & channel->commit_count_mask);
1099 //ust//         if (offsets->reserve_commit_diff == 0) {
1100 //ust//                 /* Next buffer not corrupted. */
1101 //ust//                 if (mode == FORCE_ACTIVE
1102 //ust//                     && !channel->overwrite
1103 //ust//                     && offsets->begin - atomic_long_read(&buf->consumed)
1104 //ust//                        >= channel->alloc_size) {
1105 //ust//                         /*
1106 //ust//                          * We do not overwrite non consumed buffers and we are
1107 //ust//                          * full : ignore switch while tracing is active.
1108 //ust//                          */
1109 //ust//                         return -1;
1110 //ust//                 }
1111 //ust//         } else {
1112 //ust//                 /*
1113 //ust//                  * Next subbuffer corrupted. Force pushing reader even in normal
1114 //ust//                  * mode
1115 //ust//                  */
1116 //ust//         }
1117 //ust//         offsets->end = offsets->begin;
1118 //ust//         return 0;
1119 //ust// }
1120 //ust//
1121 //ust// static inline void ltt_reserve_push_reader(
1122 //ust//                 struct ust_channel *channel,
1123 //ust//                 struct ust_buffer *buf,
1124 //ust//                 struct ltt_reserve_switch_offsets *offsets)
1125 //ust// {
1126 //ust//         long consumed_old, consumed_new;
1127 //ust//
1128 //ust//         do {
1129 //ust//                 consumed_old = atomic_long_read(&buf->consumed);
1130 //ust//                 /*
1131 //ust//                  * If buffer is in overwrite mode, push the reader consumed
1132 //ust//                  * count if the write position has reached it and we are not
1133 //ust//                  * at the first iteration (don't push the reader farther than
1134 //ust//                  * the writer). This operation can be done concurrently by many
1135 //ust//                  * writers in the same buffer, the writer being at the farthest
1136 //ust//                  * write position sub-buffer index in the buffer being the one
1137 //ust//                  * which will win this loop.
1138 //ust//                  * If the buffer is not in overwrite mode, pushing the reader
1139 //ust//                  * only happens if a sub-buffer is corrupted.
1140 //ust//                  */
1141 //ust//                 if ((SUBBUF_TRUNC(offsets->end-1, buf->chan)
1142 //ust//                    - SUBBUF_TRUNC(consumed_old, buf->chan))
1143 //ust//                    >= channel->alloc_size)
1144 //ust//                         consumed_new = SUBBUF_ALIGN(consumed_old, buf->chan);
1145 //ust//                 else {
1146 //ust//                         consumed_new = consumed_old;
1147 //ust//                         break;
1148 //ust//                 }
1149 //ust//         } while (atomic_long_cmpxchg(&buf->consumed, consumed_old,
1150 //ust//                         consumed_new) != consumed_old);
1151 //ust//
1152 //ust//         if (consumed_old != consumed_new) {
1153 //ust//                 /*
1154 //ust//                  * Reader pushed : we are the winner of the push, we can
1155 //ust//                  * therefore reequilibrate reserve and commit. Atomic increment
1156 //ust//                  * of the commit count permits other writers to play around
1157 //ust//                  * with this variable before us. We keep track of
1158 //ust//                  * corrupted_subbuffers even in overwrite mode :
1159 //ust//                  * we never want to write over a non completely committed
1160 //ust//                  * sub-buffer : possible causes : the buffer size is too low
1161 //ust//                  * compared to the unordered data input, or there is a writer
1162 //ust//                  * that died between the reserve and the commit.
1163 //ust//                  */
1164 //ust//                 if (offsets->reserve_commit_diff) {
1165 //ust//                         /*
1166 //ust//                          * We have to alter the sub-buffer commit count.
1167 //ust//                          * We do not deliver the previous subbuffer, given it
1168 //ust//                          * was either corrupted or not consumed (overwrite
1169 //ust//                          * mode).
1170 //ust//                          */
1171 //ust//                         local_add(offsets->reserve_commit_diff,
1172 //ust//                                   &buf->commit_count[
1173 //ust//                                         SUBBUF_INDEX(offsets->begin,
1174 //ust//                                                      buf->chan)]);
1175 //ust//                         if (!channel->overwrite
1176 //ust//                             || offsets->reserve_commit_diff
1177 //ust//                                != channel->subbuf_size) {
1178 //ust//                                 /*
1179 //ust//                                  * The reserve commit diff was not subbuf_size :
1180 //ust//                                  * it means the subbuffer was partly written to
1181 //ust//                                  * and is therefore corrupted. If it is multiple
1182 //ust//                                  * of subbuffer size and we are in flight
1183 //ust//                                  * recorder mode, we are skipping over a whole
1184 //ust//                                  * subbuffer.
1185 //ust//                                  */
1186 //ust//                                 local_inc(&buf->corrupted_subbuffers);
1187 //ust//                         }
1188 //ust//                 }
1189 //ust//         }
1190 //ust// }
1191 //ust//
1192 //ust// /**
1193 //ust//  * ltt_relay_reserve_slot - Atomic slot reservation in a LTTng buffer.
1194 //ust//  * @trace: the trace structure to log to.
1195 //ust//  * @ltt_channel: channel structure
1196 //ust//  * @transport_data: data structure specific to ltt relay
1197 //ust//  * @data_size: size of the variable length data to log.
1198 //ust//  * @slot_size: pointer to total size of the slot (out)
1199 //ust//  * @buf_offset : pointer to reserved buffer offset (out)
1200 //ust//  * @tsc: pointer to the tsc at the slot reservation (out)
1201 //ust//  * @cpu: cpuid
1202 //ust//  *
1203 //ust//  * Return : -ENOSPC if not enough space, else returns 0.
1204 //ust//  * It will take care of sub-buffer switching.
1205 //ust//  */
1206 //ust// static notrace int ltt_relay_reserve_slot(struct ust_trace *trace,
1207 //ust//                 struct ust_channel *channel, void **transport_data,
1208 //ust//                 size_t data_size, size_t *slot_size, long *buf_offset, u64 *tsc,
1209 //ust//                 unsigned int *rflags, int largest_align, int cpu)
1210 //ust// {
1211 //ust//         struct ust_buffer *buf = *transport_data = channel->buf[cpu];
1212 //ust//         struct ltt_reserve_switch_offsets offsets;
1213 //ust//
1214 //ust//         offsets.reserve_commit_diff = 0;
1215 //ust//         offsets.size = 0;
1216 //ust//
1217 //ust//         /*
1218 //ust//          * Perform retryable operations.
1219 //ust//          */
1220 //ust//         if (ltt_nesting > 4) {
1221 //ust//                 local_inc(&buf->events_lost);
1222 //ust//                 return -EPERM;
1223 //ust//         }
1224 //ust//         do {
1225 //ust//                 if (ltt_relay_try_reserve(channel, buf, &offsets, data_size, tsc, rflags,
1226 //ust//                                 largest_align))
1227 //ust//                         return -ENOSPC;
1228 //ust//         } while (local_cmpxchg(&buf->offset, offsets.old,
1229 //ust//                         offsets.end) != offsets.old);
1230 //ust//
1231 //ust//         /*
1232 //ust//          * Atomically update last_tsc. This update races against concurrent
1233 //ust//          * atomic updates, but the race will always cause supplementary full TSC
1234 //ust//          * events, never the opposite (missing a full TSC event when it would be
1235 //ust//          * needed).
1236 //ust//          */
1237 //ust//         save_last_tsc(buf, *tsc);
1238 //ust//
1239 //ust//         /*
1240 //ust//          * Push the reader if necessary
1241 //ust//          */
1242 //ust//         ltt_reserve_push_reader(channel, buf, &offsets);
1243 //ust//
1244 //ust//         /*
1245 //ust//          * Switch old subbuffer if needed.
1246 //ust//          */
1247 //ust//         if (offsets.end_switch_old)
1248 //ust//                 ltt_reserve_switch_old_subbuf(channel, buf, &offsets, tsc);
1249 //ust//
1250 //ust//         /*
1251 //ust//          * Populate new subbuffer.
1252 //ust//          */
1253 //ust//         if (offsets.begin_switch)
1254 //ust//                 ltt_reserve_switch_new_subbuf(channel, buf, &offsets, tsc);
1255 //ust//
1256 //ust//         if (offsets.end_switch_current)
1257 //ust//                 ltt_reserve_end_switch_current(channel, buf, &offsets, tsc);
1258 //ust//
1259 //ust//         *slot_size = offsets.size;
1260 //ust//         *buf_offset = offsets.begin + offsets.before_hdr_pad;
1261 //ust//         return 0;
1262 //ust// }
1263 //ust//
1264 //ust// /*
1265 //ust//  * Force a sub-buffer switch for a per-cpu buffer. This operation is
1266 //ust//  * completely reentrant : can be called while tracing is active with
1267 //ust//  * absolutely no lock held.
1268 //ust//  *
1269 //ust//  * Note, however, that as a local_cmpxchg is used for some atomic
1270 //ust//  * operations, this function must be called from the CPU which owns the buffer
1271 //ust//  * for a ACTIVE flush.
1272 //ust//  */
1273 //ust// static notrace void ltt_force_switch(struct ust_buffer *buf,
1274 //ust//                 enum force_switch_mode mode)
1275 //ust// {
1276 //ust//         struct ust_channel *channel = buf->chan;
1277 //ust//         struct ltt_reserve_switch_offsets offsets;
1278 //ust//         u64 tsc;
1279 //ust//
1280 //ust//         offsets.reserve_commit_diff = 0;
1281 //ust//         offsets.size = 0;
1282 //ust//
1283 //ust//         /*
1284 //ust//          * Perform retryable operations.
1285 //ust//          */
1286 //ust//         do {
1287 //ust//                 if (ltt_relay_try_switch(mode, channel, buf, &offsets, &tsc))
1288 //ust//                         return;
1289 //ust//         } while (local_cmpxchg(&buf->offset, offsets.old,
1290 //ust//                         offsets.end) != offsets.old);
1291 //ust//
1292 //ust//         /*
1293 //ust//          * Atomically update last_tsc. This update races against concurrent
1294 //ust//          * atomic updates, but the race will always cause supplementary full TSC
1295 //ust//          * events, never the opposite (missing a full TSC event when it would be
1296 //ust//          * needed).
1297 //ust//          */
1298 //ust//         save_last_tsc(buf, tsc);
1299 //ust//
1300 //ust//         /*
1301 //ust//          * Push the reader if necessary
1302 //ust//          */
1303 //ust//         if (mode == FORCE_ACTIVE)
1304 //ust//                 ltt_reserve_push_reader(channel, buf, &offsets);
1305 //ust//
1306 //ust//         /*
1307 //ust//          * Switch old subbuffer if needed.
1308 //ust//          */
1309 //ust//         if (offsets.end_switch_old)
1310 //ust//                 ltt_reserve_switch_old_subbuf(channel, buf, &offsets, &tsc);
1311 //ust//
1312 //ust//         /*
1313 //ust//          * Populate new subbuffer.
1314 //ust//          */
1315 //ust//         if (mode == FORCE_ACTIVE)
1316 //ust//                 ltt_reserve_switch_new_subbuf(channel, buf, &offsets, &tsc);
1317 //ust// }
1318
1319 /*
1320  * ltt_reserve_switch_old_subbuf: switch old subbuffer
1321  *
1322  * Concurrency safe because we are the last and only thread to alter this
1323  * sub-buffer. As long as it is not delivered and read, no other thread can
1324  * alter the offset, alter the reserve_count or call the
1325  * client_buffer_end_callback on this sub-buffer.
1326  *
1327  * The only remaining threads could be the ones with pending commits. They will
1328  * have to do the deliver themselves.  Not concurrency safe in overwrite mode.
1329  * We detect corrupted subbuffers with commit and reserve counts. We keep a
1330  * corrupted sub-buffers count and push the readers across these sub-buffers.
1331  *
1332  * Not concurrency safe if a writer is stalled in a subbuffer and another writer
1333  * switches in, finding out it's corrupted.  The result will be than the old
1334  * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
1335  * will be declared corrupted too because of the commit count adjustment.
1336  *
1337  * Note : offset_old should never be 0 here.
1338  */
1339 static void ltt_reserve_switch_old_subbuf(
1340                 struct ust_channel *chan, struct ust_buffer *buf,
1341                 struct ltt_reserve_switch_offsets *offsets, u64 *tsc)
1342 {
1343         long oldidx = SUBBUF_INDEX(offsets->old - 1, chan);
1344         long commit_count, padding_size;
1345
1346         padding_size = chan->subbuf_size
1347                         - (SUBBUF_OFFSET(offsets->old - 1, chan) + 1);
1348         ltt_buffer_end(buf, *tsc, offsets->old, oldidx);
1349
1350         /*
1351          * Must write slot data before incrementing commit count.
1352          * This compiler barrier is upgraded into a smp_wmb() by the IPI
1353          * sent by get_subbuf() when it does its smp_rmb().
1354          */
1355         barrier();
1356         local_add(padding_size,
1357                   &buf->commit_count[oldidx].cc);
1358         commit_count = local_read(&buf->commit_count[oldidx].cc);
1359         ltt_check_deliver(chan, buf, offsets->old - 1, commit_count, oldidx);
1360         ltt_write_commit_counter(buf, oldidx,
1361                 offsets->old, commit_count, padding_size);
1362 }
1363
1364 /*
1365  * ltt_reserve_switch_new_subbuf: Populate new subbuffer.
1366  *
1367  * This code can be executed unordered : writers may already have written to the
1368  * sub-buffer before this code gets executed, caution.  The commit makes sure
1369  * that this code is executed before the deliver of this sub-buffer.
1370  */
1371 static void ltt_reserve_switch_new_subbuf(
1372                 struct ust_channel *chan, struct ust_buffer *buf,
1373                 struct ltt_reserve_switch_offsets *offsets, u64 *tsc)
1374 {
1375         long beginidx = SUBBUF_INDEX(offsets->begin, chan);
1376         long commit_count;
1377
1378         ltt_buffer_begin(buf, *tsc, beginidx);
1379
1380         /*
1381          * Must write slot data before incrementing commit count.
1382          * This compiler barrier is upgraded into a smp_wmb() by the IPI
1383          * sent by get_subbuf() when it does its smp_rmb().
1384          */
1385         barrier();
1386         local_add(ltt_subbuffer_header_size(),
1387                   &buf->commit_count[beginidx].cc);
1388         commit_count = local_read(&buf->commit_count[beginidx].cc);
1389         /* Check if the written buffer has to be delivered */
1390         ltt_check_deliver(chan, buf, offsets->begin, commit_count, beginidx);
1391         ltt_write_commit_counter(buf, beginidx,
1392                 offsets->begin, commit_count, ltt_subbuffer_header_size());
1393 }
1394
1395 /*
1396  * ltt_reserve_end_switch_current: finish switching current subbuffer
1397  *
1398  * Concurrency safe because we are the last and only thread to alter this
1399  * sub-buffer. As long as it is not delivered and read, no other thread can
1400  * alter the offset, alter the reserve_count or call the
1401  * client_buffer_end_callback on this sub-buffer.
1402  *
1403  * The only remaining threads could be the ones with pending commits. They will
1404  * have to do the deliver themselves.  Not concurrency safe in overwrite mode.
1405  * We detect corrupted subbuffers with commit and reserve counts. We keep a
1406  * corrupted sub-buffers count and push the readers across these sub-buffers.
1407  *
1408  * Not concurrency safe if a writer is stalled in a subbuffer and another writer
1409  * switches in, finding out it's corrupted.  The result will be than the old
1410  * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
1411  * will be declared corrupted too because of the commit count adjustment.
1412  */
1413 static void ltt_reserve_end_switch_current(
1414                 struct ust_channel *chan,
1415                 struct ust_buffer *buf,
1416                 struct ltt_reserve_switch_offsets *offsets, u64 *tsc)
1417 {
1418         long endidx = SUBBUF_INDEX(offsets->end - 1, chan);
1419         long commit_count, padding_size;
1420
1421         padding_size = chan->subbuf_size
1422                         - (SUBBUF_OFFSET(offsets->end - 1, chan) + 1);
1423
1424         ltt_buffer_end(buf, *tsc, offsets->end, endidx);
1425
1426         /*
1427          * Must write slot data before incrementing commit count.
1428          * This compiler barrier is upgraded into a smp_wmb() by the IPI
1429          * sent by get_subbuf() when it does its smp_rmb().
1430          */
1431         barrier();
1432         local_add(padding_size,
1433                   &buf->commit_count[endidx].cc);
1434         commit_count = local_read(&buf->commit_count[endidx].cc);
1435         ltt_check_deliver(chan, buf,
1436                 offsets->end - 1, commit_count, endidx);
1437         ltt_write_commit_counter(buf, endidx,
1438                 offsets->end, commit_count, padding_size);
1439 }
1440
1441 /*
1442  * Returns :
1443  * 0 if ok
1444  * !0 if execution must be aborted.
1445  */
1446 static int ltt_relay_try_switch_slow(
1447                 enum force_switch_mode mode,
1448                 struct ust_channel *chan,
1449                 struct ust_buffer *buf,
1450                 struct ltt_reserve_switch_offsets *offsets,
1451                 u64 *tsc)
1452 {
1453         long subbuf_index;
1454         long reserve_commit_diff;
1455
1456         offsets->begin = local_read(&buf->offset);
1457         offsets->old = offsets->begin;
1458         offsets->begin_switch = 0;
1459         offsets->end_switch_old = 0;
1460
1461         *tsc = trace_clock_read64();
1462
1463         if (SUBBUF_OFFSET(offsets->begin, buf->chan) != 0) {
1464                 offsets->begin = SUBBUF_ALIGN(offsets->begin, buf->chan);
1465                 offsets->end_switch_old = 1;
1466         } else {
1467                 /* we do not have to switch : buffer is empty */
1468                 return -1;
1469         }
1470         if (mode == FORCE_ACTIVE)
1471                 offsets->begin += ltt_subbuffer_header_size();
1472         /*
1473          * Always begin_switch in FORCE_ACTIVE mode.
1474          * Test new buffer integrity
1475          */
1476         subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
1477         reserve_commit_diff =
1478                 (BUFFER_TRUNC(offsets->begin, buf->chan)
1479                  >> chan->n_subbufs_order)
1480                 - (local_read(&buf->commit_count[subbuf_index].cc_sb)
1481                         & chan->commit_count_mask);
1482         if (reserve_commit_diff == 0) {
1483                 /* Next buffer not corrupted. */
1484                 if (mode == FORCE_ACTIVE
1485                     && !chan->overwrite
1486                     && offsets->begin - atomic_long_read(&buf->consumed)
1487                        >= chan->alloc_size) {
1488                         /*
1489                          * We do not overwrite non consumed buffers and we are
1490                          * full : ignore switch while tracing is active.
1491                          */
1492                         return -1;
1493                 }
1494         } else {
1495                 /*
1496                  * Next subbuffer corrupted. Force pushing reader even in normal
1497                  * mode
1498                  */
1499         }
1500         offsets->end = offsets->begin;
1501         return 0;
1502 }
1503
1504 /*
1505  * Force a sub-buffer switch for a per-cpu buffer. This operation is
1506  * completely reentrant : can be called while tracing is active with
1507  * absolutely no lock held.
1508  *
1509  * Note, however, that as a local_cmpxchg is used for some atomic
1510  * operations, this function must be called from the CPU which owns the buffer
1511  * for a ACTIVE flush.
1512  */
1513 void ltt_force_switch_lockless_slow(struct ust_buffer *buf,
1514                 enum force_switch_mode mode)
1515 {
1516         struct ust_channel *chan = buf->chan;
1517         struct ltt_reserve_switch_offsets offsets;
1518         u64 tsc;
1519
1520         offsets.size = 0;
1521
1522         /*
1523          * Perform retryable operations.
1524          */
1525         do {
1526                 if (ltt_relay_try_switch_slow(mode, chan, buf,
1527                                 &offsets, &tsc))
1528                         return;
1529         } while (local_cmpxchg(&buf->offset, offsets.old,
1530                         offsets.end) != offsets.old);
1531
1532         /*
1533          * Atomically update last_tsc. This update races against concurrent
1534          * atomic updates, but the race will always cause supplementary full TSC
1535          * events, never the opposite (missing a full TSC event when it would be
1536          * needed).
1537          */
1538         save_last_tsc(buf, tsc);
1539
1540         /*
1541          * Push the reader if necessary
1542          */
1543         if (mode == FORCE_ACTIVE) {
1544                 ltt_reserve_push_reader(chan, buf, offsets.end - 1);
1545 //ust//         ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.end - 1, chan));
1546         }
1547
1548         /*
1549          * Switch old subbuffer if needed.
1550          */
1551         if (offsets.end_switch_old) {
1552 //ust//         ltt_clear_noref_flag(rchan, buf, SUBBUF_INDEX(offsets.old - 1, rchan));
1553                 ltt_reserve_switch_old_subbuf(chan, buf, &offsets, &tsc);
1554         }
1555
1556         /*
1557          * Populate new subbuffer.
1558          */
1559         if (mode == FORCE_ACTIVE)
1560                 ltt_reserve_switch_new_subbuf(chan, buf, &offsets, &tsc);
1561 }
1562
1563 /*
1564  * Returns :
1565  * 0 if ok
1566  * !0 if execution must be aborted.
1567  */
1568 static int ltt_relay_try_reserve_slow(struct ust_channel *chan, struct ust_buffer *buf,
1569                 struct ltt_reserve_switch_offsets *offsets, size_t data_size,
1570                 u64 *tsc, unsigned int *rflags, int largest_align)
1571 {
1572         long reserve_commit_diff;
1573
1574         offsets->begin = local_read(&buf->offset);
1575         offsets->old = offsets->begin;
1576         offsets->begin_switch = 0;
1577         offsets->end_switch_current = 0;
1578         offsets->end_switch_old = 0;
1579
1580         *tsc = trace_clock_read64();
1581         if (last_tsc_overflow(buf, *tsc))
1582                 *rflags = LTT_RFLAG_ID_SIZE_TSC;
1583
1584         if (unlikely(SUBBUF_OFFSET(offsets->begin, buf->chan) == 0)) {
1585                 offsets->begin_switch = 1;              /* For offsets->begin */
1586         } else {
1587                 offsets->size = ust_get_header_size(chan,
1588                                         offsets->begin, data_size,
1589                                         &offsets->before_hdr_pad, *rflags);
1590                 offsets->size += ltt_align(offsets->begin + offsets->size,
1591                                            largest_align)
1592                                  + data_size;
1593                 if (unlikely((SUBBUF_OFFSET(offsets->begin, buf->chan) +
1594                              offsets->size) > buf->chan->subbuf_size)) {
1595                         offsets->end_switch_old = 1;    /* For offsets->old */
1596                         offsets->begin_switch = 1;      /* For offsets->begin */
1597                 }
1598         }
1599         if (unlikely(offsets->begin_switch)) {
1600                 long subbuf_index;
1601
1602                 /*
1603                  * We are typically not filling the previous buffer completely.
1604                  */
1605                 if (likely(offsets->end_switch_old))
1606                         offsets->begin = SUBBUF_ALIGN(offsets->begin,
1607                                                       buf->chan);
1608                 offsets->begin = offsets->begin + ltt_subbuffer_header_size();
1609                 /* Test new buffer integrity */
1610                 subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
1611                 reserve_commit_diff =
1612                   (BUFFER_TRUNC(offsets->begin, buf->chan)
1613                    >> chan->n_subbufs_order)
1614                   - (local_read(&buf->commit_count[subbuf_index].cc_sb)
1615                                 & chan->commit_count_mask);
1616                 if (likely(reserve_commit_diff == 0)) {
1617                         /* Next buffer not corrupted. */
1618                         if (unlikely(!chan->overwrite &&
1619                                 (SUBBUF_TRUNC(offsets->begin, buf->chan)
1620                                  - SUBBUF_TRUNC(atomic_long_read(
1621                                                         &buf->consumed),
1622                                                 buf->chan))
1623                                 >= chan->alloc_size)) {
1624                                 /*
1625                                  * We do not overwrite non consumed buffers
1626                                  * and we are full : event is lost.
1627                                  */
1628                                 local_inc(&buf->events_lost);
1629                                 return -1;
1630                         } else {
1631                                 /*
1632                                  * next buffer not corrupted, we are either in
1633                                  * overwrite mode or the buffer is not full.
1634                                  * It's safe to write in this new subbuffer.
1635                                  */
1636                         }
1637                 } else {
1638                         /*
1639                          * Next subbuffer corrupted. Drop event in normal and
1640                          * overwrite mode. Caused by either a writer OOPS or
1641                          * too many nested writes over a reserve/commit pair.
1642                          */
1643                         local_inc(&buf->events_lost);
1644                         return -1;
1645                 }
1646                 offsets->size = ust_get_header_size(chan,
1647                                         offsets->begin, data_size,
1648                                         &offsets->before_hdr_pad, *rflags);
1649                 offsets->size += ltt_align(offsets->begin + offsets->size,
1650                                            largest_align)
1651                                  + data_size;
1652                 if (unlikely((SUBBUF_OFFSET(offsets->begin, buf->chan)
1653                              + offsets->size) > buf->chan->subbuf_size)) {
1654                         /*
1655                          * Event too big for subbuffers, report error, don't
1656                          * complete the sub-buffer switch.
1657                          */
1658                         local_inc(&buf->events_lost);
1659                         return -1;
1660                 } else {
1661                         /*
1662                          * We just made a successful buffer switch and the event
1663                          * fits in the new subbuffer. Let's write.
1664                          */
1665                 }
1666         } else {
1667                 /*
1668                  * Event fits in the current buffer and we are not on a switch
1669                  * boundary. It's safe to write.
1670                  */
1671         }
1672         offsets->end = offsets->begin + offsets->size;
1673
1674         if (unlikely((SUBBUF_OFFSET(offsets->end, buf->chan)) == 0)) {
1675                 /*
1676                  * The offset_end will fall at the very beginning of the next
1677                  * subbuffer.
1678                  */
1679                 offsets->end_switch_current = 1;        /* For offsets->begin */
1680         }
1681         return 0;
1682 }
1683
1684 /**
1685  * ltt_relay_reserve_slot_lockless_slow - Atomic slot reservation in a buffer.
1686  * @trace: the trace structure to log to.
1687  * @ltt_channel: channel structure
1688  * @transport_data: data structure specific to ltt relay
1689  * @data_size: size of the variable length data to log.
1690  * @slot_size: pointer to total size of the slot (out)
1691  * @buf_offset : pointer to reserved buffer offset (out)
1692  * @tsc: pointer to the tsc at the slot reservation (out)
1693  * @cpu: cpuid
1694  *
1695  * Return : -ENOSPC if not enough space, else returns 0.
1696  * It will take care of sub-buffer switching.
1697  */
1698 int ltt_reserve_slot_lockless_slow(struct ust_trace *trace,
1699                 struct ust_channel *chan, void **transport_data,
1700                 size_t data_size, size_t *slot_size, long *buf_offset, u64 *tsc,
1701                 unsigned int *rflags, int largest_align, int cpu)
1702 {
1703         struct ust_buffer *buf = chan->buf[cpu];
1704         struct ltt_reserve_switch_offsets offsets;
1705
1706         offsets.size = 0;
1707
1708         do {
1709                 if (unlikely(ltt_relay_try_reserve_slow(chan, buf, &offsets,
1710                                 data_size, tsc, rflags, largest_align)))
1711                         return -ENOSPC;
1712         } while (unlikely(local_cmpxchg(&buf->offset, offsets.old,
1713                         offsets.end) != offsets.old));
1714
1715         /*
1716          * Atomically update last_tsc. This update races against concurrent
1717          * atomic updates, but the race will always cause supplementary full TSC
1718          * events, never the opposite (missing a full TSC event when it would be
1719          * needed).
1720          */
1721         save_last_tsc(buf, *tsc);
1722
1723         /*
1724          * Push the reader if necessary
1725          */
1726         ltt_reserve_push_reader(chan, buf, offsets.end - 1);
1727
1728         /*
1729          * Clear noref flag for this subbuffer.
1730          */
1731 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.end - 1, chan));
1732
1733         /*
1734          * Switch old subbuffer if needed.
1735          */
1736         if (unlikely(offsets.end_switch_old)) {
1737 //ust//         ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.old - 1, chan));
1738                 ltt_reserve_switch_old_subbuf(chan, buf, &offsets, tsc);
1739         }
1740
1741         /*
1742          * Populate new subbuffer.
1743          */
1744         if (unlikely(offsets.begin_switch))
1745                 ltt_reserve_switch_new_subbuf(chan, buf, &offsets, tsc);
1746
1747         if (unlikely(offsets.end_switch_current))
1748                 ltt_reserve_end_switch_current(chan, buf, &offsets, tsc);
1749
1750         *slot_size = offsets.size;
1751         *buf_offset = offsets.begin + offsets.before_hdr_pad;
1752         return 0;
1753 }
1754
1755 static struct ltt_transport ust_relay_transport = {
1756         .name = "ustrelay",
1757         .ops = {
1758                 .create_channel = ust_buffers_create_channel,
1759                 .finish_channel = ltt_relay_finish_channel,
1760                 .remove_channel = ltt_relay_remove_channel,
1761                 .wakeup_channel = ltt_relay_async_wakeup_chan,
1762         },
1763 };
1764
1765 static char initialized = 0;
1766
1767 void __attribute__((constructor)) init_ustrelay_transport(void)
1768 {
1769         if(!initialized) {
1770                 ltt_transport_register(&ust_relay_transport);
1771                 initialized = 1;
1772         }
1773 }
1774
1775 static void __attribute__((destructor)) ust_buffers_exit(void)
1776 {
1777         ltt_transport_unregister(&ust_relay_transport);
1778 }
1779
1780 size_t ltt_write_event_header_slow(struct ust_trace *trace,
1781                 struct ust_channel *channel,
1782                 struct ust_buffer *buf, long buf_offset,
1783                 u16 eID, u32 event_size,
1784                 u64 tsc, unsigned int rflags)
1785 {
1786         struct ltt_event_header header;
1787         u16 small_size;
1788
1789         switch (rflags) {
1790         case LTT_RFLAG_ID_SIZE_TSC:
1791                 header.id_time = 29 << LTT_TSC_BITS;
1792                 break;
1793         case LTT_RFLAG_ID_SIZE:
1794                 header.id_time = 30 << LTT_TSC_BITS;
1795                 break;
1796         case LTT_RFLAG_ID:
1797                 header.id_time = 31 << LTT_TSC_BITS;
1798                 break;
1799         }
1800
1801         header.id_time |= (u32)tsc & LTT_TSC_MASK;
1802         ust_buffers_write(buf, buf_offset, &header, sizeof(header));
1803         buf_offset += sizeof(header);
1804
1805         switch (rflags) {
1806         case LTT_RFLAG_ID_SIZE_TSC:
1807                 small_size = (u16)min_t(u32, event_size, LTT_MAX_SMALL_SIZE);
1808                 ust_buffers_write(buf, buf_offset,
1809                         &eID, sizeof(u16));
1810                 buf_offset += sizeof(u16);
1811                 ust_buffers_write(buf, buf_offset,
1812                         &small_size, sizeof(u16));
1813                 buf_offset += sizeof(u16);
1814                 if (small_size == LTT_MAX_SMALL_SIZE) {
1815                         ust_buffers_write(buf, buf_offset,
1816                                 &event_size, sizeof(u32));
1817                         buf_offset += sizeof(u32);
1818                 }
1819                 buf_offset += ltt_align(buf_offset, sizeof(u64));
1820                 ust_buffers_write(buf, buf_offset,
1821                         &tsc, sizeof(u64));
1822                 buf_offset += sizeof(u64);
1823                 break;
1824         case LTT_RFLAG_ID_SIZE:
1825                 small_size = (u16)min_t(u32, event_size, LTT_MAX_SMALL_SIZE);
1826                 ust_buffers_write(buf, buf_offset,
1827                         &eID, sizeof(u16));
1828                 buf_offset += sizeof(u16);
1829                 ust_buffers_write(buf, buf_offset,
1830                         &small_size, sizeof(u16));
1831                 buf_offset += sizeof(u16);
1832                 if (small_size == LTT_MAX_SMALL_SIZE) {
1833                         ust_buffers_write(buf, buf_offset,
1834                                 &event_size, sizeof(u32));
1835                         buf_offset += sizeof(u32);
1836                 }
1837                 break;
1838         case LTT_RFLAG_ID:
1839                 ust_buffers_write(buf, buf_offset,
1840                         &eID, sizeof(u16));
1841                 buf_offset += sizeof(u16);
1842                 break;
1843         }
1844
1845         return buf_offset;
1846 }