[Boost-commit] svn:boost r55327 - in sandbox: boost/mapreduce boost/mapreduce/intermediates libs/mapreduce/test

Subject: [Boost-commit] svn:boost r55327 - in sandbox: boost/mapreduce boost/mapreduce/intermediates libs/mapreduce/test
From: cdm.henderson_at_[hidden]
Date: 2009-07-31 17:36:01

Author: chenderson
Date: 2009-07-31 17:35:59 EDT (Fri, 31 Jul 2009)
New Revision: 55327
URL: http://svn.boost.org/trac/boost/changeset/55327

Log:
Added memory-based intermediate handling policy
Text files modified: 
   sandbox/boost/mapreduce/hash_partitioner.hpp         |     2 +                                       
   sandbox/boost/mapreduce/intermediates.hpp            |     7 +----                                   
   sandbox/boost/mapreduce/intermediates/local_disk.hpp |    46 ++++++++++++++++++++++++++++++++------- 
   sandbox/boost/mapreduce/job.hpp                      |    22 ++++++-------------                     
   sandbox/libs/mapreduce/test/mrtest.cpp               |    25 ++++++++++++---------                   
   5 files changed, 62 insertions(+), 40 deletions(-)

Modified: sandbox/boost/mapreduce/hash_partitioner.hpp
==============================================================================
--- sandbox/boost/mapreduce/hash_partitioner.hpp	(original)
+++ sandbox/boost/mapreduce/hash_partitioner.hpp	2009-07-31 17:35:59 EDT (Fri, 31 Jul 2009)
@@ -12,6 +12,8 @@
 
 #ifndef BOOST_MAPREDUCE_HASH_PARTITONER_HPP
 #define BOOST_MAPREDUCE_HASH_PARTITONER_HPP
