agent.c

Go to the documentation of this file.

/* $Id: agent.c,v 1.159 2010/04/23 15:29:40 tbrady Exp $ */
/* $UTK_Copyright: $ */

#include <stdlib.h>
#include <stdio.h>
#include <netdb.h>
#include <strings.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/un.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <string.h>
#include <signal.h>
#include <time.h>
#include <unistd.h>
#include <glob.h>

#include "utility.h"
#include "proxylib.h"
#include "comm_basics.h"
#include "comm_data.h"
#include "server.h"
#include "agent.h"
#include "problem.h"
#include "comm_encode.h"
#include "gs_storage.h"
#include "mfork.h"
#ifdef GS_SMART_GRIDSOLVE
#include "gs_smart_task_graph.h"
#include "gs_smart_app_pm.h"
#endif

gs_agent_scheduler_t gs_agent_parse_scheduler_name(char *);

char 
  *agent_cfg = NULL,
  GRIDSOLVE_SQLITE_DB[FN_LEN] = "",
  GRIDSOLVE_SENSOR_USOCK[FN_LEN] = "",
  GRIDSOLVE_STATS_FILE[FN_LEN] = "",
  GRIDSOLVE_SCHED_LOCK_FILE[FN_LEN] = "";

gs_agent_stats_t
  gs_agent_stats;

gs_agent_scheduler_t
  gs_agent_scheduler_selection;

int GS_SENSORFD;
FILE *GS_SENSORFILE;

int
  global_taskid = 0,
  htm_scheduler_sync = 1,
  keep_track_of_task_events = 0;

struct timeval
  global_start_time = {0, 0};

sqlite3 *global_db = NULL;

static void
gs_agent_generic_signal_handler(int sig)
{
  if(sig == SIGHUP) {
    gs_info_t *sa_list;

    LOGPRINTF("SIGHUP: reading configuration file '%s'\n", agent_cfg);

    if(gs_parse_config_file(agent_cfg, &sa_list) == 0) {
      if(sa_list) {
        gs_info_t *p;

        for(p = sa_list; p != NULL; p = p->next) {
          if(!strcasecmp(p->type, "GS_AGENT_SCHEDULER"))
            gs_agent_scheduler_selection = gs_agent_parse_scheduler_name(p->value);
          else if(!strcasecmp(p->type, "GS_HTM_SYNC"))
            htm_scheduler_sync = atoi(p->value);

          free(p->type);
          free(p->value);
        }

        free(sa_list);
      }
    }

    return;
  }

  ERRPRINTF("Agent terminating on signal %d.\n", sig);
  unlink(GRIDSOLVE_STATS_FILE);
  unlink(GRIDSOLVE_SQLITE_DB);
  unlink(GRIDSOLVE_SCHED_LOCK_FILE);
  exit(0);
}

static void
gs_temp_sighup_handler(int sig)
{
  kill(getppid(), SIGHUP);
  return;
}

double
get_time_since_startup()
{
  struct timeval tv;
  double cur_time, start_time;

  gettimeofday(&tv, NULL);

  cur_time = (double)tv.tv_sec + (((double)tv.tv_usec) / 1000000.0);
  start_time = (double)global_start_time.tv_sec + 
       (((double)global_start_time.tv_usec) / 1000000.0);

  return cur_time - start_time;
}

int
gs_agent_create_info_dir(gs_agent_t *gs_agent)
{
  ipaddr_t ipaddress;
  char dummy_componentid[CID_LEN];

  ipaddress = proxy_get_my_ipaddr();

  if(gs_create_info_dir(dummy_componentid, ipaddress,
       gs_agent->port, &(gs_agent->infodir)) < 0) {
    ERRPRINTF("Creating agent infodir failed.\n");
    return -1;
  }

  return 0;
}

gs_agent_t *
gs_agent_init(char *cfg)
{
  gs_agent_t *gs_agent;
  int pid;
  
  gettimeofday(&global_start_time, NULL);

  gs_agent = (gs_agent_t *) calloc(1, sizeof(gs_agent_t));

  if(!gs_agent) return NULL;

  /* Setup agent */
  gs_agent->hostname = gs_get_machine_name();
  gs_agent->servers = NULL;
  gs_agent->dsig = pvmgetdsig();
  gs_agent->port = getenv_int("GRIDSOLVE_AGENT_PORT", GRIDSOLVE_AGENT_PORT_DEFAULT);

  if(gs_agent_create_info_dir(gs_agent) < 0) {
    ERRPRINTF("Could not create agent infodir.\n");
    return NULL;
  }

  pid = getpid();

  gs_clean_up_old_temp_files(gs_agent->infodir);

  snprintf(GRIDSOLVE_SQLITE_DB, FN_LEN, "%s/%s.%d", gs_agent->infodir, GRIDSOLVE_SQLITE_DB_PREFIX, pid);
  snprintf(GRIDSOLVE_STATS_FILE, FN_LEN, "%s/%s.%d", gs_agent->infodir, GRIDSOLVE_STATS_FILE_PREFIX, pid);
  snprintf(GRIDSOLVE_SENSOR_USOCK, FN_LEN, "%s/%s.%d", gs_agent->infodir, GRIDSOLVE_SENSOR_USOCK_PREFIX, pid);
  snprintf(GRIDSOLVE_SCHED_LOCK_FILE, FN_LEN, "%s/%s.%d", gs_agent->infodir, GRIDSOLVE_SCHED_LOCK_FILE_PREFIX, pid);

  gs_init_stats_file(gs_agent, GRIDSOLVE_STATS_FILE);

  gs_agent->mysql.host = GRIDSOLVE_DEFAULT_MYSQL_HOST;
  gs_agent->mysql.user = GRIDSOLVE_DEFAULT_MYSQL_USER;
  gs_agent->mysql.passwd = GRIDSOLVE_DEFAULT_MYSQL_PASSWD;
  gs_agent->mysql.db_name = GRIDSOLVE_DEFAULT_MYSQL_DB;
  gs_agent->mysql.port = GRIDSOLVE_DEFAULT_MYSQL_PORT;
  gs_agent->mysql.unix_socket = GRIDSOLVE_DEFAULT_MYSQL_UNIX_SOCKET;

  if(cfg) {
    gs_info_t *sa_list;

    if(gs_parse_config_file(cfg, &sa_list) == 0) {
      if(sa_list) {
        gs_info_t *p;

        for(p = sa_list; p != NULL; p = p->next) {
          if(!strcasecmp(p->type, "GS_MYSQL_HOST"))
            gs_agent->mysql.host = p->value;
          else if(!strcasecmp(p->type, "GS_MYSQL_USER"))
            gs_agent->mysql.user = p->value;
          else if(!strcasecmp(p->type, "GS_MYSQL_PASSWD"))
            gs_agent->mysql.passwd = p->value;
          else if(!strcasecmp(p->type, "GS_MYSQL_DB"))
            gs_agent->mysql.db_name = p->value;
          else if(!strcasecmp(p->type, "GS_MYSQL_PORT"))
            gs_agent->mysql.port = atoi(p->value);
          else if(!strcasecmp(p->type, "GS_MYSQL_UNIX_SOCKET"))
            gs_agent->mysql.unix_socket = p->value;
          else if(!strcasecmp(p->type, "GS_AGENT_SCHEDULER"))
            gs_agent_scheduler_selection = gs_agent_parse_scheduler_name(p->value);
          else if(!strcasecmp(p->type, "GS_HTM_SYNC"))
            htm_scheduler_sync = atoi(p->value);

          /* free only the type field */
          free(p->type);
        }

        free(sa_list);
      }
    }
  }

  LOGPRINTF("Initialized agent %s\n", gs_agent->hostname);

  return (gs_agent);
}

int
gs_init_stats_file(gs_agent_t *gs_agent, char *fname)
{
  int fd;

  if(!gs_agent || !fname) {
    ERRPRINTF("Invalid args\n");
    return -1;
  }

  if((fd = open(fname, O_RDWR | O_CREAT | O_TRUNC, 0600)) < 0) {
    ERRPRINTF("Could not create stats file '%s'\n", fname);
    return -1;
  }

  gettimeofday(&gs_agent_stats.start_time, NULL);
  strncpy(gs_agent_stats.hostname, gs_agent->hostname, GS_MAX_NAMELEN);
  gs_agent_stats.hostname[GS_MAX_NAMELEN-1] = '\0';
  gs_agent_stats.port = gs_agent->port;

  write(fd, &gs_agent_stats, sizeof(gs_agent_stats_t));

  close(fd);
  return 0;
}

