Fix: add missing poll.h include
[lttng-ust.git] / libringbuffer / ring_buffer_frontend.c
index 8c8126025878ed72929bf6ecd8ebedf9b2806148..aeb7f7d94a10a9c4b63d098df1191ba4cdcef47e 100644 (file)
@@ -62,6 +62,7 @@
 #include <urcu/compiler.h>
 #include <urcu/ref.h>
 #include <urcu/tls-compat.h>
+#include <poll.h>
 #include <helper.h>
 
 #include "smp.h"
@@ -84,6 +85,8 @@
 #define LTTNG_UST_RB_SIG_READ          SIGRTMIN + 1
 #define LTTNG_UST_RB_SIG_TEARDOWN      SIGRTMIN + 2
 #define CLOCKID                CLOCK_MONOTONIC
+#define LTTNG_UST_RING_BUFFER_GET_RETRY                10
+#define LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS   10
 
 /*
  * Use POSIX SHM: shm_open(3) and shm_unlink(3).
@@ -1067,8 +1070,7 @@ int lib_ring_buffer_get_subbuf(struct lttng_ust_lib_ring_buffer *buf,
        struct channel *chan = shmp(handle, buf->backend.chan);
        const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
        unsigned long consumed_cur, consumed_idx, commit_count, write_offset;
-       int ret;
-       int finalized;
+       int ret, finalized, nr_retry = LTTNG_UST_RING_BUFFER_GET_RETRY;
 
 retry:
        finalized = CMM_ACCESS_ONCE(buf->finalized);
@@ -1103,20 +1105,72 @@ retry:
 
        /*
         * Check that the subbuffer we are trying to consume has been
-        * already fully committed.
+        * already fully committed. There are a few causes that can make
+        * this unavailability situation occur:
+        *
+        * Temporary (short-term) situation:
+        * - Application is running on a different CPU, between reserve
+        *   and commit ring buffer operations,
+        * - Application is preempted between reserve and commit ring
+        *   buffer operations,
+        *
+        * Long-term situation:
+        * - Application is stopped (SIGSTOP) between reserve and commit
+        *   ring buffer operations. Could eventually be resumed by
+        *   SIGCONT.
+        * - Application is killed (SIGTERM, SIGINT, SIGKILL) between
+        *   reserve and commit ring buffer operation.
+        *
+        * From a consumer perspective, handling short-term
+        * unavailability situations is performed by retrying a few
+        * times after a delay. Handling long-term unavailability
+        * situations is handled by failing to get the sub-buffer.
+        *
+        * In all of those situations, if the application is taking a
+        * long time to perform its commit after ring buffer space
+        * reservation, we can end up in a situation where the producer
+        * will fill the ring buffer and try to write into the same
+        * sub-buffer again (which has a missing commit). This is
+        * handled by the producer in the sub-buffer switch handling
+        * code of the reserve routine by detecting unbalanced
+        * reserve/commit counters and discarding all further events
+        * until the situation is resolved in those situations. Two
+        * scenarios can occur:
+        *
+        * 1) The application causing the reserve/commit counters to be
+        *    unbalanced has been terminated. In this situation, all
+        *    further events will be discarded in the buffers, and no
+        *    further buffer data will be readable by the consumer
+        *    daemon. Tearing down the UST tracing session and starting
+        *    anew is a work-around for those situations. Note that this
+        *    only affects per-UID tracing. In per-PID tracing, the
+        *    application vanishes with the termination, and therefore
+        *    no more data needs to be written to the buffers.
+        * 2) The application causing the unbalance has been delayed for
+        *    a long time, but will eventually try to increment the
+        *    commit counter after eventually writing to the sub-buffer.
+        *    This situation can cause events to be discarded until the
+        *    application resumes its operations.
         */
        if (((commit_count - chan->backend.subbuf_size)
             & chan->commit_count_mask)
-           - (buf_trunc(consumed_cur, chan)
+           - (buf_trunc(consumed, chan)
               >> chan->backend.num_subbuf_order)
-           != 0)
-               goto nodata;
+           != 0) {
+               if (nr_retry-- > 0) {
+                       if (nr_retry <= (LTTNG_UST_RING_BUFFER_GET_RETRY >> 1))
+                               (void) poll(NULL, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS);
+                       goto retry;
+               } else {
+                       goto nodata;
+               }
+       }
 
        /*
         * Check that we are not about to read the same subbuffer in
         * which the writer head is.
         */
-       if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
+       if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed, chan)
            == 0)
                goto nodata;
 
@@ -1126,12 +1180,23 @@ retry:
         * the writer is getting access to a subbuffer we were trying to get
         * access to. Also checks that the "consumed" buffer count we are
         * looking for matches the one contained in the subbuffer id.
