libringbuffer/frontend_internal.h

   1 #ifndef _LTTNG_RING_BUFFER_FRONTEND_INTERNAL_H
   2 #define _LTTNG_RING_BUFFER_FRONTEND_INTERNAL_H
   3
   4 /*
   5  * libringbuffer/frontend_internal.h
   6  *
   7  * Ring Buffer Library Synchronization Header (internal helpers).
   8  *
   9  * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Lesser General Public
  13  * License as published by the Free Software Foundation; only
  14  * version 2.1 of the License.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  19  * Lesser General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Lesser General Public
  22  * License along with this library; if not, write to the Free Software
  23  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  24  *
  25  *
  26  * Author:
  27  *      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  28  *
  29  * See ring_buffer_frontend.c for more information on wait-free algorithms.
  30  *
  31  * Dual LGPL v2.1/GPL v2 license.
  32  */
  33
  34 #include <urcu/compiler.h>
  35 #include <urcu/tls-compat.h>
  36 #include <signal.h>
  37 #include <stdint.h>
  38 #include <pthread.h>
  39
  40 #include <lttng/ringbuffer-config.h>
  41 #include "backend_types.h"
  42 #include "frontend_types.h"
  43 #include "shm.h"
  44
  45 /* Buffer offset macros */
  46
  47 /* buf_trunc mask selects only the buffer number. */
  48 static inline
  49 unsigned long buf_trunc(unsigned long offset, struct channel *chan)
  50 {
  51         return offset & ~(chan->backend.buf_size - 1);
  52
  53 }
  54
  55 /* Select the buffer number value (counter). */
  56 static inline
  57 unsigned long buf_trunc_val(unsigned long offset, struct channel *chan)
  58 {
  59         return buf_trunc(offset, chan) >> chan->backend.buf_size_order;
  60 }
  61
  62 /* buf_offset mask selects only the offset within the current buffer. */
  63 static inline
  64 unsigned long buf_offset(unsigned long offset, struct channel *chan)
  65 {
  66         return offset & (chan->backend.buf_size - 1);
  67 }
  68
  69 /* subbuf_offset mask selects the offset within the current subbuffer. */
  70 static inline
  71 unsigned long subbuf_offset(unsigned long offset, struct channel *chan)
  72 {
  73         return offset & (chan->backend.subbuf_size - 1);
  74 }
  75
  76 /* subbuf_trunc mask selects the subbuffer number. */
  77 static inline
  78 unsigned long subbuf_trunc(unsigned long offset, struct channel *chan)
  79 {
  80         return offset & ~(chan->backend.subbuf_size - 1);
  81 }
  82
  83 /* subbuf_align aligns the offset to the next subbuffer. */
  84 static inline
  85 unsigned long subbuf_align(unsigned long offset, struct channel *chan)
  86 {
  87         return (offset + chan->backend.subbuf_size)
  88                & ~(chan->backend.subbuf_size - 1);
  89 }
  90
  91 /* subbuf_index returns the index of the current subbuffer within the buffer. */
  92 static inline
  93 unsigned long subbuf_index(unsigned long offset, struct channel *chan)
  94 {
  95         return buf_offset(offset, chan) >> chan->backend.subbuf_size_order;
  96 }
  97
  98 /*
  99  * Last TSC comparison functions. Check if the current TSC overflows tsc_bits
 100  * bits from the last TSC read. When overflows are detected, the full 64-bit
 101  * timestamp counter should be written in the record header. Reads and writes
 102  * last_tsc atomically.
 103  */
 104
 105 #if (CAA_BITS_PER_LONG == 32)
 106 static inline
 107 void save_last_tsc(const struct lttng_ust_lib_ring_buffer_config *config,
 108                    struct lttng_ust_lib_ring_buffer *buf, uint64_t tsc)
 109 {
 110         if (config->tsc_bits == 0 || config->tsc_bits == 64)
 111                 return;
 112
 113         /*
 114          * Ensure the compiler performs this update in a single instruction.
 115          */
 116         v_set(config, &buf->last_tsc, (unsigned long)(tsc >> config->tsc_bits));
 117 }
 118
 119 static inline
 120 int last_tsc_overflow(const struct lttng_ust_lib_ring_buffer_config *config,
 121                       struct lttng_ust_lib_ring_buffer *buf, uint64_t tsc)
 122 {
 123         unsigned long tsc_shifted;
 124
 125         if (config->tsc_bits == 0 || config->tsc_bits == 64)
 126                 return 0;
 127
 128         tsc_shifted = (unsigned long)(tsc >> config->tsc_bits);
 129         if (caa_unlikely(tsc_shifted
 130                      - (unsigned long)v_read(config, &buf->last_tsc)))
 131                 return 1;
 132         else
 133                 return 0;
 134 }
 135 #else
 136 static inline
 137 void save_last_tsc(const struct lttng_ust_lib_ring_buffer_config *config,
 138                    struct lttng_ust_lib_ring_buffer *buf, uint64_t tsc)
 139 {
 140         if (config->tsc_bits == 0 || config->tsc_bits == 64)
 141                 return;
 142
 143         v_set(config, &buf->last_tsc, (unsigned long)tsc);
 144 }
 145
 146 static inline
 147 int last_tsc_overflow(const struct lttng_ust_lib_ring_buffer_config *config,
 148                       struct lttng_ust_lib_ring_buffer *buf, uint64_t tsc)
 149 {
 150         if (config->tsc_bits == 0 || config->tsc_bits == 64)
 151                 return 0;
 152
 153         if (caa_unlikely((tsc - v_read(config, &buf->last_tsc))
 154                      >> config->tsc_bits))
 155                 return 1;
 156         else
 157                 return 0;
 158 }
 159 #endif
 160
 161 extern
 162 int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx,
 163                 void *client_ctx);
 164
 165 extern
 166 void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf,
 167                                  enum switch_mode mode,
 168                                  struct lttng_ust_shm_handle *handle);
 169
 170 void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_lib_ring_buffer_config *config,
 171                                    struct lttng_ust_lib_ring_buffer *buf,
 172                                    struct channel *chan,
 173                                    unsigned long offset,
 174                                    unsigned long commit_count,
 175                                    unsigned long idx,
 176                                    struct lttng_ust_shm_handle *handle,
 177                                    uint64_t tsc);
 178
 179 /* Buffer write helpers */
 180
 181 static inline
 182 void lib_ring_buffer_reserve_push_reader(struct lttng_ust_lib_ring_buffer *buf,
 183                                          struct channel *chan,
 184                                          unsigned long offset)
 185 {
 186         unsigned long consumed_old, consumed_new;
 187
 188         do {
 189                 consumed_old = uatomic_read(&buf->consumed);
 190                 /*
 191                  * If buffer is in overwrite mode, push the reader consumed
 192                  * count if the write position has reached it and we are not
 193                  * at the first iteration (don't push the reader farther than
 194                  * the writer). This operation can be done concurrently by many
 195                  * writers in the same buffer, the writer being at the farthest
 196                  * write position sub-buffer index in the buffer being the one
 197                  * which will win this loop.
 198                  */
 199                 if (caa_unlikely(subbuf_trunc(offset, chan)
 200                               - subbuf_trunc(consumed_old, chan)
 201                              >= chan->backend.buf_size))
 202                         consumed_new = subbuf_align(consumed_old, chan);
 203                 else
 204                         return;
 205         } while (caa_unlikely(uatomic_cmpxchg(&buf->consumed, consumed_old,
 206                                               consumed_new) != consumed_old));
 207 }
 208
 209 /*
 210  * Move consumed position to the beginning of subbuffer in which the
 211  * write offset is. Should only be used on ring buffers that are not
 212  * actively being written into, because clear_reader does not take into
 213  * account the commit counters when moving the consumed position, which
 214  * can make concurrent trace producers or consumers observe consumed
 215  * position further than the write offset, which breaks ring buffer
 216  * algorithm guarantees.
 217  */
 218 static inline
 219 void lib_ring_buffer_clear_reader(struct lttng_ust_lib_ring_buffer *buf,
 220                                   struct lttng_ust_shm_handle *handle)
 221 {
 222         struct channel *chan;
 223         const struct lttng_ust_lib_ring_buffer_config *config;
 224         unsigned long offset, consumed_old, consumed_new;
 225
 226         chan = shmp(handle, buf->backend.chan);
 227         if (!chan)
 228                 return;
 229         config = &chan->backend.config;
 230
 231         do {
 232                 offset = v_read(config, &buf->offset);
 233                 consumed_old = uatomic_read(&buf->consumed);
 234                 CHAN_WARN_ON(chan, (long) (subbuf_trunc(offset, chan)
 235                                 - subbuf_trunc(consumed_old, chan))
 236                                 < 0);
 237                 consumed_new = subbuf_trunc(offset, chan);
 238         } while (caa_unlikely(uatomic_cmpxchg(&buf->consumed, consumed_old,
 239                                               consumed_new) != consumed_old));
 240 }
 241
 242 static inline
 243 int lib_ring_buffer_pending_data(const struct lttng_ust_lib_ring_buffer_config *config,
 244                                  struct lttng_ust_lib_ring_buffer *buf,
 245                                  struct channel *chan)
 246 {
 247         return !!subbuf_offset(v_read(config, &buf->offset), chan);
 248 }
 249
 250 static inline
 251 unsigned long lib_ring_buffer_get_data_size(const struct lttng_ust_lib_ring_buffer_config *config,
 252                                             struct lttng_ust_lib_ring_buffer *buf,
 253                                             unsigned long idx,
 254                                             struct lttng_ust_shm_handle *handle)
 255 {
 256         return subbuffer_get_data_size(config, &buf->backend, idx, handle);
 257 }
 258
 259 /*
 260  * Check if all space reservation in a buffer have been committed. This helps
 261  * knowing if an execution context is nested (for per-cpu buffers only).
 262  * This is a very specific ftrace use-case, so we keep this as "internal" API.
 263  */
 264 static inline
 265 int lib_ring_buffer_reserve_committed(const struct lttng_ust_lib_ring_buffer_config *config,
 266                                       struct lttng_ust_lib_ring_buffer *buf,
 267                                       struct channel *chan,
 268                                       struct lttng_ust_shm_handle *handle)
 269 {
 270         unsigned long offset, idx, commit_count;
 271         struct commit_counters_hot *cc_hot;
 272
 273         CHAN_WARN_ON(chan, config->alloc != RING_BUFFER_ALLOC_PER_CPU);
 274         CHAN_WARN_ON(chan, config->sync != RING_BUFFER_SYNC_PER_CPU);
 275
 276         /*
 277          * Read offset and commit count in a loop so they are both read
 278          * atomically wrt interrupts. By deal with interrupt concurrency by
 279          * restarting both reads if the offset has been pushed. Note that given
 280          * we only have to deal with interrupt concurrency here, an interrupt
 281          * modifying the commit count will also modify "offset", so it is safe
 282          * to only check for offset modifications.
 283          */
 284         do {
 285                 offset = v_read(config, &buf->offset);
 286                 idx = subbuf_index(offset, chan);
 287                 cc_hot = shmp_index(handle, buf->commit_hot, idx);
 288                 if (caa_unlikely(!cc_hot))
 289                         return 0;
 290                 commit_count = v_read(config, &cc_hot->cc);
 291         } while (offset != v_read(config, &buf->offset));
 292
 293         return ((buf_trunc(offset, chan) >> chan->backend.num_subbuf_order)
 294                      - (commit_count & chan->commit_count_mask) == 0);
 295 }
 296
 297 /*
 298  * Receive end of subbuffer TSC as parameter. It has been read in the
 299  * space reservation loop of either reserve or switch, which ensures it
 300  * progresses monotonically with event records in the buffer. Therefore,
 301  * it ensures that the end timestamp of a subbuffer is <= begin
 302  * timestamp of the following subbuffers.
 303  */
 304 static inline
 305 void lib_ring_buffer_check_deliver(const struct lttng_ust_lib_ring_buffer_config *config,
 306                                    struct lttng_ust_lib_ring_buffer *buf,
 307                                    struct channel *chan,
 308                                    unsigned long offset,
 309                                    unsigned long commit_count,
 310                                    unsigned long idx,
 311                                    struct lttng_ust_shm_handle *handle,
 312                                    uint64_t tsc)
 313 {
 314         unsigned long old_commit_count = commit_count
 315                                          - chan->backend.subbuf_size;
 316
 317         /* Check if all commits have been done */
 318         if (caa_unlikely((buf_trunc(offset, chan) >> chan->backend.num_subbuf_order)
 319                      - (old_commit_count & chan->commit_count_mask) == 0))
 320                 lib_ring_buffer_check_deliver_slow(config, buf, chan, offset,
 321                         commit_count, idx, handle, tsc);
 322 }
 323
 324 /*
 325  * lib_ring_buffer_write_commit_counter
 326  *
 327  * For flight recording. must be called after commit.
 328  * This function increments the subbuffer's commit_seq counter each time the
 329  * commit count reaches back the reserve offset (modulo subbuffer size). It is
 330  * useful for crash dump.
 331  */
 332 static inline
 333 void lib_ring_buffer_write_commit_counter(const struct lttng_ust_lib_ring_buffer_config *config,
 334                                           struct lttng_ust_lib_ring_buffer *buf,
 335                                           struct channel *chan,
 336                                           unsigned long buf_offset,
 337                                           unsigned long commit_count,
 338                                           struct lttng_ust_shm_handle *handle,
 339                                           struct commit_counters_hot *cc_hot)
 340 {
 341         unsigned long commit_seq_old;
 342
 343         if (config->oops != RING_BUFFER_OOPS_CONSISTENCY)
 344                 return;
 345
 346         /*
 347          * subbuf_offset includes commit_count_mask. We can simply
 348          * compare the offsets within the subbuffer without caring about
 349          * buffer full/empty mismatch because offset is never zero here
 350          * (subbuffer header and record headers have non-zero length).
 351          */
 352         if (caa_unlikely(subbuf_offset(buf_offset - commit_count, chan)))
 353                 return;
 354
 355         commit_seq_old = v_read(config, &cc_hot->seq);
 356         if (caa_likely((long) (commit_seq_old - commit_count) < 0))
 357                 v_set(config, &cc_hot->seq, commit_count);
 358 }
 359
 360 extern int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer *buf,
 361                                   struct channel_backend *chanb, int cpu,
 362                                   struct lttng_ust_shm_handle *handle,
 363                                   struct shm_object *shmobj);
 364 extern void lib_ring_buffer_free(struct lttng_ust_lib_ring_buffer *buf,
 365                                  struct lttng_ust_shm_handle *handle);
 366
 367 /* Keep track of trap nesting inside ring buffer code */
 368 extern DECLARE_URCU_TLS(unsigned int, lib_ring_buffer_nesting);
 369
 370 #endif /* _LTTNG_RING_BUFFER_FRONTEND_INTERNAL_H */