proxy_server.c

Go to the documentation of this file.

/* $Id: proxy_server.c,v 1.28 2008/12/10 19:02:04 seymour Exp $ */
/* $UTK_Copyright: $ */

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

#ifndef _REENTRANT
#define _REENTRANT
#endif

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/errno.h>
#include <memory.h>
#include <signal.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/ioctl.h>

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

#include "symtab.h"
#include "queue.h"
#include "proxy_utils.h"
#include "proxy_server.h"

SYMTABLE *accept_table;
pthread_mutex_t accept_tab_mutex;
QUEUE *removed_components_queue;

void
proxy_print_component_tag(FILE *f, char *prefix, char *ctag, char *suffix)
{
  int id_size, port_size;
  in_port_t server_port;
  CID server_id;

  id_size = sizeof(server_id.id);
  port_size = sizeof(server_port);

  memcpy(&server_port, ctag + id_size, port_size);

  if(prefix)
    fprintf(f, "%s ", prefix);
  proxy_print_componentID(f, ctag);
  fprintf(f, "/%d", ntohs(server_port));
  if(suffix)
    fprintf(f, " %s", suffix);
  fprintf(f, "\n");
}

int 
proxy_daemon_init(char *root, char *logfile) {
  pid_t pid;
  int fd;

  if(!root || !logfile)
    return -1;

  pid = fork();

  if(pid < 0)
    return -1;

  if(pid != 0) /* parent exits */
    exit(0);

  setsid();   /* become session leader */

  pid = fork();

  if(pid < 0)
    return -1;

  if(pid != 0) /* 1st child exits */
    exit(0);

  chdir(root);
  umask(0);

  /* Remap stdin to /dev/null */
  fd = open("/dev/null", O_RDWR|O_CREAT, 0644);
  if(fd < 0) {
    perror("open()");
    return -1;
  }
  close(0);
  if(dup2(fd, 0) < 0) {
    fprintf(stderr,"dup2 failed");
    perror("dup2()");
    return -1;
  }
  close(fd);

  fprintf(stderr, "\n"
    "==============================================================\n"
    "|       -- Giving up terminal control! --\n|\n"
    "| All further terminal output will be sent to the log file:\n"
    "|\t\"%s\"\n"
    "=============================================================="
    "\n\n", logfile);

  /* Open logfile */
  remove(logfile);
  fd = open(logfile, O_RDWR|O_CREAT, 0644);
  if(fd < 0){
    fprintf(stderr,"dup2 failed");
    perror("open()");
    return -1;
  }

  /* Remap stdout and stderr to logfile */
  close(1);
  if(dup2(fd, 1) < 0){
    fprintf(stderr,"dup2 failed");
    perror("dup2()");
    return -1;
  }
  close(2);
  if(dup2(fd, 2) < 0){
    fprintf(stderr,"dup2 failed");
    perror("dup2()");
    return -1;
  }
  close(fd);

  return 0;
}

int 
proxy_block_sigpipe()
{
  sigset_t sigvec;

  if (sigemptyset(&sigvec) < 0 ||
      sigaddset(&sigvec, SIGPIPE) < 0 || 
      pthread_sigmask(SIG_BLOCK, &sigvec, NULL) != 0) 
  {
    fprintf(stderr, "Unable to set thread signal mask.  Aborting\n");
    return -1;
  }

  return 0;
}

int proxy_krb5_enabled = FALSE;

static int 
getenv_int(char* name, int defval)
{
  char *envstr = NULL;
  long int longval = -1;
  char *endptr;
  extern int errno;

  if (name == NULL) return defval;

  /* Env variable does not exist */
  if ((envstr = getenv(name)) == NULL) return defval;

  /* Convert to long, checking for errors */
  longval = strtol(envstr, &endptr, 10);
  if ((errno == ERANGE) || (longval==0 && endptr==envstr))
    return defval;

  return (int)longval;
}

int
proxy_server_parse_cmd_line(int argc, char **argv, char **logfile, 
  int *daemon, int *kerberos)
{
  int c;

  *logfile = NULL;
  *daemon = 1;
  *kerberos = 0;

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

#define PROXY_SERVER_USAGE_STR "Usage: proxy_server [-l logfile] [-c] [-k]"

  while((c = getopt(argc,argv,"kcl:")) != EOF) {
    switch(c) {
      case 'l':
        *logfile = strdup(optarg);
        break;
      case 'c':
        *daemon = 0;
        break;
      case 'k':
        *kerberos = 1;
        break;
      case '?':
        return -1;
        break;
      default:
        fprintf(stderr,"Bad arg: '%c'.\n",c);
        return -1;
    }
  }

  return 0;
}

