* [PATCH v3 1/2] ring_buffer: Do no not complete benchmark reader too early
2015-09-07 12:38 [PATCH v3 0/2] ring_buffer: Make the benchmark slightly more safe Petr Mladek
@ 2015-09-07 12:38 ` Petr Mladek
2015-09-07 12:38 ` [PATCH v3 2/2] ring_buffer: Fix more races when terminating the producer in the benchmark Petr Mladek
1 sibling, 0 replies; 3+ messages in thread
From: Petr Mladek @ 2015-09-07 12:38 UTC (permalink / raw)
To: Steven Rostedt; +Cc: Ingo Molnar, Jiri Kosina, linux-kernel, Petr Mladek
It seems that complete(&read_done) might be called too early
in some situations.
1st scenario:
-------------
CPU0 CPU1
ring_buffer_producer_thread()
wake_up_process(consumer);
wait_for_completion(&read_start);
ring_buffer_consumer_thread()
complete(&read_start);
ring_buffer_producer()
# producing data in
# the do-while cycle
ring_buffer_consumer();
# reading data
# got error
# set kill_test = 1;
set_current_state(
TASK_INTERRUPTIBLE);
if (reader_finish) # false
schedule();
# producer still in the middle of
# do-while cycle
if (consumer && !(cnt % wakeup_interval))
wake_up_process(consumer);
# spurious wakeup
while (!reader_finish &&
!kill_test)
# leaving because
# kill_test == 1
reader_finish = 0;
complete(&read_done);
1st BANG: We might access uninitialized "read_done" if this is the
the first round.
# producer finally leaving
# the do-while cycle because kill_test == 1;
if (consumer) {
reader_finish = 1;
wake_up_process(consumer);
wait_for_completion(&read_done);
2nd BANG: This will never complete because consumer already did
the completion.
2nd scenario:
-------------
CPU0 CPU1
ring_buffer_producer_thread()
wake_up_process(consumer);
wait_for_completion(&read_start);
ring_buffer_consumer_thread()
complete(&read_start);
ring_buffer_producer()
# CPU3 removes the module <--- difference from
# and stops producer <--- the 1st scenario
if (kthread_should_stop())
kill_test = 1;
ring_buffer_consumer();
while (!reader_finish &&
!kill_test)
# kill_test == 1 => we never go
# into the top level while()
reader_finish = 0;
complete(&read_done);
# producer still in the middle of
# do-while cycle
if (consumer && !(cnt % wakeup_interval))
wake_up_process(consumer);
# spurious wakeup
while (!reader_finish &&
!kill_test)
# leaving because kill_test == 1
reader_finish = 0;
complete(&read_done);
BANG: We are in the same "bang" situations as in the 1st scenario.
Root of the problem:
--------------------
ring_buffer_consumer() must complete "read_done" only when "reader_finish"
variable is set. It must not be skipped due to other conditions.
Note that we still must keep the check for "reader_finish" in a loop
because there might be spurious wakeups as described in the
above scenarios.
Solution:
----------
The top level cycle in ring_buffer_consumer() will finish only when
"reader_finish" is set. The data will be read in "while-do" cycle
so that they are not read after an error (kill_test == 1)
or a spurious wake up.
In addition, "reader_finish" is manipulated by the producer thread.
Therefore we add READ_ONCE() to make sure that the fresh value is
read in each cycle. Also we add the corresponding barrier
to synchronize the sleep check.
Next we set the state back to TASK_RUNNING for the situation where we
did not sleep.
Just from paranoid reasons, we initialize both completions statically.
This is safer, in case there are other races that we are unaware of.
As a side effect we could remove the memory barrier from
ring_buffer_producer_thread(). IMHO, this was the reason for
the barrier. ring_buffer_reset() uses spin locks that should
provide the needed memory barrier for using the buffer.
Signed-off-by: Petr Mladek <pmladek@suse.com>
---
kernel/trace/ring_buffer_benchmark.c | 25 ++++++++++++++++---------
1 file changed, 16 insertions(+), 9 deletions(-)
diff --git a/kernel/trace/ring_buffer_benchmark.c b/kernel/trace/ring_buffer_benchmark.c
index a1503a027ee2..9ea7949366b3 100644
--- a/kernel/trace/ring_buffer_benchmark.c
+++ b/kernel/trace/ring_buffer_benchmark.c
@@ -24,8 +24,8 @@ struct rb_page {
static int wakeup_interval = 100;
static int reader_finish;
-static struct completion read_start;
-static struct completion read_done;
+static DECLARE_COMPLETION(read_start);
+static DECLARE_COMPLETION(read_done);
static struct ring_buffer *buffer;
static struct task_struct *producer;
@@ -178,10 +178,14 @@ static void ring_buffer_consumer(void)
read_events ^= 1;
read = 0;
- while (!reader_finish && !kill_test) {
- int found;
+ /*
+ * Continue running until the producer specifically asks to stop
+ * and is ready for the completion.
+ */
+ while (!READ_ONCE(reader_finish)) {
+ int found = 1;
- do {
+ while (found && !kill_test) {
int cpu;
found = 0;
@@ -195,17 +199,23 @@ static void ring_buffer_consumer(void)
if (kill_test)
break;
+
if (stat == EVENT_FOUND)
found = 1;
+
}
- } while (found && !kill_test);
+ }
+ /* Wait till the producer wakes us up when there is more data
+ * available or when the producer wants us to finish reading.
+ */
set_current_state(TASK_INTERRUPTIBLE);
if (reader_finish)
break;
schedule();
}
+ __set_current_state(TASK_RUNNING);
reader_finish = 0;
complete(&read_done);
}
@@ -389,13 +399,10 @@ static int ring_buffer_consumer_thread(void *arg)
static int ring_buffer_producer_thread(void *arg)
{
- init_completion(&read_start);
-
while (!kthread_should_stop() && !kill_test) {
ring_buffer_reset(buffer);
if (consumer) {
- smp_wmb();
wake_up_process(consumer);
wait_for_completion(&read_start);
}
--
1.8.5.6
^ permalink raw reply [flat|nested] 3+ messages in thread* [PATCH v3 2/2] ring_buffer: Fix more races when terminating the producer in the benchmark
2015-09-07 12:38 [PATCH v3 0/2] ring_buffer: Make the benchmark slightly more safe Petr Mladek
2015-09-07 12:38 ` [PATCH v3 1/2] ring_buffer: Do no not complete benchmark reader too early Petr Mladek
@ 2015-09-07 12:38 ` Petr Mladek
1 sibling, 0 replies; 3+ messages in thread
From: Petr Mladek @ 2015-09-07 12:38 UTC (permalink / raw)
To: Steven Rostedt; +Cc: Ingo Molnar, Jiri Kosina, linux-kernel, Petr Mladek
The commit b44754d8262d3aab8 ("ring_buffer: Allow to exit the ring
buffer benchmark immediately") added a hack into ring_buffer_producer()
that set @kill_test when kthread_should_stop() returned true. It improved
the situation a lot. It stopped the kthread in most cases because
the producer spent most of the time in the patched while cycle.
But there are still few possible races when kthread_should_stop()
is set outside of the cycle. Then we do not set @kill_test and
some other checks pass.
This patch adds a better fix. It renames @test_kill/TEST_KILL() into
a better descriptive @test_error/TEST_ERROR(). Also it introduces
break_test() function that checks for both @test_error and
kthread_should_stop().
The new function is used in the producer when the check for @test_error
is not enough. It is not used in the consumer because its state
is manipulated by the producer via the "reader_finish" variable.
Also we add a missing check into ring_buffer_producer_thread()
between setting TASK_INTERRUPTIBLE and calling schedule_timeout().
Otherwise, we might miss a wakeup from kthread_stop().
Signed-off-by: Petr Mladek <pmladek@suse.com>
---
kernel/trace/ring_buffer_benchmark.c | 54 +++++++++++++++++++-----------------
1 file changed, 29 insertions(+), 25 deletions(-)
diff --git a/kernel/trace/ring_buffer_benchmark.c b/kernel/trace/ring_buffer_benchmark.c
index 9ea7949366b3..9e00fd178226 100644
--- a/kernel/trace/ring_buffer_benchmark.c
+++ b/kernel/trace/ring_buffer_benchmark.c
@@ -60,12 +60,12 @@ MODULE_PARM_DESC(consumer_fifo, "fifo prio for consumer");
static int read_events;
-static int kill_test;
+static int test_error;
-#define KILL_TEST() \
+#define TEST_ERROR() \
do { \
- if (!kill_test) { \
- kill_test = 1; \
+ if (!test_error) { \
+ test_error = 1; \
WARN_ON(1); \
} \
} while (0)
@@ -75,6 +75,11 @@ enum event_status {
EVENT_DROPPED,
};
+static bool break_test(void)
+{
+ return test_error || kthread_should_stop();
+}
+
static enum event_status read_event(int cpu)
{
struct ring_buffer_event *event;
@@ -87,7 +92,7 @@ static enum event_status read_event(int cpu)
entry = ring_buffer_event_data(event);
if (*entry != cpu) {
- KILL_TEST();
+ TEST_ERROR();
return EVENT_DROPPED;
}
@@ -115,10 +120,10 @@ static enum event_status read_page(int cpu)
rpage = bpage;
/* The commit may have missed event flags set, clear them */
commit = local_read(&rpage->commit) & 0xfffff;
- for (i = 0; i < commit && !kill_test; i += inc) {
+ for (i = 0; i < commit && !test_error ; i += inc) {
if (i >= (PAGE_SIZE - offsetof(struct rb_page, data))) {
- KILL_TEST();
+ TEST_ERROR();
break;
}
@@ -128,7 +133,7 @@ static enum event_status read_page(int cpu)
case RINGBUF_TYPE_PADDING:
/* failed writes may be discarded events */
if (!event->time_delta)
- KILL_TEST();
+ TEST_ERROR();
inc = event->array[0] + 4;
break;
case RINGBUF_TYPE_TIME_EXTEND:
@@ -137,12 +142,12 @@ static enum event_status read_page(int cpu)
case 0:
entry = ring_buffer_event_data(event);
if (*entry != cpu) {
- KILL_TEST();
+ TEST_ERROR();
break;
}
read++;
if (!event->array[0]) {
- KILL_TEST();
+ TEST_ERROR();
break;
}
inc = event->array[0] + 4;
@@ -150,17 +155,17 @@ static enum event_status read_page(int cpu)
default:
entry = ring_buffer_event_data(event);
if (*entry != cpu) {
- KILL_TEST();
+ TEST_ERROR();
break;
}
read++;
inc = ((event->type_len + 1) * 4);
}
- if (kill_test)
+ if (test_error)
break;
if (inc <= 0) {
- KILL_TEST();
+ TEST_ERROR();
break;
}
}
@@ -185,7 +190,7 @@ static void ring_buffer_consumer(void)
while (!READ_ONCE(reader_finish)) {
int found = 1;
- while (found && !kill_test) {
+ while (found && !test_error) {
int cpu;
found = 0;
@@ -197,7 +202,7 @@ static void ring_buffer_consumer(void)
else
stat = read_page(cpu);
- if (kill_test)
+ if (test_error)
break;
if (stat == EVENT_FOUND)
@@ -273,10 +278,7 @@ static void ring_buffer_producer(void)
if (cnt % wakeup_interval)
cond_resched();
#endif
- if (kthread_should_stop())
- kill_test = 1;
-
- } while (ktime_before(end_time, timeout) && !kill_test);
+ } while (ktime_before(end_time, timeout) && !break_test());
trace_printk("End ring buffer hammer\n");
if (consumer) {
@@ -297,7 +299,7 @@ static void ring_buffer_producer(void)
entries = ring_buffer_entries(buffer);
overruns = ring_buffer_overruns(buffer);
- if (kill_test && !kthread_should_stop())
+ if (test_error)
trace_printk("ERROR!\n");
if (!disable_reader) {
@@ -378,15 +380,14 @@ static void wait_to_die(void)
static int ring_buffer_consumer_thread(void *arg)
{
- while (!kthread_should_stop() && !kill_test) {
+ while (!break_test()) {
complete(&read_start);
ring_buffer_consumer();
set_current_state(TASK_INTERRUPTIBLE);
- if (kthread_should_stop() || kill_test)
+ if (break_test())
break;
-
schedule();
}
__set_current_state(TASK_RUNNING);
@@ -399,7 +400,7 @@ static int ring_buffer_consumer_thread(void *arg)
static int ring_buffer_producer_thread(void *arg)
{
- while (!kthread_should_stop() && !kill_test) {
+ while (!break_test()) {
ring_buffer_reset(buffer);
if (consumer) {
@@ -408,15 +409,18 @@ static int ring_buffer_producer_thread(void *arg)
}
ring_buffer_producer();
- if (kill_test)
+ if (break_test())
goto out_kill;
trace_printk("Sleeping for 10 secs\n");
set_current_state(TASK_INTERRUPTIBLE);
+ if (break_test())
+ goto out_kill;
schedule_timeout(HZ * SLEEP_TIME);
}
out_kill:
+ __set_current_state(TASK_RUNNING);
if (!kthread_should_stop())
wait_to_die();
--
1.8.5.6
^ permalink raw reply [flat|nested] 3+ messages in thread