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