00001 /*
00002  *      Copyright 1996, University Corporation for Atmospheric Research
00003  *      See netcdf/COPYRIGHT file for copying and redistribution conditions.
00004  */
00005 
00006 #include "nc.h"
00007 #include "rnd.h"
00008 #include <stdlib.h>
00009 #include <string.h>
00010 #include <assert.h>
00011 #include "ncx.h"
00012 #if defined(LOCKNUMREC) /* && _CRAYMPP */
00013 #  include <mpp/shmem.h>
00014 #  include <intrinsics.h>
00015 #endif
00016 
00017 /* list of open netcdf's */
00018 static NC *NClist = NULL;
00019 
00020 static void
00021 add_to_NCList(NC *ncp)
00022 {
00023         assert(ncp != NULL);
00024 
00025         ncp->prev = NULL;
00026         if(NClist != NULL)
00027                 NClist->prev = ncp;
00028         ncp->next = NClist;
00029         NClist = ncp;
00030 }
00031 
00032 static void
00033 del_from_NCList(NC *ncp)
00034 {
00035         assert(ncp != NULL);
00036 
00037         if(NClist == ncp)
00038         {
00039                 assert(ncp->prev == NULL);
00040                 NClist = ncp->next;
00041         }
00042         else
00043         {
00044                 assert(ncp->prev != NULL);
00045                 ncp->prev->next = ncp->next;
00046         }
00047 
00048         if(ncp->next != NULL)
00049                 ncp->next->prev = ncp->prev;
00050 
00051         ncp->next = NULL;
00052         ncp->prev = NULL;
00053 }
00054 
00055 
00056 int
00057 NC_check_id(int ncid, NC **ncpp)
00058 {
00059         NC *ncp;
00060 
00061         if(ncid >= 0)
00062         {
00063                 for(ncp = NClist; ncp != NULL; ncp = ncp->next)
00064                 {
00065                         if(ncp->nciop->fd == ncid)
00066                         {
00067                                 *ncpp = ncp;
00068                                 return NC_NOERR; /* normal return */
00069                         }
00070                 }
00071         }
00072 
00073         /* else, not found */
00074         return NC_EBADID;
00075 }
00076 
00077 
00078 static void
00079 free_NC(NC *ncp)
00080 {
00081         if(ncp == NULL)
00082                 return;
00083         free_NC_dimarrayV(&ncp->dims);
00084         free_NC_attrarrayV(&ncp->attrs);
00085         free_NC_vararrayV(&ncp->vars);
00086 #if _CRAYMPP && defined(LOCKNUMREC)
00087         shfree(ncp);
00088 #else
00089         free(ncp);
00090 #endif /* _CRAYMPP && LOCKNUMREC */
00091 }
00092 
00093 
00094 static NC *
00095 new_NC(const size_t *chunkp)
00096 {
00097         NC *ncp;
00098 
00099 #if _CRAYMPP && defined(LOCKNUMREC)
00100         ncp = (NC *) shmalloc(sizeof(NC));
00101 #else
00102         ncp = (NC *) malloc(sizeof(NC));
00103 #endif /* _CRAYMPP && LOCKNUMREC */
00104         if(ncp == NULL)
00105                 return NULL;
00106         (void) memset(ncp, 0, sizeof(NC));
00107 
00108         ncp->xsz = MIN_NC_XSZ;
00109         assert(ncp->xsz == ncx_len_NC(ncp));
00110         
00111         ncp->chunk = chunkp != NULL ? *chunkp : NC_SIZEHINT_DEFAULT;
00112 
00113         return ncp;
00114 }
00115 
00116 
00117 static NC *
00118 dup_NC(const NC *ref)
00119 {
00120         NC *ncp;
00121 
00122 #if _CRAYMPP && defined(LOCKNUMREC)
00123         ncp = (NC *) shmalloc(sizeof(NC));
00124 #else
00125         ncp = (NC *) malloc(sizeof(NC));
00126 #endif /* _CRAYMPP && LOCKNUMREC */
00127         if(ncp == NULL)
00128                 return NULL;
00129         (void) memset(ncp, 0, sizeof(NC));
00130 
00131         if(dup_NC_dimarrayV(&ncp->dims, &ref->dims) != NC_NOERR)
00132                 goto err;
00133         if(dup_NC_attrarrayV(&ncp->attrs, &ref->attrs) != NC_NOERR)
00134                 goto err;
00135         if(dup_NC_vararrayV(&ncp->vars, &ref->vars) != NC_NOERR)
00136                 goto err;
00137 
00138         ncp->xsz = ref->xsz;
00139         ncp->begin_var = ref->begin_var;
00140         ncp->begin_rec = ref->begin_rec;
00141         ncp->recsize = ref->recsize;
00142         NC_set_numrecs(ncp, NC_get_numrecs(ref));
00143         return ncp;
00144 err:
00145         free_NC(ncp);
00146         return NULL;
00147 }
00148 
00149 
00150 /*
00151  *  Verify that this is a user nc_type
00152  * Formerly
00153 NCcktype()
00154  * Sense of the return is changed.
00155  */
00156 int
00157 nc_cktype(nc_type type)
00158 {
00159         switch((int)type){
00160         case NC_BYTE:
00161         case NC_CHAR:
00162         case NC_SHORT:
00163         case NC_INT:
00164         case NC_FLOAT:
00165         case NC_DOUBLE:
00166                 return(NC_NOERR);
00167         }
00168         return(NC_EBADTYPE);
00169 }
00170 
00171 
00172 /*
00173  * How many objects of 'type'
00174  * will fit into xbufsize?
00175  */
00176 size_t
00177 ncx_howmany(nc_type type, size_t xbufsize)
00178 {
00179         switch(type){
00180         case NC_BYTE:
00181         case NC_CHAR:
00182                 return xbufsize;
00183         case NC_SHORT:
00184                 return xbufsize/X_SIZEOF_SHORT;
00185         case NC_INT:
00186                 return xbufsize/X_SIZEOF_INT;
00187         case NC_FLOAT:
00188                 return xbufsize/X_SIZEOF_FLOAT;
00189         case NC_DOUBLE:
00190                 return xbufsize/X_SIZEOF_DOUBLE;
00191         }
00192         assert("ncx_howmany: Bad type" == 0);
00193         return(0);
00194 }
00195 
00196 #define D_RNDUP(x, align) _RNDUP(x, (off_t)(align))
00197 
00198 /*
00199  * Compute each variable's 'begin' offset,
00200  * update 'begin_rec' as well.
00201  */
00202 static void
00203 NC_begins(NC *ncp,
00204         size_t h_minfree, size_t v_align,
00205         size_t v_minfree, size_t r_align)
00206 {
00207         size_t ii;
00208         off_t index = 0;
00209         NC_var **vpp;
00210         NC_var *last = NULL;
00211 
00212         if(v_align == NC_ALIGN_CHUNK)
00213                 v_align = ncp->chunk;
00214         if(r_align == NC_ALIGN_CHUNK)
00215                 r_align = ncp->chunk;
00216 
00217         ncp->xsz = ncx_len_NC(ncp);
00218 
00219         if(ncp->vars.nelems == 0) 
00220                 return;
00221 
00222         /* only (re)calculate begin_var if there is not sufficient space in header
00223            or start of non-record variables is not aligned as requested by valign */
00224         if (ncp->begin_var < ncp->xsz + h_minfree ||
00225             ncp->begin_var != D_RNDUP(ncp->begin_var, v_align) ) 
00226         {
00227           index = (off_t) ncp->xsz;
00228           ncp->begin_var = D_RNDUP(index, v_align);
00229           if(ncp->begin_var < index + h_minfree)
00230           {
00231             ncp->begin_var = D_RNDUP(index + (off_t)h_minfree, v_align);
00232           }
00233         }
00234         index = ncp->begin_var;
00235 
00236         /* loop thru vars, first pass is for the 'non-record' vars */
00237         vpp = ncp->vars.value;
00238         for(ii = 0; ii < ncp->vars.nelems ; ii++, vpp++)
00239         {
00240                 if( IS_RECVAR(*vpp) )
00241                 {
00242                         /* skip record variables on this pass */
00243                         continue;
00244                 }
00245 #if 0
00246 fprintf(stderr, "    VAR %d %s: %ld\n", ii, (*vpp)->name->cp, (long)index);
00247 #endif
00248                 (*vpp)->begin = index;
00249                 index += (*vpp)->len;
00250         }
00251 
00252         /* only (re)calculate begin_rec if there is not sufficient
00253            space at end of non-record variables or if start of record
00254            variables is not aligned as requested by r_align */
00255         if (ncp->begin_rec < index + v_minfree ||
00256             ncp->begin_rec != D_RNDUP(ncp->begin_rec, r_align) )
00257         {
00258           ncp->begin_rec = D_RNDUP(index, r_align);
00259           if(ncp->begin_rec < index + v_minfree)
00260           {
00261             ncp->begin_rec = D_RNDUP(index + (off_t)v_minfree, r_align);
00262           }
00263         }
00264         index = ncp->begin_rec;
00265 
00266         ncp->recsize = 0;
00267 
00268         /* loop thru vars, second pass is for the 'record' vars */
00269         vpp = (NC_var **)ncp->vars.value;
00270         for(ii = 0; ii < ncp->vars.nelems; ii++, vpp++)
00271         {
00272                 if( !IS_RECVAR(*vpp) )
00273                 {
00274                         /* skip non-record variables on this pass */
00275                         continue;
00276                 }
00277 
00278 #if 0
00279 fprintf(stderr, "    REC %d %s: %ld\n", ii, (*vpp)->name->cp, (long)index);
00280 #endif
00281                 (*vpp)->begin = index;
00282                 index += (*vpp)->len;
00283                 ncp->recsize += (*vpp)->len;
00284                 last = (*vpp);
00285         }
00286 
00287         /*
00288          * for special case of exactly one record variable, pack value
00289          */
00290         if(last != NULL && ncp->recsize == last->len)
00291                 ncp->recsize = *last->dsizes * last->xsz;
00292 
00293         if(NC_IsNew(ncp))
00294                 NC_set_numrecs(ncp, 0);
00295 
00296 }
00297 
00298 
00299 /*
00300  * Read just the numrecs member.
00301  * (A relatively expensive way to do things.)
00302  */
00303 int
00304 read_numrecs(NC *ncp)
00305 {
00306         int status = NC_NOERR;
00307         const void *xp;
00308         size_t nrecs = NC_get_numrecs(ncp);
00309 
00310         assert(!NC_indef(ncp));
00311 
00312 #define NC_NUMRECS_OFFSET 4
00313 #define NC_NUMRECS_EXTENT 4
00314         status = ncp->nciop->get(ncp->nciop,
00315                  NC_NUMRECS_OFFSET, NC_NUMRECS_EXTENT, 0, (void **)&xp);
00316                                         /* cast away const */
00317         if(status != NC_NOERR)
00318                 return status;
00319 
00320         status = ncx_get_size_t(&xp, &nrecs);
00321 
00322         (void) ncp->nciop->rel(ncp->nciop, NC_NUMRECS_OFFSET, 0);
00323 
00324         if(status == NC_NOERR)
00325         {
00326                 NC_set_numrecs(ncp, nrecs);
00327                 fClr(ncp->flags, NC_NDIRTY);
00328         }
00329 
00330         return status;
00331 }
00332 
00333 
00334 /*
00335  * Write out just the numrecs member.
00336  * (A relatively expensive way to do things.)
00337  */
00338 int
00339 write_numrecs(NC *ncp)
00340 {
00341         int status = NC_NOERR;
00342         void *xp;
00343 
00344         assert(!NC_readonly(ncp));
00345         assert(!NC_indef(ncp));
00346 
00347         status = ncp->nciop->get(ncp->nciop,
00348                  NC_NUMRECS_OFFSET, NC_NUMRECS_EXTENT, RGN_WRITE, &xp);
00349         if(status != NC_NOERR)
00350                 return status;
00351 
00352         {
00353                 const size_t nrecs = NC_get_numrecs(ncp);
00354                 status = ncx_put_size_t(&xp, &nrecs);
00355         }
00356 
00357         (void) ncp->nciop->rel(ncp->nciop, NC_NUMRECS_OFFSET, RGN_MODIFIED);
00358 
00359         if(status == NC_NOERR)
00360                 fClr(ncp->flags, NC_NDIRTY);
00361 
00362         return status;
00363 }
00364 
00365 
00366 /*
00367  * Read in the header
00368  * It is expensive.
00369  */
00370 static int
00371 read_NC(NC *ncp)
00372 {
00373         int status = NC_NOERR;
00374 
00375         free_NC_dimarrayV(&ncp->dims);
00376         free_NC_attrarrayV(&ncp->attrs);
00377         free_NC_vararrayV(&ncp->vars);
00378 
00379         status = nc_get_NC(ncp);
00380 
00381         if(status == NC_NOERR)
00382                 fClr(ncp->flags, NC_NDIRTY | NC_HDIRTY);
00383 
00384         return status;
00385 }
00386 
00387 
00388 /*
00389  * Write out the header
00390  */
00391 static int
00392 write_NC(NC *ncp)
00393 {
00394         int status = NC_NOERR;
00395 
00396         assert(!NC_readonly(ncp));
00397 
00398         status = ncx_put_NC(ncp, NULL, 0, 0);
00399 
00400         if(status == NC_NOERR)
00401                 fClr(ncp->flags, NC_NDIRTY | NC_HDIRTY);
00402 
00403         return status;
00404 }
00405 
00406 
00407 /*
00408  * Write the header or the numrecs if necessary.
00409  */
00410 int
00411 NC_sync(NC *ncp)
00412 {
00413         assert(!NC_readonly(ncp));
00414 
00415         if(NC_hdirty(ncp))
00416         {
00417                 return write_NC(ncp);
00418         }
00419         /* else */
00420 
00421         if(NC_ndirty(ncp))
00422         {
00423                 return write_numrecs(ncp);
00424         }
00425         /* else */
00426 
00427         return NC_NOERR;
00428 }
00429 
00430 
00431 /*
00432  * Initialize the 'non-record' variables.
00433  */
00434 static int
00435 fillerup(NC *ncp)
00436 {
00437         int status = NC_NOERR;
00438         size_t ii;
00439         NC_var **varpp;
00440 
00441         assert(!NC_readonly(ncp));
00442         assert(NC_dofill(ncp));
00443 
00444         /* loop thru vars */
00445         varpp = ncp->vars.value;
00446         for(ii = 0; ii < ncp->vars.nelems; ii++, varpp++)
00447         {
00448                 if(IS_RECVAR(*varpp))
00449                 {
00450                         /* skip record variables */
00451                         continue;
00452                 }
00453 
00454                 status = fill_NC_var(ncp, *varpp, 0);
00455                 if(status != NC_NOERR)
00456                         break;
00457         }
00458         return status;
00459 }
00460 
00461 /* Begin endef */
00462 
00463 /*
00464  */
00465 static int
00466 fill_added_recs(NC *gnu, NC *old)
00467 {
00468         NC_var ** const gnu_varpp = (NC_var **)gnu->vars.value;
00469 
00470         const int old_nrecs = (int) NC_get_numrecs(old);
00471         int recno = 0;
00472         for(; recno < old_nrecs; recno++)
00473         {
00474                 int varid = (int)old->vars.nelems;
00475                 for(; varid < (int)gnu->vars.nelems; varid++)
00476                 {
00477                         const NC_var *const gnu_varp = *(gnu_varpp + varid);
00478                         if(!IS_RECVAR(gnu_varp))
00479                         {
00480                                 /* skip non-record variables */
00481                                 continue;
00482                         }
00483                         /* else */
00484                         {
00485                         const int status = fill_NC_var(gnu, gnu_varp, recno);
00486                         if(status != NC_NOERR)
00487                                 return status;
00488                         }
00489                 }
00490         }
00491 
00492         return NC_NOERR;
00493 }
00494 
00495 /*
00496  */
00497 static int
00498 fill_added(NC *gnu, NC *old)
00499 {
00500         NC_var ** const gnu_varpp = (NC_var **)gnu->vars.value;
00501         int varid = (int)old->vars.nelems;
00502 
00503         for(; varid < (int)gnu->vars.nelems; varid++)
00504         {
00505                 const NC_var *const gnu_varp = *(gnu_varpp + varid);
00506                 if(IS_RECVAR(gnu_varp))
00507                 {
00508                         /* skip record variables */
00509                         continue;
00510                 }
00511                 /* else */
00512                 {
00513                 const int status = fill_NC_var(gnu, gnu_varp, 0);
00514                 if(status != NC_NOERR)
00515                         return status;
00516                 }
00517         }
00518 
00519         return NC_NOERR;
00520 }
00521 
00522 
00523 /*
00524  * Move the records "out". 
00525  * Fill as needed.
00526  */
00527 static int
00528 move_recs_r(NC *gnu, NC *old)
00529 {
00530         int status;
00531         int recno;
00532         int varid;
00533         NC_var **gnu_varpp = (NC_var **)gnu->vars.value;
00534         NC_var **old_varpp = (NC_var **)old->vars.value;
00535         NC_var *gnu_varp;
00536         NC_var *old_varp;
00537         off_t gnu_off;
00538         off_t old_off;
00539         const size_t old_nrecs = NC_get_numrecs(old);
00540         
00541         /* Don't parallelize this loop */
00542         for(recno = (int)old_nrecs -1; recno >= 0; recno--)
00543         {
00544         /* Don't parallelize this loop */
00545         for(varid = (int)old->vars.nelems -1; varid >= 0; varid--)
00546         {
00547                 gnu_varp = *(gnu_varpp + varid);
00548                 if(!IS_RECVAR(gnu_varp))
00549                 {
00550                         /* skip non-record variables on this pass */
00551                         continue;
00552                 }
00553                 /* else */
00554 
00555                 /* else, a pre-existing variable */
00556                 old_varp = *(old_varpp + varid);
00557                 gnu_off = gnu_varp->begin + (off_t)(gnu->recsize * recno);
00558                 old_off = old_varp->begin + (off_t)(old->recsize * recno);
00559 
00560                 if(gnu_off == old_off)
00561                         continue;       /* nothing to do */
00562 
00563                 assert(gnu_off > old_off);
00564         
00565                 status = gnu->nciop->move(gnu->nciop, gnu_off, old_off,
00566                          old_varp->len, 0);
00567 
00568                 if(status != NC_NOERR)
00569                         return status;
00570                 
00571         }
00572         }
00573 
00574         NC_set_numrecs(gnu, old_nrecs);
00575 
00576         return NC_NOERR;
00577 }
00578 
00579 
00580 /*
00581  * Move the "non record" variables "out". 
00582  * Fill as needed.
00583  */
00584 static int
00585 move_vars_r(NC *gnu, NC *old)
00586 {
00587         int status;
00588         int varid;
00589         NC_var **gnu_varpp = (NC_var **)gnu->vars.value;
00590         NC_var **old_varpp = (NC_var **)old->vars.value;
00591         NC_var *gnu_varp;
00592         NC_var *old_varp;
00593         off_t gnu_off;
00594         off_t old_off;
00595         
00596         /* Don't parallelize this loop */
00597         for(varid = (int)old->vars.nelems -1;
00598                  varid >= 0; varid--)
00599         {
00600                 gnu_varp = *(gnu_varpp + varid);
00601                 if(IS_RECVAR(gnu_varp))
00602                 {
00603                         /* skip record variables on this pass */
00604                         continue;
00605                 }
00606                 /* else */
00607 
00608                 old_varp = *(old_varpp + varid);
00609                 gnu_off = gnu_varp->begin;
00610                 old_off = old_varp->begin;
00611         
00612                 if(gnu_off == old_off)
00613                         continue;       /* nothing to do */
00614 
00615                 assert(gnu_off > old_off);
00616 
00617                 status = gnu->nciop->move(gnu->nciop, gnu_off, old_off,
00618                          old_varp->len, 0);
00619 
00620                 if(status != NC_NOERR)
00621                         return status;
00622                 
00623         }
00624 
00625         return NC_NOERR;
00626 }
00627 
00628 
00629 /*
00630  *  End define mode.
00631  *  Common code for ncendef, ncclose(endef)
00632  *  Flushes I/O buffers.
00633  */
00634 static int
00635 NC_endef(NC *ncp,
00636         size_t h_minfree, size_t v_align,
00637         size_t v_minfree, size_t r_align)
00638 {
00639         int status = NC_NOERR;
00640 
00641         assert(!NC_readonly(ncp));
00642         assert(NC_indef(ncp));
00643 
00644         NC_begins(ncp, h_minfree, v_align, v_minfree, r_align);
00645 
00646         if(ncp->old != NULL)
00647         {
00648                 /* a plain redef, not a create */
00649                 assert(!NC_IsNew(ncp));
00650                 assert(fIsSet(ncp->flags, NC_INDEF));
00651                 assert(ncp->begin_rec >= ncp->old->begin_rec);
00652                 assert(ncp->begin_var >= ncp->old->begin_var);
00653 
00654                 if(ncp->vars.nelems != 0)
00655                 {
00656                 if(ncp->begin_rec > ncp->old->begin_rec)
00657                 {
00658                         status = move_recs_r(ncp, ncp->old);
00659                         if(status != NC_NOERR)
00660                                 return status;
00661                         if(ncp->begin_var > ncp->old->begin_var)
00662                         {
00663                                 status = move_vars_r(ncp, ncp->old);
00664                                 if(status != NC_NOERR)
00665                                         return status;
00666                         } 
00667                         /* else if (ncp->begin_var == ncp->old->begin_var) { NOOP } */
00668                 }
00669                 else
00670                 {       /* Even if (ncp->begin_rec == ncp->old->begin_rec)
00671                            and     (ncp->begin_var == ncp->old->begin_var)
00672                            might still have added a new record variable */
00673                         if(ncp->recsize > ncp->old->recsize)
00674                         {
00675                                 status = move_recs_r(ncp, ncp->old);
00676                                 if(status != NC_NOERR)
00677                                       return status;
00678                         }
00679                 }
00680                 }
00681         }
00682 
00683         status = write_NC(ncp);
00684         if(status != NC_NOERR)
00685                 return status;
00686 
00687         if(NC_dofill(ncp))
00688         {
00689                 if(NC_IsNew(ncp))
00690                 {
00691                         status = fillerup(ncp);
00692                         if(status != NC_NOERR)
00693                                 return status;
00694                         
00695                 }
00696                 else if(ncp->vars.nelems > ncp->old->vars.nelems)
00697                 {
00698                         status = fill_added(ncp, ncp->old);
00699                         if(status != NC_NOERR)
00700                                 return status;
00701                         status = fill_added_recs(ncp, ncp->old);
00702                         if(status != NC_NOERR)
00703                                 return status;
00704                 }
00705         }
00706 
00707         if(ncp->old != NULL)
00708         {
00709                 free_NC(ncp->old);
00710                 ncp->old = NULL;
00711         }
00712 
00713         fClr(ncp->flags, NC_CREAT | NC_INDEF);
00714 
00715         return ncp->nciop->sync(ncp->nciop);
00716 }
00717 
00718 #ifdef LOCKNUMREC
00719 static int
00720 NC_init_pe(NC *ncp, int basepe) {
00721         if (basepe < 0 || basepe >= _num_pes()) {
00722                 return NC_EINVAL; /* invalid base pe */
00723         }
00724         /* initialize common values */
00725         ncp->lock[LOCKNUMREC_VALUE] = 0;
00726         ncp->lock[LOCKNUMREC_LOCK] = 0;
00727         ncp->lock[LOCKNUMREC_SERVING] = 0;
00728         ncp->lock[LOCKNUMREC_BASEPE] =  basepe;
00729         return NC_NOERR;
00730 }
00731 #endif
00732 
00733 /* Public */
00734 
00735 int
00736 nc__create(const char * path, int ioflags, size_t initialsz,
00737         size_t *chunksizehintp, int *ncid_ptr)
00738 {
00739         return nc__create_mp(path, ioflags, initialsz, 0,
00740                 chunksizehintp, ncid_ptr);
00741 }
00742 
00743 int
00744 nc__create_mp(const char * path, int ioflags, size_t initialsz, int basepe,
00745         size_t *chunksizehintp, int *ncid_ptr)
00746 {
00747         NC *ncp;
00748         int status;
00749         void *xp = NULL;
00750 
00751 #if ALWAYS_NC_SHARE /* DEBUG */
00752         fSet(ioflags, NC_SHARE);
00753 #endif
00754 
00755         ncp = new_NC(chunksizehintp);
00756         if(ncp == NULL)
00757                 return NC_ENOMEM;
00758 
00759 #if defined(LOCKNUMREC) /* && _CRAYMPP */
00760         if (status = NC_init_pe(ncp, basepe)) {
00761                 return status;
00762         }
00763 #else
00764         /*
00765          * !_CRAYMPP, only pe 0 is valid
00766          */
00767         if(basepe != 0)
00768                 return NC_EINVAL;
00769 #endif
00770         assert(ncp->xsz == ncx_len_NC(ncp));
00771         
00772         status = ncio_create(path, ioflags,
00773                 initialsz,
00774                 0, ncp->xsz, &ncp->chunk,
00775                 &ncp->nciop, &xp);
00776         if(status != NC_NOERR)
00777         {
00778                 /* translate error status */
00779                 if(status == EEXIST)
00780                         status = NC_EEXIST;
00781                 goto unwind_alloc;
00782         }
00783 
00784         assert(ncp->flags == 0);
00785         fSet(ncp->flags, NC_CREAT);
00786 
00787         if(fIsSet(ncp->nciop->ioflags, NC_SHARE))
00788         {
00789                 /*
00790                  * NC_SHARE implies sync up the number of records as well.
00791                  * (File format version one.)
00792                  * Note that other header changes are not shared
00793                  * automatically.  Some sort of IPC (external to this package)
00794                  * would be used to trigger a call to nc_sync().
00795                  */
00796                 fSet(ncp->flags, NC_NSYNC);
00797         }
00798 
00799         status = ncx_put_NC(ncp, &xp, 0, ncp->xsz);
00800         if(status != NC_NOERR)
00801                 goto unwind_ioc;
00802 
00803         add_to_NCList(ncp);
00804 
00805         if(chunksizehintp != NULL)
00806                 *chunksizehintp = ncp->chunk;
00807         *ncid_ptr = ncp->nciop->fd;
00808         return NC_NOERR;
00809 
00810 unwind_ioc:
00811         (void) ncio_close(ncp->nciop, 1); /* N.B.: unlink */
00812         ncp->nciop = NULL;
00813         /*FALLTHRU*/
00814 unwind_alloc:
00815         free_NC(ncp);
00816         return status;
00817 }
00818 
00819 int
00820 nc_create(const char * path, int ioflags, int *ncid_ptr)
00821 {
00822         return nc__create(path, ioflags, 0, NULL, ncid_ptr);
00823 }
00824 
00825 
00826 int
00827 nc__open(const char * path, int ioflags,
00828         size_t *chunksizehintp, int *ncid_ptr)
00829 {
00830         return nc__open_mp(path, ioflags, 0,
00831                 chunksizehintp, ncid_ptr);
00832 }
00833 
00834 int
00835 nc__open_mp(const char * path, int ioflags, int basepe,
00836         size_t *chunksizehintp, int *ncid_ptr)
00837 {
00838         NC *ncp;
00839         int status;
00840 
00841 #if ALWAYS_NC_SHARE /* DEBUG */
00842         fSet(ioflags, NC_SHARE);
00843 #endif
00844 
00845         ncp = new_NC(chunksizehintp);
00846         if(ncp == NULL)
00847                 return NC_ENOMEM;
00848 
00849 #if defined(LOCKNUMREC) /* && _CRAYMPP */
00850         if (status = NC_init_pe(ncp, basepe)) {
00851                 return status;
00852         }
00853 #else
00854         /*
00855          * !_CRAYMPP, only pe 0 is valid
00856          */
00857         if(basepe != 0)
00858                 return NC_EINVAL;
00859 #endif
00860         status = ncio_open(path, ioflags,
00861                 0, 0, &ncp->chunk,
00862                 &ncp->nciop, 0);
00863         if(status)
00864                 goto unwind_alloc;
00865 
00866         assert(ncp->flags == 0);
00867 
00868         if(fIsSet(ncp->nciop->ioflags, NC_SHARE))
00869         {
00870                 /*
00871                  * NC_SHARE implies sync up the number of records as well.
00872                  * (File format version one.)
00873                  * Note that other header changes are not shared
00874                  * automatically.  Some sort of IPC (external to this package)
00875                  * would be used to trigger a call to nc_sync().
00876                  */
00877                 fSet(ncp->flags, NC_NSYNC);
00878         }
00879 
00880         status = nc_get_NC(ncp);
00881         if(status != NC_NOERR)
00882                 goto unwind_ioc;
00883 
00884         add_to_NCList(ncp);
00885 
00886         if(chunksizehintp != NULL)
00887                 *chunksizehintp = ncp->chunk;
00888         *ncid_ptr = ncp->nciop->fd;
00889         return NC_NOERR;
00890 
00891 unwind_ioc:
00892         (void) ncio_close(ncp->nciop, 0);
00893         ncp->nciop = NULL;
00894         /*FALLTHRU*/
00895 unwind_alloc:
00896         free_NC(ncp);
00897         return status;
00898 }
00899 
00900 int
00901 nc_open(const char * path, int ioflags, int *ncid_ptr)
00902 {
00903         return nc__open(path, ioflags, NULL, ncid_ptr);
00904 }
00905 
00906 
00907 int
00908 nc__enddef(int ncid,
00909         size_t h_minfree, size_t v_align,
00910         size_t v_minfree, size_t r_align)
00911 {
00912         int status;
00913         NC *ncp;
00914 
00915         status = NC_check_id(ncid, &ncp); 
00916         if(status != NC_NOERR)
00917                 return status;
00918 
00919         if(!NC_indef(ncp))
00920                 return(NC_ENOTINDEFINE);
00921 
00922         return (NC_endef(ncp, h_minfree, v_align, v_minfree, r_align));
00923 }
00924 
00925 int
00926 nc_enddef(int ncid)
00927 {
00928         int status;
00929         NC *ncp;
00930 
00931         status = NC_check_id(ncid, &ncp); 
00932         if(status != NC_NOERR)
00933                 return status;
00934 
00935         if(!NC_indef(ncp))
00936                 return(NC_ENOTINDEFINE);
00937 
00938         /* return(NC_endef(ncp, 0, 4096, 0, 4096)); */
00939         return (NC_endef(ncp, 0, 1, 0, 1));
00940 }
00941 
00942 
00943 int
00944 nc_close(int ncid)
00945 {
00946         int status = NC_NOERR;
00947         NC *ncp; 
00948 
00949         status = NC_check_id(ncid, &ncp); 
00950         if(status != NC_NOERR)
00951                 return status;
00952 
00953         if(NC_indef(ncp))
00954         {
00955                 status = NC_endef(ncp, 0, 1, 0, 1); /* TODO: defaults */
00956                 if(status != NC_NOERR )
00957                 {
00958                         (void) nc_abort(ncid);
00959                         return status;
00960                 }
00961         }
00962         else if(!NC_readonly(ncp))
00963         {
00964                 status = NC_sync(ncp);
00965         }
00966 
00967         (void) ncio_close(ncp->nciop, 0);
00968         ncp->nciop = NULL;
00969 
00970         del_from_NCList(ncp);
00971 
00972         free_NC(ncp);
00973 
00974         return status;
00975 }
00976 
00977 
00978 int
00979 nc_delete(const char * path)
00980 {
00981         return nc_delete_mp(path, 0);
00982 }
00983 
00984 int
00985 nc_delete_mp(const char * path, int basepe)
00986 {
00987         NC *ncp;
00988         int status;
00989         size_t chunk = 512;
00990 
00991         ncp = new_NC(&chunk);
00992         if(ncp == NULL)
00993                 return NC_ENOMEM;
00994         
00995 #if defined(LOCKNUMREC) /* && _CRAYMPP */
00996         if (status = NC_init_pe(ncp, basepe)) {
00997                 return status;
00998         }
00999 #else
01000         /*
01001          * !_CRAYMPP, only pe 0 is valid
01002          */
01003         if(basepe != 0)
01004                 return NC_EINVAL;
01005 #endif
01006         status = ncio_open(path, NC_NOWRITE,
01007                 0, 0, &ncp->chunk,
01008                 &ncp->nciop, 0);
01009         if(status)
01010                 goto unwind_alloc;
01011 
01012         assert(ncp->flags == 0);
01013 
01014         status = nc_get_NC(ncp);
01015         if(status != NC_NOERR)
01016         {
01017                 /* Not a netcdf file, don't delete */
01018                 /* ??? is this the right semantic? what if it was just too big? */
01019                 (void) ncio_close(ncp->nciop, 0);
01020         }
01021         else
01022         {
01023                 /* ncio_close does the unlink */
01024                 status = ncio_close(ncp->nciop, 1); /* ncio_close does the unlink */
01025         }
01026 
01027         ncp->nciop = NULL;
01028 unwind_alloc:
01029         free_NC(ncp);
01030         return status;
01031 }
01032 
01033 
01034 /*
01035  * In data mode, same as ncclose.
01036  * In define mode, restore previous definition.
01037  * In create, remove the file.
01038  */
01039 int
01040 nc_abort(int ncid)
01041 {
01042         int status;
01043         NC *ncp;
01044         int doUnlink = 0;
01045 
01046         status = NC_check_id(ncid, &ncp); 
01047         if(status != NC_NOERR)
01048                 return status;
01049 
01050         doUnlink = NC_IsNew(ncp);
01051 
01052         if(ncp->old != NULL)
01053         {
01054                 /* a plain redef, not a create */
01055                 assert(!NC_IsNew(ncp));
01056                 assert(fIsSet(ncp->flags, NC_INDEF));
01057                 free_NC(ncp->old);
01058                 ncp->old = NULL;
01059                 fClr(ncp->flags, NC_INDEF);
01060         }
01061         else if(!NC_readonly(ncp))
01062         {
01063                 status = NC_sync(ncp);
01064                 if(status != NC_NOERR)
01065                         return status;
01066         }
01067 
01068 
01069         (void) ncio_close(ncp->nciop, doUnlink);
01070         ncp->nciop = NULL;
01071 
01072         del_from_NCList(ncp);
01073 
01074         free_NC(ncp);
01075 
01076         return NC_NOERR;
01077 }
01078 
01079 
01080 int
01081 nc_redef(int ncid)
01082 {
01083         int status;
01084         NC *ncp;
01085 
01086         status = NC_check_id(ncid, &ncp); 
01087         if(status != NC_NOERR)
01088                 return status;
01089 
01090         if(NC_readonly(ncp))
01091                 return NC_EPERM;
01092 
01093         if(NC_indef(ncp))
01094                 return NC_EINDEFINE;
01095 
01096         
01097         if(fIsSet(ncp->nciop->ioflags, NC_SHARE))
01098         {
01099                 /* read in from disk */
01100                 status = NC_sync(ncp);
01101                 if(status != NC_NOERR)
01102                         return status;
01103         }
01104 
01105         ncp->old = dup_NC(ncp);
01106         if(ncp->old == NULL)
01107                 return NC_ENOMEM;
01108 
01109         fSet(ncp->flags, NC_INDEF);
01110 
01111         return NC_NOERR;
01112 }
01113 
01114 
01115 int
01116 nc_inq(int ncid,
01117         int *ndimsp,
01118         int *nvarsp,
01119         int *nattsp,
01120         int *xtendimp)
01121 {
01122         int status;
01123         NC *ncp;
01124 
01125         status = NC_check_id(ncid, &ncp); 
01126         if(status != NC_NOERR)
01127                 return status;
01128 
01129         if(ndimsp != NULL)
01130                 *ndimsp = (int) ncp->dims.nelems;
01131         if(nvarsp != NULL)
01132                 *nvarsp = (int) ncp->vars.nelems;
01133         if(nattsp != NULL)
01134                 *nattsp = (int) ncp->attrs.nelems;
01135         if(xtendimp != NULL)
01136                 *xtendimp = find_NC_Udim(&ncp->dims, NULL);
01137 
01138         return NC_NOERR;
01139 }
01140 
01141 int 
01142 nc_inq_ndims(int ncid, int *ndimsp)
01143 {
01144         int status;
01145         NC *ncp;
01146 
01147         status = NC_check_id(ncid, &ncp); 
01148         if(status != NC_NOERR)
01149                 return status;
01150 
01151         if(ndimsp != NULL)
01152                 *ndimsp = (int) ncp->dims.nelems;
01153 
01154         return NC_NOERR;
01155 }
01156 
01157 int 
01158 nc_inq_nvars(int ncid, int *nvarsp)
01159 {
01160         int status;
01161         NC *ncp;
01162 
01163         status = NC_check_id(ncid, &ncp); 
01164         if(status != NC_NOERR)
01165                 return status;
01166 
01167         if(nvarsp != NULL)
01168                 *nvarsp = (int) ncp->vars.nelems;
01169 
01170         return NC_NOERR;
01171 }
01172 
01173 int 
01174 nc_inq_natts(int ncid, int *nattsp)
01175 {
01176         int status;
01177         NC *ncp;
01178 
01179         status = NC_check_id(ncid, &ncp); 
01180         if(status != NC_NOERR)
01181                 return status;
01182 
01183         if(nattsp != NULL)
01184                 *nattsp = (int) ncp->attrs.nelems;
01185 
01186         return NC_NOERR;
01187 }
01188 
01189 int 
01190 nc_inq_unlimdim(int ncid, int *xtendimp)
01191 {
01192         int status;
01193         NC *ncp;
01194 
01195         status = NC_check_id(ncid, &ncp); 
01196         if(status != NC_NOERR)
01197                 return status;
01198 
01199         if(xtendimp != NULL)
01200                 *xtendimp = find_NC_Udim(&ncp->dims, NULL);
01201 
01202         return NC_NOERR;
01203 }
01204 
01205 
01206 int
01207 nc_sync(int ncid)
01208 {
01209         int status;
01210         NC *ncp;
01211 
01212         status = NC_check_id(ncid, &ncp); 
01213         if(status != NC_NOERR)
01214                 return status;
01215 
01216         if(NC_indef(ncp))
01217                 return NC_EINDEFINE;
01218 
01219         if(NC_readonly(ncp))
01220         {
01221                 return read_NC(ncp);
01222         }
01223         /* else, read/write */
01224 
01225         status = NC_sync(ncp);
01226         if(status != NC_NOERR)
01227                 return status;
01228 
01229         return ncp->nciop->sync(ncp->nciop);
01230 }
01231 
01232 
01233 int
01234 nc_set_fill(int ncid,
01235         int fillmode, int *old_mode_ptr)
01236 {
01237         int status;
01238         NC *ncp;
01239         int oldmode;
01240 
01241         status = NC_check_id(ncid, &ncp); 
01242         if(status != NC_NOERR)
01243                 return status;
01244 
01245         if(NC_readonly(ncp))
01246                 return NC_EPERM;
01247 
01248         oldmode = fIsSet(ncp->flags, NC_NOFILL) ? NC_NOFILL : NC_FILL;
01249 
01250         if(fillmode == NC_NOFILL)
01251         {
01252                 fSet(ncp->flags, NC_NOFILL);
01253         }
01254         else if(fillmode == NC_FILL)
01255         {
01256                 if(fIsSet(ncp->flags, NC_NOFILL))
01257                 {
01258                         /*
01259                          * We are changing back to fill mode
01260                          * so do a sync
01261                          */
01262                         status = NC_sync(ncp);
01263                         if(status != NC_NOERR)
01264                                 return status;
01265                 }
01266                 fClr(ncp->flags, NC_NOFILL);
01267         }
01268         else
01269         {
01270                 return NC_EINVAL; /* Invalid fillmode */
01271         }
01272 
01273         if(old_mode_ptr != NULL)
01274                 *old_mode_ptr = oldmode;
01275 
01276         return NC_NOERR;
01277 }
01278 
01279 #ifdef LOCKNUMREC
01280 
01281 /* create function versions of the NC_*_numrecs macros */
01282 size_t NC_get_numrecs(const NC *ncp) {
01283         shmem_t numrec;
01284         shmem_short_get(&numrec, (shmem_t *) ncp->lock + LOCKNUMREC_VALUE, 1,
01285                 ncp->lock[LOCKNUMREC_BASEPE]);
01286         return (size_t) numrec;
01287 }
01288 
01289 void NC_set_numrecs(NC *ncp, size_t nrecs) {
01290         shmem_t numrec = (shmem_t) nrecs;
01291         /* update local value too */
01292         ncp->lock[LOCKNUMREC_VALUE] = (ushmem_t) numrec;
01293         shmem_short_put((shmem_t *) ncp->lock + LOCKNUMREC_VALUE, &numrec, 1,
01294                 ncp->lock[LOCKNUMREC_BASEPE]);
01295 }
01296 
01297 void NC_increase_numrecs(NC *ncp, size_t nrecs) {
01298         /* this is only called in one place that's already protected
01299          * by a lock ... so don't worry about it */
01300         if (nrecs > NC_get_numrecs(ncp))
01301                 NC_set_numrecs(ncp, nrecs);
01302 }
01303 
01304 #endif /* LOCKNUMREC */
01305 
01306 /* everyone in communicator group will be executing this */
01307 /*ARGSUSED*/
01308 int
01309 nc_set_base_pe(int ncid, int pe)
01310 {
01311 #if _CRAYMPP && defined(LOCKNUMREC)
01312         int status;
01313         NC *ncp;
01314         shmem_t numrecs;
01315 
01316         if ((status = NC_check_id(ncid, &ncp)) != NC_NOERR) {
01317                 return status;
01318         }
01319         if (pe < 0 || pe >= _num_pes()) {
01320                 return NC_EINVAL; /* invalid base pe */
01321         }
01322 
01323         numrecs = (shmem_t) NC_get_numrecs(ncp);
01324 
01325         ncp->lock[LOCKNUMREC_VALUE] = (ushmem_t) numrecs;
01326 
01327         /* update serving & lock values for a "smooth" transition */
01328         /* note that the "real" server will being doing this as well */
01329         /* as all the rest in the group */
01330         /* must have syncronization before & after this step */
01331         shmem_short_get(
01332                 (shmem_t *) ncp->lock + LOCKNUMREC_SERVING,
01333                 (shmem_t *) ncp->lock + LOCKNUMREC_SERVING,
01334                 1, ncp->lock[LOCKNUMREC_BASEPE]);
01335 
01336         shmem_short_get(
01337                 (shmem_t *) ncp->lock + LOCKNUMREC_LOCK,
01338                 (shmem_t *) ncp->lock + LOCKNUMREC_LOCK,
01339                 1, ncp->lock[LOCKNUMREC_BASEPE]);
01340 
01341         /* complete transition */
01342         ncp->lock[LOCKNUMREC_BASEPE] = (ushmem_t) pe;
01343 
01344 #endif /* _CRAYMPP && LOCKNUMREC */
01345         return NC_NOERR;
01346 }
01347 
01348 /*ARGSUSED*/
01349 int
01350 nc_inq_base_pe(int ncid, int *pe)
01351 {
01352 #if _CRAYMPP && defined(LOCKNUMREC)
01353         int status;
01354         NC *ncp;
01355 
01356         if ((status = NC_check_id(ncid, &ncp)) != NC_NOERR) {
01357                 return status;
01358         }
01359 
01360         *pe = (int) ncp->lock[LOCKNUMREC_BASEPE];
01361 #else
01362         /*
01363          * !_CRAYMPP, only pe 0 is valid
01364          */
01365         *pe = 0;
01366 #endif /* _CRAYMPP && LOCKNUMREC */
01367         return NC_NOERR;
01368 }