Changeset 38438

Timestamp:

Jun 12, 2015, 3:32:16 PM (11 years ago)

Author:

eugene

Message:

update dvomerge to limit the number of simultaneous clients; api update to relastro, relphot

Location:

branches/eam_branches/ohana.20150429/src

Files:

• dvomerge/include/dvomerge.h (modified) (1 diff)
• dvomerge/src/args.c (modified) (1 diff)
• dvomerge/src/dvomergeUpdate_catalogs.c (modified) (4 diffs)
• libdvo/include/dvo.h (modified) (1 diff)
• libdvo/src/HostTable.c (modified) (2 diffs)
• relastro/src/UpdateObjectOffsets.c (modified) (1 diff)
• relphot/src/reload_catalogs.c (modified) (1 diff)

branches/eam_branches/ohana.20150429/src/dvomerge/include/dvomerge.h

@@ -41 +41 @@
 int    REPLACE_BY_PHOTCODE;
 int    FORCE_MERGE;
+int    MAX_CLIENTS;
 char  *UPDATE_CATFORMAT;
 char  *UPDATE_CATCOMPRESS;

branches/eam_branches/ohana.20150429/src/dvomerge/src/args.c

@@ -41 +41 @@
   if ((N = get_argument (*argc, argv, "-matched-tables"))) {
     MATCHED_TABLES = TRUE;
+    remove_argument (N, argc, argv);
+  }
+
+  /* limit the impact of a dvomerge -parallel */
+  MAX_CLIENTS = 10;
+  if ((N = get_argument (*argc, argv, "-max-clients"))) {
+    remove_argument (N, argc, argv);
+    MAX_CLIENTS = atoi(argv[N]);
     remove_argument (N, argc, argv);
   }

branches/eam_branches/ohana.20150429/src/dvomerge/src/dvomergeUpdate_catalogs.c

@@ -231 +231 @@
 }
 
+int dvomergeUpdate_parallel_group (HostTableGroup *group, char *absinput, char *absoutput);
+
 // launch the dvomergeUpdate_client jobs to the parallel hosts
 int dvomergeUpdate_parallel (char *input, char *output, SkyTable *outsky, IDmapType *IDmap) {
@@ -257 +259 @@
   }    
 
+  int Ngroups;
+  HostTableGroup *groups = HostTableGroupsMaxNumber (table, &Ngroups, MAX_CLIENTS);
+  // 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 < table->Nhosts; i++) {
+  for (i = 0; i < Ngroups; i++) {
+    // update only a max of MaxClient machines at a time
+    dvomergeUpdate_parallel_group (&groups[i], absinput, absoutput);
+  }
+  return TRUE;
+}
+
+int dvomergeUpdate_parallel_group (HostTableGroup *group, char *absinput, char *absoutput) {
+
+  int i;
+  for (i = 0; i < group->Nhosts; i++) {
 
     // ensure that the paths are absolute path names
-    char *tmppath = abspath (table->hosts[i].pathname, DVO_MAX_PATH);
-    free (table->hosts[i].pathname);
-    table->hosts[i].pathname = tmppath;
+    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, table->hosts[i].hostID, table->hosts[i].pathname, 
+              absinput, absoutput, group->hosts[i][0].hostID, group->hosts[i][0].pathname, 
               UserPatch.Rmin, UserPatch.Rmax, UserPatch.Dmin, UserPatch.Dmax, RADIUS
       );
@@ -294 +310 @@
       int status = system (command);
       if (status) {
-        fprintf (stderr, "ERROR running photdbc_client\n");
+        fprintf (stderr, "ERROR running dvomerge_client\n");
         exit (2);
       }
@@ -300 +316 @@
       // launch the job on the remote machine (no handshake)
       int errorInfo = 0;
-      int pid = rconnect ("ssh", table->hosts[i].hostname, command, table->hosts[i].stdio, &errorInfo, FALSE);
+      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", table->hosts[i].hostname, errorInfo);
+        if (DEBUG) fprintf (stderr, "failure to start %s (error %d)\n", group->hosts[i][0].hostname, errorInfo);
         exit (1);
       }
-      table->hosts[i].pid = pid; // save for future reference
+      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");
+    fprintf (stderr, "run the dvomerge_client commands above.  when these are done, hit return\n");
     getchar();
   }
   if (!PARALLEL_MANUAL && !PARALLEL_SERIAL) {
-    int status = HostTableWaitJobsGetIO (table, __FILE__, __LINE__, VERBOSE);
+    int status = HostTableGroupWaitJobsGetIO (group, __FILE__, __LINE__, VERBOSE);
     if (!status) {
       fprintf (stderr, "error running one of the remote clients\n");

branches/eam_branches/ohana.20150429/src/libdvo/include/dvo.h

@@ -1091 +1091 @@
 void InitHost (HostInfo *host);
 
-HostTableGroup *HostTableGroups (HostTable *table, int *ngroups);
+HostTableGroup *HostTableGroupsUniqueMachines (HostTable *table, int *ngroups);
+HostTableGroup *HostTableGroupsMaxNumber (HostTable *table, int *ngroups, int Nmax);
+
 int HostTableGroupWaitJobsGetIO (HostTableGroup *table, char *file, int lineno, int VERBOSE);
 

branches/eam_branches/ohana.20150429/src/libdvo/src/HostTable.c

@@ -155 +155 @@
 
 // split a host table into Ngroups, each with a unique set of hosts
-HostTableGroup *HostTableGroups (HostTable *table, int *ngroups) {
+HostTableGroup *HostTableGroupsUniqueMachines (HostTable *table, int *ngroups) {
 
   // identify the unique host names and the number of times they each exist
@@ -210 +210 @@
         foundHost[k] = TRUE;
       }
+    }
+  }
+  *ngroups = Ngroups;
+  return groups;
+}
+
+// split a host table into Ngroups, each with a unique set of hosts
+HostTableGroup *HostTableGroupsMaxNumber (HostTable *table, int *ngroups, int Nmax) {
+
+  int i, j;
+
+  int Ngroups = (table->Nhosts % Nmax) ? (int)(table->Nhosts / Nmax + 1) : table->Nhosts / Nmax;
+
+  HostTableGroup *groups = NULL;
+  ALLOCATE (groups, HostTableGroup, Ngroups);
+
+  // in each group, attempt to add one of each unique host
+  int k = 0;
+  for (i = 0; i < Ngroups; i++) {
+    groups[i].Nhosts = 0;
+    ALLOCATE (groups[i].hosts, HostInfo *, Nmax);
+    for (j = 0; j < Nmax; j++) {
+      groups[i].hosts[groups[i].Nhosts] = &table->hosts[k];
+      groups[i].Nhosts ++;
+      k++;
     }
   }

branches/eam_branches/ohana.20150429/src/relastro/src/UpdateObjectOffsets.c

@@ -102 +102 @@
 
   int Ngroups;
-  HostTableGroup *groups = HostTableGroups (table, &Ngroups);
+  HostTableGroup *groups = HostTableGroupsUniqueMachines (table, &Ngroups);
   // split the table into Ngroups, each with a unique set of machines (this avoids overloading the remote host machines)
 

branches/eam_branches/ohana.20150429/src/relphot/src/reload_catalogs.c

@@ -171 +171 @@
 
   int Ngroups;
-  HostTableGroup *groups = HostTableGroups (table, &Ngroups);
+  HostTableGroup *groups = HostTableGroupsUniqueMachines (table, &Ngroups);
   // split the table into Ngroups, each with a unique set of machines (this avoids overloading the remote host machines)