X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=lib%2Fringbuffer%2Fring_buffer_frontend.c;h=b25ce7df6058d9dea6898062f4e6acbf6601c7b8;hb=9a9c9b6e05e8477c3edcf41a7f2697d1e588ce4d;hp=2626afa31e3ea9f33a7e9ebbab98bd91d510b723;hpb=74b9312477538922f7ce3194b37ed3e0c5d7ba16;p=lttng-modules.git

diff --git a/lib/ringbuffer/ring_buffer_frontend.c b/lib/ringbuffer/ring_buffer_frontend.c
index 2626afa3..b25ce7df 100644
--- a/lib/ringbuffer/ring_buffer_frontend.c
+++ b/lib/ringbuffer/ring_buffer_frontend.c
@@ -60,6 +60,8 @@
 #include "../../wrapper/ringbuffer/frontend.h"
 #include "../../wrapper/ringbuffer/iterator.h"
 #include "../../wrapper/ringbuffer/nohz.h"
+#include "../../wrapper/atomic.h"
+#include "../../wrapper/percpu-defs.h"
 
 /*
  * Internal structure representing offsets to use at a sub-buffer switch.
@@ -89,6 +91,9 @@ EXPORT_PER_CPU_SYMBOL(lib_ring_buffer_nesting);
 static
 void lib_ring_buffer_print_errors(struct channel *chan,
 				  struct lib_ring_buffer *buf, int cpu);
+static
+void _lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf,
+		enum switch_mode mode);
 
 /*
  * Must be called under cpu hotplug protection.
@@ -402,7 +407,7 @@ static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf)
  *	Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
  */
 static
-int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
+int lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
 					      unsigned long action,
 					      void *hcpu)
 {
@@ -497,16 +502,16 @@ static int notrace ring_buffer_tick_nohz_callback(struct notifier_block *nb,
 		raw_spin_unlock(&buf->raw_tick_nohz_spinlock);
 		break;
 	case TICK_NOHZ_STOP:
-		spin_lock(&__get_cpu_var(ring_buffer_nohz_lock));
+		spin_lock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock));
 		lib_ring_buffer_stop_switch_timer(buf);
 		lib_ring_buffer_stop_read_timer(buf);
-		spin_unlock(&__get_cpu_var(ring_buffer_nohz_lock));
+		spin_unlock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock));
 		break;
 	case TICK_NOHZ_RESTART:
-		spin_lock(&__get_cpu_var(ring_buffer_nohz_lock));
+		spin_lock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock));
 		lib_ring_buffer_start_read_timer(buf);
 		lib_ring_buffer_start_switch_timer(buf);
-		spin_unlock(&__get_cpu_var(ring_buffer_nohz_lock));
+		spin_unlock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock));
 		break;
 	}
 
@@ -583,8 +588,68 @@ static void channel_unregister_notifiers(struct channel *chan)
 	channel_backend_unregister_notifiers(&chan->backend);
 }
 
