Testing multithreaded transaction counting methods

.gitignore

test
*~

README.md

Testing multithreaded transaction counting methods

Using an older i7 that is otherwise only lightly loaded, 600s of run time, 60s integration time, microsecond resolution for time calculations.

Results

Using the old method:

600001505
432762 2.59657e+08
264129000
1.01722
66016000 65970000 66179000 65964000

Using the new method with an additional global lock:

600004084
350254 2.10152e+08
211158000
1.00479
53029000 52797000 52380000 52952000

Using the new method without the additional lock:

600001210
2.62319e+06 1.57392e+09
1649829000
1.04823
411782000 412009000 413407000 412631000

So the code is inaccurate in this extreme scenario by about 5% but at the same time about 6x faster than the old method.

Makefile

all:	test


test:	test.cpp
	$(CXX) $< -o $@ -lboost_system -lboost_thread

test.cpp

#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/thread.hpp>
#include <iostream>
#include <time.h>

using namespace std;



#define OLD_UTXNPSEC                    0
// set to zero to use non-locking behavior
#define NEW_PROPERLY_LOCKED             0

static const int INTERVAL_SECONDS = 600;

int64_t GetTimeMicros()
{
    int64_t now = (boost::posix_time::microsec_clock::universal_time() -
                      boost::posix_time::ptime(boost::gregorian::date(1970, 1, 1)))
                      .total_microseconds();
    assert(now > 0);
    return now;
}

static boost::mutex cs_txPerSec;

#if OLD_UTXNPSEC
#warning "old"
static double nTxPerSec;

void UpdateTransactionsPerSecond()
{
    boost::mutex::scoped_lock lock(cs_txPerSec);

    static int64_t nLastTime = GetTimeMicros();
    double nSecondsToAverage = 60; // Length of time in seconds to smooth the tx rate over
    int64_t nNow = GetTimeMicros();

    // Decay the previous tx rate.
    int64_t nDeltaTime = (nNow - nLastTime) / 1e6;
    if (nDeltaTime > 0)
    {
        nTxPerSec -= (nTxPerSec / nSecondsToAverage) * nDeltaTime;
        nLastTime = nNow;
    }

    // Add the new tx to the rate
    nTxPerSec += 1 / nSecondsToAverage; // The amount that the new tx will add to the tx rate
    if (nTxPerSec < 0)
        nTxPerSec = 0;
}
#else
#warning "new"
static std::atomic<double> nTxPerSec{0};

void UpdateTransactionsPerSecond()
{
#if NEW_PROPERLY_LOCKED
#warning "properly locked"
    boost::mutex::scoped_lock lock(cs_txPerSec);
#endif
    static std::atomic<double> nTxnsProcessed{0};
    static std::atomic<int64_t> nLastTime{GetTimeMicros()};

    double nSecondsToAverage = 60; // Length of time in seconds to smooth the tx rate over
    int64_t nNow = GetTimeMicros();

    double nNewTxns = 0.0;
    double nOldTxns = nTxnsProcessed.load();
    do
    {
        // decay the number of transactions over nSecondsToAverage and then add "1" for
        // the new transaction.
        nNewTxns = nOldTxns * std::pow(1.0 - 1.0 / nSecondsToAverage, (double)(nNow - nLastTime) / 1000000.) + 1;
    } while (!nTxnsProcessed.compare_exchange_weak(nOldTxns, nNewTxns));

    int64_t tmpTime = nLastTime;
    while (nNow > tmpTime)
    {
        nLastTime.compare_exchange_weak(tmpTime, nNow);
    }

    nTxPerSec.store(nNewTxns / nSecondsToAverage);
}
#endif

struct UpdaterThread {
    uint64_t count;
    UpdaterThread() : count(0) {}
    void run () {
        int64_t t = GetTimeMicros();
        while (GetTimeMicros()-t < INTERVAL_SECONDS*1000000) {
            for (size_t i=0; i < 1000; i++) {
                UpdateTransactionsPerSecond();
                count++;
            }
        }
    }
};

int main(const int argc, const char **argv) {
    UpdaterThread a, b, c, d;
    int64_t begin = GetTimeMicros();
    boost::thread
        ta{boost::bind(&UpdaterThread::run, &a)},
        tb{boost::bind(&UpdaterThread::run, &b)},
        tc{boost::bind(&UpdaterThread::run, &c)},
        td{boost::bind(&UpdaterThread::run, &d)};

    ta.join(); tb.join(); tc.join(); td.join();
    int64_t end = GetTimeMicros();
    cout << end-begin << endl;
#if OLD_UTXNPSEC
    double txpersec = nTxPerSec;
#else
    double txpersec = nTxPerSec.load();
#endif
    cout << txpersec << ' ' << txpersec * INTERVAL_SECONDS << endl;
    uint64_t total_count = a.count + b.count + c.count + d.count;
    cout << total_count << endl;
    cout << (total_count / (txpersec * INTERVAL_SECONDS)) << endl;
    cout << a.count << ' ' << b.count << ' ' << c.count << ' ' << d.count << endl;
}