mapreduce.h

Go to the documentation of this file.

#ifndef MAPREDUCE_H
#define MAPREDUCE_H



#include <starpu.h>
#ifdef SKEPU_OPENCL 
    #include <starpu_opencl.h>
#endif  



#ifdef SKEPU_OPENCL
#include <string>
#include <vector>
#ifdef USE_MAC_OPENCL
#include <OpenCL/opencl.h>
#else
#include <CL/cl.h>
#endif
#include "src/device_cl.h"
#endif

#include "src/environment.h"
#include "skepu/task.h"
#include "skepu/vector.h"
#include "src/operator_macros.h"
#include "src/exec_plan.h"

namespace skepu
{

template <typename MapFunc, typename ReduceFunc, typename T, skepu::FuncType c_type=UNARY>
class MapReduce: public Task
{

public:

    MapReduce(MapFunc* mapFunc, ReduceFunc* reduceFunc);
    
    MapReduce(MapFunc* mapFunc, ReduceFunc* reduceFunc, Vector<T> *in, T *out);

    ~MapReduce();
    
    void *get_result()
    {
        if(isScalarStarPU)
        {
            starpu_data_unregister(scalar_result_handle); // unregister as this data can be modified
            isScalarStarPU = false;
        }

        return output;
    }
    

    void setInput(Vector<T> *in)
    {
        input = in;
    }
    
    void setOutput(T *out)
    {
        if(isScalarStarPU)
            starpu_data_acquire(scalar_result_handle, STARPU_W);
            
        output = out;
        
        if(isScalarStarPU)
            starpu_data_release(scalar_result_handle);
    } 
    
    T getOutput()
    {
        if(isScalarStarPU)
        {
            T _output;
            starpu_data_acquire(scalar_result_handle, STARPU_R); // acquire as we know that it is just a read access…
            _output = *output;
            starpu_data_release(scalar_result_handle); // release acquire before returning control…
            return _output;
        }
        return *output;
    }

    void run_async();

    void setConstant(T constant1) {m_mapFunc->setConstant(constant1);}

    void finishAll() {m_environment->finishAll();}
    void setExecPlan(ExecPlan& plan) {m_execPlan = plan;}

public:
    Environment<int>* m_environment;
    MapFunc* m_mapFunc;
    ReduceFunc* m_reduceFunc;
    ExecPlan m_execPlan;
    
#ifdef USE_STARPU_HISTORY_MODEL    
    starpu_perfmodel *mapred1_perf_model;
#endif
    
    
    struct starpu_codelet *codelet;
    
    starpu_data_handle_t  scalar_result_handle;
    
    bool isScalarStarPU;

public:
    T operator()(Vector<T>& input);
    
    T operator()(Vector<T>& input, int parts);
    
public:
    static void cpu_func(void *buffers[], void *arg);

#ifdef SKEPU_OPENMP
public:
    static void omp_func(void *buffers[], void *arg);
#endif

#ifdef SKEPU_CUDA
public:
    static void cuda_func(void *buffers[], void *arg);
#endif

#ifdef SKEPU_OPENCL
public:
    static void opencl_func(void *buffers[], void *arg);
    
public:
    char kernelNameMap[1024];
    char kernelNameRed[1024];
    
    struct starpu_opencl_program opencl_codelet;

    void replaceText(std::string& text, std::string find, std::string replace);
    void createOpenCLProgram();
#endif

private:
    std::string perfmodel_str;
    
    Vector<T> *input;
    T *output;
    Vector<T> *temp_output;
};





template <typename MapFunc, typename ReduceFunc, typename T>
class MapReduce<MapFunc, ReduceFunc, T, skepu::BINARY> : public Task
{

public:

    MapReduce(MapFunc* mapFunc, ReduceFunc* reduceFunc);
    
    MapReduce(MapFunc* mapFunc, ReduceFunc* reduceFunc, Vector<T> *in1, Vector<T> *in2, T *out);

    ~MapReduce();
    
    void *get_result()
    {
        if(isScalarStarPU)
        {
            starpu_data_unregister(scalar_result_handle); // unregister as this data can be modified
            isScalarStarPU = false;
        }

        return output;
    }
    

    void setInput1(Vector<T> *in)
    {
        input1 = in;
    }
    
    void setInput2(Vector<T> *in)
    {
        input2 = in;
    }
    
    
    void setOutput(T *out)
    {
        if(isScalarStarPU)
            starpu_data_acquire(scalar_result_handle, STARPU_W);
            
        output = out;
        
        if(isScalarStarPU)
            starpu_data_release(scalar_result_handle);
    } 
    
    T getOutput()
    {
        if(isScalarStarPU)
        {
            T _output;
            starpu_data_acquire(scalar_result_handle, STARPU_R); // acquire as we know that it is just a read access…
            _output = *output;
            starpu_data_release(scalar_result_handle); // release acquire before returning control…
            return _output;
        }
        return *output;
    }
    
    void run_async();

    void setConstant(T constant1) {m_mapFunc->setConstant(constant1);}