int
gs_agent_process_availability_request(int sock)
{
  gs_server_t srv;
  char *req;
  int s;

  if(gs_recv_string(sock, &req) < 0) {
    ERRPRINTF("Error reading request string.\n");
    return -1;
  }

  if(gs_decode_availability_request(req, &srv) < 0) {
    ERRPRINTF("Couldn't decode availability request.\n");
    if(req) free(req);
    return -1;
  }

  if(req) free(req);

  s = gs_connect_to_host(srv.componentid, srv.ipaddress, srv.port,
                            srv.proxyip, srv.proxyport);

  if(s == INVALID_SOCKET) {
    ERRPRINTF("Server failed connectivity test.\n");
    return -1;
  }

  if(gs_send_tag(s, GS_PROT_OK) < 0) {
    ERRPRINTF("Error sending response tag.\n");
    return -1;
  }

  gs_close_socket(s);

  return 0;
}

int
gs_agent_process_problem_list(gs_agent_t *gs_agent, int sock)
{
  gs_problem_t **problem_list = NULL;
  int i, count;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  count = gs_get_all_problems(gs_agent, &problem_list, &count);

  if(count < 0) {
    DBGPRINTF("failed to get list of all problems\n");
    return -1;
  }

  /* Send number of problems to the client */
  if(gs_send_int(sock, count) < 0) {
    DBGPRINTF("failed to send number of problems\n");
    return -1;
  }

  for(i = 0; i < count; i++) {
    char *prob = NULL;
    if((gs_encode_problem(&prob, problem_list[i]) < 0) ||
       (gs_send_string(sock, prob) < 0)) {
      FREE(prob);
      DBGPRINTF("Failed to send problem list\n");
      return -1;
    }

    FREE(prob);
  }

  for(i = 0; i < count; i++)
    gs_free_problem(problem_list[i]);
  FREE(problem_list);
  return 0;
}

int
gs_agent_process_server_list(gs_agent_t *gs_agent, int sock)
{
  gs_server_t **server_list = NULL;
  int i, count;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  count = gs_get_all_servers(gs_agent, &server_list, &count);

  if(count < 0) {
    DBGPRINTF("failed to get list of all servers\n");
    return -1;
  }

  /* Send number of servers to the client */
  if(gs_send_int(sock, count) < 0) {
    DBGPRINTF("failed to send number of servers\n");
    return -1;
  }

  for(i = 0; i < count; i++) {
    char *srv = NULL;
    DBGPRINTF("Encoding server: %s.\n", server_list[i]->hostname);
    if((gs_encode_server(&srv, server_list[i]) < 0) ||
       (gs_send_string(sock, srv) < 0)) {
      FREE(srv);
      DBGPRINTF("Failed to send server list \n");
      return -1;
    }

    FREE(srv);
  }

  for(i = 0; i < count; i++)
    gs_server_free(server_list[i]);
  FREE(server_list);

  return 0;
}

int
gs_agent_process_server_ping_update(gs_agent_t *gs_agent, int sock)
{
  char *cid = NULL, *msg = NULL;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  if(gs_recv_string(sock, &cid) < 0) {
    ERRPRINTF("gs_recv_string (of cid) failed\n");
    return -1;
  }

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("gs_recv_string (of msg) failed\n");
    return -1;
  }

  if(gs_update_ping_list(gs_agent, cid, msg) < 0) {
    ERRPRINTF("failed to update ping list\n");
    return -1;
  }

  return 0;
}

int
gs_agent_process_server_ping_list(gs_agent_t *gs_agent, int sock)
{
  gs_server_t **server_list = NULL;
  char *cid = NULL;
  int i, count;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  if(gs_recv_string(sock, &cid) < 0) {
    DBGPRINTF("gs_recv_string failed\n");
    return -1;
  }

  count = gs_get_server_ping_list(gs_agent, &server_list, cid, &count);

  free(cid);

  if(count < 0) {
    DBGPRINTF("failed to get list of all servers\n");
    return -1;
  }

  /* Send number of servers to the requesting server */
  if(gs_send_int(sock, count) < 0) {
    DBGPRINTF("failed to send number of servers\n");
    return -1;
  }

  for(i = 0; i < count; i++) {
    char *srv = NULL;
    DBGPRINTF("Encoding server: %s.\n", server_list[i]->hostname);
    if((gs_encode_server(&srv, server_list[i]) < 0) ||
       (gs_send_string(sock, srv) < 0)) {
      FREE(srv);
      DBGPRINTF("Failed to send server list \n");
      return -1;
    }

    FREE(srv);
  }

  for(i = 0; i < count; i++)
    gs_server_free(server_list[i]);
  FREE(server_list);

  return 0;
}

int
gs_agent_process_problem_desc(gs_agent_t * gs_agent, int sock)
{
  char *xml_problem = NULL;
  gs_server_t **server_list = NULL;
  gs_problem_t *problem = NULL;
  int i, rc, count = 0;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  problem = (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t));
  if(!problem) {
    DBGPRINTF("Could not malloc new problem.\n");
    return -1;
  }

  if(gs_recv_string(sock, &(problem->name)) < 0) {
    DBGPRINTF("gs_recv_string failed\n");
    FREE(problem);
    return -1;
  }
  DBGPRINTF("prob name = '%s'\n", problem->name);

  count =
      gs_get_server_list(gs_agent, problem, NULL, &server_list,
                         &count);

  rc = 0;

  if(count <= 0) {
    DBGPRINTF("could not find problem %s.\n", problem->name);
    if(gs_send_string(sock, "not found") < 0) {
      DBGPRINTF("Failed to send problem\n");
      rc = -1;
    }
  }
  else {
    DBGPRINTF("Encoding problem: %s.\n", problem->name);
    if((gs_encode_problem(&xml_problem, problem) < 0) ||
       (gs_send_string(sock, xml_problem) < 0)) {
      DBGPRINTF("Failed to send problem\n");
      rc = -1;
    }
  }

  FREE(xml_problem);

  for(i = 0; i < count; i++)
    gs_server_free(server_list[i]);
  free(server_list);

  gs_free_problem(problem);

  return rc;
}