+        *
+        * The short-lived race window described here can be affected by
+        * application signals and preemption, thus requiring to bound
+        * the loop to a maximum number of retry.
         */
        ret = update_read_sb_index(config, &buf->backend, &chan->backend,
                                   consumed_idx, buf_trunc_val(consumed, chan),
                                   handle);
-       if (ret)
-               goto retry;
+       if (ret) {
+               if (nr_retry-- > 0) {
+                       if (nr_retry <= (LTTNG_UST_RING_BUFFER_GET_RETRY >> 1))
+                               (void) poll(NULL, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS);
+                       goto retry;
+               } else {
+                       goto nodata;
+               }
+       }
        subbuffer_id_clear_noref(config, &buf->backend.buf_rsb.id);
 
        buf->get_subbuf_consumed = consumed;
@@ -1333,11 +1398,10 @@ void lib_ring_buffer_switch_old_start(struct lttng_ust_lib_ring_buffer *buf,
        commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
        /* Check if the written buffer has to be delivered */
        lib_ring_buffer_check_deliver(config, buf, chan, offsets->old,
-                                     commit_count, oldidx, handle);
+                                     commit_count, oldidx, handle, tsc);
        lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
-                                            offsets->old, commit_count,
-                                            config->cb.subbuffer_header_size(),
-                                            handle);
+                       offsets->old + config->cb.subbuffer_header_size(),
+                       commit_count, handle);
 }
 
 /*
@@ -1372,10 +1436,9 @@ void lib_ring_buffer_switch_old_end(struct lttng_ust_lib_ring_buffer *buf,
        v_add(config, padding_size, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
        commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
        lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
-                                     commit_count, oldidx, handle);
+                                     commit_count, oldidx, handle, tsc);
        lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
-                                            offsets->old, commit_count,
-                                            padding_size, handle);
+                       offsets->old + padding_size, commit_count, handle);
 }
 
 /*
@@ -1408,11 +1471,10 @@ void lib_ring_buffer_switch_new_start(struct lttng_ust_lib_ring_buffer *buf,
        commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, beginidx)->cc);
        /* Check if the written buffer has to be delivered */
        lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin,
-                                     commit_count, beginidx, handle);
+                                     commit_count, beginidx, handle, tsc);
        lib_ring_buffer_write_commit_counter(config, buf, chan, beginidx,
-                                            offsets->begin, commit_count,
-                                            config->cb.subbuffer_header_size(),
-                                            handle);
+                       offsets->begin + config->cb.subbuffer_header_size(),
+                       commit_count, handle);
 }
 
 /*
@@ -1449,10 +1511,11 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
                                    struct lttng_ust_lib_ring_buffer *buf,
                                    struct channel *chan,
                                    struct switch_offsets *offsets,
-                                   uint64_t *tsc)
+                                   uint64_t *tsc,
+                                   struct lttng_ust_shm_handle *handle)
 {
        const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
-       unsigned long off;
+       unsigned long off, reserve_commit_diff;
 
        offsets->begin = v_read(config, &buf->offset);
        offsets->old = offsets->begin;
@@ -1477,36 +1540,69 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
         * timestamps) are visible to the reader. This is required for
         * quiescence guarantees for the fusion merge.
         */
