Re: [Qemu-devel] [PATCH 02/12] qemu-thread: add QemuEvent

[Paolo Bonzini <pbonzini@redhat.com>]

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

Il 16/05/2013 12:15, Stefan Hajnoczi ha scritto:
> On Wed, May 15, 2013 at 05:48:47PM +0200, Paolo Bonzini wrote:
>> This emulates Win32 manual-reset events using futexes or conditional
>> variables.  Typical ways to use them are with multi-producer,
>> single-consumer data structures, to test for a complex condition whose
>> elements come from different threads:
>>
>>     for (;;) {
>>         qemu_event_reset(ev);
>>         ... test complex condition ...
>>         if (condition is true) {
>>             break;
>>         }
>>         qemu_event_wait(ev);
>>     }
>>
>> Alternatively:
>>
>>     ... compute condition ...
>>     if (condition) {
>>         do {
>>             qemu_event_wait(ev);
>>             qemu_event_reset(ev);
>>             ... compute condition ...
>>         } while(condition);
>>         qemu_event_set(ev);
>>     }
>>
>> QemuEvent provides a very fast userspace path in the common case when
>> no other thread is waiting, or the event is not changing state.  It
>> is used to report RCU quiescent states to the thread calling
>> synchronize_rcu (the latter being the single consumer), and to report
>> call_rcu invocations to the thread that receives them.
> 
> It would be nice to describe the need for the Linux futex code.  pthread
> mutex/condvars are implemented in terms of futexes already, so how much
> benefit is there - I thought they stay in userspace in the non-contended
> case too?

Yes, but they are still measurably slower, around 20%.  I don't have
around the program I wrote for QemuEvent, because I did the measurement
~2 years ago.  However, here is one that tests a similar synchronization
primitive (not exactly the same as QemuEvent).  You can run it like this:

$ ./a.out -c 100 10 4 40 # with mutex/condvar
4 child processes, avg think time 100 msec
Avg event distance 10 msec. Running for 40 sec
........................................
waits		 1404	0.663 us (fast)
  slow waits	 152
signal		 3890	0.784 us
  fast path	 0                             <<<< (this is bogus)

$ ./a.out 100 10 4 40 # with futex
4 child processes, avg think time 100 msec
Avg event distance 10 msec. Running for 40 sec
........................................
waits		 1383	0.635 us (fast)
  slow waits	 147
signal		 3924	0.640 us
  fast path	 3791


Paolo