int
gs_agent_process_problem_submit(gs_agent_t * gs_agent, int sock)
{
  char *msg = NULL, cid_string[2 * CID_LEN + 1];
  char *client_criteria = NULL;
  char *xml_problem = NULL;
  gs_server_t **server_list = NULL;
  gs_problem_t *problem = NULL;
  int i, count = 0, fd;
  int sender_dsig = 0;
  int sender_major = -1;
  int scalar_args_to_be_transferred = 0;
  int problem_desc_to_be_transferred = 0;
  time_t clock;
  struct tm *now;
  char subtime[128];

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  problem = (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t));
  if(!problem) {
    DBGPRINTF("Could not malloc new problem.\n");
    return -1;
  }

  if(gs_recv_string(sock, &msg) < 0) {
    DBGPRINTF("gs_agent_process_problem_submit: gs_recv_string failed\n");
    return -1;
  }
  DBGPRINTF("gs_agent_process_problem_submit: msg = '%s'\n", msg);

  if(gs_decode_problem_submit_request(msg, &(problem->name), &sender_dsig, &client_criteria) < 0) {
    free(msg);
    DBGPRINTF("gs_agent_process_problem_submit: failed to decode request\n");
    return -1;
  }
  DBGPRINTF("problem->name = '%s'\n", problem->name);

  free(msg);

  /* Get the servers and fill in the problem structure.  This must be done
     before the problem can be transferred. */
  count =
      gs_get_server_list(gs_agent, problem, client_criteria, &server_list,
                         &count);

  if(client_criteria) {
    free(client_criteria);
    client_criteria = NULL;
  }

  /* Transfer problem descriptor if needed */
  if(gs_recv_int(sock, &problem_desc_to_be_transferred) < 0)
    goto error_comm;
  if(problem_desc_to_be_transferred == 1) {
    if(count == 0)
      problem->description = strdup(GS_UNKNOWN_PROB);
    if(gs_encode_problem(&xml_problem, problem) < 0)
      goto error_comm;
    if(gs_send_string(sock, xml_problem) < 0)
      goto error_comm;
    FREE(xml_problem);

    /* if the server count is 0, we have sent a problem struct
     * with the description set to GS_UNKNOWN_PROB.  When the
     * client receives this, it will abort the request, so we
     * bail out here also.
     */
    if(count == 0) {
      gs_free_problem(problem);
      return 0;
    } 
  }

  /* Get scalar args from the client if they are available; used for
     evaluating complexity expression */
  if(gs_recv_int(sock, &scalar_args_to_be_transferred) < 0)
    goto error_comm;

  if(scalar_args_to_be_transferred == 1) {
    if(gs_recv_input_scalar_args (sock, problem, sender_dsig, 
         gs_agent->dsig, &sender_major) < 0)
      goto error_comm;

    if(gs_receiver_compute_arg_sizes(problem, GS_IN) < 0) {
      ERRPRINTF("error computing argument sizes.\n");
      return -1;
    }
  }

  if((fd = gs_open_locked_file_timeout(GRIDSOLVE_SCHED_LOCK_FILE, F_WRLCK, 
             O_RDWR | O_CREAT, GS_SCHED_LOCK_TIMEOUT)) < 0) {
    ERRPRINTF("Timed out on scheduler lock.  Continuing w/o scheduling.\n");
  }
  else {
    /* Agent side scheduling using the problem complexity and server
       information occurs here.  The scheduler reorders the server list
       in order of best server to worst. */
    DBGPRINTF("Calling the agent side scheduler\n");
    if(gs_agent_scheduler(problem, count, server_list) < 0) {
      DBGPRINTF("Could not sort servers\n");
      gs_unlock_file(fd);
      return -1;
    }

    gs_unlock_file(fd);
  }
  
  if(count <= 0) {
    DBGPRINTF("No servers for problem '%s'\n", problem->name);
    count = 0;
  }

  /* Send task id to the client */
  if(gs_send_int(sock, global_taskid) < 0)
    goto error_comm;

  /* Send number of servers to the client */
  if(gs_send_int(sock, count) < 0)
    goto error_comm;

  /* Send servers and problem to client */
  if(count > 0) {
    char dummy_taskid[TASK_ID_LEN];
    double start_time;

    for(i = 0; i < count; i++) {
      char *srv = NULL;
      DBGPRINTF("Encoding server: %s.\n", server_list[i]->hostname);
      if((gs_encode_server(&srv, server_list[i]) < 0) ||
         (gs_send_string(sock, srv) < 0)) {
        FREE(srv);
        DBGPRINTF("Failed to send server list \n");
        return -1;
      }

      FREE(srv);
    }

    DBGPRINTF("Encoding problem: %s.\n", problem->name);
    if((gs_encode_problem(&xml_problem, problem) < 0) ||
       (gs_send_string(sock, xml_problem) < 0)) {
      FREE(xml_problem);
      DBGPRINTF("Failed to send problem\n");
      return -1;
    }
    FREE(xml_problem);

    /* assign task to first server in list */

    if(scalar_args_to_be_transferred) {
      struct timeval tv;

      gettimeofday(&tv, NULL);
      
      proxy_cid_to_str(cid_string, server_list[0]->componentid);
      start_time = get_time_since_startup();

      srand48((double) tv.tv_usec);

      for(i = 0; i < TASK_ID_LEN-1; i++)
        dummy_taskid[i] = (char) ((int)(drand48() * (double)('z' - 'a')) + 'a');
      dummy_taskid[TASK_ID_LEN-1] = 0;

      if(gs_insert_submitted_task_guess(cid_string, dummy_taskid, global_taskid,
           start_time, server_list[0]->score, 0.0, start_time, 0, 0) < 0)
        ERRPRINTF("Warning: error inserting task\n");
    }
  }

  LOGPRINTF("Agent %s got %s request from client and sent %d servers\n",
            gs_agent->hostname, problem->name, count);
  
  clock = time(0);
  now = localtime(&clock);
  snprintf(subtime, 1024, "[(%02d/%02d/%04d) %02d:%02d:%02d]",
           now->tm_mon+1,now->tm_mday,now->tm_year+1900, 
           now->tm_hour,now->tm_min,now->tm_sec);

  /* Heuristic: Temporarily increase the workload on the first server.
   * This assumes that the first server will be assigned the work and
   * we want to avoid assigning more work there till an new updated
   * workload report is received. */
  if (server_list != NULL && count > 0) {
    server_list[0]->workload += 100;
    if (gs_update_workload(gs_agent, server_list[0]) < 0) 
      DBGPRINTF("Could not temporarily update workload for the first server \n");
    proxy_cid_to_str(cid_string, server_list[0]->componentid);
  }
  
  for(i = 0; i < count; i++)
    gs_server_free(server_list[i]);
  if(server_list) free(server_list);

  gs_free_problem_and_data(problem);

  return 0;

error_comm:
  for(i = 0; i < count; i++)
    gs_server_free(server_list[i]);
  if(server_list) free(server_list);
  gs_free_problem_and_data(problem);
  ERRPRINTF("Error handling problem submission\n");
  return -1;
}

int
gs_agent_update_all_perf_models(char *model_update, char *srv_cid)
{
  int i, num_updates;
  char *mcopy, *tok;

  if(!model_update)
    return -1;

  mcopy = strdup(model_update);

  if(!mcopy)
    return -1;

  tok = strtok(mcopy, "\n");

  num_updates = atoi(tok);

  for(i=0;i<num_updates;i++) {
    char *name, *expr;

    tok = strtok(NULL, "\n");

    if(gs_decode_model_update(tok, &name, &expr) < 0) {
      ERRPRINTF("Error parsing performance model update\n");
      continue;
    }

    if(gs_update_perf_expr(srv_cid, name, expr) < 0) {
      ERRPRINTF("Error updating performance model\n");
    }

    free(name);
    free(expr);
  }

  free(mcopy);

  return 0;
}

int
gs_agent_process_workload_report(gs_agent_t * gs_agent, int sock)
{
  char *msg, temp_cid[CID_LEN * 2 + 1];
  gs_server_t srv;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  if(gs_recv_string(sock, &msg) < 0) {
    DBGPRINTF("gs_agent_process_workload_report: gs_recv_string failed\n");
    return -1;
  }

  if(gs_decode_workload_report(msg, &srv.workload, &srv.nproblems, 
      temp_cid) < 0) 
  {
    DBGPRINTF("Failed to decode workload report\n");
    free(msg);
    return -1;
  }

  free(msg);

  if(gs_recv_string(sock, &msg) < 0) {
    DBGPRINTF("gs_agent_process_workload_report: gs_recv_string failed\n");
    return -1;
  }

  gs_agent_update_all_perf_models(msg, temp_cid);

  free(msg);

  proxy_str_to_cid(srv.componentid, temp_cid);
  DBGPRINTF("gs_agent_process_workload_report: workload = %d serverID %s\n",
            srv.workload, temp_cid);

  if(gs_update_workload(gs_agent, &srv) < 0) {
    DBGPRINTF("gs_agent_process_workload_report: report from unknown server\n");
    if(gs_send_tag(sock, GS_PROT_UNKNOWN_SERVER) < 0)
      return -1;
  }
  else {                        /* Everything is OK */
    if(gs_send_tag(sock, GS_PROT_OK) < 0)
      return -1;
  }

  DBGPRINTF("Agent %s handled a workload report setting server %s to %d\n",
            gs_agent->hostname, temp_cid, srv.workload);
  return 0;
}

int
gs_agent_process_server_registration(gs_agent_t * gs_agent, int sock)
{
  char *serverstr = NULL;
  gs_server_t *gs_server;
  time_t clock;
  struct tm *now;
  char subtime[128];

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  gs_server = (gs_server_t *) CALLOC(1, sizeof(gs_server_t));

  if(!gs_server) {
    ERRPRINTF("Failed to malloc server struct\n");
    return -1;
  }

  DBGPRINTF("Entering \n");

  /* Receive server string and convert into structure */
  if((gs_recv_string(sock, &serverstr) < 0) ||
     (gs_decode_server(serverstr, gs_server) < 0))
    goto error;

  FREE(serverstr);

  gs_server->last_update = time(NULL);

  /* Actual storage routine */
  ASSERT_EXPR((gs_add_server(gs_agent, gs_server) >= 0), goto error);

  /* Send reply tag */
  if(gs_send_tag(sock, GS_PROT_OK) < 0)
    goto error;

  LOGPRINTF("Agent %s registered server %s\n", gs_agent->hostname, gs_server->hostname);
  clock = time(0);
  now = localtime(&clock);
  snprintf(subtime, 1024, "[(%02d/%02d/%04d) %02d:%02d:%02d]",
           now->tm_mon+1,now->tm_mday,now->tm_year+1900, 
           now->tm_hour,now->tm_min,now->tm_sec);

  gs_server_free(gs_server);

  return 0;

error:
  ERRPRINTF("Error registering server\n");
  gs_server_free(gs_server);
  FREE(serverstr);
  gs_send_tag(sock, GS_PROT_ERROR);
  return -1;
}

