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