From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: (majordomo@vger.kernel.org) by vger.kernel.org via listexpand id S934032AbbJIIk4 (ORCPT ); Fri, 9 Oct 2015 04:40:56 -0400 Received: from mailhost.informatik.uni-bremen.de ([134.102.201.18]:49711 "EHLO mailhost.informatik.uni-bremen.de" rhost-flags-OK-OK-OK-OK) by vger.kernel.org with ESMTP id S932346AbbJIIkw (ORCPT ); Fri, 9 Oct 2015 04:40:52 -0400 From: =?UTF-8?Q?Felix_H=c3=bcbner?= Subject: PROBLEM: Concurrency issue in sem_lock Cc: manfred@colorfullife.com, bitbucket@online.de, riel@redhat.com, dbueso@suse.de To: linux-kernel@vger.kernel.org Message-ID: <561777E3.1060804@informatik.uni-bremen.de> Date: Fri, 9 Oct 2015 10:16:35 +0200 User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:38.0) Gecko/20100101 Thunderbird/38.3.0 MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 8bit Sender: linux-kernel-owner@vger.kernel.org List-ID: X-Mailing-List: linux-kernel@vger.kernel.org Hi all, I have just reported a concurrency issue in the implementation of sem_lock, see https://bugzilla.kernel.org/show_bug.cgi?id=105651 Please set me to CC for answers/comments. [1.] One line summary of the problem: Concurrency issue in sem_lock [2.] Full description of the problem/report: In line 376 (file ipc/sem.c) sem_wait_array is called with pre condition sma->complex_count!=0 resulting in an immediate return. The following scenario is possible, allowing P0 and P1 to be in the critical section (section between sem_lock und sem_unlock) at the same time: P0 is a process performing up with semtimedop on single semaphore 1. P1 is a process performing up with semtimedop on single semaphore 1. P2 is a process performing down with semtimedop on semaphores 1 and 2. start condition: sma->complex_count=0 All line references apply to the file ipc/sem.c. # P0 runs sem_lock up to line 332: if(sma->complex_count==0) evaluates to true static inline int sem_lock(struct sem_array *sma, struct sembuf *sops, int nsops) { struct sem *sem; if (nsops != 1) { ... } sem = sma->sem_base + sops->sem_num; if (sma->complex_count == 0) { # P1 runs sem_lock up to line 329. static inline int sem_lock(struct sem_array *sma, struct sembuf *sops, int nsops) { struct sem *sem; if (nsops != 1) { ... } sem = sma->sem_base + sops->sem_num; # P2 runs sem_lock to (including) line 310. static inline int sem_lock(struct sem_array *sma, struct sembuf *sops, int nsops) { struct sem *sem; if (nsops != 1) { ipc_lock_object(&sma->sem_perm); sem_wait_array(sma); # P0 does spin_lock(&sem->lock); in line 336. spin_lock(&sem->lock); # P2 performs rest of semtimedop, increments complex_count and ends up in line 1961 and starts to sleep. return -1; } ##semtimedop: locknum = sem_lock(sma, sops, nsops); ... error = perform_atomic_semop(sma, &queue); if (error == 0) { ... } if (error <= 0) ... if (nsops == 1) { ... } else { if (!sma->complex_count) merge_queues(sma); if (alter) list_add_tail(&queue.list, &sma->pending_alter); else list_add_tail(&queue.list, &sma->pending_const); sma->complex_count++; } queue.status = -EINTR; queue.sleeper = current; sleep_again: __set_current_state(TASK_INTERRUPTIBLE); sem_unlock(sma, locknum); rcu_read_unlock(); if (timeout) jiffies_left = schedule_timeout(jiffies_left); else schedule(); # P1 evaluates if(sem->complex_count==0) in line 331 and ends up in line 361. if (sma->complex_count == 0) { ... } # P0 evaluates if(!spin_is_locked(&sma->sem_perm.lock)) in line 339 and ends up after line 346. if (!spin_is_locked(&sma->sem_perm.lock)) { ipc_smp_acquire__after_spin_is_unlocked(); # P1 performs ipc_lock_object(&sma->sem_perm); line 362; it then evaluates if(sma->complex_count==0) and executes the else statement, calls sem_wait_array in line 376, which returns erroneously. ipc_lock_object(&sma->sem_perm); if (sma->complex_count == 0) { ... } else { sem_wait_array(sma); static void sem_wait_array(struct sem_array*sma) { int i; struct sem *sem; if (sma->complex_count) { return; } return -1; } # P1 runs the rest of semtimedop and wakes P2 up in do_smart_update (line 1911), update_queue and unlink_queue respectively. This reduces complex_count. locknum = sem_lock(sma, sops, nsops); ... ... error = perform_atomic_semop(sma, &queue); if (error == 0) { if (alter) do_smart_update(sma, sops, nsops, 1, &tasks); # P0 now evaluates if(sma->complex_count==0) in line 353 and is able to enter the critical section, while P1 is still in do_smart_update(). if (sma->complex_count == 0) { /* fast path successful! */ return sops->sem_num; } Thanks to Manfred Spraul for his comments on the problem description. [3.] Keywords (i.e., modules, networking, kernel): IPC, sem_lock [4.] Kernel information [4.1.] Kernel version (from /proc/version): 4.3.0-rc4 [5.] Most recent kernel version which did not have the bug: According to Manfred, the attached patch, causing the bug, was first added in version 3.12 and then backported. [X.] Patch causing the bug: >>From f5346dce3b37c00f9b6e22bee35d22783d11de7c Mon Sep 17 00:00:00 2001 From: Manfred Spraul Date: Sat, 14 Sep 2013 21:45:57 +0200 Subject: [PATCH 2/3] ipc/sem.c: optimize sem_lock(). Operations that need access to the whole array must guarantee that there are no simple operations ongoing. Right now this is achieved by spin_unlock_wait(sem->lock) on all semaphores. If complex_count is nonzero, then this spin_unlock_wait() is not necessary, because it was already performed in the past by the thread that increased complex_count and even though sem_perm.lock was dropped inbetween, no simple operation could have started, because simple operations cannot start when complex_count is non-zero. Signed-off-by: Manfred Spraul Cc: Mike Galbraith Cc: Rik van Riel Cc: Davidlohr Bueso --- ipc/sem.c | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/ipc/sem.c b/ipc/sem.c index 4a92c04..e20658d 100644 --- a/ipc/sem.c +++ b/ipc/sem.c @@ -257,12 +257,20 @@ static void sem_rcu_free(struct rcu_head *head) * Caller must own sem_perm.lock. * New simple ops cannot start, because simple ops first check * that sem_perm.lock is free. + * that a) sem_perm.lock is free and b) complex_count is 0. */ static void sem_wait_array(struct sem_array *sma) { int i; struct sem *sem; + if (sma->complex_count) { + /* The thread that increased sma->complex_count waited on + * all sem->lock locks. Thus we don't need to wait again. + */ + return; + } + for (i = 0; i < sma->sem_nsems; i++) { sem = sma->sem_base + i; spin_unlock_wait(&sem->lock); -- 1.8.3.1 Best regards, Felix Hübner -- Felix Hübner Arbeitsgruppe Betriebssysteme, Verteilte Systeme (Research Group Operating Systems, Distributed Systems) FB3 Mathematik und Informatik (Department of Mathematics and Computer Science) Universität Bremen ROOM: MZH 8200 MAIL: felixh@informatik.uni-bremen.de TEL : +49 (0)421 / 218 63966 WEB : http://www.informatik.uni-bremen.de/agbs