cpuworker.c 19 KB

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