src/or/cpuworker.c

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