gsgrpc.c

Go to the documentation of this file.

/*
 * $Id: gsgrpc.c,v 1.175 2010/04/27 16:43:56 tbrady Exp $ 
 */
/*
 * $UTK_Copyright: $ 
 */

#include <stdio.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

#define ERROR_MAIN

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif /* HAVE_CONFIG_H */

#include "portability.h"
#include "grpc.h"
#include "icl_hash.h"
#ifdef GS_SMART_GRIDSOLVE
#include "gs_smart_clib.h"
#include "gs_smart_mapping_graph.h"
#include "gs_smart_mapping_heuristics.h"
#include "gs_smart_app_pm_builder.h"
#endif

#ifndef TRUE
#define TRUE 1
#endif
#ifndef FALSE
#define FALSE 0
#endif

/*
 * va_copy is not portable.  GNU suggests the following work-around -JL
 * 10/13/2004 
 */
#if ! defined va_copy
#if defined __va_copy
#define va_copy __va_copy
#else
#define va_copy(dst, src) memcpy (&dst, &src, sizeof(va_list))
#endif
#endif




#ifdef GS_SMART_GRIDSOLVE

/*
 * call type indicates whether the one pass or two pass method for
 * mapping a group of tasks has been called
 */
  static int call_type;




 /*
  * map_call_type indicates whether the group of task to be mapped has been specified
  * explicitly as a parameter to the gs_smart_map function or implicitly by using either 
  * paranthesis or the gs_smart_end_map call.
  */
  static int group_type;  


/*
 * smart_phase indicates the operation are we currently doing 
 * (building, mapping, executing stored handles, executing called handles or standard gridsolve execution)
 * default is set to standard gridsolve execution
 */
  static int smart_phase=GS_SMART_STANDARD_EXEC;  



/*
 * smart_app_num is a unique number given to each execution of an application
 * this number is appended on to the file names where remote arguments are stored.
 * this is to ensure that file names of remote arguments are unique.
 * in addition the process id and unique argument id is also appended to the file name.
 */
  static int smart_app_num;


/*
 * handle_task_count variable is used when executing called handles 
 * in the second  pass on a two pass call type.
 * the variable is used to determine which handle the called handle is in 
 * the group of mapped handles.
 * this information is used to retrieve the relevant mapping information for
 * the called handle.
 */

  static int total_nb_mapped_tasks=0;
  

 static int nb_mapped_tasks_executed=0;

 /* type of communication scheme to implement in task graph*/ 
 char * set_comm_type;


#endif

/*
 * current_pass indicates which pass is currently been undertaken.
 *
 * for the two pass call type the first pass  (current_pass=0) will be the
 * task discovery phase.
 * the second pass (current_pass=1) will be first the mapping phase 
 * and then executing called handles phase.
 *
 * for the one pass call type the first pass will involve  all phases.
 * first the building phase, when all tasks in the group have been called
 * it enters the mapping phase. This is followed by the executing stored handles 
 * phase.
 */
  static int current_pass=0;




char *GRPC_ERROR_MESSAGES[] = {
  "Success",
  "GRPC client not initialized yet",
  "GRPC client already initialized",
  "Specified configuration file not found",
  "An error occurred parsing or processing the configuration file",
  "GRPC client cannot find any server",
  "GRPC client cannot find the function on the default server",
  "Function handle is not valid",
  "Session ID is not valid",
  "RPC invocation refused by the server",
  "Communication with the server failed somehow",
  "The specified session failed",
  "Call has not completed",
  "No calls have completed",
  "Internal error detected",
  "Error description string requested for an unknown error code",
  "Highest error code; used to bound error codes [does not denote actual error]"
};

/* note that minor error codes begin at 256, so to reference the string
 * in this array, just subtract GRPC_NO_MINOR_ERROR from the index
 * (e.g. GRPC_MINOR_ERROR_MESSAGES[err-GRPC_NO_MINOR_ERROR]).
 */
char *GRPC_MINOR_ERROR_MESSAGES[] = {
  "No minor error specified",
  "Function handle is NULL",
  "Function name is NULL",
  "Argument stack is NULL",
  "Out of memory",
  "Hostname could not be resolved",
  "GridSolve agent not set or unavailable",
  "Failed to encode submit request",
  "Failed to encode solve request",
  "Failed to decode problem structure",
  "Failed to encode problem structure",
  "Failed to decode server structure",
  "Failed to encode server structure",
  "Could not parse host name",
  "Could not establish connection to server",
  "Could not kill data transfer process",
  "Invalid serialized request",
  "Could not duplicate problem structure",
  "Profiling is not enabled",
  "Invalid (or NULL) major",
  "Invalid LANGUAGE",
  "No match found for session ID",
  "Iterator is NULL",
  "Could not setup farming args",
  "Could not build argument stack for farming",
  "Agent or server has incompatible version",
  "DSI Disabled",
  "DSI Manage Error",
  "DSI Unknown File",
  "DSI Allocation Error",
  "DSI Permission Denied",
  "DSI Load Error",
  "DSI Store Error",
  "DSI Internal Error",
  "Highest error code; used to bound error codes [does not denote actual error]"
};

#define GS_PORT_DELIM '/'
#define GS_PROXY_DELIM '%'
#define GS_MAX_PINGS 5
#define GS_DEFAULT_CACHE_TTL 300
#define GS_DEFAULT_TIME_THRESH 1.0

grpc_error_t grpc_errno = GRPC_NO_ERROR;
gs_service_error_enum_t grpc_minor_errno = GRPC_NO_MINOR_ERROR;
int grpc_client_major = 'r';
int grpc_client_lang = GS_CALL_FROM_C;
int grpc_user_set_major = FALSE;
grpc_profile_t *grpc_profile_next = NULL;
char grpc_user[GRPC_USER_INFO_LEN];
char grpc_host[GRPC_USER_INFO_LEN];
char grpc_domain[GRPC_USER_INFO_LEN];
char grpc_cid_str[2*CID_LEN+1];
int grpc_measure_comm = 1;
int grpc_measure_comm_num_servers = GS_MAX_PINGS;
int grpc_measure_comm_cache_ttl = GS_DEFAULT_CACHE_TTL;
double grpc_measure_comm_time_thresh = GS_DEFAULT_TIME_THRESH;
icl_hash_t *grpc_comm_cache = NULL;

static grpc_request_t *grpc_outstanding_requests[MAX_GRPC_REQUESTS];
static grpc_profile_t *grpc_profile_info[MAX_GRPC_REQUESTS];
static grpc_error_t grpc_errors[MAX_GRPC_REQUESTS];
static grpc_sessionid_t grpc_last_failed_sid[MAX_GRPC_REQUESTS];
static int grpc_last_failed_idx = -1;
static gs_service_error_enum_t grpc_minor_errors[MAX_GRPC_REQUESTS];
static int grpc_initialized = FALSE;
static agent_host_info_t *agent_resolved = NULL;

/*
 * Local function prototypes 
 */

static grpc_error_t
  gs_call_common(grpc_function_handle_t *, grpc_sessionid_t *, gs_va_list *, 
     void **, int),
  gs_call_common_ft(grpc_function_handle_t *, grpc_sessionid_t *, gs_va_list *, 
     void **, int),
  grpc_function_handle_common(grpc_function_handle_t *, char *, char *),
  gs_resubmit_common(int);

static int 
  gs_notify_agent_of_failure(grpc_function_handle_t *, char *),
  gs_notify_agent_of_cancel(grpc_function_handle_t *, char *),
  gs_get_server_mapping(char *, grpc_function_handle_t *, gs_va_list *,
                        void **, int, int, int),
  gs_parse_host_port(char *, ipaddr_t *, in_port_t *),
  gs_parse_host_info(char *, ipaddr_t *, in_port_t *, ipaddr_t *, in_port_t *,
                     char *), 
  grpc_request_destruct(grpc_request_t *), 
  gs_free_handle_server_list(grpc_function_handle_t *);

static void
  grpc_function_handle_clear(grpc_function_handle_t *);

#define GRPC_CLEAR_ERRORS() do { \
  memset(grpc_errors, 0, MAX_GRPC_REQUESTS * sizeof(*grpc_errors));\
  memset(grpc_minor_errors, 0, MAX_GRPC_REQUESTS * sizeof(*grpc_minor_errors));\
  } while(0);

#define GRPC_FAIL_IF_NOT_INITIALIZED(retval) \
  if(!grpc_initialized) { \
    grpc_errno = GRPC_NOT_INITIALIZED; \
    grpc_minor_errno = GRPC_NO_MINOR_ERROR; \
    return retval; \
  }

SOCKET
gs_connect_to_agent()
{
  char agent_cid[CID_LEN];
  struct timeval cur;

  /* all ones will match any component ID */
  memset(agent_cid, 0xFF, CID_LEN);

  gettimeofday(&cur, NULL);

  /* check if the cached entry has expired */

  if(agent_resolved) {
    if(cur.tv_sec - agent_resolved->creation_time.tv_sec > AGENT_IP_EXPIRATION) {
      free(agent_resolved->hostname);
      free(agent_resolved);
      agent_resolved = NULL;
    }
  }

  if(!agent_resolved) {
    struct hostent *hp;
    char *tmp;

    agent_resolved = (agent_host_info_t *) malloc(sizeof(agent_host_info_t));

    if(!agent_resolved) {
      grpc_errno = GRPC_OTHER_ERROR_CODE;
      grpc_minor_errno = GRPC_OUT_OF_MEMORY;
      return INVALID_SOCKET;
    }

    tmp = getenv("GRIDSOLVE_AGENT");

    agent_resolved->hostname = tmp ? strdup(tmp) : NULL;

    if(!agent_resolved->hostname) {
      grpc_errno = GRPC_OTHER_ERROR_CODE;
      grpc_minor_errno = GRPC_AGENT_NOT_SET;
      return INVALID_SOCKET;
    }

    agent_resolved->port =
        getenv_int("GRIDSOLVE_AGENT_PORT", GRIDSOLVE_AGENT_PORT_DEFAULT);

    if((hp = gethostbyname(agent_resolved->hostname)) == NULL) {
      ERRPRINTF("Could not resolve host name '%s'\n", agent_resolved->hostname);
      perror("gethostbyname()");
      grpc_errno = GRPC_RPC_REFUSED;
      grpc_minor_errno = GRPC_AGENT_NOT_SET;
      return INVALID_SOCKET;
    }

    memcpy((void *) &(agent_resolved->ipaddr), hp->h_addr_list[0], sizeof(ipaddr_t));

    agent_resolved->creation_time = cur;
  }

  return gs_connect_to_host(agent_cid, agent_resolved->ipaddr, agent_resolved->port, 0, 0);
}

grpc_error_t
grpc_process_config_file(char *config_file_name)
{
  gs_info_t *attr_list = NULL;
  gs_struct_stat stbuf;

  if(!config_file_name)
    GRPC_RETURN(GRPC_CONFIGFILE_NOT_FOUND, GRPC_NO_MINOR_ERROR);

  if(gs_stat(config_file_name, &stbuf) < 0)
    GRPC_RETURN(GRPC_CONFIGFILE_NOT_FOUND, GRPC_NO_MINOR_ERROR);

  if(gs_parse_config_file(config_file_name, &attr_list) != 0)
    GRPC_RETURN(GRPC_CONFIGFILE_ERROR, GRPC_NO_MINOR_ERROR);

  if(attr_list) {
    gs_info_t *p;

    for(p = attr_list; p != NULL; p = p->next) {
      if(!strcasecmp(p->type, "MEASURE_COMM"))
        grpc_measure_comm = strcasecmp(p->value, "y") == 0;
      else if(!strcasecmp(p->type, "MEASURE_COMM_NUM_SERVERS"))
        grpc_measure_comm_num_servers = atoi(p->value);
      else if(!strcasecmp(p->type, "MEASURE_COMM_CACHE_TTL"))
        grpc_measure_comm_cache_ttl = atoi(p->value);
      else if(!strcasecmp(p->type, "MEASURE_COMM_TIME_THRESH"))
        grpc_measure_comm_time_thresh = atof(p->value);
    }
  }

  gs_free_infolist(attr_list);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_initialize(char *config_file_name)
{
  PROXY_COMPONENTADDR proxy_info;
  char *username=NULL, *tmp_host;
  int i;

#ifdef GS_SMART_GRIDSOLVE
  /*
   * Reads and increments the application file number.
   * This number is used to give each remote argument a
   * unique file name.
   */ 
  if(gs_smart_read_update_app_no_file("/tmp/gs_app_num", &smart_app_num)<0)
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

#endif



  /*
   * Repeated initializations return GRPC_ALREADY_INITIALIZED 
   */
  if(grpc_initialized == TRUE)
    GRPC_RETURN(GRPC_ALREADY_INITIALIZED, GRPC_NO_MINOR_ERROR);

  if(config_file_name)
    if(grpc_process_config_file(config_file_name) != GRPC_NO_ERROR)
      GRPC_RETURN(grpc_errno, GRPC_NO_MINOR_ERROR);

  GRPC_CLEAR_ERRORS();

  grpc_errno = GRPC_NO_ERROR;
  grpc_minor_errno = GRPC_NO_MINOR_ERROR;

  for(i = 0; i < MAX_GRPC_REQUESTS; i++) {
    grpc_outstanding_requests[i] = NULL;
    grpc_profile_info[i] = NULL;
    grpc_errors[i] = GRPC_NO_ERROR;
    grpc_minor_errors[i] = GRPC_NO_MINOR_ERROR;
  }

  /* Startup winsock in win32 */
  initialize_sockets();

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

  if((username = gs_get_login_name())) {
    strncpy(grpc_user, username, GRPC_USER_INFO_LEN);
    grpc_user[GRPC_USER_INFO_LEN-1] = '\0';
    free(username);
  }
  else {
    strncpy(grpc_user, "unknown_user", GRPC_USER_INFO_LEN);
    grpc_user[GRPC_USER_INFO_LEN-1] = '\0';
  }

  tmp_host = gs_get_machine_name();
  if(!tmp_host) {
    strncpy(grpc_host, "unknown_host", GRPC_USER_INFO_LEN);
    grpc_host[GRPC_USER_INFO_LEN-1] = '\0';
    strncpy(grpc_domain, "unknown_domain", GRPC_USER_INFO_LEN);
    grpc_domain[GRPC_USER_INFO_LEN-1] = '\0';
  }
  else {
    char *p;

    strncpy(grpc_host, tmp_host, GRPC_USER_INFO_LEN);
    grpc_host[GRPC_USER_INFO_LEN-1] = '\0';
        
    if((p = strstr(grpc_host, "."))) {
      *p = '\0';
      strncpy(grpc_domain, p+1, GRPC_USER_INFO_LEN);
    }
    else
      strncpy(grpc_domain, "unknown_domain", GRPC_USER_INFO_LEN);

    grpc_domain[GRPC_USER_INFO_LEN-1] = '\0';

    free(tmp_host);
  }

  proxy_info = proxy_get_local_addr();

  proxy_cid_to_str(grpc_cid_str, proxy_info.ID);

  grpc_comm_cache = icl_hash_create(37, NULL);

  grpc_initialized = TRUE;

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_finalize()
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  cleanup_sockets();

  grpc_initialized = FALSE;

  if(grpc_comm_cache)
    icl_hash_destroy(grpc_comm_cache, free, free);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_set_criteria(grpc_function_handle_t *handle, char *c)
{
  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);

  if(c) {
    handle->criteria = strdup(c);
    if(!handle->criteria) {
      ERRPRINTF("Can't allocate memory, strdup failed\n");
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
    }
  }
  else {
    /*
     * Client has set criteria to NULL.  If there was already some criteria
     * set, then free it first. 
     */
    if(handle->criteria)
      free(handle->criteria);

    handle->criteria = NULL;
  }

  /*
   * since the criteria have changed, request that the next call using this
   * handle gets a new binding (i.e. server list). 
   */
  handle->bind_servers_at_call_time = 1;

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

static void
grpc_function_handle_clear(grpc_function_handle_t *handle)
{
  if(!handle)
    return;

  handle->func_name = NULL;
  handle->num_servers = 0;
  handle->srv_idx = 0;
  handle->server_list = NULL;
  handle->problem_desc = NULL;
  handle->criteria = NULL;
  handle->bind_servers_at_call_time = 0;
  handle->agent_taskid = -1;
  handle->num_calls = 0;
}

grpc_error_t
grpc_function_handle_default(grpc_function_handle_t *handle, char *func_name)
{
  grpc_function_handle_clear(handle);
  if(handle)
    handle->bind_servers_at_call_time = 0;
  return grpc_function_handle_common(handle, NULL, func_name);
}

grpc_error_t
grpc_function_handle_default_ns(grpc_function_handle_t *handle,
                                char *func_name)
{
  grpc_function_handle_clear(handle);
  /*
   * Use any non-blank hostname and set handle->bind_servers_at_call_time to
   * 1 
   */
  if(handle)
    handle->bind_servers_at_call_time = 1;
  return grpc_function_handle_common(handle, "bind_servers_at_call_time",
                                     func_name);
}

grpc_error_t
grpc_function_handle_init(grpc_function_handle_t *handle, char *host_name,
                          char *func_name)
{

#ifdef GS_SMART_GRIDSOLVE

  /*
   * if the task graph is being built then we do not want to bind the servers at call time
   */
  if(smart_phase==GS_SMART_TASK_DISCOVERY){
    return grpc_function_handle_default(handle, func_name);
  }

   
  /*
   * if we are in executing mode then we don't need to set up handles, just return with no errors
   */
  if(smart_phase==GS_SMART_EXEC_CALLED_HANDLES) {
    return grpc_function_handle_default(handle, func_name);
      GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }

  /*
   * else if we are not building or executing (i.e. we are executing in the standard gridsolve approach)
   * then just set up the handle_init as normal
   */
  else if(smart_phase==GS_SMART_STANDARD_EXEC)
  {   
    grpc_function_handle_clear(handle);
    if((handle) && strcmp(host_name, "bind_servers_at_call_time") == 0)
      handle->bind_servers_at_call_time = 1;
    else if(handle)
      handle->bind_servers_at_call_time = 0;
    return grpc_function_handle_common(handle, host_name, func_name);
  }
  
  /*
   * if none of the above set up handle_init as normal
   * 
   */
  else
  {   
    grpc_function_handle_clear(handle);
    if((handle) && strcmp(host_name, "bind_servers_at_call_time") == 0)
      handle->bind_servers_at_call_time = 1;
    else if(handle)
      handle->bind_servers_at_call_time = 0;
    return grpc_function_handle_common(handle, host_name, func_name);
  }



#else


  grpc_function_handle_clear(handle);
  if((handle) && strcmp(host_name, "bind_servers_at_call_time") == 0)
    handle->bind_servers_at_call_time = 1;
  else if(handle)
    handle->bind_servers_at_call_time = 0;
  return grpc_function_handle_common(handle, host_name, func_name);
#endif
}

static grpc_error_t
grpc_function_handle_common(grpc_function_handle_t *handle, char *host_name,
                            char *func_name)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);

  if(!func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  if(host_name) {
    int i, parse_host = 0;
    gs_server_t *srv;

    handle->srv_idx = 0;
    handle->num_servers = 1;
    handle->server_list = (gs_server_t **) malloc(sizeof(gs_server_t *));

    if(!handle->server_list)
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

    handle->server_list[handle->srv_idx] =
        (gs_server_t *) CALLOC(1, sizeof(gs_server_t));

    if(!handle->server_list[handle->srv_idx])
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

    for(i = 0; i < (int)strlen(host_name); i++) {
      if(host_name[i] == GS_PORT_DELIM || host_name[i] == GS_PROXY_DELIM) {
        parse_host = 1;
        break;
      }
    }

    srv = handle->server_list[handle->srv_idx];
    srv->arch = strdup("unknown");
    srv->problemlist = NULL;
    srv->agenthost = NULL;
    srv->sa_list = NULL;

    if(parse_host) {
      if(gs_parse_host_info(host_name, &srv->ipaddress, &srv->port,
                            &srv->proxyip, &srv->proxyport,
                            srv->componentid) < 0) {
        ERRPRINTF("Error parsing info\n");
        GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_PARSE_HOST_INFO);
      }
      srv->hostname = strdup("unspecified");
    }
    else {
      srv->hostname = strdup(host_name);
      srv->port =
          getenv_int("GRIDSOLVE_SERVER_PORT",
                     GRIDSOLVE_SERVER_PORT_DEFAULT);

      if(!handle->bind_servers_at_call_time) {
        struct hostent *hp;

        if((hp = gethostbyname(host_name)) == NULL) {
          ERRPRINTF("Can't resolve host name '%s'\n", host_name);
          GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_RESOLV_FAILED);
        }

        memcpy((void *) &(srv->ipaddress), hp->h_addr_list[0], 
          sizeof(srv->ipaddress));

        /* since there was only hostname specified, componentID is
         * unavailable.  So initialize it with the 'wildcard' value.
         */

        memset(srv->componentid, 0xFF, CID_LEN);
      }
    }

    handle->func_name = strdup(func_name);
    handle->problem_desc = NULL;
  }
  else {
    /*
     * Use NULLS as the arg and argptr and -1 as the dsig to indicate that
     * the input args are NOT to be transferred at this time. They are not
     * known/bound yet in the standard GridRPC protocol, so they could not be 
     * transferred yet. 
     */
    if(gs_get_server_mapping(func_name, handle, NULL, NULL, -1,
          grpc_client_lang, grpc_client_major) < 0) {
      if(grpc_errno == GRPC_SERVER_NOT_FOUND ||
         grpc_errno == GRPC_NOT_INITIALIZED ||
         grpc_errno == GRPC_FUNCTION_NOT_FOUND)
        GRPC_RETURN(grpc_errno, grpc_minor_errno);
 
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);
    }

    GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_function_handle_destruct(grpc_function_handle_t *handle)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!handle)
    GRPC_RETURN(GRPC_INVALID_FUNCTION_HANDLE, GRPC_NULL_FUNCTION_HANDLE);

  if(handle->func_name)
    free(handle->func_name);
  gs_free_problem(handle->problem_desc);
  gs_free_handle_server_list(handle);

  if(handle->criteria) {
    free(handle->criteria);
    handle->criteria = NULL;
  }

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_get_handle(grpc_function_handle_t **handle, grpc_sessionid_t sessionId)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionId < 0) || (sessionId >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(!grpc_outstanding_requests[sessionId])
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  *handle = grpc_outstanding_requests[sessionId]->handle;

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_request_t *
grpc_get_request(grpc_sessionid_t sessionId)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(NULL);

  if((sessionId < 0) || (sessionId >= MAX_GRPC_REQUESTS))
    return NULL;

  if(!grpc_outstanding_requests[sessionId])
    return NULL;

  return grpc_outstanding_requests[sessionId];
}

