mariadb/ft/tests/test_partitioned_counter.cc

/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
#ident "$Id$"
#ident "Copyright (c) 2007-2012 Tokutek Inc.  All rights reserved."
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."

/* This code can either test the PARTITIONED_COUNTER abstraction or it can time various implementations. */

/* Try to make counter that requires no cache misses to increment, and to get the value can be slow.
 * I don't care much about races between the readers and writers on the counter.
 *
 * The problem: We observed that incrementing a counter with multiple threads is quite expensive.
 * Here are some performance numbers:
 * Machines:  mork or mindy (Intel Xeon L5520 2.27GHz)
 *            bradley's 4-core laptop laptop (Intel Core i7-2640M 2.80GHz) sandy bridge
 *            alf       16-core server (xeon E5-2665 2.4GHz) sandybridge
 *
 *      mork  mindy  bradley  alf
 *      0.3ns  1.07ns  1.27ns  0.61ns   to do a ++, but it's got a race in it.
 *     28.0ns 20.47ns 18.75ns 34.15ns   to do a sync_fetch_and_add().
 *      0.4ns  0.29ns  0.71ns  0.19ns   to do with a single version of a counter
 *             0.33ns  0.69ns  0.18ns   pure thread-local variable (no way to add things up)
 *             0.76ns  2.40ns  0.54ns   partitioned_counter.c (using gcc link-time optimization, otherwise the function call overwhelms everything)
 *      
 * 
 * How it works.  Each thread has a thread-local counter structure with an integer in it.  To increment, we increment the thread-local structure.
 *   The other operation is to query the counters to get the sum of all the thread-local variables.
 *   The first time a pthread increments the variable we add the variable to a linked list.
 *   When a pthread ends, we use the pthread_key destructor to remove the variable from the linked list.  We also have to remember the sum of everything.
 *    that has been removed from the list.
 *   To get the sum we add the sum of the destructed items, plus everything in the list.
 *
 */

#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <unistd.h>
#include <valgrind/helgrind.h>
#include "toku_assert.h"
#include "partitioned_counter.h"
#include "memory.h"
#include "test.h"

// The test code includes the fastest version I could figure out to make, implemented below.

struct counter_s {
    bool inited;
    volatile int counter;
    struct counter_s *prev, *next;
    int myid;
};
static __thread struct counter_s counter = {false,0, NULL,NULL,0};

static int finished_counter=0; // counter for all threads that are done.

// We use a single mutex for anything complex.  We'd like to use a mutex per partitioned counter, but we must cope with the possibility of a race between
// a terminating pthread (which calls destroy_counter()), and a call to the counter destructor.  So we use a global mutex.
static pthread_mutex_t pc_mutex = PTHREAD_MUTEX_INITIALIZER; 
static struct counter_s *head=NULL;
static pthread_key_t   counter_key;

static void pc_lock (void)
// Effect: Lock the pc mutex.  
{
    int r = pthread_mutex_lock(&pc_mutex);
    assert(r==0);
}

static void pc_unlock (void)
// Effect: Unlock the pc mutex.
{
    int r = pthread_mutex_unlock(&pc_mutex);
    assert(r==0);
}

static void destroy_counter (void *counterp)
// Effect: This is the function passed to pthread_key_create that is to run whenever a thread terminates.
//   The thread-local part of the counter must be copied into the shared state, and the thread-local part of the counter must be
//   removed from the linked list of all thread-local parts.
{
    assert((struct counter_s*)counterp==&counter);
    pc_lock();
    if (counter.prev==NULL) {
	assert(head==&counter);
	head = counter.next;
    } else {
	counter.prev->next = counter.next;
    }
    if (counter.next!=NULL) {
	counter.next->prev = counter.prev;
    }
    finished_counter += counter.counter;
    HELGRIND_VALGRIND_HG_ENABLE_CHECKING(&counter.counter, sizeof(counter.counter)); // stop ignoring races
    //printf("finished counter now %d\n", finished_counter);
    pc_unlock();
}

static int idcounter=0;

