IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 33475


Ignore:
Timestamp:
Mar 10, 2012, 2:56:07 PM (14 years ago)
Author:
eugene
Message:

remove 2nd library link to various ohana libs; create HostTableWaitJobsGetIO function & use in high-speed

Location:
branches/eam_branches/ipp-20111122/Ohana/src
Files:
6 edited

Legend:

Unmodified
Added
Removed

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

    r33454 r33475  
    242242} SkyTableDataFlags;
    243243
     244typedef enum {
     245  HOST_STDIN = 0,
     246  HOST_STDOUT = 1,
     247  HOST_STDERR = 2,
     248} HostInfoIOfd;
     249
    244250typedef struct {
    245251  char *hostname;             // name of remote machine
    … …  
    249255  int pid;                    // remote process ID
    250256  int status;                 // job exit status
     257  IOBuffer stdout;
     258  IOBuffer stderr;
    251259} HostInfo;
    252260
    … …  
    614622HostTable    *HostTableLoad (char *catdir, char *rootname);
    615623int HostTableWaitJobs (HostTable *table, char *file, int lineno);
     624int HostTableWaitJobsGetIO (HostTable *table, char *file, int lineno);
    616625
    617626// functions to support tiny versions of Average and Measure

  • branches/eam_branches/ipp-20111122/Ohana/src/libdvo/src/HostTable.c

    r33474 r33475  
    99    hosts[i].hostname = NULL;
    1010    hosts[i].pathname = NULL;
    11     hosts[i].stdio[0] = -1;
    12     hosts[i].stdio[1] = -1;
    13     hosts[i].stdio[2] = -1;
     11    hosts[i].stdio[HOST_STDIN] = -1;
     12    hosts[i].stdio[HOST_STDOUT] = -1;
     13    hosts[i].stdio[HOST_STDERR] = -1;
    1414    hosts[i].pid = 0;
    1515  }
    … …  
    9191    hosts[Nhosts].pathname = strcreate(tmppath);
    9292   
     93    InitIOBuffer (&hosts[Nhosts].stdout, 1000);
     94    InitIOBuffer (&hosts[Nhosts].stderr, 1000);
     95
    9396    Nhosts ++;
    9497    if (Nhosts >= NHOSTS) {
    … …  
    146149      }
    147150    }
     151
     152    // when the host has finished, close the open sockets
    148153
    149154    // find the host which has finished
    … …  
    158163      IOBuffer buffer;
    159164      InitIOBuffer (&buffer, 100);
    160       EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[1]);
     165      EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[HOST_STDOUT]);
    161166      fprintf (stderr, "--- stdout from %s ---\n", table->hosts[j].hostname);
    162167      write (STDOUT_FILENO, buffer.buffer, buffer.Nbuffer);
    … …  
    164169         
    165170      InitIOBuffer (&buffer, 100);
    166       EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[2]);
     171      EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[HOST_STDERR]);
    167172      fprintf (stderr, "--- stderr from %s ---\n", table->hosts[j].hostname);
    168173      write (STDOUT_FILENO, buffer.buffer, buffer.Nbuffer);
    … …  
    192197int HostTableWaitJobsGetIO (HostTable *table, char *file, int lineno) {
    193198
    194   int i;
    195 
    196199  // we have launched table->Nhosts jobs; wait for all of them to complete...
    197200  // if one (N) failed to launch, we will get an ECHILD error from the last (N) calls
    … …  
    201204  // timeouts for both
    202205
     206  // add all hosts' sockets to the fd_sets
     207  fd_set rdSet, wtSet;
     208  FD_ZERO (&rdSet);
     209  FD_ZERO (&wtSet);
     210
     211  // XXX can I set the fd_sets once, since I am not actually closing the fd's?
     212
     213  int i;
     214  int Nmax = 0;
     215  for (i = 0; i < table->Nhosts; i++) {
     216    FD_SET (table->hosts[i].stdio[HOST_STDIN], &wtSet);
     217    Nmax = MAX (Nmax, table->hosts[i].stdio[HOST_STDIN]);
     218    FD_SET (table->hosts[i].stdio[HOST_STDOUT], &rdSet);
     219    Nmax = MAX (Nmax, table->hosts[i].stdio[HOST_STDOUT]);
     220    FD_SET (table->hosts[i].stdio[HOST_STDERR], &rdSet);
     221    Nmax = MAX (Nmax, table->hosts[i].stdio[HOST_STDERR]);
     222  }   
     223  Nmax ++;
     224
     225  int Nfound = 0;
     226
     227  // this loop has 2 chunks: (a) check for I/O + (b) check for jobs done
    203228  while (1) {
    204229
    205     // is anyone done?
    206     int status = 0;
    207 
    208     // XXX we'll need to pass in WNOHANG and keep retrying if we want to have a timeout...
    209     int pid = waitpid (-1, &status, WNOHANG);
    210     if (pid) {
    211       if (pid == -1) {
    212         switch (errno) {
    213           case ECHILD:
    214             done = TRUE;
    215             // no more children, need to exit ...
    216             break;
    217           default:
    218             fprintf (stderr, "programming error (2)? %s %d", file, lineno);
    219             exit (2);
     230    // Wait up to 0.5 second for host to provide I/O
     231    // timeout gets mucked: need to reset before each select
     232    struct timeval timeout;
     233    timeout.tv_sec = 10;
     234    timeout.tv_usec = 500000;
     235
     236    int status = select (Nmax, &rdSet, &wtSet, NULL, &timeout);
     237    if (status == -1) {
     238      perror("select()");
     239      exit (2);
     240    }
     241
     242    // we have some sockets to check, check sockets for all hosts
     243    for (i = 0; (status > 0) && (i < table->Nhosts); i++) {
     244      if (FALSE && FD_ISSET (table->hosts[i].stdio[HOST_STDIN], &wtSet)) {
     245        // this host is waiting for input : this is an error, so exit
     246        fprintf (stderr, "host %s is waiting for input\n", table->hosts[i].hostname);
     247        abort();
     248      }
     249     
     250      if (FD_ISSET (table->hosts[i].stdio[HOST_STDOUT], &rdSet)) {
     251        // this host has waiting output : read to buffer, and dump if necessary
     252        ReadtoIOBuffer (&table->hosts[i].stdout, table->hosts[i].stdio[HOST_STDOUT]);
     253        if (table->hosts[i].stdout.Nbuffer > 0x10000) {
     254          fprintf (stdout, "--- stdout from %s ---\n", table->hosts[i].hostname);
     255          write (STDOUT_FILENO, table->hosts[i].stdout.buffer, table->hosts[i].stdout.Nbuffer);
     256          fprintf (stdout, "\n");
    220257        }
    221       } else {
    222         // handle
    223      
    224 
    225 
    226     // find the host which has finished
    227     int j;
    228     int found = FALSE;
    229     for (j = 0; j < table->Nhosts; j++) {
    230       if (table->hosts[j].pid != pid) continue;
    231       found = TRUE;
    232       // check on the status of this and report any output?
    233       fprintf (stderr, "job finished for %s (%d)\n", table->hosts[j].hostname, pid);
    234       // read the stderr and stdout
    235       IOBuffer buffer;
    236       InitIOBuffer (&buffer, 100);
    237       EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[1]);
    238       fprintf (stderr, "--- stdout from %s ---\n", table->hosts[j].hostname);
    239       write (STDOUT_FILENO, buffer.buffer, buffer.Nbuffer);
    240       fprintf (stderr, "\n");
    241          
    242       InitIOBuffer (&buffer, 100);
    243       EmptyIOBuffer (&buffer, 100, table->hosts[j].stdio[2]);
    244       fprintf (stderr, "--- stderr from %s ---\n", table->hosts[j].hostname);
    245       write (STDOUT_FILENO, buffer.buffer, buffer.Nbuffer);
    246       fprintf (stderr, "\n");
    247       if (WIFEXITED(status)) {
    248         fprintf (stderr, "normal completion, exit status is %d\n", WEXITSTATUS(status));
    249         table->hosts[j].status = WEXITSTATUS(status);
    250         if (table->hosts[j].status) {
    251           fprintf (stderr, "job failed on %s\n", table->hosts[j].hostname);
     258      }
     259
     260      if (FD_ISSET (table->hosts[i].stdio[HOST_STDERR], &rdSet)) {
     261        // this host has waiting output : read to buffer, and dump if necessary
     262        ReadtoIOBuffer (&table->hosts[i].stderr, table->hosts[i].stdio[HOST_STDERR]);
     263        if (table->hosts[i].stderr.Nbuffer > 0x10000) {
     264          fprintf (stdout, "--- stderr from %s ---\n", table->hosts[i].hostname);
     265          write (STDOUT_FILENO, table->hosts[i].stderr.buffer, table->hosts[i].stderr.Nbuffer);
     266          fprintf (stdout, "\n");
     267        }
     268      }
     269    }
     270
     271    // now check if any children have finished...
     272    while (TRUE) {
     273      int status = 0;
     274      int pid = waitpid (-1, &status, WNOHANG);
     275      if (!pid) {
     276        break; // no outstanding jobs have finished
     277        fprintf (stderr, "no hosts to harvest\n");
     278        usleep (500000);
     279      }
     280      if ((pid == -1) && (errno == ECHILD)) goto escape; // no more jobs on which to wait
     281      if ((pid == -1) && (errno != ECHILD)) {
     282        fprintf (stderr, "programming error (2)? %s %d", file, lineno);
     283        exit (2);
     284      }
     285
     286      // find the host which has finished
     287      int j;
     288      int found = FALSE;
     289      for (j = 0; (j < table->Nhosts) && !found; j++) {
     290        if (table->hosts[j].pid != pid) continue;
     291        found = TRUE;
     292
     293        // check on the status of this and report any output?
     294        fprintf (stdout, "job finished for %s (%d)\n", table->hosts[j].hostname, pid);
     295
     296        // read stdout
     297        EmptyIOBuffer (&table->hosts[j].stdout, 100, table->hosts[j].stdio[HOST_STDOUT]);
     298        fprintf (stdout, "--- stdout from %s ---\n", table->hosts[j].hostname);
     299        write (STDOUT_FILENO, table->hosts[j].stdout.buffer, table->hosts[j].stdout.Nbuffer);
     300        fprintf (stdout, "\n");
     301           
     302        // read stderr
     303        EmptyIOBuffer (&table->hosts[j].stderr, 100, table->hosts[j].stdio[HOST_STDERR]);
     304        fprintf (stdout, "--- stderr from %s ---\n", table->hosts[j].hostname);
     305        write (STDOUT_FILENO, table->hosts[j].stderr.buffer, table->hosts[j].stderr.Nbuffer);
     306        fprintf (stdout, "\n");
     307
     308        if (WIFEXITED(status)) {
     309          fprintf (stdout, "normal completion, exit status is %d\n", WEXITSTATUS(status));
     310          table->hosts[j].status = WEXITSTATUS(status);
     311          if (table->hosts[j].status) {
     312            fprintf (stdout, "job failed on %s\n", table->hosts[j].hostname);
     313          }
     314        } else {
     315          table->hosts[j].status = -1;
     316          fprintf (stdout, "job exited abnormally on %s\n", table->hosts[j].hostname);
    252317          continue;
    253318        }
    254       } else {
    255         table->hosts[j].status = -1;
    256         fprintf (stderr, "job exited abnormally on %s\n", table->hosts[j].hostname);
    257         continue;
    258       }
    259     }
    260     if (!found) {
    261       fprintf (stderr, "Programming error: failed to matched finished job to known host!\n");
    262       exit (2);
    263     }
    264   }
     319      }
     320      if (!found) {
     321        fprintf (stderr, "Programming error: failed to matched finished job to known host!\n");
     322        exit (2);
     323      }
     324      Nfound ++;
     325      if (Nfound == table->Nhosts) goto escape; // we've harvested all jobs
     326    }
     327  }
     328
     329escape:
    265330  return TRUE;
    266331}

  • branches/eam_branches/ipp-20111122/Ohana/src/relastro/Makefile

    r33448 r33475  
    7171#$(SRC)/write_coords.$(ARCH).o        \
    7272
    73 $(RELASTRO): $(INC)/relastro.h $(KAPA_INCS)
    74 $(BIN)/relastro.$(ARCH): $(RELASTRO) $(KAPA_LIBS)
     73$(RELASTRO): $(INC)/relastro.h
     74$(BIN)/relastro.$(ARCH): $(RELASTRO)
    7575
    7676RELASTRO_CLIENT = \
    … …  
    122122#$(SRC)/StarMaps.$(ARCH).o    \
    123123
    124 $(RELASTRO_CLIENT): $(INC)/relastro.h $(KAPA_INCS)
    125 $(BIN)/relastro_client.$(ARCH): $(RELASTRO_CLIENT) $(KAPA_LIBS)
     124$(RELASTRO_CLIENT): $(INC)/relastro.h
     125$(BIN)/relastro_client.$(ARCH): $(RELASTRO_CLIENT)
    126126
    127127testparallax: $(BIN)/testparallax.$(ARCH)
    … …  
    134134$(SRC)/testparallax.$(ARCH).o
    135135
    136 $(TESTPAR): $(INC)/relastro.h $(KAPA_INCS)
    137 $(BIN)/testparallax.$(ARCH): $(TESTPAR) $(KAPA_LIBS)
     136$(TESTPAR): $(INC)/relastro.h
     137$(BIN)/testparallax.$(ARCH): $(TESTPAR)

  • branches/eam_branches/ipp-20111122/Ohana/src/relastro/include/relastro.h

    r33448 r33475  
    418418int BrightCatalogSplit (CatalogSplitter *catalogs, BrightCatalog *bcatalog);
    419419int BrightCatalogSplitFree (CatalogSplitter *catalogs);
     420
     421PhotCode **ParsePhotcodeList (char *rawlist, int *nphotcodes);

  • branches/eam_branches/ipp-20111122/Ohana/src/relastro/src/high_speed_catalogs.c

    r33473 r33475  
    161161  }
    162162  if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
    163     HostTableWaitJobs (table, __FILE__, __LINE__);
     163    HostTableWaitJobsGetIO (table, __FILE__, __LINE__);
    164164  }
    165165

  • branches/eam_branches/ipp-20111122/Ohana/src/relastro/src/initialize.c

    r33473 r33475  
    22
    33void initialize (int argc, char **argv) {
    4 
    5   int NPHOTCODES;
    6   char *codename, *ptr, *list;
    7 
    8   ptr = NULL;
    94
    105  ConfigInit (&argc, argv);
    … …  
    4843void initialize_client (int argc, char **argv) {
    4944
    50   int NPHOTCODES;
    51   char *codename, *ptr, *list;
    52 
    53   ptr = NULL;
    54 
    5545  ConfigInit (&argc, argv);
    5646  args_client (argc, argv);
    … …  
    7363}
    7464
    75 PhotCode *ParsePhotcodeList (char *rawlist, int *nphotcodes) {
     65PhotCode **ParsePhotcodeList (char *rawlist, int *nphotcodes) {
    7666
    7767  *nphotcodes = 0;
    7868  if (!rawlist) return NULL;
    7969
    80   Nphotcodes = 0;
    81   photcodes = NULL;
     70  int Nphotcodes = 0;
    8271  int NPHOTCODES = 10;
     72  PhotCode **photcodes = NULL;
    8373  ALLOCATE (photcodes, PhotCode *, NPHOTCODES);
    8474
Note: See TracChangeset for help on using the changeset viewer.