grpc_error_t
grpc_call(grpc_function_handle_t *handle, ...)
{

#ifdef GS_SMART_GRIDSOLVE

  if(smart_phase==GS_SMART_EXEC_TASK_FAIL){
    return GRPC_NO_ERROR;
  }
#endif

  gs_va_list argptr;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  va_start(argptr.args, handle);

  return gs_call_common(handle, NULL, &argptr, NULL, TRUE);
}

grpc_error_t
grpc_call_async(grpc_function_handle_t *handle, 
  grpc_sessionid_t *sessionId, ...)
{

#ifdef GS_SMART_GRIDSOLVE
  if(smart_phase==GS_SMART_EXEC_TASK_FAIL){
    return GRPC_NO_ERROR;
  }

#endif


  gs_va_list argptr;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  va_start(argptr.args, sessionId);

  return gs_call_common(handle, sessionId, &argptr, NULL, FALSE);
}


/*************************************************************************/
/*------ new GridRPC API functions for GridSolve request sequencing -----*/
/*************************************************************************/
grpc_error_t
grpc_call_arg_list(grpc_function_handle_t *handle, gs_va_list *list) {
        return gs_call_common(handle, NULL, list, NULL, TRUE);
}

grpc_error_t grpc_call_arg_list_async(grpc_function_handle_t *handle,
            grpc_sessionid_t *sessionId, gs_va_list *list) {
        return gs_call_common(handle, sessionId, list, NULL, FALSE);
}
/*************************************************************************/


grpc_error_t
grpc_call_arg_stack(grpc_function_handle_t *handle, grpc_arg_stack *stack)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!stack || !stack->args)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_ARGSTACK);
  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  return gs_call_common(handle, NULL, NULL, stack->args, TRUE);
}

grpc_error_t
grpc_call_arg_stack_async(grpc_function_handle_t *handle, 
   grpc_sessionid_t *sessionId, grpc_arg_stack *stack)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!stack || !stack->args) 
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_ARGSTACK);
  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  return gs_call_common(handle, sessionId, NULL, stack->args, FALSE);
}

static int
gs_req_child_waitpid(grpc_request_t *req, int options)
{
  pid_t pid, tmp_pid;
  int stat;

  pid = waitpid(req->s_pid, &stat, options);

  switch (pid) {
    case WAIT_ABANDONED:        /* -1 */
      /*
       * if we reach here, most likely it's because the user has their own
       * signal handler catching SIGCHLD signals and they're calling waitpid
       * on the process forked by grpc, so it's gone by the time we get here. 
       * in any case, just clear the s_pid field so we won't wait on it again
       * and return normally. 
       */
      tmp_pid = req->s_pid;
      req->s_pid = -1;
      return tmp_pid;

    case WAIT_TIMEOUT:      /* 0 */
      /*
       * the child process is still sending the data 
       */
      return 0;

    default:
      /*
       * the child process has completed 
       */
      tmp_pid = req->s_pid;
      req->s_pid = -1;
      return tmp_pid;
  }
}

grpc_error_t
grpc_probe(grpc_sessionid_t sessionID)
{
  grpc_function_handle_t *handle;
  gs_server_t *srv;
  SOCKET sock;
  int tag;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(!grpc_outstanding_requests[sessionID])
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(grpc_outstanding_requests[sessionID]->s_pid > 0) {
    if(gs_req_child_waitpid(grpc_outstanding_requests[sessionID], GS_WNOHANG) == 0) {
      /*
       * the child is still sending data to the parent 
       */
      GRPC_RETURN(GRPC_NOT_COMPLETED, GRPC_NO_MINOR_ERROR);
    }
  }

  handle = grpc_outstanding_requests[sessionID]->handle;

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);

  srv = handle->server_list[handle->srv_idx];

  sock = gs_connect_to_host(srv->componentid, srv->ipaddress, srv->port,
                            srv->proxyip, srv->proxyport);

  if(sock == INVALID_SOCKET)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SERVER_CONNECTION);

  if((gs_send_tag(sock, GS_PROT_PROBE_REQUEST) < 0) ||
     (gs_send_string(sock, VERSION) < 0)) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  }

  if(gs_recv_tag(sock, &tag) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  }
    
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      grpc_minor_errno = GRPC_VERSION_MISMATCH;
    else
      grpc_minor_errno = GRPC_NO_MINOR_ERROR;
    
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);
  } 

  if(gs_send_string(sock, grpc_outstanding_requests[sessionID]->request_id) < 0)
    goto probe_failure;

  if(gs_recv_tag(sock, &tag) < 0)
    goto probe_failure;

  proxy_close(sock);

  switch (tag) {
    case GS_PROT_OK:
      GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
    case GS_SVC_ERR_NOT_FINISHED:
      GRPC_RETURN(GRPC_NOT_COMPLETED, GRPC_NO_MINOR_ERROR);
    default:
      ERRPRINTF("Probe failed: %s\n", gs_service_error[tag]);
      if(gs_notify_agent_of_failure(handle, 
           grpc_outstanding_requests[sessionID]->request_id) < 0)
        ERRPRINTF("Warning: failed to notify agent of server failure.\n");
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, tag);
  }

probe_failure:
  proxy_close(sock);
  if(gs_notify_agent_of_failure(handle, 
       grpc_outstanding_requests[sessionID]->request_id) < 0)
    ERRPRINTF("Warning: failed to notify agent of server failure.\n");
  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  
  /*
   * shouldn't reach here 
   */
  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_probe_or(grpc_sessionid_t *idArray, size_t length, grpc_sessionid_t *idPtr)
{   
  gs_service_error_enum_t retval_minor;
  grpc_error_t retval;
  int i; 

  *idPtr = GRPC_SESSIONID_VOID;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);
  
  /*
   * first validate the numbers 
   */
  for(i = 0; i < (int)length; i++)
    if((idArray[i] < 0) || (idArray[i] >= MAX_GRPC_REQUESTS))
      GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  retval = GRPC_NONE_COMPLETED;
  retval_minor = GRPC_NO_MINOR_ERROR;

  for(i = 0; i < (int)length; i++) {
    grpc_sessionid_t s = idArray[i];
      
    if(grpc_outstanding_requests[s] != NULL) {
      grpc_error_t r;

      r = grpc_probe(s);

      if(r == GRPC_NO_ERROR) {
        *idPtr = s;
        GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
      }
      else if(r != GRPC_NOT_COMPLETED) {
        retval = r;
        retval_minor = grpc_minor_errno;
      }
    }
  }

  GRPC_RETURN(retval, retval_minor);
}

grpc_error_t
grpc_cancel(grpc_sessionid_t sessionID)
{
  grpc_function_handle_t *handle;
  gs_server_t *srv;
  SOCKET sock;
  int tag;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(!grpc_outstanding_requests[sessionID])
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(gs_notify_agent_of_cancel(grpc_outstanding_requests[sessionID]->handle, 
       grpc_outstanding_requests[sessionID]->request_id) < 0)
    ERRPRINTF("Warning: failed to notify agent of job cancellation.\n");

  if(grpc_outstanding_requests[sessionID]->s_pid > 0) {
    if(gs_req_child_waitpid(grpc_outstanding_requests[sessionID], GS_WNOHANG) == 0) {
      /*
       * the child is still sending data to the parent.  kill it. 
       */

      kill(grpc_outstanding_requests[sessionID]->s_pid, GS_SIGKILL);
    }
  }

  handle = grpc_outstanding_requests[sessionID]->handle;

  if(!handle)
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  srv = handle->server_list[handle->srv_idx];

  sock = gs_connect_to_host(srv->componentid, srv->ipaddress, srv->port,
                            srv->proxyip, srv->proxyport);

  if(sock == INVALID_SOCKET)
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_SERVER_CONNECTION);

  if((gs_send_tag(sock, GS_PROT_KILL_JOB) < 0) ||
     (gs_send_string(sock, VERSION) < 0)) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  }
            
  if(gs_recv_tag(sock, &tag) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  } 
    
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      grpc_minor_errno = GRPC_VERSION_MISMATCH;
    else
      grpc_minor_errno = GRPC_NO_MINOR_ERROR;

    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);
  } 

  if(gs_send_string(sock, grpc_outstanding_requests[sessionID]->request_id) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  }

  if(gs_recv_tag(sock, &tag) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
  }

  proxy_close(sock);

  switch (tag) {
    case GS_PROT_OK:
      grpc_request_destruct(grpc_outstanding_requests[sessionID]);
      free(grpc_outstanding_requests[sessionID]);
      grpc_outstanding_requests[sessionID] = NULL;
      grpc_profile_info[sessionID] = NULL;
      GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
    default:
      ERRPRINTF("cancel failed: %s\n", gs_service_error[tag]);
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, tag);
  }

  /*
   * shouldn't reach here 
   */
  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_cancel_all(void)
{
  grpc_error_t last_minor_error = GRPC_NO_MINOR_ERROR;
  int i, saw_failure = FALSE;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  for(i = 0; i < MAX_GRPC_REQUESTS; i++)
    if(grpc_outstanding_requests[i] != NULL)
      if(grpc_cancel(i) != GRPC_NO_ERROR) {
        saw_failure = TRUE;
        last_minor_error = grpc_minor_errno;
      }

  if(saw_failure)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, last_minor_error);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_wait(grpc_sessionid_t sessionID)
{
#ifdef GS_SMART_GRIDSOLVE
  if(smart_phase==GS_SMART_EXEC_TASK_FAIL){
    return GRPC_NO_ERROR;
  }

#endif
  gs_service_error_enum_t save_minor_errno;
  grpc_error_t status;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(!grpc_outstanding_requests[sessionID])
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  status = gs_wait_common(grpc_outstanding_requests[sessionID]);
  save_minor_errno = grpc_minor_errno;

#ifdef GS_PROFILING
  if(grpc_profile_info[sessionID]) {
    GRPC_TIMER_STOP(grpc_profile_info[sessionID]);
    grpc_profile_info[sessionID]->recv_output =
        GRPC_TIMER_ELAPSED(grpc_profile_info[sessionID]);
  }
#endif

  if(status != GRPC_NO_ERROR) {
    grpc_last_failed_idx = 0;
    grpc_last_failed_sid[grpc_last_failed_idx] = sessionID;
  }
  else 
    grpc_last_failed_idx = -1;

  grpc_request_destruct(grpc_outstanding_requests[sessionID]);
  free(grpc_outstanding_requests[sessionID]);
  grpc_outstanding_requests[sessionID] = NULL;
  grpc_profile_info[sessionID] = NULL;

  GRPC_RETURN(status, save_minor_errno);
}