int 
main(int argc, char *argv[])
{
  int listen_sock, daemon;
  int proxy_listen_port = -1;
  socklen_t clilen;
  struct sockaddr_in tcp_serv, cliaddr;
  void *proxy_generic_conn_handler(void *);
  void *proxy_purge_old_components(void *);
  char *logfile;
  pthread_t tid;

  /* Parse command line args */
  if(proxy_server_parse_cmd_line(argc, argv, &logfile, &daemon, 
       &proxy_krb5_enabled) < 0) 
  {
    fprintf(stderr, "%s\n", PROXY_SERVER_USAGE_STR);
    exit(EXIT_FAILURE);
  }

#ifdef KERBEROS5
  if (proxy_check_krb5_envvars() != 0)
    exit(EXIT_FAILURE);
#else
  if(proxy_krb5_enabled) {
    fprintf(stderr, "Warning!  Kerberos was not enabled during compilation ");
    fprintf(stderr, "(-k ignored).\n\n");
  }
#endif

  /* if logfile wasn't specified on the command line, use default */
  if(!logfile)
    logfile = PROXY_LOGFILE;

  /* Make current process into a daemon */
  if(daemon && proxy_daemon_init(".", logfile) < 0) {
    fprintf(stderr, "Failed to start proxy server as a daemon.\n");
    exit(EXIT_FAILURE);
  }

  if (proxy_block_sigpipe() < 0) {
    fprintf(stderr, "Couldn't block SIGPIPE.  Aborting.\n");
    exit(EXIT_FAILURE);
  }

  removed_components_queue = new_queue();
  if (!removed_components_queue) {
    fprintf(stderr, "Bad news!  out of memory allocating queue\n");
    exit(EXIT_FAILURE);
  }

  /* create a hash table containing the components that have connected */
  accept_table = new_symtable(PROXY_ACCEPT_TABLE_SIZE);

  if (!accept_table) {
    fprintf(stderr, "Bad news!  out of memory allocating table\n");
    exit(EXIT_FAILURE);
  }

  if (pthread_mutex_init(&accept_tab_mutex, NULL)) {
    fprintf(stderr, "Bad news!  cannot create mutex\n");
    exit(EXIT_FAILURE);
  }

  proxy_listen_port = getenv_int("PROXY_LISTEN_PORT", PROXY_LISTEN_PORT_DEFAULT);
  if (proxy_init_sock(&listen_sock, &tcp_serv, "tcp", FALSE, proxy_listen_port) < 0) {
    fprintf(stderr, "Failed to create socket\n");
    perror("proxy_init_sock");
    exit(EXIT_FAILURE);
  }

  if(pthread_create(&tid, NULL, &proxy_purge_old_components, NULL)) {
    fprintf(stderr, "Warning: error creating new thread.\n");
    perror("pthread_create");
  }

  if(pthread_detach(tid)) 
    fprintf(stderr,"Warning: could not detach thread after creation.\n");

  printf("proxy IP %u\n", proxy_get_my_ipaddr());
  printf("proxy listening on port %d\n", ntohs(tcp_serv.sin_port));

  if (listen(listen_sock, 50) < 0) {
    perror("listen");
    exit(EXIT_FAILURE);
  }

  for (;;) {
    int *connfd;

    clilen = sizeof(cliaddr);

    /* malloc to avoid race between accept and new thread */
    connfd = (int *) malloc(sizeof(int));

    if (!connfd) {
      fprintf(stderr, "Cannot malloc.  going to sleep for a while.\n");
      sleep(3);
      continue;
    }

    *connfd = accept(listen_sock, (struct sockaddr *) &cliaddr, &clilen);

    if (*connfd < 0) {
      free(connfd);
      perror("accept");
      continue;
    }

    printf("new client connected!\n");

    if (proxy_auth(*connfd) < 0) {
      fprintf(stderr, "Warning: failed to authenticate client.\n");
      close(*connfd);
      continue;
    }

    if (pthread_create(&tid, NULL, &proxy_generic_conn_handler, connfd)) {
      fprintf(stderr, "Warning: error creating new thread.\n");
      close(*connfd);
      free(connfd);
      continue;
    }

    if(pthread_detach(tid)) 
      fprintf(stderr,"Warning: could not detach thread after creation.\n");
  }
}

