tor/src/or/cpuworker.c

/* Copyright (c) 2003-2004, Roger Dingledine.
 * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
 * Copyright (c) 2007-2012, The Tor Project, Inc. */
/* See LICENSE for licensing information */

/**
 * \file cpuworker.c
 * \brief Implements a farm of 'CPU worker' processes to perform
 * CPU-intensive tasks in another thread or process, to not
 * interrupt the main thread.
 *
 * Right now, we only use this for processing onionskins.
 **/

#include "or.h"
#include "buffers.h"
#include "channel.h"
#include "channeltls.h"
#include "circuitbuild.h"
#include "circuitlist.h"
#include "config.h"
#include "connection.h"
#include "cpuworker.h"
#include "main.h"
#include "onion.h"
#include "router.h"

/** The maximum number of cpuworker processes we will keep around. */
#define MAX_CPUWORKERS 16
/** The minimum number of cpuworker processes we will keep around. */
#define MIN_CPUWORKERS 1

/** The tag specifies which circuit this onionskin was from. */
#define TAG_LEN 10

/** How many cpuworkers we have running right now. */
static int num_cpuworkers=0;
/** How many of the running cpuworkers have an assigned task right now. */
static int num_cpuworkers_busy=0;
/** We need to spawn new cpuworkers whenever we rotate the onion keys
 * on platforms where execution contexts==processes.  This variable stores
 * the last time we got a key rotation event. */
static time_t last_rotation_time=0;

static void cpuworker_main(void *data) ATTR_NORETURN;
static int spawn_cpuworker(void);
static void spawn_enough_cpuworkers(void);
static void process_pending_task(connection_t *cpuworker);

/** Initialize the cpuworker subsystem.
 */
void
cpu_init(void)
{
  cpuworkers_rotate();
}

/** Called when we're done sending a request to a cpuworker. */
int
connection_cpu_finished_flushing(connection_t *conn)
{
  tor_assert(conn);
  tor_assert(conn->type == CONN_TYPE_CPUWORKER);
  return 0;
}

/** Pack global_id and circ_id; set *tag to the result. (See note on
 * cpuworker_main for wire format.) */
static void
tag_pack(uint8_t *tag, uint64_t chan_id, circid_t circ_id)
{
  /*XXXX RETHINK THIS WHOLE MESS !!!! !NM NM NM NM*/
  /*XXXX DOUBLEPLUSTHIS!!!! AS AS AS AS*/
  set_uint64(tag, chan_id);
  set_uint16(tag+8, circ_id);
}

/** Unpack <b>tag</b> into addr, port, and circ_id.
 */
static void
tag_unpack(const uint8_t *tag, uint64_t *chan_id, circid_t *circ_id)
{
  *chan_id = get_uint64(tag);
  *circ_id = get_uint16(tag+8);
}

/** DOCDOC */
#define CPUWORKER_REQUEST_MAGIC 0xda4afeed
#define CPUWORKER_REPLY_MAGIC 0x5eedf00d

/**DOCDOC*/
typedef struct cpuworker_request_t {
  uint32_t magic;
  /** Opaque tag to identify the job */
  uint8_t tag[TAG_LEN];
  uint8_t task;

  create_cell_t create_cell;
  /* Turn the above into a tagged union if needed. */
} cpuworker_request_t;

/**DOCDOC*/
typedef struct cpuworker_reply_t {
  uint32_t magic;
  uint8_t tag[TAG_LEN];
  uint8_t success;

  created_cell_t created_cell;
  uint8_t keys[CPATH_KEY_MATERIAL_LEN];
  uint8_t rend_auth_material[DIGEST_LEN];
} cpuworker_reply_t;

/** Called when the onion key has changed and we need to spawn new
 * cpuworkers.  Close all currently idle cpuworkers, and mark the last
 * rotation time as now.
 */
void
cpuworkers_rotate(void)
{
  connection_t *cpuworker;
  while ((cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
                                                   CPUWORKER_STATE_IDLE))) {
    connection_mark_for_close(cpuworker);
    --num_cpuworkers;
  }
  last_rotation_time = time(NULL);
  if (server_mode(get_options()))
    spawn_enough_cpuworkers();
}

/** If the cpuworker closes the connection,
 * mark it as closed and spawn a new one as needed. */
int
connection_cpu_reached_eof(connection_t *conn)
{
  log_warn(LD_GENERAL,"Read eof. CPU worker died unexpectedly.");
  if (conn->state != CPUWORKER_STATE_IDLE) {
    /* the circ associated with this cpuworker will have to wait until
     * it gets culled in run_connection_housekeeping(), since we have
     * no way to find out which circ it was. */
    log_warn(LD_GENERAL,"...and it left a circuit queued; abandoning circ.");
    num_cpuworkers_busy--;
  }
  num_cpuworkers--;
  spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up
                                spinning. */
  connection_mark_for_close(conn);
  return 0;
}

