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