int 
proxy_auth(int s)
{
  proxy_tag_t response;

#ifdef KERBEROS5
  if (proxy_krb5_enabled) {
    response = PROXY_KRB5_AUTH_REQUIRED;
    if (write(s, &response, sizeof(response)) < 0)
      return -1;

    /* get credentials from client */
    if (proxy_recv_krb5_credentials(s) < 0) {
      /* 
       * if we didn't like the credentials; tell client that the
       * authentication failed, and this will be the error code
       * associated with the problem request.  otherwise, fall
       * through to other error checking below
       */
      response = PROXY_AUTH_FAILED;

      if (write(s, &response, sizeof(response)) == -1)
    return -1;

      return -1;
    }

    response = PROXY_AUTH_ACCEPTED;
  } 
  else
    response = PROXY_AUTH_ACCEPTED;
#else
  response = PROXY_AUTH_ACCEPTED;
#endif

  if (write(s, &response, sizeof(response)) < 0)
    return -1;

  return 0;
}

void
dump_component(char *key, void *var)
{
  COMPONENT *item = (COMPONENT *)var;

  proxy_print_component_tag(stdout, "COMPONENT key =", key, NULL);

  printf("\tID = ");
  proxy_print_componentID(stdout, item->id.id);
  printf("\n");
  printf("\tport = %d\n", ntohs(item->port));
  printf("\tsockfd = %d\n", item->sockfd);

  return;
}

void *
proxy_purge_old_components(void *arg)
{
  COMPONENT *component;

  for(;;) {
    sleep(PROXY_PURGE_FREQ);

    if (pthread_mutex_lock(&accept_tab_mutex) == 0) {
      printf("\n");
      printf("--------- ACCEPT TABLE [%d items] ---------\n", accept_table->num_items);
      hash_dump(accept_table, dump_component);
      printf("------------------------------------------\n");
      printf("\n");

      pthread_mutex_unlock(&accept_tab_mutex);
    }

    if(pthread_mutex_lock(removed_components_queue->mutex) == 0) {

      /* free every component in the queue */

      while((component = (COMPONENT *)dequeue(removed_components_queue))) {
        destroy_queue(component->msg_queue);
        free(component);
      }

      pthread_mutex_unlock(removed_components_queue->mutex);
    } else {
      fprintf(stderr, "Warning: couldn't lock removed components queue.\n");
    }
  }

  return NULL;
}

void *
proxy_generic_conn_handler(void *arg)
{
  int s, n;
  proxy_tag_t tag;

  s = *((int *) arg);
  free(arg);

  n = proxy_tread(s, (void *) &tag, 1, PROXY_TIMEOUT_DEFAULT);

  if (n <= 0) {
    fprintf(stderr, "Warning: error reading tag.  Closing connection.\n");
    close(s);
    return NULL;
  }

  printf("tag = %d\n", tag);

  switch (tag) {

    case PROXY_CONTROL_CONNECTION:
      proxy_handle_control_connection(s);
      break;

    case PROXY_CONNECT:
      proxy_handle_connection_request(s);
      break;

    /* this case is for testing purposes.. remove later. */
    case 'Z':
      fprintf(stderr, "TERMINATING...\n");
      exit(EXIT_SUCCESS);
      break;

    default:
      fprintf(stderr, "Warning: unknown tag %d\n", tag);
      close(s);
      break;
  }

  return NULL;
}

COMPONENT *
proxy_new_component()
{
  COMPONENT *new;

  new = (COMPONENT *) malloc(sizeof(COMPONENT));

  if (!new)
    return NULL;

  new->id.id[0] = '\0';
  new->port = 0;
  new->sockfd = 0;
  new->to_be_freed = FALSE;
  new->msg_queue = NULL;

  return new;
}

