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