1 /* Copyright 2003-2004 Roger Dingledine.
2 * Copyright 2004 Roger Dingledine, Nick Mathewson. */
3 /* See LICENSE for licensing information */
5 const char cpuworker_c_id
[] = "$Id$";
9 * \brief Run computation-intensive tasks (generally for crypto) in
10 * a separate execution context. [OR only.]
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 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)
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;
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
);
43 /** Initialize the cpuworker subsystem.
46 last_rotation_time
=time(NULL
);
47 spawn_enough_cpuworkers();
50 /** Called when we're done sending a request to a cpuworker. */
51 int connection_cpu_finished_flushing(connection_t
*conn
) {
53 tor_assert(conn
->type
== CONN_TYPE_CPUWORKER
);
54 connection_stop_writing(conn
);
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
;
66 /** Unpack <b>tag</b> into addr, port, and circ_id.
68 static void tag_unpack(const char *tag
, uint32_t *addr
, uint16_t *port
, uint16_t *circ_id
) {
70 char addrbuf
[INET_NTOA_BUF_LEN
];
72 *addr
= *(const uint32_t *)tag
;
73 *port
= *(const uint16_t *)(tag
+4);
74 *circ_id
= *(const uint16_t *)(tag
+6);
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
);
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.
85 void cpuworkers_rotate(void)
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
);
93 last_rotation_time
= time(NULL
);
94 spawn_enough_cpuworkers();
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
--;
109 spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up spinning. */
110 connection_mark_for_close(conn
);
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.
118 int connection_cpu_process_inbuf(connection_t
*conn
) {
120 unsigned char buf
[LEN_ONION_RESPONSE
];
124 connection_t
*p_conn
;
128 tor_assert(conn
->type
== CONN_TYPE_CPUWORKER
);
130 if (!buf_datalen(conn
->inbuf
))
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
);
138 connection_fetch_from_buf(&success
,1,conn
);
139 connection_fetch_from_buf(buf
,LEN_ONION_RESPONSE
-1,conn
);
141 /* parse out the circ it was talking about */
142 tag_unpack(buf
, &addr
, &port
, &circ_id
);
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
);
149 circ
= circuit_get_by_circ_id_conn(circ_id
, p_conn
);
152 log_fn(LOG_INFO
,"decoding onionskin failed. Closing.");
154 circuit_mark_for_close(circ
);
155 goto done_processing
;
158 log_fn(LOG_INFO
,"processed onion for a circ that's gone. Dropping.");
159 goto done_processing
;
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
;
167 log_fn(LOG_DEBUG
,"onionskin_answer succeeded. Yay.");
169 tor_assert(0); /* don't ask me to do handshakes yet */
173 conn
->state
= CPUWORKER_STATE_IDLE
;
174 num_cpuworkers_busy
--;
175 if (conn
->timestamp_created
< last_rotation_time
) {
176 connection_mark_for_close(conn
);
178 spawn_enough_cpuworkers();
180 process_pending_task(conn
);
185 /** Implement a cpuworker. 'data' is an fdarray as returned by socketpair.
186 * Read and writes from fdarray[1]. Reads requests, writes answers.
189 * Task type [1 byte, always CPUWORKER_TASK_ONION]
191 * Onionskin challenge ONIONSKIN_CHALLENGE_LEN
193 * Success/failure [1 byte, boolean.]
195 * Onionskin challenge ONIONSKIN_REPLY_LEN
196 * Negotiated keys KEY_LEN*2+DIGEST_LEN*2
198 * (Note: this _should_ be by addr/port, since we're concerned with specific
199 * connections, not with routers (where we'd use identity).)
201 static int cpuworker_main(void *data
) {
202 unsigned char question
[ONIONSKIN_CHALLENGE_LEN
];
203 unsigned char question_type
;
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
];
212 crypto_pk_env_t
*onion_key
= NULL
, *last_onion_key
= NULL
;
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 */
222 dup_onion_keys(&onion_key
, &last_onion_key
);
227 if ((r
= recv(fd
, &question_type
, 1, 0)) != 1) {
228 // log_fn(LOG_ERR,"read type failed. Exiting.");
230 log_fn(LOG_INFO
,"CPU worker exiting because Tor process closed connection (either rotated keys or died).");
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
)));
237 tor_assert(question_type
== CPUWORKER_TASK_ONION
);
239 if (read_all(fd
, tag
, TAG_LEN
, 1) != TAG_LEN
) {
240 log_fn(LOG_ERR
,"read tag failed. Exiting.");
244 if (read_all(fd
, question
, ONIONSKIN_CHALLENGE_LEN
, 1) != ONIONSKIN_CHALLENGE_LEN
) {
245 log_fn(LOG_ERR
,"read question failed. Exiting.");
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) {
253 log_fn(LOG_INFO
,"onion_skin_server_handshake failed.");
254 memset(buf
,0,LEN_ONION_RESPONSE
); /* send all zeros for failure */
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);
263 if (write_all(fd
, buf
, LEN_ONION_RESPONSE
, 1) != LEN_ONION_RESPONSE
) {
264 log_fn(LOG_ERR
,"writing response buf failed. Exiting.");
267 log_fn(LOG_DEBUG
,"finished writing response.");
272 crypto_free_pk_env(onion_key
);
274 crypto_free_pk_env(last_onion_key
);
276 return 0; /* windows wants this function to return an int */
279 /** Launch a new cpuworker.
281 static int spawn_cpuworker(void) {
286 fdarray
= tor_malloc(sizeof(int)*2);
287 if (tor_socketpair(AF_UNIX
, SOCK_STREAM
, 0, fdarray
) < 0) {
288 log(LOG_ERR
, "Couldn't construct socketpair: %s",
289 tor_socket_strerror(tor_socket_errno(-1)));
296 spawn_func(cpuworker_main
, (void*)fdarray
);
297 log_fn(LOG_DEBUG
,"just spawned a worker.");
298 #ifndef TOR_IS_MULTITHREADED
299 tor_close_socket(fdarray
[1]); /* we don't need the worker's side of the pipe */
303 conn
= connection_new(CONN_TYPE_CPUWORKER
);
305 set_socket_nonblocking(fd
);
307 /* set up conn so it's got all the data we need to remember */
309 conn
->address
= tor_strdup("localhost");
311 if (connection_add(conn
) < 0) { /* no space, forget it */
312 log_fn(LOG_WARN
,"connection_add failed. Giving up.");
313 connection_free(conn
); /* this closes fd */
317 conn
->state
= CPUWORKER_STATE_IDLE
;
318 connection_start_reading(conn
);
320 return 0; /* success */
323 /** If we have too few or too many active cpuworkers, try to spawn new ones
326 static void spawn_enough_cpuworkers(void) {
327 int num_cpuworkers_needed
= get_options()->NumCpus
;
329 if (num_cpuworkers_needed
< MIN_CPUWORKERS
)
330 num_cpuworkers_needed
= MIN_CPUWORKERS
;
331 if (num_cpuworkers_needed
> MAX_CPUWORKERS
)
332 num_cpuworkers_needed
= MAX_CPUWORKERS
;
334 while (num_cpuworkers
< num_cpuworkers_needed
) {
335 if (spawn_cpuworker() < 0) {
336 log_fn(LOG_WARN
,"spawn failed!");
343 /** Take a pending task from the queue and assign it to 'cpuworker'. */
344 static void process_pending_task(connection_t
*cpuworker
) {
347 tor_assert(cpuworker
);
349 /* for now only process onion tasks */
351 circ
= onion_next_task();
354 if (assign_to_cpuworker(cpuworker
, CPUWORKER_TASK_ONION
, circ
) < 0)
355 log_fn(LOG_WARN
,"assign_to_cpuworker failed. Ignoring.");
358 /** if cpuworker is defined, assert that he's idle, and use him. else,
359 * look for an idle cpuworker and use him. if none idle, queue task onto
360 * the pending onion list and return.
361 * If question_type is CPUWORKER_TASK_ONION then task is a circ.
362 * No other question_types are allowed.
364 int assign_to_cpuworker(connection_t
*cpuworker
, unsigned char question_type
,
369 tor_assert(question_type
== CPUWORKER_TASK_ONION
);
371 if (question_type
== CPUWORKER_TASK_ONION
) {
374 if (num_cpuworkers_busy
== num_cpuworkers
) {
375 log_fn(LOG_DEBUG
,"No idle cpuworkers. Queuing.");
376 if (onion_pending_add(circ
) < 0)
382 cpuworker
= connection_get_by_type_state(CONN_TYPE_CPUWORKER
, CPUWORKER_STATE_IDLE
);
384 tor_assert(cpuworker
);
387 log_fn(LOG_INFO
,"circ->p_conn gone. Failing circ.");
390 tag_pack(tag
, circ
->p_conn
->addr
, circ
->p_conn
->port
, circ
->p_circ_id
);
392 cpuworker
->state
= CPUWORKER_STATE_BUSY_ONION
;
393 num_cpuworkers_busy
++;
395 connection_write_to_buf(&question_type
, 1, cpuworker
);
396 connection_write_to_buf(tag
, sizeof(tag
), cpuworker
);
397 connection_write_to_buf(circ
->onionskin
, ONIONSKIN_CHALLENGE_LEN
, cpuworker
);