static inline void increment (void) 
{
    if (!counter.inited) {
        pc_lock();
        struct counter_s *cp = &counter;
	{ int r = pthread_setspecific(counter_key, cp); assert(r==0); }
	cp->prev = NULL;
	cp->next = head;
	if (head!=NULL) {
	    head->prev = cp;
	}
        head = cp;
#ifdef __INTEL_COMPILER
        __memory_barrier(); // for some reason I don't understand, ICC needs a memory barrier here. -Bradley
#endif
	cp->counter = 0;
	cp->inited = true;
	cp->myid = idcounter++;
	HELGRIND_VALGRIND_HG_DISABLE_CHECKING(&counter.counter, sizeof(counter.counter)); // the counter increment is kind of racy.
        pc_unlock();
    }
    counter.counter++;
}

static int getvals (void) {
    pc_lock();
    int sum=finished_counter;
    for (struct counter_s *p=head; p; p=p->next) {
	sum+=p->counter;
    }
    pc_unlock();
    return sum;
}
    
/**********************************************************************************/
/* And now for some actual test code.                                             */
/**********************************************************************************/

static const int N=10000000;
static const int T=20;


PARTITIONED_COUNTER pc;
static void *pc_doit (void *v) {
    for (int i=0; i<N; i++) {
	pc.increment(1);
    }
    //printf("val=%ld\n", read_partitioned_counter(pc));
    return v;
}

static void* new_doit (void* v) {
    for (int i=0; i<N; i++) {
	increment();
	//if (i%0x2000 == 0) sched_yield();
    }
    if (0) printf("done id=%d, getvals=%d\n", counter.myid, getvals());
    return v;
}

static int oldcounter=0;

static void* old_doit (void* v) {
    for (int i=0; i<N; i++) {
	(void)__sync_fetch_and_add(&oldcounter, 1);
	//if (i%0x1000 == 0) sched_yield();
    }
    return v;
}

static volatile int oldcounter_nonatomic=0;

static void* old_doit_nonatomic (void* v) {
    for (int i=0; i<N; i++) {
	oldcounter_nonatomic++;
	//if (i%0x1000 == 0) sched_yield();
    }
    return v;
}

static __thread volatile int thread_local_counter=0;
static void* tl_doit (void *v) {
    for (int i=0; i<N; i++) {
	thread_local_counter++;
    }
    return v;
}

static float tdiff (struct timeval *start, struct timeval *end) {
    return (end->tv_sec-start->tv_sec) +1e-6*(end->tv_usec - start->tv_usec);
}

static void pt_create (pthread_t *thread, void *(*f)(void*), void *extra) {
    int r = pthread_create(thread, NULL, f, extra);
    assert(r==0);
}

static void pt_join (pthread_t thread, void *expect_extra) {
    void *result;
    int r = pthread_join(thread, &result);
    assert(r==0);
    assert(result==expect_extra);
}

static void timeit (const char *description, void* (*f)(void*)) {
    struct timeval start, end;
    pthread_t threads[T];
    gettimeofday(&start, 0);
    for (int i=0; i<T; i++) {
	pt_create(&threads[i], f, NULL);
    }
    for (int i=0; i<T; i++) {
	pt_join(threads[i], NULL);
    }
    gettimeofday(&end, 0);
    printf("%-9s Time=%.6fs (%7.3fns per increment)\n", description, tdiff(&start, &end), (1e9*tdiff(&start, &end)/T)/N);
}

static int verboseness_cmdarg=0;
static bool time_cmdarg=false;

static void parse_args (int argc, const char *argv[]) {
    const char *progname = argv[1];
    argc--; argv++;
    while (argc>0) {
	if (strcmp(argv[0], "-v")==0) verboseness_cmdarg++;
	else if (strcmp(argv[0], "--time")==0) time_cmdarg=true;
	else {
	    printf("Usage: %s [-v] [--time]\n Default is to run tests.  --time produces timing output.\n", progname);
	    exit(1);
	}
	argc--; argv++;
    }
}

