[Boost-commit] svn:boost r82094 - in trunk: boost/thread/pthread libs/thread/build libs/thread/example libs/thread/test

Subject: [Boost-commit] svn:boost r82094 - in trunk: boost/thread/pthread libs/thread/build libs/thread/example libs/thread/test
From: vicente.botet_at_[hidden]
Date: 2012-12-19 05:50:24

Author: viboes
Date: 2012-12-19 05:50:23 EST (Wed, 19 Dec 2012)
New Revision: 82094
URL: http://svn.boost.org/trac/boost/changeset/82094

Log:
Thread: #7422: don't use internal_mutex when interruptions not enabled
Added:
   trunk/libs/thread/example/perf_condition_variable.cpp   (contents, props changed)
Text files modified: 
   trunk/boost/thread/pthread/condition_variable.hpp     |    33 +++++++++++++++++++++++++--------       
   trunk/boost/thread/pthread/condition_variable_fwd.hpp |     8 ++++++++                                
   trunk/libs/thread/build/Jamfile.v2                    |     1 +                                       
   trunk/libs/thread/test/Jamfile.v2                     |     8 +++++---                                
   4 files changed, 39 insertions(+), 11 deletions(-)

Modified: trunk/boost/thread/pthread/condition_variable.hpp
==============================================================================
--- trunk/boost/thread/pthread/condition_variable.hpp	(original)
+++ trunk/boost/thread/pthread/condition_variable.hpp	2012-12-19 05:50:23 EST (Wed, 19 Dec 2012)
@@ -57,18 +57,28 @@
 
     inline void condition_variable::wait(unique_lock<mutex>& m)
     {
+#if defined BOOST_THREAD_THROW_IF_PRECONDITION_NOT_SATISFIED
+        if(! m.owns_lock())
+        {
+            boost::throw_exception(condition_error(-1, "boost::condition_variable::wait precondition"));
+        }
+#endif
         int res=0;
         {
-            thread_cv_detail::lock_on_exit<unique_lock<mutex> > guard;
 #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
+            thread_cv_detail::lock_on_exit<unique_lock<mutex> > guard;
             detail::interruption_checker check_for_interruption(&internal_mutex,&cond);
-#else
-            boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
-#endif
             guard.activate(m);
             do {
               res = pthread_cond_wait(&cond,&internal_mutex);
             } while (res == EINTR);
+#else
+            //boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
+            pthread_mutex_t* the_mutex = m.mutex()->native_handle();
+            do {
+              res = pthread_cond_wait(&cond,the_mutex);
+            } while (res == EINTR);
+#endif
         }
 #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
         this_thread::interruption_point();
@@ -83,21 +93,24 @@
                 unique_lock<mutex>& m,
                 struct timespec const &timeout)
     {
+#if defined BOOST_THREAD_THROW_IF_PRECONDITION_NOT_SATISFIED
         if (!m.owns_lock())
         {
             boost::throw_exception(condition_error(EPERM, "condition_variable do_wait_until: mutex not locked"));
         }
-
+#endif
         thread_cv_detail::lock_on_exit<unique_lock<mutex> > guard;
         int cond_res;
         {
 #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
             detail::interruption_checker check_for_interruption(&internal_mutex,&cond);
-#else
-            boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
-#endif
             guard.activate(m);
             cond_res=pthread_cond_timedwait(&cond,&internal_mutex,&timeout);
+#else
+            //boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
+            pthread_mutex_t* the_mutex = m.mutex()->native_handle();
+            cond_res=pthread_cond_timedwait(&cond,the_mutex,&timeout);
+#endif
         }
 #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
         this_thread::interruption_point();
@@ -115,13 +128,17 @@
 
     inline void condition_variable::notify_one() BOOST_NOEXCEPT
     {
+#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
         boost::pthread::pthread_mutex_scoped_lock internal_lock(&internal_mutex);
+#endif
         BOOST_VERIFY(!pthread_cond_signal(&cond));
     }
 
     inline void condition_variable::notify_all() BOOST_NOEXCEPT
     {
+#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
         boost::pthread::pthread_mutex_scoped_lock internal_lock(&internal_mutex);
+#endif
         BOOST_VERIFY(!pthread_cond_broadcast(&cond));
     }
 

Modified: trunk/boost/thread/pthread/condition_variable_fwd.hpp
==============================================================================
--- trunk/boost/thread/pthread/condition_variable_fwd.hpp	(original)
+++ trunk/boost/thread/pthread/condition_variable_fwd.hpp	2012-12-19 05:50:23 EST (Wed, 19 Dec 2012)
@@ -32,7 +32,9 @@
     class condition_variable
     {
     private:
+#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
         pthread_mutex_t internal_mutex;
+#endif
         pthread_cond_t cond;
 
     public:
@@ -53,25 +55,31 @@
       BOOST_THREAD_NO_COPYABLE(condition_variable)
         condition_variable()
         {
+#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
             int const res=pthread_mutex_init(&internal_mutex,NULL);
             if(res)
             {
                 boost::throw_exception(thread_resource_error(res, "boost:: condition_variable constructor failed in pthread_mutex_init"));
             }
+#endif
             int const res2=pthread_cond_init(&cond,NULL);
             if(res2)
             {
+#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
                 BOOST_VERIFY(!pthread_mutex_destroy(&internal_mutex));
+#endif
                 boost::throw_exception(thread_resource_error(res2, "boost:: condition_variable constructor failed in pthread_cond_init"));
             }
         }
         ~condition_variable()
         {
             int ret;
+#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
             do {
               ret = pthread_mutex_destroy(&internal_mutex);
             } while (ret == EINTR);
             BOOST_ASSERT(!ret);
+#endif
             do {
               ret = pthread_cond_destroy(&cond);
             } while (ret == EINTR);

Modified: trunk/libs/thread/build/Jamfile.v2
==============================================================================
--- trunk/libs/thread/build/Jamfile.v2	(original)
+++ trunk/libs/thread/build/Jamfile.v2	2012-12-19 05:50:23 EST (Wed, 19 Dec 2012)
@@ -115,6 +115,7 @@
       <link>shared:<define>BOOST_THREAD_BUILD_DLL=1
       <define>BOOST_THREAD_THROW_IF_PRECONDITION_NOT_SATISFIED
       #<define>BOOST_SYSTEM_NO_DEPRECATED
+      #<define>BOOST_THREAD_DONT_PROVIDE_INTERRUPTIONS 
       <library>/boost/system//boost_system
     ;
 

Added: trunk/libs/thread/example/perf_condition_variable.cpp
==============================================================================
--- (empty file)
+++ trunk/libs/thread/example/perf_condition_variable.cpp	2012-12-19 05:50:23 EST (Wed, 19 Dec 2012)
@@ -0,0 +1,237 @@
+//  (C) Copyright 2012 Vicente J. Botet Escriba
+//
+//  Distributed under the Boost Software License, Version 1.0. (See
+//  accompanying file LICENSE_1_0.txt or copy at
+//  http://www.boost.org/LICENSE_1_0.txt)
+//
+// This performance test is based on the performance test provided by maxim.yegorushkin
+// at https://svn.boost.org/trac/boost/ticket/7422
+
+#define BOOST_THREAD_DONT_PROVIDE_INTERRUPTIONS
+
+#include <boost/thread/condition_variable.hpp>
+#include <boost/thread/mutex.hpp>
+#include <boost/chrono/stopwatches/simple_stopwatch.hpp>
+
+#include <condition_variable>
+#include <future>
+#include <limits>
+#include <cstdio>
+#include <thread>
+#include <mutex>
+#include <vector>
+
+////////////////////////////////////////////////////////////////////////////////////////////////
+
+namespace
+{
+  ////////////////////////////////////////////////////////////////////////////////////////////////
+
+//  class Stopwatch
+//  {
+//  public:
+//    typedef long long rep;
+//
+//    static rep now()
+//    {
+//      timespec ts;
+//      if (clock_gettime(CLOCK_MONOTONIC, &ts)) abort();
+//      return ts.tv_sec * rep(1000000000) + ts.tv_nsec;
+//    }
+//
+//    Stopwatch() :
+//      start_(now())
+//    {
+//    }
+//
+//    rep elapsed() const
+//    {
+//      return now() - start_;
+//    }
+//
+//  private:
+//    rep start_;
+//  };
+
+  typedef boost::chrono::simple_stopwatch<> Stopwatch;
+  ////////////////////////////////////////////////////////////////////////////////////////////////
+
+  struct BoostTypes
+  {
+    typedef boost::condition_variable condition_variable;
+    typedef boost::mutex mutex;
+    typedef boost::mutex::scoped_lock scoped_lock;
+  };
+
+  struct StdTypes
+  {
+    typedef std::condition_variable condition_variable;
+    typedef std::mutex mutex;
+    typedef std::unique_lock<std::mutex> scoped_lock;
+  };
+
+  template <class Types>
+  struct SharedData: Types
+  {
+    unsigned const iterations;
+    unsigned counter;
+    unsigned semaphore;
+    typename Types::condition_variable cnd;
+    typename Types::mutex mtx;
+    Stopwatch::rep producer_time;
+
+    SharedData(unsigned iterations, unsigned consumers) :
+      iterations(iterations), counter(), semaphore(consumers) // Initialize to the number of consumers. (*)
+          , producer_time()
+    {
+    }
+  };
+
+
+  ////////////////////////////////////////////////////////////////////////////////////////////////
+
+  template <class S>
+  void producer_thread(S* shared_data)
+  {
+    Stopwatch sw;
+
+    unsigned const consumers = shared_data->semaphore; // (*)
+    for (unsigned i = shared_data->iterations; i--;)
+    {
+      {
+        typename S::scoped_lock lock(shared_data->mtx);
+        // Wait till all consumers signal.
+        while (consumers != shared_data->semaphore)
+        {
+          shared_data->cnd.wait(lock);
+        }
+        shared_data->semaphore = 0;
+        // Signal consumers.
+        ++shared_data->counter;
+      }
+      shared_data->cnd.notify_all();
+    }
+
+    shared_data->producer_time = sw.elapsed().count();
+  }
+
+  template <class S>
+  void consumer_thread(S* shared_data)
+  {
+    unsigned counter = 0;
+    while (counter != shared_data->iterations)
+    {
+      {
+        typename S::scoped_lock lock(shared_data->mtx);
+        // Wait till the producer signals.
+        while (counter == shared_data->counter)
+        {
+          shared_data->cnd.wait(lock);
+        }
+        counter = shared_data->counter;
+        // Signal the producer.
+        ++shared_data->semaphore;
+      }
+      shared_data->cnd.notify_all();
+    }
+  }
+
+  ////////////////////////////////////////////////////////////////////////////////////////////////
+
+  template <class Types>
+  Stopwatch::rep benchmark_ping_pong(unsigned consumer_count)
+  {
+    typedef SharedData<Types> S;
+
+    auto best_producer_time = std::numeric_limits<Stopwatch::rep>::max();
+
+    std::vector<std::thread> consumers
+    { consumer_count };
+
+    // Run the benchmark 10 times and report the best time.
+    for (int times = 10; times--;)
+    {
+      S shared_data
+      { 100000, consumer_count };
+
+      // Start the consumers.
+      for (unsigned i = 0; i < consumer_count; ++i)
+        consumers[i] = std::thread
+        { consumer_thread<S> , &shared_data };
+      // Start the producer and wait till it finishes.
+      std::thread
+      { producer_thread<S> , &shared_data }.join();
+      // Wait till consumers finish.
+      for (unsigned i = 0; i < consumer_count; ++i)
+        consumers[i].join();
+
+      best_producer_time = std::min(best_producer_time, shared_data.producer_time);
+
+    }
+
+    return best_producer_time;
+  }
+
+////////////////////////////////////////////////////////////////////////////////////////////////
+
+} // namespace
+
+////////////////////////////////////////////////////////////////////////////////////////////////
+
+// sudo chrt -f 99 /usr/bin/time -f "\n***\ntime: %E\ncontext switches: %c\nwaits: %w" /home/max/otsquant/build/Linux-x86_64-64.g++-release/test/test
+
+/*
+
+ Producer-consumer ping-pong tests. It aims to benchmark condition variables with and without
+ thread cancellation support by comparing the time it took to complete the benchmark.
+
+ Condition variable with thread cancellation support is boost::condition_variable from
+ boost-1.51. Without - std::condition_variable that comes with gcc-4.7.2.
+
+ One producer, one to CONSUMER_MAX consumers. The benchmark calls
+ condition_variable::notify_all() without holding a mutex to maximize contention within this
+ function. Each benchmark for a number of consumers is run three times and the best time is
+ picked to get rid of outliers.
+
+ The results are reported for each benchmark for a number of consumers. The most important number
+ is (std - boost) / std * 100. Positive numbers are when boost::condition_variable is faster,
+ negative it is slower.
+
+ */
+
+int main()
+{
+  std::printf("MAIN\n");
+  enum
+  {
+    CONSUMER_MAX = 2
+  };
+
+  struct
+  {
+    Stopwatch::rep boost, std;
+  } best_times[CONSUMER_MAX] = {};
+
+  for (unsigned i = 1; i <= CONSUMER_MAX; ++i)
+  {
+    auto& b = best_times[i - 1];
+    std::printf("STD: %d\n", i);
+    b.std = benchmark_ping_pong<StdTypes> (i);
+    std::printf("BOOST: %d\n", i);
+    b.boost = benchmark_ping_pong<BoostTypes> (i);
+
+    std::printf("consumers:                 %4d\n", i);
+    std::printf("best std producer time:   %15.9fsec\n", b.std * 1e-9);
+    std::printf("best boost producer time: %15.9fsec\n", b.boost * 1e-9);
+    std::printf("(std - boost) / std:       %7.2f%%\n", (b.std - b.boost) * 100. / b.std);
+  }
+
+  printf("\ncsv:\n\n");
+  printf("consumers,(std-boost)/std,std,boost\n");
+  for (unsigned i = 1; i <= CONSUMER_MAX; ++i)
+  {
+    auto& b = best_times[i - 1];
+    printf("%d,%f,%lld,%lld\n", i, (b.std - b.boost) * 100. / b.std, b.std, b.boost);
+  }
+  return 1;
+}

Modified: trunk/libs/thread/test/Jamfile.v2
==============================================================================
--- trunk/libs/thread/test/Jamfile.v2	(original)
+++ trunk/libs/thread/test/Jamfile.v2	2012-12-19 05:50:23 EST (Wed, 19 Dec 2012)
@@ -126,9 +126,9 @@
     [ run $(sources) ../src/tss_null.cpp ../build//boost_thread/<link>static
         : : :
       : $(name)_lib ]
-    [ run $(sources) ../build//boost_thread : : :
-      <define>BOOST_THREAD_DONT_PROVIDE_INTERRUPTIONS 
-      : $(name)_noit ]
+    #[ run $(sources) ../build//boost_thread : : :
+    #  <define>BOOST_THREAD_DONT_PROVIDE_INTERRUPTIONS 
+    #  : $(name)_noit ]
     ;
 }
 
@@ -662,6 +662,7 @@
     explicit ts_ ;
     test-suite ts_
     :
+
           #[ thread-run2-noit ./sync/futures/future/then_pass.cpp : future__then_p ]
           #[ thread-run ../example/test_so.cpp ]
           #[ thread-run ../example/test_so2.cpp ]
@@ -670,6 +671,7 @@
           #[ thread-run test_7665.cpp ]
           #[ thread-run test_7666.cpp ]
           #[ thread-run ../example/unwrap.cpp ]
+          [ thread-run ../example/perf_condition_variable.cpp ]
     ;
 
 }