int
gs_agent_task_terminated(char *server_cid, char *request_id, int agent_taskid,
  double run_time)
{
  gs_htm_task *task;
  double end_time;

  end_time = get_time_since_startup();

  if(gs_insert_completed_task(server_cid, request_id, agent_taskid, end_time, 
      0.0, 0.0, end_time, 0, 1) < 0)
    ERRPRINTF("Warning: could not insert completed task\n");

  task = (gs_htm_task *) malloc(sizeof(gs_htm_task));

  if(!task) {
    ERRPRINTF("malloc...\n");
    /* in this case don't need to return an error since it's not fatal */
    return 0;
  }

  if(gs_get_task_by_agent_taskid(agent_taskid, task) < 0) {
    free(task);
    /* also not fatal */
    return 0;
  }

  LOGPRINTF("task %d: total time = %lf, server run time = %lf\n", agent_taskid,
    end_time-task->start, run_time);

  free(task);

  return 0;
}

int
gs_agent_process_notify_cancel(int sock)
{
  char *msg, *server_cid, *request_id;
  int agent_taskid;

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  LOGPRINTF("msg = '%s'\n", msg);
  SENSORPRINTF("CANCEL %s\n", msg);

  /* if not keeping track of task events, bail out now. */
  if(!keep_track_of_task_events) {
    free(msg);
    return 0;
  }

  server_cid = (char *)malloc(strlen(msg));

  if(!server_cid) {
    ERRPRINTF("malloc failed.\n");
    return -1;
  }

  request_id = (char *)malloc(strlen(msg));

  if(!request_id) {
    ERRPRINTF("malloc failed.\n");
    free(server_cid);
    return -1;
  }

  if(gs_decode_cancel_notification(msg, &server_cid, &request_id, 
       &agent_taskid) < 0)
    ERRPRINTF("Could not decode cancel notifiction.\n");
  else
    gs_agent_task_terminated(server_cid, request_id, agent_taskid, 0.0);

  free(request_id);
  free(server_cid);
  free(msg);

  return 0;
}

int
gs_agent_process_notify_failure(int sock)
{
  char *msg, *server_cid, *request_id;
  int agent_taskid;

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  LOGPRINTF("msg = '%s'\n", msg);
  SENSORPRINTF("FAIL %s\n", msg);

  /* if not keeping track of task events, bail out now. */
  if(!keep_track_of_task_events) {
    free(msg);
    return 0;
  }

  server_cid = (char *)malloc(strlen(msg));

  if(!server_cid) {
    ERRPRINTF("malloc failed.\n");
    return -1;
  }

  request_id = (char *)malloc(strlen(msg));

  if(!request_id) {
    ERRPRINTF("malloc failed.\n");
    free(server_cid);
    return -1;
  }

  if(gs_decode_cancel_notification(msg, &server_cid, &request_id, 
       &agent_taskid) < 0)
    ERRPRINTF("Could not decode failure notifiction.\n");
  else
    gs_agent_task_terminated(server_cid, request_id, agent_taskid, 0.0);

  free(request_id);
  free(server_cid);
  free(msg);

  return 0;
}

int
gs_agent_process_notify_complete(int sock)
{
  char *msg, *server_cid, *request_id;
  double service_et;
  int agent_taskid;

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  LOGPRINTF("msg = '%s'\n", msg);
  SENSORPRINTF("COMPLE %s\n", msg);

  /* if not keeping track of task events, bail out now. */
  if(!keep_track_of_task_events) {
    free(msg);
    return 0;
  }

  server_cid = (char *)malloc(strlen(msg));

  if(!server_cid) {
    ERRPRINTF("malloc failed.\n");
    return -1;
  }

  request_id = (char *)malloc(strlen(msg));

  if(!request_id) {
    ERRPRINTF("malloc failed.\n");
    free(server_cid);
    return -1;
  }

  if(gs_decode_problem_complete_notification(msg, &server_cid, &request_id,
       &agent_taskid, &service_et) < 0)
    ERRPRINTF("Could not decode completion notifiction.\n");
  else
    gs_agent_task_terminated(server_cid, request_id, agent_taskid, service_et);

  free(request_id);
  free(server_cid);
  free(msg);

  return 0;
}

int
gs_dump_task_list(char *server_cid)
{
  gs_htm_task **tasks = NULL;
  int i, count;

  gs_get_tasks_for_server(server_cid, &tasks, &count, 1);

  if(count < 0) {
    ERRPRINTF("failed to get list of all tasks\n");
    return -1;
  }

  printf("task list for '%s':\n", server_cid);

  for(i=0;i<count;i++) {
    printf("%s:\n", tasks[i]->id);
    printf("start=%10.2lf, duration=%10.2lf, remaining=%10.2lf, end=%10.2lf, active=%d, finished=%d\n",
      tasks[i]->start, tasks[i]->duration, tasks[i]->remaining,
      tasks[i]->end, tasks[i]->active, tasks[i]->finished);

    free(tasks[i]);
  }

  free(tasks);

  return 0;
}

int
gs_agent_reassign_task(char *cid_string, char *taskid, int agent_taskid, 
  double duration, double agent_est_time)
{
  double start_time;
  gs_htm_task *task;

  task = (gs_htm_task *) malloc(sizeof(gs_htm_task));

  if(!task) {
    ERRPRINTF("malloc...\n");
    return -1;
  }

  start_time = get_time_since_startup();

  if(agent_taskid == -1) {

    if(gs_insert_submitted_task(cid_string, taskid, agent_taskid, start_time,
          duration, agent_est_time, start_time, 0, 0) < 0)
    {
      free(task);
      return -1;
    }
  }
  else {
    if(gs_get_task_by_agent_taskid(agent_taskid, task) < 0) {

      /* the task was not already assigned, so insert it now */

      if(gs_insert_submitted_task(cid_string, taskid, agent_taskid,
           start_time, duration, agent_est_time, start_time, 0, 0) < 0)
      {
        free(task);
        return -1;
      }
    }
    else {
      /* found the task in the database, just update the server assignment
       * if necessary.
       */

      LOGPRINTF("end time from db = %g, agent est end time = %g\n",
          task->end, agent_est_time);

      if(gs_update_task(cid_string, task->id, taskid, agent_taskid, task->start,
           duration, task->remaining, task->end, task->active,
           task->finished) < 0)
      {
        free(task);
        return -1;
      }
    }
  }

  free(task);

  return 0;
}

