drivers/staging/lttng/lib/ringbuffer/ring_buffer_frontend.c

   1 /*
   2  * ring_buffer_frontend.c
   3  *
   4  * (C) Copyright 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * Ring buffer wait-free buffer synchronization. Producer-consumer and flight
   7  * recorder (overwrite) modes. See thesis:
   8  *
   9  * Desnoyers, Mathieu (2009), "Low-Impact Operating System Tracing", Ph.D.
  10  * dissertation, Ecole Polytechnique de Montreal.
  11  * http://www.lttng.org/pub/thesis/desnoyers-dissertation-2009-12.pdf
  12  *
  13  * - Algorithm presentation in Chapter 5:
  14  *     "Lockless Multi-Core High-Throughput Buffering".
  15  * - Algorithm formal verification in Section 8.6:
  16  *     "Formal verification of LTTng"
  17  *
  18  * Author:
  19  *      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  20  *
  21  * Inspired from LTT and RelayFS:
  22  *  Karim Yaghmour <karim@opersys.com>
  23  *  Tom Zanussi <zanussi@us.ibm.com>
  24  *  Bob Wisniewski <bob@watson.ibm.com>
  25  * And from K42 :
  26  *  Bob Wisniewski <bob@watson.ibm.com>
  27  *
  28  * Buffer reader semantic :
  29  *
  30  * - get_subbuf_size
  31  * while buffer is not finalized and empty
  32  *   - get_subbuf
  33  *     - if return value != 0, continue
  34  *   - splice one subbuffer worth of data to a pipe
  35  *   - splice the data from pipe to disk/network
  36  *   - put_subbuf
  37  *
  38  * Dual LGPL v2.1/GPL v2 license.
  39  */
  40
  41 #include <linux/delay.h>
  42 #include <linux/module.h>
  43 #include <linux/percpu.h>
  44
  45 #include "../../wrapper/ringbuffer/config.h"
  46 #include "../../wrapper/ringbuffer/backend.h"
  47 #include "../../wrapper/ringbuffer/frontend.h"
  48 #include "../../wrapper/ringbuffer/iterator.h"
  49 #include "../../wrapper/ringbuffer/nohz.h"
  50
  51 /*
  52  * Internal structure representing offsets to use at a sub-buffer switch.
  53  */
  54 struct switch_offsets {
  55         unsigned long begin, end, old;
  56         size_t pre_header_padding, size;
  57         uint switch_new_start:1, switch_new_end:1, switch_old_start:1,
  58                      switch_old_end:1;
  59 };
  60
  61 #ifdef CONFIG_NO_HZ
  62 enum tick_nohz_val {
  63         TICK_NOHZ_STOP,
  64         TICK_NOHZ_FLUSH,
  65         TICK_NOHZ_RESTART,
  66 };
  67
  68 static ATOMIC_NOTIFIER_HEAD(tick_nohz_notifier);
  69 #endif /* CONFIG_NO_HZ */
  70
  71 static DEFINE_PER_CPU(spinlock_t, ring_buffer_nohz_lock);
  72
  73 DEFINE_PER_CPU(unsigned int, lib_ring_buffer_nesting);
  74 EXPORT_PER_CPU_SYMBOL(lib_ring_buffer_nesting);
  75
  76 static
  77 void lib_ring_buffer_print_errors(struct channel *chan,
  78                                   struct lib_ring_buffer *buf, int cpu);
  79
  80 /*
  81  * Must be called under cpu hotplug protection.
  82  */
  83 void lib_ring_buffer_free(struct lib_ring_buffer *buf)
  84 {
  85         struct channel *chan = buf->backend.chan;
  86
  87         lib_ring_buffer_print_errors(chan, buf, buf->backend.cpu);
  88         kfree(buf->commit_hot);
  89         kfree(buf->commit_cold);
  90
  91         lib_ring_buffer_backend_free(&buf->backend);
  92 }
  93
  94 /**
  95  * lib_ring_buffer_reset - Reset ring buffer to initial values.
  96  * @buf: Ring buffer.
  97  *
  98  * Effectively empty the ring buffer. Should be called when the buffer is not
  99  * used for writing. The ring buffer can be opened for reading, but the reader
 100  * should not be using the iterator concurrently with reset. The previous
 101  * current iterator record is reset.
 102  */
 103 void lib_ring_buffer_reset(struct lib_ring_buffer *buf)
 104 {
 105         struct channel *chan = buf->backend.chan;
 106         const struct lib_ring_buffer_config *config = chan->backend.config;
 107         unsigned int i;
 108
 109         /*
 110          * Reset iterator first. It will put the subbuffer if it currently holds
 111          * it.
 112          */
 113         lib_ring_buffer_iterator_reset(buf);
 114         v_set(config, &buf->offset, 0);
 115         for (i = 0; i < chan->backend.num_subbuf; i++) {
 116                 v_set(config, &buf->commit_hot[i].cc, 0);
 117                 v_set(config, &buf->commit_hot[i].seq, 0);
 118                 v_set(config, &buf->commit_cold[i].cc_sb, 0);
 119         }
 120         atomic_long_set(&buf->consumed, 0);
 121         atomic_set(&buf->record_disabled, 0);
 122         v_set(config, &buf->last_tsc, 0);
 123         lib_ring_buffer_backend_reset(&buf->backend);
 124         /* Don't reset number of active readers */
 125         v_set(config, &buf->records_lost_full, 0);
 126         v_set(config, &buf->records_lost_wrap, 0);
 127         v_set(config, &buf->records_lost_big, 0);
 128         v_set(config, &buf->records_count, 0);
 129         v_set(config, &buf->records_overrun, 0);
 130         buf->finalized = 0;
 131 }
 132 EXPORT_SYMBOL_GPL(lib_ring_buffer_reset);
 133
 134 /**
 135  * channel_reset - Reset channel to initial values.
 136  * @chan: Channel.
 137  *
 138  * Effectively empty the channel. Should be called when the channel is not used
 139  * for writing. The channel can be opened for reading, but the reader should not
 140  * be using the iterator concurrently with reset. The previous current iterator
 141  * record is reset.
 142  */
 143 void channel_reset(struct channel *chan)
 144 {
 145         /*
 146          * Reset iterators first. Will put the subbuffer if held for reading.
 147          */
 148         channel_iterator_reset(chan);
 149         atomic_set(&chan->record_disabled, 0);
 150         /* Don't reset commit_count_mask, still valid */
 151         channel_backend_reset(&chan->backend);
 152         /* Don't reset switch/read timer interval */
 153         /* Don't reset notifiers and notifier enable bits */
 154         /* Don't reset reader reference count */
 155 }
 156 EXPORT_SYMBOL_GPL(channel_reset);
 157
 158 /*
 159  * Must be called under cpu hotplug protection.
 160  */
 161 int lib_ring_buffer_create(struct lib_ring_buffer *buf,
 162                            struct channel_backend *chanb, int cpu)
 163 {
 164         const struct lib_ring_buffer_config *config = chanb->config;
 165         struct channel *chan = container_of(chanb, struct channel, backend);
 166         void *priv = chanb->priv;
 167         size_t subbuf_header_size;
 168         u64 tsc;
 169         int ret;
 170
 171         /* Test for cpu hotplug */
 172         if (buf->backend.allocated)
 173                 return 0;
 174
 175         /*
 176          * Paranoia: per cpu dynamic allocation is not officially documented as
 177          * zeroing the memory, so let's do it here too, just in case.
 178          */
 179         memset(buf, 0, sizeof(*buf));
 180
 181         ret = lib_ring_buffer_backend_create(&buf->backend, &chan->backend, cpu);
 182         if (ret)
 183                 return ret;
 184
 185         buf->commit_hot =
 186                 kzalloc_node(ALIGN(sizeof(*buf->commit_hot)
 187                                    * chan->backend.num_subbuf,
 188                                    1 << INTERNODE_CACHE_SHIFT),
 189                         GFP_KERNEL, cpu_to_node(max(cpu, 0)));
 190         if (!buf->commit_hot) {
 191                 ret = -ENOMEM;
 192                 goto free_chanbuf;
 193         }
 194
 195         buf->commit_cold =
 196                 kzalloc_node(ALIGN(sizeof(*buf->commit_cold)
 197                                    * chan->backend.num_subbuf,
 198                                    1 << INTERNODE_CACHE_SHIFT),
 199                         GFP_KERNEL, cpu_to_node(max(cpu, 0)));
 200         if (!buf->commit_cold) {
 201                 ret = -ENOMEM;
 202                 goto free_commit;
 203         }
 204
 205         init_waitqueue_head(&buf->read_wait);
 206         init_waitqueue_head(&buf->write_wait);
 207         raw_spin_lock_init(&buf->raw_tick_nohz_spinlock);
 208
 209         /*
 210          * Write the subbuffer header for first subbuffer so we know the total
 211          * duration of data gathering.
 212          */
 213         subbuf_header_size = config->cb.subbuffer_header_size();
 214         v_set(config, &buf->offset, subbuf_header_size);
 215         subbuffer_id_clear_noref(config, &buf->backend.buf_wsb[0].id);
 216         tsc = config->cb.ring_buffer_clock_read(buf->backend.chan);
 217         config->cb.buffer_begin(buf, tsc, 0);
 218         v_add(config, subbuf_header_size, &buf->commit_hot[0].cc);
 219
 220         if (config->cb.buffer_create) {
 221                 ret = config->cb.buffer_create(buf, priv, cpu, chanb->name);
 222                 if (ret)
 223                         goto free_init;
 224         }
 225
 226         /*
 227          * Ensure the buffer is ready before setting it to allocated and setting
 228          * the cpumask.
 229          * Used for cpu hotplug vs cpumask iteration.
 230          */
 231         smp_wmb();
 232         buf->backend.allocated = 1;
 233
 234         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 235                 CHAN_WARN_ON(chan, cpumask_test_cpu(cpu,
 236                              chan->backend.cpumask));
 237                 cpumask_set_cpu(cpu, chan->backend.cpumask);
 238         }
 239
 240         return 0;
 241
 242         /* Error handling */
 243 free_init:
 244         kfree(buf->commit_cold);
 245 free_commit:
 246         kfree(buf->commit_hot);
 247 free_chanbuf:
 248         lib_ring_buffer_backend_free(&buf->backend);
 249         return ret;
 250 }
 251
 252 static void switch_buffer_timer(unsigned long data)
 253 {
 254         struct lib_ring_buffer *buf = (struct lib_ring_buffer *)data;
 255         struct channel *chan = buf->backend.chan;
 256         const struct lib_ring_buffer_config *config = chan->backend.config;
 257
 258         /*
 259          * Only flush buffers periodically if readers are active.
 260          */
 261         if (atomic_long_read(&buf->active_readers))
 262                 lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
 263
 264         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 265                 mod_timer_pinned(&buf->switch_timer,
 266                                  jiffies + chan->switch_timer_interval);
 267         else
 268                 mod_timer(&buf->switch_timer,
 269                           jiffies + chan->switch_timer_interval);
 270 }
 271
 272 /*
 273  * Called with ring_buffer_nohz_lock held for per-cpu buffers.
 274  */
 275 static void lib_ring_buffer_start_switch_timer(struct lib_ring_buffer *buf)
 276 {
 277         struct channel *chan = buf->backend.chan;
 278         const struct lib_ring_buffer_config *config = chan->backend.config;
 279
 280         if (!chan->switch_timer_interval || buf->switch_timer_enabled)
 281                 return;
 282         init_timer(&buf->switch_timer);
 283         buf->switch_timer.function = switch_buffer_timer;
 284         buf->switch_timer.expires = jiffies + chan->switch_timer_interval;
 285         buf->switch_timer.data = (unsigned long)buf;
 286         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 287                 add_timer_on(&buf->switch_timer, buf->backend.cpu);
 288         else
 289                 add_timer(&buf->switch_timer);
 290         buf->switch_timer_enabled = 1;
 291 }
 292
 293 /*
 294  * Called with ring_buffer_nohz_lock held for per-cpu buffers.
 295  */
 296 static void lib_ring_buffer_stop_switch_timer(struct lib_ring_buffer *buf)
 297 {
 298         struct channel *chan = buf->backend.chan;
 299
 300         if (!chan->switch_timer_interval || !buf->switch_timer_enabled)
 301                 return;
 302
 303         del_timer_sync(&buf->switch_timer);
 304         buf->switch_timer_enabled = 0;
 305 }
 306
 307 /*
 308  * Polling timer to check the channels for data.
 309  */
 310 static void read_buffer_timer(unsigned long data)
 311 {
 312         struct lib_ring_buffer *buf = (struct lib_ring_buffer *)data;
 313         struct channel *chan = buf->backend.chan;
 314         const struct lib_ring_buffer_config *config = chan->backend.config;
 315
 316         CHAN_WARN_ON(chan, !buf->backend.allocated);
 317
 318         if (atomic_long_read(&buf->active_readers)
 319             && lib_ring_buffer_poll_deliver(config, buf, chan)) {
 320                 wake_up_interruptible(&buf->read_wait);
 321                 wake_up_interruptible(&chan->read_wait);
 322         }
 323
 324         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 325                 mod_timer_pinned(&buf->read_timer,
 326                                  jiffies + chan->read_timer_interval);
 327         else
 328                 mod_timer(&buf->read_timer,
 329                           jiffies + chan->read_timer_interval);
 330 }
 331
 332 /*
 333  * Called with ring_buffer_nohz_lock held for per-cpu buffers.
 334  */
 335 static void lib_ring_buffer_start_read_timer(struct lib_ring_buffer *buf)
 336 {
 337         struct channel *chan = buf->backend.chan;
 338         const struct lib_ring_buffer_config *config = chan->backend.config;
 339
 340         if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
 341             || !chan->read_timer_interval
 342             || buf->read_timer_enabled)
 343                 return;
 344
 345         init_timer(&buf->read_timer);
 346         buf->read_timer.function = read_buffer_timer;
 347         buf->read_timer.expires = jiffies + chan->read_timer_interval;
 348         buf->read_timer.data = (unsigned long)buf;
 349
 350         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 351                 add_timer_on(&buf->read_timer, buf->backend.cpu);
 352         else
 353                 add_timer(&buf->read_timer);
 354         buf->read_timer_enabled = 1;
 355 }
 356
 357 /*
 358  * Called with ring_buffer_nohz_lock held for per-cpu buffers.
 359  */
 360 static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf)
 361 {
 362         struct channel *chan = buf->backend.chan;
 363         const struct lib_ring_buffer_config *config = chan->backend.config;
 364
 365         if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
 366             || !chan->read_timer_interval
 367             || !buf->read_timer_enabled)
 368                 return;
 369
 370         del_timer_sync(&buf->read_timer);
 371         /*
 372          * do one more check to catch data that has been written in the last
 373          * timer period.
 374          */
 375         if (lib_ring_buffer_poll_deliver(config, buf, chan)) {
 376                 wake_up_interruptible(&buf->read_wait);
 377                 wake_up_interruptible(&chan->read_wait);
 378         }
 379         buf->read_timer_enabled = 0;
 380 }
 381
 382 #ifdef CONFIG_HOTPLUG_CPU
 383 /**
 384  *      lib_ring_buffer_cpu_hp_callback - CPU hotplug callback
 385  *      @nb: notifier block
 386  *      @action: hotplug action to take
 387  *      @hcpu: CPU number
 388  *
 389  *      Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
 390  */
 391 static
 392 int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
 393                                               unsigned long action,
 394                                               void *hcpu)
 395 {
 396         unsigned int cpu = (unsigned long)hcpu;
 397         struct channel *chan = container_of(nb, struct channel,
 398                                             cpu_hp_notifier);
 399         struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
 400         const struct lib_ring_buffer_config *config = chan->backend.config;
 401
 402         if (!chan->cpu_hp_enable)
 403                 return NOTIFY_DONE;
 404
 405         CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL);
 406
 407         switch (action) {
 408         case CPU_DOWN_FAILED:
 409         case CPU_DOWN_FAILED_FROZEN:
 410         case CPU_ONLINE:
 411         case CPU_ONLINE_FROZEN:
 412                 wake_up_interruptible(&chan->hp_wait);
 413                 lib_ring_buffer_start_switch_timer(buf);
 414                 lib_ring_buffer_start_read_timer(buf);
 415                 return NOTIFY_OK;
 416
 417         case CPU_DOWN_PREPARE:
 418         case CPU_DOWN_PREPARE_FROZEN:
 419                 lib_ring_buffer_stop_switch_timer(buf);
 420                 lib_ring_buffer_stop_read_timer(buf);
 421                 return NOTIFY_OK;
 422
 423         case CPU_DEAD:
 424         case CPU_DEAD_FROZEN:
 425                 /*
 426                  * Performing a buffer switch on a remote CPU. Performed by
 427                  * the CPU responsible for doing the hotunplug after the target
 428                  * CPU stopped running completely. Ensures that all data
 429                  * from that remote CPU is flushed.
 430                  */
 431                 lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
 432                 return NOTIFY_OK;
 433
 434         default:
 435                 return NOTIFY_DONE;
 436         }
 437 }
 438 #endif
 439
 440 #if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER)
 441 /*
 442  * For per-cpu buffers, call the reader wakeups before switching the buffer, so
 443  * that wake-up-tracing generated events are flushed before going idle (in
 444  * tick_nohz). We test if the spinlock is locked to deal with the race where
 445  * readers try to sample the ring buffer before we perform the switch. We let
 446  * the readers retry in that case. If there is data in the buffer, the wake up
 447  * is going to forbid the CPU running the reader thread from going idle.
 448  */
 449 static int notrace ring_buffer_tick_nohz_callback(struct notifier_block *nb,
 450                                                   unsigned long val,
 451                                                   void *data)
 452 {
 453         struct channel *chan = container_of(nb, struct channel,
 454                                             tick_nohz_notifier);
 455         const struct lib_ring_buffer_config *config = chan->backend.config;
 456         struct lib_ring_buffer *buf;
 457         int cpu = smp_processor_id();
 458
 459         if (config->alloc != RING_BUFFER_ALLOC_PER_CPU) {
 460                 /*
 461                  * We don't support keeping the system idle with global buffers
 462                  * and streaming active. In order to do so, we would need to
 463                  * sample a non-nohz-cpumask racelessly with the nohz updates
 464                  * without adding synchronization overhead to nohz. Leave this
 465                  * use-case out for now.
 466                  */
 467                 return 0;
 468         }
 469
 470         buf = channel_get_ring_buffer(config, chan, cpu);
 471         switch (val) {
 472         case TICK_NOHZ_FLUSH:
 473                 raw_spin_lock(&buf->raw_tick_nohz_spinlock);
 474                 if (config->wakeup == RING_BUFFER_WAKEUP_BY_TIMER
 475                     && chan->read_timer_interval
 476                     && atomic_long_read(&buf->active_readers)
 477                     && (lib_ring_buffer_poll_deliver(config, buf, chan)
 478                         || lib_ring_buffer_pending_data(config, buf, chan))) {
 479                         wake_up_interruptible(&buf->read_wait);
 480                         wake_up_interruptible(&chan->read_wait);
 481                 }
 482                 if (chan->switch_timer_interval)
 483                         lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
 484                 raw_spin_unlock(&buf->raw_tick_nohz_spinlock);
 485                 break;
 486         case TICK_NOHZ_STOP:
 487                 spin_lock(&__get_cpu_var(ring_buffer_nohz_lock));
 488                 lib_ring_buffer_stop_switch_timer(buf);
 489                 lib_ring_buffer_stop_read_timer(buf);
 490                 spin_unlock(&__get_cpu_var(ring_buffer_nohz_lock));
 491                 break;
 492         case TICK_NOHZ_RESTART:
 493                 spin_lock(&__get_cpu_var(ring_buffer_nohz_lock));
 494                 lib_ring_buffer_start_read_timer(buf);
 495                 lib_ring_buffer_start_switch_timer(buf);
 496                 spin_unlock(&__get_cpu_var(ring_buffer_nohz_lock));
 497                 break;
 498         }
 499
 500         return 0;
 501 }
 502
 503 void notrace lib_ring_buffer_tick_nohz_flush(void)
 504 {
 505         atomic_notifier_call_chain(&tick_nohz_notifier, TICK_NOHZ_FLUSH,
 506                                    NULL);
 507 }
 508
 509 void notrace lib_ring_buffer_tick_nohz_stop(void)
 510 {
 511         atomic_notifier_call_chain(&tick_nohz_notifier, TICK_NOHZ_STOP,
 512                                    NULL);
 513 }
 514
 515 void notrace lib_ring_buffer_tick_nohz_restart(void)
 516 {
 517         atomic_notifier_call_chain(&tick_nohz_notifier, TICK_NOHZ_RESTART,
 518                                    NULL);
 519 }
 520 #endif /* defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER) */
 521
 522 /*
 523  * Holds CPU hotplug.
 524  */
 525 static void channel_unregister_notifiers(struct channel *chan)
 526 {
 527         const struct lib_ring_buffer_config *config = chan->backend.config;
 528         int cpu;
 529
 530         channel_iterator_unregister_notifiers(chan);
 531         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 532 #ifdef CONFIG_NO_HZ
 533                 /*
 534                  * Remove the nohz notifier first, so we are certain we stop
 535                  * the timers.
 536                  */
 537                 atomic_notifier_chain_unregister(&tick_nohz_notifier,
 538                                                  &chan->tick_nohz_notifier);
 539                 /*
 540                  * ring_buffer_nohz_lock will not be needed below, because
 541                  * we just removed the notifiers, which were the only source of
 542                  * concurrency.
 543                  */
 544 #endif /* CONFIG_NO_HZ */
 545 #ifdef CONFIG_HOTPLUG_CPU
 546                 get_online_cpus();
 547                 chan->cpu_hp_enable = 0;
 548                 for_each_online_cpu(cpu) {
 549                         struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
 550                                                               cpu);
 551                         lib_ring_buffer_stop_switch_timer(buf);
 552                         lib_ring_buffer_stop_read_timer(buf);
 553                 }
 554                 put_online_cpus();
 555                 unregister_cpu_notifier(&chan->cpu_hp_notifier);
 556 #else
 557                 for_each_possible_cpu(cpu) {
 558                         struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
 559                                                               cpu);
 560                         lib_ring_buffer_stop_switch_timer(buf);
 561                         lib_ring_buffer_stop_read_timer(buf);
 562                 }
 563 #endif
 564         } else {
 565                 struct lib_ring_buffer *buf = chan->backend.buf;
 566
 567                 lib_ring_buffer_stop_switch_timer(buf);
 568                 lib_ring_buffer_stop_read_timer(buf);
 569         }
 570         channel_backend_unregister_notifiers(&chan->backend);
 571 }
 572
 573 static void channel_free(struct channel *chan)
 574 {
 575         channel_iterator_free(chan);
 576         channel_backend_free(&chan->backend);
 577         kfree(chan);
 578 }
 579
 580 /**
 581  * channel_create - Create channel.
 582  * @config: ring buffer instance configuration
 583  * @name: name of the channel
 584  * @priv: ring buffer client private data
 585  * @buf_addr: pointer the the beginning of the preallocated buffer contiguous
 586  *            address mapping. It is used only by RING_BUFFER_STATIC
 587  *            configuration. It can be set to NULL for other backends.
 588  * @subbuf_size: subbuffer size
 589  * @num_subbuf: number of subbuffers
 590  * @switch_timer_interval: Time interval (in us) to fill sub-buffers with
 591  *                         padding to let readers get those sub-buffers.
 592  *                         Used for live streaming.
 593  * @read_timer_interval: Time interval (in us) to wake up pending readers.
 594  *
 595  * Holds cpu hotplug.
 596  * Returns NULL on failure.
 597  */
 598 struct channel *channel_create(const struct lib_ring_buffer_config *config,
 599                    const char *name, void *priv, void *buf_addr,
 600                    size_t subbuf_size,
 601                    size_t num_subbuf, unsigned int switch_timer_interval,
 602                    unsigned int read_timer_interval)
 603 {
 604         int ret, cpu;
 605         struct channel *chan;
 606
 607         if (lib_ring_buffer_check_config(config, switch_timer_interval,
 608                                          read_timer_interval))
 609                 return NULL;
 610
 611         chan = kzalloc(sizeof(struct channel), GFP_KERNEL);
 612         if (!chan)
 613                 return NULL;
 614
 615         ret = channel_backend_init(&chan->backend, name, config, priv,
 616                                    subbuf_size, num_subbuf);
 617         if (ret)
 618                 goto error;
 619
 620         ret = channel_iterator_init(chan);
 621         if (ret)
 622                 goto error_free_backend;
 623
 624         chan->commit_count_mask = (~0UL >> chan->backend.num_subbuf_order);
 625         chan->switch_timer_interval = usecs_to_jiffies(switch_timer_interval);
 626         chan->read_timer_interval = usecs_to_jiffies(read_timer_interval);
 627         kref_init(&chan->ref);
 628         init_waitqueue_head(&chan->read_wait);
 629         init_waitqueue_head(&chan->hp_wait);
 630
 631         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 632 #if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER)
 633                 /* Only benefit from NO_HZ idle with per-cpu buffers for now. */
 634                 chan->tick_nohz_notifier.notifier_call =
 635                         ring_buffer_tick_nohz_callback;
 636                 chan->tick_nohz_notifier.priority = ~0U;
 637                 atomic_notifier_chain_register(&tick_nohz_notifier,
 638                                        &chan->tick_nohz_notifier);
 639 #endif /* defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER) */
 640
 641                 /*
 642                  * In case of non-hotplug cpu, if the ring-buffer is allocated
 643                  * in early initcall, it will not be notified of secondary cpus.
 644                  * In that off case, we need to allocate for all possible cpus.
 645                  */
 646 #ifdef CONFIG_HOTPLUG_CPU
 647                 chan->cpu_hp_notifier.notifier_call =
 648                                 lib_ring_buffer_cpu_hp_callback;
 649                 chan->cpu_hp_notifier.priority = 6;
 650                 register_cpu_notifier(&chan->cpu_hp_notifier);
 651
 652                 get_online_cpus();
 653                 for_each_online_cpu(cpu) {
 654                         struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
 655                                                                cpu);
 656                         spin_lock(&per_cpu(ring_buffer_nohz_lock, cpu));
 657                         lib_ring_buffer_start_switch_timer(buf);
 658                         lib_ring_buffer_start_read_timer(buf);
 659                         spin_unlock(&per_cpu(ring_buffer_nohz_lock, cpu));
 660                 }
 661                 chan->cpu_hp_enable = 1;
 662                 put_online_cpus();
 663 #else
 664                 for_each_possible_cpu(cpu) {
 665                         struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
 666                                                               cpu);
 667                         spin_lock(&per_cpu(ring_buffer_nohz_lock, cpu));
 668                         lib_ring_buffer_start_switch_timer(buf);
 669                         lib_ring_buffer_start_read_timer(buf);
 670                         spin_unlock(&per_cpu(ring_buffer_nohz_lock, cpu));
 671                 }
 672 #endif
 673         } else {
 674                 struct lib_ring_buffer *buf = chan->backend.buf;
 675
 676                 lib_ring_buffer_start_switch_timer(buf);
 677                 lib_ring_buffer_start_read_timer(buf);
 678         }
 679
 680         return chan;
 681
 682 error_free_backend:
 683         channel_backend_free(&chan->backend);
 684 error:
 685         kfree(chan);
 686         return NULL;
 687 }
 688 EXPORT_SYMBOL_GPL(channel_create);
 689
 690 static
 691 void channel_release(struct kref *kref)
 692 {
 693         struct channel *chan = container_of(kref, struct channel, ref);
 694         channel_free(chan);
 695 }
 696
 697 /**
 698  * channel_destroy - Finalize, wait for q.s. and destroy channel.
 699  * @chan: channel to destroy
 700  *
 701  * Holds cpu hotplug.
 702  * Call "destroy" callback, finalize channels, and then decrement the
 703  * channel reference count.  Note that when readers have completed data
 704  * consumption of finalized channels, get_subbuf() will return -ENODATA.
 705  * They should release their handle at that point.  Returns the private
 706  * data pointer.
 707  */
 708 void *channel_destroy(struct channel *chan)
 709 {
 710         int cpu;
 711         const struct lib_ring_buffer_config *config = chan->backend.config;
 712         void *priv;
 713
 714         channel_unregister_notifiers(chan);
 715
 716         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 717                 /*
 718                  * No need to hold cpu hotplug, because all notifiers have been
 719                  * unregistered.
 720                  */
 721                 for_each_channel_cpu(cpu, chan) {
 722                         struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
 723                                                               cpu);
 724
 725                         if (config->cb.buffer_finalize)
 726                                 config->cb.buffer_finalize(buf,
 727                                                            chan->backend.priv,
 728                                                            cpu);
 729                         if (buf->backend.allocated)
 730                                 lib_ring_buffer_switch_slow(buf, SWITCH_FLUSH);
 731                         /*
 732                          * Perform flush before writing to finalized.
 733                          */
 734                         smp_wmb();
 735                         ACCESS_ONCE(buf->finalized) = 1;
 736                         wake_up_interruptible(&buf->read_wait);
 737                 }
 738         } else {
 739                 struct lib_ring_buffer *buf = chan->backend.buf;
 740
 741                 if (config->cb.buffer_finalize)
 742                         config->cb.buffer_finalize(buf, chan->backend.priv, -1);
 743                 if (buf->backend.allocated)
 744                         lib_ring_buffer_switch_slow(buf, SWITCH_FLUSH);
 745                 /*
 746                  * Perform flush before writing to finalized.
 747                  */
 748                 smp_wmb();
 749                 ACCESS_ONCE(buf->finalized) = 1;
 750                 wake_up_interruptible(&buf->read_wait);
 751         }
 752         ACCESS_ONCE(chan->finalized) = 1;
 753         wake_up_interruptible(&chan->hp_wait);
 754         wake_up_interruptible(&chan->read_wait);
 755         priv = chan->backend.priv;
 756         kref_put(&chan->ref, channel_release);
 757         return priv;
 758 }
 759 EXPORT_SYMBOL_GPL(channel_destroy);
 760
 761 struct lib_ring_buffer *channel_get_ring_buffer(
 762                                         const struct lib_ring_buffer_config *config,
 763                                         struct channel *chan, int cpu)
 764 {
 765         if (config->alloc == RING_BUFFER_ALLOC_GLOBAL)
 766                 return chan->backend.buf;
 767         else
 768                 return per_cpu_ptr(chan->backend.buf, cpu);
 769 }
 770 EXPORT_SYMBOL_GPL(channel_get_ring_buffer);
 771
 772 int lib_ring_buffer_open_read(struct lib_ring_buffer *buf)
 773 {
 774         struct channel *chan = buf->backend.chan;
 775
 776         if (!atomic_long_add_unless(&buf->active_readers, 1, 1))
 777                 return -EBUSY;
 778         kref_get(&chan->ref);
 779         smp_mb__after_atomic_inc();
 780         return 0;
 781 }
 782 EXPORT_SYMBOL_GPL(lib_ring_buffer_open_read);
 783
 784 void lib_ring_buffer_release_read(struct lib_ring_buffer *buf)
 785 {
 786         struct channel *chan = buf->backend.chan;
 787
 788         CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
 789         smp_mb__before_atomic_dec();
 790         atomic_long_dec(&buf->active_readers);
 791         kref_put(&chan->ref, channel_release);
 792 }
 793 EXPORT_SYMBOL_GPL(lib_ring_buffer_release_read);
 794
 795 /*
 796  * Promote compiler barrier to a smp_mb().
 797  * For the specific ring buffer case, this IPI call should be removed if the
 798  * architecture does not reorder writes.  This should eventually be provided by
 799  * a separate architecture-specific infrastructure.
 800  */
 801 static void remote_mb(void *info)
 802 {
 803         smp_mb();
 804 }
 805
 806 /**
 807  * lib_ring_buffer_snapshot - save subbuffer position snapshot (for read)
 808  * @buf: ring buffer
 809  * @consumed: consumed count indicating the position where to read
 810  * @produced: produced count, indicates position when to stop reading
 811  *
 812  * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
 813  * data to read at consumed position, or 0 if the get operation succeeds.
 814  * Busy-loop trying to get data if the tick_nohz sequence lock is held.
 815  */
 816
 817 int lib_ring_buffer_snapshot(struct lib_ring_buffer *buf,
 818                              unsigned long *consumed, unsigned long *produced)
 819 {
 820         struct channel *chan = buf->backend.chan;
 821         const struct lib_ring_buffer_config *config = chan->backend.config;
 822         unsigned long consumed_cur, write_offset;
 823         int finalized;
 824
 825 retry:
 826         finalized = ACCESS_ONCE(buf->finalized);
 827         /*
 828          * Read finalized before counters.
 829          */
 830         smp_rmb();
 831         consumed_cur = atomic_long_read(&buf->consumed);
 832         /*
 833          * No need to issue a memory barrier between consumed count read and
 834          * write offset read, because consumed count can only change
 835          * concurrently in overwrite mode, and we keep a sequence counter
 836          * identifier derived from the write offset to check we are getting
 837          * the same sub-buffer we are expecting (the sub-buffers are atomically
 838          * "tagged" upon writes, tags are checked upon read).
 839          */
 840         write_offset = v_read(config, &buf->offset);
 841
 842         /*
 843          * Check that we are not about to read the same subbuffer in
 844          * which the writer head is.
 845          */
 846         if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
 847             == 0)
 848                 goto nodata;
 849
 850         *consumed = consumed_cur;
 851         *produced = subbuf_trunc(write_offset, chan);
 852
 853         return 0;
 854
 855 nodata:
 856         /*
 857          * The memory barriers __wait_event()/wake_up_interruptible() take care
 858          * of "raw_spin_is_locked" memory ordering.
 859          */
 860         if (finalized)
 861                 return -ENODATA;
 862         else if (raw_spin_is_locked(&buf->raw_tick_nohz_spinlock))
 863                 goto retry;
 864         else
 865                 return -EAGAIN;
 866 }
 867 EXPORT_SYMBOL_GPL(lib_ring_buffer_snapshot);
 868
 869 /**
 870  * lib_ring_buffer_put_snapshot - move consumed counter forward
 871  *
 872  * Should only be called from consumer context.
 873  * @buf: ring buffer
 874  * @consumed_new: new consumed count value
 875  */
 876 void lib_ring_buffer_move_consumer(struct lib_ring_buffer *buf,
 877                                    unsigned long consumed_new)
 878 {
 879         struct lib_ring_buffer_backend *bufb = &buf->backend;
 880         struct channel *chan = bufb->chan;
 881         unsigned long consumed;
 882
 883         CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
 884
 885         /*
 886          * Only push the consumed value forward.
 887          * If the consumed cmpxchg fails, this is because we have been pushed by
 888          * the writer in flight recorder mode.
 889          */
 890         consumed = atomic_long_read(&buf->consumed);
 891         while ((long) consumed - (long) consumed_new < 0)
 892                 consumed = atomic_long_cmpxchg(&buf->consumed, consumed,
 893                                                consumed_new);
 894         /* Wake-up the metadata producer */
 895         wake_up_interruptible(&buf->write_wait);
 896 }
 897 EXPORT_SYMBOL_GPL(lib_ring_buffer_move_consumer);
 898
 899 /**
 900  * lib_ring_buffer_get_subbuf - get exclusive access to subbuffer for reading
 901  * @buf: ring buffer
 902  * @consumed: consumed count indicating the position where to read
 903  *
 904  * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
 905  * data to read at consumed position, or 0 if the get operation succeeds.
 906  * Busy-loop trying to get data if the tick_nohz sequence lock is held.
 907  */
 908 int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf,
 909                                unsigned long consumed)
 910 {
 911         struct channel *chan = buf->backend.chan;
 912         const struct lib_ring_buffer_config *config = chan->backend.config;
 913         unsigned long consumed_cur, consumed_idx, commit_count, write_offset;
 914         int ret;
 915         int finalized;
 916
 917 retry:
 918         finalized = ACCESS_ONCE(buf->finalized);
 919         /*
 920          * Read finalized before counters.
 921          */
 922         smp_rmb();
 923         consumed_cur = atomic_long_read(&buf->consumed);
 924         consumed_idx = subbuf_index(consumed, chan);
 925         commit_count = v_read(config, &buf->commit_cold[consumed_idx].cc_sb);
 926         /*
 927          * Make sure we read the commit count before reading the buffer
 928          * data and the write offset. Correct consumed offset ordering
 929          * wrt commit count is insured by the use of cmpxchg to update
 930          * the consumed offset.
 931          * smp_call_function_single can fail if the remote CPU is offline,
 932          * this is OK because then there is no wmb to execute there.
 933          * If our thread is executing on the same CPU as the on the buffers
 934          * belongs to, we don't have to synchronize it at all. If we are
 935          * migrated, the scheduler will take care of the memory barriers.
 936          * Normally, smp_call_function_single() should ensure program order when
 937          * executing the remote function, which implies that it surrounds the
 938          * function execution with :
 939          * smp_mb()
 940          * send IPI
 941          * csd_lock_wait
 942          *                recv IPI
 943          *                smp_mb()
 944          *                exec. function
 945          *                smp_mb()
 946          *                csd unlock
 947          * smp_mb()
 948          *
 949          * However, smp_call_function_single() does not seem to clearly execute
 950          * such barriers. It depends on spinlock semantic to provide the barrier
 951          * before executing the IPI and, when busy-looping, csd_lock_wait only
 952          * executes smp_mb() when it has to wait for the other CPU.
 953          *
 954          * I don't trust this code. Therefore, let's add the smp_mb() sequence
 955          * required ourself, even if duplicated. It has no performance impact
 956          * anyway.
 957          *
 958          * smp_mb() is needed because smp_rmb() and smp_wmb() only order read vs
 959          * read and write vs write. They do not ensure core synchronization. We
 960          * really have to ensure total order between the 3 barriers running on
 961          * the 2 CPUs.
 962          */
 963         if (config->ipi == RING_BUFFER_IPI_BARRIER) {
 964                 if (config->sync == RING_BUFFER_SYNC_PER_CPU
 965                     && config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 966                         if (raw_smp_processor_id() != buf->backend.cpu) {
 967                                 /* Total order with IPI handler smp_mb() */
 968                                 smp_mb();
 969                                 smp_call_function_single(buf->backend.cpu,
 970                                                          remote_mb, NULL, 1);
 971                                 /* Total order with IPI handler smp_mb() */
 972                                 smp_mb();
 973                         }
 974                 } else {
 975                         /* Total order with IPI handler smp_mb() */
 976                         smp_mb();
 977                         smp_call_function(remote_mb, NULL, 1);
 978                         /* Total order with IPI handler smp_mb() */
 979                         smp_mb();
 980                 }
 981         } else {
 982                 /*
 983                  * Local rmb to match the remote wmb to read the commit count
 984                  * before the buffer data and the write offset.
 985                  */
 986                 smp_rmb();
 987         }
 988
 989         write_offset = v_read(config, &buf->offset);
 990
 991         /*
 992          * Check that the buffer we are getting is after or at consumed_cur
 993          * position.
 994          */
 995         if ((long) subbuf_trunc(consumed, chan)
 996             - (long) subbuf_trunc(consumed_cur, chan) < 0)
 997                 goto nodata;
 998
 999         /*
1000          * Check that the subbuffer we are trying to consume has been
1001          * already fully committed.
1002          */
1003         if (((commit_count - chan->backend.subbuf_size)
1004              & chan->commit_count_mask)
1005             - (buf_trunc(consumed_cur, chan)
1006                >> chan->backend.num_subbuf_order)
1007             != 0)
1008                 goto nodata;
1009
1010         /*
1011          * Check that we are not about to read the same subbuffer in
1012          * which the writer head is.
1013          */
1014         if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
1015             == 0)
1016                 goto nodata;
1017
1018         /*
1019          * Failure to get the subbuffer causes a busy-loop retry without going
1020          * to a wait queue. These are caused by short-lived race windows where
1021          * the writer is getting access to a subbuffer we were trying to get
1022          * access to. Also checks that the "consumed" buffer count we are
1023          * looking for matches the one contained in the subbuffer id.
1024          */
1025         ret = update_read_sb_index(config, &buf->backend, &chan->backend,
1026                                    consumed_idx, buf_trunc_val(consumed, chan));
1027         if (ret)
1028                 goto retry;
1029         subbuffer_id_clear_noref(config, &buf->backend.buf_rsb.id);
1030
1031         buf->get_subbuf_consumed = consumed;
1032         buf->get_subbuf = 1;
1033
1034         return 0;
1035
1036 nodata:
1037         /*
1038          * The memory barriers __wait_event()/wake_up_interruptible() take care
1039          * of "raw_spin_is_locked" memory ordering.
1040          */
1041         if (finalized)
1042                 return -ENODATA;
1043         else if (raw_spin_is_locked(&buf->raw_tick_nohz_spinlock))
1044                 goto retry;
1045         else
1046                 return -EAGAIN;
1047 }
1048 EXPORT_SYMBOL_GPL(lib_ring_buffer_get_subbuf);
1049
1050 /**
1051  * lib_ring_buffer_put_subbuf - release exclusive subbuffer access
1052  * @buf: ring buffer
1053  */
1054 void lib_ring_buffer_put_subbuf(struct lib_ring_buffer *buf)
1055 {
1056         struct lib_ring_buffer_backend *bufb = &buf->backend;
1057         struct channel *chan = bufb->chan;
1058         const struct lib_ring_buffer_config *config = chan->backend.config;
1059         unsigned long read_sb_bindex, consumed_idx, consumed;
1060
1061         CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
1062
1063         if (!buf->get_subbuf) {
1064                 /*
1065                  * Reader puts a subbuffer it did not get.
1066                  */
1067                 CHAN_WARN_ON(chan, 1);
1068                 return;
1069         }
1070         consumed = buf->get_subbuf_consumed;
1071         buf->get_subbuf = 0;
1072
1073         /*
1074          * Clear the records_unread counter. (overruns counter)
1075          * Can still be non-zero if a file reader simply grabbed the data
1076          * without using iterators.
1077          * Can be below zero if an iterator is used on a snapshot more than
1078          * once.
1079          */
1080         read_sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
1081         v_add(config, v_read(config,
1082                              &bufb->array[read_sb_bindex]->records_unread),
1083               &bufb->records_read);
1084         v_set(config, &bufb->array[read_sb_bindex]->records_unread, 0);
1085         CHAN_WARN_ON(chan, config->mode == RING_BUFFER_OVERWRITE
1086                      && subbuffer_id_is_noref(config, bufb->buf_rsb.id));
1087         subbuffer_id_set_noref(config, &bufb->buf_rsb.id);
1088
1089         /*
1090          * Exchange the reader subbuffer with the one we put in its place in the
1091          * writer subbuffer table. Expect the original consumed count. If
1092          * update_read_sb_index fails, this is because the writer updated the
1093          * subbuffer concurrently. We should therefore keep the subbuffer we
1094          * currently have: it has become invalid to try reading this sub-buffer
1095          * consumed count value anyway.
1096          */
1097         consumed_idx = subbuf_index(consumed, chan);
1098         update_read_sb_index(config, &buf->backend, &chan->backend,
1099                              consumed_idx, buf_trunc_val(consumed, chan));
1100         /*
1101          * update_read_sb_index return value ignored. Don't exchange sub-buffer
1102          * if the writer concurrently updated it.
1103          */
1104 }
1105 EXPORT_SYMBOL_GPL(lib_ring_buffer_put_subbuf);
1106
1107 /*
1108  * cons_offset is an iterator on all subbuffer offsets between the reader
1109  * position and the writer position. (inclusive)
1110  */
1111 static
1112 void lib_ring_buffer_print_subbuffer_errors(struct lib_ring_buffer *buf,
1113                                             struct channel *chan,
1114                                             unsigned long cons_offset,
1115                                             int cpu)
1116 {
1117         const struct lib_ring_buffer_config *config = chan->backend.config;
1118         unsigned long cons_idx, commit_count, commit_count_sb;
1119
1120         cons_idx = subbuf_index(cons_offset, chan);
1121         commit_count = v_read(config, &buf->commit_hot[cons_idx].cc);
1122         commit_count_sb = v_read(config, &buf->commit_cold[cons_idx].cc_sb);
1123
1124         if (subbuf_offset(commit_count, chan) != 0)
1125                 printk(KERN_WARNING
1126                        "ring buffer %s, cpu %d: "
1127                        "commit count in subbuffer %lu,\n"
1128                        "expecting multiples of %lu bytes\n"
1129                        "  [ %lu bytes committed, %lu bytes reader-visible ]\n",
1130                        chan->backend.name, cpu, cons_idx,
1131                        chan->backend.subbuf_size,
1132                        commit_count, commit_count_sb);
1133
1134         printk(KERN_DEBUG "ring buffer: %s, cpu %d: %lu bytes committed\n",
1135                chan->backend.name, cpu, commit_count);
1136 }
1137
1138 static
1139 void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf,
1140                                          struct channel *chan,
1141                                          void *priv, int cpu)
1142 {
1143         const struct lib_ring_buffer_config *config = chan->backend.config;
1144         unsigned long write_offset, cons_offset;
1145
1146         /*
1147          * No need to order commit_count, write_offset and cons_offset reads
1148          * because we execute at teardown when no more writer nor reader
1149          * references are left.
1150          */
1151         write_offset = v_read(config, &buf->offset);
1152         cons_offset = atomic_long_read(&buf->consumed);
1153         if (write_offset != cons_offset)
1154                 printk(KERN_DEBUG
1155                        "ring buffer %s, cpu %d: "
1156                        "non-consumed data\n"
1157                        "  [ %lu bytes written, %lu bytes read ]\n",
1158                        chan->backend.name, cpu, write_offset, cons_offset);
1159
1160         for (cons_offset = atomic_long_read(&buf->consumed);
1161              (long) (subbuf_trunc((unsigned long) v_read(config, &buf->offset),
1162                                   chan)
1163                      - cons_offset) > 0;
1164              cons_offset = subbuf_align(cons_offset, chan))
1165                 lib_ring_buffer_print_subbuffer_errors(buf, chan, cons_offset,
1166                                                        cpu);
1167 }
1168
1169 static
1170 void lib_ring_buffer_print_errors(struct channel *chan,
1171                                   struct lib_ring_buffer *buf, int cpu)
1172 {
1173         const struct lib_ring_buffer_config *config = chan->backend.config;
1174         void *priv = chan->backend.priv;
1175
1176         printk(KERN_DEBUG "ring buffer %s, cpu %d: %lu records written, "
1177                           "%lu records overrun\n",
1178                           chan->backend.name, cpu,
1179                           v_read(config, &buf->records_count),
1180                           v_read(config, &buf->records_overrun));
1181
1182         if (v_read(config, &buf->records_lost_full)
1183             || v_read(config, &buf->records_lost_wrap)
1184             || v_read(config, &buf->records_lost_big))
1185                 printk(KERN_WARNING
1186                        "ring buffer %s, cpu %d: records were lost. Caused by:\n"
1187                        "  [ %lu buffer full, %lu nest buffer wrap-around, "
1188                        "%lu event too big ]\n",
1189                        chan->backend.name, cpu,
1190                        v_read(config, &buf->records_lost_full),
1191                        v_read(config, &buf->records_lost_wrap),
1192                        v_read(config, &buf->records_lost_big));
1193
1194         lib_ring_buffer_print_buffer_errors(buf, chan, priv, cpu);
1195 }
1196
1197 /*
1198  * lib_ring_buffer_switch_old_start: Populate old subbuffer header.
1199  *
1200  * Only executed when the buffer is finalized, in SWITCH_FLUSH.
1201  */
1202 static
1203 void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf,
1204                                       struct channel *chan,
1205                                       struct switch_offsets *offsets,
1206                                       u64 tsc)
1207 {
1208         const struct lib_ring_buffer_config *config = chan->backend.config;
1209         unsigned long oldidx = subbuf_index(offsets->old, chan);
1210         unsigned long commit_count;
1211
1212         config->cb.buffer_begin(buf, tsc, oldidx);
1213
1214         /*
1215          * Order all writes to buffer before the commit count update that will
1216          * determine that the subbuffer is full.
1217          */
1218         if (config->ipi == RING_BUFFER_IPI_BARRIER) {
1219                 /*
1220                  * Must write slot data before incrementing commit count.  This
1221                  * compiler barrier is upgraded into a smp_mb() by the IPI sent
1222                  * by get_subbuf().
1223                  */
1224                 barrier();
1225         } else
1226                 smp_wmb();
1227         v_add(config, config->cb.subbuffer_header_size(),
1228               &buf->commit_hot[oldidx].cc);
1229         commit_count = v_read(config, &buf->commit_hot[oldidx].cc);
1230         /* Check if the written buffer has to be delivered */
1231         lib_ring_buffer_check_deliver(config, buf, chan, offsets->old,
1232                                       commit_count, oldidx);
1233         lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
1234                                              offsets->old, commit_count,
1235                                              config->cb.subbuffer_header_size());
1236 }
1237
1238 /*
1239  * lib_ring_buffer_switch_old_end: switch old subbuffer
1240  *
1241  * Note : offset_old should never be 0 here. It is ok, because we never perform
1242  * buffer switch on an empty subbuffer in SWITCH_ACTIVE mode. The caller
1243  * increments the offset_old value when doing a SWITCH_FLUSH on an empty
1244  * subbuffer.
1245  */
1246 static
1247 void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf,
1248                                     struct channel *chan,
1249                                     struct switch_offsets *offsets,
1250                                     u64 tsc)
1251 {
1252         const struct lib_ring_buffer_config *config = chan->backend.config;
1253         unsigned long oldidx = subbuf_index(offsets->old - 1, chan);
1254         unsigned long commit_count, padding_size, data_size;
1255
1256         data_size = subbuf_offset(offsets->old - 1, chan) + 1;
1257         padding_size = chan->backend.subbuf_size - data_size;
1258         subbuffer_set_data_size(config, &buf->backend, oldidx, data_size);
1259
1260         /*
1261          * Order all writes to buffer before the commit count update that will
1262          * determine that the subbuffer is full.
1263          */
1264         if (config->ipi == RING_BUFFER_IPI_BARRIER) {
1265                 /*
1266                  * Must write slot data before incrementing commit count.  This
1267                  * compiler barrier is upgraded into a smp_mb() by the IPI sent
1268                  * by get_subbuf().
1269                  */
1270                 barrier();
1271         } else
1272                 smp_wmb();
1273         v_add(config, padding_size, &buf->commit_hot[oldidx].cc);
1274         commit_count = v_read(config, &buf->commit_hot[oldidx].cc);
1275         lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
1276                                       commit_count, oldidx);
1277         lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
1278                                              offsets->old, commit_count,
1279                                              padding_size);
1280 }
1281
1282 /*
1283  * lib_ring_buffer_switch_new_start: Populate new subbuffer.
1284  *
1285  * This code can be executed unordered : writers may already have written to the
1286  * sub-buffer before this code gets executed, caution.  The commit makes sure
1287  * that this code is executed before the deliver of this sub-buffer.
1288  */
1289 static
1290 void lib_ring_buffer_switch_new_start(struct lib_ring_buffer *buf,
1291                                       struct channel *chan,
1292                                       struct switch_offsets *offsets,
1293                                       u64 tsc)
1294 {
1295         const struct lib_ring_buffer_config *config = chan->backend.config;
1296         unsigned long beginidx = subbuf_index(offsets->begin, chan);
1297         unsigned long commit_count;
1298
1299         config->cb.buffer_begin(buf, tsc, beginidx);
1300
1301         /*
1302          * Order all writes to buffer before the commit count update that will
1303          * determine that the subbuffer is full.
1304          */
1305         if (config->ipi == RING_BUFFER_IPI_BARRIER) {
1306                 /*
1307                  * Must write slot data before incrementing commit count.  This
1308                  * compiler barrier is upgraded into a smp_mb() by the IPI sent
1309                  * by get_subbuf().
1310                  */
1311                 barrier();
1312         } else
1313                 smp_wmb();
1314         v_add(config, config->cb.subbuffer_header_size(),
1315               &buf->commit_hot[beginidx].cc);
1316         commit_count = v_read(config, &buf->commit_hot[beginidx].cc);
1317         /* Check if the written buffer has to be delivered */
1318         lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin,
1319                                       commit_count, beginidx);
1320         lib_ring_buffer_write_commit_counter(config, buf, chan, beginidx,
1321                                              offsets->begin, commit_count,
1322                                              config->cb.subbuffer_header_size());
1323 }
1324
1325 /*
1326  * lib_ring_buffer_switch_new_end: finish switching current subbuffer
1327  *
1328  * The only remaining threads could be the ones with pending commits. They will
1329  * have to do the deliver themselves.
1330  */
1331 static
1332 void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf,
1333                                             struct channel *chan,
1334                                             struct switch_offsets *offsets,
1335                                             u64 tsc)
1336 {
1337         const struct lib_ring_buffer_config *config = chan->backend.config;
1338         unsigned long endidx = subbuf_index(offsets->end - 1, chan);
1339         unsigned long commit_count, padding_size, data_size;
1340
1341         data_size = subbuf_offset(offsets->end - 1, chan) + 1;
1342         padding_size = chan->backend.subbuf_size - data_size;
1343         subbuffer_set_data_size(config, &buf->backend, endidx, data_size);
1344
1345         /*
1346          * Order all writes to buffer before the commit count update that will
1347          * determine that the subbuffer is full.
1348          */
1349         if (config->ipi == RING_BUFFER_IPI_BARRIER) {
1350                 /*
1351                  * Must write slot data before incrementing commit count.  This
1352                  * compiler barrier is upgraded into a smp_mb() by the IPI sent
1353                  * by get_subbuf().
1354                  */
1355                 barrier();
1356         } else
1357                 smp_wmb();
1358         v_add(config, padding_size, &buf->commit_hot[endidx].cc);
1359         commit_count = v_read(config, &buf->commit_hot[endidx].cc);
1360         lib_ring_buffer_check_deliver(config, buf, chan, offsets->end - 1,
1361                                   commit_count, endidx);
1362         lib_ring_buffer_write_commit_counter(config, buf, chan, endidx,
1363                                              offsets->end, commit_count,
1364                                              padding_size);
1365 }
1366
1367 /*
1368  * Returns :
1369  * 0 if ok
1370  * !0 if execution must be aborted.
1371  */
1372 static
1373 int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
1374                                     struct lib_ring_buffer *buf,
1375                                     struct channel *chan,
1376                                     struct switch_offsets *offsets,
1377                                     u64 *tsc)
1378 {
1379         const struct lib_ring_buffer_config *config = chan->backend.config;
1380         unsigned long off;
1381
1382         offsets->begin = v_read(config, &buf->offset);
1383         offsets->old = offsets->begin;
1384         offsets->switch_old_start = 0;
1385         off = subbuf_offset(offsets->begin, chan);
1386
1387         *tsc = config->cb.ring_buffer_clock_read(chan);
1388
1389         /*
1390          * Ensure we flush the header of an empty subbuffer when doing the
1391          * finalize (SWITCH_FLUSH). This ensures that we end up knowing the
1392          * total data gathering duration even if there were no records saved
1393          * after the last buffer switch.
1394          * In SWITCH_ACTIVE mode, switch the buffer when it contains events.
1395          * SWITCH_ACTIVE only flushes the current subbuffer, dealing with end of
1396          * subbuffer header as appropriate.
1397          * The next record that reserves space will be responsible for
1398          * populating the following subbuffer header. We choose not to populate
1399          * the next subbuffer header here because we want to be able to use
1400          * SWITCH_ACTIVE for periodical buffer flush and CPU tick_nohz stop
1401          * buffer flush, which must guarantee that all the buffer content
1402          * (records and header timestamps) are visible to the reader. This is
1403          * required for quiescence guarantees for the fusion merge.
1404          */
1405         if (mode == SWITCH_FLUSH || off > 0) {
1406                 if (unlikely(off == 0)) {
1407                         /*
1408                          * The client does not save any header information.
1409                          * Don't switch empty subbuffer on finalize, because it
1410                          * is invalid to deliver a completely empty subbuffer.
1411                          */
1412                         if (!config->cb.subbuffer_header_size())
1413                                 return -1;
1414                         /*
1415                          * Need to write the subbuffer start header on finalize.
1416                          */
1417                         offsets->switch_old_start = 1;
1418                 }
1419                 offsets->begin = subbuf_align(offsets->begin, chan);
1420         } else
1421                 return -1;      /* we do not have to switch : buffer is empty */
1422         /* Note: old points to the next subbuf at offset 0 */
1423         offsets->end = offsets->begin;
1424         return 0;
1425 }
1426
1427 /*
1428  * Force a sub-buffer switch. This operation is completely reentrant : can be
1429  * called while tracing is active with absolutely no lock held.
1430  *
1431  * Note, however, that as a v_cmpxchg is used for some atomic
1432  * operations, this function must be called from the CPU which owns the buffer
1433  * for a ACTIVE flush.
1434  */
1435 void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode mode)
1436 {
1437         struct channel *chan = buf->backend.chan;
1438         const struct lib_ring_buffer_config *config = chan->backend.config;
1439         struct switch_offsets offsets;
1440         unsigned long oldidx;
1441         u64 tsc;
1442
1443         offsets.size = 0;
1444
1445         /*
1446          * Perform retryable operations.
1447          */
1448         do {
1449                 if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets,
1450                                                     &tsc))
1451                         return; /* Switch not needed */
1452         } while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end)
1453                  != offsets.old);
1454
1455         /*
1456          * Atomically update last_tsc. This update races against concurrent
1457          * atomic updates, but the race will always cause supplementary full TSC
1458          * records, never the opposite (missing a full TSC record when it would
1459          * be needed).
1460          */
1461         save_last_tsc(config, buf, tsc);
1462
1463         /*
1464          * Push the reader if necessary
1465          */
1466         lib_ring_buffer_reserve_push_reader(buf, chan, offsets.old);
1467
1468         oldidx = subbuf_index(offsets.old, chan);
1469         lib_ring_buffer_clear_noref(config, &buf->backend, oldidx);
1470
1471         /*
1472          * May need to populate header start on SWITCH_FLUSH.
1473          */
1474         if (offsets.switch_old_start) {
1475                 lib_ring_buffer_switch_old_start(buf, chan, &offsets, tsc);
1476                 offsets.old += config->cb.subbuffer_header_size();
1477         }
1478
1479         /*
1480          * Switch old subbuffer.
1481          */
1482         lib_ring_buffer_switch_old_end(buf, chan, &offsets, tsc);
1483 }
1484 EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_slow);
1485
1486 /*
1487  * Returns :
1488  * 0 if ok
1489  * -ENOSPC if event size is too large for packet.
1490  * -ENOBUFS if there is currently not enough space in buffer for the event.
1491  * -EIO if data cannot be written into the buffer for any other reason.
1492  */
1493 static
1494 int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
1495                                      struct channel *chan,
1496                                      struct switch_offsets *offsets,
1497                                      struct lib_ring_buffer_ctx *ctx)
1498 {
1499         const struct lib_ring_buffer_config *config = chan->backend.config;
1500         unsigned long reserve_commit_diff;
1501
1502         offsets->begin = v_read(config, &buf->offset);
1503         offsets->old = offsets->begin;
1504         offsets->switch_new_start = 0;
1505         offsets->switch_new_end = 0;
1506         offsets->switch_old_end = 0;
1507         offsets->pre_header_padding = 0;
1508
1509         ctx->tsc = config->cb.ring_buffer_clock_read(chan);
1510         if ((int64_t) ctx->tsc == -EIO)
1511                 return -EIO;
1512
1513         if (last_tsc_overflow(config, buf, ctx->tsc))
1514                 ctx->rflags |= RING_BUFFER_RFLAG_FULL_TSC;
1515
1516         if (unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) {
1517                 offsets->switch_new_start = 1;          /* For offsets->begin */
1518         } else {
1519                 offsets->size = config->cb.record_header_size(config, chan,
1520                                                 offsets->begin,
1521                                                 &offsets->pre_header_padding,
1522                                                 ctx);
1523                 offsets->size +=
1524                         lib_ring_buffer_align(offsets->begin + offsets->size,
1525                                               ctx->largest_align)
1526                         + ctx->data_size;
1527                 if (unlikely(subbuf_offset(offsets->begin, chan) +
1528                              offsets->size > chan->backend.subbuf_size)) {
1529                         offsets->switch_old_end = 1;    /* For offsets->old */
1530                         offsets->switch_new_start = 1;  /* For offsets->begin */
1531                 }
1532         }
1533         if (unlikely(offsets->switch_new_start)) {
1534                 unsigned long sb_index;
1535
1536                 /*
1537                  * We are typically not filling the previous buffer completely.
1538                  */
1539                 if (likely(offsets->switch_old_end))
1540                         offsets->begin = subbuf_align(offsets->begin, chan);
1541                 offsets->begin = offsets->begin
1542                                  + config->cb.subbuffer_header_size();
1543                 /* Test new buffer integrity */
1544                 sb_index = subbuf_index(offsets->begin, chan);
1545                 reserve_commit_diff =
1546                   (buf_trunc(offsets->begin, chan)
1547                    >> chan->backend.num_subbuf_order)
1548                   - ((unsigned long) v_read(config,
1549                                             &buf->commit_cold[sb_index].cc_sb)
1550                      & chan->commit_count_mask);
1551                 if (likely(reserve_commit_diff == 0)) {
1552                         /* Next subbuffer not being written to. */
1553                         if (unlikely(config->mode != RING_BUFFER_OVERWRITE &&
1554                                 subbuf_trunc(offsets->begin, chan)
1555                                  - subbuf_trunc((unsigned long)
1556                                      atomic_long_read(&buf->consumed), chan)
1557                                 >= chan->backend.buf_size)) {
1558                                 /*
1559                                  * We do not overwrite non consumed buffers
1560                                  * and we are full : record is lost.
1561                                  */
1562                                 v_inc(config, &buf->records_lost_full);
1563                                 return -ENOBUFS;
1564                         } else {
1565                                 /*
1566                                  * Next subbuffer not being written to, and we
1567                                  * are either in overwrite mode or the buffer is
1568                                  * not full. It's safe to write in this new
1569                                  * subbuffer.
1570                                  */
1571                         }
1572                 } else {
1573                         /*
1574                          * Next subbuffer reserve offset does not match the
1575                          * commit offset. Drop record in producer-consumer and
1576                          * overwrite mode. Caused by either a writer OOPS or too
1577                          * many nested writes over a reserve/commit pair.
1578                          */
1579                         v_inc(config, &buf->records_lost_wrap);
1580                         return -EIO;
1581                 }
1582                 offsets->size =
1583                         config->cb.record_header_size(config, chan,
1584                                                 offsets->begin,
1585                                                 &offsets->pre_header_padding,
1586                                                 ctx);
1587                 offsets->size +=
1588                         lib_ring_buffer_align(offsets->begin + offsets->size,
1589                                               ctx->largest_align)
1590                         + ctx->data_size;
1591                 if (unlikely(subbuf_offset(offsets->begin, chan)
1592                              + offsets->size > chan->backend.subbuf_size)) {
1593                         /*
1594                          * Record too big for subbuffers, report error, don't
1595                          * complete the sub-buffer switch.
1596                          */
1597                         v_inc(config, &buf->records_lost_big);
1598                         return -ENOSPC;
1599                 } else {
1600                         /*
1601                          * We just made a successful buffer switch and the
1602                          * record fits in the new subbuffer. Let's write.
1603                          */
1604                 }
1605         } else {
1606                 /*
1607                  * Record fits in the current buffer and we are not on a switch
1608                  * boundary. It's safe to write.
1609                  */
1610         }
1611         offsets->end = offsets->begin + offsets->size;
1612
1613         if (unlikely(subbuf_offset(offsets->end, chan) == 0)) {
1614                 /*
1615                  * The offset_end will fall at the very beginning of the next
1616                  * subbuffer.
1617                  */
1618                 offsets->switch_new_end = 1;    /* For offsets->begin */
1619         }
1620         return 0;
1621 }
1622
1623 /**
1624  * lib_ring_buffer_reserve_slow - Atomic slot reservation in a buffer.
1625  * @ctx: ring buffer context.
1626  *
1627  * Return : -NOBUFS if not enough space, -ENOSPC if event size too large,
1628  * -EIO for other errors, else returns 0.
1629  * It will take care of sub-buffer switching.
1630  */
1631 int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx)
1632 {
1633         struct channel *chan = ctx->chan;
1634         const struct lib_ring_buffer_config *config = chan->backend.config;
1635         struct lib_ring_buffer *buf;
1636         struct switch_offsets offsets;
1637         int ret;
1638
1639         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
1640                 buf = per_cpu_ptr(chan->backend.buf, ctx->cpu);
1641         else
1642                 buf = chan->backend.buf;
1643         ctx->buf = buf;
1644
1645         offsets.size = 0;
1646
1647         do {
1648                 ret = lib_ring_buffer_try_reserve_slow(buf, chan, &offsets,
1649                                                        ctx);
1650                 if (unlikely(ret))
1651                         return ret;
1652         } while (unlikely(v_cmpxchg(config, &buf->offset, offsets.old,
1653                                     offsets.end)
1654                           != offsets.old));
1655
1656         /*
1657          * Atomically update last_tsc. This update races against concurrent
1658          * atomic updates, but the race will always cause supplementary full TSC
1659          * records, never the opposite (missing a full TSC record when it would
1660          * be needed).
1661          */
1662         save_last_tsc(config, buf, ctx->tsc);
1663
1664         /*
1665          * Push the reader if necessary
1666          */
1667         lib_ring_buffer_reserve_push_reader(buf, chan, offsets.end - 1);
1668
1669         /*
1670          * Clear noref flag for this subbuffer.
1671          */
1672         lib_ring_buffer_clear_noref(config, &buf->backend,
1673                                     subbuf_index(offsets.end - 1, chan));
1674
1675         /*
1676          * Switch old subbuffer if needed.
1677          */
1678         if (unlikely(offsets.switch_old_end)) {
1679                 lib_ring_buffer_clear_noref(config, &buf->backend,
1680                                             subbuf_index(offsets.old - 1, chan));
1681                 lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx->tsc);
1682         }
1683
1684         /*
1685          * Populate new subbuffer.
1686          */
1687         if (unlikely(offsets.switch_new_start))
1688                 lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx->tsc);
1689
1690         if (unlikely(offsets.switch_new_end))
1691                 lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx->tsc);
1692
1693         ctx->slot_size = offsets.size;
1694         ctx->pre_offset = offsets.begin;
1695         ctx->buf_offset = offsets.begin + offsets.pre_header_padding;
1696         return 0;
1697 }
1698 EXPORT_SYMBOL_GPL(lib_ring_buffer_reserve_slow);
1699
1700 int __init init_lib_ring_buffer_frontend(void)
1701 {
1702         int cpu;
1703
1704         for_each_possible_cpu(cpu)
1705                 spin_lock_init(&per_cpu(ring_buffer_nohz_lock, cpu));
1706         return 0;
1707 }
1708
1709 module_init(init_lib_ring_buffer_frontend);
1710
1711 void __exit exit_lib_ring_buffer_frontend(void)
1712 {
1713 }
1714
1715 module_exit(exit_lib_ring_buffer_frontend);