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