block/fvd-load.c

   1 /*
   2  * Copyright (c) 2010-2011 IBM
   3  *
   4  * Authors:
   5  *         Chunqiang Tang <ctang@us.ibm.com>
   6  *
   7  * This work is licensed under the terms of the GNU GPL, version 2.
   8  * See the COPYING file in the top-level directory.
   9  */
  10
  11 /*=============================================================================
  12  *  A short description: this FVD module implements loading data from a
  13  *  compact image.
  14  *============================================================================*/
  15
  16 static void aio_wrapper_bh (void *opaque);
  17 static void finish_load_data_from_compact_image (void *opaque, int ret);
  18 static inline FvdAIOCB *init_load_acb (FvdAIOCB * parent_acb,
  19                                        BlockDriverState * bs,
  20                                        int64_t sector_num,
  21                                        QEMUIOVector * orig_qiov, int nb_sectors,
  22                                        BlockDriverCompletionFunc * cb,
  23                                        void *opaque);
  24
  25 static inline BlockDriverAIOCB *load_data (FvdAIOCB * parent_acb,
  26                                            BlockDriverState * bs,
  27                                            int64_t sector_num,
  28                                            QEMUIOVector * orig_qiov,
  29                                            int nb_sectors,
  30                                            BlockDriverCompletionFunc * cb,
  31                                            void *opaque)
  32 {
  33     BDRVFvdState *s = bs->opaque;
  34
  35     if (!s->table) {
  36         /* Load directly since it is not a compact image. */
  37         return bdrv_aio_readv (s->fvd_data, s->data_offset + sector_num,
  38                                orig_qiov, nb_sectors, cb, opaque);
  39     } else {
  40         return load_data_from_compact_image (NULL, parent_acb, bs, sector_num,
  41                                              orig_qiov, nb_sectors, cb, opaque);
  42     }
  43 }
  44
  45 static BlockDriverAIOCB *
  46 load_data_from_compact_image (FvdAIOCB * acb, FvdAIOCB * parent_acb,
  47                               BlockDriverState * bs, int64_t sector_num,
  48                               QEMUIOVector * orig_qiov, int nb_sectors,
  49                               BlockDriverCompletionFunc * cb, void *opaque)
  50 {
  51     BDRVFvdState *s = bs->opaque;
  52     const uint32_t first_chunk = sector_num / s->chunk_size;
  53     const uint32_t last_chunk = (sector_num + nb_sectors - 1) / s->chunk_size;
  54     uint32_t chunk;
  55     int64_t start_sec;
  56     int i;
  57
  58     if (first_chunk == last_chunk) {
  59         goto handle_one_continuous_region;
  60     }
  61
  62     /* Count the number of qiov and iov needed to cover the continuous regions
  63      * of the compact image. */
  64     int iov_index = 0;
  65     size_t iov_left = orig_qiov->iov[0].iov_len;
  66     uint8_t *iov_buf = orig_qiov->iov[0].iov_base;
  67     int nqiov = 0;
  68     int nziov = 0;        /* Number of empty regions. */
  69     int niov = 0;
  70     uint32_t prev = READ_TABLE2 (s->table[first_chunk]);
  71
  72     /* Amount of data in the first chunk. */
  73     int nb = s->chunk_size - (sector_num % s->chunk_size);
  74
  75     for (chunk = first_chunk + 1; chunk <= last_chunk; chunk++) {
  76         uint32_t current = READ_TABLE2 (s->table[chunk]);
  77         int64_t data_size;
  78         if (chunk < last_chunk) {
  79             data_size = s->chunk_size;
  80         } else {
  81             data_size = (sector_num + nb_sectors) % s->chunk_size;
  82             if (data_size == 0) {
  83                 data_size = s->chunk_size;
  84             }
  85         }
  86
  87         if ((IS_EMPTY (current) && IS_EMPTY (prev)) ||
  88             (!IS_EMPTY (prev) && !IS_EMPTY (current) && current == prev + 1)) {
  89             nb += data_size;        /* Belong to the previous continuous region. */
  90         } else {
  91             /* Terminate the previous continuous region. */
  92             if (IS_EMPTY (prev)) {
  93                 /* Skip this empty region. */
  94                 count_iov (orig_qiov->iov, &iov_index, &iov_buf,
  95                            &iov_left, nb * 512);
  96                 nziov++;
  97             } else {
  98                 niov += count_iov (orig_qiov->iov, &iov_index, &iov_buf,
  99                                    &iov_left, nb * 512);
 100                 nqiov++;
 101             }
 102             nb = data_size;        /* Data in the new region. */
 103         }
 104         prev = current;
 105     }
 106
 107     if (nqiov == 0 && nziov == 0) {
 108         /* All data can be read in one qiov. Reuse orig_qiov. */
 109       handle_one_continuous_region:
 110         if (IS_EMPTY (s->table[first_chunk])) {
 111             /* Fill qiov with zeros. */
 112             for (i = 0; i < orig_qiov->niov; i++) {
 113                 memset (orig_qiov->iov[i].iov_base,
 114                         0, orig_qiov->iov[i].iov_len);
 115             }
 116
 117             /* Use a bh to invoke the callback. */
 118             if (!acb) {
 119                 if (!(acb = my_qemu_aio_get (&fvd_aio_pool, bs, cb, opaque))) {
 120                     return NULL;
 121                 }
 122                 COPY_UUID (acb, parent_acb);
 123             }
 124
 125             QDEBUG ("LOAD: acb%llu-%p  load_fill_all_with_zeros\n",
 126                     acb->uuid, acb);
 127             acb->type = OP_WRAPPER;
 128             acb->wrapper.bh = qemu_bh_new (aio_wrapper_bh, acb);
 129             qemu_bh_schedule (acb->wrapper.bh);
 130             return &acb->common;
 131         }
 132
 133         /* A non-empty region. */
 134         start_sec = READ_TABLE (s->table[first_chunk]) * s->chunk_size +
 135                                     (sector_num % s->chunk_size);
 136         if (!acb) {
 137             if (parent_acb) {
 138                 QDEBUG ("LOAD: acb%llu-%p  "
 139                         "load_directly_as_one_continuous_region\n",
 140                         parent_acb->uuid, acb);
 141             }
 142             return bdrv_aio_readv (s->fvd_data, s->data_offset + start_sec,
 143                                    orig_qiov, nb_sectors, cb, opaque);
 144         }
 145
 146         QDEBUG ("LOAD: acb%llu-%p  load_directly_as_one_continuous_region\n",
 147                 acb->uuid, acb);
 148         acb->load.num_children = 1;
 149         acb->load.one_child.hd_acb =
 150             bdrv_aio_readv (s->fvd_data, s->data_offset + start_sec, orig_qiov,
 151                             nb_sectors, finish_load_data_from_compact_image,
 152                             &acb->load.one_child);
 153         if (acb->load.one_child.hd_acb) {
 154             acb->load.one_child.acb = acb;
 155             return &acb->common;
 156         } else {
 157             my_qemu_aio_unref (acb);
 158             return NULL;
 159         }
 160     }
 161
 162     /* qiov for the last continuous region. */
 163     if (!IS_EMPTY (prev)) {
 164         niov += count_iov (orig_qiov->iov, &iov_index, &iov_buf,
 165                            &iov_left, nb * 512);
 166         nqiov++;
 167         ASSERT (iov_index == orig_qiov->niov - 1 && iov_left == 0);
 168     }
 169
 170     /* Need to submit multiple requests to the lower layer. Initialize acb. */
 171     if (!acb && !(acb = init_load_acb (parent_acb, bs, sector_num,
 172                                        orig_qiov, nb_sectors, cb, opaque))) {
 173         return NULL;
 174     }
 175     acb->load.num_children = nqiov;
 176
 177     /* Allocate memory and create multiple requests. */
 178     acb->load.children = my_qemu_malloc ((sizeof (CompactChildCB) +
 179                                           sizeof (QEMUIOVector)) * nqiov +
 180                                          sizeof (struct iovec) * niov);
 181     QEMUIOVector *q = (QEMUIOVector *) (acb->load.children + nqiov);
 182     struct iovec *v = (struct iovec *) (q + nqiov);
 183
 184     /* Set up iov and qiov. */
 185     nqiov = 0;
 186     iov_index = 0;
 187     iov_left = orig_qiov->iov[0].iov_len;
 188     iov_buf = orig_qiov->iov[0].iov_base;
 189     nb = s->chunk_size - (sector_num % s->chunk_size); /* Data in first chunk.*/
 190     prev = READ_TABLE2 (s->table[first_chunk]);
 191
 192     /* if (IS_EMPTY(prev)), start_sec will not be used later, and hence safe. */
 193     start_sec = prev * s->chunk_size + (sector_num % s->chunk_size);
 194
 195     for (chunk = first_chunk + 1; chunk <= last_chunk; chunk++) {
 196         uint32_t current = READ_TABLE2 (s->table[chunk]);
 197         int64_t data_size;
 198         if (chunk < last_chunk) {
 199             data_size = s->chunk_size;
 200         } else {
 201             data_size = (sector_num + nb_sectors) % s->chunk_size;
 202             if (data_size == 0) {
 203                 data_size = s->chunk_size;
 204             }
 205         }
 206
 207         if ((IS_EMPTY (prev) && IS_EMPTY (current)) ||
 208             (!IS_EMPTY (prev) && !IS_EMPTY (current) && current == prev + 1)) {
 209             nb += data_size;        /* Continue the previous region. */
 210         } else {
 211             /* Terminate the previous continuous region. */
 212             if (IS_EMPTY (prev)) {
 213                 zero_iov (orig_qiov->iov, &iov_index, &iov_buf, &iov_left,
 214                           nb * 512);        /* Fill iov data with zeros. */
 215             } else {
 216                 niov = setup_iov (orig_qiov->iov, v, &iov_index, &iov_buf,
 217                                   &iov_left, nb * 512);
 218                 qemu_iovec_init_external (q, v, niov);
 219                 QDEBUG ("LOAD: acb%llu-%p  create_child %d sector_num=%" PRId64
 220                         " nb_sectors=%d niov=%d\n", acb->uuid, acb, nqiov,
 221                         start_sec, nb, niov);
 222                 acb->load.children[nqiov].hd_acb =
 223                     bdrv_aio_readv (s->fvd_data, s->data_offset + start_sec, q,
 224                                     nb, finish_load_data_from_compact_image,
 225                                     &acb->load.children[nqiov]);
 226                 if (!acb->load.children[nqiov].hd_acb) {
 227                     goto fail;
 228                 }
 229                 acb->load.children[nqiov].acb = acb;
 230                 v += niov;
 231                 q++;
 232                 nqiov++;
 233             }
 234
 235             nb = data_size;
 236
 237             /* if (IS_EMPTY(current)), start_sec will not be used later. */
 238             start_sec = current * s->chunk_size;
 239         }
 240         prev = current;
 241     }
 242
 243     /* The last continuous region. */
 244     if (IS_EMPTY (prev)) {
 245         zero_iov (orig_qiov->iov, &iov_index, &iov_buf, &iov_left, nb * 512);
 246     } else {
 247         niov = setup_iov (orig_qiov->iov, v, &iov_index, &iov_buf,
 248                           &iov_left, nb * 512);
 249         qemu_iovec_init_external (q, v, niov);
 250         QDEBUG ("LOAD: acb%llu-%p  create_child %d sector_num=%" PRId64
 251                 " nb_sectors=%d niov=%d\n", acb->uuid, acb, nqiov, start_sec,
 252                 nb, niov);
 253         acb->load.children[nqiov].hd_acb =
 254             bdrv_aio_readv (s->fvd_data, s->data_offset + start_sec, q, nb,
 255                             finish_load_data_from_compact_image,
 256                             &acb->load.children[nqiov]);
 257         if (!acb->load.children[nqiov].hd_acb) {
 258             goto fail;
 259         }
 260         acb->load.children[nqiov].acb = acb;
 261     }
 262     ASSERT (iov_index == orig_qiov->niov - 1 && iov_left == 0);
 263
 264     return &acb->common;
 265
 266   fail:
 267     for (i = 0; i < nqiov; i++) {
 268         bdrv_aio_cancel (acb->load.children[i].hd_acb);
 269     }
 270     my_qemu_free (acb->load.children);
 271     my_qemu_aio_unref (acb);
 272     return NULL;
 273 }
 274
 275 static void aio_wrapper_bh (void *opaque)
 276 {
 277     FvdAIOCB *acb = opaque;
 278     acb->common.cb (acb->common.opaque, 0);
 279     qemu_bh_delete (acb->wrapper.bh);
 280     my_qemu_aio_unref (acb);
 281 }
 282
 283 static void finish_load_data_from_compact_image (void *opaque, int ret)
 284 {
 285     CompactChildCB *child = opaque;
 286     FvdAIOCB *acb = child->acb;
 287
 288     /* Now fvd_store_compact_cancel(), if invoked, won't cancel this child
 289      * request. */
 290     child->hd_acb = NULL;
 291
 292     if (acb->load.ret == 0) {
 293         acb->load.ret = ret;
 294     } else {
 295         QDEBUG ("LOAD: acb%llu-%p  load_child=%d total_children=%d "
 296                 "error ret=%d\n", acb->uuid, acb, acb->load.finished_children,
 297                 acb->load.num_children, ret);
 298     }
 299
 300     acb->load.finished_children++;
 301     if (acb->load.finished_children < acb->load.num_children) {
 302         QDEBUG ("LOAD: acb%llu-%p  load_finished_children=%d "
 303                 "total_children=%d\n", acb->uuid, acb,
 304                 acb->load.finished_children, acb->load.num_children);
 305         return;
 306     }
 307
 308     QDEBUG ("LOAD: acb%llu-%p  load_last_child_finished ret=%d\n", acb->uuid,
 309             acb, acb->load.ret);
 310     acb->common.cb (acb->common.opaque, acb->load.ret);
 311     if (acb->load.children) {
 312         my_qemu_free (acb->load.children);
 313     }
 314     my_qemu_aio_unref (acb);
 315 }
 316
 317 static inline FvdAIOCB *init_load_acb (FvdAIOCB * parent_acb,
 318                                        BlockDriverState * bs,
 319                                        int64_t sector_num,
 320                                        QEMUIOVector * orig_qiov,
 321                                        int nb_sectors,
 322                                        BlockDriverCompletionFunc * cb,
 323                                        void *opaque)
 324 {
 325     FvdAIOCB *const acb = my_qemu_aio_get (&fvd_aio_pool, bs, cb, opaque);
 326     if (!acb) {
 327         return NULL;
 328     }
 329     acb->type = OP_LOAD_COMPACT;
 330     acb->sector_num = sector_num;
 331     acb->nb_sectors = nb_sectors;
 332     acb->load.parent_acb = parent_acb;
 333     acb->load.finished_children = 0;
 334     acb->load.children = NULL;
 335     acb->load.one_child.hd_acb = NULL;
 336     acb->load.orig_qiov = orig_qiov;
 337     acb->load.ret = 0;
 338     COPY_UUID (acb, parent_acb);
 339     return acb;
 340 }
 341
 342 #if 0
 343 static void fvd_wrapper_cancel (FvdAIOCB * acb)
 344 {
 345     qemu_bh_cancel (acb->wrapper.bh);
 346     qemu_bh_delete (acb->wrapper.bh);
 347     my_qemu_aio_unref (acb);
 348 }
 349
 350 static void fvd_load_compact_cancel (FvdAIOCB * acb)
 351 {
 352     if (acb->load.children) {
 353         int i;
 354         for (i = 0; i < acb->load.num_children; i++) {
 355             if (acb->load.children[i].hd_acb) {
 356                 bdrv_aio_cancel (acb->load.children[i].hd_acb);
 357             }
 358         }
 359         my_qemu_free (acb->load.children);
 360     }
 361     if (acb->load.one_child.hd_acb) {
 362         bdrv_aio_cancel (acb->load.one_child.hd_acb);
 363     }
 364     my_qemu_aio_unref (acb);
 365 }
 366 #endif
 367