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