[Boost-commit] svn:boost r55510 - sandbox/libs/mapreduce/test

Subject: [Boost-commit] svn:boost r55510 - sandbox/libs/mapreduce/test
From: cdm.henderson_at_[hidden]
Date: 2009-08-10 15:28:10

Author: chenderson
Date: 2009-08-10 15:28:09 EDT (Mon, 10 Aug 2009)
New Revision: 55510
URL: http://svn.boost.org/trac/boost/changeset/55510

Log:
Multi-test addition
Text files modified: 
   sandbox/libs/mapreduce/test/mrtest.cpp |   271 +++++++++++++++++++++------------------ 
   1 files changed, 149 insertions(+), 122 deletions(-)

Modified: sandbox/libs/mapreduce/test/mrtest.cpp
==============================================================================
--- sandbox/libs/mapreduce/test/mrtest.cpp	(original)
+++ sandbox/libs/mapreduce/test/mrtest.cpp	2009-08-10 15:28:09 EDT (Mon, 10 Aug 2009)
@@ -12,9 +12,6 @@
 
 // configuration options
 #define WORD_COUNT_MEMORY_MAP_FILE
-#define USE_WORDCOUNT_COMBINER
-#define USE_IN_MEMORY_INTERMEDIATES
-//#define WRITE_OUTPUT_FILES
 
 #if defined(_DEBUG)
 #   if 0
@@ -161,33 +158,7 @@
     }
 }
 
-
-class combiner;
-
-typedef map_task<map_value_type> map_task_type;
-
-typedef
-boost::mapreduce::job<
-    wordcount::map_task_type
-  , wordcount::reduce_task
-#ifdef USE_WORDCOUNT_COMBINER
-  , wordcount::combiner
-#else
-  , boost::mapreduce::null_combiner
-#endif
-  , boost::mapreduce::datasource::directory_iterator<wordcount::map_task_type>
-#ifdef USE_IN_MEMORY_INTERMEDIATES
-  , boost::mapreduce::intermediates::in_memory<wordcount::map_task_type, wordcount::reduce_task>
-#else
-  , boost::mapreduce::intermediates::local_disk<wordcount::map_task_type, wordcount::reduce_task>
-#endif
-#if defined(USE_IN_MEMORY_INTERMEDIATES)  &&  defined(WRITE_OUTPUT_FILES)
-  , boost::mapreduce::intermediates::reduce_file_output<wordcount::map_task_type, wordcount::reduce_task>
-#endif
-> job;
-
-
-
+template<typename ReduceTask>
 class combiner
 {
   public:
@@ -198,19 +169,19 @@
         intermediate_store.combine(instance);
     }
 
-    void start(job::reduce_task_type::key_type const &)
+    void start(typename ReduceTask::key_type const &)
     {
         total_ = 0;
     }
 
     template<typename IntermediateStore>
-    void finish(job::reduce_task_type::key_type const &key, IntermediateStore &intermediate_store)
+    void finish(typename ReduceTask::key_type const &key, IntermediateStore &intermediate_store)
     {
         if (total_ > 0)
             intermediate_store.insert(key, total_);
     }
 
-    void operator()(job::reduce_task_type::value_type const &value)
+    void operator()(typename ReduceTask::value_type const &value)
     {
         total_ += value;
     }
@@ -222,9 +193,122 @@
     unsigned total_;
 };
 
+typedef map_task<map_value_type> map_task_type;
+
 }   // namespace wordcount
 
 