int
gs_agent_process_notify_submit(int sock)
{
  char *msg, *problem_name, *server_cid, *user_name, *host_name, 
    *client_cid, *request_id;
  int agent_taskid;
  double est_time, agent_est_time;

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  if(!msg) {
    ERRPRINTF("Bad notification message.\n");
    return -1;
  }

  LOGPRINTF("msg = '%s'\n", msg);
  SENSORPRINTF("SUBMIT %s\n", msg);

  /* if not keeping track of task events, bail out now. */
  if(!keep_track_of_task_events) {
    free(msg);
    return 0;
  }

  if(gs_decode_problem_solve_notification(msg, &problem_name, &est_time,
      &server_cid, &user_name, &host_name, &client_cid, &request_id, 
      &agent_taskid, &agent_est_time) < 0) {
    free(msg);
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  LOGPRINTF("in agent, est time = %lf, agent est time = %lf\n", est_time, agent_est_time);

  if(gs_agent_reassign_task(server_cid, request_id, agent_taskid, est_time, agent_est_time) < 0)
    ERRPRINTF("Warning: could not assign task\n");
    
  free(msg);
  free(problem_name);
  free(server_cid);
  free(user_name);
  free(host_name );
  free(client_cid);
  free(request_id);

  return 0;
}

int
gs_agent_process_problem_registration(gs_agent_t * gs_agent, int sock)
{
  char *problemstr, *temp_cid, **model_strings, **removed_probs;
  gs_problem_t **problem = NULL;
  int i, num_services, num_removed;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  if(gs_recv_string(sock, &temp_cid) < 0) {
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  if(gs_recv_int(sock, &num_services) < 0) {
    ERRPRINTF("Error communicating with server.\n");
    return -1;
  }

  problem = (gs_problem_t **)calloc(num_services, sizeof(gs_problem_t *));
  if(!problem) {
    ERRPRINTF("malloc failed.\n");
    return -1;
  }

  model_strings = (char **)calloc(num_services+1, sizeof(char *));
  if(!model_strings) {
    ERRPRINTF("malloc failed.\n");
    return -1;
  }

  removed_probs = (char **)calloc(num_services+1, sizeof(char *));
  if(!removed_probs) {
    ERRPRINTF("malloc failed.\n");
    return -1;
  }

  i = 0;
  for(;;) {
    /* Receive string and convert into structure */
    if(gs_recv_string(sock, &problemstr) < 0) {
      ERRPRINTF("Error communicating with the server\n");
      goto error;
    }

    /* check whether this is the last problem */
    if(!strcmp(problemstr, GS_END_PROB_REG)) {
      FREE(problemstr);
      break;
    }

    problem[i] = (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t));

    if(!problem[i]) {
      ERRPRINTF("Failed to allocate problem struct\n");
      goto error;
    }

    if(gs_decode_problem(problemstr, problem[i]) < 0) {
      ERRPRINTF("Error decoding problem struct\n");
      FREE(problemstr);
      goto error;
    }

    FREE(problemstr);

    LOGPRINTF("Received problem %s\n", problem[i]->name);

    i++;
  }

  /* num_services is sent as the total number of services on disk at the server,
   * but the actual number of sent services is 'i'.
   */
  num_services = i;

  if(num_services > 0)
    LOGPRINTF("Received %d new services\n", num_services);

  i = 0;
  for(;;) {
    if(gs_recv_string(sock, &model_strings[i]) < 0) {
      ERRPRINTF("Error communicating with the server\n");
      goto error;
    }
    
    /* check whether this is the last problem */
    if(!strcmp(model_strings[i], GS_END_PROB_REG))
      break;

    i++;
  }

  i = 0;
  for(;;) {

    if(gs_recv_string(sock, &removed_probs[i]) < 0) {
      ERRPRINTF("Error communicating with the server\n");
      goto error;
    }

    /* check whether this is the last problem */
    if(!strcmp(removed_probs[i], GS_END_PROB_REG)) {
      break;
    }

    i++;
  }

  num_removed = i;

  /* Actual storage */
  if(gs_register_problem_changes(gs_agent, problem, num_services,
        model_strings, removed_probs, num_removed, temp_cid) < 0) {
    ERRPRINTF("Failed to add problem to database\n");
    goto error;
  }

  /* Send reply tag */
  if(gs_send_tag(sock, GS_PROT_OK) < 0)
    goto error;

  FREE(temp_cid);
  return 0;

error:
  ERRPRINTF("Agent could not process problem registration \n");

  if(problem) {
    for(i=0;i<num_services;i++)
      if(problem[i])
        gs_free_problem(problem[i]);
    free(problem);
  }

  FREE(temp_cid);
  gs_send_tag(sock, GS_PROT_ERROR);
  return -1;
}

int
gs_agent_process_kill_agent(gs_agent_t *gs_agent, int sock)
{
  char *msg;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  /* If authenticated, authorize to kill */

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("Error receving Password \n");
    return -1;
  }

  if(!strcmp(msg, "GridSolve")) {
    free(msg);

    if(gs_send_tag(sock, GS_PROT_OK) < 0) {
      ERRPRINTF("Error sending confirmation \n");
      return -1;
    }

    ERRPRINTF("Agent terminating...\n");


    gs_close_socket(gs_agent->sock);

    kill(getppid(),SIGTERM);
  } 
  else {
    free(msg);

    if(gs_send_tag(sock, GS_PROT_ERROR) < 0) {
      ERRPRINTF("Error sending confirmation \n");
      return -1;
    }
  }

  return 0;
}


int
gs_agent_process_kill_server(gs_agent_t *gs_agent, int sock)
{
  gs_server_t **server_list = NULL;
  int i, count, tag;
  char *msg;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  if(gs_recv_string(sock, &msg) < 0) {
    ERRPRINTF("Error receving Server\n");
    return -1;
  }

  if((strlen(msg) >= 2) && !strncmp(msg, "-c", 2)) {
    LOGPRINTF("GS_PROT_KILL_SERVER request for server cid '%s': ", msg+2);

    server_list = (gs_server_t **) calloc(1, sizeof(gs_server_t *));
    if(!server_list) {
      ERRPRINTF("malloc failed\n");
      free(msg);
      return -1;
    }

    server_list[0] = (gs_server_t *) calloc(1, sizeof(gs_server_t));
    if(!server_list[0]) {
      ERRPRINTF("malloc failed\n");
      free(server_list);
      free(msg);
      return -1;
    }

    if(gs_get_server_by_cid(gs_agent, msg+2, server_list[0]) < 0) {
      ERRPRINTF("No such server: '%s'.\n", msg+2);

      free(server_list[0]);
      free(server_list);
      free(msg);

      if(gs_send_tag(sock, GS_PROT_UNKNOWN_SERVER) < 0) {
        ERRPRINTF("Unsuccessful (Sending No Server tag)\n");
        return -1;
      }

      return -1;
    }

    count = 1;
  }
  else {
    LOGPRINTF("GS_PROT_KILL_SERVER request for server '%s': ", msg);

    count = gs_get_all_servers_by_hostname(gs_agent, msg, &server_list, &count);

    if(count < 1) {
      ERRPRINTF("No such server: '%s'.\n", msg);

      free(msg);

      if(gs_send_tag(sock, GS_PROT_UNKNOWN_SERVER) < 0) {
        ERRPRINTF("Unsuccessful (Sending No Server tag)\n");
        return -1;
      }

      return -1;
    }

    if(count > 1) {
      ERRPRINTF("Multiple matches for server: '%s'.\n", msg);

      free(msg);

      for(i = 0; i < count; i++)
        gs_server_free(server_list[i]);
      FREE(server_list);

      if(gs_send_tag(sock, GS_PROT_MULTIPLE_SERVER) < 0) {
        ERRPRINTF("Unsuccessful (Sending Multiple Server tag)\n");
        return -1;
      }

      return -1;
    }
  }

  free(msg);

  if(gs_send_tag(sock, GS_PROT_OK) < 0) {
    ERRPRINTF("Unsuccessful (Sending No Server)\n");
    return -1;
  }

  if(gs_encode_server(&msg, server_list[0]) < 0) {
    ERRPRINTF("Unsuccessful (Encoding Server)\n");
    gs_server_free(server_list[0]);
    return -1;
  }

  if(gs_send_string(sock, msg) < 0) {
    ERRPRINTF("Unsuccessful (Sending Server)\n");
    free(msg);
    gs_server_free(server_list[0]);
    return -1;
  }

  free(msg);

  if(gs_recv_tag(sock, &tag) < 0) {
    ERRPRINTF("Error receving Confirmation\n");
    return -1;
  }

  if(tag != GS_PROT_OK) {
    ERRPRINTF("Unsuccessful (rejected response from client).\n");
    gs_server_free(server_list[0]);
    return -1;
  }

  gs_delete_server(gs_agent, server_list[0]);
  gs_server_free(server_list[0]);

  LOGPRINTF("Server successfully killed.\n");

  FREE(server_list);
  return 0;
}



#ifdef GS_SMART_GRIDSOLVE

int gs_smart_fault_update_pm(gs_agent_t * gs_agent, int sock){
 


  char *serverstr = NULL;
  gs_server_t *gs_server;
  time_t clock;
  struct tm *now;
  char subtime[128];

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  gs_server = (gs_server_t *) CALLOC(1, sizeof(gs_server_t));

  if(!gs_server) {
    ERRPRINTF("Failed to malloc server struct\n");
    return -1;
  }

  DBGPRINTF("Entering \n");

  /* Receive server string and convert into structure */
  if((gs_recv_string(sock, &serverstr) < 0) ||
     (gs_decode_server(serverstr, gs_server) < 0))
    return -1;

  FREE(serverstr);


 
  LOGPRINTF("SMART: Server %s has failed during execution .\n", gs_server->hostname);
      
  if(gs_delete_server(gs_agent, gs_server) < 0){
    ERRPRINTF("Failed to delete server '%s'\n", gs_server->hostname);
    return -1;        
  }
  LOGPRINTF("SMART: Server %s has been removed from database .\n", gs_server->hostname);

  gs_server_free(gs_server);


  if(gs_send_tag(sock, GS_PROT_OK) < 0)
    return -1;


}










/*
 * This function adds the idl information to the application performance model.
 * This idl information is added to the problem structure of each app_pm node when 
 * the gs_get_server_list is called.
 *
 *
 * @param gs_agent --- The agent data structure
 * @param app_pm -- The application performance model
 *
 * @returns 0 on success, -1 on failure
 *
 */