+static void lib_ring_buffer_set_quiescent(struct lib_ring_buffer *buf)
+{
+	if (!buf->quiescent) {
+		buf->quiescent = true;
+		_lib_ring_buffer_switch_remote(buf, SWITCH_FLUSH);
+	}
+}
+
+static void lib_ring_buffer_clear_quiescent(struct lib_ring_buffer *buf)
+{
+	buf->quiescent = false;
+}
+
+void lib_ring_buffer_set_quiescent_channel(struct channel *chan)
+{
+	int cpu;
+	const struct lib_ring_buffer_config *config = &chan->backend.config;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		get_online_cpus();
+		for_each_channel_cpu(cpu, chan) {
+			struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							      cpu);
+
+			lib_ring_buffer_set_quiescent(buf);
+		}
+		put_online_cpus();
+	} else {
+		struct lib_ring_buffer *buf = chan->backend.buf;
+
+		lib_ring_buffer_set_quiescent(buf);
+	}
+}
+EXPORT_SYMBOL_GPL(lib_ring_buffer_set_quiescent_channel);
+
+void lib_ring_buffer_clear_quiescent_channel(struct channel *chan)
+{
+	int cpu;
+	const struct lib_ring_buffer_config *config = &chan->backend.config;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		get_online_cpus();
+		for_each_channel_cpu(cpu, chan) {
+			struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							      cpu);
+
+			lib_ring_buffer_clear_quiescent(buf);
+		}
+		put_online_cpus();
+	} else {
+		struct lib_ring_buffer *buf = chan->backend.buf;
+
+		lib_ring_buffer_clear_quiescent(buf);
+	}
+}
+EXPORT_SYMBOL_GPL(lib_ring_buffer_clear_quiescent_channel);
+
 static void channel_free(struct channel *chan)
 {
+	if (chan->backend.release_priv_ops) {
+		chan->backend.release_priv_ops(chan->backend.priv_ops);
+	}
 	channel_iterator_free(chan);
 	channel_backend_free(&chan->backend);
 	kfree(chan);
@@ -740,7 +805,7 @@ void *channel_destroy(struct channel *chan)
 							   chan->backend.priv,
 							   cpu);
 			if (buf->backend.allocated)
-				lib_ring_buffer_switch_slow(buf, SWITCH_FLUSH);
+				lib_ring_buffer_set_quiescent(buf);
 			/*
 			 * Perform flush before writing to finalized.
 			 */
@@ -754,7 +819,7 @@ void *channel_destroy(struct channel *chan)
 		if (config->cb.buffer_finalize)
 			config->cb.buffer_finalize(buf, chan->backend.priv, -1);
 		if (buf->backend.allocated)
-			lib_ring_buffer_switch_slow(buf, SWITCH_FLUSH);
+			lib_ring_buffer_set_quiescent(buf);
 		/*
 		 * Perform flush before writing to finalized.
 		 */
@@ -789,7 +854,7 @@ int lib_ring_buffer_open_read(struct lib_ring_buffer *buf)
 	if (!atomic_long_add_unless(&buf->active_readers, 1, 1))
 		return -EBUSY;
 	kref_get(&chan->ref);
-	smp_mb__after_atomic_inc();
+	lttng_smp_mb__after_atomic();
 	return 0;
 }
 EXPORT_SYMBOL_GPL(lib_ring_buffer_open_read);
@@ -799,7 +864,7 @@ void lib_ring_buffer_release_read(struct lib_ring_buffer *buf)
 	struct channel *chan = buf->backend.chan;
 
 	CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
-	smp_mb__before_atomic_dec();
+	lttng_smp_mb__before_atomic();
 	atomic_long_dec(&buf->active_readers);
 	kref_put(&chan->ref, channel_release);
 }
@@ -927,6 +992,13 @@ int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf,
 	int ret;
 	int finalized;
 
+	if (buf->get_subbuf) {
+		/*
+		 * Reader is trying to get a subbuffer twice.
+		 */
+		CHAN_WARN_ON(chan, 1);
+		return -EBUSY;
+	}
 retry:
 	finalized = ACCESS_ONCE(buf->finalized);
 	/*
@@ -1015,7 +1087,7 @@ retry:
 	 */
 	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;
@@ -1024,7 +1096,7 @@ retry:
 	 * 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;
 
@@ -1217,7 +1289,8 @@ void lib_ring_buffer_print_errors(struct channel *chan,
 /*
  * lib_ring_buffer_switch_old_start: Populate old subbuffer header.
  *
- * Only executed when the buffer is finalized, in SWITCH_FLUSH.
+ * Only executed by SWITCH_FLUSH, which can be issued while tracing is active
+ * or at buffer finalization (destroy).
  */
 static
 void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf,
@@ -1249,10 +1322,10 @@ void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf,
 	commit_count = v_read(config, &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);
+				      commit_count, oldidx, tsc);
 	lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
