pthread condition variables

OS_ALWAYS_INLINE
static inline pthread_ulock_cond_state_u *
_pthread_ulock_cond_state(pthread_cond_t *cond)
{
	return (pthread_ulock_cond_state_u *)&cond->c_seq[cond->misalign];
}

int
pthread_condattr_init(pthread_condattr_t *attr)
{
	attr->sig = _PTHREAD_COND_ATTR_SIG;
	attr->pshared = _PTHREAD_DEFAULT_PSHARED;
	return 0;
}

int
pthread_condattr_destroy(pthread_condattr_t *attr)
{
	attr->sig = _PTHREAD_NO_SIG;
	return 0;
}

int
pthread_condattr_getpshared(const pthread_condattr_t *attr, int *pshared)
{
	int res = EINVAL;
	if (attr->sig == _PTHREAD_COND_ATTR_SIG) {
		*pshared = (int)attr->pshared;
		res = 0;
	}
	return res;
}

int
pthread_condattr_setpshared(pthread_condattr_t *attr, int pshared)
{
	int res = EINVAL;
	if (attr->sig == _PTHREAD_COND_ATTR_SIG) {
		if (pshared == PTHREAD_PROCESS_PRIVATE || pshared == PTHREAD_PROCESS_SHARED) {
			attr->pshared = pshared;
			res = 0;
		}
	}
	return res;
}

int
pthread_cond_timedwait_relative_np(pthread_cond_t *cond, pthread_mutex_t *mutex,
		const struct timespec *abstime)
{
	return _pthread_cond_wait(cond, mutex, abstime, 1,
			PTHREAD_CONFORM_UNIX03_NOCANCEL);
}

