cpuworker.c 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571
  1. /* Copyright (c) 2003-2004, Roger Dingledine.
  2. * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
  3. * Copyright (c) 2007-2015, The Tor Project, Inc. */
  4. /* See LICENSE for licensing information */
  5. /**
  6. * \file cpuworker.c
  7. * \brief Uses the workqueue/threadpool code to farm CPU-intensive activities
  8. * out to subprocesses.
  9. *
  10. * Right now, we only use this for processing onionskins.
  11. **/
  12. #include "or.h"
  13. #include "channel.h"
  14. #include "circuitbuild.h"
  15. #include "circuitlist.h"
  16. #include "connection_or.h"
  17. #include "config.h"
  18. #include "cpuworker.h"
  19. #include "main.h"
  20. #include "onion.h"
  21. #include "rephist.h"
  22. #include "router.h"
  23. #include "workqueue.h"
  24. #ifdef HAVE_EVENT2_EVENT_H
  25. #include <event2/event.h>
  26. #else
  27. #include <event.h>
  28. #endif
  29. static void queue_pending_tasks(void);
  30. typedef struct worker_state_s {
  31. int generation;
  32. server_onion_keys_t *onion_keys;
  33. } worker_state_t;
  34. static void *
  35. worker_state_new(void *arg)
  36. {
  37. worker_state_t *ws;
  38. (void)arg;
  39. ws = tor_malloc_zero(sizeof(worker_state_t));
  40. ws->onion_keys = server_onion_keys_new();
  41. return ws;
  42. }
  43. static void
  44. worker_state_free(void *arg)
  45. {
  46. worker_state_t *ws = arg;
  47. server_onion_keys_free(ws->onion_keys);
  48. tor_free(ws);
  49. }
  50. static replyqueue_t *replyqueue = NULL;
  51. static threadpool_t *threadpool = NULL;
  52. static struct event *reply_event = NULL;
  53. static tor_weak_rng_t request_sample_rng = TOR_WEAK_RNG_INIT;
  54. static int total_pending_tasks = 0;
  55. static int max_pending_tasks = 128;
  56. static void
  57. replyqueue_process_cb(evutil_socket_t sock, short events, void *arg)
  58. {
  59. replyqueue_t *rq = arg;
  60. (void) sock;
  61. (void) events;
  62. replyqueue_process(rq);
  63. }
  64. /** Initialize the cpuworker subsystem. It is OK to call this more than once
  65. * during Tor's lifetime.
  66. */
  67. void
  68. cpu_init(void)
  69. {
  70. if (!replyqueue) {
  71. replyqueue = replyqueue_new(0);
  72. }
  73. if (!reply_event) {
  74. reply_event = tor_event_new(tor_libevent_get_base(),
  75. replyqueue_get_socket(replyqueue),
  76. EV_READ|EV_PERSIST,
  77. replyqueue_process_cb,
  78. replyqueue);
  79. event_add(reply_event, NULL);
  80. }
  81. if (!threadpool) {
  82. threadpool = threadpool_new(get_num_cpus(get_options()),
  83. replyqueue,
  84. worker_state_new,
  85. worker_state_free,
  86. NULL);
  87. }
  88. /* Total voodoo. Can we make this more sensible? */
  89. max_pending_tasks = get_num_cpus(get_options()) * 64;
  90. crypto_seed_weak_rng(&request_sample_rng);
  91. }
  92. /** Magic numbers to make sure our cpuworker_requests don't grow any
  93. * mis-framing bugs. */
  94. #define CPUWORKER_REQUEST_MAGIC 0xda4afeed
  95. #define CPUWORKER_REPLY_MAGIC 0x5eedf00d
  96. /** A request sent to a cpuworker. */
  97. typedef struct cpuworker_request_t {
  98. /** Magic number; must be CPUWORKER_REQUEST_MAGIC. */
  99. uint32_t magic;
  100. /** Flag: Are we timing this request? */
  101. unsigned timed : 1;
  102. /** If we're timing this request, when was it sent to the cpuworker? */
  103. struct timeval started_at;
  104. /** A create cell for the cpuworker to process. */
  105. create_cell_t create_cell;
  106. /* Turn the above into a tagged union if needed. */
  107. } cpuworker_request_t;
  108. /** A reply sent by a cpuworker. */
  109. typedef struct cpuworker_reply_t {
  110. /** Magic number; must be CPUWORKER_REPLY_MAGIC. */
  111. uint32_t magic;
  112. /** True iff we got a successful request. */
  113. uint8_t success;
  114. /** Are we timing this request? */
  115. unsigned int timed : 1;
  116. /** What handshake type was the request? (Used for timing) */
  117. uint16_t handshake_type;
  118. /** When did we send the request to the cpuworker? */
  119. struct timeval started_at;
  120. /** Once the cpuworker received the request, how many microseconds did it
  121. * take? (This shouldn't overflow; 4 billion micoseconds is over an hour,
  122. * and we'll never have an onion handshake that takes so long.) */
  123. uint32_t n_usec;
  124. /** Output of processing a create cell
  125. *
  126. * @{
  127. */
  128. /** The created cell to send back. */
  129. created_cell_t created_cell;
  130. /** The keys to use on this circuit. */
  131. uint8_t keys[CPATH_KEY_MATERIAL_LEN];
  132. /** Input to use for authenticating introduce1 cells. */
  133. uint8_t rend_auth_material[DIGEST_LEN];
  134. } cpuworker_reply_t;
  135. typedef struct cpuworker_job_u {
  136. or_circuit_t *circ;
  137. union {
  138. cpuworker_request_t request;
  139. cpuworker_reply_t reply;
  140. } u;
  141. } cpuworker_job_t;
  142. static workqueue_reply_t
  143. update_state_threadfn(void *state_, void *work_)
  144. {
  145. worker_state_t *state = state_;
  146. worker_state_t *update = work_;
  147. server_onion_keys_free(state->onion_keys);
  148. state->onion_keys = update->onion_keys;
  149. update->onion_keys = NULL;
  150. ++state->generation;
  151. return WQ_RPL_REPLY;
  152. }
  153. /** Called when the onion key has changed so update all CPU worker(s) with
  154. * new function pointers with which a new state will be generated.
  155. */
  156. void
  157. cpuworkers_rotate_keyinfo(void)
  158. {
  159. if (!threadpool) {
  160. /* If we're a client, then we won't have cpuworkers, and we won't need
  161. * to tell them to rotate their state.
  162. */
  163. return;
  164. }
  165. if (threadpool_queue_update(threadpool,
  166. worker_state_new,
  167. update_state_threadfn,
  168. worker_state_free,
  169. NULL)) {
  170. log_warn(LD_OR, "Failed to queue key update for worker threads.");
  171. }
  172. }
  173. /** Indexed by handshake type: how many onionskins have we processed and
  174. * counted of that type? */
  175. static uint64_t onionskins_n_processed[MAX_ONION_HANDSHAKE_TYPE+1];
  176. /** Indexed by handshake type, corresponding to the onionskins counted in
  177. * onionskins_n_processed: how many microseconds have we spent in cpuworkers
  178. * processing that kind of onionskin? */
  179. static uint64_t onionskins_usec_internal[MAX_ONION_HANDSHAKE_TYPE+1];
  180. /** Indexed by handshake type, corresponding to onionskins counted in
  181. * onionskins_n_processed: how many microseconds have we spent waiting for
  182. * cpuworkers to give us answers for that kind of onionskin?
  183. */
  184. static uint64_t onionskins_usec_roundtrip[MAX_ONION_HANDSHAKE_TYPE+1];
  185. /** If any onionskin takes longer than this, we clip them to this
  186. * time. (microseconds) */
  187. #define MAX_BELIEVABLE_ONIONSKIN_DELAY (2*1000*1000)
  188. /** Return true iff we'd like to measure a handshake of type
  189. * <b>onionskin_type</b>. Call only from the main thread. */
  190. static int
  191. should_time_request(uint16_t onionskin_type)
  192. {
  193. /* If we've never heard of this type, we shouldn't even be here. */
  194. if (onionskin_type > MAX_ONION_HANDSHAKE_TYPE)
  195. return 0;
  196. /* Measure the first N handshakes of each type, to ensure we have a
  197. * sample */
  198. if (onionskins_n_processed[onionskin_type] < 4096)
  199. return 1;
  200. /** Otherwise, measure with P=1/128. We avoid doing this for every
  201. * handshake, since the measurement itself can take a little time. */
  202. return tor_weak_random_one_in_n(&request_sample_rng, 128);
  203. }
  204. /** Return an estimate of how many microseconds we will need for a single
  205. * cpuworker to to process <b>n_requests</b> onionskins of type
  206. * <b>onionskin_type</b>. */
  207. uint64_t
  208. estimated_usec_for_onionskins(uint32_t n_requests, uint16_t onionskin_type)
  209. {
  210. if (onionskin_type > MAX_ONION_HANDSHAKE_TYPE) /* should be impossible */
  211. return 1000 * (uint64_t)n_requests;
  212. if (PREDICT_UNLIKELY(onionskins_n_processed[onionskin_type] < 100)) {
  213. /* Until we have 100 data points, just asssume everything takes 1 msec. */
  214. return 1000 * (uint64_t)n_requests;
  215. } else {
  216. /* This can't overflow: we'll never have more than 500000 onionskins
  217. * measured in onionskin_usec_internal, and they won't take anything near
  218. * 1 sec each, and we won't have anything like 1 million queued
  219. * onionskins. But that's 5e5 * 1e6 * 1e6, which is still less than
  220. * UINT64_MAX. */
  221. return (onionskins_usec_internal[onionskin_type] * n_requests) /
  222. onionskins_n_processed[onionskin_type];
  223. }
  224. }
  225. /** Compute the absolute and relative overhead of using the cpuworker
  226. * framework for onionskins of type <b>onionskin_type</b>.*/
  227. static int
  228. get_overhead_for_onionskins(uint32_t *usec_out, double *frac_out,
  229. uint16_t onionskin_type)
  230. {
  231. uint64_t overhead;
  232. *usec_out = 0;
  233. *frac_out = 0.0;
  234. if (onionskin_type > MAX_ONION_HANDSHAKE_TYPE) /* should be impossible */
  235. return -1;
  236. if (onionskins_n_processed[onionskin_type] == 0 ||
  237. onionskins_usec_internal[onionskin_type] == 0 ||
  238. onionskins_usec_roundtrip[onionskin_type] == 0)
  239. return -1;
  240. overhead = onionskins_usec_roundtrip[onionskin_type] -
  241. onionskins_usec_internal[onionskin_type];
  242. *usec_out = (uint32_t)(overhead / onionskins_n_processed[onionskin_type]);
  243. *frac_out = U64_TO_DBL(overhead) / onionskins_usec_internal[onionskin_type];
  244. return 0;
  245. }
  246. /** If we've measured overhead for onionskins of type <b>onionskin_type</b>,
  247. * log it. */
  248. void
  249. cpuworker_log_onionskin_overhead(int severity, int onionskin_type,
  250. const char *onionskin_type_name)
  251. {
  252. uint32_t overhead;
  253. double relative_overhead;
  254. int r;
  255. r = get_overhead_for_onionskins(&overhead, &relative_overhead,
  256. onionskin_type);
  257. if (!overhead || r<0)
  258. return;
  259. log_fn(severity, LD_OR,
  260. "%s onionskins have averaged %u usec overhead (%.2f%%) in "
  261. "cpuworker code ",
  262. onionskin_type_name, (unsigned)overhead, relative_overhead*100);
  263. }
  264. /** Handle a reply from the worker threads. */
  265. static void
  266. cpuworker_onion_handshake_replyfn(void *work_)
  267. {
  268. cpuworker_job_t *job = work_;
  269. cpuworker_reply_t rpl;
  270. or_circuit_t *circ = NULL;
  271. tor_assert(total_pending_tasks > 0);
  272. --total_pending_tasks;
  273. /* Could avoid this, but doesn't matter. */
  274. memcpy(&rpl, &job->u.reply, sizeof(rpl));
  275. tor_assert(rpl.magic == CPUWORKER_REPLY_MAGIC);
  276. if (rpl.timed && rpl.success &&
  277. rpl.handshake_type <= MAX_ONION_HANDSHAKE_TYPE) {
  278. /* Time how long this request took. The handshake_type check should be
  279. needless, but let's leave it in to be safe. */
  280. struct timeval tv_end, tv_diff;
  281. int64_t usec_roundtrip;
  282. tor_gettimeofday(&tv_end);
  283. timersub(&tv_end, &rpl.started_at, &tv_diff);
  284. usec_roundtrip = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec;
  285. if (usec_roundtrip >= 0 &&
  286. usec_roundtrip < MAX_BELIEVABLE_ONIONSKIN_DELAY) {
  287. ++onionskins_n_processed[rpl.handshake_type];
  288. onionskins_usec_internal[rpl.handshake_type] += rpl.n_usec;
  289. onionskins_usec_roundtrip[rpl.handshake_type] += usec_roundtrip;
  290. if (onionskins_n_processed[rpl.handshake_type] >= 500000) {
  291. /* Scale down every 500000 handshakes. On a busy server, that's
  292. * less impressive than it sounds. */
  293. onionskins_n_processed[rpl.handshake_type] /= 2;
  294. onionskins_usec_internal[rpl.handshake_type] /= 2;
  295. onionskins_usec_roundtrip[rpl.handshake_type] /= 2;
  296. }
  297. }
  298. }
  299. circ = job->circ;
  300. log_debug(LD_OR,
  301. "Unpacking cpuworker reply %p, circ=%p, success=%d",
  302. job, circ, rpl.success);
  303. if (circ->base_.magic == DEAD_CIRCUIT_MAGIC) {
  304. /* The circuit was supposed to get freed while the reply was
  305. * pending. Instead, it got left for us to free so that we wouldn't freak
  306. * out when the job->circ field wound up pointing to nothing. */
  307. log_debug(LD_OR, "Circuit died while reply was pending. Freeing memory.");
  308. circ->base_.magic = 0;
  309. tor_free(circ);
  310. goto done_processing;
  311. }
  312. circ->workqueue_entry = NULL;
  313. if (TO_CIRCUIT(circ)->marked_for_close) {
  314. /* We already marked this circuit; we can't call it open. */
  315. log_debug(LD_OR,"circuit is already marked.");
  316. goto done_processing;
  317. }
  318. if (rpl.success == 0) {
  319. log_debug(LD_OR,
  320. "decoding onionskin failed. "
  321. "(Old key or bad software.) Closing.");
  322. circuit_mark_for_close(TO_CIRCUIT(circ), END_CIRC_REASON_TORPROTOCOL);
  323. goto done_processing;
  324. }
  325. if (onionskin_answer(circ,
  326. &rpl.created_cell,
  327. (const char*)rpl.keys,
  328. rpl.rend_auth_material) < 0) {
  329. log_warn(LD_OR,"onionskin_answer failed. Closing.");
  330. circuit_mark_for_close(TO_CIRCUIT(circ), END_CIRC_REASON_INTERNAL);
  331. goto done_processing;
  332. }
  333. log_debug(LD_OR,"onionskin_answer succeeded. Yay.");
  334. done_processing:
  335. memwipe(&rpl, 0, sizeof(rpl));
  336. memwipe(job, 0, sizeof(*job));
  337. tor_free(job);
  338. queue_pending_tasks();
  339. }
  340. /** Implementation function for onion handshake requests. */
  341. static workqueue_reply_t
  342. cpuworker_onion_handshake_threadfn(void *state_, void *work_)
  343. {
  344. worker_state_t *state = state_;
  345. cpuworker_job_t *job = work_;
  346. /* variables for onion processing */
  347. server_onion_keys_t *onion_keys = state->onion_keys;
  348. cpuworker_request_t req;
  349. cpuworker_reply_t rpl;
  350. memcpy(&req, &job->u.request, sizeof(req));
  351. tor_assert(req.magic == CPUWORKER_REQUEST_MAGIC);
  352. memset(&rpl, 0, sizeof(rpl));
  353. const create_cell_t *cc = &req.create_cell;
  354. created_cell_t *cell_out = &rpl.created_cell;
  355. struct timeval tv_start = {0,0}, tv_end;
  356. int n;
  357. rpl.timed = req.timed;
  358. rpl.started_at = req.started_at;
  359. rpl.handshake_type = cc->handshake_type;
  360. if (req.timed)
  361. tor_gettimeofday(&tv_start);
  362. n = onion_skin_server_handshake(cc->handshake_type,
  363. cc->onionskin, cc->handshake_len,
  364. onion_keys,
  365. cell_out->reply,
  366. rpl.keys, CPATH_KEY_MATERIAL_LEN,
  367. rpl.rend_auth_material);
  368. if (n < 0) {
  369. /* failure */
  370. log_debug(LD_OR,"onion_skin_server_handshake failed.");
  371. memset(&rpl, 0, sizeof(rpl));
  372. rpl.success = 0;
  373. } else {
  374. /* success */
  375. log_debug(LD_OR,"onion_skin_server_handshake succeeded.");
  376. cell_out->handshake_len = n;
  377. switch (cc->cell_type) {
  378. case CELL_CREATE:
  379. cell_out->cell_type = CELL_CREATED; break;
  380. case CELL_CREATE2:
  381. cell_out->cell_type = CELL_CREATED2; break;
  382. case CELL_CREATE_FAST:
  383. cell_out->cell_type = CELL_CREATED_FAST; break;
  384. default:
  385. tor_assert(0);
  386. return WQ_RPL_SHUTDOWN;
  387. }
  388. rpl.success = 1;
  389. }
  390. rpl.magic = CPUWORKER_REPLY_MAGIC;
  391. if (req.timed) {
  392. struct timeval tv_diff;
  393. int64_t usec;
  394. tor_gettimeofday(&tv_end);
  395. timersub(&tv_end, &tv_start, &tv_diff);
  396. usec = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec;
  397. if (usec < 0 || usec > MAX_BELIEVABLE_ONIONSKIN_DELAY)
  398. rpl.n_usec = MAX_BELIEVABLE_ONIONSKIN_DELAY;
  399. else
  400. rpl.n_usec = (uint32_t) usec;
  401. }
  402. memcpy(&job->u.reply, &rpl, sizeof(rpl));
  403. memwipe(&req, 0, sizeof(req));
  404. memwipe(&rpl, 0, sizeof(req));
  405. return WQ_RPL_REPLY;
  406. }
  407. /** Take pending tasks from the queue and assign them to cpuworkers. */
  408. static void
  409. queue_pending_tasks(void)
  410. {
  411. or_circuit_t *circ;
  412. create_cell_t *onionskin = NULL;
  413. while (total_pending_tasks < max_pending_tasks) {
  414. circ = onion_next_task(&onionskin);
  415. if (!circ)
  416. return;
  417. if (assign_onionskin_to_cpuworker(circ, onionskin))
  418. log_warn(LD_OR,"assign_to_cpuworker failed. Ignoring.");
  419. }
  420. }
  421. /** Try to tell a cpuworker to perform the public key operations necessary to
  422. * respond to <b>onionskin</b> for the circuit <b>circ</b>.
  423. *
  424. * Return 0 if we successfully assign the task, or -1 on failure.
  425. */
  426. int
  427. assign_onionskin_to_cpuworker(or_circuit_t *circ,
  428. create_cell_t *onionskin)
  429. {
  430. workqueue_entry_t *queue_entry;
  431. cpuworker_job_t *job;
  432. cpuworker_request_t req;
  433. int should_time;
  434. tor_assert(threadpool);
  435. if (!circ->p_chan) {
  436. log_info(LD_OR,"circ->p_chan gone. Failing circ.");
  437. tor_free(onionskin);
  438. return -1;
  439. }
  440. if (total_pending_tasks >= max_pending_tasks) {
  441. log_debug(LD_OR,"No idle cpuworkers. Queuing.");
  442. if (onion_pending_add(circ, onionskin) < 0) {
  443. tor_free(onionskin);
  444. return -1;
  445. }
  446. return 0;
  447. }
  448. if (connection_or_digest_is_known_relay(circ->p_chan->identity_digest))
  449. rep_hist_note_circuit_handshake_assigned(onionskin->handshake_type);
  450. should_time = should_time_request(onionskin->handshake_type);
  451. memset(&req, 0, sizeof(req));
  452. req.magic = CPUWORKER_REQUEST_MAGIC;
  453. req.timed = should_time;
  454. memcpy(&req.create_cell, onionskin, sizeof(create_cell_t));
  455. tor_free(onionskin);
  456. if (should_time)
  457. tor_gettimeofday(&req.started_at);
  458. job = tor_malloc_zero(sizeof(cpuworker_job_t));
  459. job->circ = circ;
  460. memcpy(&job->u.request, &req, sizeof(req));
  461. memwipe(&req, 0, sizeof(req));
  462. ++total_pending_tasks;
  463. queue_entry = threadpool_queue_work(threadpool,
  464. cpuworker_onion_handshake_threadfn,
  465. cpuworker_onion_handshake_replyfn,
  466. job);
  467. if (!queue_entry) {
  468. log_warn(LD_BUG, "Couldn't queue work on threadpool");
  469. tor_free(job);
  470. return -1;
  471. }
  472. log_debug(LD_OR, "Queued task %p (qe=%p, circ=%p)",
  473. job, queue_entry, job->circ);
  474. circ->workqueue_entry = queue_entry;
  475. return 0;
  476. }
  477. /** If <b>circ</b> has a pending handshake that hasn't been processed yet,
  478. * remove it from the worker queue. */
  479. void
  480. cpuworker_cancel_circ_handshake(or_circuit_t *circ)
  481. {
  482. cpuworker_job_t *job;
  483. if (circ->workqueue_entry == NULL)
  484. return;
  485. job = workqueue_entry_cancel(circ->workqueue_entry);
  486. if (job) {
  487. /* It successfully cancelled. */
  488. memwipe(job, 0xe0, sizeof(*job));
  489. tor_free(job);
  490. tor_assert(total_pending_tasks > 0);
  491. --total_pending_tasks;
  492. /* if (!job), this is done in cpuworker_onion_handshake_replyfn. */
  493. circ->workqueue_entry = NULL;
  494. }
  495. }