Report only the top 10 ports in exit-port stats.
[tor/rransom.git] / src / or / cpuworker.c
blobae8d69f6ec8889cb0a1de14b0c0ff683ef55fc1b
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 */
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.
12 * Right now, we only use this for processing onionskins.
13 **/
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"
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
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)
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;
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);
51 /** Initialize the cpuworker subsystem.
53 void
54 cpu_init(void)
56 cpuworkers_rotate();
59 /** Called when we're done sending a request to a cpuworker. */
60 int
61 connection_cpu_finished_flushing(connection_t *conn)
63 tor_assert(conn);
64 tor_assert(conn->type == CONN_TYPE_CPUWORKER);
65 connection_stop_writing(conn);
66 return 0;
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)
74 /*XXXX RETHINK THIS WHOLE MESS !!!! !NM NM NM NM*/
75 set_uint64(tag, conn_id);
76 set_uint16(tag+8, circ_id);
79 /** Unpack <b>tag</b> into addr, port, and circ_id.
81 static void
82 tag_unpack(const char *tag, uint64_t *conn_id, circid_t *circ_id)
84 *conn_id = get_uint64(tag);
85 *circ_id = get_uint16(tag+8);
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.
92 void
93 cpuworkers_rotate(void)
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;
101 last_rotation_time = time(NULL);
102 if (server_mode(get_options()))
103 spawn_enough_cpuworkers();
106 /** If the cpuworker closes the connection,
107 * mark it as closed and spawn a new one as needed. */
109 connection_cpu_reached_eof(connection_t *conn)
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--;
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;
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.
131 connection_cpu_process_inbuf(connection_t *conn)
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;
141 tor_assert(conn);
142 tor_assert(conn->type == CONN_TYPE_CPUWORKER);
144 if (!buf_datalen(conn->inbuf))
145 return 0;
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);
152 connection_fetch_from_buf(&success,1,conn);
153 connection_fetch_from_buf(buf,LEN_ONION_RESPONSE-1,conn);
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);
163 if (p_conn)
164 circ = circuit_get_by_circid_orconn(circ_id, p_conn);
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;
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.
180 log_debug(LD_OR,"processed onion for a circ that's gone. Dropping.");
181 goto done_processing;
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;
190 log_debug(LD_OR,"onionskin_answer succeeded. Yay.");
191 } else {
192 tor_assert(0); /* don't ask me to do handshakes yet */
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);
205 return 0;
208 /** Implement a cpuworker. 'data' is an fdarray as returned by socketpair.
209 * Read and writes from fdarray[1]. Reads requests, writes answers.
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
221 * (Note: this _should_ be by addr/port, since we're concerned with specific
222 * connections, not with routers (where we'd use identity).)
224 static void
225 cpuworker_main(void *data)
227 char question[ONIONSKIN_CHALLENGE_LEN];
228 uint8_t question_type;
229 int *fdarray = data;
230 int fd;
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;
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);
248 dup_onion_keys(&onion_key, &last_onion_key);
250 for (;;) {
251 ssize_t r;
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)));
266 goto end;
268 tor_assert(question_type == CPUWORKER_TASK_ONION);
270 if (read_all(fd, tag, TAG_LEN, 1) != TAG_LEN) {
271 log_err(LD_BUG,"read tag failed. Exiting.");
272 goto end;
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;
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);
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;
302 log_debug(LD_OR,"finished writing response.");
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();
315 /** Launch a new cpuworker. Return 0 if we're happy, -1 if we failed.
317 static int
318 spawn_cpuworker(void)
320 int *fdarray;
321 int fd;
322 connection_t *conn;
323 int err;
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;
333 tor_assert(fdarray[0] >= 0);
334 tor_assert(fdarray[1] >= 0);
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
344 conn = connection_new(CONN_TYPE_CPUWORKER, AF_UNIX);
346 set_socket_nonblocking(fd);
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");
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;
358 conn->state = CPUWORKER_STATE_IDLE;
359 connection_start_reading(conn);
361 return 0; /* success */
364 /** If we have too few or too many active cpuworkers, try to spawn new ones
365 * or kill idle ones.
367 static void
368 spawn_enough_cpuworkers(void)
370 int num_cpuworkers_needed = get_options()->NumCPUs;
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;
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;
382 num_cpuworkers++;
386 /** Take a pending task from the queue and assign it to 'cpuworker'. */
387 static void
388 process_pending_task(connection_t *cpuworker)
390 or_circuit_t *circ;
391 char *onionskin = NULL;
393 tor_assert(cpuworker);
395 /* for now only process onion tasks */
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.");
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)
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.
415 static void
416 cull_wedged_cpuworkers(void)
418 time_t now = time(NULL);
419 smartlist_t *conns = get_connection_array();
420 SMARTLIST_FOREACH(conns, connection_t *, conn,
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);
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>.
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.
444 assign_onionskin_to_cpuworker(connection_t *cpuworker,
445 or_circuit_t *circ, char *onionskin)
447 char qbuf[1];
448 char tag[TAG_LEN];
450 cull_wedged_cpuworkers();
451 spawn_enough_cpuworkers();
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;
460 return 0;
463 if (!cpuworker)
464 cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
465 CPUWORKER_STATE_IDLE);
467 tor_assert(cpuworker);
469 if (!circ->p_conn) {
470 log_info(LD_OR,"circ->p_conn gone. Failing circ.");
471 tor_free(onionskin);
472 return -1;
474 tag_pack(tag, circ->p_conn->_base.global_identifier,
475 circ->p_circ_id);
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++;
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);
490 return 0;