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