+template<typename Job>
+void run_test(boost::mapreduce::specification spec)
+{
+    boost::mapreduce::results result;
+
+    std::cout << "\n" << typeid(Job).name() << "\n";
+
+    try
+    {
+#ifdef RUN_SEQUENTIAL_MAP_REDUCE
+        std::cout << "\nRunning Sequential MapReduce...";
+
+        spec.map_tasks = 1;
+        spec.reduce_tasks = 1;
+
+        Job job(datasource, spec);
+        job.run<boost::mapreduce::schedule_policy::sequential<Job> >(result);
+        std::cout << "\nSequential MapReduce Finished.";
+#else
+        std::cout << "\nRunning CPU Parallel MapReduce...";
+
+        // 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<Job>(spec, result);
+
+        typename Job::datasource_type datasource(spec);
+        Job job(datasource, spec);
+        job.run<boost::mapreduce::schedule_policy::cpu_parallel<Job> >(result);
+        std::cout << "\nCPU Parallel MapReduce Finished.\n";
+#endif
+
+        std::cout << "\nMapReduce statistics:";
+        std::cout << "\n  MapReduce job runtime                     : " << result.job_runtime << " seconds, of which...";
+        std::cout << "\n    Map phase runtime                       : " << result.map_runtime << " seconds";
+        std::cout << "\n    Reduce phase runtime                    : " << result.reduce_runtime << " seconds";
+        std::cout << "\n\n  Map:";
+        std::cout << "\n    Total Map keys                          : " << result.counters.map_keys_executed;
+        std::cout << "\n    Map keys processed                      : " << result.counters.map_keys_completed;
+        std::cout << "\n    Map key processing errors               : " << result.counters.map_key_errors;
+        std::cout << "\n    Number of Map Tasks run (in parallel)   : " << result.counters.actual_map_tasks;
+        std::cout << "\n    Fastest Map key processed in            : " << *std::min_element(result.map_times.begin(), result.map_times.end()) << " seconds";
+        std::cout << "\n    Slowest Map key processed in            : " << *std::max_element(result.map_times.begin(), result.map_times.end()) << " seconds";
+        std::cout << "\n    Average time to process Map keys        : " << std::accumulate(result.map_times.begin(), result.map_times.end(), boost::posix_time::time_duration()) / result.map_times.size() << " seconds";
+
+        std::cout << "\n\n  Reduce:";
+        std::cout << "\n    Total Reduce keys                       : " << result.counters.reduce_keys_executed;
+        std::cout << "\n    Reduce keys processed                   : " << result.counters.reduce_keys_completed;
+        std::cout << "\n    Reduce key processing errors            : " << result.counters.reduce_key_errors;
+        std::cout << "\n    Number of Reduce Tasks run (in parallel): " << result.counters.actual_reduce_tasks;
+        std::cout << "\n    Number of Result Files                  : " << result.counters.num_result_files;
+        if (result.reduce_times.size() > 0)
+        {
+            std::cout << "\n    Fastest Reduce key processed in         : " << *std::min_element(result.reduce_times.begin(), result.reduce_times.end()) << " seconds";
+            std::cout << "\n    Slowest Reduce key processed in         : " << *std::max_element(result.reduce_times.begin(), result.reduce_times.end()) << " seconds";
+            std::cout << "\n    Average time to process Reduce keys     : " << std::accumulate(result.reduce_times.begin(), result.reduce_times.end(), boost::posix_time::time_duration()) / result.map_times.size() << " seconds";
+        }
+
+        typename Job::const_result_iterator it  = job.begin_results();
+        typename Job::const_result_iterator ite = job.end_results();
+        if (it != ite)
+        {
+            typedef std::list<typename Job::keyvalue_t> frequencies_t;
+            frequencies_t frequencies;
+            frequencies.push_back(*it);
+            frequencies_t::reverse_iterator it_smallest = frequencies.rbegin();
+            for (++it; it!=ite; ++it)
+            {
+                if (frequencies.size() < 10)    // show top 10
+                {
+                    frequencies.push_back(*it);
+                    if (it->second < it_smallest->second)
+                        it_smallest = frequencies.rbegin();
+                }
+                else if (it->second > it_smallest->second)
+                {
+                    *it_smallest = *it;
+                    it_smallest = std::min_element(frequencies.rbegin(), frequencies.rend(), boost::mapreduce::detail::less_2nd<typename Job::keyvalue_t>);
+                }
+            }
+
+            frequencies.sort(boost::mapreduce::detail::greater_2nd<typename Job::keyvalue_t>);
+            std::cout << "\n\nMapReduce results:";
+            for (frequencies_t::const_iterator freq=frequencies.begin(); freq!=frequencies.end(); ++freq)
+                std::cout << "\n" << freq->first << "\t" << freq->second;
+        }
+    }
+    catch (std::exception &e)
+    {
+        std::cout << std::endl << "Error running MapReduce: " << e.what();
+    }
+}
+
+//typedef
+//boost::mapreduce::job<
+//    wordcount::map_task_type
+//  , wordcount::reduce_task
+//#ifdef USE_WORDCOUNT_COMBINER
+//  , wordcount::combiner<wordcount::reduce_task>
+//#else
+//  , boost::mapreduce::null_combiner
+//#endif
+//  , boost::mapreduce::datasource::directory_iterator<wordcount::map_task_type>
+//#ifdef USE_IN_MEMORY_INTERMEDIATES
+//  , boost::mapreduce::intermediates::in_memory<wordcount::map_task_type, wordcount::reduce_task>
+//#else
+//  , boost::mapreduce::intermediates::local_disk<wordcount::map_task_type, wordcount::reduce_task>
+//#endif
+//#if defined(USE_IN_MEMORY_INTERMEDIATES)  &&  defined(WRITE_OUTPUT_FILES)
+//  , boost::mapreduce::intermediates::reduce_file_output<wordcount::map_task_type, wordcount::reduce_task>
+//#endif
+//> job;
 
 int main(int argc, char **argv)
 {
@@ -241,27 +325,8 @@
     }
 
     boost::mapreduce::specification spec;
