[PATCH 0/9] printk: Cleanups and softlockup avoidance

[PATCH 0/9] printk: Cleanups and softlockup avoidance

[Jan Kara]

  Hello,

  this is another piece of the printk softlockup saga series. Let me first
remind the problem:

Currently, console_unlock() prints messages from kernel printk buffer to
console while the buffer is non-empty. When serial console is attached,
printing is slow and thus other CPUs in the system have plenty of time
to append new messages to the buffer while one CPU is printing. Thus the
CPU can spend unbounded amount of time doing printing in console_unlock().
This is especially serious since vprintk_emit() calls console_unlock()
with interrupts disabled.
    
In practice users have observed a CPU can spend tens of seconds printing
in console_unlock() (usually during boot when hundreds of SCSI devices
are discovered) resulting in RCU stalls (CPU doing printing doesn't
reach quiescent state for a long time), softlockup reports (IPIs for the
printing CPU don't get served and thus other CPUs are spinning waiting
for the printing CPU to process IPIs), and eventually a machine death
(as messages from stalls and lockups append to printk buffer faster than
we are able to print). So these machines are unable to boot with serial
console attached. Also during artificial stress testing SATA disk
disappears from the system because its interrupts aren't served for too
long.
---

Since my previous attempts to fix softlockups in printk under heavy load met
some resistance, I've decided to try a different approach - do not let
CPU out of the console_unlock() loop until there's someone else to take over
the printing.

This patch set implements that idea. It is organized as follows:

First three patches are cleanups of block layer and improvement of
smp_call_function_single() to use lockless lists.  These patches are already
queued in block tree so they are here only for completeness.

Patches 4-5 implement __smp_call_function_any() to IPI any CPU from given
cpumask with own csd structure provided.

Patches 6-8 are the printk cleanup patches I have already posted. They make
sense on their own so even if patch 9 is considered too problematic / needing
more work please consider merging these three.

Patch 9 implements the hand over of console_sem when CPU has printed over
printk.offload_chars characters and another CPU is in
console_trylock_for_printk() and also sending IPI to some other CPU to come and
take over printing if no printk has been called for a long time.

What do you guys think?

						Merry Christmas ;)
								Honza

[PATCH 1/9] block: Stop abusing csd.list for fifo_time

[Jan Kara]

Block layer currently abuses rq->csd.list.next for storing fifo_time.
That is a terrible hack and completely unnecessary as well. Union
achieves the same space saving in a cleaner way.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 block/cfq-iosched.c      | 8 ++++----
 block/deadline-iosched.c | 8 ++++----
 include/linux/blkdev.h   | 1 +
 include/linux/elevator.h | 6 ------
 4 files changed, 9 insertions(+), 14 deletions(-)