int gs_smart_add_idl_info_to_app_pm(gs_agent_t * gs_agent , gs_smart_app_pm * app_pm){
  gs_server_t **server_list;
  int count;
  int i;
  char * app_pm_problem;
  
  for(i=0; i<app_pm->nb_tasks;i++){
    app_pm->tasks[i]->problem = (gs_problem_t *)calloc(1, sizeof(gs_problem_t));
    if(!app_pm->tasks[i]->problem){
      ERRPRINTF("SMART: Error allocating to problem %d\n", i);
      return -1;
    }
    app_pm->tasks[i]->problem->name=app_pm->tasks[i]->nickname;
    if(gs_encode_problem(&app_pm_problem, app_pm->tasks[i]->problem) < 0){
        ERRPRINTF("SMART: Error encoding problem\n");
        return -1;
      }

    count = gs_get_server_list(gs_agent, app_pm->tasks[i]->problem, NULL, &server_list, &count);
  }
  return 0;

}


/*
 * This function adds the list of servers that can execute each remote task 
 * to each remote task node of the task graph. 
 *
 *
 * @param gs_agent --- The agent data structure
 * @param task -- The task graph structure which represents the group of task
 *                which will be mapped.
 *
 * @returns 0 on success, -1 on failure
 *
 */



int gs_smart_add_servers_to_tg(gs_agent_t * gs_agent, gs_smart_tg * task_graph){
  int i, count, node_index=0;
  gs_server_t ***server_list;
  char *client_criteria = NULL;
  server_list= (gs_server_t ***)calloc(task_graph->nb_nodes , sizeof(gs_server_t **));
  if(!server_list){
    ERRPRINTF("SMART: Error allocating to server list\n");
    return -1;
  }
  for(i=0; i<task_graph->nb_nodes;i++){
    if(task_graph->task_nodes[i]->node_type==GS_SMART_TG_REM_TASK_NODE){
      count = gs_get_server_list(gs_agent, task_graph->task_nodes[i]->tg_rem_task_node->problem, client_criteria, &server_list[node_index], &count);
      if( ( !server_list[node_index] ) || ( count<=0 ) ){
        ERRPRINTF("SMART: Error receiving server list or no servers in list\n");
        return -1;
      }
      task_graph->task_nodes[i]->tg_rem_task_node->nb_servers=count;
      task_graph->task_nodes[i]->tg_rem_task_node->avail_servers=server_list[node_index];
      node_index++;
    }
  }
  return 0;
}



int
gs_smart_agent_process_app_pm_model(gs_agent_t * gs_agent, int sock){
  gs_smart_tg * tg;
  gs_server_t **server_list = NULL;
  gs_smart_app_pm *  app_pm;
  char * app_pm_str=NULL;
  int tag, i, count;
  int graph_type;
  char * comm_type=NULL; 
 /*
   * Set graph type to the type of
   * task graph that should be generated.
   * This variable determines what dependencies
   * get anaylzed when generating the task graph
   * If set to GS_SMART_NO_DEP_GRAPH, then the
   * task graph will be generated with no
   * dependencies and therefore the mapping heuristic
   * is unable to generate solution which implement
   * remote communication.
   */

//  graph_type=GS_SMART_NO_DEP_GRAPH;
// graph_type=GS_SMART_P2P_GRAPH;
//  graph_type=GS_SMART_SRV_BRDCST_GRAPH;


  if(gs_recv_string(sock, &(comm_type)) < 0) {
    ERRPRINTF("SMART: Error receiving xml info string\n");
    return -1;
  }
  if(strcmp(comm_type,"enable_remote_comm")==0){
    graph_type=GS_SMART_CL_SRV_BRDCST_GRAPH;
  }
  else if(strcmp(comm_type, "no_dep")==0){
    graph_type=GS_SMART_NO_DEP_GRAPH;
  }
  else if(strcmp(comm_type, "server_comm")==0){
    graph_type=GS_SMART_SRV_BRDCST_GRAPH;
  }
  else if(strcmp(comm_type, "disable_remote_comm")==0){
    graph_type=GS_SMART_CLNT_BRDCAST_COMM;
  }
  else{
    graph_type=GS_SMART_CL_SRV_BRDCST_GRAPH;
  }
  app_pm=(gs_smart_app_pm *)calloc(1,sizeof(gs_smart_app_pm));
  
  if(!gs_agent) {
    ERRPRINTF("SMART: Invalid arg: null agent\n");
    if(app_pm_str) free(app_pm_str);
    if(app_pm) gs_smart_app_pm_free(app_pm);
    return -1;
  }

  if(gs_recv_string(sock, &(app_pm_str)) < 0) {
    ERRPRINTF("SMART: Error receiving app_pm string\n");
    if(app_pm_str) free(app_pm_str);
    if(app_pm) gs_smart_app_pm_free(app_pm);
    return -1;
  }
  
  if(gs_smart_decode_app_pm(app_pm_str,app_pm)<0){
    ERRPRINTF("SMART: Error decoding app_pm string\n");
    if(app_pm_str) free(app_pm_str);
    if(app_pm) gs_smart_app_pm_free(app_pm);
    return -1;
  }

  if(app_pm_str) free(app_pm_str);
  tg = (gs_smart_tg *)calloc(1, sizeof(gs_smart_tg ));

  if(!tg){
    ERRPRINTF("SMART: Error allocating memory for task graph\n");
    if(app_pm_str) free(app_pm_str);
    if(app_pm) gs_smart_app_pm_free(app_pm);
    if(tg) gs_smart_tg_free(tg);
    return -1;
  }

  tag=GS_SMART_GRAPH_OK;

  gs_problem_t **problem_list;
  problem_list=(gs_problem_t **)calloc(app_pm->nb_tasks , sizeof(gs_problem_t *));

  if(!problem_list){
    ERRPRINTF("SMART: Error allocating to problem list\n");
    return -1;
  }

  for(i=0; i<app_pm->nb_tasks;i++){
    problem_list[i] = (gs_problem_t *)calloc(1, sizeof(gs_problem_t));
    if(!problem_list[i]){
      ERRPRINTF("SMART: Error allocating to problem %d\n", i);
      return -1;
    }
  }
  
  if(gs_smart_add_idl_info_to_app_pm(gs_agent, app_pm)<0){
    ERRPRINTF("SMART: Error getting problem list for application performance model\n");
    return -1;
  }

  if(gs_smart_tg_construct_tg(app_pm, tg, graph_type)<0){
    ERRPRINTF("SMART: Constructing task nodes for task graph\n");
     tag=GS_SMART_GRAPH_FAIL;
  }

  if(tag==GS_SMART_GRAPH_OK){
    if(gs_smart_add_servers_to_tg(gs_agent, tg)<0){
      ERRPRINTF("SMART: Error adding servers to task graph\n");
      tag=GS_SMART_GRAPH_FAIL;
    }
  }
  if(gs_send_tag(sock, tag)<0){
    ERRPRINTF("SMART: Error sending tag\n");
    if(tg) gs_smart_tg_free(tg);
    return -1;
  }
  if(tag==GS_SMART_GRAPH_OK){
    DBGPRINTF("SMART: Task graph created successfully\n");
  }
  if(tag==GS_SMART_GRAPH_FAIL){
    ERRPRINTF("SMART: Error creating task graph. This may be due to no servers registered\n");
    if(tg) gs_smart_tg_free(tg);
    return -1;
  }

  if(gs_smart_send_tg(sock, tg)<0){
    ERRPRINTF("SMART: gs_send_task_graph failed\n");
    if(tg) gs_smart_tg_free(tg);
    return -1;
  }

  count = gs_get_all_servers(gs_agent, &server_list, &count);
  if(count < 0) {
    ERRPRINTF("failed to get list of all servers\n");
    return -1;
  }

  if(gs_smart_send_all_servers(sock, server_list, count)<0){
    ERRPRINTF("SMART: Error sending server link info\n");
    return -1;
  }

  for(i = 0; i < count; i++)
    gs_server_free(server_list[i]);
    
  FREE(server_list);

  if(gs_send_tag(sock, GS_PROT_OK) < 0)
    return -1;


  if(app_pm){
    if(gs_smart_app_pm_free(app_pm)<0){
      ERRPRINTF("SMART : Error freeing application performance model\n");
      return -1;
    }
  }

  if(tg){
    if(gs_smart_tg_free(tg)<0){
      ERRPRINTF("SMART: Error Freeing task graph\n");
      return -1;
    }
  }

  return 0;
}

#endif