+
+#include <boost/functional/hash.hpp>
  
 namespace boost {
 

Modified: sandbox/boost/mapreduce/intermediates.hpp
==============================================================================
--- sandbox/boost/mapreduce/intermediates.hpp	(original)
+++ sandbox/boost/mapreduce/intermediates.hpp	2009-07-31 17:35:59 EDT (Fri, 31 Jul 2009)
@@ -10,9 +10,6 @@
 // For more information, see http://www.boost.org/libs/mapreduce/
 //
  
-//!!!
-#ifdef USE_IN_MEMORY_INTERMEDIATES
-#include <boost/mapreduce/intermediates/in_memory.hpp>
-#endif  // USE_IN_MEMORY_INTERMEDIATES
-
+#include <boost/mapreduce/hash_partitioner.hpp>
 #include <boost/mapreduce/intermediates/local_disk.hpp>
+#include <boost/mapreduce/intermediates/in_memory.hpp>

Modified: sandbox/boost/mapreduce/intermediates/local_disk.hpp
==============================================================================
--- sandbox/boost/mapreduce/intermediates/local_disk.hpp	(original)
+++ sandbox/boost/mapreduce/intermediates/local_disk.hpp	2009-07-31 17:35:59 EDT (Fri, 31 Jul 2009)
@@ -13,10 +13,8 @@
 #ifndef BOOST_MAPREDUCE_LOCAL_DISK_INTERMEDIATES_HPP
 #define BOOST_MAPREDUCE_LOCAL_DISK_INTERMEDIATES_HPP
 
-#include <iomanip>                  // setw
-#include <fstream>		// linux
-#include <boost/unordered_map.hpp>
-#include <boost/mapreduce/hash_partitioner.hpp>
+#include <iomanip>      // setw
+#include <fstream>      // linux
 
 namespace boost {
 
@@ -93,8 +91,35 @@
 
 namespace intermediates {
 
+template<typename MapTask, typename ReduceTask>
+class reduce_file_output
+{
+  public:
+    reduce_file_output(std::string const &output_filespec,
+                       unsigned    const  partition,
+                       unsigned    const  num_partitions)
+    {
+        std::ostringstream filename;
+        filename << output_filespec << partition+1 << "_of_" << num_partitions;
+        filename_ = filename.str();
+        output_file_.open(filename_.c_str());
+    }
+
+    void operator()(typename MapTask::intermediate_key_type const &key,
+                    typename ReduceTask::value_type         const &value)
+    {
+        output_file_ << key << "\t" << value << "\n";
+    }
+
+  private:
+    std::string   filename_;
+    std::ofstream output_file_;
+};
+
+
 template<
     typename MapTask,
+    typename ReduceTask,
     typename SortFn=mapreduce::detail::file_sorter,
     typename MergeFn=mapreduce::detail::file_merger,
     typename PartitionFn=mapreduce::hash_partitioner>
@@ -102,11 +127,7 @@
 {
   private:
     typedef
-#ifdef _DEBUG
     std::map<
-#else
-    boost::unordered_map<
-#endif
         size_t,                                     // hash value of intermediate key (R)
         std::pair<
             std::string,                            // filename
@@ -115,6 +136,7 @@
 
   public:
     typedef MapTask map_task_type;
+    typedef reduce_file_output<MapTask, ReduceTask> store_result_type;
 
     local_disk(unsigned const num_partitions)
       : num_partitions_(num_partitions)
@@ -141,7 +163,13 @@
                       typename map_task_type::intermediate_value_type const &value)
     {
         unsigned const partition = partitioner_(key, num_partitions_);
-        intermediates_t::iterator it = intermediate_files_.insert(make_pair(partition, intermediates_t::mapped_type())).first;
+
+        intermediates_t::iterator it =
+            intermediate_files_.insert(
+                make_pair(
+                    partition,
+                    intermediates_t::mapped_type())).first;
+
         if (it->second.first.empty())
         {
             it->second.first = platform::get_temporary_filename();

Modified: sandbox/boost/mapreduce/job.hpp
==============================================================================
--- sandbox/boost/mapreduce/job.hpp	(original)
+++ sandbox/boost/mapreduce/job.hpp	2009-07-31 17:35:59 EDT (Fri, 31 Jul 2009)
@@ -44,7 +44,8 @@
          typename ReduceTask,
          typename Combiner=null_combiner,
          typename Datasource=datasource::directory_iterator<MapTask>,
-         typename IntermediateStore=intermediates::local_disk<MapTask> >
+         typename IntermediateStore=intermediates::in_memory<MapTask, ReduceTask>,
+         typename StoreResult=typename IntermediateStore::store_result_type>
 class job : private boost::noncopyable
 {
   public:
@@ -99,19 +100,16 @@
       public:
         reduce_task_runner(
             std::string   const &output_filespec,
-            size_t        const  partition,
-            size_t        const  num_partitions)
+            unsigned      const  partition,
+            unsigned      const  num_partitions)
+          : store_result_(output_filespec, partition, num_partitions)
         {
-            std::ostringstream filename;
-            filename << output_filespec << partition+1 << "_of_" << num_partitions;
-            filename_ = filename.str();
-            output_file_.open(filename_.c_str());
         }
 
         void emit(typename map_task_type::intermediate_key_type const &key,
                   typename reduce_task_type::value_type         const &value)
         {
-            output_file_ << key << "\t" << value << "\n";
+            store_result_(key, value);
         }
 
         template<typename It>
@@ -120,14 +118,8 @@
             reduce_task_type::reduce(*this, key, it, ite);
         }
 
-        std::string const &filename(void) const
-        {
-            return filename_;
-        }
-
       private:
-        std::string   filename_;
-        std::ofstream output_file_;
+        StoreResult store_result_;
     };
 
     job(datasource_type &datasource, specification const &spec)

Modified: sandbox/libs/mapreduce/test/mrtest.cpp
==============================================================================
--- sandbox/libs/mapreduce/test/mrtest.cpp	(original)
+++ sandbox/libs/mapreduce/test/mrtest.cpp	2009-07-31 17:35:59 EDT (Fri, 31 Jul 2009)
@@ -9,6 +9,17 @@
 //
 // For more information, see http://www.boost.org/libs/mapreduce/
 //
+
+// configuration options
+#define WORD_COUNT_MEMORY_MAP_FILE
+#define USE_WORDCOUNT_COMBINER
+#define USE_IN_MEMORY_INTERMEDIATES
+
+#if defined(_DEBUG)
+#   define RUN_SEQUENTIAL_MAP_REDUCE
+#else
+#   define BOOST_DISABLE_ASSERTS
+#endif
  
 #if !defined(_DEBUG) &&  !defined(BOOST_DISABLE_ASSERTS)
 #   pragma message("Warning: BOOST_DISABLE_ASSERTS not defined")
@@ -26,15 +37,6 @@
 #include <crtdbg.h>
 #endif
 
-// configuration options
-#define WORD_COUNT_MEMORY_MAP_FILE
-#define USE_WORDCOUNT_COMBINER
-//#define USE_IN_MEMORY_INTERMEDIATES
-
-#if defined(_DEBUG)
-#define RUN_SEQUENTIAL_MAP_REDUCE
-#endif
-
 namespace wordcount {
 
 typedef
@@ -172,9 +174,9 @@
 #else
   , boost::mapreduce::null_combiner
 #endif
-#ifdef USE_IN_MEMORY_INTERMEDIATES
+#ifndef USE_IN_MEMORY_INTERMEDIATES
   , boost::mapreduce::datasource::directory_iterator<wordcount::map_task_type>
-  , boost::mapreduce::intermediates::in_memory<wordcount::map_task_type>
+  , boost::mapreduce::intermediates::local_disk<wordcount::map_task_type, wordcount::reduce_task>
 #endif
 > job;
 
@@ -244,6 +246,7 @@
         std::cout << "\nFinished.";
 #else
         std::cout << "\nRunning CPU Parallel MapReduce...";
+        spec.reduce_tasks = 1;
 
         if (argc > 2)
             spec.map_tasks = atoi(argv[2]);