util/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$"
/*======
This file is part of PerconaFT.


Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.

    PerconaFT is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2,
    as published by the Free Software Foundation.

    PerconaFT is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.

----------------------------------------

    PerconaFT is free software: you can redistribute it and/or modify
    it under the terms of the GNU Affero General Public License, version 3,
    as published by the Free Software Foundation.

    PerconaFT is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Affero General Public License for more details.

    You should have received a copy of the GNU Affero General Public License
    along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
======= */

#ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."

/* 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
 *     1.22ns  1.07ns  1.27ns  0.61ns   to do a ++, but it's got a race in it.
 *    27.11ns 20.47ns 18.75ns 34.15ns   to do a sync_fetch_and_add().
 *     0.26ns  0.29ns  0.71ns  0.19ns   to do with a single version of a counter
 *     0.35ns  0.33ns  0.69ns  0.18ns   pure thread-local variable (no way to add things up)
 *             0.76ns  1.50ns  0.35ns   partitioned_counter.c (using link-time optimization, otherwise the function all overwhelms everything)
 *     2.21ns          3.32ns  0.70ns   partitioned_counter.c (using gcc, the C version at r46097, not C++)  This one is a little slower because it has an extra branch in it.
 *
 * Surprisingly, compiling this code without -fPIC doesn't make it any faster (even the pure thread-local variable is the same).  -fPIC access to
 * thread-local variables look slower since they have a function all, but they don't seem to be any slower in practice.  In fact, even the puretl-ptr test
 * which simply increments a thread-local pointer is basically the same speed as accessing thread_local variable.
 *
 * 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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <unistd.h>
#include <toku_race_tools.h>
#include <toku_assert.h>
#include <portability/toku_atomic.h>
#include <memory.h>
#include <util/partitioned_counter.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;
    TOKU_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;
	cp->counter = 0;
	cp->inited = true;
	cp->myid = idcounter++;
	TOKU_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++) {
	increment_partitioned_counter(pc, 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)toku_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("%-10s Time=%.6fs (%7.3fns per increment)\n", description, tdiff(&start, &end), (1e9*tdiff(&start, &end)/T)/N);
}

// Do a measurement where it really is only a pointer dereference to increment the variable, which is thread local.
static void* tl_doit_ptr (void *v) {
    volatile uint64_t *p = (uint64_t *)v;
    for (int i=0; i<N; i++) {
	(*p)++;
    }
    return v;
}


static void timeit_with_thread_local_pointer (const char *description, void* (*f)(void*)) {
    struct timeval start, end;
    pthread_t threads[T];
    struct { uint64_t values[8] __attribute__((__aligned__(64))); } values[T]; // pad to different cache lines.
    gettimeofday(&start, 0);
    for (int i=0; i<T; i++) {
        values[i].values[0]=0;
	pt_create(&threads[i], f, &values[i].values[0]);
    }
    for (int i=0; i<T; i++) {
	pt_join(threads[i], &values[i].values[0]);
    }
    gettimeofday(&end, 0);
    printf("%-10s 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_with_thread_local_pointer("puretl-ptr", tl_doit_ptr);
    pc = create_partitioned_counter();
    timeit("pc",       pc_doit);
    destroy_partitioned_counter(pc);
}

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

static void *reader_test_fun (void *ta_v) {
    struct test_arguments *ta = (struct test_arguments *)ta_v;
    uint64_t lastval = 0;
    while (ta->unfinished_count>0) {
	uint64_t thisval = read_partitioned_counter(ta->pc);
	assert(lastval <= thisval);
	assert(thisval <= ta->limit+2);
	lastval = thisval;
	if (verboseness_cmdarg && (0==(thisval & (thisval-1)))) printf("ufc=%" PRIu64 " Thisval=%" PRIu64 "\n", ta->unfinished_count,thisval);
    }
    uint64_t thisval = read_partitioned_counter(ta->pc);
    assert(thisval==ta->limit+2); // we incremented two extra times in the test
    return ta_v;
}

static void *writer_test_fun (void *ta_v) {
    struct test_arguments *ta = (struct test_arguments *)ta_v;
    for (uint64_t i=0; i<ta->total_increment_per_writer; i++) {
	if (i%1000 == 0) sched_yield();
	increment_partitioned_counter(ta->pc, 1);
    }
    uint64_t c __attribute__((__unused__)) = toku_sync_fetch_and_sub(&ta->unfinished_count, 1);
    return ta_v;
}


static void do_testit (void) {
    const int NGROUPS = 2;
    uint64_t limits[NGROUPS];
    limits [0] = 2000000;
    limits [1] = 1000000;
    uint64_t n_writers[NGROUPS];
    n_writers[0] = 20;
    n_writers[1] = 40;
    struct test_arguments tas[NGROUPS];
    pthread_t reader_threads[NGROUPS];
    pthread_t *writer_threads[NGROUPS];
    for (int i=0; i<NGROUPS; i++) {
        tas[i].pc                         = create_partitioned_counter();
	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]);
        increment_partitioned_counter(tas[i].pc, 1); // make sure that the long-lived thread also increments the partitioned counter, to test for #5321.
	MALLOC_N(n_writers[i], writer_threads[i]);
	for (uint64_t j=0; j<n_writers[i] ; j++) {
	    pt_create(&writer_threads[i][j], writer_test_fun, &tas[i]);
	}
        increment_partitioned_counter(tas[i].pc, 1); // make sure that the long-lived thread also increments the partitioned counter, to test for #5321.
    }
    for (int i=0; i<NGROUPS; i++) {
	pt_join(reader_threads[i], &tas[i]);
	for (uint64_t j=0; j<n_writers[i] ; j++) {
	    pt_join(writer_threads[i][j], &tas[i]);
	}
	toku_free(writer_threads[i]);
        destroy_partitioned_counter(tas[i].pc);
    }
}

volatile int spinwait=0;
static void* test2_fun (void* mypc_v) {
    PARTITIONED_COUNTER mypc = (PARTITIONED_COUNTER)mypc_v;
    increment_partitioned_counter(mypc, 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;
    TOKU_VALGRIND_HG_DISABLE_CHECKING(&spinwait, sizeof(spinwait)); // this is a racy volatile variable.
    {
        PARTITIONED_COUNTER mypc = create_partitioned_counter();
        increment_partitioned_counter(mypc, 1); // make sure that the long-lived thread also increments the partitioned counter, to test for #5321.
        pt_create(&t, test2_fun, mypc);
        while(spinwait==0); // wait until he incremented the counter.
        increment_partitioned_counter(mypc, -1);
        assert(read_partitioned_counter(mypc)==3);
        destroy_partitioned_counter(mypc);
    } // 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;
}