You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

job.c 24KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989
  1. /*
  2. * Background jobs (long-running operations)
  3. *
  4. * Copyright (c) 2011 IBM Corp.
  5. * Copyright (c) 2012, 2018 Red Hat, Inc.
  6. *
  7. * Permission is hereby granted, free of charge, to any person obtaining a copy
  8. * of this software and associated documentation files (the "Software"), to deal
  9. * in the Software without restriction, including without limitation the rights
  10. * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11. * copies of the Software, and to permit persons to whom the Software is
  12. * furnished to do so, subject to the following conditions:
  13. *
  14. * The above copyright notice and this permission notice shall be included in
  15. * all copies or substantial portions of the Software.
  16. *
  17. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18. * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19. * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  20. * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21. * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22. * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  23. * THE SOFTWARE.
  24. */
  25. #include "qemu/osdep.h"
  26. #include "qapi/error.h"
  27. #include "qemu/job.h"
  28. #include "qemu/id.h"
  29. #include "qemu/main-loop.h"
  30. #include "block/aio-wait.h"
  31. #include "trace-root.h"
  32. #include "qapi/qapi-events-job.h"
  33. static QLIST_HEAD(, Job) jobs = QLIST_HEAD_INITIALIZER(jobs);
  34. /* Job State Transition Table */
  35. bool JobSTT[JOB_STATUS__MAX][JOB_STATUS__MAX] = {
  36. /* U, C, R, P, Y, S, W, D, X, E, N */
  37. /* U: */ [JOB_STATUS_UNDEFINED] = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0},
  38. /* C: */ [JOB_STATUS_CREATED] = {0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1},
  39. /* R: */ [JOB_STATUS_RUNNING] = {0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0},
  40. /* P: */ [JOB_STATUS_PAUSED] = {0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0},
  41. /* Y: */ [JOB_STATUS_READY] = {0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0},
  42. /* S: */ [JOB_STATUS_STANDBY] = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
  43. /* W: */ [JOB_STATUS_WAITING] = {0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0},
  44. /* D: */ [JOB_STATUS_PENDING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
  45. /* X: */ [JOB_STATUS_ABORTING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
  46. /* E: */ [JOB_STATUS_CONCLUDED] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1},
  47. /* N: */ [JOB_STATUS_NULL] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
  48. };
  49. bool JobVerbTable[JOB_VERB__MAX][JOB_STATUS__MAX] = {
  50. /* U, C, R, P, Y, S, W, D, X, E, N */
  51. [JOB_VERB_CANCEL] = {0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0},
  52. [JOB_VERB_PAUSE] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  53. [JOB_VERB_RESUME] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  54. [JOB_VERB_SET_SPEED] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  55. [JOB_VERB_COMPLETE] = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
  56. [JOB_VERB_FINALIZE] = {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0},
  57. [JOB_VERB_DISMISS] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0},
  58. };
  59. /* Transactional group of jobs */
  60. struct JobTxn {
  61. /* Is this txn being cancelled? */
  62. bool aborting;
  63. /* List of jobs */
  64. QLIST_HEAD(, Job) jobs;
  65. /* Reference count */
  66. int refcnt;
  67. };
  68. /* Right now, this mutex is only needed to synchronize accesses to job->busy
  69. * and job->sleep_timer, such as concurrent calls to job_do_yield and
  70. * job_enter. */
  71. static QemuMutex job_mutex;
  72. static void job_lock(void)
  73. {
  74. qemu_mutex_lock(&job_mutex);
  75. }
  76. static void job_unlock(void)
  77. {
  78. qemu_mutex_unlock(&job_mutex);
  79. }
  80. static void __attribute__((__constructor__)) job_init(void)
  81. {
  82. qemu_mutex_init(&job_mutex);
  83. }
  84. JobTxn *job_txn_new(void)
  85. {
  86. JobTxn *txn = g_new0(JobTxn, 1);
  87. QLIST_INIT(&txn->jobs);
  88. txn->refcnt = 1;
  89. return txn;
  90. }
  91. static void job_txn_ref(JobTxn *txn)
  92. {
  93. txn->refcnt++;
  94. }
  95. void job_txn_unref(JobTxn *txn)
  96. {
  97. if (txn && --txn->refcnt == 0) {
  98. g_free(txn);
  99. }
  100. }
  101. void job_txn_add_job(JobTxn *txn, Job *job)
  102. {
  103. if (!txn) {
  104. return;
  105. }
  106. assert(!job->txn);
  107. job->txn = txn;
  108. QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
  109. job_txn_ref(txn);
  110. }
  111. static void job_txn_del_job(Job *job)
  112. {
  113. if (job->txn) {
  114. QLIST_REMOVE(job, txn_list);
  115. job_txn_unref(job->txn);
  116. job->txn = NULL;
  117. }
  118. }
  119. static int job_txn_apply(JobTxn *txn, int fn(Job *))
  120. {
  121. Job *job, *next;
  122. int rc = 0;
  123. QLIST_FOREACH_SAFE(job, &txn->jobs, txn_list, next) {
  124. rc = fn(job);
  125. if (rc) {
  126. break;
  127. }
  128. }
  129. return rc;
  130. }
  131. bool job_is_internal(Job *job)
  132. {
  133. return (job->id == NULL);
  134. }
  135. static void job_state_transition(Job *job, JobStatus s1)
  136. {
  137. JobStatus s0 = job->status;
  138. assert(s1 >= 0 && s1 < JOB_STATUS__MAX);
  139. trace_job_state_transition(job, job->ret,
  140. JobSTT[s0][s1] ? "allowed" : "disallowed",
  141. JobStatus_str(s0), JobStatus_str(s1));
  142. assert(JobSTT[s0][s1]);
  143. job->status = s1;
  144. if (!job_is_internal(job) && s1 != s0) {
  145. qapi_event_send_job_status_change(job->id, job->status);
  146. }
  147. }
  148. int job_apply_verb(Job *job, JobVerb verb, Error **errp)
  149. {
  150. JobStatus s0 = job->status;
  151. assert(verb >= 0 && verb < JOB_VERB__MAX);
  152. trace_job_apply_verb(job, JobStatus_str(s0), JobVerb_str(verb),
  153. JobVerbTable[verb][s0] ? "allowed" : "prohibited");
  154. if (JobVerbTable[verb][s0]) {
  155. return 0;
  156. }
  157. error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'",
  158. job->id, JobStatus_str(s0), JobVerb_str(verb));
  159. return -EPERM;
  160. }
  161. JobType job_type(const Job *job)
  162. {
  163. return job->driver->job_type;
  164. }
  165. const char *job_type_str(const Job *job)
  166. {
  167. return JobType_str(job_type(job));
  168. }
  169. bool job_is_cancelled(Job *job)
  170. {
  171. return job->cancelled;
  172. }
  173. bool job_is_ready(Job *job)
  174. {
  175. switch (job->status) {
  176. case JOB_STATUS_UNDEFINED:
  177. case JOB_STATUS_CREATED:
  178. case JOB_STATUS_RUNNING:
  179. case JOB_STATUS_PAUSED:
  180. case JOB_STATUS_WAITING:
  181. case JOB_STATUS_PENDING:
  182. case JOB_STATUS_ABORTING:
  183. case JOB_STATUS_CONCLUDED:
  184. case JOB_STATUS_NULL:
  185. return false;
  186. case JOB_STATUS_READY:
  187. case JOB_STATUS_STANDBY:
  188. return true;
  189. default:
  190. g_assert_not_reached();
  191. }
  192. return false;
  193. }
  194. bool job_is_completed(Job *job)
  195. {
  196. switch (job->status) {
  197. case JOB_STATUS_UNDEFINED:
  198. case JOB_STATUS_CREATED:
  199. case JOB_STATUS_RUNNING:
  200. case JOB_STATUS_PAUSED:
  201. case JOB_STATUS_READY:
  202. case JOB_STATUS_STANDBY:
  203. return false;
  204. case JOB_STATUS_WAITING:
  205. case JOB_STATUS_PENDING:
  206. case JOB_STATUS_ABORTING:
  207. case JOB_STATUS_CONCLUDED:
  208. case JOB_STATUS_NULL:
  209. return true;
  210. default:
  211. g_assert_not_reached();
  212. }
  213. return false;
  214. }
  215. static bool job_started(Job *job)
  216. {
  217. return job->co;
  218. }
  219. static bool job_should_pause(Job *job)
  220. {
  221. return job->pause_count > 0;
  222. }
  223. Job *job_next(Job *job)
  224. {
  225. if (!job) {
  226. return QLIST_FIRST(&jobs);
  227. }
  228. return QLIST_NEXT(job, job_list);
  229. }
  230. Job *job_get(const char *id)
  231. {
  232. Job *job;
  233. QLIST_FOREACH(job, &jobs, job_list) {
  234. if (job->id && !strcmp(id, job->id)) {
  235. return job;
  236. }
  237. }
  238. return NULL;
  239. }
  240. static void job_sleep_timer_cb(void *opaque)
  241. {
  242. Job *job = opaque;
  243. job_enter(job);
  244. }
  245. void *job_create(const char *job_id, const JobDriver *driver, JobTxn *txn,
  246. AioContext *ctx, int flags, BlockCompletionFunc *cb,
  247. void *opaque, Error **errp)
  248. {
  249. Job *job;
  250. if (job_id) {
  251. if (flags & JOB_INTERNAL) {
  252. error_setg(errp, "Cannot specify job ID for internal job");
  253. return NULL;
  254. }
  255. if (!id_wellformed(job_id)) {
  256. error_setg(errp, "Invalid job ID '%s'", job_id);
  257. return NULL;
  258. }
  259. if (job_get(job_id)) {
  260. error_setg(errp, "Job ID '%s' already in use", job_id);
  261. return NULL;
  262. }
  263. } else if (!(flags & JOB_INTERNAL)) {
  264. error_setg(errp, "An explicit job ID is required");
  265. return NULL;
  266. }
  267. job = g_malloc0(driver->instance_size);
  268. job->driver = driver;
  269. job->id = g_strdup(job_id);
  270. job->refcnt = 1;
  271. job->aio_context = ctx;
  272. job->busy = false;
  273. job->paused = true;
  274. job->pause_count = 1;
  275. job->auto_finalize = !(flags & JOB_MANUAL_FINALIZE);
  276. job->auto_dismiss = !(flags & JOB_MANUAL_DISMISS);
  277. job->cb = cb;
  278. job->opaque = opaque;
  279. notifier_list_init(&job->on_finalize_cancelled);
  280. notifier_list_init(&job->on_finalize_completed);
  281. notifier_list_init(&job->on_pending);
  282. notifier_list_init(&job->on_ready);
  283. job_state_transition(job, JOB_STATUS_CREATED);
  284. aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
  285. QEMU_CLOCK_REALTIME, SCALE_NS,
  286. job_sleep_timer_cb, job);
  287. QLIST_INSERT_HEAD(&jobs, job, job_list);
  288. /* Single jobs are modeled as single-job transactions for sake of
  289. * consolidating the job management logic */
  290. if (!txn) {
  291. txn = job_txn_new();
  292. job_txn_add_job(txn, job);
  293. job_txn_unref(txn);
  294. } else {
  295. job_txn_add_job(txn, job);
  296. }
  297. return job;
  298. }
  299. void job_ref(Job *job)
  300. {
  301. ++job->refcnt;
  302. }
  303. void job_unref(Job *job)
  304. {
  305. if (--job->refcnt == 0) {
  306. assert(job->status == JOB_STATUS_NULL);
  307. assert(!timer_pending(&job->sleep_timer));
  308. assert(!job->txn);
  309. if (job->driver->free) {
  310. job->driver->free(job);
  311. }
  312. QLIST_REMOVE(job, job_list);
  313. error_free(job->err);
  314. g_free(job->id);
  315. g_free(job);
  316. }
  317. }
  318. void job_progress_update(Job *job, uint64_t done)
  319. {
  320. job->progress_current += done;
  321. }
  322. void job_progress_set_remaining(Job *job, uint64_t remaining)
  323. {
  324. job->progress_total = job->progress_current + remaining;
  325. }
  326. void job_progress_increase_remaining(Job *job, uint64_t delta)
  327. {
  328. job->progress_total += delta;
  329. }
  330. void job_event_cancelled(Job *job)
  331. {
  332. notifier_list_notify(&job->on_finalize_cancelled, job);
  333. }
  334. void job_event_completed(Job *job)
  335. {
  336. notifier_list_notify(&job->on_finalize_completed, job);
  337. }
  338. static void job_event_pending(Job *job)
  339. {
  340. notifier_list_notify(&job->on_pending, job);
  341. }
  342. static void job_event_ready(Job *job)
  343. {
  344. notifier_list_notify(&job->on_ready, job);
  345. }
  346. static void job_event_idle(Job *job)
  347. {
  348. notifier_list_notify(&job->on_idle, job);
  349. }
  350. void job_enter_cond(Job *job, bool(*fn)(Job *job))
  351. {
  352. if (!job_started(job)) {
  353. return;
  354. }
  355. if (job->deferred_to_main_loop) {
  356. return;
  357. }
  358. job_lock();
  359. if (job->busy) {
  360. job_unlock();
  361. return;
  362. }
  363. if (fn && !fn(job)) {
  364. job_unlock();
  365. return;
  366. }
  367. assert(!job->deferred_to_main_loop);
  368. timer_del(&job->sleep_timer);
  369. job->busy = true;
  370. job_unlock();
  371. aio_co_enter(job->aio_context, job->co);
  372. }
  373. void job_enter(Job *job)
  374. {
  375. job_enter_cond(job, NULL);
  376. }
  377. /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
  378. * Reentering the job coroutine with job_enter() before the timer has expired
  379. * is allowed and cancels the timer.
  380. *
  381. * If @ns is (uint64_t) -1, no timer is scheduled and job_enter() must be
  382. * called explicitly. */
  383. static void coroutine_fn job_do_yield(Job *job, uint64_t ns)
  384. {
  385. job_lock();
  386. if (ns != -1) {
  387. timer_mod(&job->sleep_timer, ns);
  388. }
  389. job->busy = false;
  390. job_event_idle(job);
  391. job_unlock();
  392. qemu_coroutine_yield();
  393. /* Set by job_enter_cond() before re-entering the coroutine. */
  394. assert(job->busy);
  395. }
  396. void coroutine_fn job_pause_point(Job *job)
  397. {
  398. assert(job && job_started(job));
  399. if (!job_should_pause(job)) {
  400. return;
  401. }
  402. if (job_is_cancelled(job)) {
  403. return;
  404. }
  405. if (job->driver->pause) {
  406. job->driver->pause(job);
  407. }
  408. if (job_should_pause(job) && !job_is_cancelled(job)) {
  409. JobStatus status = job->status;
  410. job_state_transition(job, status == JOB_STATUS_READY
  411. ? JOB_STATUS_STANDBY
  412. : JOB_STATUS_PAUSED);
  413. job->paused = true;
  414. job_do_yield(job, -1);
  415. job->paused = false;
  416. job_state_transition(job, status);
  417. }
  418. if (job->driver->resume) {
  419. job->driver->resume(job);
  420. }
  421. }
  422. void job_yield(Job *job)
  423. {
  424. assert(job->busy);
  425. /* Check cancellation *before* setting busy = false, too! */
  426. if (job_is_cancelled(job)) {
  427. return;
  428. }
  429. if (!job_should_pause(job)) {
  430. job_do_yield(job, -1);
  431. }
  432. job_pause_point(job);
  433. }
  434. void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
  435. {
  436. assert(job->busy);
  437. /* Check cancellation *before* setting busy = false, too! */
  438. if (job_is_cancelled(job)) {
  439. return;
  440. }
  441. if (!job_should_pause(job)) {
  442. job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
  443. }
  444. job_pause_point(job);
  445. }
  446. /* Assumes the block_job_mutex is held */
  447. static bool job_timer_not_pending(Job *job)
  448. {
  449. return !timer_pending(&job->sleep_timer);
  450. }
  451. void job_pause(Job *job)
  452. {
  453. job->pause_count++;
  454. }
  455. void job_resume(Job *job)
  456. {
  457. assert(job->pause_count > 0);
  458. job->pause_count--;
  459. if (job->pause_count) {
  460. return;
  461. }
  462. /* kick only if no timer is pending */
  463. job_enter_cond(job, job_timer_not_pending);
  464. }
  465. void job_user_pause(Job *job, Error **errp)
  466. {
  467. if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
  468. return;
  469. }
  470. if (job->user_paused) {
  471. error_setg(errp, "Job is already paused");
  472. return;
  473. }
  474. job->user_paused = true;
  475. job_pause(job);
  476. }
  477. bool job_user_paused(Job *job)
  478. {
  479. return job->user_paused;
  480. }
  481. void job_user_resume(Job *job, Error **errp)
  482. {
  483. assert(job);
  484. if (!job->user_paused || job->pause_count <= 0) {
  485. error_setg(errp, "Can't resume a job that was not paused");
  486. return;
  487. }
  488. if (job_apply_verb(job, JOB_VERB_RESUME, errp)) {
  489. return;
  490. }
  491. if (job->driver->user_resume) {
  492. job->driver->user_resume(job);
  493. }
  494. job->user_paused = false;
  495. job_resume(job);
  496. }
  497. static void job_do_dismiss(Job *job)
  498. {
  499. assert(job);
  500. job->busy = false;
  501. job->paused = false;
  502. job->deferred_to_main_loop = true;
  503. job_txn_del_job(job);
  504. job_state_transition(job, JOB_STATUS_NULL);
  505. job_unref(job);
  506. }
  507. void job_dismiss(Job **jobptr, Error **errp)
  508. {
  509. Job *job = *jobptr;
  510. /* similarly to _complete, this is QMP-interface only. */
  511. assert(job->id);
  512. if (job_apply_verb(job, JOB_VERB_DISMISS, errp)) {
  513. return;
  514. }
  515. job_do_dismiss(job);
  516. *jobptr = NULL;
  517. }
  518. void job_early_fail(Job *job)
  519. {
  520. assert(job->status == JOB_STATUS_CREATED);
  521. job_do_dismiss(job);
  522. }
  523. static void job_conclude(Job *job)
  524. {
  525. job_state_transition(job, JOB_STATUS_CONCLUDED);
  526. if (job->auto_dismiss || !job_started(job)) {
  527. job_do_dismiss(job);
  528. }
  529. }
  530. static void job_update_rc(Job *job)
  531. {
  532. if (!job->ret && job_is_cancelled(job)) {
  533. job->ret = -ECANCELED;
  534. }
  535. if (job->ret) {
  536. if (!job->err) {
  537. error_setg(&job->err, "%s", strerror(-job->ret));
  538. }
  539. job_state_transition(job, JOB_STATUS_ABORTING);
  540. }
  541. }
  542. static void job_commit(Job *job)
  543. {
  544. assert(!job->ret);
  545. if (job->driver->commit) {
  546. job->driver->commit(job);
  547. }
  548. }
  549. static void job_abort(Job *job)
  550. {
  551. assert(job->ret);
  552. if (job->driver->abort) {
  553. job->driver->abort(job);
  554. }
  555. }
  556. static void job_clean(Job *job)
  557. {
  558. if (job->driver->clean) {
  559. job->driver->clean(job);
  560. }
  561. }
  562. static int job_finalize_single(Job *job)
  563. {
  564. assert(job_is_completed(job));
  565. /* Ensure abort is called for late-transactional failures */
  566. job_update_rc(job);
  567. if (!job->ret) {
  568. job_commit(job);
  569. } else {
  570. job_abort(job);
  571. }
  572. job_clean(job);
  573. if (job->cb) {
  574. job->cb(job->opaque, job->ret);
  575. }
  576. /* Emit events only if we actually started */
  577. if (job_started(job)) {
  578. if (job_is_cancelled(job)) {
  579. job_event_cancelled(job);
  580. } else {
  581. job_event_completed(job);
  582. }
  583. }
  584. job_txn_del_job(job);
  585. job_conclude(job);
  586. return 0;
  587. }
  588. static void job_cancel_async(Job *job, bool force)
  589. {
  590. if (job->user_paused) {
  591. /* Do not call job_enter here, the caller will handle it. */
  592. if (job->driver->user_resume) {
  593. job->driver->user_resume(job);
  594. }
  595. job->user_paused = false;
  596. assert(job->pause_count > 0);
  597. job->pause_count--;
  598. }
  599. job->cancelled = true;
  600. /* To prevent 'force == false' overriding a previous 'force == true' */
  601. job->force_cancel |= force;
  602. }
  603. static void job_completed_txn_abort(Job *job)
  604. {
  605. AioContext *outer_ctx = job->aio_context;
  606. AioContext *ctx;
  607. JobTxn *txn = job->txn;
  608. Job *other_job;
  609. if (txn->aborting) {
  610. /*
  611. * We are cancelled by another job, which will handle everything.
  612. */
  613. return;
  614. }
  615. txn->aborting = true;
  616. job_txn_ref(txn);
  617. /* We can only hold the single job's AioContext lock while calling
  618. * job_finalize_single() because the finalization callbacks can involve
  619. * calls of AIO_WAIT_WHILE(), which could deadlock otherwise. */
  620. aio_context_release(outer_ctx);
  621. /* Other jobs are effectively cancelled by us, set the status for
  622. * them; this job, however, may or may not be cancelled, depending
  623. * on the caller, so leave it. */
  624. QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
  625. if (other_job != job) {
  626. ctx = other_job->aio_context;
  627. aio_context_acquire(ctx);
  628. job_cancel_async(other_job, false);
  629. aio_context_release(ctx);
  630. }
  631. }
  632. while (!QLIST_EMPTY(&txn->jobs)) {
  633. other_job = QLIST_FIRST(&txn->jobs);
  634. ctx = other_job->aio_context;
  635. aio_context_acquire(ctx);
  636. if (!job_is_completed(other_job)) {
  637. assert(job_is_cancelled(other_job));
  638. job_finish_sync(other_job, NULL, NULL);
  639. }
  640. job_finalize_single(other_job);
  641. aio_context_release(ctx);
  642. }
  643. aio_context_acquire(outer_ctx);
  644. job_txn_unref(txn);
  645. }
  646. static int job_prepare(Job *job)
  647. {
  648. if (job->ret == 0 && job->driver->prepare) {
  649. job->ret = job->driver->prepare(job);
  650. job_update_rc(job);
  651. }
  652. return job->ret;
  653. }
  654. static int job_needs_finalize(Job *job)
  655. {
  656. return !job->auto_finalize;
  657. }
  658. static void job_do_finalize(Job *job)
  659. {
  660. int rc;
  661. assert(job && job->txn);
  662. /* prepare the transaction to complete */
  663. rc = job_txn_apply(job->txn, job_prepare);
  664. if (rc) {
  665. job_completed_txn_abort(job);
  666. } else {
  667. job_txn_apply(job->txn, job_finalize_single);
  668. }
  669. }
  670. void job_finalize(Job *job, Error **errp)
  671. {
  672. assert(job && job->id);
  673. if (job_apply_verb(job, JOB_VERB_FINALIZE, errp)) {
  674. return;
  675. }
  676. job_do_finalize(job);
  677. }
  678. static int job_transition_to_pending(Job *job)
  679. {
  680. job_state_transition(job, JOB_STATUS_PENDING);
  681. if (!job->auto_finalize) {
  682. job_event_pending(job);
  683. }
  684. return 0;
  685. }
  686. void job_transition_to_ready(Job *job)
  687. {
  688. job_state_transition(job, JOB_STATUS_READY);
  689. job_event_ready(job);
  690. }
  691. static void job_completed_txn_success(Job *job)
  692. {
  693. JobTxn *txn = job->txn;
  694. Job *other_job;
  695. job_state_transition(job, JOB_STATUS_WAITING);
  696. /*
  697. * Successful completion, see if there are other running jobs in this
  698. * txn.
  699. */
  700. QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
  701. if (!job_is_completed(other_job)) {
  702. return;
  703. }
  704. assert(other_job->ret == 0);
  705. }
  706. job_txn_apply(txn, job_transition_to_pending);
  707. /* If no jobs need manual finalization, automatically do so */
  708. if (job_txn_apply(txn, job_needs_finalize) == 0) {
  709. job_do_finalize(job);
  710. }
  711. }
  712. static void job_completed(Job *job)
  713. {
  714. assert(job && job->txn && !job_is_completed(job));
  715. job_update_rc(job);
  716. trace_job_completed(job, job->ret);
  717. if (job->ret) {
  718. job_completed_txn_abort(job);
  719. } else {
  720. job_completed_txn_success(job);
  721. }
  722. }
  723. /** Useful only as a type shim for aio_bh_schedule_oneshot. */
  724. static void job_exit(void *opaque)
  725. {
  726. Job *job = (Job *)opaque;
  727. AioContext *ctx = job->aio_context;
  728. aio_context_acquire(ctx);
  729. /* This is a lie, we're not quiescent, but still doing the completion
  730. * callbacks. However, completion callbacks tend to involve operations that
  731. * drain block nodes, and if .drained_poll still returned true, we would
  732. * deadlock. */
  733. job->busy = false;
  734. job_event_idle(job);
  735. job_completed(job);
  736. aio_context_release(ctx);
  737. }
  738. /**
  739. * All jobs must allow a pause point before entering their job proper. This
  740. * ensures that jobs can be paused prior to being started, then resumed later.
  741. */
  742. static void coroutine_fn job_co_entry(void *opaque)
  743. {
  744. Job *job = opaque;
  745. assert(job && job->driver && job->driver->run);
  746. job_pause_point(job);
  747. job->ret = job->driver->run(job, &job->err);
  748. job->deferred_to_main_loop = true;
  749. job->busy = true;
  750. aio_bh_schedule_oneshot(qemu_get_aio_context(), job_exit, job);
  751. }
  752. void job_start(Job *job)
  753. {
  754. assert(job && !job_started(job) && job->paused &&
  755. job->driver && job->driver->run);
  756. job->co = qemu_coroutine_create(job_co_entry, job);
  757. job->pause_count--;
  758. job->busy = true;
  759. job->paused = false;
  760. job_state_transition(job, JOB_STATUS_RUNNING);
  761. aio_co_enter(job->aio_context, job->co);
  762. }
  763. void job_cancel(Job *job, bool force)
  764. {
  765. if (job->status == JOB_STATUS_CONCLUDED) {
  766. job_do_dismiss(job);
  767. return;
  768. }
  769. job_cancel_async(job, force);
  770. if (!job_started(job)) {
  771. job_completed(job);
  772. } else if (job->deferred_to_main_loop) {
  773. job_completed_txn_abort(job);
  774. } else {
  775. job_enter(job);
  776. }
  777. }
  778. void job_user_cancel(Job *job, bool force, Error **errp)
  779. {
  780. if (job_apply_verb(job, JOB_VERB_CANCEL, errp)) {
  781. return;
  782. }
  783. job_cancel(job, force);
  784. }
  785. /* A wrapper around job_cancel() taking an Error ** parameter so it may be
  786. * used with job_finish_sync() without the need for (rather nasty) function
  787. * pointer casts there. */
  788. static void job_cancel_err(Job *job, Error **errp)
  789. {
  790. job_cancel(job, false);
  791. }
  792. int job_cancel_sync(Job *job)
  793. {
  794. return job_finish_sync(job, &job_cancel_err, NULL);
  795. }
  796. void job_cancel_sync_all(void)
  797. {
  798. Job *job;
  799. AioContext *aio_context;
  800. while ((job = job_next(NULL))) {
  801. aio_context = job->aio_context;
  802. aio_context_acquire(aio_context);
  803. job_cancel_sync(job);
  804. aio_context_release(aio_context);
  805. }
  806. }
  807. int job_complete_sync(Job *job, Error **errp)
  808. {
  809. return job_finish_sync(job, job_complete, errp);
  810. }
  811. void job_complete(Job *job, Error **errp)
  812. {
  813. /* Should not be reachable via external interface for internal jobs */
  814. assert(job->id);
  815. if (job_apply_verb(job, JOB_VERB_COMPLETE, errp)) {
  816. return;
  817. }
  818. if (job->pause_count || job_is_cancelled(job) || !job->driver->complete) {
  819. error_setg(errp, "The active block job '%s' cannot be completed",
  820. job->id);
  821. return;
  822. }
  823. job->driver->complete(job, errp);
  824. }
  825. int job_finish_sync(Job *job, void (*finish)(Job *, Error **errp), Error **errp)
  826. {
  827. Error *local_err = NULL;
  828. int ret;
  829. job_ref(job);
  830. if (finish) {
  831. finish(job, &local_err);
  832. }
  833. if (local_err) {
  834. error_propagate(errp, local_err);
  835. job_unref(job);
  836. return -EBUSY;
  837. }
  838. AIO_WAIT_WHILE(job->aio_context,
  839. (job_enter(job), !job_is_completed(job)));
  840. ret = (job_is_cancelled(job) && job->ret == 0) ? -ECANCELED : job->ret;
  841. job_unref(job);
  842. return ret;
  843. }