libringbuffer/ring_buffer_backend.c

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