blockjob.c

   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 "qemu-common.h"
  28 #include "trace.h"
  29 #include "block/block.h"
  30 #include "block/blockjob.h"
  31 #include "block/block_int.h"
  32 #include "sysemu/block-backend.h"
  33 #include "qapi/qmp/qerror.h"
  34 #include "qapi/qmp/qjson.h"
  35 #include "qemu/coroutine.h"
  36 #include "qemu/id.h"
  37 #include "qmp-commands.h"
  38 #include "qemu/timer.h"
  39 #include "qapi-event.h"
  40
  41 /* Transactional group of block jobs */
  42 struct BlockJobTxn {
  43
  44     /* Is this txn being cancelled? */
  45     bool aborting;
  46
  47     /* List of jobs */
  48     QLIST_HEAD(, BlockJob) jobs;
  49
  50     /* Reference count */
  51     int refcnt;
  52 };
  53
  54 static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs);
  55
  56 BlockJob *block_job_next(BlockJob *job)
  57 {
  58     if (!job) {
  59         return QLIST_FIRST(&block_jobs);
  60     }
  61     return QLIST_NEXT(job, job_list);
  62 }
  63
  64 BlockJob *block_job_get(const char *id)
  65 {
  66     BlockJob *job;
  67
  68     QLIST_FOREACH(job, &block_jobs, job_list) {
  69         if (!strcmp(id, job->id)) {
  70             return job;
  71         }
  72     }
  73
  74     return NULL;
  75 }
  76
  77 /* Normally the job runs in its BlockBackend's AioContext.  The exception is
  78  * block_job_defer_to_main_loop() where it runs in the QEMU main loop.  Code
  79  * that supports both cases uses this helper function.
  80  */
  81 static AioContext *block_job_get_aio_context(BlockJob *job)
  82 {
  83     return job->deferred_to_main_loop ?
  84            qemu_get_aio_context() :
  85            blk_get_aio_context(job->blk);
  86 }
  87
  88 static void block_job_attached_aio_context(AioContext *new_context,
  89                                            void *opaque)
  90 {
  91     BlockJob *job = opaque;
  92
  93     if (job->driver->attached_aio_context) {
  94         job->driver->attached_aio_context(job, new_context);
  95     }
  96
  97     block_job_resume(job);
  98 }
  99
 100 static void block_job_detach_aio_context(void *opaque)
 101 {
 102     BlockJob *job = opaque;
 103
 104     /* In case the job terminates during aio_poll()... */
 105     block_job_ref(job);
 106
 107     block_job_pause(job);
 108
 109     if (!job->paused) {
 110         /* If job is !job->busy this kicks it into the next pause point. */
 111         block_job_enter(job);
 112     }
 113     while (!job->paused && !job->completed) {
 114         aio_poll(block_job_get_aio_context(job), true);
 115     }
 116
 117     block_job_unref(job);
 118 }
 119
 120 void *block_job_create(const char *job_id, const BlockJobDriver *driver,
 121                        BlockDriverState *bs, int64_t speed,
 122                        BlockCompletionFunc *cb, void *opaque, Error **errp)
 123 {
 124     BlockBackend *blk;
 125     BlockJob *job;
 126
 127     assert(cb);
 128     if (bs->job) {
 129         error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
 130         return NULL;
 131     }
 132
 133     if (job_id == NULL) {
 134         job_id = bdrv_get_device_name(bs);
 135         /* Assign a default ID if the BDS does not have a device
 136          * name. We'll get rid of this soon when we finish extending
 137          * the API of all commands that create block jobs. */
 138         if (job_id[0] == '\0') {
 139             job_id = "default_job";
 140         }
 141     }
 142
 143     if (!id_wellformed(job_id)) {
 144         error_setg(errp, "Invalid job ID '%s'", job_id);
 145         return NULL;
 146     }
 147
 148     if (block_job_get(job_id)) {
 149         error_setg(errp, "Job ID '%s' already in use", job_id);
 150         return NULL;
 151     }
 152
 153     blk = blk_new();
 154     blk_insert_bs(blk, bs);
 155
 156     job = g_malloc0(driver->instance_size);
 157     error_setg(&job->blocker, "block device is in use by block job: %s",
 158                BlockJobType_lookup[driver->job_type]);
 159     bdrv_op_block_all(bs, job->blocker);
 160     bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
 161
 162     job->driver        = driver;
 163     job->id            = g_strdup(job_id);
 164     job->blk           = blk;
 165     job->cb            = cb;
 166     job->opaque        = opaque;
 167     job->busy          = true;
 168     job->refcnt        = 1;
 169     bs->job = job;
 170
 171     QLIST_INSERT_HEAD(&block_jobs, job, job_list);
 172
 173     blk_add_aio_context_notifier(blk, block_job_attached_aio_context,
 174                                  block_job_detach_aio_context, job);
 175
 176     /* Only set speed when necessary to avoid NotSupported error */
 177     if (speed != 0) {
 178         Error *local_err = NULL;
 179
 180         block_job_set_speed(job, speed, &local_err);
 181         if (local_err) {
 182             block_job_unref(job);
 183             error_propagate(errp, local_err);
 184             return NULL;
 185         }
 186     }
 187     return job;
 188 }
 189
 190 void block_job_ref(BlockJob *job)
 191 {
 192     ++job->refcnt;
 193 }
 194
 195 void block_job_unref(BlockJob *job)
 196 {
 197     if (--job->refcnt == 0) {
 198         BlockDriverState *bs = blk_bs(job->blk);
 199         bs->job = NULL;
 200         bdrv_op_unblock_all(bs, job->blocker);
 201         blk_remove_aio_context_notifier(job->blk,
 202                                         block_job_attached_aio_context,
 203                                         block_job_detach_aio_context, job);
 204         blk_unref(job->blk);
 205         error_free(job->blocker);
 206         g_free(job->id);
 207         QLIST_REMOVE(job, job_list);
 208         g_free(job);
 209     }
 210 }
 211
 212 static void block_job_completed_single(BlockJob *job)
 213 {
 214     if (!job->ret) {
 215         if (job->driver->commit) {
 216             job->driver->commit(job);
 217         }
 218     } else {
 219         if (job->driver->abort) {
 220             job->driver->abort(job);
 221         }
 222     }
 223     job->cb(job->opaque, job->ret);
 224     if (job->txn) {
 225         block_job_txn_unref(job->txn);
 226     }
 227     block_job_unref(job);
 228 }
 229
 230 static void block_job_completed_txn_abort(BlockJob *job)
 231 {
 232     AioContext *ctx;
 233     BlockJobTxn *txn = job->txn;
 234     BlockJob *other_job, *next;
 235
 236     if (txn->aborting) {
 237         /*
 238          * We are cancelled by another job, which will handle everything.
 239          */
 240         return;
 241     }
 242     txn->aborting = true;
 243     /* We are the first failed job. Cancel other jobs. */
 244     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 245         ctx = blk_get_aio_context(other_job->blk);
 246         aio_context_acquire(ctx);
 247     }
 248     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 249         if (other_job == job || other_job->completed) {
 250             /* Other jobs are "effectively" cancelled by us, set the status for
 251              * them; this job, however, may or may not be cancelled, depending
 252              * on the caller, so leave it. */
 253             if (other_job != job) {
 254                 other_job->cancelled = true;
 255             }
 256             continue;
 257         }
 258         block_job_cancel_sync(other_job);
 259         assert(other_job->completed);
 260     }
 261     QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
 262         ctx = blk_get_aio_context(other_job->blk);
 263         block_job_completed_single(other_job);
 264         aio_context_release(ctx);
 265     }
 266 }
 267
 268 static void block_job_completed_txn_success(BlockJob *job)
 269 {
 270     AioContext *ctx;
 271     BlockJobTxn *txn = job->txn;
 272     BlockJob *other_job, *next;
 273     /*
 274      * Successful completion, see if there are other running jobs in this
 275      * txn.
 276      */
 277     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 278         if (!other_job->completed) {
 279             return;
 280         }
 281     }
 282     /* We are the last completed job, commit the transaction. */
 283     QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
 284         ctx = blk_get_aio_context(other_job->blk);
 285         aio_context_acquire(ctx);
 286         assert(other_job->ret == 0);
 287         block_job_completed_single(other_job);
 288         aio_context_release(ctx);
 289     }
 290 }
 291
 292 void block_job_completed(BlockJob *job, int ret)
 293 {
 294     assert(blk_bs(job->blk)->job == job);
 295     assert(!job->completed);
 296     job->completed = true;
 297     job->ret = ret;
 298     if (!job->txn) {
 299         block_job_completed_single(job);
 300     } else if (ret < 0 || block_job_is_cancelled(job)) {
 301         block_job_completed_txn_abort(job);
 302     } else {
 303         block_job_completed_txn_success(job);
 304     }
 305 }
 306
 307 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
 308 {
 309     Error *local_err = NULL;
 310
 311     if (!job->driver->set_speed) {
 312         error_setg(errp, QERR_UNSUPPORTED);
 313         return;
 314     }
 315     job->driver->set_speed(job, speed, &local_err);
 316     if (local_err) {
 317         error_propagate(errp, local_err);
 318         return;
 319     }
 320
 321     job->speed = speed;
 322 }
 323
 324 void block_job_complete(BlockJob *job, Error **errp)
 325 {
 326     if (job->pause_count || job->cancelled || !job->driver->complete) {
 327         error_setg(errp, "The active block job '%s' cannot be completed",
 328                    job->id);
 329         return;
 330     }
 331
 332     job->driver->complete(job, errp);
 333 }
 334
 335 void block_job_pause(BlockJob *job)
 336 {
 337     job->pause_count++;
 338 }
 339
 340 static bool block_job_should_pause(BlockJob *job)
 341 {
 342     return job->pause_count > 0;
 343 }
 344
 345 void coroutine_fn block_job_pause_point(BlockJob *job)
 346 {
 347     if (!block_job_should_pause(job)) {
 348         return;
 349     }
 350     if (block_job_is_cancelled(job)) {
 351         return;
 352     }
 353
 354     if (job->driver->pause) {
 355         job->driver->pause(job);
 356     }
 357
 358     if (block_job_should_pause(job) && !block_job_is_cancelled(job)) {
 359         job->paused = true;
 360         job->busy = false;
 361         qemu_coroutine_yield(); /* wait for block_job_resume() */
 362         job->busy = true;
 363         job->paused = false;
 364     }
 365
 366     if (job->driver->resume) {
 367         job->driver->resume(job);
 368     }
 369 }
 370
 371 void block_job_resume(BlockJob *job)
 372 {
 373     assert(job->pause_count > 0);
 374     job->pause_count--;
 375     if (job->pause_count) {
 376         return;
 377     }
 378     block_job_enter(job);
 379 }
 380
 381 void block_job_enter(BlockJob *job)
 382 {
 383     if (job->co && !job->busy) {
 384         qemu_coroutine_enter(job->co, NULL);
 385     }
 386 }
 387
 388 void block_job_cancel(BlockJob *job)
 389 {
 390     job->cancelled = true;
 391     block_job_iostatus_reset(job);
 392     block_job_enter(job);
 393 }
 394
 395 bool block_job_is_cancelled(BlockJob *job)
 396 {
 397     return job->cancelled;
 398 }
 399
 400 void block_job_iostatus_reset(BlockJob *job)
 401 {
 402     job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
 403     if (job->driver->iostatus_reset) {
 404         job->driver->iostatus_reset(job);
 405     }
 406 }
 407
 408 static int block_job_finish_sync(BlockJob *job,
 409                                  void (*finish)(BlockJob *, Error **errp),
 410                                  Error **errp)
 411 {
 412     Error *local_err = NULL;
 413     int ret;
 414
 415     assert(blk_bs(job->blk)->job == job);
 416
 417     block_job_ref(job);
 418     finish(job, &local_err);
 419     if (local_err) {
 420         error_propagate(errp, local_err);
 421         block_job_unref(job);
 422         return -EBUSY;
 423     }
 424     while (!job->completed) {
 425         aio_poll(block_job_get_aio_context(job), true);
 426     }
 427     ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret;
 428     block_job_unref(job);
 429     return ret;
 430 }
 431
 432 /* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
 433  * used with block_job_finish_sync() without the need for (rather nasty)
 434  * function pointer casts there. */
 435 static void block_job_cancel_err(BlockJob *job, Error **errp)
 436 {
 437     block_job_cancel(job);
 438 }
 439
 440 int block_job_cancel_sync(BlockJob *job)
 441 {
 442     return block_job_finish_sync(job, &block_job_cancel_err, NULL);
 443 }
 444
 445 void block_job_cancel_sync_all(void)
 446 {
 447     BlockJob *job;
 448     AioContext *aio_context;
 449
 450     while ((job = QLIST_FIRST(&block_jobs))) {
 451         aio_context = blk_get_aio_context(job->blk);
 452         aio_context_acquire(aio_context);
 453         block_job_cancel_sync(job);
 454         aio_context_release(aio_context);
 455     }
 456 }
 457
 458 int block_job_complete_sync(BlockJob *job, Error **errp)
 459 {
 460     return block_job_finish_sync(job, &block_job_complete, errp);
 461 }
 462
 463 void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
 464 {
 465     assert(job->busy);
 466
 467     /* Check cancellation *before* setting busy = false, too!  */
 468     if (block_job_is_cancelled(job)) {
 469         return;
 470     }
 471
 472     job->busy = false;
 473     if (!block_job_should_pause(job)) {
 474         co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns);
 475     }
 476     job->busy = true;
 477
 478     block_job_pause_point(job);
 479 }
 480
 481 void block_job_yield(BlockJob *job)
 482 {
 483     assert(job->busy);
 484
 485     /* Check cancellation *before* setting busy = false, too!  */
 486     if (block_job_is_cancelled(job)) {
 487         return;
 488     }
 489
 490     job->busy = false;
 491     if (!block_job_should_pause(job)) {
 492         qemu_coroutine_yield();
 493     }
 494     job->busy = true;
 495
 496     block_job_pause_point(job);
 497 }
 498
 499 BlockJobInfo *block_job_query(BlockJob *job)
 500 {
 501     BlockJobInfo *info = g_new0(BlockJobInfo, 1);
 502     info->type      = g_strdup(BlockJobType_lookup[job->driver->job_type]);
 503     info->device    = g_strdup(job->id);
 504     info->len       = job->len;
 505     info->busy      = job->busy;
 506     info->paused    = job->pause_count > 0;
 507     info->offset    = job->offset;
 508     info->speed     = job->speed;
 509     info->io_status = job->iostatus;
 510     info->ready     = job->ready;
 511     return info;
 512 }
 513
 514 static void block_job_iostatus_set_err(BlockJob *job, int error)
 515 {
 516     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
 517         job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
 518                                           BLOCK_DEVICE_IO_STATUS_FAILED;
 519     }
 520 }
 521
 522 void block_job_event_cancelled(BlockJob *job)
 523 {
 524     qapi_event_send_block_job_cancelled(job->driver->job_type,
 525                                         job->id,
 526                                         job->len,
 527                                         job->offset,
 528                                         job->speed,
 529                                         &error_abort);
 530 }
 531
 532 void block_job_event_completed(BlockJob *job, const char *msg)
 533 {
 534     qapi_event_send_block_job_completed(job->driver->job_type,
 535                                         job->id,
 536                                         job->len,
 537                                         job->offset,
 538                                         job->speed,
 539                                         !!msg,
 540                                         msg,
 541                                         &error_abort);
 542 }
 543
 544 void block_job_event_ready(BlockJob *job)
 545 {
 546     job->ready = true;
 547
 548     qapi_event_send_block_job_ready(job->driver->job_type,
 549                                     job->id,
 550                                     job->len,
 551                                     job->offset,
 552                                     job->speed, &error_abort);
 553 }
 554
 555 BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
 556                                         int is_read, int error)
 557 {
 558     BlockErrorAction action;
 559
 560     switch (on_err) {
 561     case BLOCKDEV_ON_ERROR_ENOSPC:
 562         action = (error == ENOSPC) ?
 563                  BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
 564         break;
 565     case BLOCKDEV_ON_ERROR_STOP:
 566         action = BLOCK_ERROR_ACTION_STOP;
 567         break;
 568     case BLOCKDEV_ON_ERROR_REPORT:
 569         action = BLOCK_ERROR_ACTION_REPORT;
 570         break;
 571     case BLOCKDEV_ON_ERROR_IGNORE:
 572         action = BLOCK_ERROR_ACTION_IGNORE;
 573         break;
 574     default:
 575         abort();
 576     }
 577     qapi_event_send_block_job_error(job->id,
 578                                     is_read ? IO_OPERATION_TYPE_READ :
 579                                     IO_OPERATION_TYPE_WRITE,
 580                                     action, &error_abort);
 581     if (action == BLOCK_ERROR_ACTION_STOP) {
 582         /* make the pause user visible, which will be resumed from QMP. */
 583         job->user_paused = true;
 584         block_job_pause(job);
 585         block_job_iostatus_set_err(job, error);
 586     }
 587     return action;
 588 }
 589
 590 typedef struct {
 591     BlockJob *job;
 592     QEMUBH *bh;
 593     AioContext *aio_context;
 594     BlockJobDeferToMainLoopFn *fn;
 595     void *opaque;
 596 } BlockJobDeferToMainLoopData;
 597
 598 static void block_job_defer_to_main_loop_bh(void *opaque)
 599 {
 600     BlockJobDeferToMainLoopData *data = opaque;
 601     AioContext *aio_context;
 602
 603     qemu_bh_delete(data->bh);
 604
 605     /* Prevent race with block_job_defer_to_main_loop() */
 606     aio_context_acquire(data->aio_context);
 607
 608     /* Fetch BDS AioContext again, in case it has changed */
 609     aio_context = blk_get_aio_context(data->job->blk);
 610     aio_context_acquire(aio_context);
 611
 612     data->job->deferred_to_main_loop = false;
 613     data->fn(data->job, data->opaque);
 614
 615     aio_context_release(aio_context);
 616
 617     aio_context_release(data->aio_context);
 618
 619     g_free(data);
 620 }
 621
 622 void block_job_defer_to_main_loop(BlockJob *job,
 623                                   BlockJobDeferToMainLoopFn *fn,
 624                                   void *opaque)
 625 {
 626     BlockJobDeferToMainLoopData *data = g_malloc(sizeof(*data));
 627     data->job = job;
 628     data->bh = qemu_bh_new(block_job_defer_to_main_loop_bh, data);
 629     data->aio_context = blk_get_aio_context(job->blk);
 630     data->fn = fn;
 631     data->opaque = opaque;
 632     job->deferred_to_main_loop = true;
 633
 634     qemu_bh_schedule(data->bh);
 635 }
 636
 637 BlockJobTxn *block_job_txn_new(void)
 638 {
 639     BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
 640     QLIST_INIT(&txn->jobs);
 641     txn->refcnt = 1;
 642     return txn;
 643 }
 644
 645 static void block_job_txn_ref(BlockJobTxn *txn)
 646 {
 647     txn->refcnt++;
 648 }
 649
 650 void block_job_txn_unref(BlockJobTxn *txn)
 651 {
 652     if (txn && --txn->refcnt == 0) {
 653         g_free(txn);
 654     }
 655 }
 656
 657 void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
 658 {
 659     if (!txn) {
 660         return;
 661     }
 662
 663     assert(!job->txn);
 664     job->txn = txn;
 665
 666     QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
 667     block_job_txn_ref(txn);
 668 }