14 ThreadBlockerWhileUsedByStrand

ThreadBlockerWhileUsedByStrand is defined purely in a header file. It is only used by the test TestAsioSyncCloseStrand in the (old) txMessage. Therefore it can probably be removed from ceda-core.

Intended to be used by an object using a single strand for its async operations, for the purpose of implementing a synchronous close function that blocks until all the pending async operations have completed.

For this purpose a manual reset event is used which is initially in the nonsignalled state and becomes signalled when a count of the number of pending async operations falls to zero.

This transition to zero can only occur once, to ensure this the count is normally decremented by the implementation of the close function itself.

It is assumed operator++() and operator--() are only called by completion handlers of a single strand and therefore there is no need to protect the counter with a mutex.

ThreadBlockerWhileUsedByStrand.h

// ThreadBlockerWhileUsedByStrand.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2013

#pragma once
#ifndef Ceda_cxThread_ThreadBlockerWhileUsedByStrand_H
#define Ceda_cxThread_ThreadBlockerWhileUsedByStrand_H

#include "cxThread.h"
#include "ManualResetEvent.h"
#include "Ceda/cxUtils/CedaAssert.h"

namespace ceda
{
class ThreadBlockerWhileUsedByStrand
{
    cxNotCloneable(ThreadBlockerWhileUsedByStrand)
    
public:
    explicit ThreadBlockerWhileUsedByStrand(int count) : count_(count) 
    {
        cxAssert(count_ > 0);
    }
    ~ThreadBlockerWhileUsedByStrand()
    {
        cxAssert(count_ == 0);
    }

    // Not thread safe
    int operator++() 
    {
        // 0 to 1 transitions are not permitted because we only allow one transition to 0 
        // (we don't allow the event to return to non signalled)
        cxAssert(count_ > 0);       
        return ++count_;
    }

    // Not thread safe
    int operator--()
    {
        cxAssert(count_ > 0);
        if (--count_ == 0)
        {
            event_.Signal();
        }
        return count_;
    }

    // Postfix
    int operator++(int) { return operator++()-1; }
    int operator--(int) { return operator--()+1; }

    void Wait()
    {
        event_.Wait();    
    }
private:
    ManualResetEvent event_;     // Initially not signalled, becomes signalled when count falls to zero
                                 // Once signalled never returns to not signalled

    int count_;                  // Not thread safe
};
} // namespace ceda
#endif // include guard

TestAsioSyncCloseStrand.cpp

// TestAsioSyncCloseStrand.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2016

#include "Ceda/cxThread/ThreadBlockerWhileUsedByStrand.h"
#include "Ceda/cxUtils/Tracer.h"
#include "Ceda/cxUtils/TestTimer.h"
#include <boost/asio.hpp>
#include <atomic>
#include <thread>
#include <functional>

const int NUM_COUNT = 10;
const int NUM_ASYNC_TASKS = 10;

class AsioSyncCloseStrand
{
public:
    AsioSyncCloseStrand(boost::asio::io_service& io_service);
    void Close();
private:
    void AsyncTask();
    void IncrementCounters();
private:
    boost::asio::io_service::strand strand_;
    ceda::ThreadBlockerWhileUsedByStrand blocker_;

    int numAsyncCalls_;

    // The pure CPU task involves incrementing all the values in count, in a way that tends to expose racing conditions
    int count_[NUM_COUNT];
};

AsioSyncCloseStrand::AsioSyncCloseStrand(boost::asio::io_service& io_service) :
    strand_(io_service),
    blocker_(2),
    numAsyncCalls_(0)
{
    for (int i=0 ; i < NUM_COUNT ; ++i)
    {
        count_[i] = 0;    
    }
    AsyncTask();
}

void AsioSyncCloseStrand::IncrementCounters()
{
    int temp[NUM_COUNT];

    // read all the counters
    for (int i=0 ; i < NUM_COUNT ; ++i)
    {
        temp[i] = count_[i];
    }

    // write all the counters
    for (int i=0 ; i < NUM_COUNT ; ++i)
    {
        count_[i] = temp[i]+1;
    }
}

void AsioSyncCloseStrand::AsyncTask()
{
    strand_.post(
        [this]()
        {
            IncrementCounters();
            if (++numAsyncCalls_ < NUM_ASYNC_TASKS)
            {
                AsyncTask();    
            }
            else
            {
                --blocker_;
            }
        });
}

void AsioSyncCloseStrand::Close()
{
    strand_.post(
        [this]()
        {
            IncrementCounters();
            --blocker_;
        });

    blocker_.Wait();

    for (int i=0 ; i < NUM_COUNT ; ++i)
    {
        cxAlwaysAssert(count_[i] == NUM_ASYNC_TASKS+1);
    }
}

void TestAsioSyncCloseStrand(double timeForTestInSecs)
{
    Tracer() << "TestAsioSyncCloseStrand\n";
    ceda::TraceIndenter indent(4);

    const int numThread = 8;
    const int count = 1000;

    boost::asio::io_service io_service;
    std::thread threads[numThread];
    
    {
        // When the io_service::run method is called without a work object, it will return right away. 
        // We want the threads to only exit when told to do so.
        boost::asio::io_service::work work(io_service);

        for (int i=0 ; i < numThread ; ++i)
        {
            threads[i] = std::thread([&io_service] { io_service.run(); });
        }

        ceda::HPTimer timer;
        ceda::ssize_t loop = 0;
        while(timer.GetElapsedTimeInSeconds() < timeForTestInSecs)
        {
            for (int i=0 ; i < count ; ++i)
            {
                AsioSyncCloseStrand test1(io_service);
                AsioSyncCloseStrand test2(io_service);
                AsioSyncCloseStrand test3(io_service);
                AsioSyncCloseStrand test4(io_service);
                test1.Close();
                test2.Close();
                test3.Close();
                test4.Close();
            }
            ++loop;
        }
        double numIncrementsPerStrand = NUM_COUNT*(NUM_ASYNC_TASKS+1);
        double numIncrementsPerIteration = numIncrementsPerStrand * 4 * count;
        double numIncrements = loop * numIncrementsPerIteration;
        double rate = numIncrements / timer.GetElapsedTimeInSeconds();

        timer.Done() << "threads=" << numThread << " concurrent strands=4" << " iterations=" << loop << " rate=" << rate/1000000.0 << " MHz";
    }

    for (int i=0 ; i < numThread ; ++i)
    {
        threads[i].join();
    }
}