First draft of new synccall, seems to be working

First draft of new synccall, seems to be working

[Rich Felker]

[-- Attachment #1: Type: text/plain, Size: 820 bytes --]

I just wrote the first draft of the new __synccall, used for
multithreaded set*id(). All the concepts seem sound and seem to be
working. It's missing the code to block creation of unboundedly more
threads while __synccall is running, but otherwise everything's in
place. Both the /proc/self/task processing logic and the PI futex
logic seem to be working as expected. I haven't yet tested in the
presence of multiple realtime priorities where there's a risk of some
threads never getting a chance to run; I need a framework with a
high-RT-prio supervisor process to kill the test if it runs too long,
because I don't want to be rebooting boxes over and over to test. :)

Attached is the current draft of the code, as a source file rather
than a patch since the whole file is basically rewritten. Comments
welcome.

Rich

[-- Attachment #2: synccall.c --]
[-- Type: text/plain, Size: 3767 bytes --]

#include "pthread_impl.h"
#include <semaphore.h>
#include <unistd.h>
#include <dirent.h>
#include <string.h>
#include <ctype.h>
#include "futex.h"
#include <../dirent/__dirent.h>

static struct chain {
	struct chain *next;
	int tid;
	sem_t start_sem, next_sem, finish_sem;
} *head;

static int synccall_lock[2];
static int data_lock[2];
static int cur_tid;
static void (*callback)(void *), *context;

static void handler(int sig)
{
	struct chain ch;
	int old_errno = errno;

	sem_init(&ch.start_sem, 0, 0);
	sem_init(&ch.next_sem, 0, 0);
	sem_init(&ch.finish_sem, 0, 0);

	LOCK(data_lock);
	ch.tid = __syscall(SYS_gettid);
	ch.next = head;
	head = &ch;
	UNLOCK(data_lock);
	if (a_cas(&cur_tid, ch.tid, 0) == (ch.tid | 0x80000000))
		__syscall(SYS_futex, &cur_tid, FUTEX_UNLOCK_PI|FUTEX_PRIVATE);

	sem_wait(&ch.start_sem);
	callback(context);
	sem_post(&ch.next_sem);
	sem_wait(&ch.finish_sem);

	errno = old_errno;
}


static int is_member(struct chain *p, int tid)
{
	while (p) {
		if (p->tid == tid) return 1;
		p = p->next;
	}
	return 0;
}

void __synccall(void (*func)(void *), void *ctx)
{
	sigset_t oldmask;
	int cs;
	int pid, self;
	int i;
	DIR dir = {0};
	struct dirent *de;
	struct sigaction sa = { .sa_flags = 0, .sa_handler = handler };
	struct chain *cp, *next;
	struct timespec ts;

	__block_all_sigs(&oldmask);
	LOCK(synccall_lock);
	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cs);

	LOCK(data_lock);
	head = 0;
	callback = func;
	context = ctx;
	UNLOCK(data_lock);

	if (!libc.threaded) goto single_threaded;

	/* Block even impl-internal signals */
	memset(&sa.sa_mask, -1, sizeof sa.sa_mask);
	__libc_sigaction(SIGSYNCCALL, &sa, 0);

	pid = __syscall(SYS_getpid);
	self = __syscall(SYS_gettid);

	/* Since opendir is not AS-safe, the DIR needs to be setup manually
	 * in automatic storage. Thankfully this is easy. */
	dir.fd = open("/proc/self/task", O_RDONLY|O_DIRECTORY|O_CLOEXEC);
	if (dir.fd < 0) goto out;

	/* Initially send one signal per counted thread. But since we can't
	 * synchronize with thread creation/exit here, there could be too
	 * few signals. This initial signaling is just an optimization, not
	 * part of the logic. */
	for (i=libc.threads_minus_1; i; i--)
		__syscall(SYS_kill, pid, SIGSYNCCALL);

	/* Loop scanning the kernel-provided thread list until it shows no
	 * threads that have not already replied to the signal. */
	for (;;) {
		int miss_cnt = 0;
		while ((de = readdir(&dir))) {
			if (!isdigit(de->d_name[0])) continue;
			int tid = atoi(de->d_name);
			if (tid == self) continue;
			LOCK(data_lock);
			if (is_member(head, tid)) {
				UNLOCK(data_lock);
				continue;
			}
			UNLOCK(data_lock);

			__syscall(SYS_tgkill, pid, tid, SIGSYNCCALL);

			/* The FUTEX_LOCK_PI operation is used to loan priority
			 * to the target thread, which otherwise may be unable
			 * to run. Timeout is necessary because there is a race
			 * condition where the tid may be reused by a different
			 * process. */
			clock_gettime(CLOCK_REALTIME, &ts);
			ts.tv_nsec += 10000000;
			if (ts.tv_nsec >= 1000000000) {
				ts.tv_sec++;
				ts.tv_nsec -= 1000000000;
			}
			a_store(&cur_tid, tid);
			__syscall(SYS_futex, &cur_tid, FUTEX_LOCK_PI|FUTEX_PRIVATE, 0, &ts);

			LOCK(data_lock);
			if (cur_tid != self && !is_member(head, tid))
				miss_cnt++;
			UNLOCK(data_lock);
		}
		if (!miss_cnt) break;
		rewinddir(&dir);
	}
	close(dir.fd);

	for (cp=head; cp; cp=next) {
		next = cp->next;
		sem_post(&cp->start_sem);
		sem_wait(&cp->next_sem);
	}

	sa.sa_handler = SIG_IGN;
	__libc_sigaction(SIGSYNCCALL, &sa, 0);

single_threaded:
	func(ctx);

	for (cp=head; cp; cp=next) {
		next = cp->next;
		sem_post(&cp->finish_sem);
	}

out:
	pthread_setcancelstate(cs, 0);
	UNLOCK(synccall_lock);
	__restore_sigs(&oldmask);
}

Re: First draft of new synccall, seems to be working

[Rich Felker]

[-- Attachment #1: Type: text/plain, Size: 1233 bytes --]

On Wed, Jan 14, 2015 at 11:33:15PM -0500, Rich Felker wrote:
> I just wrote the first draft of the new __synccall, used for
> multithreaded set*id(). All the concepts seem sound and seem to be
> working. It's missing the code to block creation of unboundedly more
> threads while __synccall is running, but otherwise everything's in
> place. Both the /proc/self/task processing logic and the PI futex
> logic seem to be working as expected. I haven't yet tested in the
> presence of multiple realtime priorities where there's a risk of some
> threads never getting a chance to run; I need a framework with a
> high-RT-prio supervisor process to kill the test if it runs too long,
> because I don't want to be rebooting boxes over and over to test. :)
> 
> Attached is the current draft of the code, as a source file rather
> than a patch since the whole file is basically rewritten. Comments
> welcome.

I had some synchronization logic errors (which would have been masked
by the timeout anyway, but still wrong) in the old version which I've
corrected, and I realized that most of the internal locking could be
replaced with atomics, which I've done. I still need to write the
pthread_create blocking.

New version attached.

Rich

[-- Attachment #2: synccall.c --]
[-- Type: text/plain, Size: 4218 bytes --]

#include "pthread_impl.h"
#include <semaphore.h>
#include <unistd.h>
#include <dirent.h>
#include <string.h>
#include <ctype.h>
#include "futex.h"
#include "atomic.h"
#include "../dirent/__dirent.h"

static struct chain {
	struct chain *next;
	int tid;
	sem_t target_sem, caller_sem;
} *head;

static int synccall_lock[2];
static int target_tid;
static void (*callback)(void *), *context;

static void handler(int sig)
{
	struct chain ch;
	int old_errno = errno;

	sem_init(&ch.target_sem, 0, 0);
	sem_init(&ch.caller_sem, 0, 0);

	ch.tid = __syscall(SYS_gettid);

	do ch.next = head;
	while (a_cas_p(&head, ch.next, &ch) != ch.next);

	if (a_cas(&target_tid, ch.tid, 0) == (ch.tid | 0x80000000))
		__syscall(SYS_futex, &target_tid, FUTEX_UNLOCK_PI|FUTEX_PRIVATE);

	sem_wait(&ch.target_sem);
	callback(context);
	sem_post(&ch.caller_sem);
	sem_wait(&ch.target_sem);

	errno = old_errno;
}

void __synccall(void (*func)(void *), void *ctx)
{
	sigset_t oldmask;
	int cs, i, r, pid, self;;
	DIR dir = {0};
	struct dirent *de;
	struct sigaction sa = { .sa_flags = 0, .sa_handler = handler };
	struct chain *cp, *next;
	struct timespec ts;

	__block_all_sigs(&oldmask);
	LOCK(synccall_lock);
	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cs);

	head = 0;
	callback = func;
	context = ctx;
	a_barrier();

	if (!libc.threaded) goto single_threaded;

	/* Block even impl-internal signals */
	memset(&sa.sa_mask, -1, sizeof sa.sa_mask);
	__libc_sigaction(SIGSYNCCALL, &sa, 0);

	pid = __syscall(SYS_getpid);
	self = __syscall(SYS_gettid);

	/* Since opendir is not AS-safe, the DIR needs to be setup manually
	 * in automatic storage. Thankfully this is easy. */
	dir.fd = open("/proc/self/task", O_RDONLY|O_DIRECTORY|O_CLOEXEC);
	if (dir.fd < 0) goto out;

	/* Initially send one signal per counted thread. But since we can't
	 * synchronize with thread creation/exit here, there could be too
	 * few signals. This initial signaling is just an optimization, not
	 * part of the logic. */
	for (i=libc.threads_minus_1; i; i--)
		__syscall(SYS_kill, pid, SIGSYNCCALL);

	/* Loop scanning the kernel-provided thread list until it shows no
	 * threads that have not already replied to the signal. */
	for (;;) {
		int miss_cnt = 0;
		while ((de = readdir(&dir))) {
			if (!isdigit(de->d_name[0])) continue;
			int tid = atoi(de->d_name);
			if (tid == self) continue;

			/* Set the target thread as the PI futex owner before
			 * checking if it's in the list of caught threads. If it
			 * adds itself to the list after we check for it, then
			 * it will see its own tid in the PI futex and perform
			 * the unlock operation. */
			a_store(&target_tid, tid);

			a_barrier();
			for (cp = head; cp && cp->tid != tid; cp=cp->next);

			/* Thread-already-caught is a success condition. */
			if (cp) continue;

			r = -__syscall(SYS_tgkill, pid, tid, SIGSYNCCALL);

			/* Target thread exit is a success condition. */
			if (r == ESRCH) continue;

			/* The FUTEX_LOCK_PI operation is used to loan priority
			 * to the target thread, which otherwise may be unable
			 * to run. Timeout is necessary because there is a race
			 * condition where the tid may be reused by a different
			 * process. */
			clock_gettime(CLOCK_REALTIME, &ts);
			ts.tv_nsec += 10000000;
			if (ts.tv_nsec >= 1000000000) {
				ts.tv_sec++;
				ts.tv_nsec -= 1000000000;
			}
			r = -__syscall(SYS_futex, &target_tid,
				FUTEX_LOCK_PI|FUTEX_PRIVATE, 0, &ts);

			/* Obtaining the lock means the thread responded. ESRCH
			 * means the target thread exited, which is okay too. */
			if (!r || r == ESRCH) continue;

			miss_cnt++;
		}
		if (!miss_cnt) break;
		rewinddir(&dir);
	}
	close(dir.fd);

	/* Serialize execution of callback in caught threads. */
	for (cp=head; cp; cp=cp->next) {
		sem_post(&cp->target_sem);
		sem_wait(&cp->caller_sem);
	}

	sa.sa_handler = SIG_IGN;
	__libc_sigaction(SIGSYNCCALL, &sa, 0);

single_threaded:
	func(ctx);

	/* Only release the caught threads once all threads, including the
	 * caller, have returned from the callback function. */
	for (cp=head; cp; cp=next) {
		next = cp->next;
		sem_post(&cp->target_sem);
	}

out:
	pthread_setcancelstate(cs, 0);
	UNLOCK(synccall_lock);
	__restore_sigs(&oldmask);
}