-					     offsets->old, commit_count,
-					     config->cb.subbuffer_header_size());
+			offsets->old + config->cb.subbuffer_header_size(),
+			commit_count);
 }
 
 /*
@@ -1293,10 +1366,9 @@ void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf,
 	v_add(config, padding_size, &buf->commit_hot[oldidx].cc);
 	commit_count = v_read(config, &buf->commit_hot[oldidx].cc);
 	lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
-				      commit_count, oldidx);
+				      commit_count, oldidx, tsc);
 	lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
-					     offsets->old, commit_count,
-					     padding_size);
+			offsets->old + padding_size, commit_count);
 }
 
 /*
@@ -1336,17 +1408,19 @@ void lib_ring_buffer_switch_new_start(struct lib_ring_buffer *buf,
 	commit_count = v_read(config, &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);
+				      commit_count, beginidx, tsc);
 	lib_ring_buffer_write_commit_counter(config, buf, chan, beginidx,
-					     offsets->begin, commit_count,
-					     config->cb.subbuffer_header_size());
+			offsets->begin + config->cb.subbuffer_header_size(),
+			commit_count);
 }
 
 /*
  * lib_ring_buffer_switch_new_end: finish switching current subbuffer
  *
- * The only remaining threads could be the ones with pending commits. They will
- * have to do the deliver themselves.
+ * Calls subbuffer_set_data_size() to set the data size of the current
+ * sub-buffer. We do not need to perform check_deliver nor commit here,
+ * since this task will be done by the "commit" of the event for which
+ * we are currently doing the space reservation.
  */
 static
 void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf,
@@ -1355,33 +1429,11 @@ void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf,
 					    u64 tsc)
 {
 	const struct lib_ring_buffer_config *config = &chan->backend.config;
-	unsigned long endidx = subbuf_index(offsets->end - 1, chan);
-	unsigned long commit_count, padding_size, data_size;
+	unsigned long endidx, data_size;
 
+	endidx = subbuf_index(offsets->end - 1, chan);
 	data_size = subbuf_offset(offsets->end - 1, chan) + 1;
-	padding_size = chan->backend.subbuf_size - data_size;
 	subbuffer_set_data_size(config, &buf->backend, endidx, data_size);
-
-	/*
-	 * Order all writes to buffer before the commit count update that will
-	 * determine that the subbuffer is full.
-	 */
-	if (config->ipi == RING_BUFFER_IPI_BARRIER) {
-		/*
-		 * Must write slot data before incrementing commit count.  This
-		 * compiler barrier is upgraded into a smp_mb() by the IPI sent
-		 * by get_subbuf().
-		 */
-		barrier();
-	} else
-		smp_wmb();
-	v_add(config, padding_size, &buf->commit_hot[endidx].cc);
-	commit_count = v_read(config, &buf->commit_hot[endidx].cc);
-	lib_ring_buffer_check_deliver(config, buf, chan, offsets->end - 1,
-				  commit_count, endidx);
-	lib_ring_buffer_write_commit_counter(config, buf, chan, endidx,
-					     offsets->end, commit_count,
-					     padding_size);
 }
 
 /*
@@ -1397,7 +1449,7 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
 				    u64 *tsc)
 {
 	const struct 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;
@@ -1422,36 +1474,70 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
 	 * (records and header timestamps) are visible to the reader. This is
 	 * required for quiescence guarantees for the fusion merge.
 	 */