grpc_error_t
grpc_wait_and(grpc_sessionid_t *idArray, size_t length)
{
  int i, tmp_idx, requests_remaining = TRUE, saw_failure = FALSE;
  grpc_sessionid_t tmp_sid[MAX_GRPC_REQUESTS];

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  tmp_idx = -1;

  /*
   * first validate the numbers 
   */
  for(i = 0; i < (int)length; i++)
    if((idArray[i] < 0) || (idArray[i] >= MAX_GRPC_REQUESTS))
      GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  while(requests_remaining) {
    requests_remaining = FALSE;

    for(i = 0; i < (int)length; i++) {
      grpc_sessionid_t s = idArray[i];

      if(grpc_outstanding_requests[s] != NULL) {
        requests_remaining = TRUE;

        /* if the job is no longer running (i.e. it either successfully
         * completed or it failed) call wait and check the error status.
         */

        if(grpc_probe(s) != GRPC_NOT_COMPLETED) {
          if(grpc_wait(s) != GRPC_NO_ERROR) {
            saw_failure = TRUE;
            grpc_errors[s] = grpc_errno;
            grpc_minor_errors[s] = grpc_minor_errno;
            tmp_idx++;
            tmp_sid[tmp_idx] = s;
          }
        }
      }
    }

    sleep(1);
  }

  /* copy failed session IDs to the global array */
  grpc_last_failed_idx = tmp_idx;
  memcpy(grpc_last_failed_sid, tmp_sid, (tmp_idx+1) * sizeof(*tmp_sid));

  GRPC_RETURN(saw_failure ? GRPC_SESSION_FAILED : GRPC_NO_ERROR,
     GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_wait_or(grpc_sessionid_t *idArray, size_t length, grpc_sessionid_t *idPtr)
{
  int i, jobCount = 0;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  /*
   * first validate the numbers 
   */
  for(i = 0; i < (int)length; i++)
    if((idArray[i] < 0) || (idArray[i] >= MAX_GRPC_REQUESTS))
      GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  for(;;) {

    for(i = 0; i < (int)length; i++) {
      grpc_sessionid_t s = idArray[i];

      if(grpc_outstanding_requests[s] != NULL) {
        /* if the job is no longer running (i.e. it either successfully
         * completed or it failed) call wait and check the error status.
         */

        if(grpc_probe(s) != GRPC_NOT_COMPLETED) {
          grpc_error_t r;

          *idPtr = s;

          r = grpc_wait(s);

          if(r != GRPC_NO_ERROR) {
            grpc_errors[s] = grpc_errno;
            grpc_minor_errors[s] = grpc_minor_errno;
          }

          GRPC_RETURN(r, grpc_minor_errno);
        }

        jobCount++;
      }
    }

    if(jobCount == 0) {
      *idPtr = -1;
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }

    sleep(1);
  }

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_wait_all(void)
{
  grpc_sessionid_t tmp_sid[MAX_GRPC_REQUESTS];
  int tmp_idx, saw_failure = FALSE;
  grpc_sessionid_t sid = 0;
  grpc_error_t r;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  tmp_idx = -1;

  for(;;) {
    r = grpc_wait_any(&sid);

    if(r != GRPC_NO_ERROR) {

      if(sid >= 0) {
        tmp_idx++;
        tmp_sid[tmp_idx] = sid;
      }

      /* if there are no more jobs to wait for */
      if(r == GRPC_OTHER_ERROR_CODE) {

        /* copy failed session IDs to the global array */
        grpc_last_failed_idx = tmp_idx;
        memcpy(grpc_last_failed_sid, tmp_sid, (tmp_idx+1) * sizeof(*tmp_sid));

        GRPC_RETURN(saw_failure ? GRPC_SESSION_FAILED : GRPC_NO_ERROR,
          grpc_minor_errno);
      }

      saw_failure = TRUE;
    }
  }

  /* should not reach here */
  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_wait_any(grpc_sessionid_t *idPtr)
{
  int i, jobCount = 0;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  for(;;) {
    for(i = 0; i < MAX_GRPC_REQUESTS; i++)
      if(grpc_outstanding_requests[i] != NULL) {
        /* if the job is no longer running (i.e. it either successfully
         * completed or it failed) call wait and check the error status.
         */

        if(grpc_probe(i) != GRPC_NOT_COMPLETED) {
          grpc_error_t r = grpc_wait(i);

          *idPtr = i;

          if(r != GRPC_NO_ERROR) {
            grpc_errors[i] = grpc_errno;
            grpc_minor_errors[i] = grpc_minor_errno;
          }

          GRPC_RETURN(r, grpc_minor_errno);
        }

        jobCount++;
      }

    if(jobCount == 0) {
      *idPtr = -1;
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }

    sleep(1);
  }

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

void
grpc_perror(char *str)
{
  /* check if this is a server-specific error code */
  if((grpc_minor_errno > GS_SVC_NO_ERROR) && 
     (grpc_minor_errno < GRPC_NO_MINOR_ERROR))
    fprintf(stderr, "%s: %s -- %s\n", str, grpc_error_string(grpc_errno),
       gs_service_error[grpc_minor_errno]);
  else if((grpc_minor_errno > GRPC_NO_MINOR_ERROR) &&
          (grpc_minor_errno < GRPC_LAST_MINOR_ERROR_CODE))
    fprintf(stderr, "%s: %s -- %s\n", str, grpc_error_string(grpc_errno),
       GRPC_MINOR_ERROR_MESSAGES[grpc_minor_errno-GRPC_NO_MINOR_ERROR]);
  else
    fprintf(stderr, "%s: %s\n", str, grpc_error_string(grpc_errno));
}

char *
grpc_minor_error_string(grpc_error_t minor_errno)
{
  if((minor_errno > GRPC_NO_MINOR_ERROR) &&
     (minor_errno < GRPC_LAST_MINOR_ERROR_CODE))
    return GRPC_MINOR_ERROR_MESSAGES[minor_errno-GRPC_NO_MINOR_ERROR];
  else
    return "";
}


char *
grpc_error_string(grpc_error_t error_code)
{
  if((error_code < 0) || (error_code >= GRPC_LAST_ERROR_CODE))
    return "Invalid Error Code";

  return GRPC_ERROR_MESSAGES[error_code];
}

grpc_error_t
grpc_get_error(grpc_sessionid_t sessionID)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(!grpc_outstanding_requests[sessionID])
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  return grpc_errors[sessionID];
}

grpc_error_t
grpc_get_failed_sessionid(grpc_sessionid_t *idPtr)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(grpc_last_failed_idx < 0) {
    *idPtr = GRPC_SESSIONID_VOID;
    return GRPC_NO_ERROR;
  }

  *idPtr = grpc_last_failed_sid[grpc_last_failed_idx];

  grpc_last_failed_idx--;

  return GRPC_NO_ERROR;
}

grpc_arg_stack *
grpc_arg_stack_new(int max)
{
  grpc_arg_stack *tmp;

  tmp = (grpc_arg_stack *) malloc(sizeof(grpc_arg_stack));

  if(!tmp)
    return NULL;

  tmp->max_size = max;
  tmp->top = -1;
  tmp->args = (void **) calloc(max, sizeof(void *));

  if(!tmp->args) {
    free(tmp);
    return NULL;
  }

  return tmp;
}

int
grpc_arg_stack_push_arg(grpc_arg_stack *stack, void *arg)
{
  if(!stack)
    return -1;

  if(stack->top == (stack->max_size - 1))
    return -1;

  stack->args[++stack->top] = arg;

  return 0;
}

void *
grpc_arg_stack_pop_arg(grpc_arg_stack *stack)
{
  if(!stack)
    return NULL;

  if(stack->top == -1)
    return NULL;

  return stack->args[stack->top--];
}

int
grpc_arg_stack_destruct(grpc_arg_stack *stack)
{
  if(!stack)
    return -1;

  if(stack->args)
    free(stack->args);

  free(stack);

  return 0;
}

grpc_error_t
grpc_serialize_request(grpc_sessionid_t sessionID, char **str)
{
  char *server_str, *problem_str;
  grpc_function_handle_t *handle;
  grpc_request_t *req;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  req = grpc_outstanding_requests[sessionID];

  if(!req)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  handle = req->handle;

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  if(gs_encode_server(&server_str, handle->server_list[handle->srv_idx]) < 0)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SERVER_ENCODE);

  if(!server_str)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  if(gs_encode_problem(&problem_str, handle->problem_desc) < 0)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_PROBLEM_ENCODE);

  if(!problem_str) {
    free(server_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  *str =
      dstring_sprintf
      ("<request id=\"%s\">\n  <handle func_name=\"%s\">\n%s\n%s\n  </handle>\n</request>\n",
       req->request_id, handle->func_name, server_str, problem_str);

  free(server_str);
  free(problem_str);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_deserialize_request(char *str, grpc_request_t *req)
{
  char *p, *endp, *tmp_str;
  int i, len;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  tmp_str = strdup(str);
  if(!tmp_str)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  p = tmp_str;
  len = strlen(tmp_str);

  for(i = 0; (tmp_str[i] != '\n') && (i < len); i++)
    /*
     * spin 
     */ ;

  if(i == len) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_SERIALIZED_REQ);
  }

  tmp_str[i] = '\0';

  req->request_id = CALLOC(strlen(p), sizeof(char));
  if(!req->request_id) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  sscanf(p, "<request id=\"%[^\"]\">", req->request_id);

  req->handle = (grpc_function_handle_t *) CALLOC(1, 
     sizeof(grpc_function_handle_t));
  if(!req->handle) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  p = tmp_str + i + 1;
  p = strstr(p, "<handle");

  for(; (tmp_str[i] != '\n') && (i < len); i++)
    /*
     * spin 
     */ ;

  if(i == len) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_SERIALIZED_REQ);
  }

  tmp_str[i] = '\0';

  req->handle->func_name = CALLOC(strlen(p), sizeof(char));
  if(!req->handle->func_name) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  sscanf(p, "<handle func_name=\"%[^\"]\">", req->handle->func_name);

  p = tmp_str + i + 1;

  endp = strstr(p, "<problem name");
  if(!endp) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_SERIALIZED_REQ);
  }

  *(endp - 1) = '\0';

  req->handle->srv_idx = 0;
  req->handle->num_servers = 1;
  req->handle->server_list =
      (gs_server_t **) CALLOC(1, sizeof(gs_server_t *));
  if(!req->handle->server_list) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  req->handle->server_list[req->handle->srv_idx] =
      (gs_server_t *) CALLOC(1, sizeof(gs_server_t));
  if(!req->handle->server_list[req->handle->srv_idx]) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }
  req->s_pid = -1;

  if(gs_decode_server(p, req->handle->server_list[req->handle->srv_idx]) < 0) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SERVER_DECODE);
  }

  p = endp;

  endp = strstr(p, "</problem>");
  if(!endp) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_SERIALIZED_REQ);
  }

  endp += strlen("</problem>");
  *endp = '\0';

  req->problem = (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t));
  if(!req->problem) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  if(gs_decode_problem(p, req->problem) < 0) {
    free(tmp_str);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_PROBLEM_DECODE);
  }

  free(tmp_str);

  req->handle->problem_desc = (gs_problem_t *) malloc(sizeof(gs_problem_t));

  if(!req->handle->problem_desc)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  if(gs_dup_problem(req->handle->problem_desc, req->problem) < 0)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_DUP_PROBLEM_FAILED);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_retrieve(char *request_str, ...)
{
  gs_service_error_enum_t save_minor_errno;
  grpc_request_t *req;
  gs_va_list argptr;
  grpc_error_t status;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  va_start(argptr.args, request_str);

  req = (grpc_request_t *) malloc(sizeof(grpc_request_t));

  if(!req)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  if(grpc_deserialize_request(request_str, req) != GRPC_NO_ERROR)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);

  if(gs_sender_compute_arg_sizes(&argptr, NULL, req->problem,
       grpc_client_lang, grpc_client_major) < 0)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);

  status = gs_wait_common(req);
  save_minor_errno = grpc_minor_errno;

  grpc_function_handle_destruct(req->handle);
  free(req->handle);
  grpc_request_destruct(req);
  free(req);

  GRPC_RETURN(status, save_minor_errno);
}

int
gs_measure_network_perf(grpc_function_handle_t *handle)
{
  int i, len, ns, nping;
  char *msg;
  grpc_profile_t profile_tmp;

  memset(&profile_tmp, 0, sizeof(grpc_profile_t));

#ifdef GS_PROFILING
  GRPC_TIMER_START(&profile_tmp);
#endif

  ns = MIN(handle->num_servers, grpc_measure_comm_num_servers);
  len = 32 * 1024;

  msg = (char *)malloc(len);

  if(!msg)
    return -1;

  nping = 0;

  for(i=0;i<handle->num_servers;i++) {
    char srv_cid[2*CID_LEN+1];
    grpc_cache_entry_t *ent;
    double elapsed_time;

    proxy_cid_to_str(srv_cid, handle->server_list[i]->componentid);

    if(grpc_comm_cache) {
      ent = (grpc_cache_entry_t *) icl_hash_find(grpc_comm_cache, srv_cid);

      /* check if this entry is out of date */
      if(ent) {
        struct timeval ctv;

        gettimeofday(&ctv, NULL);

        if(ctv.tv_sec - ent->tv.tv_sec > grpc_measure_comm_cache_ttl)
          ent = NULL;
      }
    }
    else
      ent = NULL;

    if(ent) {
      handle->server_list[i]->comm_bw = ent->comm_bw; 
    }
    else {
      if(gs_do_ping(handle->server_list[i], msg, len, &elapsed_time) < 0) {
        handle->server_list[i]->comm_bw = 0.0;
      }
      else {
        handle->server_list[i]->comm_bw = len / elapsed_time;

        if(grpc_comm_cache) {
          ent = (grpc_cache_entry_t *)malloc(sizeof(grpc_cache_entry_t));

          if(ent) {
            char *cid_dup;

            cid_dup = strdup(srv_cid);

            if(!cid_dup) {
              free(ent);
            }
            else {
              void *old_ent;

              gettimeofday(&(ent->tv), NULL);
              ent->comm_bw = handle->server_list[i]->comm_bw;

              icl_hash_update_insert(grpc_comm_cache, cid_dup, ent, &old_ent);
            }
          }
        }
      }

      nping++;
    }
  
    if(nping == ns)
      break;
  }

  free(msg);

#ifdef GS_PROFILING
  GRPC_TIMER_STOP(&profile_tmp);
  if(grpc_profile_next)
    grpc_profile_next->measure_comm = GRPC_TIMER_ELAPSED(&profile_tmp);
#endif

  return 0;
}

static int
gs_server_compare_total_time(const void *p1, const void *p2)
{
  gs_server_t *s1, *s2;
  double s1_time, s2_time;

  if(!p1 || !p2) return 0;

  s1 = *((gs_server_t **) p1);
  s2 = *((gs_server_t **) p2);

  s1_time = s1->score + s1->comm_time;
  s2_time = s2->score + s2->comm_time;

  /* if the measurements are pretty close, then we just
   * consider them equal.
   */
  if(abs(s1_time - s2_time) < grpc_measure_comm_time_thresh)
    return 0;

  if(s1_time > s2_time)
    return 1;
  if(s1_time < s2_time)
    return -1;

  return 0;
}

int
gs_sort_servers_on_comp_plus_comm(grpc_function_handle_t *handle, int data_size)
{
  gs_server_t *srv;
  double avg_bw, sum;
  int i, nnzs;

  sum = avg_bw = 0.0;
  nnzs = 0;

  for(i=0;i<handle->num_servers;i++) {
    srv = handle->server_list[i];

    if(srv->comm_bw != 0.0) {
      sum += srv->comm_bw;
      nnzs++;
    }
  }

  if(nnzs > 0)
    avg_bw = sum / (double) nnzs;
 
  if(avg_bw == 0.0)
    avg_bw = 20000.0;
    
  for(i=0;i<handle->num_servers;i++) {
    srv = handle->server_list[i];

    if(srv->comm_bw == 0.0)
      srv->comm_time = data_size / avg_bw;
    else
      srv->comm_time = data_size / srv->comm_bw;
  }

  /* Sort the servers */
  qsort(handle->server_list, handle->num_servers, sizeof(gs_server_t *),
     gs_server_compare_total_time);

  return 0;
}

int
gs_get_total_arg_size(grpc_function_handle_t *handle)
{
  int my_dsig, total_bytes;
  gs_argument_t *argptr;

  my_dsig = pvmgetdsig();
  total_bytes = 0;
    
  for(argptr=handle->problem_desc->arglist;argptr;argptr=argptr->next) {
    int elsize = 0;

    /* For input/output args, double the elsize since it will be
     * a round trip.  Workspace args are not transferred, so use zero
     * for them.  For variable length output, we have no idea what the
     * size will be, so just choose some small value.
     */

    switch(argptr->inout) {
      case GS_INOUT:
        elsize = 2 * argptr->rows * argptr->cols * 
                 gs_get_element_size(argptr->datatype, my_dsig);
        break;
      case GS_WORKSPACE:
        elsize = 0;
        break;
      case GS_VAROUT:
        elsize = 8;
        break;
      default:
        elsize = argptr->rows * argptr->cols * 
                 gs_get_element_size(argptr->datatype, my_dsig);
    }

    total_bytes += elsize;
  }

  return total_bytes;
}