int
gs_agent_process_message(gs_agent_t * gs_agent, int sock)
{
  char *version_str;
  int tag, retval;

  if((gs_recv_tag(sock, &tag)) < 0) {
    ERRPRINTF("could not get tag\n");
    return -1;
  }

  if((gs_recv_string(sock, &version_str)) < 0) {
    ERRPRINTF("could not read version string\n");
    return -1;
  }

  if(gs_versions_incompatible(version_str, VERSION))
    retval = gs_send_tag(sock, GS_PROT_VERSION_MISMATCH);
  else
    retval = gs_send_tag(sock, GS_PROT_OK);

  free(version_str);

  if(retval < 0) {
    ERRPRINTF("could not send response tag.\n");
    return -1;
  }

  if(gs_storage_init(gs_agent) < 0) {
    ERRPRINTF("Could not init database.\n");
    return -1;
  }
  
  retval = -1;

  switch (tag) {
    case GS_PROT_PROBLEM_SUBMIT:
      retval = gs_agent_process_problem_submit(gs_agent, sock);
      break;
    case GS_PROT_SERVER_REGISTER:
      retval = gs_agent_process_server_registration(gs_agent, sock);
      break;
    case GS_PROT_PROBLEM_REGISTER:
      retval = gs_agent_process_problem_registration(gs_agent, sock);
      break;
    case GS_PROT_WORKLOAD_REPORT:
      retval = gs_agent_process_workload_report(gs_agent, sock);
      break;
    case GS_PROT_KILL_AGENT:
      retval = gs_agent_process_kill_agent(gs_agent, sock);
      break;
    case GS_PROT_KILL_SERVER:
      retval = gs_agent_process_kill_server(gs_agent, sock);
      break;
    case GS_PROT_SERVER_LIST:
      retval = gs_agent_process_server_list(gs_agent, sock);
      break;
    case GS_PROT_SERVER_PING_LIST:
      retval = gs_agent_process_server_ping_list(gs_agent, sock);
      break;
    case GS_PROT_SERVER_PING_UPDATE:
      retval = gs_agent_process_server_ping_update(gs_agent, sock);
      break;
    case GS_PROT_PROBLEM_LIST:
      retval = gs_agent_process_problem_list(gs_agent, sock);
      break;
    case GS_PROT_PROBLEM_DESC:
      retval = gs_agent_process_problem_desc(gs_agent, sock);
      break;
    case GS_PROT_NOTIFY_SUBMIT:
      retval = gs_agent_process_notify_submit(sock);
      break;
    case GS_PROT_NOTIFY_FAILURE:
      retval = gs_agent_process_notify_failure(sock);
      break;
    case GS_PROT_NOTIFY_COMPLETE:
      retval = gs_agent_process_notify_complete(sock);
      break;
    case GS_PROT_NOTIFY_CANCEL:
      retval = gs_agent_process_notify_cancel(sock);
      break;
    case GS_PROT_AVAILABILITY_REQ:
      retval = gs_agent_process_availability_request(sock);
      break;
#ifdef GS_SMART_GRIDSOLVE
    case GS_PROT_APP_PM_MODEL:
      retval = gs_smart_agent_process_app_pm_model(gs_agent, sock);
      break;
    case  GS_SMART_FAULT_UPDATE_PM:
      retval = gs_smart_fault_update_pm(gs_agent, sock);
      break;
#endif
    default:
      LOGPRINTF("Unknown tag %d\n", tag);
      break;
  }

  gs_storage_finalize(gs_agent);

  return retval;
}

int
gs_agent_listen_and_process_messages(gs_agent_t *gs_agent)
{
  int sock = -1;
  int accept_error = 0;

  if(!gs_agent) {
    ERRPRINTF("Invalid arg: null agent\n");
    return -1;
  }

  /* Initialize proxy library */
  proxy_init("");

  gs_agent->sock = gs_establish_socket(&(gs_agent->port), TRUE);

  if(gs_agent->sock == -1) {
    ERRPRINTF("Could not bind to port %d \n", gs_agent->port);
    return (-1);
  }
  LOGPRINTF("Agent %s listening at port %d\n",
            gs_agent->hostname, gs_agent->port);

  gs_listen_on_socket(gs_agent->sock);

  for(;;) {
    pid_t pid;

    /* if accept fails, go back to while loop */
    if((sock = gs_accept_connection(gs_agent->sock)) == -1) {
      /* Print only one copy of this error message */
      if(accept_error != 1)
        ERRPRINTF
            ("Failed to accept connection on socket, return to listening \n");
      accept_error = 1;
      continue;
    }
    else {
      accept_error = 0;
    }

    /* increment task id before forking. */
    global_taskid++;

    pid = fork();

    switch(pid) {
      case -1:     /* error */
        ERRPRINTF("Failed to fork\n");
        gs_close_socket(sock);
        continue;
      case 0:      /* child */
        gs_close_socket(gs_agent->sock);
        gs_agent_process_message(gs_agent, sock);
        _exit(0);
      default:
        gs_close_socket(sock);
    }

    fflush(stderr);
    fflush(stdout);
  }

  return 0;
}

gs_agent_scheduler_t
gs_agent_parse_scheduler_name(char *name)
{
  if(!name) return GS_INVALID_SCHEDULER;

  if(!strcasecmp(name, "default_mct"))
    return GS_DEFAULT_MCT;

  /* if using one of the HTM schedulers, set the flag to enable
   * keeping track of task events (start time, end time, etc).
   */
  if(!strncasecmp(name, "htm", 3)) {
    keep_track_of_task_events = 1;

    if(!strcasecmp(name, "htm_ml"))
      return GS_HTM_ML;

    if(!strcasecmp(name, "htm_msf"))
      return GS_HTM_MSF;

    if(!strcasecmp(name, "htm_hmct"))
      return GS_HTM_HMCT;

    if(!strcasecmp(name, "htm_mp"))
      return GS_HTM_MP;
  }

  return GS_INVALID_SCHEDULER;
}

int
gs_agent_parse_cmd_line(int argc, char **argv, char **logfile, int *daemon,
  int *httpd_port, char **config_file)
{
  char *httpd_port_env;
  int c;

  if(!argv || !logfile || !daemon) {
    ERRPRINTF("Invalid args\n");
    return -1;
  }

  *logfile = NULL;
  *daemon = 1;
  *config_file = NULL;

  /* check if the GRIDSOLVE_HTTPD_PORT is set to "disable" */

  if((httpd_port_env = getenv("GRIDSOLVE_HTTPD_PORT")) &&
     (!strcmp(httpd_port_env, "disable")))
    *httpd_port = -1;
  else
    *httpd_port = getenv_int("GRIDSOLVE_HTTPD_PORT", 
         GRIDSOLVE_HTTPD_PORT_DEFAULT);

  /* when making changes to the command line args, update 
   * GS_AGENT_USAGE_STR so the usage information is printed 
   * correctly upon error.
   */

#define GS_AGENT_USAGE_STR "Usage: GS_agent [-l logfile] [-c] [-w httpd port] [-s config file]"

  while((c = getopt(argc,argv,"cl:w:s:")) != EOF) {
    switch(c) {
      case 'w':
        if(!strcmp(optarg, "disable"))
          *httpd_port = -1;
        else
          *httpd_port = atoi(optarg);
        break;
      case 's':
        *config_file = strdup(optarg);
        break;
      case 'l':
        *logfile = strdup(optarg);
        break;
      case 'c':
        *daemon = 0;
        break;
      case '?':
        return -1;
        break;
      default:
        ERRPRINTF("Bad arg: '%c'.\n",c);
        return -1;
    }
  }

  /* Setup default server config file name if necessary */
  if(!*config_file) {
    char *gridsolve_root;

    if((gridsolve_root = getenv("GRIDSOLVE_ROOT")) == NULL)
      gridsolve_root = GRIDSOLVE_TOP_BUILD_DIR;
    if(!gridsolve_root) {
      ERRPRINTF("Warning: GRIDSOLVE_ROOT unknown, assuming cwd.\n");
      gridsolve_root = strdup(".");
    }

    *config_file = dstring_sprintf("%s/agent_config", gridsolve_root);
    if(!*config_file) {
      fprintf(stderr,"Error generating agent config file name.\n");
      exit(EXIT_FAILURE);
    }
  } 

  return 0;
}

int
gs_spawn_http_server(gs_agent_t *gs_agent, int httpd_port)
{
  void **httpd_args;
  int *port;
  pid_t pid;

  httpd_args = (void **) malloc(2 * sizeof(void *));
  if(!httpd_args) {
    ERRPRINTF("Failed to allocate memory for child args\n");
    return -1;
  }

  port = (int *)malloc(sizeof(int));
  if(!port) {
    ERRPRINTF("Failed to allocate memory for port arg\n");
    return -1;
  }

  *port = httpd_port;

  httpd_args[0] = port;
  httpd_args[1] = gs_agent;

  /* fork http server process */
  pid = mfork(gs_agent_httpd, -1, httpd_args, NULL, NULL, NULL, 10);

  if(pid < 0) {
    fprintf(stderr,"Failed to fork http server.\n");
    return -1;
  }

  return 0;
}

