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