[Boost-commit] svn:boost r48636 - in sandbox/SOC/2007/signals: boost/dataflow/signals/component boost/dataflow/utility libs/dataflow/build/xcodeide/dataflow.xcodeproj libs/dataflow/example/threadpool

Subject: [Boost-commit] svn:boost r48636 - in sandbox/SOC/2007/signals: boost/dataflow/signals/component boost/dataflow/utility libs/dataflow/build/xcodeide/dataflow.xcodeproj libs/dataflow/example/threadpool
From: stipe_at_[hidden]
Date: 2008-09-06 16:05:08

Author: srajko
Date: 2008-09-06 16:05:07 EDT (Sat, 06 Sep 2008)
New Revision: 48636
URL: http://svn.boost.org/trac/boost/changeset/48636

Log:
new threadpool example
Added:
   sandbox/SOC/2007/signals/boost/dataflow/utility/bind_functor.hpp   (contents, props changed)
   sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/threadpool_component_example.cpp   (contents, props changed)
Text files modified: 
   sandbox/SOC/2007/signals/boost/dataflow/signals/component/filter_base.hpp                |     4 ++--                                    
   sandbox/SOC/2007/signals/libs/dataflow/build/xcodeide/dataflow.xcodeproj/project.pbxproj |     4 ++++                                    
   sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/Jamfile                        |     1 +                                       
   3 files changed, 7 insertions(+), 2 deletions(-)

Modified: sandbox/SOC/2007/signals/boost/dataflow/signals/component/filter_base.hpp
==============================================================================
--- sandbox/SOC/2007/signals/boost/dataflow/signals/component/filter_base.hpp	(original)
+++ sandbox/SOC/2007/signals/boost/dataflow/signals/component/filter_base.hpp	2008-09-06 16:05:07 EDT (Sat, 06 Sep 2008)
@@ -11,7 +11,7 @@
 #include <boost/dataflow/support/port/port_adapter.hpp>
 
 #include <boost/signal.hpp>
-
+#include <boost/mpl/vector.hpp>
 
 namespace boost {  namespace signals {
 
@@ -136,7 +136,7 @@
     };
 };
 