int
gs_spawn_server_expiration(gs_agent_t *gs_agent, int timeout)
{
  extern void gs_server_expiration_pre(void**);
  extern void gs_server_expiration_post(void**);
  void **se_args;
  pid_t pid;
  int *tout;

  se_args = (void **) malloc(2 * sizeof(void *));
  if(!se_args) {
    ERRPRINTF("Failed to allocate memory for child args\n");
    return -1;
  }

  tout = (int *) malloc(sizeof(int));
  if(!tout) {
    ERRPRINTF("Failed to allocate memory for timeout arg\n");
    return -1;
  }

  *tout = timeout;

  se_args[0] = gs_agent;
  se_args[1] = tout;
  
  if(gs_signal(SIGHUP, gs_temp_sighup_handler) == SIG_ERR)
    ERRPRINTF("error setting up temp signal handler\n");

  pid = mfork(gs_server_expiration, GS_EXPIRE_FREQUENCY, se_args,
    gs_server_expiration_pre, NULL, gs_server_expiration_post, 30);

  if(gs_signal(SIGHUP, gs_agent_generic_signal_handler) == SIG_ERR)
    ERRPRINTF("error restoring signal handler\n");

  if(pid < 0) {
    ERRPRINTF("Failed to fork server expiration process.\n");
    return -1;
  }

  return 0;
}

int
gs_spawn_sensor(gs_agent_t *gs_agent)
{
  struct sockaddr_un sensoraddr;
  void **sens_args;
  pid_t pid;
  int err;

  sens_args = (void **) malloc(2 * sizeof(void *));
  if(!sens_args) {
    ERRPRINTF("Failed to allocate memory for child args\n");
    return -1;
  }
  sens_args[0] = NULL;
  sens_args[1] = gs_agent;

  /* fork sensor process  */
  pid = mfork(gs_agent_sensor_run, -1, sens_args,
    gs_agent_sensor_pre, gs_agent_sensor_post,
    gs_agent_sensor_exit, 30);

  if(pid < 0) {
    ERRPRINTF("Failed to fork sensor process.\n");
    return -1;
  }

  /* connect to sensor */
  GS_SENSORFD = socket(PF_UNIX, SOCK_STREAM, 0);
  if ( GS_SENSORFD < 0 )
  {
    ERRPRINTF("Could not open socket to sensor.\n");
    return -1;
  }

  memset(&sensoraddr, 0x0, sizeof(struct sockaddr_un));
  sensoraddr.sun_family = PF_UNIX;
  strcpy(sensoraddr.sun_path, GRIDSOLVE_SENSOR_USOCK);

  err = connect(GS_SENSORFD, (struct sockaddr*) &sensoraddr,
                sizeof(struct sockaddr_un));
  if ( err < 0 )
  {
    ERRPRINTF("Warning: Connect on sensor socket failed.");
    ERRPRINTF("  logging events to stdout.\n");
    GS_SENSORFILE = stdout;
  }
  else {
    GS_SENSORFILE = fdopen(GS_SENSORFD, "w+");
    LOGPRINTF("Connected to sensor.\n");
  }

  return 0;
}

int
main(int argc, char **argv)
{
  gs_agent_t *gs_agent = NULL;
  int status = -1, daemon;
  int httpd_port;
  char *logfile, *gridsolve_root;

  gs_agent_scheduler_selection = GS_DEFAULT_MCT;

  gs_setup_signal_handlers(gs_agent_generic_signal_handler);

  if((gridsolve_root = getenv("GRIDSOLVE_ROOT")) == NULL) 
    gridsolve_root = GRIDSOLVE_TOP_BUILD_DIR;

  if(!gridsolve_root) {
    ERRPRINTF("Warning: GRIDSOLVE_ROOT unknown, assuming cwd.\n");
    gridsolve_root = strdup(".");
  }

  if(gs_agent_parse_cmd_line(argc, argv, &logfile, &daemon, &httpd_port,
       &agent_cfg) < 0) {
    fprintf(stderr, "%s\n", GS_AGENT_USAGE_STR);
    exit(EXIT_FAILURE);
  }

  if(!logfile) {
    logfile = dstring_sprintf("%s/gs_agent.log", gridsolve_root);
    if(!logfile) {
      ERRPRINTF("Error generating log file name.\n");
      exit(EXIT_FAILURE);
    }
  }

  if(daemon && gs_daemon_init(gridsolve_root, logfile) < 0) {
    fprintf(stderr, "Failed to start agent as a daemon.\n");
    exit(EXIT_FAILURE);
  }

  gs_agent = gs_agent_init(agent_cfg);
  if(!gs_agent) {
    ERRPRINTF("Failed to initialize agent\n");
    exit(EXIT_FAILURE);
  }

  if(gs_agent_scheduler_selection == GS_INVALID_SCHEDULER) {
    fprintf(stderr, "Invalid scheduler type specified.  Legal values are:\n");
    fprintf(stderr, "  default_mct -- original gridsolve scheduler\n");
    fprintf(stderr, "  htm_ml -- HTM minimum length\n");
    fprintf(stderr, "  htm_msf -- HTM minimum sumflow\n");
    fprintf(stderr, "  htm_hmct -- HTM historical minimum completion time\n");
    fprintf(stderr, "  htm_mp -- HTM minimum perturbation\n");
    exit(EXIT_FAILURE);
  }

  gs_log_init(argv[0], -1, -1, "agent");

  /* if using MySQL, no need to spawn the db manager (for sqlite) */
#ifdef GS_USE_MYSQL
  if(gs_mysql_init_db(gs_agent) < 0) {
    printf("MySQL db could not be initialized\n");
    exit(-1);
  }
#else
  if(gs_init_db(&global_db) < 0) {
    ERRPRINTF("SQLite could not be initialized\n");
    exit(EXIT_FAILURE);
  }
#endif

  if(gs_spawn_sensor(gs_agent) < 0) {
    ERRPRINTF("Failed to spawn sensor\n");
    exit(EXIT_FAILURE);
  }

  if(gs_spawn_server_expiration(gs_agent, GS_EXPIRE_TIMEOUT) < 0) {
    ERRPRINTF("Failed to spawn server expiration process\n");
    exit(EXIT_FAILURE);
  }

  if(httpd_port >= 0)
    if(gs_spawn_http_server(gs_agent, httpd_port) < 0) {
      ERRPRINTF("Failed to spawn http server\n");
      exit(EXIT_FAILURE);
    }

  status = gs_agent_listen_and_process_messages(gs_agent);

  LOGPRINTF("Agent %s exiting with status %d\n", gs_agent->hostname, status);

  exit(status);
}

int
gs_agent_init_conns(gs_agent_conn_t *conns)
{
  int i;

  if(!conns) {
    ERRPRINTF("Invalid arg: null connections ptr\n");
    return -1;
  }
  
  for(i=0; i < GS_AGENT_MAX_FD; i++)
    conns->fd[i] = 0; 
  conns->maxfd = 0;
    
  return 0;
}

int
gs_agent_add_conn(gs_agent_conn_t *conns, int fd)
{
  if(!conns) {
    ERRPRINTF("Invalid arg: null connections ptr\n");
    return -1;
  } 
    
  if(fd > GS_AGENT_MAX_FD) 
    return -1;

  conns->fd[fd] = 1;

  if(conns->maxfd < fd)
    conns->maxfd = fd;

  return 0;
}

int
gs_agent_del_conn(gs_agent_conn_t *conns, int fd)
{
  int i;

  if(!conns) {
    ERRPRINTF("Invalid arg: null connections ptr\n");
    return -1;
  }

  conns->fd[fd] = 0;

  if(fd == conns->maxfd) {
    for(i=fd;i>=0;i--)
      if(conns->fd[i]) {
        conns->maxfd = i;
        break;
      }

    if(i < 0)
      conns->maxfd = 0;
  }

  return 0;
}

int
gs_agent_setup_fd_sets(gs_agent_conn_t *connections, int listensock,
  int extrasock, fd_set *allset, int *maxfd)
{
  int i;

  if(!connections || !allset || !maxfd) {
    ERRPRINTF("Invalid args\n");
    return -1;
  }

  *maxfd = extrasock;
  if(listensock > *maxfd)
    *maxfd = listensock;

  FD_ZERO(allset);
  if(extrasock >= 0)
    FD_SET(extrasock, allset);
  if(listensock >= 0)
    FD_SET(listensock, allset);

  for(i=0; i <= connections->maxfd; i++) {
    if(connections->fd[i]) {
      FD_SET(i, allset);
      if(i > *maxfd)
        *maxfd = i;
    }
  }

  return 0;
}