-       if (mode == SWITCH_FLUSH || off > 0) {
-               if (caa_unlikely(off == 0)) {
-                       /*
-                        * A final flush that encounters an empty
-                        * sub-buffer cannot switch buffer if a
-                        * reader is located within this sub-buffer.
-                        * Anyway, the purpose of final flushing of a
-                        * sub-buffer at offset 0 is to handle the case
-                        * of entirely empty stream.
-                        */
-                       if (caa_unlikely(subbuf_trunc(offsets->begin, chan)
-                                        - subbuf_trunc((unsigned long)
-                                            uatomic_read(&buf->consumed), chan)
-                                       >= chan->backend.buf_size))
-                               return -1;
-                       /*
-                        * The client does not save any header information.
-                        * Don't switch empty subbuffer on finalize, because it
-                        * is invalid to deliver a completely empty subbuffer.
-                        */
-                       if (!config->cb.subbuffer_header_size())
+       if (mode != SWITCH_FLUSH && !off)
+               return -1;      /* we do not have to switch : buffer is empty */
+
+       if (caa_unlikely(off == 0)) {
+               unsigned long sb_index, commit_count;
+
+               /*
+                * We are performing a SWITCH_FLUSH. At this stage, there are no
+                * concurrent writes into the buffer.
+                *
+                * The client does not save any header information.  Don't
+                * switch empty subbuffer on finalize, because it is invalid to
+                * deliver a completely empty subbuffer.
+                */
+               if (!config->cb.subbuffer_header_size())
+                       return -1;
+
+               /* Test new buffer integrity */
+               sb_index = subbuf_index(offsets->begin, chan);
+               commit_count = v_read(config,
+                               &shmp_index(handle, buf->commit_cold,
+                                       sb_index)->cc_sb);
+               reserve_commit_diff =
+                 (buf_trunc(offsets->begin, chan)
+                  >> chan->backend.num_subbuf_order)
+                 - (commit_count & chan->commit_count_mask);
+               if (caa_likely(reserve_commit_diff == 0)) {
+                       /* Next subbuffer not being written to. */
+                       if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
+                               subbuf_trunc(offsets->begin, chan)
+                                - subbuf_trunc((unsigned long)
+                                    uatomic_read(&buf->consumed), chan)
+                               >= chan->backend.buf_size)) {
+                               /*
+                                * We do not overwrite non consumed buffers
+                                * and we are full : don't switch.
+                                */
                                return -1;
+                       } else {
+                               /*
+                                * Next subbuffer not being written to, and we
+                                * are either in overwrite mode or the buffer is
+                                * not full. It's safe to write in this new
+                                * subbuffer.
+                                */
+                       }
+               } else {
                        /*
-                        * Need to write the subbuffer start header on finalize.
+                        * Next subbuffer reserve offset does not match the
+                        * commit offset. Don't perform switch in
+                        * producer-consumer and overwrite mode.  Caused by
+                        * either a writer OOPS or too many nested writes over a
+                        * reserve/commit pair.
                         */
-                       offsets->switch_old_start = 1;
+                       return -1;
                }
-               offsets->begin = subbuf_align(offsets->begin, chan);
-       } else
-               return -1;      /* we do not have to switch : buffer is empty */
+
+               /*
+                * Need to write the subbuffer start header on finalize.
+                */
+               offsets->switch_old_start = 1;
+       }
+       offsets->begin = subbuf_align(offsets->begin, chan);
        /* Note: old points to the next subbuf at offset 0 */
        offsets->end = offsets->begin;
        return 0;
@@ -1536,7 +1632,7 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf, enum swi
         */
        do {
                if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets,
-                                                   &tsc))
+                                                   &tsc, handle))
                        return; /* Switch not needed */
        } while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end)
                 != offsets.old);
@@ -1586,9 +1682,10 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
 {
        const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
        struct lttng_ust_shm_handle *handle = ctx->handle;
-       unsigned long reserve_commit_diff;
+       unsigned long reserve_commit_diff, offset_cmp;
 
-       offsets->begin = v_read(config, &buf->offset);
+retry:
+       offsets->begin = offset_cmp = v_read(config, &buf->offset);
        offsets->old = offsets->begin;
        offsets->switch_new_start = 0;
        offsets->switch_new_end = 0;
@@ -1620,7 +1717,7 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
                }
        }
        if (caa_unlikely(offsets->switch_new_start)) {
-               unsigned long sb_index;
+               unsigned long sb_index, commit_count;
 
                /*
                 * We are typically not filling the previous buffer completely.
@@ -1631,12 +1728,32 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
                                 + config->cb.subbuffer_header_size();
                /* Test new buffer integrity */
                sb_index = subbuf_index(offsets->begin, chan);
+               /*
+                * Read buf->offset before buf->commit_cold[sb_index].cc_sb.
+                * lib_ring_buffer_check_deliver() has the matching
+                * memory barriers required around commit_cold cc_sb
+                * updates to ensure reserve and commit counter updates
+                * are not seen reordered when updated by another CPU.
+                */
+               cmm_smp_rmb();
+               commit_count = v_read(config,
+                               &shmp_index(handle, buf->commit_cold,
+                                       sb_index)->cc_sb);
+               /* Read buf->commit_cold[sb_index].cc_sb before buf->offset. */
+               cmm_smp_rmb();
+               if (caa_unlikely(offset_cmp != v_read(config, &buf->offset))) {
+                       /*
+                        * The reserve counter have been concurrently updated
+                        * while we read the commit counter. This means the
+                        * commit counter we read might not match buf->offset
+                        * due to concurrent update. We therefore need to retry.
+                        */
+                       goto retry;
+               }
                reserve_commit_diff =
                  (buf_trunc(offsets->begin, chan)
                   >> chan->backend.num_subbuf_order)
-                 - ((unsigned long) v_read(config,
-                                           &shmp_index(handle, buf->commit_cold, sb_index)->cc_sb)
-                    & chan->commit_count_mask);
+                 - (commit_count & chan->commit_count_mask);
                if (caa_likely(reserve_commit_diff == 0)) {
                        /* Next subbuffer not being written to. */
                        if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
@@ -1671,7 +1788,8 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
 
                        /*
                         * Next subbuffer reserve offset does not match the
-                        * commit offset. Drop record in producer-consumer and
+                        * commit offset, and this did not involve update to the
+                        * reserve counter. Drop record in producer-consumer and
                         * overwrite mode. Caused by either a writer OOPS or too
                         * many nested writes over a reserve/commit pair.
                         */
This page took 0.027161 seconds and 4 git commands to generate.