ipc: Make the ipc code -rt aware

RT serializes the code with the (rt)spinlock but keeps preemption
enabled. Some parts of the code need to be atomic
nevertheless. 

Protect it with preempt_disable/enable_rt pairts.
Signed-off-by: Ingo Molnar <mingo@elte.hu>
Signed-off-by: Thomas Gleixner <tglx@linutronix.de>

ipc/mqueue.c

@@ -820,12 +820,17 @@ static inline void pipelined_send(struct mqueue_inode_info *info,
 				  struct msg_msg *message,
 				  struct ext_wait_queue *receiver)
 {
+	/*
+	 * Keep them in one critical section for PREEMPT_RT:
+	 */
+	preempt_disable_rt();
 	receiver->msg = message;
 	list_del(&receiver->list);
 	receiver->state = STATE_PENDING;
 	wake_up_process(receiver->task);
 	smp_wmb();
 	receiver->state = STATE_READY;
+	preempt_enable_nort();
 }
 
 /* pipelined_receive() - if there is task waiting in sys_mq_timedsend()

ipc/msg.c

@@ -259,12 +259,20 @@ static void expunge_all(struct msg_queue *msq, int res)
 	while (tmp != &msq->q_receivers) {
 		struct msg_receiver *msr;
 
+		/*
+		 * Make sure that the wakeup doesnt preempt
+		 * this CPU prematurely. (on PREEMPT_RT)
+		 */
+		preempt_disable_rt();
+
 		msr = list_entry(tmp, struct msg_receiver, r_list);
 		tmp = tmp->next;
 		msr->r_msg = NULL;
 		wake_up_process(msr->r_tsk);
 		smp_mb();
 		msr->r_msg = ERR_PTR(res);
+
+		preempt_enable_rt();
 	}
 }
 
@@ -611,6 +619,12 @@ static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg)
 		    !security_msg_queue_msgrcv(msq, msg, msr->r_tsk,
 					       msr->r_msgtype, msr->r_mode)) {
 
+			/*
+			 * Make sure that the wakeup doesnt preempt
+			 * this CPU prematurely. (on PREEMPT_RT)
+			 */
+			preempt_disable_rt();
+
 			list_del(&msr->r_list);
 			if (msr->r_maxsize < msg->m_ts) {
 				msr->r_msg = NULL;
@@ -624,9 +638,11 @@ static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg)
 				wake_up_process(msr->r_tsk);
 				smp_mb();
 				msr->r_msg = msg;
+				preempt_enable_rt();
 
 				return 1;
 			}
+			preempt_enable_rt();
 		}
 	}
 	return 0;

ipc/sem.c

@@ -414,6 +414,11 @@ static void update_queue (struct sem_array * sma)
 		if (error <= 0) {
 			struct sem_queue *n;
 
+			/*
+			 * make sure that the wakeup doesnt preempt
+			 * _this_ cpu prematurely. (on preempt_rt)
+			 */
+			preempt_disable_rt();
 			/*
 			 * Continue scanning. The next operation
 			 * that must be checked depends on the type of the
@@ -450,6 +455,7 @@ static void update_queue (struct sem_array * sma)
 			 */
 			smp_wmb();
 			q->status = error;
+			preempt_enable_rt();
 			q = n;
 		} else {
 			q = list_entry(q->list.next, struct sem_queue, list);