block/sheepdog.c

   1 /*
   2  * Copyright (C) 2009-2010 Nippon Telegraph and Telephone Corporation.
   3  *
   4  * This program is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU General Public License version
   6  * 2 as published by the Free Software Foundation.
   7  *
   8  * You should have received a copy of the GNU General Public License
   9  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  10  *
  11  * Contributions after 2012-01-13 are licensed under the terms of the
  12  * GNU GPL, version 2 or (at your option) any later version.
  13  */
  14
  15 #include "qemu/osdep.h"
  16 #include "qapi/error.h"
  17 #include "qemu/uri.h"
  18 #include "qemu/error-report.h"
  19 #include "qemu/sockets.h"
  20 #include "block/block_int.h"
  21 #include "sysemu/block-backend.h"
  22 #include "qemu/bitops.h"
  23 #include "qemu/cutils.h"
  24
  25 #define SD_PROTO_VER 0x01
  26
  27 #define SD_DEFAULT_ADDR "localhost"
  28 #define SD_DEFAULT_PORT 7000
  29
  30 #define SD_OP_CREATE_AND_WRITE_OBJ  0x01
  31 #define SD_OP_READ_OBJ       0x02
  32 #define SD_OP_WRITE_OBJ      0x03
  33 /* 0x04 is used internally by Sheepdog */
  34
  35 #define SD_OP_NEW_VDI        0x11
  36 #define SD_OP_LOCK_VDI       0x12
  37 #define SD_OP_RELEASE_VDI    0x13
  38 #define SD_OP_GET_VDI_INFO   0x14
  39 #define SD_OP_READ_VDIS      0x15
  40 #define SD_OP_FLUSH_VDI      0x16
  41 #define SD_OP_DEL_VDI        0x17
  42 #define SD_OP_GET_CLUSTER_DEFAULT   0x18
  43
  44 #define SD_FLAG_CMD_WRITE    0x01
  45 #define SD_FLAG_CMD_COW      0x02
  46 #define SD_FLAG_CMD_CACHE    0x04 /* Writeback mode for cache */
  47 #define SD_FLAG_CMD_DIRECT   0x08 /* Don't use cache */
  48
  49 #define SD_RES_SUCCESS       0x00 /* Success */
  50 #define SD_RES_UNKNOWN       0x01 /* Unknown error */
  51 #define SD_RES_NO_OBJ        0x02 /* No object found */
  52 #define SD_RES_EIO           0x03 /* I/O error */
  53 #define SD_RES_VDI_EXIST     0x04 /* Vdi exists already */
  54 #define SD_RES_INVALID_PARMS 0x05 /* Invalid parameters */
  55 #define SD_RES_SYSTEM_ERROR  0x06 /* System error */
  56 #define SD_RES_VDI_LOCKED    0x07 /* Vdi is locked */
  57 #define SD_RES_NO_VDI        0x08 /* No vdi found */
  58 #define SD_RES_NO_BASE_VDI   0x09 /* No base vdi found */
  59 #define SD_RES_VDI_READ      0x0A /* Cannot read requested vdi */
  60 #define SD_RES_VDI_WRITE     0x0B /* Cannot write requested vdi */
  61 #define SD_RES_BASE_VDI_READ 0x0C /* Cannot read base vdi */
  62 #define SD_RES_BASE_VDI_WRITE   0x0D /* Cannot write base vdi */
  63 #define SD_RES_NO_TAG        0x0E /* Requested tag is not found */
  64 #define SD_RES_STARTUP       0x0F /* Sheepdog is on starting up */
  65 #define SD_RES_VDI_NOT_LOCKED   0x10 /* Vdi is not locked */
  66 #define SD_RES_SHUTDOWN      0x11 /* Sheepdog is shutting down */
  67 #define SD_RES_NO_MEM        0x12 /* Cannot allocate memory */
  68 #define SD_RES_FULL_VDI      0x13 /* we already have the maximum vdis */
  69 #define SD_RES_VER_MISMATCH  0x14 /* Protocol version mismatch */
  70 #define SD_RES_NO_SPACE      0x15 /* Server has no room for new objects */
  71 #define SD_RES_WAIT_FOR_FORMAT  0x16 /* Waiting for a format operation */
  72 #define SD_RES_WAIT_FOR_JOIN    0x17 /* Waiting for other nodes joining */
  73 #define SD_RES_JOIN_FAILED   0x18 /* Target node had failed to join sheepdog */
  74 #define SD_RES_HALT          0x19 /* Sheepdog is stopped serving IO request */
  75 #define SD_RES_READONLY      0x1A /* Object is read-only */
  76
  77 /*
  78  * Object ID rules
  79  *
  80  *  0 - 19 (20 bits): data object space
  81  * 20 - 31 (12 bits): reserved data object space
  82  * 32 - 55 (24 bits): vdi object space
  83  * 56 - 59 ( 4 bits): reserved vdi object space
  84  * 60 - 63 ( 4 bits): object type identifier space
  85  */
  86
  87 #define VDI_SPACE_SHIFT   32
  88 #define VDI_BIT (UINT64_C(1) << 63)
  89 #define VMSTATE_BIT (UINT64_C(1) << 62)
  90 #define MAX_DATA_OBJS (UINT64_C(1) << 20)
  91 #define MAX_CHILDREN 1024
  92 #define SD_MAX_VDI_LEN 256
  93 #define SD_MAX_VDI_TAG_LEN 256
  94 #define SD_NR_VDIS   (1U << 24)
  95 #define SD_DATA_OBJ_SIZE (UINT64_C(1) << 22)
  96 #define SD_MAX_VDI_SIZE (SD_DATA_OBJ_SIZE * MAX_DATA_OBJS)
  97 #define SD_DEFAULT_BLOCK_SIZE_SHIFT 22
  98 /*
  99  * For erasure coding, we use at most SD_EC_MAX_STRIP for data strips and
 100  * (SD_EC_MAX_STRIP - 1) for parity strips
 101  *
 102  * SD_MAX_COPIES is sum of number of data strips and parity strips.
 103  */
 104 #define SD_EC_MAX_STRIP 16
 105 #define SD_MAX_COPIES (SD_EC_MAX_STRIP * 2 - 1)
 106
 107 #define SD_INODE_SIZE (sizeof(SheepdogInode))
 108 #define CURRENT_VDI_ID 0
 109
 110 #define LOCK_TYPE_NORMAL 0
 111 #define LOCK_TYPE_SHARED 1      /* for iSCSI multipath */
 112
 113 typedef struct SheepdogReq {
 114     uint8_t proto_ver;
 115     uint8_t opcode;
 116     uint16_t flags;
 117     uint32_t epoch;
 118     uint32_t id;
 119     uint32_t data_length;
 120     uint32_t opcode_specific[8];
 121 } SheepdogReq;
 122
 123 typedef struct SheepdogRsp {
 124     uint8_t proto_ver;
 125     uint8_t opcode;
 126     uint16_t flags;
 127     uint32_t epoch;
 128     uint32_t id;
 129     uint32_t data_length;
 130     uint32_t result;
 131     uint32_t opcode_specific[7];
 132 } SheepdogRsp;
 133
 134 typedef struct SheepdogObjReq {
 135     uint8_t proto_ver;
 136     uint8_t opcode;
 137     uint16_t flags;
 138     uint32_t epoch;
 139     uint32_t id;
 140     uint32_t data_length;
 141     uint64_t oid;
 142     uint64_t cow_oid;
 143     uint8_t copies;
 144     uint8_t copy_policy;
 145     uint8_t reserved[6];
 146     uint64_t offset;
 147 } SheepdogObjReq;
 148
 149 typedef struct SheepdogObjRsp {
 150     uint8_t proto_ver;
 151     uint8_t opcode;
 152     uint16_t flags;
 153     uint32_t epoch;
 154     uint32_t id;
 155     uint32_t data_length;
 156     uint32_t result;
 157     uint8_t copies;
 158     uint8_t copy_policy;
 159     uint8_t reserved[2];
 160     uint32_t pad[6];
 161 } SheepdogObjRsp;
 162
 163 typedef struct SheepdogVdiReq {
 164     uint8_t proto_ver;
 165     uint8_t opcode;
 166     uint16_t flags;
 167     uint32_t epoch;
 168     uint32_t id;
 169     uint32_t data_length;
 170     uint64_t vdi_size;
 171     uint32_t base_vdi_id;
 172     uint8_t copies;
 173     uint8_t copy_policy;
 174     uint8_t store_policy;
 175     uint8_t block_size_shift;
 176     uint32_t snapid;
 177     uint32_t type;
 178     uint32_t pad[2];
 179 } SheepdogVdiReq;
 180
 181 typedef struct SheepdogVdiRsp {
 182     uint8_t proto_ver;
 183     uint8_t opcode;
 184     uint16_t flags;
 185     uint32_t epoch;
 186     uint32_t id;
 187     uint32_t data_length;
 188     uint32_t result;
 189     uint32_t rsvd;
 190     uint32_t vdi_id;
 191     uint32_t pad[5];
 192 } SheepdogVdiRsp;
 193
 194 typedef struct SheepdogClusterRsp {
 195     uint8_t proto_ver;
 196     uint8_t opcode;
 197     uint16_t flags;
 198     uint32_t epoch;
 199     uint32_t id;
 200     uint32_t data_length;
 201     uint32_t result;
 202     uint8_t nr_copies;
 203     uint8_t copy_policy;
 204     uint8_t block_size_shift;
 205     uint8_t __pad1;
 206     uint32_t __pad2[6];
 207 } SheepdogClusterRsp;
 208
 209 typedef struct SheepdogInode {
 210     char name[SD_MAX_VDI_LEN];
 211     char tag[SD_MAX_VDI_TAG_LEN];
 212     uint64_t ctime;
 213     uint64_t snap_ctime;
 214     uint64_t vm_clock_nsec;
 215     uint64_t vdi_size;
 216     uint64_t vm_state_size;
 217     uint16_t copy_policy;
 218     uint8_t nr_copies;
 219     uint8_t block_size_shift;
 220     uint32_t snap_id;
 221     uint32_t vdi_id;
 222     uint32_t parent_vdi_id;
 223     uint32_t child_vdi_id[MAX_CHILDREN];
 224     uint32_t data_vdi_id[MAX_DATA_OBJS];
 225 } SheepdogInode;
 226
 227 #define SD_INODE_HEADER_SIZE offsetof(SheepdogInode, data_vdi_id)
 228
 229 /*
 230  * 64 bit FNV-1a non-zero initial basis
 231  */
 232 #define FNV1A_64_INIT ((uint64_t)0xcbf29ce484222325ULL)
 233
 234 /*
 235  * 64 bit Fowler/Noll/Vo FNV-1a hash code
 236  */
 237 static inline uint64_t fnv_64a_buf(void *buf, size_t len, uint64_t hval)
 238 {
 239     unsigned char *bp = buf;
 240     unsigned char *be = bp + len;
 241     while (bp < be) {
 242         hval ^= (uint64_t) *bp++;
 243         hval += (hval << 1) + (hval << 4) + (hval << 5) +
 244             (hval << 7) + (hval << 8) + (hval << 40);
 245     }
 246     return hval;
 247 }
 248
 249 static inline bool is_data_obj_writable(SheepdogInode *inode, unsigned int idx)
 250 {
 251     return inode->vdi_id == inode->data_vdi_id[idx];
 252 }
 253
 254 static inline bool is_data_obj(uint64_t oid)
 255 {
 256     return !(VDI_BIT & oid);
 257 }
 258
 259 static inline uint64_t data_oid_to_idx(uint64_t oid)
 260 {
 261     return oid & (MAX_DATA_OBJS - 1);
 262 }
 263
 264 static inline uint32_t oid_to_vid(uint64_t oid)
 265 {
 266     return (oid & ~VDI_BIT) >> VDI_SPACE_SHIFT;
 267 }
 268
 269 static inline uint64_t vid_to_vdi_oid(uint32_t vid)
 270 {
 271     return VDI_BIT | ((uint64_t)vid << VDI_SPACE_SHIFT);
 272 }
 273
 274 static inline uint64_t vid_to_vmstate_oid(uint32_t vid, uint32_t idx)
 275 {
 276     return VMSTATE_BIT | ((uint64_t)vid << VDI_SPACE_SHIFT) | idx;
 277 }
 278
 279 static inline uint64_t vid_to_data_oid(uint32_t vid, uint32_t idx)
 280 {
 281     return ((uint64_t)vid << VDI_SPACE_SHIFT) | idx;
 282 }
 283
 284 static inline bool is_snapshot(struct SheepdogInode *inode)
 285 {
 286     return !!inode->snap_ctime;
 287 }
 288
 289 static inline size_t count_data_objs(const struct SheepdogInode *inode)
 290 {
 291     return DIV_ROUND_UP(inode->vdi_size,
 292                         (1UL << inode->block_size_shift));
 293 }
 294
 295 #undef DPRINTF
 296 #ifdef DEBUG_SDOG
 297 #define DEBUG_SDOG_PRINT 1
 298 #else
 299 #define DEBUG_SDOG_PRINT 0
 300 #endif
 301 #define DPRINTF(fmt, args...)                                           \
 302     do {                                                                \
 303         if (DEBUG_SDOG_PRINT) {                                         \
 304             fprintf(stderr, "%s %d: " fmt, __func__, __LINE__, ##args); \
 305         }                                                               \
 306     } while (0)
 307
 308 typedef struct SheepdogAIOCB SheepdogAIOCB;
 309
 310 typedef struct AIOReq {
 311     SheepdogAIOCB *aiocb;
 312     unsigned int iov_offset;
 313
 314     uint64_t oid;
 315     uint64_t base_oid;
 316     uint64_t offset;
 317     unsigned int data_len;
 318     uint8_t flags;
 319     uint32_t id;
 320     bool create;
 321
 322     QLIST_ENTRY(AIOReq) aio_siblings;
 323 } AIOReq;
 324
 325 enum AIOCBState {
 326     AIOCB_WRITE_UDATA,
 327     AIOCB_READ_UDATA,
 328     AIOCB_FLUSH_CACHE,
 329     AIOCB_DISCARD_OBJ,
 330 };
 331
 332 #define AIOCBOverlapping(x, y)                                 \
 333     (!(x->max_affect_data_idx < y->min_affect_data_idx          \
 334        || y->max_affect_data_idx < x->min_affect_data_idx))
 335
 336 struct SheepdogAIOCB {
 337     BlockAIOCB common;
 338
 339     QEMUIOVector *qiov;
 340
 341     int64_t sector_num;
 342     int nb_sectors;
 343
 344     int ret;
 345     enum AIOCBState aiocb_type;
 346
 347     Coroutine *coroutine;
 348     void (*aio_done_func)(SheepdogAIOCB *);
 349
 350     bool cancelable;
 351     int nr_pending;
 352
 353     uint32_t min_affect_data_idx;
 354     uint32_t max_affect_data_idx;
 355
 356     /*
 357      * The difference between affect_data_idx and dirty_data_idx:
 358      * affect_data_idx represents range of index of all request types.
 359      * dirty_data_idx represents range of index updated by COW requests.
 360      * dirty_data_idx is used for updating an inode object.
 361      */
 362     uint32_t min_dirty_data_idx;
 363     uint32_t max_dirty_data_idx;
 364
 365     QLIST_ENTRY(SheepdogAIOCB) aiocb_siblings;
 366 };
 367
 368 typedef struct BDRVSheepdogState {
 369     BlockDriverState *bs;
 370     AioContext *aio_context;
 371
 372     SheepdogInode inode;
 373
 374     char name[SD_MAX_VDI_LEN];
 375     bool is_snapshot;
 376     uint32_t cache_flags;
 377     bool discard_supported;
 378
 379     char *host_spec;
 380     bool is_unix;
 381     int fd;
 382
 383     CoMutex lock;
 384     Coroutine *co_send;
 385     Coroutine *co_recv;
 386
 387     uint32_t aioreq_seq_num;
 388
 389     /* Every aio request must be linked to either of these queues. */
 390     QLIST_HEAD(inflight_aio_head, AIOReq) inflight_aio_head;
 391     QLIST_HEAD(failed_aio_head, AIOReq) failed_aio_head;
 392
 393     CoQueue overlapping_queue;
 394     QLIST_HEAD(inflight_aiocb_head, SheepdogAIOCB) inflight_aiocb_head;
 395 } BDRVSheepdogState;
 396
 397 typedef struct BDRVSheepdogReopenState {
 398     int fd;
 399     int cache_flags;
 400 } BDRVSheepdogReopenState;
 401
 402 static const char * sd_strerror(int err)
 403 {
 404     int i;
 405
 406     static const struct {
 407         int err;
 408         const char *desc;
 409     } errors[] = {
 410         {SD_RES_SUCCESS, "Success"},
 411         {SD_RES_UNKNOWN, "Unknown error"},
 412         {SD_RES_NO_OBJ, "No object found"},
 413         {SD_RES_EIO, "I/O error"},
 414         {SD_RES_VDI_EXIST, "VDI exists already"},
 415         {SD_RES_INVALID_PARMS, "Invalid parameters"},
 416         {SD_RES_SYSTEM_ERROR, "System error"},
 417         {SD_RES_VDI_LOCKED, "VDI is already locked"},
 418         {SD_RES_NO_VDI, "No vdi found"},
 419         {SD_RES_NO_BASE_VDI, "No base VDI found"},
 420         {SD_RES_VDI_READ, "Failed read the requested VDI"},
 421         {SD_RES_VDI_WRITE, "Failed to write the requested VDI"},
 422         {SD_RES_BASE_VDI_READ, "Failed to read the base VDI"},
 423         {SD_RES_BASE_VDI_WRITE, "Failed to write the base VDI"},
 424         {SD_RES_NO_TAG, "Failed to find the requested tag"},
 425         {SD_RES_STARTUP, "The system is still booting"},
 426         {SD_RES_VDI_NOT_LOCKED, "VDI isn't locked"},
 427         {SD_RES_SHUTDOWN, "The system is shutting down"},
 428         {SD_RES_NO_MEM, "Out of memory on the server"},
 429         {SD_RES_FULL_VDI, "We already have the maximum vdis"},
 430         {SD_RES_VER_MISMATCH, "Protocol version mismatch"},
 431         {SD_RES_NO_SPACE, "Server has no space for new objects"},
 432         {SD_RES_WAIT_FOR_FORMAT, "Sheepdog is waiting for a format operation"},
 433         {SD_RES_WAIT_FOR_JOIN, "Sheepdog is waiting for other nodes joining"},
 434         {SD_RES_JOIN_FAILED, "Target node had failed to join sheepdog"},
 435         {SD_RES_HALT, "Sheepdog is stopped serving IO request"},
 436         {SD_RES_READONLY, "Object is read-only"},
 437     };
 438
 439     for (i = 0; i < ARRAY_SIZE(errors); ++i) {
 440         if (errors[i].err == err) {
 441             return errors[i].desc;
 442         }
 443     }
 444
 445     return "Invalid error code";
 446 }
 447
 448 /*
 449  * Sheepdog I/O handling:
 450  *
 451  * 1. In sd_co_rw_vector, we send the I/O requests to the server and
 452  *    link the requests to the inflight_list in the
 453  *    BDRVSheepdogState.  The function exits without waiting for
 454  *    receiving the response.
 455  *
 456  * 2. We receive the response in aio_read_response, the fd handler to
 457  *    the sheepdog connection.  If metadata update is needed, we send
 458  *    the write request to the vdi object in sd_write_done, the write
 459  *    completion function.  We switch back to sd_co_readv/writev after
 460  *    all the requests belonging to the AIOCB are finished.
 461  */
 462
 463 static inline AIOReq *alloc_aio_req(BDRVSheepdogState *s, SheepdogAIOCB *acb,
 464                                     uint64_t oid, unsigned int data_len,
 465                                     uint64_t offset, uint8_t flags, bool create,
 466                                     uint64_t base_oid, unsigned int iov_offset)
 467 {
 468     AIOReq *aio_req;
 469
 470     aio_req = g_malloc(sizeof(*aio_req));
 471     aio_req->aiocb = acb;
 472     aio_req->iov_offset = iov_offset;
 473     aio_req->oid = oid;
 474     aio_req->base_oid = base_oid;
 475     aio_req->offset = offset;
 476     aio_req->data_len = data_len;
 477     aio_req->flags = flags;
 478     aio_req->id = s->aioreq_seq_num++;
 479     aio_req->create = create;
 480
 481     acb->nr_pending++;
 482     return aio_req;
 483 }
 484
 485 static inline void free_aio_req(BDRVSheepdogState *s, AIOReq *aio_req)
 486 {
 487     SheepdogAIOCB *acb = aio_req->aiocb;
 488
 489     acb->cancelable = false;
 490     QLIST_REMOVE(aio_req, aio_siblings);
 491     g_free(aio_req);
 492
 493     acb->nr_pending--;
 494 }
 495
 496 static void coroutine_fn sd_finish_aiocb(SheepdogAIOCB *acb)
 497 {
 498     qemu_coroutine_enter(acb->coroutine);
 499     qemu_aio_unref(acb);
 500 }
 501
 502 /*
 503  * Check whether the specified acb can be canceled
 504  *
 505  * We can cancel aio when any request belonging to the acb is:
 506  *  - Not processed by the sheepdog server.
 507  *  - Not linked to the inflight queue.
 508  */
 509 static bool sd_acb_cancelable(const SheepdogAIOCB *acb)
 510 {
 511     BDRVSheepdogState *s = acb->common.bs->opaque;
 512     AIOReq *aioreq;
 513
 514     if (!acb->cancelable) {
 515         return false;
 516     }
 517
 518     QLIST_FOREACH(aioreq, &s->inflight_aio_head, aio_siblings) {
 519         if (aioreq->aiocb == acb) {
 520             return false;
 521         }
 522     }
 523
 524     return true;
 525 }
 526
 527 static void sd_aio_cancel(BlockAIOCB *blockacb)
 528 {
 529     SheepdogAIOCB *acb = (SheepdogAIOCB *)blockacb;
 530     BDRVSheepdogState *s = acb->common.bs->opaque;
 531     AIOReq *aioreq, *next;
 532
 533     if (sd_acb_cancelable(acb)) {
 534         /* Remove outstanding requests from failed queue.  */
 535         QLIST_FOREACH_SAFE(aioreq, &s->failed_aio_head, aio_siblings,
 536                            next) {
 537             if (aioreq->aiocb == acb) {
 538                 free_aio_req(s, aioreq);
 539             }
 540         }
 541
 542         assert(acb->nr_pending == 0);
 543         if (acb->common.cb) {
 544             acb->common.cb(acb->common.opaque, -ECANCELED);
 545         }
 546         sd_finish_aiocb(acb);
 547     }
 548 }
 549
 550 static const AIOCBInfo sd_aiocb_info = {
 551     .aiocb_size     = sizeof(SheepdogAIOCB),
 552     .cancel_async   = sd_aio_cancel,
 553 };
 554
 555 static SheepdogAIOCB *sd_aio_setup(BlockDriverState *bs, QEMUIOVector *qiov,
 556                                    int64_t sector_num, int nb_sectors)
 557 {
 558     SheepdogAIOCB *acb;
 559     uint32_t object_size;
 560     BDRVSheepdogState *s = bs->opaque;
 561
 562     object_size = (UINT32_C(1) << s->inode.block_size_shift);
 563
 564     acb = qemu_aio_get(&sd_aiocb_info, bs, NULL, NULL);
 565
 566     acb->qiov = qiov;
 567
 568     acb->sector_num = sector_num;
 569     acb->nb_sectors = nb_sectors;
 570
 571     acb->aio_done_func = NULL;
 572     acb->cancelable = true;
 573     acb->coroutine = qemu_coroutine_self();
 574     acb->ret = 0;
 575     acb->nr_pending = 0;
 576
 577     acb->min_affect_data_idx = acb->sector_num * BDRV_SECTOR_SIZE / object_size;
 578     acb->max_affect_data_idx = (acb->sector_num * BDRV_SECTOR_SIZE +
 579                               acb->nb_sectors * BDRV_SECTOR_SIZE) / object_size;
 580
 581     acb->min_dirty_data_idx = UINT32_MAX;
 582     acb->max_dirty_data_idx = 0;
 583
 584     return acb;
 585 }
 586
 587 /* Return -EIO in case of error, file descriptor on success */
 588 static int connect_to_sdog(BDRVSheepdogState *s, Error **errp)
 589 {
 590     int fd;
 591
 592     if (s->is_unix) {
 593         fd = unix_connect(s->host_spec, errp);
 594     } else {
 595         fd = inet_connect(s->host_spec, errp);
 596
 597         if (fd >= 0) {
 598             int ret = socket_set_nodelay(fd);
 599             if (ret < 0) {
 600                 error_report("%s", strerror(errno));
 601             }
 602         }
 603     }
 604
 605     if (fd >= 0) {
 606         qemu_set_nonblock(fd);
 607     } else {
 608         fd = -EIO;
 609     }
 610
 611     return fd;
 612 }
 613
 614 /* Return 0 on success and -errno in case of error */
 615 static coroutine_fn int send_co_req(int sockfd, SheepdogReq *hdr, void *data,
 616                                     unsigned int *wlen)
 617 {
 618     int ret;
 619
 620     ret = qemu_co_send(sockfd, hdr, sizeof(*hdr));
 621     if (ret != sizeof(*hdr)) {
 622         error_report("failed to send a req, %s", strerror(errno));
 623         return -errno;
 624     }
 625
 626     ret = qemu_co_send(sockfd, data, *wlen);
 627     if (ret != *wlen) {
 628         error_report("failed to send a req, %s", strerror(errno));
 629         return -errno;
 630     }
 631
 632     return ret;
 633 }
 634
 635 static void restart_co_req(void *opaque)
 636 {
 637     Coroutine *co = opaque;
 638
 639     qemu_coroutine_enter(co);
 640 }
 641
 642 typedef struct SheepdogReqCo {
 643     int sockfd;
 644     BlockDriverState *bs;
 645     AioContext *aio_context;
 646     SheepdogReq *hdr;
 647     void *data;
 648     unsigned int *wlen;
 649     unsigned int *rlen;
 650     int ret;
 651     bool finished;
 652 } SheepdogReqCo;
 653
 654 static coroutine_fn void do_co_req(void *opaque)
 655 {
 656     int ret;
 657     Coroutine *co;
 658     SheepdogReqCo *srco = opaque;
 659     int sockfd = srco->sockfd;
 660     SheepdogReq *hdr = srco->hdr;
 661     void *data = srco->data;
 662     unsigned int *wlen = srco->wlen;
 663     unsigned int *rlen = srco->rlen;
 664
 665     co = qemu_coroutine_self();
 666     aio_set_fd_handler(srco->aio_context, sockfd, false,
 667                        NULL, restart_co_req, co);
 668
 669     ret = send_co_req(sockfd, hdr, data, wlen);
 670     if (ret < 0) {
 671         goto out;
 672     }
 673
 674     aio_set_fd_handler(srco->aio_context, sockfd, false,
 675                        restart_co_req, NULL, co);
 676
 677     ret = qemu_co_recv(sockfd, hdr, sizeof(*hdr));
 678     if (ret != sizeof(*hdr)) {
 679         error_report("failed to get a rsp, %s", strerror(errno));
 680         ret = -errno;
 681         goto out;
 682     }
 683
 684     if (*rlen > hdr->data_length) {
 685         *rlen = hdr->data_length;
 686     }
 687
 688     if (*rlen) {
 689         ret = qemu_co_recv(sockfd, data, *rlen);
 690         if (ret != *rlen) {
 691             error_report("failed to get the data, %s", strerror(errno));
 692             ret = -errno;
 693             goto out;
 694         }
 695     }
 696     ret = 0;
 697 out:
 698     /* there is at most one request for this sockfd, so it is safe to
 699      * set each handler to NULL. */
 700     aio_set_fd_handler(srco->aio_context, sockfd, false,
 701                        NULL, NULL, NULL);
 702
 703     srco->ret = ret;
 704     srco->finished = true;
 705     if (srco->bs) {
 706         bdrv_wakeup(srco->bs);
 707     }
 708 }
 709
 710 /*
 711  * Send the request to the sheep in a synchronous manner.
 712  *
 713  * Return 0 on success, -errno in case of error.
 714  */
 715 static int do_req(int sockfd, BlockDriverState *bs, SheepdogReq *hdr,
 716                   void *data, unsigned int *wlen, unsigned int *rlen)
 717 {
 718     Coroutine *co;
 719     SheepdogReqCo srco = {
 720         .sockfd = sockfd,
 721         .aio_context = bs ? bdrv_get_aio_context(bs) : qemu_get_aio_context(),
 722         .bs = bs,
 723         .hdr = hdr,
 724         .data = data,
 725         .wlen = wlen,
 726         .rlen = rlen,
 727         .ret = 0,
 728         .finished = false,
 729     };
 730
 731     if (qemu_in_coroutine()) {
 732         do_co_req(&srco);
 733     } else {
 734         co = qemu_coroutine_create(do_co_req, &srco);
 735         if (bs) {
 736             qemu_coroutine_enter(co);
 737             BDRV_POLL_WHILE(bs, !srco.finished);
 738         } else {
 739             qemu_coroutine_enter(co);
 740             while (!srco.finished) {
 741                 aio_poll(qemu_get_aio_context(), true);
 742             }
 743         }
 744     }
 745
 746     return srco.ret;
 747 }
 748
 749 static void coroutine_fn add_aio_request(BDRVSheepdogState *s, AIOReq *aio_req,
 750                                          struct iovec *iov, int niov,
 751                                          enum AIOCBState aiocb_type);
 752 static void coroutine_fn resend_aioreq(BDRVSheepdogState *s, AIOReq *aio_req);
 753 static int reload_inode(BDRVSheepdogState *s, uint32_t snapid, const char *tag);
 754 static int get_sheep_fd(BDRVSheepdogState *s, Error **errp);
 755 static void co_write_request(void *opaque);
 756
 757 static coroutine_fn void reconnect_to_sdog(void *opaque)
 758 {
 759     BDRVSheepdogState *s = opaque;
 760     AIOReq *aio_req, *next;
 761
 762     aio_set_fd_handler(s->aio_context, s->fd, false, NULL,
 763                        NULL, NULL);
 764     close(s->fd);
 765     s->fd = -1;
 766
 767     /* Wait for outstanding write requests to be completed. */
 768     while (s->co_send != NULL) {
 769         co_write_request(opaque);
 770     }
 771
 772     /* Try to reconnect the sheepdog server every one second. */
 773     while (s->fd < 0) {
 774         Error *local_err = NULL;
 775         s->fd = get_sheep_fd(s, &local_err);
 776         if (s->fd < 0) {
 777             DPRINTF("Wait for connection to be established\n");
 778             error_report_err(local_err);
 779             co_aio_sleep_ns(bdrv_get_aio_context(s->bs), QEMU_CLOCK_REALTIME,
 780                             1000000000ULL);
 781         }
 782     };
 783
 784     /*
 785      * Now we have to resend all the request in the inflight queue.  However,
 786      * resend_aioreq() can yield and newly created requests can be added to the
 787      * inflight queue before the coroutine is resumed.  To avoid mixing them, we
 788      * have to move all the inflight requests to the failed queue before
 789      * resend_aioreq() is called.
 790      */
 791     QLIST_FOREACH_SAFE(aio_req, &s->inflight_aio_head, aio_siblings, next) {
 792         QLIST_REMOVE(aio_req, aio_siblings);
 793         QLIST_INSERT_HEAD(&s->failed_aio_head, aio_req, aio_siblings);
 794     }
 795
 796     /* Resend all the failed aio requests. */
 797     while (!QLIST_EMPTY(&s->failed_aio_head)) {
 798         aio_req = QLIST_FIRST(&s->failed_aio_head);
 799         QLIST_REMOVE(aio_req, aio_siblings);
 800         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
 801         resend_aioreq(s, aio_req);
 802     }
 803 }
 804
 805 /*
 806  * Receive responses of the I/O requests.
 807  *
 808  * This function is registered as a fd handler, and called from the
 809  * main loop when s->fd is ready for reading responses.
 810  */
 811 static void coroutine_fn aio_read_response(void *opaque)
 812 {
 813     SheepdogObjRsp rsp;
 814     BDRVSheepdogState *s = opaque;
 815     int fd = s->fd;
 816     int ret;
 817     AIOReq *aio_req = NULL;
 818     SheepdogAIOCB *acb;
 819     uint64_t idx;
 820
 821     /* read a header */
 822     ret = qemu_co_recv(fd, &rsp, sizeof(rsp));
 823     if (ret != sizeof(rsp)) {
 824         error_report("failed to get the header, %s", strerror(errno));
 825         goto err;
 826     }
 827
 828     /* find the right aio_req from the inflight aio list */
 829     QLIST_FOREACH(aio_req, &s->inflight_aio_head, aio_siblings) {
 830         if (aio_req->id == rsp.id) {
 831             break;
 832         }
 833     }
 834     if (!aio_req) {
 835         error_report("cannot find aio_req %x", rsp.id);
 836         goto err;
 837     }
 838
 839     acb = aio_req->aiocb;
 840
 841     switch (acb->aiocb_type) {
 842     case AIOCB_WRITE_UDATA:
 843         /* this coroutine context is no longer suitable for co_recv
 844          * because we may send data to update vdi objects */
 845         s->co_recv = NULL;
 846         if (!is_data_obj(aio_req->oid)) {
 847             break;
 848         }
 849         idx = data_oid_to_idx(aio_req->oid);
 850
 851         if (aio_req->create) {
 852             /*
 853              * If the object is newly created one, we need to update
 854              * the vdi object (metadata object).  min_dirty_data_idx
 855              * and max_dirty_data_idx are changed to include updated
 856              * index between them.
 857              */
 858             if (rsp.result == SD_RES_SUCCESS) {
 859                 s->inode.data_vdi_id[idx] = s->inode.vdi_id;
 860                 acb->max_dirty_data_idx = MAX(idx, acb->max_dirty_data_idx);
 861                 acb->min_dirty_data_idx = MIN(idx, acb->min_dirty_data_idx);
 862             }
 863         }
 864         break;
 865     case AIOCB_READ_UDATA:
 866         ret = qemu_co_recvv(fd, acb->qiov->iov, acb->qiov->niov,
 867                             aio_req->iov_offset, rsp.data_length);
 868         if (ret != rsp.data_length) {
 869             error_report("failed to get the data, %s", strerror(errno));
 870             goto err;
 871         }
 872         break;
 873     case AIOCB_FLUSH_CACHE:
 874         if (rsp.result == SD_RES_INVALID_PARMS) {
 875             DPRINTF("disable cache since the server doesn't support it\n");
 876             s->cache_flags = SD_FLAG_CMD_DIRECT;
 877             rsp.result = SD_RES_SUCCESS;
 878         }
 879         break;
 880     case AIOCB_DISCARD_OBJ:
 881         switch (rsp.result) {
 882         case SD_RES_INVALID_PARMS:
 883             error_report("sheep(%s) doesn't support discard command",
 884                          s->host_spec);
 885             rsp.result = SD_RES_SUCCESS;
 886             s->discard_supported = false;
 887             break;
 888         default:
 889             break;
 890         }
 891     }
 892
 893     switch (rsp.result) {
 894     case SD_RES_SUCCESS:
 895         break;
 896     case SD_RES_READONLY:
 897         if (s->inode.vdi_id == oid_to_vid(aio_req->oid)) {
 898             ret = reload_inode(s, 0, "");
 899             if (ret < 0) {
 900                 goto err;
 901             }
 902         }
 903         if (is_data_obj(aio_req->oid)) {
 904             aio_req->oid = vid_to_data_oid(s->inode.vdi_id,
 905                                            data_oid_to_idx(aio_req->oid));
 906         } else {
 907             aio_req->oid = vid_to_vdi_oid(s->inode.vdi_id);
 908         }
 909         resend_aioreq(s, aio_req);
 910         goto out;
 911     default:
 912         acb->ret = -EIO;
 913         error_report("%s", sd_strerror(rsp.result));
 914         break;
 915     }
 916
 917     free_aio_req(s, aio_req);
 918     if (!acb->nr_pending) {
 919         /*
 920          * We've finished all requests which belong to the AIOCB, so
 921          * we can switch back to sd_co_readv/writev now.
 922          */
 923         acb->aio_done_func(acb);
 924     }
 925 out:
 926     s->co_recv = NULL;
 927     return;
 928 err:
 929     s->co_recv = NULL;
 930     reconnect_to_sdog(opaque);
 931 }
 932
 933 static void co_read_response(void *opaque)
 934 {
 935     BDRVSheepdogState *s = opaque;
 936
 937     if (!s->co_recv) {
 938         s->co_recv = qemu_coroutine_create(aio_read_response, opaque);
 939     }
 940
 941     qemu_coroutine_enter(s->co_recv);
 942 }
 943
 944 static void co_write_request(void *opaque)
 945 {
 946     BDRVSheepdogState *s = opaque;
 947
 948     qemu_coroutine_enter(s->co_send);
 949 }
 950
 951 /*
 952  * Return a socket descriptor to read/write objects.
 953  *
 954  * We cannot use this descriptor for other operations because
 955  * the block driver may be on waiting response from the server.
 956  */
 957 static int get_sheep_fd(BDRVSheepdogState *s, Error **errp)
 958 {
 959     int fd;
 960
 961     fd = connect_to_sdog(s, errp);
 962     if (fd < 0) {
 963         return fd;
 964     }
 965
 966     aio_set_fd_handler(s->aio_context, fd, false,
 967                        co_read_response, NULL, s);
 968     return fd;
 969 }
 970
 971 static int sd_parse_uri(BDRVSheepdogState *s, const char *filename,
 972                         char *vdi, uint32_t *snapid, char *tag)
 973 {
 974     URI *uri;
 975     QueryParams *qp = NULL;
 976     int ret = 0;
 977
 978     uri = uri_parse(filename);
 979     if (!uri) {
 980         return -EINVAL;
 981     }
 982
 983     /* transport */
 984     if (!strcmp(uri->scheme, "sheepdog")) {
 985         s->is_unix = false;
 986     } else if (!strcmp(uri->scheme, "sheepdog+tcp")) {
 987         s->is_unix = false;
 988     } else if (!strcmp(uri->scheme, "sheepdog+unix")) {
 989         s->is_unix = true;
 990     } else {
 991         ret = -EINVAL;
 992         goto out;
 993     }
 994
 995     if (uri->path == NULL || !strcmp(uri->path, "/")) {
 996         ret = -EINVAL;
 997         goto out;
 998     }
 999     pstrcpy(vdi, SD_MAX_VDI_LEN, uri->path + 1);
1000
1001     qp = query_params_parse(uri->query);
1002     if (qp->n > 1 || (s->is_unix && !qp->n) || (!s->is_unix && qp->n)) {
1003         ret = -EINVAL;
1004         goto out;
1005     }
1006
1007     if (s->is_unix) {
1008         /* sheepdog+unix:///vdiname?socket=path */
1009         if (uri->server || uri->port || strcmp(qp->p[0].name, "socket")) {
1010             ret = -EINVAL;
1011             goto out;
1012         }
1013         s->host_spec = g_strdup(qp->p[0].value);
1014     } else {
1015         /* sheepdog[+tcp]://[host:port]/vdiname */
1016         s->host_spec = g_strdup_printf("%s:%d", uri->server ?: SD_DEFAULT_ADDR,
1017                                        uri->port ?: SD_DEFAULT_PORT);
1018     }
1019
1020     /* snapshot tag */
1021     if (uri->fragment) {
1022         *snapid = strtoul(uri->fragment, NULL, 10);
1023         if (*snapid == 0) {
1024             pstrcpy(tag, SD_MAX_VDI_TAG_LEN, uri->fragment);
1025         }
1026     } else {
1027         *snapid = CURRENT_VDI_ID; /* search current vdi */
1028     }
1029
1030 out:
1031     if (qp) {
1032         query_params_free(qp);
1033     }
1034     uri_free(uri);
1035     return ret;
1036 }
1037
1038 /*
1039  * Parse a filename (old syntax)
1040  *
1041  * filename must be one of the following formats:
1042  *   1. [vdiname]
1043  *   2. [vdiname]:[snapid]
1044  *   3. [vdiname]:[tag]
1045  *   4. [hostname]:[port]:[vdiname]
1046  *   5. [hostname]:[port]:[vdiname]:[snapid]
1047  *   6. [hostname]:[port]:[vdiname]:[tag]
1048  *
1049  * You can boot from the snapshot images by specifying `snapid` or
1050  * `tag'.
1051  *
1052  * You can run VMs outside the Sheepdog cluster by specifying
1053  * `hostname' and `port' (experimental).
1054  */
1055 static int parse_vdiname(BDRVSheepdogState *s, const char *filename,
1056                          char *vdi, uint32_t *snapid, char *tag)
1057 {
1058     char *p, *q, *uri;
1059     const char *host_spec, *vdi_spec;
1060     int nr_sep, ret;
1061
1062     strstart(filename, "sheepdog:", &filename);
1063     p = q = g_strdup(filename);
1064
1065     /* count the number of separators */
1066     nr_sep = 0;
1067     while (*p) {
1068         if (*p == ':') {
1069             nr_sep++;
1070         }
1071         p++;
1072     }
1073     p = q;
1074
1075     /* use the first two tokens as host_spec. */
1076     if (nr_sep >= 2) {
1077         host_spec = p;
1078         p = strchr(p, ':');
1079         p++;
1080         p = strchr(p, ':');
1081         *p++ = '\0';
1082     } else {
1083         host_spec = "";
1084     }
1085
1086     vdi_spec = p;
1087
1088     p = strchr(vdi_spec, ':');
1089     if (p) {
1090         *p++ = '#';
1091     }
1092
1093     uri = g_strdup_printf("sheepdog://%s/%s", host_spec, vdi_spec);
1094
1095     ret = sd_parse_uri(s, uri, vdi, snapid, tag);
1096
1097     g_free(q);
1098     g_free(uri);
1099
1100     return ret;
1101 }
1102
1103 static int find_vdi_name(BDRVSheepdogState *s, const char *filename,
1104                          uint32_t snapid, const char *tag, uint32_t *vid,
1105                          bool lock, Error **errp)
1106 {
1107     int ret, fd;
1108     SheepdogVdiReq hdr;
1109     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1110     unsigned int wlen, rlen = 0;
1111     char buf[SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN];
1112
1113     fd = connect_to_sdog(s, errp);
1114     if (fd < 0) {
1115         return fd;
1116     }
1117
1118     /* This pair of strncpy calls ensures that the buffer is zero-filled,
1119      * which is desirable since we'll soon be sending those bytes, and
1120      * don't want the send_req to read uninitialized data.
1121      */
1122     strncpy(buf, filename, SD_MAX_VDI_LEN);
1123     strncpy(buf + SD_MAX_VDI_LEN, tag, SD_MAX_VDI_TAG_LEN);
1124
1125     memset(&hdr, 0, sizeof(hdr));
1126     if (lock) {
1127         hdr.opcode = SD_OP_LOCK_VDI;
1128         hdr.type = LOCK_TYPE_NORMAL;
1129     } else {
1130         hdr.opcode = SD_OP_GET_VDI_INFO;
1131     }
1132     wlen = SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN;
1133     hdr.proto_ver = SD_PROTO_VER;
1134     hdr.data_length = wlen;
1135     hdr.snapid = snapid;
1136     hdr.flags = SD_FLAG_CMD_WRITE;
1137
1138     ret = do_req(fd, s->bs, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1139     if (ret) {
1140         error_setg_errno(errp, -ret, "cannot get vdi info");
1141         goto out;
1142     }
1143
1144     if (rsp->result != SD_RES_SUCCESS) {
1145         error_setg(errp, "cannot get vdi info, %s, %s %" PRIu32 " %s",
1146                    sd_strerror(rsp->result), filename, snapid, tag);
1147         if (rsp->result == SD_RES_NO_VDI) {
1148             ret = -ENOENT;
1149         } else if (rsp->result == SD_RES_VDI_LOCKED) {
1150             ret = -EBUSY;
1151         } else {
1152             ret = -EIO;
1153         }
1154         goto out;
1155     }
1156     *vid = rsp->vdi_id;
1157
1158     ret = 0;
1159 out:
1160     closesocket(fd);
1161     return ret;
1162 }
1163
1164 static void coroutine_fn add_aio_request(BDRVSheepdogState *s, AIOReq *aio_req,
1165                                          struct iovec *iov, int niov,
1166                                          enum AIOCBState aiocb_type)
1167 {
1168     int nr_copies = s->inode.nr_copies;
1169     SheepdogObjReq hdr;
1170     unsigned int wlen = 0;
1171     int ret;
1172     uint64_t oid = aio_req->oid;
1173     unsigned int datalen = aio_req->data_len;
1174     uint64_t offset = aio_req->offset;
1175     uint8_t flags = aio_req->flags;
1176     uint64_t old_oid = aio_req->base_oid;
1177     bool create = aio_req->create;
1178
1179     if (!nr_copies) {
1180         error_report("bug");
1181     }
1182
1183     memset(&hdr, 0, sizeof(hdr));
1184
1185     switch (aiocb_type) {
1186     case AIOCB_FLUSH_CACHE:
1187         hdr.opcode = SD_OP_FLUSH_VDI;
1188         break;
1189     case AIOCB_READ_UDATA:
1190         hdr.opcode = SD_OP_READ_OBJ;
1191         hdr.flags = flags;
1192         break;
1193     case AIOCB_WRITE_UDATA:
1194         if (create) {
1195             hdr.opcode = SD_OP_CREATE_AND_WRITE_OBJ;
1196         } else {
1197             hdr.opcode = SD_OP_WRITE_OBJ;
1198         }
1199         wlen = datalen;
1200         hdr.flags = SD_FLAG_CMD_WRITE | flags;
1201         break;
1202     case AIOCB_DISCARD_OBJ:
1203         hdr.opcode = SD_OP_WRITE_OBJ;
1204         hdr.flags = SD_FLAG_CMD_WRITE | flags;
1205         s->inode.data_vdi_id[data_oid_to_idx(oid)] = 0;
1206         offset = offsetof(SheepdogInode,
1207                           data_vdi_id[data_oid_to_idx(oid)]);
1208         oid = vid_to_vdi_oid(s->inode.vdi_id);
1209         wlen = datalen = sizeof(uint32_t);
1210         break;
1211     }
1212
1213     if (s->cache_flags) {
1214         hdr.flags |= s->cache_flags;
1215     }
1216
1217     hdr.oid = oid;
1218     hdr.cow_oid = old_oid;
1219     hdr.copies = s->inode.nr_copies;
1220
1221     hdr.data_length = datalen;
1222     hdr.offset = offset;
1223
1224     hdr.id = aio_req->id;
1225
1226     qemu_co_mutex_lock(&s->lock);
1227     s->co_send = qemu_coroutine_self();
1228     aio_set_fd_handler(s->aio_context, s->fd, false,
1229                        co_read_response, co_write_request, s);
1230     socket_set_cork(s->fd, 1);
1231
1232     /* send a header */
1233     ret = qemu_co_send(s->fd, &hdr, sizeof(hdr));
1234     if (ret != sizeof(hdr)) {
1235         error_report("failed to send a req, %s", strerror(errno));
1236         goto out;
1237     }
1238
1239     if (wlen) {
1240         ret = qemu_co_sendv(s->fd, iov, niov, aio_req->iov_offset, wlen);
1241         if (ret != wlen) {
1242             error_report("failed to send a data, %s", strerror(errno));
1243         }
1244     }
1245 out:
1246     socket_set_cork(s->fd, 0);
1247     aio_set_fd_handler(s->aio_context, s->fd, false,
1248                        co_read_response, NULL, s);
1249     s->co_send = NULL;
1250     qemu_co_mutex_unlock(&s->lock);
1251 }
1252
1253 static int read_write_object(int fd, BlockDriverState *bs, char *buf,
1254                              uint64_t oid, uint8_t copies,
1255                              unsigned int datalen, uint64_t offset,
1256                              bool write, bool create, uint32_t cache_flags)
1257 {
1258     SheepdogObjReq hdr;
1259     SheepdogObjRsp *rsp = (SheepdogObjRsp *)&hdr;
1260     unsigned int wlen, rlen;
1261     int ret;
1262
1263     memset(&hdr, 0, sizeof(hdr));
1264
1265     if (write) {
1266         wlen = datalen;
1267         rlen = 0;
1268         hdr.flags = SD_FLAG_CMD_WRITE;
1269         if (create) {
1270             hdr.opcode = SD_OP_CREATE_AND_WRITE_OBJ;
1271         } else {
1272             hdr.opcode = SD_OP_WRITE_OBJ;
1273         }
1274     } else {
1275         wlen = 0;
1276         rlen = datalen;
1277         hdr.opcode = SD_OP_READ_OBJ;
1278     }
1279
1280     hdr.flags |= cache_flags;
1281
1282     hdr.oid = oid;
1283     hdr.data_length = datalen;
1284     hdr.offset = offset;
1285     hdr.copies = copies;
1286
1287     ret = do_req(fd, bs, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1288     if (ret) {
1289         error_report("failed to send a request to the sheep");
1290         return ret;
1291     }
1292
1293     switch (rsp->result) {
1294     case SD_RES_SUCCESS:
1295         return 0;
1296     default:
1297         error_report("%s", sd_strerror(rsp->result));
1298         return -EIO;
1299     }
1300 }
1301
1302 static int read_object(int fd, BlockDriverState *bs, char *buf,
1303                        uint64_t oid, uint8_t copies,
1304                        unsigned int datalen, uint64_t offset,
1305                        uint32_t cache_flags)
1306 {
1307     return read_write_object(fd, bs, buf, oid, copies,
1308                              datalen, offset, false,
1309                              false, cache_flags);
1310 }
1311
1312 static int write_object(int fd, BlockDriverState *bs, char *buf,
1313                         uint64_t oid, uint8_t copies,
1314                         unsigned int datalen, uint64_t offset, bool create,
1315                         uint32_t cache_flags)
1316 {
1317     return read_write_object(fd, bs, buf, oid, copies,
1318                              datalen, offset, true,
1319                              create, cache_flags);
1320 }
1321
1322 /* update inode with the latest state */
1323 static int reload_inode(BDRVSheepdogState *s, uint32_t snapid, const char *tag)
1324 {
1325     Error *local_err = NULL;
1326     SheepdogInode *inode;
1327     int ret = 0, fd;
1328     uint32_t vid = 0;
1329
1330     fd = connect_to_sdog(s, &local_err);
1331     if (fd < 0) {
1332         error_report_err(local_err);
1333         return -EIO;
1334     }
1335
1336     inode = g_malloc(SD_INODE_HEADER_SIZE);
1337
1338     ret = find_vdi_name(s, s->name, snapid, tag, &vid, false, &local_err);
1339     if (ret) {
1340         error_report_err(local_err);
1341         goto out;
1342     }
1343
1344     ret = read_object(fd, s->bs, (char *)inode, vid_to_vdi_oid(vid),
1345                       s->inode.nr_copies, SD_INODE_HEADER_SIZE, 0,
1346                       s->cache_flags);
1347     if (ret < 0) {
1348         goto out;
1349     }
1350
1351     if (inode->vdi_id != s->inode.vdi_id) {
1352         memcpy(&s->inode, inode, SD_INODE_HEADER_SIZE);
1353     }
1354
1355 out:
1356     g_free(inode);
1357     closesocket(fd);
1358
1359     return ret;
1360 }
1361
1362 static void coroutine_fn resend_aioreq(BDRVSheepdogState *s, AIOReq *aio_req)
1363 {
1364     SheepdogAIOCB *acb = aio_req->aiocb;
1365
1366     aio_req->create = false;
1367
1368     /* check whether this request becomes a CoW one */
1369     if (acb->aiocb_type == AIOCB_WRITE_UDATA && is_data_obj(aio_req->oid)) {
1370         int idx = data_oid_to_idx(aio_req->oid);
1371
1372         if (is_data_obj_writable(&s->inode, idx)) {
1373             goto out;
1374         }
1375
1376         if (s->inode.data_vdi_id[idx]) {
1377             aio_req->base_oid = vid_to_data_oid(s->inode.data_vdi_id[idx], idx);
1378             aio_req->flags |= SD_FLAG_CMD_COW;
1379         }
1380         aio_req->create = true;
1381     }
1382 out:
1383     if (is_data_obj(aio_req->oid)) {
1384         add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
1385                         acb->aiocb_type);
1386     } else {
1387         struct iovec iov;
1388         iov.iov_base = &s->inode;
1389         iov.iov_len = sizeof(s->inode);
1390         add_aio_request(s, aio_req, &iov, 1, AIOCB_WRITE_UDATA);
1391     }
1392 }
1393
1394 static void sd_detach_aio_context(BlockDriverState *bs)
1395 {
1396     BDRVSheepdogState *s = bs->opaque;
1397
1398     aio_set_fd_handler(s->aio_context, s->fd, false, NULL,
1399                        NULL, NULL);
1400 }
1401
1402 static void sd_attach_aio_context(BlockDriverState *bs,
1403                                   AioContext *new_context)
1404 {
1405     BDRVSheepdogState *s = bs->opaque;
1406
1407     s->aio_context = new_context;
1408     aio_set_fd_handler(new_context, s->fd, false,
1409                        co_read_response, NULL, s);
1410 }
1411
1412 /* TODO Convert to fine grained options */
1413 static QemuOptsList runtime_opts = {
1414     .name = "sheepdog",
1415     .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
1416     .desc = {
1417         {
1418             .name = "filename",
1419             .type = QEMU_OPT_STRING,
1420             .help = "URL to the sheepdog image",
1421         },
1422         { /* end of list */ }
1423     },
1424 };
1425
1426 static int sd_open(BlockDriverState *bs, QDict *options, int flags,
1427                    Error **errp)
1428 {
1429     int ret, fd;
1430     uint32_t vid = 0;
1431     BDRVSheepdogState *s = bs->opaque;
1432     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1433     uint32_t snapid;
1434     char *buf = NULL;
1435     QemuOpts *opts;
1436     Error *local_err = NULL;
1437     const char *filename;
1438
1439     s->bs = bs;
1440     s->aio_context = bdrv_get_aio_context(bs);
1441
1442     opts = qemu_opts_create(&runtime_opts, NULL, 0, &error_abort);
1443     qemu_opts_absorb_qdict(opts, options, &local_err);
1444     if (local_err) {
1445         error_propagate(errp, local_err);
1446         ret = -EINVAL;
1447         goto out;
1448     }
1449
1450     filename = qemu_opt_get(opts, "filename");
1451
1452     QLIST_INIT(&s->inflight_aio_head);
1453     QLIST_INIT(&s->failed_aio_head);
1454     QLIST_INIT(&s->inflight_aiocb_head);
1455     s->fd = -1;
1456
1457     memset(vdi, 0, sizeof(vdi));
1458     memset(tag, 0, sizeof(tag));
1459
1460     if (strstr(filename, "://")) {
1461         ret = sd_parse_uri(s, filename, vdi, &snapid, tag);
1462     } else {
1463         ret = parse_vdiname(s, filename, vdi, &snapid, tag);
1464     }
1465     if (ret < 0) {
1466         error_setg(errp, "Can't parse filename");
1467         goto out;
1468     }
1469     s->fd = get_sheep_fd(s, errp);
1470     if (s->fd < 0) {
1471         ret = s->fd;
1472         goto out;
1473     }
1474
1475     ret = find_vdi_name(s, vdi, snapid, tag, &vid, true, errp);
1476     if (ret) {
1477         goto out;
1478     }
1479
1480     /*
1481      * QEMU block layer emulates writethrough cache as 'writeback + flush', so
1482      * we always set SD_FLAG_CMD_CACHE (writeback cache) as default.
1483      */
1484     s->cache_flags = SD_FLAG_CMD_CACHE;
1485     if (flags & BDRV_O_NOCACHE) {
1486         s->cache_flags = SD_FLAG_CMD_DIRECT;
1487     }
1488     s->discard_supported = true;
1489
1490     if (snapid || tag[0] != '\0') {
1491         DPRINTF("%" PRIx32 " snapshot inode was open.\n", vid);
1492         s->is_snapshot = true;
1493     }
1494
1495     fd = connect_to_sdog(s, errp);
1496     if (fd < 0) {
1497         ret = fd;
1498         goto out;
1499     }
1500
1501     buf = g_malloc(SD_INODE_SIZE);
1502     ret = read_object(fd, s->bs, buf, vid_to_vdi_oid(vid),
1503                       0, SD_INODE_SIZE, 0, s->cache_flags);
1504
1505     closesocket(fd);
1506
1507     if (ret) {
1508         error_setg(errp, "Can't read snapshot inode");
1509         goto out;
1510     }
1511
1512     memcpy(&s->inode, buf, sizeof(s->inode));
1513
1514     bs->total_sectors = s->inode.vdi_size / BDRV_SECTOR_SIZE;
1515     pstrcpy(s->name, sizeof(s->name), vdi);
1516     qemu_co_mutex_init(&s->lock);
1517     qemu_co_queue_init(&s->overlapping_queue);
1518     qemu_opts_del(opts);
1519     g_free(buf);
1520     return 0;
1521 out:
1522     aio_set_fd_handler(bdrv_get_aio_context(bs), s->fd,
1523                        false, NULL, NULL, NULL);
1524     if (s->fd >= 0) {
1525         closesocket(s->fd);
1526     }
1527     qemu_opts_del(opts);
1528     g_free(buf);
1529     return ret;
1530 }
1531
1532 static int sd_reopen_prepare(BDRVReopenState *state, BlockReopenQueue *queue,
1533                              Error **errp)
1534 {
1535     BDRVSheepdogState *s = state->bs->opaque;
1536     BDRVSheepdogReopenState *re_s;
1537     int ret = 0;
1538
1539     re_s = state->opaque = g_new0(BDRVSheepdogReopenState, 1);
1540
1541     re_s->cache_flags = SD_FLAG_CMD_CACHE;
1542     if (state->flags & BDRV_O_NOCACHE) {
1543         re_s->cache_flags = SD_FLAG_CMD_DIRECT;
1544     }
1545
1546     re_s->fd = get_sheep_fd(s, errp);
1547     if (re_s->fd < 0) {
1548         ret = re_s->fd;
1549         return ret;
1550     }
1551
1552     return ret;
1553 }
1554
1555 static void sd_reopen_commit(BDRVReopenState *state)
1556 {
1557     BDRVSheepdogReopenState *re_s = state->opaque;
1558     BDRVSheepdogState *s = state->bs->opaque;
1559
1560     if (s->fd) {
1561         aio_set_fd_handler(s->aio_context, s->fd, false,
1562                            NULL, NULL, NULL);
1563         closesocket(s->fd);
1564     }
1565
1566     s->fd = re_s->fd;
1567     s->cache_flags = re_s->cache_flags;
1568
1569     g_free(state->opaque);
1570     state->opaque = NULL;
1571
1572     return;
1573 }
1574
1575 static void sd_reopen_abort(BDRVReopenState *state)
1576 {
1577     BDRVSheepdogReopenState *re_s = state->opaque;
1578     BDRVSheepdogState *s = state->bs->opaque;
1579
1580     if (re_s == NULL) {
1581         return;
1582     }
1583
1584     if (re_s->fd) {
1585         aio_set_fd_handler(s->aio_context, re_s->fd, false,
1586                            NULL, NULL, NULL);
1587         closesocket(re_s->fd);
1588     }
1589
1590     g_free(state->opaque);
1591     state->opaque = NULL;
1592
1593     return;
1594 }
1595
1596 static int do_sd_create(BDRVSheepdogState *s, uint32_t *vdi_id, int snapshot,
1597                         Error **errp)
1598 {
1599     SheepdogVdiReq hdr;
1600     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1601     int fd, ret;
1602     unsigned int wlen, rlen = 0;
1603     char buf[SD_MAX_VDI_LEN];
1604
1605     fd = connect_to_sdog(s, errp);
1606     if (fd < 0) {
1607         return fd;
1608     }
1609
1610     /* FIXME: would it be better to fail (e.g., return -EIO) when filename
1611      * does not fit in buf?  For now, just truncate and avoid buffer overrun.
1612      */
1613     memset(buf, 0, sizeof(buf));
1614     pstrcpy(buf, sizeof(buf), s->name);
1615
1616     memset(&hdr, 0, sizeof(hdr));
1617     hdr.opcode = SD_OP_NEW_VDI;
1618     hdr.base_vdi_id = s->inode.vdi_id;
1619
1620     wlen = SD_MAX_VDI_LEN;
1621
1622     hdr.flags = SD_FLAG_CMD_WRITE;
1623     hdr.snapid = snapshot;
1624
1625     hdr.data_length = wlen;
1626     hdr.vdi_size = s->inode.vdi_size;
1627     hdr.copy_policy = s->inode.copy_policy;
1628     hdr.copies = s->inode.nr_copies;
1629     hdr.block_size_shift = s->inode.block_size_shift;
1630
1631     ret = do_req(fd, NULL, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1632
1633     closesocket(fd);
1634
1635     if (ret) {
1636         error_setg_errno(errp, -ret, "create failed");
1637         return ret;
1638     }
1639
1640     if (rsp->result != SD_RES_SUCCESS) {
1641         error_setg(errp, "%s, %s", sd_strerror(rsp->result), s->inode.name);
1642         return -EIO;
1643     }
1644
1645     if (vdi_id) {
1646         *vdi_id = rsp->vdi_id;
1647     }
1648
1649     return 0;
1650 }
1651
1652 static int sd_prealloc(const char *filename, Error **errp)
1653 {
1654     BlockBackend *blk = NULL;
1655     BDRVSheepdogState *base = NULL;
1656     unsigned long buf_size;
1657     uint32_t idx, max_idx;
1658     uint32_t object_size;
1659     int64_t vdi_size;
1660     void *buf = NULL;
1661     int ret;
1662
1663     blk = blk_new_open(filename, NULL, NULL,
1664                        BDRV_O_RDWR | BDRV_O_PROTOCOL, errp);
1665     if (blk == NULL) {
1666         ret = -EIO;
1667         goto out_with_err_set;
1668     }
1669
1670     blk_set_allow_write_beyond_eof(blk, true);
1671
1672     vdi_size = blk_getlength(blk);
1673     if (vdi_size < 0) {
1674         ret = vdi_size;
1675         goto out;
1676     }
1677
1678     base = blk_bs(blk)->opaque;
1679     object_size = (UINT32_C(1) << base->inode.block_size_shift);
1680     buf_size = MIN(object_size, SD_DATA_OBJ_SIZE);
1681     buf = g_malloc0(buf_size);
1682
1683     max_idx = DIV_ROUND_UP(vdi_size, buf_size);
1684
1685     for (idx = 0; idx < max_idx; idx++) {
1686         /*
1687          * The created image can be a cloned image, so we need to read
1688          * a data from the source image.
1689          */
1690         ret = blk_pread(blk, idx * buf_size, buf, buf_size);
1691         if (ret < 0) {
1692             goto out;
1693         }
1694         ret = blk_pwrite(blk, idx * buf_size, buf, buf_size, 0);
1695         if (ret < 0) {
1696             goto out;
1697         }
1698     }
1699
1700     ret = 0;
1701 out:
1702     if (ret < 0) {
1703         error_setg_errno(errp, -ret, "Can't pre-allocate");
1704     }
1705 out_with_err_set:
1706     if (blk) {
1707         blk_unref(blk);
1708     }
1709     g_free(buf);
1710
1711     return ret;
1712 }
1713
1714 /*
1715  * Sheepdog support two kinds of redundancy, full replication and erasure
1716  * coding.
1717  *
1718  * # create a fully replicated vdi with x copies
1719  * -o redundancy=x (1 <= x <= SD_MAX_COPIES)
1720  *
1721  * # create a erasure coded vdi with x data strips and y parity strips
1722  * -o redundancy=x:y (x must be one of {2,4,8,16} and 1 <= y < SD_EC_MAX_STRIP)
1723  */
1724 static int parse_redundancy(BDRVSheepdogState *s, const char *opt)
1725 {
1726     struct SheepdogInode *inode = &s->inode;
1727     const char *n1, *n2;
1728     long copy, parity;
1729     char p[10];
1730
1731     pstrcpy(p, sizeof(p), opt);
1732     n1 = strtok(p, ":");
1733     n2 = strtok(NULL, ":");
1734
1735     if (!n1) {
1736         return -EINVAL;
1737     }
1738
1739     copy = strtol(n1, NULL, 10);
1740     if (copy > SD_MAX_COPIES || copy < 1) {
1741         return -EINVAL;
1742     }
1743     if (!n2) {
1744         inode->copy_policy = 0;
1745         inode->nr_copies = copy;
1746         return 0;
1747     }
1748
1749     if (copy != 2 && copy != 4 && copy != 8 && copy != 16) {
1750         return -EINVAL;
1751     }
1752
1753     parity = strtol(n2, NULL, 10);
1754     if (parity >= SD_EC_MAX_STRIP || parity < 1) {
1755         return -EINVAL;
1756     }
1757
1758     /*
1759      * 4 bits for parity and 4 bits for data.
1760      * We have to compress upper data bits because it can't represent 16
1761      */
1762     inode->copy_policy = ((copy / 2) << 4) + parity;
1763     inode->nr_copies = copy + parity;
1764
1765     return 0;
1766 }
1767
1768 static int parse_block_size_shift(BDRVSheepdogState *s, QemuOpts *opt)
1769 {
1770     struct SheepdogInode *inode = &s->inode;
1771     uint64_t object_size;
1772     int obj_order;
1773
1774     object_size = qemu_opt_get_size_del(opt, BLOCK_OPT_OBJECT_SIZE, 0);
1775     if (object_size) {
1776         if ((object_size - 1) & object_size) {    /* not a power of 2? */
1777             return -EINVAL;
1778         }
1779         obj_order = ctz32(object_size);
1780         if (obj_order < 20 || obj_order > 31) {
1781             return -EINVAL;
1782         }
1783         inode->block_size_shift = (uint8_t)obj_order;
1784     }
1785
1786     return 0;
1787 }
1788
1789 static int sd_create(const char *filename, QemuOpts *opts,
1790                      Error **errp)
1791 {
1792     int ret = 0;
1793     uint32_t vid = 0;
1794     char *backing_file = NULL;
1795     char *buf = NULL;
1796     BDRVSheepdogState *s;
1797     char tag[SD_MAX_VDI_TAG_LEN];
1798     uint32_t snapid;
1799     uint64_t max_vdi_size;
1800     bool prealloc = false;
1801
1802     s = g_new0(BDRVSheepdogState, 1);
1803
1804     memset(tag, 0, sizeof(tag));
1805     if (strstr(filename, "://")) {
1806         ret = sd_parse_uri(s, filename, s->name, &snapid, tag);
1807     } else {
1808         ret = parse_vdiname(s, filename, s->name, &snapid, tag);
1809     }
1810     if (ret < 0) {
1811         error_setg(errp, "Can't parse filename");
1812         goto out;
1813     }
1814
1815     s->inode.vdi_size = ROUND_UP(qemu_opt_get_size_del(opts, BLOCK_OPT_SIZE, 0),
1816                                  BDRV_SECTOR_SIZE);
1817     backing_file = qemu_opt_get_del(opts, BLOCK_OPT_BACKING_FILE);
1818     buf = qemu_opt_get_del(opts, BLOCK_OPT_PREALLOC);
1819     if (!buf || !strcmp(buf, "off")) {
1820         prealloc = false;
1821     } else if (!strcmp(buf, "full")) {
1822         prealloc = true;
1823     } else {
1824         error_setg(errp, "Invalid preallocation mode: '%s'", buf);
1825         ret = -EINVAL;
1826         goto out;
1827     }
1828
1829     g_free(buf);
1830     buf = qemu_opt_get_del(opts, BLOCK_OPT_REDUNDANCY);
1831     if (buf) {
1832         ret = parse_redundancy(s, buf);
1833         if (ret < 0) {
1834             error_setg(errp, "Invalid redundancy mode: '%s'", buf);
1835             goto out;
1836         }
1837     }
1838     ret = parse_block_size_shift(s, opts);
1839     if (ret < 0) {
1840         error_setg(errp, "Invalid object_size."
1841                          " obect_size needs to be power of 2"
1842                          " and be limited from 2^20 to 2^31");
1843         goto out;
1844     }
1845
1846     if (backing_file) {
1847         BlockBackend *blk;
1848         BDRVSheepdogState *base;
1849         BlockDriver *drv;
1850
1851         /* Currently, only Sheepdog backing image is supported. */
1852         drv = bdrv_find_protocol(backing_file, true, NULL);
1853         if (!drv || strcmp(drv->protocol_name, "sheepdog") != 0) {
1854             error_setg(errp, "backing_file must be a sheepdog image");
1855             ret = -EINVAL;
1856             goto out;
1857         }
1858
1859         blk = blk_new_open(backing_file, NULL, NULL,
1860                            BDRV_O_PROTOCOL, errp);
1861         if (blk == NULL) {
1862             ret = -EIO;
1863             goto out;
1864         }
1865
1866         base = blk_bs(blk)->opaque;
1867
1868         if (!is_snapshot(&base->inode)) {
1869             error_setg(errp, "cannot clone from a non snapshot vdi");
1870             blk_unref(blk);
1871             ret = -EINVAL;
1872             goto out;
1873         }
1874         s->inode.vdi_id = base->inode.vdi_id;
1875         blk_unref(blk);
1876     }
1877
1878     s->aio_context = qemu_get_aio_context();
1879
1880     /* if block_size_shift is not specified, get cluster default value */
1881     if (s->inode.block_size_shift == 0) {
1882         SheepdogVdiReq hdr;
1883         SheepdogClusterRsp *rsp = (SheepdogClusterRsp *)&hdr;
1884         Error *local_err = NULL;
1885         int fd;
1886         unsigned int wlen = 0, rlen = 0;
1887
1888         fd = connect_to_sdog(s, &local_err);
1889         if (fd < 0) {
1890             error_report_err(local_err);
1891             ret = -EIO;
1892             goto out;
1893         }
1894
1895         memset(&hdr, 0, sizeof(hdr));
1896         hdr.opcode = SD_OP_GET_CLUSTER_DEFAULT;
1897         hdr.proto_ver = SD_PROTO_VER;
1898
1899         ret = do_req(fd, NULL, (SheepdogReq *)&hdr,
1900                      NULL, &wlen, &rlen);
1901         closesocket(fd);
1902         if (ret) {
1903             error_setg_errno(errp, -ret, "failed to get cluster default");
1904             goto out;
1905         }
1906         if (rsp->result == SD_RES_SUCCESS) {
1907             s->inode.block_size_shift = rsp->block_size_shift;
1908         } else {
1909             s->inode.block_size_shift = SD_DEFAULT_BLOCK_SIZE_SHIFT;
1910         }
1911     }
1912
1913     max_vdi_size = (UINT64_C(1) << s->inode.block_size_shift) * MAX_DATA_OBJS;
1914
1915     if (s->inode.vdi_size > max_vdi_size) {
1916         error_setg(errp, "An image is too large."
1917                          " The maximum image size is %"PRIu64 "GB",
1918                          max_vdi_size / 1024 / 1024 / 1024);
1919         ret = -EINVAL;
1920         goto out;
1921     }
1922
1923     ret = do_sd_create(s, &vid, 0, errp);
1924     if (ret) {
1925         goto out;
1926     }
1927
1928     if (prealloc) {
1929         ret = sd_prealloc(filename, errp);
1930     }
1931 out:
1932     g_free(backing_file);
1933     g_free(buf);
1934     g_free(s);
1935     return ret;
1936 }
1937
1938 static void sd_close(BlockDriverState *bs)
1939 {
1940     Error *local_err = NULL;
1941     BDRVSheepdogState *s = bs->opaque;
1942     SheepdogVdiReq hdr;
1943     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1944     unsigned int wlen, rlen = 0;
1945     int fd, ret;
1946
1947     DPRINTF("%s\n", s->name);
1948
1949     fd = connect_to_sdog(s, &local_err);
1950     if (fd < 0) {
1951         error_report_err(local_err);
1952         return;
1953     }
1954
1955     memset(&hdr, 0, sizeof(hdr));
1956
1957     hdr.opcode = SD_OP_RELEASE_VDI;
1958     hdr.type = LOCK_TYPE_NORMAL;
1959     hdr.base_vdi_id = s->inode.vdi_id;
1960     wlen = strlen(s->name) + 1;
1961     hdr.data_length = wlen;
1962     hdr.flags = SD_FLAG_CMD_WRITE;
1963
1964     ret = do_req(fd, s->bs, (SheepdogReq *)&hdr,
1965                  s->name, &wlen, &rlen);
1966
1967     closesocket(fd);
1968
1969     if (!ret && rsp->result != SD_RES_SUCCESS &&
1970         rsp->result != SD_RES_VDI_NOT_LOCKED) {
1971         error_report("%s, %s", sd_strerror(rsp->result), s->name);
1972     }
1973
1974     aio_set_fd_handler(bdrv_get_aio_context(bs), s->fd,
1975                        false, NULL, NULL, NULL);
1976     closesocket(s->fd);
1977     g_free(s->host_spec);
1978 }
1979
1980 static int64_t sd_getlength(BlockDriverState *bs)
1981 {
1982     BDRVSheepdogState *s = bs->opaque;
1983
1984     return s->inode.vdi_size;
1985 }
1986
1987 static int sd_truncate(BlockDriverState *bs, int64_t offset)
1988 {
1989     Error *local_err = NULL;
1990     BDRVSheepdogState *s = bs->opaque;
1991     int ret, fd;
1992     unsigned int datalen;
1993     uint64_t max_vdi_size;
1994
1995     max_vdi_size = (UINT64_C(1) << s->inode.block_size_shift) * MAX_DATA_OBJS;
1996     if (offset < s->inode.vdi_size) {
1997         error_report("shrinking is not supported");
1998         return -EINVAL;
1999     } else if (offset > max_vdi_size) {
2000         error_report("too big image size");
2001         return -EINVAL;
2002     }
2003
2004     fd = connect_to_sdog(s, &local_err);
2005     if (fd < 0) {
2006         error_report_err(local_err);
2007         return fd;
2008     }
2009
2010     /* we don't need to update entire object */
2011     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
2012     s->inode.vdi_size = offset;
2013     ret = write_object(fd, s->bs, (char *)&s->inode,
2014                        vid_to_vdi_oid(s->inode.vdi_id), s->inode.nr_copies,
2015                        datalen, 0, false, s->cache_flags);
2016     close(fd);
2017
2018     if (ret < 0) {
2019         error_report("failed to update an inode.");
2020     }
2021
2022     return ret;
2023 }
2024
2025 /*
2026  * This function is called after writing data objects.  If we need to
2027  * update metadata, this sends a write request to the vdi object.
2028  * Otherwise, this switches back to sd_co_readv/writev.
2029  */
2030 static void coroutine_fn sd_write_done(SheepdogAIOCB *acb)
2031 {
2032     BDRVSheepdogState *s = acb->common.bs->opaque;
2033     struct iovec iov;
2034     AIOReq *aio_req;
2035     uint32_t offset, data_len, mn, mx;
2036
2037     mn = acb->min_dirty_data_idx;
2038     mx = acb->max_dirty_data_idx;
2039     if (mn <= mx) {
2040         /* we need to update the vdi object. */
2041         offset = sizeof(s->inode) - sizeof(s->inode.data_vdi_id) +
2042             mn * sizeof(s->inode.data_vdi_id[0]);
2043         data_len = (mx - mn + 1) * sizeof(s->inode.data_vdi_id[0]);
2044
2045         acb->min_dirty_data_idx = UINT32_MAX;
2046         acb->max_dirty_data_idx = 0;
2047
2048         iov.iov_base = &s->inode;
2049         iov.iov_len = sizeof(s->inode);
2050         aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
2051                                 data_len, offset, 0, false, 0, offset);
2052         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
2053         add_aio_request(s, aio_req, &iov, 1, AIOCB_WRITE_UDATA);
2054
2055         acb->aio_done_func = sd_finish_aiocb;
2056         acb->aiocb_type = AIOCB_WRITE_UDATA;
2057         return;
2058     }
2059
2060     sd_finish_aiocb(acb);
2061 }
2062
2063 /* Delete current working VDI on the snapshot chain */
2064 static bool sd_delete(BDRVSheepdogState *s)
2065 {
2066     Error *local_err = NULL;
2067     unsigned int wlen = SD_MAX_VDI_LEN, rlen = 0;
2068     SheepdogVdiReq hdr = {
2069         .opcode = SD_OP_DEL_VDI,
2070         .base_vdi_id = s->inode.vdi_id,
2071         .data_length = wlen,
2072         .flags = SD_FLAG_CMD_WRITE,
2073     };
2074     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
2075     int fd, ret;
2076
2077     fd = connect_to_sdog(s, &local_err);
2078     if (fd < 0) {
2079         error_report_err(local_err);
2080         return false;
2081     }
2082
2083     ret = do_req(fd, s->bs, (SheepdogReq *)&hdr,
2084                  s->name, &wlen, &rlen);
2085     closesocket(fd);
2086     if (ret) {
2087         return false;
2088     }
2089     switch (rsp->result) {
2090     case SD_RES_NO_VDI:
2091         error_report("%s was already deleted", s->name);
2092         /* fall through */
2093     case SD_RES_SUCCESS:
2094         break;
2095     default:
2096         error_report("%s, %s", sd_strerror(rsp->result), s->name);
2097         return false;
2098     }
2099
2100     return true;
2101 }
2102
2103 /*
2104  * Create a writable VDI from a snapshot
2105  */
2106 static int sd_create_branch(BDRVSheepdogState *s)
2107 {
2108     Error *local_err = NULL;
2109     int ret, fd;
2110     uint32_t vid;
2111     char *buf;
2112     bool deleted;
2113
2114     DPRINTF("%" PRIx32 " is snapshot.\n", s->inode.vdi_id);
2115
2116     buf = g_malloc(SD_INODE_SIZE);
2117
2118     /*
2119      * Even If deletion fails, we will just create extra snapshot based on
2120      * the working VDI which was supposed to be deleted. So no need to
2121      * false bail out.
2122      */
2123     deleted = sd_delete(s);
2124     ret = do_sd_create(s, &vid, !deleted, &local_err);
2125     if (ret) {
2126         error_report_err(local_err);
2127         goto out;
2128     }
2129
2130     DPRINTF("%" PRIx32 " is created.\n", vid);
2131
2132     fd = connect_to_sdog(s, &local_err);
2133     if (fd < 0) {
2134         error_report_err(local_err);
2135         ret = fd;
2136         goto out;
2137     }
2138
2139     ret = read_object(fd, s->bs, buf, vid_to_vdi_oid(vid),
2140                       s->inode.nr_copies, SD_INODE_SIZE, 0, s->cache_flags);
2141
2142     closesocket(fd);
2143
2144     if (ret < 0) {
2145         goto out;
2146     }
2147
2148     memcpy(&s->inode, buf, sizeof(s->inode));
2149
2150     s->is_snapshot = false;
2151     ret = 0;
2152     DPRINTF("%" PRIx32 " was newly created.\n", s->inode.vdi_id);
2153
2154 out:
2155     g_free(buf);
2156
2157     return ret;
2158 }
2159
2160 /*
2161  * Send I/O requests to the server.
2162  *
2163  * This function sends requests to the server, links the requests to
2164  * the inflight_list in BDRVSheepdogState, and exits without
2165  * waiting the response.  The responses are received in the
2166  * `aio_read_response' function which is called from the main loop as
2167  * a fd handler.
2168  *
2169  * Returns 1 when we need to wait a response, 0 when there is no sent
2170  * request and -errno in error cases.
2171  */
2172 static int coroutine_fn sd_co_rw_vector(void *p)
2173 {
2174     SheepdogAIOCB *acb = p;
2175     int ret = 0;
2176     unsigned long len, done = 0, total = acb->nb_sectors * BDRV_SECTOR_SIZE;
2177     unsigned long idx;
2178     uint32_t object_size;
2179     uint64_t oid;
2180     uint64_t offset;
2181     BDRVSheepdogState *s = acb->common.bs->opaque;
2182     SheepdogInode *inode = &s->inode;
2183     AIOReq *aio_req;
2184
2185     if (acb->aiocb_type == AIOCB_WRITE_UDATA && s->is_snapshot) {
2186         /*
2187          * In the case we open the snapshot VDI, Sheepdog creates the
2188          * writable VDI when we do a write operation first.
2189          */
2190         ret = sd_create_branch(s);
2191         if (ret) {
2192             acb->ret = -EIO;
2193             goto out;
2194         }
2195     }
2196
2197     object_size = (UINT32_C(1) << inode->block_size_shift);
2198     idx = acb->sector_num * BDRV_SECTOR_SIZE / object_size;
2199     offset = (acb->sector_num * BDRV_SECTOR_SIZE) % object_size;
2200
2201     /*
2202      * Make sure we don't free the aiocb before we are done with all requests.
2203      * This additional reference is dropped at the end of this function.
2204      */
2205     acb->nr_pending++;
2206
2207     while (done != total) {
2208         uint8_t flags = 0;
2209         uint64_t old_oid = 0;
2210         bool create = false;
2211
2212         oid = vid_to_data_oid(inode->data_vdi_id[idx], idx);
2213
2214         len = MIN(total - done, object_size - offset);
2215
2216         switch (acb->aiocb_type) {
2217         case AIOCB_READ_UDATA:
2218             if (!inode->data_vdi_id[idx]) {
2219                 qemu_iovec_memset(acb->qiov, done, 0, len);
2220                 goto done;
2221             }
2222             break;
2223         case AIOCB_WRITE_UDATA:
2224             if (!inode->data_vdi_id[idx]) {
2225                 create = true;
2226             } else if (!is_data_obj_writable(inode, idx)) {
2227                 /* Copy-On-Write */
2228                 create = true;
2229                 old_oid = oid;
2230                 flags = SD_FLAG_CMD_COW;
2231             }
2232             break;
2233         case AIOCB_DISCARD_OBJ:
2234             /*
2235              * We discard the object only when the whole object is
2236              * 1) allocated 2) trimmed. Otherwise, simply skip it.
2237              */
2238             if (len != object_size || inode->data_vdi_id[idx] == 0) {
2239                 goto done;
2240             }
2241             break;
2242         default:
2243             break;
2244         }
2245
2246         if (create) {
2247             DPRINTF("update ino (%" PRIu32 ") %" PRIu64 " %" PRIu64 " %ld\n",
2248                     inode->vdi_id, oid,
2249                     vid_to_data_oid(inode->data_vdi_id[idx], idx), idx);
2250             oid = vid_to_data_oid(inode->vdi_id, idx);
2251             DPRINTF("new oid %" PRIx64 "\n", oid);
2252         }
2253
2254         aio_req = alloc_aio_req(s, acb, oid, len, offset, flags, create,
2255                                 old_oid,
2256                                 acb->aiocb_type == AIOCB_DISCARD_OBJ ?
2257                                 0 : done);
2258         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
2259
2260         add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
2261                         acb->aiocb_type);
2262     done:
2263         offset = 0;
2264         idx++;
2265         done += len;
2266     }
2267 out:
2268     if (!--acb->nr_pending) {
2269         return acb->ret;
2270     }
2271     return 1;
2272 }
2273
2274 static bool check_overlapping_aiocb(BDRVSheepdogState *s, SheepdogAIOCB *aiocb)
2275 {
2276     SheepdogAIOCB *cb;
2277
2278     QLIST_FOREACH(cb, &s->inflight_aiocb_head, aiocb_siblings) {
2279         if (AIOCBOverlapping(aiocb, cb)) {
2280             return true;
2281         }
2282     }
2283
2284     QLIST_INSERT_HEAD(&s->inflight_aiocb_head, aiocb, aiocb_siblings);
2285     return false;
2286 }
2287
2288 static coroutine_fn int sd_co_writev(BlockDriverState *bs, int64_t sector_num,
2289                         int nb_sectors, QEMUIOVector *qiov)
2290 {
2291     SheepdogAIOCB *acb;
2292     int ret;
2293     int64_t offset = (sector_num + nb_sectors) * BDRV_SECTOR_SIZE;
2294     BDRVSheepdogState *s = bs->opaque;
2295
2296     if (offset > s->inode.vdi_size) {
2297         ret = sd_truncate(bs, offset);
2298         if (ret < 0) {
2299             return ret;
2300         }
2301     }
2302
2303     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors);
2304     acb->aio_done_func = sd_write_done;
2305     acb->aiocb_type = AIOCB_WRITE_UDATA;
2306
2307 retry:
2308     if (check_overlapping_aiocb(s, acb)) {
2309         qemu_co_queue_wait(&s->overlapping_queue);
2310         goto retry;
2311     }
2312
2313     ret = sd_co_rw_vector(acb);
2314     if (ret <= 0) {
2315         QLIST_REMOVE(acb, aiocb_siblings);
2316         qemu_co_queue_restart_all(&s->overlapping_queue);
2317         qemu_aio_unref(acb);
2318         return ret;
2319     }
2320
2321     qemu_coroutine_yield();
2322
2323     QLIST_REMOVE(acb, aiocb_siblings);
2324     qemu_co_queue_restart_all(&s->overlapping_queue);
2325
2326     return acb->ret;
2327 }
2328
2329 static coroutine_fn int sd_co_readv(BlockDriverState *bs, int64_t sector_num,
2330                        int nb_sectors, QEMUIOVector *qiov)
2331 {
2332     SheepdogAIOCB *acb;
2333     int ret;
2334     BDRVSheepdogState *s = bs->opaque;
2335
2336     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors);
2337     acb->aiocb_type = AIOCB_READ_UDATA;
2338     acb->aio_done_func = sd_finish_aiocb;
2339
2340 retry:
2341     if (check_overlapping_aiocb(s, acb)) {
2342         qemu_co_queue_wait(&s->overlapping_queue);
2343         goto retry;
2344     }
2345
2346     ret = sd_co_rw_vector(acb);
2347     if (ret <= 0) {
2348         QLIST_REMOVE(acb, aiocb_siblings);
2349         qemu_co_queue_restart_all(&s->overlapping_queue);
2350         qemu_aio_unref(acb);
2351         return ret;
2352     }
2353
2354     qemu_coroutine_yield();
2355
2356     QLIST_REMOVE(acb, aiocb_siblings);
2357     qemu_co_queue_restart_all(&s->overlapping_queue);
2358     return acb->ret;
2359 }
2360
2361 static int coroutine_fn sd_co_flush_to_disk(BlockDriverState *bs)
2362 {
2363     BDRVSheepdogState *s = bs->opaque;
2364     SheepdogAIOCB *acb;
2365     AIOReq *aio_req;
2366
2367     if (s->cache_flags != SD_FLAG_CMD_CACHE) {
2368         return 0;
2369     }
2370
2371     acb = sd_aio_setup(bs, NULL, 0, 0);
2372     acb->aiocb_type = AIOCB_FLUSH_CACHE;
2373     acb->aio_done_func = sd_finish_aiocb;
2374
2375     aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
2376                             0, 0, 0, false, 0, 0);
2377     QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
2378     add_aio_request(s, aio_req, NULL, 0, acb->aiocb_type);
2379
2380     qemu_coroutine_yield();
2381     return acb->ret;
2382 }
2383
2384 static int sd_snapshot_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info)
2385 {
2386     Error *local_err = NULL;
2387     BDRVSheepdogState *s = bs->opaque;
2388     int ret, fd;
2389     uint32_t new_vid;
2390     SheepdogInode *inode;
2391     unsigned int datalen;
2392
2393     DPRINTF("sn_info: name %s id_str %s s: name %s vm_state_size %" PRId64 " "
2394             "is_snapshot %d\n", sn_info->name, sn_info->id_str,
2395             s->name, sn_info->vm_state_size, s->is_snapshot);
2396
2397     if (s->is_snapshot) {
2398         error_report("You can't create a snapshot of a snapshot VDI, "
2399                      "%s (%" PRIu32 ").", s->name, s->inode.vdi_id);
2400
2401         return -EINVAL;
2402     }
2403
2404     DPRINTF("%s %s\n", sn_info->name, sn_info->id_str);
2405
2406     s->inode.vm_state_size = sn_info->vm_state_size;
2407     s->inode.vm_clock_nsec = sn_info->vm_clock_nsec;
2408     /* It appears that inode.tag does not require a NUL terminator,
2409      * which means this use of strncpy is ok.
2410      */
2411     strncpy(s->inode.tag, sn_info->name, sizeof(s->inode.tag));
2412     /* we don't need to update entire object */
2413     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
2414     inode = g_malloc(datalen);
2415
2416     /* refresh inode. */
2417     fd = connect_to_sdog(s, &local_err);
2418     if (fd < 0) {
2419         error_report_err(local_err);
2420         ret = fd;
2421         goto cleanup;
2422     }
2423
2424     ret = write_object(fd, s->bs, (char *)&s->inode,
2425                        vid_to_vdi_oid(s->inode.vdi_id), s->inode.nr_copies,
2426                        datalen, 0, false, s->cache_flags);
2427     if (ret < 0) {
2428         error_report("failed to write snapshot's inode.");
2429         goto cleanup;
2430     }
2431
2432     ret = do_sd_create(s, &new_vid, 1, &local_err);
2433     if (ret < 0) {
2434         error_reportf_err(local_err,
2435                           "failed to create inode for snapshot: ");
2436         goto cleanup;
2437     }
2438
2439     ret = read_object(fd, s->bs, (char *)inode,
2440                       vid_to_vdi_oid(new_vid), s->inode.nr_copies, datalen, 0,
2441                       s->cache_flags);
2442
2443     if (ret < 0) {
2444         error_report("failed to read new inode info. %s", strerror(errno));
2445         goto cleanup;
2446     }
2447
2448     memcpy(&s->inode, inode, datalen);
2449     DPRINTF("s->inode: name %s snap_id %x oid %x\n",
2450             s->inode.name, s->inode.snap_id, s->inode.vdi_id);
2451
2452 cleanup:
2453     g_free(inode);
2454     closesocket(fd);
2455     return ret;
2456 }
2457
2458 /*
2459  * We implement rollback(loadvm) operation to the specified snapshot by
2460  * 1) switch to the snapshot
2461  * 2) rely on sd_create_branch to delete working VDI and
2462  * 3) create a new working VDI based on the specified snapshot
2463  */
2464 static int sd_snapshot_goto(BlockDriverState *bs, const char *snapshot_id)
2465 {
2466     BDRVSheepdogState *s = bs->opaque;
2467     BDRVSheepdogState *old_s;
2468     char tag[SD_MAX_VDI_TAG_LEN];
2469     uint32_t snapid = 0;
2470     int ret = 0;
2471
2472     old_s = g_new(BDRVSheepdogState, 1);
2473
2474     memcpy(old_s, s, sizeof(BDRVSheepdogState));
2475
2476     snapid = strtoul(snapshot_id, NULL, 10);
2477     if (snapid) {
2478         tag[0] = 0;
2479     } else {
2480         pstrcpy(tag, sizeof(tag), snapshot_id);
2481     }
2482
2483     ret = reload_inode(s, snapid, tag);
2484     if (ret) {
2485         goto out;
2486     }
2487
2488     ret = sd_create_branch(s);
2489     if (ret) {
2490         goto out;
2491     }
2492
2493     g_free(old_s);
2494
2495     return 0;
2496 out:
2497     /* recover bdrv_sd_state */
2498     memcpy(s, old_s, sizeof(BDRVSheepdogState));
2499     g_free(old_s);
2500
2501     error_report("failed to open. recover old bdrv_sd_state.");
2502
2503     return ret;
2504 }
2505
2506 #define NR_BATCHED_DISCARD 128
2507
2508 static bool remove_objects(BDRVSheepdogState *s)
2509 {
2510     int fd, i = 0, nr_objs = 0;
2511     Error *local_err = NULL;
2512     int ret = 0;
2513     bool result = true;
2514     SheepdogInode *inode = &s->inode;
2515
2516     fd = connect_to_sdog(s, &local_err);
2517     if (fd < 0) {
2518         error_report_err(local_err);
2519         return false;
2520     }
2521
2522     nr_objs = count_data_objs(inode);
2523     while (i < nr_objs) {
2524         int start_idx, nr_filled_idx;
2525
2526         while (i < nr_objs && !inode->data_vdi_id[i]) {
2527             i++;
2528         }
2529         start_idx = i;
2530
2531         nr_filled_idx = 0;
2532         while (i < nr_objs && nr_filled_idx < NR_BATCHED_DISCARD) {
2533             if (inode->data_vdi_id[i]) {
2534                 inode->data_vdi_id[i] = 0;
2535                 nr_filled_idx++;
2536             }
2537
2538             i++;
2539         }
2540
2541         ret = write_object(fd, s->bs,
2542                            (char *)&inode->data_vdi_id[start_idx],
2543                            vid_to_vdi_oid(s->inode.vdi_id), inode->nr_copies,
2544                            (i - start_idx) * sizeof(uint32_t),
2545                            offsetof(struct SheepdogInode,
2546                                     data_vdi_id[start_idx]),
2547                            false, s->cache_flags);
2548         if (ret < 0) {
2549             error_report("failed to discard snapshot inode.");
2550             result = false;
2551             goto out;
2552         }
2553     }
2554
2555 out:
2556     closesocket(fd);
2557     return result;
2558 }
2559
2560 static int sd_snapshot_delete(BlockDriverState *bs,
2561                               const char *snapshot_id,
2562                               const char *name,
2563                               Error **errp)
2564 {
2565     unsigned long snap_id = 0;
2566     char snap_tag[SD_MAX_VDI_TAG_LEN];
2567     Error *local_err = NULL;
2568     int fd, ret;
2569     char buf[SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN];
2570     BDRVSheepdogState *s = bs->opaque;
2571     unsigned int wlen = SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN, rlen = 0;
2572     uint32_t vid;
2573     SheepdogVdiReq hdr = {
2574         .opcode = SD_OP_DEL_VDI,
2575         .data_length = wlen,
2576         .flags = SD_FLAG_CMD_WRITE,
2577     };
2578     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
2579
2580     if (!remove_objects(s)) {
2581         return -1;
2582     }
2583
2584     memset(buf, 0, sizeof(buf));
2585     memset(snap_tag, 0, sizeof(snap_tag));
2586     pstrcpy(buf, SD_MAX_VDI_LEN, s->name);
2587     ret = qemu_strtoul(snapshot_id, NULL, 10, &snap_id);
2588     if (ret || snap_id > UINT32_MAX) {
2589         error_setg(errp, "Invalid snapshot ID: %s",
2590                          snapshot_id ? snapshot_id : "<null>");
2591         return -EINVAL;
2592     }
2593
2594     if (snap_id) {
2595         hdr.snapid = (uint32_t) snap_id;
2596     } else {
2597         pstrcpy(snap_tag, sizeof(snap_tag), snapshot_id);
2598         pstrcpy(buf + SD_MAX_VDI_LEN, SD_MAX_VDI_TAG_LEN, snap_tag);
2599     }
2600
2601     ret = find_vdi_name(s, s->name, snap_id, snap_tag, &vid, true,
2602                         &local_err);
2603     if (ret) {
2604         return ret;
2605     }
2606
2607     fd = connect_to_sdog(s, &local_err);
2608     if (fd < 0) {
2609         error_report_err(local_err);
2610         return -1;
2611     }
2612
2613     ret = do_req(fd, s->bs, (SheepdogReq *)&hdr,
2614                  buf, &wlen, &rlen);
2615     closesocket(fd);
2616     if (ret) {
2617         return ret;
2618     }
2619
2620     switch (rsp->result) {
2621     case SD_RES_NO_VDI:
2622         error_report("%s was already deleted", s->name);
2623     case SD_RES_SUCCESS:
2624         break;
2625     default:
2626         error_report("%s, %s", sd_strerror(rsp->result), s->name);
2627         return -1;
2628     }
2629
2630     return ret;
2631 }
2632
2633 static int sd_snapshot_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab)
2634 {
2635     Error *local_err = NULL;
2636     BDRVSheepdogState *s = bs->opaque;
2637     SheepdogReq req;
2638     int fd, nr = 1024, ret, max = BITS_TO_LONGS(SD_NR_VDIS) * sizeof(long);
2639     QEMUSnapshotInfo *sn_tab = NULL;
2640     unsigned wlen, rlen;
2641     int found = 0;
2642     static SheepdogInode inode;
2643     unsigned long *vdi_inuse;
2644     unsigned int start_nr;
2645     uint64_t hval;
2646     uint32_t vid;
2647
2648     vdi_inuse = g_malloc(max);
2649
2650     fd = connect_to_sdog(s, &local_err);
2651     if (fd < 0) {
2652         error_report_err(local_err);
2653         ret = fd;
2654         goto out;
2655     }
2656
2657     rlen = max;
2658     wlen = 0;
2659
2660     memset(&req, 0, sizeof(req));
2661
2662     req.opcode = SD_OP_READ_VDIS;
2663     req.data_length = max;
2664
2665     ret = do_req(fd, s->bs, &req, vdi_inuse, &wlen, &rlen);
2666
2667     closesocket(fd);
2668     if (ret) {
2669         goto out;
2670     }
2671
2672     sn_tab = g_new0(QEMUSnapshotInfo, nr);
2673
2674     /* calculate a vdi id with hash function */
2675     hval = fnv_64a_buf(s->name, strlen(s->name), FNV1A_64_INIT);
2676     start_nr = hval & (SD_NR_VDIS - 1);
2677
2678     fd = connect_to_sdog(s, &local_err);
2679     if (fd < 0) {
2680         error_report_err(local_err);
2681         ret = fd;
2682         goto out;
2683     }
2684
2685     for (vid = start_nr; found < nr; vid = (vid + 1) % SD_NR_VDIS) {
2686         if (!test_bit(vid, vdi_inuse)) {
2687             break;
2688         }
2689
2690         /* we don't need to read entire object */
2691         ret = read_object(fd, s->bs, (char *)&inode,
2692                           vid_to_vdi_oid(vid),
2693                           0, SD_INODE_SIZE - sizeof(inode.data_vdi_id), 0,
2694                           s->cache_flags);
2695
2696         if (ret) {
2697             continue;
2698         }
2699
2700         if (!strcmp(inode.name, s->name) && is_snapshot(&inode)) {
2701             sn_tab[found].date_sec = inode.snap_ctime >> 32;
2702             sn_tab[found].date_nsec = inode.snap_ctime & 0xffffffff;
2703             sn_tab[found].vm_state_size = inode.vm_state_size;
2704             sn_tab[found].vm_clock_nsec = inode.vm_clock_nsec;
2705
2706             snprintf(sn_tab[found].id_str, sizeof(sn_tab[found].id_str),
2707                      "%" PRIu32, inode.snap_id);
2708             pstrcpy(sn_tab[found].name,
2709                     MIN(sizeof(sn_tab[found].name), sizeof(inode.tag)),
2710                     inode.tag);
2711             found++;
2712         }
2713     }
2714
2715     closesocket(fd);
2716 out:
2717     *psn_tab = sn_tab;
2718
2719     g_free(vdi_inuse);
2720
2721     if (ret < 0) {
2722         return ret;
2723     }
2724
2725     return found;
2726 }
2727
2728 static int do_load_save_vmstate(BDRVSheepdogState *s, uint8_t *data,
2729                                 int64_t pos, int size, int load)
2730 {
2731     Error *local_err = NULL;
2732     bool create;
2733     int fd, ret = 0, remaining = size;
2734     unsigned int data_len;
2735     uint64_t vmstate_oid;
2736     uint64_t offset;
2737     uint32_t vdi_index;
2738     uint32_t vdi_id = load ? s->inode.parent_vdi_id : s->inode.vdi_id;
2739     uint32_t object_size = (UINT32_C(1) << s->inode.block_size_shift);
2740
2741     fd = connect_to_sdog(s, &local_err);
2742     if (fd < 0) {
2743         error_report_err(local_err);
2744         return fd;
2745     }
2746
2747     while (remaining) {
2748         vdi_index = pos / object_size;
2749         offset = pos % object_size;
2750
2751         data_len = MIN(remaining, object_size - offset);
2752
2753         vmstate_oid = vid_to_vmstate_oid(vdi_id, vdi_index);
2754
2755         create = (offset == 0);
2756         if (load) {
2757             ret = read_object(fd, s->bs, (char *)data, vmstate_oid,
2758                               s->inode.nr_copies, data_len, offset,
2759                               s->cache_flags);
2760         } else {
2761             ret = write_object(fd, s->bs, (char *)data, vmstate_oid,
2762                                s->inode.nr_copies, data_len, offset, create,
2763                                s->cache_flags);
2764         }
2765
2766         if (ret < 0) {
2767             error_report("failed to save vmstate %s", strerror(errno));
2768             goto cleanup;
2769         }
2770
2771         pos += data_len;
2772         data += data_len;
2773         remaining -= data_len;
2774     }
2775     ret = size;
2776 cleanup:
2777     closesocket(fd);
2778     return ret;
2779 }
2780
2781 static int sd_save_vmstate(BlockDriverState *bs, QEMUIOVector *qiov,
2782                            int64_t pos)
2783 {
2784     BDRVSheepdogState *s = bs->opaque;
2785     void *buf;
2786     int ret;
2787
2788     buf = qemu_blockalign(bs, qiov->size);
2789     qemu_iovec_to_buf(qiov, 0, buf, qiov->size);
2790     ret = do_load_save_vmstate(s, (uint8_t *) buf, pos, qiov->size, 0);
2791     qemu_vfree(buf);
2792
2793     return ret;
2794 }
2795
2796 static int sd_load_vmstate(BlockDriverState *bs, QEMUIOVector *qiov,
2797                            int64_t pos)
2798 {
2799     BDRVSheepdogState *s = bs->opaque;
2800     void *buf;
2801     int ret;
2802
2803     buf = qemu_blockalign(bs, qiov->size);
2804     ret = do_load_save_vmstate(s, buf, pos, qiov->size, 1);
2805     qemu_iovec_from_buf(qiov, 0, buf, qiov->size);
2806     qemu_vfree(buf);
2807
2808     return ret;
2809 }
2810
2811
2812 static coroutine_fn int sd_co_pdiscard(BlockDriverState *bs, int64_t offset,
2813                                       int count)
2814 {
2815     SheepdogAIOCB *acb;
2816     BDRVSheepdogState *s = bs->opaque;
2817     int ret;
2818     QEMUIOVector discard_iov;
2819     struct iovec iov;
2820     uint32_t zero = 0;
2821
2822     if (!s->discard_supported) {
2823         return 0;
2824     }
2825
2826     memset(&discard_iov, 0, sizeof(discard_iov));
2827     memset(&iov, 0, sizeof(iov));
2828     iov.iov_base = &zero;
2829     iov.iov_len = sizeof(zero);
2830     discard_iov.iov = &iov;
2831     discard_iov.niov = 1;
2832     if (!QEMU_IS_ALIGNED(offset | count, BDRV_SECTOR_SIZE)) {
2833         return -ENOTSUP;
2834     }
2835     acb = sd_aio_setup(bs, &discard_iov, offset >> BDRV_SECTOR_BITS,
2836                        count >> BDRV_SECTOR_BITS);
2837     acb->aiocb_type = AIOCB_DISCARD_OBJ;
2838     acb->aio_done_func = sd_finish_aiocb;
2839
2840 retry:
2841     if (check_overlapping_aiocb(s, acb)) {
2842         qemu_co_queue_wait(&s->overlapping_queue);
2843         goto retry;
2844     }
2845
2846     ret = sd_co_rw_vector(acb);
2847     if (ret <= 0) {
2848         QLIST_REMOVE(acb, aiocb_siblings);
2849         qemu_co_queue_restart_all(&s->overlapping_queue);
2850         qemu_aio_unref(acb);
2851         return ret;
2852     }
2853
2854     qemu_coroutine_yield();
2855
2856     QLIST_REMOVE(acb, aiocb_siblings);
2857     qemu_co_queue_restart_all(&s->overlapping_queue);
2858
2859     return acb->ret;
2860 }
2861
2862 static coroutine_fn int64_t
2863 sd_co_get_block_status(BlockDriverState *bs, int64_t sector_num, int nb_sectors,
2864                        int *pnum, BlockDriverState **file)
2865 {
2866     BDRVSheepdogState *s = bs->opaque;
2867     SheepdogInode *inode = &s->inode;
2868     uint32_t object_size = (UINT32_C(1) << inode->block_size_shift);
2869     uint64_t offset = sector_num * BDRV_SECTOR_SIZE;
2870     unsigned long start = offset / object_size,
2871                   end = DIV_ROUND_UP((sector_num + nb_sectors) *
2872                                      BDRV_SECTOR_SIZE, object_size);
2873     unsigned long idx;
2874     int64_t ret = BDRV_BLOCK_DATA | BDRV_BLOCK_OFFSET_VALID | offset;
2875
2876     for (idx = start; idx < end; idx++) {
2877         if (inode->data_vdi_id[idx] == 0) {
2878             break;
2879         }
2880     }
2881     if (idx == start) {
2882         /* Get the longest length of unallocated sectors */
2883         ret = 0;
2884         for (idx = start + 1; idx < end; idx++) {
2885             if (inode->data_vdi_id[idx] != 0) {
2886                 break;
2887             }
2888         }
2889     }
2890
2891     *pnum = (idx - start) * object_size / BDRV_SECTOR_SIZE;
2892     if (*pnum > nb_sectors) {
2893         *pnum = nb_sectors;
2894     }
2895     if (ret > 0 && ret & BDRV_BLOCK_OFFSET_VALID) {
2896         *file = bs;
2897     }
2898     return ret;
2899 }
2900
2901 static int64_t sd_get_allocated_file_size(BlockDriverState *bs)
2902 {
2903     BDRVSheepdogState *s = bs->opaque;
2904     SheepdogInode *inode = &s->inode;
2905     uint32_t object_size = (UINT32_C(1) << inode->block_size_shift);
2906     unsigned long i, last = DIV_ROUND_UP(inode->vdi_size, object_size);
2907     uint64_t size = 0;
2908
2909     for (i = 0; i < last; i++) {
2910         if (inode->data_vdi_id[i] == 0) {
2911             continue;
2912         }
2913         size += object_size;
2914     }
2915     return size;
2916 }
2917
2918 static QemuOptsList sd_create_opts = {
2919     .name = "sheepdog-create-opts",
2920     .head = QTAILQ_HEAD_INITIALIZER(sd_create_opts.head),
2921     .desc = {
2922         {
2923             .name = BLOCK_OPT_SIZE,
2924             .type = QEMU_OPT_SIZE,
2925             .help = "Virtual disk size"
2926         },
2927         {
2928             .name = BLOCK_OPT_BACKING_FILE,
2929             .type = QEMU_OPT_STRING,
2930             .help = "File name of a base image"
2931         },
2932         {
2933             .name = BLOCK_OPT_PREALLOC,
2934             .type = QEMU_OPT_STRING,
2935             .help = "Preallocation mode (allowed values: off, full)"
2936         },
2937         {
2938             .name = BLOCK_OPT_REDUNDANCY,
2939             .type = QEMU_OPT_STRING,
2940             .help = "Redundancy of the image"
2941         },
2942         {
2943             .name = BLOCK_OPT_OBJECT_SIZE,
2944             .type = QEMU_OPT_SIZE,
2945             .help = "Object size of the image"
2946         },
2947         { /* end of list */ }
2948     }
2949 };
2950
2951 static BlockDriver bdrv_sheepdog = {
2952     .format_name    = "sheepdog",
2953     .protocol_name  = "sheepdog",
2954     .instance_size  = sizeof(BDRVSheepdogState),
2955     .bdrv_needs_filename = true,
2956     .bdrv_file_open = sd_open,
2957     .bdrv_reopen_prepare    = sd_reopen_prepare,
2958     .bdrv_reopen_commit     = sd_reopen_commit,
2959     .bdrv_reopen_abort      = sd_reopen_abort,
2960     .bdrv_close     = sd_close,
2961     .bdrv_create    = sd_create,
2962     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2963     .bdrv_getlength = sd_getlength,
2964     .bdrv_get_allocated_file_size = sd_get_allocated_file_size,
2965     .bdrv_truncate  = sd_truncate,
2966
2967     .bdrv_co_readv  = sd_co_readv,
2968     .bdrv_co_writev = sd_co_writev,
2969     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2970     .bdrv_co_pdiscard = sd_co_pdiscard,
2971     .bdrv_co_get_block_status = sd_co_get_block_status,
2972
2973     .bdrv_snapshot_create   = sd_snapshot_create,
2974     .bdrv_snapshot_goto     = sd_snapshot_goto,
2975     .bdrv_snapshot_delete   = sd_snapshot_delete,
2976     .bdrv_snapshot_list     = sd_snapshot_list,
2977
2978     .bdrv_save_vmstate  = sd_save_vmstate,
2979     .bdrv_load_vmstate  = sd_load_vmstate,
2980
2981     .bdrv_detach_aio_context = sd_detach_aio_context,
2982     .bdrv_attach_aio_context = sd_attach_aio_context,
2983
2984     .create_opts    = &sd_create_opts,
2985 };
2986
2987 static BlockDriver bdrv_sheepdog_tcp = {
2988     .format_name    = "sheepdog",
2989     .protocol_name  = "sheepdog+tcp",
2990     .instance_size  = sizeof(BDRVSheepdogState),
2991     .bdrv_needs_filename = true,
2992     .bdrv_file_open = sd_open,
2993     .bdrv_reopen_prepare    = sd_reopen_prepare,
2994     .bdrv_reopen_commit     = sd_reopen_commit,
2995     .bdrv_reopen_abort      = sd_reopen_abort,
2996     .bdrv_close     = sd_close,
2997     .bdrv_create    = sd_create,
2998     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2999     .bdrv_getlength = sd_getlength,
3000     .bdrv_get_allocated_file_size = sd_get_allocated_file_size,
3001     .bdrv_truncate  = sd_truncate,
3002
3003     .bdrv_co_readv  = sd_co_readv,
3004     .bdrv_co_writev = sd_co_writev,
3005     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
3006     .bdrv_co_pdiscard = sd_co_pdiscard,
3007     .bdrv_co_get_block_status = sd_co_get_block_status,
3008
3009     .bdrv_snapshot_create   = sd_snapshot_create,
3010     .bdrv_snapshot_goto     = sd_snapshot_goto,
3011     .bdrv_snapshot_delete   = sd_snapshot_delete,
3012     .bdrv_snapshot_list     = sd_snapshot_list,
3013
3014     .bdrv_save_vmstate  = sd_save_vmstate,
3015     .bdrv_load_vmstate  = sd_load_vmstate,
3016
3017     .bdrv_detach_aio_context = sd_detach_aio_context,
3018     .bdrv_attach_aio_context = sd_attach_aio_context,
3019
3020     .create_opts    = &sd_create_opts,
3021 };
3022
3023 static BlockDriver bdrv_sheepdog_unix = {
3024     .format_name    = "sheepdog",
3025     .protocol_name  = "sheepdog+unix",
3026     .instance_size  = sizeof(BDRVSheepdogState),
3027     .bdrv_needs_filename = true,
3028     .bdrv_file_open = sd_open,
3029     .bdrv_reopen_prepare    = sd_reopen_prepare,
3030     .bdrv_reopen_commit     = sd_reopen_commit,
3031     .bdrv_reopen_abort      = sd_reopen_abort,
3032     .bdrv_close     = sd_close,
3033     .bdrv_create    = sd_create,
3034     .bdrv_has_zero_init = bdrv_has_zero_init_1,
3035     .bdrv_getlength = sd_getlength,
3036     .bdrv_get_allocated_file_size = sd_get_allocated_file_size,
3037     .bdrv_truncate  = sd_truncate,
3038
3039     .bdrv_co_readv  = sd_co_readv,
3040     .bdrv_co_writev = sd_co_writev,
3041     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
3042     .bdrv_co_pdiscard = sd_co_pdiscard,
3043     .bdrv_co_get_block_status = sd_co_get_block_status,
3044
3045     .bdrv_snapshot_create   = sd_snapshot_create,
3046     .bdrv_snapshot_goto     = sd_snapshot_goto,
3047     .bdrv_snapshot_delete   = sd_snapshot_delete,
3048     .bdrv_snapshot_list     = sd_snapshot_list,
3049
3050     .bdrv_save_vmstate  = sd_save_vmstate,
3051     .bdrv_load_vmstate  = sd_load_vmstate,
3052
3053     .bdrv_detach_aio_context = sd_detach_aio_context,
3054     .bdrv_attach_aio_context = sd_attach_aio_context,
3055
3056     .create_opts    = &sd_create_opts,
3057 };
3058
3059 static void bdrv_sheepdog_init(void)
3060 {
3061     bdrv_register(&bdrv_sheepdog);
3062     bdrv_register(&bdrv_sheepdog_tcp);
3063     bdrv_register(&bdrv_sheepdog_unix);
3064 }
3065 block_init(bdrv_sheepdog_init);