[Boost-commit] svn:boost r55343 - in sandbox: boost/mapreduce boost/mapreduce/intermediates libs/mapreduce/examples/wordcount libs/mapreduce/test

Subject: [Boost-commit] svn:boost r55343 - in sandbox: boost/mapreduce boost/mapreduce/intermediates libs/mapreduce/examples/wordcount libs/mapreduce/test
From: cdm.henderson_at_[hidden]
Date: 2009-08-01 10:15:07

Author: chenderson
Date: 2009-08-01 10:15:05 EDT (Sat, 01 Aug 2009)
New Revision: 55343
URL: http://svn.boost.org/trac/boost/changeset/55343

Log:
Refined the type definitions for key/value pair types in Map and Reduce Tasks
Text files modified: 
   sandbox/boost/mapreduce/intermediates/in_memory.hpp     |    15 ++++++++-------                         
   sandbox/boost/mapreduce/intermediates/local_disk.hpp    |    25 +++++++++++++------------               
   sandbox/boost/mapreduce/job.hpp                         |    21 ++++++++-------------                   
   sandbox/libs/mapreduce/examples/wordcount/wordcount.cpp |    14 ++++++--------                          
   sandbox/libs/mapreduce/test/mrtest.cpp                  |    29 +++++++++++++++++------------           
   5 files changed, 52 insertions(+), 52 deletions(-)

Modified: sandbox/boost/mapreduce/intermediates/in_memory.hpp
==============================================================================
--- sandbox/boost/mapreduce/intermediates/in_memory.hpp	(original)
+++ sandbox/boost/mapreduce/intermediates/in_memory.hpp	2009-08-01 10:15:05 EDT (Sat, 01 Aug 2009)
@@ -29,8 +29,8 @@
     {
     }
 
-    void operator()(typename MapTask::intermediate_key_type const &/*key*/,
-                    typename ReduceTask::value_type         const &/*value*/)
+    void operator()(typename ReduceTask::key_type   const &/*key*/,
+                    typename ReduceTask::value_type const &/*value*/)
     {
     }
 };
@@ -46,12 +46,13 @@
     typedef
     std::vector<
         std::map<
-            typename MapTask::intermediate_key_type,
-            std::list<typename MapTask::intermediate_value_type> > >
+            typename ReduceTask::key_type,
+            std::list<typename ReduceTask::value_type> > >
     intermediates_t;
 
   public:
-    typedef MapTask map_task_type;
+    typedef MapTask    map_task_type;
+    typedef ReduceTask reduce_task_type;
     typedef reduce_null_output<MapTask, ReduceTask> store_result_type;
 
     in_memory(unsigned const num_partitions)
@@ -99,8 +100,8 @@
     }
 
 
-    bool const insert(typename map_task_type::intermediate_key_type   const &key,
-                      typename map_task_type::intermediate_value_type const &value)
+    bool const insert(typename reduce_task_type::key_type   const &key,
+                      typename reduce_task_type::value_type const &value)
     {
         typename intermediates_t::value_type &map = intermediates_[partitioner_(key, num_partitions_)];
 

Modified: sandbox/boost/mapreduce/intermediates/local_disk.hpp
==============================================================================
--- sandbox/boost/mapreduce/intermediates/local_disk.hpp	(original)
+++ sandbox/boost/mapreduce/intermediates/local_disk.hpp	2009-08-01 10:15:05 EDT (Sat, 01 Aug 2009)
@@ -108,8 +108,8 @@
         output_file_.open(filename_.c_str());
     }
 
-    void operator()(typename MapTask::intermediate_key_type const &key,
-                    typename ReduceTask::value_type         const &value)
+    void operator()(typename ReduceTask::key_type   const &key,
+                    typename ReduceTask::value_type const &value)
     {
         output_file_ << key << "\t" << value << "\n";
     }
@@ -138,7 +138,8 @@
     intermediates_t;
 
   public:
-    typedef MapTask map_task_type;
+    typedef MapTask    map_task_type;
+    typedef ReduceTask reduce_task_type;
     typedef reduce_file_output<MapTask, ReduceTask> store_result_type;
 
     local_disk(unsigned const num_partitions)
@@ -162,8 +163,8 @@
         }
     }
 