OS_ALWAYS_INLINE
static inline int
_pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr,
		uint32_t sig)
{
	volatile uint64_t *c_lsseqaddr;
	volatile uint32_t *c_lseqcnt, *c_useqcnt, *c_sseqcnt;

	cond->busy = NULL;
	cond->c_seq[0] = 0;
	cond->c_seq[1] = 0;
	cond->c_seq[2] = 0;
	cond->unused = 0;

	// TODO: PTHREAD_STRICT candidate
	cond->misalign = (((uintptr_t)&cond->c_seq[0]) & 0x7) != 0;
	COND_GETSEQ_ADDR(cond, &c_lsseqaddr, &c_lseqcnt, &c_useqcnt, &c_sseqcnt);
	*c_sseqcnt = PTH_RWS_CV_CBIT; // set Sword to 0c

	if (attr) {
		cond->pshared = attr->pshared;
	} else {
		cond->pshared = _PTHREAD_DEFAULT_PSHARED;
	}

	// Ensure all contents are properly set before setting signature.
#if defined(__LP64__)
	// For binary compatibility reasons we cannot require natural alignment of
	// the 64bit 'sig' long value in the struct. rdar://problem/21610439
	cond->sig._pad = 0;
… more in source

OS_ALWAYS_INLINE
static int
_pthread_cond_check_signature(pthread_cond_t *cond, uint32_t sig_current,
		uint32_t *sig_inout)
{
	int res = 0;
	switch (sig_current) {
	case _PTHREAD_COND_SIG_init:
		__builtin_unreachable();
		break;
	case _PTHREAD_COND_SIG_pristine:
		if (*sig_inout != _PTHREAD_COND_SIG_pristine) {
			os_atomic_store(&cond->sig.val, *sig_inout, relaxed);
		}
		break;
	case _PTHREAD_COND_SIG_psynch:
	case _PTHREAD_COND_SIG_ulock:
		if (*sig_inout == _PTHREAD_COND_SIG_pristine) {
			*sig_inout = sig_current;
		} else if (*sig_inout != sig_current) {
			PTHREAD_INTERNAL_CRASH(0, "Mixed ulock and psych condvar use");
		}
		break;
	default:
		// TODO: PTHREAD_STRICT candidate
		res = EINVAL;
		break;
	}

	return res;
… more in source

OS_NOINLINE
static int
_pthread_cond_check_init_slow(pthread_cond_t *cond, uint32_t *sig_inout)
{
	int res;
	_pthread_lock_lock(&cond->lock);

	uint32_t sig_current = os_atomic_load(&cond->sig.val, relaxed);
	if (sig_current == _PTHREAD_COND_SIG_init) {
		res = _pthread_cond_init(cond, NULL, *sig_inout);
	} else {
		res = _pthread_cond_check_signature(cond, sig_current, sig_inout);
	}

	_pthread_lock_unlock(&cond->lock);
	return res;
}

OS_ALWAYS_INLINE
static inline int
_pthread_cond_check_init(pthread_cond_t *cond, uint32_t *sig_inout)
{
	uint32_t sig_current = os_atomic_load(&cond->sig.val, relaxed);
	if (sig_current == _PTHREAD_COND_SIG_init) {
		return _pthread_cond_check_init_slow(cond, sig_inout);
	} else {
		return _pthread_cond_check_signature(cond, sig_current, sig_inout);
	}
}

PTHREAD_NOEXPORT_VARIANT
int
pthread_cond_destroy(pthread_cond_t *cond)
{
	int res = EINVAL;
	uint32_t sig = os_atomic_load(&cond->sig.val, relaxed);
	switch (sig) {
	case _PTHREAD_COND_SIG_psynch:
		_pthread_lock_lock(&cond->lock);

		uint64_t oldval64, newval64;
		uint32_t lcntval, ucntval, scntval;
		volatile uint64_t *c_lsseqaddr;
		volatile uint32_t *c_lseqcnt, *c_useqcnt, *c_sseqcnt;

		COND_GETSEQ_ADDR(cond, &c_lsseqaddr, &c_lseqcnt, &c_useqcnt, &c_sseqcnt);

		do {
			lcntval = *c_lseqcnt;
			ucntval = *c_useqcnt;
			scntval = *c_sseqcnt;

			// validate it is not busy
			if ((lcntval & PTHRW_COUNT_MASK) != (scntval & PTHRW_COUNT_MASK)) {
				//res = EBUSY;
				break;
			}
			oldval64 = (((uint64_t)scntval) << 32);
			oldval64 |= lcntval;
			newval64 = oldval64;
… more in source

OS_ALWAYS_INLINE
static inline int
_pthread_psynch_cond_signal(pthread_cond_t *cond, bool broadcast,
		mach_port_t thread)
{
	uint32_t updateval;
	uint32_t diffgen;
	uint32_t ulval;

	uint64_t oldval64, newval64;
	uint32_t lcntval, ucntval, scntval;
	volatile uint64_t *c_lsseqaddr;
	volatile uint32_t *c_lseqcnt, *c_useqcnt, *c_sseqcnt;

	int retry_count = 0, uretry_count = 0;
	int ucountreset = 0;

	COND_GETSEQ_ADDR(cond, &c_lsseqaddr, &c_lseqcnt, &c_useqcnt, &c_sseqcnt);

	bool retry;
	do {
		retry = false;

		lcntval = *c_lseqcnt;
		ucntval = *c_useqcnt;
		scntval = *c_sseqcnt;
		diffgen = 0;
		ulval = 0;

		if (((lcntval & PTHRW_COUNT_MASK) == (scntval & PTHRW_COUNT_MASK)) ||
… more in source

OS_ALWAYS_INLINE
static inline int
_pthread_ulock_cond_signal(pthread_cond_t *cond, bool broadcast,
		mach_port_t thread)
{
	pthread_ulock_cond_state_u *state = _pthread_ulock_cond_state(cond);

	pthread_ulock_cond_state_u oldstate, newstate;
	// release to pair with acquire after wait
	os_atomic_rmw_loop(&state->val, oldstate.val, newstate.val, release, {
		if (!oldstate.waiters || oldstate.waiters == oldstate.signal) {
			os_atomic_rmw_loop_give_up(return 0);
		}

		newstate = (pthread_ulock_cond_state_u){
			.seq = oldstate.seq + 1,
			.waiters = oldstate.waiters,
			.signal = broadcast ? oldstate.waiters :
					MIN(oldstate.signal + 1, oldstate.waiters),
		};
	});

	PTHREAD_TRACE(ulcond_signal, cond, oldstate.val, newstate.val, broadcast);

	// Priority hole: if we're pre-empted here, nobody else can signal the
	// waiter we took responsibility for signaling by incrementing the signal
	// count.

	if (oldstate.signal < oldstate.waiters) {
		uint32_t wake_op = UL_COMPARE_AND_WAIT | ULF_NO_ERRNO;
… more in source

OS_ALWAYS_INLINE
static inline int
_pthread_cond_signal(pthread_cond_t *cond, bool broadcast, mach_port_t thread)
{
	uint32_t sig = _PTHREAD_COND_SIG_pristine;
	int res = _pthread_cond_check_init(cond, &sig);
	if (res != 0 || sig == _PTHREAD_COND_SIG_pristine) {
		return res;
	}

	switch (sig) {
	case _PTHREAD_COND_SIG_psynch:
		return _pthread_psynch_cond_signal(cond, broadcast, thread);
	case _PTHREAD_COND_SIG_ulock:
		return _pthread_ulock_cond_signal(cond, broadcast, thread);
	default:
		PTHREAD_INTERNAL_CRASH(sig, "impossible cond signature");
	}
}

PTHREAD_NOEXPORT_VARIANT
int
pthread_cond_broadcast(pthread_cond_t *cond)
{
	return _pthread_cond_signal(cond, true, MACH_PORT_NULL);
}

PTHREAD_NOEXPORT_VARIANT
int
pthread_cond_signal_thread_np(pthread_cond_t *cond, pthread_t thread)
{
	mach_port_t mp = MACH_PORT_NULL;
	if (thread) {
		mp = pthread_mach_thread_np((_Nonnull pthread_t)thread);
	}
	return _pthread_cond_signal(cond, false, mp);
}

PTHREAD_NOEXPORT_VARIANT
int
pthread_cond_signal(pthread_cond_t *cond)
{
	return _pthread_cond_signal(cond, false, MACH_PORT_NULL);
}

static int
_pthread_psynch_cond_wait(pthread_cond_t *cond,
			pthread_mutex_t *mutex,
			const struct timespec *then,
			pthread_conformance_t conforming)
{
	uint32_t mtxgen, mtxugen, flags=0, updateval;
	uint32_t lcntval, ucntval, scntval;
	uint32_t nlval, ulval, savebits;
	volatile uint64_t *c_lsseqaddr;
	volatile uint32_t *c_lseqcnt, *c_useqcnt, *c_sseqcnt;
	uint64_t oldval64, newval64, mugen, cvlsgen;
	uint32_t *npmtx = NULL;

	COND_GETSEQ_ADDR(cond, &c_lsseqaddr, &c_lseqcnt, &c_useqcnt, &c_sseqcnt);

	do {
		lcntval = *c_lseqcnt;
		ucntval = *c_useqcnt;
		scntval = *c_sseqcnt;

		oldval64 = (((uint64_t)scntval) << 32);
		oldval64 |= lcntval;

		/* remove c and p bits on S word */
		savebits = scntval & PTH_RWS_CV_BITSALL;
		ulval = (scntval & PTHRW_COUNT_MASK);
		nlval = lcntval + PTHRW_INC;
		newval64 = (((uint64_t)ulval) << 32);
		newval64 |= nlval;
… more in source

static int
_pthread_ulock_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex,
		const struct timespec *then, pthread_conformance_t conforming)
{
	bool cancelable = (conforming == PTHREAD_CONFORM_UNIX03_CANCELABLE);

	uint64_t timeout_ns = 0;
	if (then->tv_sec || then->tv_nsec) {
		// psynch compatibility: cast and bitwise-truncate tv_nsec
		uint64_t fraction_ns = ((uint32_t)then->tv_nsec) & 0x3fffffff;
		if (os_mul_and_add_overflow(then->tv_sec, NSEC_PER_SEC, fraction_ns,
				&timeout_ns)) {
			// saturate (can't wait longer than 584 years...)
			timeout_ns = UINT64_MAX;
		}
	}

	pthread_ulock_cond_state_u *state = _pthread_ulock_cond_state(cond);

	pthread_ulock_cond_state_u origstate = {
		.val = os_atomic_add(&state->val, _PTHREAD_COND_WAITERS_INC, relaxed)
	};

	int rc = _pthread_mutex_ulock_unlock(mutex);
	if (rc) {
		return _pthread_ulock_cond_wait_complete(state, NULL, rc);
	}

	PTHREAD_TRACE(ulcond_wait, cond, origstate.val, timeout_ns, 0);
… more in source

static int
_pthread_ulock_cond_wait_complete(pthread_ulock_cond_state_u *state,
		pthread_mutex_t *mutex, int rc)
{
	if (mutex) {
		// XXX Check this return value?  Historically we haven't, but if rc == 0
		// we could promote the return value to this one.
		_pthread_mutex_ulock_lock(mutex, false);
	}

	pthread_ulock_cond_state_u oldstate, newstate;
	// acquire to pair with release upon signal
	os_atomic_rmw_loop(&state->val, oldstate.val, newstate.val, acquire, {
		newstate = (pthread_ulock_cond_state_u){
			.seq = oldstate.seq,
			.waiters = oldstate.waiters - 1,
			.signal = oldstate.signal ? oldstate.signal - 1 : 0,
		};
	});

	return rc;
}

PTHREAD_NOEXPORT OS_NOINLINE
int
_pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex,
			const struct timespec *abstime, int isRelative,
			pthread_conformance_t conforming)
{
	int res;
	struct timespec then = { 0, 0 };
	bool timeout_elapsed = false;

	if (!_pthread_mutex_check_signature(mutex) &&
			!_pthread_mutex_check_signature_init(mutex)) {
		return EINVAL;
	}

	bool ulock = _pthread_mutex_uses_ulock(mutex);
	uint32_t sig = ulock ? _PTHREAD_COND_SIG_ulock : _PTHREAD_COND_SIG_psynch;
	res = _pthread_cond_check_init(cond, &sig);
	if (res != 0) {
		return res;
	}

	if (conforming == PTHREAD_CONFORM_UNIX03_CANCELABLE) {
		pthread_testcancel();
	}

	/* send relative time to kernel */
	if (abstime) {
		if (abstime->tv_nsec < 0 || abstime->tv_nsec >= NSEC_PER_SEC) {
			// TODO: PTHREAD_STRICT candidate
… more in source

static void
_pthread_ulock_cond_cleanup(void *arg)
{
	struct pthread_ulock_cond_cancel_ctx_s *ctx = arg;
	pthread_ulock_cond_state_u *state = _pthread_ulock_cond_state(ctx->cond);

	(void)_pthread_ulock_cond_wait_complete(state, ctx->mutex, 0);

	// "A thread that has been unblocked because it has been canceled while
	// blocked in a call to pthread_cond_timedwait() or pthread_cond_wait()
	// shall not consume any condition signal that may be directed concurrently
	// at the condition variable if there are other threads blocked on the
	// condition variable."
	//
	// Since we have no way to know if we've eaten somebody else's signal, just
	// signal again pessimistically.
	pthread_cond_signal(ctx->cond);
}

static void
_pthread_psynch_cond_cleanup(void *arg)
{
	pthread_cond_t *cond = (pthread_cond_t *)arg;
	pthread_t thread = pthread_self();
	pthread_mutex_t *mutex;

// 4597450: begin
	if (!thread->canceled) {
		return;
	}
// 4597450: end

	mutex = cond->busy;

	// add unlock ref to show one less waiter
	_pthread_cond_updateval(cond, mutex, thread->cancel_error, 0);

	/*
	** Can't do anything if this fails -- we're on the way out
	*/
	if (mutex != NULL) {
		(void)pthread_mutex_lock(mutex);
	}
}

static void
_pthread_cond_updateval(pthread_cond_t *cond, pthread_mutex_t *mutex,
		int error, uint32_t updateval)
{
	int needclearpre;

	uint32_t diffgen, nsval;
	uint64_t oldval64, newval64;
	uint32_t lcntval, ucntval, scntval;
	volatile uint64_t *c_lsseqaddr;
	volatile uint32_t *c_lseqcnt, *c_useqcnt, *c_sseqcnt;

	if (error != 0) {
		updateval = PTHRW_INC;
		if (error & ECVCLEARED) {
			updateval |= PTH_RWS_CV_CBIT;
		}
		if (error & ECVPREPOST) {
			updateval |= PTH_RWS_CV_PBIT;
		}
	}

	COND_GETSEQ_ADDR(cond, &c_lsseqaddr, &c_lseqcnt, &c_useqcnt, &c_sseqcnt);

	do {
		lcntval = *c_lseqcnt;
		ucntval = *c_useqcnt;
		scntval = *c_sseqcnt;
		nsval = 0;
		needclearpre = 0;
… more in source

PTHREAD_NOEXPORT_VARIANT
int
pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)
{
	_pthread_lock_init(&cond->lock);
	return _pthread_cond_init(cond, attr, _PTHREAD_COND_SIG_pristine);
}

struct pthread_ulock_cond_cancel_ctx_s {
	pthread_cond_t *cond;
	pthread_mutex_t *mutex;
}

typedef union { uint64_t val; struct { uint32_t seq; uint16_t waiters; uint16_t signal; }; } pthread_ulock_cond_state_u;