include/ringbuffer/backend_internal.h

   1 /* SPDX-License-Identifier: (GPL-2.0-only OR LGPL-2.1-only)
   2  *
   3  * lib/ringbuffer/backend_internal.h
   4  *
   5  * Ring buffer backend (internal helpers).
   6  *
   7  * Copyright (C) 2008-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   8  */
   9
  10 #ifndef _LIB_RING_BUFFER_BACKEND_INTERNAL_H
  11 #define _LIB_RING_BUFFER_BACKEND_INTERNAL_H
  12
  13 #include <ringbuffer/config.h>
  14 #include <ringbuffer/backend_types.h>
  15 #include <ringbuffer/frontend_types.h>
  16 #include <linux/string.h>
  17 #include <linux/uaccess.h>
  18
  19 /* Ring buffer backend API presented to the frontend */
  20
  21 /* Ring buffer and channel backend create/free */
  22
  23 int lib_ring_buffer_backend_create(struct lib_ring_buffer_backend *bufb,
  24                                    struct channel_backend *chan, int cpu);
  25 void channel_backend_unregister_notifiers(struct channel_backend *chanb);
  26 void lib_ring_buffer_backend_free(struct lib_ring_buffer_backend *bufb);
  27 int channel_backend_init(struct channel_backend *chanb,
  28                          const char *name,
  29                          const struct lib_ring_buffer_config *config,
  30                          void *priv, size_t subbuf_size,
  31                          size_t num_subbuf);
  32 void channel_backend_free(struct channel_backend *chanb);
  33
  34 void lib_ring_buffer_backend_reset(struct lib_ring_buffer_backend *bufb);
  35 void channel_backend_reset(struct channel_backend *chanb);
  36
  37 int lib_ring_buffer_backend_init(void);
  38 void lib_ring_buffer_backend_exit(void);
  39
  40 extern void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb,
  41                                    size_t offset, const void *src, size_t len,
  42                                    size_t pagecpy);
  43 extern void _lib_ring_buffer_memset(struct lib_ring_buffer_backend *bufb,
  44                                     size_t offset, int c, size_t len,
  45                                     size_t pagecpy);
  46 extern void _lib_ring_buffer_strcpy(struct lib_ring_buffer_backend *bufb,
  47                                    size_t offset, const char *src, size_t len,
  48                                    size_t pagecpy, int pad);
  49 extern void _lib_ring_buffer_copy_from_user_inatomic(struct lib_ring_buffer_backend *bufb,
  50                                             size_t offset, const void *src,
  51                                             size_t len, size_t pagecpy);
  52 extern void _lib_ring_buffer_strcpy_from_user_inatomic(struct lib_ring_buffer_backend *bufb,
  53                 size_t offset, const char __user *src, size_t len,
  54                 size_t pagecpy, int pad);
  55
  56 /*
  57  * Subbuffer ID bits for overwrite mode. Need to fit within a single word to be
  58  * exchanged atomically.
  59  *
  60  * Top half word, except lowest bit, belongs to "offset", which is used to keep
  61  * to count the produced buffers.  For overwrite mode, this provides the
  62  * consumer with the capacity to read subbuffers in order, handling the
  63  * situation where producers would write up to 2^15 buffers (or 2^31 for 64-bit
  64  * systems) concurrently with a single execution of get_subbuf (between offset
  65  * sampling and subbuffer ID exchange).
  66  */
  67
  68 #define HALF_ULONG_BITS         (BITS_PER_LONG >> 1)
  69
  70 #define SB_ID_OFFSET_SHIFT      (HALF_ULONG_BITS + 1)
  71 #define SB_ID_OFFSET_COUNT      (1UL << SB_ID_OFFSET_SHIFT)
  72 #define SB_ID_OFFSET_MASK       (~(SB_ID_OFFSET_COUNT - 1))
  73 /*
  74  * Lowest bit of top word half belongs to noref. Used only for overwrite mode.
  75  */
  76 #define SB_ID_NOREF_SHIFT       (SB_ID_OFFSET_SHIFT - 1)
  77 #define SB_ID_NOREF_COUNT       (1UL << SB_ID_NOREF_SHIFT)
  78 #define SB_ID_NOREF_MASK        SB_ID_NOREF_COUNT
  79 /*
  80  * In overwrite mode: lowest half of word is used for index.
  81  * Limit of 2^16 subbuffers per buffer on 32-bit, 2^32 on 64-bit.
  82  * In producer-consumer mode: whole word used for index.
  83  */
  84 #define SB_ID_INDEX_SHIFT       0
  85 #define SB_ID_INDEX_COUNT       (1UL << SB_ID_INDEX_SHIFT)
  86 #define SB_ID_INDEX_MASK        (SB_ID_NOREF_COUNT - 1)
  87
  88 /*
  89  * Construct the subbuffer id from offset, index and noref. Use only the index
  90  * for producer-consumer mode (offset and noref are only used in overwrite
  91  * mode).
  92  */
  93 static inline
  94 unsigned long subbuffer_id(const struct lib_ring_buffer_config *config,
  95                            unsigned long offset, unsigned long noref,
  96                            unsigned long index)
  97 {
  98         if (config->mode == RING_BUFFER_OVERWRITE)
  99                 return (offset << SB_ID_OFFSET_SHIFT)
 100                        | (noref << SB_ID_NOREF_SHIFT)
 101                        | index;
 102         else
 103                 return index;
 104 }
 105
 106 /*
 107  * Compare offset with the offset contained within id. Return 1 if the offset
 108  * bits are identical, else 0.
 109  */
 110 static inline
 111 int subbuffer_id_compare_offset(const struct lib_ring_buffer_config *config,
 112                                 unsigned long id, unsigned long offset)
 113 {
 114         return (id & SB_ID_OFFSET_MASK) == (offset << SB_ID_OFFSET_SHIFT);
 115 }
 116
 117 static inline
 118 unsigned long subbuffer_id_get_index(const struct lib_ring_buffer_config *config,
 119                                      unsigned long id)
 120 {
 121         if (config->mode == RING_BUFFER_OVERWRITE)
 122                 return id & SB_ID_INDEX_MASK;
 123         else
 124                 return id;
 125 }
 126
 127 static inline
 128 unsigned long subbuffer_id_is_noref(const struct lib_ring_buffer_config *config,
 129                                     unsigned long id)
 130 {
 131         if (config->mode == RING_BUFFER_OVERWRITE)
 132                 return !!(id & SB_ID_NOREF_MASK);
 133         else
 134                 return 1;
 135 }
 136
 137 /*
 138  * Only used by reader on subbuffer ID it has exclusive access to. No volatile
 139  * needed.
 140  */
 141 static inline
 142 void subbuffer_id_set_noref(const struct lib_ring_buffer_config *config,
 143                             unsigned long *id)
 144 {
 145         if (config->mode == RING_BUFFER_OVERWRITE)
 146                 *id |= SB_ID_NOREF_MASK;
 147 }
 148
 149 static inline
 150 void subbuffer_id_set_noref_offset(const struct lib_ring_buffer_config *config,
 151                                    unsigned long *id, unsigned long offset)
 152 {
 153         unsigned long tmp;
 154
 155         if (config->mode == RING_BUFFER_OVERWRITE) {
 156                 tmp = *id;
 157                 tmp &= ~SB_ID_OFFSET_MASK;
 158                 tmp |= offset << SB_ID_OFFSET_SHIFT;
 159                 tmp |= SB_ID_NOREF_MASK;
 160                 /* Volatile store, read concurrently by readers. */
 161                 WRITE_ONCE(*id, tmp);
 162         }
 163 }
 164
 165 /* No volatile access, since already used locally */
 166 static inline
 167 void subbuffer_id_clear_noref(const struct lib_ring_buffer_config *config,
 168                               unsigned long *id)
 169 {
 170         if (config->mode == RING_BUFFER_OVERWRITE)
 171                 *id &= ~SB_ID_NOREF_MASK;
 172 }
 173
 174 /*
 175  * For overwrite mode, cap the number of subbuffers per buffer to:
 176  * 2^16 on 32-bit architectures
 177  * 2^32 on 64-bit architectures
 178  * This is required to fit in the index part of the ID. Return 0 on success,
 179  * -EPERM on failure.
 180  */
 181 static inline
 182 int subbuffer_id_check_index(const struct lib_ring_buffer_config *config,
 183                              unsigned long num_subbuf)
 184 {
 185         if (config->mode == RING_BUFFER_OVERWRITE)
 186                 return (num_subbuf > (1UL << HALF_ULONG_BITS)) ? -EPERM : 0;
 187         else
 188                 return 0;
 189 }
 190
 191 static inline
 192 void lib_ring_buffer_backend_get_pages(const struct lib_ring_buffer_config *config,
 193                         struct lib_ring_buffer_ctx *ctx,
 194                         struct lib_ring_buffer_backend_pages **backend_pages)
 195 {
 196         struct lib_ring_buffer_backend *bufb = &ctx->buf->backend;
 197         struct channel_backend *chanb = &ctx->chan->backend;
 198         size_t sbidx, offset = ctx->buf_offset;
 199         unsigned long sb_bindex, id;
 200         struct lib_ring_buffer_backend_pages *rpages;
 201
 202         offset &= chanb->buf_size - 1;
 203         sbidx = offset >> chanb->subbuf_size_order;
 204         id = bufb->buf_wsb[sbidx].id;
 205         sb_bindex = subbuffer_id_get_index(config, id);
 206         rpages = bufb->array[sb_bindex];
 207         CHAN_WARN_ON(ctx->chan,
 208                      config->mode == RING_BUFFER_OVERWRITE
 209                      && subbuffer_id_is_noref(config, id));
 210         *backend_pages = rpages;
 211 }
 212
 213 /* Get backend pages from cache. */
 214 static inline
 215 struct lib_ring_buffer_backend_pages *
 216         lib_ring_buffer_get_backend_pages_from_ctx(const struct lib_ring_buffer_config *config,
 217                 struct lib_ring_buffer_ctx *ctx)
 218 {
 219         return ctx->backend_pages;
 220 }
 221
 222 /*
 223  * The ring buffer can count events recorded and overwritten per buffer,
 224  * but it is disabled by default due to its performance overhead.
 225  */
 226 #ifdef LTTNG_RING_BUFFER_COUNT_EVENTS
 227 static inline
 228 void subbuffer_count_record(const struct lib_ring_buffer_config *config,
 229                             struct lib_ring_buffer_backend *bufb,
 230                             unsigned long idx)
 231 {
 232         unsigned long sb_bindex;
 233
 234         sb_bindex = subbuffer_id_get_index(config, bufb->buf_wsb[idx].id);
 235         v_inc(config, &bufb->array[sb_bindex]->records_commit);
 236 }
 237 #else /* LTTNG_RING_BUFFER_COUNT_EVENTS */
 238 static inline
 239 void subbuffer_count_record(const struct lib_ring_buffer_config *config,
 240                             struct lib_ring_buffer_backend *bufb,
 241                             unsigned long idx)
 242 {
 243 }
 244 #endif /* #else LTTNG_RING_BUFFER_COUNT_EVENTS */
 245
 246 /*
 247  * Reader has exclusive subbuffer access for record consumption. No need to
 248  * perform the decrement atomically.
 249  */
 250 static inline
 251 void subbuffer_consume_record(const struct lib_ring_buffer_config *config,
 252                               struct lib_ring_buffer_backend *bufb)
 253 {
 254         unsigned long sb_bindex;
 255
 256         sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
 257         CHAN_WARN_ON(bufb->chan,
 258                      !v_read(config, &bufb->array[sb_bindex]->records_unread));
 259         /* Non-atomic decrement protected by exclusive subbuffer access */
 260         _v_dec(config, &bufb->array[sb_bindex]->records_unread);
 261         v_inc(config, &bufb->records_read);
 262 }
 263
 264 static inline
 265 unsigned long subbuffer_get_records_count(
 266                                 const struct lib_ring_buffer_config *config,
 267                                 struct lib_ring_buffer_backend *bufb,
 268                                 unsigned long idx)
 269 {
 270         unsigned long sb_bindex;
 271
 272         sb_bindex = subbuffer_id_get_index(config, bufb->buf_wsb[idx].id);
 273         return v_read(config, &bufb->array[sb_bindex]->records_commit);
 274 }
 275
 276 /*
 277  * Must be executed at subbuffer delivery when the writer has _exclusive_
 278  * subbuffer access. See lib_ring_buffer_check_deliver() for details.
 279  * lib_ring_buffer_get_records_count() must be called to get the records
 280  * count before this function, because it resets the records_commit
 281  * count.
 282  */
 283 static inline
 284 unsigned long subbuffer_count_records_overrun(
 285                                 const struct lib_ring_buffer_config *config,
 286                                 struct lib_ring_buffer_backend *bufb,
 287                                 unsigned long idx)
 288 {
 289         struct lib_ring_buffer_backend_pages *pages;
 290         unsigned long overruns, sb_bindex;
 291
 292         sb_bindex = subbuffer_id_get_index(config, bufb->buf_wsb[idx].id);
 293         pages = bufb->array[sb_bindex];
 294         overruns = v_read(config, &pages->records_unread);
 295         v_set(config, &pages->records_unread,
 296               v_read(config, &pages->records_commit));
 297         v_set(config, &pages->records_commit, 0);
 298
 299         return overruns;
 300 }
 301
 302 static inline
 303 void subbuffer_set_data_size(const struct lib_ring_buffer_config *config,
 304                              struct lib_ring_buffer_backend *bufb,
 305                              unsigned long idx,
 306                              unsigned long data_size)
 307 {
 308         struct lib_ring_buffer_backend_pages *pages;
 309         unsigned long sb_bindex;
 310
 311         sb_bindex = subbuffer_id_get_index(config, bufb->buf_wsb[idx].id);
 312         pages = bufb->array[sb_bindex];
 313         pages->data_size = data_size;
 314 }
 315
 316 static inline
 317 unsigned long subbuffer_get_read_data_size(
 318                                 const struct lib_ring_buffer_config *config,
 319                                 struct lib_ring_buffer_backend *bufb)
 320 {
 321         struct lib_ring_buffer_backend_pages *pages;
 322         unsigned long sb_bindex;
 323
 324         sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
 325         pages = bufb->array[sb_bindex];
 326         return pages->data_size;
 327 }
 328
 329 static inline
 330 unsigned long subbuffer_get_data_size(
 331                                 const struct lib_ring_buffer_config *config,
 332                                 struct lib_ring_buffer_backend *bufb,
 333                                 unsigned long idx)
 334 {
 335         struct lib_ring_buffer_backend_pages *pages;
 336         unsigned long sb_bindex;
 337
 338         sb_bindex = subbuffer_id_get_index(config, bufb->buf_wsb[idx].id);
 339         pages = bufb->array[sb_bindex];
 340         return pages->data_size;
 341 }
 342
 343 static inline
 344 void subbuffer_inc_packet_count(const struct lib_ring_buffer_config *config,
 345                                 struct lib_ring_buffer_backend *bufb,
 346                                 unsigned long idx)
 347 {
 348         bufb->buf_cnt[idx].seq_cnt++;
 349 }
 350
 351 /**
 352  * lib_ring_buffer_clear_noref - Clear the noref subbuffer flag, called by
 353  *                               writer.
 354  */
 355 static inline
 356 void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config,
 357                                  struct lib_ring_buffer_backend *bufb,
 358                                  unsigned long idx)
 359 {
 360         unsigned long id, new_id;
 361
 362         if (config->mode != RING_BUFFER_OVERWRITE)
 363                 return;
 364
 365         /*
 366          * Performing a volatile access to read the sb_pages, because we want to
 367          * read a coherent version of the pointer and the associated noref flag.
 368          */
 369         id = READ_ONCE(bufb->buf_wsb[idx].id);
 370         for (;;) {
 371                 /* This check is called on the fast path for each record. */
 372                 if (likely(!subbuffer_id_is_noref(config, id))) {
 373                         /*
 374                          * Store after load dependency ordering the writes to
 375                          * the subbuffer after load and test of the noref flag
 376                          * matches the memory barrier implied by the cmpxchg()
 377                          * in update_read_sb_index().
 378                          */
 379                         return; /* Already writing to this buffer */
 380                 }
 381                 new_id = id;
 382                 subbuffer_id_clear_noref(config, &new_id);
 383                 new_id = cmpxchg(&bufb->buf_wsb[idx].id, id, new_id);
 384                 if (likely(new_id == id))
 385                         break;
 386                 id = new_id;
 387         }
 388 }
 389
 390 /**
 391  * lib_ring_buffer_set_noref_offset - Set the noref subbuffer flag and offset,
 392  *                                    called by writer.
 393  */
 394 static inline
 395 void lib_ring_buffer_set_noref_offset(const struct lib_ring_buffer_config *config,
 396                                       struct lib_ring_buffer_backend *bufb,
 397                                       unsigned long idx, unsigned long offset)
 398 {
 399         if (config->mode != RING_BUFFER_OVERWRITE)
 400                 return;
 401
 402         /*
 403          * Because ring_buffer_set_noref() is only called by a single thread
 404          * (the one which updated the cc_sb value), there are no concurrent
 405          * updates to take care of: other writers have not updated cc_sb, so
 406          * they cannot set the noref flag, and concurrent readers cannot modify
 407          * the pointer because the noref flag is not set yet.
 408          * The smp_wmb() in ring_buffer_commit() takes care of ordering writes
 409          * to the subbuffer before this set noref operation.
 410          * subbuffer_set_noref() uses a volatile store to deal with concurrent
 411          * readers of the noref flag.
 412          */
 413         CHAN_WARN_ON(bufb->chan,
 414                      subbuffer_id_is_noref(config, bufb->buf_wsb[idx].id));
 415         /*
 416          * Memory barrier that ensures counter stores are ordered before set
 417          * noref and offset.
 418          */
 419         smp_mb();
 420         subbuffer_id_set_noref_offset(config, &bufb->buf_wsb[idx].id, offset);
 421 }
 422
 423 /**
 424  * update_read_sb_index - Read-side subbuffer index update.
 425  */
 426 static inline
 427 int update_read_sb_index(const struct lib_ring_buffer_config *config,
 428                          struct lib_ring_buffer_backend *bufb,
 429                          struct channel_backend *chanb,
 430                          unsigned long consumed_idx,
 431                          unsigned long consumed_count)
 432 {
 433         unsigned long old_id, new_id;
 434
 435         if (config->mode == RING_BUFFER_OVERWRITE) {
 436                 /*
 437                  * Exchange the target writer subbuffer with our own unused
 438                  * subbuffer. No need to use READ_ONCE() here to read the
 439                  * old_wpage, because the value read will be confirmed by the
 440                  * following cmpxchg().
 441                  */
 442                 old_id = bufb->buf_wsb[consumed_idx].id;
 443                 if (unlikely(!subbuffer_id_is_noref(config, old_id)))
 444                         return -EAGAIN;
 445                 /*
 446                  * Make sure the offset count we are expecting matches the one
 447                  * indicated by the writer.
 448                  */
 449                 if (unlikely(!subbuffer_id_compare_offset(config, old_id,
 450                                                           consumed_count)))
 451                         return -EAGAIN;
 452                 CHAN_WARN_ON(bufb->chan,
 453                              !subbuffer_id_is_noref(config, bufb->buf_rsb.id));
 454                 subbuffer_id_set_noref_offset(config, &bufb->buf_rsb.id,
 455                                               consumed_count);
 456                 new_id = cmpxchg(&bufb->buf_wsb[consumed_idx].id, old_id,
 457                                  bufb->buf_rsb.id);
 458                 if (unlikely(old_id != new_id))
 459                         return -EAGAIN;
 460                 bufb->buf_rsb.id = new_id;
 461         } else {
 462                 /* No page exchange, use the writer page directly */
 463                 bufb->buf_rsb.id = bufb->buf_wsb[consumed_idx].id;
 464         }
 465         return 0;
 466 }
 467
 468 static inline __attribute__((always_inline))
 469 void lttng_inline_memcpy(void *dest, const void *src,
 470                 unsigned long len)
 471 {
 472         switch (len) {
 473         case 1:
 474                 *(uint8_t *) dest = *(const uint8_t *) src;
 475                 break;
 476         case 2:
 477                 *(uint16_t *) dest = *(const uint16_t *) src;
 478                 break;
 479         case 4:
 480                 *(uint32_t *) dest = *(const uint32_t *) src;
 481                 break;
 482         case 8:
 483                 *(uint64_t *) dest = *(const uint64_t *) src;
 484                 break;
 485         default:
 486                 memcpy(dest, src, len);
 487         }
 488 }
 489
 490 /*
 491  * Use the architecture-specific memcpy implementation for constant-sized
 492  * inputs, but rely on an inline memcpy for length statically unknown.
 493  * The function call to memcpy is just way too expensive for a fast path.
 494  */
 495 #define lib_ring_buffer_do_copy(config, dest, src, len)         \
 496 do {                                                            \
 497         size_t __len = (len);                                   \
 498         if (__builtin_constant_p(len))                          \
 499                 memcpy(dest, src, __len);                       \
 500         else                                                    \
 501                 lttng_inline_memcpy(dest, src, __len);          \
 502 } while (0)
 503
 504 /*
 505  * We use __copy_from_user_inatomic to copy userspace data since we already
 506  * did the access_ok for the whole range.
 507  *
 508  * Return 0 if OK, nonzero on error.
 509  */
 510 static inline
 511 unsigned long lib_ring_buffer_do_copy_from_user_inatomic(void *dest,
 512                                                 const void __user *src,
 513                                                 unsigned long len)
 514 {
 515         return __copy_from_user_inatomic(dest, src, len);
 516 }
 517
 518 /*
 519  * write len bytes to dest with c
 520  */
 521 static inline
 522 void lib_ring_buffer_do_memset(char *dest, int c,
 523         unsigned long len)
 524 {
 525         unsigned long i;
 526
 527         for (i = 0; i < len; i++)
 528                 dest[i] = c;
 529 }
 530
 531 #endif /* _LIB_RING_BUFFER_BACKEND_INTERNAL_H */