    void finishAll() {m_environment->finishAll();}
    void setExecPlan(ExecPlan& plan) {m_execPlan = plan;}

public:
    Environment<int>* m_environment;
    MapFunc* m_mapFunc;
    ReduceFunc* m_reduceFunc;
    ExecPlan m_execPlan;
    
#ifdef USE_STARPU_HISTORY_MODEL    
    starpu_perfmodel *mapred2_perf_model;
#endif    
    struct starpu_codelet *codelet;
    
    starpu_data_handle_t  scalar_result_handle;
    
    bool isScalarStarPU;

public:
    T operator()(Vector<T>& input1, Vector<T>& input2);
    
    T operator()(Vector<T>& input1, Vector<T>& input2, int parts);
    
public:
    static void cpu_func(void *buffers[], void *arg);

#ifdef SKEPU_OPENMP
public:
    static void omp_func(void *buffers[], void *arg);
#endif

#ifdef SKEPU_CUDA
public:
    static void cuda_func(void *buffers[], void *arg);
#endif

#ifdef SKEPU_OPENCL
public:
    static void opencl_func(void *buffers[], void *arg);
    
public:
    char kernelNameMap[1024];
    char kernelNameRed[1024];
    
    struct starpu_opencl_program opencl_codelet;

    void replaceText(std::string& text, std::string find, std::string replace);
    void createOpenCLProgram();
#endif

private:
    std::string perfmodel_str;
    
    Vector<T> *input1;
    Vector<T> *input2;
    T *output;
    Vector<T> *temp_output;
};



template <typename MapFunc, typename ReduceFunc, typename T>
class MapReduce<MapFunc, ReduceFunc, T, TERNARY>: public Task
{

public:

    MapReduce(MapFunc* mapFunc, ReduceFunc* reduceFunc);
    
    MapReduce(MapFunc* mapFunc, ReduceFunc* reduceFunc, Vector<T> *in1, Vector<T> *in2, Vector<T> *in3, T *out);

    ~MapReduce();
    
    void *get_result()
    {
        if(isScalarStarPU)
        {
            starpu_data_unregister(scalar_result_handle); // unregister as this data can be modified
            isScalarStarPU = false;
        }

        return output;
    }
    

    void setInput1(Vector<T> *in)
    {
        input1 = in;
    }
    
    void setInput2(Vector<T> *in)
    {
        input2 = in;
    }
    
    void setInput3(Vector<T> *in)
    {
        input3 = in;
    }
    
    void setOutput(T *out)
    {
        if(isScalarStarPU)
            starpu_data_acquire(scalar_result_handle, STARPU_W);
            
        output = out;
        
        if(isScalarStarPU)
            starpu_data_release(scalar_result_handle);
    } 
    
    T getOutput()
    {
        if(isScalarStarPU)
        {
            T _output;
            starpu_data_acquire(scalar_result_handle, STARPU_R); // acquire as we know that it is just a read access…
            _output = *output;
            starpu_data_release(scalar_result_handle); // release acquire before returning control…
            return _output;
        }
        return *output;
    }
    
    void run_async();

    void setConstant(T constant1) {m_mapFunc->setConstant(constant1);}

    void finishAll() {m_environment->finishAll();}
    void setExecPlan(ExecPlan& plan) {m_execPlan = plan;}

public:
    Environment<int>* m_environment;
    MapFunc* m_mapFunc;
    ReduceFunc* m_reduceFunc;
    ExecPlan m_execPlan;   

#ifdef USE_STARPU_HISTORY_MODEL    
    starpu_perfmodel *mapred3_perf_model;
#endif
    
    struct starpu_codelet *codelet;
    
    starpu_data_handle_t  scalar_result_handle;
    
    bool isScalarStarPU;

public:
    T operator()(Vector<T>& input1, Vector<T>& input2, Vector<T>& input3);
    
    T operator()(Vector<T>& input1, Vector<T>& input2, Vector<T>& input3, int parts);
    
public:
    static void cpu_func(void *buffers[], void *arg);

#ifdef SKEPU_OPENMP
public:
    static void omp_func(void *buffers[], void *arg);
#endif

#ifdef SKEPU_CUDA
public:
    static void cuda_func(void *buffers[], void *arg);
#endif

#ifdef SKEPU_OPENCL
public:
    static void opencl_func(void *buffers[], void *arg);
    
public:
    char kernelNameMap[1024];
    char kernelNameRed[1024];
    
    struct starpu_opencl_program opencl_codelet;
    
    void replaceText(std::string& text, std::string find, std::string replace);
    void createOpenCLProgram();
#endif

private:
    std::string perfmodel_str;
    
    Vector<T> *input1;
    Vector<T> *input2;
    Vector<T> *input3;
    T *output;
    Vector<T> *temp_output;
};

}

#include "src/mapreduce.inl"

#include "src/mapreduce_cpu.inl"

#ifdef SKEPU_OPENMP
#include "src/mapreduce_omp.inl"
#endif

#ifdef SKEPU_OPENCL
#include "src/mapreduce_cl.inl"
#endif

#ifdef SKEPU_CUDA
#include "src/mapreduce_cu.inl"
#endif

#endif