Ticket #11895: Source.cpp

#include <deque>
#include <iostream>
#include <boost/function.hpp>
#include <boost/asio/io_service.hpp>
#include <boost/asio/strand.hpp>
#include <boost/chrono/chrono.hpp>
#include <boost/thread/thread.hpp>

using namespace std;
using namespace boost::asio;
using namespace boost::chrono;
using namespace boost::this_thread;
using boost::mutex;
using boost::thread;
using boost::thread_group;

// simplified strand implementation that fixes scheduling issue
struct CStrand
{
        CStrand(io_service& IoService) :
                IoService_(IoService),
                bSignaled_(false) { }

        void post(function<void()> Fxn)
        {
                mutex::scoped_lock Lock(Mutex_);
                PendingWorkQueue_.push_back(move(Fxn));
                if (bSignaled_)
                        return;
                bSignaled_ = true;
                Lock.unlock();
                IoService_.post([this] { service(); });
        }

        // don't give control back to the io_service until the queue is drained!
        void service(void)
        {
                for (;;) {
                        deque<function<void()>> WorkQueue;
                        mutex::scoped_lock Lock(Mutex_);
                        WorkQueue.swap(PendingWorkQueue_);
                        Lock.unlock();
                        for (auto& Fxn : WorkQueue)
                                Fxn();
                        Lock.lock();
                        if (PendingWorkQueue_.empty()) {
                                bSignaled_ = false;
                                return;
                        }
                }
        }

        io_service& IoService_;

        mutable mutex Mutex_;
        bool bSignaled_;
        deque<function<void()>> PendingWorkQueue_;
};


template<typename TStrand>
void Test(const string& sName)
{
        cout << "__________" << sName << "__________" << endl << endl;

        io_service IoService;
        auto pWork = make_unique<io_service::work>(IoService);
        auto Cores = thread::hardware_concurrency();

        thread_group Threads;
        for (unsigned i = 0; i < Cores; ++i)
                Threads.create_thread([&]{ IoService.run(); });

        TStrand Strand(IoService);
        mutex Mutex;

        // post() a thread-proportional amount of work to the io_service and the strand
        // for 4 cores that would be 30 seconds of work on the io_service and 10 second of work on the strand.
        // optimally, this should only take 10 seconds in parallel but you'll see that's not the case.
        auto Start = high_resolution_clock::now();
        for (int i = 0, Count = 0; i < 10; ++i) {
                for (unsigned u = 0; u < Cores - 1; ++u) {
                        ++Count;
                        IoService.post([&, Count] {
                                mutex::scoped_lock Lock(Mutex);
                                cout << "io_service " << Count << endl;
                                Lock.unlock();
                                sleep_for(seconds(1));
                        });
                }
                ++Count;
                Strand.post([&, Count] {
                        mutex::scoped_lock Lock(Mutex);
                        cout << "strand " << Count << endl;
                        Lock.unlock();
                        sleep_for(seconds(1));
                });
        }
        pWork.reset();
        Threads.join_all();
        auto Stop = high_resolution_clock::now();

        auto Elapsed = duration_cast<seconds>(Stop - Start);
        cout << endl << Elapsed.count() << " seconds" << endl << endl;
}


void main()
{
        Test<strand>("strand");
        Test<CStrand>("CStrand");
        cin.get();
}