libringbuffer/ring_buffer_backend.c

   1 /*
   2  * SPDX-License-Identifier: LGPL-2.1-only
   3  *
   4  * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  */
   6
   7 #define _LGPL_SOURCE
   8 #include <stddef.h>
   9 #include <stdint.h>
  10 #include <unistd.h>
  11 #include <urcu/arch.h>
  12 #include <limits.h>
  13
  14 #include <lttng/align.h>
  15 #include <lttng/ringbuffer-config.h>
  16 #include "vatomic.h"
  17 #include "backend.h"
  18 #include "frontend.h"
  19 #include "smp.h"
  20 #include "shm.h"
  21
  22 /**
  23  * lib_ring_buffer_backend_allocate - allocate a channel buffer
  24  * @config: ring buffer instance configuration
  25  * @buf: the buffer struct
  26  * @size: total size of the buffer
  27  * @num_subbuf: number of subbuffers
  28  * @extra_reader_sb: need extra subbuffer for reader
  29  */
  30 static
  31 int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_config *config,
  32                                      struct lttng_ust_lib_ring_buffer_backend *bufb,
  33                                      size_t size, size_t num_subbuf,
  34                                      int extra_reader_sb,
  35                                      struct lttng_ust_shm_handle *handle,
  36                                      struct shm_object *shmobj)
  37 {
  38         struct channel_backend *chanb;
  39         unsigned long subbuf_size, mmap_offset = 0;
  40         unsigned long num_subbuf_alloc;
  41         unsigned long i;
  42         long page_size;
  43
  44         chanb = &shmp(handle, bufb->chan)->backend;
  45         if (!chanb)
  46                 return -EINVAL;
  47
  48         subbuf_size = chanb->subbuf_size;
  49         num_subbuf_alloc = num_subbuf;
  50
  51         if (extra_reader_sb)
  52                 num_subbuf_alloc++;
  53
  54         page_size = LTTNG_UST_PAGE_SIZE;
  55         if (page_size <= 0) {
  56                 goto page_size_error;
  57         }
  58
  59         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp));
  60         set_shmp(bufb->array, zalloc_shm(shmobj,
  61                         sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp) * num_subbuf_alloc));
  62         if (caa_unlikely(!shmp(handle, bufb->array)))
  63                 goto array_error;
  64
  65         /*
  66          * This is the largest element (the buffer pages) which needs to
  67          * be aligned on page size.
  68          */
  69         align_shm(shmobj, page_size);
  70         set_shmp(bufb->memory_map, zalloc_shm(shmobj,
  71                         subbuf_size * num_subbuf_alloc));
  72         if (caa_unlikely(!shmp(handle, bufb->memory_map)))
  73                 goto memory_map_error;
  74
  75         /* Allocate backend pages array elements */
  76         for (i = 0; i < num_subbuf_alloc; i++) {
  77                 align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages));
  78                 set_shmp(shmp_index(handle, bufb->array, i)->shmp,
  79                         zalloc_shm(shmobj,
  80                                 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages)));
  81                 if (!shmp(handle, shmp_index(handle, bufb->array, i)->shmp))
  82                         goto free_array;
  83         }
  84
  85         /* Allocate write-side subbuffer table */
  86         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer));
  87         set_shmp(bufb->buf_wsb, zalloc_shm(shmobj,
  88                                 sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer)
  89                                 * num_subbuf));
  90         if (caa_unlikely(!shmp(handle, bufb->buf_wsb)))
  91                 goto free_array;
  92
  93         for (i = 0; i < num_subbuf; i++) {
  94                 struct lttng_ust_lib_ring_buffer_backend_subbuffer *sb;
  95
  96                 sb = shmp_index(handle, bufb->buf_wsb, i);
  97                 if (!sb)
  98                         goto free_array;
  99                 sb->id = subbuffer_id(config, 0, 1, i);
 100         }
 101
 102         /* Assign read-side subbuffer table */
 103         if (extra_reader_sb)
 104                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
 105                                                 num_subbuf_alloc - 1);
 106         else
 107                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 108
 109         /* Allocate subbuffer packet counter table */
 110         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_counts));
 111         set_shmp(bufb->buf_cnt, zalloc_shm(shmobj,
 112                                 sizeof(struct lttng_ust_lib_ring_buffer_backend_counts)
 113                                 * num_subbuf));
 114         if (caa_unlikely(!shmp(handle, bufb->buf_cnt)))
 115                 goto free_wsb;
 116
 117         /* Assign pages to page index */
 118         for (i = 0; i < num_subbuf_alloc; i++) {
 119                 struct lttng_ust_lib_ring_buffer_backend_pages_shmp *sbp;
 120                 struct lttng_ust_lib_ring_buffer_backend_pages *pages;
 121                 struct shm_ref ref;
 122
 123                 ref.index = bufb->memory_map._ref.index;
 124                 ref.offset = bufb->memory_map._ref.offset;
 125                 ref.offset += i * subbuf_size;
 126
 127                 sbp = shmp_index(handle, bufb->array, i);
 128                 if (!sbp)
 129                         goto free_array;
 130                 pages = shmp(handle, sbp->shmp);
 131                 if (!pages)
 132                         goto free_array;
 133                 set_shmp(pages->p, ref);
 134                 if (config->output == RING_BUFFER_MMAP) {
 135                         pages->mmap_offset = mmap_offset;
 136                         mmap_offset += subbuf_size;
 137                 }
 138         }
 139         return 0;
 140
 141 free_wsb:
 142         /* bufb->buf_wsb will be freed by shm teardown */
 143 free_array:
 144         /* bufb->array[i] will be freed by shm teardown */
 145 memory_map_error:
 146         /* bufb->array will be freed by shm teardown */
 147 array_error:
 148 page_size_error:
 149         return -ENOMEM;
 150 }
 151
 152 int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend *bufb,
 153                                    struct channel_backend *chanb, int cpu,
 154                                    struct lttng_ust_shm_handle *handle,
 155                                    struct shm_object *shmobj)
 156 {
 157         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 158
 159         set_shmp(bufb->chan, handle->chan._ref);
 160         bufb->cpu = cpu;
 161
 162         return lib_ring_buffer_backend_allocate(config, bufb, chanb->buf_size,
 163                                                 chanb->num_subbuf,
 164                                                 chanb->extra_reader_sb,
 165                                                 handle, shmobj);
 166 }
 167
 168 void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend *bufb,
 169                                    struct lttng_ust_shm_handle *handle)
 170 {
 171         struct channel_backend *chanb;
 172         const struct lttng_ust_lib_ring_buffer_config *config;
 173         unsigned long num_subbuf_alloc;
 174         unsigned int i;
 175
 176         chanb = &shmp(handle, bufb->chan)->backend;
 177         if (!chanb)
 178                 return;
 179         config = &chanb->config;
 180
 181         num_subbuf_alloc = chanb->num_subbuf;
 182         if (chanb->extra_reader_sb)
 183                 num_subbuf_alloc++;
 184
 185         for (i = 0; i < chanb->num_subbuf; i++) {
 186                 struct lttng_ust_lib_ring_buffer_backend_subbuffer *sb;
 187
 188                 sb = shmp_index(handle, bufb->buf_wsb, i);
 189                 if (!sb)
 190                         return;
 191                 sb->id = subbuffer_id(config, 0, 1, i);
 192         }
 193         if (chanb->extra_reader_sb)
 194                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
 195                                                 num_subbuf_alloc - 1);
 196         else
 197                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 198
 199         for (i = 0; i < num_subbuf_alloc; i++) {
 200                 struct lttng_ust_lib_ring_buffer_backend_pages_shmp *sbp;
 201                 struct lttng_ust_lib_ring_buffer_backend_pages *pages;
 202
 203                 sbp = shmp_index(handle, bufb->array, i);
 204                 if (!sbp)
 205                         return;
 206                 pages = shmp(handle, sbp->shmp);
 207                 if (!pages)
 208                         return;
 209                 /* Don't reset mmap_offset */
 210                 v_set(config, &pages->records_commit, 0);
 211                 v_set(config, &pages->records_unread, 0);
 212                 pages->data_size = 0;
 213                 /* Don't reset backend page and virt addresses */
 214         }
 215         /* Don't reset num_pages_per_subbuf, cpu, allocated */
 216         v_set(config, &bufb->records_read, 0);
 217 }
 218
 219 /*
 220  * The frontend is responsible for also calling ring_buffer_backend_reset for
 221  * each buffer when calling channel_backend_reset.
 222  */
 223 void channel_backend_reset(struct channel_backend *chanb)
 224 {
 225         struct channel *chan = caa_container_of(chanb, struct channel, backend);
 226         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 227
 228         /*
 229          * Don't reset buf_size, subbuf_size, subbuf_size_order,
 230          * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
 231          * priv, notifiers, config, cpumask and name.
 232          */
 233         chanb->start_tsc = config->cb.ring_buffer_clock_read(chan);
 234 }
 235
 236 /**
 237  * channel_backend_init - initialize a channel backend
 238  * @chanb: channel backend
 239  * @name: channel name
 240  * @config: client ring buffer configuration
 241  * @parent: dentry of parent directory, %NULL for root directory
 242  * @subbuf_size: size of sub-buffers (> page size, power of 2)
 243  * @num_subbuf: number of sub-buffers (power of 2)
 244  * @lttng_ust_shm_handle: shared memory handle
 245  * @stream_fds: stream file descriptors.
 246  *
 247  * Returns channel pointer if successful, %NULL otherwise.
 248  *
 249  * Creates per-cpu channel buffers using the sizes and attributes
 250  * specified.  The created channel buffer files will be named
 251  * name_0...name_N-1.  File permissions will be %S_IRUSR.
 252  *
 253  * Called with CPU hotplug disabled.
 254  */
 255 int channel_backend_init(struct channel_backend *chanb,
 256                          const char *name,
 257                          const struct lttng_ust_lib_ring_buffer_config *config,
 258                          size_t subbuf_size, size_t num_subbuf,
 259                          struct lttng_ust_shm_handle *handle,
 260                          const int *stream_fds)
 261 {
 262         struct channel *chan = caa_container_of(chanb, struct channel, backend);
 263         unsigned int i;
 264         int ret;
 265         size_t shmsize = 0, num_subbuf_alloc;
 266         long page_size;
 267
 268         if (!name)
 269                 return -EPERM;
 270
 271         page_size = LTTNG_UST_PAGE_SIZE;
 272         if (page_size <= 0) {
 273                 return -ENOMEM;
 274         }
 275         /* Check that the subbuffer size is larger than a page. */
 276         if (subbuf_size < page_size)
 277                 return -EINVAL;
 278
 279         /*
 280          * Make sure the number of subbuffers and subbuffer size are
 281          * power of 2, and nonzero.
 282          */
 283         if (!subbuf_size || (subbuf_size & (subbuf_size - 1)))
 284                 return -EINVAL;
 285         if (!num_subbuf || (num_subbuf & (num_subbuf - 1)))
 286                 return -EINVAL;
 287         /*
 288          * Overwrite mode buffers require at least 2 subbuffers per
 289          * buffer.
 290          */
 291         if (config->mode == RING_BUFFER_OVERWRITE && num_subbuf < 2)
 292                 return -EINVAL;
 293
 294         ret = subbuffer_id_check_index(config, num_subbuf);
 295         if (ret)
 296                 return ret;
 297
 298         chanb->buf_size = num_subbuf * subbuf_size;
 299         chanb->subbuf_size = subbuf_size;
 300         chanb->buf_size_order = get_count_order(chanb->buf_size);
 301         chanb->subbuf_size_order = get_count_order(subbuf_size);
 302         chanb->num_subbuf_order = get_count_order(num_subbuf);
 303         chanb->extra_reader_sb =
 304                         (config->mode == RING_BUFFER_OVERWRITE) ? 1 : 0;
 305         chanb->num_subbuf = num_subbuf;
 306         strncpy(chanb->name, name, NAME_MAX);
 307         chanb->name[NAME_MAX - 1] = '\0';
 308         memcpy(&chanb->config, config, sizeof(*config));
 309
 310         /* Per-cpu buffer size: control (prior to backend) */
 311         shmsize = lttng_ust_offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer));
 312         shmsize += sizeof(struct lttng_ust_lib_ring_buffer);
 313         shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct commit_counters_hot));
 314         shmsize += sizeof(struct commit_counters_hot) * num_subbuf;
 315         shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct commit_counters_cold));
 316         shmsize += sizeof(struct commit_counters_cold) * num_subbuf;
 317         /* Sampled timestamp end */
 318         shmsize += lttng_ust_offset_align(shmsize, __alignof__(uint64_t));
 319         shmsize += sizeof(uint64_t) * num_subbuf;
 320
 321         /* Per-cpu buffer size: backend */
 322         /* num_subbuf + 1 is the worse case */
 323         num_subbuf_alloc = num_subbuf + 1;
 324         shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp));
 325         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp) * num_subbuf_alloc;
 326         shmsize += lttng_ust_offset_align(shmsize, page_size);
 327         shmsize += subbuf_size * num_subbuf_alloc;
 328         shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages));
 329         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_pages) * num_subbuf_alloc;
 330         shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer));
 331         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer) * num_subbuf;
 332         shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_counts));
 333         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_counts) * num_subbuf;
 334
 335         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 336                 struct lttng_ust_lib_ring_buffer *buf;
 337                 /*
 338                  * We need to allocate for all possible cpus.
 339                  */
 340                 for_each_possible_cpu(i) {
 341                         struct shm_object *shmobj;
 342
 343                         shmobj = shm_object_table_alloc(handle->table, shmsize,
 344                                         SHM_OBJECT_SHM, stream_fds[i], i);
 345                         if (!shmobj)
 346                                 goto end;
 347                         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer));
 348                         set_shmp(chanb->buf[i].shmp, zalloc_shm(shmobj, sizeof(struct lttng_ust_lib_ring_buffer)));
 349                         buf = shmp(handle, chanb->buf[i].shmp);
 350                         if (!buf)
 351                                 goto end;
 352                         set_shmp(buf->self, chanb->buf[i].shmp._ref);
 353                         ret = lib_ring_buffer_create(buf, chanb, i,
 354                                         handle, shmobj);
 355                         if (ret)
 356                                 goto free_bufs; /* cpu hotplug locked */
 357                 }
 358         } else {
 359                 struct shm_object *shmobj;
 360                 struct lttng_ust_lib_ring_buffer *buf;
 361
 362                 shmobj = shm_object_table_alloc(handle->table, shmsize,
 363                                         SHM_OBJECT_SHM, stream_fds[0], -1);
 364                 if (!shmobj)
 365                         goto end;
 366                 align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer));
 367                 set_shmp(chanb->buf[0].shmp, zalloc_shm(shmobj, sizeof(struct lttng_ust_lib_ring_buffer)));
 368                 buf = shmp(handle, chanb->buf[0].shmp);
 369                 if (!buf)
 370                         goto end;
 371                 set_shmp(buf->self, chanb->buf[0].shmp._ref);
 372                 ret = lib_ring_buffer_create(buf, chanb, -1,
 373                                         handle, shmobj);
 374                 if (ret)
 375                         goto free_bufs;
 376         }
 377         chanb->start_tsc = config->cb.ring_buffer_clock_read(chan);
 378
 379         return 0;
 380
 381 free_bufs:
 382         /* We only free the buffer data upon shm teardown */
 383 end:
 384         return -ENOMEM;
 385 }
 386
 387 /**
 388  * channel_backend_free - destroy the channel
 389  * @chan: the channel
 390  *
 391  * Destroy all channel buffers and frees the channel.
 392  */
 393 void channel_backend_free(struct channel_backend *chanb,
 394                           struct lttng_ust_shm_handle *handle)
 395 {
 396         /* SHM teardown takes care of everything */
 397 }
 398
 399 /**
 400  * lib_ring_buffer_read - read data from ring_buffer_buffer.
 401  * @bufb : buffer backend
 402  * @offset : offset within the buffer
 403  * @dest : destination address
 404  * @len : length to copy to destination
 405  *
 406  * Should be protected by get_subbuf/put_subbuf.
 407  * Returns the length copied.
 408  */
 409 size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb, size_t offset,
 410                             void *dest, size_t len, struct lttng_ust_shm_handle *handle)
 411 {
 412         struct channel_backend *chanb;
 413         const struct lttng_ust_lib_ring_buffer_config *config;
 414         ssize_t orig_len;
 415         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 416         struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
 417         unsigned long sb_bindex, id;
 418         void *src;
 419
 420         chanb = &shmp(handle, bufb->chan)->backend;
 421         if (!chanb)
 422                 return 0;
 423         config = &chanb->config;
 424         orig_len = len;
 425         offset &= chanb->buf_size - 1;
 426
 427         if (caa_unlikely(!len))
 428                 return 0;
 429         id = bufb->buf_rsb.id;
 430         sb_bindex = subbuffer_id_get_index(config, id);
 431         rpages = shmp_index(handle, bufb->array, sb_bindex);
 432         if (!rpages)
 433                 return 0;
 434         /*
 435          * Underlying layer should never ask for reads across
 436          * subbuffers.
 437          */
 438         CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 439         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 440                      && subbuffer_id_is_noref(config, id));
 441         backend_pages = shmp(handle, rpages->shmp);
 442         if (!backend_pages)
 443                 return 0;
 444         src = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
 445         if (caa_unlikely(!src))
 446                 return 0;
 447         memcpy(dest, src, len);
 448         return orig_len;
 449 }
 450
 451 /**
 452  * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
 453  * @bufb : buffer backend
 454  * @offset : offset within the buffer
 455  * @dest : destination address
 456  * @len : destination's length
 457  *
 458  * Return string's length, or -EINVAL on error.
 459  * Should be protected by get_subbuf/put_subbuf.
 460  * Destination length should be at least 1 to hold '\0'.
 461  */
 462 int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend *bufb, size_t offset,
 463                               void *dest, size_t len, struct lttng_ust_shm_handle *handle)
 464 {
 465         struct channel_backend *chanb;
 466         const struct lttng_ust_lib_ring_buffer_config *config;
 467         ssize_t string_len, orig_offset;
 468         char *str;
 469         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 470         struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
 471         unsigned long sb_bindex, id;
 472
 473         chanb = &shmp(handle, bufb->chan)->backend;
 474         if (!chanb)
 475                 return -EINVAL;
 476         config = &chanb->config;
 477         if (caa_unlikely(!len))
 478                 return -EINVAL;
 479         offset &= chanb->buf_size - 1;
 480         orig_offset = offset;
 481         id = bufb->buf_rsb.id;
 482         sb_bindex = subbuffer_id_get_index(config, id);
 483         rpages = shmp_index(handle, bufb->array, sb_bindex);
 484         if (!rpages)
 485                 return -EINVAL;
 486         /*
 487          * Underlying layer should never ask for reads across
 488          * subbuffers.
 489          */
 490         CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 491         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 492                      && subbuffer_id_is_noref(config, id));
 493         backend_pages = shmp(handle, rpages->shmp);
 494         if (!backend_pages)
 495                 return -EINVAL;
 496         str = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
 497         if (caa_unlikely(!str))
 498                 return -EINVAL;
 499         string_len = strnlen(str, len);
 500         if (dest && len) {
 501                 memcpy(dest, str, string_len);
 502                 ((char *)dest)[0] = 0;
 503         }
 504         return offset - orig_offset;
 505 }
 506
 507 /**
 508  * lib_ring_buffer_read_offset_address - get address of a buffer location
 509  * @bufb : buffer backend
 510  * @offset : offset within the buffer.
 511  *
 512  * Return the address where a given offset is located (for read).
 513  * Should be used to get the current subbuffer header pointer. Given we know
 514  * it's never on a page boundary, it's safe to read/write directly
 515  * from/to this address, as long as the read/write is never bigger than
 516  * a page size.
 517  */
 518 void *lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
 519                                           size_t offset,
 520                                           struct lttng_ust_shm_handle *handle)
 521 {
 522         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 523         struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
 524         struct channel_backend *chanb;
 525         const struct lttng_ust_lib_ring_buffer_config *config;
 526         unsigned long sb_bindex, id;
 527
 528         chanb = &shmp(handle, bufb->chan)->backend;
 529         if (!chanb)
 530                 return NULL;
 531         config = &chanb->config;
 532         offset &= chanb->buf_size - 1;
 533         id = bufb->buf_rsb.id;
 534         sb_bindex = subbuffer_id_get_index(config, id);
 535         rpages = shmp_index(handle, bufb->array, sb_bindex);
 536         if (!rpages)
 537                 return NULL;
 538         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 539                      && subbuffer_id_is_noref(config, id));
 540         backend_pages = shmp(handle, rpages->shmp);
 541         if (!backend_pages)
 542                 return NULL;
 543         return shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
 544 }
 545
 546 /**
 547  * lib_ring_buffer_offset_address - get address of a location within the buffer
 548  * @bufb : buffer backend
 549  * @offset : offset within the buffer.
 550  *
 551  * Return the address where a given offset is located.
 552  * Should be used to get the current subbuffer header pointer. Given we know
 553  * it's always at the beginning of a page, it's safe to write directly to this
 554  * address, as long as the write is never bigger than a page size.
 555  */
 556 void *lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
 557                                      size_t offset,
 558                                      struct lttng_ust_shm_handle *handle)
 559 {
 560         size_t sbidx;
 561         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 562         struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
 563         struct channel_backend *chanb;
 564         const struct lttng_ust_lib_ring_buffer_config *config;
 565         unsigned long sb_bindex, id;
 566         struct lttng_ust_lib_ring_buffer_backend_subbuffer *sb;
 567
 568         chanb = &shmp(handle, bufb->chan)->backend;
 569         if (!chanb)
 570                 return NULL;
 571         config = &chanb->config;
 572         offset &= chanb->buf_size - 1;
 573         sbidx = offset >> chanb->subbuf_size_order;
 574         sb = shmp_index(handle, bufb->buf_wsb, sbidx);
 575         if (!sb)
 576                 return NULL;
 577         id = sb->id;
 578         sb_bindex = subbuffer_id_get_index(config, id);
 579         rpages = shmp_index(handle, bufb->array, sb_bindex);
 580         if (!rpages)
 581                 return NULL;
 582         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 583                      && subbuffer_id_is_noref(config, id));
 584         backend_pages = shmp(handle, rpages->shmp);
 585         if (!backend_pages)
 586                 return NULL;
 587         return shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
 588 }