COMPONENT *
proxy_get_request_header(int connfd, CID * client_id)
{
  int id_size, port_size, to;
  COMPONENT *component;
  in_port_t dest_port;
  HASHNODE *ht;
  CID dest_id;
  char component_tag[sizeof(CID) + sizeof(dest_port) + 1];

  to = PROXY_TIMEOUT_DEFAULT;
  id_size = sizeof(client_id->id);
  port_size = sizeof(dest_port);

  if (proxy_tread(connfd, (void *) client_id->id, id_size, to) <= 0 ||
      proxy_tread(connfd, (void *) dest_id.id, id_size, to) <= 0 ||
      proxy_tread(connfd, (void *) &dest_port, port_size, to) <= 0) {
    fprintf(stderr, "Error reading IDs.  Aborting connection.\n");
    return NULL;
  }

  /* component_tag is composed of CID/port */
  memcpy(component_tag, dest_id.id, id_size);
  memcpy(component_tag + id_size, &dest_port, port_size);
  *(component_tag + id_size + port_size) = '\0';

  proxy_print_component_tag(stdout, "CONNECT REQ dest = ", 
    component_tag, NULL);

  if (pthread_mutex_lock(&accept_tab_mutex) == 0) {
    ht = hash_lookup(accept_table, (void *) component_tag);

    if (!ht) {
      proxy_tag_t response = PROXY_CONNECTION_REFUSED;

      fprintf(stderr, "Component entry not found!\n");
      pthread_mutex_unlock(&accept_tab_mutex);
      write(connfd, &response, sizeof(response));
      return NULL;
    }

    component = (COMPONENT *) (ht->item);

    printf("found entry.. socket desc = %d\n", component->sockfd);

    pthread_mutex_unlock(&accept_tab_mutex);
  } else {
    fprintf(stderr, "proxy_get_request_header(): Error locking.\n");
    return NULL;
  }

  return component;
}

int 
proxy_send_cid_to_control_conn(int s, COMPONENT * c, CID client_id, 
  in_port_t port)
{
  /* must enqueue request to control connection and signal notempty */

  printf("waiting for msg queue lock\n");

  if (pthread_mutex_lock(c->msg_queue->mutex) == 0) {
    COMPONENT *client_component;

    client_component = proxy_new_component();

    /* make sure this component isn't destined to be freed */
    if (c->to_be_freed || !client_component) {
      proxy_tag_t response = PROXY_CONNECTION_REFUSED;

      if (client_component)
    free(client_component);
      pthread_mutex_unlock(c->msg_queue->mutex);
      write(s, &response, sizeof(response));
      return -1;
    }

    memcpy(client_component->id.id, client_id.id, sizeof(client_id.id));
    client_component->port = port;

    printf("putting component in queue\n");

    enqueue(c->msg_queue, (void *) client_component);
    pthread_cond_signal(c->msg_queue->notEmpty);
    pthread_mutex_unlock(c->msg_queue->mutex);
  } else {
    fprintf(stderr, "Error locking.\n");
    return -1;
  }

  return 0;
}

int 
proxy_wait_for_server_conn(int connfd, int sock2)
{
  int maxfd, nready;
  fd_set allset, rset;
  struct timeval tv;
  int newfd = -1;
  struct sockaddr_in cliaddr;
  socklen_t clilen;

  /* now listen for incoming connection from server behind nat */

  maxfd = sock2;
  FD_ZERO(&allset);
  FD_SET(sock2, &allset);

  rset = allset;
  tv.tv_sec = 15;
  tv.tv_usec = 0;

  printf("waiting for 2nd connection\n");
  nready = select(maxfd + 1, &rset, NULL, NULL, &tv);

  if (nready <= 0) {
    proxy_tag_t response = PROXY_CONNECTION_REFUSED;

    printf("select error %d (0 means timed out)\n", nready);
    write(connfd, &response, sizeof(response));
    return -1;
  }

  if (FD_ISSET(sock2, &rset)) {
    clilen = sizeof(cliaddr);

    printf("something happened on the second listening socket!\n");

    newfd = accept(sock2, (struct sockaddr *) &cliaddr, &clilen);

    printf("after accept, new fd = %d\n", newfd);

    if (proxy_auth(newfd) < 0) {
      fprintf(stderr, "Warning: failed to authenticate client.\n");
      close(newfd);
    }
  }

  return newfd;
}

