nco/mpncecat.c File Reference

#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#include "nco_getopt.h"
#include <netcdf.h>
#include "libnco.h"

Include dependency graph for mpncecat.c:

Go to the source code of this file.

Defines
#define	MAIN_PROGRAM_FILE
Functions
int	main (int argc, char **argv)

---

Define Documentation

#define MAIN_PROGRAM_FILE

Definition at line 61 of file mpncecat.c.

---

Function Documentation

int main (  int  argc, char **  argv )

Definition at line 65 of file mpncecat.c. References var_sct_tag::cnt, dmn_sct_tag::cnt, copyright_prn(), var_sct_tag::dim, var_sct_tag::dmn_id, var_sct_tag::end, dmn_sct_tag::end, EXIT_FAILURE, EXIT_SUCCESS, False, FILE, getopt_long, var_sct_tag::id, dmn_sct_tag::id, idx_all_wrk_ass, int_CEWI, dmn_sct_tag::is_crd_dmn, dmn_sct_tag::is_rec_dmn, var_sct_tag::is_rec_var, lst_prs_2D(), MPI_Comm, MPI_Info, msg_bfr_lng, msg_tag_tkn_wrt_rqs, msg_tag_tkn_wrt_rsp, msg_tag_wrk_rqs, msg_tag_wrk_rsp, option::name, var_sct_tag::nbr_dim, NC_FORMAT_64BIT, NC_FORMAT_CLASSIC, NC_FORMAT_NETCDF4, NC_GLOBAL, dmn_sct_tag::nc_id, NC_MAX_DIMS, NC_MPIIO, NC_NETCDF4, NC_NOERR, NC_NOFILL, NC_NOWRITE, NC_SHARE, NC_WRITE, nco_att_cpy(), nco_bool, nco_close(), nco_cmd_ln_sng(), nco_cnt_run(), nco_cnv_ccm_ccsm_cf_inq(), nco_create_mode_prs(), nco_dmn_dfn(), nco_dmn_dpl(), nco_dmn_fll(), nco_dmn_lmt_mrg(), nco_dmn_lst_ass_var(), nco_dmn_lst_free(), nco_dmn_xrf(), nco_enddef(), nco_err_exit(), nco_exit(), nco_exit_gracefully(), nco_fl_cmp_err_chk(), nco_fl_lst_att_cat(), nco_fl_lst_mk(), nco_fl_mk_lcl(), nco_fl_mv(), nco_fl_nm_prs(), nco_fl_out_cls(), nco_fl_out_open(), nco_fl_rm(), nco_free(), nco_hst_att_cat(), nco_inq(), nco_lbr_vrs_prn(), nco_lmt_evl(), nco_lmt_lst_free(), nco_lmt_prs(), nco_lst_comma2hash(), nco_lst_srt_nm_id(), nco_malloc(), nco_mpi_att_cat(), nco_nm_id_lst_free(), nco_open(), nco_openmp_ini(), nco_pck_map_nil, nco_pck_plc_nil, nco_put_var1(), nco_put_vara(), nco_realloc(), NCO_REC_DMN_UNDEFINED, nco_set_fill(), nco_sng_lst_free(), nco_spn_lck_brk, nco_spn_lck_us, nco_thr_att_cat(), nco_usg_prn(), nco_var_dfn(), nco_var_dpl(), nco_var_fll(), nco_var_get(), nco_var_lst_add_crd(), nco_var_lst_ass_crd_add(), nco_var_lst_dvd(), nco_var_lst_free(), nco_var_lst_mk(), nco_var_lst_xcl(), nco_var_mtd_refresh(), nco_var_srt_zero(), nco_var_val_cpy(), nco_xrf_dmn(), nco_xrf_var(), dmn_sct_tag::nm, no_argument, NULL_CEWI, omp_get_thread_num(), optarg, optind, prg_nm_get(), prg_prs(), required_argument, rnk_mgr, var_sct_tag::srt, dmn_sct_tag::srt, var_sct_tag::sz, dmn_sct_tag::sz, var_sct_tag::sz_rec, tkn_wrt_rqs_dny, tkn_wrt_rqs_ntv, tkn_wrt_rqs_xcp, True, var_sct_tag::type, var_sct_tag::val, option::val, dmn_sct_tag::val, ptr_unn::vp, wrk_id_bfr_lng, and dmn_sct_tag::xrf. { nco_bool EXCLUDE_INPUT_LIST=False; /* Option c */ nco_bool EXTRACT_ALL_COORDINATES=False; /* Option c */ nco_bool EXTRACT_ASSOCIATED_COORDINATES=True; /* Option C */ nco_bool FILE_RETRIEVED_FROM_REMOTE_LOCATION; nco_bool FL_LST_IN_APPEND=True; /* Option H */ nco_bool FL_LST_IN_FROM_STDIN=False; /* [flg] fl_lst_in comes from stdin */ nco_bool FORCE_APPEND=False; /* Option A */ nco_bool FORCE_OVERWRITE=False; /* Option O */ nco_bool FORTRAN_IDX_CNV=False; /* Option F */ nco_bool HISTORY_APPEND=True; /* Option h */ nco_bool CNV_CCM_CCSM_CF; nco_bool REMOVE_REMOTE_FILES_AFTER_PROCESSING=True; /* Option R */ char **fl_lst_abb=NULL; /* Option a */ char **fl_lst_in; char **var_lst_in=NULL_CEWI; char *cmd_ln; char *fl_in=NULL; char *fl_out=NULL; /* Option o */ char *fl_out_tmp=NULL; /* MPI CEWI */ char *fl_pth=NULL; /* Option p */ char *fl_pth_lcl=NULL; /* Option l */ char *lmt_arg[NC_MAX_DIMS]; char *opt_crr=NULL; /* [sng] String representation of current long-option name */ char *optarg_lcl=NULL; /* [sng] Local copy of system optarg */ char *time_bfr_srt; const char * const CVS_Id="$Id: mpncecat.c,v 1.35 2006/02/19 00:42:34 zender Exp $"; const char * const CVS_Revision="$Revision: 1.35 $"; const char * const opt_sht_lst="4ACcD:d:FHhl:n:Oo:p:rRSt:v:x-:"; dmn_sct *rec_dmn; dmn_sct **dim; dmn_sct **dmn_out; extern char *optarg; extern int optind; /* Using naked stdin/stdout/stderr in parallel region generates warning Copy appropriate filehandle to variable scoped shared in parallel clause */ FILE * const fp_stderr=stderr; /* [fl] stderr filehandle CEWI */ FILE * const fp_stdout=stdout; /* [fl] stdout filehandle CEWI */ int *in_id_arr; int abb_arg_nbr=0; int fl_idx; int fl_nbr=0; int fl_out_fmt=NC_FORMAT_CLASSIC; /* [enm] Output file format */ int fll_md_old; /* [enm] Old fill mode */ int idx; int in_id; int lmt_nbr=0; /* Option d. NB: lmt_nbr gets incremented */ int nbr_dmn_fl; int nbr_dmn_xtr; int nbr_var_fix; /* nbr_var_fix gets incremented */ int nbr_var_fl; int nbr_var_prc; /* nbr_var_prc gets incremented */ int nbr_xtr=0; /* nbr_xtr won't otherwise be set for -c with no -v */ int opt; int out_id; int rcd=NC_NOERR; /* [rcd] Return code */ int rec_dmn_id=NCO_REC_DMN_UNDEFINED; int thr_idx; /* [idx] Index of current thread */ int thr_nbr=int_CEWI; /* [nbr] Thread number Option t */ int var_lst_in_nbr=0; lmt_sct **lmt; long idx_rec_out=0L; /* idx_rec_out gets incremented */ nm_id_sct *dmn_lst; nm_id_sct *xtr_lst=NULL; /* xtr_lst may be alloc()'d from NULL with -c option */ time_t time_crr_time_t; var_sct **var; var_sct **var_fix; var_sct **var_fix_out; var_sct **var_out; var_sct **var_prc; var_sct **var_prc_out; #ifdef ENABLE_MPI /* Declare all MPI-specific variables here */ MPI_Comm mpi_cmm=MPI_COMM_WORLD; /* [prc] Communicator */ #if defined(ENABLE_NETCDF4) || defined(ENABLE_PNETCDF) MPI_Info mpi_nfo=MPI_INFO_NULL; /* [sct] File geometry hints */ #endif /* !ENABLE_NETCDF4 || !ENABLE_PNETCDF */ MPI_Status mpi_stt; /* [enm] Status check to decode msg_tag_typ */ bool TKN_WRT_FREE=True; /* [flg] Write-access to output file is available */ const double tkn_wrt_rqs_ntv=0.04; /* [s] Token request interval */ const int msg_bfr_lng=3; /* [nbr] Number of elements in msg_bfr */ const int wrk_id_bfr_lng=1; /* [nbr] Number of elements in wrk_id_bfr */ int fl_nm_lng; /* [nbr] Output file name length */ int msg_bfr[msg_bfr_lng]; /* [bfr] Buffer containing var, idx, tkn_wrt_rsp */ int msg_tag_typ; /* [enm] MPI message tag type */ int prc_rnk; /* [idx] Process rank */ int prc_nbr=0; /* [nbr] Number of MPI processes */ int tkn_wrt_rsp; /* [enm] Response to request for write token */ int var_wrt_nbr=0; /* [nbr] Variables written to output file until now */ int rnk_wrk; /* [idx] Worker rank */ int wrk_id_bfr[wrk_id_bfr_lng]; /* [bfr] Buffer for rnk_wrk */ #endif /* !ENABLE_MPI */ static struct option opt_lng[]= { /* Structure ordered by short option key if possible */ /* Long options with no argument, no short option counterpart */ /* Long options with argument, no short option counterpart */ {"fl_fmt",required_argument,0,0}, {"file_format",required_argument,0,0}, /* Long options with short counterparts */ {"4",no_argument,0,'4'}, {"64bit",no_argument,0,'4'}, {"netcdf4",no_argument,0,'4'}, {"append",no_argument,0,'A'}, {"coords",no_argument,0,'c'}, {"crd",no_argument,0,'c'}, {"no-coords",no_argument,0,'C'}, {"no-crd",no_argument,0,'C'}, {"debug",required_argument,0,'D'}, {"dbg_lvl",required_argument,0,'D'}, {"dimension",required_argument,0,'d'}, {"dmn",required_argument,0,'d'}, {"fortran",no_argument,0,'F'}, {"ftn",no_argument,0,'F'}, {"fl_lst_in",no_argument,0,'H'}, {"file_list",no_argument,0,'H'}, {"history",no_argument,0,'h'}, {"hst",no_argument,0,'h'}, {"local",required_argument,0,'l'}, {"lcl",required_argument,0,'l'}, {"nintap",required_argument,0,'n'}, {"overwrite",no_argument,0,'O'}, {"ovr",no_argument,0,'O'}, {"output",required_argument,0,'o'}, {"fl_out",required_argument,0,'o'}, {"path",required_argument,0,'p'}, {"retain",no_argument,0,'R'}, {"rtn",no_argument,0,'R'}, {"revision",no_argument,0,'r'}, {"suspend", no_argument,0,'S'}, {"thr_nbr",required_argument,0,'t'}, {"threads",required_argument,0,'t'}, {"omp_num_threads",required_argument,0,'t'}, {"variable",required_argument,0,'v'}, {"version",no_argument,0,'r'}, {"vrs",no_argument,0,'r'}, {"exclude",no_argument,0,'x'}, {"xcl",no_argument,0,'x'}, {"help",no_argument,0,'?'}, {0,0,0,0} }; /* end opt_lng */ int opt_idx=0; /* Index of current long option into opt_lng array */ #ifdef ENABLE_MPI /* MPI Initialization */ MPI_Init(&argc,&argv); MPI_Comm_size(mpi_cmm,&prc_nbr); MPI_Comm_rank(mpi_cmm,&prc_rnk); #endif /* !ENABLE_MPI */ /* Start clock and save command line */ cmd_ln=nco_cmd_ln_sng(argc,argv); time_crr_time_t=time((time_t *)NULL); time_bfr_srt=ctime(&time_crr_time_t); time_bfr_srt=time_bfr_srt; /* Avoid compiler warning until variable is used for something */ /* Get program name and set program enum (e.g., prg=ncra) */ prg_nm=prg_prs(argv[0],&prg); /* Parse command line arguments */ while(1){ /* getopt_long_only() allows one dash to prefix long options */ opt=getopt_long(argc,argv,opt_sht_lst,opt_lng,&opt_idx); /* NB: access to opt_crr is only valid when long_opt is detected */ if(opt == EOF) break; /* Parse positional arguments once getopt_long() returns EOF */ opt_crr=(char *)strdup(opt_lng[opt_idx].name); /* Process long options without short option counterparts */ if(opt == 0){ if(!strcmp(opt_crr,"fl_fmt") || !strcmp(opt_crr,"file_format")) rcd=nco_create_mode_prs(optarg,&fl_out_fmt); } /* opt != 0 */ /* Process short options */ switch(opt){ case 0: /* Long options have already been processed, return */ break; case '4': /* [flg] Catch-all to prescribe output storage format */ if(!strcmp(opt_crr,"64bit")) fl_out_fmt=NC_FORMAT_64BIT; else fl_out_fmt=NC_FORMAT_NETCDF4; break; case 'A': /* Toggle FORCE_APPEND */ FORCE_APPEND=!FORCE_APPEND; break; case 'C': /* Extract all coordinates associated with extracted variables? */ EXTRACT_ASSOCIATED_COORDINATES=False; break; case 'c': EXTRACT_ALL_COORDINATES=True; break; case 'D': /* Debugging level. Default is 0. */ dbg_lvl=(unsigned short)strtol(optarg,(char **)NULL,10); break; case 'd': /* Copy argument for later processing */ lmt_arg[lmt_nbr]=(char *)strdup(optarg); lmt_nbr++; break; case 'F': /* Toggle index convention. Default is 0-based arrays (C-style). */ FORTRAN_IDX_CNV=!FORTRAN_IDX_CNV; break; case 'H': /* Toggle writing input file list attribute */ FL_LST_IN_APPEND=!FL_LST_IN_APPEND; break; case 'h': /* Toggle appending to history global attribute */ HISTORY_APPEND=!HISTORY_APPEND; break; case 'l': /* Local path prefix for files retrieved from remote file system */ fl_pth_lcl=(char *)strdup(optarg); break; case 'n': /* NINTAP-style abbreviation of files to process */ fl_lst_abb=lst_prs_2D(optarg,",",&abb_arg_nbr); if(abb_arg_nbr < 1 || abb_arg_nbr > 5){ (void)fprintf(stdout,"%s: ERROR Incorrect abbreviation for file list\n",prg_nm); (void)nco_usg_prn(); nco_exit(EXIT_FAILURE); } /* end if */ break; case 'O': /* Toggle FORCE_OVERWRITE */ FORCE_OVERWRITE=!FORCE_OVERWRITE; break; case 'o': /* Name of output file */ fl_out=(char *)strdup(optarg); break; case 'p': /* Common file path */ fl_pth=(char *)strdup(optarg); break; case 'R': /* Toggle removal of remotely-retrieved-files. Default is True. */ REMOVE_REMOTE_FILES_AFTER_PROCESSING=!REMOVE_REMOTE_FILES_AFTER_PROCESSING; break; case 'r': /* Print CVS program information and copyright notice */ (void)copyright_prn(CVS_Id,CVS_Revision); (void)nco_lbr_vrs_prn(); nco_exit(EXIT_SUCCESS); break; #ifdef ENABLE_MPI case 'S': /* Suspend with signal handler to facilitate debugging */ if(signal(SIGUSR1,nco_cnt_run) == SIG_ERR) (void)fprintf(fp_stdout,"%s: ERROR Could not install suspend handler.\n",prg_nm); while(!nco_spn_lck_brk) usleep(nco_spn_lck_us); /* Spinlock. fxm: should probably insert a sched_yield */ break; #endif /* !ENABLE_MPI */ case 't': /* Thread number */ thr_nbr=(int)strtol(optarg,(char **)NULL,10); break; case 'v': /* Variables to extract/exclude */ /* Replace commas with hashes when within braces (convert back later) */ optarg_lcl=(char *)strdup(optarg); (void)nco_lst_comma2hash(optarg_lcl); var_lst_in=lst_prs_2D(optarg_lcl,",",&var_lst_in_nbr); optarg_lcl=(char *)nco_free(optarg_lcl); nbr_xtr=var_lst_in_nbr; break; case 'x': /* Exclude rather than extract variables specified with -v */ EXCLUDE_INPUT_LIST=True; break; case '?': /* Print proper usage */ (void)nco_usg_prn(); nco_exit(EXIT_SUCCESS); break; case '-': /* Long options are not allowed */ (void)fprintf(stderr,"%s: ERROR Long options are not available in this build. Use single letter options instead.\n",prg_nm_get()); nco_exit(EXIT_FAILURE); break; default: /* Print proper usage */ (void)nco_usg_prn(); nco_exit(EXIT_FAILURE); break; } /* end switch */ if(opt_crr != NULL) opt_crr=(char *)nco_free(opt_crr); } /* end while loop */ /* Process positional arguments and fill in filenames */ fl_lst_in=nco_fl_lst_mk(argv,argc,optind,&fl_nbr,&fl_out,&FL_LST_IN_FROM_STDIN); /* Make uniform list of user-specified dimension limits */ lmt=nco_lmt_prs(lmt_nbr,lmt_arg); /* Initialize thread information */ thr_nbr=nco_openmp_ini(thr_nbr); in_id_arr=(int *)nco_malloc(thr_nbr*sizeof(int)); /* Parse filename */ fl_in=nco_fl_nm_prs(fl_in,0,&fl_nbr,fl_lst_in,abb_arg_nbr,fl_lst_abb,fl_pth); /* Make sure file is on local system and is readable or die trying */ fl_in=nco_fl_mk_lcl(fl_in,fl_pth_lcl,&FILE_RETRIEVED_FROM_REMOTE_LOCATION); /* Open file for reading */ rcd=nco_open(fl_in,NC_NOWRITE,&in_id); /* Get number of variables, dimensions, and record dimension ID of input file */ (void)nco_inq(in_id,&nbr_dmn_fl,&nbr_var_fl,(int *)NULL,&rec_dmn_id); /* Form initial extraction list which may include extended regular expressions */ xtr_lst=nco_var_lst_mk(in_id,nbr_var_fl,var_lst_in,EXTRACT_ALL_COORDINATES,&nbr_xtr); /* Change included variables to excluded variables */ if(EXCLUDE_INPUT_LIST) xtr_lst=nco_var_lst_xcl(in_id,nbr_var_fl,xtr_lst,&nbr_xtr); /* Add all coordinate variables to extraction list */ if(EXTRACT_ALL_COORDINATES) xtr_lst=nco_var_lst_add_crd(in_id,nbr_dmn_fl,xtr_lst,&nbr_xtr); /* Make sure coordinates associated extracted variables are also on extraction list */ if(EXTRACT_ASSOCIATED_COORDINATES) xtr_lst=nco_var_lst_ass_crd_add(in_id,xtr_lst,&nbr_xtr); /* Sort extraction list by variable ID for fastest I/O */ if(nbr_xtr > 1) xtr_lst=nco_lst_srt_nm_id(xtr_lst,nbr_xtr,False); /* We now have final list of variables to extract. Phew. */ /* Find coordinate/dimension values associated with user-specified limits NB: nco_lmt_evl() with same nc_id contains OpenMP critical region */ for(idx=0;idx<lmt_nbr;idx++) (void)nco_lmt_evl(in_id,lmt[idx],0L,FORTRAN_IDX_CNV); /* Find dimensions associated with variables to be extracted */ dmn_lst=nco_dmn_lst_ass_var(in_id,xtr_lst,nbr_xtr,&nbr_dmn_xtr); /* Fill in dimension structure for all extracted dimensions */ dim=(dmn_sct **)nco_malloc(nbr_dmn_xtr*sizeof(dmn_sct *)); for(idx=0;idx<nbr_dmn_xtr;idx++) dim[idx]=nco_dmn_fll(in_id,dmn_lst[idx].id,dmn_lst[idx].nm); /* Dimension list no longer needed */ dmn_lst=nco_nm_id_lst_free(dmn_lst,nbr_dmn_xtr); /* Merge hyperslab limit information into dimension structures */ if(lmt_nbr > 0) (void)nco_dmn_lmt_mrg(dim,nbr_dmn_xtr,lmt,lmt_nbr); /* Duplicate input dimension structures for output dimension structures */ dmn_out=(dmn_sct **)nco_malloc(nbr_dmn_xtr*sizeof(dmn_sct *)); for(idx=0;idx<nbr_dmn_xtr;idx++){ dmn_out[idx]=nco_dmn_dpl(dim[idx]); (void)nco_dmn_xrf(dim[idx],dmn_out[idx]); } /* end loop over idx */ /* Is this an CCM/CCSM/CF-format history tape? */ CNV_CCM_CCSM_CF=nco_cnv_ccm_ccsm_cf_inq(in_id); /* Fill in variable structure list for all extracted variables */ var=(var_sct **)nco_malloc(nbr_xtr*sizeof(var_sct *)); var_out=(var_sct **)nco_malloc(nbr_xtr*sizeof(var_sct *)); for(idx=0;idx<nbr_xtr;idx++){ var[idx]=nco_var_fll(in_id,xtr_lst[idx].id,xtr_lst[idx].nm,dim,nbr_dmn_xtr); var_out[idx]=nco_var_dpl(var[idx]); (void)nco_xrf_var(var[idx],var_out[idx]); (void)nco_xrf_dmn(var_out[idx]); } /* end loop over idx */ /* Extraction list no longer needed */ xtr_lst=nco_nm_id_lst_free(xtr_lst,nbr_xtr); /* Divide variable lists into lists of fixed variables and variables to be processed */ (void)nco_var_lst_dvd(var,var_out,nbr_xtr,CNV_CCM_CCSM_CF,nco_pck_plc_nil,nco_pck_map_nil,(dmn_sct **)NULL,0,&var_fix,&var_fix_out,&nbr_var_fix,&var_prc,&var_prc_out,&nbr_var_prc); #ifdef ENABLE_MPI if(prc_rnk == rnk_mgr){ /* MPI manager code */ #endif /* !ENABLE_MPI */ /* Open output file */ fl_out_tmp=nco_fl_out_open(fl_out,FORCE_APPEND,FORCE_OVERWRITE,fl_out_fmt,&out_id); /* Copy global attributes */ (void)nco_att_cpy(in_id,out_id,NC_GLOBAL,NC_GLOBAL,True); /* Catenate time-stamped command line to "history" global attribute */ if(HISTORY_APPEND) (void)nco_hst_att_cat(out_id,cmd_ln); if(thr_nbr > 0 && HISTORY_APPEND) (void)nco_thr_att_cat(out_id,thr_nbr); #ifdef ENABLE_MPI /* Initialize MPI task information */ if(prc_nbr > 0 && HISTORY_APPEND) (void)nco_mpi_att_cat(out_id,prc_nbr); #endif /* !ENABLE_MPI */ /* Add input file list global attribute */ if(FL_LST_IN_APPEND && HISTORY_APPEND && FL_LST_IN_FROM_STDIN) (void)nco_fl_lst_att_cat(out_id,fl_lst_in,fl_nbr); #ifdef ENABLE_MPI } /* prc_rnk != rnk_mgr */ #endif /* !ENABLE_MPI */ /* ncecat-specific operations */ if(True){ /* Always construct new "record" dimension from scratch */ rec_dmn=(dmn_sct *)nco_malloc(sizeof(dmn_sct)); rec_dmn->nm=(char *)strdup("record"); rec_dmn->id=-1; rec_dmn->nc_id=-1; rec_dmn->xrf=NULL; rec_dmn->val.vp=NULL; rec_dmn->is_crd_dmn=False; rec_dmn->is_rec_dmn=True; rec_dmn->sz=0L; rec_dmn->cnt=0L; rec_dmn->srt=0L; rec_dmn->end=rec_dmn->sz-1L; /* Change existing record dimension, if any, to regular dimension */ for(idx=0;idx<nbr_dmn_xtr;idx++){ /* Is any input dimension a record dimension? */ if(dmn_out[idx]->is_rec_dmn){ dmn_out[idx]->is_rec_dmn=False; break; } /* end if */ } /* end loop over idx */ /* Add record dimension to end of dimension list */ nbr_dmn_xtr++; dmn_out=(dmn_sct **)nco_realloc(dmn_out,nbr_dmn_xtr*sizeof(dmn_sct **)); dmn_out[nbr_dmn_xtr-1]=rec_dmn; } /* end if */ #ifdef ENABLE_MPI if(prc_rnk == rnk_mgr){ /* MPI manager code */ #endif /* !ENABLE_MPI */ /* Define dimensions in output file */ (void)nco_dmn_dfn(fl_out,out_id,dmn_out,nbr_dmn_xtr); #ifdef ENABLE_MPI } /* prc_rnk != rnk_mgr */ #endif /* !ENABLE_MPI */ if(True){ /* Prepend record dimension to beginning of all vectors for processed variables */ for(idx=0;idx<nbr_var_prc;idx++){ var_prc_out[idx]->nbr_dim++; var_prc_out[idx]->is_rec_var=True; var_prc_out[idx]->sz_rec=var_prc_out[idx]->sz; /* Allocate space to hold dimension IDs */ var_prc_out[idx]->dim=(dmn_sct **)nco_realloc(var_prc_out[idx]->dim,var_prc_out[idx]->nbr_dim*sizeof(dmn_sct *)); var_prc_out[idx]->dmn_id=(int *)nco_realloc(var_prc_out[idx]->dmn_id,var_prc_out[idx]->nbr_dim*sizeof(int)); var_prc_out[idx]->cnt=(long *)nco_realloc(var_prc_out[idx]->cnt,var_prc_out[idx]->nbr_dim*sizeof(long)); var_prc_out[idx]->srt=(long *)nco_realloc(var_prc_out[idx]->srt,var_prc_out[idx]->nbr_dim*sizeof(long)); var_prc_out[idx]->end=(long *)nco_realloc(var_prc_out[idx]->end,var_prc_out[idx]->nbr_dim*sizeof(long)); /* Move current array by one to make room for new record dimension info */ (void)memmove((void *)(var_prc_out[idx]->dim+1),(void *)(var_prc_out[idx]->dim),(var_prc_out[idx]->nbr_dim-1)*sizeof(dmn_sct *)); (void)memmove((void *)(var_prc_out[idx]->dmn_id+1),(void *)(var_prc_out[idx]->dmn_id),(var_prc_out[idx]->nbr_dim-1)*sizeof(int)); (void)memmove((void *)(var_prc_out[idx]->cnt+1),(void *)(var_prc_out[idx]->cnt),(var_prc_out[idx]->nbr_dim-1)*sizeof(long)); (void)memmove((void *)(var_prc_out[idx]->srt+1),(void *)(var_prc_out[idx]->srt),(var_prc_out[idx]->nbr_dim-1)*sizeof(long)); (void)memmove((void *)(var_prc_out[idx]->end+1),(void *)(var_prc_out[idx]->end),(var_prc_out[idx]->nbr_dim-1)*sizeof(long)); /* Insert value for new record dimension */ var_prc_out[idx]->dim[0]=rec_dmn; var_prc_out[idx]->dmn_id[0]=rec_dmn->id; var_prc_out[idx]->cnt[0]=1L; var_prc_out[idx]->srt[0]=-1L; var_prc_out[idx]->end[0]=-1L; } /* end loop over idx */ } /* end if */ #ifdef ENABLE_MPI if(prc_rnk == rnk_mgr){ /* MPI manager code */ #endif /* !ENABLE_MPI */ /* Define variables in output file, copy their attributes */ (void)nco_var_dfn(in_id,fl_out,out_id,var_out,nbr_xtr,(dmn_sct **)NULL,(int)0,nco_pck_plc_nil,nco_pck_map_nil); #ifdef ENABLE_MPI } /* prc_rnk != rnk_mgr */ #endif /* !ENABLE_MPI */ /* Zero start vectors for all output variables */ (void)nco_var_srt_zero(var_out,nbr_xtr); #ifdef ENABLE_MPI if(prc_rnk == rnk_mgr){ /* prc_rnk != rnk_mgr */ #endif /* !ENABLE_MPI */ /* Turn off default filling behavior to enhance efficiency */ rcd=nco_set_fill(out_id,NC_NOFILL,&fll_md_old); /* Take output file out of define mode */ (void)nco_enddef(out_id); #ifdef ENABLE_MPI } /* prc_rnk != rnk_mgr */ /* Manager obtains output filename and broadcasts to workers */ if(prc_rnk == rnk_mgr) fl_nm_lng=(int)strlen(fl_out_tmp); MPI_Bcast(&fl_nm_lng,1,MPI_INT,0,mpi_cmm); if(prc_rnk != rnk_mgr) fl_out_tmp=(char *)malloc((fl_nm_lng+1)*sizeof(char)); MPI_Bcast(fl_out_tmp,fl_nm_lng+1,MPI_CHAR,0,mpi_cmm); if(prc_rnk == rnk_mgr){ /* MPI manager code */ TKN_WRT_FREE=False; #endif /* !ENABLE_MPI */ /* Copy variable data for non-processed variables */ (void)nco_var_val_cpy(in_id,out_id,var_fix,nbr_var_fix); #ifdef ENABLE_MPI /* Close output file so workers can open it */ nco_close(out_id); TKN_WRT_FREE=True; } /* prc_rnk != rnk_mgr */ #endif /* !ENABLE_MPI */ /* Close first input netCDF file */ (void)nco_close(in_id); /* Loop over input files */ for(fl_idx=0;fl_idx<fl_nbr;fl_idx++){ #ifdef ENABLE_MPI MPI_Barrier(mpi_cmm); #endif /* !ENABLE_MPI */ /* Parse filename */ if(fl_idx != 0) fl_in=nco_fl_nm_prs(fl_in,fl_idx,(int *)NULL,fl_lst_in,abb_arg_nbr,fl_lst_abb,fl_pth); if(dbg_lvl > 0) (void)fprintf(fp_stderr,"\nInput file %d is %s; ",fl_idx,fl_in); /* Make sure file is on local system and is readable or die trying */ if(fl_idx != 0) fl_in=nco_fl_mk_lcl(fl_in,fl_pth_lcl,&FILE_RETRIEVED_FROM_REMOTE_LOCATION); if(dbg_lvl > 0) (void)fprintf(fp_stderr,"local file %s:\n",fl_in); /* Open file once per thread to improve caching */ for(thr_idx=0;thr_idx<thr_nbr;thr_idx++) rcd=nco_open(fl_in,NC_NOWRITE,in_id_arr+thr_idx); #if 0 /* fxm: netCDF4: Change to independent variable reads? */ #ifdef ENABLE_NETCDF4 rcd=nco_open_par(fl_in,NC_MPIIO|NC_NETCDF4,mpi_cmm,mpi_nfo,&in_id); #endif /* !ENABLE_NETCDF4 */ #ifdef ENABLE_PNETCDF rcd=ncompi_open(mpi_cmm,fl_in,NC_NOWRITE,mpi_nfo,&in_id); #endif /* !ENABLE_PNETCDF */ #endif /* !0 */ /* Perform various error-checks on input file */ if(False) (void)nco_fl_cmp_err_chk(); #ifdef ENABLE_MPI if(prc_rnk == rnk_mgr){ /* MPI manager code */ /* Compensate for incrementing on each worker's first message */ var_wrt_nbr=-prc_nbr+1; idx=0; /* While variables remain to be processed or written... */ while(var_wrt_nbr < nbr_var_prc){ /* Receive message from any worker */ MPI_Recv(wrk_id_bfr,wrk_id_bfr_lng,MPI_INT,MPI_ANY_SOURCE,MPI_ANY_TAG,mpi_cmm,&mpi_stt); /* Obtain MPI message tag type */ msg_tag_typ=mpi_stt.MPI_TAG; /* Get sender's prc_rnk */ rnk_wrk=wrk_id_bfr[0]; /* Allocate next variable, if any, to worker */ if(msg_tag_typ == msg_tag_wrk_rqs){ var_wrt_nbr++; /* [nbr] Number of variables written */ /* Worker closed output file before sending msg_tag_wrk_rqs */ TKN_WRT_FREE=True; if(idx > nbr_var_prc-1){ msg_bfr[0]=idx_all_wrk_ass; /* [enm] All variables already assigned */ msg_bfr[1]=out_id; /* Output file ID */ }else{ /* Tell requesting worker to allocate space for next variable */ msg_bfr[0]=idx; /* [idx] Variable to be processed */ msg_bfr[1]=out_id; /* Output file ID */ msg_bfr[2]=var_prc_out[idx]->id; /* [id] Variable ID in output file */ /* Point to next variable on list */ idx++; } /* endif idx */ MPI_Send(msg_bfr,msg_bfr_lng,MPI_INT,rnk_wrk,msg_tag_wrk_rsp,mpi_cmm); /* msg_tag_typ != msg_tag_wrk_rqs */ }else if(msg_tag_typ == msg_tag_tkn_wrt_rqs){ /* Allocate token if free, else ask worker to try later */ if(TKN_WRT_FREE){ TKN_WRT_FREE=False; msg_bfr[0]=tkn_wrt_rqs_xcp; /* Accept request for write token */ }else{ msg_bfr[0]=tkn_wrt_rqs_dny; /* Deny request for write token */ } /* !TKN_WRT_FREE */ MPI_Send(msg_bfr,msg_bfr_lng,MPI_INT,rnk_wrk,msg_tag_tkn_wrt_rsp,mpi_cmm); } /* msg_tag_typ != msg_tag_tkn_wrt_rqs */ } /* end while var_wrt_nbr < nbr_var_prc */ }else{ /* prc_rnk != rnk_mgr, end Manager code begin Worker code */ wrk_id_bfr[0]=prc_rnk; while(1){ /* While work remains... */ /* Send msg_tag_wrk_rqs */ wrk_id_bfr[0]=prc_rnk; MPI_Send(wrk_id_bfr,wrk_id_bfr_lng,MPI_INT,rnk_mgr,msg_tag_wrk_rqs,mpi_cmm); /* Receive msg_tag_wrk_rsp */ MPI_Recv(msg_bfr,msg_bfr_lng,MPI_INT,0,msg_tag_wrk_rsp,mpi_cmm,&mpi_stt); idx=msg_bfr[0]; out_id=msg_bfr[1]; if(idx == idx_all_wrk_ass) break; else{ var_prc_out[idx]->id=msg_bfr[2]; /* Process this variable same as UP code */ #else /* !ENABLE_MPI */ /* OpenMP with threading over variables, not files */ #ifdef _OPENMP #pragma omp parallel for default(none) private(idx,in_id) shared(dbg_lvl,fl_nbr,idx_rec_out,in_id_arr,nbr_var_prc,out_id,var_prc,var_prc_out) #endif /* !_OPENMP */ /* Process all variables in current file */ for(idx=0;idx<nbr_var_prc;idx++){ #endif /* !ENABLE_MPI */ /* Common code for UP and MPI */ /* fxm: requires C99 as is? */ in_id=in_id_arr[omp_get_thread_num()]; if(dbg_lvl > 1) (void)fprintf(fp_stderr,"%s, ",var_prc[idx]->nm); if(dbg_lvl > 0) (void)fflush(fp_stderr); /* Variables may have different ID, missing_value, type, in each file */ (void)nco_var_mtd_refresh(in_id,var_prc[idx]); /* Retrieve variable from disk into memory */ /* NB: nco_var_get() with same nc_id contains OpenMP critical region */ (void)nco_var_get(in_id,var_prc[idx]); /* Size of record dimension is 1 in output file */ var_prc_out[idx]->cnt[0]=1L; var_prc_out[idx]->srt[0]=idx_rec_out; #ifdef ENABLE_MPI /* Obtain token and prepare to write */ while(1){ /* Send msg_tag_tkn_wrt_rqs repeatedly until token obtained */ wrk_id_bfr[0]=prc_rnk; MPI_Send(wrk_id_bfr,wrk_id_bfr_lng,MPI_INT,rnk_mgr,msg_tag_tkn_wrt_rqs,mpi_cmm); MPI_Recv(msg_bfr,msg_bfr_lng,MPI_INT,rnk_mgr,msg_tag_tkn_wrt_rsp,mpi_cmm,&mpi_stt); tkn_wrt_rsp=msg_bfr[0]; /* Wait then re-send request */ if(tkn_wrt_rsp == tkn_wrt_rqs_dny) sleep(tkn_wrt_rqs_ntv); else break; } /* end while loop waiting for write token */ /* Worker has token---prepare to write */ if(tkn_wrt_rsp == tkn_wrt_rqs_xcp){ rcd=nco_open(fl_out_tmp,NC_WRITE|NC_SHARE,&out_id); /* Turn off default filling behavior to enhance efficiency */ rcd=nco_set_fill(out_id,NC_NOFILL,&fll_md_old); #else /* !ENABLE_MPI */ #ifdef _OPENMP #pragma omp critical #endif /* _OPENMP */ #endif /* !ENABLE_MPI */ { /* begin OpenMP critical */ /* Write variable into current record in output file */ if(var_prc[idx]->nbr_dim == 0){ (void)nco_put_var1(out_id,var_prc_out[idx]->id,var_prc_out[idx]->srt,var_prc[idx]->val.vp,var_prc[idx]->type); }else{ /* end if variable is a scalar */ (void)nco_put_vara(out_id,var_prc_out[idx]->id,var_prc_out[idx]->srt,var_prc_out[idx]->cnt,var_prc[idx]->val.vp,var_prc[idx]->type); } /* end if variable is array */ /* Free current input buffer */ var_prc[idx]->val.vp=nco_free(var_prc[idx]->val.vp); } /* end OpenMP critical */ #ifdef ENABLE_MPI /* Close output file and increment written counter */ nco_close(out_id); var_wrt_nbr++; } /* endif tkn_wrt_rqs_xcp */ } /* end else !idx_all_wrk_ass */ } /* end while loop requesting work/token */ } /* endif Worker */ #else /* !ENABLE_MPI */ } /* end (OpenMP parallel for) loop over idx */ #endif /* !ENABLE_MPI */ idx_rec_out++; /* [idx] Index of current record in output file (0 is first, ...) */ if(dbg_lvl > 1) (void)fprintf(stderr,"\n"); /* Close input netCDF file */ for(thr_idx=0;thr_idx<thr_nbr;thr_idx++) nco_close(in_id_arr[thr_idx]); /* Remove local copy of file */ if(FILE_RETRIEVED_FROM_REMOTE_LOCATION && REMOVE_REMOTE_FILES_AFTER_PROCESSING) (void)nco_fl_rm(fl_in); #ifdef ENABLE_MPI MPI_Barrier(mpi_cmm); #endif /* !ENABLE_MPI */ } /* end loop over fl_idx */ #ifdef ENABLE_MPI /* Manager moves output file (closed by workers) from temporary to permanent location */ if(prc_rnk == rnk_mgr) (void)nco_fl_mv(fl_out_tmp,fl_out); #else /* !ENABLE_MPI */ /* Close output file and move it from temporary to permanent location */ (void)nco_fl_out_cls(fl_out,fl_out_tmp,out_id); #endif /* end !ENABLE_MPI */ /* ncecat-specific memory cleanup */ /* NCO-generic clean-up */ /* Free individual strings/arrays */ if(cmd_ln != NULL) cmd_ln=(char *)nco_free(cmd_ln); if(fl_in != NULL) fl_in=(char *)nco_free(fl_in); if(fl_out != NULL) fl_out=(char *)nco_free(fl_out); if(fl_out_tmp != NULL) fl_out_tmp=(char *)nco_free(fl_out_tmp); if(fl_pth != NULL) fl_pth=(char *)nco_free(fl_pth); if(fl_pth_lcl != NULL) fl_pth_lcl=(char *)nco_free(fl_pth_lcl); if(in_id_arr != NULL) in_id_arr=(int *)nco_free(in_id_arr); /* Free lists of strings */ if(fl_lst_in != NULL && fl_lst_abb == NULL) fl_lst_in=nco_sng_lst_free(fl_lst_in,fl_nbr); if(fl_lst_in != NULL && fl_lst_abb != NULL) fl_lst_in=nco_sng_lst_free(fl_lst_in,1); if(fl_lst_abb != NULL) fl_lst_abb=nco_sng_lst_free(fl_lst_abb,abb_arg_nbr); if(var_lst_in_nbr > 0) var_lst_in=nco_sng_lst_free(var_lst_in,var_lst_in_nbr); /* Free limits */ for(idx=0;idx<lmt_nbr;idx++) lmt_arg[idx]=(char *)nco_free(lmt_arg[idx]); if(lmt_nbr > 0) lmt=nco_lmt_lst_free(lmt,lmt_nbr); /* Free dimension lists */ if(nbr_dmn_xtr > 0) dim=nco_dmn_lst_free(dim,nbr_dmn_xtr-1); /* NB: ncecat has one fewer input than output dimension */ if(nbr_dmn_xtr > 0) dmn_out=nco_dmn_lst_free(dmn_out,nbr_dmn_xtr); /* Free variable lists */ if(nbr_xtr > 0) var=nco_var_lst_free(var,nbr_xtr); if(nbr_xtr > 0) var_out=nco_var_lst_free(var_out,nbr_xtr); var_prc=(var_sct **)nco_free(var_prc); var_prc_out=(var_sct **)nco_free(var_prc_out); var_fix=(var_sct **)nco_free(var_fix); var_fix_out=(var_sct **)nco_free(var_fix_out); #ifdef ENABLE_MPI MPI_Finalize(); #endif /* !ENABLE_MPI */ if(rcd != NC_NOERR) nco_err_exit(rcd,"main"); nco_exit_gracefully(); return EXIT_SUCCESS; } /* end main() */

---