src/or/cpuworker.c

   1 /* Copyright 2003-2004 Roger Dingledine.
   2  * Copyright 2004 Roger Dingledine, Nick Mathewson. */
   3 /* See LICENSE for licensing information */
   4 /* $Id$ */
   5 const char cpuworker_c_id[] = "$Id$";
   6
   7 /**
   8  * \file cpuworker.c
   9  * \brief Run computation-intensive tasks (generally for crypto) in
  10  * a separate execution context. [OR only.]
  11  *
  12  * Right now, we only use this for processing onionskins.
  13  **/
  14
  15 #include "or.h"
  16
  17 /** The maximum number of cpuworker processes we will keep around. */
  18 #define MAX_CPUWORKERS 16
  19 /** The minimum number of cpuworker processes we will keep around. */
  20 #define MIN_CPUWORKERS 1
  21
  22 /** The tag specifies which circuit this onionskin was from. */
  23 #define TAG_LEN 8
  24 /** How many bytes are sent from tor to the cpuworker? */
  25 #define LEN_ONION_QUESTION (1+TAG_LEN+ONIONSKIN_CHALLENGE_LEN)
  26 /** How many bytes are sent from the cpuworker back to tor? */
  27 #define LEN_ONION_RESPONSE (1+TAG_LEN+ONIONSKIN_REPLY_LEN+40+32)
  28
  29 /** How many cpuworkers we have running right now. */
  30 static int num_cpuworkers=0;
  31 /** How many of the running cpuworkers have an assigned task right now. */
  32 static int num_cpuworkers_busy=0;
  33 /** We need to spawn new cpuworkers whenever we rotate the onion keys
  34  * on platforms where execution contexts==processes.  This variable stores
  35  * the last time we got a key rotation event. */
  36 static time_t last_rotation_time=0;
  37
  38 static int cpuworker_main(void *data);
  39 static int spawn_cpuworker(void);
  40 static void spawn_enough_cpuworkers(void);
  41 static void process_pending_task(connection_t *cpuworker);
  42
  43 /** Initialize the cpuworker subsystem.
  44  */
  45 void cpu_init(void) {
  46   last_rotation_time=time(NULL);
  47   spawn_enough_cpuworkers();
  48 }
  49
  50 /** Called when we're done sending a request to a cpuworker. */
  51 int connection_cpu_finished_flushing(connection_t *conn) {
  52   tor_assert(conn);
  53   tor_assert(conn->type == CONN_TYPE_CPUWORKER);
  54   connection_stop_writing(conn);
  55   return 0;
  56 }
  57
  58 /** Pack addr,port,and circ_id; set *tag to the result. (See note on
  59  * cpuworker_main for wire format.) */
  60 static void tag_pack(char *tag, uint32_t addr, uint16_t port, uint16_t circ_id) {
  61   *(uint32_t *)tag     = addr;
  62   *(uint16_t *)(tag+4) = port;
  63   *(uint16_t *)(tag+6) = circ_id;
  64 }
  65
  66 /** Unpack <b>tag</b> into addr, port, and circ_id.
  67  */
  68 static void tag_unpack(const char *tag, uint32_t *addr, uint16_t *port, uint16_t *circ_id) {
  69   struct in_addr in;
  70
  71   *addr    = *(const uint32_t *)tag;
  72   *port    = *(const uint16_t *)(tag+4);
  73   *circ_id = *(const uint16_t *)(tag+6);
  74
  75   in.s_addr = htonl(*addr);
  76   log_fn(LOG_DEBUG,"onion was from %s:%d, circ_id %d.", inet_ntoa(in), *port, *circ_id);
  77 }
  78
  79 /** Called when the onion key has changed and we need to spawn new
  80  * cpuworkers.  Close all currently idle cpuworkers, and mark the last
  81  * rotation time as now.
  82  */
  83 void cpuworkers_rotate(void)
  84 {
  85   connection_t *cpuworker;
  86   while ((cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
  87                                                    CPUWORKER_STATE_IDLE))) {
  88     connection_mark_for_close(cpuworker);
  89     --num_cpuworkers;
  90   }
  91   last_rotation_time = time(NULL);
  92   spawn_enough_cpuworkers();
  93 }
  94
  95 /** If the cpuworker closes the connection,
  96  * mark it as closed and spawn a new one as needed. */
  97 int connection_cpu_reached_eof(connection_t *conn) {
  98   log_fn(LOG_WARN,"Read eof. Worker died unexpectedly.");
  99   if (conn->state != CPUWORKER_STATE_IDLE) {
 100     /* the circ associated with this cpuworker will have to wait until
 101      * it gets culled in run_connection_housekeeping(), since we have
 102      * no way to find out which circ it was. */
 103     log_fn(LOG_WARN,"...and it left a circuit queued; abandoning circ.");
 104     num_cpuworkers_busy--;
 105   }
 106   num_cpuworkers--;
 107   spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up spinning. */
 108   connection_mark_for_close(conn);
 109   return 0;
 110 }
 111
 112 /** Called when we get data from a cpuworker.  If the answer is not complete,
 113  * wait for a complete answer. If the answer is complete,
 114  * process it as appropriate.
 115  */
 116 int connection_cpu_process_inbuf(connection_t *conn) {
 117   char success;
 118   unsigned char buf[LEN_ONION_RESPONSE];
 119   uint32_t addr;
 120   uint16_t port;
 121   uint16_t circ_id;
 122   connection_t *p_conn;
 123   circuit_t *circ;
 124
 125   tor_assert(conn);
 126   tor_assert(conn->type == CONN_TYPE_CPUWORKER);
 127
 128   if (!buf_datalen(conn->inbuf))
 129     return 0;
 130
 131   if (conn->state == CPUWORKER_STATE_BUSY_ONION) {
 132     if (buf_datalen(conn->inbuf) < LEN_ONION_RESPONSE) /* entire answer available? */
 133       return 0; /* not yet */
 134     tor_assert(buf_datalen(conn->inbuf) == LEN_ONION_RESPONSE);
 135
 136     connection_fetch_from_buf(&success,1,conn);
 137     connection_fetch_from_buf(buf,LEN_ONION_RESPONSE-1,conn);
 138
 139     /* parse out the circ it was talking about */
 140     tag_unpack(buf, &addr, &port, &circ_id);
 141     circ = NULL;
 142     /* (Here we use connection_exact_get_by_addr_port rather than
 143      * get_by_identity_digest: we want a specific port here in
 144      * case there are multiple connections.) */
 145     p_conn = connection_exact_get_by_addr_port(addr,port);
 146     if (p_conn)
 147       circ = circuit_get_by_circ_id_conn(circ_id, p_conn);
 148
 149     if (success == 0) {
 150       log_fn(LOG_INFO,"decoding onionskin failed. Closing.");
 151       if (circ)
 152         circuit_mark_for_close(circ);
 153       goto done_processing;
 154     }
 155     if (!circ) {
 156       log_fn(LOG_INFO,"processed onion for a circ that's gone. Dropping.");
 157       goto done_processing;
 158     }
 159     tor_assert(circ->p_conn);
 160     if (onionskin_answer(circ, buf+TAG_LEN, buf+TAG_LEN+ONIONSKIN_REPLY_LEN) < 0) {
 161       log_fn(LOG_WARN,"onionskin_answer failed. Closing.");
 162       circuit_mark_for_close(circ);
 163       goto done_processing;
 164     }
 165     log_fn(LOG_DEBUG,"onionskin_answer succeeded. Yay.");
 166   } else {
 167     tor_assert(0); /* don't ask me to do handshakes yet */
 168   }
 169
 170 done_processing:
 171   conn->state = CPUWORKER_STATE_IDLE;
 172   num_cpuworkers_busy--;
 173   if (conn->timestamp_created < last_rotation_time) {
 174     connection_mark_for_close(conn);
 175     num_cpuworkers--;
 176     spawn_enough_cpuworkers();
 177   } else {
 178     process_pending_task(conn);
 179   }
 180   return 0;
 181 }
 182
 183 /** Implement a cpuworker.  'data' is an fdarray as returned by socketpair.
 184  * Read and writes from fdarray[1].  Reads requests, writes answers.
 185  *
 186  *   Request format:
 187  *          Task type           [1 byte, always CPUWORKER_TASK_ONION]
 188  *          Opaque tag          TAG_LEN
 189  *          Onionskin challenge ONIONSKIN_CHALLENGE_LEN
 190  *   Response format:
 191  *          Success/failure     [1 byte, boolean.]
 192  *          Opaque tag          TAG_LEN
 193  *          Onionskin challenge ONIONSKIN_REPLY_LEN
 194  *          Negotiated keys     KEY_LEN*2+DIGEST_LEN*2
 195  *
 196  *  (Note: this _should_ be by addr/port, since we're concerned with specific
 197  * connections, not with routers (where we'd use identity).)
 198  */
 199 static int cpuworker_main(void *data) {
 200   unsigned char question[ONIONSKIN_CHALLENGE_LEN];
 201   unsigned char question_type;
 202   int *fdarray = data;
 203   int fd;
 204
 205   /* variables for onion processing */
 206   unsigned char keys[40+32];
 207   unsigned char reply_to_proxy[ONIONSKIN_REPLY_LEN];
 208   unsigned char buf[LEN_ONION_RESPONSE];
 209   char tag[TAG_LEN];
 210   crypto_pk_env_t *onion_key = NULL, *last_onion_key = NULL;
 211
 212   fd = fdarray[1]; /* this side is ours */
 213 #ifndef TOR_IS_MULTITHREADED
 214   tor_close_socket(fdarray[0]); /* this is the side of the socketpair the parent uses */
 215   connection_free_all(); /* so the child doesn't hold the parent's fd's open */
 216   handle_signals(0); /* ignore interrupts from the keyboard, etc */
 217 #endif
 218   tor_free(data);
 219
 220   dup_onion_keys(&onion_key, &last_onion_key);
 221
 222   for (;;) {
 223     int r;
 224
 225     if ((r = recv(fd, &question_type, 1, 0)) != 1) {
 226 //      log_fn(LOG_ERR,"read type failed. Exiting.");
 227       if (r == 0) {
 228         log_fn(LOG_INFO,"CPU worker exiting because Tor process closed connection (either rotated keys or died).");
 229       } else {
 230         log_fn(LOG_INFO,"CPU worker editing because of error on connection to Tor process.");
 231         log_fn(LOG_INFO,"(Error on %d was %s)", fd, tor_socket_strerror(tor_socket_errno(fd)));
 232       }
 233       goto end;
 234     }
 235     tor_assert(question_type == CPUWORKER_TASK_ONION);
 236
 237     if (read_all(fd, tag, TAG_LEN, 1) != TAG_LEN) {
 238       log_fn(LOG_ERR,"read tag failed. Exiting.");
 239       goto end;
 240     }
 241
 242     if (read_all(fd, question, ONIONSKIN_CHALLENGE_LEN, 1) != ONIONSKIN_CHALLENGE_LEN) {
 243       log_fn(LOG_ERR,"read question failed. Exiting.");
 244       goto end;
 245     }
 246
 247     if (question_type == CPUWORKER_TASK_ONION) {
 248       if (onion_skin_server_handshake(question, onion_key, last_onion_key,
 249           reply_to_proxy, keys, 40+32) < 0) {
 250         /* failure */
 251         log_fn(LOG_INFO,"onion_skin_server_handshake failed.");
 252         memset(buf,0,LEN_ONION_RESPONSE); /* send all zeros for failure */
 253       } else {
 254         /* success */
 255         log_fn(LOG_DEBUG,"onion_skin_server_handshake succeeded.");
 256         buf[0] = 1; /* 1 means success */
 257         memcpy(buf+1,tag,TAG_LEN);
 258         memcpy(buf+1+TAG_LEN,reply_to_proxy,ONIONSKIN_REPLY_LEN);
 259         memcpy(buf+1+TAG_LEN+ONIONSKIN_REPLY_LEN,keys,40+32);
 260       }
 261       if (write_all(fd, buf, LEN_ONION_RESPONSE, 1) != LEN_ONION_RESPONSE) {
 262         log_fn(LOG_ERR,"writing response buf failed. Exiting.");
 263         spawn_exit();
 264       }
 265       log_fn(LOG_DEBUG,"finished writing response.");
 266     }
 267   }
 268  end:
 269   if (onion_key)
 270     crypto_free_pk_env(onion_key);
 271   if (last_onion_key)
 272     crypto_free_pk_env(last_onion_key);
 273   spawn_exit();
 274   return 0; /* windows wants this function to return an int */
 275 }
 276
 277 /** Launch a new cpuworker.
 278  */
 279 static int spawn_cpuworker(void) {
 280   int *fdarray;
 281   int fd;
 282   connection_t *conn;
 283
 284   fdarray = tor_malloc(sizeof(int)*2);
 285   if (tor_socketpair(AF_UNIX, SOCK_STREAM, 0, fdarray) < 0) {
 286     log(LOG_ERR, "Couldn't construct socketpair: %s",
 287         tor_socket_strerror(tor_socket_errno(-1)));
 288     tor_cleanup();
 289     tor_free(fdarray);
 290     exit(1);
 291   }
 292
 293   fd = fdarray[0];
 294   spawn_func(cpuworker_main, (void*)fdarray);
 295   log_fn(LOG_DEBUG,"just spawned a worker.");
 296 #ifndef TOR_IS_MULTITHREADED
 297   tor_close_socket(fdarray[1]); /* we don't need the worker's side of the pipe */
 298   tor_free(fdarray);
 299 #endif
 300
 301   conn = connection_new(CONN_TYPE_CPUWORKER);
 302
 303   set_socket_nonblocking(fd);
 304
 305   /* set up conn so it's got all the data we need to remember */
 306   conn->s = fd;
 307   conn->address = tor_strdup("localhost");
 308
 309   if (connection_add(conn) < 0) { /* no space, forget it */
 310     log_fn(LOG_WARN,"connection_add failed. Giving up.");
 311     connection_free(conn); /* this closes fd */
 312     return -1;
 313   }
 314
 315   conn->state = CPUWORKER_STATE_IDLE;
 316   connection_start_reading(conn);
 317
 318   return 0; /* success */
 319 }
 320
 321 /** If we have too few or too many active cpuworkers, try to spawn new ones
 322  * or kill idle ones.
 323  */
 324 static void spawn_enough_cpuworkers(void) {
 325   int num_cpuworkers_needed = get_options()->NumCpus;
 326
 327   if (num_cpuworkers_needed < MIN_CPUWORKERS)
 328     num_cpuworkers_needed = MIN_CPUWORKERS;
 329   if (num_cpuworkers_needed > MAX_CPUWORKERS)
 330     num_cpuworkers_needed = MAX_CPUWORKERS;
 331
 332   while (num_cpuworkers < num_cpuworkers_needed) {
 333     if (spawn_cpuworker() < 0) {
 334       log_fn(LOG_WARN,"spawn failed!");
 335       return;
 336     }
 337     num_cpuworkers++;
 338   }
 339 }
 340
 341 /** Take a pending task from the queue and assign it to 'cpuworker'. */
 342 static void process_pending_task(connection_t *cpuworker) {
 343   circuit_t *circ;
 344
 345   tor_assert(cpuworker);
 346
 347   /* for now only process onion tasks */
 348
 349   circ = onion_next_task();
 350   if (!circ)
 351     return;
 352   if (assign_to_cpuworker(cpuworker, CPUWORKER_TASK_ONION, circ) < 0)
 353     log_fn(LOG_WARN,"assign_to_cpuworker failed. Ignoring.");
 354 }
 355
 356 /** if cpuworker is defined, assert that he's idle, and use him. else,
 357  * look for an idle cpuworker and use him. if none idle, queue task onto
 358  * the pending onion list and return.
 359  * If question_type is CPUWORKER_TASK_ONION then task is a circ.
 360  * No other question_types are allowed.
 361  */
 362 int assign_to_cpuworker(connection_t *cpuworker, unsigned char question_type,
 363                         void *task) {
 364   circuit_t *circ;
 365   char tag[TAG_LEN];
 366
 367   tor_assert(question_type == CPUWORKER_TASK_ONION);
 368
 369   if (question_type == CPUWORKER_TASK_ONION) {
 370     circ = task;
 371
 372     if (num_cpuworkers_busy == num_cpuworkers) {
 373       log_fn(LOG_DEBUG,"No idle cpuworkers. Queuing.");
 374       if (onion_pending_add(circ) < 0)
 375         return -1;
 376       return 0;
 377     }
 378
 379     if (!cpuworker)
 380       cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER, CPUWORKER_STATE_IDLE);
 381
 382     tor_assert(cpuworker);
 383
 384     if (!circ->p_conn) {
 385       log_fn(LOG_INFO,"circ->p_conn gone. Failing circ.");
 386       return -1;
 387     }
 388     tag_pack(tag, circ->p_conn->addr, circ->p_conn->port, circ->p_circ_id);
 389
 390     cpuworker->state = CPUWORKER_STATE_BUSY_ONION;
 391     num_cpuworkers_busy++;
 392
 393     connection_write_to_buf(&question_type, 1, cpuworker);
 394     connection_write_to_buf(tag, sizeof(tag), cpuworker);
 395     connection_write_to_buf(circ->onionskin, ONIONSKIN_CHALLENGE_LEN, cpuworker);
 396   }
 397   return 0;
 398 }
 399