Forward propagation computes a parallel partial sum by dividing the input into chunks. Each chunk first computes its own partial sums; the accumulated total from the preceding chunk is then added to its results. section 8.5.3
Update a chunk’s last element first. One thread can then adjust the remaining elements while another starts updating the next chunk. This keeps cores useful when chunks contain many elements, but leaves processors waiting when there are as many cores as elements. section 8.5.3
Listing 8.11. Calculating partial sums in parallel by dividing the problem
template<typename Iterator>
void parallel_partial_sum(Iterator first,Iterator last)
{
typedef typename Iterator::value_type value_type;
struct process_chunk #1
{
void operator()(Iterator begin,Iterator last,
std::future<value_type>* previous_end_value,
std::promise<value_type>* end_value)
{
try
{
Iterator end=last;
++end;
std::partial_sum(begin,end,begin); #2
if(previous_end_value) #3
{
value_type& addend=previous_end_value->get(); #4
*last+=addend; #5
if(end_value)
{
end_value->set_value(*last); #6
}
std::for_each(begin,last,[addend](value_type& item) #7
{
item+=addend;
});
}
else if(end_value)
{
end_value->set_value(*last); #8
}
}
catch(...) #9
{
if(end_value)
{
end_value->set_exception(std::current_exception()); #10
}
else
{
throw; #11
}
}
}
};
unsigned long const length=std::distance(first,last);
if(!length)
return;
unsigned long const min_per_thread=25; #12
unsigned long const max_threads=
(length+min_per_thread-1)/min_per_thread;
unsigned long const hardware_threads=
std::thread::hardware_concurrency();
unsigned long const num_threads=
std::min(hardware_threads!=0?hardware_threads:2,max_threads);
unsigned long const block_size=length/num_threads;
typedef typename Iterator::value_type value_type;
std::vector<std::thread> threads(num_threads-1); #13
std::vector<std::promise<value_type> >
end_values(num_threads-1); #14
std::vector<std::future<value_type> >
previous_end_values; #15
previous_end_values.reserve(num_threads-1); #16
join_threads joiner(threads);
Iterator block_start=first;
for(unsigned long i=0;i<(num_threads-1);++i)
{
Iterator block_last=block_start;
std::advance(block_last,block_size-1); #17
threads[i]=std::thread(process_chunk(), #18
block_start,block_last,
(i!=0)?&previous_end_values[i-1]:0,
&end_values[i]);
block_start=block_last;
++block_start; #19
previous_end_values.push_back(end_values[i].get_future()); #20
}
Iterator final_element=block_start;
std::advance(final_element,std::distance(block_start,last)-1); #21
process_chunk()(block_start,final_element, #22
(num_threads>1)?&previous_end_values.back():0,
0);
}
!@%STYLE%@!
{"css":"{\"css\": \"font-weight: bold;\"}","target":"[[{\"line\":5,\"ch\":73},{\"line\":5,\"ch\":75}],[{\"line\":38,\"ch\":73},{\"line\":38,\"ch\":75}],[{\"line\":42,\"ch\":72},{\"line\":42,\"ch\":74}],[{\"line\":50,\"ch\":72},{\"line\":50,\"ch\":74}],[{\"line\":59,\"ch\":72},{\"line\":59,\"ch\":74}],[{\"line\":61,\"ch\":72},{\"line\":61,\"ch\":74}],[{\"line\":63,\"ch\":72},{\"line\":63,\"ch\":74}],[{\"line\":64,\"ch\":72},{\"line\":64,\"ch\":74}],[{\"line\":70,\"ch\":72},{\"line\":70,\"ch\":74}],[{\"line\":71,\"ch\":72},{\"line\":71,\"ch\":74}],[{\"line\":76,\"ch\":72},{\"line\":76,\"ch\":74}],[{\"line\":15,\"ch\":73},{\"line\":15,\"ch\":75}],[{\"line\":77,\"ch\":72},{\"line\":77,\"ch\":74}],[{\"line\":80,\"ch\":72},{\"line\":80,\"ch\":74}],[{\"line\":81,\"ch\":72},{\"line\":81,\"ch\":74}],[{\"line\":16,\"ch\":73},{\"line\":16,\"ch\":75}],[{\"line\":18,\"ch\":73},{\"line\":18,\"ch\":75}],[{\"line\":19,\"ch\":73},{\"line\":19,\"ch\":75}],[{\"line\":22,\"ch\":73},{\"line\":22,\"ch\":75}],[{\"line\":24,\"ch\":73},{\"line\":24,\"ch\":75}],[{\"line\":31,\"ch\":73},{\"line\":31,\"ch\":75}],[{\"line\":34,\"ch\":73},{\"line\":34,\"ch\":75}],[{\"line\":38,\"ch\":73},{\"line\":38,\"ch\":76}],[{\"line\":42,\"ch\":72},{\"line\":42,\"ch\":75}],[{\"line\":50,\"ch\":72},{\"line\":50,\"ch\":75}],[{\"line\":59,\"ch\":72},{\"line\":59,\"ch\":75}],[{\"line\":61,\"ch\":72},{\"line\":61,\"ch\":75}],[{\"line\":63,\"ch\":72},{\"line\":63,\"ch\":75}],[{\"line\":64,\"ch\":72},{\"line\":64,\"ch\":75}],[{\"line\":70,\"ch\":72},{\"line\":70,\"ch\":75}],[{\"line\":71,\"ch\":72},{\"line\":71,\"ch\":75}],[{\"line\":76,\"ch\":72},{\"line\":76,\"ch\":75}],[{\"line\":77,\"ch\":72},{\"line\":77,\"ch\":75}],[{\"line\":80,\"ch\":72},{\"line\":80,\"ch\":75}],[{\"line\":81,\"ch\":72},{\"line\":81,\"ch\":75}]]"}
!@%STYLE%@!
The C++ implementation uses std::partial_sum for each chunk and a future to obtain the preceding chunk’s accumulated value. section 8.5.3
When more parallelism is needed
For inputs 1 through 9, an alternative propagates sums over increasing distances: adjacent elements first, then distances of two, four, and so on. It reaches 1, 3, 6, 10, 15, 21, 28, 36, 45 while allowing more processors to update entries each round. section 8.5.3