/** Called when we get data from a cpuworker.  If the answer is not complete,
 * wait for a complete answer. If the answer is complete,
 * process it as appropriate.
 */
int
connection_cpu_process_inbuf(connection_t *conn)
{
  uint64_t chan_id;
  circid_t circ_id;
  channel_t *p_chan = NULL;
  circuit_t *circ;

  tor_assert(conn);
  tor_assert(conn->type == CONN_TYPE_CPUWORKER);

  if (!connection_get_inbuf_len(conn))
    return 0;

  if (conn->state == CPUWORKER_STATE_BUSY_ONION) {
    cpuworker_reply_t rpl;
    if (connection_get_inbuf_len(conn) < sizeof(cpuworker_reply_t))
      return 0; /* not yet */
    tor_assert(connection_get_inbuf_len(conn) == sizeof(cpuworker_reply_t));

    connection_fetch_from_buf((void*)&rpl,sizeof(cpuworker_reply_t),conn);

    tor_assert(rpl.magic == CPUWORKER_REPLY_MAGIC);
    /* parse out the circ it was talking about */
    tag_unpack(rpl.tag, &chan_id, &circ_id);
    circ = NULL;
    log_debug(LD_OR,
              "Unpacking cpuworker reply, chan_id is " U64_FORMAT
              ", circ_id is %d",
              U64_PRINTF_ARG(chan_id), circ_id);
    p_chan = channel_find_by_global_id(chan_id);

    if (p_chan)
      circ = circuit_get_by_circid_channel(circ_id, p_chan);

    if (rpl.success == 0) {
      log_debug(LD_OR,
                "decoding onionskin failed. "
                "(Old key or bad software.) Closing.");
      if (circ)
        circuit_mark_for_close(circ, END_CIRC_REASON_TORPROTOCOL);
      goto done_processing;
    }
    if (!circ) {
      /* This happens because somebody sends us a destroy cell and the
       * circuit goes away, while the cpuworker is working. This is also
       * why our tag doesn't include a pointer to the circ, because we'd
       * never know if it's still valid.
       */
      log_debug(LD_OR,"processed onion for a circ that's gone. Dropping.");
      goto done_processing;
    }
    tor_assert(! CIRCUIT_IS_ORIGIN(circ));
    if (onionskin_answer(TO_OR_CIRCUIT(circ),
                         rpl.created_cell.cell_type,
                         (const char*)rpl.created_cell.reply,
                         rpl.created_cell.handshake_len,
                         (const char*)rpl.keys,
                         rpl.rend_auth_material) < 0) {
      log_warn(LD_OR,"onionskin_answer failed. Closing.");
      circuit_mark_for_close(circ, END_CIRC_REASON_INTERNAL);
      goto done_processing;
    }
    log_debug(LD_OR,"onionskin_answer succeeded. Yay.");
  } else {
    tor_assert(0); /* don't ask me to do handshakes yet */
  }

 done_processing:
  conn->state = CPUWORKER_STATE_IDLE;
  num_cpuworkers_busy--;
  if (conn->timestamp_created < last_rotation_time) {
    connection_mark_for_close(conn);
    num_cpuworkers--;
    spawn_enough_cpuworkers();
  } else {
    process_pending_task(conn);
  }
  return 0;
}

/** Implement a cpuworker.  'data' is an fdarray as returned by socketpair.
 * Read and writes from fdarray[1].  Reads requests, writes answers.
 *
 *   Request format:
 *          cpuworker_request_t.
 *   Response format:
 *          cpuworker_reply_t
 */
