parallel_driver.h

Go to the documentation of this file.

/*
**  parallel_driver.h
**
**  This is part of Shiokaze, a research-oriented fluid solver for computer graphics.
**  Created by Ryoichi Ando <rand@nii.ac.jp> on Feb 1, 2018.
**
**  Permission is hereby granted, free of charge, to any person obtaining a copy of
**  this software and associated documentation files (the "Software"), to deal in
**  the Software without restriction, including without limitation the rights to use,
**  copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
**  Software, and to permit persons to whom the Software is furnished to do so,
**  subject to the following conditions:
**
**  The above copyright notice and this permission notice shall be included in all copies
**  or substantial portions of the Software.
**
**  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
**  INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
**  PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
**  HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
**  CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
**  OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
//
#ifndef SHKZ_PARALLEL_DRIVER_H
#define SHKZ_PARALLEL_DRIVER_H
//
#include <algorithm>
#include <vector>
#include <shiokaze/core/credit.h>
#include <shiokaze/core/configurable.h>
#include <shiokaze/math/shape.h>
#include "parallel_core.h"
#include "loop_splitter.h"
//
SHKZ_BEGIN_NAMESPACE
//
#define shkz_default_parallel_name  "stdthread"
#define shkz_default_splitter_name  "sequential_splitter"
//
class parallel_driver : public configurable, public credit {
public:
    //
    LONG_NAME("Parallel Driver")
    ARGUMENT_NAME("Parallel")
    parallel_driver ( recursive_configurable *parent, std::string parallel_name=shkz_default_parallel_name, std::string splitter_name=shkz_default_splitter_name ) : m_parallel_name(parallel_name), m_splitter_name(splitter_name) {
        if( parent ) parent->add_child(this);
        else setup_now();
    }
    parallel_driver ( std::string parallel_name=shkz_default_parallel_name, std::string splitter_name=shkz_default_splitter_name ) : parallel_driver(nullptr,parallel_name,splitter_name) {
        setup_now();
    }
    int get_thread_num() const {
        return m_maximal_threads;
    }
    void set_thread_num( int maximal_threads ) {
        m_maximal_threads = maximal_threads;
    }
    const parallel_core *get() const {
        return m_parallel_dispatcher.get();
    }
    void run( const std::vector<std::function<void()> > &functions ) {
        if( m_maximal_threads > 1 ) {
            m_parallel_dispatcher->run(functions);
        } else {
            for( auto f : functions ) f();
        }
    }
    void for_each_byte_safe( size_t size, std::function<void(size_t n, int thread_index)> func ) const {
        size_t size_div_8 = (size+7) / 8;
        for_each( size_div_8, [&]( size_t n_div_8, int thread_index ) {
            for( char i=0; i<8; ++i ) {
                size_t n = 8*n_div_8+i;
                if( n < size ) {
                    func(n,thread_index);
                }
            }
        });
    }
    void for_each( size_t size, std::function<void(size_t n, int thread_index)> func ) const {
        //
        if( size ) {
            int num_threads = m_maximal_threads;
            if( num_threads > size ) num_threads = size;
            if( num_threads > 1 ) {
                //
                assert(m_loop_splitter.get());
                //
                const void *context = m_loop_splitter->new_context(size,num_threads);
                auto start_func = m_loop_splitter->get_start_func(context);
                auto advance_func = m_loop_splitter->get_advance_func(context);
                //
                m_parallel_dispatcher->for_each(func,
                    [&](int q) { return start_func(context,q); },
                    [&](size_t &n, int q){ return advance_func(context,n,q); },
                num_threads);
                //
                m_loop_splitter->delete_context(context);
            } else {
                for( size_t n=0; n<size; ++n ) {
                    func(n,0);
                }
            }
        }
    }
    void for_each( size_t size, std::function<void(size_t n)> func ) const {
        for_each(size,[&]( size_t n, int thread_n ) {
            func(n);
        });
    }
    void for_each( const shape2 &shape, std::function<void(int i, int j, int thread_index)> func ) const {
        for_each(shape.count(),[&]( size_t n, int thread_index ) {
            vec2i pi = shape.decode(n);
            func(pi[0],pi[1],thread_index);
        });
    }
    void for_each( const shape2 &shape, std::function<void(int i, int j)> func ) const {
        for_each(shape,[&]( int i, int j, int thread_index ) {
            func(i,j);
        });
    }
    void for_each( const shape3 &shape, std::function<void(int i, int j, int k, int thread_index)> func ) const {
        for_each(shape.count(),[&]( size_t n, int thread_index ) {
            vec3i pi = shape.decode(n);
            func(pi[0],pi[1],pi[2],thread_index);
        });
    }
    void for_each( const shape3 &shape, std::function<void(int i, int j, int k)> func ) const {
        for_each(shape,[&]( int i, int j, int k, int thread_index ) {
            func(i,j,k);
        });
    }
    //
private:
    //
    std::string m_parallel_name, m_splitter_name;
    parallel_ptr m_parallel_dispatcher;
    loop_splitter_ptr m_loop_splitter;
    int m_maximal_threads {NUM_THREAD};
    //
    virtual void load( configuration &config ) override {
        configuration::auto_group group(config,*this);
        m_parallel_dispatcher = parallel_core::quick_load_module(config,m_parallel_name);
        m_loop_splitter = loop_splitter::quick_load_module(config,m_splitter_name);
    }
    //
    virtual void configure( configuration &config ) override {
        configuration::auto_group group(config,*this);
        config.get_integer("Threads",m_maximal_threads,"Number of maximal threads");
        m_parallel_dispatcher->recursive_configure(config);
        m_loop_splitter->recursive_configure(config);
    }
};
//
SHKZ_END_NAMESPACE
//
#endif
//