Changeset 38271

Timestamp:

May 13, 2015, 11:37:34 AM (11 years ago)

Author:

eugene

Message:

allow dvomerge to operate on machine groups (one client per machine at a time)

File:

• branches/eam_branches/ipp-20150419/Ohana/src/dvomerge/src/dvomergeUpdate_catalogs.c (modified) (4 diffs)

branches/eam_branches/ipp-20150419/Ohana/src/dvomerge/src/dvomergeUpdate_catalogs.c

@@ -1 +1 @@
 # include "dvomerge.h"
 # define DEBUG 1
+# define MACHINE_GROUPS 1
 
 // in parallel mode, the target database may be distributed, while the input database is
@@ -9 +10 @@
 
 int dvomergeUpdate_parallel (char *input, char *output, SkyTable *outsky, IDmapType *IDmap);
+int dvomergeUpdate_parallel_group (HostTableGroup *group, char *absinput, char *absoutput, SkyTable *sky);
+int dvomergeUpdate_parallel_table (HostTable *table, char *absinput, char *absoutput);
 
 int dvomergeUpdate_catalogs (char *input, char *output, SkyTable *outsky, SkyList *inlist, int NsecfiltInput, int NsecfiltOutput, IDmapType *IDmap, int *secfiltMap) {
@@ -254 +257 @@
   }    
 
+# if (MACHINE_GROUPS)
+  int Ngroups;
+  HostTableGroup *groups = HostTableGroups (table, &Ngroups);
+  // split the table into Ngroups, each with a unique set of machines (this avoids overloading the remote host machines)
+
+  int i;
+  for (i = 0; i < Ngroups; i++) {
+    // update only a group of unique machines at a time
+    dvomergeUpdate_parallel_group (&groups[i], absinput, absoutput, outsky);
+  }
+# else
+  dvomergeUpdate_parallel_table (table, absinput, absoutput);
+# endif
+
+  return TRUE;
+}
+
+// launch the dvomergeUpdate_client jobs to the parallel hosts
+int dvomergeUpdate_parallel_table (HostTable *table, char *absinput, char *absoutput) {
+
   int i;
   for (i = 0; i < table->Nhosts; i++) {
@@ -320 +343 @@
   return TRUE;
 }      
+
+// launch the dvomergeUpdate_client jobs to the parallel hosts
+int dvomergeUpdate_parallel_group (HostTableGroup *group, char *absinput, char *absoutput, SkyTable *sky) {
+
+  int i, j;
+  for (i = 0; i < group->Nhosts; i++) {
+
+    if (sky->Nregions < group->Nhosts) {
+      // do any of the regions want this host?
+      int wantThisHost = FALSE;
+      for (j = 0; j < sky->Nregions; j++) {
+        if (HostTableTestHost (&sky->regions[j], group->hosts[i][0].hostID)) {
+          wantThisHost = TRUE;
+          break;
+        }
+      }
+      if (!wantThisHost) {
+        // fprintf (stderr, "skip host %s\n", group->hosts[i][0].hostname);
+        continue;
+      }
+    }
+
+    // ensure that the paths are absolute path names
+    char *tmppath = abspath (group->hosts[i][0].pathname, DVO_MAX_PATH);
+    free (group->hosts[i][0].pathname);
+    group->hosts[i][0].pathname = tmppath;
+
+    // options / arguments that can affect relastro_client -update-objects:
+    char command[DVO_MAX_PATH];
+    snprintf (command, DVO_MAX_PATH, "dvomerge_client %s into %s -hostID %d -hostdir %s -region %f %f %f %f -D ADDSTAR_RADIUS %f", 
+              absinput, absoutput, group->hosts[i][0].hostID, group->hosts[i][0].pathname, 
+              UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax, RADIUS
+      );
+
+    char tmpline[DVO_MAX_PATH];
+    if (VERBOSE)             { snprintf (tmpline, DVO_MAX_PATH, "%s -v",                command); strcpy (command, tmpline); }
+    if (VERIFY)              { snprintf (tmpline, DVO_MAX_PATH, "%s -verify",           command); strcpy (command, tmpline); }
+    if (VERIFY_CATALOG_ONLY) { snprintf (tmpline, DVO_MAX_PATH, "%s -verify-catalogs",  command); strcpy (command, tmpline); }
+    if (REPLACE_BY_PHOTCODE) { snprintf (tmpline, DVO_MAX_PATH, "%s -replace",          command); strcpy (command, tmpline); }
+    if (PARALLEL_INPUT)      { snprintf (tmpline, DVO_MAX_PATH, "%s -parallel-input",   command); strcpy (command, tmpline); }
+    if (FORCE_MERGE)         { snprintf (tmpline, DVO_MAX_PATH, "%s -force-merge",      command); strcpy (command, tmpline); }
+    if (MATCHED_TABLES)      { snprintf (tmpline, DVO_MAX_PATH, "%s -matched-tables",   command); strcpy (command, tmpline); }
+    if (UPDATE_CATFORMAT)    { snprintf (tmpline, DVO_MAX_PATH, "%s -update-catformat %s", command, UPDATE_CATFORMAT); strcpy (command, tmpline); }
+
+    // add some config variables:
+    snprintf (tmpline, DVO_MAX_PATH, "%s -D CATMODE %s",    command, CATMODE);   strcpy (command, tmpline);
+    snprintf (tmpline, DVO_MAX_PATH, "%s -D CATFORMAT %s",  command, CATFORMAT); strcpy (command, tmpline);
+    snprintf (tmpline, DVO_MAX_PATH, "%s -D SKY_DEPTH %d",  command, SKY_DEPTH); strcpy (command, tmpline);
+
+    fprintf (stderr, "command: %s\n", command);
+
+    if (PARALLEL_MANUAL) continue;
+
+    if (PARALLEL_SERIAL) {
+      int status = system (command);
+      if (status) {
+        fprintf (stderr, "ERROR running photdbc_client\n");
+        exit (2);
+      }
+    } else {
+      // launch the job on the remote machine (no handshake)
+      int errorInfo = 0;
+      int pid = rconnect ("ssh", group->hosts[i][0].hostname, command, group->hosts[i][0].stdio, &errorInfo, FALSE);
+      if (!pid) {
+        if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", group->hosts[i][0].hostname, errorInfo);
+        exit (1);
+      }
+      group->hosts[i][0].pid = pid; // save for future reference
+    }
+  }
+
+  if (PARALLEL_MANUAL) {
+    fprintf (stderr, "run the photdbc_client commands above.  when these are done, hit return\n");
+    getchar();
+  }
+  if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
+    int status = HostTableGroupWaitJobsGetIO (group, __FILE__, __LINE__, VERBOSE);
+    if (!status) {
+      fprintf (stderr, "error running one of the remote clients\n");
+      return status;
+    }
+  }
+
+  return TRUE;
+}