src/lib/ringbuffer/ring_buffer_iterator.c

   1 /* SPDX-License-Identifier: (GPL-2.0-only OR LGPL-2.1-only)
   2  *
   3  * ring_buffer_iterator.c
   4  *
   5  * Ring buffer and channel iterators. Get each event of a channel in order. Uses
   6  * a prio heap for per-cpu buffers, giving a O(log(NR_CPUS)) algorithmic
   7  * complexity for the "get next event" operation.
   8  *
   9  * Copyright (C) 2010-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  10  */
  11
  12 #include <ringbuffer/iterator.h>
  13 #include <wrapper/cpu.h>
  14 #include <wrapper/file.h>
  15 #include <wrapper/uaccess.h>
  16 #include <linux/jiffies.h>
  17 #include <linux/delay.h>
  18 #include <linux/module.h>
  19
  20 /*
  21  * Safety factor taking into account internal kernel interrupt latency.
  22  * Assuming 250ms worse-case latency.
  23  */
  24 #define MAX_SYSTEM_LATENCY      250
  25
  26 /*
  27  * Maximum delta expected between trace clocks. At most 1 jiffy delta.
  28  */
  29 #define MAX_CLOCK_DELTA         (jiffies_to_usecs(1) * 1000)
  30
  31 /**
  32  * lib_ring_buffer_get_next_record - Get the next record in a buffer.
  33  * @chan: channel
  34  * @buf: buffer
  35  *
  36  * Returns the size of the event read, -EAGAIN if buffer is empty, -ENODATA if
  37  * buffer is empty and finalized. The buffer must already be opened for reading.
  38  */
  39 ssize_t lib_ring_buffer_get_next_record(struct lttng_kernel_ring_buffer_channel *chan,
  40                                         struct lttng_kernel_ring_buffer *buf)
  41 {
  42         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
  43         struct lttng_kernel_ring_buffer_iter *iter = &buf->iter;
  44         int ret;
  45
  46 restart:
  47         switch (iter->state) {
  48         case ITER_GET_SUBBUF:
  49                 ret = lib_ring_buffer_get_next_subbuf(buf);
  50                 if (ret && !LTTNG_READ_ONCE(buf->finalized)
  51                     && config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
  52                         /*
  53                          * Use "pull" scheme for global buffers. The reader
  54                          * itself flushes the buffer to "pull" data not visible
  55                          * to readers yet. Flush current subbuffer and re-try.
  56                          *
  57                          * Per-CPU buffers rather use a "push" scheme because
  58                          * the IPI needed to flush all CPU's buffers is too
  59                          * costly. In the "push" scheme, the reader waits for
  60                          * the writer periodic timer to flush the
  61                          * buffers (keeping track of a quiescent state
  62                          * timestamp). Therefore, the writer "pushes" data out
  63                          * of the buffers rather than letting the reader "pull"
  64                          * data from the buffer.
  65                          */
  66                         lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
  67                         ret = lib_ring_buffer_get_next_subbuf(buf);
  68                 }
  69                 if (ret)
  70                         return ret;
  71                 iter->consumed = buf->cons_snapshot;
  72                 iter->data_size = lib_ring_buffer_get_read_data_size(config, buf);
  73                 iter->read_offset = iter->consumed;
  74                 /* skip header */
  75                 iter->read_offset += config->cb.subbuffer_header_size();
  76                 iter->state = ITER_TEST_RECORD;
  77                 goto restart;
  78         case ITER_TEST_RECORD:
  79                 if (iter->read_offset - iter->consumed >= iter->data_size) {
  80                         iter->state = ITER_PUT_SUBBUF;
  81                 } else {
  82                         CHAN_WARN_ON(chan, !config->cb.record_get);
  83                         config->cb.record_get(config, chan, buf,
  84                                               iter->read_offset,
  85                                               &iter->header_len,
  86                                               &iter->payload_len,
  87                                               &iter->timestamp);
  88                         iter->read_offset += iter->header_len;
  89                         subbuffer_consume_record(config, &buf->backend);
  90                         iter->state = ITER_NEXT_RECORD;
  91                         return iter->payload_len;
  92                 }
  93                 goto restart;
  94         case ITER_NEXT_RECORD:
  95                 iter->read_offset += iter->payload_len;
  96                 iter->state = ITER_TEST_RECORD;
  97                 goto restart;
  98         case ITER_PUT_SUBBUF:
  99                 lib_ring_buffer_put_next_subbuf(buf);
 100                 iter->state = ITER_GET_SUBBUF;
 101                 goto restart;
 102         default:
 103                 CHAN_WARN_ON(chan, 1);  /* Should not happen */
 104                 return -EPERM;
 105         }
 106 }
 107 EXPORT_SYMBOL_GPL(lib_ring_buffer_get_next_record);
 108
 109 void lib_ring_buffer_put_current_record(struct lttng_kernel_ring_buffer *buf)
 110 {
 111         struct lttng_kernel_ring_buffer_iter *iter;
 112
 113         if (!buf)
 114                 return;
 115         iter = &buf->iter;
 116         if (iter->state != ITER_NEXT_RECORD)
 117                 return;
 118         iter->read_offset += iter->payload_len;
 119         iter->state = ITER_TEST_RECORD;
 120         if (iter->read_offset - iter->consumed >= iter->data_size) {
 121                 lib_ring_buffer_put_next_subbuf(buf);
 122                 iter->state = ITER_GET_SUBBUF;
 123         }
 124 }
 125 EXPORT_SYMBOL_GPL(lib_ring_buffer_put_current_record);
 126
 127 static int buf_is_higher(void *a, void *b)
 128 {
 129         struct lttng_kernel_ring_buffer *bufa = a;
 130         struct lttng_kernel_ring_buffer *bufb = b;
 131
 132         /* Consider lowest timestamps to be at the top of the heap */
 133         return (bufa->iter.timestamp < bufb->iter.timestamp);
 134 }
 135
 136 static
 137 void lib_ring_buffer_get_empty_buf_records(const struct lttng_kernel_ring_buffer_config *config,
 138                                            struct lttng_kernel_ring_buffer_channel *chan)
 139 {
 140         struct lttng_ptr_heap *heap = &chan->iter.heap;
 141         struct lttng_kernel_ring_buffer *buf, *tmp;
 142         ssize_t len;
 143
 144         list_for_each_entry_safe(buf, tmp, &chan->iter.empty_head,
 145                                  iter.empty_node) {
 146                 len = lib_ring_buffer_get_next_record(chan, buf);
 147
 148                 /*
 149                  * Deal with -EAGAIN and -ENODATA.
 150                  * len >= 0 means record contains data.
 151                  * -EBUSY should never happen, because we support only one
 152                  * reader.
 153                  */
 154                 switch (len) {
 155                 case -EAGAIN:
 156                         /* Keep node in empty list */
 157                         break;
 158                 case -ENODATA:
 159                         /*
 160                          * Buffer is finalized. Don't add to list of empty
 161                          * buffer, because it has no more data to provide, ever.
 162                          */
 163                         list_del(&buf->iter.empty_node);
 164                         break;
 165                 case -EBUSY:
 166                         CHAN_WARN_ON(chan, 1);
 167                         break;
 168                 default:
 169                         /*
 170                          * Insert buffer into the heap, remove from empty buffer
 171                          * list.
 172                          */
 173                         CHAN_WARN_ON(chan, len < 0);
 174                         list_del(&buf->iter.empty_node);
 175                         CHAN_WARN_ON(chan, lttng_heap_insert(heap, buf));
 176                 }
 177         }
 178 }
 179
 180 static
 181 void lib_ring_buffer_wait_for_qs(const struct lttng_kernel_ring_buffer_config *config,
 182                                  struct lttng_kernel_ring_buffer_channel *chan)
 183 {
 184         u64 timestamp_qs;
 185         unsigned long wait_msecs;
 186
 187         /*
 188          * No need to wait if no empty buffers are present.
 189          */
 190         if (list_empty(&chan->iter.empty_head))
 191                 return;
 192
 193         timestamp_qs = config->cb.ring_buffer_clock_read(chan);
 194         /*
 195          * We need to consider previously empty buffers.
 196          * Do a get next buf record on each of them. Add them to
 197          * the heap if they have data. If at least one of them
 198          * don't have data, we need to wait for
 199          * switch_timer_interval + MAX_SYSTEM_LATENCY (so we are sure the
 200          * buffers have been switched either by the timer or idle entry) and
 201          * check them again, adding them if they have data.
 202          */
 203         lib_ring_buffer_get_empty_buf_records(config, chan);
 204
 205         /*
 206          * No need to wait if no empty buffers are present.
 207          */
 208         if (list_empty(&chan->iter.empty_head))
 209                 return;
 210
 211         /*
 212          * We need to wait for the buffer switch timer to run. If the
 213          * CPU is idle, idle entry performed the switch.
 214          * TODO: we could optimize further by skipping the sleep if all
 215          * empty buffers belong to idle or offline cpus.
 216          */
 217         wait_msecs = jiffies_to_msecs(chan->switch_timer_interval);
 218         wait_msecs += MAX_SYSTEM_LATENCY;
 219         msleep(wait_msecs);
 220         lib_ring_buffer_get_empty_buf_records(config, chan);
 221         /*
 222          * Any buffer still in the empty list here cannot possibly
 223          * contain an event with a timestamp prior to "timestamp_qs".
 224          * The new quiescent state timestamp is the one we grabbed
 225          * before waiting for buffer data.  It is therefore safe to
 226          * ignore empty buffers up to last_qs timestamp for fusion
 227          * merge.
 228          */
 229         chan->iter.last_qs = timestamp_qs;
 230 }
 231
 232 /**
 233  * channel_get_next_record - Get the next record in a channel.
 234  * @chan: channel
 235  * @ret_buf: the buffer in which the event is located (output)
 236  *
 237  * Returns the size of new current event, -EAGAIN if all buffers are empty,
 238  * -ENODATA if all buffers are empty and finalized. The channel must already be
 239  * opened for reading.
 240  */
 241
 242 ssize_t channel_get_next_record(struct lttng_kernel_ring_buffer_channel *chan,
 243                                 struct lttng_kernel_ring_buffer **ret_buf)
 244 {
 245         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 246         struct lttng_kernel_ring_buffer *buf;
 247         struct lttng_ptr_heap *heap;
 248         ssize_t len;
 249
 250         if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
 251                 *ret_buf = channel_get_ring_buffer(config, chan, 0);
 252                 return lib_ring_buffer_get_next_record(chan, *ret_buf);
 253         }
 254
 255         heap = &chan->iter.heap;
 256
 257         /*
 258          * get next record for topmost buffer.
 259          */
 260         buf = lttng_heap_maximum(heap);
 261         if (buf) {
 262                 len = lib_ring_buffer_get_next_record(chan, buf);
 263                 /*
 264                  * Deal with -EAGAIN and -ENODATA.
 265                  * len >= 0 means record contains data.
 266                  */
 267                 switch (len) {
 268                 case -EAGAIN:
 269                         buf->iter.timestamp = 0;
 270                         list_add(&buf->iter.empty_node, &chan->iter.empty_head);
 271                         /* Remove topmost buffer from the heap */
 272                         CHAN_WARN_ON(chan, lttng_heap_remove(heap) != buf);
 273                         break;
 274                 case -ENODATA:
 275                         /*
 276                          * Buffer is finalized. Remove buffer from heap and
 277                          * don't add to list of empty buffer, because it has no
 278                          * more data to provide, ever.
 279                          */
 280                         CHAN_WARN_ON(chan, lttng_heap_remove(heap) != buf);
 281                         break;
 282                 case -EBUSY:
 283                         CHAN_WARN_ON(chan, 1);
 284                         break;
 285                 default:
 286                         /*
 287                          * Reinsert buffer into the heap. Note that heap can be
 288                          * partially empty, so we need to use
 289                          * lttng_heap_replace_max().
 290                          */
 291                         CHAN_WARN_ON(chan, len < 0);
 292                         CHAN_WARN_ON(chan, lttng_heap_replace_max(heap, buf) != buf);
 293                         break;
 294                 }
 295         }
 296
 297         buf = lttng_heap_maximum(heap);
 298         if (!buf || buf->iter.timestamp > chan->iter.last_qs) {
 299                 /*
 300                  * Deal with buffers previously showing no data.
 301                  * Add buffers containing data to the heap, update
 302                  * last_qs.
 303                  */
 304                 lib_ring_buffer_wait_for_qs(config, chan);
 305         }
 306
 307         *ret_buf = buf = lttng_heap_maximum(heap);
 308         if (buf) {
 309                 /*
 310                  * If this warning triggers, you probably need to check your
 311                  * system interrupt latency. Typical causes: too many printk()
 312                  * output going to a serial console with interrupts off.
 313                  * Allow for MAX_CLOCK_DELTA ns timestamp delta going backward.
 314                  * Observed on SMP KVM setups with trace_clock().
 315                  */
 316                 if (chan->iter.last_timestamp
 317                     > (buf->iter.timestamp + MAX_CLOCK_DELTA)) {
 318                         printk(KERN_WARNING "LTTng: ring_buffer: timestamps going "
 319                                "backward. Last time %llu ns, cpu %d, "
 320                                "current time %llu ns, cpu %d, "
 321                                "delta %llu ns.\n",
 322                                chan->iter.last_timestamp, chan->iter.last_cpu,
 323                                buf->iter.timestamp, buf->backend.cpu,
 324                                chan->iter.last_timestamp - buf->iter.timestamp);
 325                         CHAN_WARN_ON(chan, 1);
 326                 }
 327                 chan->iter.last_timestamp = buf->iter.timestamp;
 328                 chan->iter.last_cpu = buf->backend.cpu;
 329                 return buf->iter.payload_len;
 330         } else {
 331                 /* Heap is empty */
 332                 if (list_empty(&chan->iter.empty_head))
 333                         return -ENODATA;        /* All buffers finalized */
 334                 else
 335                         return -EAGAIN;         /* Temporarily empty */
 336         }
 337 }
 338 EXPORT_SYMBOL_GPL(channel_get_next_record);
 339
 340 static
 341 void lib_ring_buffer_iterator_init(struct lttng_kernel_ring_buffer_channel *chan, struct lttng_kernel_ring_buffer *buf)
 342 {
 343         if (buf->iter.allocated)
 344                 return;
 345
 346         buf->iter.allocated = 1;
 347         if (chan->iter.read_open && !buf->iter.read_open) {
 348                 CHAN_WARN_ON(chan, lib_ring_buffer_open_read(buf) != 0);
 349                 buf->iter.read_open = 1;
 350         }
 351
 352         /* Add to list of buffers without any current record */
 353         if (chan->backend.config.alloc == RING_BUFFER_ALLOC_PER_CPU)
 354                 list_add(&buf->iter.empty_node, &chan->iter.empty_head);
 355 }
 356
 357 #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0))
 358
 359 int lttng_cpuhp_rb_iter_online(unsigned int cpu,
 360                 struct lttng_cpuhp_node *node)
 361 {
 362         struct lttng_kernel_ring_buffer_channel *chan = container_of(node, struct lttng_kernel_ring_buffer_channel,
 363                                             cpuhp_iter_online);
 364         struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
 365         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 366
 367         CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL);
 368
 369         lib_ring_buffer_iterator_init(chan, buf);
 370         return 0;
 371 }
 372 EXPORT_SYMBOL_GPL(lttng_cpuhp_rb_iter_online);
 373
 374 #else /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */
 375
 376 #ifdef CONFIG_HOTPLUG_CPU
 377 static
 378 int channel_iterator_cpu_hotplug(struct notifier_block *nb,
 379                                            unsigned long action,
 380                                            void *hcpu)
 381 {
 382         unsigned int cpu = (unsigned long)hcpu;
 383         struct lttng_kernel_ring_buffer_channel *chan = container_of(nb, struct lttng_kernel_ring_buffer_channel,
 384                                             hp_iter_notifier);
 385         struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
 386         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 387
 388         if (!chan->hp_iter_enable)
 389                 return NOTIFY_DONE;
 390
 391         CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL);
 392
 393         switch (action) {
 394         case CPU_DOWN_FAILED:
 395         case CPU_DOWN_FAILED_FROZEN:
 396         case CPU_ONLINE:
 397         case CPU_ONLINE_FROZEN:
 398                 lib_ring_buffer_iterator_init(chan, buf);
 399                 return NOTIFY_OK;
 400         default:
 401                 return NOTIFY_DONE;
 402         }
 403 }
 404 #endif
 405
 406 #endif /* #else #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */
 407
 408 int channel_iterator_init(struct lttng_kernel_ring_buffer_channel *chan)
 409 {
 410         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 411         struct lttng_kernel_ring_buffer *buf;
 412
 413         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 414                 int ret;
 415
 416                 INIT_LIST_HEAD(&chan->iter.empty_head);
 417                 ret = lttng_heap_init(&chan->iter.heap,
 418                                 num_possible_cpus(),
 419                                 GFP_KERNEL, buf_is_higher);
 420                 if (ret)
 421                         return ret;
 422
 423 #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0))
 424                 chan->cpuhp_iter_online.component = LTTNG_RING_BUFFER_ITER;
 425                 ret = cpuhp_state_add_instance(lttng_rb_hp_online,
 426                         &chan->cpuhp_iter_online.node);
 427                 if (ret)
 428                         return ret;
 429 #else /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */
 430                 {
 431                         int cpu;
 432
 433                         /*
 434                          * In case of non-hotplug cpu, if the ring-buffer is allocated
 435                          * in early initcall, it will not be notified of secondary cpus.
 436                          * In that off case, we need to allocate for all possible cpus.
 437                          */
 438 #ifdef CONFIG_HOTPLUG_CPU
 439                         chan->hp_iter_notifier.notifier_call =
 440                                 channel_iterator_cpu_hotplug;
 441                         chan->hp_iter_notifier.priority = 10;
 442                         register_cpu_notifier(&chan->hp_iter_notifier);
 443
 444                         lttng_cpus_read_lock();
 445                         for_each_online_cpu(cpu) {
 446                                 buf = per_cpu_ptr(chan->backend.buf, cpu);
 447                                 lib_ring_buffer_iterator_init(chan, buf);
 448                         }
 449                         chan->hp_iter_enable = 1;
 450                         lttng_cpus_read_unlock();
 451 #else
 452                         for_each_possible_cpu(cpu) {
 453                                 buf = per_cpu_ptr(chan->backend.buf, cpu);
 454                                 lib_ring_buffer_iterator_init(chan, buf);
 455                         }
 456 #endif
 457                 }
 458 #endif /* #else #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */
 459         } else {
 460                 buf = channel_get_ring_buffer(config, chan, 0);
 461                 lib_ring_buffer_iterator_init(chan, buf);
 462         }
 463         return 0;
 464 }
 465
 466 void channel_iterator_unregister_notifiers(struct lttng_kernel_ring_buffer_channel *chan)
 467 {
 468         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 469
 470         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 471 #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0))
 472                 {
 473                         int ret;
 474
 475                         ret = cpuhp_state_remove_instance(lttng_rb_hp_online,
 476                                 &chan->cpuhp_iter_online.node);
 477                         WARN_ON(ret);
 478                 }
 479 #else /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */
 480                 chan->hp_iter_enable = 0;
 481                 unregister_cpu_notifier(&chan->hp_iter_notifier);
 482 #endif /* #else #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */
 483         }
 484 }
 485
 486 void channel_iterator_free(struct lttng_kernel_ring_buffer_channel *chan)
 487 {
 488         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 489
 490         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 491                 lttng_heap_free(&chan->iter.heap);
 492 }
 493
 494 int lib_ring_buffer_iterator_open(struct lttng_kernel_ring_buffer *buf)
 495 {
 496         struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan;
 497         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 498         CHAN_WARN_ON(chan, config->output != RING_BUFFER_ITERATOR);
 499         return lib_ring_buffer_open_read(buf);
 500 }
 501 EXPORT_SYMBOL_GPL(lib_ring_buffer_iterator_open);
 502
 503 /*
 504  * Note: Iterators must not be mixed with other types of outputs, because an
 505  * iterator can leave the buffer in "GET" state, which is not consistent with
 506  * other types of output (mmap, splice, raw data read).
 507  */
 508 void lib_ring_buffer_iterator_release(struct lttng_kernel_ring_buffer *buf)
 509 {
 510         lib_ring_buffer_release_read(buf);
 511 }
 512 EXPORT_SYMBOL_GPL(lib_ring_buffer_iterator_release);
 513
 514 int channel_iterator_open(struct lttng_kernel_ring_buffer_channel *chan)
 515 {
 516         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 517         struct lttng_kernel_ring_buffer *buf;
 518         int ret = 0, cpu;
 519
 520         CHAN_WARN_ON(chan, config->output != RING_BUFFER_ITERATOR);
 521
 522         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 523                 lttng_cpus_read_lock();
 524                 /* Allow CPU hotplug to keep track of opened reader */
 525                 chan->iter.read_open = 1;
 526                 for_each_channel_cpu(cpu, chan) {
 527                         buf = channel_get_ring_buffer(config, chan, cpu);
 528                         ret = lib_ring_buffer_iterator_open(buf);
 529                         if (ret)
 530                                 goto error;
 531                         buf->iter.read_open = 1;
 532                 }
 533                 lttng_cpus_read_unlock();
 534         } else {
 535                 buf = channel_get_ring_buffer(config, chan, 0);
 536                 ret = lib_ring_buffer_iterator_open(buf);
 537         }
 538         return ret;
 539 error:
 540         /* Error should always happen on CPU 0, hence no close is required. */
 541         CHAN_WARN_ON(chan, cpu != 0);
 542         lttng_cpus_read_unlock();
 543         return ret;
 544 }
 545 EXPORT_SYMBOL_GPL(channel_iterator_open);
 546
 547 void channel_iterator_release(struct lttng_kernel_ring_buffer_channel *chan)
 548 {
 549         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 550         struct lttng_kernel_ring_buffer *buf;
 551         int cpu;
 552
 553         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 554                 lttng_cpus_read_lock();
 555                 for_each_channel_cpu(cpu, chan) {
 556                         buf = channel_get_ring_buffer(config, chan, cpu);
 557                         if (buf->iter.read_open) {
 558                                 lib_ring_buffer_iterator_release(buf);
 559                                 buf->iter.read_open = 0;
 560                         }
 561                 }
 562                 chan->iter.read_open = 0;
 563                 lttng_cpus_read_unlock();
 564         } else {
 565                 buf = channel_get_ring_buffer(config, chan, 0);
 566                 lib_ring_buffer_iterator_release(buf);
 567         }
 568 }
 569 EXPORT_SYMBOL_GPL(channel_iterator_release);
 570
 571 void lib_ring_buffer_iterator_reset(struct lttng_kernel_ring_buffer *buf)
 572 {
 573         struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan;
 574
 575         if (buf->iter.state != ITER_GET_SUBBUF)
 576                 lib_ring_buffer_put_next_subbuf(buf);
 577         buf->iter.state = ITER_GET_SUBBUF;
 578         /* Remove from heap (if present). */
 579         if (lttng_heap_cherrypick(&chan->iter.heap, buf))
 580                 list_add(&buf->iter.empty_node, &chan->iter.empty_head);
 581         buf->iter.timestamp = 0;
 582         buf->iter.header_len = 0;
 583         buf->iter.payload_len = 0;
 584         buf->iter.consumed = 0;
 585         buf->iter.read_offset = 0;
 586         buf->iter.data_size = 0;
 587         /* Don't reset allocated and read_open */
 588 }
 589
 590 void channel_iterator_reset(struct lttng_kernel_ring_buffer_channel *chan)
 591 {
 592         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 593         struct lttng_kernel_ring_buffer *buf;
 594         int cpu;
 595
 596         /* Empty heap, put into empty_head */
 597         while ((buf = lttng_heap_remove(&chan->iter.heap)) != NULL)
 598                 list_add(&buf->iter.empty_node, &chan->iter.empty_head);
 599
 600         for_each_channel_cpu(cpu, chan) {
 601                 buf = channel_get_ring_buffer(config, chan, cpu);
 602                 lib_ring_buffer_iterator_reset(buf);
 603         }
 604         /* Don't reset read_open */
 605         chan->iter.last_qs = 0;
 606         chan->iter.last_timestamp = 0;
 607         chan->iter.last_cpu = 0;
 608         chan->iter.len_left = 0;
 609 }
 610
 611 /*
 612  * Ring buffer payload extraction read() implementation.
 613  */
 614 static
 615 ssize_t channel_ring_buffer_file_read(struct file *filp,
 616                                       char __user *user_buf,
 617                                       size_t count,
 618                                       loff_t *ppos,
 619                                       struct lttng_kernel_ring_buffer_channel *chan,
 620                                       struct lttng_kernel_ring_buffer *buf,
 621                                       int fusionmerge)
 622 {
 623         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 624         size_t read_count = 0, read_offset;
 625         ssize_t len;
 626
 627         might_sleep();
 628         if (!lttng_access_ok(VERIFY_WRITE, user_buf, count))
 629                 return -EFAULT;
 630
 631         /* Finish copy of previous record */
 632         if (*ppos != 0) {
 633                 if (read_count < count) {
 634                         len = chan->iter.len_left;
 635                         read_offset = *ppos;
 636                         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU
 637                             && fusionmerge)
 638                                 buf = lttng_heap_maximum(&chan->iter.heap);
 639                         CHAN_WARN_ON(chan, !buf);
 640                         goto skip_get_next;
 641                 }
 642         }
 643
 644         while (read_count < count) {
 645                 size_t copy_len, space_left;
 646
 647                 if (fusionmerge)
 648                         len = channel_get_next_record(chan, &buf);
 649                 else
 650                         len = lib_ring_buffer_get_next_record(chan, buf);
 651 len_test:
 652                 if (len < 0) {
 653                         /*
 654                          * Check if buffer is finalized (end of file).
 655                          */
 656                         if (len == -ENODATA) {
 657                                 /* A 0 read_count will tell about end of file */
 658                                 goto nodata;
 659                         }
 660                         if (filp->f_flags & O_NONBLOCK) {
 661                                 if (!read_count)
 662                                         read_count = -EAGAIN;
 663                                 goto nodata;
 664                         } else {
 665                                 int error;
 666
 667                                 /*
 668                                  * No data available at the moment, return what
 669                                  * we got.
 670                                  */
 671                                 if (read_count)
 672                                         goto nodata;
 673
 674                                 /*
 675                                  * Wait for returned len to be >= 0 or -ENODATA.
 676                                  */
 677                                 if (fusionmerge)
 678                                         error = wait_event_interruptible(
 679                                           chan->read_wait,
 680                                           ((len = channel_get_next_record(chan,
 681                                                 &buf)), len != -EAGAIN));
 682                                 else
 683                                         error = wait_event_interruptible(
 684                                           buf->read_wait,
 685                                           ((len = lib_ring_buffer_get_next_record(
 686                                                   chan, buf)), len != -EAGAIN));
 687                                 CHAN_WARN_ON(chan, len == -EBUSY);
 688                                 if (error) {
 689                                         read_count = error;
 690                                         goto nodata;
 691                                 }
 692                                 CHAN_WARN_ON(chan, len < 0 && len != -ENODATA);
 693                                 goto len_test;
 694                         }
 695                 }
 696                 read_offset = buf->iter.read_offset;
 697 skip_get_next:
 698                 space_left = count - read_count;
 699                 if (len <= space_left) {
 700                         copy_len = len;
 701                         chan->iter.len_left = 0;
 702                         *ppos = 0;
 703                 } else {
 704                         copy_len = space_left;
 705                         chan->iter.len_left = len - copy_len;
 706                         *ppos = read_offset + copy_len;
 707                 }
 708                 if (__lib_ring_buffer_copy_to_user(&buf->backend, read_offset,
 709                                                &user_buf[read_count],
 710                                                copy_len)) {
 711                         /*
 712                          * Leave the len_left and ppos values at their current
 713                          * state, as we currently have a valid event to read.
 714                          */
 715                         return -EFAULT;
 716                 }
 717                 read_count += copy_len;
 718         }
 719         goto put_record;
 720
 721 nodata:
 722         *ppos = 0;
 723         chan->iter.len_left = 0;
 724 put_record:
 725         lib_ring_buffer_put_current_record(buf);
 726         return read_count;
 727 }
 728
 729 /**
 730  * lib_ring_buffer_file_read - Read buffer record payload.
 731  * @filp: file structure pointer.
 732  * @buffer: user buffer to read data into.
 733  * @count: number of bytes to read.
 734  * @ppos: file read position.
 735  *
 736  * Returns a negative value on error, or the number of bytes read on success.
 737  * ppos is used to save the position _within the current record_ between calls
 738  * to read().
 739  */
 740 static
 741 ssize_t lib_ring_buffer_file_read(struct file *filp,
 742                                   char __user *user_buf,
 743                                   size_t count,
 744                                   loff_t *ppos)
 745 {
 746         struct inode *inode = filp->lttng_f_dentry->d_inode;
 747         struct lttng_kernel_ring_buffer *buf = inode->i_private;
 748         struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan;
 749
 750         return channel_ring_buffer_file_read(filp, user_buf, count, ppos,
 751                                              chan, buf, 0);
 752 }
 753
 754 /**
 755  * channel_file_read - Read channel record payload.
 756  * @filp: file structure pointer.
 757  * @buffer: user buffer to read data into.
 758  * @count: number of bytes to read.
 759  * @ppos: file read position.
 760  *
 761  * Returns a negative value on error, or the number of bytes read on success.
 762  * ppos is used to save the position _within the current record_ between calls
 763  * to read().
 764  */
 765 static
 766 ssize_t channel_file_read(struct file *filp,
 767                           char __user *user_buf,
 768                           size_t count,
 769                           loff_t *ppos)
 770 {
 771         struct inode *inode = filp->lttng_f_dentry->d_inode;
 772         struct lttng_kernel_ring_buffer_channel *chan = inode->i_private;
 773         const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config;
 774
 775         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 776                 return channel_ring_buffer_file_read(filp, user_buf, count,
 777                                                      ppos, chan, NULL, 1);
 778         else {
 779                 struct lttng_kernel_ring_buffer *buf =
 780                         channel_get_ring_buffer(config, chan, 0);
 781                 return channel_ring_buffer_file_read(filp, user_buf, count,
 782                                                      ppos, chan, buf, 0);
 783         }
 784 }
 785
 786 static
 787 int lib_ring_buffer_file_open(struct inode *inode, struct file *file)
 788 {
 789         struct lttng_kernel_ring_buffer *buf = inode->i_private;
 790         int ret;
 791
 792         ret = lib_ring_buffer_iterator_open(buf);
 793         if (ret)
 794                 return ret;
 795
 796         file->private_data = buf;
 797         ret = nonseekable_open(inode, file);
 798         if (ret)
 799                 goto release_iter;
 800         return 0;
 801
 802 release_iter:
 803         lib_ring_buffer_iterator_release(buf);
 804         return ret;
 805 }
 806
 807 static
 808 int lib_ring_buffer_file_release(struct inode *inode, struct file *file)
 809 {
 810         struct lttng_kernel_ring_buffer *buf = inode->i_private;
 811
 812         lib_ring_buffer_iterator_release(buf);
 813         return 0;
 814 }
 815
 816 static
 817 int channel_file_open(struct inode *inode, struct file *file)
 818 {
 819         struct lttng_kernel_ring_buffer_channel *chan = inode->i_private;
 820         int ret;
 821
 822         ret = channel_iterator_open(chan);
 823         if (ret)
 824                 return ret;
 825
 826         file->private_data = chan;
 827         ret = nonseekable_open(inode, file);
 828         if (ret)
 829                 goto release_iter;
 830         return 0;
 831
 832 release_iter:
 833         channel_iterator_release(chan);
 834         return ret;
 835 }
 836
 837 static
 838 int channel_file_release(struct inode *inode, struct file *file)
 839 {
 840         struct lttng_kernel_ring_buffer_channel *chan = inode->i_private;
 841
 842         channel_iterator_release(chan);
 843         return 0;
 844 }
 845
 846 const struct file_operations channel_payload_file_operations = {
 847         .owner = THIS_MODULE,
 848         .open = channel_file_open,
 849         .release = channel_file_release,
 850         .read = channel_file_read,
 851         .llseek = vfs_lib_ring_buffer_no_llseek,
 852 };
 853 EXPORT_SYMBOL_GPL(channel_payload_file_operations);
 854
 855 const struct file_operations lib_ring_buffer_payload_file_operations = {
 856         .owner = THIS_MODULE,
 857         .open = lib_ring_buffer_file_open,
 858         .release = lib_ring_buffer_file_release,
 859         .read = lib_ring_buffer_file_read,
 860         .llseek = vfs_lib_ring_buffer_no_llseek,
 861 };
 862 EXPORT_SYMBOL_GPL(lib_ring_buffer_payload_file_operations);