-    boost::mapreduce::results       result;
 
     spec.input_directory = argv[1];
-    wordcount::job::datasource_type datasource(spec);
-
-    std::cout << "\n" << std::max(1,(int)boost::thread::hardware_concurrency()) << " CPU cores";
-    std::cout << "\n" << typeid(wordcount::job).name() << "\n";
-
-#ifdef RUN_SEQUENTIAL_MAP_REDUCE
-    std::cout << "\nRunning Sequential MapReduce...";
-
-    spec.map_tasks = 1;
-    spec.reduce_tasks = 1;
-
-    wordcount::job job(datasource, spec);
-    try
-    {
-        job.run<boost::mapreduce::schedule_policy::sequential<wordcount::job> >(result);
-        std::cout << "\nSequential MapReduce Finished.";
-#else
-    std::cout << "\nRunning CPU Parallel MapReduce...";
 
     if (argc > 2)
         spec.map_tasks = atoi(argv[2]);
@@ -271,76 +336,38 @@
     else
         spec.reduce_tasks = std::max(1U,boost::thread::hardware_concurrency());
 
-    // 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);
-    try
-    {
-        job.run<boost::mapreduce::schedule_policy::cpu_parallel<wordcount::job> >(result);
-        std::cout << "\nCPU Parallel MapReduce Finished.\n";
-#endif
-    }
-    catch (std::exception &e)
-    {
-        std::cout << std::endl << "Error running MapReduce: " << e.what();
-    }
-
-    std::cout << "\nMapReduce statistics:";
-    std::cout << "\n  MapReduce job runtime                     : " << result.job_runtime << " seconds, of which...";
-    std::cout << "\n    Map phase runtime                       : " << result.map_runtime << " seconds";
-    std::cout << "\n    Reduce phase runtime                    : " << result.reduce_runtime << " seconds";
-    std::cout << "\n\n  Map:";
-    std::cout << "\n    Total Map keys                          : " << result.counters.map_keys_executed;
-    std::cout << "\n    Map keys processed                      : " << result.counters.map_keys_completed;
-    std::cout << "\n    Map key processing errors               : " << result.counters.map_key_errors;
-    std::cout << "\n    Number of Map Tasks run (in parallel)   : " << result.counters.actual_map_tasks;
-    std::cout << "\n    Fastest Map key processed in            : " << *std::min_element(result.map_times.begin(), result.map_times.end()) << " seconds";
-    std::cout << "\n    Slowest Map key processed in            : " << *std::max_element(result.map_times.begin(), result.map_times.end()) << " seconds";
-    std::cout << "\n    Average time to process Map keys        : " << std::accumulate(result.map_times.begin(), result.map_times.end(), boost::posix_time::time_duration()) / result.map_times.size() << " seconds";
-
-    std::cout << "\n\n  Reduce:";
-    std::cout << "\n    Total Reduce keys                       : " << result.counters.reduce_keys_executed;
-    std::cout << "\n    Reduce keys processed                   : " << result.counters.reduce_keys_completed;
-    std::cout << "\n    Reduce key processing errors            : " << result.counters.reduce_key_errors;
-    std::cout << "\n    Number of Reduce Tasks run (in parallel): " << result.counters.actual_reduce_tasks;
-    std::cout << "\n    Number of Result Files                  : " << result.counters.num_result_files;
-    if (result.reduce_times.size() > 0)
-    {
-        std::cout << "\n    Fastest Reduce key processed in         : " << *std::min_element(result.reduce_times.begin(), result.reduce_times.end()) << " seconds";
-        std::cout << "\n    Slowest Reduce key processed in         : " << *std::max_element(result.reduce_times.begin(), result.reduce_times.end()) << " seconds";
-        std::cout << "\n    Average time to process Reduce keys     : " << std::accumulate(result.reduce_times.begin(), result.reduce_times.end(), boost::posix_time::time_duration()) / result.map_times.size() << " seconds";
-    }
-
-    wordcount::job::const_result_iterator it  = job.begin_results();
-    wordcount::job::const_result_iterator ite = job.end_results();
-    if (it != ite)
-    {
-        typedef std::list<wordcount::job::keyvalue_t> frequencies_t;
-        frequencies_t frequencies;
-        frequencies.push_back(*it);
-        frequencies_t::reverse_iterator it_smallest = frequencies.rbegin();
-        for (++it; it!=ite; ++it)
-        {
-            if (frequencies.size() < 10)    // show top 10
-            {
-                frequencies.push_back(*it);
-                if (it->second < it_smallest->second)
-                    it_smallest = frequencies.rbegin();
-            }
-            else if (it->second > it_smallest->second)
-            {
-                *it_smallest = *it;
-                it_smallest = std::min_element(frequencies.rbegin(), frequencies.rend(), boost::mapreduce::detail::less_2nd<wordcount::job::keyvalue_t>);
-            }
-        }
-
-        frequencies.sort(boost::mapreduce::detail::greater_2nd<wordcount::job::keyvalue_t>);
-        std::cout << "\n\nMapReduce results:";
-        for (frequencies_t::const_iterator freq=frequencies.begin(); freq!=frequencies.end(); ++freq)
-            std::cout << "\n" << freq->first << "\t" << freq->second;
-    }
+    std::cout << "\n" << std::max(1,(int)boost::thread::hardware_concurrency()) << " CPU cores";
+    run_test<
+        boost::mapreduce::job<
+            wordcount::map_task_type
+          , wordcount::reduce_task>
+        >(spec);
+
+    run_test<
+        boost::mapreduce::job<
+            wordcount::map_task_type
+          , wordcount::reduce_task
+          , wordcount::combiner<wordcount::reduce_task> >
+        >(spec);
+
+    run_test<
+        boost::mapreduce::job<
+            wordcount::map_task_type
+          , wordcount::reduce_task
+          , wordcount::combiner<wordcount::reduce_task>
+          , boost::mapreduce::datasource::directory_iterator<wordcount::map_task_type>
+          , boost::mapreduce::intermediates::in_memory<wordcount::map_task_type, wordcount::reduce_task>
+          , boost::mapreduce::intermediates::reduce_file_output<wordcount::map_task_type, wordcount::reduce_task> >
+        >(spec);
+
+    run_test<
+        boost::mapreduce::job<
+            wordcount::map_task_type
+          , wordcount::reduce_task
+          , wordcount::combiner<wordcount::reduce_task>
+          , boost::mapreduce::datasource::directory_iterator<wordcount::map_task_type>
+          , boost::mapreduce::intermediates::local_disk<wordcount::map_task_type, wordcount::reduce_task> >
+        >(spec);
 
     return 0;
 }