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