static int
gs_get_server_mapping(char *func_name, grpc_function_handle_t *handle,
   gs_va_list *args, void **argstack, int my_dsig,
   int lang, int major)
{
  char *msg = NULL;
  int tag, i, argsize;
  SOCKET sock;
  int scalar_args_to_be_transferred = 0;
  int problem_desc_to_be_transferred = 0;
  handle->agent_taskid = -1;

  grpc_minor_errno = GRPC_NO_MINOR_ERROR;

  if(!func_name) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    grpc_minor_errno = GRPC_NULL_FUNCTION_NAME;
    return -1;
  }

  if((sock = gs_connect_to_agent()) == INVALID_SOCKET) {
    grpc_errno = GRPC_RPC_REFUSED;
    grpc_minor_errno = GRPC_AGENT_NOT_SET;
    return -1;
  }

  if((gs_send_tag(sock, GS_PROT_PROBLEM_SUBMIT) < 0) ||
     (gs_send_string(sock, VERSION) < 0))
    goto error_communication_failed;

  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;
  
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      grpc_minor_errno = GRPC_VERSION_MISMATCH;
    else
      grpc_minor_errno = GRPC_NO_MINOR_ERROR;

    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return -1;
  }

  if(gs_encode_problem_submit_request(&msg, func_name, my_dsig, handle->criteria) < 0) {
    grpc_minor_errno = GRPC_SUBMIT_REQ_ENCODE;
    goto error_client_internal_error;
  }

  if(gs_send_string(sock, msg) < 0)
    goto error_communication_failed;
  FREE(msg);
  msg = NULL;

  /*
   * Receive the problem description if necessary 
   */
  if(handle->problem_desc != NULL) {
    problem_desc_to_be_transferred = 0;
    if(gs_send_int(sock, problem_desc_to_be_transferred) < 0)
      goto error_communication_failed;
  }
  else {
    problem_desc_to_be_transferred = 1;
    if(gs_send_int(sock, problem_desc_to_be_transferred) < 0)
      goto error_communication_failed;
    if(gs_recv_string(sock, &msg) < 0)
      goto error_communication_failed;
    if((handle->problem_desc =
        (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t))) == NULL) {
      grpc_minor_errno = GRPC_OUT_OF_MEMORY;
      goto error_client_internal_error;
    }
    if(gs_decode_problem(msg, handle->problem_desc) < 0) {
      grpc_minor_errno = GRPC_PROBLEM_DECODE;
      goto error_communication_failed;
    }
    FREE(msg);
  }

  if(!strcmp(handle->problem_desc->description, GS_UNKNOWN_PROB)) {
    DBGPRINTF("Client Error: problem not found\n");
    if(sock != INVALID_SOCKET)
      proxy_close(sock);
    gs_free_problem(handle->problem_desc);
    handle->problem_desc = NULL;
    grpc_errno = GRPC_FUNCTION_NOT_FOUND;
    return -1;
  }

  /*
   * If one of args,argstack,my_dsig is passed in, then we need to send input 
   * scalars so that the agent can do agent side scheduling 
   */
  if(args || argstack || my_dsig != -1) {
    gs_va_list *vlptr, tmp_va_list;

    scalar_args_to_be_transferred = 1;

    if(gs_send_int(sock, scalar_args_to_be_transferred) < 0)
      goto error_communication_failed;

    /*
     * make a copy of the va_list if necessary before the call to
     * gs_send_input_scalar_args() since it seems to get clobbered and causes 
     * problems to reuse it for the actual submission. 
     */

    if(args) {
      va_copy(tmp_va_list.args, args->args);
      vlptr = &tmp_va_list;
    }
    else
      vlptr = NULL;

    if(gs_send_input_scalar_args(vlptr, argstack, sock, handle->problem_desc,
                                 my_dsig, grpc_client_lang,
                                 grpc_client_major) < 0)
      goto error_communication_failed;
  }
  else {
    scalar_args_to_be_transferred = 0;
    if(gs_send_int(sock, scalar_args_to_be_transferred) < 0)
      goto error_communication_failed;
  }

  /*
   * Receive the agent's task id.
   */
  if(gs_recv_int(sock, &(handle->agent_taskid)) < 0)
    goto error_communication_failed;

  /* agent task id useless if not transferring args */

  if(!scalar_args_to_be_transferred)
    handle->agent_taskid = -1;

  /*
   * Receive number of servers 
   */
  if(gs_recv_int(sock, &(handle->num_servers)) < 0)
    goto error_communication_failed;

  /*
   * If no servers return error 
   */
  if(handle->num_servers <= 0)
    goto error_server_not_found;

  if(handle->server_list)
    gs_free_handle_server_list(handle);

  handle->server_list = (gs_server_t **)
      CALLOC(handle->num_servers, sizeof(gs_server_t *));
  if(!handle->server_list) {
    grpc_minor_errno = GRPC_OUT_OF_MEMORY;
    goto error_client_internal_error;
  }

  for(i = 0; i < handle->num_servers; i++) {
    handle->server_list[i] = (gs_server_t *) CALLOC(1, sizeof(gs_server_t));
    if(gs_recv_string(sock, &msg) < 0)
      goto error_communication_failed;
    if(gs_decode_server(msg, handle->server_list[i]) < 0) {
      grpc_minor_errno = GRPC_SERVER_DECODE;
      goto error_client_internal_error;
    }

    FREE(msg);
  }

#ifdef GS_PROFILING
  if(grpc_profile_next)
    grpc_profile_next->measure_comm = 0.0;
#endif

  if(grpc_measure_comm) {
    argsize = gs_get_total_arg_size(handle);

    if(argsize > GS_ARG_SIZE_THRESH)
      if(gs_measure_network_perf(handle) == 0)
        gs_sort_servers_on_comp_plus_comm(handle, argsize);
  }

  if(handle->num_servers > 0) {
    if(gs_recv_string(sock, &msg) < 0)
      goto error_communication_failed;

    if(handle->problem_desc)
      gs_free_problem(handle->problem_desc);

    handle->problem_desc = (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t));
    if(!handle->problem_desc) {
      grpc_minor_errno = GRPC_OUT_OF_MEMORY;
      goto error_client_internal_error;
    }
    if(gs_decode_problem(msg, handle->problem_desc) < 0) {
      grpc_minor_errno = GRPC_PROBLEM_DECODE;
      goto error_client_internal_error;
    }
    FREE(msg);
  }

  proxy_close(sock);

  handle->srv_idx = 0;

  handle->func_name = strdup(func_name);

  DBGPRINTF("Got %d servers from agent for problem %s\n",
            handle->num_servers, func_name);

  return 0;

error_communication_failed:
  DBGPRINTF("Client Error: Communication failed\n");
  FREE(msg);
  if(sock != INVALID_SOCKET)
    proxy_close(sock);
  grpc_errno = GRPC_COMMUNICATION_FAILED;
  return -1;

error_client_internal_error:
  DBGPRINTF("Client Error: Client internal error\n");
  FREE(msg);
  if(sock != INVALID_SOCKET)
    proxy_close(sock);
  grpc_errno = GRPC_OTHER_ERROR_CODE;
  return -1;

error_server_not_found:
  DBGPRINTF("Client Error: Server not found\n");
  FREE(msg);
  if(sock != INVALID_SOCKET)
    proxy_close(sock);
  grpc_errno = GRPC_SERVER_NOT_FOUND;
  return -1;
}


static int
gs_parse_host_port(char *str, ipaddr_t *ip, in_port_t * port)
{
  struct hostent *hp;
  char *tok, *host_tok = NULL, *port_tok = NULL, delim[2];

  delim[0] = GS_PORT_DELIM;
  delim[1] = '\0';

  tok = strtok(str, delim);
  if(!tok)
    goto gs_parse_host_port_error;

  host_tok = strdup(tok);
  if(!host_tok)
    goto gs_parse_host_port_error;

  tok = strtok(NULL, delim);
  if(!tok)
    goto gs_parse_host_port_error;

  port_tok = strdup(tok);
  if(!port_tok)
    goto gs_parse_host_port_error;

  if((hp = gethostbyname(host_tok)) == NULL)
    goto gs_parse_host_port_error;

  memcpy((void *) ip, hp->h_addr_list[0], sizeof(*ip));

  *port = atoi(port_tok);

  if(host_tok)
    free(host_tok);
  if(port_tok)
    free(port_tok);
  return 0;

gs_parse_host_port_error:
  if(host_tok)
    free(host_tok);
  if(port_tok)
    free(port_tok);
  return -1;
}

static int
gs_parse_host_info(char *host_str, ipaddr_t *ip, in_port_t * port,
                   ipaddr_t *proxyip, in_port_t * proxyport, char *cid)
{
  char *tok, *component_tok = NULL, *proxy_tok = NULL, delim[2];

  delim[0] = GS_PROXY_DELIM;
  delim[1] = '\0';

  /*
   * first get the part before the proxy delimiter.  this will be the server
   * host name and port. 
   */

  tok = strtok(host_str, delim);
  if(!tok)
    goto gs_parse_host_info_error;

  /*
   * copy this token for use later since strtok trashes things 
   */

  component_tok = strdup(tok);
  if(!component_tok)
    goto gs_parse_host_info_error;

  /*
   * now get the proxy part of the string 
   */
  tok = strtok(NULL, delim);
  if(!tok) {
    /*
     * no proxy specified 
     */
    proxy_tok = NULL;
    *proxyip = 0;
    *proxyport = 0;
    memset(cid, 0xFF, CID_LEN);
  }
  else {
    proxy_tok = strdup(tok);
    if(!proxy_tok)
      goto gs_parse_host_info_error;

    if(gs_parse_host_port(proxy_tok, proxyip, proxyport) < 0)
      goto gs_parse_host_info_error;
  }

  /*
   * check whether the server host name looks like a component ID 
   */

  if(component_tok[0] == '0' &&
     (component_tok[1] == 'x' || component_tok[1] == 'X')) {
    char delim[2];

    if(*proxyip == 0) {
      ERRPRINTF("If specifying a cid, proxy must be specified\n");
      goto gs_parse_host_info_error;
    }

    delim[0] = GS_PORT_DELIM;
    delim[1] = '\0';

    tok = strtok(component_tok, delim);
    if(!tok)
      goto gs_parse_host_info_error;

    proxy_str_to_cid(cid, tok);
    *ip = 0;

    tok = strtok(NULL, delim);
    if(!tok)
      goto gs_parse_host_info_error;

    *port = atoi(tok);
  }
  else {
    if(*proxyip != 0) {
      ERRPRINTF
          ("If proxy is used, component ID must be specified (not IP)\n");
      goto gs_parse_host_info_error;
    }

    memset(cid, 0xFF, CID_LEN);
    if(gs_parse_host_port(component_tok, ip, port) < 0)
      goto gs_parse_host_info_error;
  }

  if(component_tok)
    free(component_tok);
  if(proxy_tok)
    free(proxy_tok);

  return 0;

gs_parse_host_info_error:
  if(component_tok)
    free(component_tok);
  if(proxy_tok)
    free(proxy_tok);
  return -1;
}

static int
get_next_request_id()
{
  int i;

  for(i = 0; i < MAX_GRPC_REQUESTS; i++) {
    if(grpc_outstanding_requests[i] == NULL)
      return i;
  }

  return -1;
}

static int
gs_handle_blocking_call(grpc_function_handle_t *handle, gs_va_list *argptr,
                    void **argstack, SOCKET sock, int my_dsig)
{
  int status_tag;



#ifdef GS_SMART_GRIDSOLVE

  /*
   * Sending has_smart_arg_comm variable to the assigned server.
   * This variable specifies whether the remote task requires 
   * remote communication. This variable will be set to 1,
   * if the inputs are stored remotely or if the output arguments
   * should be sent remotely or cached locally.
   * 
   */ 

  if(gs_send_int(sock, handle->problem_desc->has_smart_arg_comm)<0){
    ERRPRINTF("SMART: Error sending has smart comm\n");
    return -1;
  }
 
 
  if(handle->problem_desc->has_smart_arg_comm){
   /*
    * If the remote task requires remote communication, then the mapping
    * information is sent to the server. This mapping info outlines
    * which inputs should be received from client, server file or server
    * memory. It also outlines which should be sent to client, stored
    * locally or sent remotely to another or 
    * multiple servers.
    */  
    if(gs_smart_send_map_info(sock, handle->problem_desc)<0){
      ERRPRINTF("SMART : Error sending mapping info\n");
      return -1;
    }
   /*
    * The client sends only the input arguments to the server
    * which have been outlined in the mapping solution.
    */  
    if(gs_smart_send_input_args(argptr, argstack, sock, handle->problem_desc,
                        my_dsig, grpc_client_lang, grpc_client_major) < 0)
      return -1;



     int pid; 
     pid=fork();
      if(pid==-1){
        ERRPRINTF("SMART: Out of memory could not fork\n");
        return -1;
      }

      if(pid==0){
        /*
         * Send input arguments from client to other servers as outlined by 
         * mapping solution
         */ 
        if(gs_smart_send_input_args_remotely(sock, handle->problem_desc, my_dsig)<0){
          ERRPRINTF("SMART : Error sending smart sending arguments\n");
          return -1;
        }
        _exit(0);
      }

 
  }
  else{
    if(gs_send_input_args(argptr, argstack, sock, handle->problem_desc,
                        my_dsig, grpc_client_lang, grpc_client_major) < 0)
      return -1;
  }
#else
  if(gs_send_input_args(argptr, argstack, sock, handle->problem_desc,
                        my_dsig, grpc_client_lang, grpc_client_major) < 0)
    return -1;
#endif



#ifdef GS_PROFILING
  if(grpc_profile_next) {
    GRPC_TIMER_STOP(grpc_profile_next);
    grpc_profile_next->send_input = GRPC_TIMER_ELAPSED(grpc_profile_next);
    GRPC_TIMER_START(grpc_profile_next);
  }
#endif

  if(gs_wait_for_output(sock) < 0) {
    ERRPRINTF("error waiting for output to be ready\n");
    return -1;
  }

  if(gs_recv_tag(sock, &status_tag) < 0)
    return -1;

  if(status_tag != GS_PROT_OK) {
    ERRPRINTF("Service failed: %s\n", gs_service_error[status_tag]);
    return -1;
  }


#ifdef GS_SMART_GRIDSOLVE
  if(handle->problem_desc->has_smart_arg_comm==1){
   /*
    * The client receives only the output arguments from the server
    * which have been outlined in the mapping solution.
    */  
    if(gs_smart_recv_output_args(sock, handle->problem_desc, handle->server_list[handle->srv_idx]->data_format, 
                                 my_dsig) < 0){
      ERRPRINTF("SMART: Error trying to receive output args\n");
      return -1;
    }
  }
  else{
    if(gs_recv_output_args(sock, handle->problem_desc,
                         handle->server_list[handle->srv_idx]->data_format,
                         my_dsig) < 0)
      return -1;
  }
#else
  if(gs_recv_output_args(sock, handle->problem_desc,
                         handle->server_list[handle->srv_idx]->data_format,
                         my_dsig) < 0)
    return -1;
#endif


#ifdef GS_PROFILING
  if(grpc_profile_next) {
    GRPC_TIMER_STOP(grpc_profile_next);
    grpc_profile_next->recv_output = GRPC_TIMER_ELAPSED(grpc_profile_next);
  }
#endif

  return 0;
}

static int
gs_handle_nonblocking_call(grpc_function_handle_t *handle, 
   grpc_sessionid_t *sessionId, gs_va_list *argptr,
   void **argstack, SOCKET sock, int my_dsig, char *request_id)
{
  int idx = get_next_request_id();
  grpc_request_t *new_req;
  int retval = 0;


#ifdef GS_SMART_GRIDSOLVE
  /*
   * Sending has_smart_arg_comm variable to the assigned server.
   * This variable specifies whether the remote task requires 
   * remote communication. This variable will be set to 1,
   * if the inputs are stored remotely or if the output arguments
   * should be sent remotely or cached locally.
   * 
   */ 
  if(gs_send_int(sock, handle->problem_desc->has_smart_arg_comm)<0){
    ERRPRINTF("SMART: Error sending has smart comm\n");
    return -1;
  }


  if(handle->problem_desc->has_smart_arg_comm){
   /*
    * If the remote task requires remote communication, then the mapping
    * information is sent to the server. This mapping info outlines
    * which inputs should be received from client, server file or server
    * memory. It also outlines which should be sent to client, stored
    * locally or sent remotely to another or 
    * multiple servers.
    */  
    if(gs_smart_send_map_info(sock, handle->problem_desc)<0){
      ERRPRINTF("SMART : Error sending mapping info\n");
      return -1;
    }
  }  

#endif




  /*
   * before forking, call gs_send_input_scalar_args() so that the arg sizes
   * will be computed and stored in this problem struct.  it's better to do
   * this prior to forking because we'd otherwise need to communicate the
   * sizes back to the parent process. 
   */

  if(gs_send_input_scalar_args(argptr, argstack, sock, handle->problem_desc,
                               my_dsig, grpc_client_lang,
                               grpc_client_major) < 0)
    return -1;

  new_req = (grpc_request_t *) malloc(sizeof(grpc_request_t));
  if(!new_req || (idx < 0))
    return -1;

  new_req->request_id = request_id;
  new_req->handle = handle;

  /*
   * we need to dup the problem in case the user submits subsequent requests
   * with different arguments. 
   */
  new_req->problem = (gs_problem_t *) malloc(sizeof(gs_problem_t));
  if(!new_req->problem)
    return -1;

  if(gs_dup_problem(new_req->problem, handle->problem_desc) < 0)
    return -1;

  grpc_outstanding_requests[idx] = new_req;
  grpc_profile_info[idx] = grpc_profile_next;


/* NOTE: On Win32 we do not fork to send the arguments, we send them in this process */
/* TODO: ayk, should replace WIN32 with ifndef HAVE_FORK */
#if defined (WIN32) || defined (WINNT)
  grpc_outstanding_requests[idx]->s_pid = 0;
  retval = gs_send_input_nonscalar_args(sock, handle->problem_desc, my_dsig);
  proxy_close(sock);
#else
  grpc_outstanding_requests[idx]->s_pid = fork();

  switch (grpc_outstanding_requests[idx]->s_pid) {
    case -1:
      ERRPRINTF("Failed to fork process to send input data\n");
      grpc_request_destruct_free_clear(idx);
      return -1;

    case 0:                    /* child */

#ifdef GS_SMART_GRIDSOLVE
      if(handle->problem_desc->has_smart_arg_comm==1){
       /*
        * The client sends only the input arguments to the server 
        * which have been outlined in the mapping solution.
        */  
        if(gs_smart_send_input_nonscalar_args(sock, handle->problem_desc, my_dsig) < 0)
          _exit(-1);
     

        int pid; 
        pid=fork();
        if(pid==-1){
          ERRPRINTF("SMART: Out of memory could not fork\n");
          return -1;
        }

        if(pid==0){
         /*
          * Send input arguments from client to other servers as outlined by mapping solution
          */ 
          if(gs_smart_send_input_args_remotely(sock ,handle->problem_desc, my_dsig)<0){
            ERRPRINTF("SMART : Error sending smart sending arguments\n");
            return -1;
          }
          _exit(0);
        }

      }
      else{
        if(gs_send_input_nonscalar_args(sock, handle->problem_desc, my_dsig) < 0)
          _exit(-1);
      
      }
#else

      if(gs_send_input_nonscalar_args(sock, handle->problem_desc, my_dsig) < 0)
        _exit(-1);
#endif
      proxy_close(sock);
      _exit(0);

    default:                   /* parent */
      proxy_close(sock);
  }
#endif

  *sessionId = idx;
  return retval;
}

