MPISync.hpp

Go to the documentation of this file.

/*
 *  This file is part of AQUAgpusph, a free CFD program based on SPH.
 *  Copyright (C) 2012  Jose Luis Cercos Pita <jl.cercos@upm.es>
 *
 *  AQUAgpusph is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  AQUAgpusph is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with AQUAgpusph.  If not, see <http://www.gnu.org/licenses/>.
 */

 
#ifndef MPISYNC_H_INCLUDED
#define MPISYNC_H_INCLUDED
 
#ifndef HAVE_MPI
#error MPI not available
#endif
 
#include <mpi.h>
 
#include "CalcServer.hpp"
#include "Kernel.hpp"
#include "RadixSort.hpp"
#include "Reduction.hpp"
#include "UnSort.hpp"
#include "SetScalar.hpp"
#include "Set.hpp"
 
namespace Aqua {
namespace CalcServer {

class MPISync : public Aqua::CalcServer::Tool
{
  public:
        MPISync(const std::string name,
                const std::string mask,
                const std::vector<std::string> fields,
                const std::vector<unsigned int> procs,
                bool once = false);

        ~MPISync();

        void setup();
 
  protected:
        cl_event _execute(const std::vector<cl_event> events);
 
  private:
        void variables();

        void setupSort();

        void setupFieldSort(InputOutput::ArrayVariable* field);

        void setupSenders();

        void setupReceivers();

        std::string _mask_name;
        InputOutput::ArrayVariable* _mask;

        std::vector<std::string> _field_names;
        std::vector<InputOutput::ArrayVariable*> _fields;
        std::vector<InputOutput::ArrayVariable*> _fields_sorted;

        std::vector<uicl> _procs;

        InputOutput::ArrayVariable* _unsorted_id;

        InputOutput::ArrayVariable* _sorted_id;

        RadixSort* _sort;

        std::vector<UnSort*> _field_sorters;

        size_t _n;
 
  public:
        class Exchanger
        {
          public:
                Exchanger(const std::string name,
                          const std::string vars_prefix,
                          InputOutput::ArrayVariable* mask,
                          const std::vector<InputOutput::ArrayVariable*> fields,
                          const std::vector<void*> field_hosts,
                          const uicl proc);

                ~Exchanger() {}

                inline const std::string name() const { return _name; }

                inline uicl proc() const { return _proc; }

                typedef struct
                {
                        size_t n;
                        MPI_Datatype t;
                } MPIType;

                static const MPIType typeToMPI(std::string t);
 
          protected:
                std::string _var_prefix;
                
                InputOutput::ArrayVariable* _mask;

                std::vector<InputOutput::ArrayVariable*> _fields;

                uicl _proc;

                size_t _n;

                std::vector<void*> _fields_host;
 
          private:
                std::string _name;
        };

        class Sender : public Exchanger
        {
          public:
                Sender(const std::string name,
                       const std::string vars_prefix,
                       InputOutput::ArrayVariable* mask,
                       const std::vector<InputOutput::ArrayVariable*> fields,
                       const std::vector<void*> field_hosts,
                       const uicl proc);

                ~Sender();

                void execute(EventProfile* profiler);
 
          private:
                void setupSubMaskMems();

                void setupOpenCL(const std::string kernel_name);

                void setupReduction(const std::string var_name);

                InputOutput::Variable* _n_offset;

                InputOutput::ArrayVariable* _n_offset_mask;

                cl_kernel _n_offset_kernel;

                Reduction* _n_offset_reduction;

                InputOutput::Variable* _n_send;

                InputOutput::ArrayVariable* _n_send_mask;

                cl_kernel _n_send_kernel;

                Reduction* _n_send_reduction;

                size_t _global_work_size;
                size_t _local_work_size;
        };

        class Receiver : public Exchanger
        {
          public:
                Receiver(const std::string name,
                         const std::string vars_prefix,
                         InputOutput::ArrayVariable* mask,
                         const std::vector<InputOutput::ArrayVariable*> fields,
                         const std::vector<void*> field_hosts,
                         const uicl proc,
                         InputOutput::Variable* n_offset);

                ~Receiver();

                void execute(EventProfile* profiler);
 
          private:
                void setupOpenCL();

                cl_kernel _kernel;

                InputOutput::Variable* _n_offset;

                size_t _local_work_size;
        };
 
  private:
        std::vector<void*> _fields_send;

        std::vector<Sender*> _senders;

        Set* _mask_reinit;

        InputOutput::Variable* _n_offset_recv;

        SetScalar* _n_offset_recv_reinit;

        std::vector<void*> _fields_recv;

        std::vector<Receiver*> _receivers;
};
 
}
} // namespace
 
#endif // MPISYNC_H_INCLUDED