src/or/cpuworker.c

   1 /* Copyright (c) 2003-2004, Roger Dingledine.
   2  * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
   3  * Copyright (c) 2007-2008, The Tor Project, Inc. */
   4 /* See LICENSE for licensing information */
   5 /* $Id$ */
   6 const char cpuworker_c_id[] =
   7   "$Id$";
   8
   9 /**
  10  * \file cpuworker.c
  11  * \brief Implements a farm of 'CPU worker' processes to perform
  12  * CPU-intensive tasks in another thread or process, to not
  13  * interrupt the main thread.
  14  *
  15  * Right now, we only use this for processing onionskins.
  16  **/
  17
  18 #include "or.h"
  19
  20 /** The maximum number of cpuworker processes we will keep around. */
  21 #define MAX_CPUWORKERS 16
  22 /** The minimum number of cpuworker processes we will keep around. */
  23 #define MIN_CPUWORKERS 1
  24
  25 /** The tag specifies which circuit this onionskin was from. */
  26 #define TAG_LEN 10
  27 /** How many bytes are sent from the cpuworker back to tor? */
  28 #define LEN_ONION_RESPONSE \
  29   (1+TAG_LEN+ONIONSKIN_REPLY_LEN+CPATH_KEY_MATERIAL_LEN)
  30
  31 /** How many cpuworkers we have running right now. */
  32 static int num_cpuworkers=0;
  33 /** How many of the running cpuworkers have an assigned task right now. */
  34 static int num_cpuworkers_busy=0;
  35 /** We need to spawn new cpuworkers whenever we rotate the onion keys
  36  * on platforms where execution contexts==processes.  This variable stores
  37  * the last time we got a key rotation event. */
  38 static time_t last_rotation_time=0;
  39
  40 static void cpuworker_main(void *data) ATTR_NORETURN;
  41 static int spawn_cpuworker(void);
  42 static void spawn_enough_cpuworkers(void);
  43 static void process_pending_task(connection_t *cpuworker);
  44
  45 /** Initialize the cpuworker subsystem.
  46  */
  47 void
  48 cpu_init(void)
  49 {
  50   cpuworkers_rotate();
  51 }
  52
  53 /** Called when we're done sending a request to a cpuworker. */
  54 int
  55 connection_cpu_finished_flushing(connection_t *conn)
  56 {
  57   tor_assert(conn);
  58   tor_assert(conn->type == CONN_TYPE_CPUWORKER);
  59   connection_stop_writing(conn);
  60   return 0;
  61 }
  62
  63 /** Pack global_id and circ_id; set *tag to the result. (See note on
  64  * cpuworker_main for wire format.) */
  65 static void
  66 tag_pack(char *tag, uint64_t conn_id, uint16_t circ_id)
  67 {
  68   *(uint64_t*)tag = conn_id;
  69   *(uint16_t*)(tag+8) = circ_id;
  70 }
  71
  72 /** Unpack <b>tag</b> into addr, port, and circ_id.
  73  */
  74 static void
  75 tag_unpack(const char *tag, uint64_t *conn_id, uint16_t *circ_id)
  76 {
  77   *conn_id = *(const uint64_t *)tag;
  78   *circ_id = *(const uint16_t *)(tag+8);
  79 }
  80
  81 /** Called when the onion key has changed and we need to spawn new
  82  * cpuworkers.  Close all currently idle cpuworkers, and mark the last
  83  * rotation time as now.
  84  */
  85 void
  86 cpuworkers_rotate(void)
  87 {
  88   connection_t *cpuworker;
  89   while ((cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
  90                                                    CPUWORKER_STATE_IDLE))) {
  91     connection_mark_for_close(cpuworker);
  92     --num_cpuworkers;
  93   }
  94   last_rotation_time = time(NULL);
  95   if (server_mode(get_options()))
  96     spawn_enough_cpuworkers();
  97 }
  98
  99 /** If the cpuworker closes the connection,
 100  * mark it as closed and spawn a new one as needed. */
 101 int
 102 connection_cpu_reached_eof(connection_t *conn)
 103 {
 104   log_warn(LD_GENERAL,"Read eof. CPU worker died unexpectedly.");
 105   if (conn->state != CPUWORKER_STATE_IDLE) {
 106     /* the circ associated with this cpuworker will have to wait until
 107      * it gets culled in run_connection_housekeeping(), since we have
 108      * no way to find out which circ it was. */
 109     log_warn(LD_GENERAL,"...and it left a circuit queued; abandoning circ.");
 110     num_cpuworkers_busy--;
 111   }
 112   num_cpuworkers--;
 113   spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up
 114                                 spinning. */
 115   connection_mark_for_close(conn);
 116   return 0;
 117 }
 118
 119 /** Called when we get data from a cpuworker.  If the answer is not complete,
 120  * wait for a complete answer. If the answer is complete,
 121  * process it as appropriate.
 122  */
 123 int
 124 connection_cpu_process_inbuf(connection_t *conn)
 125 {
 126   char success;
 127   char buf[LEN_ONION_RESPONSE];
 128   uint64_t conn_id;
 129   uint16_t circ_id;
 130   connection_t *tmp_conn;
 131   or_connection_t *p_conn = NULL;
 132   circuit_t *circ;
 133
 134   tor_assert(conn);
 135   tor_assert(conn->type == CONN_TYPE_CPUWORKER);
 136
 137   if (!buf_datalen(conn->inbuf))
 138     return 0;
 139
 140   if (conn->state == CPUWORKER_STATE_BUSY_ONION) {
 141     if (buf_datalen(conn->inbuf) < LEN_ONION_RESPONSE) /* answer available? */
 142       return 0; /* not yet */
 143     tor_assert(buf_datalen(conn->inbuf) == LEN_ONION_RESPONSE);
 144
 145     connection_fetch_from_buf(&success,1,conn);
 146     connection_fetch_from_buf(buf,LEN_ONION_RESPONSE-1,conn);
 147
 148     /* parse out the circ it was talking about */
 149     tag_unpack(buf, &conn_id, &circ_id);
 150     circ = NULL;
 151     tmp_conn = connection_get_by_global_id(conn_id);
 152     if (tmp_conn && !tmp_conn->marked_for_close &&
 153         tmp_conn->type == CONN_TYPE_OR)
 154       p_conn = TO_OR_CONN(tmp_conn);
 155
 156     if (p_conn)
 157       circ = circuit_get_by_circid_orconn(circ_id, p_conn);
 158
 159     if (success == 0) {
 160       log_debug(LD_OR,
 161                 "decoding onionskin failed. "
 162                 "(Old key or bad software.) Closing.");
 163       if (circ)
 164         circuit_mark_for_close(circ, END_CIRC_REASON_TORPROTOCOL);
 165       goto done_processing;
 166     }
 167     if (!circ) {
 168       /* This happens because somebody sends us a destroy cell and the
 169        * circuit goes away, while the cpuworker is working. This is also
 170        * why our tag doesn't include a pointer to the circ, because we'd
 171        * never know if it's still valid.
 172        */
 173       log_debug(LD_OR,"processed onion for a circ that's gone. Dropping.");
 174       goto done_processing;
 175     }
 176     tor_assert(! CIRCUIT_IS_ORIGIN(circ));
 177     if (onionskin_answer(TO_OR_CIRCUIT(circ), CELL_CREATED, buf+TAG_LEN,
 178                          buf+TAG_LEN+ONIONSKIN_REPLY_LEN) < 0) {
 179       log_warn(LD_OR,"onionskin_answer failed. Closing.");
 180       circuit_mark_for_close(circ, END_CIRC_REASON_INTERNAL);
 181       goto done_processing;
 182     }
 183     log_debug(LD_OR,"onionskin_answer succeeded. Yay.");
 184   } else {
 185     tor_assert(0); /* don't ask me to do handshakes yet */
 186   }
 187
 188 done_processing:
 189   conn->state = CPUWORKER_STATE_IDLE;
 190   num_cpuworkers_busy--;
 191   if (conn->timestamp_created < last_rotation_time) {
 192     connection_mark_for_close(conn);
 193     num_cpuworkers--;
 194     spawn_enough_cpuworkers();
 195   } else {
 196     process_pending_task(conn);
 197   }
 198   return 0;
 199 }
 200
 201 /** Implement a cpuworker.  'data' is an fdarray as returned by socketpair.
 202  * Read and writes from fdarray[1].  Reads requests, writes answers.
 203  *
 204  *   Request format:
 205  *          Task type           [1 byte, always CPUWORKER_TASK_ONION]
 206  *          Opaque tag          TAG_LEN
 207  *          Onionskin challenge ONIONSKIN_CHALLENGE_LEN
 208  *   Response format:
 209  *          Success/failure     [1 byte, boolean.]
 210  *          Opaque tag          TAG_LEN
 211  *          Onionskin challenge ONIONSKIN_REPLY_LEN
 212  *          Negotiated keys     KEY_LEN*2+DIGEST_LEN*2
 213  *
 214  *  (Note: this _should_ be by addr/port, since we're concerned with specific
 215  * connections, not with routers (where we'd use identity).)
 216  */
 217 static void
 218 cpuworker_main(void *data)
 219 {
 220   char question[ONIONSKIN_CHALLENGE_LEN];
 221   uint8_t question_type;
 222   int *fdarray = data;
 223   int fd;
 224
 225   /* variables for onion processing */
 226   char keys[CPATH_KEY_MATERIAL_LEN];
 227   char reply_to_proxy[ONIONSKIN_REPLY_LEN];
 228   char buf[LEN_ONION_RESPONSE];
 229   char tag[TAG_LEN];
 230   crypto_pk_env_t *onion_key = NULL, *last_onion_key = NULL;
 231
 232   fd = fdarray[1]; /* this side is ours */
 233 #ifndef TOR_IS_MULTITHREADED
 234   tor_close_socket(fdarray[0]); /* this is the side of the socketpair the
 235                                  * parent uses */
 236   tor_free_all(1); /* so the child doesn't hold the parent's fd's open */
 237   handle_signals(0); /* ignore interrupts from the keyboard, etc */
 238 #endif
 239   tor_free(data);
 240
 241   dup_onion_keys(&onion_key, &last_onion_key);
 242
 243   for (;;) {
 244     ssize_t r;
 245
 246     if ((r = recv(fd, &question_type, 1, 0)) != 1) {
 247 //      log_fn(LOG_ERR,"read type failed. Exiting.");
 248       if (r == 0) {
 249         log_info(LD_OR,
 250                  "CPU worker exiting because Tor process closed connection "
 251                  "(either rotated keys or died).");
 252       } else {
 253         log_info(LD_OR,
 254                  "CPU worker exiting because of error on connection to Tor "
 255                  "process.");
 256         log_info(LD_OR,"(Error on %d was %s)",
 257                  fd, tor_socket_strerror(tor_socket_errno(fd)));
 258       }
 259       goto end;
 260     }
 261     tor_assert(question_type == CPUWORKER_TASK_ONION);
 262
 263     if (read_all(fd, tag, TAG_LEN, 1) != TAG_LEN) {
 264       log_err(LD_BUG,"read tag failed. Exiting.");
 265       goto end;
 266     }
 267
 268     if (read_all(fd, question, ONIONSKIN_CHALLENGE_LEN, 1) !=
 269         ONIONSKIN_CHALLENGE_LEN) {
 270       log_err(LD_BUG,"read question failed. Exiting.");
 271       goto end;
 272     }
 273
 274     if (question_type == CPUWORKER_TASK_ONION) {
 275       if (onion_skin_server_handshake(question, onion_key, last_onion_key,
 276           reply_to_proxy, keys, CPATH_KEY_MATERIAL_LEN) < 0) {
 277         /* failure */
 278         log_debug(LD_OR,"onion_skin_server_handshake failed.");
 279         *buf = 0; /* indicate failure in first byte */
 280         memcpy(buf+1,tag,TAG_LEN);
 281         /* send all zeros as answer */
 282         memset(buf+1+TAG_LEN, 0, LEN_ONION_RESPONSE-(1+TAG_LEN));
 283       } else {
 284         /* success */
 285         log_debug(LD_OR,"onion_skin_server_handshake succeeded.");
 286         buf[0] = 1; /* 1 means success */
 287         memcpy(buf+1,tag,TAG_LEN);
 288         memcpy(buf+1+TAG_LEN,reply_to_proxy,ONIONSKIN_REPLY_LEN);
 289         memcpy(buf+1+TAG_LEN+ONIONSKIN_REPLY_LEN,keys,CPATH_KEY_MATERIAL_LEN);
 290       }
 291       if (write_all(fd, buf, LEN_ONION_RESPONSE, 1) != LEN_ONION_RESPONSE) {
 292         log_err(LD_BUG,"writing response buf failed. Exiting.");
 293         goto end;
 294       }
 295       log_debug(LD_OR,"finished writing response.");
 296     }
 297   }
 298  end:
 299   if (onion_key)
 300     crypto_free_pk_env(onion_key);
 301   if (last_onion_key)
 302     crypto_free_pk_env(last_onion_key);
 303   tor_close_socket(fd);
 304   crypto_thread_cleanup();
 305   spawn_exit();
 306 }
 307
 308 /** Launch a new cpuworker. Return 0 if we're happy, -1 if we failed.
 309  */
 310 static int
 311 spawn_cpuworker(void)
 312 {
 313   int *fdarray;
 314   int fd;
 315   connection_t *conn;
 316   int err;
 317
 318   fdarray = tor_malloc(sizeof(int)*2);
 319   if ((err = tor_socketpair(AF_UNIX, SOCK_STREAM, 0, fdarray)) < 0) {
 320     log_warn(LD_NET, "Couldn't construct socketpair for cpuworker: %s",
 321              tor_socket_strerror(-err));
 322     tor_free(fdarray);
 323     return -1;
 324   }
 325
 326   tor_assert(fdarray[0] >= 0);
 327   tor_assert(fdarray[1] >= 0);
 328
 329   fd = fdarray[0];
 330   spawn_func(cpuworker_main, (void*)fdarray);
 331   log_debug(LD_OR,"just spawned a cpu worker.");
 332 #ifndef TOR_IS_MULTITHREADED
 333   tor_close_socket(fdarray[1]); /* don't need the worker's side of the pipe */
 334   tor_free(fdarray);
 335 #endif
 336
 337   conn = connection_new(CONN_TYPE_CPUWORKER, AF_UNIX);
 338
 339   set_socket_nonblocking(fd);
 340
 341   /* set up conn so it's got all the data we need to remember */
 342   conn->s = fd;
 343   conn->address = tor_strdup("localhost");
 344
 345   if (connection_add(conn) < 0) { /* no space, forget it */
 346     log_warn(LD_NET,"connection_add for cpuworker failed. Giving up.");
 347     connection_free(conn); /* this closes fd */
 348     return -1;
 349   }
 350
 351   conn->state = CPUWORKER_STATE_IDLE;
 352   connection_start_reading(conn);
 353
 354   return 0; /* success */
 355 }
 356
 357 /** If we have too few or too many active cpuworkers, try to spawn new ones
 358  * or kill idle ones.
 359  */
 360 static void
 361 spawn_enough_cpuworkers(void)
 362 {
 363   int num_cpuworkers_needed = get_options()->NumCpus;
 364
 365   if (num_cpuworkers_needed < MIN_CPUWORKERS)
 366     num_cpuworkers_needed = MIN_CPUWORKERS;
 367   if (num_cpuworkers_needed > MAX_CPUWORKERS)
 368     num_cpuworkers_needed = MAX_CPUWORKERS;
 369
 370   while (num_cpuworkers < num_cpuworkers_needed) {
 371     if (spawn_cpuworker() < 0) {
 372       log_warn(LD_GENERAL,"Cpuworker spawn failed. Will try again later.");
 373       return;
 374     }
 375     num_cpuworkers++;
 376   }
 377 }
 378
 379 /** Take a pending task from the queue and assign it to 'cpuworker'. */
 380 static void
 381 process_pending_task(connection_t *cpuworker)
 382 {
 383   or_circuit_t *circ;
 384   char *onionskin = NULL;
 385
 386   tor_assert(cpuworker);
 387
 388   /* for now only process onion tasks */
 389
 390   circ = onion_next_task(&onionskin);
 391   if (!circ)
 392     return;
 393   if (assign_onionskin_to_cpuworker(cpuworker, circ, onionskin))
 394     log_warn(LD_OR,"assign_to_cpuworker failed. Ignoring.");
 395 }
 396
 397 /** How long should we let a cpuworker stay busy before we give
 398  * up on it and decide that we have a bug or infinite loop?
 399  * This value is high because some servers with low memory/cpu
 400  * sometimes spend an hour or more swapping, and Tor starves. */
 401 #define CPUWORKER_BUSY_TIMEOUT (60*60*12)
 402
 403 /** We have a bug that I can't find. Sometimes, very rarely, cpuworkers get
 404  * stuck in the 'busy' state, even though the cpuworker process thinks of
 405  * itself as idle. I don't know why. But here's a workaround to kill any
 406  * cpuworker that's been busy for more than CPUWORKER_BUSY_TIMEOUT.
 407  */
 408 static void
 409 cull_wedged_cpuworkers(void)
 410 {
 411   time_t now = time(NULL);
 412   smartlist_t *conns = get_connection_array();
 413   SMARTLIST_FOREACH(conns, connection_t *, conn,
 414   {
 415     if (!conn->marked_for_close &&
 416         conn->type == CONN_TYPE_CPUWORKER &&
 417         conn->state == CPUWORKER_STATE_BUSY_ONION &&
 418         conn->timestamp_lastwritten + CPUWORKER_BUSY_TIMEOUT < now) {
 419       log_notice(LD_BUG,
 420                  "closing wedged cpuworker. Can somebody find the bug?");
 421       num_cpuworkers_busy--;
 422       num_cpuworkers--;
 423       connection_mark_for_close(conn);
 424     }
 425   });
 426 }
 427
 428 /** Try to tell a cpuworker to perform the public key operations necessary to
 429  * respond to <b>onionskin</b> for the circuit <b>circ</b>.
 430  *
 431  * If <b>cpuworker</b> is defined, assert that he's idle, and use him. Else,
 432  * look for an idle cpuworker and use him. If none idle, queue task onto the
 433  * pending onion list and return.  Return 0 if we successfully assign the
 434  * task, or -1 on failure.
 435  */
 436 int
 437 assign_onionskin_to_cpuworker(connection_t *cpuworker,
 438                               or_circuit_t *circ, char *onionskin)
 439 {
 440   char qbuf[1];
 441   char tag[TAG_LEN];
 442
 443   cull_wedged_cpuworkers();
 444   spawn_enough_cpuworkers();
 445
 446   if (1) {
 447     if (num_cpuworkers_busy == num_cpuworkers) {
 448       log_debug(LD_OR,"No idle cpuworkers. Queuing.");
 449       if (onion_pending_add(circ, onionskin) < 0)
 450         return -1;
 451       return 0;
 452     }
 453
 454     if (!cpuworker)
 455       cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
 456                                                CPUWORKER_STATE_IDLE);
 457
 458     tor_assert(cpuworker);
 459
 460     if (!circ->p_conn) {
 461       log_info(LD_OR,"circ->p_conn gone. Failing circ.");
 462       tor_free(onionskin);
 463       return -1;
 464     }
 465     tag_pack(tag, circ->p_conn->_base.global_identifier,
 466              circ->p_circ_id);
 467
 468     cpuworker->state = CPUWORKER_STATE_BUSY_ONION;
 469     /* touch the lastwritten timestamp, since that's how we check to
 470      * see how long it's been since we asked the question, and sometimes
 471      * we check before the first call to connection_handle_write(). */
 472     cpuworker->timestamp_lastwritten = time(NULL);
 473     num_cpuworkers_busy++;
 474
 475     qbuf[0] = CPUWORKER_TASK_ONION;
 476     connection_write_to_buf(qbuf, 1, cpuworker);
 477     connection_write_to_buf(tag, sizeof(tag), cpuworker);
 478     connection_write_to_buf(onionskin, ONIONSKIN_CHALLENGE_LEN, cpuworker);
 479     tor_free(onionskin);
 480   }
 481   return 0;
 482 }
 483