cpuworker.c 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572
  1. /* Copyright (c) 2003-2004, Roger Dingledine.
  2. * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
  3. * Copyright (c) 2007-2016, 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. * The multithreading backend for this module is in workqueue.c; this module
  11. * specializes workqueue.c.
  12. *
  13. * Right now, we only use this for processing onionskins, and invoke it mostly
  14. * from onion.c.
  15. **/
  16. #include "or.h"
  17. #include "channel.h"
  18. #include "circuitbuild.h"
  19. #include "circuitlist.h"
  20. #include "connection_or.h"
  21. #include "config.h"
  22. #include "cpuworker.h"
  23. #include "main.h"
  24. #include "onion.h"
  25. #include "rephist.h"
  26. #include "router.h"
  27. #include "workqueue.h"
  28. #include <event2/event.h>
  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. worker_state_free(update);
  151. ++state->generation;
  152. return WQ_RPL_REPLY;
  153. }
  154. /** Called when the onion key has changed so update all CPU worker(s) with
  155. * new function pointers with which a new state will be generated.
  156. */
  157. void
  158. cpuworkers_rotate_keyinfo(void)
  159. {
  160. if (!threadpool) {
  161. /* If we're a client, then we won't have cpuworkers, and we won't need
  162. * to tell them to rotate their state.
  163. */
  164. return;
  165. }
  166. if (threadpool_queue_update(threadpool,
  167. worker_state_new,
  168. update_state_threadfn,
  169. worker_state_free,
  170. NULL)) {
  171. log_warn(LD_OR, "Failed to queue key update for worker threads.");
  172. }
  173. }
  174. /** Indexed by handshake type: how many onionskins have we processed and
  175. * counted of that type? */
  176. static uint64_t onionskins_n_processed[MAX_ONION_HANDSHAKE_TYPE+1];
  177. /** Indexed by handshake type, corresponding to the onionskins counted in
  178. * onionskins_n_processed: how many microseconds have we spent in cpuworkers
  179. * processing that kind of onionskin? */
  180. static uint64_t onionskins_usec_internal[MAX_ONION_HANDSHAKE_TYPE+1];
  181. /** Indexed by handshake type, corresponding to onionskins counted in
  182. * onionskins_n_processed: how many microseconds have we spent waiting for
  183. * cpuworkers to give us answers for that kind of onionskin?
  184. */
  185. static uint64_t onionskins_usec_roundtrip[MAX_ONION_HANDSHAKE_TYPE+1];
  186. /** If any onionskin takes longer than this, we clip them to this
  187. * time. (microseconds) */
  188. #define MAX_BELIEVABLE_ONIONSKIN_DELAY (2*1000*1000)
  189. /** Return true iff we'd like to measure a handshake of type
  190. * <b>onionskin_type</b>. Call only from the main thread. */
  191. static int
  192. should_time_request(uint16_t onionskin_type)
  193. {
  194. /* If we've never heard of this type, we shouldn't even be here. */
  195. if (onionskin_type > MAX_ONION_HANDSHAKE_TYPE)
  196. return 0;
  197. /* Measure the first N handshakes of each type, to ensure we have a
  198. * sample */
  199. if (onionskins_n_processed[onionskin_type] < 4096)
  200. return 1;
  201. /** Otherwise, measure with P=1/128. We avoid doing this for every
  202. * handshake, since the measurement itself can take a little time. */
  203. return tor_weak_random_one_in_n(&request_sample_rng, 128);
  204. }
  205. /** Return an estimate of how many microseconds we will need for a single
  206. * cpuworker to to process <b>n_requests</b> onionskins of type
  207. * <b>onionskin_type</b>. */
  208. uint64_t
  209. estimated_usec_for_onionskins(uint32_t n_requests, uint16_t onionskin_type)
  210. {
  211. if (onionskin_type > MAX_ONION_HANDSHAKE_TYPE) /* should be impossible */
  212. return 1000 * (uint64_t)n_requests;
  213. if (PREDICT_UNLIKELY(onionskins_n_processed[onionskin_type] < 100)) {
  214. /* Until we have 100 data points, just asssume everything takes 1 msec. */
  215. return 1000 * (uint64_t)n_requests;
  216. } else {
  217. /* This can't overflow: we'll never have more than 500000 onionskins
  218. * measured in onionskin_usec_internal, and they won't take anything near
  219. * 1 sec each, and we won't have anything like 1 million queued
  220. * onionskins. But that's 5e5 * 1e6 * 1e6, which is still less than
  221. * UINT64_MAX. */
  222. return (onionskins_usec_internal[onionskin_type] * n_requests) /
  223. onionskins_n_processed[onionskin_type];
  224. }
  225. }
  226. /** Compute the absolute and relative overhead of using the cpuworker
  227. * framework for onionskins of type <b>onionskin_type</b>.*/
  228. static int
  229. get_overhead_for_onionskins(uint32_t *usec_out, double *frac_out,
  230. uint16_t onionskin_type)
  231. {
  232. uint64_t overhead;
  233. *usec_out = 0;
  234. *frac_out = 0.0;
  235. if (onionskin_type > MAX_ONION_HANDSHAKE_TYPE) /* should be impossible */
  236. return -1;
  237. if (onionskins_n_processed[onionskin_type] == 0 ||
  238. onionskins_usec_internal[onionskin_type] == 0 ||
  239. onionskins_usec_roundtrip[onionskin_type] == 0)
  240. return -1;
  241. overhead = onionskins_usec_roundtrip[onionskin_type] -
  242. onionskins_usec_internal[onionskin_type];
  243. *usec_out = (uint32_t)(overhead / onionskins_n_processed[onionskin_type]);
  244. *frac_out = U64_TO_DBL(overhead) / onionskins_usec_internal[onionskin_type];
  245. return 0;
  246. }
  247. /** If we've measured overhead for onionskins of type <b>onionskin_type</b>,
  248. * log it. */
  249. void
  250. cpuworker_log_onionskin_overhead(int severity, int onionskin_type,
  251. const char *onionskin_type_name)
  252. {
  253. uint32_t overhead;
  254. double relative_overhead;
  255. int r;
  256. r = get_overhead_for_onionskins(&overhead, &relative_overhead,
  257. onionskin_type);
  258. if (!overhead || r<0)
  259. return;
  260. log_fn(severity, LD_OR,
  261. "%s onionskins have averaged %u usec overhead (%.2f%%) in "
  262. "cpuworker code ",
  263. onionskin_type_name, (unsigned)overhead, relative_overhead*100);
  264. }
  265. /** Handle a reply from the worker threads. */
  266. static void
  267. cpuworker_onion_handshake_replyfn(void *work_)
  268. {
  269. cpuworker_job_t *job = work_;
  270. cpuworker_reply_t rpl;
  271. or_circuit_t *circ = NULL;
  272. tor_assert(total_pending_tasks > 0);
  273. --total_pending_tasks;
  274. /* Could avoid this, but doesn't matter. */
  275. memcpy(&rpl, &job->u.reply, sizeof(rpl));
  276. tor_assert(rpl.magic == CPUWORKER_REPLY_MAGIC);
  277. if (rpl.timed && rpl.success &&
  278. rpl.handshake_type <= MAX_ONION_HANDSHAKE_TYPE) {
  279. /* Time how long this request took. The handshake_type check should be
  280. needless, but let's leave it in to be safe. */
  281. struct timeval tv_end, tv_diff;
  282. int64_t usec_roundtrip;
  283. tor_gettimeofday(&tv_end);
  284. timersub(&tv_end, &rpl.started_at, &tv_diff);
  285. usec_roundtrip = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec;
  286. if (usec_roundtrip >= 0 &&
  287. usec_roundtrip < MAX_BELIEVABLE_ONIONSKIN_DELAY) {
  288. ++onionskins_n_processed[rpl.handshake_type];
  289. onionskins_usec_internal[rpl.handshake_type] += rpl.n_usec;
  290. onionskins_usec_roundtrip[rpl.handshake_type] += usec_roundtrip;
  291. if (onionskins_n_processed[rpl.handshake_type] >= 500000) {
  292. /* Scale down every 500000 handshakes. On a busy server, that's
  293. * less impressive than it sounds. */
  294. onionskins_n_processed[rpl.handshake_type] /= 2;
  295. onionskins_usec_internal[rpl.handshake_type] /= 2;
  296. onionskins_usec_roundtrip[rpl.handshake_type] /= 2;
  297. }
  298. }
  299. }
  300. circ = job->circ;
  301. log_debug(LD_OR,
  302. "Unpacking cpuworker reply %p, circ=%p, success=%d",
  303. job, circ, rpl.success);
  304. if (circ->base_.magic == DEAD_CIRCUIT_MAGIC) {
  305. /* The circuit was supposed to get freed while the reply was
  306. * pending. Instead, it got left for us to free so that we wouldn't freak
  307. * out when the job->circ field wound up pointing to nothing. */
  308. log_debug(LD_OR, "Circuit died while reply was pending. Freeing memory.");
  309. circ->base_.magic = 0;
  310. tor_free(circ);
  311. goto done_processing;
  312. }
  313. circ->workqueue_entry = NULL;
  314. if (TO_CIRCUIT(circ)->marked_for_close) {
  315. /* We already marked this circuit; we can't call it open. */
  316. log_debug(LD_OR,"circuit is already marked.");
  317. goto done_processing;
  318. }
  319. if (rpl.success == 0) {
  320. log_debug(LD_OR,
  321. "decoding onionskin failed. "
  322. "(Old key or bad software.) Closing.");
  323. circuit_mark_for_close(TO_CIRCUIT(circ), END_CIRC_REASON_TORPROTOCOL);
  324. goto done_processing;
  325. }
  326. if (onionskin_answer(circ,
  327. &rpl.created_cell,
  328. (const char*)rpl.keys,
  329. rpl.rend_auth_material) < 0) {
  330. log_warn(LD_OR,"onionskin_answer failed. Closing.");
  331. circuit_mark_for_close(TO_CIRCUIT(circ), END_CIRC_REASON_INTERNAL);
  332. goto done_processing;
  333. }
  334. log_debug(LD_OR,"onionskin_answer succeeded. Yay.");
  335. done_processing:
  336. memwipe(&rpl, 0, sizeof(rpl));
  337. memwipe(job, 0, sizeof(*job));
  338. tor_free(job);
  339. queue_pending_tasks();
  340. }
  341. /** Implementation function for onion handshake requests. */
  342. static workqueue_reply_t
  343. cpuworker_onion_handshake_threadfn(void *state_, void *work_)
  344. {
  345. worker_state_t *state = state_;
  346. cpuworker_job_t *job = work_;
  347. /* variables for onion processing */
  348. server_onion_keys_t *onion_keys = state->onion_keys;
  349. cpuworker_request_t req;
  350. cpuworker_reply_t rpl;
  351. memcpy(&req, &job->u.request, sizeof(req));
  352. tor_assert(req.magic == CPUWORKER_REQUEST_MAGIC);
  353. memset(&rpl, 0, sizeof(rpl));
  354. const create_cell_t *cc = &req.create_cell;
  355. created_cell_t *cell_out = &rpl.created_cell;
  356. struct timeval tv_start = {0,0}, tv_end;
  357. int n;
  358. rpl.timed = req.timed;
  359. rpl.started_at = req.started_at;
  360. rpl.handshake_type = cc->handshake_type;
  361. if (req.timed)
  362. tor_gettimeofday(&tv_start);
  363. n = onion_skin_server_handshake(cc->handshake_type,
  364. cc->onionskin, cc->handshake_len,
  365. onion_keys,
  366. cell_out->reply,
  367. rpl.keys, CPATH_KEY_MATERIAL_LEN,
  368. rpl.rend_auth_material);
  369. if (n < 0) {
  370. /* failure */
  371. log_debug(LD_OR,"onion_skin_server_handshake failed.");
  372. memset(&rpl, 0, sizeof(rpl));
  373. rpl.success = 0;
  374. } else {
  375. /* success */
  376. log_debug(LD_OR,"onion_skin_server_handshake succeeded.");
  377. cell_out->handshake_len = n;
  378. switch (cc->cell_type) {
  379. case CELL_CREATE:
  380. cell_out->cell_type = CELL_CREATED; break;
  381. case CELL_CREATE2:
  382. cell_out->cell_type = CELL_CREATED2; break;
  383. case CELL_CREATE_FAST:
  384. cell_out->cell_type = CELL_CREATED_FAST; break;
  385. default:
  386. tor_assert(0);
  387. return WQ_RPL_SHUTDOWN;
  388. }
  389. rpl.success = 1;
  390. }
  391. rpl.magic = CPUWORKER_REPLY_MAGIC;
  392. if (req.timed) {
  393. struct timeval tv_diff;
  394. int64_t usec;
  395. tor_gettimeofday(&tv_end);
  396. timersub(&tv_end, &tv_start, &tv_diff);
  397. usec = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec;
  398. if (usec < 0 || usec > MAX_BELIEVABLE_ONIONSKIN_DELAY)
  399. rpl.n_usec = MAX_BELIEVABLE_ONIONSKIN_DELAY;
  400. else
  401. rpl.n_usec = (uint32_t) usec;
  402. }
  403. memcpy(&job->u.reply, &rpl, sizeof(rpl));
  404. memwipe(&req, 0, sizeof(req));
  405. memwipe(&rpl, 0, sizeof(req));
  406. return WQ_RPL_REPLY;
  407. }
  408. /** Take pending tasks from the queue and assign them to cpuworkers. */
  409. static void
  410. queue_pending_tasks(void)
  411. {
  412. or_circuit_t *circ;
  413. create_cell_t *onionskin = NULL;
  414. while (total_pending_tasks < max_pending_tasks) {
  415. circ = onion_next_task(&onionskin);
  416. if (!circ)
  417. return;
  418. if (assign_onionskin_to_cpuworker(circ, onionskin))
  419. log_warn(LD_OR,"assign_to_cpuworker failed. Ignoring.");
  420. }
  421. }
  422. /** Try to tell a cpuworker to perform the public key operations necessary to
  423. * respond to <b>onionskin</b> for the circuit <b>circ</b>.
  424. *
  425. * Return 0 if we successfully assign the task, or -1 on failure.
  426. */
  427. int
  428. assign_onionskin_to_cpuworker(or_circuit_t *circ,
  429. create_cell_t *onionskin)
  430. {
  431. workqueue_entry_t *queue_entry;
  432. cpuworker_job_t *job;
  433. cpuworker_request_t req;
  434. int should_time;
  435. tor_assert(threadpool);
  436. if (!circ->p_chan) {
  437. log_info(LD_OR,"circ->p_chan gone. Failing circ.");
  438. tor_free(onionskin);
  439. return -1;
  440. }
  441. if (total_pending_tasks >= max_pending_tasks) {
  442. log_debug(LD_OR,"No idle cpuworkers. Queuing.");
  443. if (onion_pending_add(circ, onionskin) < 0) {
  444. tor_free(onionskin);
  445. return -1;
  446. }
  447. return 0;
  448. }
  449. if (connection_or_digest_is_known_relay(circ->p_chan->identity_digest))
  450. rep_hist_note_circuit_handshake_assigned(onionskin->handshake_type);
  451. should_time = should_time_request(onionskin->handshake_type);
  452. memset(&req, 0, sizeof(req));
  453. req.magic = CPUWORKER_REQUEST_MAGIC;
  454. req.timed = should_time;
  455. memcpy(&req.create_cell, onionskin, sizeof(create_cell_t));
  456. tor_free(onionskin);
  457. if (should_time)
  458. tor_gettimeofday(&req.started_at);
  459. job = tor_malloc_zero(sizeof(cpuworker_job_t));
  460. job->circ = circ;
  461. memcpy(&job->u.request, &req, sizeof(req));
  462. memwipe(&req, 0, sizeof(req));
  463. ++total_pending_tasks;
  464. queue_entry = threadpool_queue_work(threadpool,
  465. cpuworker_onion_handshake_threadfn,
  466. cpuworker_onion_handshake_replyfn,
  467. job);
  468. if (!queue_entry) {
  469. log_warn(LD_BUG, "Couldn't queue work on threadpool");
  470. tor_free(job);
  471. return -1;
  472. }
  473. log_debug(LD_OR, "Queued task %p (qe=%p, circ=%p)",
  474. job, queue_entry, job->circ);
  475. circ->workqueue_entry = queue_entry;
  476. return 0;
  477. }
  478. /** If <b>circ</b> has a pending handshake that hasn't been processed yet,
  479. * remove it from the worker queue. */
  480. void
  481. cpuworker_cancel_circ_handshake(or_circuit_t *circ)
  482. {
  483. cpuworker_job_t *job;
  484. if (circ->workqueue_entry == NULL)
  485. return;
  486. job = workqueue_entry_cancel(circ->workqueue_entry);
  487. if (job) {
  488. /* It successfully cancelled. */
  489. memwipe(job, 0xe0, sizeof(*job));
  490. tor_free(job);
  491. tor_assert(total_pending_tasks > 0);
  492. --total_pending_tasks;
  493. /* if (!job), this is done in cpuworker_onion_handshake_replyfn. */
  494. circ->workqueue_entry = NULL;
  495. }
  496. }