MPISync.h

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.h>
#include <CalcServer/Kernel.h>
#include <CalcServer/RadixSort.h>
#include <CalcServer/Reduction.h>
#include <CalcServer/UnSort.h>
#include <CalcServer/SetScalar.h>
#include <CalcServer/Set.h>
 
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<unsigned int> _procs;
 
        InputOutput::ArrayVariable* _unsorted_id;
 
        InputOutput::ArrayVariable* _sorted_id;
 
        RadixSort* _sort;
 
        std::vector<UnSort*> _field_sorters;
 
        unsigned int _n;
 
  public:
        class Exchanger
        {
          public:
                Exchanger(const std::string name,
                          InputOutput::ArrayVariable* mask,
                          const std::vector<InputOutput::ArrayVariable*> fields,
                          const std::vector<void*> field_hosts,
                          const unsigned int proc);
 
                ~Exchanger();
 
                const std::string name() { return _name; }
 
                typedef struct
                {
                        unsigned int n;
                        MPI_Datatype t;
                } MPIType;
 
                static const MPIType typeToMPI(std::string t);
 
          protected:
                InputOutput::ArrayVariable* _mask;
 
                std::vector<InputOutput::ArrayVariable*> _fields;
 
                unsigned int _proc;
 
                unsigned int _n;
 
                std::vector<void*> _fields_host;
 
          private:
                std::string _name;
        };
 
        class Sender : public Exchanger
        {
          public:
                Sender(const std::string name,
                       InputOutput::ArrayVariable* mask,
                       const std::vector<InputOutput::ArrayVariable*> fields,
                       const std::vector<void*> field_hosts,
                       const unsigned int proc);
 
                ~Sender();
 
                void execute(void);
 
          private:
                void setupSubMaskMems();
 
                void setupOpenCL(const std::string kernel_name);
 
                void setupReduction(const std::string var_name);
 
                InputOutput::UIntVariable* _n_offset;
 
                InputOutput::ArrayVariable* _n_offset_mask;
 
                cl_kernel _n_offset_kernel;
 
                Reduction* _n_offset_reduction;
 
                InputOutput::UIntVariable* _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,
                         InputOutput::ArrayVariable* mask,
                         const std::vector<InputOutput::ArrayVariable*> fields,
                         const std::vector<void*> field_hosts,
                         const unsigned int proc,
                         InputOutput::UIntVariable* n_offset);
 
                ~Receiver();
 
                void execute(void);
 
          private:
                void setupOpenCL();
 
                cl_kernel _kernel;
 
                InputOutput::UIntVariable* _n_offset;
 
                size_t _local_work_size;
        };
 
  private:
        std::vector<void*> _fields_send;
 
        std::vector<Sender*> _senders;
 
        Set* _mask_reinit;
 
        InputOutput::UIntVariable* _n_offset_recv;
 
        SetScalar* _n_offset_recv_reinit;
 
        std::vector<void*> _fields_recv;
 
        std::vector<Receiver*> _receivers;
};
 
}
} // namespace
 
#endif // MPISYNC_H_INCLUDED