[-- Attachment #2: evc.c --]
[-- Type: text/x-c, Size: 6107 bytes --]

#define _GNU_SOURCE 1
#include "pthread.h"
#include <limits.h>
#include <string.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <math.h>

#ifdef __linux__
#include <sys/syscall.h>
#include <linux/futex.h>
#define HAVE_FUTEX 1
#endif

#ifdef _WIN32
#include <windows.h>
#include <mmsystem.h>
#else
#define HAVE_AFFINITY 1
#endif

typedef struct EvCounter {
    int ctr;
    int waiters;
    int fast_signal;
    pthread_mutex_t lock;
    pthread_cond_t cond;
} EvCounter;

typedef int EvCounterState;

void evcounter_get(EvCounterState *state, EvCounter *evcounter)
{
    *state = evcounter->ctr;
}

#ifdef HAVE_FUTEX
#define futex(...)      syscall(__NR_futex, __VA_ARGS__)

void _evcounter_init(EvCounter *evcounter)
{
    evcounter->ctr = evcounter->waiters = 0;
}

int _evcounter_wait(EvCounterState *state, EvCounter *evcounter)
{
    int fast = 1;
    __sync_fetch_and_add(&evcounter->waiters, 1);
    while (*state == evcounter->ctr) {
        futex(&evcounter->ctr, FUTEX_WAIT_PRIVATE, *state, NULL);
        fast = 0;
    }
    __sync_fetch_and_add(&evcounter->waiters, -1);
    *state = evcounter->ctr;
    return fast;
}

void _evcounter_signal(EvCounter *evcounter)
{
    __sync_fetch_and_add(&evcounter->ctr, 1);
    if (evcounter->waiters != 0) {
        futex(&evcounter->ctr, FUTEX_WAKE_PRIVATE, INT_MAX);
    } else {
	evcounter->fast_signal++;
    }
}
#endif

void _evcounter_init_cond(EvCounter *evcounter)
{
    evcounter->ctr = 0;
    pthread_mutex_init (&evcounter->lock, NULL);
    pthread_cond_init (&evcounter->cond, NULL);
}

int _evcounter_wait_cond(EvCounterState *state, EvCounter *evcounter)
{
    int fast = 1;
    pthread_mutex_lock(&evcounter->lock);
    while (*state == evcounter->ctr) {
        pthread_cond_wait(&evcounter->cond, &evcounter->lock);
        fast = 0;
    }
    pthread_mutex_unlock(&evcounter->lock);
    *state = evcounter->ctr;
    return fast;
}

void _evcounter_signal_cond(EvCounter *evcounter)
{
    pthread_mutex_lock(&evcounter->lock);
    evcounter->ctr++;
    pthread_cond_broadcast(&evcounter->cond);
    pthread_mutex_unlock(&evcounter->lock);
}


void (*evcounter_init)(EvCounter *);
int (*evcounter_wait)(EvCounterState *, EvCounter *);
void (*evcounter_signal)(EvCounter *);

EvCounter evc;
volatile int stop = 0;
int affinity = 0;

long t_wait;
long n_slow_wait;
long n_wait;
long t_signal;
long n_signal;

#define CLOCK_RES (sizeof(long) == 8 ? 1000000000LL : 1000000LL)

static inline long long getclk()
{
#ifndef _WIN32
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (ts.tv_sec * 1000000000LL + ts.tv_nsec) / (1000000000LL / CLOCK_RES);
#else
    static LARGE_INTEGER freq, init;
    LARGE_INTEGER counter;
    if (!init.QuadPart) {
        QueryPerformanceFrequency(&freq);
        QueryPerformanceCounter(&init);
    }
    QueryPerformanceCounter(&counter);
    return (counter.QuadPart - init.QuadPart) * CLOCK_RES / freq.QuadPart;
#endif
}

#ifdef _WIN32
void sleep(int secs)
{
    Sleep (secs * 1000);
}
#endif

void exp_usleep(int avg)
{
    if (avg) {
	double x = rand() / ((double)RAND_MAX + 1.0);
	int usecs = (int) -avg * log(x);
#ifdef _WIN32
	Sleep(usecs / 1000);
#else
	usleep(usecs);
#endif
    }
}

void *generator(void *pavg)
{
    int avg = *(int *)pavg;
    while (stop != 2) {
	exp_usleep(avg);
	long long t1 = getclk();
	evcounter_signal(&evc);
	t_signal += getclk() - t1;
	n_signal++;
    }
    return NULL;
}

void *consumer(void *pavg)
{
    EvCounterState s;
    int avg = *(int *)pavg;
    evcounter_get(&s, &evc);
    while (stop == 0) {
	long long t1 = getclk();
	if (evcounter_wait(&s, &evc)) {
	    __sync_fetch_and_add (&t_wait, getclk() - t1);
	    __sync_fetch_and_add (&n_wait, 1);
        } else {
            __sync_fetch_and_add(&n_slow_wait, 1);
        }
	exp_usleep(avg);
    }
    return NULL;
}

void
create_thread(pthread_t *p, void *(*f)(void *), void *arg)
{
    pthread_create(p, NULL, f, arg);
#ifdef HAVE_AFFINITY
    if (affinity) {
        static int nproc, i;
        cpu_set_t cpu_set;
        if (nproc == 0)
	    nproc = sysconf (_SC_NPROCESSORS_ONLN);

        CPU_ZERO(&cpu_set);
        CPU_SET(i % nproc, &cpu_set);
        i++;

        pthread_setaffinity_np(*p, sizeof(cpu_set), &cpu_set);
    }
#endif
}

int main(int argc, char **argv)
{
#ifdef HAVE_FUTEX
    evcounter_init = _evcounter_init;
    evcounter_wait = _evcounter_wait;
    evcounter_signal = _evcounter_signal;
#else
    evcounter_init = _evcounter_init_cond;
    evcounter_wait = _evcounter_wait_cond;
    evcounter_signal = _evcounter_signal_cond;
#endif
    while (argv[1] && argv[1][0] == '-') { 
        if (strchr(argv[1], 'c')) {
	    evcounter_init = _evcounter_init_cond;
	    evcounter_wait = _evcounter_wait_cond;
	    evcounter_signal = _evcounter_signal_cond;
        }
        if (!strcmp(argv[1], "-a")) {
            affinity = 1;
        }
	argc--, argv++;
    }

    evcounter_init (&evc);

    srand(time(NULL));

    int think_avg = argc > 1 ? atoi(argv[1]) * 1000 : 100000;
    int throughput_avg = argc > 2 ? atoi(argv[2]) * 1000 : 10000;
    int n = argc > 3 ? atoi(argv[3]) : 10;
    int len = argc > 4 ? atoi(argv[4]) : 10;

    printf ("%d child processes, avg think time %d msec\n", n, think_avg/1000);
    printf ("Avg event distance %d msec. Running for %d sec\n",
	    throughput_avg/1000, len);

    pthread_t g, c[n];
    create_thread(&g, generator, &throughput_avg);

    int i;
    for (i = 0; i < n; i++)
	create_thread(&c[i], consumer, &think_avg);

    for (i = 0; i < len; i++) {
	write(1, ".", 1);
	sleep (1);
    }

    write (1, "\n", 1);

    stop = 1;
    for (i = 0; i < n; i++)
	pthread_join(c[i], NULL);
    stop = 2;
    pthread_join(g, NULL);

    printf ("waits\t\t %ld	%.3f us (fast case)\n", n_wait,
            (double)t_wait/(CLOCK_RES/1000000.0)/n_wait);
    printf ("  slow waits\t %ld\n", n_slow_wait);
    printf ("signal\t\t %ld	%.3f us\n", n_signal,
            (double)t_signal/(CLOCK_RES/1000000.0)/n_signal);
    printf ("  fast path\t %ld\n", evc.fast_signal);
    exit (0);
}