1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26#include "qemu/osdep.h"
27#include "block/block.h"
28#include "block/blockjob_int.h"
29#include "block/block_int.h"
30#include "block/trace.h"
31#include "sysemu/block-backend.h"
32#include "qapi/error.h"
33#include "qapi/qapi-events-block-core.h"
34#include "qapi/qmp/qerror.h"
35#include "qemu/coroutine.h"
36#include "qemu/main-loop.h"
37#include "qemu/timer.h"
38
39static bool is_block_job(Job *job)
40{
41 return job_type(job) == JOB_TYPE_BACKUP ||
42 job_type(job) == JOB_TYPE_COMMIT ||
43 job_type(job) == JOB_TYPE_MIRROR ||
44 job_type(job) == JOB_TYPE_STREAM;
45}
46
47BlockJob *block_job_next_locked(BlockJob *bjob)
48{
49 Job *job = bjob ? &bjob->job : NULL;
50 GLOBAL_STATE_CODE();
51
52 do {
53 job = job_next_locked(job);
54 } while (job && !is_block_job(job));
55
56 return job ? container_of(job, BlockJob, job) : NULL;
57}
58
59BlockJob *block_job_get_locked(const char *id)
60{
61 Job *job = job_get_locked(id);
62 GLOBAL_STATE_CODE();
63
64 if (job && is_block_job(job)) {
65 return container_of(job, BlockJob, job);
66 } else {
67 return NULL;
68 }
69}
70
71BlockJob *block_job_get(const char *id)
72{
73 JOB_LOCK_GUARD();
74 return block_job_get_locked(id);
75}
76
77void block_job_free(Job *job)
78{
79 BlockJob *bjob = container_of(job, BlockJob, job);
80 GLOBAL_STATE_CODE();
81
82 block_job_remove_all_bdrv(bjob);
83 ratelimit_destroy(&bjob->limit);
84 error_free(bjob->blocker);
85}
86
87static char *child_job_get_parent_desc(BdrvChild *c)
88{
89 BlockJob *job = c->opaque;
90 return g_strdup_printf("%s job '%s'", job_type_str(&job->job), job->job.id);
91}
92
93static void child_job_drained_begin(BdrvChild *c)
94{
95 BlockJob *job = c->opaque;
96 job_pause(&job->job);
97}
98
99static bool child_job_drained_poll(BdrvChild *c)
100{
101 BlockJob *bjob = c->opaque;
102 Job *job = &bjob->job;
103 const BlockJobDriver *drv = block_job_driver(bjob);
104
105
106
107
108 WITH_JOB_LOCK_GUARD() {
109 if (!job->busy || job_is_completed_locked(job)) {
110 return false;
111 }
112 }
113
114
115
116 if (drv->drained_poll) {
117 return drv->drained_poll(bjob);
118 } else {
119 return true;
120 }
121}
122
123static void child_job_drained_end(BdrvChild *c, int *drained_end_counter)
124{
125 BlockJob *job = c->opaque;
126 job_resume(&job->job);
127}
128
129typedef struct BdrvStateChildJobContext {
130 AioContext *new_ctx;
131 BlockJob *job;
132} BdrvStateChildJobContext;
133
134static void child_job_set_aio_ctx_commit(void *opaque)
135{
136 BdrvStateChildJobContext *s = opaque;
137 BlockJob *job = s->job;
138
139 job_set_aio_context(&job->job, s->new_ctx);
140}
141
142static TransactionActionDrv change_child_job_context = {
143 .commit = child_job_set_aio_ctx_commit,
144 .clean = g_free,
145};
146
147static bool child_job_change_aio_ctx(BdrvChild *c, AioContext *ctx,
148 GHashTable *visited, Transaction *tran,
149 Error **errp)
150{
151 BlockJob *job = c->opaque;
152 BdrvStateChildJobContext *s;
153 GSList *l;
154
155 for (l = job->nodes; l; l = l->next) {
156 BdrvChild *sibling = l->data;
157 if (!bdrv_child_change_aio_context(sibling, ctx, visited,
158 tran, errp)) {
159 return false;
160 }
161 }
162
163 s = g_new(BdrvStateChildJobContext, 1);
164 *s = (BdrvStateChildJobContext) {
165 .new_ctx = ctx,
166 .job = job,
167 };
168
169 tran_add(tran, &change_child_job_context, s);
170 return true;
171}
172
173static AioContext *child_job_get_parent_aio_context(BdrvChild *c)
174{
175 BlockJob *job = c->opaque;
176 IO_CODE();
177 JOB_LOCK_GUARD();
178
179 return job->job.aio_context;
180}
181
182static const BdrvChildClass child_job = {
183 .get_parent_desc = child_job_get_parent_desc,
184 .drained_begin = child_job_drained_begin,
185 .drained_poll = child_job_drained_poll,
186 .drained_end = child_job_drained_end,
187 .change_aio_ctx = child_job_change_aio_ctx,
188 .stay_at_node = true,
189 .get_parent_aio_context = child_job_get_parent_aio_context,
190};
191
192void block_job_remove_all_bdrv(BlockJob *job)
193{
194 GLOBAL_STATE_CODE();
195
196
197
198
199
200
201 while (job->nodes) {
202 GSList *l = job->nodes;
203 BdrvChild *c = l->data;
204
205 job->nodes = l->next;
206
207 bdrv_op_unblock_all(c->bs, job->blocker);
208 bdrv_root_unref_child(c);
209
210 g_slist_free_1(l);
211 }
212}
213
214bool block_job_has_bdrv(BlockJob *job, BlockDriverState *bs)
215{
216 GSList *el;
217 GLOBAL_STATE_CODE();
218
219 for (el = job->nodes; el; el = el->next) {
220 BdrvChild *c = el->data;
221 if (c->bs == bs) {
222 return true;
223 }
224 }
225
226 return false;
227}
228
229int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs,
230 uint64_t perm, uint64_t shared_perm, Error **errp)
231{
232 BdrvChild *c;
233 bool need_context_ops;
234 GLOBAL_STATE_CODE();
235
236 bdrv_ref(bs);
237
238 need_context_ops = bdrv_get_aio_context(bs) != job->job.aio_context;
239
240 if (need_context_ops && job->job.aio_context != qemu_get_aio_context()) {
241 aio_context_release(job->job.aio_context);
242 }
243 c = bdrv_root_attach_child(bs, name, &child_job, 0, perm, shared_perm, job,
244 errp);
245 if (need_context_ops && job->job.aio_context != qemu_get_aio_context()) {
246 aio_context_acquire(job->job.aio_context);
247 }
248 if (c == NULL) {
249 return -EPERM;
250 }
251
252 job->nodes = g_slist_prepend(job->nodes, c);
253 bdrv_op_block_all(bs, job->blocker);
254
255 return 0;
256}
257
258
259static void block_job_on_idle_locked(Notifier *n, void *opaque)
260{
261 aio_wait_kick();
262}
263
264bool block_job_is_internal(BlockJob *job)
265{
266 return (job->job.id == NULL);
267}
268
269const BlockJobDriver *block_job_driver(BlockJob *job)
270{
271 return container_of(job->job.driver, BlockJobDriver, job_driver);
272}
273
274
275static bool job_timer_pending(Job *job)
276{
277 return timer_pending(&job->sleep_timer);
278}
279
280bool block_job_set_speed_locked(BlockJob *job, int64_t speed, Error **errp)
281{
282 const BlockJobDriver *drv = block_job_driver(job);
283 int64_t old_speed = job->speed;
284
285 GLOBAL_STATE_CODE();
286
287 if (job_apply_verb_locked(&job->job, JOB_VERB_SET_SPEED, errp) < 0) {
288 return false;
289 }
290 if (speed < 0) {
291 error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "speed",
292 "a non-negative value");
293 return false;
294 }
295
296 ratelimit_set_speed(&job->limit, speed, BLOCK_JOB_SLICE_TIME);
297
298 job->speed = speed;
299
300 if (drv->set_speed) {
301 job_unlock();
302 drv->set_speed(job, speed);
303 job_lock();
304 }
305
306 if (speed && speed <= old_speed) {
307 return true;
308 }
309
310
311 job_enter_cond_locked(&job->job, job_timer_pending);
312
313 return true;
314}
315
316static bool block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
317{
318 JOB_LOCK_GUARD();
319 return block_job_set_speed_locked(job, speed, errp);
320}
321
322int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
323{
324 IO_CODE();
325 return ratelimit_calculate_delay(&job->limit, n);
326}
327
328BlockJobInfo *block_job_query_locked(BlockJob *job, Error **errp)
329{
330 BlockJobInfo *info;
331 uint64_t progress_current, progress_total;
332
333 GLOBAL_STATE_CODE();
334
335 if (block_job_is_internal(job)) {
336 error_setg(errp, "Cannot query QEMU internal jobs");
337 return NULL;
338 }
339
340 progress_get_snapshot(&job->job.progress, &progress_current,
341 &progress_total);
342
343 info = g_new0(BlockJobInfo, 1);
344 info->type = g_strdup(job_type_str(&job->job));
345 info->device = g_strdup(job->job.id);
346 info->busy = job->job.busy;
347 info->paused = job->job.pause_count > 0;
348 info->offset = progress_current;
349 info->len = progress_total;
350 info->speed = job->speed;
351 info->io_status = job->iostatus;
352 info->ready = job_is_ready_locked(&job->job),
353 info->status = job->job.status;
354 info->auto_finalize = job->job.auto_finalize;
355 info->auto_dismiss = job->job.auto_dismiss;
356 if (job->job.ret) {
357 info->has_error = true;
358 info->error = job->job.err ?
359 g_strdup(error_get_pretty(job->job.err)) :
360 g_strdup(strerror(-job->job.ret));
361 }
362 return info;
363}
364
365
366static void block_job_iostatus_set_err_locked(BlockJob *job, int error)
367{
368 if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
369 job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
370 BLOCK_DEVICE_IO_STATUS_FAILED;
371 }
372}
373
374
375static void block_job_event_cancelled_locked(Notifier *n, void *opaque)
376{
377 BlockJob *job = opaque;
378 uint64_t progress_current, progress_total;
379
380 if (block_job_is_internal(job)) {
381 return;
382 }
383
384 progress_get_snapshot(&job->job.progress, &progress_current,
385 &progress_total);
386
387 qapi_event_send_block_job_cancelled(job_type(&job->job),
388 job->job.id,
389 progress_total,
390 progress_current,
391 job->speed);
392}
393
394
395static void block_job_event_completed_locked(Notifier *n, void *opaque)
396{
397 BlockJob *job = opaque;
398 const char *msg = NULL;
399 uint64_t progress_current, progress_total;
400
401 if (block_job_is_internal(job)) {
402 return;
403 }
404
405 if (job->job.ret < 0) {
406 msg = error_get_pretty(job->job.err);
407 }
408
409 progress_get_snapshot(&job->job.progress, &progress_current,
410 &progress_total);
411
412 qapi_event_send_block_job_completed(job_type(&job->job),
413 job->job.id,
414 progress_total,
415 progress_current,
416 job->speed,
417 !!msg,
418 msg);
419}
420
421
422static void block_job_event_pending_locked(Notifier *n, void *opaque)
423{
424 BlockJob *job = opaque;
425
426 if (block_job_is_internal(job)) {
427 return;
428 }
429
430 qapi_event_send_block_job_pending(job_type(&job->job),
431 job->job.id);
432}
433
434
435static void block_job_event_ready_locked(Notifier *n, void *opaque)
436{
437 BlockJob *job = opaque;
438 uint64_t progress_current, progress_total;
439
440 if (block_job_is_internal(job)) {
441 return;
442 }
443
444 progress_get_snapshot(&job->job.progress, &progress_current,
445 &progress_total);
446
447 qapi_event_send_block_job_ready(job_type(&job->job),
448 job->job.id,
449 progress_total,
450 progress_current,
451 job->speed);
452}
453
454
455void *block_job_create(const char *job_id, const BlockJobDriver *driver,
456 JobTxn *txn, BlockDriverState *bs, uint64_t perm,
457 uint64_t shared_perm, int64_t speed, int flags,
458 BlockCompletionFunc *cb, void *opaque, Error **errp)
459{
460 BlockJob *job;
461 int ret;
462 GLOBAL_STATE_CODE();
463
464 if (job_id == NULL && !(flags & JOB_INTERNAL)) {
465 job_id = bdrv_get_device_name(bs);
466 }
467
468 job = job_create(job_id, &driver->job_driver, txn, bdrv_get_aio_context(bs),
469 flags, cb, opaque, errp);
470 if (job == NULL) {
471 return NULL;
472 }
473
474 assert(is_block_job(&job->job));
475 assert(job->job.driver->free == &block_job_free);
476 assert(job->job.driver->user_resume == &block_job_user_resume);
477
478 ratelimit_init(&job->limit);
479
480 job->finalize_cancelled_notifier.notify = block_job_event_cancelled_locked;
481 job->finalize_completed_notifier.notify = block_job_event_completed_locked;
482 job->pending_notifier.notify = block_job_event_pending_locked;
483 job->ready_notifier.notify = block_job_event_ready_locked;
484 job->idle_notifier.notify = block_job_on_idle_locked;
485
486 WITH_JOB_LOCK_GUARD() {
487 notifier_list_add(&job->job.on_finalize_cancelled,
488 &job->finalize_cancelled_notifier);
489 notifier_list_add(&job->job.on_finalize_completed,
490 &job->finalize_completed_notifier);
491 notifier_list_add(&job->job.on_pending, &job->pending_notifier);
492 notifier_list_add(&job->job.on_ready, &job->ready_notifier);
493 notifier_list_add(&job->job.on_idle, &job->idle_notifier);
494 }
495
496 error_setg(&job->blocker, "block device is in use by block job: %s",
497 job_type_str(&job->job));
498
499 ret = block_job_add_bdrv(job, "main node", bs, perm, shared_perm, errp);
500 if (ret < 0) {
501 goto fail;
502 }
503
504 bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
505
506 if (!block_job_set_speed(job, speed, errp)) {
507 goto fail;
508 }
509
510 return job;
511
512fail:
513 job_early_fail(&job->job);
514 return NULL;
515}
516
517void block_job_iostatus_reset_locked(BlockJob *job)
518{
519 GLOBAL_STATE_CODE();
520 if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
521 return;
522 }
523 assert(job->job.user_paused && job->job.pause_count > 0);
524 job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
525}
526
527static void block_job_iostatus_reset(BlockJob *job)
528{
529 JOB_LOCK_GUARD();
530 block_job_iostatus_reset_locked(job);
531}
532
533void block_job_user_resume(Job *job)
534{
535 BlockJob *bjob = container_of(job, BlockJob, job);
536 GLOBAL_STATE_CODE();
537 block_job_iostatus_reset(bjob);
538}
539
540BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
541 int is_read, int error)
542{
543 BlockErrorAction action;
544 IO_CODE();
545
546 switch (on_err) {
547 case BLOCKDEV_ON_ERROR_ENOSPC:
548 case BLOCKDEV_ON_ERROR_AUTO:
549 action = (error == ENOSPC) ?
550 BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
551 break;
552 case BLOCKDEV_ON_ERROR_STOP:
553 action = BLOCK_ERROR_ACTION_STOP;
554 break;
555 case BLOCKDEV_ON_ERROR_REPORT:
556 action = BLOCK_ERROR_ACTION_REPORT;
557 break;
558 case BLOCKDEV_ON_ERROR_IGNORE:
559 action = BLOCK_ERROR_ACTION_IGNORE;
560 break;
561 default:
562 abort();
563 }
564 if (!block_job_is_internal(job)) {
565 qapi_event_send_block_job_error(job->job.id,
566 is_read ? IO_OPERATION_TYPE_READ :
567 IO_OPERATION_TYPE_WRITE,
568 action);
569 }
570 if (action == BLOCK_ERROR_ACTION_STOP) {
571 WITH_JOB_LOCK_GUARD() {
572 if (!job->job.user_paused) {
573 job_pause_locked(&job->job);
574
575
576
577
578 job->job.user_paused = true;
579 }
580 block_job_iostatus_set_err_locked(job, error);
581 }
582 }
583 return action;
584}
585
586AioContext *block_job_get_aio_context(BlockJob *job)
587{
588 GLOBAL_STATE_CODE();
589 return job->job.aio_context;
590}
591