Re: [PATCH 11/12] Generic semaphore implementation

[Andrew Morton <akpm@linux-foundation.org>]

On Thu, 28 Feb 2008 01:34:00 -0500 Matthew Wilcox <matthew@wil.cx> wrote:

> Semaphores are no longer performance-critical, so a generic C
> implementation is better for maintainability, debuggability and
> extensibility.
> 

This make all architectures use the lib/semaphore-sleepers.c
implementation.  The major difference is that down() and up() will now
unconditionally do a spin_lock_irq[save]() on the not-contended fastpath,
whereas the hand-optimised arch-specific implementations would avoid that.

Yes, hopefully not many sempahores are used on fastpaths any more, and a
suitable fix for any which _are_ so used is usually convert-to-mutex.

And spin_lock_irqsave() probably isn't hugely slower than an atomic-dec
anyway.



A few nits, most or all of which pertain to the old code:

> --- /dev/null
> +++ b/kernel/semaphore.c
> @@ -0,0 +1,204 @@
> +/*
> + * Copyright (c) 2008 Intel Corporation
> + * Author: Matthew Wilcox <willy@linux.intel.com>
> + *
> + * Distributed under the terms of the GNU GPL, version 2
> + */
> +
> +#include <linux/compiler.h>
> +#include <linux/kernel.h>
> +#include <linux/module.h>
> +#include <linux/sched.h>
> +#include <linux/semaphore.h>
> +#include <linux/spinlock.h>
> +
> +/*
> + * Some notes on the implementation:
> + *
> + * down_trylock() and up() can be called from interrupt context.
> + * So we have to disable interrupts when taking the lock.
> + *
> + * The ->count variable, if positive, defines how many more tasks can
> + * acquire the semaphore.  If negative, it represents how many tasks are
> + * waiting on the semaphore (*).  If zero, no tasks are waiting, and no more
> + * tasks can acquire the semaphore.
> + *
> + * (*) Except for the window between one task calling up() and the task
> + * sleeping in a __down_common() waking up.  In order to avoid a third task
> + * coming in and stealing the second task's wakeup, we leave the ->count
> + * negative.  If we have a more complex situation, the ->count may become
> + * zero or negative (eg a semaphore with count = 2, three tasks attempt to
> + * acquire it, one sleeps, two finish and call up(), the second task to call
> + * up() notices that the list is empty and just increments count).
> + */
> +
> +static void noinline __down(struct semaphore *sem);
> +static int noinline __down_interruptible(struct semaphore *sem);
> +static int noinline __down_killable(struct semaphore *sem);
> +static void noinline __up(struct semaphore *sem);