static int
gs_notify_agent_of_cancel(grpc_function_handle_t *handle, char *reqid)
{
  char *msg, srv_cid[2*CID_LEN+1];
  SOCKET sock;
  int tag, agent_taskid;

#if !defined (WIN32) && !defined (WINNT)
  pid_t childpid;
  childpid = fork();

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

  if(childpid > 0)
    return 0;
#endif

  if(handle->num_calls > 1)
    agent_taskid = -1;
  else
    agent_taskid = handle->agent_taskid;

  /* all code after here is executed in the child process */

  if((sock = gs_connect_to_agent()) == INVALID_SOCKET) {
    grpc_errno = GRPC_RPC_REFUSED;
    _exit(-1);
  }

  proxy_cid_to_str(srv_cid, handle->server_list[handle->srv_idx]->componentid);

  if((gs_send_tag(sock, GS_PROT_NOTIFY_CANCEL) < 0) ||
     (gs_send_string(sock, VERSION) < 0)) {
    ERRPRINTF("failed to send tag\n");
    close_socket(sock);
    _exit(-1);
  }

  if(gs_recv_tag(sock, &tag) < 0) {
    ERRPRINTF("Error communicating with agent.\n");
    close_socket(sock);
    _exit(-1);
  }

  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      ERRPRINTF("Error: Agent is an incompatible version\n");
    else
      ERRPRINTF("Error: Agent refused with code %d\n", tag);
    close_socket(sock);
    _exit(-1);
  }

  if(gs_encode_cancel_notification(&msg, srv_cid, reqid, agent_taskid) < 0)
  {
    ERRPRINTF("Failed to encode message\n");
    close_socket(sock);
    _exit(-1);
  }

  if(gs_send_string(sock, msg) < 0) {
    ERRPRINTF("failed to send tag\n");
    free(msg);
    close_socket(sock);
    _exit(-1);
  }

  free(msg);
#if !defined (WIN32) && !defined (WINNT)
  close_socket(sock);
  _exit(0);
#endif
  return 0;
}

static int
gs_notify_agent_of_failure(grpc_function_handle_t *handle, char *reqid)
{
  char *msg, srv_cid[2*CID_LEN+1];
  SOCKET sock;
  int tag, agent_taskid;

#if !defined (WIN32) && !defined (WINNT)
  pid_t childpid;
  childpid = fork();

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

  if(childpid > 0)
    return 0;
#endif /* WIN32 */

  if(handle->num_calls > 1)
    agent_taskid = -1;
  else
    agent_taskid = handle->agent_taskid;

  /* all code after here is executed in the child process */

  if((sock = gs_connect_to_agent()) == INVALID_SOCKET) {
    grpc_errno = GRPC_RPC_REFUSED;
    _exit(-1);
  }

  proxy_cid_to_str(srv_cid, handle->server_list[handle->srv_idx]->componentid);

  if((gs_send_tag(sock, GS_PROT_NOTIFY_FAILURE) < 0) ||
     (gs_send_string(sock, VERSION) < 0)) {
    ERRPRINTF("failed to send tag\n");
    close_socket(sock);
    _exit(-1);
  }

  if(gs_recv_tag(sock, &tag) < 0) {
    ERRPRINTF("Error communicating with agent.\n");
    close_socket(sock);
    _exit(-1);
  }

  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      ERRPRINTF("Error: Agent is an incompatible version\n");
    else
      ERRPRINTF("Error: Agent refused with code %d\n", tag);
    close_socket(sock);
    _exit(-1);
  }

  if(gs_encode_failure_notification(&msg, srv_cid, reqid, agent_taskid) < 0)
  {
    ERRPRINTF("Failed to encode message\n");
    close_socket(sock);
    _exit(-1);
  }

  if(gs_send_string(sock, msg) < 0) {
    ERRPRINTF("failed to send tag\n");
    free(msg);
    close_socket(sock);
    _exit(-1);
  }

  free(msg);
#if !defined (WIN32) && !defined (WINNT)
  close(sock);
  _exit(0);
#endif /* WIN32 */
  return 0;
}

static grpc_error_t
gs_call_common(grpc_function_handle_t *handle, grpc_sessionid_t *sessionId,
    gs_va_list *argptr, void **argstack, int blocking)
{
  int tag, agent_taskid;
  SOCKET sock;
  int my_dsig;
  char *msg, *request_id = NULL;
  gs_server_t *srv;

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  grpc_minor_errno = GRPC_NO_MINOR_ERROR;


#ifdef GS_SMART_GRIDSOLVE
/*
 * when the user has explicitly speficied the number of tasks to build
 * check if there are still tasks to be built. if so set the phase to building.
 */
  if(group_type==GS_SMART_EXPLICIT_GROUP){
    if( gs_smart_get_nb_auto_tasks_left_to_build()>0 ){
      smart_phase=GS_SMART_TASK_DISCOVERY;
    }
  }

 /*
  * if we are still executing mapped tasks 
  * (i.e. when nb_mapped_tasks_executed<total_nb_mapped_tasks)
  * then embed the mapping information into this task handle.
  * If smart_phase has been set to GS_SMART_STANDARD_EXEC
  * then it means the task has already failed execution on 
  * a different server. Therefore we do not implement
  * the mapping for this task and execute it in the
  * standard gs protocol. 
  */

 if((nb_mapped_tasks_executed<total_nb_mapped_tasks) && 
    (smart_phase!=GS_SMART_STANDARD_EXEC) && (smart_phase!=GS_SMART_TASK_DISCOVERY)){
   if(gs_smart_mg_embed_mapping(handle, nb_mapped_tasks_executed)<0){
     ERRPRINTF("SMART : Error embedding mapping\n");
     return -1;
   }
 }
#endif

#ifdef GS_PROFILING
  if(grpc_profile_next) {
    /*
     * clear profiling info 
     */
    memset(grpc_profile_next, 0, sizeof(grpc_profile_t));

    GRPC_TIMER_START(grpc_profile_next);
  }
#endif



#ifdef GS_PROFILING
  if(grpc_profile_next) {
    /*
     * clear profiling info 
     */
    memset(grpc_profile_next, 0, sizeof(grpc_profile_t));

    GRPC_TIMER_START(grpc_profile_next);
  }
#endif

  my_dsig = pvmgetdsig();

  /*
   * Check if the servers are to be bound now.  If so talk with the agent.
   * This variable was set when the handle was initialized 
   */
 
#ifdef GS_SMART_GRIDSOLVE

/*
 * if the current phase is the building phase, store the handle and exit
 * the gs_call_common_function
 */
  if( ( smart_phase==GS_SMART_TASK_DISCOVERY ) ){       
    /*
     * If the bind_servers_at_call_time parameter has been specified
     * i.e. the user has requested that the task gets mapped at 
     * run-time, then the handle's problem desc will not be initialised
     * for the task discovery phase. In this case we get the problem
     * description here.  And then when the task is called for execution
     * the problem description will not be retrieved like it normally would.
     * 
     */ 
    if(handle->bind_servers_at_call_time == 1) {
      char *tmp_name;

      /*
       * get_server_mapping will over-write both server_list and func_name.  we 
       * can go ahead and free the server_list now, but we will have to save a
       * pointer to func_name and free it afterwards. 
       */
      gs_free_handle_server_list(handle);
      tmp_name = handle->func_name;
      handle->func_name = NULL;

      /*
       * Get server mapping is a full transaction with the agent that can can
       * include the input arguments if they are provided 
       */
      if(gs_get_server_mapping(tmp_name, handle, argptr, argstack,
            my_dsig, grpc_client_lang, grpc_client_major) < 0) {
        /* restore the handle's function name */
        handle->func_name = tmp_name;
        handle->valid_mapping=0;
        GRPC_RETURN(grpc_errno, grpc_minor_errno);
      }
      handle->valid_mapping=1;
      free(tmp_name);

      /*
       * Keep this binding from now on, so that fault tolerance works correctly 
       */
      handle->bind_servers_at_call_time = 0;
    }
    
    if(gs_smart_store_handle(handle, sessionId, argptr, blocking, grpc_client_lang, grpc_client_major, group_type)<0){
       ERRPRINTF("SMART : Error storing handle \n");
       return -1;
    }

    /*
     * if the user has explicitly specified and the last task in that group has just
     * been built, then execute stored handles and exit the gs_call_common function.
     */
    if(group_type==GS_SMART_EXPLICIT_GROUP){
      if( gs_smart_get_nb_auto_tasks_left_to_build()==0 ){
         
        smart_phase=GS_SMART_EXEC_STORED_HANDLES;    
       /*
        TODO : reintroduce execute stored handles for explicit calls to gs_smart_map
        */
       smart_phase=GS_SMART_STANDARD_EXEC;
     }
   }
   handle->valid_mapping=1;
   return 0;
 }

 if(smart_phase==GS_SMART_STANDARD_EXEC){
   if(handle->bind_servers_at_call_time == 1) {
     char *tmp_name;

     /*
      * get_server_mapping will over-write both server_list and func_name.  we 
      * can go ahead and free the server_list now, but we will have to save a
      * pointer to func_name and free it afterwards. 
      */
     gs_free_handle_server_list(handle);
     tmp_name = handle->func_name;
     handle->func_name = NULL;

     /*
      * Get server mapping is a full transaction with the agent that can can
      * include the input arguments if they are provided 
      */
     if(gs_get_server_mapping(tmp_name, handle, argptr, argstack,
           my_dsig, grpc_client_lang, grpc_client_major) < 0) {
       /* restore the handle's function name */
       handle->func_name = tmp_name;
       GRPC_RETURN(grpc_errno, grpc_minor_errno);
     }

     free(tmp_name);

     /*
      * Keep this binding from now on, so that fault tolerance works correctly 
      */
     handle->bind_servers_at_call_time = 0;
   }  
   handle->valid_mapping=1;
 } 



#else
 if(handle->bind_servers_at_call_time == 1) {
    char *tmp_name;

    /*
     * get_server_mapping will over-write both server_list and func_name.  we 
     * can go ahead and free the server_list now, but we will have to save a
     * pointer to func_name and free it afterwards. 
     */
    gs_free_handle_server_list(handle);
    tmp_name = handle->func_name;
    handle->func_name = NULL;

    /*
     * Get server mapping is a full transaction with the agent that can can
     * include the input arguments if they are provided 
     */
    if(gs_get_server_mapping(tmp_name, handle, argptr, argstack,
          my_dsig, grpc_client_lang, grpc_client_major) < 0) {
      /* restore the handle's function name */
      handle->func_name = tmp_name;
      GRPC_RETURN(grpc_errno, grpc_minor_errno);
    }

    free(tmp_name);

    /*
     * Keep this binding from now on, so that fault tolerance works correctly 
     */
    handle->bind_servers_at_call_time = 0;
  }
#endif

#ifdef GS_SMART_GRIDSOLVE
/*
  if(!handle->valid_mapping){
      GRPC_RETURN(grpc_errno, grpc_minor_errno);
  }
*/
#endif
  srv = handle->server_list[handle->srv_idx];

#ifdef GS_PROFILING
  if(grpc_profile_next) {
    GRPC_TIMER_STOP(grpc_profile_next);
    grpc_profile_next->agent_comm = GRPC_TIMER_ELAPSED(grpc_profile_next);
    GRPC_TIMER_START(grpc_profile_next);
  }
#endif

  /* compare greater than 0 here; elsewhere greater than 1.  this is because
   * at this point we haven't incremented yet.
   */
  if(handle->num_calls > 0)
    agent_taskid = -1;
  else
    agent_taskid = handle->agent_taskid;


#ifdef GS_SMART_GRIDSOLVE

printf("\n\n\n\n");
printf(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n");
printf("Executing task %s on server %s:%d\n", handle->problem_desc->name, srv->hostname, srv->port);

#endif

  sock = gs_connect_to_host(srv->componentid, srv->ipaddress, srv->port,
                            srv->proxyip, srv->proxyport);



/*
 * Implementing fault tolerance for SmartGridSolve
 */ 

 if(sock == INVALID_SOCKET){
#ifdef GS_SMART_GRIDSOLVE
  if(smart_phase!=GS_SMART_STANDARD_EXEC){
    smart_phase=GS_SMART_EXEC_TASK_FAIL;

    /*
     * SmartGridSolve fault tolerance
     * Failed to connect to server
     * Sending failed server to agent to be removed from pm
     */

  printf("SMART: Server %s has failed during execution .\n", srv->hostname);
     if((sock = gs_connect_to_agent()) == INVALID_SOCKET)
       GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_AGENT_NOT_SET);


  if((gs_send_tag(sock, GS_SMART_FAULT_UPDATE_PM) < 0) ||
     (gs_send_string(sock, VERSION) < 0) ||
     (gs_recv_tag(sock, &tag) < 0))
    goto error_communication_failed;
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH) {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_VERSION_MISMATCH);
    } else {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }
  }


    char *srv_str = NULL;
    if((gs_encode_server(&srv_str, srv) < 0) ||
       (gs_send_string(sock, srv_str) < 0)) {
      FREE(srv);
      DBGPRINTF("Failed to send server list \n");
      return -1;
    }

    FREE(srv_str);

    

  printf("SMART: Server %s has been removed from network graph .\n", srv->hostname);
  printf("SMART: Remapping group of tasks.\n");


  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;

 


    return GRPC_NO_ERROR;
  }
  else{
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_SERVER_CONNECTION);
  } 
#else 
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_SERVER_CONNECTION);
#endif
  }

  if(gs_encode_problem_solve_request(&msg, handle->func_name, 
     grpc_user, grpc_host, grpc_domain, grpc_cid_str, my_dsig, 
     agent_taskid, srv->score) < 0) {
    grpc_minor_errno = GRPC_SOLVE_REQ_ENCODE;
    goto error_client_internal_error;
  }

  if(handle->problem_desc == NULL)
    tag = GS_PROT_PROBLEM_SOLVE_ASSIGNED;
  else
    tag = GS_PROT_PROBLEM_SOLVE;

  if((gs_send_tag(sock, tag) < 0) ||
     (gs_send_string(sock, VERSION) < 0))
    goto error_communication_failed;

  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;
  
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      grpc_minor_errno = GRPC_VERSION_MISMATCH;
    else
      grpc_minor_errno = GRPC_NO_MINOR_ERROR;

    goto error_client_internal_error;
  }

  tag = (blocking) ? GS_PROT_BLOCKING : GS_PROT_NONBLOCKING;

  if(gs_send_tag(sock, tag) < 0)
    goto error_communication_failed;

  if(gs_send_string(sock, msg) < 0)
    goto error_communication_failed;

  FREE(msg);

  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;

  if(tag != GS_PROT_OK) {
    ERRPRINTF("Error occured\n");
    switch (tag) {
      case GS_SVC_ERR_EXEC:
        proxy_close(sock);
        GRPC_RETURN(GRPC_FUNCTION_NOT_FOUND, GRPC_NO_MINOR_ERROR);
      default:
        ERRPRINTF("%s\n", gs_service_error[tag]);
        proxy_close(sock);
        GRPC_RETURN(GRPC_OTHER_ERROR_CODE, tag);
    }
  }

  if(gs_recv_string(sock, &request_id) < 0)
    goto error_communication_failed;

  if(handle->problem_desc == NULL) {
    /*
     * this must be an assigned server request, so get the problem desc from 
     * the server.  also since the client won't know what architecture the
     * server is, the server will send its data signature now. 
     */
    if(gs_recv_string(sock, &msg) < 0)
      goto error_communication_failed;
    sscanf(msg, "%d", &(handle->server_list[handle->srv_idx]->data_format));
    free(msg);

    if(gs_recv_string(sock, &msg) < 0)
      goto error_communication_failed;
    handle->problem_desc = (gs_problem_t *) CALLOC(1, sizeof(gs_problem_t));
    if(gs_decode_problem(msg, handle->problem_desc) < 0) {
      grpc_minor_errno = GRPC_PROBLEM_DECODE;
      goto error_client_internal_error;
    }
    FREE(msg);
  }

  handle->num_calls++;
#ifdef GS_SMART_GRIDSOLVE
  if((nb_mapped_tasks_executed<total_nb_mapped_tasks) &
    (smart_phase==GS_SMART_STANDARD_EXEC)){
    /*
     * If previous execution of this mapped tasks 
     * failed then set the remote comm variable to 0
     */
    handle->problem_desc->has_smart_arg_comm=0;

  }
