IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 36562


Ignore:
Timestamp:
Feb 26, 2014, 1:52:04 PM (12 years ago)
Author:
eugene
Message:

relphot -parallel-regions basically works now including for parallel databases

Location:
branches/eam_branches/ipp-20140206/Ohana/src
Files:
7 edited

Legend:

Unmodified
Added
Removed

  • branches/eam_branches/ipp-20140206/Ohana/src/libdvo/include/dvo.h

    r36558 r36562  
    303303  int pid;                    // remote process ID
    304304  int status;
     305  IOBuffer stdout;
     306  IOBuffer stderr;
    305307
    306308  off_t Nimage;
    … …  
    899901RegionHostTable *RegionHostTableLoad (char *catdir, char *rootname);
    900902int RegionHostTableWaitJobs (RegionHostTable *regionHosts, char *file, int lineno);
     903int RegionHostTableWaitJobsGetIO (RegionHostTable *regionHosts, char *file, int lineno, int VERBOSE);
    901904
    902905# endif // DVO_H

  • branches/eam_branches/ipp-20140206/Ohana/src/libdvo/src/RegionHostTable.c

    r36558 r36562  
    122122    hosts[Nhosts].hostID = ID;
    123123    hosts[Nhosts].hostname = strcreate(tmphost);
     124
     125    InitIOBuffer (&hosts[Nhosts].stdout, 1000);
     126    InitIOBuffer (&hosts[Nhosts].stderr, 1000);
    124127
    125128    hosts[Nhosts].Rmin = Rmin;
    … …  
    169172
    170173// wait for all children to complete, report output to stdout
    171 int RegionHostTableWaitJobs (RegionHostTable *regionHosts, char *file, int lineno) {
     174int RegionHostTableWaitJobs (RegionHostTable *table, char *file, int lineno) {
    172175
    173176  int i;
    174177
    175   // we have launched regionHosts->Nhosts jobs; wait for all of them to complete...
     178  // we have launched table->Nhosts jobs; wait for all of them to complete...
    176179  // if one (N) failed to launch, we will get an ECHILD error from the last (N) calls
    177180  int done = FALSE;
    178   for (i = 0; !done  && (i < regionHosts->Nhosts); i++) {
     181  for (i = 0; !done  && (i < table->Nhosts); i++) {
    179182    int status = 0;
    180183
    … …  
    202205    int Nout, j;
    203206    int found = FALSE;
    204     for (j = 0; j < regionHosts->Nhosts; j++) {
    205       if (regionHosts->hosts[j].pid != pid) continue;
     207    for (j = 0; j < table->Nhosts; j++) {
     208      if (table->hosts[j].pid != pid) continue;
    206209      found = TRUE;
    207210      // check on the status of this and report any output?
    208       fprintf (stderr, "job finished for %s (%d)\n", regionHosts->hosts[j].hostname, pid);
     211      fprintf (stderr, "job finished for %s (%d)\n", table->hosts[j].hostname, pid);
    209212      // read the stderr and stdout
    210213      IOBuffer buffer;
    211214      InitIOBuffer (&buffer, 100);
    212       EmptyIOBuffer (&buffer, 100, regionHosts->hosts[j].stdio[HOST_STDOUT]);
    213       fprintf (stderr, "--- stdout from %s ---\n", regionHosts->hosts[j].hostname);
     215      EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[HOST_STDOUT]);
     216      fprintf (stderr, "--- stdout from %s ---\n", table->hosts[j].hostname);
    214217      Nout = write (STDOUT_FILENO, buffer.buffer, buffer.Nbuffer);
    215218      if (Nout != buffer.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
    … …  
    217220         
    218221      InitIOBuffer (&buffer, 100);
    219       EmptyIOBuffer (&buffer, 100, regionHosts->hosts[j].stdio[HOST_STDERR]);
    220       fprintf (stderr, "--- stderr from %s ---\n", regionHosts->hosts[j].hostname);
     222      EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[HOST_STDERR]);
     223      fprintf (stderr, "--- stderr from %s ---\n", table->hosts[j].hostname);
    221224      Nout = write (STDOUT_FILENO, buffer.buffer, buffer.Nbuffer);
    222225      if (Nout != buffer.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
    … …  
    224227      if (WIFEXITED(status)) {
    225228        fprintf (stderr, "normal completion, exit status is %d\n", WEXITSTATUS(status));
    226         regionHosts->hosts[j].status = WEXITSTATUS(status);
    227         if (regionHosts->hosts[j].status) {
    228           fprintf (stderr, "job failed on %s\n", regionHosts->hosts[j].hostname);
     229        table->hosts[j].status = WEXITSTATUS(status);
     230        if (table->hosts[j].status) {
     231          fprintf (stderr, "job failed on %s\n", table->hosts[j].hostname);
    229232          continue;
    230233        }
    231234      } else {
    232         regionHosts->hosts[j].status = -1;
    233         fprintf (stderr, "job exited abnormally on %s\n", regionHosts->hosts[j].hostname);
     235        table->hosts[j].status = -1;
     236        fprintf (stderr, "job exited abnormally on %s\n", table->hosts[j].hostname);
    234237        continue;
    235238      }
    … …  
    243246}
    244247
     248
     249// wait for all children to complete, report output to stdout
     250int RegionHostTableWaitJobsGetIO (RegionHostTable *table, char *file, int lineno, int VERBOSE) {
     251
     252  // we have launched table->Nhosts jobs; wait for all of them to complete...
     253  // if one (N) failed to launch, we will get an ECHILD error from the last (N) calls
     254
     255  // we need to read any data waiting on stderr or stdout from these jobs, or the overfull
     256  // buffers can cause a problem.  we alternate between 'select' and 'waitpid' calls with
     257  // timeouts for both
     258
     259  // add all hosts' sockets to the fd_sets
     260  fd_set rdSet, wtSet;
     261  FD_ZERO (&rdSet);
     262  FD_ZERO (&wtSet);
     263
     264  // XXX can I set the fd_sets once, since I am not actually closing the fd's?
     265
     266  int globalStatus = TRUE;
     267
     268  int i;
     269  int Nmax = 0;
     270  for (i = 0; i < table->Nhosts; i++) {
     271    if (!table->hosts[i].pid) continue; // any unconnected hosts should be skipped
     272    FD_SET (table->hosts[i].stdio[HOST_STDIN], &wtSet);
     273    Nmax = MAX (Nmax, table->hosts[i].stdio[HOST_STDIN]);
     274    FD_SET (table->hosts[i].stdio[HOST_STDOUT], &rdSet);
     275    Nmax = MAX (Nmax, table->hosts[i].stdio[HOST_STDOUT]);
     276    FD_SET (table->hosts[i].stdio[HOST_STDERR], &rdSet);
     277    Nmax = MAX (Nmax, table->hosts[i].stdio[HOST_STDERR]);
     278  }   
     279  Nmax ++;
     280
     281  // need the list of connected hosts for exit test below
     282  int Nrunning = 0;
     283  for (i = 0; i < table->Nhosts; i++) {
     284    if (!table->hosts[i].pid) continue; // any unconnected hosts should be skipped
     285    Nrunning ++;
     286  }
     287
     288  int Nfound = 0;
     289
     290  // this loop has 2 chunks: (a) check for I/O + (b) check for jobs done
     291  while (1) {
     292
     293    // Wait up to 0.5 second for host to provide I/O
     294    // timeout gets mucked: need to reset before each select
     295    struct timeval timeout;
     296    timeout.tv_sec = 10;
     297    timeout.tv_usec = 500000;
     298
     299    int status = select (Nmax, NULL, &wtSet, NULL, &timeout);
     300    if (status == -1) {
     301      perror("select()");
     302      exit (2);
     303    }
     304
     305    // we have some sockets to check, check sockets for all hosts
     306    for (i = 0; (status > 0) && (i < table->Nhosts); i++) {
     307      if (!table->hosts[i].pid) continue; // any unconnected hosts should be skipped
     308
     309      if (FALSE && FD_ISSET (table->hosts[i].stdio[HOST_STDIN], &wtSet)) {
     310        // this host is waiting for input : this is an error, so exit
     311        fprintf (stderr, "host %s is waiting for input\n", table->hosts[i].hostname);
     312        abort();
     313      }
     314     
     315      if ((table->hosts[i].stdio[HOST_STDOUT] > 0) && FD_ISSET (table->hosts[i].stdio[HOST_STDOUT], &rdSet)) {
     316        // this host has waiting output : read to buffer, and dump if necessary
     317        ReadtoIOBuffer (&table->hosts[i].stdout, table->hosts[i].stdio[HOST_STDOUT]);
     318        // if (table->hosts[i].stdout.Nbuffer > 0x10000) {
     319        if (table->hosts[i].stdout.Nbuffer > 0x1000) {
     320          int printHead = VERBOSE || (table->hosts[i].stdout.Nbuffer > 0);
     321          if (printHead) fprintf (stdout, "--- stdout from %s --- (%d bytes, v1)\n", table->hosts[i].hostname, table->hosts[i].stdout.Nbuffer);
     322          int Nout = write (STDOUT_FILENO, table->hosts[i].stdout.buffer, table->hosts[i].stdout.Nbuffer);
     323          if (Nout != table->hosts[i].stdout.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
     324          FlushIOBuffer (&table->hosts[i].stdout);
     325          if (printHead) fprintf (stdout, "\n");
     326        }
     327      }
     328
     329      if ((table->hosts[i].stdio[HOST_STDERR] > 0) && FD_ISSET (table->hosts[i].stdio[HOST_STDERR], &rdSet)) {
     330        // this host has waiting output : read to buffer, and dump if necessary
     331        ReadtoIOBuffer (&table->hosts[i].stderr, table->hosts[i].stdio[HOST_STDERR]);
     332        // if (table->hosts[i].stderr.Nbuffer > 0x10000) {
     333        if (table->hosts[i].stderr.Nbuffer > 0x1000) {
     334          int printHead = VERBOSE || (table->hosts[i].stderr.Nbuffer > 0);
     335          if (printHead) fprintf (stdout, "--- stderr from %s --- (%d bytes, v1)\n", table->hosts[i].hostname, table->hosts[i].stderr.Nbuffer);
     336          int Nout = write (STDOUT_FILENO, table->hosts[i].stderr.buffer, table->hosts[i].stderr.Nbuffer);
     337          if (Nout != table->hosts[i].stderr.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
     338          FlushIOBuffer (&table->hosts[i].stderr);
     339          if (printHead) fprintf (stdout, "\n");
     340        }
     341      }
     342    }
     343
     344    // now check if any children have finished...
     345    while (TRUE) {
     346      int status = 0;
     347      int pid = waitpid (-1, &status, WNOHANG);
     348      if (!pid) {
     349        // fprintf (stderr, "no hosts to harvest\n");
     350        usleep (500000);
     351        break; // no outstanding jobs have finished
     352      }
     353      if ((pid == -1) && (errno == ECHILD)) goto escape; // no more jobs on which to wait
     354      if ((pid == -1) && (errno != ECHILD)) {
     355        fprintf (stderr, "programming error (2)? %s %d", file, lineno);
     356        exit (2);
     357      }
     358
     359      // find the host which has finished
     360      int found = FALSE;
     361      for (i = 0; (i < table->Nhosts) && !found; i++) {
     362        if (table->hosts[i].pid != pid) continue;
     363        found = TRUE;
     364
     365        RegionHostInfo *host = &table->hosts[i];
     366
     367        // check on the status of this and report any output?
     368        if (VERBOSE) fprintf (stdout, "job finished for %s (%d)\n", host->hostname, pid);
     369
     370        // read stdout
     371        int printHead;
     372        printHead = VERBOSE || (host->stdout.Nbuffer > 0);
     373        EmptyIOBuffer (&host->stdout, 100, host->stdio[HOST_STDOUT]);
     374        if (printHead) fprintf (stdout, "--- stdout from %s --- (%d bytes, v2)\n", host->hostname, host->stdout.Nbuffer);
     375        int Nout = write (STDOUT_FILENO, host->stdout.buffer, host->stdout.Nbuffer);
     376        if (Nout != host->stdout.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
     377        FlushIOBuffer (&host->stdout);
     378        if (printHead) fprintf (stdout, "\n");
     379           
     380        // read stderr
     381        printHead = VERBOSE || (host->stderr.Nbuffer > 0);
     382        EmptyIOBuffer (&host->stderr, 100, host->stdio[HOST_STDERR]);
     383        if (printHead) fprintf (stdout, "--- stderr from %s --- (%d bytes, v2)\n", host->hostname, host->stderr.Nbuffer);
     384        Nout = write (STDOUT_FILENO, host->stderr.buffer, host->stderr.Nbuffer);
     385        if (Nout != host->stderr.Nbuffer) { fprintf (stderr, "(error writing log?)\n"); }
     386        FlushIOBuffer (&host->stderr);
     387        if (printHead) fprintf (stdout, "\n");
     388
     389        if (WIFEXITED(status)) {
     390          if (VERBOSE) fprintf (stdout, "normal completion, exit status is %d\n", WEXITSTATUS(status));
     391          host->status = WEXITSTATUS(status);
     392          if (host->status) {
     393            fprintf (stdout, "job failed on %s\n", host->hostname);
     394            globalStatus = FALSE;
     395          }
     396        } else {
     397          host->status = -1;
     398          fprintf (stdout, "job exited abnormally on %s\n", host->hostname);
     399          globalStatus = FALSE;
     400          continue;
     401        }
     402      }
     403      if (!found) {
     404        fprintf (stderr, "Programming error: failed to matched finished job to known host!\n");
     405        exit (2);
     406      }
     407      Nfound ++;
     408      if (Nfound == Nrunning) goto escape; // we've harvested all jobs
     409    }
     410  }
     411
     412escape:
     413
     414  // close all opened connections
     415  for (i = 0; i < table->Nhosts; i++) {
     416    if (!table->hosts[i].pid) continue; // any unconnected hosts should be skipped
     417    close (table->hosts[i].stdio[HOST_STDIN]);
     418    close (table->hosts[i].stdio[HOST_STDOUT]);
     419    close (table->hosts[i].stdio[HOST_STDERR]);
     420  }
     421
     422  return globalStatus;
     423}

  • branches/eam_branches/ipp-20140206/Ohana/src/relphot/src/BrightCatalog.c

    r33651 r36562  
    101101      measure[i].catID     = catID[i];
    102102      measure[i].photcode  = photcode[i];
     103      measure[i].myDet     = FALSE;
    103104    }
    104105    fprintf (stderr, "loaded data for %lld measure\n", (long long) Nrow);
    … …  
    141142    GET_COLUMN(flags,         "FLAGS",       int);
    142143    GET_COLUMN(catID,         "CAT_ID",      int);
     144    GET_COLUMN(objID,         "OBJ_ID",      int);
    143145    gfits_free_header (&theader);
    144146    gfits_free_table  (&ftable);
    … …  
    153155      average[i].flags          = flags[i];
    154156      average[i].catID          = catID[i];
     157      average[i].objID          = objID[i];
     158      average[i].nOwn           = 0;
    155159    }
    156160    fprintf (stderr, "loaded data for %lld average\n", (long long) Nrow);
    … …  
    162166    free (flags         );
    163167    free (catID         );
     168    free (objID         );
    164169
    165170    catalog->average = average;
    … …  
    370375    gfits_define_bintable_column (&theader, "J", "FLAGS",       "flags",                  NULL,    1.0, 0.0);
    371376    gfits_define_bintable_column (&theader, "J", "CAT_ID",      "catalog ref",            NULL,    1.0, 0.0);
     377    gfits_define_bintable_column (&theader, "J", "OBJ_ID",      "object ref",             NULL,    1.0, 0.0);
    372378
    373379    // generate the output array that carries the data
    … …  
    381387    int   *flags          ; ALLOCATE (flags,         int,    catalog->Naverage);
    382388    int   *catID          ; ALLOCATE (catID,         int,    catalog->Naverage);
     389    int   *objID          ; ALLOCATE (objID,         int,    catalog->Naverage);
    383390
    384391    // assign the storage arrays
    … …  
    391398      flags[i]          = average[i].flags;
    392399      catID[i]          = average[i].catID;
     400      objID[i]          = average[i].objID;
    393401    }
    394402
    … …  
    400408    gfits_set_bintable_column (&theader, &ftable, "FLAGS",       flags,         catalog->Naverage);
    401409    gfits_set_bintable_column (&theader, &ftable, "CAT_ID",      catID,         catalog->Naverage);
     410    gfits_set_bintable_column (&theader, &ftable, "OBJ_ID",      objID,         catalog->Naverage);
    402411
    403412    free (R             );
    … …  
    407416    free (flags         );
    408417    free (catID         );
     418    free (objID         );
    409419
    410420    gfits_fwrite_Theader (f, &theader);

  • branches/eam_branches/ipp-20140206/Ohana/src/relphot/src/args.c

    r36558 r36562  
    387387    if ((N = get_argument (argc, argv, "-parallel-regions-manual"))) {
    388388      remove_argument (N, &argc, argv);
    389       PARALLEL_REGIONS_MANUAL = FALSE;
     389      PARALLEL_REGIONS_MANUAL = TRUE;
    390390    }
    391391  }

  • branches/eam_branches/ipp-20140206/Ohana/src/relphot/src/client_logger.c

    r33963 r36562  
    1313  int fd = mkstemp (filename);
    1414  if (fd == -1) {
    15     fprintf (stderr, "failed to open client logger, exiting\n");
     15    fprintf (stderr, "failed to open client logger %s, exiting\n", filename);
    1616    exit (50);
    1717  }

  • branches/eam_branches/ipp-20140206/Ohana/src/relphot/src/launch_region_hosts.c

    r36558 r36562  
    9292    getchar();
    9393  } else {
    94     RegionHostTableWaitJobs (regionHosts, __FILE__, __LINE__);
     94    RegionHostTableWaitJobsGetIO (regionHosts, __FILE__, __LINE__, VERBOSE);
    9595  }
    9696 
    97   return TRUE;
     97  int status = TRUE;
     98  for (i = 0; i < regionHosts->Nhosts; i++) {
     99    status = status && (regionHosts->hosts[i].status == 0);
     100  }
     101
     102  return status;
    98103}

  • branches/eam_branches/ipp-20140206/Ohana/src/relphot/src/relphot_parallel_regions.c

    r36550 r36562  
    2121  sprintf (flatcorrfile, "%s/flatcorr.fits", CATDIR);
    2222  FlatCorrectionTable *flatcorr = FlatCorrectionLoad (flatcorrfile, VERBOSE);
    23 
    24   SkyTable *sky = SkyTableLoadOptimal (CATDIR, SKY_TABLE, GSCFILE, TRUE, SKY_DEPTH, VERBOSE);
    25   SkyTableSetFilenames (sky, CATDIR, "cpt");
    26   SkyList *skylist = SkyListByBounds (sky, -1, regionHosts->Rmin, regionHosts->Rmax, regionHosts->Dmin, regionHosts->Dmax);
    2723
    2824  /* register database handle with shutdown procedure */
    … …  
    5450
    5551  /* launch processing on the parallel region hosts */
    56   launch_region_hosts (regionHosts);
     52  if (!launch_region_hosts (regionHosts)) Shutdown ("error launching region hosts");
    5753
    5854  // retrieve updated image parameters from the remote hosts (also set Image.mcal)
    59   slurp_image_mags (regionHosts, -1);
     55  if (!slurp_image_mags (regionHosts, -1)) Shutdown ("error loading image updates");
    6056
    6157  if (!UPDATE) {
    … …  
    6561    exit (0);
    6662  }
     63
     64  SkyTable *sky = SkyTableLoadOptimal (CATDIR, SKY_TABLE, GSCFILE, TRUE, SKY_DEPTH, VERBOSE);
     65  SkyTableSetFilenames (sky, CATDIR, "cpt");
     66  SkyList *skylist = SkyListByBounds (sky, -1, regionHosts->Rmin, regionHosts->Rmax, regionHosts->Dmin, regionHosts->Dmax);
    6767
    6868  /* update catalogs (in parallel) */
Note: See TracChangeset for help on using the changeset viewer.