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