-template<typename Filter, typename Signals, typename InSignatures>
+template<typename Filter, typename Signals, typename InSignatures = mpl::vector<> >
 class filter_base : public dataflow::component<filter_component_traits<Filter, Signals, InSignatures> >
 {
 public:

Added: sandbox/SOC/2007/signals/boost/dataflow/utility/bind_functor.hpp
==============================================================================
--- (empty file)
+++ sandbox/SOC/2007/signals/boost/dataflow/utility/bind_functor.hpp	2008-09-06 16:05:07 EDT (Sat, 06 Sep 2008)
@@ -0,0 +1,54 @@
+// Copyright 2008 Stjepan Rajko.
+// 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)
+
+#ifndef BOOST_DATAFLOW__UTILITY__BIND_FUNCTOR_HPP
+#define BOOST_DATAFLOW__UTILITY__BIND_FUNCTOR_HPP
+
+
+#include <boost/bind.hpp>
+#include <boost/bind/placeholders.hpp>
+#include <boost/function.hpp>
+
+
+namespace boost { namespace dataflow {
+
+namespace utility {
+
+    struct bind_functor
+    {
+        typedef boost::function<void()> result_type;
+        
+        template<typename T0>
+        result_type operator()(const T0 &t0)
+        {
+            return boost::bind(t0);
+        }
+
+        template<typename T0, typename T1>
+        result_type operator()(const T0 &t0, const T1 &t1)
+        {
+            return boost::bind(t0, t1);
+        }
+
+        template<typename T0, typename T1, typename T2>
+        result_type operator()(const T0 &t0, const T1 &t1, const T2 &t2)
+        {
+            return boost::bind(t0, t1, t2);
+        }
+
+        template<typename T0, typename T1, typename T2, typename T3>
+        result_type operator()(const T0 &t0, const T1 &t1, const T2 &t2, const T3 &t3)
+        {
+            return boost::bind(t0, t1, t2, t3);
+        }
+        
+        //...
+    };
+
+} // namespace utility
+
+} } // namespace boost::dataflow
+
+#endif // BOOST_DATAFLOW__UTILITY__BIND_FUNCTOR_HPP

Modified: sandbox/SOC/2007/signals/libs/dataflow/build/xcodeide/dataflow.xcodeproj/project.pbxproj
==============================================================================
--- sandbox/SOC/2007/signals/libs/dataflow/build/xcodeide/dataflow.xcodeproj/project.pbxproj	(original)
+++ sandbox/SOC/2007/signals/libs/dataflow/build/xcodeide/dataflow.xcodeproj/project.pbxproj	2008-09-06 16:05:07 EDT (Sat, 06 Sep 2008)
@@ -143,6 +143,7 @@
                 089C098C0E0B11FF00397123 /* test_network.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = test_network.cpp; sourceTree = "<group>"; };
                 089CDA940D832AD200731C70 /* tutorial.qbk */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; path = tutorial.qbk; sourceTree = "<group>"; };
                 089CDAA90D8333CC00731C70 /* unary_operation.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; path = unary_operation.hpp; sourceTree = "<group>"; };
+		089DE1580E730C100038115E /* bind_functor.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; path = bind_functor.hpp; sourceTree = "<group>"; };
                 089E78C20E132E220008C0BB /* dynamic_multi_port.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; path = dynamic_multi_port.hpp; sourceTree = "<group>"; };
                 089F845A0E2CFAB900F6B668 /* lazy_sequence.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; path = lazy_sequence.hpp; sourceTree = "<group>"; };
                 089F845E0E2D032E00F6B668 /* test_lazy_sequence.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = test_lazy_sequence.cpp; sourceTree = "<group>"; };
@@ -200,6 +201,7 @@
                 08A364C10E129DC7001E6002 /* test_dynamic_port.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = test_dynamic_port.cpp; sourceTree = "<group>"; };
                 08A438900E295AE7009845FD /* component.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = component.cpp; sourceTree = "<group>"; };
                 08A439B80E295B8B009845FD /* Jamfile */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.jam; path = Jamfile; sourceTree = "<group>"; };
+		08A6948E0E71D7100065ECFD /* threadpool_component_example.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = threadpool_component_example.cpp; sourceTree = "<group>"; };
                 08A6B2650E25A566005539F2 /* blueprint_bank.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = blueprint_bank.cpp; sourceTree = "<group>"; };
                 08A6B2660E25A566005539F2 /* blueprint_bank.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; path = blueprint_bank.hpp; sourceTree = "<group>"; };
                 08A6B2670E25A566005539F2 /* blueprint_component.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = blueprint_component.cpp; sourceTree = "<group>"; };
@@ -757,6 +759,7 @@
                                 08E228DF0E6207EC00D1C2AF /* boost */,
                                 08E228FE0E62084100D1C2AF /* Jamfile */,
                                 08E229020E6208AC00D1C2AF /* threadpool_example.cpp */,
+				08A6948E0E71D7100065ECFD /* threadpool_component_example.cpp */,
                         );
                         path = threadpool;
                         sourceTree = "<group>";
@@ -1052,6 +1055,7 @@
                                 089AE6390D79D95C00AB9DA8 /* member_function_signature.hpp */,
                                 08C9D7DA0D83C5EB00354FF8 /* bind_mem_fn_overload.hpp */,
                                 08AD8AC40D84E3A9008A9764 /* has_call_operator.hpp */,
+				089DE1580E730C100038115E /* bind_functor.hpp */,
                         );
                         path = utility;
                         sourceTree = "<group>";

Modified: sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/Jamfile
==============================================================================
--- sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/Jamfile	(original)
+++ sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/Jamfile	2008-09-06 16:05:07 EDT (Sat, 06 Sep 2008)
@@ -13,3 +13,4 @@
     ;
 
 exe threadpool_example : threadpool_example.cpp ;
+exe threadpool_component_example : threadpool_component_example.cpp ;