-    bool const insert(typename map_task_type::intermediate_key_type   const &key,
-                      typename map_task_type::intermediate_value_type const &value)
+    bool const insert(typename reduce_task_type::key_type   const &key,
+                      typename reduce_task_type::value_type const &value)
     {
         unsigned const partition = partitioner_(key, num_partitions_);
 
@@ -208,7 +209,7 @@
             std::ifstream infile(infilename.c_str());
             while (!infile.eof())
             {
-                typename map_task_type::intermediate_value_type value;
+                typename reduce_task_type::value_type value;
                 if (read_record(infile, key, value))
                 {
                     if (key != last_key  &&  key.length() > 0)
@@ -276,10 +277,10 @@
     template<typename Callback>
     void reduce(unsigned const partition, Callback &callback, results &result)
     {
-        typename map_task_type::intermediate_key_type   key;
-        typename map_task_type::intermediate_key_type   last_key;
-        typename map_task_type::intermediate_value_type value;
-        std::list<typename map_task_type::intermediate_value_type> values;
+        typename reduce_task_type::key_type   key;
+        typename reduce_task_type::key_type   last_key;
+        typename reduce_task_type::value_type value;
+        std::list<typename reduce_task_type::value_type> values;
 
         std::list<std::string> filenames;
         intermediates_t::const_iterator it = intermediate_files_.find(partition);
@@ -316,8 +317,8 @@
 
   protected:
     static bool const read_record(std::ifstream &infile,
-                                  typename map_task_type::intermediate_key_type   &key,
-                                  typename map_task_type::intermediate_value_type &value)
+                                  typename reduce_task_type::key_type   &key,
+                                  typename reduce_task_type::value_type &value)
     {
 #if defined(__SGI_STL_PORT)
         size_t keylen;

Modified: sandbox/boost/mapreduce/job.hpp
==============================================================================
--- sandbox/boost/mapreduce/job.hpp	(original)
+++ sandbox/boost/mapreduce/job.hpp	2009-08-01 10:15:05 EDT (Sat, 01 Aug 2009)
@@ -19,24 +19,19 @@
 
 template<typename T> size_t length(T const &str);
 
-template<
-    typename MapKey,
-    typename MapValue,
-    typename IntermediateKey,
-    typename IntermediateValue>
+template<typename MapKey, typename MapValue>
 class map_task
 {
   public:
     typedef MapKey              key_type;
     typedef MapValue            value_type;             
-    typedef IntermediateKey     intermediate_key_type;
-    typedef IntermediateValue   intermediate_value_type;
 };
 
-template<typename ReduceValue>
+template<typename ReduceKey, typename ReduceValue>
 class reduce_task
 {
   public:
+    typedef ReduceKey   key_type;
     typedef ReduceValue value_type;
 };
 
@@ -78,8 +73,8 @@
             return *this;
         }
 
-        bool const emit_intermediate(typename map_task_type::intermediate_key_type   const &key,
-                                     typename map_task_type::intermediate_value_type const &value)
+        bool const emit_intermediate(typename reduce_task_type::key_type   const &key,
+                                     typename reduce_task_type::value_type const &value)
         {
             return intermediate_store_.insert(key, value);
         }
@@ -105,14 +100,14 @@
         {
         }
 
-        void emit(typename map_task_type::intermediate_key_type const &key,
-                  typename reduce_task_type::value_type         const &value)
+        void emit(typename reduce_task_type::key_type   const &key,
+                  typename reduce_task_type::value_type const &value)
         {
             store_result_(key, value);
         }
 
         template<typename It>
-        void operator()(typename map_task_type::intermediate_key_type const &key, It it, It ite)
+        void operator()(typename reduce_task_type::key_type const &key, It it, It ite)
         {
             reduce_task_type::reduce(*this, key, it, ite);
         }

Modified: sandbox/libs/mapreduce/examples/wordcount/wordcount.cpp
==============================================================================
--- sandbox/libs/mapreduce/examples/wordcount/wordcount.cpp	(original)
+++ sandbox/libs/mapreduce/examples/wordcount/wordcount.cpp	2009-08-01 10:15:05 EDT (Sat, 01 Aug 2009)
@@ -29,10 +29,8 @@
 namespace wordcount {
 
 struct map_task : public boost::mapreduce::map_task<
-                             std::string,                           // MapKey
-                             std::pair<char const *, char const *>, // MapValue
-                             std::string,                           // IntermediateKey
-                             unsigned>                              // IntermediateValue
+                             std::string,                            // MapKey
+                             std::pair<char const *, char const *> > // MapValue
 {
     template<typename Runtime>
     static void map(Runtime &runtime, std::string const &/*key*/, value_type &value)
@@ -79,7 +77,7 @@
     }
 };
 
-struct reduce_task : public boost::mapreduce::reduce_task<unsigned>
+struct reduce_task : public boost::mapreduce::reduce_task<std::string, unsigned>
 {
     template<typename Runtime, typename It>
     static void reduce(Runtime &runtime, std::string const &key, It it, It const ite)
@@ -98,19 +96,19 @@
         intermediate_store.combine(instance);
     }
 
-    void start(map_task::intermediate_key_type const &)
+    void start(reduce_task::key_type const &)
     {
         total_ = 0;
     }
 
     template<typename IntermediateStore>
-    void finish(map_task::intermediate_key_type const &key, IntermediateStore &intermediate_store)
+    void finish(reduce_task::key_type const &key, IntermediateStore &intermediate_store)
     {
         if (total_ > 0)
             intermediate_store.insert(key, total_);
     }
 
-    void operator()(map_task::intermediate_value_type const &value)
+    void operator()(reduce_task::value_type const &value)
     {
         total_ += value;
     }

Modified: sandbox/libs/mapreduce/test/mrtest.cpp
==============================================================================
--- sandbox/libs/mapreduce/test/mrtest.cpp	(original)
+++ sandbox/libs/mapreduce/test/mrtest.cpp	2009-08-01 10:15:05 EDT (Sat, 01 Aug 2009)
@@ -50,18 +50,14 @@
 map_value_type;
 
 template<typename T>
-struct map_task : public boost::mapreduce::map_task<
-                             std::string,            // MapKey
-                             map_value_type,         // MapValue
-                             std::string,            // IntermediateKey
-                             unsigned>               // IntermediateValue
+struct map_task
+  : public boost::mapreduce::map_task<std::string, map_value_type>
 {
     template<typename Runtime>
     static void map(Runtime &runtime, std::string const &key, T &value);
 };
-typedef map_task<map_value_type> map_task_type;
 
-struct reduce_task : public boost::mapreduce::reduce_task<unsigned>
+struct reduce_task : public boost::mapreduce::reduce_task<std::string, unsigned>
 {
     template<typename Runtime, typename It>
     static void reduce(Runtime &runtime, std::string const &key, It it, It const ite)
@@ -167,6 +163,8 @@
 
 class combiner;
 
+typedef map_task<map_value_type> map_task_type;
+
 typedef
 boost::mapreduce::job<
     wordcount::map_task_type
@@ -197,19 +195,19 @@
         intermediate_store.combine(instance);
     }
 
-    void start(job::map_task_type::intermediate_key_type const &)
+    void start(job::reduce_task_type::key_type const &)
     {
         total_ = 0;
     }
 
     template<typename IntermediateStore>
-    void finish(job::map_task_type::intermediate_key_type const &key, IntermediateStore &intermediate_store)
+    void finish(job::reduce_task_type::key_type const &key, IntermediateStore &intermediate_store)
     {
         if (total_ > 0)
             intermediate_store.insert(key, total_);
     }
 
-    void operator()(job::map_task_type::intermediate_value_type const &value)
+    void operator()(job::reduce_task_type::value_type const &value)
     {
         total_ += value;
     }
@@ -256,7 +254,7 @@
         spec.map_tasks = 1;
         spec.reduce_tasks = 1;
 
-        wordcount::job      job(datasource, spec);
+        wordcount::job job(datasource, spec);
         job.run<boost::mapreduce::schedule_policy::sequential<wordcount::job> >(result);
         std::cout << "\nFinished.";
 #else
@@ -266,7 +264,12 @@
         if (argc > 2)
             spec.map_tasks = atoi(argv[2]);
 
-        boost::mapreduce::run<wordcount::job>(spec, result);
+        // this method can be called, but since we want access to the result data,
+        // we need to have a job object to interrogate
+        //boost::mapreduce::run<wordcount::job>(spec, result);
+
+        wordcount::job job(datasource, spec);
+        job.run<boost::mapreduce::schedule_policy::cpu_parallel<wordcount::job> >(result);
         std::cout << "\nCPU Parallel MapReduce Finished.";
 #endif
     }
@@ -275,6 +278,8 @@
         std::cout << std::endl << "Error: " << e.what();
     }
 
+    typedef std::pair<wordcount::reduce_task::key_type, wordcount::reduce_task::value_type> keyvalue_t;
+
     std::cout << std::endl << "\n" << "MapReduce statistics:";
     std::cout << "\n  " << "MapReduce job runtime                     : " << result.job_runtime << " seconds, of which...";
     std::cout << "\n  " << "  Map phase runtime                       : " << result.map_runtime << " seconds";