src/or/cpuworker.c

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