int 
proxy_handle_data_transfer(int client_fd, int serv_fd)
{
  int maxfd, nready;
  fd_set allset, rset;

  maxfd = serv_fd > client_fd ? serv_fd : client_fd;

  FD_ZERO(&allset);
  FD_SET(serv_fd, &allset);
  FD_SET(client_fd, &allset);

  for (;;) {
    int n;
    char buf[PROXY_XFER_CHUNKSIZE];

    rset = allset;

    nready = select(maxfd + 1, &rset, NULL, NULL, NULL);

    if (nready < 0) {

      if (errno == EINTR)
    continue;

      return -1;
    }

    if (nready == 0)
      continue;

    if (FD_ISSET(client_fd, &rset)) {
      n = proxy_read_timeout(client_fd, buf, PROXY_XFER_CHUNKSIZE, 
        PROXY_TIMEOUT_DEFAULT);

      if (n <= 0) {
    printf("seems the requesting host (client) broke connection\n");
    break;
      }

      n = write(serv_fd, buf, n);
    } else if (FD_ISSET(serv_fd, &rset)) {
      n = proxy_read_timeout(serv_fd, buf, PROXY_XFER_CHUNKSIZE, 
        PROXY_TIMEOUT_DEFAULT);

      if (n <= 0) {
    printf("seems the proxied host (server) broke connection\n");
    break;
      }

      n = write(client_fd, buf, n);
    }
  }

  return 0;
}

int 
proxy_check_client_match(int data_conn_fd, int client_fd, 
  in_port_t port, CID client_id)
{
  proxy_tag_t response = PROXY_CONNECTION_ACCEPTED;
  in_port_t new_port;
  CID new_cid;
  int n;

  printf("sending connection accepted to client\n");
  n = write(client_fd, &response, sizeof(response));

  if (n <= 0) {
    fprintf(stderr, "Error writing response to client\n");
    return -1;
  }

  printf("getting response\n");

  n = proxy_tread(data_conn_fd, (char *)&response, sizeof(response), 
    PROXY_TIMEOUT_DEFAULT);
  if (n <= 0) {
    fprintf(stderr, "Error getting response from client\n");
    return -1;
  }

  if (response != PROXY_CONNECT_REPLY) {
    fprintf(stderr, "Didn't get the expected tag!!\n");
    return -1;
  }

  printf("getting cid\n");
  n = proxy_tread(data_conn_fd, (char *)&(new_cid.id), sizeof(new_cid.id), 
    PROXY_TIMEOUT_DEFAULT);
  if (n <= 0) {
    fprintf(stderr, "Error getting ID from client\n");
    return -1;
  }

  printf("getting port\n");
  n = proxy_tread(data_conn_fd, (char *)&new_port, sizeof(new_port), 
    PROXY_TIMEOUT_DEFAULT);
  if (n <= 0) {
    fprintf(stderr, "Error getting port from client\n");
    return -1;
  }

  /* check for a bad match */
  if (memcmp(client_id.id, new_cid.id, sizeof(new_cid.id))
      || (port != new_port)) {
    fprintf(stderr, "BAD MATCH!!\n");
    response = PROXY_CONNECTION_REFUSED;
    write(data_conn_fd, &response, sizeof(response));

    return -1;
  }

  printf("IDs match, now sending connection accepted msg to server.\n");

  /* found a match.. send connection accepted to server */
  response = PROXY_CONNECTION_ACCEPTED;
  n = write(data_conn_fd, &response, sizeof(response));
  if (n <= 0) {
    fprintf(stderr, "Error writing response to client\n");
    return -1;
  }

  return 0;
}

int 
proxy_handle_connection_request(int connfd)
{
  int new_incoming_sock, data_conn_fd;
  struct sockaddr_in tcp_serv;
  CID client_id;
  COMPONENT *component;

  if (proxy_block_sigpipe() < 0)
    return -1;

  component = proxy_get_request_header(connfd, &client_id);

  if (!component) {
    fprintf(stderr, "Error getting component header.  Aborting connection.\n");
    close(connfd);
    return -1;
  }

  if (proxy_init_sock(&new_incoming_sock, &tcp_serv, "tcp", FALSE, 0) < 0) {
    fprintf(stderr, "Error creating new incoming socket.\n");
    close(connfd);
    return -1;
  }

  printf("new port  -->  %d\n", ntohs(tcp_serv.sin_port));

  if (listen(new_incoming_sock, 5) < 0) {
    proxy_tag_t response = PROXY_CONNECTION_REFUSED;

    write(connfd, &response, sizeof(response));
    close(connfd);
    return -1;
  }

  if (proxy_send_cid_to_control_conn(connfd, component, client_id, tcp_serv.sin_port) < 0) {
    fprintf(stderr, "Error queueing control id.\n");
    close(connfd);
    return -1;
  }

  data_conn_fd = proxy_wait_for_server_conn(connfd, new_incoming_sock);

  if (data_conn_fd > 0) {
    if (proxy_check_client_match(data_conn_fd, connfd, tcp_serv.sin_port, client_id) < 0) {
      printf("bad match\n");
    } else
      proxy_handle_data_transfer(connfd, data_conn_fd);

    close(data_conn_fd);
  }

  close(connfd);
  close(new_incoming_sock);

  return 0;
}

