Line data    Source code

       1             : /* -*- mode: c++ -*- */
       2             : //# ContinuumPartitionMixin.h: Parallel continuum imaging data partitioning
       3             : //# Copyright (C) 2016
       4             : //# Associated Universities, Inc. Washington DC, USA.
       5             : //#
       6             : //# This library is free software; you can redistribute it and/or modify it
       7             : //# under the terms of the GNU Library General Public License as published by
       8             : //# the Free Software Foundation; either version 2 of the License, or (at your
       9             : //# option) any later version.
      10             : //#
      11             : //# This library is distributed in the hope that it will be useful, but WITHOUT
      12             : //# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
      13             : //# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Library General Public
      14             : //# License for more details.
      15             : //#
      16             : //# You should have received a copy of the GNU Library General Public License
      17             : //# along with this library; if not, write to the Free Software Foundation,
      18             : //# Inc., 675 Massachusetts Ave, Cambridge, MA 02139, USA.
      19             : //#
      20             : //# Correspondence concerning AIPS++ should be addressed as follows:
      21             : //#        Internet email: casa-feedback@nrao.edu.
      22             : //#        Postal address: AIPS++ Project Office
      23             : //#                        National Radio Astronomy Observatory
      24             : //#                        520 Edgemont Road
      25             : //#                        Charlottesville, VA 22903-2475 USA
      26             : //#
      27             : #ifndef CONTINUUM_PARTITION_MIXIN_H_
      28             : #define CONTINUUM_PARTITION_MIXIN_H_
      29             : 
      30             : #include <synthesis/ImagerObjects/SynthesisUtilMethods.h>
      31             : #include <synthesis/ImagerObjects/ParallelImagerParams.h>
      32             : #include <synthesis/ImagerObjects/MultiParamFieldIterator.h>
      33             : #include <synthesis/ImagerObjects/MPIGlue.h>
      34             : #include <algorithm>
      35             : #include <unistd.h>
      36             : #include <vector>
      37             : #include <string>
      38             : 
      39             : namespace casa {
      40             : 
      41             : /**
      42             :  * Parameter and input data partitioning for parallel continuum imaging (in
      43             :  * ParallelImagerMixin).
      44             :  */
      45             : template <class T>
      46             : class ContinuumPartitionMixin
      47             :         : public T {
      48             : 
      49             : public:
      50           0 :         void concat_images(const std::string &type __attribute__((unused))) {};
      51             : 
      52             : protected:
      53             :         MPI_Comm worker_comm;
      54             : 
      55             :         int num_workers;
      56             : 
      57             :         int worker_rank;
      58             : 
      59             :         ParallelImagerParams
      60           0 :         get_params(MPI_Comm wcomm, ParallelImagerParams &initial) {
      61             : 
      62             :                 // Determine rank among parallel imaging worker processes.
      63           0 :                 worker_comm = wcomm;
      64           0 :                 if (worker_comm != MPI_COMM_NULL) {
      65           0 :                         MPI_Comm_size(worker_comm, &num_workers);
      66           0 :                         MPI_Comm_rank(worker_comm, &worker_rank);
      67             :                 } else {
      68           0 :                         num_workers = 0;
      69           0 :                         worker_rank = -1;
      70             :                 }
      71             : 
      72           0 :                 std::string cwd(getcwd(nullptr, 0));
      73           0 :                 std::vector<std::string> all_worker_suffixes;
      74           0 :                 for (int r = 0; r < num_workers; ++r)
      75           0 :                         all_worker_suffixes.push_back(".n" + std::to_string(r));
      76           0 :                 std::string my_worker_suffix =
      77           0 :                         ((num_workers > 1 && worker_rank >= 0)
      78           0 :                          ? all_worker_suffixes[worker_rank]
      79             :                          : "");
      80           0 :                 SynthesisUtilMethods util;
      81           0 :                 ParallelImagerParams result;
      82             : 
      83             :                 // selection params
      84           0 :                 result.selection =
      85           0 :                         ((worker_rank >= 0)
      86           0 :                          ? util.continuumDataPartition(initial.selection, num_workers).
      87           0 :                          rwSubRecord(std::to_string(worker_rank))
      88             :                          : casacore::Record());
      89             : 
      90             :                 // image params
      91           0 :                 if (worker_rank >= 0) {
      92           0 :                         auto modify_imagename = [&](const char *field_val) {
      93           0 :                                 return cwd + "/" + field_val + my_worker_suffix;
      94             :                         };
      95           0 :                         result.image = convert_fields(initial.image, "imagename",
      96             :                                                       modify_imagename);
      97             :                 } else {
      98           0 :                         result.image = empty_fields(initial.image, "imagename");
      99             :                 }
     100             : 
     101             :                 // grid params
     102           0 :                 if (worker_rank >= 0) {
     103           0 :                         auto modify_cfcache = [&](const char *field_val) {
     104           0 :                                 return field_val + my_worker_suffix;
     105             :                         };
     106           0 :                         result.grid =
     107           0 :                                 convert_fields(initial.grid, "cfcache", modify_cfcache);
     108             :                 } else {
     109           0 :                         result.grid = empty_fields(initial.grid, "cfcache");
     110             :                 }
     111             : 
     112             :                 // normalization params
     113           0 :                 if (worker_rank == 0 && num_workers > 1) {
     114           0 :                         auto accumulate_part_names =
     115           0 :                                 [&] (std::array<casacore::Record *,2> im_norm_par) {
     116           0 :                                 std::vector<casacore::String> part_names;
     117           0 :                                 std::string image_path =
     118           0 :                                 cwd + "/" + im_norm_par[0]->asString("imagename").c_str();
     119           0 :                                 for (auto s : all_worker_suffixes) {
     120           0 :                                         part_names.push_back(casacore::String(image_path + s));
     121             :                                 }
     122           0 :                                 im_norm_par[1]->define("partimagenames", casacore::Vector<casacore::String>(part_names));
     123           0 :                         };
     124           0 :                         std::array<casacore::Record *,2> im_norm_params =
     125           0 :                                 { &initial.image, &initial.normalization };
     126           0 :                         std::for_each(MultiParamFieldIterator<2>::begin(im_norm_params),
     127             :                                       MultiParamFieldIterator<2>::end(im_norm_params),
     128             :                                       accumulate_part_names);
     129           0 :                         result.normalization = *(im_norm_params[1]);
     130           0 :                 } else {
     131           0 :                         result.normalization =
     132           0 :                                 empty_fields(initial.normalization, "partimagenames");
     133             :                 }
     134             : 
     135             :                 // deconvolution params
     136           0 :                 result.deconvolution =
     137           0 :                         ((worker_rank == 0) ? initial.deconvolution : casacore::Record());
     138             : 
     139             :                 // weight params
     140           0 :                 result.weight = ((worker_rank >= 0) ? initial.weight : casacore::Record());
     141             : 
     142             :                 // iteration params
     143           0 :                 result.iteration = initial.iteration;
     144             : 
     145           0 :                 return result;
     146           0 :         }
     147             : 
     148             : private:
     149             : 
     150             :         // Convenience method to transform certain record fields
     151           0 :         casacore::Record convert_fields(casacore::Record &rec, const char *field,
     152             :                               std::function<std::string(const char *)> fn) {
     153           0 :                 auto modify_field_val = [&](casacore::Record &msRec) {
     154           0 :                         msRec.define(field, fn(msRec.asString(field).c_str()));
     155             :                 };
     156           0 :                 casacore::Record result(rec);
     157           0 :                 std::for_each(ParamFieldIterator::begin(&result),
     158             :                               ParamFieldIterator::end(&result),
     159             :                               modify_field_val);
     160           0 :                 return result;
     161           0 :         }
     162             : 
     163             :         // Convenience method to clear certain record fields
     164           0 :         casacore::Record empty_fields(casacore::Record &rec, const char *field) {
     165           0 :                 auto modify_field_val = [&](casacore::Record &msRec) {
     166           0 :                         msRec.defineRecord(field, casacore::Record());
     167             :                 };
     168           0 :                 casacore::Record result(rec);
     169           0 :                 std::for_each(ParamFieldIterator::begin(&result),
     170             :                               ParamFieldIterator::end(&result),
     171             :                               modify_field_val);
     172           0 :                 return result;
     173           0 :         }
     174             : };
     175             : 
     176             : } // namespace casa
     177             : 
     178             : #endif // CONTINUUM_PARTITION_MIXIN_H_