libringbuffer/ring_buffer_backend.c

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