Re: [Boost-users] [mapreduce] Prim Calculator

Subject: Re: [Boost-users] [mapreduce] Prim Calculator
From: Christian Henning (chhenning_at_[hidden])
Date: 2009-08-21 16:32:19

Where in the sandbox is mapreduce? Thanks for all your comments! I'll
get back to you once I can run the prime calculator.

On Fri, Aug 21, 2009 at 3:52 PM, Craig
Henderson<cdm.henderson_at_[hidden]> wrote:
>> -----Original Message-----
>> From: boost-users-bounces_at_[hidden] [mailto:boost-users-
>> bounces_at_[hidden]] On Behalf Of Christian Henning
>> Sent: 21 August 2009 15:05
>> To: boost-users_at_[hidden]
>> Subject: Re: [Boost-users] [mapreduce] Prim Calculator
>>
>> Hi Craig, what you're suggesting is exactly what I do. For reasons I
>> don't understand I'm getting the following assertion:
>>
>> Running Parallel Prime_Calculator MapReduce...Assertion failed:
>> map_key != typename map_task_type::key_type(), file
>> c:\chh\boost\boost\mapreduce\job.hpp, line 210
>>
>> I don't understand why that happens? Can you help me out here.
>>
>
> Hi Christian
> The assertion is an incorrect error condition which meant that the key could
> not be a default constructed valued (in the case of your numeric key, zero).
> I have updated the sandbox - thanks for finding this. Get the latest version
> of job.hpp & in_memory.hpp from the sandbox.
>
> There are a couple other problems in your implementation, too.
> * number_source constructor should initialize _current to be start, not
> zero.
> * using a global to hold prime_numbers isn't thread-safe, and adding locking
> will introduce contention. The MapReduce library provides mechanisms to
> avoid this. In reduce_task::reduce, replace the line
>    copy( it, ite, back_inserter( prime_numbers ));
> with
>    for_each(it, ite, boost::bind(&Runtime::emit, &runtime, _1, 0));
> which will emit the final results.
>
> To display the results:
>    for (prime_calculator::job::const_result_iterator
> it=job.begin_results(); it!=job.end_results(); ++it)
>        std::cout << it->first << " ";
>
> For performance, the datasource should return a range of integers so each
> task tests many prime numbers rather than one at a time. This will reduce
> lock contention in retrieving map keys.
>
> Finally, 0 & 1 are not prime numbers - this is a bug in your is_prime
> function.
>
> Regards
> -- Craig
>
> #include <algorithm>
> #include <cmath>
> #include <numeric>
>
> #include <boost/mapreduce.hpp>
>
> namespace prime_calculator
> {
>
> std::size_t is_prime( std::size_t number ) {
>    long n = static_cast<long>( number );
>
>    if( number == 0 )
>        return 1;
>
>    n = std::abs( n );
>    std::size_t sqrt_number = static_cast< std::size_t >( std::sqrt(
> static_cast< double >( n )));
>
>    for( std::size_t i = 2; i <= sqrt_number; i++ )
>    {
>        if( n % i == 0 )
>            return 0;
>    }
>
>    return 1;
> }
>
> template< typename MapTask >
> class number_source : boost::noncopyable {
> public:
>
>    number_source( std::size_t start
>                 , std::size_t end
>                 )
>    : _start( start )
>    , _end  ( end   )
>    , _current( start )     // CH
>    {}
>
>    const bool setup_key( typename MapTask::key_type& key ) const
>    {
>        if( _current < _end )
>        {
>            key = _current;
>            return true;
>        }
>        else
>        {
>            return false;
>        }
>    }
>
>    const bool get_data( typename MapTask::key_type&   key
>                       , typename MapTask::value_type& value
>                       )
>    {
>        if( _current < _end )
>        {
>            value.first  = _current;
>            value.second = _current;
>            _current++;
>
>            return true;
>        }
>        else
>        {
>            return false;
>        }
>    }
>
> private:
>
>    std::size_t _start;
>    std::size_t _end;
>    std::size_t _current;
> };
>
> struct map_task : public boost::mapreduce::map_task< std::size_t
>     // MapKey
>                                                   , std::pair< std::size_t
>                                                              , std::size_t
>                                                              >
>     // MapValue
>                                                   > {
>    template<typename Runtime>
>    static void map( Runtime& runtime
>                   , const std::size_t& /*key*/
>                   , value_type&        value
>                   )
>    {
>        runtime.emit_intermediate( is_prime( value.first ), value.first );
>    }
> };
>
> //std::vector< std::size_t > prime_numbers;     // CH
>
> struct reduce_task : public boost::mapreduce::reduce_task< std::size_t
>                                                         , unsigned
>                                                         > {
>    template< typename Runtime
>            , typename It
>            >
>    static void reduce( Runtime&           runtime
>                      , const std::size_t& key
>                      , It                 it
>                      , const It           ite
>                      )
>    {
>        if( key > 0 )
>        {
> //            copy( it, ite, back_inserter( prime_numbers ));       // CH
>            for_each(it, ite, boost::bind(&Runtime::emit, &runtime, _1, 0));
> // CH
>        }
>    }
> };
>
> typedef
> boost::mapreduce::job< prime_calculator::map_task
>                     , prime_calculator::reduce_task
>                     , boost::mapreduce::null_combiner
>                     , prime_calculator::number_source<
> prime_calculator::map_task >
>> job;
>
> } // namespace prime_calculator
>
> int main(int argc, char* argv[])
> {
>    boost::mapreduce::specification spec;
>
>    boost::mapreduce::results result;
>    prime_calculator::job::datasource_type datasource( 0, 1000 );
>
>    spec.map_tasks    = 0;
>    spec.reduce_tasks = std::max( 1U, boost::thread::hardware_concurrency()
> );
>
>    std::cout << "\nRunning Parallel Prime_Calculator MapReduce...";
>    prime_calculator::job job( datasource, spec );
>    job.run<
> boost::mapreduce::schedule_policy::cpu_parallel<prime_calculator::job> >(
> result );
>    std::cout << "\nMapReduce Finished.";
>
>    for (prime_calculator::job::const_result_iterator
> it=job.begin_results(); it!=job.end_results(); ++it)
>        std::cout << it->first << " ";
>
>        return 0;
> }
>
>
> _______________________________________________
> Boost-users mailing list
> Boost-users_at_[hidden]
> http://lists.boost.org/mailman/listinfo.cgi/boost-users
>