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