static void
cpuworker_main(void *data)
{
  /* For talking to the parent thread/process */
  tor_socket_t *fdarray = data;
  tor_socket_t fd;

  /* variables for onion processing */
  server_onion_keys_t onion_keys;
  cpuworker_request_t req;
  cpuworker_reply_t rpl;

  fd = fdarray[1]; /* this side is ours */
#ifndef TOR_IS_MULTITHREADED
  tor_close_socket(fdarray[0]); /* this is the side of the socketpair the
                                 * parent uses */
  tor_free_all(1); /* so the child doesn't hold the parent's fd's open */
  handle_signals(0); /* ignore interrupts from the keyboard, etc */
#endif
  tor_free(data);

  setup_server_onion_keys(&onion_keys);

  for (;;) {
    if (read_all(fd, (void *)&req, sizeof(req), 1) != sizeof(req)) {
      log_info(LD_OR, "read request failed. Exiting.");
      goto end;
    }
    tor_assert(req.magic == CPUWORKER_REQUEST_MAGIC);

    memset(&rpl, 0, sizeof(rpl));

    if (req.task == CPUWORKER_TASK_ONION) {
      const create_cell_t *cc = &req.create_cell;
      created_cell_t *cell_out = &rpl.created_cell;
      int n;
      n = onion_skin_server_handshake(cc->handshake_type,
                                      cc->onionskin, cc->handshake_len,
                                      &onion_keys,
                                      cell_out->reply,
                                      rpl.keys, CPATH_KEY_MATERIAL_LEN,
                                      rpl.rend_auth_material);
      if (n < 0) {
        /* failure */
        log_debug(LD_OR,"onion_skin_server_handshake failed.");
        memset(&rpl, 0, sizeof(rpl));
        memcpy(rpl.tag, req.tag, TAG_LEN);
        rpl.success = 0;
      } else {
        /* success */
        log_debug(LD_OR,"onion_skin_server_handshake succeeded.");
        memcpy(rpl.tag, req.tag, TAG_LEN);
        cell_out->handshake_len = n;
        switch (cc->cell_type) {
        case CELL_CREATE:
          cell_out->cell_type = CELL_CREATED; break;
        case CELL_CREATE2:
          cell_out->cell_type = CELL_CREATED2; break;
        case CELL_CREATE_FAST:
          cell_out->cell_type = CELL_CREATED_FAST; break;
        default:
          tor_assert(0);
          goto end;
        }
        rpl.success = 1;
      }
      rpl.magic = CPUWORKER_REPLY_MAGIC;
      if (write_all(fd, (void*)&rpl, sizeof(rpl), 1) != sizeof(rpl)) {
        log_err(LD_BUG,"writing response buf failed. Exiting.");
        goto end;
      }
      log_debug(LD_OR,"finished writing response.");
    } else if (req.task == CPUWORKER_TASK_SHUTDOWN) {
      log_info(LD_OR,"Clean shutdown: exiting");
      goto end;
    }
    memwipe(&req, 0, sizeof(req));
    memwipe(&rpl, 0, sizeof(req));
  }
 end:
  memwipe(&req, 0, sizeof(req));
  memwipe(&rpl, 0, sizeof(req));
  release_server_onion_keys(&onion_keys);
  tor_close_socket(fd);
  crypto_thread_cleanup();
  spawn_exit();
}

/** Launch a new cpuworker. Return 0 if we're happy, -1 if we failed.
 */
static int
spawn_cpuworker(void)
{
  tor_socket_t *fdarray;
  tor_socket_t fd;
  connection_t *conn;
  int err;

  fdarray = tor_malloc(sizeof(tor_socket_t)*2);
  if ((err = tor_socketpair(AF_UNIX, SOCK_STREAM, 0, fdarray)) < 0) {
    log_warn(LD_NET, "Couldn't construct socketpair for cpuworker: %s",
             tor_socket_strerror(-err));
    tor_free(fdarray);
    return -1;
  }

  tor_assert(SOCKET_OK(fdarray[0]));
  tor_assert(SOCKET_OK(fdarray[1]));

  fd = fdarray[0];
  spawn_func(cpuworker_main, (void*)fdarray);
  log_debug(LD_OR,"just spawned a cpu worker.");
#ifndef TOR_IS_MULTITHREADED
  tor_close_socket(fdarray[1]); /* don't need the worker's side of the pipe */
  tor_free(fdarray);
#endif

  conn = connection_new(CONN_TYPE_CPUWORKER, AF_UNIX);

  set_socket_nonblocking(fd);

  /* set up conn so it's got all the data we need to remember */
  conn->s = fd;
  conn->address = tor_strdup("localhost");
  tor_addr_make_unspec(&conn->addr);

  if (connection_add(conn) < 0) { /* no space, forget it */
    log_warn(LD_NET,"connection_add for cpuworker failed. Giving up.");
    connection_free(conn); /* this closes fd */
    return -1;
  }

  conn->state = CPUWORKER_STATE_IDLE;
  connection_start_reading(conn);

  return 0; /* success */
}

/** If we have too few or too many active cpuworkers, try to spawn new ones
 * or kill idle ones.
 */
static void
spawn_enough_cpuworkers(void)
{
  int num_cpuworkers_needed = get_num_cpus(get_options());

  if (num_cpuworkers_needed < MIN_CPUWORKERS)
    num_cpuworkers_needed = MIN_CPUWORKERS;
  if (num_cpuworkers_needed > MAX_CPUWORKERS)
    num_cpuworkers_needed = MAX_CPUWORKERS;

  while (num_cpuworkers < num_cpuworkers_needed) {
    if (spawn_cpuworker() < 0) {
      log_warn(LD_GENERAL,"Cpuworker spawn failed. Will try again later.");
      return;
    }
    num_cpuworkers++;
  }
}

