xref: /openbmc/qemu/blockjob.c (revision 411132c9790aae23f69add2e9f5f129880e079c0)
1  /*
2   * QEMU System Emulator block driver
3   *
4   * Copyright (c) 2011 IBM Corp.
5   * Copyright (c) 2012 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  
26  #include "qemu/osdep.h"
27  #include "block/aio-wait.h"
28  #include "block/block.h"
29  #include "block/blockjob_int.h"
30  #include "block/block_int.h"
31  #include "block/trace.h"
32  #include "sysemu/block-backend.h"
33  #include "qapi/error.h"
34  #include "qapi/qapi-events-block-core.h"
35  #include "qapi/qmp/qerror.h"
36  #include "qemu/main-loop.h"
37  #include "qemu/timer.h"
38  
39  static 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  
47  BlockJob *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  
59  BlockJob *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  
71  BlockJob *block_job_get(const char *id)
72  {
73      JOB_LOCK_GUARD();
74      return block_job_get_locked(id);
75  }
76  
77  void 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  
87  static 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  
93  static void child_job_drained_begin(BdrvChild *c)
94  {
95      BlockJob *job = c->opaque;
96      job_pause(&job->job);
97  }
98  
99  static 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      /* An inactive or completed job doesn't have any pending requests. Jobs
106       * with !job->busy are either already paused or have a pause point after
107       * being reentered, so no job driver code will run before they pause. */
108      WITH_JOB_LOCK_GUARD() {
109          if (!job->busy || job_is_completed_locked(job)) {
110              return false;
111          }
112      }
113  
114      /* Otherwise, assume that it isn't fully stopped yet, but allow the job to
115       * override this assumption. */
116      if (drv->drained_poll) {
117          return drv->drained_poll(bjob);
118      } else {
119          return true;
120      }
121  }
122  
123  static void child_job_drained_end(BdrvChild *c)
124  {
125      BlockJob *job = c->opaque;
126      job_resume(&job->job);
127  }
128  
129  typedef struct BdrvStateChildJobContext {
130      AioContext *new_ctx;
131      BlockJob *job;
132  } BdrvStateChildJobContext;
133  
134  static 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  
142  static TransactionActionDrv change_child_job_context = {
143      .commit = child_job_set_aio_ctx_commit,
144      .clean = g_free,
145  };
146  
147  static 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  
173  static 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  
182  static 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  
192  void block_job_remove_all_bdrv(BlockJob *job)
193  {
194      GLOBAL_STATE_CODE();
195      /*
196       * bdrv_root_unref_child() may reach child_job_[can_]set_aio_ctx(),
197       * which will also traverse job->nodes, so consume the list one by
198       * one to make sure that such a concurrent access does not attempt
199       * to process an already freed BdrvChild.
200       */
201      aio_context_release(job->job.aio_context);
202      bdrv_graph_wrlock(NULL);
203      aio_context_acquire(job->job.aio_context);
204      while (job->nodes) {
205          GSList *l = job->nodes;
206          BdrvChild *c = l->data;
207  
208          job->nodes = l->next;
209  
210          bdrv_op_unblock_all(c->bs, job->blocker);
211          bdrv_root_unref_child(c);
212  
213          g_slist_free_1(l);
214      }
215      bdrv_graph_wrunlock_ctx(job->job.aio_context);
216  }
217  
218  bool block_job_has_bdrv(BlockJob *job, BlockDriverState *bs)
219  {
220      GSList *el;
221      GLOBAL_STATE_CODE();
222  
223      for (el = job->nodes; el; el = el->next) {
224          BdrvChild *c = el->data;
225          if (c->bs == bs) {
226              return true;
227          }
228      }
229  
230      return false;
231  }
232  
233  int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs,
234                         uint64_t perm, uint64_t shared_perm, Error **errp)
235  {
236      BdrvChild *c;
237      AioContext *ctx = bdrv_get_aio_context(bs);
238      bool need_context_ops;
239      GLOBAL_STATE_CODE();
240  
241      bdrv_ref(bs);
242  
243      need_context_ops = ctx != job->job.aio_context;
244  
245      if (need_context_ops) {
246          if (job->job.aio_context != qemu_get_aio_context()) {
247              aio_context_release(job->job.aio_context);
248          }
249          aio_context_acquire(ctx);
250      }
251      c = bdrv_root_attach_child(bs, name, &child_job, 0, perm, shared_perm, job,
252                                 errp);
253      if (need_context_ops) {
254          aio_context_release(ctx);
255          if (job->job.aio_context != qemu_get_aio_context()) {
256              aio_context_acquire(job->job.aio_context);
257          }
258      }
259      if (c == NULL) {
260          return -EPERM;
261      }
262  
263      job->nodes = g_slist_prepend(job->nodes, c);
264      bdrv_op_block_all(bs, job->blocker);
265  
266      return 0;
267  }
268  
269  /* Called with job_mutex lock held. */
270  static void block_job_on_idle_locked(Notifier *n, void *opaque)
271  {
272      aio_wait_kick();
273  }
274  
275  bool block_job_is_internal(BlockJob *job)
276  {
277      return (job->job.id == NULL);
278  }
279  
280  const BlockJobDriver *block_job_driver(BlockJob *job)
281  {
282      return container_of(job->job.driver, BlockJobDriver, job_driver);
283  }
284  
285  /* Assumes the job_mutex is held */
286  static bool job_timer_pending(Job *job)
287  {
288      return timer_pending(&job->sleep_timer);
289  }
290  
291  bool block_job_set_speed_locked(BlockJob *job, int64_t speed, Error **errp)
292  {
293      const BlockJobDriver *drv = block_job_driver(job);
294      int64_t old_speed = job->speed;
295  
296      GLOBAL_STATE_CODE();
297  
298      if (job_apply_verb_locked(&job->job, JOB_VERB_SET_SPEED, errp) < 0) {
299          return false;
300      }
301      if (speed < 0) {
302          error_setg(errp, QERR_INVALID_PARAMETER_VALUE, "speed",
303                     "a non-negative value");
304          return false;
305      }
306  
307      ratelimit_set_speed(&job->limit, speed, BLOCK_JOB_SLICE_TIME);
308  
309      job->speed = speed;
310  
311      if (drv->set_speed) {
312          job_unlock();
313          drv->set_speed(job, speed);
314          job_lock();
315      }
316  
317      if (speed && speed <= old_speed) {
318          return true;
319      }
320  
321      /* kick only if a timer is pending */
322      job_enter_cond_locked(&job->job, job_timer_pending);
323  
324      return true;
325  }
326  
327  static bool block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
328  {
329      JOB_LOCK_GUARD();
330      return block_job_set_speed_locked(job, speed, errp);
331  }
332  
333  void block_job_change_locked(BlockJob *job, BlockJobChangeOptions *opts,
334                               Error **errp)
335  {
336      const BlockJobDriver *drv = block_job_driver(job);
337  
338      GLOBAL_STATE_CODE();
339  
340      if (job_apply_verb_locked(&job->job, JOB_VERB_CHANGE, errp)) {
341          return;
342      }
343  
344      if (drv->change) {
345          job_unlock();
346          drv->change(job, opts, errp);
347          job_lock();
348      } else {
349          error_setg(errp, "Job type does not support change");
350      }
351  }
352  
353  void block_job_ratelimit_processed_bytes(BlockJob *job, uint64_t n)
354  {
355      IO_CODE();
356      ratelimit_calculate_delay(&job->limit, n);
357  }
358  
359  void block_job_ratelimit_sleep(BlockJob *job)
360  {
361      uint64_t delay_ns;
362  
363      /*
364       * Sleep at least once. If the job is reentered early, keep waiting until
365       * we've waited for the full time that is necessary to keep the job at the
366       * right speed.
367       *
368       * Make sure to recalculate the delay after each (possibly interrupted)
369       * sleep because the speed can change while the job has yielded.
370       */
371      do {
372          delay_ns = ratelimit_calculate_delay(&job->limit, 0);
373          job_sleep_ns(&job->job, delay_ns);
374      } while (delay_ns && !job_is_cancelled(&job->job));
375  }
376  
377  BlockJobInfo *block_job_query_locked(BlockJob *job, Error **errp)
378  {
379      BlockJobInfo *info;
380      uint64_t progress_current, progress_total;
381      const BlockJobDriver *drv = block_job_driver(job);
382  
383      GLOBAL_STATE_CODE();
384  
385      if (block_job_is_internal(job)) {
386          error_setg(errp, "Cannot query QEMU internal jobs");
387          return NULL;
388      }
389  
390      progress_get_snapshot(&job->job.progress, &progress_current,
391                            &progress_total);
392  
393      info = g_new0(BlockJobInfo, 1);
394      info->type      = job_type(&job->job);
395      info->device    = g_strdup(job->job.id);
396      info->busy      = job->job.busy;
397      info->paused    = job->job.pause_count > 0;
398      info->offset    = progress_current;
399      info->len       = progress_total;
400      info->speed     = job->speed;
401      info->io_status = job->iostatus;
402      info->ready     = job_is_ready_locked(&job->job),
403      info->status    = job->job.status;
404      info->auto_finalize = job->job.auto_finalize;
405      info->auto_dismiss  = job->job.auto_dismiss;
406      if (job->job.ret) {
407          info->error = job->job.err ?
408                          g_strdup(error_get_pretty(job->job.err)) :
409                          g_strdup(strerror(-job->job.ret));
410      }
411      if (drv->query) {
412          job_unlock();
413          drv->query(job, info);
414          job_lock();
415      }
416      return info;
417  }
418  
419  /* Called with job lock held */
420  static void block_job_iostatus_set_err_locked(BlockJob *job, int error)
421  {
422      if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
423          job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
424                                            BLOCK_DEVICE_IO_STATUS_FAILED;
425      }
426  }
427  
428  /* Called with job_mutex lock held. */
429  static void block_job_event_cancelled_locked(Notifier *n, void *opaque)
430  {
431      BlockJob *job = opaque;
432      uint64_t progress_current, progress_total;
433  
434      if (block_job_is_internal(job)) {
435          return;
436      }
437  
438      progress_get_snapshot(&job->job.progress, &progress_current,
439                            &progress_total);
440  
441      qapi_event_send_block_job_cancelled(job_type(&job->job),
442                                          job->job.id,
443                                          progress_total,
444                                          progress_current,
445                                          job->speed);
446  }
447  
448  /* Called with job_mutex lock held. */
449  static void block_job_event_completed_locked(Notifier *n, void *opaque)
450  {
451      BlockJob *job = opaque;
452      const char *msg = NULL;
453      uint64_t progress_current, progress_total;
454  
455      if (block_job_is_internal(job)) {
456          return;
457      }
458  
459      if (job->job.ret < 0) {
460          msg = error_get_pretty(job->job.err);
461      }
462  
463      progress_get_snapshot(&job->job.progress, &progress_current,
464                            &progress_total);
465  
466      qapi_event_send_block_job_completed(job_type(&job->job),
467                                          job->job.id,
468                                          progress_total,
469                                          progress_current,
470                                          job->speed,
471                                          msg);
472  }
473  
474  /* Called with job_mutex lock held. */
475  static void block_job_event_pending_locked(Notifier *n, void *opaque)
476  {
477      BlockJob *job = opaque;
478  
479      if (block_job_is_internal(job)) {
480          return;
481      }
482  
483      qapi_event_send_block_job_pending(job_type(&job->job),
484                                        job->job.id);
485  }
486  
487  /* Called with job_mutex lock held. */
488  static void block_job_event_ready_locked(Notifier *n, void *opaque)
489  {
490      BlockJob *job = opaque;
491      uint64_t progress_current, progress_total;
492  
493      if (block_job_is_internal(job)) {
494          return;
495      }
496  
497      progress_get_snapshot(&job->job.progress, &progress_current,
498                            &progress_total);
499  
500      qapi_event_send_block_job_ready(job_type(&job->job),
501                                      job->job.id,
502                                      progress_total,
503                                      progress_current,
504                                      job->speed);
505  }
506  
507  
508  void *block_job_create(const char *job_id, const BlockJobDriver *driver,
509                         JobTxn *txn, BlockDriverState *bs, uint64_t perm,
510                         uint64_t shared_perm, int64_t speed, int flags,
511                         BlockCompletionFunc *cb, void *opaque, Error **errp)
512  {
513      BlockJob *job;
514      int ret;
515      GLOBAL_STATE_CODE();
516  
517      bdrv_graph_wrlock(bs);
518  
519      if (job_id == NULL && !(flags & JOB_INTERNAL)) {
520          job_id = bdrv_get_device_name(bs);
521      }
522  
523      job = job_create(job_id, &driver->job_driver, txn, bdrv_get_aio_context(bs),
524                       flags, cb, opaque, errp);
525      if (job == NULL) {
526          bdrv_graph_wrunlock(bs);
527          return NULL;
528      }
529  
530      assert(is_block_job(&job->job));
531      assert(job->job.driver->free == &block_job_free);
532      assert(job->job.driver->user_resume == &block_job_user_resume);
533  
534      ratelimit_init(&job->limit);
535  
536      job->finalize_cancelled_notifier.notify = block_job_event_cancelled_locked;
537      job->finalize_completed_notifier.notify = block_job_event_completed_locked;
538      job->pending_notifier.notify = block_job_event_pending_locked;
539      job->ready_notifier.notify = block_job_event_ready_locked;
540      job->idle_notifier.notify = block_job_on_idle_locked;
541  
542      WITH_JOB_LOCK_GUARD() {
543          notifier_list_add(&job->job.on_finalize_cancelled,
544                            &job->finalize_cancelled_notifier);
545          notifier_list_add(&job->job.on_finalize_completed,
546                            &job->finalize_completed_notifier);
547          notifier_list_add(&job->job.on_pending, &job->pending_notifier);
548          notifier_list_add(&job->job.on_ready, &job->ready_notifier);
549          notifier_list_add(&job->job.on_idle, &job->idle_notifier);
550      }
551  
552      error_setg(&job->blocker, "block device is in use by block job: %s",
553                 job_type_str(&job->job));
554  
555      ret = block_job_add_bdrv(job, "main node", bs, perm, shared_perm, errp);
556      if (ret < 0) {
557          goto fail;
558      }
559  
560      bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
561  
562      if (!block_job_set_speed(job, speed, errp)) {
563          goto fail;
564      }
565  
566      bdrv_graph_wrunlock(bs);
567      return job;
568  
569  fail:
570      bdrv_graph_wrunlock(bs);
571      job_early_fail(&job->job);
572      return NULL;
573  }
574  
575  void block_job_iostatus_reset_locked(BlockJob *job)
576  {
577      GLOBAL_STATE_CODE();
578      if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
579          return;
580      }
581      assert(job->job.user_paused && job->job.pause_count > 0);
582      job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
583  }
584  
585  static void block_job_iostatus_reset(BlockJob *job)
586  {
587      JOB_LOCK_GUARD();
588      block_job_iostatus_reset_locked(job);
589  }
590  
591  void block_job_user_resume(Job *job)
592  {
593      BlockJob *bjob = container_of(job, BlockJob, job);
594      GLOBAL_STATE_CODE();
595      block_job_iostatus_reset(bjob);
596  }
597  
598  BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
599                                          int is_read, int error)
600  {
601      BlockErrorAction action;
602      IO_CODE();
603  
604      switch (on_err) {
605      case BLOCKDEV_ON_ERROR_ENOSPC:
606      case BLOCKDEV_ON_ERROR_AUTO:
607          action = (error == ENOSPC) ?
608                   BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
609          break;
610      case BLOCKDEV_ON_ERROR_STOP:
611          action = BLOCK_ERROR_ACTION_STOP;
612          break;
613      case BLOCKDEV_ON_ERROR_REPORT:
614          action = BLOCK_ERROR_ACTION_REPORT;
615          break;
616      case BLOCKDEV_ON_ERROR_IGNORE:
617          action = BLOCK_ERROR_ACTION_IGNORE;
618          break;
619      default:
620          abort();
621      }
622      if (!block_job_is_internal(job)) {
623          qapi_event_send_block_job_error(job->job.id,
624                                          is_read ? IO_OPERATION_TYPE_READ :
625                                          IO_OPERATION_TYPE_WRITE,
626                                          action);
627      }
628      if (action == BLOCK_ERROR_ACTION_STOP) {
629          WITH_JOB_LOCK_GUARD() {
630              if (!job->job.user_paused) {
631                  job_pause_locked(&job->job);
632                  /*
633                   * make the pause user visible, which will be
634                   * resumed from QMP.
635                   */
636                  job->job.user_paused = true;
637              }
638              block_job_iostatus_set_err_locked(job, error);
639          }
640      }
641      return action;
642  }
643  
644  AioContext *block_job_get_aio_context(BlockJob *job)
645  {
646      GLOBAL_STATE_CODE();
647      return job->job.aio_context;
648  }
649