-	if (mode == SWITCH_FLUSH || off > 0) {
-		if (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 (unlikely(subbuf_trunc(offsets->begin, chan)
-					- subbuf_trunc((unsigned long)
-						atomic_long_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 (unlikely(off == 0)) {
+		unsigned long sb_index, commit_count;
+
+		/*
+		 * We are performing a SWITCH_FLUSH. There may be concurrent
+		 * writes into the buffer if e.g. invoked while performing a
+		 * snapshot on an active trace.
+		 *
+		 * If the client does not save any header information (sub-buffer
+		 * header size == 0), 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,
+				&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 (likely(reserve_commit_diff == 0)) {
+			/* Next subbuffer not being written to. */
+			if (unlikely(config->mode != RING_BUFFER_OVERWRITE &&
+				subbuf_trunc(offsets->begin, chan)
+				 - subbuf_trunc((unsigned long)
+				     atomic_long_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;
@@ -1516,6 +1602,70 @@ void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode m
 }
 EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_slow);
 
+struct switch_param {
+	struct lib_ring_buffer *buf;
+	enum switch_mode mode;
+};
+
+static void remote_switch(void *info)
+{
+	struct switch_param *param = info;
+	struct lib_ring_buffer *buf = param->buf;
+
+	lib_ring_buffer_switch_slow(buf, param->mode);
+}
+
+static void _lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf,
+		enum switch_mode mode)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct lib_ring_buffer_config *config = &chan->backend.config;
+	int ret;
+	struct switch_param param;
+
+	/*
+	 * With global synchronization we don't need to use the IPI scheme.
+	 */
+	if (config->sync == RING_BUFFER_SYNC_GLOBAL) {
+		lib_ring_buffer_switch_slow(buf, mode);
+		return;
+	}
+
+	/*
+	 * Taking lock on CPU hotplug to ensure two things: first, that the
+	 * target cpu is not taken concurrently offline while we are within
+	 * smp_call_function_single() (I don't trust that get_cpu() on the
+	 * _local_ CPU actually inhibit CPU hotplug for the _remote_ CPU (to be
+	 * confirmed)). Secondly, if it happens that the CPU is not online, our
+	 * own call to lib_ring_buffer_switch_slow() needs to be protected from
+	 * CPU hotplug handlers, which can also perform a remote subbuffer
+	 * switch.
+	 */
+	get_online_cpus();
+	param.buf = buf;
+	param.mode = mode;
+	ret = smp_call_function_single(buf->backend.cpu,
+				 remote_switch, &param, 1);
+	if (ret) {
+		/* Remote CPU is offline, do it ourself. */
+		lib_ring_buffer_switch_slow(buf, mode);
+	}
+	put_online_cpus();
+}
+
+void lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf)
+{
+	_lib_ring_buffer_switch_remote(buf, SWITCH_ACTIVE);
+}
+EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_remote);
+
+/* Switch sub-buffer even if current sub-buffer is empty. */
+void lib_ring_buffer_switch_remote_empty(struct lib_ring_buffer *buf)
+{
+	_lib_ring_buffer_switch_remote(buf, SWITCH_FLUSH);
+}
+EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_remote_empty);
+
 /*
  * Returns :
  * 0 if ok
@@ -1530,9 +1680,10 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
 				     struct lib_ring_buffer_ctx *ctx)
 {
 	const struct lib_ring_buffer_config *config = &chan->backend.config;
-	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;
@@ -1564,7 +1715,7 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
 		}
 	}
 	if (unlikely(offsets->switch_new_start)) {
-		unsigned long sb_index;
+		unsigned long sb_index, commit_count;
 
 		/*
 		 * We are typically not filling the previous buffer completely.
@@ -1575,12 +1726,31 @@ int lib_ring_buffer_try_reserve_slow(struct 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.
+		 */
+		smp_rmb();
+		commit_count = v_read(config,
+				&buf->commit_cold[sb_index].cc_sb);
+		/* Read buf->commit_cold[sb_index].cc_sb before buf->offset. */
+		smp_rmb();
+		if (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,
-					    &buf->commit_cold[sb_index].cc_sb)
-		     & chan->commit_count_mask);
+		  - (commit_count & chan->commit_count_mask);
 		if (likely(reserve_commit_diff == 0)) {
 			/* Next subbuffer not being written to. */
 			if (unlikely(config->mode != RING_BUFFER_OVERWRITE &&
@@ -1605,9 +1775,10 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf,
 		} else {
 			/*
 			 * Next subbuffer reserve offset does not match the
-			 * commit offset. Drop record in producer-consumer and
-			 * overwrite mode. Caused by either a writer OOPS or too
-			 * many nested writes over a reserve/commit pair.
+			 * 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.
 			 */
 			v_inc(config, &buf->records_lost_wrap);
 			return -EIO;