/** Take a pending task from the queue and assign it to 'cpuworker'. */
static void
process_pending_task(connection_t *cpuworker)
{
  or_circuit_t *circ;
  create_cell_t *onionskin = NULL;

  tor_assert(cpuworker);

  /* for now only process onion tasks */

  circ = onion_next_task(&onionskin);
  if (!circ)
    return;
  if (assign_onionskin_to_cpuworker(cpuworker, circ, onionskin))
    log_warn(LD_OR,"assign_to_cpuworker failed. Ignoring.");
}

/** How long should we let a cpuworker stay busy before we give
 * up on it and decide that we have a bug or infinite loop?
 * This value is high because some servers with low memory/cpu
 * sometimes spend an hour or more swapping, and Tor starves. */
#define CPUWORKER_BUSY_TIMEOUT (60*60*12)

/** We have a bug that I can't find. Sometimes, very rarely, cpuworkers get
 * stuck in the 'busy' state, even though the cpuworker process thinks of
 * itself as idle. I don't know why. But here's a workaround to kill any
 * cpuworker that's been busy for more than CPUWORKER_BUSY_TIMEOUT.
 */
static void
cull_wedged_cpuworkers(void)
{
  time_t now = time(NULL);
  smartlist_t *conns = get_connection_array();
  SMARTLIST_FOREACH_BEGIN(conns, connection_t *, conn) {
    if (!conn->marked_for_close &&
        conn->type == CONN_TYPE_CPUWORKER &&
        conn->state == CPUWORKER_STATE_BUSY_ONION &&
        conn->timestamp_lastwritten + CPUWORKER_BUSY_TIMEOUT < now) {
      log_notice(LD_BUG,
                 "closing wedged cpuworker. Can somebody find the bug?");
      num_cpuworkers_busy--;
      num_cpuworkers--;
      connection_mark_for_close(conn);
    }
  } SMARTLIST_FOREACH_END(conn);
}

/** Try to tell a cpuworker to perform the public key operations necessary to
 * respond to <b>onionskin</b> for the circuit <b>circ</b>.
 *
 * If <b>cpuworker</b> is defined, assert that he's idle, and use him. Else,
 * look for an idle cpuworker and use him. If none idle, queue task onto the
 * pending onion list and return.  Return 0 if we successfully assign the
 * task, or -1 on failure.
 */
int
assign_onionskin_to_cpuworker(connection_t *cpuworker,
                              or_circuit_t *circ,
                              create_cell_t *onionskin)
{
  cpuworker_request_t req;
  time_t now = approx_time();
  static time_t last_culled_cpuworkers = 0;

  /* Checking for wedged cpuworkers requires a linear search over all
   * connections, so let's do it only once a minute.
   */
#define CULL_CPUWORKERS_INTERVAL 60

  if (last_culled_cpuworkers + CULL_CPUWORKERS_INTERVAL <= now) {
    cull_wedged_cpuworkers();
    spawn_enough_cpuworkers();
    last_culled_cpuworkers = now;
  }

  if (1) {
    if (num_cpuworkers_busy == num_cpuworkers) {
      log_debug(LD_OR,"No idle cpuworkers. Queuing.");
      if (onion_pending_add(circ, onionskin) < 0) {
        tor_free(onionskin);
        return -1;
      }
      return 0;
    }

    if (!cpuworker)
      cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
                                               CPUWORKER_STATE_IDLE);

    tor_assert(cpuworker);

    if (!circ->p_chan) {
      log_info(LD_OR,"circ->p_chan gone. Failing circ.");
      tor_free(onionskin);
      return -1;
    }

    memset(&req, 0, sizeof(req));
    req.magic = CPUWORKER_REQUEST_MAGIC;
    tag_pack(req.tag, circ->p_chan->global_identifier,
             circ->p_circ_id);

    cpuworker->state = CPUWORKER_STATE_BUSY_ONION;
    /* touch the lastwritten timestamp, since that's how we check to
     * see how long it's been since we asked the question, and sometimes
     * we check before the first call to connection_handle_write(). */
    cpuworker->timestamp_lastwritten = time(NULL);
    num_cpuworkers_busy++;

    req.task = CPUWORKER_TASK_ONION;
    memcpy(&req.create_cell, onionskin, sizeof(create_cell_t));

    tor_free(onionskin);

    connection_write_to_buf((void*)&req, sizeof(req), cpuworker);
    memwipe(&req, 0, sizeof(req));
  }
  return 0;
}