Added: sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/threadpool_component_example.cpp
==============================================================================
--- (empty file)
+++ sandbox/SOC/2007/signals/libs/dataflow/example/threadpool/threadpool_component_example.cpp	2008-09-06 16:05:07 EDT (Sat, 06 Sep 2008)
@@ -0,0 +1,246 @@
+// Copyright 2008 Stjepan Rajko.
+// 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)
+
+#include <iostream>
+#include <cstdlib>
+#include <stdexcept>
+
+#include <boost/future/future.hpp>
+#include <boost/fusion/include/fused.hpp>
+#include <boost/fusion/include/join.hpp>
+#include <boost/fusion/include/make_vector.hpp>
+#define signalslib signals
+#define signals signals
+#include <boost/thread_safe_signal.hpp>
+//#include <boost/signals.hpp>
+
+#include <boost/dataflow/signals/component/storage.hpp>
+#include <boost/dataflow/signals/component/function.hpp>
+#include <boost/dataflow/signals/connection/operators.hpp>
+#include <boost/dataflow/utility/bind_mem_fn.hpp>
+#include <boost/dataflow/utility/bind_functor.hpp>
+
+#include "boost/tp/fifo.hpp"
+#include "boost/tp/lazy.hpp"
+#include "boost/tp/pool.hpp"
+#include "boost/tp/poolsize.hpp"
+#include "boost/tp/bounded_channel.hpp"
+#include "boost/tp/priority.hpp"
+
+
+namespace tp = boost::tp;
+
+template<typename Threadpool, typename Component>
+class async_component;
+
+namespace detail {
+
+    template<typename Threadpool, typename Component>
+    class async_component_impl
+    // filter_base is like filter, but it doesn't come with it's own signal.
+    // instead, the derived class must provide a default_signal member function
+    // that refers to the default signal.  We will use this to return the
+    // default signal of the underlying component.
+         : public boost::signals::filter_base<async_component<Threadpool, Component>, typename Component::signal_type >
+    {
+    public:
+        typedef typename Component::signature_type signature_type;
+        typedef typename Component::signal_type signal_type;
+        typedef void result_type;
+        
+        template<typename T0>
+        async_component_impl(Threadpool &threadpool, const T0 &t0)
+            : m_component(t0), m_threadpool(threadpool)
+        {
+            // record the appropriate operator() overload of Component into m_component_function,
+            // so we can submit it as a task later
+            typedef typename boost::dataflow::utility::slot_type<signature_type, Component>::type mem_fn_type;
+
+            m_component_function = boost::dataflow::utility::bind_mem_fn<mem_fn_type, Component>
+                (static_cast<mem_fn_type>(&Component::operator()), m_component);
+        }
+        
+        template <class Seq>
+        void operator()(const Seq &vec_par) const
+        {
+            // add the next function as a task in the pool
+            std::cout << "adding task" << std::endl;
+            // bind_functor is just a function object that calls bind
+            // we create a fused version so we can call it with a fusion sequence
+            boost::fusion::fused<boost::dataflow::utility::bind_functor> fused_bind;
+            
+            // submit the task (the first parameter to bind is the function,
+            // and the rest are the bound function arguments).
+            boost::tp::task< void > t(
+                m_threadpool.submit(
+                    fused_bind(
+                        boost::fusion::join(
+                            boost::fusion::make_vector(m_component_function),
+                            vec_par
+                    )   )
+                    ));
+        }
+        // with this, anything that connects to the async_component's default
+        // signal will actually connect to the default signal of the underlying
+        // component
+        typename Component::signal_type &default_signal()
+        {
+            namespace dataflow = boost::dataflow;
+            return dataflow::get_default_port<
+                    dataflow::args::left,
+                    dataflow::signals::connect_mechanism, 
+                    dataflow::signals::tag
+                > (m_component);
+        }
+
+    private:
+        Component m_component;
+        Threadpool &m_threadpool;
+        boost::function<signature_type> m_component_function;
+    };
+}
+
+
+// our new async_component class - it will create a new task for the underlying
+// component when it's operator() is called.
+template<typename Threadpool, typename Component>
+class async_component : public boost::fusion::unfused_inherited<
+    detail::async_component_impl<Threadpool,Component>,
+    typename boost::function_types::parameter_types<typename Component::signature_type>::type >
+{
+    typedef boost::fusion::unfused_inherited<
+        detail::async_component_impl<Threadpool,Component>,
+        typename boost::function_types::parameter_types<typename Component::signature_type>::type>
+        base_type;
+public:
+    template<typename T0>
+    async_component(Threadpool &threadpool, const T0 &t0)
+        : base_type(threadpool, t0)
+    {}
+};
+
+// just an operation to work with
+int inc_fn(int x)
+{
+    std::cout << "filter: " << x+1 << std::endl;
+    return x+1;
+}
+
+// a component that displays text, waits a second, and then sends a signal
+class printer : public boost::signals::filter<printer, void()>
+{
+public:
+    printer(const std::string &text)
+        : m_text(text)
+    {}
+    void operator()()
+    {
+        std::cout << m_text << std::endl;
+        boost::this_thread::sleep(boost::posix_time::seconds(1));
+        out();
+    }
+private:
+    std::string m_text;
+};
+
+// a function to submit the first task
+template<typename Threadpool, typename Next>
+void submit(Threadpool &threadpool, Next &next)
+{
+    tp::task< void > task(
+        threadpool.submit(
+            boost::bind(&Next::send, boost::ref(next))));
+}
+
+int main( int argc, char *argv[])
+{
+   typedef 
+        tp::pool<
+            tp::fixed,
+            tp::unbounded_channel< tp::fifo >
+        > threadpool_type;
+        
+    threadpool_type pool(tp::max_poolsize(5));
+
+    typedef boost::signals::storage<void(int)> source_type;
+    typedef boost::signals::function<void(int), int(int)> filter_type;
+    typedef async_component<threadpool_type, filter_type> async_filter_type;
+
+    // our components
+    source_type source(1);
+    async_filter_type increase1(pool, inc_fn);
+    filter_type increase2(inc_fn);
+    filter_type increase3(inc_fn);
+    async_filter_type increase4(pool, inc_fn);
+    async_filter_type increase5(pool, inc_fn);
+    
+    // our network
+    //  increase1 >>= increase2 >>= increase3 will be in its own thread
+    //  increase3 will be in its own thread
+    //  increase4 will be in its own thread
+    source
+        | (increase1 >>= increase2 >>= increase3)
+        | (increase4 >>= increase5);
+
+    // submit the first task
+    submit(pool, source);
+           
+    // wait a little
+    boost::this_thread::sleep(boost::posix_time::seconds(1));
+
+    // --------------------------------------------------------------
+
+/*
+   typedef 
+        tp::pool<
+            tp::fixed,
+            tp::unbounded_channel< tp::priority< int > >
+        > priority_threadpool_type;
+        
+    priority_threadpool_type priority_pool(tp::max_poolsize(5));
+    
+    typedef boost::signals::storage<void()> print_starter_type;
+    typedef async_component<priority_threadpool_type, printer> async_printer_type;
+    
+    print_starter_type print_starter;
+    async_printer_type print1(priority_pool, "World");
+    async_printer_type print2(priority_pool, " ");
+    async_printer_type print3(priority_pool, "Hello");
+    async_printer_type print4(priority_pool, "!");
+    
+    print_starter
+        | print1
+        | print2
+        | print3
+        | print4;
+        
+    // submit the first task
+    submit(priority_pool, print_starter);
+
+    // wait a little
+    boost::this_thread::sleep(boost::posix_time::seconds(1));
+*/
+        
+    // --------------------------------------------------------------
+
+    typedef boost::signals::storage<void()> tick_starter_type;
+    typedef async_component<threadpool_type, printer> ticker_type;
+    
+    // our components
+    tick_starter_type tick_starter;
+    ticker_type ticker1(pool, "tick 1..."), ticker2(pool, "tick 2..."), ticker3(pool, "tick 3...");
+
+    // our network
+    tick_starter >>= ticker1 >>= ticker2 >>= ticker3 >>= ticker1;
+    
+    // submit the first task
+    submit(pool, tick_starter);
+
+    // wait a little
+    boost::this_thread::sleep(boost::posix_time::seconds(5));
+
+    pool.shutdown();
+    return 0;
+}