IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Ignore:
Timestamp:
Dec 10, 2012, 12:07:22 PM (14 years ago)
Author:
Serge CHASTEL
Message:

Pantasks/Condor branch

Location:
branches/sc_branches/pantasks_condor
Files:
1 edited
1 copied

Legend:

Unmodified
Added
Removed

  • branches/sc_branches/pantasks_condor

  • branches/sc_branches/pantasks_condor/pantasks/ControllerOps.c

    r34088 r34783  
    99static IOBuffer stdout_buffer;
    1010static IOBuffer stderr_buffer;
     11#ifndef IPP_CONDOR
    1112static int ControllerPID = 0;
     13#else
     14#include "ipp_condor.h"
     15#include <condor/condor.nsmap>
     16#include <shell.h>
     17static struct soap soap;
     18#endif
    1219
    1320/* local static variables to track the controller host properties */
    … …  
    1825static int NHOSTS = 0;
    1926
     27#ifdef IPP_CONDOR
     28#include "ipp_condor.h"
     29static int hosts_list_modified = TRUE;
     30static char* hosts_requirement = NULL;
     31static int is_host_in_list(char *hostname);
     32#endif
     33
    2034int AddHost (char *hostname, int max_threads) {
    21 
    2235  int N;
    23 
     36#ifdef IPP_CONDOR
     37  int index;
     38  index = is_host_in_list(hostname);
     39  if (index != -1) {
     40      if (hosts[index].max_threads != max_threads) {
     41          fprintf(stderr, "Hostname [%s] already declared. Changing max_threads from %d to %d",
     42                 hostname,
     43                 hosts[index].max_threads,
     44                 max_threads);
     45          hosts[index].max_threads = max_threads;
     46      } else {
     47          fprintf(stderr, "Hostname [%s] already declared. Aborting", hostname);
     48      }
     49      return (TRUE);
     50  }
     51#endif
    2452  N = Nhosts;
    2553  Nhosts ++;
    26 
    2754  CHECK_REALLOCATE (hosts, Host, NHOSTS, Nhosts, 16);
    28 
    2955  hosts[N].hostname = strcreate (hostname);
    3056  hosts[N].max_threads = max_threads;
    31  
     57#ifdef IPP_CONDOR
     58  fprintf(stderr, "Adding host %s (%d)\n", hostname, max_threads);
     59  hosts_list_modified = TRUE; //Tell get_hosts_requirements to rebuild the hosts based requirement
     60#endif
    3261  return (TRUE);
    3362}
    … …  
    3766// but perhaps mark it in the client_thread?
    3867int DeleteHost (char *hostname) {
    39 
    4068  int i, j;
    41 
    42   for (i = 0; i < Nhosts; i++) {
     69  fprintf(stderr, "Attempting to delete %s\n", hostname);
     70#ifdef IPP_CONDOR
     71  int index;
     72  index = is_host_in_list(hostname);
     73  if (index == -1) {
     74      return (FALSE);
     75  }
     76#endif
     77  for (i = 0; i < Nhosts; i++) { //Optimize this for condor
    4378    if (strcmp (hosts[i].hostname, hostname)) continue;
    44    
    4579    // delete this one
    4680    free (hosts[i].hostname);
    … …  
    5084    }
    5185    Nhosts --;
     86#ifdef IPP_CONDOR
     87    hosts_list_modified = TRUE; //Tell get_hosts_requirements to rebuild the hosts based requirement
     88#endif
    5289    return (TRUE);
    5390  }
    … …  
    88125    // float dtime = DTIME (stop, start);
    89126    /* if (VerboseMode()) gprint (GP_ERR, "delete job %f\n", dtime); */
    90   } 
     127  }
    91128  return (TRUE);
    92129}
    … …  
    95132
    96133  int status;
    97   char cmd[128]; 
     134  char cmd[128];
    98135  IOBuffer buffer;
    99136
    … …  
    104141  return (status);
    105142}
    106  
     143
    107144/* ask controller about job status */
    108145int CheckControllerJobStatus (Job *job) {
    … …  
    169206}
    170207
    171 /* we read Nbytes from the host, then watch for the prompt */ 
     208/* we read Nbytes from the host, then watch for the prompt */
    172209int GetJobOutput (char *cmd, int pid, IOBuffer *buffer, int Nbytes) {
    173  
     210
    174211  int i, status, Nstart;
    175212  char *line;
    … …  
    222259/* submitting a job to the controller automatically starts controller */
    223260int SubmitControllerJob (Job *job) {
    224 
     261#ifndef IPP_CONDOR
     262  fprintf(stderr, "Submitting job to pcontrol");
     263  fprintf(stdout, "Submitting job to pcontrol");
    225264  int i, Nchar, status;
    226265  char *cmd, *p, string[64];
    227266  IOBuffer buffer;
    228 
    229   if (job[0].task[0].host == NULL) return (FALSE);
    230 
     267  if (job[0].task[0].host == NULL) return (FALSE);
    231268  if (!StartController ()) {
    232269    gprint (GP_ERR, "failure to start pcontrol\n");
    233270    return (FALSE);
    234271  }
    235 
    236272  /** construct the line to be sent to the controller **/
    237 
    238273  /* determine the total line length */
    239274  Nchar = 0;
    … …  
    247282  ALLOCATE (cmd, char, Nchar);
    248283  bzero (cmd, Nchar);
    249 
    250284  /* construct the controller command portion */
    251285  if (!strcasecmp (job[0].task[0].host, "ANYHOST")) {
    … …  
    258292    }
    259293  }
    260  
    261294  if (job[0].priority) {
    262295    char tmp[64];
    … …  
    264297    strcat (cmd, tmp);
    265298  }
    266 
    267299  /* add the command arguments */
    268300  for (i = 0; i < job[0].task[0].argc; i++) {
    … …  
    270302    strcat (cmd, job[0].task[0].argv[i]);
    271303  }
    272 
    273304  // This function is called by the JobTaskThread via SubmitJob.  We need to unlock the
    274305  // JobTaskLock to avoid a dead lock with the JobTaskLock called in CheckController
    … …  
    280311  ControlUnlock(__func__);
    281312  JobTaskLock();
    282 
    283 
    284313  /* extract the job PID from the controller response */
    285314  p = memstr (buffer.buffer, "JobID", buffer.Nbuffer);
    … …  
    292321  sscanf (p, "%*s %s", string);
    293322  FreeIOBuffer (&buffer);
    294 
    295323  job[0].pid = atoi (string);
    296324  if (job[0].pid < 0) {
    297325    return (FALSE);
    298326  }
     327#else
     328  struct ns1__Transaction *transaction;
     329  int clusterId;
     330  //Create a Transaction
     331  transaction = beginTransaction(&soap, 10);
     332  //Get a clusterId
     333  clusterId = newCluster(&soap, transaction);
     334  fprintf(stderr, "ClusterId = %d\n", clusterId);
     335  fprintf(stdout, "Submitting job to HTCondor\n");
     336  char *args = paste_args(job[0].task[0].argc-1, job[0].task[0].argv+1);
     337  submitJob(&soap, transaction, clusterId, 0,
     338            job[0].task[0].argv[0],
     339            args,
     340            get_hosts_requirements());
     341  commitTransaction(&soap, transaction);
     342#endif
    299343  return (TRUE);
    300344}
    301345
    302346int StartController () {
    303 
     347#ifndef IPP_CONDOR
    304348  char *p;
    305349  char **argv, cmd[128];
    … …  
    307351  int stdin_fd[2], stdout_fd[2], stderr_fd[2];
    308352  IOBuffer buffer;
    309 
    310353  if (ControllerStatus) return (TRUE);
    311 
    312354  if (VarConfig ("CONTROLLER", "%s", cmd) == NULL) strcpy (cmd, "pcontrol");
    313 
    314355  if (hosts == NULL) {
    315356    NHOSTS = 16;
    316357    ALLOCATE (hosts, Host, NHOSTS);
    317358  }
    318 
    319359  bzero (stdin_fd,  2*sizeof(int));
    320360  bzero (stdout_fd, 2*sizeof(int));
    321361  bzero (stderr_fd, 2*sizeof(int));
    322 
    323362  if (pipe (stdin_fd)  < 0) goto pipe_error;
    324363  if (pipe (stdout_fd) < 0) goto pipe_error;
    325364  if (pipe (stderr_fd) < 0) goto pipe_error;
    326 
    327365  ALLOCATE (argv, char *, 2);
    328366  argv[0] = cmd;
    329367  argv[1] = 0;
    330 
    331368  pid = fork ();
    332369  if (!pid) { /* must be child process */
    333370    gprint (GP_LOG, "starting controller connection\n");
    334 
    335371    /* close the other ends of the pipes */
    336372    close (stdin_fd[1]);
    337373    close (stdout_fd[0]);
    338374    close (stderr_fd[0]);
    339 
    340375    /* tie our ends of the pipes to stdin, stdout, stderr */
    341376    dup2 (stdin_fd[0],  STDIN_FILENO);
    342377    dup2 (stdout_fd[1], STDOUT_FILENO);
    343378    dup2 (stderr_fd[1], STDERR_FILENO);
    344 
    345379    /* set all three unblocking */
    346380    setvbuf (stdin,  (char *) NULL, _IONBF, BUFSIZ);
    347381    setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ);
    348382    setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ);
    349 
    350     status = execvp (argv[0], argv);
     383    status = execvp (argv[0], argv);
    351384    exit (1);
    352385  }
    353386  free (argv);
    354 
    355387  /* close the other ends of the pipes */
    356388  close (stdin_fd[0]);  stdin_fd[0]  = 0;
    357389  close (stdout_fd[1]); stdout_fd[1] = 0;
    358390  close (stderr_fd[1]); stderr_fd[1] = 0;
    359 
    360391  /* make the pipes non-blocking */
    361392  fcntl (stdin_fd[1],  F_SETFL, O_NONBLOCK);
    362393  fcntl (stdout_fd[0], F_SETFL, O_NONBLOCK);
    363394  fcntl (stderr_fd[0], F_SETFL, O_NONBLOCK);
    364 
    365395  /* perform handshake with controller to verify alive & running */
    366396  /** this handshake is similar to ControllerCommand, but has important differences **/
    367397  InitIOBuffer (&buffer, 0x100);
    368 
    369398  /* send handshake command */
    370399  status = write_fmt (stdin_fd[1], "echo CONNECTED\n");
    371400  if ((status == -1) && (errno == EPIPE)) goto pipe_error;
    372 
    373401  /* try to get evidence connection is alive - wait upto a few seconds */
    374402  /* connection is likely slow; don't bother with nanosleep here */
    … …  
    383411  if (status == -1) goto io_error;
    384412  FreeIOBuffer (&buffer);
    385 
    386413  /* set local static vars to pipe connections */
    387414  stdin_cntl  = stdin_fd[1];
    388415  stdout_cntl = stdout_fd[0];
    389416  stderr_cntl = stderr_fd[0];
    390 
    391417  InitIOBuffer (&stdout_buffer, 0x100);
    392418  InitIOBuffer (&stderr_buffer, 0x100);
    393 
    394419  ControllerPID = pid;
     420#else
     421  if (ControllerStatus == FALSE) {
     422      //Initiate the HTcondor connection
     423      soap_init(&soap);
     424      soap_set_namespaces(&soap, condor_namespaces);
     425  }
     426#endif
    395427  ControllerStatus = TRUE;
    396428  gprint (GP_LOG, "Connected\n");
    397429  return (TRUE);
    398 
     430#ifndef IPP_CONDOR
    399431pipe_error:
    400432  perror ("pipe error:");
    401433  goto close_pipes;
    402 
    403434io_error:
    404435  gprint (GP_ERR, "timeout while connecting\n");
    405436  goto close_pipes;
    406 
    407437close_pipes:
    408438  if (stdin_fd[0]  != 0) close (stdin_fd[0]);
    … …  
    413443  if (stderr_fd[1] != 0) close (stderr_fd[1]);
    414444  return (FALSE);
     445#endif
    415446}
    416447
    417448int ControllerCommand (char *cmd, char *response, IOBuffer *buffer) {
    418 
     449#ifndef IPP_CONDOR
    419450  int i, j, status;
    420451  char *line;
    421452  struct timespec request, remain;
    422 
     453  //fprintf(stderr, "ControllerCommand: [%s]\n", cmd);
    423454  /* avoid blocking on waitpid, test every 100 usec, up to 50 msec */
    424455  request.tv_sec = 0;
    425456  request.tv_nsec = 100000;
    426 
    427457  ReadtoIOBuffer (buffer, stdout_cntl);
    428458  FlushIOBuffer (buffer);
    429 
    430459  /* send command, is pipe still open? */
    431460  status = write_fmt (stdin_cntl, "%s\n", cmd);
    … …  
    438467    return (FALSE);
    439468  }
    440  
    441469  /* for commands which don't return a prompt, don't look for one */
    442470  if (response == NULL) {
    443471    return (TRUE);
    444472  }
    445 
    446473  /* watch for response - wait up to 1 second */
    447474  line = NULL;
    … …  
    457484      fprintf (stderr, "controller is down (EOF), restarting\n");
    458485      if (!RestartController ()) {
    459         return (FALSE);
     486        return (FALSE);
    460487      }
    461488      return (FALSE);
    … …  
    470497    return (FALSE);
    471498  }
    472 
    473499  /* need to strip off the prompt */
    474500  line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer);
    … …  
    478504  }
    479505  if (VerboseMode()) fprintf (stderr, "message received, %d cycles\n", i);
     506#endif
    480507  return (TRUE);
    481508}
    … …  
    498525      break;
    499526  }
    500  
     527
    501528  /* read stderr buffer */
    502529  Nread = ReadtoIOBuffer (&stderr_buffer, stderr_cntl);
    … …  
    576603
    577604int QuitController () {
    578 
     605#ifndef IPP_CONDOR
    579606  char cmd[128];
    580607  IOBuffer buffer;
    581 
    582608  if (!ControllerStatus) return (TRUE);
    583 
    584609  sprintf (cmd, "quit");
    585610  InitIOBuffer (&buffer, 0x100);
    586611  ControllerCommand (cmd, NULL, &buffer);
    587612  FreeIOBuffer (&buffer);
    588 
    589   /* the quit command does not return a prompt,
     613  /* the quit command does not return a prompt,
    590614     check that the controller exited */
    591615  StopController ();
     616#else
     617  soap_destroy(&soap);
     618  soap_end(&soap);
     619  soap_done(&soap);
     620#endif
    592621  return (TRUE);
    593622}
    594623
    595624int StopController () {
    596 
     625#ifndef IPP_CONDOR
    597626  int i, waitstatus, result;
    598 
    599627  if (!ControllerStatus) return (TRUE);
    600 
    601628  ControllerStatus = FALSE;
    602629  result = waitpid (ControllerPID, &waitstatus, WNOHANG);
    … …  
    611638  FreeIOBuffer (&stdout_buffer);
    612639  FreeIOBuffer (&stderr_buffer);
     640#endif
    613641  return (TRUE);
    614642}
    615643
    616644int RestartController () {
    617 
     645#ifndef IPP_CONDOR
    618646  int i, status;
    619647  char command[256];
    620648  IOBuffer buffer;
    621 
    622649  gprint (GP_ERR, "attempting to restart pcontrol\n");
    623650  if (!StartController ()) {
    … …  
    625652    return (FALSE);
    626653  }
    627 
    628654  InitIOBuffer (&buffer, 0x100);
    629 
    630655  status = TRUE;
    631 
    632656  // XXX lock the host table? no: that would risk a dead lock between client and controller threads:
    633657  gprint (GP_ERR, "pcontrol restarted, reloading hosts\n");
    … …  
    637661    status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer);
    638662  }
    639 
    640663  if (status) gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG);
    641 
    642   FreeIOBuffer (&buffer);
    643 
    644   return (TRUE);
    645 }
     664  FreeIOBuffer (&buffer);
     665#endif
     666  return (TRUE);
     667}
     668
     669#ifdef IPP_CONDOR
     670char* get_hosts_requirements(void) {
     671    int i;
     672    int hr_length;
     673    char* base_requirement_pattern_b = " (machine == \"";
     674    char* base_requirement_pattern_e = "\")";
     675    char* base_requirement_pattern_or = " || ";
     676    if ( (hosts_requirement != NULL) && (hosts_list_modified == FALSE) ) {
     677        return hosts_requirement;
     678    }
     679    if (hosts_requirement != NULL) {
     680        free(hosts_requirement);
     681    }
     682    printf("Generating hosts based requirement\n");
     683    hosts_requirement = strcreate("");
     684    hr_length = strlen(hosts_requirement);
     685    if (Nhosts > 0) {
     686        hosts_requirement = opihi_append(hosts_requirement, &hr_length, "( ", NULL);
     687        for (i = 0; i < Nhosts; i++) {
     688            hosts_requirement = opihi_append(hosts_requirement, &hr_length,
     689                                             base_requirement_pattern_b, NULL);
     690            hosts_requirement = opihi_append(hosts_requirement, &hr_length,
     691                                             hosts[i].hostname, NULL);
     692            hosts_requirement = opihi_append(hosts_requirement, &hr_length,
     693                                             base_requirement_pattern_e, NULL);
     694            if (i != Nhosts-1) {
     695                hosts_requirement = opihi_append(hosts_requirement, &hr_length,
     696                                                 base_requirement_pattern_or, NULL);
     697            }
     698        }
     699    } else { // No host defined: leave it empty
     700        //pass
     701    }
     702    hosts_list_modified = FALSE;
     703    printf("end of Generating hosts based requirement\n");
     704    return hosts_requirement;
     705}
     706
     707static int is_host_in_list(char *hostname) {
     708    int i;
     709    for (i = 0; i < Nhosts; i++) {
     710        if (strcmp(hosts[i].hostname, hostname) == 0) {
     711            return i;
     712        }
     713    }
     714    return -1;
     715}
     716
     717char *get_host_status(char* hostname) {
     718    return getHostStatus(&soap, hostname);
     719}
     720
     721#endif
Note: See TracChangeset for help on using the changeset viewer.