diff --git a/block/cfq-iosched.c b/block/cfq-iosched.c
index 4d5cec1ad80d..37ea6045fdc6 100644
--- a/block/cfq-iosched.c
+++ b/block/cfq-iosched.c
@@ -2382,10 +2382,10 @@ cfq_merged_requests(struct request_queue *q, struct request *rq,
 	 * reposition in fifo if next is older than rq
 	 */
 	if (!list_empty(&rq->queuelist) && !list_empty(&next->queuelist) &&
-	    time_before(rq_fifo_time(next), rq_fifo_time(rq)) &&
+	    time_before(next->fifo_time, rq->fifo_time) &&
 	    cfqq == RQ_CFQQ(next)) {
 		list_move(&rq->queuelist, &next->queuelist);
-		rq_set_fifo_time(rq, rq_fifo_time(next));
+		rq->fifo_time = next->fifo_time;
 	}
 
 	if (cfqq->next_rq == next)
@@ -2829,7 +2829,7 @@ static struct request *cfq_check_fifo(struct cfq_queue *cfqq)
 		return NULL;
 
 	rq = rq_entry_fifo(cfqq->fifo.next);
-	if (time_before(jiffies, rq_fifo_time(rq)))
+	if (time_before(jiffies, rq->fifo_time))
 		rq = NULL;
 
 	cfq_log_cfqq(cfqq->cfqd, cfqq, "fifo=%p", rq);
@@ -3942,7 +3942,7 @@ static void cfq_insert_request(struct request_queue *q, struct request *rq)
 	cfq_log_cfqq(cfqd, cfqq, "insert_request");
 	cfq_init_prio_data(cfqq, RQ_CIC(rq));
 
-	rq_set_fifo_time(rq, jiffies + cfqd->cfq_fifo_expire[rq_is_sync(rq)]);
+	rq->fifo_time = jiffies + cfqd->cfq_fifo_expire[rq_is_sync(rq)];
 	list_add_tail(&rq->queuelist, &cfqq->fifo);
 	cfq_add_rq_rb(rq);
 	cfqg_stats_update_io_add(RQ_CFQG(rq), cfqd->serving_group,
diff --git a/block/deadline-iosched.c b/block/deadline-iosched.c
index 9ef66406c625..a753df2b3fc2 100644
--- a/block/deadline-iosched.c
+++ b/block/deadline-iosched.c
@@ -106,7 +106,7 @@ deadline_add_request(struct request_queue *q, struct request *rq)
 	/*
 	 * set expire time and add to fifo list
 	 */
-	rq_set_fifo_time(rq, jiffies + dd->fifo_expire[data_dir]);
+	rq->fifo_time = jiffies + dd->fifo_expire[data_dir];
 	list_add_tail(&rq->queuelist, &dd->fifo_list[data_dir]);
 }
 
@@ -174,9 +174,9 @@ deadline_merged_requests(struct request_queue *q, struct request *req,
 	 * and move into next position (next will be deleted) in fifo
 	 */
 	if (!list_empty(&req->queuelist) && !list_empty(&next->queuelist)) {
-		if (time_before(rq_fifo_time(next), rq_fifo_time(req))) {
+		if (time_before(next->fifo_time, req->fifo_time)) {
 			list_move(&req->queuelist, &next->queuelist);
-			rq_set_fifo_time(req, rq_fifo_time(next));
+			req->fifo_time = next->fifo_time;
 		}
 	}
 
@@ -230,7 +230,7 @@ static inline int deadline_check_fifo(struct deadline_data *dd, int ddir)
 	/*
 	 * rq is expired!
 	 */
-	if (time_after_eq(jiffies, rq_fifo_time(rq)))
+	if (time_after_eq(jiffies, rq->fifo_time))
 		return 1;
 
 	return 0;
diff --git a/include/linux/blkdev.h b/include/linux/blkdev.h
index 1b135d49b279..744b5fb13a9b 100644
--- a/include/linux/blkdev.h
+++ b/include/linux/blkdev.h
@@ -102,6 +102,7 @@ struct request {
 	union {
 		struct call_single_data csd;
 		struct work_struct mq_flush_data;
+		unsigned long fifo_time;
 	};
 
 	struct request_queue *q;
diff --git a/include/linux/elevator.h b/include/linux/elevator.h
index 306dd8cd0b6f..0bdfd46f4735 100644
--- a/include/linux/elevator.h
+++ b/include/linux/elevator.h
@@ -202,12 +202,6 @@ enum {
 #define rq_end_sector(rq)	(blk_rq_pos(rq) + blk_rq_sectors(rq))
 #define rb_entry_rq(node)	rb_entry((node), struct request, rb_node)
 
-/*
- * Hack to reuse the csd.list list_head as the fifo time holder while
- * the request is in the io scheduler. Saves an unsigned long in rq.
- */
-#define rq_fifo_time(rq)	((unsigned long) (rq)->csd.list.next)
-#define rq_set_fifo_time(rq,exp)	((rq)->csd.list.next = (void *) (exp))
 #define rq_entry_fifo(ptr)	list_entry((ptr), struct request, queuelist)
 #define rq_fifo_clear(rq)	do {		\
 	list_del_init(&(rq)->queuelist);	\
-- 
1.8.1.4

Re: [PATCH 1/9] block: Stop abusing csd.list for fifo_time

[Frederic Weisbecker]

On Mon, Dec 23, 2013 at 09:39:22PM +0100, Jan Kara wrote:
> Block layer currently abuses rq->csd.list.next for storing fifo_time.
> That is a terrible hack and completely unnecessary as well. Union
> achieves the same space saving in a cleaner way.
> 
> Signed-off-by: Jan Kara <jack@suse.cz>

Hi Jan,

Taken as is, the patch is fine and it builds.
But later when I finally get rid of csd->list in a subsequent patch,
rq_fifo_clear() callers break the build.

This is because rq_fifo_clear() initialize the csd->list and I'm not
sure how to fix that leftover because I am not clear about the purpose
of that INIT_LIST_HEAD(): is it to reset fifo time or to prepare for
an IPI to be queued?

All in one it looks buggy because if this is to prepare for the IPI,
it's useless as csd.list is not a list head but just a node. Otherwise if it
is to reset fifo_time it's wrong because INIT_LIST_HEAD doesn't initialize
to 0.

Anyway so I did a braindead fix by removing the whole INIT_LIST_HEAD()
from rq_fifo_clear(), see the patch below. Now to be honest I have no idea
what I'm doing.

---
>From 112bcbb73076dd1374541eec9b554410dd0e73e0 Mon Sep 17 00:00:00 2001
From: Jan Kara <jack@suse.cz>
Date: Mon, 23 Dec 2013 21:39:22 +0100
Subject: [PATCH] block: Stop abusing csd.list for fifo_time

Block layer currently abuses rq->csd.list.next for storing fifo_time.
That is a terrible hack and completely unnecessary as well. Union
achieves the same space saving in a cleaner way.

Signed-off-by: Jan Kara <jack@suse.cz>
Signed-off-by: Frederic Weisbecker <fweisbec@gmail.com>
---
 block/cfq-iosched.c      |  8 ++++----
 block/deadline-iosched.c |  8 ++++----
 include/linux/blkdev.h   |  1 +
 include/linux/elevator.h | 12 ++----------
 4 files changed, 11 insertions(+), 18 deletions(-)

diff --git a/block/cfq-iosched.c b/block/cfq-iosched.c
index 744833b..5873e4a 100644
--- a/block/cfq-iosched.c
+++ b/block/cfq-iosched.c
@@ -2367,10 +2367,10 @@ cfq_merged_requests(struct request_queue *q, struct request *rq,
 	 * reposition in fifo if next is older than rq
 	 */
 	if (!list_empty(&rq->queuelist) && !list_empty(&next->queuelist) &&
-	    time_before(rq_fifo_time(next), rq_fifo_time(rq)) &&
+	    time_before(next->fifo_time, rq->fifo_time) &&
 	    cfqq == RQ_CFQQ(next)) {
 		list_move(&rq->queuelist, &next->queuelist);
-		rq_set_fifo_time(rq, rq_fifo_time(next));
+		rq->fifo_time = next->fifo_time;
 	}
 
 	if (cfqq->next_rq == next)
@@ -2814,7 +2814,7 @@ static struct request *cfq_check_fifo(struct cfq_queue *cfqq)
 		return NULL;
 
 	rq = rq_entry_fifo(cfqq->fifo.next);
-	if (time_before(jiffies, rq_fifo_time(rq)))
+	if (time_before(jiffies, rq->fifo_time))
 		rq = NULL;
 
 	cfq_log_cfqq(cfqq->cfqd, cfqq, "fifo=%p", rq);
@@ -3927,7 +3927,7 @@ static void cfq_insert_request(struct request_queue *q, struct request *rq)
 	cfq_log_cfqq(cfqd, cfqq, "insert_request");
 	cfq_init_prio_data(cfqq, RQ_CIC(rq));
 
-	rq_set_fifo_time(rq, jiffies + cfqd->cfq_fifo_expire[rq_is_sync(rq)]);
+	rq->fifo_time = jiffies + cfqd->cfq_fifo_expire[rq_is_sync(rq)];
 	list_add_tail(&rq->queuelist, &cfqq->fifo);
 	cfq_add_rq_rb(rq);
 	cfqg_stats_update_io_add(RQ_CFQG(rq), cfqd->serving_group,
diff --git a/block/deadline-iosched.c b/block/deadline-iosched.c
index 9ef6640..a753df2 100644
--- a/block/deadline-iosched.c
+++ b/block/deadline-iosched.c
@@ -106,7 +106,7 @@ deadline_add_request(struct request_queue *q, struct request *rq)
 	/*
 	 * set expire time and add to fifo list
 	 */
-	rq_set_fifo_time(rq, jiffies + dd->fifo_expire[data_dir]);
+	rq->fifo_time = jiffies + dd->fifo_expire[data_dir];
 	list_add_tail(&rq->queuelist, &dd->fifo_list[data_dir]);
 }
 
@@ -174,9 +174,9 @@ deadline_merged_requests(struct request_queue *q, struct request *req,
 	 * and move into next position (next will be deleted) in fifo
 	 */
 	if (!list_empty(&req->queuelist) && !list_empty(&next->queuelist)) {
-		if (time_before(rq_fifo_time(next), rq_fifo_time(req))) {
+		if (time_before(next->fifo_time, req->fifo_time)) {
 			list_move(&req->queuelist, &next->queuelist);
-			rq_set_fifo_time(req, rq_fifo_time(next));
+			req->fifo_time = next->fifo_time;
 		}
 	}
 
@@ -230,7 +230,7 @@ static inline int deadline_check_fifo(struct deadline_data *dd, int ddir)
 	/*
 	 * rq is expired!
 	 */
-	if (time_after_eq(jiffies, rq_fifo_time(rq)))
+	if (time_after_eq(jiffies, rq->fifo_time))
 		return 1;
 
 	return 0;
diff --git a/include/linux/blkdev.h b/include/linux/blkdev.h
index 8678c43..dda98e3 100644
--- a/include/linux/blkdev.h
+++ b/include/linux/blkdev.h
@@ -99,6 +99,7 @@ struct request {
 	union {
 		struct call_single_data csd;
 		struct work_struct mq_flush_data;
+		unsigned long fifo_time;
 	};
 
 	struct request_queue *q;
diff --git a/include/linux/elevator.h b/include/linux/elevator.h
index 306dd8c..0b87f4e 100644
--- a/include/linux/elevator.h
+++ b/include/linux/elevator.h
@@ -202,17 +202,9 @@ enum {
 #define rq_end_sector(rq)	(blk_rq_pos(rq) + blk_rq_sectors(rq))
 #define rb_entry_rq(node)	rb_entry((node), struct request, rb_node)
 
-/*
- * Hack to reuse the csd.list list_head as the fifo time holder while
- * the request is in the io scheduler. Saves an unsigned long in rq.
- */
-#define rq_fifo_time(rq)	((unsigned long) (rq)->csd.list.next)
-#define rq_set_fifo_time(rq,exp)	((rq)->csd.list.next = (void *) (exp))
 #define rq_entry_fifo(ptr)	list_entry((ptr), struct request, queuelist)
-#define rq_fifo_clear(rq)	do {		\
-	list_del_init(&(rq)->queuelist);	\
-	INIT_LIST_HEAD(&(rq)->csd.list);	\
-	} while (0)
+//CHECKME twice
+#define rq_fifo_clear(rq)	list_del_init(&(rq)->queuelist);
 
 #else /* CONFIG_BLOCK */
 
-- 
1.8.3.1

Re: [PATCH 1/9] block: Stop abusing csd.list for fifo_time

[Jan Kara]

On Sat 01-02-14 17:48:27, Frederic Weisbecker wrote:
> On Mon, Dec 23, 2013 at 09:39:22PM +0100, Jan Kara wrote:
> > Block layer currently abuses rq->csd.list.next for storing fifo_time.
> > That is a terrible hack and completely unnecessary as well. Union
> > achieves the same space saving in a cleaner way.
> > 
> > Signed-off-by: Jan Kara <jack@suse.cz>
> 
> Hi Jan,
> 
> Taken as is, the patch is fine and it builds.
> But later when I finally get rid of csd->list in a subsequent patch,
> rq_fifo_clear() callers break the build.
> 
> This is because rq_fifo_clear() initialize the csd->list and I'm not
> sure how to fix that leftover because I am not clear about the purpose
> of that INIT_LIST_HEAD(): is it to reset fifo time or to prepare for
> an IPI to be queued?
  I'm convinced it is there to prepare IPI to be queued. So just removing
the initialization as you did should be the right thing to do. You can
easily verify that it is correct - you boot the kernel, switch to 'deadline'
IO scheduler by doing
  echo 'deadline' >/sys/block/sda/queue/scheduler
and then do some IO. If it doesn't blow up, it is correct.

> All in one it looks buggy because if this is to prepare for the IPI,
> it's useless as csd.list is not a list head but just a node. Otherwise if it
> is to reset fifo_time it's wrong because INIT_LIST_HEAD doesn't initialize
> to 0.
  Yup, I think it is useless.

									Honza

> Anyway so I did a braindead fix by removing the whole INIT_LIST_HEAD()
> from rq_fifo_clear(), see the patch below. Now to be honest I have no idea
> what I'm doing.
> 
> ---
> From 112bcbb73076dd1374541eec9b554410dd0e73e0 Mon Sep 17 00:00:00 2001
> From: Jan Kara <jack@suse.cz>
> Date: Mon, 23 Dec 2013 21:39:22 +0100
> Subject: [PATCH] block: Stop abusing csd.list for fifo_time
> 
> Block layer currently abuses rq->csd.list.next for storing fifo_time.
> That is a terrible hack and completely unnecessary as well. Union
> achieves the same space saving in a cleaner way.
> 
> Signed-off-by: Jan Kara <jack@suse.cz>
> Signed-off-by: Frederic Weisbecker <fweisbec@gmail.com>
> ---
>  block/cfq-iosched.c      |  8 ++++----
>  block/deadline-iosched.c |  8 ++++----
>  include/linux/blkdev.h   |  1 +
>  include/linux/elevator.h | 12 ++----------
>  4 files changed, 11 insertions(+), 18 deletions(-)
> 
> diff --git a/block/cfq-iosched.c b/block/cfq-iosched.c
> index 744833b..5873e4a 100644
> --- a/block/cfq-iosched.c
> +++ b/block/cfq-iosched.c
> @@ -2367,10 +2367,10 @@ cfq_merged_requests(struct request_queue *q, struct request *rq,
>  	 * reposition in fifo if next is older than rq
>  	 */
>  	if (!list_empty(&rq->queuelist) && !list_empty(&next->queuelist) &&
> -	    time_before(rq_fifo_time(next), rq_fifo_time(rq)) &&
> +	    time_before(next->fifo_time, rq->fifo_time) &&
>  	    cfqq == RQ_CFQQ(next)) {
>  		list_move(&rq->queuelist, &next->queuelist);
> -		rq_set_fifo_time(rq, rq_fifo_time(next));
> +		rq->fifo_time = next->fifo_time;
>  	}
>  
>  	if (cfqq->next_rq == next)
> @@ -2814,7 +2814,7 @@ static struct request *cfq_check_fifo(struct cfq_queue *cfqq)
>  		return NULL;
>  
>  	rq = rq_entry_fifo(cfqq->fifo.next);
> -	if (time_before(jiffies, rq_fifo_time(rq)))
> +	if (time_before(jiffies, rq->fifo_time))
>  		rq = NULL;
>  
>  	cfq_log_cfqq(cfqq->cfqd, cfqq, "fifo=%p", rq);
> @@ -3927,7 +3927,7 @@ static void cfq_insert_request(struct request_queue *q, struct request *rq)
>  	cfq_log_cfqq(cfqd, cfqq, "insert_request");
>  	cfq_init_prio_data(cfqq, RQ_CIC(rq));
>  
> -	rq_set_fifo_time(rq, jiffies + cfqd->cfq_fifo_expire[rq_is_sync(rq)]);
> +	rq->fifo_time = jiffies + cfqd->cfq_fifo_expire[rq_is_sync(rq)];
>  	list_add_tail(&rq->queuelist, &cfqq->fifo);
>  	cfq_add_rq_rb(rq);
>  	cfqg_stats_update_io_add(RQ_CFQG(rq), cfqd->serving_group,
> diff --git a/block/deadline-iosched.c b/block/deadline-iosched.c
> index 9ef6640..a753df2 100644
> --- a/block/deadline-iosched.c
> +++ b/block/deadline-iosched.c
> @@ -106,7 +106,7 @@ deadline_add_request(struct request_queue *q, struct request *rq)
>  	/*
>  	 * set expire time and add to fifo list
>  	 */
> -	rq_set_fifo_time(rq, jiffies + dd->fifo_expire[data_dir]);
> +	rq->fifo_time = jiffies + dd->fifo_expire[data_dir];
>  	list_add_tail(&rq->queuelist, &dd->fifo_list[data_dir]);
>  }
>  
> @@ -174,9 +174,9 @@ deadline_merged_requests(struct request_queue *q, struct request *req,
>  	 * and move into next position (next will be deleted) in fifo
>  	 */
>  	if (!list_empty(&req->queuelist) && !list_empty(&next->queuelist)) {
> -		if (time_before(rq_fifo_time(next), rq_fifo_time(req))) {
> +		if (time_before(next->fifo_time, req->fifo_time)) {
>  			list_move(&req->queuelist, &next->queuelist);
> -			rq_set_fifo_time(req, rq_fifo_time(next));
> +			req->fifo_time = next->fifo_time;
>  		}
>  	}
>  
> @@ -230,7 +230,7 @@ static inline int deadline_check_fifo(struct deadline_data *dd, int ddir)
>  	/*
>  	 * rq is expired!
>  	 */
> -	if (time_after_eq(jiffies, rq_fifo_time(rq)))
> +	if (time_after_eq(jiffies, rq->fifo_time))
>  		return 1;
>  
>  	return 0;
> diff --git a/include/linux/blkdev.h b/include/linux/blkdev.h
> index 8678c43..dda98e3 100644
> --- a/include/linux/blkdev.h
> +++ b/include/linux/blkdev.h
> @@ -99,6 +99,7 @@ struct request {
>  	union {
>  		struct call_single_data csd;
>  		struct work_struct mq_flush_data;
> +		unsigned long fifo_time;
>  	};
>  
>  	struct request_queue *q;
> diff --git a/include/linux/elevator.h b/include/linux/elevator.h
> index 306dd8c..0b87f4e 100644
> --- a/include/linux/elevator.h
> +++ b/include/linux/elevator.h
> @@ -202,17 +202,9 @@ enum {
>  #define rq_end_sector(rq)	(blk_rq_pos(rq) + blk_rq_sectors(rq))
>  #define rb_entry_rq(node)	rb_entry((node), struct request, rb_node)
>  
> -/*
> - * Hack to reuse the csd.list list_head as the fifo time holder while
> - * the request is in the io scheduler. Saves an unsigned long in rq.
> - */
> -#define rq_fifo_time(rq)	((unsigned long) (rq)->csd.list.next)
> -#define rq_set_fifo_time(rq,exp)	((rq)->csd.list.next = (void *) (exp))
>  #define rq_entry_fifo(ptr)	list_entry((ptr), struct request, queuelist)
> -#define rq_fifo_clear(rq)	do {		\
> -	list_del_init(&(rq)->queuelist);	\
> -	INIT_LIST_HEAD(&(rq)->csd.list);	\
> -	} while (0)
> +//CHECKME twice
> +#define rq_fifo_clear(rq)	list_del_init(&(rq)->queuelist);
>  
>  #else /* CONFIG_BLOCK */
>  
> -- 
> 1.8.3.1
> 
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

Re: [PATCH 1/9] block: Stop abusing csd.list for fifo_time

[Frederic Weisbecker]

On Mon, Feb 03, 2014 at 03:48:29PM +0100, Jan Kara wrote:
> On Sat 01-02-14 17:48:27, Frederic Weisbecker wrote:
> > On Mon, Dec 23, 2013 at 09:39:22PM +0100, Jan Kara wrote:
> > > Block layer currently abuses rq->csd.list.next for storing fifo_time.
> > > That is a terrible hack and completely unnecessary as well. Union
> > > achieves the same space saving in a cleaner way.
> > > 
> > > Signed-off-by: Jan Kara <jack@suse.cz>
> > 
> > Hi Jan,
> > 
> > Taken as is, the patch is fine and it builds.
> > But later when I finally get rid of csd->list in a subsequent patch,
> > rq_fifo_clear() callers break the build.
> > 
> > This is because rq_fifo_clear() initialize the csd->list and I'm not
> > sure how to fix that leftover because I am not clear about the purpose
> > of that INIT_LIST_HEAD(): is it to reset fifo time or to prepare for
> > an IPI to be queued?
>   I'm convinced it is there to prepare IPI to be queued. So just removing
> the initialization as you did should be the right thing to do. You can
> easily verify that it is correct - you boot the kernel, switch to 'deadline'
> IO scheduler by doing
>   echo 'deadline' >/sys/block/sda/queue/scheduler
> and then do some IO. If it doesn't blow up, it is correct.

Ok that seem to work :)

> 
> > All in one it looks buggy because if this is to prepare for the IPI,
> > it's useless as csd.list is not a list head but just a node. Otherwise if it
> > is to reset fifo_time it's wrong because INIT_LIST_HEAD doesn't initialize
> > to 0.
>   Yup, I think it is useless.

Ok then I'll apply this change. I'm just moving it to a separate patch to lower
the chance of it being missed.

Thanks.

[PATCH 2/9] block: Stop abusing rq->csd.list in blk-softirq

[Jan Kara]

Abusing rq->csd.list for a list of requests to complete is rather ugly.
Especially since using queuelist should be safe and much cleaner.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 block/blk-softirq.c | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/block/blk-softirq.c b/block/blk-softirq.c
index 57790c1a97eb..7ea5534096d5 100644
--- a/block/blk-softirq.c
+++ b/block/blk-softirq.c
@@ -30,8 +30,8 @@ static void blk_done_softirq(struct softirq_action *h)
 	while (!list_empty(&local_list)) {
 		struct request *rq;
 
-		rq = list_entry(local_list.next, struct request, csd.list);
-		list_del_init(&rq->csd.list);
+		rq = list_entry(local_list.next, struct request, queuelist);
+		list_del_init(&rq->queuelist);
 		rq->q->softirq_done_fn(rq);
 	}
 }
@@ -45,9 +45,9 @@ static void trigger_softirq(void *data)
 
 	local_irq_save(flags);
 	list = this_cpu_ptr(&blk_cpu_done);
-	list_add_tail(&rq->csd.list, list);
+	list_add_tail(&rq->queuelist, list);
 
-	if (list->next == &rq->csd.list)
+	if (list->next == &rq->queuelist)
 		raise_softirq_irqoff(BLOCK_SOFTIRQ);
 
 	local_irq_restore(flags);
@@ -136,7 +136,7 @@ void __blk_complete_request(struct request *req)
 		struct list_head *list;
 do_local:
 		list = this_cpu_ptr(&blk_cpu_done);
-		list_add_tail(&req->csd.list, list);
+		list_add_tail(&req->queuelist, list);
 
 		/*
 		 * if the list only contains our just added request,
@@ -144,7 +144,7 @@ do_local:
 		 * entries there, someone already raised the irq but it
 		 * hasn't run yet.
 		 */
-		if (list->next == &req->csd.list)
+		if (list->next == &req->queuelist)
 			raise_softirq_irqoff(BLOCK_SOFTIRQ);
 	} else if (raise_blk_irq(ccpu, req))
 		goto do_local;
-- 
1.8.1.4

Re: [PATCH 2/9] block: Stop abusing rq->csd.list in blk-softirq

[Frederic Weisbecker]

Hi Jan,

I'm currently working on some cleanups on IPI code too and working on top
of these patches, just have a few comments:

On Mon, Dec 23, 2013 at 09:39:23PM +0100, Jan Kara wrote:
> Abusing rq->csd.list for a list of requests to complete is rather ugly.
> Especially since using queuelist should be safe and much cleaner.

It would be nice to have a few more details that explain why doing so is safe
wrt a block request lifecycle. At least something that tells why rq->queuelist
can't be ever used concurrently by the time we send the IPI and we trigger/raise
the softirq.

> 
> Signed-off-by: Jan Kara <jack@suse.cz>
> ---
>  block/blk-softirq.c | 12 ++++++------
>  1 file changed, 6 insertions(+), 6 deletions(-)
> 
> diff --git a/block/blk-softirq.c b/block/blk-softirq.c
> index 57790c1a97eb..7ea5534096d5 100644
> --- a/block/blk-softirq.c
> +++ b/block/blk-softirq.c
> @@ -30,8 +30,8 @@ static void blk_done_softirq(struct softirq_action *h)
>  	while (!list_empty(&local_list)) {
>  		struct request *rq;
>  
> -		rq = list_entry(local_list.next, struct request, csd.list);
> -		list_del_init(&rq->csd.list);
> +		rq = list_entry(local_list.next, struct request, queuelist);
> +		list_del_init(&rq->queuelist);
>  		rq->q->softirq_done_fn(rq);
>  	}
>  }
> @@ -45,9 +45,9 @@ static void trigger_softirq(void *data)
>  
>  	local_irq_save(flags);
>  	list = this_cpu_ptr(&blk_cpu_done);
> -	list_add_tail(&rq->csd.list, list);
> +	list_add_tail(&rq->queuelist, list);

And given that's an alternate use of rq->queuelist, perhaps add a comment
to unconfuse people.

Thanks.

>  
> -	if (list->next == &rq->csd.list)
> +	if (list->next == &rq->queuelist)
>  		raise_softirq_irqoff(BLOCK_SOFTIRQ);
>  
>  	local_irq_restore(flags);
> @@ -136,7 +136,7 @@ void __blk_complete_request(struct request *req)
>  		struct list_head *list;
>  do_local:
>  		list = this_cpu_ptr(&blk_cpu_done);
> -		list_add_tail(&req->csd.list, list);
> +		list_add_tail(&req->queuelist, list);
>  
>  		/*
>  		 * if the list only contains our just added request,
> @@ -144,7 +144,7 @@ do_local:
>  		 * entries there, someone already raised the irq but it
>  		 * hasn't run yet.
>  		 */
> -		if (list->next == &req->csd.list)
> +		if (list->next == &req->queuelist)
>  			raise_softirq_irqoff(BLOCK_SOFTIRQ);
>  	} else if (raise_blk_irq(ccpu, req))
>  		goto do_local;
> -- 
> 1.8.1.4
> 

Re: [PATCH 2/9] block: Stop abusing rq->csd.list in blk-softirq

[Jan Kara]

  Hi,

On Thu 30-01-14 13:39:18, Frederic Weisbecker wrote:
> I'm currently working on some cleanups on IPI code too and working on top
> of these patches, just have a few comments:
  Great, thanks!

> On Mon, Dec 23, 2013 at 09:39:23PM +0100, Jan Kara wrote:
> > Abusing rq->csd.list for a list of requests to complete is rather ugly.
> > Especially since using queuelist should be safe and much cleaner.
> 
> It would be nice to have a few more details that explain why doing so is safe
> wrt a block request lifecycle. At least something that tells why rq->queuelist
> can't be ever used concurrently by the time we send the IPI and we trigger/raise
> the softirq.
  Sure. Should I send the patch to you with an updated changelog and added
comment you requested?

> > Signed-off-by: Jan Kara <jack@suse.cz>
> > ---
> >  block/blk-softirq.c | 12 ++++++------
> >  1 file changed, 6 insertions(+), 6 deletions(-)
> > 
> > diff --git a/block/blk-softirq.c b/block/blk-softirq.c
> > index 57790c1a97eb..7ea5534096d5 100644
> > --- a/block/blk-softirq.c
> > +++ b/block/blk-softirq.c
> > @@ -30,8 +30,8 @@ static void blk_done_softirq(struct softirq_action *h)
> >  	while (!list_empty(&local_list)) {
> >  		struct request *rq;
> >  
> > -		rq = list_entry(local_list.next, struct request, csd.list);
> > -		list_del_init(&rq->csd.list);
> > +		rq = list_entry(local_list.next, struct request, queuelist);
> > +		list_del_init(&rq->queuelist);
> >  		rq->q->softirq_done_fn(rq);
> >  	}
> >  }
> > @@ -45,9 +45,9 @@ static void trigger_softirq(void *data)
> >  
> >  	local_irq_save(flags);
> >  	list = this_cpu_ptr(&blk_cpu_done);
> > -	list_add_tail(&rq->csd.list, list);
> > +	list_add_tail(&rq->queuelist, list);
> 
> And given that's an alternate use of rq->queuelist, perhaps add a comment
> to unconfuse people.
  Good idea, will do.

								Honza

> >  
> > -	if (list->next == &rq->csd.list)
> > +	if (list->next == &rq->queuelist)
> >  		raise_softirq_irqoff(BLOCK_SOFTIRQ);
> >  
> >  	local_irq_restore(flags);
> > @@ -136,7 +136,7 @@ void __blk_complete_request(struct request *req)
> >  		struct list_head *list;
> >  do_local:
> >  		list = this_cpu_ptr(&blk_cpu_done);
> > -		list_add_tail(&req->csd.list, list);
> > +		list_add_tail(&req->queuelist, list);
> >  
> >  		/*
> >  		 * if the list only contains our just added request,
> > @@ -144,7 +144,7 @@ do_local:
> >  		 * entries there, someone already raised the irq but it
> >  		 * hasn't run yet.
> >  		 */
> > -		if (list->next == &req->csd.list)
> > +		if (list->next == &req->queuelist)
> >  			raise_softirq_irqoff(BLOCK_SOFTIRQ);
> >  	} else if (raise_blk_irq(ccpu, req))
> >  		goto do_local;
> > -- 
> > 1.8.1.4
> > 
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

Re: [PATCH 2/9] block: Stop abusing rq->csd.list in blk-softirq

[Frederic Weisbecker]

On Thu, Jan 30, 2014 at 04:45:23PM +0100, Jan Kara wrote:
>   Hi,
> 
> On Thu 30-01-14 13:39:18, Frederic Weisbecker wrote:
> > I'm currently working on some cleanups on IPI code too and working on top
> > of these patches, just have a few comments:
>   Great, thanks!
> 
> > On Mon, Dec 23, 2013 at 09:39:23PM +0100, Jan Kara wrote:
> > > Abusing rq->csd.list for a list of requests to complete is rather ugly.
> > > Especially since using queuelist should be safe and much cleaner.
> > 
> > It would be nice to have a few more details that explain why doing so is safe
> > wrt a block request lifecycle. At least something that tells why rq->queuelist
> > can't be ever used concurrently by the time we send the IPI and we trigger/raise
> > the softirq.
>   Sure. Should I send the patch to you with an updated changelog and added
> comment you requested?

Yeah that would be nice!

For more precision, I would like to apply these:

1) block: Stop abusing csd.list for fifo_time
2) block: Stop abusing rq->csd.list in blk-softirq
3) kernel: use lockless list for smp_call_function_single()
4) smp: Teach __smp_call_function_single() to check for offline cpus

This is because I need to tweak a bit the IPI code for some nohz
functionnality. I'm not sure about the result but in any case these
llist based cleanups look very nice, so lets try to push them. I'm also
working on some consolidation between __smp_call_function_single()
and smp_call_function_single() since they share almost the same code.

Also this shouldn't conflict with Andrew's tree if he applies these as well
since -mm is based on -next. And the printk part should still go through his
tree I think.

> 
> > > Signed-off-by: Jan Kara <jack@suse.cz>
> > > ---
> > >  block/blk-softirq.c | 12 ++++++------
> > >  1 file changed, 6 insertions(+), 6 deletions(-)
> > > 
> > > diff --git a/block/blk-softirq.c b/block/blk-softirq.c
> > > index 57790c1a97eb..7ea5534096d5 100644
> > > --- a/block/blk-softirq.c
> > > +++ b/block/blk-softirq.c
> > > @@ -30,8 +30,8 @@ static void blk_done_softirq(struct softirq_action *h)
> > >  	while (!list_empty(&local_list)) {
> > >  		struct request *rq;
> > >  
> > > -		rq = list_entry(local_list.next, struct request, csd.list);
> > > -		list_del_init(&rq->csd.list);
> > > +		rq = list_entry(local_list.next, struct request, queuelist);
> > > +		list_del_init(&rq->queuelist);
> > >  		rq->q->softirq_done_fn(rq);
> > >  	}
> > >  }
> > > @@ -45,9 +45,9 @@ static void trigger_softirq(void *data)
> > >  
> > >  	local_irq_save(flags);
> > >  	list = this_cpu_ptr(&blk_cpu_done);
> > > -	list_add_tail(&rq->csd.list, list);
> > > +	list_add_tail(&rq->queuelist, list);
> > 
> > And given that's an alternate use of rq->queuelist, perhaps add a comment
> > to unconfuse people.
>   Good idea, will do.

Thanks!

Re: [PATCH 2/9] block: Stop abusing rq->csd.list in blk-softirq

[Jan Kara]

[-- Attachment #1: Type: text/plain, Size: 2029 bytes --]

On Thu 30-01-14 18:01:20, Frederic Weisbecker wrote:
> On Thu, Jan 30, 2014 at 04:45:23PM +0100, Jan Kara wrote:
> >   Hi,
> > 
> > On Thu 30-01-14 13:39:18, Frederic Weisbecker wrote:
> > > I'm currently working on some cleanups on IPI code too and working on top
> > > of these patches, just have a few comments:
> >   Great, thanks!
> > 
> > > On Mon, Dec 23, 2013 at 09:39:23PM +0100, Jan Kara wrote:
> > > > Abusing rq->csd.list for a list of requests to complete is rather ugly.
> > > > Especially since using queuelist should be safe and much cleaner.
> > > 
> > > It would be nice to have a few more details that explain why doing so is safe
> > > wrt a block request lifecycle. At least something that tells why rq->queuelist
> > > can't be ever used concurrently by the time we send the IPI and we trigger/raise
> > > the softirq.
> >   Sure. Should I send the patch to you with an updated changelog and added
> > comment you requested?
> 
> Yeah that would be nice!
  OK, the updated patch is attached.

> For more precision, I would like to apply these:
> 
> 1) block: Stop abusing csd.list for fifo_time
> 2) block: Stop abusing rq->csd.list in blk-softirq
> 3) kernel: use lockless list for smp_call_function_single()
> 4) smp: Teach __smp_call_function_single() to check for offline cpus
> 
> This is because I need to tweak a bit the IPI code for some nohz
> functionnality. I'm not sure about the result but in any case these
> llist based cleanups look very nice, so lets try to push them. I'm also
> working on some consolidation between __smp_call_function_single()
> and smp_call_function_single() since they share almost the same code.
> 
> Also this shouldn't conflict with Andrew's tree if he applies these as well
> since -mm is based on -next. And the printk part should still go through his
> tree I think.
  Sure, that should be no problem. Jens might have the patch somewhere in