COMPONENT *
proxy_init_server_component(int connfd, char *component_tag)
{
  COMPONENT *component;
  CID server_id;
  in_port_t server_port;
  int id_size, to;
  int port_size;

  to = PROXY_TIMEOUT_DEFAULT;
  id_size = sizeof(server_id.id);
  port_size = sizeof(server_port);

  component = proxy_new_component();
  if (component)
    component->msg_queue = new_queue();

  if (!component || !component->msg_queue) {
    fprintf(stderr, "Cannot malloc, aborting connection\n");
    if (component) {
      if (component->msg_queue)
    destroy_queue(component->msg_queue);
      free(component);
    }
    return NULL;
  }

  printf("id size = %d, port size = %d\n", id_size, port_size);

  if (proxy_tread(connfd, (void *) server_id.id, id_size, to) < 0 ||
      proxy_tread(connfd, (void *) &server_port, port_size, to) < 0) {
    if (component->msg_queue)
      destroy_queue(component->msg_queue);
    free(component);
    return NULL;
  }

  printf("port = %d\n", ntohs(server_port));

  /* component_tag is composed of CID/port */
  memcpy(component_tag, server_id.id, id_size);
  memcpy(component_tag + id_size, &server_port, port_size);
  *(component_tag + id_size + port_size) = '\0';

  proxy_print_component_tag(stdout, "component_tag =", component_tag, NULL);

  component->id = server_id;
  component->port = server_port;
  component->sockfd = connfd;

  if (pthread_mutex_lock(&accept_tab_mutex) == 0) {
    HASHNODE *ht;

    ht = hash_lookup(accept_table, (void *) component_tag);

    if(ht) {
      proxy_print_component_tag(stderr, "Component ", component_tag,
        " already exists in the table!\n");
      pthread_mutex_unlock(&accept_tab_mutex);
      return NULL;
    }

    hash_insert(accept_table, (void *) component, (void *) component_tag);
    pthread_mutex_unlock(&accept_tab_mutex);
  } else {
    if (component->msg_queue)
      destroy_queue(component->msg_queue);
    free(component);
    return NULL;
  }

  return component;
}

COMPONENT *
proxy_wait_for_queue_message(int connfd, COMPONENT * component)
{
  COMPONENT *item;
  int keepalive_timer;

  keepalive_timer = PROXY_KEEPALIVE_FREQ;

  if (pthread_mutex_lock(component->msg_queue->mutex) == 0) {
    while (is_empty(component->msg_queue)) {
      struct timespec timeout;
      struct timeval tv;
      int rv;

      gettimeofday(&tv, NULL);

      timeout.tv_sec = tv.tv_sec + PROXY_MSG_QUEUE_FREQ;
      timeout.tv_nsec = 0;

      /* wait until a message is placed in the queue or until the timeout elapses. */
      rv = pthread_cond_timedwait(component->msg_queue->notEmpty,
                  component->msg_queue->mutex, &timeout);

      keepalive_timer -= PROXY_MSG_QUEUE_FREQ;

      if (rv != 0) {
        int remove_srv = 0;

    /* if we timed out, check if the server is still there. */

        if(keepalive_timer <= 0) {
          keepalive_timer = PROXY_KEEPALIVE_FREQ;
      remove_srv = (proxy_send_keepalive(connfd) < 0);
        }
        else {
      remove_srv = (proxy_is_something_on_socket(connfd) == 0);
        }

    if(remove_srv) {
      char component_tag[sizeof(CID) + sizeof(in_port_t) + 1];
      int id_size = sizeof(component->id);
      int port_size = sizeof(component->port);

      printf("server must have gone away!\n");

      /* component_tag is composed of CID/port */
      memcpy(component_tag, &(component->id), id_size);
      memcpy(component_tag + id_size, &(component->port), port_size);
      *(component_tag + id_size + port_size) = '\0';

      proxy_delete_server_tag(component_tag, component);
          close(component->sockfd);

      return NULL;
    }
      }
    }
    item = (COMPONENT *) dequeue(component->msg_queue);
    pthread_mutex_unlock(component->msg_queue->mutex);
  } else
    return NULL;

  return item;
}

