cpuworker.c 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327
  1. /* Copyright 2003 Roger Dingledine. */
  2. /* See LICENSE for licensing information */
  3. /* $Id$ */
  4. #include "or.h"
  5. extern or_options_t options; /* command-line and config-file options */
  6. #define MAX_CPUWORKERS 16
  7. #define MIN_CPUWORKERS 1
  8. #define TAG_LEN 8
  9. #define LEN_ONION_QUESTION (1+TAG_LEN+ONIONSKIN_CHALLENGE_LEN)
  10. #define LEN_ONION_RESPONSE (1+TAG_LEN+ONIONSKIN_REPLY_LEN+40+32)
  11. static int num_cpuworkers=0;
  12. static int num_cpuworkers_busy=0;
  13. static time_t last_rotation_time=0;
  14. int cpuworker_main(void *data);
  15. static int spawn_cpuworker(void);
  16. static void spawn_enough_cpuworkers(void);
  17. static void process_pending_task(connection_t *cpuworker);
  18. void cpu_init(void) {
  19. last_rotation_time=time(NULL);
  20. spawn_enough_cpuworkers();
  21. }
  22. int connection_cpu_finished_flushing(connection_t *conn) {
  23. tor_assert(conn && conn->type == CONN_TYPE_CPUWORKER);
  24. connection_stop_writing(conn);
  25. return 0;
  26. }
  27. static void tag_pack(char *tag, uint32_t addr, uint16_t port, uint16_t circ_id) {
  28. *(uint32_t *)tag = addr;
  29. *(uint16_t *)(tag+4) = port;
  30. *(uint16_t *)(tag+6) = circ_id;
  31. }
  32. static void tag_unpack(char *tag, uint32_t *addr, uint16_t *port, uint16_t *circ_id) {
  33. struct in_addr in;
  34. *addr = *(uint32_t *)tag;
  35. *port = *(uint16_t *)(tag+4);
  36. *circ_id = *(uint16_t *)(tag+6);
  37. in.s_addr = htonl(*addr);
  38. log_fn(LOG_DEBUG,"onion was from %s:%d, circ_id %d.", inet_ntoa(in), *port, *circ_id);
  39. }
  40. void cpuworkers_rotate(void)
  41. {
  42. connection_t *cpuworker;
  43. while ((cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER,
  44. CPUWORKER_STATE_IDLE))) {
  45. connection_mark_for_close(cpuworker,0);
  46. --num_cpuworkers;
  47. }
  48. last_rotation_time = time(NULL);
  49. spawn_enough_cpuworkers();
  50. }
  51. int connection_cpu_process_inbuf(connection_t *conn) {
  52. char success;
  53. unsigned char buf[LEN_ONION_RESPONSE];
  54. uint32_t addr;
  55. uint16_t port;
  56. uint16_t circ_id;
  57. connection_t *p_conn;
  58. circuit_t *circ;
  59. tor_assert(conn && conn->type == CONN_TYPE_CPUWORKER);
  60. if(conn->inbuf_reached_eof) {
  61. log_fn(LOG_WARN,"Read eof. Worker died unexpectedly.");
  62. if(conn->state != CPUWORKER_STATE_IDLE) {
  63. /* the circ associated with this cpuworker will have to wait until
  64. * it gets culled in run_connection_housekeeping(), since we have
  65. * no way to find out which circ it was. */
  66. log_fn(LOG_WARN,"...and it left a circuit queued; abandoning circ.");
  67. num_cpuworkers_busy--;
  68. }
  69. num_cpuworkers--;
  70. spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up spinning. */
  71. connection_mark_for_close(conn,0);
  72. return 0;
  73. }
  74. if(conn->state == CPUWORKER_STATE_BUSY_ONION) {
  75. if(buf_datalen(conn->inbuf) < LEN_ONION_RESPONSE) /* entire answer available? */
  76. return 0; /* not yet */
  77. tor_assert(buf_datalen(conn->inbuf) == LEN_ONION_RESPONSE);
  78. connection_fetch_from_buf(&success,1,conn);
  79. connection_fetch_from_buf(buf,LEN_ONION_RESPONSE-1,conn);
  80. /* parse out the circ it was talking about */
  81. tag_unpack(buf, &addr, &port, &circ_id);
  82. circ = NULL;
  83. p_conn = connection_exact_get_by_addr_port(addr,port);
  84. if(p_conn)
  85. circ = circuit_get_by_circ_id_conn(circ_id, p_conn);
  86. if(success == 0) {
  87. log_fn(LOG_WARN,"decoding onionskin failed. Closing.");
  88. if(circ)
  89. circuit_mark_for_close(circ);
  90. goto done_processing;
  91. }
  92. if(!circ) {
  93. log_fn(LOG_INFO,"processed onion for a circ that's gone. Dropping.");
  94. goto done_processing;
  95. }
  96. tor_assert(circ->p_conn);
  97. if(onionskin_answer(circ, buf+TAG_LEN, buf+TAG_LEN+ONIONSKIN_REPLY_LEN) < 0) {
  98. log_fn(LOG_WARN,"onionskin_answer failed. Closing.");
  99. circuit_mark_for_close(circ);
  100. goto done_processing;
  101. }
  102. log_fn(LOG_DEBUG,"onionskin_answer succeeded. Yay.");
  103. } else {
  104. tor_assert(0); /* don't ask me to do handshakes yet */
  105. }
  106. done_processing:
  107. conn->state = CPUWORKER_STATE_IDLE;
  108. num_cpuworkers_busy--;
  109. if (conn->timestamp_created < last_rotation_time) {
  110. connection_mark_for_close(conn,0);
  111. num_cpuworkers--;
  112. spawn_enough_cpuworkers();
  113. } else {
  114. process_pending_task(conn);
  115. }
  116. return 0;
  117. }
  118. int cpuworker_main(void *data) {
  119. unsigned char question[ONIONSKIN_CHALLENGE_LEN];
  120. unsigned char question_type;
  121. int *fdarray = data;
  122. int fd;
  123. /* variables for onion processing */
  124. unsigned char keys[40+32];
  125. unsigned char reply_to_proxy[ONIONSKIN_REPLY_LEN];
  126. unsigned char buf[LEN_ONION_RESPONSE];
  127. char tag[TAG_LEN];
  128. crypto_pk_env_t *onion_key = NULL, *last_onion_key = NULL;
  129. close(fdarray[0]); /* this is the side of the socketpair the parent uses */
  130. fd = fdarray[1]; /* this side is ours */
  131. #ifndef MS_WINDOWS
  132. connection_free_all(); /* so the child doesn't hold the parent's fd's open */
  133. #endif
  134. /* XXXX WINDOWS lock here. */
  135. onion_key = crypto_pk_dup_key(get_onion_key());
  136. if (get_previous_onion_key())
  137. last_onion_key = crypto_pk_dup_key(get_previous_onion_key());
  138. for(;;) {
  139. if(recv(fd, &question_type, 1, 0) != 1) {
  140. // log_fn(LOG_ERR,"read type failed. Exiting.");
  141. log_fn(LOG_INFO,"cpuworker exiting because tor process died.");
  142. goto end;
  143. }
  144. tor_assert(question_type == CPUWORKER_TASK_ONION);
  145. if(read_all(fd, tag, TAG_LEN, 1) != TAG_LEN) {
  146. log_fn(LOG_ERR,"read tag failed. Exiting.");
  147. goto end;
  148. }
  149. if(read_all(fd, question, ONIONSKIN_CHALLENGE_LEN, 1) != ONIONSKIN_CHALLENGE_LEN) {
  150. log_fn(LOG_ERR,"read question failed. Exiting.");
  151. goto end;
  152. }
  153. if(question_type == CPUWORKER_TASK_ONION) {
  154. if(onion_skin_server_handshake(question, onion_key, last_onion_key,
  155. reply_to_proxy, keys, 40+32) < 0) {
  156. /* failure */
  157. log_fn(LOG_WARN,"onion_skin_server_handshake failed.");
  158. memset(buf,0,LEN_ONION_RESPONSE); /* send all zeros for failure */
  159. } else {
  160. /* success */
  161. log_fn(LOG_INFO,"onion_skin_server_handshake succeeded.");
  162. buf[0] = 1; /* 1 means success */
  163. memcpy(buf+1,tag,TAG_LEN);
  164. memcpy(buf+1+TAG_LEN,reply_to_proxy,ONIONSKIN_REPLY_LEN);
  165. memcpy(buf+1+TAG_LEN+ONIONSKIN_REPLY_LEN,keys,40+32);
  166. }
  167. if(write_all(fd, buf, LEN_ONION_RESPONSE, 1) != LEN_ONION_RESPONSE) {
  168. log_fn(LOG_ERR,"writing response buf failed. Exiting.");
  169. spawn_exit();
  170. }
  171. log_fn(LOG_DEBUG,"finished writing response.");
  172. }
  173. }
  174. end:
  175. if (onion_key)
  176. crypto_free_pk_env(onion_key);
  177. if (last_onion_key)
  178. crypto_free_pk_env(last_onion_key);
  179. spawn_exit();
  180. return 0; /* windows wants this function to return an int */
  181. }
  182. static int spawn_cpuworker(void) {
  183. int fd[2];
  184. connection_t *conn;
  185. if(tor_socketpair(AF_UNIX, SOCK_STREAM, 0, fd) < 0) {
  186. log(LOG_ERR, "Couldn't construct socketpair: %s", strerror(errno));
  187. exit(1);
  188. }
  189. spawn_func(cpuworker_main, (void*)fd);
  190. log_fn(LOG_DEBUG,"just spawned a worker.");
  191. close(fd[1]); /* we don't need the worker's side of the pipe */
  192. conn = connection_new(CONN_TYPE_CPUWORKER);
  193. set_socket_nonblocking(fd[0]);
  194. /* set up conn so it's got all the data we need to remember */
  195. conn->s = fd[0];
  196. conn->address = tor_strdup("localhost");
  197. if(connection_add(conn) < 0) { /* no space, forget it */
  198. log_fn(LOG_WARN,"connection_add failed. Giving up.");
  199. connection_free(conn); /* this closes fd[0] */
  200. return -1;
  201. }
  202. conn->state = CPUWORKER_STATE_IDLE;
  203. connection_start_reading(conn);
  204. return 0; /* success */
  205. }
  206. static void spawn_enough_cpuworkers(void) {
  207. int num_cpuworkers_needed = options.NumCpus;
  208. if(num_cpuworkers_needed < MIN_CPUWORKERS)
  209. num_cpuworkers_needed = MIN_CPUWORKERS;
  210. if(num_cpuworkers_needed > MAX_CPUWORKERS)
  211. num_cpuworkers_needed = MAX_CPUWORKERS;
  212. while(num_cpuworkers < num_cpuworkers_needed) {
  213. if(spawn_cpuworker() < 0) {
  214. log_fn(LOG_WARN,"spawn failed!");
  215. return;
  216. }
  217. num_cpuworkers++;
  218. }
  219. }
  220. static void process_pending_task(connection_t *cpuworker) {
  221. circuit_t *circ;
  222. tor_assert(cpuworker);
  223. /* for now only process onion tasks */
  224. circ = onion_next_task();
  225. if(!circ)
  226. return;
  227. if(assign_to_cpuworker(cpuworker, CPUWORKER_TASK_ONION, circ) < 0)
  228. log_fn(LOG_WARN,"assign_to_cpuworker failed. Ignoring.");
  229. }
  230. /* if cpuworker is defined, assert that he's idle, and use him. else,
  231. * look for an idle cpuworker and use him. if none idle, queue task onto
  232. * the pending onion list and return.
  233. * If question_type is CPUWORKER_TASK_ONION then task is a circ.
  234. * No other question_types are allowed.
  235. */
  236. int assign_to_cpuworker(connection_t *cpuworker, unsigned char question_type,
  237. void *task) {
  238. circuit_t *circ;
  239. char tag[TAG_LEN];
  240. tor_assert(question_type == CPUWORKER_TASK_ONION);
  241. if(question_type == CPUWORKER_TASK_ONION) {
  242. circ = task;
  243. if(num_cpuworkers_busy == num_cpuworkers) {
  244. log_fn(LOG_DEBUG,"No idle cpuworkers. Queuing.");
  245. if(onion_pending_add(circ) < 0)
  246. return -1;
  247. return 0;
  248. }
  249. if (!cpuworker)
  250. cpuworker = connection_get_by_type_state(CONN_TYPE_CPUWORKER, CPUWORKER_STATE_IDLE);
  251. tor_assert(cpuworker);
  252. if(!circ->p_conn) {
  253. log_fn(LOG_INFO,"circ->p_conn gone. Failing circ.");
  254. return -1;
  255. }
  256. tag_pack(tag, circ->p_conn->addr, circ->p_conn->port, circ->p_circ_id);
  257. cpuworker->state = CPUWORKER_STATE_BUSY_ONION;
  258. num_cpuworkers_busy++;
  259. connection_write_to_buf(&question_type, 1, cpuworker);
  260. connection_write_to_buf(tag, sizeof(tag), cpuworker);
  261. connection_write_to_buf(circ->onionskin, ONIONSKIN_CHALLENGE_LEN, cpuworker);
  262. }
  263. return 0;
  264. }
  265. /*
  266. Local Variables:
  267. mode:c
  268. indent-tabs-mode:nil
  269. c-basic-offset:2
  270. End:
  271. */