the linux-block.git but any conflict should be easy to resolve.

								Honza
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

[-- Attachment #2: 0001-block-Stop-abusing-rq-csd.list-in-blk-softirq.patch --]
[-- Type: text/x-patch, Size: 2297 bytes --]

>From a7782fa4cee73a0581eded978eacf52cb0db1ec7 Mon Sep 17 00:00:00 2001
From: Jan Kara <jack@suse.cz>
Date: Tue, 17 Dec 2013 23:47:41 +0100
Subject: [PATCH] block: Stop abusing rq->csd.list in blk-softirq

Abusing rq->csd.list for a list of requests to complete is rather ugly.
We use rq->queuelist instead which is much cleaner. It is safe because
queuelist is used by the block layer only for requests waiting to be
submitted to a device. Thus it is unused when irq reports the request IO
is finished.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 block/blk-softirq.c | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/block/blk-softirq.c b/block/blk-softirq.c
index 57790c1a97eb..b5c37d96cf0e 100644
--- a/block/blk-softirq.c
+++ b/block/blk-softirq.c
@@ -30,8 +30,8 @@ static void blk_done_softirq(struct softirq_action *h)
 	while (!list_empty(&local_list)) {
 		struct request *rq;
 
-		rq = list_entry(local_list.next, struct request, csd.list);
-		list_del_init(&rq->csd.list);
+		rq = list_entry(local_list.next, struct request, queuelist);
+		list_del_init(&rq->queuelist);
 		rq->q->softirq_done_fn(rq);
 	}
 }
@@ -45,9 +45,14 @@ static void trigger_softirq(void *data)
 
 	local_irq_save(flags);
 	list = this_cpu_ptr(&blk_cpu_done);
-	list_add_tail(&rq->csd.list, list);
+	/*
+	 * We reuse queuelist for a list of requests to process. Since the
+	 * queuelist is used by the block layer only for requests waiting to be
+	 * submitted to the device it is unused now.
+	 */
+	list_add_tail(&rq->queuelist, list);
 
-	if (list->next == &rq->csd.list)
+	if (list->next == &rq->queuelist)
 		raise_softirq_irqoff(BLOCK_SOFTIRQ);
 
 	local_irq_restore(flags);
@@ -136,7 +141,7 @@ void __blk_complete_request(struct request *req)
 		struct list_head *list;
 do_local:
 		list = this_cpu_ptr(&blk_cpu_done);
-		list_add_tail(&req->csd.list, list);
+		list_add_tail(&req->queuelist, list);
 
 		/*
 		 * if the list only contains our just added request,
@@ -144,7 +149,7 @@ do_local:
 		 * entries there, someone already raised the irq but it
 		 * hasn't run yet.
 		 */
-		if (list->next == &req->csd.list)
+		if (list->next == &req->queuelist)
 			raise_softirq_irqoff(BLOCK_SOFTIRQ);
 	} else if (raise_blk_irq(ccpu, req))
 		goto do_local;
-- 
1.8.1.4

Re: [PATCH 2/9] block: Stop abusing rq->csd.list in blk-softirq

[Frederic Weisbecker]

On Thu, Jan 30, 2014 at 11:12:41PM +0100, Jan Kara wrote:
> On Thu 30-01-14 18:01:20, Frederic Weisbecker wrote:
> > On Thu, Jan 30, 2014 at 04:45:23PM +0100, Jan Kara wrote:
> > >   Hi,
> > > 
> > > On Thu 30-01-14 13:39:18, Frederic Weisbecker wrote:
> > > > I'm currently working on some cleanups on IPI code too and working on top
> > > > of these patches, just have a few comments:
> > >   Great, thanks!
> > > 
> > > > On Mon, Dec 23, 2013 at 09:39:23PM +0100, Jan Kara wrote:
> > > > > Abusing rq->csd.list for a list of requests to complete is rather ugly.
> > > > > Especially since using queuelist should be safe and much cleaner.
> > > > 
> > > > It would be nice to have a few more details that explain why doing so is safe
> > > > wrt a block request lifecycle. At least something that tells why rq->queuelist
> > > > can't be ever used concurrently by the time we send the IPI and we trigger/raise
> > > > the softirq.
> > >   Sure. Should I send the patch to you with an updated changelog and added
> > > comment you requested?
> > 
> > Yeah that would be nice!
>   OK, the updated patch is attached.

Applied, thanks!

Note that the llist use in smp.c patch from Christoph has been merged
upstream today. But it keeps list_head in a union so I applied your changes
that:

1) remove list_head from smp.c
2) use llist_for_each_entry_safe()

in seperate delta patches. Anyway, I'll send the series soonish.

[PATCH 3/9] kernel: use lockless list for smp_call_function_single()

[Jan Kara]

From: Christoph Hellwig <hch@lst.de>

Make smp_call_function_single and friends more efficient by using
a lockless list.

Signed-off-by: Christoph Hellwig <hch@lst.de>
Signed-off-by: Jan Kara <jack@suse.cz>
---
 include/linux/elevator.h |  2 +-
 include/linux/smp.h      |  3 ++-
 kernel/smp.c             | 51 ++++++++++--------------------------------------
 3 files changed, 13 insertions(+), 43 deletions(-)

diff --git a/include/linux/elevator.h b/include/linux/elevator.h
index 0bdfd46f4735..06860e2a25c3 100644
--- a/include/linux/elevator.h
+++ b/include/linux/elevator.h
@@ -205,7 +205,7 @@ enum {
 #define rq_entry_fifo(ptr)	list_entry((ptr), struct request, queuelist)
 #define rq_fifo_clear(rq)	do {		\
 	list_del_init(&(rq)->queuelist);	\
-	INIT_LIST_HEAD(&(rq)->csd.list);	\
+	(rq)->csd.llist.next = NULL;		\
 	} while (0)
 
 #else /* CONFIG_BLOCK */
diff --git a/include/linux/smp.h b/include/linux/smp.h
index 5da22ee42e16..9a1b8ba05924 100644
--- a/include/linux/smp.h
+++ b/include/linux/smp.h
@@ -11,12 +11,13 @@
 #include <linux/list.h>
 #include <linux/cpumask.h>
 #include <linux/init.h>