int
proxy_send_conn_request_to_server(int connfd, char *component_tag,
                COMPONENT * component, COMPONENT * item)
{
  proxy_tag_t response = PROXY_CONNECT_REQUEST;

  printf("going to write to the server now (port = %d)...\n", ntohs(item->port));
  if (write(connfd, &response, sizeof(response)) < 0 ||
      write(connfd, item->id.id, sizeof(item->id.id)) < 0 ||
      write(connfd, &item->port, sizeof(item->port)) < 0) {
    fprintf(stderr, "Can't write to server, aborting.\n");

    proxy_delete_server_tag(component_tag, component);

    return -1;
  }

  return 0;
}

int 
proxy_delete_server_tag(char *component_tag, COMPONENT * component)
{

  proxy_print_component_tag(stdout, "going to attempt to delete component: ",
    component_tag, NULL);

  if (pthread_mutex_lock(&accept_tab_mutex) == 0) {
    HASHNODE *ht;

    printf("going to delete component_tag now...\n");
    ht = hash_delete(accept_table, (void *) component_tag);

    if (!ht)
      printf("warning: expected to find component in hash table\n");
    else if (component != (COMPONENT *) (ht->item))
      printf("warning: wrong component! \n");

    if(ht) free(ht);
    pthread_mutex_unlock(&accept_tab_mutex);
  } else
    return -1;

  component->to_be_freed = TRUE;

  /* put component in queue to be freed later */

  if (pthread_mutex_lock(removed_components_queue->mutex) == 0) {

    printf("putting component in removal queue\n");

    enqueue(removed_components_queue, (void *) component);
    pthread_mutex_unlock(removed_components_queue->mutex);
  } else {
    fprintf(stderr, "Warning: couldn't lock removed components queue.\n");
  }

  return 0;
}

int 
proxy_handle_control_connection(int connfd)
{
  COMPONENT *component;
  char component_tag[sizeof(CID) + sizeof(in_port_t) + 1];

  if (proxy_block_sigpipe() < 0)
    return -1;

  printf("i am a thread handling a control connection %d\n", connfd);

  component = proxy_init_server_component(connfd, component_tag);
  if (!component) {
    fprintf(stderr, "Error intializing server component.\n");
    close(connfd);
    return -1;
  }

  for (;;) {
    COMPONENT *item;

    item = proxy_wait_for_queue_message(connfd, component);

    if (!item) {
      close(connfd);
      return -1;
    }

    if (proxy_send_conn_request_to_server(connfd, component_tag, component, item) < 0) {
      free(item);
      close(connfd);
      return -1;
    }

    /* free the queue item.  the msg_queue field isn't initialized for queue items, so
     * there's no need to dispose of it here. */
    free(item);
  }

  return 0;
}

int 
proxy_init_sock(int *sfd, struct sockaddr_in *serv, char *proto, int nb, int port)
{
  int r, flag = 1;
  socklen_t namelen;
  char istcp = strcmp(proto, "udp");

  if ((*sfd = socket(AF_INET, istcp ? SOCK_STREAM : SOCK_DGRAM, 0)) < 0)
    return *sfd;

  if ((r = setsockopt(*sfd, SOL_SOCKET, SO_REUSEADDR, (char *) &flag, sizeof(flag))) < 0)
    return r;

  if(istcp)
    setsockopt(*sfd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int));

  serv->sin_family = AF_INET;
  serv->sin_addr.s_addr = htonl(INADDR_ANY);
  serv->sin_port = htons(port);

  if ((r = bind(*sfd, (struct sockaddr *) serv, sizeof(*serv))) < 0)
    return r;

  namelen = sizeof(*serv);
  if ((r = getsockname(*sfd, (struct sockaddr *) serv, &namelen)) < 0)
    return r;

  if (nb)
    if ((r = fcntl(*sfd, F_SETFL, O_NDELAY)) < 0)   /* make socket non-blocking */
      return r;

  return 0;
}

#ifdef KERBEROS5
int 
proxy_check_krb5_envvars()
{

  if (getenv("GRIDSOLVE_KEYTAB") == NULL) {
    fprintf(stderr, "Environment variable GRIDSOLVE_KEYTAB not defined\n");
    return -1;
  }

  if (getenv("GRIDSOLVE_USERS") == NULL) {
    fprintf(stderr, "Environment variable GRIDSOLVE_USERS not defined\n");
    return -1;
  }

  return 0;
}
#endif