[RFC] slot allocator - waitqueue use review needed (Re: Orangefs ABI documentation)

[Al Viro <viro@ZenIV.linux.org.uk>]

FWIW, I think I have a kinda-sorta solution for bufmap slot allocation/waiting;
if somebody has a better idea, I would love to drop the variant below.  And
I would certainly appreciate review - I hate messing with waitqueue primitives
and I know how easy it is to fuck those up ;-/

Below is a mockup of that thing:

/*  Three possible states: absent, installed and shutting down.
 *  install(map, count, bitmap) sets it up
 *  get(map)/put(map, slot) allocate and free resp.
 *  mark_dead(map) moves to shutdown state - no new allocations succeed until
 *  we reinstall it.
 *  run_down(map) waits for all allocations to be released; in the end, we
 *  are in the "absent" state again.
 *
 *  get() is not allowed to take longer than slot_timeout_secs seconds total;
 *  if the thing gets shut down and reinstalled during the wait, we are OK
 *  as long as reinstall comes within restart_timeout_secs.  For orangefs
 *  those default to 15 minutes and 30 seconds resp...
 */
struct slot_map {
	int c;			// absent -> -1
				// installed and full -> 0
				// installed with n slots free -> n
				// shutting down, with n slots in use -> -1-n
	wait_queue_head_t q;	// q.lock protects everything here.
	int count;
	unsigned long *map;
};

void install(struct slot_map *m, int count, unsigned long *map)
{
	spin_lock(&m->q.lock);
	m->c = m->count = count;
	m->map = map;
	wake_up_all_locked(&m->q);
	spin_unlock(&m->q.lock);
}

void mark_killed(struct slot_map *m)
{
	spin_lock(&m->q.lock);
	m->c -= m->count + 1;
	spin_unlock(&m->q.lock);
}

void run_down(struct slot_map *m)
{
	DEFINE_WAIT(wait);
	spin_lock(&m->q.lock);
#if 0
	// we don't have wait_event_locked(); might be worth adding.
	wait_event_locked(&m->q, m->c == -1);
#else
	// or we can open-code it
	if (m->c != -1) {
		for (;;) {
			if (likely(list_empty(&wait.task_list)))
				__add_wait_queue_tail(&m->q, &wait);
			set_current_state(TASK_UNINTERRUPTIBLE);

			if (m->c == -1)
				break;

			spin_unlock(&m->q.lock);
			schedule();
			spin_lock(&m->q.lock);
		}
		__remove_wait_queue(&m->q, &wait);
		__set_current_state(TASK_RUNNING);
	}
#endif
	m->map = NULL;
	spin_unlock(&m->q.lock);
}

void put(struct slot_map *m, int slot)
{
	int v;
	spin_lock(&m->q.lock);
	__clear_bit(slot, m->map);
	v = ++m->c;
	if (unlikely(v == 1))	/* no free slots -> one free slot */
		wake_up_locked(&m->q);
	else if (unlikely(v == -1))	/* finished dying */
		wake_up_all_locked(&m->q);
	spin_unlock(&m->q.lock);
}

static int wait_for_free(struct slot_map *m)
{
	long left = slot_timeout_secs * HZ;
	DEFINE_WAIT(wait);

#if 0
	// the trouble is, there's no wait_event_interruptible_timeout_locked()
	// might be worth adding...
	do {
		if (m->c > 0)
			break;
		if (m->c < 0) {
			/* we are waiting for map to be installed */
			/* it would better be there soon, or we go away */
			long n = left, t;
			if (n > restart_timeout_secs * HZ)
				n = restart_timeout_secs * HZ;
			t = wait_event_interruptible_timeout_locked(&m->q,
						m->c > 0, n);
			if (unlikely(t < 0) || (!t && m->c < 0))
				left = t;
			else
				left = t + (left - n);
		} else {
			/* just waiting for a slot to come free */
			left = wait_event_interruptible_timeout_locked(&m->q,
						m->c > 0, left);
		}
	} while (left > 0);
#else
	// or we can open-code it
	do {
		if (likely(list_empty(&wait.task_list)))
			__add_wait_queue_tail_exclusive(&m->q, &wait);
		set_current_state(TASK_INTERRUPTIBLE);

		if (m->c > 0)
			break;

		if (m->c < 0) {
			/* we are waiting for map to be installed */
			/* it would better be there soon, or we go away */
			long n = left, t;
			if (n > ORANGEFS_BUFMAP_WAIT_TIMEOUT_SECS * HZ)
				n = ORANGEFS_BUFMAP_WAIT_TIMEOUT_SECS * HZ;
			spin_unlock(&m->q.lock);
			t = schedule_timeout(n);
			spin_lock(&m->q.lock);
			if (unlikely(t < 0) || (!t && m->c < 0))
				left = t;
			else
				left = t + (left - n);
		} else {
			/* just waiting for a slot to come free */
			spin_unlock(&m->q.lock);
			left = schedule_timeout(left);
			spin_lock(&m->q.lock);
		}
	} while (left > 0);

	__remove_wait_queue(&m->q, &wait);
	__set_current_state(TASK_RUNNING);
#endif
	if (likely(left > 0))
		return 0;

	return left < 0 ? -EINTR : -ETIMEDOUT;
}

int get(struct slot_map *m)
{
	int res = 0;
	spin_lock(&m->q.lock);
	if (unlikely(m->c <= 0))
		res = wait_for_free(m);
	if (likely(!res)) {
		m->c--;
		res = find_first_zero_bit(m->map, m->count);
		__set_bit(res, m->map);
	}
	spin_unlock(&m->q.lock);
	return res;
}