+#include <linux/llist.h>
 
 extern void cpu_idle(void);
 
 typedef void (*smp_call_func_t)(void *info);
 struct call_single_data {
-	struct list_head list;
+	struct llist_node llist;
 	smp_call_func_t func;
 	void *info;
 	u16 flags;
diff --git a/kernel/smp.c b/kernel/smp.c
index bd9f94028838..47b415e16c24 100644
--- a/kernel/smp.c
+++ b/kernel/smp.c
@@ -28,12 +28,7 @@ struct call_function_data {
 
 static DEFINE_PER_CPU_SHARED_ALIGNED(struct call_function_data, cfd_data);
 
-struct call_single_queue {
-	struct list_head	list;
-	raw_spinlock_t		lock;
-};
-
-static DEFINE_PER_CPU_SHARED_ALIGNED(struct call_single_queue, call_single_queue);
+static DEFINE_PER_CPU_SHARED_ALIGNED(struct llist_head, call_single_queue);
 
 static int
 hotplug_cfd(struct notifier_block *nfb, unsigned long action, void *hcpu)
@@ -85,12 +80,8 @@ void __init call_function_init(void)
 	void *cpu = (void *)(long)smp_processor_id();
 	int i;
 
-	for_each_possible_cpu(i) {
-		struct call_single_queue *q = &per_cpu(call_single_queue, i);
-
-		raw_spin_lock_init(&q->lock);
-		INIT_LIST_HEAD(&q->list);
-	}
+	for_each_possible_cpu(i)
+		init_llist_head(&per_cpu(call_single_queue, i));
 
 	hotplug_cfd(&hotplug_cfd_notifier, CPU_UP_PREPARE, cpu);
 	register_cpu_notifier(&hotplug_cfd_notifier);
@@ -141,18 +132,9 @@ static void csd_unlock(struct call_single_data *csd)
  */
 static void generic_exec_single(int cpu, struct call_single_data *csd, int wait)
 {
-	struct call_single_queue *dst = &per_cpu(call_single_queue, cpu);
-	unsigned long flags;
-	int ipi;
-
 	if (wait)
 		csd->flags |= CSD_FLAG_WAIT;
 
-	raw_spin_lock_irqsave(&dst->lock, flags);
-	ipi = list_empty(&dst->list);
-	list_add_tail(&csd->list, &dst->list);
-	raw_spin_unlock_irqrestore(&dst->lock, flags);
-
 	/*
 	 * The list addition should be visible before sending the IPI
 	 * handler locks the list to pull the entry off it because of
@@ -164,7 +146,7 @@ static void generic_exec_single(int cpu, struct call_single_data *csd, int wait)
 	 * locking and barrier primitives. Generic code isn't really
 	 * equipped to do the right thing...
 	 */
-	if (ipi)
+	if (llist_add(&csd->llist, &per_cpu(call_single_queue, cpu)))
 		arch_send_call_function_single_ipi(cpu);
 
 	if (wait)
@@ -177,26 +159,19 @@ static void generic_exec_single(int cpu, struct call_single_data *csd, int wait)
  */
 void generic_smp_call_function_single_interrupt(void)
 {
-	struct call_single_queue *q = &__get_cpu_var(call_single_queue);
-	LIST_HEAD(list);
+	struct llist_node *entry;
+	struct call_single_data *csd, *csd_next;
 
 	/*
 	 * Shouldn't receive this interrupt on a cpu that is not yet online.
 	 */
 	WARN_ON_ONCE(!cpu_online(smp_processor_id()));
 
-	raw_spin_lock(&q->lock);
-	list_replace_init(&q->list, &list);
-	raw_spin_unlock(&q->lock);
-
-	while (!list_empty(&list)) {
-		struct call_single_data *csd;
-
-		csd = list_entry(list.next, struct call_single_data, list);
-		list_del(&csd->list);
+	entry = llist_del_all(&__get_cpu_var(call_single_queue));
+	entry = llist_reverse_order(entry);
 
+	llist_for_each_entry_safe(csd, csd_next, entry, llist) {
 		csd->func(csd->info);
-
 		csd_unlock(csd);
 	}
 }
@@ -411,17 +386,11 @@ void smp_call_function_many(const struct cpumask *mask,
 
 	for_each_cpu(cpu, cfd->cpumask) {
 		struct call_single_data *csd = per_cpu_ptr(cfd->csd, cpu);
-		struct call_single_queue *dst =
-					&per_cpu(call_single_queue, cpu);
-		unsigned long flags;
 
 		csd_lock(csd);
 		csd->func = func;
 		csd->info = info;
-
-		raw_spin_lock_irqsave(&dst->lock, flags);
-		list_add_tail(&csd->list, &dst->list);
-		raw_spin_unlock_irqrestore(&dst->lock, flags);
+		llist_add(&csd->llist, &per_cpu(call_single_queue, cpu));
 	}
 
 	/* Send a message to all CPUs in the map */
-- 
1.8.1.4

Re: [PATCH 3/9] kernel: use lockless list for smp_call_function_single()

[Frederic Weisbecker]

2013/12/23 Jan Kara <jack@suse.cz>:
> From: Christoph Hellwig <hch@lst.de>
>
> Make smp_call_function_single and friends more efficient by using
> a lockless list.
>
> Signed-off-by: Christoph Hellwig <hch@lst.de>
> Signed-off-by: Jan Kara <jack@suse.cz>

FWIW, I really like that patch.

Reviewed-by: Frederic Weisbecker <fweisbec@gmail.com>

[PATCH 4/9] smp: Teach __smp_call_function_single() to check for offline cpus

[Jan Kara]

Align __smp_call_function_single() with smp_call_function_single() so
that it also checks whether requested cpu is still online.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 include/linux/smp.h |  3 +--
 kernel/smp.c        | 11 +++++++----
 kernel/up.c         |  5 +++--
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/include/linux/smp.h b/include/linux/smp.h
index 9a1b8ba05924..1e8c72100eda 100644
--- a/include/linux/smp.h
+++ b/include/linux/smp.h
@@ -50,8 +50,7 @@ void on_each_cpu_cond(bool (*cond_func)(int cpu, void *info),
 		smp_call_func_t func, void *info, bool wait,
 		gfp_t gfp_flags);
 
-void __smp_call_function_single(int cpuid, struct call_single_data *data,
-				int wait);
+int __smp_call_function_single(int cpu, struct call_single_data *csd, int wait);
 
 #ifdef CONFIG_SMP
 
diff --git a/kernel/smp.c b/kernel/smp.c
index 47b415e16c24..2efcfa1d11ee 100644
--- a/kernel/smp.c
+++ b/kernel/smp.c
@@ -284,18 +284,18 @@ EXPORT_SYMBOL_GPL(smp_call_function_any);
 /**
  * __smp_call_function_single(): Run a function on a specific CPU
  * @cpu: The CPU to run on.
- * @data: Pre-allocated and setup data structure
+ * @csd: Pre-allocated and setup data structure
  * @wait: If true, wait until function has completed on specified CPU.
  *
  * Like smp_call_function_single(), but allow caller to pass in a
  * pre-allocated data structure. Useful for embedding @data inside
  * other structures, for instance.
  */
-void __smp_call_function_single(int cpu, struct call_single_data *csd,
-				int wait)
+int __smp_call_function_single(int cpu, struct call_single_data *csd, int wait)
 {
 	unsigned int this_cpu;
 	unsigned long flags;
+	int err = 0;
 
 	this_cpu = get_cpu();
 	/*
@@ -311,11 +311,14 @@ void __smp_call_function_single(int cpu, struct call_single_data *csd,
 		local_irq_save(flags);
 		csd->func(csd->info);
 		local_irq_restore(flags);
-	} else {
+	} else if ((unsigned)cpu < nr_cpu_ids && cpu_online(cpu)) {
 		csd_lock(csd);
 		generic_exec_single(cpu, csd, wait);
+	} else {
+		err = -ENXIO;	/* CPU not online */
 	}
 	put_cpu();
+	return err;
 }
 EXPORT_SYMBOL_GPL(__smp_call_function_single);
 
diff --git a/kernel/up.c b/kernel/up.c
index 509403e3fbc6..cdf03d16840e 100644
--- a/kernel/up.c
+++ b/kernel/up.c
@@ -22,14 +22,15 @@ int smp_call_function_single(int cpu, void (*func) (void *info), void *info,
 }
 EXPORT_SYMBOL(smp_call_function_single);
 
-void __smp_call_function_single(int cpu, struct call_single_data *csd,
-				int wait)
+int __smp_call_function_single(int cpu, struct call_single_data *csd,
+			       int wait)
 {
 	unsigned long flags;
 
 	local_irq_save(flags);
 	csd->func(csd->info);
 	local_irq_restore(flags);
+	return 0;
 }
 EXPORT_SYMBOL(__smp_call_function_single);
 
-- 
1.8.1.4

Re: [PATCH 4/9] smp: Teach __smp_call_function_single() to check for offline cpus

[Steven Rostedt]

On Mon, 23 Dec 2013 21:39:25 +0100
Jan Kara <jack@suse.cz> wrote:

> Align __smp_call_function_single() with smp_call_function_single() so
> that it also checks whether requested cpu is still online.
> 
> Signed-off-by: Jan Kara <jack@suse.cz>

Reviewed-by: Steven Rostedt <rostedt@goodmis.org>

-- Steve

[PATCH 5/9] smp: Provide __smp_call_function_any()

[Jan Kara]

Provide function to call given function on any cpu from a given cpu mask
while providing own csd structure.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 include/linux/smp.h |  9 +++++++
 kernel/smp.c        | 67 +++++++++++++++++++++++++++++++++++++++--------------
 2 files changed, 58 insertions(+), 18 deletions(-)

diff --git a/include/linux/smp.h b/include/linux/smp.h
index 1e8c72100eda..37aa794a93ce 100644
--- a/include/linux/smp.h
+++ b/include/linux/smp.h
@@ -100,6 +100,8 @@ void smp_call_function_many(const struct cpumask *mask,
 
 int smp_call_function_any(const struct cpumask *mask,
 			  smp_call_func_t func, void *info, int wait);
+int __smp_call_function_any(const struct cpumask *mask,
+			    struct call_single_data *csd, int wait);
 
 void kick_all_cpus_sync(void);
 
@@ -149,6 +151,13 @@ smp_call_function_any(const struct cpumask *mask, smp_call_func_t func,
 	return smp_call_function_single(0, func, info, wait);
 }
 
+static inline int
+__smp_call_function_any(const struct cpumask *mask,
+			struct call_single_data *csd, int wait)
+{
+	return __smp_call_function_single(0, csd, wait);
+}
+
 static inline void kick_all_cpus_sync(void) {  }
 
 #endif /* !SMP */
diff --git a/kernel/smp.c b/kernel/smp.c
index 2efcfa1d11ee..4c44554ff5b6 100644
--- a/kernel/smp.c
+++ b/kernel/smp.c
@@ -238,6 +238,27 @@ int smp_call_function_single(int cpu, smp_call_func_t func, void *info,
 }
 EXPORT_SYMBOL(smp_call_function_single);
 
+static unsigned int pick_any_cpu(const struct cpumask *mask)
+{
+	unsigned int cpu = smp_processor_id();
+	const struct cpumask *nodemask;
+
+	/* Try for same CPU (cheapest) */
+	if (cpumask_test_cpu(cpu, mask))
+		return cpu;
+
+	/* Try for same node. */
+	nodemask = cpumask_of_node(cpu_to_node(cpu));
+	for (cpu = cpumask_first_and(nodemask, mask); cpu < nr_cpu_ids;
+	     cpu = cpumask_next_and(cpu, nodemask, mask)) {
+		if (cpu_online(cpu))
+			return cpu;
+	}
+
+	/* Any online will do */
+	return cpumask_any_and(mask, cpu_online_mask);
+}
+
 /*
  * smp_call_function_any - Run a function on any of the given cpus
  * @mask: The mask of cpus it can run on.
@@ -256,27 +277,13 @@ int smp_call_function_any(const struct cpumask *mask,
 			  smp_call_func_t func, void *info, int wait)
 {
 	unsigned int cpu;
-	const struct cpumask *nodemask;
 	int ret;
 
-	/* Try for same CPU (cheapest) */
-	cpu = get_cpu();
-	if (cpumask_test_cpu(cpu, mask))
-		goto call;
-
-	/* Try for same node. */
-	nodemask = cpumask_of_node(cpu_to_node(cpu));
-	for (cpu = cpumask_first_and(nodemask, mask); cpu < nr_cpu_ids;
-	     cpu = cpumask_next_and(cpu, nodemask, mask)) {
-		if (cpu_online(cpu))
-			goto call;
-	}
-
-	/* Any online will do: smp_call_function_single handles nr_cpu_ids. */
-	cpu = cpumask_any_and(mask, cpu_online_mask);
-call:
+	preempt_disable();
+	cpu = pick_any_cpu(mask);
+	/* smp_call_function_single handles nr_cpu_ids. */
 	ret = smp_call_function_single(cpu, func, info, wait);
-	put_cpu();
+	preempt_enable();
 	return ret;
 }
 EXPORT_SYMBOL_GPL(smp_call_function_any);
@@ -322,6 +329,30 @@ int __smp_call_function_single(int cpu, struct call_single_data *csd, int wait)
 }
 EXPORT_SYMBOL_GPL(__smp_call_function_single);
 
+/*
+ * __smp_call_function_any - Run a function on any of the given cpus
+ * @mask: The mask of cpus it can run on.
+ * @csd: Pre-allocated and setup data structure
+ * @wait: If true, wait until function has completed.
+ *
+ * Like smp_call_function_any() but allow caller to pass in a pre-allocated
+ * data structure.
+ */
+int __smp_call_function_any(const struct cpumask *mask,
+			    struct call_single_data *csd, int wait)
+{
+	unsigned int cpu;
+	int ret;
+
+	preempt_disable();
+	cpu = pick_any_cpu(mask);
+	/* smp_call_function_single handles nr_cpu_ids. */
+	ret = __smp_call_function_single(cpu, csd, wait);
+	preempt_enable();
+	return ret;
+}
+EXPORT_SYMBOL_GPL(__smp_call_function_any);
+
 /**
  * smp_call_function_many(): Run a function on a set of other CPUs.
  * @mask: The set of cpus to run on (only runs on online subset).
-- 
1.8.1.4

Re: [PATCH 5/9] smp: Provide __smp_call_function_any()

[Steven Rostedt]

On Mon, 23 Dec 2013 21:39:26 +0100
Jan Kara <jack@suse.cz> wrote:

> Provide function to call given function on any cpu from a given cpu mask
> while providing own csd structure.
> 
> Signed-off-by: Jan Kara <jack@suse.cz>

Reviewed-by: Steven Rostedt <rostedt@goodmis.org>

-- Steve

[PATCH 6/9] printk: Release lockbuf_lock before calling console_trylock_for_printk()

[Jan Kara]

There's no reason to hold lockbuf_lock when entering
console_trylock_for_printk(). The first thing this function does is
calling down_trylock(console_sem) and if that fails it immediately
unlocks lockbuf_lock. So lockbuf_lock isn't needed for that branch.
When down_trylock() succeeds, the rest of console_trylock() is OK
without lockbuf_lock (it is called without it from other places), and
the only remaining thing in console_trylock_for_printk() is
can_use_console() call. For that call console_sem is enough (it
iterates all consoles and checks CON_ANYTIME flag).

So we drop logbuf_lock before entering console_trylock_for_printk()
which simplifies the code.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 kernel/printk/printk.c | 49 +++++++++++++++++--------------------------------
 1 file changed, 17 insertions(+), 32 deletions(-)

diff --git a/kernel/printk/printk.c b/kernel/printk/printk.c
index be7c86bae576..72ec3637115f 100644
--- a/kernel/printk/printk.c
+++ b/kernel/printk/printk.c
@@ -250,9 +250,6 @@ static char __log_buf[__LOG_BUF_LEN] __aligned(LOG_ALIGN);
 static char *log_buf = __log_buf;
 static u32 log_buf_len = __LOG_BUF_LEN;
 
-/* cpu currently holding logbuf_lock */
-static volatile unsigned int logbuf_cpu = UINT_MAX;
-
 /* human readable text of the record */
 static char *log_text(const struct printk_log *msg)
 {
@@ -1338,36 +1335,22 @@ static inline int can_use_console(unsigned int cpu)
  * messages from a 'printk'. Return true (and with the
  * console_lock held, and 'console_locked' set) if it
  * is successful, false otherwise.
- *
- * This gets called with the 'logbuf_lock' spinlock held and
- * interrupts disabled. It should return with 'lockbuf_lock'
- * released but interrupts still disabled.
  */
 static int console_trylock_for_printk(unsigned int cpu)
-	__releases(&logbuf_lock)
 {
-	int retval = 0, wake = 0;
-
-	if (console_trylock()) {
-		retval = 1;
-
-		/*
-		 * If we can't use the console, we need to release
-		 * the console semaphore by hand to avoid flushing
-		 * the buffer. We need to hold the console semaphore
-		 * in order to do this test safely.
-		 */
-		if (!can_use_console(cpu)) {
-			console_locked = 0;
-			wake = 1;
-			retval = 0;
-		}
-	}
-	logbuf_cpu = UINT_MAX;
-	raw_spin_unlock(&logbuf_lock);
-	if (wake)
+	if (!console_trylock())
+		return 0;
+	/*
+	 * If we can't use the console, we need to release the console
+	 * semaphore by hand to avoid flushing the buffer. We need to hold the
+	 * console semaphore in order to do this test safely.
+	 */
+	if (!can_use_console(cpu)) {
+		console_locked = 0;
 		up(&console_sem);
-	return retval;
+		return 0;
+	}
+	return 1;
 }
 
 int printk_delay_msec __read_mostly;
@@ -1500,6 +1483,9 @@ asmlinkage int vprintk_emit(int facility, int level,
 	unsigned long flags;
 	int this_cpu;
 	int printed_len = 0;
+	/* cpu currently holding logbuf_lock in this function */
+	static volatile unsigned int logbuf_cpu = UINT_MAX;
+
 
 	boot_delay_msec(level);
 	printk_delay();
@@ -1612,13 +1598,12 @@ asmlinkage int vprintk_emit(int facility, int level,
 	}
 	printed_len += text_len;
 
+	logbuf_cpu = UINT_MAX;
+	raw_spin_unlock(&logbuf_lock);
 	/*
 	 * Try to acquire and then immediately release the console semaphore.
 	 * The release will print out buffers and wake up /dev/kmsg and syslog()
 	 * users.
-	 *
-	 * The console_trylock_for_printk() function will release 'logbuf_lock'
-	 * regardless of whether it actually gets the console semaphore or not.
 	 */
 	if (console_trylock_for_printk(this_cpu))
 		console_unlock();
-- 
1.8.1.4

Re: [PATCH 6/9] printk: Release lockbuf_lock before calling console_trylock_for_printk()

[Steven Rostedt]

On Mon, 23 Dec 2013 21:39:27 +0100
Jan Kara <jack@suse.cz> wrote:

> There's no reason to hold lockbuf_lock when entering
> console_trylock_for_printk(). The first thing this function does is
> calling down_trylock(console_sem) and if that fails it immediately
> unlocks lockbuf_lock. So lockbuf_lock isn't needed for that branch.
> When down_trylock() succeeds, the rest of console_trylock() is OK
> without lockbuf_lock (it is called without it from other places), and
> the only remaining thing in console_trylock_for_printk() is
> can_use_console() call. For that call console_sem is enough (it
> iterates all consoles and checks CON_ANYTIME flag).
> 
> So we drop logbuf_lock before entering console_trylock_for_printk()
> which simplifies the code.

I'm very nervous about this change. The interlocking between console
lock and logbuf_lock seems to be very subtle. Especially the comment
where logbuf_lock is defined:

/*
 * The logbuf_lock protects kmsg buffer, indices, counters. It is also
 * used in interesting ways to provide interlocking in console_unlock();
 */

Unfortunately, it does not specify what those "interesting ways" are.


Now what I think this does is to make sure whoever wrote to the logbuf
first, does the flushing. With your change we now have:

	CPU 0				CPU 1
	-----				-----
   printk("blah");
   lock(logbuf_lock);

				printk("bazinga!");
				lock(logbuf_lock);
				<blocked>

   unlock(logbuf_lock);
   < NMI comes in delays CPU>

				<get logbuf_lock>
				unlock(logbuf_lock)
				console_trylock_for_printk()
				console_unlock();
				< dumps output >

  
Now is this a bad thing? I don't know. But the current locking will
make sure that the first writer into logbuf_lock gets to do the
dumping, and all the others will just add onto that writer.

Your change now lets the second or third or whatever writer into printk
be the one that dumps the log.

Again, this may not be a big deal, but as printk is such a core part of
the Linux kernel, and this is a very subtle change, I rather be very
cautious here and try to think what can go wrong when this happens.

-- Steve

Re: [PATCH 6/9] printk: Release lockbuf_lock before calling console_trylock_for_printk()

[Jan Kara]

On Thu 02-01-14 20:53:05, Steven Rostedt wrote:
> On Mon, 23 Dec 2013 21:39:27 +0100
> Jan Kara <jack@suse.cz> wrote:
> 
> > There's no reason to hold lockbuf_lock when entering
> > console_trylock_for_printk(). The first thing this function does is
> > calling down_trylock(console_sem) and if that fails it immediately
> > unlocks lockbuf_lock. So lockbuf_lock isn't needed for that branch.
> > When down_trylock() succeeds, the rest of console_trylock() is OK
> > without lockbuf_lock (it is called without it from other places), and
> > the only remaining thing in console_trylock_for_printk() is
> > can_use_console() call. For that call console_sem is enough (it
> > iterates all consoles and checks CON_ANYTIME flag).
> > 
> > So we drop logbuf_lock before entering console_trylock_for_printk()
> > which simplifies the code.
> 
> I'm very nervous about this change. The interlocking between console
> lock and logbuf_lock seems to be very subtle. Especially the comment
> where logbuf_lock is defined:
> 
> /*
>  * The logbuf_lock protects kmsg buffer, indices, counters. It is also
>  * used in interesting ways to provide interlocking in console_unlock();
>  */
> 
> Unfortunately, it does not specify what those "interesting ways" are.
  Hum, yes. So I was digging in history and the comment was added by Andrew
Morton in early 2002 when converting console_lock to console_sem +
logbuf_lock. I'm sure he remembers all the details ;) It is part of commit
a880f45a48be2956d2c78a839c472287d54435c1 in linux-history.git.

Looking into that commit I think the comment refers to the following trick:
printk()
	/* This stops the holder of console_sem just where we want him */
	spin_lock_irqsave(&logbuf_lock, flags);
	...
	if (!down_trylock(&console_sem)) {
		/*
		 * We own the drivers.  We can drop the spinlock and let
		 * release_console_sem() print the text
		 */
		spin_unlock_irqrestore(&logbuf_lock, flags);
		...
	} else {
		/*
		 * Someone else owns the drivers.  We drop the spinlock, which
		 * allows the semaphore holder to proceed and to call the
		 * console drivers with the output which we just produced.
		 */
		spin_unlock_irqrestore(&logbuf_lock, flags);
	}

release_console_sem() (equivalent of today's console_unlock()):
	for ( ; ; ) {
		spin_lock_irqsave(&logbuf_lock, flags);
...
		if (con_start == log_end)
			break;                  /* Nothing to print */
...
		spin_unlock_irqrestore(&logbuf_lock, flags);
		call_console_drivers(_con_start, _log_end);
	}
	up(&console_sem);
	spin_unlock_irqrestore(&logbuf_lock, flags);

This interesting combination of console_sem and logbuf_lock locking makes
sure we cannot exit the loop in release_console_sem() before printk()
decides whether it should do printing or not. So the appended message gets
reliably printed either by current holder of console_sem or by CPU in
printk(). Apparently this trick got broken sometime later and then fixed up
again by rechecking 'console_seq != log_next_seq' after releasing
console_sem. So I think the comment isn't valid anymore.

> Now what I think this does is to make sure whoever wrote to the logbuf
> first, does the flushing. With your change we now have:
> 
> 	CPU 0				CPU 1
> 	-----				-----
>    printk("blah");
>    lock(logbuf_lock);
> 
> 				printk("bazinga!");
> 				lock(logbuf_lock);
> 				<blocked>
> 
>    unlock(logbuf_lock);
>    < NMI comes in delays CPU>
> 
> 				<get logbuf_lock>
> 				unlock(logbuf_lock)
> 				console_trylock_for_printk()
> 				console_unlock();
> 				< dumps output >
> 
>   
> Now is this a bad thing? I don't know. But the current locking will
> make sure that the first writer into logbuf_lock gets to do the
> dumping, and all the others will just add onto that writer.
> 
> Your change now lets the second or third or whatever writer into printk
> be the one that dumps the log.
  I agree and I admit I didn't think about this. But given how printk
buffering works this doesn't seem to be any problem at all. But I can
comment about this in the changelog.

> Again, this may not be a big deal, but as printk is such a core part of
> the Linux kernel, and this is a very subtle change, I rather be very
> cautious here and try to think what can go wrong when this happens.
  Sure. Thanks for review!

								Honza
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

[PATCH 7/9] printk: Enable interrupts before calling console_trylock_for_printk()

[Jan Kara]

We need interrupts disabled when calling console_trylock_for_printk()
only so that cpu id we pass to can_use_console() remains valid (for
other things console_sem provides all the exclusion we need and
deadlocks on console_sem due to interrupts are impossible because we use
down_trylock()).  However if we are rescheduled, we are guaranteed to
run on an online cpu so we can easily just get the cpu id in
can_use_console().

We can loose a bit of performance when we enable interrupts in
vprintk_emit() and then disable them again in console_unlock() but OTOH
it can somewhat reduce interrupt latency caused by console_unlock()
especially since later in the patch series we will want to spin on
console_sem in console_trylock_for_printk().

Signed-off-by: Jan Kara <jack@suse.cz>
---
 kernel/printk/printk.c | 29 ++++++++++++++++++-----------
 1 file changed, 18 insertions(+), 11 deletions(-)

diff --git a/kernel/printk/printk.c b/kernel/printk/printk.c
index 72ec3637115f..ab0d24a9ab7c 100644
--- a/kernel/printk/printk.c
+++ b/kernel/printk/printk.c
@@ -1320,10 +1320,9 @@ static int have_callable_console(void)
 /*
  * Can we actually use the console at this time on this cpu?
  *
- * Console drivers may assume that per-cpu resources have
- * been allocated. So unless they're explicitly marked as
- * being able to cope (CON_ANYTIME) don't call them until
- * this CPU is officially up.
+ * Console drivers may assume that per-cpu resources have been allocated. So
+ * unless they're explicitly marked as being able to cope (CON_ANYTIME) don't
+ * call them until this CPU is officially up.
  */
 static inline int can_use_console(unsigned int cpu)
 {
@@ -1336,8 +1335,10 @@ static inline int can_use_console(unsigned int cpu)
  * console_lock held, and 'console_locked' set) if it
  * is successful, false otherwise.
  */
-static int console_trylock_for_printk(unsigned int cpu)
+static int console_trylock_for_printk(void)
 {
+	unsigned int cpu = smp_processor_id();
+
 	if (!console_trylock())
 		return 0;
 	/*
@@ -1507,7 +1508,8 @@ asmlinkage int vprintk_emit(int facility, int level,
 		 */
 		if (!oops_in_progress && !lockdep_recursing(current)) {
 			recursion_bug = 1;
-			goto out_restore_irqs;
+			local_irq_restore(flags);
+			return 0;
 		}
 		zap_locks();
 	}
@@ -1600,17 +1602,22 @@ asmlinkage int vprintk_emit(int facility, int level,
 
 	logbuf_cpu = UINT_MAX;
 	raw_spin_unlock(&logbuf_lock);
+	lockdep_on();
+	local_irq_restore(flags);
+
+	/*
+	 * Disable preemption to avoid being preempted while holding
+	 * console_sem which would prevent anyone from printing to console
+	 */
+	preempt_disable();
 	/*
 	 * Try to acquire and then immediately release the console semaphore.
 	 * The release will print out buffers and wake up /dev/kmsg and syslog()
 	 * users.
 	 */
-	if (console_trylock_for_printk(this_cpu))
+	if (console_trylock_for_printk())
 		console_unlock();
-
-	lockdep_on();
-out_restore_irqs:
-	local_irq_restore(flags);
+	preempt_enable();
 
 	return printed_len;
 }
-- 
1.8.1.4

[PATCH 8/9] printk: Remove separate printk_sched buffers and use printk buf instead

[Jan Kara]

From: Steven Rostedt <rostedt@goodmis.org>

To prevent deadlocks with doing a printk inside the scheduler,
printk_sched() was created. The issue is that printk has a console_sem
that it can grab and release. The release does a wake up if there's a
task pending on the sem, and this wake up grabs the rq locks that is
held in the scheduler. This leads to a possible deadlock if the wake up
uses the same rq as the one with the rq lock held already.

What printk_sched() does is to save the printk write in a per cpu buffer
and sets the PRINTK_PENDING_SCHED flag. On a timer tick, if this flag is
set, the printk() is done against the buffer.

There's a couple of issues with this approach.

1) If two printk_sched()s are called before the tick, the second one
will overwrite the first one.

2) The temporary buffer is 512 bytes and is per cpu. This is a quite a
bit of space wasted for something that is seldom used.

In order to remove this, the printk_sched() can use the printk buffer
instead, and delay the console_trylock()/console_unlock() to the queued
work.

Because printk_sched() would then be taking the logbuf_lock, the
logbuf_lock must not be held while doing anything that may call into the
scheduler functions, which includes wake ups. Unfortunately, printk()
also has a console_sem that it uses, and on release, the
up(&console_sem) may do a wake up of any pending waiters. This must be
avoided while holding the logbuf_lock.

Signed-off-by: Steven Rostedt <rostedt@goodmis.org>
Signed-off-by: Jan Kara <jack@suse.cz>
---
 kernel/printk/printk.c | 43 ++++++++++++++++++++++++++-----------------
 1 file changed, 26 insertions(+), 17 deletions(-)

diff --git a/kernel/printk/printk.c b/kernel/printk/printk.c
index ab0d24a9ab7c..bfbe1a5f6804 100644
--- a/kernel/printk/printk.c
+++ b/kernel/printk/printk.c
@@ -208,6 +208,9 @@ struct printk_log {
 /*
  * The logbuf_lock protects kmsg buffer, indices, counters. It is also
  * used in interesting ways to provide interlocking in console_unlock();
+ * This can be taken within the scheduler's rq lock. It must be released
+ * before calling console_unlock() or anything else that might wake up
+ * a process.
  */
 static DEFINE_RAW_SPINLOCK(logbuf_lock);
 
@@ -1479,14 +1482,19 @@ asmlinkage int vprintk_emit(int facility, int level,
 	static int recursion_bug;
 	static char textbuf[LOG_LINE_MAX];
 	char *text = textbuf;
-	size_t text_len;
+	size_t text_len = 0;
 	enum log_flags lflags = 0;
 	unsigned long flags;
 	int this_cpu;
 	int printed_len = 0;
+	bool in_sched = false;
 	/* cpu currently holding logbuf_lock in this function */
 	static volatile unsigned int logbuf_cpu = UINT_MAX;
 
+	if (level == -2) {
+		level = -1;
+		in_sched = true;
+	}
 
 	boot_delay_msec(level);
 	printk_delay();
@@ -1533,7 +1541,12 @@ asmlinkage int vprintk_emit(int facility, int level,
 	 * The printf needs to come first; we need the syslog
 	 * prefix which might be passed-in as a parameter.
 	 */
-	text_len = vscnprintf(text, sizeof(textbuf), fmt, args);
+	if (in_sched)
+		text_len = scnprintf(text, sizeof(textbuf),
+				     KERN_WARNING "[sched_delayed] ");
+
+	text_len += vscnprintf(text + text_len,
+			       sizeof(textbuf) - text_len, fmt, args);
 
 	/* mark and strip a trailing newline */
 	if (text_len && text[text_len-1] == '\n') {
@@ -1605,6 +1618,10 @@ asmlinkage int vprintk_emit(int facility, int level,
 	lockdep_on();
 	local_irq_restore(flags);
 
+	/* If called from the scheduler, we can not call up(). */
+	if (in_sched)
+		return printed_len;
+
 	/*
 	 * Disable preemption to avoid being preempted while holding
 	 * console_sem which would prevent anyone from printing to console
@@ -2426,21 +2443,19 @@ late_initcall(printk_late_init);
 /*
  * Delayed printk version, for scheduler-internal messages:
  */
-#define PRINTK_BUF_SIZE		512
-
 #define PRINTK_PENDING_WAKEUP	0x01
-#define PRINTK_PENDING_SCHED	0x02
+#define PRINTK_PENDING_OUTPUT	0x02
 
 static DEFINE_PER_CPU(int, printk_pending);
-static DEFINE_PER_CPU(char [PRINTK_BUF_SIZE], printk_sched_buf);
 
 static void wake_up_klogd_work_func(struct irq_work *irq_work)
 {
 	int pending = __this_cpu_xchg(printk_pending, 0);
 
-	if (pending & PRINTK_PENDING_SCHED) {
-		char *buf = __get_cpu_var(printk_sched_buf);
-		pr_warn("[sched_delayed] %s", buf);
+	if (pending & PRINTK_PENDING_OUTPUT) {
+		/* If trylock fails, someone else is doing the printing */
+		if (console_trylock())
+			console_unlock();
 	}
 
 	if (pending & PRINTK_PENDING_WAKEUP)
@@ -2464,21 +2479,15 @@ void wake_up_klogd(void)
 
 int printk_sched(const char *fmt, ...)
 {
-	unsigned long flags;
 	va_list args;
-	char *buf;
 	int r;
 
-	local_irq_save(flags);
-	buf = __get_cpu_var(printk_sched_buf);
-
 	va_start(args, fmt);
-	r = vsnprintf(buf, PRINTK_BUF_SIZE, fmt, args);
+	r = vprintk_emit(0, -2, NULL, 0, fmt, args);
 	va_end(args);
 
-	__this_cpu_or(printk_pending, PRINTK_PENDING_SCHED);
+	__this_cpu_or(printk_pending, PRINTK_PENDING_OUTPUT);
 	irq_work_queue(&__get_cpu_var(wake_up_klogd_work));
-	local_irq_restore(flags);
 
 	return r;
 }
-- 
1.8.1.4

[PATCH 9/9] printk: Hand over printing to console if printing too long

[Jan Kara]

Currently, console_unlock() prints messages from kernel printk buffer to
console while the buffer is non-empty. When serial console is attached,
printing is slow and thus other CPUs in the system have plenty of time
to append new messages to the buffer while one CPU is printing. Thus the
CPU can spend unbounded amount of time doing printing in console_unlock().
This is especially serious problem if the printk() calling
console_unlock() was called with interrupts disabled.

In practice users have observed a CPU can spend tens of seconds printing
in console_unlock() (usually during boot when hundreds of SCSI devices
are discovered) resulting in RCU stalls (CPU doing printing doesn't
reach quiescent state for a long time), softlockup reports (IPIs for the
printing CPU don't get served and thus other CPUs are spinning waiting
for the printing CPU to process IPIs), and eventually a machine death
(as messages from stalls and lockups append to printk buffer faster than
we are able to print). So these machines are unable to boot with serial
console attached. Also during artificial stress testing SATA disk
disappears from the system because its interrupts aren't served for too
long.

This patch implements a mechanism where after printing specified number
of characters (tunable as a kernel parameter printk.offload_chars), CPU
doing printing asks for help by setting a 'hand over' state. The CPU
still keeps printing until another CPU running printk() or a CPU being
pinged by an IPI comes and takes over printing.  This way no CPU should
spend printing too long if there is heavy printk traffic.

Signed-off-by: Jan Kara <jack@suse.cz>
---
 Documentation/kernel-parameters.txt |  16 +++
 kernel/printk/printk.c              | 201 ++++++++++++++++++++++++++++++++----
 2 files changed, 198 insertions(+), 19 deletions(-)

diff --git a/Documentation/kernel-parameters.txt b/Documentation/kernel-parameters.txt
index 50680a59a2ff..5124bc82a9f5 100644
--- a/Documentation/kernel-parameters.txt
+++ b/Documentation/kernel-parameters.txt
@@ -2557,6 +2557,22 @@ bytes respectively. Such letter suffixes can also be entirely omitted.
 			Format: <bool>  (1/Y/y=enable, 0/N/n=disable)
 			default: disabled
 
+	printk.offload_chars=
+			Printing to console can be relatively slow especially
+			in case of serial console. When there is intensive
+			printing happening from several cpus (as is the case
+			during boot), a cpu can be spending significant time
+			(seconds or more) doing printing. To avoid softlockups,
+			lost interrupts, and similar problems other cpus
+			will take over printing after the currently printing
+			cpu has printed 'printk.offload_chars' characters.
+			Higher value means possibly longer interrupt and other
+			latencies but lower latency of printing and lower
+			chance something goes wrong during system crash and
+			important message is not printed.
+			Format: <number>
+			default: 0 (disabled)
+
 	printk.time=	Show timing data prefixed to each printk message line
 			Format: <bool>  (1/Y/y=enable, 0/N/n=disable)
 
diff --git a/kernel/printk/printk.c b/kernel/printk/printk.c
index bfbe1a5f6804..8abbb5373999 100644
--- a/kernel/printk/printk.c
+++ b/kernel/printk/printk.c
@@ -84,6 +84,45 @@ static DEFINE_SEMAPHORE(console_sem);
 struct console *console_drivers;
 EXPORT_SYMBOL_GPL(console_drivers);
 
+/*
+ * State of printing to console.
+ * 0 - noone is printing
+ * 1 - the CPU doing printing is happy doing so
+ * 2 - the printing CPU wants some other CPU to take over
+ * 3 - some CPU is waiting to take over printing
+ *
+ * Allowed state transitions are:
+ * 0 -> 1, 1 -> 0, 1 -> 2, 2 -> 0, 2 -> 3, 3 -> 0
+ * All state transitions except for 2 -> 3 are done by the holder of
+ * console_sem. Transition 2 -> 3 happens using cmpxchg from a task not owning
+ * console_sem. Thus it can race with other state transitions from state 2.
+ * However these races are harmless since the only transition we can race with
+ * is 2 -> 0. If cmpxchg comes after we have moved from state 2, it does
+ * nothing and we end in state 0. If cmpxchg comes before, we end in state 0 as
+ * desired.
+ */
+static enum {
+	PS_NONE,
+	PS_PRINTING,
+	PS_HANDOVER,
+	PS_WAITING
+} printing_state;
+/* CPU which is handing over printing */
+static unsigned int hand_over_cpu;
+/*
+ * Structure for IPI to hand printing to another CPU. We have actually two
+ * structures for the case we need to send IPI from an IPI handler...
+ */
+static void printk_take_over(void *info);
+static struct call_single_data hand_over_csd[2] = {
+	{ .func = printk_take_over, },
+	{ .func = printk_take_over, }
+};
+/* Index of csd to use for sending IPI now */
+static int current_csd;
+/* Set if there is IPI pending to take over printing */
+static bool printk_ipi_sent;
+
 #ifdef CONFIG_LOCKDEP
 static struct lockdep_map console_lock_dep_map = {
 	.name = "console_lock"
@@ -253,6 +292,18 @@ static char __log_buf[__LOG_BUF_LEN] __aligned(LOG_ALIGN);
 static char *log_buf = __log_buf;
 static u32 log_buf_len = __LOG_BUF_LEN;
 
+/*
+ * How many characters can we print in one call of printk before asking
+ * other cpus to continue printing (0 means infinity). Tunable via
+ * printk.offload_chars kernel parameter.
+ */
+static unsigned int __read_mostly printk_offload_chars = 0;
+
+module_param_named(offload_chars, printk_offload_chars, uint,
+		   S_IRUGO | S_IWUSR);
+MODULE_PARM_DESC(offload_chars, "offload printing to console to a different"
+	" cpu after this number of characters");
+
 /* human readable text of the record */
 static char *log_text(const struct printk_log *msg)
 {
@@ -1342,8 +1393,40 @@ static int console_trylock_for_printk(void)
 {
 	unsigned int cpu = smp_processor_id();
 
-	if (!console_trylock())
-		return 0;
+	if (!console_trylock()) {
+		int state;
+
+		if (printing_state != PS_HANDOVER || console_suspended)
+			return 0;
+		smp_rmb();	/* Paired with smp_wmb() in cpu_stop_printing */
+		/*
+		 * Avoid deadlocks when CPU holding console_sem takes an
+		 * interrupt which does printk.
+		 */
+		if (hand_over_cpu == cpu)
+			return 0;
+
+		state = cmpxchg(&printing_state, PS_HANDOVER, PS_WAITING);
+		if (state != PS_HANDOVER)
+			return 0;
+
+		/*
+		 * Since PS_HANDOVER state is set only in console_unlock()
+		 * we shouldn't spin for long here. And we cannot sleep because
+		 * the printk() might be called from atomic context.
+		 */
+		while (!console_trylock()) {
+			if (console_suspended)
+				return 0;
+			/*
+			 * Someone else took console_sem? Exit as we don't want
+			 * to spin for a long time here.
+			 */
+			if (ACCESS_ONCE(printing_state) == PS_PRINTING)
+				return 0;
+			__delay(1);
+		}
+	}
 	/*
 	 * If we can't use the console, we need to release the console
 	 * semaphore by hand to avoid flushing the buffer. We need to hold the
@@ -1351,6 +1434,7 @@ static int console_trylock_for_printk(void)
 	 */
 	if (!can_use_console(cpu)) {
 		console_locked = 0;
+		printing_state = PS_NONE;
 		up(&console_sem);
 		return 0;
 	}
@@ -1944,6 +2028,7 @@ void console_lock(void)
 		return;
 	console_locked = 1;
 	console_may_schedule = 1;
+	printing_state = PS_PRINTING;
 	mutex_acquire(&console_lock_dep_map, 0, 0, _RET_IP_);
 }
 EXPORT_SYMBOL(console_lock);
@@ -1966,6 +2051,7 @@ int console_trylock(void)
 	}
 	console_locked = 1;
 	console_may_schedule = 0;
+	printing_state = PS_PRINTING;
 	mutex_acquire(&console_lock_dep_map, 0, 1, _RET_IP_);
 	return 1;
 }
@@ -2005,15 +2091,77 @@ out:
 	raw_spin_unlock_irqrestore(&logbuf_lock, flags);
 }
 
+/* Handler for IPI to take over printing from another CPU */
+static void printk_take_over(void *info)
+{
+	/*
+	 * We have to clear printk_ipi_sent only after we succeed / fail the
+	 * trylock. That way we make sure there is at most one IPI spinning
+	 * on console_sem and thus we cannot deadlock on csd_lock
+	 */
+	if (console_trylock_for_printk()) {
+		printk_ipi_sent = false;
+		/* Switch csd as the current one is locked until we finish */
+		current_csd ^= 1;
+		console_unlock();
+	} else
+		printk_ipi_sent = false;
+}
+
+/*
+ * Returns true iff there is other cpu waiting to take over printing. This
+ * function also takes are of changing printing_state if we want to hand over
+ * printing to some other cpu.
+ */
+static bool cpu_stop_printing(int printed_chars)
+{
+	cpumask_var_t mask;
+
+	/* Oops? Print everything now to maximize chances user will see it */
+	if (oops_in_progress)
+		return false;
+	/* Someone is waiting. Stop printing. */
+	if (printing_state == PS_WAITING)
+		return true;
+	if (!printk_offload_chars || printed_chars <= printk_offload_chars)
+		return false;
+	if (printing_state == PS_PRINTING) {
+		hand_over_cpu = smp_processor_id();
+		/* Paired with smp_rmb() in console_trylock_for_printk() */
+		smp_wmb();
+		printing_state = PS_HANDOVER;
+		return false;
+	}
+	/*
+	 * We ping another CPU with IPI only if noone took over printing for a
+	 * long time - we prefer other printk() to take over printing since it
+	 * has chances to happen from a better context than IPI handler.
+	 */
+	if (!printk_ipi_sent && printed_chars > 2 * printk_offload_chars) {
+		struct call_single_data *csd = &hand_over_csd[current_csd];
+
+		/* Ping another cpu to take printing from us */
+		cpumask_copy(mask, cpu_online_mask);
+		cpumask_clear_cpu(hand_over_cpu, mask);
+		if (!cpumask_empty(mask)) {
+			printk_ipi_sent = true;
+			__smp_call_function_any(mask, csd, 0);
+		}
+	}
+	return false;
+}
+
 /**
  * console_unlock - unlock the console system
  *
  * Releases the console_lock which the caller holds on the console system
  * and the console driver list.
  *
- * While the console_lock was held, console output may have been buffered
- * by printk().  If this is the case, console_unlock(); emits
- * the output prior to releasing the lock.
+ * While the console_lock was held, console output may have been buffered by
+ * printk(). If this is the case, console_unlock() emits the output prior to
+ * releasing the lock. However we need not write all the data in the buffer if
+ * we would hog the CPU for too long. In such case we try to hand over printing
+ * to a different cpu.
  *
  * If there is output waiting, we wake /dev/kmsg and syslog() users.
  *
@@ -2026,6 +2174,8 @@ void console_unlock(void)
 	unsigned long flags;
 	bool wake_klogd = false;
 	bool retry;
+	bool hand_over = false;
+	int printed_chars = 0;
 
 	if (console_suspended) {
 		up(&console_sem);
@@ -2042,6 +2192,11 @@ again:
 		size_t len;
 		int level;
 
+		if (cpu_stop_printing(printed_chars)) {
+			hand_over = true;
+			break;
+		}
+
 		raw_spin_lock_irqsave(&logbuf_lock, flags);
 		if (seen_seq != log_next_seq) {
 			wake_klogd = true;
@@ -2055,8 +2210,10 @@ again:
 			console_prev = 0;
 		}
 skip:
-		if (console_seq == log_next_seq)
+		if (console_seq == log_next_seq) {
+			raw_spin_unlock(&logbuf_lock);
 			break;
+		}
 
 		msg = log_from_idx(console_idx);
 		if (msg->flags & LOG_NOCONS) {
@@ -2087,31 +2244,37 @@ skip:
 		stop_critical_timings();	/* don't trace print latency */
 		call_console_drivers(level, text, len);
 		start_critical_timings();
+		printed_chars += len;
 		local_irq_restore(flags);
 	}
-	console_locked = 0;
-	mutex_release(&console_lock_dep_map, 1, _RET_IP_);
 
 	/* Release the exclusive_console once it is used */
 	if (unlikely(exclusive_console))
 		exclusive_console = NULL;
 
-	raw_spin_unlock(&logbuf_lock);
-
+	printing_state = PS_NONE;
+	console_locked = 0;
+	mutex_release(&console_lock_dep_map, 1, _RET_IP_);
 	up(&console_sem);
 
 	/*
-	 * Someone could have filled up the buffer again, so re-check if there's
-	 * something to flush. In case we cannot trylock the console_sem again,
-	 * there's a new owner and the console_unlock() from them will do the
-	 * flush, no worries.
+	 * Subtlety: We have interrupts disabled iff hand_over == false (to
+	 * save one cli/sti pair in the fast path.
 	 */
-	raw_spin_lock(&logbuf_lock);
-	retry = console_seq != log_next_seq;
-	raw_spin_unlock_irqrestore(&logbuf_lock, flags);
+	if (!hand_over) {
+		/*
+		 * Someone could have filled up the buffer again, so re-check
+		 * if there's something to flush. In case we cannot trylock the
+		 * console_sem again, there's a new owner and the
+		 * console_unlock() from them will do the flush, no worries.
+		 */
+		raw_spin_lock(&logbuf_lock);
+		retry = console_seq != log_next_seq;
+		raw_spin_unlock_irqrestore(&logbuf_lock, flags);
 
-	if (retry && console_trylock())
-		goto again;
+		if (retry && console_trylock())
+			goto again;
+	}
 
 	if (wake_klogd)
 		wake_up_klogd();
-- 
1.8.1.4

Re: [PATCH 9/9] printk: Hand over printing to console if printing too long

[Andrew Morton]

On Mon, 23 Dec 2013 21:39:30 +0100 Jan Kara <jack@suse.cz> wrote:

> Currently, console_unlock() prints messages from kernel printk buffer to
> console while the buffer is non-empty. When serial console is attached,
> printing is slow and thus other CPUs in the system have plenty of time
> to append new messages to the buffer while one CPU is printing. Thus the
> CPU can spend unbounded amount of time doing printing in console_unlock().
> This is especially serious problem if the printk() calling
> console_unlock() was called with interrupts disabled.
> 
> In practice users have observed a CPU can spend tens of seconds printing
> in console_unlock() (usually during boot when hundreds of SCSI devices
> are discovered) resulting in RCU stalls (CPU doing printing doesn't
> reach quiescent state for a long time), softlockup reports (IPIs for the
> printing CPU don't get served and thus other CPUs are spinning waiting
> for the printing CPU to process IPIs), and eventually a machine death
> (as messages from stalls and lockups append to printk buffer faster than
> we are able to print). So these machines are unable to boot with serial
> console attached. Also during artificial stress testing SATA disk
> disappears from the system because its interrupts aren't served for too
> long.
> 
> This patch implements a mechanism where after printing specified number
> of characters (tunable as a kernel parameter printk.offload_chars), CPU
> doing printing asks for help by setting a 'hand over' state. The CPU
> still keeps printing until another CPU running printk() or a CPU being
> pinged by an IPI comes and takes over printing.  This way no CPU should
> spend printing too long if there is heavy printk traffic.

It all seems to rely on luck?  If there are 100k characters queued and
all the other CPUs stop calling printk(), the CPU which is left in
printk is screwed, isn't it?  If so, perhaps it can send an async IPI
to ask for help?

Re: [PATCH 9/9] printk: Hand over printing to console if printing too long

[Jan Kara]

On Sat 04-01-14 23:57:43, Andrew Morton wrote:
> On Mon, 23 Dec 2013 21:39:30 +0100 Jan Kara <jack@suse.cz> wrote:
> 
> > Currently, console_unlock() prints messages from kernel printk buffer to
> > console while the buffer is non-empty. When serial console is attached,
> > printing is slow and thus other CPUs in the system have plenty of time
> > to append new messages to the buffer while one CPU is printing. Thus the
> > CPU can spend unbounded amount of time doing printing in console_unlock().
> > This is especially serious problem if the printk() calling
> > console_unlock() was called with interrupts disabled.
> > 
> > In practice users have observed a CPU can spend tens of seconds printing
> > in console_unlock() (usually during boot when hundreds of SCSI devices
> > are discovered) resulting in RCU stalls (CPU doing printing doesn't
> > reach quiescent state for a long time), softlockup reports (IPIs for the
> > printing CPU don't get served and thus other CPUs are spinning waiting
> > for the printing CPU to process IPIs), and eventually a machine death
> > (as messages from stalls and lockups append to printk buffer faster than
> > we are able to print). So these machines are unable to boot with serial
> > console attached. Also during artificial stress testing SATA disk
> > disappears from the system because its interrupts aren't served for too
> > long.
> > 
> > This patch implements a mechanism where after printing specified number
> > of characters (tunable as a kernel parameter printk.offload_chars), CPU
> > doing printing asks for help by setting a 'hand over' state. The CPU
> > still keeps printing until another CPU running printk() or a CPU being
> > pinged by an IPI comes and takes over printing.  This way no CPU should
> > spend printing too long if there is heavy printk traffic.
> 
> It all seems to rely on luck?  If there are 100k characters queued and
> all the other CPUs stop calling printk(), the CPU which is left in
> printk is screwed, isn't it?  If so, perhaps it can send an async IPI
> to ask for help?
  Let me cite a sentence from the changelog:
"... until another CPU running printk() or a CPU being pinged by an IPI
comes and takes over printing."

  So sending IPI (async one) to another CPU to come and take over printing
is already implemented :). Do you have a better suggestion how to make that
more obvious in the changelog?

								Honza
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

Re: [PATCH 9/9] printk: Hand over printing to console if printing too long

[Jan Kara]

On Mon 06-01-14 10:46:08, Jan Kara wrote:
> On Sat 04-01-14 23:57:43, Andrew Morton wrote:
> > On Mon, 23 Dec 2013 21:39:30 +0100 Jan Kara <jack@suse.cz> wrote:
> > 
> > > Currently, console_unlock() prints messages from kernel printk buffer to
> > > console while the buffer is non-empty. When serial console is attached,
> > > printing is slow and thus other CPUs in the system have plenty of time
> > > to append new messages to the buffer while one CPU is printing. Thus the
> > > CPU can spend unbounded amount of time doing printing in console_unlock().
> > > This is especially serious problem if the printk() calling
> > > console_unlock() was called with interrupts disabled.
> > > 
> > > In practice users have observed a CPU can spend tens of seconds printing
> > > in console_unlock() (usually during boot when hundreds of SCSI devices
> > > are discovered) resulting in RCU stalls (CPU doing printing doesn't
> > > reach quiescent state for a long time), softlockup reports (IPIs for the
> > > printing CPU don't get served and thus other CPUs are spinning waiting
> > > for the printing CPU to process IPIs), and eventually a machine death
> > > (as messages from stalls and lockups append to printk buffer faster than
> > > we are able to print). So these machines are unable to boot with serial
> > > console attached. Also during artificial stress testing SATA disk
> > > disappears from the system because its interrupts aren't served for too
> > > long.
> > > 
> > > This patch implements a mechanism where after printing specified number
> > > of characters (tunable as a kernel parameter printk.offload_chars), CPU
> > > doing printing asks for help by setting a 'hand over' state. The CPU
> > > still keeps printing until another CPU running printk() or a CPU being
> > > pinged by an IPI comes and takes over printing.  This way no CPU should
> > > spend printing too long if there is heavy printk traffic.
> > 
> > It all seems to rely on luck?  If there are 100k characters queued and
> > all the other CPUs stop calling printk(), the CPU which is left in
> > printk is screwed, isn't it?  If so, perhaps it can send an async IPI
> > to ask for help?
>   Let me cite a sentence from the changelog:
> "... until another CPU running printk() or a CPU being pinged by an IPI
> comes and takes over printing."
> 
>   So sending IPI (async one) to another CPU to come and take over printing
> is already implemented :). Do you have a better suggestion how to make that
> more obvious in the changelog?
  Ping Andrew, did you have a look at the patch?

								Honza
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

Re: [PATCH 9/9] printk: Hand over printing to console if printing too long

[Andrew Morton]

On Mon, 23 Dec 2013 21:39:30 +0100 Jan Kara <jack@suse.cz> wrote:

> Currently, console_unlock() prints messages from kernel printk buffer to
> console while the buffer is non-empty. When serial console is attached,
> printing is slow and thus other CPUs in the system have plenty of time
> to append new messages to the buffer while one CPU is printing. Thus the
> CPU can spend unbounded amount of time doing printing in console_unlock().
> This is especially serious problem if the printk() calling
> console_unlock() was called with interrupts disabled.
> 
> In practice users have observed a CPU can spend tens of seconds printing
> in console_unlock() (usually during boot when hundreds of SCSI devices
> are discovered) resulting in RCU stalls (CPU doing printing doesn't
> reach quiescent state for a long time), softlockup reports (IPIs for the
> printing CPU don't get served and thus other CPUs are spinning waiting
> for the printing CPU to process IPIs), and eventually a machine death
> (as messages from stalls and lockups append to printk buffer faster than
> we are able to print). So these machines are unable to boot with serial
> console attached. Also during artificial stress testing SATA disk
> disappears from the system because its interrupts aren't served for too
> long.
> 
> This patch implements a mechanism where after printing specified number
> of characters (tunable as a kernel parameter printk.offload_chars), CPU
> doing printing asks for help by setting a 'hand over' state. The CPU
> still keeps printing until another CPU running printk() or a CPU being
> pinged by an IPI comes and takes over printing.  This way no CPU should
> spend printing too long if there is heavy printk traffic.
> 
> ...
>
> --- a/kernel/printk/printk.c
> +++ b/kernel/printk/printk.c
> @@ -84,6 +84,45 @@ static DEFINE_SEMAPHORE(console_sem);
>  struct console *console_drivers;
>  EXPORT_SYMBOL_GPL(console_drivers);
>  
> +/*
> + * State of printing to console.
> + * 0 - noone is printing
> + * 1 - the CPU doing printing is happy doing so
> + * 2 - the printing CPU wants some other CPU to take over
> + * 3 - some CPU is waiting to take over printing
> + *
> + * Allowed state transitions are:
> + * 0 -> 1, 1 -> 0, 1 -> 2, 2 -> 0, 2 -> 3, 3 -> 0
> + * All state transitions except for 2 -> 3 are done by the holder of
> + * console_sem. Transition 2 -> 3 happens using cmpxchg from a task not owning
> + * console_sem. Thus it can race with other state transitions from state 2.
> + * However these races are harmless since the only transition we can race with
> + * is 2 -> 0. If cmpxchg comes after we have moved from state 2, it does
> + * nothing and we end in state 0. If cmpxchg comes before, we end in state 0 as
> + * desired.
> + */

This comment is great, but would be much better if "0"-"3" were
replaced with their PS_foo representations.

The locking issue is regrettable.  What's the problem with getting full
console_sem coverage?

The mixture of cmpxchg with non-atomic reads and writes makes things
significantly more difficult.

> +static enum {
> +	PS_NONE,
> +	PS_PRINTING,
> +	PS_HANDOVER,
> +	PS_WAITING
> +} printing_state;
> +/* CPU which is handing over printing */
> +static unsigned int hand_over_cpu;
> +/*
> + * Structure for IPI to hand printing to another CPU. We have actually two
> + * structures for the case we need to send IPI from an IPI handler...
> + */
> +static void printk_take_over(void *info);
> +static struct call_single_data hand_over_csd[2] = {
> +	{ .func = printk_take_over, },
> +	{ .func = printk_take_over, }
> +};
> +/* Index of csd to use for sending IPI now */
> +static int current_csd;

Locking for this?

> +/* Set if there is IPI pending to take over printing */
> +static bool printk_ipi_sent;

And this?

>  #ifdef CONFIG_LOCKDEP
>  static struct lockdep_map console_lock_dep_map = {
>  	.name = "console_lock"
> 
> ...
>
> @@ -1342,8 +1393,40 @@ static int console_trylock_for_printk(void)
>  {
>  	unsigned int cpu = smp_processor_id();
>  
> -	if (!console_trylock())
> -		return 0;
> +	if (!console_trylock()) {
> +		int state;
> +
> +		if (printing_state != PS_HANDOVER || console_suspended)
> +			return 0;
> +		smp_rmb();	/* Paired with smp_wmb() in cpu_stop_printing */
> +		/*
> +		 * Avoid deadlocks when CPU holding console_sem takes an
> +		 * interrupt which does printk.
> +		 */
> +		if (hand_over_cpu == cpu)
> +			return 0;
> +
> +		state = cmpxchg(&printing_state, PS_HANDOVER, PS_WAITING);
> +		if (state != PS_HANDOVER)
> +			return 0;
> +
> +		/*
> +		 * Since PS_HANDOVER state is set only in console_unlock()
> +		 * we shouldn't spin for long here.

"shouldn't" is ambiguous here.  Suggest replacing it with "won't".

>		    And we cannot sleep because
> +		 * the printk() might be called from atomic context.
> +		 */

console_trylock_for_printk() is called under logbuf_lock, isn't it? 
We're atomic here regardless of the printk() caller's state.  That's
why smp_processor_id() was OK.

> +		while (!console_trylock()) {
> +			if (console_suspended)
> +				return 0;
> +			/*
> +			 * Someone else took console_sem? Exit as we don't want
> +			 * to spin for a long time here.
> +			 */
> +			if (ACCESS_ONCE(printing_state) == PS_PRINTING)

Is this appropriate use of ACCESS_ONCE?  What is the ACCESS_ONCE()
trying to achieve?

> +				return 0;
> +			__delay(1);
> +		}
> +	}
>  	/*
>  	 * If we can't use the console, we need to release the console
>  	 * semaphore by hand to avoid flushing the buffer. We need to hold the
> 
> ...
>
> @@ -2005,15 +2091,77 @@ out:
>  	raw_spin_unlock_irqrestore(&logbuf_lock, flags);
>  }
>  
> +/* Handler for IPI to take over printing from another CPU */
> +static void printk_take_over(void *info)
> +{
> +	/*
> +	 * We have to clear printk_ipi_sent only after we succeed / fail the
> +	 * trylock. That way we make sure there is at most one IPI spinning
> +	 * on console_sem and thus we cannot deadlock on csd_lock
> +	 */
> +	if (console_trylock_for_printk()) {

erk, scared.  We're in interrupt and console_trylock_for_printk() can
loop for arbitrarily long durations.  printk_take_over() is called
asynchronously and the system could be in any state at all.

> +		printk_ipi_sent = false;
> +		/* Switch csd as the current one is locked until we finish */
> +		current_csd ^= 1;

So current_csd is protected by console_sem?  As is printk_ipi_sent?

> +		console_unlock();

So it's via this console_unlock() that the current CPU starts printing?
Within IPI context?  It's worth documenting this a bit.

> +	} else
> +		printk_ipi_sent = false;
> +}
> +
> +/*
> + * Returns true iff there is other cpu waiting to take over printing. This
> + * function also takes are of changing printing_state if we want to hand over

"care"

> + * printing to some other cpu.
> + */
> +static bool cpu_stop_printing(int printed_chars)
> +{
> +	cpumask_var_t mask;
> +
> +	/* Oops? Print everything now to maximize chances user will see it */
> +	if (oops_in_progress)
> +		return false;
> +	/* Someone is waiting. Stop printing. */
> +	if (printing_state == PS_WAITING)
> +		return true;
> +	if (!printk_offload_chars || printed_chars <= printk_offload_chars)

Off-by-one?  Should that be "<"?

> +		return false;
> +	if (printing_state == PS_PRINTING) {
> +		hand_over_cpu = smp_processor_id();
> +		/* Paired with smp_rmb() in console_trylock_for_printk() */
> +		smp_wmb();
> +		printing_state = PS_HANDOVER;

So console_sem must be held by the caller?  Worth documenting this.

Again, the race with cmpxchg is worrisome.  Perhaps document its
(non-)effects here?

> +		return false;
> +	}
> +	/*
> +	 * We ping another CPU with IPI only if noone took over printing for a
> +	 * long time - we prefer other printk() to take over printing since it
> +	 * has chances to happen from a better context than IPI handler.
> +	 */
> +	if (!printk_ipi_sent && printed_chars > 2 * printk_offload_chars) {

What is the "2 *" doing?  I don't recall seeing a description of this.

> +		struct call_single_data *csd = &hand_over_csd[current_csd];

I didn't really understand why we need two call_single_data's.

> +
> +		/* Ping another cpu to take printing from us */
> +		cpumask_copy(mask, cpu_online_mask);
> +		cpumask_clear_cpu(hand_over_cpu, mask);
> +		if (!cpumask_empty(mask)) {

So what happens if this was the only online CPU?  We blow a chunk of
CPU time in cpu_stop_printing() for each printed char?  Not a problem I
guess.

> +			printk_ipi_sent = true;
> +			__smp_call_function_any(mask, csd, 0);

The IPI is sent to all other online CPUs.  I wonder if that was overkill.

> +		}
> +	}
> +	return false;
> +}
> +
> 
> ...
>

Re: [PATCH 9/9] printk: Hand over printing to console if printing too long

[Jan Kara]

On Wed 15-01-14 14:23:47, Andrew Morton wrote:
> On Mon, 23 Dec 2013 21:39:30 +0100 Jan Kara <jack@suse.cz> wrote:
> 
> > Currently, console_unlock() prints messages from kernel printk buffer to
> > console while the buffer is non-empty. When serial console is attached,
> > printing is slow and thus other CPUs in the system have plenty of time
> > to append new messages to the buffer while one CPU is printing. Thus the
> > CPU can spend unbounded amount of time doing printing in console_unlock().
> > This is especially serious problem if the printk() calling
> > console_unlock() was called with interrupts disabled.
> > 
> > In practice users have observed a CPU can spend tens of seconds printing
> > in console_unlock() (usually during boot when hundreds of SCSI devices
> > are discovered) resulting in RCU stalls (CPU doing printing doesn't
> > reach quiescent state for a long time), softlockup reports (IPIs for the
> > printing CPU don't get served and thus other CPUs are spinning waiting
> > for the printing CPU to process IPIs), and eventually a machine death
> > (as messages from stalls and lockups append to printk buffer faster than
> > we are able to print). So these machines are unable to boot with serial
> > console attached. Also during artificial stress testing SATA disk
> > disappears from the system because its interrupts aren't served for too
> > long.
> > 
> > This patch implements a mechanism where after printing specified number
> > of characters (tunable as a kernel parameter printk.offload_chars), CPU
> > doing printing asks for help by setting a 'hand over' state. The CPU
> > still keeps printing until another CPU running printk() or a CPU being
> > pinged by an IPI comes and takes over printing.  This way no CPU should
> > spend printing too long if there is heavy printk traffic.
> > 
> > ...
> >
> > --- a/kernel/printk/printk.c
> > +++ b/kernel/printk/printk.c
> > @@ -84,6 +84,45 @@ static DEFINE_SEMAPHORE(console_sem);
> >  struct console *console_drivers;
> >  EXPORT_SYMBOL_GPL(console_drivers);
> >  
> > +/*
> > + * State of printing to console.
> > + * 0 - noone is printing
> > + * 1 - the CPU doing printing is happy doing so
> > + * 2 - the printing CPU wants some other CPU to take over
> > + * 3 - some CPU is waiting to take over printing
> > + *
> > + * Allowed state transitions are:
> > + * 0 -> 1, 1 -> 0, 1 -> 2, 2 -> 0, 2 -> 3, 3 -> 0
> > + * All state transitions except for 2 -> 3 are done by the holder of
> > + * console_sem. Transition 2 -> 3 happens using cmpxchg from a task not owning
> > + * console_sem. Thus it can race with other state transitions from state 2.
> > + * However these races are harmless since the only transition we can race with
> > + * is 2 -> 0. If cmpxchg comes after we have moved from state 2, it does
> > + * nothing and we end in state 0. If cmpxchg comes before, we end in state 0 as
> > + * desired.
> > + */
> 
> This comment is great, but would be much better if "0"-"3" were
> replaced with their PS_foo representations.
  Sure, will do.

> The locking issue is regrettable.  What's the problem with getting full
> console_sem coverage?
>
> The mixture of cmpxchg with non-atomic reads and writes makes things
> significantly more difficult.
  Yes, I understand. The thing is you need a way for a CPU to say "I'm here
to help you, please release console_sem". Clearly this cannot happen under
console_sem. But now when I look at it from distance, there are two ways how
we can make the protocol more standard:
1) protect printing_state by a dedicated spinlock. That makes everything
rather standard although more heavy weight.
2) use a separate bool variable to indicate someone is spinning on
   console_sem. I guess I'll go for this.

> > +static enum {
> > +	PS_NONE,
> > +	PS_PRINTING,
> > +	PS_HANDOVER,
> > +	PS_WAITING
> > +} printing_state;
> > +/* CPU which is handing over printing */
> > +static unsigned int hand_over_cpu;
> > +/*
> > + * Structure for IPI to hand printing to another CPU. We have actually two
> > + * structures for the case we need to send IPI from an IPI handler...
> > + */
> > +static void printk_take_over(void *info);
> > +static struct call_single_data hand_over_csd[2] = {
> > +	{ .func = printk_take_over, },
> > +	{ .func = printk_take_over, }
> > +};
> > +/* Index of csd to use for sending IPI now */
> > +static int current_csd;
> 
> Locking for this?
  console_sem.

> > +/* Set if there is IPI pending to take over printing */
> > +static bool printk_ipi_sent;
> 
> And this?
  Good question, this deserves documentation if nothing else. Since the
variable is bool, reads & writes of it are atomic. So it's really only
about the value of the variable matching the fact whether we sent IPI
asking for help or not. And that is achieved by:
  * setting printk_ipi_sent to true only if is is currently false and doing
    that under console_sem
  * setting printk_ipi_sent to false only from IPI handler of that IPI.
    Since we always send the IPI to at most one CPU, this setting cannot
    race.

> >  #ifdef CONFIG_LOCKDEP
> >  static struct lockdep_map console_lock_dep_map = {
> >  	.name = "console_lock"
> > 
> > ...
> >
> > @@ -1342,8 +1393,40 @@ static int console_trylock_for_printk(void)
> >  {
> >  	unsigned int cpu = smp_processor_id();
> >  
> > -	if (!console_trylock())
> > -		return 0;
> > +	if (!console_trylock()) {
> > +		int state;
> > +
> > +		if (printing_state != PS_HANDOVER || console_suspended)
> > +			return 0;
> > +		smp_rmb();	/* Paired with smp_wmb() in cpu_stop_printing */
> > +		/*
> > +		 * Avoid deadlocks when CPU holding console_sem takes an
> > +		 * interrupt which does printk.
> > +		 */
> > +		if (hand_over_cpu == cpu)
> > +			return 0;
> > +
> > +		state = cmpxchg(&printing_state, PS_HANDOVER, PS_WAITING);
> > +		if (state != PS_HANDOVER)
> > +			return 0;
> > +
> > +		/*
> > +		 * Since PS_HANDOVER state is set only in console_unlock()
> > +		 * we shouldn't spin for long here.
> 
> "shouldn't" is ambiguous here.  Suggest replacing it with "won't".
  OK.

> >		    And we cannot sleep because
> > +		 * the printk() might be called from atomic context.
> > +		 */
> 
> console_trylock_for_printk() is called under logbuf_lock, isn't it? 
> We're atomic here regardless of the printk() caller's state.  That's
> why smp_processor_id() was OK.
  Preceding patches in this series changed vprintk_emit() to unlock
logbuf_lock and enable interrupts (well, local_irq_restore()) before
calling into console_trylock_for_printk() and console_unlock(). So we
should run in the same locking context in which printk() was called.

Sadly that isn't a solution to lockup problems (tested - most notably
because the system dies if RCU period doesn't happen often enough on every
CPU) but it should help to significantly reduce irq latencies on average.

> > +		while (!console_trylock()) {
> > +			if (console_suspended)
> > +				return 0;
> > +			/*
> > +			 * Someone else took console_sem? Exit as we don't want
> > +			 * to spin for a long time here.
> > +			 */
> > +			if (ACCESS_ONCE(printing_state) == PS_PRINTING)
> 
> Is this appropriate use of ACCESS_ONCE?  What is the ACCESS_ONCE()
> trying to achieve?
  I wanted to force compiler to reload printing_state variable on each loop
iteration. But maybe I'm completely wrong and it isn't needed here. Or
maybe something else should be here? It just seemed to me that without
anything, compiler could "optimize" the loop to do the check just during
the first iteration.

> > +				return 0;
> > +			__delay(1);
> > +		}
> > +	}
> >  	/*
> >  	 * If we can't use the console, we need to release the console
> >  	 * semaphore by hand to avoid flushing the buffer. We need to hold the
> > 
> > ...
> >
> > @@ -2005,15 +2091,77 @@ out:
> >  	raw_spin_unlock_irqrestore(&logbuf_lock, flags);
> >  }
> >  
> > +/* Handler for IPI to take over printing from another CPU */
> > +static void printk_take_over(void *info)
> > +{
> > +	/*
> > +	 * We have to clear printk_ipi_sent only after we succeed / fail the
> > +	 * trylock. That way we make sure there is at most one IPI spinning
> > +	 * on console_sem and thus we cannot deadlock on csd_lock
> > +	 */
> > +	if (console_trylock_for_printk()) {
> 
> erk, scared.  We're in interrupt and console_trylock_for_printk() can
> loop for arbitrarily long durations.  printk_take_over() is called
> asynchronously and the system could be in any state at all.
  So what mechanism would you prefer to use to ask for help (and as you
noted in your previous mail, we need some mechanism to not rely on luck
that someone calls printk()). The options I see are:
1) raw IPI as in this patch.
   + simple lockless code to send IPI
   - there is a trouble that if your are printing from IPI and want to send
     IPI to another CPU to take over printing, you need a separate CSD
     structure to avoid deadlocking on csd_lock
   - runs in interrupt context
2) irq_work - rather similar to raw IPI although you get slightly less raw
   interface for slightly more complex code so you might need to be more
   careful to avoid trouble.
3) normal work
   + you run in normal context
   - the code isn't lockless anymore so if there is printk() in that code
     (currently there isn't) we have to be *really* careful to avoid
     deadlocks
   - similarly to IPI you will need more work structs to avoid problems
     due to reentering works (possibly deadlocks on pwq->pool->lock,
     certainly you would end up executing the work always on the same CPU).
4) separate kernel threads just for taking over printing
   + you run in normal context
   + you don't need any special locking
   - you have these additional threads lurking on the system mostly doing
     nothing

Pick your poison... Thoughts?

> > +		printk_ipi_sent = false;
> > +		/* Switch csd as the current one is locked until we finish */
> > +		current_csd ^= 1;
> 
> So current_csd is protected by console_sem?  As is printk_ipi_sent?
  Answered above.

> > +		console_unlock();
> 
> So it's via this console_unlock() that the current CPU starts printing?
> Within IPI context?  It's worth documenting this a bit.
  Yes, I will add documentation.

> > +	} else
> > +		printk_ipi_sent = false;
> > +}
> > +
> > +/*
> > + * Returns true iff there is other cpu waiting to take over printing. This
> > + * function also takes are of changing printing_state if we want to hand over
> 
> "care"
> 
> > + * printing to some other cpu.
> > + */
> > +static bool cpu_stop_printing(int printed_chars)
> > +{
> > +	cpumask_var_t mask;
> > +
> > +	/* Oops? Print everything now to maximize chances user will see it */
> > +	if (oops_in_progress)
> > +		return false;
> > +	/* Someone is waiting. Stop printing. */
> > +	if (printing_state == PS_WAITING)
> > +		return true;
> > +	if (!printk_offload_chars || printed_chars <= printk_offload_chars)
> 
> Off-by-one?  Should that be "<"?
  Doesn't really matter... but yes.

> > +		return false;
> > +	if (printing_state == PS_PRINTING) {
> > +		hand_over_cpu = smp_processor_id();
> > +		/* Paired with smp_rmb() in console_trylock_for_printk() */
> > +		smp_wmb();
> > +		printing_state = PS_HANDOVER;
> 
> So console_sem must be held by the caller?  Worth documenting this.
  OK.

> Again, the race with cmpxchg is worrisome.  Perhaps document its
> (non-)effects here?
  I'll see if using separate bool (and thus avoiding cmpxchg) will make
things more obvious. And add some documentation here.

> > +		return false;
> > +	}
> > +	/*
> > +	 * We ping another CPU with IPI only if noone took over printing for a
> > +	 * long time - we prefer other printk() to take over printing since it
> > +	 * has chances to happen from a better context than IPI handler.
> > +	 */
> > +	if (!printk_ipi_sent && printed_chars > 2 * printk_offload_chars) {
> 
> What is the "2 *" doing?  I don't recall seeing a description of this.
  Yeah, I should describe this somewhere more highlevel than in this
comment. As the comment describes, IPI isn't really a good context to do
printing from (unknown locks held, interrupts disabled on the cpu) so we
want to use it only as a last resort. If we send IPI immediately after
printing printk_offload_chars, there is really low chance another printk()
will take over printing - IPI will be almost always faster. So we have an
interval between printing printk_offload_chars and 2 * printk_offload_chars
where we ask for help but don't send the IPI yet.

> > +		struct call_single_data *csd = &hand_over_csd[current_csd];
> 
> I didn't really understand why we need two call_single_data's.
  The thing is: while IPI is executing, call_single_data structure is
locked. So if we are printing from the IPI and want to ask for help, we
cannot really use the call_single_data structure which was used to send IPI
we are running. That's why we have to have at least two...

> > +
> > +		/* Ping another cpu to take printing from us */
> > +		cpumask_copy(mask, cpu_online_mask);
> > +		cpumask_clear_cpu(hand_over_cpu, mask);
> > +		if (!cpumask_empty(mask)) {
> 
> So what happens if this was the only online CPU?  We blow a chunk of
> CPU time in cpu_stop_printing() for each printed char?  Not a problem I
> guess.
  For every printed line, yes. But the thing is that on single CPU system
you do not really accumulate large backlog in the printk buffer (I presume
interrupts are the only way to accumulate any backlog) so you never get
here. Well, sysrq-t is one special case and cpu hotplug another but these are
cornercases I'm willing to dismiss.

> > +			printk_ipi_sent = true;
> > +			__smp_call_function_any(mask, csd, 0);
> 
> The IPI is sent to all other online CPUs.  I wonder if that was overkill.
  No, it is sent to 'any other online CPU' - we pick one and send IPI
there.

Thanks for review!

								Honza
-- 
Jan Kara <jack@suse.cz>
SUSE Labs, CR

[PATCH 10/10] printk: debug: Slow down printing to 9600 bauds

[Jan Kara]

Signed-off-by: Jan Kara <jack@suse.cz>
---
 include/trace/events/printk.h | 42 +++++++++++++++++++++++++++++++++++
 kernel/printk/printk.c        | 51 ++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 92 insertions(+), 1 deletion(-)

diff --git a/include/trace/events/printk.h b/include/trace/events/printk.h
index c008bc99f9fa..2bd0c0e241f7 100644
--- a/include/trace/events/printk.h
+++ b/include/trace/events/printk.h
@@ -22,6 +22,48 @@ TRACE_EVENT(console,
 
 	TP_printk("%s", __get_str(msg))
 );
+
+DECLARE_EVENT_CLASS(printk_class,
+	TP_PROTO(int u),
+	TP_ARGS(u),
+	TP_STRUCT__entry(
+	        __field(        int, u)
+	),
+	TP_fast_assign(
+	        __entry->u       = u;
+	),
+	TP_printk("arg=%d", __entry->u)
+);
+
+DEFINE_EVENT(printk_class, printk_hand_over,
+	TP_PROTO(int u),
+	TP_ARGS(u)
+);
+
+DEFINE_EVENT(printk_class, printk_ask_help,
+	TP_PROTO(int u),
+	TP_ARGS(u)
+);
+
+DEFINE_EVENT(printk_class, printk_send_ipi,
+	TP_PROTO(int u),
+	TP_ARGS(u)
+);
+
+DEFINE_EVENT(printk_class, printk_ipi_received,
+	TP_PROTO(int u),
+	TP_ARGS(u)
+);
+
+DEFINE_EVENT(printk_class, printk_set_wait,
+	TP_PROTO(int u),
+	TP_ARGS(u)
+);
+
+DEFINE_EVENT(printk_class, printk_sem_spin,
+	TP_PROTO(int u),
+	TP_ARGS(u)
+);
 #endif /* _TRACE_PRINTK_H */
 
 /* This part must be outside protection */
diff --git a/kernel/printk/printk.c b/kernel/printk/printk.c
index 8abbb5373999..e78fe2dc280f 100644
--- a/kernel/printk/printk.c
+++ b/kernel/printk/printk.c
@@ -1337,6 +1337,15 @@ static void call_console_drivers(int level, const char *text, size_t len)
 	}
 }
 
+static void printk_echo(char *txt)
+{
+	unsigned long flags;
+
+	local_irq_save(flags);
+	call_console_drivers(0, txt, strlen(txt));
+	local_irq_restore(flags);
+}
+
 /*
  * Zap console related locks when oopsing. Only zap at most once
  * every 10 seconds, to leave time for slow consoles to print a
@@ -1406,10 +1415,14 @@ static int console_trylock_for_printk(void)
 		if (hand_over_cpu == cpu)
 			return 0;
 
+		printk_echo("Setting PS_WAITING\n");
+		trace_printk_set_wait(0);
 		state = cmpxchg(&printing_state, PS_HANDOVER, PS_WAITING);
 		if (state != PS_HANDOVER)
 			return 0;
 
+		printk_echo("Spinning on console_sem\n");
+		trace_printk_sem_spin(0);
 		/*
 		 * Since PS_HANDOVER state is set only in console_unlock()
 		 * we shouldn't spin for long here. And we cannot sleep because
@@ -1426,6 +1439,7 @@ static int console_trylock_for_printk(void)
 				return 0;
 			__delay(1);
 		}
+		printk_echo("Got console_sem\n");
 	}
 	/*
 	 * If we can't use the console, we need to release the console
@@ -1574,6 +1588,7 @@ asmlinkage int vprintk_emit(int facility, int level,
 	bool in_sched = false;
 	/* cpu currently holding logbuf_lock in this function */
 	static volatile unsigned int logbuf_cpu = UINT_MAX;
+	bool irq_off = irqs_disabled();
 
 	if (level == -2) {
 		level = -1;
@@ -1628,6 +1643,8 @@ asmlinkage int vprintk_emit(int facility, int level,
 	if (in_sched)
 		text_len = scnprintf(text, sizeof(textbuf),
 				     KERN_WARNING "[sched_delayed] ");
+	if (irq_off)
+		text[text_len++] = 'X';
 
 	text_len += vscnprintf(text + text_len,
 			       sizeof(textbuf) - text_len, fmt, args);
@@ -2094,6 +2111,8 @@ out:
 /* Handler for IPI to take over printing from another CPU */
 static void printk_take_over(void *info)
 {
+	printk_echo("IPI received\n");
+	trace_printk_ipi_received(smp_processor_id());
 	/*
 	 * We have to clear printk_ipi_sent only after we succeed / fail the
 	 * trylock. That way we make sure there is at most one IPI spinning
@@ -2116,13 +2135,17 @@ static void printk_take_over(void *info)
 static bool cpu_stop_printing(int printed_chars)
 {
 	cpumask_var_t mask;
+	char buf[80];
 
 	/* Oops? Print everything now to maximize chances user will see it */
 	if (oops_in_progress)
 		return false;
 	/* Someone is waiting. Stop printing. */
-	if (printing_state == PS_WAITING)
+	if (printing_state == PS_WAITING) {
+		printk_echo("Handing over printing\n");
+		trace_printk_hand_over(0);
 		return true;
+	}
 	if (!printk_offload_chars || printed_chars <= printk_offload_chars)
 		return false;
 	if (printing_state == PS_PRINTING) {
@@ -2130,8 +2153,13 @@ static bool cpu_stop_printing(int printed_chars)
 		/* Paired with smp_rmb() in console_trylock_for_printk() */
 		smp_wmb();
 		printing_state = PS_HANDOVER;
+		printk_echo("Asking for help\n");
+		trace_printk_ask_help(0);
 		return false;
 	}
+
+	sprintf(buf, "Checking IPI: %d %d > %d\n", (int)printk_ipi_sent, printed_chars, 2*printk_offload_chars);
+	printk_echo(buf);
 	/*
 	 * We ping another CPU with IPI only if noone took over printing for a
 	 * long time - we prefer other printk() to take over printing since it
@@ -2140,6 +2168,8 @@ static bool cpu_stop_printing(int printed_chars)
 	if (!printk_ipi_sent && printed_chars > 2 * printk_offload_chars) {
 		struct call_single_data *csd = &hand_over_csd[current_csd];
 
+		trace_printk_send_ipi(hand_over_cpu);
+		printk_echo("Sending IPI\n");
 		/* Ping another cpu to take printing from us */
 		cpumask_copy(mask, cpu_online_mask);
 		cpumask_clear_cpu(hand_over_cpu, mask);
@@ -2245,6 +2275,7 @@ skip:
 		call_console_drivers(level, text, len);
 		start_critical_timings();
 		printed_chars += len;
+		mdelay(len);
 		local_irq_restore(flags);
 	}
 
@@ -2588,9 +2619,27 @@ int unregister_console(struct console *console)
 }
 EXPORT_SYMBOL(unregister_console);
 
+void do_print(struct work_struct *work)
+{
+	char buf[75];
+	int i;
+
+	sprintf(buf, "%p: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n", work);
+	for (i = 0; i < 15; i++)
+		printk(buf);
+}
+
+static struct delayed_work print_work[100];
+
 static int __init printk_late_init(void)
 {
 	struct console *con;
+	int i;
+
+	for (i = 0; i < 100; i++) {
+		INIT_DELAYED_WORK(&print_work[i], do_print);
+		schedule_delayed_work(&print_work[i], HZ * 180);
+	}
 
 	for_each_console(con) {
 		if (!keep_bootcon && con->flags & CON_BOOT) {
-- 
1.8.1.4