#endif

  if(blocking) {
    if(gs_handle_blocking_call(handle, argptr, argstack, sock, my_dsig) < 0)
    {
      if(gs_notify_agent_of_failure(handle, request_id) < 0)
        ERRPRINTF("Warning: failed to notify agent of server failure.\n");
      FREE(request_id);
      goto error_communication_failed;
    }
#ifdef GS_SMART_GRIDSOLVE
   /*
    * after this blocking task is executed, check if there are still more 
    * mapped tasks to execute. if so then increment the number of mapped tasks executed
    */    
    if(nb_mapped_tasks_executed<total_nb_mapped_tasks){
      nb_mapped_tasks_executed++;
    }
    /*
     * if the last mapped task has just been executed
     * then reset counters
     */ 
    else{
      //free mapping and maybe task graph
      nb_mapped_tasks_executed=0;
      total_nb_mapped_tasks=0;
    }
#endif


    FREE(request_id);
  }
  else {
    if(gs_handle_nonblocking_call(handle, sessionId, argptr, argstack, sock,
          my_dsig, request_id) < 0)
    {
      if(gs_notify_agent_of_failure(handle, request_id) < 0)
        ERRPRINTF("Warning: failed to notify agent of server failure.\n");
      goto error_communication_failed;
    }

#ifdef GS_SMART_GRIDSOLVE
   /*
    * after this blocking task is executed, check if there are still more 
    * mapped tasks to execute. if so then increment the number of mapped tasks executed
    */    
    if(nb_mapped_tasks_executed<total_nb_mapped_tasks){
      nb_mapped_tasks_executed++;
    }
    /*
     * if the last mapped task has just been executed
     * then reset counters
     */ 
    else{
      //free mapping and maybe task graph
      nb_mapped_tasks_executed=0;
      total_nb_mapped_tasks=0;
    }
#endif
    FREE(msg);
    GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }

  FREE(msg)
  proxy_close(sock);
  return GRPC_NO_ERROR;


error_communication_failed:
#ifdef GS_SMART_GRIDSOLVE
  if(smart_phase!=GS_SMART_STANDARD_EXEC){
    smart_phase=GS_SMART_EXEC_TASK_FAIL;
    if(sock != INVALID_SOCKET){
      proxy_close(sock);
    }
    FREE(msg);

  printf("SMART: Server %s has failed during execution .\n", srv->hostname);
    /*
     * SmartGridSolve fault tolerance
     * Failed to communicate with servers (inputs, outputs, tags etc.)
     * Sending failed server to agent to be removed from pm
     */
    
     if((sock = gs_connect_to_agent()) == INVALID_SOCKET)
       GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_AGENT_NOT_SET);
  
     if((gs_send_tag(sock, GS_SMART_FAULT_UPDATE_PM) < 0) ||
        (gs_send_string(sock, VERSION) < 0) ||
        (gs_recv_tag(sock, &tag) < 0))
          goto error_communication_failed;
        
     if(tag != GS_PROT_OK) {
       if(tag == GS_PROT_VERSION_MISMATCH) {
         GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_VERSION_MISMATCH);
       } else {
        GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
      }
    }


    char *srv_str = NULL;
    if((gs_encode_server(&srv_str, srv) < 0) ||
       (gs_send_string(sock, srv_str) < 0)) {
      FREE(srv);
      DBGPRINTF("Failed to send server list \n");
      return -1;
    }


    
  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;

  printf("SMART: Server %s has been removed from network graph .\n", srv->hostname);
  printf("SMART: Remapping of the group of tasks will now be implemented .\n");
    
FREE(srv_str);
    return GRPC_NO_ERROR;

  }
  else{
    if(sock != INVALID_SOCKET){
      proxy_close(sock);
    }
    FREE(msg);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, grpc_minor_errno);
  }
#else
  if(sock != INVALID_SOCKET){
    proxy_close(sock);
  }
  FREE(msg);
  GRPC_RETURN(GRPC_COMMUNICATION_FAILED, grpc_minor_errno);
#endif

error_client_internal_error:
  if(sock != INVALID_SOCKET)
    proxy_close(sock);
  FREE(msg);
  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);
}

grpc_error_t
gs_wait_common(grpc_request_t *req)
{
  int my_dsig;
  SOCKET sock;
  int tag;
  gs_server_t *srv;

  grpc_minor_errno = GRPC_NO_MINOR_ERROR;

  my_dsig = pvmgetdsig();

  if(!req->handle)
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);


#ifdef GS_SMART_GRIDSOLVE
  /*
   * Count the number of output arguments that should be 
   * received from the server as outlined in the mapping 
   * solution. If there are none the function returns.
   */ 
  
  gs_argument_t *argptr;
  int nb_args_to_receive=0;
  for(argptr= req->handle->problem_desc->arglist;argptr!=NULL;argptr=argptr->next) {
    if(argptr->objecttype!=GS_SCALAR){
      if( ( argptr->inout==GS_INOUT ) || ( argptr->inout==GS_OUT ) ) {
        if(argptr->output_arg_sent_remotely==0) nb_args_to_receive++;
      }
    }
  }
  if(nb_args_to_receive==0){
    GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }
#endif


  if(req->s_pid > 0)
    gs_req_child_waitpid(req, GS_WBLOCK);

  srv = req->handle->server_list[req->handle->srv_idx];

  sock = gs_connect_to_host(srv->componentid, srv->ipaddress,
                            srv->port, srv->proxyip, srv->proxyport);

  if(sock == INVALID_SOCKET)
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_SERVER_CONNECTION);

  if((gs_send_tag(sock, GS_PROT_AWAIT_COMPLETION) < 0) ||
     (gs_send_string(sock, VERSION) < 0))
    goto error_communication_failed;

  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;

  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      grpc_minor_errno = GRPC_VERSION_MISMATCH;
    else
      grpc_minor_errno = GRPC_NO_MINOR_ERROR;

    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);
  }

  if(gs_send_string(sock, req->request_id) < 0)
    goto error_communication_failed;


#ifdef GS_SMART_GRIDSOLVE
  
 /*
  * For each output argument send the variable output_arg_sent_remotely to 
  * the server. This specifies whether the output arg is sent remotely or
  * back to the client as outlined by the mapping solution.  
  */  

  for(argptr= req->handle->problem_desc->arglist;argptr!=NULL;argptr=argptr->next) {
    if(argptr->objecttype!=GS_SCALAR){
      if( ( argptr->inout==GS_INOUT ) || ( argptr->inout==GS_OUT ) ) {
        if(gs_send_int(sock, argptr->output_arg_sent_remotely) < 0) {
          ERRPRINTF("gs_send output remote param\n");
          return -1;         
        }
      }
    }
  }


#endif



  if(gs_wait_for_output(sock) < 0)
    goto error_communication_failed;

  if(gs_recv_tag(sock, &tag) < 0)
    goto error_communication_failed;

  if(tag != GS_PROT_OK) {
    ERRPRINTF("Service failed: %s\n", gs_service_error[tag]);
    grpc_minor_errno = tag;
    goto error_protocol_error;
  }



#ifdef GS_SMART_GRIDSOLVE
 /*
  * The client receives only the output arguments from the server
  * which have been outlined in the mapping solution.
  */
  
  if(gs_smart_recv_output_args_from_server(sock, req->problem, req->handle->problem_desc,
              req->handle->server_list[req->handle->srv_idx]->data_format,  my_dsig) < 0){
    goto error_communication_failed;
  }

#else

  if(gs_recv_output_args(sock, req->problem,
       req->handle->server_list[req->handle->srv_idx]->data_format, 
       my_dsig) < 0)
    goto error_communication_failed;
#endif

  proxy_close(sock);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);

error_communication_failed:
  if(sock != INVALID_SOCKET)
    proxy_close(sock);
  if(gs_notify_agent_of_failure(req->handle, req->request_id) < 0)
    ERRPRINTF("Warning: failed to notify agent of server failure.\n");
  GRPC_RETURN(GRPC_COMMUNICATION_FAILED, grpc_minor_errno);

error_protocol_error:
  if(sock != INVALID_SOCKET)
    proxy_close(sock);
  if(gs_notify_agent_of_failure(req->handle, req->request_id) < 0)
    ERRPRINTF("Warning: failed to notify agent of server failure.\n");
  GRPC_RETURN(GRPC_SESSION_FAILED, grpc_minor_errno);
}

static int
grpc_request_destruct(grpc_request_t *req)
{
  if(!req)
    return -1;

  /*
   * don't free handle since it's only a pointer to the user's data. we'll
   * let them free it. 
   */
  if(req->request_id)
    free(req->request_id);
  req->request_id = NULL;
  if(req->problem)
    gs_free_problem(req->problem);

  return 0;
}

int
grpc_request_destruct_free_clear(grpc_sessionid_t sessionId)
{
  int rc = -1;

  if((sessionId < 0) || (sessionId >= MAX_GRPC_REQUESTS))
    return 0;

  rc = grpc_request_destruct(grpc_outstanding_requests[sessionId]);
  if(rc != 0)
    return -1;
  if(grpc_outstanding_requests[sessionId])
    free(grpc_outstanding_requests[sessionId]);
  grpc_outstanding_requests[sessionId] = NULL;
  grpc_profile_info[sessionId] = NULL;

  return 0;
}

grpc_error_t
grpc_call_ft(grpc_function_handle_t *handle, ...)
{
  gs_va_list argptr;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  va_start(argptr.args, handle);

  return gs_call_common_ft(handle, NULL, &argptr, NULL, TRUE);
}

grpc_error_t
grpc_call_valist_ft(grpc_function_handle_t *handle, gs_va_list *argptr)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  return gs_call_common_ft(handle, NULL, argptr, NULL, TRUE);
}

grpc_error_t
grpc_call_async_ft(grpc_function_handle_t *handle, 
   grpc_sessionid_t *sessionId, ...)
{
  gs_va_list argptr;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  va_start(argptr.args, sessionId);

  return gs_call_common_ft(handle, sessionId, &argptr, NULL, FALSE);
}

grpc_error_t
grpc_call_valist_async_ft(grpc_function_handle_t *handle, 
   grpc_sessionid_t *sessionId, gs_va_list *argptr)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  return gs_call_common_ft(handle, sessionId, argptr, NULL, FALSE);
}

grpc_error_t
grpc_call_arg_stack_ft(grpc_function_handle_t *handle, grpc_arg_stack *stack)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!stack || !stack->args)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_ARGSTACK);
  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  return gs_call_common_ft(handle, NULL, NULL, stack->args, TRUE);
}

grpc_error_t
grpc_call_arg_stack_async_ft(grpc_function_handle_t *handle,
   grpc_sessionid_t *sessionId, grpc_arg_stack *stack)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!stack || !stack->args)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_ARGSTACK);
  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  return gs_call_common_ft(handle, sessionId, NULL, stack->args, FALSE);
}

