[Boost-commit] svn:boost r55857 - in sandbox/task/boost/task: . detail

Subject: [Boost-commit] svn:boost r55857 - in sandbox/task/boost/task: . detail
From: oliver.kowalke_at_[hidden]
Date: 2009-08-29 03:16:04

Author: olli
Date: 2009-08-29 03:16:03 EDT (Sat, 29 Aug 2009)
New Revision: 55857
URL: http://svn.boost.org/trac/boost/changeset/55857

Log:
- fiber corrected

Text files modified: 
   sandbox/task/boost/task/detail/fiber_posix.hpp |     8 ++--                                    
   sandbox/task/boost/task/detail/worker.hpp      |    60 ++++++++++++++++++++++----------------- 
   sandbox/task/boost/task/new_thread.hpp         |    21 ++++++++++++-                           
   3 files changed, 57 insertions(+), 32 deletions(-)

Modified: sandbox/task/boost/task/detail/fiber_posix.hpp
==============================================================================
--- sandbox/task/boost/task/detail/fiber_posix.hpp	(original)
+++ sandbox/task/boost/task/detail/fiber_posix.hpp	2009-08-29 03:16:03 EDT (Sat, 29 Aug 2009)
@@ -156,10 +156,10 @@
         bool ready() const
         { return uninitialized_() || ready_(); }
 
-    	bool running() const
+    bool running() const
         { return running_(); }
 
-   	 bool exited() const
+   	bool exited() const
         { return exited_(); }
 
         void switch_to( sptr_t & to)
@@ -173,8 +173,8 @@
 
         void run()
         {
-		BOOST_ASSERT( uninitialized_() || ready_() );
-		if ( uninitialized_() ) init_();
+		BOOST_ASSERT( uninitialized_() );
+		init_();
                 BOOST_ASSERT( ready_() );
                 state_ = st_running;
                 run_();

Modified: sandbox/task/boost/task/detail/worker.hpp
==============================================================================
--- sandbox/task/boost/task/detail/worker.hpp	(original)
+++ sandbox/task/boost/task/detail/worker.hpp	2009-08-29 03:16:03 EDT (Sat, 29 Aug 2009)
@@ -122,21 +122,6 @@
                 BOOST_ASSERT( ca.empty() );
         }
 
-	void try_blocked_fibers_()
-	{
-		if ( ums_.has_blocked() )
-		{
-			fiber::sptr_t this_fib = fib_;
-			ums_.put_runnable( this_fib);
-			fiber::sptr_t blocked_fib;
-			ums_.try_take_blocked( blocked_fib);
-			BOOST_ASSERT( blocked_fib);
-			fib_ = blocked_fib;
-			this_fib->switch_to( blocked_fib);
-			fib_ = this_fib;
-		}
-	}
-
         bool take_global_callable_(
                         callable & ca,
                         posix_time::time_duration const& asleep)
@@ -153,6 +138,7 @@
                 std::size_t idx( rnd_idx_() );
                 for ( std::size_t j( 0); j < pool_.wg_.size(); ++j)
                 {
+			// TODO: not thread-safe -> segfault
                         Worker other( pool_.wg_[idx]);
                         if ( this_thread::get_id() == other.get_id() ) continue;
                         if ( ++idx >= pool_.wg_.size() ) idx = 0;
@@ -162,8 +148,29 @@
                 return false;
         }
 
+	bool try_blocked_fibers_()
+	{
+		if ( ums_.has_blocked() )
+		{
+			fiber::sptr_t this_fib = fib_;
+			BOOST_ASSERT( this_fib);
+			ums_.put_runnable( this_fib);
+			fiber::sptr_t blocked_fib;
+			ums_.try_take_blocked( blocked_fib);
+			BOOST_ASSERT( blocked_fib);
+			fib_ = blocked_fib;
+			BOOST_ASSERT( this_fib->running() );
+			this_fib->switch_to( blocked_fib);
+			BOOST_ASSERT( this_fib->running() );
+			fib_ = this_fib;
+			return true;
+		}
+		return false;
+	}
+
         void process_( bool all)
         {
+		try_blocked_fibers_();
                 callable ca;
                 if ( all ? try_take_local_callable_( ca) || 
                                         try_steal_other_callable_( ca) ||
@@ -196,13 +203,14 @@
                         else
                                 this_thread::yield();
                 }
-		try_blocked_fibers_();
         }
 
         void run_()
-	{
+	{	
+		BOOST_ASSERT( fib_->running() );
                 while ( ! shutdown_() )
                         process_( true);
+		BOOST_ASSERT( fib_->running() );
         }
 
         bool shutdown_()
@@ -282,19 +290,17 @@
 
                 ums_.attach();
 
-		fiber::sptr_t fib(
-			fiber::create(
-				bind( & worker_object::run_, this),
-				stack_size_) );
-		fib_.swap( fib);
+		fib_ = fiber::create(
+			bind( & worker_object::run_, this),
+			stack_size_);
                 fib_->run();
-		fib_.reset();
+		BOOST_ASSERT( fib_->exited() );
         }
 
         void reschedule_until( function< bool() > const& pred)
         {
                 while ( ! pred() )
-			process_( false);
+			block();
         }
 
         void block()
@@ -306,11 +312,13 @@
                         ums_.try_take_runnable( runnable_fib);
                 else
                         runnable_fib = fiber::create(
-					bind( & worker_object::run_, this),
-					stack_size_);
+				bind( & worker_object::run_, this),
+				stack_size_);
                 BOOST_ASSERT( runnable_fib);
                 fib_ = runnable_fib;
+		BOOST_ASSERT( this_fib->running() );
                 this_fib->switch_to( runnable_fib);
+		BOOST_ASSERT( this_fib->running() );
                 fib_ = this_fib;
         }
 };

Modified: sandbox/task/boost/task/new_thread.hpp
==============================================================================
--- sandbox/task/boost/task/new_thread.hpp	(original)
+++ sandbox/task/boost/task/new_thread.hpp	2009-08-29 03:16:03 EDT (Sat, 29 Aug 2009)
@@ -43,8 +43,25 @@
 
 }
 
-struct new_thread
+class new_thread
 {
+private:
+	struct wrapper
+	{
+		callable	ca;
+		
+		wrapper( callable const& ca_)
+		: ca( ca_)
+		{}
+		
+		void operator()()
+		{
+			ca();
+			ca.clear();
+		}
+	};
+	
+public:
         template< typename R >
         handle< R > operator()( task< R > t)
         {
@@ -53,7 +70,7 @@
                 handle< R > h( ctx.get_handle( f) );
                 callable ca( ctx.get_callable( boost::move( t) ) );
                 shared_ptr< thread > thrd(
-			new thread( ca),
+			new thread( wrapper( ca) ),
                         detail::joiner() );
                 ca.reset( thrd);