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

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999
  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. void job_drain(Job *job)
  447. {
  448. /* If job is !busy this kicks it into the next pause point. */
  449. job_enter(job);
  450. if (job->driver->drain) {
  451. job->driver->drain(job);
  452. }
  453. }
  454. /* Assumes the block_job_mutex is held */
  455. static bool job_timer_not_pending(Job *job)
  456. {
  457. return !timer_pending(&job->sleep_timer);
  458. }
  459. void job_pause(Job *job)
  460. {
  461. job->pause_count++;
  462. }
  463. void job_resume(Job *job)
  464. {
  465. assert(job->pause_count > 0);
  466. job->pause_count--;
  467. if (job->pause_count) {
  468. return;
  469. }
  470. /* kick only if no timer is pending */
  471. job_enter_cond(job, job_timer_not_pending);
  472. }
  473. void job_user_pause(Job *job, Error **errp)
  474. {
  475. if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
  476. return;
  477. }
  478. if (job->user_paused) {
  479. error_setg(errp, "Job is already paused");
  480. return;
  481. }
  482. job->user_paused = true;
  483. job_pause(job);
  484. }
  485. bool job_user_paused(Job *job)
  486. {
  487. return job->user_paused;
  488. }
  489. void job_user_resume(Job *job, Error **errp)
  490. {
  491. assert(job);
  492. if (!job->user_paused || job->pause_count <= 0) {
  493. error_setg(errp, "Can't resume a job that was not paused");
  494. return;
  495. }
  496. if (job_apply_verb(job, JOB_VERB_RESUME, errp)) {
  497. return;
  498. }
  499. if (job->driver->user_resume) {
  500. job->driver->user_resume(job);
  501. }
  502. job->user_paused = false;
  503. job_resume(job);
  504. }
  505. static void job_do_dismiss(Job *job)
  506. {
  507. assert(job);
  508. job->busy = false;
  509. job->paused = false;
  510. job->deferred_to_main_loop = true;
  511. job_txn_del_job(job);
  512. job_state_transition(job, JOB_STATUS_NULL);
  513. job_unref(job);
  514. }
  515. void job_dismiss(Job **jobptr, Error **errp)
  516. {
  517. Job *job = *jobptr;
  518. /* similarly to _complete, this is QMP-interface only. */
  519. assert(job->id);
  520. if (job_apply_verb(job, JOB_VERB_DISMISS, errp)) {
  521. return;
  522. }
  523. job_do_dismiss(job);
  524. *jobptr = NULL;
  525. }
  526. void job_early_fail(Job *job)
  527. {
  528. assert(job->status == JOB_STATUS_CREATED);
  529. job_do_dismiss(job);
  530. }
  531. static void job_conclude(Job *job)
  532. {
  533. job_state_transition(job, JOB_STATUS_CONCLUDED);
  534. if (job->auto_dismiss || !job_started(job)) {
  535. job_do_dismiss(job);
  536. }
  537. }
  538. static void job_update_rc(Job *job)
  539. {
  540. if (!job->ret && job_is_cancelled(job)) {
  541. job->ret = -ECANCELED;
  542. }
  543. if (job->ret) {
  544. if (!job->err) {
  545. error_setg(&job->err, "%s", strerror(-job->ret));
  546. }
  547. job_state_transition(job, JOB_STATUS_ABORTING);
  548. }
  549. }
  550. static void job_commit(Job *job)
  551. {
  552. assert(!job->ret);
  553. if (job->driver->commit) {
  554. job->driver->commit(job);
  555. }
  556. }
  557. static void job_abort(Job *job)
  558. {
  559. assert(job->ret);
  560. if (job->driver->abort) {
  561. job->driver->abort(job);
  562. }
  563. }
  564. static void job_clean(Job *job)
  565. {
  566. if (job->driver->clean) {
  567. job->driver->clean(job);
  568. }
  569. }
  570. static int job_finalize_single(Job *job)
  571. {
  572. assert(job_is_completed(job));
  573. /* Ensure abort is called for late-transactional failures */
  574. job_update_rc(job);
  575. if (!job->ret) {
  576. job_commit(job);
  577. } else {
  578. job_abort(job);
  579. }
  580. job_clean(job);
  581. if (job->cb) {
  582. job->cb(job->opaque, job->ret);
  583. }
  584. /* Emit events only if we actually started */
  585. if (job_started(job)) {
  586. if (job_is_cancelled(job)) {
  587. job_event_cancelled(job);
  588. } else {
  589. job_event_completed(job);
  590. }
  591. }
  592. job_txn_del_job(job);
  593. job_conclude(job);
  594. return 0;
  595. }
  596. static void job_cancel_async(Job *job, bool force)
  597. {
  598. if (job->user_paused) {
  599. /* Do not call job_enter here, the caller will handle it. */
  600. if (job->driver->user_resume) {
  601. job->driver->user_resume(job);
  602. }
  603. job->user_paused = false;
  604. assert(job->pause_count > 0);
  605. job->pause_count--;
  606. }
  607. job->cancelled = true;
  608. /* To prevent 'force == false' overriding a previous 'force == true' */
  609. job->force_cancel |= force;
  610. }
  611. static void job_completed_txn_abort(Job *job)
  612. {
  613. AioContext *outer_ctx = job->aio_context;
  614. AioContext *ctx;
  615. JobTxn *txn = job->txn;
  616. Job *other_job;
  617. if (txn->aborting) {
  618. /*
  619. * We are cancelled by another job, which will handle everything.
  620. */
  621. return;
  622. }
  623. txn->aborting = true;
  624. job_txn_ref(txn);
  625. /* We can only hold the single job's AioContext lock while calling
  626. * job_finalize_single() because the finalization callbacks can involve
  627. * calls of AIO_WAIT_WHILE(), which could deadlock otherwise. */
  628. aio_context_release(outer_ctx);
  629. /* Other jobs are effectively cancelled by us, set the status for
  630. * them; this job, however, may or may not be cancelled, depending
  631. * on the caller, so leave it. */
  632. QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
  633. if (other_job != job) {
  634. ctx = other_job->aio_context;
  635. aio_context_acquire(ctx);
  636. job_cancel_async(other_job, false);
  637. aio_context_release(ctx);
  638. }
  639. }
  640. while (!QLIST_EMPTY(&txn->jobs)) {
  641. other_job = QLIST_FIRST(&txn->jobs);
  642. ctx = other_job->aio_context;
  643. aio_context_acquire(ctx);
  644. if (!job_is_completed(other_job)) {
  645. assert(job_is_cancelled(other_job));
  646. job_finish_sync(other_job, NULL, NULL);
  647. }
  648. job_finalize_single(other_job);
  649. aio_context_release(ctx);
  650. }
  651. aio_context_acquire(outer_ctx);
  652. job_txn_unref(txn);
  653. }
  654. static int job_prepare(Job *job)
  655. {
  656. if (job->ret == 0 && job->driver->prepare) {
  657. job->ret = job->driver->prepare(job);
  658. job_update_rc(job);
  659. }
  660. return job->ret;
  661. }
  662. static int job_needs_finalize(Job *job)
  663. {
  664. return !job->auto_finalize;
  665. }
  666. static void job_do_finalize(Job *job)
  667. {
  668. int rc;
  669. assert(job && job->txn);
  670. /* prepare the transaction to complete */
  671. rc = job_txn_apply(job->txn, job_prepare);
  672. if (rc) {
  673. job_completed_txn_abort(job);
  674. } else {
  675. job_txn_apply(job->txn, job_finalize_single);
  676. }
  677. }
  678. void job_finalize(Job *job, Error **errp)
  679. {
  680. assert(job && job->id);
  681. if (job_apply_verb(job, JOB_VERB_FINALIZE, errp)) {
  682. return;
  683. }
  684. job_do_finalize(job);
  685. }
  686. static int job_transition_to_pending(Job *job)
  687. {
  688. job_state_transition(job, JOB_STATUS_PENDING);
  689. if (!job->auto_finalize) {
  690. job_event_pending(job);
  691. }
  692. return 0;
  693. }
  694. void job_transition_to_ready(Job *job)
  695. {
  696. job_state_transition(job, JOB_STATUS_READY);
  697. job_event_ready(job);
  698. }
  699. static void job_completed_txn_success(Job *job)
  700. {
  701. JobTxn *txn = job->txn;
  702. Job *other_job;
  703. job_state_transition(job, JOB_STATUS_WAITING);
  704. /*
  705. * Successful completion, see if there are other running jobs in this
  706. * txn.
  707. */
  708. QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
  709. if (!job_is_completed(other_job)) {
  710. return;
  711. }
  712. assert(other_job->ret == 0);
  713. }
  714. job_txn_apply(txn, job_transition_to_pending);
  715. /* If no jobs need manual finalization, automatically do so */
  716. if (job_txn_apply(txn, job_needs_finalize) == 0) {
  717. job_do_finalize(job);
  718. }
  719. }
  720. static void job_completed(Job *job)
  721. {
  722. assert(job && job->txn && !job_is_completed(job));
  723. job_update_rc(job);
  724. trace_job_completed(job, job->ret);
  725. if (job->ret) {
  726. job_completed_txn_abort(job);
  727. } else {
  728. job_completed_txn_success(job);
  729. }
  730. }
  731. /** Useful only as a type shim for aio_bh_schedule_oneshot. */
  732. static void job_exit(void *opaque)
  733. {
  734. Job *job = (Job *)opaque;
  735. AioContext *ctx = job->aio_context;
  736. aio_context_acquire(ctx);
  737. /* This is a lie, we're not quiescent, but still doing the completion
  738. * callbacks. However, completion callbacks tend to involve operations that
  739. * drain block nodes, and if .drained_poll still returned true, we would
  740. * deadlock. */
  741. job->busy = false;
  742. job_event_idle(job);
  743. job_completed(job);
  744. aio_context_release(ctx);
  745. }
  746. /**
  747. * All jobs must allow a pause point before entering their job proper. This
  748. * ensures that jobs can be paused prior to being started, then resumed later.
  749. */
  750. static void coroutine_fn job_co_entry(void *opaque)
  751. {
  752. Job *job = opaque;
  753. assert(job && job->driver && job->driver->run);
  754. job_pause_point(job);
  755. job->ret = job->driver->run(job, &job->err);
  756. job->deferred_to_main_loop = true;
  757. job->busy = true;
  758. aio_bh_schedule_oneshot(qemu_get_aio_context(), job_exit, job);
  759. }
  760. void job_start(Job *job)
  761. {
  762. assert(job && !job_started(job) && job->paused &&
  763. job->driver && job->driver->run);
  764. job->co = qemu_coroutine_create(job_co_entry, job);
  765. job->pause_count--;
  766. job->busy = true;
  767. job->paused = false;
  768. job_state_transition(job, JOB_STATUS_RUNNING);
  769. aio_co_enter(job->aio_context, job->co);
  770. }
  771. void job_cancel(Job *job, bool force)
  772. {
  773. if (job->status == JOB_STATUS_CONCLUDED) {
  774. job_do_dismiss(job);
  775. return;
  776. }
  777. job_cancel_async(job, force);
  778. if (!job_started(job)) {
  779. job_completed(job);
  780. } else if (job->deferred_to_main_loop) {
  781. job_completed_txn_abort(job);
  782. } else {
  783. job_enter(job);
  784. }
  785. }
  786. void job_user_cancel(Job *job, bool force, Error **errp)
  787. {
  788. if (job_apply_verb(job, JOB_VERB_CANCEL, errp)) {
  789. return;
  790. }
  791. job_cancel(job, force);
  792. }
  793. /* A wrapper around job_cancel() taking an Error ** parameter so it may be
  794. * used with job_finish_sync() without the need for (rather nasty) function
  795. * pointer casts there. */
  796. static void job_cancel_err(Job *job, Error **errp)
  797. {
  798. job_cancel(job, false);
  799. }
  800. int job_cancel_sync(Job *job)
  801. {
  802. return job_finish_sync(job, &job_cancel_err, NULL);
  803. }
  804. void job_cancel_sync_all(void)
  805. {
  806. Job *job;
  807. AioContext *aio_context;
  808. while ((job = job_next(NULL))) {
  809. aio_context = job->aio_context;
  810. aio_context_acquire(aio_context);
  811. job_cancel_sync(job);
  812. aio_context_release(aio_context);
  813. }
  814. }
  815. int job_complete_sync(Job *job, Error **errp)
  816. {
  817. return job_finish_sync(job, job_complete, errp);
  818. }
  819. void job_complete(Job *job, Error **errp)
  820. {
  821. /* Should not be reachable via external interface for internal jobs */
  822. assert(job->id);
  823. if (job_apply_verb(job, JOB_VERB_COMPLETE, errp)) {
  824. return;
  825. }
  826. if (job->pause_count || job_is_cancelled(job) || !job->driver->complete) {
  827. error_setg(errp, "The active block job '%s' cannot be completed",
  828. job->id);
  829. return;
  830. }
  831. job->driver->complete(job, errp);
  832. }
  833. int job_finish_sync(Job *job, void (*finish)(Job *, Error **errp), Error **errp)
  834. {
  835. Error *local_err = NULL;
  836. int ret;
  837. job_ref(job);
  838. if (finish) {
  839. finish(job, &local_err);
  840. }
  841. if (local_err) {
  842. error_propagate(errp, local_err);
  843. job_unref(job);
  844. return -EBUSY;
  845. }
  846. AIO_WAIT_WHILE(job->aio_context,
  847. (job_drain(job), !job_is_completed(job)));
  848. ret = (job_is_cancelled(job) && job->ret == 0) ? -ECANCELED : job->ret;
  849. job_unref(job);
  850. return ret;
  851. }