static grpc_error_t
gs_call_common_ft(grpc_function_handle_t *handle, grpc_sessionid_t *sessionId, 
   gs_va_list *argptr, void **argstack, int blocking)
{
  int c;
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  for(c = 0; c < handle->num_servers; c++) {
    grpc_error_t rv;
#ifdef GS_SMART_GRIDSOLVE
    /*
     * If currently executing a mapped group of tasks 
     * a task fails on its first server, then set
     * standard exec phase so that the execution
     * proceeds using the standard gridsolve protocol.
     */ 
    if((c>0) && (nb_mapped_tasks_executed<total_nb_mapped_tasks)){
      smart_phase=GS_SMART_STANDARD_EXEC;
    }
#endif
    rv = gs_call_common(handle, sessionId, argptr, argstack, blocking);

    if(rv == GRPC_NO_ERROR)
      GRPC_RETURN(rv, GRPC_NO_MINOR_ERROR);

    if(handle->num_servers == 0)
      break;
    handle->srv_idx = (handle->srv_idx + 1) % handle->num_servers;
  }
  handle->srv_idx = 0;
  GRPC_RETURN(GRPC_SERVER_NOT_FOUND, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_profile(grpc_profile_t *prof)
{
#ifdef GS_PROFILING
  grpc_profile_next = prof;
  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
#else

  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_PROFILING_NOT_ENABLED);
#endif
}

grpc_error_t
grpc_set_client_major(char *maj)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!maj)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_MAJOR);

  if((maj[0] == 'R') || (maj[0] == 'r')) {
    grpc_client_major = 'r';
    grpc_user_set_major = TRUE;
  }
  else if((maj[0] == 'C') || (maj[0] == 'c')) {
    grpc_client_major = 'c';
    grpc_user_set_major = TRUE;
  }
  else
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_MAJOR);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_set_default_major(char *maj)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!grpc_user_set_major)
    return grpc_set_client_major(maj);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_set_client_language(int lang)
{
  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((lang != GS_CALL_FROM_C) && (lang != GS_CALL_FROM_FORTRAN))
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_INVALID_LANG);

  grpc_client_lang = lang;

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_probe_ft(grpc_sessionid_t sessionID)
{
  grpc_function_handle_t *handle;
  grpc_error_t retval;
  int c; 

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  retval = grpc_probe(sessionID);

  /*
   * if probe returns 
   */

  if((retval == GRPC_NO_ERROR) || (retval == GRPC_NOT_COMPLETED))
    GRPC_RETURN(retval, grpc_minor_errno);

  handle = grpc_outstanding_requests[sessionID]->handle;

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  /*
   * if we reach this point, then the call to grpc_probe detected a failure
   * and we should resubmit the request to another server. 
   */

  for(c = 0; c < handle->num_servers; c++) {
    if(gs_resubmit_common(sessionID) == GRPC_NO_ERROR)
      GRPC_RETURN(GRPC_NOT_COMPLETED, GRPC_NO_MINOR_ERROR);

    if(handle->num_servers == 0)
      break;

    handle->srv_idx = (handle->srv_idx + 1) % handle->num_servers;
  }

  handle->srv_idx = 0;

  GRPC_RETURN(GRPC_SERVER_NOT_FOUND, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_wait_ft(grpc_sessionid_t sessionID)
{
  grpc_function_handle_t *handle;
  int c;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(grpc_wait(sessionID) == GRPC_NO_ERROR)
    return GRPC_NO_ERROR;

  handle = grpc_outstanding_requests[sessionID]->handle;

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  /*
   * if we reach this point, then the call to grpc_wait detected a failure
   * and we should resubmit the request to another server. 
   */

  for(c = 0; c < handle->num_servers; c++) {
    if(gs_resubmit_common(sessionID) == GRPC_NO_ERROR) {

      /*
       * ok, we were able to resubmit the job successfully, so now we wait
       * for it to complete.  if this wait fails, we keep attempting to
       * resubmit until no more servers remain. 
       */
      if(grpc_wait(sessionID) == GRPC_NO_ERROR)
        return GRPC_NO_ERROR;
    }

    if(handle->num_servers == 0)
      break;
    handle->srv_idx = (handle->srv_idx + 1) % handle->num_servers;
  }

  handle->srv_idx = 0;
  GRPC_RETURN(GRPC_SERVER_NOT_FOUND, GRPC_NO_MINOR_ERROR);
}

static grpc_error_t
gs_resubmit_common(grpc_sessionid_t sessionID)
{
  grpc_function_handle_t *handle;
  int tag;
  SOCKET sock;
  int my_dsig;
  char *msg, *request_id = NULL;
  gs_server_t *srv;

  if((sessionID < 0) || (sessionID >= MAX_GRPC_REQUESTS))
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  if(!grpc_outstanding_requests[sessionID])
    GRPC_RETURN(GRPC_INVALID_SESSION_ID, GRPC_NO_MINOR_ERROR);

  handle = grpc_outstanding_requests[sessionID]->handle;

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!handle->func_name)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_NAME);

  srv = handle->server_list[handle->srv_idx];

  sock = gs_connect_to_host(srv->componentid, srv->ipaddress, srv->port,
                            srv->proxyip, srv->proxyport);

  if(sock < 0)
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_SERVER_CONNECTION);

  my_dsig = pvmgetdsig();
  handle->agent_taskid = -1;

  if(gs_encode_problem_solve_request(&msg, handle->func_name, 
     grpc_user, grpc_host, grpc_domain, grpc_cid_str, my_dsig,
     handle->agent_taskid, srv->score) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SOLVE_REQ_ENCODE);
  }

  tag = GS_PROT_PROBLEM_SOLVE;

  if((gs_send_tag(sock, tag) < 0) ||
     (gs_send_string(sock, VERSION) < 0)) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  if(gs_recv_tag(sock, &tag) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH)
      grpc_minor_errno = GRPC_VERSION_MISMATCH;
    else
      grpc_minor_errno = GRPC_NO_MINOR_ERROR;

    proxy_close(sock);
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, grpc_minor_errno);
  }

  tag = GS_PROT_NONBLOCKING;

  if(gs_send_tag(sock, tag) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  if(gs_send_string(sock, msg) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  free(msg);

  if(gs_recv_tag(sock, &tag) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  if(tag != GS_PROT_OK) {
    switch (tag) {
      case GS_SVC_ERR_EXEC:
        proxy_close(sock);
        GRPC_RETURN(GRPC_FUNCTION_NOT_FOUND, GRPC_NO_MINOR_ERROR);
      default:
        ERRPRINTF("%s\n", gs_service_error[tag]);
        proxy_close(sock);
        GRPC_RETURN(GRPC_OTHER_ERROR_CODE, tag);
        break;
    }
  }

  if(gs_recv_string(sock, &request_id) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  if(gs_send_input_args(NULL, NULL, sock, handle->problem_desc,
                        my_dsig, grpc_client_lang, grpc_client_major) < 0) {
    proxy_close(sock);
    GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  }

  /*
   * free the old request id and assign the new one 
   */
  free(grpc_outstanding_requests[sessionID]->request_id);
  grpc_outstanding_requests[sessionID]->request_id = request_id;

  proxy_close(sock);

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

static int
gs_free_handle_server_list(grpc_function_handle_t *handle)
{
  int i;

  if(!handle)
    return -1;

  if(handle->server_list) {
    /*
     * Remove/destroy any old server mapping memory/data-structures 
     */
    for(i = 0; i < handle->num_servers; i++)
      gs_server_free(handle->server_list[i]);
    free(handle->server_list);
    handle->srv_idx = 0;
    handle->num_servers = 0;
    handle->server_list = NULL;
  }

  return 0;
}

grpc_error_t *
grpc_farm(char *iteration, char *func_name, ...)
{
  int start, end, *returned_value;
  va_list argptr;
  char *buf;

  GRPC_FAIL_IF_NOT_INITIALIZED(NULL);

  if((!iteration) || (!func_name)) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return NULL;
  }

  buf = strdup(iteration);

  if(!buf) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return NULL;
  }

  /* get the <start> end the <end> */
  if (sscanf(buf,"i=%d,%d",&start,&end) != 2)
  {
    free(buf);
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    returned_value = (int *)calloc(1,sizeof(int));
    returned_value[0] = GRPC_OTHER_ERROR_CODE;
    return returned_value;
  }

  free(buf);

  va_start(argptr, func_name);

  return grpc_farming(start, end, func_name, argptr);
}

grpc_error_t
grpc_farm_set_failure_status(int *statuses, int num_req, 
   grpc_sessionid_t *sessionids)
{
  grpc_sessionid_t failed_sid;
  int i;

  if(grpc_get_failed_sessionid(&failed_sid) == GRPC_NO_ERROR) {
    for(i = 0; i < num_req; i++) {
      if(sessionids[i] == failed_sid) {
        statuses[i] = grpc_get_error(failed_sid);
        GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
      }
    }
  }

  /* if we reach here, the session id was not found */
  GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_FARM_SESSIONID_NOT_FOUND);
}

grpc_error_t *
grpc_farming(int start, int end, char *func_name, va_list argptr)
{
  int *statuses;
  int nb_requests;
  int window_size;          /* window size = 2 * # of capable servers */
  int status;
  int i, j;
  int nb_args;
  int pending;
  gs_argument_t *argp = NULL;
  grpc_iterator_t **iterator_array = NULL;
  grpc_error_t *returned_value;
  grpc_sessionid_t *sessionids;
  grpc_function_handle_t *handles;

  GRPC_FAIL_IF_NOT_INITIALIZED(NULL);

  if(!func_name) {
    grpc_errno = GRPC_NOT_INITIALIZED;
    grpc_minor_errno = GRPC_NULL_FUNCTION_NAME;
    return NULL;
  }

  /*
   * Start the farming 
   */
  nb_requests = end - start + 1;
  statuses = (int *) calloc(nb_requests, sizeof(int));
  sessionids = (grpc_sessionid_t *) calloc(nb_requests,
     sizeof(grpc_sessionid_t));
  handles = (grpc_function_handle_t *) calloc(nb_requests,
     sizeof(grpc_function_handle_t));

  if(!statuses || !sessionids || !handles) {
    if(statuses) free(statuses);
    if(sessionids) free(sessionids);
    if(handles) free(handles);
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    grpc_minor_errno = GRPC_NO_MINOR_ERROR;
    return NULL;
  }

  if(grpc_function_handle_default(&handles[0], func_name) != GRPC_NO_ERROR) {
    ERRPRINTF("Could not get function handle\n");
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    grpc_minor_errno = GRPC_NO_MINOR_ERROR;
    return NULL;
  }

  /*
   * Getting the nb_args 
   */
  nb_args = 0;
  for(argp = handles[0].problem_desc->arglist; argp != NULL; argp = argp->next)
    nb_args++;

  window_size = handles[0].num_servers * 2;

  grpc_function_handle_destruct(&handles[0]);

  if(grpc_construct_iterator_array(argptr, nb_args, &iterator_array, start) == -1)
    ERRPRINTF("Could not construct iterator array\n");

  /*
   * Filling problem_desc->arglist->data 
   */

  DBGPRINTF("# of requests is %d \n", nb_requests);

  if(nb_requests <= window_size) {
    j = nb_requests;
    pending = 0;
  }
  else {
    j = window_size;
    pending = 1;
  }

  for(i = 0; i < j; i++)
  {
    
    if(grpc_function_handle_default(&handles[i], func_name) != 
        GRPC_NO_ERROR)
    {
      statuses[i] = grpc_errno;
      sessionids[i] = -1;
    }
    else 
      statuses[i] = grpc_send_farming_request(&handles[i], &sessionids[i], 
        iterator_array, nb_args);
  }

  if(!pending) {
    if(grpc_wait_and(sessionids, nb_requests) != GRPC_NO_ERROR)
      grpc_farm_set_failure_status(statuses, nb_requests, sessionids);
  }
  else {                        
    int num_retrieved = 0;

    while(num_retrieved < nb_requests) {
      if(grpc_wait_any(&status) != GRPC_NO_ERROR)
        grpc_farm_set_failure_status(statuses, nb_requests, sessionids);
      
      num_retrieved++;

      if(i < nb_requests) {
        if(grpc_function_handle_default(&handles[i], func_name) != 
            GRPC_NO_ERROR)
        {
          statuses[i] = grpc_errno;
          sessionids[i] = -1;
        }
        else
          statuses[i] =
            grpc_send_farming_request(&handles[i], &sessionids[i], 
               iterator_array, nb_args);
        i++;
      }
    }
  }

  status = 1;
  for(i = 0; i < nb_requests; i++) {
    if(statuses[i] != GRPC_NO_ERROR)
      status = -1;

    grpc_function_handle_destruct(&handles[i]);
  }
  free(handles);

  if(status == -1) {
    returned_value = (grpc_error_t *) calloc(nb_requests + 1, sizeof(grpc_error_t));
    returned_value[0] = GRPC_OTHER_ERROR_CODE;
    for(i = 0; i < nb_requests; i++)
      returned_value[i + 1] = statuses[i];
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    grpc_minor_errno = GRPC_NO_MINOR_ERROR;
  }
  else {
    returned_value = (grpc_error_t *) calloc(1, sizeof(grpc_error_t));
    returned_value[0] = GRPC_NO_ERROR;
    grpc_errno = GRPC_NO_ERROR;
    grpc_minor_errno = GRPC_NO_MINOR_ERROR;
  }

  return returned_value;
}


grpc_error_t
grpc_send_farming_request(grpc_function_handle_t *handle, grpc_sessionid_t *sid,
                          grpc_iterator_t **iterator_array, int nb_args)
{
  gs_service_error_enum_t save_minor_errno;
  grpc_arg_stack *stack = NULL;
  gs_argument_t *argp = NULL;
  grpc_error_t status;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!handle)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NULL_FUNCTION_HANDLE);
  if(!iterator_array)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_FARM_NULL_ITERATOR);

  status =
      grpc_setup_farming_args(handle->problem_desc, iterator_array, nb_args);
  if(status != GRPC_NO_ERROR)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_FARM_SETUP_ARGS_FAILED);

  /*
   * Creating Stack 
   */
  stack = grpc_arg_stack_new(nb_args);
  if(!stack)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_FARM_CREATE_STACK_FAILED);

  for(argp = handle->problem_desc->arglist; argp != NULL; argp = argp->next) {
    if(grpc_arg_stack_push_arg(stack, argp->data) < 0)
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_FARM_CREATE_STACK_FAILED);
  }

  /*
   * Calling solve routine with stack interface 
   */
  status = grpc_call_arg_stack_async_ft(handle, sid, stack);
  save_minor_errno = grpc_minor_errno;
  grpc_arg_stack_destruct(stack);
  GRPC_RETURN(status, save_minor_errno);
}

grpc_error_t
grpc_setup_farming_args(gs_problem_t *problem, grpc_iterator_t **it_array,
                        int nb_args)
{
  void *dataptr = NULL;
  gs_argument_t *argp = NULL;
  int i = 0;
  grpc_void_star_or_int_t *stackarray;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!problem || !it_array)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);

  /*
   * Creating Array from Iterators 
   */
  if(grpc_generate_array_from_iterators
     (problem, &stackarray, it_array, nb_args) != GRPC_NO_ERROR)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);

  /*
   * For each argument, check datatype, and attach a dataptr to the arg->data 
   */
  for(argp = problem->arglist, i = 0; argp != NULL; argp = argp->next, i++) {
    switch (argp->datatype) {
      case GS_DOUBLE:
        dataptr = (double *) calloc(1, sizeof(double));
        if(!dataptr)
          GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
        dataptr = (double *) (stackarray[i].ptr);
        break;
      case GS_SCOMPLEX:
        dataptr = (gs_scomplex *) calloc(1, sizeof(gs_scomplex));
        if(!dataptr)
          GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
        dataptr = (gs_scomplex *) (stackarray[i].ptr);
        break;
      case GS_DCOMPLEX:
        dataptr = (gs_dcomplex *) calloc(1, sizeof(gs_dcomplex));
        if(!dataptr)
          GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
        dataptr = (gs_dcomplex *) (stackarray[i].ptr);
        break;
      case GS_INT:
        dataptr = (int *) calloc(1, sizeof(int));
        if(!dataptr)
          GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
        if(argp->objecttype == GS_SCALAR)
          ((int *) dataptr)[0] = stackarray[i].i;
        else
          dataptr = (int *) (stackarray[i].ptr);
        break;
      case GS_FLOAT:
        dataptr = (float *) calloc(1, sizeof(float));
        if(!dataptr)
          GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
        /*
         * if (argp->inout == GS_IN || argp->inout == GS_INOUT) 
         */
        dataptr = (float *) (stackarray[i].ptr);
        break;
      case GS_CHAR:
        dataptr = (char *) calloc(1, sizeof(char));
        dataptr = (char *) (stackarray[i].ptr);
        break;
      default:
        ERRPRINTF("This datatype is not handled yet\n");
        GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }
    argp->data = dataptr;
  }

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_generate_array_from_iterators(gs_problem_t *pd,
                                   grpc_void_star_or_int_t **stackarray,
                                   grpc_iterator_t **iterator_array,
                                   int nb_args)
{
  grpc_void_star_or_int_t *new;
  int i;
  grpc_iterator_t *it;
  int arg_type;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if(!pd || !iterator_array)
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);

  new =
      (grpc_void_star_or_int_t *) calloc(nb_args,
                                         sizeof(grpc_void_star_or_int_t));
  for(i = 0; i < nb_args; i++) {
    it = iterator_array[i];
    arg_type = it->returned_type;
    switch (arg_type) {
      case IGNORE:
        new[i].ptr = NULL;
        break;
      case POINTER:
        new[i].ptr = grpc_get_next(it, arg_type).ptr;
        break;
      case INTEGER:
        new[i].i = grpc_get_next(it, arg_type).i;
        break;
    }
  }

  *stackarray = new;

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

grpc_error_t
grpc_construct_iterator_array(va_list argptr, int nb_args,
                              grpc_iterator_t ***iterator_array, int start)
{
  grpc_iterator_t **new;
  int i;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  new = (grpc_iterator_t **) calloc(nb_args, sizeof(grpc_iterator_t *));

  for(i = 0; i < nb_args; i++) {
    new[i] = (grpc_iterator_t *) va_arg(argptr, void *);
    new[i]->i = start;
  }
  *iterator_array = new;

  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
}

void
grpc_free_iterator(grpc_iterator_t *it)
{
  if(!it) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return;
  }

  grpc_free_specific_iterator(it->specific);
  free(it);
}

void
grpc_free_specific_iterator(grpc_specific_iterator_t *it)
{
  if(!it) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return;
  }

  switch (it->type) {
    case INT_ITERATOR:
      free(it->it.int_iterator->expression);
      free(it->it.int_iterator);
      break;
    case INT_ARRAY_ITERATOR:
      free(it->it.int_array_iterator->expression);
      free(it->it.int_array_iterator);
      break;
    case PTR_ARRAY_ITERATOR:
      free(it->it.ptr_array_iterator->expression);
      free(it->it.ptr_array_iterator);
      break;
  }

  free(it);
}

grpc_iterator_t *
grpc_int(char *s)
{
  grpc_iterator_t *it;

  if(!s) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return NULL;
  }

  it = (grpc_iterator_t *) calloc(1, sizeof(grpc_iterator_t));
  it->returned_type = INTEGER;
  it->specific =
      (grpc_specific_iterator_t *) calloc(1,
                                          sizeof(grpc_specific_iterator_t));
  it->specific->type = INT_ITERATOR;
  it->specific->it.int_iterator =
      (grpc_int_iterator_t *) calloc(1, sizeof(grpc_int_iterator_t));
  it->specific->it.int_iterator->expression = strdup(s);

  return it;
}

grpc_iterator_t *
grpc_int_array(int *array, char *expression)
{
  grpc_iterator_t *it;

  if(!expression) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return NULL;
  }

  it = (grpc_iterator_t *) calloc(1, sizeof(grpc_iterator_t));
  it->returned_type = INTEGER;
  it->specific =
      (grpc_specific_iterator_t *) calloc(1,
                                          sizeof(grpc_specific_iterator_t));
  it->specific->type = INT_ARRAY_ITERATOR;
  it->specific->it.int_array_iterator =
      (grpc_int_array_iterator_t *) calloc(1,
                                           sizeof(grpc_int_array_iterator_t));
  it->specific->it.int_array_iterator->array = array;
  it->specific->it.int_array_iterator->expression = strdup(expression);

  return it;
}

grpc_iterator_t *
grpc_ptr_array(void **array, char *expression)
{
  grpc_iterator_t *it;

  if(!expression) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return NULL;
  }

  it = (grpc_iterator_t *) calloc(1, sizeof(grpc_iterator_t));
  it->returned_type = POINTER;
  it->specific =
      (grpc_specific_iterator_t *) calloc(1,
                                          sizeof(grpc_specific_iterator_t));
  it->specific->type = PTR_ARRAY_ITERATOR;
  it->specific->it.ptr_array_iterator =
      (grpc_ptr_array_iterator_t *) calloc(1,
                                           sizeof(grpc_ptr_array_iterator_t));
  it->specific->it.ptr_array_iterator->array = array;
  it->specific->it.ptr_array_iterator->expression = strdup(expression);

  return it;
}

grpc_void_star_or_int_t
grpc_get_next(grpc_iterator_t *it, int arg_type)
{
  grpc_void_star_or_int_t stuff;
  stuff.ptr = NULL;

  if(!it) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return stuff;
  }

  stuff = grpc_get_value(it->specific, it->i, arg_type);
  (it->i)++;

  return stuff;
}

grpc_void_star_or_int_t
grpc_get_value(grpc_specific_iterator_t *it, int i, int arg_type)
{
  grpc_void_star_or_int_t stuff;
  int *x;
  stuff.ptr = NULL;

  if(!it) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    return stuff;
  }

  switch (it->type) {
    case INT_ITERATOR:
      stuff.i = grpc_eval_integer(it->it.int_iterator->expression, i);
      break;
    case INT_ARRAY_ITERATOR:
      if(arg_type == INTEGER) {
        stuff.i =
            (it->it.int_array_iterator->
             array)[grpc_eval_integer(it->it.int_array_iterator->expression,
                                      i)];
      }
      else if(arg_type == POINTER) {
        stuff.ptr = (it->it.int_array_iterator->array) +
            grpc_eval_integer(it->it.int_array_iterator->expression, i);
      }
      break;
    case PTR_ARRAY_ITERATOR:
      if(arg_type == POINTER) {
        stuff.ptr =
            (it->it.ptr_array_iterator->
             array)[grpc_eval_integer(it->it.ptr_array_iterator->expression,
                                      i)];
      }
      if(arg_type == INTEGER) {
        x = (it->it.ptr_array_iterator->
             array)[grpc_eval_integer(it->it.ptr_array_iterator->expression,
                                      i)];
        stuff.i = *x;
      }
      break;
    default:
      ERRPRINTF("Unknown ITERATOR\n");
      stuff.i = 0;
  }

  return stuff;
}

int
grpc_eval_integer(char *s, int i)
{
  return i;
}



#ifdef GS_SMART_GRIDSOLVE




