test/include/cds_test/thread.h

   1 /*
   2     This file is a part of libcds - Concurrent Data Structures library
   3
   4     (C) Copyright Maxim Khizhinsky (libcds.dev@gmail.com) 2006-2016
   5
   6     Source code repo: http://github.com/khizmax/libcds/
   7     Download: http://sourceforge.net/projects/libcds/files/
   8
   9     Redistribution and use in source and binary forms, with or without
  10     modification, are permitted provided that the following conditions are met:
  11
  12     * Redistributions of source code must retain the above copyright notice, this
  13     list of conditions and the following disclaimer.
  14
  15     * Redistributions in binary form must reproduce the above copyright notice,
  16     this list of conditions and the following disclaimer in the documentation
  17     and/or other materials provided with the distribution.
  18
  19     THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  20     AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  21     IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  22     DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  23     FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  24     DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  25     SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  26     CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  27     OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28     OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29 */
  30
  31 #ifndef CDSTEST_THREAD_H
  32 #define CDSTEST_THREAD_H
  33
  34 #include <gtest/gtest.h>
  35 #include <vector>
  36 #include <thread>
  37 #include <condition_variable>
  38 #include <mutex>
  39 #include <chrono>
  40 #include <cds/threading/model.h>
  41
  42 namespace cds_test {
  43
  44     // Forwards
  45     class thread;
  46     class thread_pool;
  47
  48     // Test thread
  49     class thread
  50     {
  51         void run();
  52
  53     protected: // thread_pool interface
  54         thread( thread const& sample );
  55
  56         virtual ~thread()
  57         {}
  58
  59         void join()
  60         {
  61             m_impl.join();
  62         }
  63
  64     protected:
  65         virtual thread * clone() = 0;
  66         virtual void test() = 0;
  67
  68         virtual void SetUp()
  69         {
  70             cds::threading::Manager::attachThread();
  71         }
  72
  73         virtual void TearDown()
  74         {
  75             cds::threading::Manager::detachThread();
  76         }
  77
  78     public:
  79         explicit thread( thread_pool& master, int type = 0 );
  80
  81         thread_pool& pool() { return m_pool; }
  82         int type() const { return m_type; }
  83         size_t id() const { return m_id;  }
  84         bool time_elapsed() const;
  85
  86     private:
  87         friend class thread_pool;
  88
  89         thread_pool&    m_pool;
  90         int             m_type;
  91         size_t          m_id;
  92         std::thread     m_impl;
  93     };
  94
  95     // Pool of test threads
  96     class thread_pool
  97     {
  98     public:
  99         explicit thread_pool( ::testing::Test& fixture )
 100             : m_fixture( fixture )
 101             , m_bRunning( false )
 102             , m_bStopped( false )
 103             , m_doneCount( 0 )
 104             , m_bTimeElapsed( false )
 105             , m_readyCount( 0 )
 106         {}
 107
 108         ~thread_pool()
 109         {
 110             clear();
 111         }
 112
 113         void add( thread * what )
 114         {
 115             m_threads.push_back( what );
 116         }
 117
 118         void add( thread * what, size_t count )
 119         {
 120             add( what );
 121             for ( size_t i = 1; i < count; ++i ) {
 122                 thread * p = what->clone();
 123                 add( p );
 124             }
 125         }
 126
 127         std::chrono::milliseconds run()
 128         {
 129             return run( std::chrono::seconds::zero() );
 130         }
 131
 132         std::chrono::milliseconds run( std::chrono::seconds duration )
 133         {
 134             m_bStopped = false;
 135             m_doneCount = 0;
 136
 137             while ( m_readyCount.load() != m_threads.size() )
 138                 std::this_thread::yield();
 139
 140             m_bTimeElapsed.store( false, std::memory_order_release );
 141
 142             auto native_duration = std::chrono::duration_cast<std::chrono::steady_clock::duration>(duration);
 143             auto time_start = std::chrono::steady_clock::now();
 144             auto const expected_end = time_start + native_duration;
 145
 146             {
 147                 scoped_lock l( m_cvMutex );
 148                 m_bRunning = true;
 149                 m_cvStart.notify_all();
 150             }
 151
 152             if ( duration != std::chrono::seconds::zero() ) {
 153                 for ( ;; ) {
 154                     std::this_thread::sleep_for( native_duration );
 155                     auto time_now = std::chrono::steady_clock::now();
 156                     if ( time_now >= expected_end )
 157                         break;
 158                     native_duration = expected_end - time_now;
 159                 }
 160             }
 161             m_bTimeElapsed.store( true, std::memory_order_release );
 162
 163             {
 164                 scoped_lock l( m_cvMutex );
 165                 while ( m_doneCount != m_threads.size() )
 166                     m_cvDone.wait( l );
 167                 m_bStopped = true;
 168             }
 169             auto time_end = std::chrono::steady_clock::now();
 170
 171             m_cvStop.notify_all();
 172
 173             for ( auto t : m_threads )
 174                 t->join();
 175
 176             return m_testDuration = std::chrono::duration_cast<std::chrono::milliseconds>(time_end - time_start);
 177         }
 178
 179         size_t size() const             { return m_threads.size(); }
 180         thread& get( size_t idx ) const { return *m_threads.at( idx ); }
 181
 182         template <typename Fixture>
 183         Fixture& fixture()
 184         {
 185             return static_cast<Fixture&>(m_fixture);
 186         }
 187
 188         std::chrono::milliseconds duration() const { return m_testDuration; }
 189
 190         void clear()
 191         {
 192             for ( auto t : m_threads )
 193                 delete t;
 194             m_threads.clear();
 195             m_bRunning = false;
 196             m_bStopped = false;
 197             m_doneCount = 0;
 198         }
 199
 200     protected: // thread interface
 201         size_t get_next_id()
 202         {
 203             return m_threads.size();
 204         }
 205
 206         void    ready_to_start( thread& /*who*/ )
 207         {
 208             // Called from test thread
 209
 210             // Wait for all thread created
 211             scoped_lock l( m_cvMutex );
 212             m_readyCount.fetch_add( 1 );
 213             while ( !m_bRunning )
 214                 m_cvStart.wait( l );
 215         }
 216
 217         void    thread_done( thread& /*who*/ )
 218         {
 219             // Called from test thread
 220
 221             {
 222                 scoped_lock l( m_cvMutex );
 223                 ++m_doneCount;
 224
 225                 // Tell pool that the thread is done
 226                 m_cvDone.notify_all();
 227
 228                 // Wait for all thread done
 229                 while ( !m_bStopped )
 230                     m_cvStop.wait( l );
 231             }
 232         }
 233
 234     private:
 235         friend class thread;
 236
 237         ::testing::Test&        m_fixture;
 238         std::vector<thread *>   m_threads;
 239
 240         typedef std::unique_lock<std::mutex> scoped_lock;
 241         std::mutex              m_cvMutex;
 242         std::condition_variable m_cvStart;
 243         std::condition_variable m_cvStop;
 244         std::condition_variable m_cvDone;
 245
 246         volatile bool   m_bRunning;
 247         volatile bool   m_bStopped;
 248         volatile size_t m_doneCount;
 249         std::atomic<bool> m_bTimeElapsed;
 250         std::atomic<size_t> m_readyCount;
 251
 252         std::chrono::milliseconds m_testDuration;
 253     };
 254
 255     inline thread::thread( thread_pool& master, int type /*= 0*/ )
 256         : m_pool( master )
 257         , m_type( type )
 258         , m_id( master.get_next_id())
 259         , m_impl( &thread::run, this )
 260     {}
 261
 262     inline thread::thread( thread const& sample )
 263         : m_pool( sample.m_pool )
 264         , m_type( sample.m_type )
 265         , m_id( m_pool.get_next_id() )
 266         , m_impl( &thread::run, this )
 267     {}
 268
 269     inline void thread::run()
 270     {
 271         SetUp();
 272         m_pool.ready_to_start( *this );
 273         test();
 274         m_pool.thread_done( *this );
 275         TearDown();
 276     }
 277
 278     inline bool thread::time_elapsed() const
 279     {
 280         return m_pool.m_bTimeElapsed.load( std::memory_order_acquire );
 281     }
 282
 283 } // namespace cds_test
 284
 285 #endif // CDSTEST_THREAD_H