libringbuffer/ring_buffer_backend.c

   1 /*
   2  * ring_buffer_backend.c
   3  *
   4  * Copyright (C) 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * Dual LGPL v2.1/GPL v2 license.
   7  */
   8
   9 #define _GNU_SOURCE
  10 #include <urcu/arch.h>
  11 #include <limits.h>
  12
  13 #include <lttng/ringbuffer-config.h>
  14 #include "vatomic.h"
  15 #include "backend.h"
  16 #include "frontend.h"
  17 #include "smp.h"
  18 #include "shm.h"
  19
  20 /**
  21  * lib_ring_buffer_backend_allocate - allocate a channel buffer
  22  * @config: ring buffer instance configuration
  23  * @buf: the buffer struct
  24  * @size: total size of the buffer
  25  * @num_subbuf: number of subbuffers
  26  * @extra_reader_sb: need extra subbuffer for reader
  27  */
  28 static
  29 int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_config *config,
  30                                      struct lttng_ust_lib_ring_buffer_backend *bufb,
  31                                      size_t size, size_t num_subbuf,
  32                                      int extra_reader_sb,
  33                                      struct lttng_ust_shm_handle *handle,
  34                                      struct shm_object *shmobj)
  35 {
  36         struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
  37         unsigned long subbuf_size, mmap_offset = 0;
  38         unsigned long num_subbuf_alloc;
  39         unsigned long i;
  40
  41         subbuf_size = chanb->subbuf_size;
  42         num_subbuf_alloc = num_subbuf;
  43
  44         if (extra_reader_sb)
  45                 num_subbuf_alloc++;
  46
  47         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp));
  48         set_shmp(bufb->array, zalloc_shm(shmobj,
  49                         sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp) * num_subbuf_alloc));
  50         if (caa_unlikely(!shmp(handle, bufb->array)))
  51                 goto array_error;
  52
  53         /*
  54          * This is the largest element (the buffer pages) which needs to
  55          * be aligned on PAGE_SIZE.
  56          */
  57         align_shm(shmobj, PAGE_SIZE);
  58         set_shmp(bufb->memory_map, zalloc_shm(shmobj,
  59                         subbuf_size * num_subbuf_alloc));
  60         if (caa_unlikely(!shmp(handle, bufb->memory_map)))
  61                 goto memory_map_error;
  62
  63         /* Allocate backend pages array elements */
  64         for (i = 0; i < num_subbuf_alloc; i++) {
  65                 align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages));
  66                 set_shmp(shmp_index(handle, bufb->array, i)->shmp,
  67                         zalloc_shm(shmobj,
  68                                 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages)));
  69                 if (!shmp(handle, shmp_index(handle, bufb->array, i)->shmp))
  70                         goto free_array;
  71         }
  72
  73         /* Allocate write-side subbuffer table */
  74         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer));
  75         set_shmp(bufb->buf_wsb, zalloc_shm(shmobj,
  76                                 sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer)
  77                                 * num_subbuf));
  78         if (caa_unlikely(!shmp(handle, bufb->buf_wsb)))
  79                 goto free_array;
  80
  81         for (i = 0; i < num_subbuf; i++)
  82                 shmp_index(handle, bufb->buf_wsb, i)->id = subbuffer_id(config, 0, 1, i);
  83
  84         /* Assign read-side subbuffer table */
  85         if (extra_reader_sb)
  86                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
  87                                                 num_subbuf_alloc - 1);
  88         else
  89                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
  90
  91         /* Assign pages to page index */
  92         for (i = 0; i < num_subbuf_alloc; i++) {
  93                 struct shm_ref ref;
  94
  95                 ref.index = bufb->memory_map._ref.index;
  96                 ref.offset = bufb->memory_map._ref.offset;
  97                 ref.offset += i * subbuf_size;
  98
  99                 set_shmp(shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->p,
 100                          ref);
 101                 if (config->output == RING_BUFFER_MMAP) {
 102                         shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->mmap_offset = mmap_offset;
 103                         mmap_offset += subbuf_size;
 104                 }
 105         }
 106
 107         return 0;
 108
 109 free_array:
 110         /* bufb->array[i] will be freed by shm teardown */
 111 memory_map_error:
 112         /* bufb->array will be freed by shm teardown */
 113 array_error:
 114         return -ENOMEM;
 115 }
 116
 117 int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend *bufb,
 118                                    struct channel_backend *chanb, int cpu,
 119                                    struct lttng_ust_shm_handle *handle,
 120                                    struct shm_object *shmobj)
 121 {
 122         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 123
 124         set_shmp(bufb->chan, handle->chan._ref);
 125         bufb->cpu = cpu;
 126
 127         return lib_ring_buffer_backend_allocate(config, bufb, chanb->buf_size,
 128                                                 chanb->num_subbuf,
 129                                                 chanb->extra_reader_sb,
 130                                                 handle, shmobj);
 131 }
 132
 133 void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend *bufb,
 134                                    struct lttng_ust_shm_handle *handle)
 135 {
 136         struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
 137         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 138         unsigned long num_subbuf_alloc;
 139         unsigned int i;
 140
 141         num_subbuf_alloc = chanb->num_subbuf;
 142         if (chanb->extra_reader_sb)
 143                 num_subbuf_alloc++;
 144
 145         for (i = 0; i < chanb->num_subbuf; i++)
 146                 shmp_index(handle, bufb->buf_wsb, i)->id = subbuffer_id(config, 0, 1, i);
 147         if (chanb->extra_reader_sb)
 148                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
 149                                                 num_subbuf_alloc - 1);
 150         else
 151                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 152
 153         for (i = 0; i < num_subbuf_alloc; i++) {
 154                 /* Don't reset mmap_offset */
 155                 v_set(config, &shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->records_commit, 0);
 156                 v_set(config, &shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->records_unread, 0);
 157                 shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->data_size = 0;
 158                 /* Don't reset backend page and virt addresses */
 159         }
 160         /* Don't reset num_pages_per_subbuf, cpu, allocated */
 161         v_set(config, &bufb->records_read, 0);
 162 }
 163
 164 /*
 165  * The frontend is responsible for also calling ring_buffer_backend_reset for
 166  * each buffer when calling channel_backend_reset.
 167  */
 168 void channel_backend_reset(struct channel_backend *chanb)
 169 {
 170         struct channel *chan = caa_container_of(chanb, struct channel, backend);
 171         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 172
 173         /*
 174          * Don't reset buf_size, subbuf_size, subbuf_size_order,
 175          * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
 176          * priv, notifiers, config, cpumask and name.
 177          */
 178         chanb->start_tsc = config->cb.ring_buffer_clock_read(chan);
 179 }
 180
 181 /**
 182  * channel_backend_init - initialize a channel backend
 183  * @chanb: channel backend
 184  * @name: channel name
 185  * @config: client ring buffer configuration
 186  * @parent: dentry of parent directory, %NULL for root directory
 187  * @subbuf_size: size of sub-buffers (> PAGE_SIZE, power of 2)
 188  * @num_subbuf: number of sub-buffers (power of 2)
 189  * @lttng_ust_shm_handle: shared memory handle
 190  *
 191  * Returns channel pointer if successful, %NULL otherwise.
 192  *
 193  * Creates per-cpu channel buffers using the sizes and attributes
 194  * specified.  The created channel buffer files will be named
 195  * name_0...name_N-1.  File permissions will be %S_IRUSR.
 196  *
 197  * Called with CPU hotplug disabled.
 198  */
 199 int channel_backend_init(struct channel_backend *chanb,
 200                          const char *name,
 201                          const struct lttng_ust_lib_ring_buffer_config *config,
 202                          size_t subbuf_size, size_t num_subbuf,
 203                          struct lttng_ust_shm_handle *handle)
 204 {
 205         struct channel *chan = caa_container_of(chanb, struct channel, backend);
 206         unsigned int i;
 207         int ret;
 208         size_t shmsize = 0, num_subbuf_alloc;
 209
 210         if (!name)
 211                 return -EPERM;
 212
 213         if (!(subbuf_size && num_subbuf))
 214                 return -EPERM;
 215
 216         /* Check that the subbuffer size is larger than a page. */
 217         if (subbuf_size < PAGE_SIZE)
 218                 return -EINVAL;
 219
 220         /*
 221          * Make sure the number of subbuffers and subbuffer size are
 222          * power of 2, and nonzero.
 223          */
 224         if (!subbuf_size || (subbuf_size & (subbuf_size - 1)))
 225                 return -EINVAL;
 226         if (!num_subbuf || (num_subbuf & (num_subbuf - 1)))
 227                 return -EINVAL;
 228
 229         ret = subbuffer_id_check_index(config, num_subbuf);
 230         if (ret)
 231                 return ret;
 232
 233         chanb->buf_size = num_subbuf * subbuf_size;
 234         chanb->subbuf_size = subbuf_size;
 235         chanb->buf_size_order = get_count_order(chanb->buf_size);
 236         chanb->subbuf_size_order = get_count_order(subbuf_size);
 237         chanb->num_subbuf_order = get_count_order(num_subbuf);
 238         chanb->extra_reader_sb =
 239                         (config->mode == RING_BUFFER_OVERWRITE) ? 1 : 0;
 240         chanb->num_subbuf = num_subbuf;
 241         strncpy(chanb->name, name, NAME_MAX);
 242         chanb->name[NAME_MAX - 1] = '\0';
 243         memcpy(&chanb->config, config, sizeof(*config));
 244
 245         /* Per-cpu buffer size: control (prior to backend) */
 246         shmsize = offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer));
 247         shmsize += sizeof(struct lttng_ust_lib_ring_buffer);
 248
 249         /* Per-cpu buffer size: backend */
 250         /* num_subbuf + 1 is the worse case */
 251         num_subbuf_alloc = num_subbuf + 1;
 252         shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp));
 253         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp) * num_subbuf_alloc;
 254         shmsize += offset_align(shmsize, PAGE_SIZE);
 255         shmsize += subbuf_size * num_subbuf_alloc;
 256         shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages));
 257         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_pages) * num_subbuf_alloc;
 258         shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer));
 259         shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer) * num_subbuf;
 260         /* Per-cpu buffer size: control (after backend) */
 261         shmsize += offset_align(shmsize, __alignof__(struct commit_counters_hot));
 262         shmsize += sizeof(struct commit_counters_hot) * num_subbuf;
 263         shmsize += offset_align(shmsize, __alignof__(struct commit_counters_cold));
 264         shmsize += sizeof(struct commit_counters_cold) * num_subbuf;
 265
 266         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 267                 struct lttng_ust_lib_ring_buffer *buf;
 268                 /*
 269                  * We need to allocate for all possible cpus.
 270                  */
 271                 for_each_possible_cpu(i) {
 272                         struct shm_object *shmobj;
 273
 274                         shmobj = shm_object_table_append(handle->table, shmsize);
 275                         if (!shmobj)
 276                                 goto end;
 277                         align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer));
 278                         set_shmp(chanb->buf[i].shmp, zalloc_shm(shmobj, sizeof(struct lttng_ust_lib_ring_buffer)));
 279                         buf = shmp(handle, chanb->buf[i].shmp);
 280                         if (!buf)
 281                                 goto end;
 282                         set_shmp(buf->self, chanb->buf[i].shmp._ref);
 283                         ret = lib_ring_buffer_create(buf, chanb, i,
 284                                         handle, shmobj);
 285                         if (ret)
 286                                 goto free_bufs; /* cpu hotplug locked */
 287                 }
 288         } else {
 289                 struct shm_object *shmobj;
 290                 struct lttng_ust_lib_ring_buffer *buf;
 291
 292                 shmobj = shm_object_table_append(handle->table, shmsize);
 293                 if (!shmobj)
 294                         goto end;
 295                 align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer));
 296                 set_shmp(chanb->buf[0].shmp, zalloc_shm(shmobj, sizeof(struct lttng_ust_lib_ring_buffer)));
 297                 buf = shmp(handle, chanb->buf[0].shmp);
 298                 if (!buf)
 299                         goto end;
 300                 set_shmp(buf->self, chanb->buf[0].shmp._ref);
 301                 ret = lib_ring_buffer_create(buf, chanb, -1,
 302                                         handle, shmobj);
 303                 if (ret)
 304                         goto free_bufs;
 305         }
 306         chanb->start_tsc = config->cb.ring_buffer_clock_read(chan);
 307
 308         return 0;
 309
 310 free_bufs:
 311         /* We only free the buffer data upon shm teardown */
 312 end:
 313         return -ENOMEM;
 314 }
 315
 316 /**
 317  * channel_backend_free - destroy the channel
 318  * @chan: the channel
 319  *
 320  * Destroy all channel buffers and frees the channel.
 321  */
 322 void channel_backend_free(struct channel_backend *chanb,
 323                           struct lttng_ust_shm_handle *handle)
 324 {
 325         /* SHM teardown takes care of everything */
 326 }
 327
 328 /**
 329  * lib_ring_buffer_read - read data from ring_buffer_buffer.
 330  * @bufb : buffer backend
 331  * @offset : offset within the buffer
 332  * @dest : destination address
 333  * @len : length to copy to destination
 334  *
 335  * Should be protected by get_subbuf/put_subbuf.
 336  * Returns the length copied.
 337  */
 338 size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb, size_t offset,
 339                             void *dest, size_t len, struct lttng_ust_shm_handle *handle)
 340 {
 341         struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
 342         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 343         ssize_t orig_len;
 344         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 345         unsigned long sb_bindex, id;
 346
 347         orig_len = len;
 348         offset &= chanb->buf_size - 1;
 349
 350         if (caa_unlikely(!len))
 351                 return 0;
 352         id = bufb->buf_rsb.id;
 353         sb_bindex = subbuffer_id_get_index(config, id);
 354         rpages = shmp_index(handle, bufb->array, sb_bindex);
 355         /*
 356          * Underlying layer should never ask for reads across
 357          * subbuffers.
 358          */
 359         CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 360         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 361                      && subbuffer_id_is_noref(config, id));
 362         memcpy(dest, shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1)), len);
 363         return orig_len;
 364 }
 365
 366 /**
 367  * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
 368  * @bufb : buffer backend
 369  * @offset : offset within the buffer
 370  * @dest : destination address
 371  * @len : destination's length
 372  *
 373  * return string's length
 374  * Should be protected by get_subbuf/put_subbuf.
 375  */
 376 int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend *bufb, size_t offset,
 377                               void *dest, size_t len, struct lttng_ust_shm_handle *handle)
 378 {
 379         struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
 380         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 381         ssize_t string_len, orig_offset;
 382         char *str;
 383         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 384         unsigned long sb_bindex, id;
 385
 386         offset &= chanb->buf_size - 1;
 387         orig_offset = offset;
 388         id = bufb->buf_rsb.id;
 389         sb_bindex = subbuffer_id_get_index(config, id);
 390         rpages = shmp_index(handle, bufb->array, sb_bindex);
 391         /*
 392          * Underlying layer should never ask for reads across
 393          * subbuffers.
 394          */
 395         CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 396         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 397                      && subbuffer_id_is_noref(config, id));
 398         str = shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1));
 399         string_len = strnlen(str, len);
 400         if (dest && len) {
 401                 memcpy(dest, str, string_len);
 402                 ((char *)dest)[0] = 0;
 403         }
 404         return offset - orig_offset;
 405 }
 406
 407 /**
 408  * lib_ring_buffer_read_offset_address - get address of a buffer location
 409  * @bufb : buffer backend
 410  * @offset : offset within the buffer.
 411  *
 412  * Return the address where a given offset is located (for read).
 413  * Should be used to get the current subbuffer header pointer. Given we know
 414  * it's never on a page boundary, it's safe to write directly to this address,
 415  * as long as the write is never bigger than a page size.
 416  */
 417 void *lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
 418                                           size_t offset,
 419                                           struct lttng_ust_shm_handle *handle)
 420 {
 421         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 422         struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
 423         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 424         unsigned long sb_bindex, id;
 425
 426         offset &= chanb->buf_size - 1;
 427         id = bufb->buf_rsb.id;
 428         sb_bindex = subbuffer_id_get_index(config, id);
 429         rpages = shmp_index(handle, bufb->array, sb_bindex);
 430         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 431                      && subbuffer_id_is_noref(config, id));
 432         return shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1));
 433 }
 434
 435 /**
 436  * lib_ring_buffer_offset_address - get address of a location within the buffer
 437  * @bufb : buffer backend
 438  * @offset : offset within the buffer.
 439  *
 440  * Return the address where a given offset is located.
 441  * Should be used to get the current subbuffer header pointer. Given we know
 442  * it's always at the beginning of a page, it's safe to write directly to this
 443  * address, as long as the write is never bigger than a page size.
 444  */
 445 void *lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
 446                                      size_t offset,
 447                                      struct lttng_ust_shm_handle *handle)
 448 {
 449         size_t sbidx;
 450         struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
 451         struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
 452         const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
 453         unsigned long sb_bindex, id;
 454
 455         offset &= chanb->buf_size - 1;
 456         sbidx = offset >> chanb->subbuf_size_order;
 457         id = shmp_index(handle, bufb->buf_wsb, sbidx)->id;
 458         sb_bindex = subbuffer_id_get_index(config, id);
 459         rpages = shmp_index(handle, bufb->array, sb_bindex);
 460         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 461                      && subbuffer_id_is_noref(config, id));
 462         return shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1));
 463 }