We conventionally do `static inline void' rather than `static void inline',
so I guess we should do `static noinline void' too.

> +int down_interruptible(struct semaphore *sem)
> +{
> +	int result = 0;
> +	might_sleep();
> +	spin_lock_irq(&sem->lock);
> +	if (unlikely(sem->count-- <= 0))
> +		result = __down_interruptible(sem);
> +	spin_unlock_irq(&sem->lock);
> +	return result;
> +}
> +EXPORT_SYMBOL(down_interruptible);

Some functions leave no blank line between end-of-locals and start-of-code...

> +int down_killable(struct semaphore *sem)
> +{
> +	int result = 0;
> +	might_sleep();
> +	spin_lock_irq(&sem->lock);
> +	if (unlikely(sem->count-- <= 0))
> +		result = __down_killable(sem);
> +	spin_unlock_irq(&sem->lock);
> +	return result;
> +}
> +EXPORT_SYMBOL(down_killable);
> +
> +/**
> + * down_trylock - try to acquire the semaphore, without waiting
> + * @sem: the semaphore to be acquired
> + *
> + * Try to acquire the semaphore atomically.  Returns 0 if the mutex has
> + * been acquired successfully and 1 if it is contended.
> + *
> + * NOTE: This return value is inverted from both spin_trylock and
> + * mutex_trylock!  Be careful about this when converting code.
> + *
> + * Unlike mutex_trylock, this function can be used from interrupt context,
> + * and the semaphore can be released by any task or interrupt.
> + */
> +int down_trylock(struct semaphore *sem)
> +{
> +	unsigned long flags;
> +	int count;
> +
> +	spin_lock_irqsave(&sem->lock, flags);
> +	count = sem->count - 1;
> +	if (likely(count >= 0))
> +		sem->count = count;
> +	spin_unlock_irqrestore(&sem->lock, flags);
> +	return (count < 0);
> +}
> +EXPORT_SYMBOL(down_trylock);

whereas some others do leave a blank line.

I think the 51% consensus is that the latter form is preferred.

> +void up(struct semaphore *sem)
> +{
> +	unsigned long flags;
> +
> +	spin_lock_irqsave(&sem->lock, flags);
> +	if (likely(sem->count >= 0))
> +		sem->count++;
> +	else
> +		__up(sem);
> +	spin_unlock_irqrestore(&sem->lock, flags);
> +}
> +EXPORT_SYMBOL(up);
> +
> +/* Functions for the contended case */
> +
> +struct semaphore_waiter {
> +	struct list_head list;
> +	struct task_struct *task;
> +	int up;
> +};
> +
> +/*
> + * Wake up a process waiting on a semaphore.  We need to call this from both
> + * __up and __down_common as it's possible to race a task into the semaphore
> + * if it comes in at just the right time between two tasks calling up() and
> + * a third task waking up.  This function assumes the wait_list is already
> + * checked for being non-empty.
> + */
> +static void noinline __sched __up_down_common(struct semaphore *sem)
> +{
> +	struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
> +						struct semaphore_waiter, list);
> +	list_del(&waiter->list);
> +	waiter->up = 1;

hm.  Do we need a barrier here?

> +	wake_up_process(waiter->task);
> +}
> +
> +/*
> + * Because this function is inlined, the 'state' parameter will be constant,
> + * and thus optimised away by the compiler.
> + */
> +static inline int __sched __down_common(struct semaphore *sem, long state)
> +{
> +	int result = 0;
> +	struct task_struct *task = current;
> +	struct semaphore_waiter waiter;
> +
> +	list_add_tail(&waiter.list, &sem->wait_list);
> +	waiter.task = task;
> +	waiter.up = 0;
> +
> +	for (;;) {
> +		if (unlikely((state == TASK_INTERRUPTIBLE &&
> +					signal_pending(task)) ||
> +			     (state == TASK_KILLABLE &&
> +					fatal_signal_pending(task))))
> +			goto interrupted;
> +		__set_task_state(task, state);
> +		spin_unlock_irq(&sem->lock);
> +		schedule();
> +		spin_lock_irq(&sem->lock);
> +		if (waiter.up)
> +			goto woken;
> +	}
> +
> + interrupted:
> +	list_del(&waiter.list);
> +	result = -EINTR;
> + woken:
> +	/*
> +	 * Account for the process which woke us up.  For the case where
> +	 * we're interrupted, we need to increment the count on our own
> +	 * behalf.  I don't believe we can hit the case where the
> +	 * sem->count hits zero, *and* there's a second task sleeping,
> +	 * but it doesn't hurt, that's not a commonly exercised path and
> +	 * it's not a performance path either.
> +	 */
> +	if (unlikely((++sem->count >= 0) && !list_empty(&sem->wait_list)))
> +		__up_down_common(sem);
> +	return result;
> +}

That is one huuuuuuuge inline.  Are we sure it's justified?

> +static void noinline __sched __down(struct semaphore *sem)
> +{
> +	__down_common(sem, TASK_UNINTERRUPTIBLE);
> +}
> +
> +static int noinline __sched __down_interruptible(struct semaphore *sem)
> +{
> +	return __down_common(sem, TASK_INTERRUPTIBLE);
> +}
> +
> +static int noinline __sched __down_killable(struct semaphore *sem)
> +{
> +	return __down_common(sem, TASK_KILLABLE);
> +}
> +
> +static void noinline __sched __up(struct semaphore *sem)
> +{
> +	if (unlikely(list_empty(&sem->wait_list)))
> +		sem->count++;
> +	else
> +		__up_down_common(sem);
> +}
> Please read the FAQ at  http://www.tux.org/lkml/