Re: Resuming work on new semaphore

[Alexander Monakov <amonakov@ispras.ru>]

On Mon, 6 Apr 2015, Alexander Monakov wrote:
> One other thing to consider.  In the absence of concurrent operations on the
> semaphore, return value of sem_getvalue should be equal to the number of times
> sem_trywait will indicate success when called repeatedly.  So if the
> implementation performs post-stealing in trywait, it should return the higher
> bound as semaphore value.  Likewise for timedwait.

If we accept the above, it follows that in the new implementation getvalue
should return not max(0, val[0] + val[1]), but rather max(0, val[0]) + val[1].

So incorporating the above into getvalue, and Rich's scheme of mirroring
timedwait into trywait (as he pointed out on IRC, my approach does not
properly work with two concurrent posters), we get:

int sem_getvalue(sem_t *restrict sem, int *restrict valp)
{
	a_barrier(); // Memory holding the semaphore should not be stale
	int val = sem->__val[0];
	val = val < 0 ? 0 : val;
	*valp = val + sem->__val[1];
	return 0;
}

int sem_trywait(sem_t *sem)
{
	if (a_fetch_add(sem->__val, -1) > 0)
		return 0;
	for (;;) {
		int wak = sem->__val[1];
		if (wak > 0 && wak == a_cas(sem->__val+1, wak, wak-1))
			return 0;
		int val = sem->__val[0];
		if (val < 0 && val == a_cas(sem->__val, val, val+1))
			break;
		a_spin();
	}
	errno = EAGAIN;
	return -1;
}

int sem_post(sem_t *sem)
{
	int val;
	do val = sem->__val[0];
	while (val != a_cas(sem->__val, val, val+!!(val<SEM_VALUE_MAX)));
	if (val < 0) {
		int priv = sem->__val[2];
		a_inc(sem->__val+1);
		__wake(sem->__val+1, 1, priv);
	}
	if (val < SEM_VALUE_MAX) return 0;
	errno = EOVERFLOW;
	return -1;
}

int sem_timedwait(sem_t *restrict sem, const struct timespec *restrict at)
{
	pthread_testcancel();

	// Do not spin if already contended (val0 is negative)
	for (int spins = 1000; spins && sem->__val[0] == 0; spins--)
		a_spin();

	if (a_fetch_add(sem->__val, -1) > 0)
		return 0;

	int priv = sem->__val[2], e = 0, ee, cs;
	pthread_setcancelstate(PTHREAD_CANCEL_MASKED, &cs);

	do {
		ee = __timedwait_cp(sem->__val+1, 0, CLOCK_REALTIME, at, priv);
		int wak = sem->__val[1];
		if (wak > 0 && wak == a_cas(sem->__val+1, wak, wak-1))
			goto done;
	} while (!ee || ee == EINTR);

	// Upon returning from wait with an error, either discount ourselves as a waiter
	// by incrementing negative val0, and propagate error, or consume a racing post
	// if val0 is non-negative, and suppress error.
	for (;;) {
		int val = sem->__val[0];
		if (val < 0 && val == a_cas(sem->__val, val, val+1))
			break;
		int wak = sem->__val[1];
		if (wak > 0 && wak == a_cas(sem->__val+1, wak, wak-1))
			goto done;
		a_spin();
	}
	e = ee;
done:
	pthread_setcancelstate(cs, 0);

	if (!e) return 0;

	if (e == ECANCELED) {
		pthread_testcancel();
		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, 0);
	}

	errno = e;
	return -1;
}