grpc_error_t
grpc_update_server_list(char **info)
{
  char dottedIP[20], proxy_dottedIP[20];
  char *msg=NULL;
  int i, num_servers, tag;
  gs_server_t **server_list;
  SOCKET sock;
  char *str1 = NULL;
  char *str2 = NULL;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sock = gs_connect_to_agent()) == INVALID_SOCKET)
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_AGENT_NOT_SET);

  if((gs_send_tag(sock, GS_PROT_SERVER_LIST) < 0) ||
     (gs_send_string(sock, VERSION) < 0) || 
     (gs_recv_tag(sock, &tag) < 0))
    goto error_communication_failed;
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH) {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_VERSION_MISMATCH);
    } else {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }
  }

  if(gs_recv_int(sock, &num_servers) < 0)
    goto error_communication_failed;
  
  str1 = dstring_sprintf("AGENT: %s [%d servers]\n", agent_resolved->hostname,
     num_servers);
  
  if(num_servers <= 0) {
    *info = str1;
    GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }

  server_list = (gs_server_t **) CALLOC(num_servers, sizeof(gs_server_t *));
  if(!server_list) 
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  for(i=0;i<num_servers;i++) {
    server_list[i] = (gs_server_t *) CALLOC(1,sizeof(gs_server_t));
    if(gs_recv_string(sock, &msg) < 0) {
      goto error_communication_failed;
    }
    if(gs_decode_server(msg, server_list[i]) < 0) {
      FREE(msg);
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SERVER_DECODE);
    }
    FREE(msg)
  }

  for(i=0;i<num_servers;i++) {
    proxy_ip_to_str(server_list[i]->ipaddress, dottedIP);
    proxy_ip_to_str(server_list[i]->proxyip, proxy_dottedIP);
    if(server_list[i]->proxyip != 0)
      str2 = dstring_sprintf("SERVER: %s (%s:%d, proxy=%s:%d)\n", server_list[i]->hostname, 
                 dottedIP, server_list[i]->port, proxy_dottedIP, 
                 server_list[i]->proxyport);
    else
      str2 = dstring_sprintf("SERVER: %s (%s:%d)\n", server_list[i]->hostname, 
                 dottedIP, server_list[i]->port);
    str1 = dstring_append_free(str1, str2);
  }

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

  if(sock != INVALID_SOCKET)  proxy_close(sock);

  *info = str1;
  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  
  
 error_communication_failed:
  DBGPRINTF("Client Error: Communication failed\n");
  FREE(msg);
  if(sock != INVALID_SOCKET)  proxy_close(sock);
  GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  
}




#endif



grpc_error_t
grpc_get_servers(char **info)
{
  char dottedIP[20], proxy_dottedIP[20];
  char *msg=NULL;
  int i, num_servers, tag;
  gs_server_t **server_list;
  SOCKET sock;
  char *str1 = NULL;
  char *str2 = NULL;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sock = gs_connect_to_agent()) == INVALID_SOCKET)
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_AGENT_NOT_SET);

  if((gs_send_tag(sock, GS_PROT_SERVER_LIST) < 0) ||
     (gs_send_string(sock, VERSION) < 0) || 
     (gs_recv_tag(sock, &tag) < 0))
    goto error_communication_failed;
  if(tag != GS_PROT_OK) {
    if(tag == GS_PROT_VERSION_MISMATCH) {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_VERSION_MISMATCH);
    } else {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }
  }

  if(gs_recv_int(sock, &num_servers) < 0)
    goto error_communication_failed;
  
  str1 = dstring_sprintf("AGENT: %s [%d servers]\n", agent_resolved->hostname,
     num_servers);
  
  if(num_servers <= 0) {
    *info = str1;
    GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }

  server_list = (gs_server_t **) CALLOC(num_servers, sizeof(gs_server_t *));
  if(!server_list) 
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);

  for(i=0;i<num_servers;i++) {
    server_list[i] = (gs_server_t *) CALLOC(1,sizeof(gs_server_t));
    if(gs_recv_string(sock, &msg) < 0) {
      goto error_communication_failed;
    }
    if(gs_decode_server(msg, server_list[i]) < 0) {
      FREE(msg);
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SERVER_DECODE);
    }
    FREE(msg)
  }

  for(i=0;i<num_servers;i++) {
    proxy_ip_to_str(server_list[i]->ipaddress, dottedIP);
    proxy_ip_to_str(server_list[i]->proxyip, proxy_dottedIP);
    if(server_list[i]->proxyip != 0)
      str2 = dstring_sprintf("SERVER: %s (%s:%d, proxy=%s:%d)\n", server_list[i]->hostname, 
                 dottedIP, server_list[i]->port, proxy_dottedIP, 
                 server_list[i]->proxyport);
    else
      str2 = dstring_sprintf("SERVER: %s (%s:%d)\n", server_list[i]->hostname, 
                 dottedIP, server_list[i]->port);
    str1 = dstring_append_free(str1, str2);
  }

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

  if(sock != INVALID_SOCKET)  proxy_close(sock);

  *info = str1;
  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  
  
 error_communication_failed:
  DBGPRINTF("Client Error: Communication failed\n");
  FREE(msg);
  if(sock != INVALID_SOCKET)  proxy_close(sock);
  GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  
}


grpc_error_t
grpc_get_problems(char **info)
{
  char *msg=NULL;
  int i, tag;
  SOCKET sock;
  char *str1 = NULL;
  char *str2 = NULL;
  int num_problems;
  gs_problem_t ** problem_list;

  GRPC_FAIL_IF_NOT_INITIALIZED(GRPC_NOT_INITIALIZED);

  if((sock = gs_connect_to_agent()) == INVALID_SOCKET)
    GRPC_RETURN(GRPC_RPC_REFUSED, GRPC_AGENT_NOT_SET);

  if((gs_send_tag(sock, GS_PROT_PROBLEM_LIST) < 0) ||
     (gs_send_string(sock, VERSION) < 0) || 
     (gs_recv_tag(sock, &tag) < 0))
    goto error_communication_failed;
  if(tag != GS_PROT_OK) {
    grpc_errno = GRPC_OTHER_ERROR_CODE;
    if(tag == GS_PROT_VERSION_MISMATCH) {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_VERSION_MISMATCH);
    } else {
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_NO_MINOR_ERROR);
    }
  }

  if(gs_recv_int(sock, &num_problems) < 0) {
    goto error_communication_failed;
  }

  str1 = dstring_sprintf("AGENT: %s [%d problems]\n", agent_resolved->hostname,
    num_problems);

  if(num_problems <= 0) {
    *info = str1;
    GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  }

  problem_list = (gs_problem_t **) CALLOC(num_problems, sizeof(gs_problem_t *));
  if(!problem_list) { 
    GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_OUT_OF_MEMORY);
  }

  for(i=0;i<num_problems;i++) {
    problem_list[i] = (gs_problem_t *) CALLOC(1,sizeof(gs_problem_t));
    if(gs_recv_string(sock, &msg) < 0) {
      goto error_communication_failed;
    }
    if(gs_decode_problem(msg, problem_list[i]) < 0) {
      FREE(msg);
      GRPC_RETURN(GRPC_OTHER_ERROR_CODE, GRPC_SERVER_DECODE);
    }
    FREE(msg);
  }

  for(i=0;i<num_problems;i++) {
    str2 = dstring_sprintf("%s\n", problem_list[i]->name);
    str1 = dstring_append_free(str1, str2);
  }
  
  for(i=0;i<num_problems;i++) 
    FREE(problem_list[i]);
  FREE(problem_list);

  if(sock != INVALID_SOCKET) proxy_close(sock);
  
  *info = str1;
  GRPC_RETURN(GRPC_NO_ERROR, GRPC_NO_MINOR_ERROR);
  
  
 error_communication_failed:
  DBGPRINTF("Client Error: Communication failed\n");
  FREE(msg);
  if(sock != INVALID_SOCKET)  proxy_close(sock);
  GRPC_RETURN(GRPC_COMMUNICATION_FAILED, GRPC_NO_MINOR_ERROR);
  
}

int gs_smart_map_two_cycles(char * mapper_name, char * comm_type){

#ifdef GS_SMART_GRIDSOLVE
  /*
   * list of handles session ids and blocking array of the group of 
   * tasks to be mapped. These structures are initialised
   * when gs_smart_get_handle_list is called which gets the list of
   * handles,session ids and blocking arrays which were initialised
   * in the previous pass through the group of tasks. 
   * This is used to generate the app_pm.
   * 
   */  
  grpc_function_handle_t ** handle_list;
  int nb_tasks;
  int ** blocking_array;
  grpc_sessionid_t ** session_ids;

  /* 
   * application performance model is used by agent 
   * to generate task graph for the group of tasks.
   */
  gs_smart_app_pm * app_pm=NULL;

  /*
   * task graph and network performance model
   * used by mapping heuristic to generate
   * a mapping solution for the group of tasks.
   */
  gs_smart_tg * tg=NULL;
  gs_smart_netpm * netpm=NULL;

  call_type=GS_SMART_TWO_PASS_CALL;  
  group_type=GS_SMART_IMPLICIT_GROUP;  

  if(!mapper_name){
    ERRPRINTF("SMART : Error mapper name is NULL\n");
    smart_phase=GS_SMART_MAP_FAIL;
    return -1;
  }

  /*
   * The first time  gs_smart_map is called 
   * the GS_SMART_TASK_DISCOVERY phase is set, the 
   * mapping parameters parameters are then set. After this we enter the first pass (pass 0)
   * through the scope of the group of tasks, this is done by returning 1.
   * In this pass task discovery is performed.
   */ 
  if((current_pass==0) && (smart_phase==GS_SMART_STANDARD_EXEC)){


    /* Free the previous handle list */
    if(gs_smart_free_handle_list()<0){
      ERRPRINTF("SMART: Error freeing handle list\n");
      return -1;
    } 

    /*** First time this function is called enter here ***/
    smart_phase=GS_SMART_TASK_DISCOVERY;
    nb_mapped_tasks_executed=0;
    total_nb_mapped_tasks=0;

    /*
     *  if mapper name is left empty use default mapper
     */
    if(strcmp(mapper_name, "")==0){    
      if(gs_smart_set_mapper_type("greedy_map")<0){
        ERRPRINTF("SMART: Default mapper not found\n");
        smart_phase=GS_SMART_MAP_FAIL;
        return -1;
      }
    }
    else{
      if(gs_smart_set_mapper_type(mapper_name)<0){
        ERRPRINTF("SMART: %s mapper not found\n", mapper_name);
        smart_phase=GS_SMART_MAP_FAIL;
        return -1;
      }
    }
    if(strcmp(comm_type, "")==0){    
      set_comm_type="enable_remote_comm";
    }
    else{
      if((strcmp(comm_type,"enable_remote_comm")==0) |
         (strcmp(comm_type, "disable_remote_comm")==0) |
         (strcmp(comm_type, "no_dep")==0) |
         (strcmp(comm_type, "server_comm")==0)){
        set_comm_type=comm_type;
      }
      else{
       ERRPRINTF("SMART : Did not recognise the second parameter to GS_SMART_MAP\n");
       set_comm_type="enable_remote_comm";
      }
    }
    return 1;
  }


  /*
   * The second time  gs_smart_map is called 
   * the GS_SMART_MAPPING phase is set.
   * Once this is done the group of tasks is mapped.
   * After this is the second pass (pass 1) 
   * through the scope of the group of tasks. Therefore 1 is returned to enter second pass.
   *  On this pass through the group of tasks, each called handle is executed 
   *  according to the mapping. This phase is called the GS_SMART_EXEC_CALLED_HANDLES phase.
   */ 

  else if((current_pass==0) && (smart_phase==GS_SMART_TASK_DISCOVERY)){
    /*** Second time this function is called enter here ***/
    smart_phase=GS_SMART_MAPPING;
   /*
    * Get the stored list of handles which was initalised on the first pass through
    * the group of tasks.  These are the handles of each task in the group of tasks
    * which are to be mapped.
    */  
   if(gs_smart_get_handle_list(&handle_list,&session_ids, &blocking_array, &nb_tasks)<0){   
      ERRPRINTF("SMART: Error getting handle list\n");
       smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }
    if((!handle_list) || (!session_ids) || (!blocking_array)) return -1;

    app_pm=(gs_smart_app_pm *)calloc(1, sizeof(gs_smart_app_pm));
    if(!app_pm){
      smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }

    /*
     * Generate the application performance model based on the stored
     * list of handles and blocking array. 
     */ 

    if(gs_smart_generate_app_pm(handle_list, blocking_array, nb_tasks, app_pm)<0){
      ERRPRINTF("SMART: Error building application performance model\n");
      smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }

    tg = (gs_smart_tg *)calloc(1,sizeof(gs_smart_tg)); 
    netpm=(gs_smart_netpm *)calloc(1, sizeof(gs_smart_netpm));
 
    if( (!tg) || (!netpm)){
      smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }

    /*
     * Send app_pm to agent. The agent then generates the task graph
     * based on the app_pm and the information in the idl files of each
     * task which is stored on the ageny. 
     *
     * The agent also sends back the a list of servers and the
     * network performance model is generated from this list of servers
     */ 
    if(gs_smart_generate_netpm_and_task_graph(app_pm,  tg, netpm,set_comm_type)){
      ERRPRINTF("SMART: Error generating task graph\n");
      smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }
 
    /* 
     * THESE FUNCTIONS ARE JUST TEMPORARY FOR DEMONSTRATION OF HOW
     * SMARTGRIDSOLVE EXTENSION WORKS 
     */
/*
    if(gs_smart_net_pm_print_to_dotgraph(netpm, "smart_graphs/netpm.dot")<0){
      ERRPRINTF("SMART : Error printing network performance model dotgraph\n");
    }
   
    if(gs_smart_tg_print_to_dotgraph(tg, "smart_graphs/task_graph.dot")<0){
      ERRPRINTF("SMART : Error printing dotgraph\n ");
      return -1;
    }
    system("dot -Tjpg smart_graphs/task_graph.dot -o smart_graphs/task_graph.jpg");
    system("dot -Tjpg smart_graphs/netpm.dot -o smart_graphs/netpm.jpg");
     
*/

    if(gs_smart_get_mapper_type(&mapper_name)<0){
      ERRPRINTF("SMART: Error getting automatic mapper type\n");
      smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }

    /*
     * Call the mapping heuristic (mapper_name) specified by the application programmer.
     * The group of tasks is mapped based  on the task graph and network performance 
     * model.  This function generates a mapping solution 
     * (best_mg in gs_smart_mapping_solution.h) which will be used when
     * executing the group of tasks.
     */      
    if(gs_smart_map_common(mapper_name, tg, netpm, &total_nb_mapped_tasks)<0){
      ERRPRINTF("SMART: Error doing automatic mapping");
      smart_phase=GS_SMART_MAP_FAIL;
      return -1;
    }

    smart_phase=GS_SMART_EXEC_CALLED_HANDLES;
    current_pass=1;
    return 1;
  }

  /*
   * The third time  gs_smart_map is called 
   * we reset the current pass variable and set
   * the smart_phase to GS_SMART_STANDARD_EXEC so that each of the called handles
   * is executed in standard GridSolve.
   * And then we exit out of the scope of the group of tasks which is done by returning 0;
   */ 
  else if((current_pass==1) && (smart_phase==GS_SMART_EXEC_CALLED_HANDLES)){
    current_pass=0;
    smart_phase=GS_SMART_STANDARD_EXEC;

    if(app_pm){
      if(gs_smart_app_pm_free(app_pm)<0){
        ERRPRINTF("SMART : Error freeing application performance model\n");
        smart_phase=GS_SMART_MAP_FAIL;
        return -1;
      }
    }
    if(tg){
      if(gs_smart_tg_free(tg)<0){
        ERRPRINTF("SMART : Error freeing task graph\n");
        smart_phase=GS_SMART_MAP_FAIL;
        return -1;
      }
    }
     if(netpm){
      if(gs_smart_netpm_free(netpm)<0){
        ERRPRINTF("SMART : Error freeing network performance model\n");
        smart_phase=GS_SMART_MAP_FAIL;
        return -1;
      }

    }
    return 0;
  } 
return 0;

#else
 
 if(current_pass==0){
    printf("SMART: Unable to map tasks as GRIDSOLVE was not configured with the SMARTGRIDSOLVE extension\n");
    printf("SMART: The application will execute normally\n");
    current_pass++;
    return 1;
  }
  if(current_pass==1){
    current_pass=0;
    return 0;
  }
#endif
 return 0;
}

/*
 * The macro in gs_smart_clib.h wraps a while loop around 
 * this function. Therefore  this means the function 
 * gs_smart_map_two cycles function and each
 * call in the scope of the parenthis is called twice.
 * The parameter "enable_remote_comm" is passed so
 * that a mapping solution may outline both client
 * and server broadcast communication, as well as 
 * standard client server communication.
 * 
 */

int gs_smart_map_X(char * mapper_name) {

#ifdef GS_SMART_GRIDSOLVE
  if(smart_phase==GS_SMART_EXEC_TASK_FAIL){
    current_pass=0;
    smart_phase=GS_SMART_STANDARD_EXEC;
  }
#endif
  return gs_smart_map_two_cycles(mapper_name, "enable_remote_comm"); 
}

/*
 * The macro in gs_smart_clib.h wraps a while loop around 
 * this function. Therefore  this means the function 
 * gs_smart_map_two_cycles function and each
 * call in the scope of the parenthis is called twice.
 * The parameter "disable_remote_comm" is passed so
 * that a mapping solution may only outline  client
 *  broadcast communication and  
 * standard client server communication.
 */

int gs_smart_map_ft_X(char * mapper_name){
  return gs_smart_map_two_cycles(mapper_name, "disable_remote_comm"); 
}

int grpc_local_X(){


#ifdef GS_SMART_GRIDSOLVE
  if(smart_phase==GS_SMART_STANDARD_EXEC){
    return 1;
  }
  else if(smart_phase==GS_SMART_EXEC_TASK_FAIL){
    return 0;
  }
  else if(smart_phase==GS_SMART_MAP_FAIL){
    return 0;
  }
  else if(smart_phase==GS_SMART_TASK_DISCOVERY){
    return 0;
  }
  else if(smart_phase==GS_SMART_MAPPING){
   return 0; 
  }
  else if(smart_phase==GS_SMART_EXEC_CALLED_HANDLES){
    return 1;
  }
#else
  return 1;
#endif
  return 1;
}




#ifdef GS_SMART_GRIDSOLVE

int gs_smart_get_current_phase(int * cur_phase){
  *cur_phase=smart_phase;
  return 0;
}

#endif