1 /* Copyright (c) 2003-2004, Roger Dingledine.
2 * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
3 * Copyright (c) 2007-2009, The Tor Project, Inc. */
4 /* See LICENSE for licensing information */
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.
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
22 /** The tag specifies which circuit this onionskin was from. */
24 /** How many bytes are sent from the cpuworker back to tor? */
25 #define LEN_ONION_RESPONSE \
26 (1+TAG_LEN+ONIONSKIN_REPLY_LEN+CPATH_KEY_MATERIAL_LEN)
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;
37 static void cpuworker_main(void *data
) ATTR_NORETURN
;
38 static int spawn_cpuworker(void);
39 static void spawn_enough_cpuworkers(void);
40 static void process_pending_task(connection_t
*cpuworker
);
42 /** Initialize the cpuworker subsystem.
50 /** Called when we're done sending a request to a cpuworker. */
52 connection_cpu_finished_flushing(connection_t
*conn
)
55 tor_assert(conn
->type
== CONN_TYPE_CPUWORKER
);
56 connection_stop_writing(conn
);
60 /** Pack global_id and circ_id; set *tag to the result. (See note on
61 * cpuworker_main for wire format.) */
63 tag_pack(char *tag
, uint64_t conn_id
, circid_t circ_id
)
65 /*XXXX RETHINK THIS WHOLE MESS !!!! !NM NM NM NM*/
66 set_uint64(tag
, conn_id
);
67 set_uint16(tag
+8, circ_id
);
70 /** Unpack <b>tag</b> into addr, port, and circ_id.
73 tag_unpack(const char *tag
, uint64_t *conn_id
, circid_t
*circ_id
)
75 *conn_id
= get_uint64(tag
);
76 *circ_id
= get_uint16(tag
+8);
79 /** Called when the onion key has changed and we need to spawn new
80 * cpuworkers. Close all currently idle cpuworkers, and mark the last
81 * rotation time as now.
84 cpuworkers_rotate(void)
86 connection_t
*cpuworker
;
87 while ((cpuworker
= connection_get_by_type_state(CONN_TYPE_CPUWORKER
,
88 CPUWORKER_STATE_IDLE
))) {
89 connection_mark_for_close(cpuworker
);
92 last_rotation_time
= time(NULL
);
93 if (server_mode(get_options()))
94 spawn_enough_cpuworkers();
97 /** If the cpuworker closes the connection,
98 * mark it as closed and spawn a new one as needed. */
100 connection_cpu_reached_eof(connection_t
*conn
)
102 log_warn(LD_GENERAL
,"Read eof. CPU worker died unexpectedly.");
103 if (conn
->state
!= CPUWORKER_STATE_IDLE
) {
104 /* the circ associated with this cpuworker will have to wait until
105 * it gets culled in run_connection_housekeeping(), since we have
106 * no way to find out which circ it was. */
107 log_warn(LD_GENERAL
,"...and it left a circuit queued; abandoning circ.");
108 num_cpuworkers_busy
--;
111 spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up
113 connection_mark_for_close(conn
);
117 /** Called when we get data from a cpuworker. If the answer is not complete,
118 * wait for a complete answer. If the answer is complete,
119 * process it as appropriate.
122 connection_cpu_process_inbuf(connection_t
*conn
)
125 char buf
[LEN_ONION_RESPONSE
];
128 connection_t
*tmp_conn
;
129 or_connection_t
*p_conn
= NULL
;
133 tor_assert(conn
->type
== CONN_TYPE_CPUWORKER
);
135 if (!buf_datalen(conn
->inbuf
))
138 if (conn
->state
== CPUWORKER_STATE_BUSY_ONION
) {
139 if (buf_datalen(conn
->inbuf
) < LEN_ONION_RESPONSE
) /* answer available? */
140 return 0; /* not yet */
141 tor_assert(buf_datalen(conn
->inbuf
) == LEN_ONION_RESPONSE
);
143 connection_fetch_from_buf(&success
,1,conn
);
144 connection_fetch_from_buf(buf
,LEN_ONION_RESPONSE
-1,conn
);
146 /* parse out the circ it was talking about */
147 tag_unpack(buf
, &conn_id
, &circ_id
);
149 tmp_conn
= connection_get_by_global_id(conn_id
);
150 if (tmp_conn
&& !tmp_conn
->marked_for_close
&&
151 tmp_conn
->type
== CONN_TYPE_OR
)
152 p_conn
= TO_OR_CONN(tmp_conn
);
155 circ
= circuit_get_by_circid_orconn(circ_id
, p_conn
);
159 "decoding onionskin failed. "
160 "(Old key or bad software.) Closing.");
162 circuit_mark_for_close(circ
, END_CIRC_REASON_TORPROTOCOL
);
163 goto done_processing
;
166 /* This happens because somebody sends us a destroy cell and the
167 * circuit goes away, while the cpuworker is working. This is also
168 * why our tag doesn't include a pointer to the circ, because we'd
169 * never know if it's still valid.
171 log_debug(LD_OR
,"processed onion for a circ that's gone. Dropping.");
172 goto done_processing
;
174 tor_assert(! CIRCUIT_IS_ORIGIN(circ
));
175 if (onionskin_answer(TO_OR_CIRCUIT(circ
), CELL_CREATED
, buf
+TAG_LEN
,
176 buf
+TAG_LEN
+ONIONSKIN_REPLY_LEN
) < 0) {
177 log_warn(LD_OR
,"onionskin_answer failed. Closing.");
178 circuit_mark_for_close(circ
, END_CIRC_REASON_INTERNAL
);
179 goto done_processing
;
181 log_debug(LD_OR
,"onionskin_answer succeeded. Yay.");
183 tor_assert(0); /* don't ask me to do handshakes yet */
187 conn
->state
= CPUWORKER_STATE_IDLE
;
188 num_cpuworkers_busy
--;
189 if (conn
->timestamp_created
< last_rotation_time
) {
190 connection_mark_for_close(conn
);
192 spawn_enough_cpuworkers();
194 process_pending_task(conn
);
199 /** Implement a cpuworker. 'data' is an fdarray as returned by socketpair.
200 * Read and writes from fdarray[1]. Reads requests, writes answers.
203 * Task type [1 byte, always CPUWORKER_TASK_ONION]
205 * Onionskin challenge ONIONSKIN_CHALLENGE_LEN
207 * Success/failure [1 byte, boolean.]
209 * Onionskin challenge ONIONSKIN_REPLY_LEN
210 * Negotiated keys KEY_LEN*2+DIGEST_LEN*2
212 * (Note: this _should_ be by addr/port, since we're concerned with specific
213 * connections, not with routers (where we'd use identity).)
216 cpuworker_main(void *data
)
218 char question
[ONIONSKIN_CHALLENGE_LEN
];
219 uint8_t question_type
;
223 /* variables for onion processing */
224 char keys
[CPATH_KEY_MATERIAL_LEN
];
225 char reply_to_proxy
[ONIONSKIN_REPLY_LEN
];
226 char buf
[LEN_ONION_RESPONSE
];
228 crypto_pk_env_t
*onion_key
= NULL
, *last_onion_key
= NULL
;
230 fd
= fdarray
[1]; /* this side is ours */
231 #ifndef TOR_IS_MULTITHREADED
232 tor_close_socket(fdarray
[0]); /* this is the side of the socketpair the
234 tor_free_all(1); /* so the child doesn't hold the parent's fd's open */
235 handle_signals(0); /* ignore interrupts from the keyboard, etc */
239 dup_onion_keys(&onion_key
, &last_onion_key
);
244 if ((r
= recv(fd
, &question_type
, 1, 0)) != 1) {
245 // log_fn(LOG_ERR,"read type failed. Exiting.");
248 "CPU worker exiting because Tor process closed connection "
249 "(either rotated keys or died).");
252 "CPU worker exiting because of error on connection to Tor "
254 log_info(LD_OR
,"(Error on %d was %s)",
255 fd
, tor_socket_strerror(tor_socket_errno(fd
)));
259 tor_assert(question_type
== CPUWORKER_TASK_ONION
);
261 if (read_all(fd
, tag
, TAG_LEN
, 1) != TAG_LEN
) {
262 log_err(LD_BUG
,"read tag failed. Exiting.");
266 if (read_all(fd
, question
, ONIONSKIN_CHALLENGE_LEN
, 1) !=
267 ONIONSKIN_CHALLENGE_LEN
) {
268 log_err(LD_BUG
,"read question failed. Exiting.");
272 if (question_type
== CPUWORKER_TASK_ONION
) {
273 if (onion_skin_server_handshake(question
, onion_key
, last_onion_key
,
274 reply_to_proxy
, keys
, CPATH_KEY_MATERIAL_LEN
) < 0) {
276 log_debug(LD_OR
,"onion_skin_server_handshake failed.");
277 *buf
= 0; /* indicate failure in first byte */
278 memcpy(buf
+1,tag
,TAG_LEN
);
279 /* send all zeros as answer */
280 memset(buf
+1+TAG_LEN
, 0, LEN_ONION_RESPONSE
-(1+TAG_LEN
));
283 log_debug(LD_OR
,"onion_skin_server_handshake succeeded.");
284 buf
[0] = 1; /* 1 means success */
285 memcpy(buf
+1,tag
,TAG_LEN
);
286 memcpy(buf
+1+TAG_LEN
,reply_to_proxy
,ONIONSKIN_REPLY_LEN
);
287 memcpy(buf
+1+TAG_LEN
+ONIONSKIN_REPLY_LEN
,keys
,CPATH_KEY_MATERIAL_LEN
);
289 if (write_all(fd
, buf
, LEN_ONION_RESPONSE
, 1) != LEN_ONION_RESPONSE
) {
290 log_err(LD_BUG
,"writing response buf failed. Exiting.");
293 log_debug(LD_OR
,"finished writing response.");
298 crypto_free_pk_env(onion_key
);
300 crypto_free_pk_env(last_onion_key
);
301 tor_close_socket(fd
);
302 crypto_thread_cleanup();
306 /** Launch a new cpuworker. Return 0 if we're happy, -1 if we failed.
309 spawn_cpuworker(void)
316 fdarray
= tor_malloc(sizeof(int)*2);
317 if ((err
= tor_socketpair(AF_UNIX
, SOCK_STREAM
, 0, fdarray
)) < 0) {
318 log_warn(LD_NET
, "Couldn't construct socketpair for cpuworker: %s",
319 tor_socket_strerror(-err
));
324 tor_assert(fdarray
[0] >= 0);
325 tor_assert(fdarray
[1] >= 0);
328 spawn_func(cpuworker_main
, (void*)fdarray
);
329 log_debug(LD_OR
,"just spawned a cpu worker.");
330 #ifndef TOR_IS_MULTITHREADED
331 tor_close_socket(fdarray
[1]); /* don't need the worker's side of the pipe */
335 conn
= connection_new(CONN_TYPE_CPUWORKER
, AF_UNIX
);
337 set_socket_nonblocking(fd
);
339 /* set up conn so it's got all the data we need to remember */
341 conn
->address
= tor_strdup("localhost");
343 if (connection_add(conn
) < 0) { /* no space, forget it */
344 log_warn(LD_NET
,"connection_add for cpuworker failed. Giving up.");
345 connection_free(conn
); /* this closes fd */
349 conn
->state
= CPUWORKER_STATE_IDLE
;
350 connection_start_reading(conn
);
352 return 0; /* success */
355 /** If we have too few or too many active cpuworkers, try to spawn new ones
359 spawn_enough_cpuworkers(void)
361 int num_cpuworkers_needed
= get_options()->NumCpus
;
363 if (num_cpuworkers_needed
< MIN_CPUWORKERS
)
364 num_cpuworkers_needed
= MIN_CPUWORKERS
;
365 if (num_cpuworkers_needed
> MAX_CPUWORKERS
)
366 num_cpuworkers_needed
= MAX_CPUWORKERS
;
368 while (num_cpuworkers
< num_cpuworkers_needed
) {
369 if (spawn_cpuworker() < 0) {
370 log_warn(LD_GENERAL
,"Cpuworker spawn failed. Will try again later.");
377 /** Take a pending task from the queue and assign it to 'cpuworker'. */
379 process_pending_task(connection_t
*cpuworker
)
382 char *onionskin
= NULL
;
384 tor_assert(cpuworker
);
386 /* for now only process onion tasks */
388 circ
= onion_next_task(&onionskin
);
391 if (assign_onionskin_to_cpuworker(cpuworker
, circ
, onionskin
))
392 log_warn(LD_OR
,"assign_to_cpuworker failed. Ignoring.");
395 /** How long should we let a cpuworker stay busy before we give
396 * up on it and decide that we have a bug or infinite loop?
397 * This value is high because some servers with low memory/cpu
398 * sometimes spend an hour or more swapping, and Tor starves. */
399 #define CPUWORKER_BUSY_TIMEOUT (60*60*12)
401 /** We have a bug that I can't find. Sometimes, very rarely, cpuworkers get
402 * stuck in the 'busy' state, even though the cpuworker process thinks of
403 * itself as idle. I don't know why. But here's a workaround to kill any
404 * cpuworker that's been busy for more than CPUWORKER_BUSY_TIMEOUT.
407 cull_wedged_cpuworkers(void)
409 time_t now
= time(NULL
);
410 smartlist_t
*conns
= get_connection_array();
411 SMARTLIST_FOREACH(conns
, connection_t
*, conn
,
413 if (!conn
->marked_for_close
&&
414 conn
->type
== CONN_TYPE_CPUWORKER
&&
415 conn
->state
== CPUWORKER_STATE_BUSY_ONION
&&
416 conn
->timestamp_lastwritten
+ CPUWORKER_BUSY_TIMEOUT
< now
) {
418 "closing wedged cpuworker. Can somebody find the bug?");
419 num_cpuworkers_busy
--;
421 connection_mark_for_close(conn
);
426 /** Try to tell a cpuworker to perform the public key operations necessary to
427 * respond to <b>onionskin</b> for the circuit <b>circ</b>.
429 * If <b>cpuworker</b> is defined, assert that he's idle, and use him. Else,
430 * look for an idle cpuworker and use him. If none idle, queue task onto the
431 * pending onion list and return. Return 0 if we successfully assign the
432 * task, or -1 on failure.
435 assign_onionskin_to_cpuworker(connection_t
*cpuworker
,
436 or_circuit_t
*circ
, char *onionskin
)
441 cull_wedged_cpuworkers();
442 spawn_enough_cpuworkers();
445 if (num_cpuworkers_busy
== num_cpuworkers
) {
446 log_debug(LD_OR
,"No idle cpuworkers. Queuing.");
447 if (onion_pending_add(circ
, onionskin
) < 0) {
455 cpuworker
= connection_get_by_type_state(CONN_TYPE_CPUWORKER
,
456 CPUWORKER_STATE_IDLE
);
458 tor_assert(cpuworker
);
461 log_info(LD_OR
,"circ->p_conn gone. Failing circ.");
465 tag_pack(tag
, circ
->p_conn
->_base
.global_identifier
,
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
++;
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
);