static void do_timeit (void) {
    { int r = pthread_key_create(&counter_key, destroy_counter); assert(r==0); } 
    printf("%d threads\n%d increments per thread\n", T, N);
    timeit("++",       old_doit_nonatomic);
    timeit("atomic++", old_doit);
    timeit("fast",     new_doit);
    timeit("puretl",   tl_doit);
    timeit("pc",       pc_doit);
}

struct test_arguments {
    PARTITIONED_COUNTER pc;
    u_int64_t           limit;
    u_int64_t           total_increment_per_writer;
    volatile u_int64_t  unfinished_count;
};

static void *reader_test_fun (void *ta_v) {
    struct test_arguments *ta = (struct test_arguments *)ta_v;
    u_int64_t lastval = 0;
    while (ta->unfinished_count>0) {
	u_int64_t thisval = ta->pc.read();
	assert(lastval <= thisval);
	assert(thisval <= ta->limit);
	lastval = thisval;
	if (verboseness_cmdarg && (0==(thisval & (thisval-1)))) printf("Thisval=%ld\n", thisval);
    }
    u_int64_t thisval = ta->pc.read();
    assert(thisval==ta->limit);
    return ta_v;
}

static void *writer_test_fun (void *ta_v) {
    struct test_arguments *ta = (struct test_arguments *)ta_v;
    for (u_int64_t i=0; i<ta->total_increment_per_writer; i++) {
	if (i%1000 == 0) sched_yield();
	ta->pc.increment(1);
    }
    __sync_fetch_and_sub(&ta->unfinished_count, 1);
    return ta_v;
}
    

static void do_testit (void) {
    const int NGROUPS = 2;
    u_int64_t limits[NGROUPS];
    limits [0] = 200000;
    limits [1] = 100000;
    u_int64_t n_writers[NGROUPS];
    n_writers[0] = 2;
    n_writers[1] = 4;
    struct test_arguments tas[NGROUPS];
    pthread_t reader_threads[NGROUPS];
    pthread_t *writer_threads[NGROUPS];
    for (int i=0; i<NGROUPS; i++) {
	tas[i].limit = limits[i];
	tas[i].unfinished_count  = n_writers[i];
	tas[i].total_increment_per_writer = limits[i]/n_writers[i];
	assert(tas[i].total_increment_per_writer * n_writers[i] == limits[i]);
	pt_create(&reader_threads[i], reader_test_fun, &tas[i]);
	MALLOC_N(n_writers[i], writer_threads[i]);
	for (u_int64_t j=0; j<n_writers[i] ; j++) {
	    pt_create(&writer_threads[i][j], writer_test_fun, &tas[i]);
	}
    }
    for (int i=0; i<NGROUPS; i++) {
	pt_join(reader_threads[i], &tas[i]);
	for (u_int64_t j=0; j<n_writers[i] ; j++) {
	    pt_join(writer_threads[i][j], &tas[i]);
	}
	toku_free(writer_threads[i]);
    }
}

volatile int spinwait=0;
static void* test2_fun (void* mypc_v) {
    PARTITIONED_COUNTER *mypc = (PARTITIONED_COUNTER*)mypc_v;
    mypc->increment(3);
    spinwait=1;
    while (spinwait==1);
    // mypc no longer points at a valid data structure.
    return NULL;
}

static void do_testit2 (void) 
// This test checks to see what happens if a thread is still live when we destruct a counter.
//   A thread increments the counter, then lets us know through a spin wait, then waits until we destroy the counter.
{
    pthread_t t;
    {
        PARTITIONED_COUNTER mypc;
        pt_create(&t, test2_fun, &mypc);
        while(spinwait==0); // wait until he incremented the counter.
        assert(mypc.read()==3);
    } // leave scope, so the counter goes away.
    spinwait=2; // tell the other guy to finish up.
    pt_join(t, NULL);
}

int test_main (int argc, const char *argv[]) {
    parse_args(argc, argv);
    if (time_cmdarg) {
	do_timeit();
    } else {
	do_testit();
        do_testit2();
    }
    return 0;
}