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-common.h"
  16 #include "qemu/uri.h"
  17 #include "qemu/error-report.h"
  18 #include "qemu/sockets.h"
  19 #include "block/block_int.h"
  20 #include "qemu/bitops.h"
  21
  22 #define SD_PROTO_VER 0x01
  23
  24 #define SD_DEFAULT_ADDR "localhost"
  25 #define SD_DEFAULT_PORT 7000
  26
  27 #define SD_OP_CREATE_AND_WRITE_OBJ  0x01
  28 #define SD_OP_READ_OBJ       0x02
  29 #define SD_OP_WRITE_OBJ      0x03
  30 /* 0x04 is used internally by Sheepdog */
  31 #define SD_OP_DISCARD_OBJ    0x05
  32
  33 #define SD_OP_NEW_VDI        0x11
  34 #define SD_OP_LOCK_VDI       0x12
  35 #define SD_OP_RELEASE_VDI    0x13
  36 #define SD_OP_GET_VDI_INFO   0x14
  37 #define SD_OP_READ_VDIS      0x15
  38 #define SD_OP_FLUSH_VDI      0x16
  39 #define SD_OP_DEL_VDI        0x17
  40
  41 #define SD_FLAG_CMD_WRITE    0x01
  42 #define SD_FLAG_CMD_COW      0x02
  43 #define SD_FLAG_CMD_CACHE    0x04 /* Writeback mode for cache */
  44 #define SD_FLAG_CMD_DIRECT   0x08 /* Don't use cache */
  45
  46 #define SD_RES_SUCCESS       0x00 /* Success */
  47 #define SD_RES_UNKNOWN       0x01 /* Unknown error */
  48 #define SD_RES_NO_OBJ        0x02 /* No object found */
  49 #define SD_RES_EIO           0x03 /* I/O error */
  50 #define SD_RES_VDI_EXIST     0x04 /* Vdi exists already */
  51 #define SD_RES_INVALID_PARMS 0x05 /* Invalid parameters */
  52 #define SD_RES_SYSTEM_ERROR  0x06 /* System error */
  53 #define SD_RES_VDI_LOCKED    0x07 /* Vdi is locked */
  54 #define SD_RES_NO_VDI        0x08 /* No vdi found */
  55 #define SD_RES_NO_BASE_VDI   0x09 /* No base vdi found */
  56 #define SD_RES_VDI_READ      0x0A /* Cannot read requested vdi */
  57 #define SD_RES_VDI_WRITE     0x0B /* Cannot write requested vdi */
  58 #define SD_RES_BASE_VDI_READ 0x0C /* Cannot read base vdi */
  59 #define SD_RES_BASE_VDI_WRITE   0x0D /* Cannot write base vdi */
  60 #define SD_RES_NO_TAG        0x0E /* Requested tag is not found */
  61 #define SD_RES_STARTUP       0x0F /* Sheepdog is on starting up */
  62 #define SD_RES_VDI_NOT_LOCKED   0x10 /* Vdi is not locked */
  63 #define SD_RES_SHUTDOWN      0x11 /* Sheepdog is shutting down */
  64 #define SD_RES_NO_MEM        0x12 /* Cannot allocate memory */
  65 #define SD_RES_FULL_VDI      0x13 /* we already have the maximum vdis */
  66 #define SD_RES_VER_MISMATCH  0x14 /* Protocol version mismatch */
  67 #define SD_RES_NO_SPACE      0x15 /* Server has no room for new objects */
  68 #define SD_RES_WAIT_FOR_FORMAT  0x16 /* Waiting for a format operation */
  69 #define SD_RES_WAIT_FOR_JOIN    0x17 /* Waiting for other nodes joining */
  70 #define SD_RES_JOIN_FAILED   0x18 /* Target node had failed to join sheepdog */
  71 #define SD_RES_HALT          0x19 /* Sheepdog is stopped serving IO request */
  72 #define SD_RES_READONLY      0x1A /* Object is read-only */
  73
  74 /*
  75  * Object ID rules
  76  *
  77  *  0 - 19 (20 bits): data object space
  78  * 20 - 31 (12 bits): reserved data object space
  79  * 32 - 55 (24 bits): vdi object space
  80  * 56 - 59 ( 4 bits): reserved vdi object space
  81  * 60 - 63 ( 4 bits): object type identifier space
  82  */
  83
  84 #define VDI_SPACE_SHIFT   32
  85 #define VDI_BIT (UINT64_C(1) << 63)
  86 #define VMSTATE_BIT (UINT64_C(1) << 62)
  87 #define MAX_DATA_OBJS (UINT64_C(1) << 20)
  88 #define MAX_CHILDREN 1024
  89 #define SD_MAX_VDI_LEN 256
  90 #define SD_MAX_VDI_TAG_LEN 256
  91 #define SD_NR_VDIS   (1U << 24)
  92 #define SD_DATA_OBJ_SIZE (UINT64_C(1) << 22)
  93 #define SD_MAX_VDI_SIZE (SD_DATA_OBJ_SIZE * MAX_DATA_OBJS)
  94
  95 #define SD_INODE_SIZE (sizeof(SheepdogInode))
  96 #define CURRENT_VDI_ID 0
  97
  98 typedef struct SheepdogReq {
  99     uint8_t proto_ver;
 100     uint8_t opcode;
 101     uint16_t flags;
 102     uint32_t epoch;
 103     uint32_t id;
 104     uint32_t data_length;
 105     uint32_t opcode_specific[8];
 106 } SheepdogReq;
 107
 108 typedef struct SheepdogRsp {
 109     uint8_t proto_ver;
 110     uint8_t opcode;
 111     uint16_t flags;
 112     uint32_t epoch;
 113     uint32_t id;
 114     uint32_t data_length;
 115     uint32_t result;
 116     uint32_t opcode_specific[7];
 117 } SheepdogRsp;
 118
 119 typedef struct SheepdogObjReq {
 120     uint8_t proto_ver;
 121     uint8_t opcode;
 122     uint16_t flags;
 123     uint32_t epoch;
 124     uint32_t id;
 125     uint32_t data_length;
 126     uint64_t oid;
 127     uint64_t cow_oid;
 128     uint8_t copies;
 129     uint8_t copy_policy;
 130     uint8_t reserved[6];
 131     uint64_t offset;
 132 } SheepdogObjReq;
 133
 134 typedef struct SheepdogObjRsp {
 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     uint32_t result;
 142     uint8_t copies;
 143     uint8_t copy_policy;
 144     uint8_t reserved[2];
 145     uint32_t pad[6];
 146 } SheepdogObjRsp;
 147
 148 typedef struct SheepdogVdiReq {
 149     uint8_t proto_ver;
 150     uint8_t opcode;
 151     uint16_t flags;
 152     uint32_t epoch;
 153     uint32_t id;
 154     uint32_t data_length;
 155     uint64_t vdi_size;
 156     uint32_t vdi_id;
 157     uint8_t copies;
 158     uint8_t copy_policy;
 159     uint8_t reserved[2];
 160     uint32_t snapid;
 161     uint32_t pad[3];
 162 } SheepdogVdiReq;
 163
 164 typedef struct SheepdogVdiRsp {
 165     uint8_t proto_ver;
 166     uint8_t opcode;
 167     uint16_t flags;
 168     uint32_t epoch;
 169     uint32_t id;
 170     uint32_t data_length;
 171     uint32_t result;
 172     uint32_t rsvd;
 173     uint32_t vdi_id;
 174     uint32_t pad[5];
 175 } SheepdogVdiRsp;
 176
 177 typedef struct SheepdogInode {
 178     char name[SD_MAX_VDI_LEN];
 179     char tag[SD_MAX_VDI_TAG_LEN];
 180     uint64_t ctime;
 181     uint64_t snap_ctime;
 182     uint64_t vm_clock_nsec;
 183     uint64_t vdi_size;
 184     uint64_t vm_state_size;
 185     uint16_t copy_policy;
 186     uint8_t nr_copies;
 187     uint8_t block_size_shift;
 188     uint32_t snap_id;
 189     uint32_t vdi_id;
 190     uint32_t parent_vdi_id;
 191     uint32_t child_vdi_id[MAX_CHILDREN];
 192     uint32_t data_vdi_id[MAX_DATA_OBJS];
 193 } SheepdogInode;
 194
 195 /*
 196  * 64 bit FNV-1a non-zero initial basis
 197  */
 198 #define FNV1A_64_INIT ((uint64_t)0xcbf29ce484222325ULL)
 199
 200 /*
 201  * 64 bit Fowler/Noll/Vo FNV-1a hash code
 202  */
 203 static inline uint64_t fnv_64a_buf(void *buf, size_t len, uint64_t hval)
 204 {
 205     unsigned char *bp = buf;
 206     unsigned char *be = bp + len;
 207     while (bp < be) {
 208         hval ^= (uint64_t) *bp++;
 209         hval += (hval << 1) + (hval << 4) + (hval << 5) +
 210             (hval << 7) + (hval << 8) + (hval << 40);
 211     }
 212     return hval;
 213 }
 214
 215 static inline bool is_data_obj_writable(SheepdogInode *inode, unsigned int idx)
 216 {
 217     return inode->vdi_id == inode->data_vdi_id[idx];
 218 }
 219
 220 static inline bool is_data_obj(uint64_t oid)
 221 {
 222     return !(VDI_BIT & oid);
 223 }
 224
 225 static inline uint64_t data_oid_to_idx(uint64_t oid)
 226 {
 227     return oid & (MAX_DATA_OBJS - 1);
 228 }
 229
 230 static inline uint64_t vid_to_vdi_oid(uint32_t vid)
 231 {
 232     return VDI_BIT | ((uint64_t)vid << VDI_SPACE_SHIFT);
 233 }
 234
 235 static inline uint64_t vid_to_vmstate_oid(uint32_t vid, uint32_t idx)
 236 {
 237     return VMSTATE_BIT | ((uint64_t)vid << VDI_SPACE_SHIFT) | idx;
 238 }
 239
 240 static inline uint64_t vid_to_data_oid(uint32_t vid, uint32_t idx)
 241 {
 242     return ((uint64_t)vid << VDI_SPACE_SHIFT) | idx;
 243 }
 244
 245 static inline bool is_snapshot(struct SheepdogInode *inode)
 246 {
 247     return !!inode->snap_ctime;
 248 }
 249
 250 #undef DPRINTF
 251 #ifdef DEBUG_SDOG
 252 #define DPRINTF(fmt, args...)                                       \
 253     do {                                                            \
 254         fprintf(stdout, "%s %d: " fmt, __func__, __LINE__, ##args); \
 255     } while (0)
 256 #else
 257 #define DPRINTF(fmt, args...)
 258 #endif
 259
 260 typedef struct SheepdogAIOCB SheepdogAIOCB;
 261
 262 typedef struct AIOReq {
 263     SheepdogAIOCB *aiocb;
 264     unsigned int iov_offset;
 265
 266     uint64_t oid;
 267     uint64_t base_oid;
 268     uint64_t offset;
 269     unsigned int data_len;
 270     uint8_t flags;
 271     uint32_t id;
 272
 273     QLIST_ENTRY(AIOReq) aio_siblings;
 274 } AIOReq;
 275
 276 enum AIOCBState {
 277     AIOCB_WRITE_UDATA,
 278     AIOCB_READ_UDATA,
 279     AIOCB_FLUSH_CACHE,
 280     AIOCB_DISCARD_OBJ,
 281 };
 282
 283 struct SheepdogAIOCB {
 284     BlockDriverAIOCB common;
 285
 286     QEMUIOVector *qiov;
 287
 288     int64_t sector_num;
 289     int nb_sectors;
 290
 291     int ret;
 292     enum AIOCBState aiocb_type;
 293
 294     Coroutine *coroutine;
 295     void (*aio_done_func)(SheepdogAIOCB *);
 296
 297     bool canceled;
 298     int nr_pending;
 299 };
 300
 301 typedef struct BDRVSheepdogState {
 302     SheepdogInode inode;
 303
 304     uint32_t min_dirty_data_idx;
 305     uint32_t max_dirty_data_idx;
 306
 307     char name[SD_MAX_VDI_LEN];
 308     bool is_snapshot;
 309     uint32_t cache_flags;
 310     bool discard_supported;
 311
 312     char *host_spec;
 313     bool is_unix;
 314     int fd;
 315
 316     CoMutex lock;
 317     Coroutine *co_send;
 318     Coroutine *co_recv;
 319
 320     uint32_t aioreq_seq_num;
 321     QLIST_HEAD(inflight_aio_head, AIOReq) inflight_aio_head;
 322     QLIST_HEAD(pending_aio_head, AIOReq) pending_aio_head;
 323 } BDRVSheepdogState;
 324
 325 static const char * sd_strerror(int err)
 326 {
 327     int i;
 328
 329     static const struct {
 330         int err;
 331         const char *desc;
 332     } errors[] = {
 333         {SD_RES_SUCCESS, "Success"},
 334         {SD_RES_UNKNOWN, "Unknown error"},
 335         {SD_RES_NO_OBJ, "No object found"},
 336         {SD_RES_EIO, "I/O error"},
 337         {SD_RES_VDI_EXIST, "VDI exists already"},
 338         {SD_RES_INVALID_PARMS, "Invalid parameters"},
 339         {SD_RES_SYSTEM_ERROR, "System error"},
 340         {SD_RES_VDI_LOCKED, "VDI is already locked"},
 341         {SD_RES_NO_VDI, "No vdi found"},
 342         {SD_RES_NO_BASE_VDI, "No base VDI found"},
 343         {SD_RES_VDI_READ, "Failed read the requested VDI"},
 344         {SD_RES_VDI_WRITE, "Failed to write the requested VDI"},
 345         {SD_RES_BASE_VDI_READ, "Failed to read the base VDI"},
 346         {SD_RES_BASE_VDI_WRITE, "Failed to write the base VDI"},
 347         {SD_RES_NO_TAG, "Failed to find the requested tag"},
 348         {SD_RES_STARTUP, "The system is still booting"},
 349         {SD_RES_VDI_NOT_LOCKED, "VDI isn't locked"},
 350         {SD_RES_SHUTDOWN, "The system is shutting down"},
 351         {SD_RES_NO_MEM, "Out of memory on the server"},
 352         {SD_RES_FULL_VDI, "We already have the maximum vdis"},
 353         {SD_RES_VER_MISMATCH, "Protocol version mismatch"},
 354         {SD_RES_NO_SPACE, "Server has no space for new objects"},
 355         {SD_RES_WAIT_FOR_FORMAT, "Sheepdog is waiting for a format operation"},
 356         {SD_RES_WAIT_FOR_JOIN, "Sheepdog is waiting for other nodes joining"},
 357         {SD_RES_JOIN_FAILED, "Target node had failed to join sheepdog"},
 358         {SD_RES_HALT, "Sheepdog is stopped serving IO request"},
 359         {SD_RES_READONLY, "Object is read-only"},
 360     };
 361
 362     for (i = 0; i < ARRAY_SIZE(errors); ++i) {
 363         if (errors[i].err == err) {
 364             return errors[i].desc;
 365         }
 366     }
 367
 368     return "Invalid error code";
 369 }
 370
 371 /*
 372  * Sheepdog I/O handling:
 373  *
 374  * 1. In sd_co_rw_vector, we send the I/O requests to the server and
 375  *    link the requests to the inflight_list in the
 376  *    BDRVSheepdogState.  The function exits without waiting for
 377  *    receiving the response.
 378  *
 379  * 2. We receive the response in aio_read_response, the fd handler to
 380  *    the sheepdog connection.  If metadata update is needed, we send
 381  *    the write request to the vdi object in sd_write_done, the write
 382  *    completion function.  We switch back to sd_co_readv/writev after
 383  *    all the requests belonging to the AIOCB are finished.
 384  */
 385
 386 static inline AIOReq *alloc_aio_req(BDRVSheepdogState *s, SheepdogAIOCB *acb,
 387                                     uint64_t oid, unsigned int data_len,
 388                                     uint64_t offset, uint8_t flags,
 389                                     uint64_t base_oid, unsigned int iov_offset)
 390 {
 391     AIOReq *aio_req;
 392
 393     aio_req = g_malloc(sizeof(*aio_req));
 394     aio_req->aiocb = acb;
 395     aio_req->iov_offset = iov_offset;
 396     aio_req->oid = oid;
 397     aio_req->base_oid = base_oid;
 398     aio_req->offset = offset;
 399     aio_req->data_len = data_len;
 400     aio_req->flags = flags;
 401     aio_req->id = s->aioreq_seq_num++;
 402
 403     acb->nr_pending++;
 404     return aio_req;
 405 }
 406
 407 static inline void free_aio_req(BDRVSheepdogState *s, AIOReq *aio_req)
 408 {
 409     SheepdogAIOCB *acb = aio_req->aiocb;
 410
 411     QLIST_REMOVE(aio_req, aio_siblings);
 412     g_free(aio_req);
 413
 414     acb->nr_pending--;
 415 }
 416
 417 static void coroutine_fn sd_finish_aiocb(SheepdogAIOCB *acb)
 418 {
 419     if (!acb->canceled) {
 420         qemu_coroutine_enter(acb->coroutine, NULL);
 421     }
 422     qemu_aio_release(acb);
 423 }
 424
 425 static void sd_aio_cancel(BlockDriverAIOCB *blockacb)
 426 {
 427     SheepdogAIOCB *acb = (SheepdogAIOCB *)blockacb;
 428
 429     /*
 430      * Sheepdog cannot cancel the requests which are already sent to
 431      * the servers, so we just complete the request with -EIO here.
 432      */
 433     acb->ret = -EIO;
 434     qemu_coroutine_enter(acb->coroutine, NULL);
 435     acb->canceled = true;
 436 }
 437
 438 static const AIOCBInfo sd_aiocb_info = {
 439     .aiocb_size = sizeof(SheepdogAIOCB),
 440     .cancel = sd_aio_cancel,
 441 };
 442
 443 static SheepdogAIOCB *sd_aio_setup(BlockDriverState *bs, QEMUIOVector *qiov,
 444                                    int64_t sector_num, int nb_sectors)
 445 {
 446     SheepdogAIOCB *acb;
 447
 448     acb = qemu_aio_get(&sd_aiocb_info, bs, NULL, NULL);
 449
 450     acb->qiov = qiov;
 451
 452     acb->sector_num = sector_num;
 453     acb->nb_sectors = nb_sectors;
 454
 455     acb->aio_done_func = NULL;
 456     acb->canceled = false;
 457     acb->coroutine = qemu_coroutine_self();
 458     acb->ret = 0;
 459     acb->nr_pending = 0;
 460     return acb;
 461 }
 462
 463 static int connect_to_sdog(BDRVSheepdogState *s)
 464 {
 465     int fd;
 466     Error *err = NULL;
 467
 468     if (s->is_unix) {
 469         fd = unix_connect(s->host_spec, &err);
 470     } else {
 471         fd = inet_connect(s->host_spec, &err);
 472
 473         if (err == NULL) {
 474             int ret = socket_set_nodelay(fd);
 475             if (ret < 0) {
 476                 error_report("%s", strerror(errno));
 477             }
 478         }
 479     }
 480
 481     if (err != NULL) {
 482         qerror_report_err(err);
 483         error_free(err);
 484     } else {
 485         qemu_set_nonblock(fd);
 486     }
 487
 488     return fd;
 489 }
 490
 491 static coroutine_fn int send_co_req(int sockfd, SheepdogReq *hdr, void *data,
 492                                     unsigned int *wlen)
 493 {
 494     int ret;
 495
 496     ret = qemu_co_send(sockfd, hdr, sizeof(*hdr));
 497     if (ret != sizeof(*hdr)) {
 498         error_report("failed to send a req, %s", strerror(errno));
 499         return ret;
 500     }
 501
 502     ret = qemu_co_send(sockfd, data, *wlen);
 503     if (ret != *wlen) {
 504         error_report("failed to send a req, %s", strerror(errno));
 505     }
 506
 507     return ret;
 508 }
 509
 510 static void restart_co_req(void *opaque)
 511 {
 512     Coroutine *co = opaque;
 513
 514     qemu_coroutine_enter(co, NULL);
 515 }
 516
 517 typedef struct SheepdogReqCo {
 518     int sockfd;
 519     SheepdogReq *hdr;
 520     void *data;
 521     unsigned int *wlen;
 522     unsigned int *rlen;
 523     int ret;
 524     bool finished;
 525 } SheepdogReqCo;
 526
 527 static coroutine_fn void do_co_req(void *opaque)
 528 {
 529     int ret;
 530     Coroutine *co;
 531     SheepdogReqCo *srco = opaque;
 532     int sockfd = srco->sockfd;
 533     SheepdogReq *hdr = srco->hdr;
 534     void *data = srco->data;
 535     unsigned int *wlen = srco->wlen;
 536     unsigned int *rlen = srco->rlen;
 537
 538     co = qemu_coroutine_self();
 539     qemu_aio_set_fd_handler(sockfd, NULL, restart_co_req, co);
 540
 541     ret = send_co_req(sockfd, hdr, data, wlen);
 542     if (ret < 0) {
 543         goto out;
 544     }
 545
 546     qemu_aio_set_fd_handler(sockfd, restart_co_req, NULL, co);
 547
 548     ret = qemu_co_recv(sockfd, hdr, sizeof(*hdr));
 549     if (ret != sizeof(*hdr)) {
 550         error_report("failed to get a rsp, %s", strerror(errno));
 551         ret = -errno;
 552         goto out;
 553     }
 554
 555     if (*rlen > hdr->data_length) {
 556         *rlen = hdr->data_length;
 557     }
 558
 559     if (*rlen) {
 560         ret = qemu_co_recv(sockfd, data, *rlen);
 561         if (ret != *rlen) {
 562             error_report("failed to get the data, %s", strerror(errno));
 563             ret = -errno;
 564             goto out;
 565         }
 566     }
 567     ret = 0;
 568 out:
 569     /* there is at most one request for this sockfd, so it is safe to
 570      * set each handler to NULL. */
 571     qemu_aio_set_fd_handler(sockfd, NULL, NULL, NULL);
 572
 573     srco->ret = ret;
 574     srco->finished = true;
 575 }
 576
 577 static int do_req(int sockfd, SheepdogReq *hdr, void *data,
 578                   unsigned int *wlen, unsigned int *rlen)
 579 {
 580     Coroutine *co;
 581     SheepdogReqCo srco = {
 582         .sockfd = sockfd,
 583         .hdr = hdr,
 584         .data = data,
 585         .wlen = wlen,
 586         .rlen = rlen,
 587         .ret = 0,
 588         .finished = false,
 589     };
 590
 591     if (qemu_in_coroutine()) {
 592         do_co_req(&srco);
 593     } else {
 594         co = qemu_coroutine_create(do_co_req);
 595         qemu_coroutine_enter(co, &srco);
 596         while (!srco.finished) {
 597             qemu_aio_wait();
 598         }
 599     }
 600
 601     return srco.ret;
 602 }
 603
 604 static int coroutine_fn add_aio_request(BDRVSheepdogState *s, AIOReq *aio_req,
 605                            struct iovec *iov, int niov, bool create,
 606                            enum AIOCBState aiocb_type);
 607 static int coroutine_fn resend_aioreq(BDRVSheepdogState *s, AIOReq *aio_req);
 608
 609
 610 static AIOReq *find_pending_req(BDRVSheepdogState *s, uint64_t oid)
 611 {
 612     AIOReq *aio_req;
 613
 614     QLIST_FOREACH(aio_req, &s->pending_aio_head, aio_siblings) {
 615         if (aio_req->oid == oid) {
 616             return aio_req;
 617         }
 618     }
 619
 620     return NULL;
 621 }
 622
 623 /*
 624  * This function searchs pending requests to the object `oid', and
 625  * sends them.
 626  */
 627 static void coroutine_fn send_pending_req(BDRVSheepdogState *s, uint64_t oid)
 628 {
 629     AIOReq *aio_req;
 630     SheepdogAIOCB *acb;
 631     int ret;
 632
 633     while ((aio_req = find_pending_req(s, oid)) != NULL) {
 634         acb = aio_req->aiocb;
 635         /* move aio_req from pending list to inflight one */
 636         QLIST_REMOVE(aio_req, aio_siblings);
 637         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
 638         ret = add_aio_request(s, aio_req, acb->qiov->iov,
 639                               acb->qiov->niov, false, acb->aiocb_type);
 640         if (ret < 0) {
 641             error_report("add_aio_request is failed");
 642             free_aio_req(s, aio_req);
 643             if (!acb->nr_pending) {
 644                 sd_finish_aiocb(acb);
 645             }
 646         }
 647     }
 648 }
 649
 650 /*
 651  * Receive responses of the I/O requests.
 652  *
 653  * This function is registered as a fd handler, and called from the
 654  * main loop when s->fd is ready for reading responses.
 655  */
 656 static void coroutine_fn aio_read_response(void *opaque)
 657 {
 658     SheepdogObjRsp rsp;
 659     BDRVSheepdogState *s = opaque;
 660     int fd = s->fd;
 661     int ret;
 662     AIOReq *aio_req = NULL;
 663     SheepdogAIOCB *acb;
 664     uint64_t idx;
 665
 666     if (QLIST_EMPTY(&s->inflight_aio_head)) {
 667         goto out;
 668     }
 669
 670     /* read a header */
 671     ret = qemu_co_recv(fd, &rsp, sizeof(rsp));
 672     if (ret != sizeof(rsp)) {
 673         error_report("failed to get the header, %s", strerror(errno));
 674         goto out;
 675     }
 676
 677     /* find the right aio_req from the inflight aio list */
 678     QLIST_FOREACH(aio_req, &s->inflight_aio_head, aio_siblings) {
 679         if (aio_req->id == rsp.id) {
 680             break;
 681         }
 682     }
 683     if (!aio_req) {
 684         error_report("cannot find aio_req %x", rsp.id);
 685         goto out;
 686     }
 687
 688     acb = aio_req->aiocb;
 689
 690     switch (acb->aiocb_type) {
 691     case AIOCB_WRITE_UDATA:
 692         /* this coroutine context is no longer suitable for co_recv
 693          * because we may send data to update vdi objects */
 694         s->co_recv = NULL;
 695         if (!is_data_obj(aio_req->oid)) {
 696             break;
 697         }
 698         idx = data_oid_to_idx(aio_req->oid);
 699
 700         if (s->inode.data_vdi_id[idx] != s->inode.vdi_id) {
 701             /*
 702              * If the object is newly created one, we need to update
 703              * the vdi object (metadata object).  min_dirty_data_idx
 704              * and max_dirty_data_idx are changed to include updated
 705              * index between them.
 706              */
 707             if (rsp.result == SD_RES_SUCCESS) {
 708                 s->inode.data_vdi_id[idx] = s->inode.vdi_id;
 709                 s->max_dirty_data_idx = MAX(idx, s->max_dirty_data_idx);
 710                 s->min_dirty_data_idx = MIN(idx, s->min_dirty_data_idx);
 711             }
 712             /*
 713              * Some requests may be blocked because simultaneous
 714              * create requests are not allowed, so we search the
 715              * pending requests here.
 716              */
 717             send_pending_req(s, aio_req->oid);
 718         }
 719         break;
 720     case AIOCB_READ_UDATA:
 721         ret = qemu_co_recvv(fd, acb->qiov->iov, acb->qiov->niov,
 722                             aio_req->iov_offset, rsp.data_length);
 723         if (ret != rsp.data_length) {
 724             error_report("failed to get the data, %s", strerror(errno));
 725             goto out;
 726         }
 727         break;
 728     case AIOCB_FLUSH_CACHE:
 729         if (rsp.result == SD_RES_INVALID_PARMS) {
 730             DPRINTF("disable cache since the server doesn't support it\n");
 731             s->cache_flags = SD_FLAG_CMD_DIRECT;
 732             rsp.result = SD_RES_SUCCESS;
 733         }
 734         break;
 735     case AIOCB_DISCARD_OBJ:
 736         switch (rsp.result) {
 737         case SD_RES_INVALID_PARMS:
 738             error_report("sheep(%s) doesn't support discard command",
 739                          s->host_spec);
 740             rsp.result = SD_RES_SUCCESS;
 741             s->discard_supported = false;
 742             break;
 743         case SD_RES_SUCCESS:
 744             idx = data_oid_to_idx(aio_req->oid);
 745             s->inode.data_vdi_id[idx] = 0;
 746             break;
 747         default:
 748             break;
 749         }
 750     }
 751
 752     switch (rsp.result) {
 753     case SD_RES_SUCCESS:
 754         break;
 755     case SD_RES_READONLY:
 756         ret = resend_aioreq(s, aio_req);
 757         if (ret == SD_RES_SUCCESS) {
 758             goto out;
 759         }
 760         /* fall through */
 761     default:
 762         acb->ret = -EIO;
 763         error_report("%s", sd_strerror(rsp.result));
 764         break;
 765     }
 766
 767     free_aio_req(s, aio_req);
 768     if (!acb->nr_pending) {
 769         /*
 770          * We've finished all requests which belong to the AIOCB, so
 771          * we can switch back to sd_co_readv/writev now.
 772          */
 773         acb->aio_done_func(acb);
 774     }
 775 out:
 776     s->co_recv = NULL;
 777 }
 778
 779 static void co_read_response(void *opaque)
 780 {
 781     BDRVSheepdogState *s = opaque;
 782
 783     if (!s->co_recv) {
 784         s->co_recv = qemu_coroutine_create(aio_read_response);
 785     }
 786
 787     qemu_coroutine_enter(s->co_recv, opaque);
 788 }
 789
 790 static void co_write_request(void *opaque)
 791 {
 792     BDRVSheepdogState *s = opaque;
 793
 794     qemu_coroutine_enter(s->co_send, NULL);
 795 }
 796
 797 /*
 798  * Return a socket discriptor to read/write objects.
 799  *
 800  * We cannot use this discriptor for other operations because
 801  * the block driver may be on waiting response from the server.
 802  */
 803 static int get_sheep_fd(BDRVSheepdogState *s)
 804 {
 805     int fd;
 806
 807     fd = connect_to_sdog(s);
 808     if (fd < 0) {
 809         return fd;
 810     }
 811
 812     qemu_aio_set_fd_handler(fd, co_read_response, NULL, s);
 813     return fd;
 814 }
 815
 816 static int sd_parse_uri(BDRVSheepdogState *s, const char *filename,
 817                         char *vdi, uint32_t *snapid, char *tag)
 818 {
 819     URI *uri;
 820     QueryParams *qp = NULL;
 821     int ret = 0;
 822
 823     uri = uri_parse(filename);
 824     if (!uri) {
 825         return -EINVAL;
 826     }
 827
 828     /* transport */
 829     if (!strcmp(uri->scheme, "sheepdog")) {
 830         s->is_unix = false;
 831     } else if (!strcmp(uri->scheme, "sheepdog+tcp")) {
 832         s->is_unix = false;
 833     } else if (!strcmp(uri->scheme, "sheepdog+unix")) {
 834         s->is_unix = true;
 835     } else {
 836         ret = -EINVAL;
 837         goto out;
 838     }
 839
 840     if (uri->path == NULL || !strcmp(uri->path, "/")) {
 841         ret = -EINVAL;
 842         goto out;
 843     }
 844     pstrcpy(vdi, SD_MAX_VDI_LEN, uri->path + 1);
 845
 846     qp = query_params_parse(uri->query);
 847     if (qp->n > 1 || (s->is_unix && !qp->n) || (!s->is_unix && qp->n)) {
 848         ret = -EINVAL;
 849         goto out;
 850     }
 851
 852     if (s->is_unix) {
 853         /* sheepdog+unix:///vdiname?socket=path */
 854         if (uri->server || uri->port || strcmp(qp->p[0].name, "socket")) {
 855             ret = -EINVAL;
 856             goto out;
 857         }
 858         s->host_spec = g_strdup(qp->p[0].value);
 859     } else {
 860         /* sheepdog[+tcp]://[host:port]/vdiname */
 861         s->host_spec = g_strdup_printf("%s:%d", uri->server ?: SD_DEFAULT_ADDR,
 862                                        uri->port ?: SD_DEFAULT_PORT);
 863     }
 864
 865     /* snapshot tag */
 866     if (uri->fragment) {
 867         *snapid = strtoul(uri->fragment, NULL, 10);
 868         if (*snapid == 0) {
 869             pstrcpy(tag, SD_MAX_VDI_TAG_LEN, uri->fragment);
 870         }
 871     } else {
 872         *snapid = CURRENT_VDI_ID; /* search current vdi */
 873     }
 874
 875 out:
 876     if (qp) {
 877         query_params_free(qp);
 878     }
 879     uri_free(uri);
 880     return ret;
 881 }
 882
 883 /*
 884  * Parse a filename (old syntax)
 885  *
 886  * filename must be one of the following formats:
 887  *   1. [vdiname]
 888  *   2. [vdiname]:[snapid]
 889  *   3. [vdiname]:[tag]
 890  *   4. [hostname]:[port]:[vdiname]
 891  *   5. [hostname]:[port]:[vdiname]:[snapid]
 892  *   6. [hostname]:[port]:[vdiname]:[tag]
 893  *
 894  * You can boot from the snapshot images by specifying `snapid` or
 895  * `tag'.
 896  *
 897  * You can run VMs outside the Sheepdog cluster by specifying
 898  * `hostname' and `port' (experimental).
 899  */
 900 static int parse_vdiname(BDRVSheepdogState *s, const char *filename,
 901                          char *vdi, uint32_t *snapid, char *tag)
 902 {
 903     char *p, *q, *uri;
 904     const char *host_spec, *vdi_spec;
 905     int nr_sep, ret;
 906
 907     strstart(filename, "sheepdog:", (const char **)&filename);
 908     p = q = g_strdup(filename);
 909
 910     /* count the number of separators */
 911     nr_sep = 0;
 912     while (*p) {
 913         if (*p == ':') {
 914             nr_sep++;
 915         }
 916         p++;
 917     }
 918     p = q;
 919
 920     /* use the first two tokens as host_spec. */
 921     if (nr_sep >= 2) {
 922         host_spec = p;
 923         p = strchr(p, ':');
 924         p++;
 925         p = strchr(p, ':');
 926         *p++ = '\0';
 927     } else {
 928         host_spec = "";
 929     }
 930
 931     vdi_spec = p;
 932
 933     p = strchr(vdi_spec, ':');
 934     if (p) {
 935         *p++ = '#';
 936     }
 937
 938     uri = g_strdup_printf("sheepdog://%s/%s", host_spec, vdi_spec);
 939
 940     ret = sd_parse_uri(s, uri, vdi, snapid, tag);
 941
 942     g_free(q);
 943     g_free(uri);
 944
 945     return ret;
 946 }
 947
 948 static int find_vdi_name(BDRVSheepdogState *s, const char *filename,
 949                          uint32_t snapid, const char *tag, uint32_t *vid,
 950                          bool lock)
 951 {
 952     int ret, fd;
 953     SheepdogVdiReq hdr;
 954     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
 955     unsigned int wlen, rlen = 0;
 956     char buf[SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN];
 957
 958     fd = connect_to_sdog(s);
 959     if (fd < 0) {
 960         return fd;
 961     }
 962
 963     /* This pair of strncpy calls ensures that the buffer is zero-filled,
 964      * which is desirable since we'll soon be sending those bytes, and
 965      * don't want the send_req to read uninitialized data.
 966      */
 967     strncpy(buf, filename, SD_MAX_VDI_LEN);
 968     strncpy(buf + SD_MAX_VDI_LEN, tag, SD_MAX_VDI_TAG_LEN);
 969
 970     memset(&hdr, 0, sizeof(hdr));
 971     if (lock) {
 972         hdr.opcode = SD_OP_LOCK_VDI;
 973     } else {
 974         hdr.opcode = SD_OP_GET_VDI_INFO;
 975     }
 976     wlen = SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN;
 977     hdr.proto_ver = SD_PROTO_VER;
 978     hdr.data_length = wlen;
 979     hdr.snapid = snapid;
 980     hdr.flags = SD_FLAG_CMD_WRITE;
 981
 982     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
 983     if (ret) {
 984         goto out;
 985     }
 986
 987     if (rsp->result != SD_RES_SUCCESS) {
 988         error_report("cannot get vdi info, %s, %s %d %s",
 989                      sd_strerror(rsp->result), filename, snapid, tag);
 990         if (rsp->result == SD_RES_NO_VDI) {
 991             ret = -ENOENT;
 992         } else {
 993             ret = -EIO;
 994         }
 995         goto out;
 996     }
 997     *vid = rsp->vdi_id;
 998
 999     ret = 0;
1000 out:
1001     closesocket(fd);
1002     return ret;
1003 }
1004
1005 static int coroutine_fn add_aio_request(BDRVSheepdogState *s, AIOReq *aio_req,
1006                            struct iovec *iov, int niov, bool create,
1007                            enum AIOCBState aiocb_type)
1008 {
1009     int nr_copies = s->inode.nr_copies;
1010     SheepdogObjReq hdr;
1011     unsigned int wlen = 0;
1012     int ret;
1013     uint64_t oid = aio_req->oid;
1014     unsigned int datalen = aio_req->data_len;
1015     uint64_t offset = aio_req->offset;
1016     uint8_t flags = aio_req->flags;
1017     uint64_t old_oid = aio_req->base_oid;
1018
1019     if (!nr_copies) {
1020         error_report("bug");
1021     }
1022
1023     memset(&hdr, 0, sizeof(hdr));
1024
1025     switch (aiocb_type) {
1026     case AIOCB_FLUSH_CACHE:
1027         hdr.opcode = SD_OP_FLUSH_VDI;
1028         break;
1029     case AIOCB_READ_UDATA:
1030         hdr.opcode = SD_OP_READ_OBJ;
1031         hdr.flags = flags;
1032         break;
1033     case AIOCB_WRITE_UDATA:
1034         if (create) {
1035             hdr.opcode = SD_OP_CREATE_AND_WRITE_OBJ;
1036         } else {
1037             hdr.opcode = SD_OP_WRITE_OBJ;
1038         }
1039         wlen = datalen;
1040         hdr.flags = SD_FLAG_CMD_WRITE | flags;
1041         break;
1042     case AIOCB_DISCARD_OBJ:
1043         hdr.opcode = SD_OP_DISCARD_OBJ;
1044         break;
1045     }
1046
1047     if (s->cache_flags) {
1048         hdr.flags |= s->cache_flags;
1049     }
1050
1051     hdr.oid = oid;
1052     hdr.cow_oid = old_oid;
1053     hdr.copies = s->inode.nr_copies;
1054
1055     hdr.data_length = datalen;
1056     hdr.offset = offset;
1057
1058     hdr.id = aio_req->id;
1059
1060     qemu_co_mutex_lock(&s->lock);
1061     s->co_send = qemu_coroutine_self();
1062     qemu_aio_set_fd_handler(s->fd, co_read_response, co_write_request, s);
1063     socket_set_cork(s->fd, 1);
1064
1065     /* send a header */
1066     ret = qemu_co_send(s->fd, &hdr, sizeof(hdr));
1067     if (ret != sizeof(hdr)) {
1068         qemu_co_mutex_unlock(&s->lock);
1069         error_report("failed to send a req, %s", strerror(errno));
1070         return -errno;
1071     }
1072
1073     if (wlen) {
1074         ret = qemu_co_sendv(s->fd, iov, niov, aio_req->iov_offset, wlen);
1075         if (ret != wlen) {
1076             qemu_co_mutex_unlock(&s->lock);
1077             error_report("failed to send a data, %s", strerror(errno));
1078             return -errno;
1079         }
1080     }
1081
1082     socket_set_cork(s->fd, 0);
1083     qemu_aio_set_fd_handler(s->fd, co_read_response, NULL, s);
1084     qemu_co_mutex_unlock(&s->lock);
1085
1086     return 0;
1087 }
1088
1089 static int read_write_object(int fd, char *buf, uint64_t oid, uint8_t copies,
1090                              unsigned int datalen, uint64_t offset,
1091                              bool write, bool create, uint32_t cache_flags)
1092 {
1093     SheepdogObjReq hdr;
1094     SheepdogObjRsp *rsp = (SheepdogObjRsp *)&hdr;
1095     unsigned int wlen, rlen;
1096     int ret;
1097
1098     memset(&hdr, 0, sizeof(hdr));
1099
1100     if (write) {
1101         wlen = datalen;
1102         rlen = 0;
1103         hdr.flags = SD_FLAG_CMD_WRITE;
1104         if (create) {
1105             hdr.opcode = SD_OP_CREATE_AND_WRITE_OBJ;
1106         } else {
1107             hdr.opcode = SD_OP_WRITE_OBJ;
1108         }
1109     } else {
1110         wlen = 0;
1111         rlen = datalen;
1112         hdr.opcode = SD_OP_READ_OBJ;
1113     }
1114
1115     hdr.flags |= cache_flags;
1116
1117     hdr.oid = oid;
1118     hdr.data_length = datalen;
1119     hdr.offset = offset;
1120     hdr.copies = copies;
1121
1122     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1123     if (ret) {
1124         error_report("failed to send a request to the sheep");
1125         return ret;
1126     }
1127
1128     switch (rsp->result) {
1129     case SD_RES_SUCCESS:
1130         return 0;
1131     default:
1132         error_report("%s", sd_strerror(rsp->result));
1133         return -EIO;
1134     }
1135 }
1136
1137 static int read_object(int fd, char *buf, uint64_t oid, uint8_t copies,
1138                        unsigned int datalen, uint64_t offset,
1139                        uint32_t cache_flags)
1140 {
1141     return read_write_object(fd, buf, oid, copies, datalen, offset, false,
1142                              false, cache_flags);
1143 }
1144
1145 static int write_object(int fd, char *buf, uint64_t oid, uint8_t copies,
1146                         unsigned int datalen, uint64_t offset, bool create,
1147                         uint32_t cache_flags)
1148 {
1149     return read_write_object(fd, buf, oid, copies, datalen, offset, true,
1150                              create, cache_flags);
1151 }
1152
1153 /* update inode with the latest state */
1154 static int reload_inode(BDRVSheepdogState *s, uint32_t snapid, const char *tag)
1155 {
1156     SheepdogInode *inode;
1157     int ret = 0, fd;
1158     uint32_t vid = 0;
1159
1160     fd = connect_to_sdog(s);
1161     if (fd < 0) {
1162         return -EIO;
1163     }
1164
1165     inode = g_malloc(sizeof(s->inode));
1166
1167     ret = find_vdi_name(s, s->name, snapid, tag, &vid, false);
1168     if (ret) {
1169         goto out;
1170     }
1171
1172     ret = read_object(fd, (char *)inode, vid_to_vdi_oid(vid),
1173                       s->inode.nr_copies, sizeof(*inode), 0, s->cache_flags);
1174     if (ret < 0) {
1175         goto out;
1176     }
1177
1178     if (inode->vdi_id != s->inode.vdi_id) {
1179         memcpy(&s->inode, inode, sizeof(s->inode));
1180     }
1181
1182 out:
1183     g_free(inode);
1184     closesocket(fd);
1185
1186     return ret;
1187 }
1188
1189 static int coroutine_fn resend_aioreq(BDRVSheepdogState *s, AIOReq *aio_req)
1190 {
1191     SheepdogAIOCB *acb = aio_req->aiocb;
1192     bool create = false;
1193     int ret;
1194
1195     ret = reload_inode(s, 0, "");
1196     if (ret < 0) {
1197         return ret;
1198     }
1199
1200     if (is_data_obj(aio_req->oid)) {
1201         aio_req->oid = vid_to_data_oid(s->inode.vdi_id,
1202                                        data_oid_to_idx(aio_req->oid));
1203     } else {
1204         aio_req->oid = vid_to_vdi_oid(s->inode.vdi_id);
1205     }
1206
1207     /* check whether this request becomes a CoW one */
1208     if (acb->aiocb_type == AIOCB_WRITE_UDATA && is_data_obj(aio_req->oid)) {
1209         int idx = data_oid_to_idx(aio_req->oid);
1210         AIOReq *areq;
1211
1212         if (s->inode.data_vdi_id[idx] == 0) {
1213             create = true;
1214             goto out;
1215         }
1216         if (is_data_obj_writable(&s->inode, idx)) {
1217             goto out;
1218         }
1219
1220         /* link to the pending list if there is another CoW request to
1221          * the same object */
1222         QLIST_FOREACH(areq, &s->inflight_aio_head, aio_siblings) {
1223             if (areq != aio_req && areq->oid == aio_req->oid) {
1224                 DPRINTF("simultaneous CoW to %" PRIx64 "\n", aio_req->oid);
1225                 QLIST_REMOVE(aio_req, aio_siblings);
1226                 QLIST_INSERT_HEAD(&s->pending_aio_head, aio_req, aio_siblings);
1227                 return SD_RES_SUCCESS;
1228             }
1229         }
1230
1231         aio_req->base_oid = vid_to_data_oid(s->inode.data_vdi_id[idx], idx);
1232         aio_req->flags |= SD_FLAG_CMD_COW;
1233         create = true;
1234     }
1235 out:
1236     if (is_data_obj(aio_req->oid)) {
1237         return add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
1238                                create, acb->aiocb_type);
1239     } else {
1240         struct iovec iov;
1241         iov.iov_base = &s->inode;
1242         iov.iov_len = sizeof(s->inode);
1243         return add_aio_request(s, aio_req, &iov, 1, false, AIOCB_WRITE_UDATA);
1244     }
1245 }
1246
1247 /* TODO Convert to fine grained options */
1248 static QemuOptsList runtime_opts = {
1249     .name = "sheepdog",
1250     .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
1251     .desc = {
1252         {
1253             .name = "filename",
1254             .type = QEMU_OPT_STRING,
1255             .help = "URL to the sheepdog image",
1256         },
1257         { /* end of list */ }
1258     },
1259 };
1260
1261 static int sd_open(BlockDriverState *bs, QDict *options, int flags,
1262                    Error **errp)
1263 {
1264     int ret, fd;
1265     uint32_t vid = 0;
1266     BDRVSheepdogState *s = bs->opaque;
1267     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1268     uint32_t snapid;
1269     char *buf = NULL;
1270     QemuOpts *opts;
1271     Error *local_err = NULL;
1272     const char *filename;
1273
1274     opts = qemu_opts_create_nofail(&runtime_opts);
1275     qemu_opts_absorb_qdict(opts, options, &local_err);
1276     if (error_is_set(&local_err)) {
1277         qerror_report_err(local_err);
1278         error_free(local_err);
1279         ret = -EINVAL;
1280         goto out;
1281     }
1282
1283     filename = qemu_opt_get(opts, "filename");
1284
1285     QLIST_INIT(&s->inflight_aio_head);
1286     QLIST_INIT(&s->pending_aio_head);
1287     s->fd = -1;
1288
1289     memset(vdi, 0, sizeof(vdi));
1290     memset(tag, 0, sizeof(tag));
1291
1292     if (strstr(filename, "://")) {
1293         ret = sd_parse_uri(s, filename, vdi, &snapid, tag);
1294     } else {
1295         ret = parse_vdiname(s, filename, vdi, &snapid, tag);
1296     }
1297     if (ret < 0) {
1298         goto out;
1299     }
1300     s->fd = get_sheep_fd(s);
1301     if (s->fd < 0) {
1302         ret = s->fd;
1303         goto out;
1304     }
1305
1306     ret = find_vdi_name(s, vdi, snapid, tag, &vid, true);
1307     if (ret) {
1308         goto out;
1309     }
1310
1311     /*
1312      * QEMU block layer emulates writethrough cache as 'writeback + flush', so
1313      * we always set SD_FLAG_CMD_CACHE (writeback cache) as default.
1314      */
1315     s->cache_flags = SD_FLAG_CMD_CACHE;
1316     if (flags & BDRV_O_NOCACHE) {
1317         s->cache_flags = SD_FLAG_CMD_DIRECT;
1318     }
1319     s->discard_supported = true;
1320
1321     if (snapid || tag[0] != '\0') {
1322         DPRINTF("%" PRIx32 " snapshot inode was open.\n", vid);
1323         s->is_snapshot = true;
1324     }
1325
1326     fd = connect_to_sdog(s);
1327     if (fd < 0) {
1328         ret = fd;
1329         goto out;
1330     }
1331
1332     buf = g_malloc(SD_INODE_SIZE);
1333     ret = read_object(fd, buf, vid_to_vdi_oid(vid), 0, SD_INODE_SIZE, 0,
1334                       s->cache_flags);
1335
1336     closesocket(fd);
1337
1338     if (ret) {
1339         goto out;
1340     }
1341
1342     memcpy(&s->inode, buf, sizeof(s->inode));
1343     s->min_dirty_data_idx = UINT32_MAX;
1344     s->max_dirty_data_idx = 0;
1345
1346     bs->total_sectors = s->inode.vdi_size / BDRV_SECTOR_SIZE;
1347     pstrcpy(s->name, sizeof(s->name), vdi);
1348     qemu_co_mutex_init(&s->lock);
1349     qemu_opts_del(opts);
1350     g_free(buf);
1351     return 0;
1352 out:
1353     qemu_aio_set_fd_handler(s->fd, NULL, NULL, NULL);
1354     if (s->fd >= 0) {
1355         closesocket(s->fd);
1356     }
1357     qemu_opts_del(opts);
1358     g_free(buf);
1359     return ret;
1360 }
1361
1362 static int do_sd_create(BDRVSheepdogState *s, char *filename, int64_t vdi_size,
1363                         uint32_t base_vid, uint32_t *vdi_id, int snapshot,
1364                         uint8_t copy_policy)
1365 {
1366     SheepdogVdiReq hdr;
1367     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1368     int fd, ret;
1369     unsigned int wlen, rlen = 0;
1370     char buf[SD_MAX_VDI_LEN];
1371
1372     fd = connect_to_sdog(s);
1373     if (fd < 0) {
1374         return fd;
1375     }
1376
1377     /* FIXME: would it be better to fail (e.g., return -EIO) when filename
1378      * does not fit in buf?  For now, just truncate and avoid buffer overrun.
1379      */
1380     memset(buf, 0, sizeof(buf));
1381     pstrcpy(buf, sizeof(buf), filename);
1382
1383     memset(&hdr, 0, sizeof(hdr));
1384     hdr.opcode = SD_OP_NEW_VDI;
1385     hdr.vdi_id = base_vid;
1386
1387     wlen = SD_MAX_VDI_LEN;
1388
1389     hdr.flags = SD_FLAG_CMD_WRITE;
1390     hdr.snapid = snapshot;
1391
1392     hdr.data_length = wlen;
1393     hdr.vdi_size = vdi_size;
1394     hdr.copy_policy = copy_policy;
1395
1396     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1397
1398     closesocket(fd);
1399
1400     if (ret) {
1401         return ret;
1402     }
1403
1404     if (rsp->result != SD_RES_SUCCESS) {
1405         error_report("%s, %s", sd_strerror(rsp->result), filename);
1406         return -EIO;
1407     }
1408
1409     if (vdi_id) {
1410         *vdi_id = rsp->vdi_id;
1411     }
1412
1413     return 0;
1414 }
1415
1416 static int sd_prealloc(const char *filename)
1417 {
1418     BlockDriverState *bs = NULL;
1419     uint32_t idx, max_idx;
1420     int64_t vdi_size;
1421     void *buf = g_malloc0(SD_DATA_OBJ_SIZE);
1422     Error *local_err = NULL;
1423     int ret;
1424
1425     ret = bdrv_file_open(&bs, filename, NULL, BDRV_O_RDWR, &local_err);
1426     if (ret < 0) {
1427         qerror_report_err(local_err);
1428         error_free(local_err);
1429         goto out;
1430     }
1431
1432     vdi_size = bdrv_getlength(bs);
1433     if (vdi_size < 0) {
1434         ret = vdi_size;
1435         goto out;
1436     }
1437     max_idx = DIV_ROUND_UP(vdi_size, SD_DATA_OBJ_SIZE);
1438
1439     for (idx = 0; idx < max_idx; idx++) {
1440         /*
1441          * The created image can be a cloned image, so we need to read
1442          * a data from the source image.
1443          */
1444         ret = bdrv_pread(bs, idx * SD_DATA_OBJ_SIZE, buf, SD_DATA_OBJ_SIZE);
1445         if (ret < 0) {
1446             goto out;
1447         }
1448         ret = bdrv_pwrite(bs, idx * SD_DATA_OBJ_SIZE, buf, SD_DATA_OBJ_SIZE);
1449         if (ret < 0) {
1450             goto out;
1451         }
1452     }
1453 out:
1454     if (bs) {
1455         bdrv_unref(bs);
1456     }
1457     g_free(buf);
1458
1459     return ret;
1460 }
1461
1462 static int sd_create(const char *filename, QEMUOptionParameter *options,
1463                      Error **errp)
1464 {
1465     int ret = 0;
1466     uint32_t vid = 0, base_vid = 0;
1467     int64_t vdi_size = 0;
1468     char *backing_file = NULL;
1469     BDRVSheepdogState *s;
1470     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1471     uint32_t snapid;
1472     bool prealloc = false;
1473     Error *local_err = NULL;
1474
1475     s = g_malloc0(sizeof(BDRVSheepdogState));
1476
1477     memset(vdi, 0, sizeof(vdi));
1478     memset(tag, 0, sizeof(tag));
1479     if (strstr(filename, "://")) {
1480         ret = sd_parse_uri(s, filename, vdi, &snapid, tag);
1481     } else {
1482         ret = parse_vdiname(s, filename, vdi, &snapid, tag);
1483     }
1484     if (ret < 0) {
1485         goto out;
1486     }
1487
1488     while (options && options->name) {
1489         if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
1490             vdi_size = options->value.n;
1491         } else if (!strcmp(options->name, BLOCK_OPT_BACKING_FILE)) {
1492             backing_file = options->value.s;
1493         } else if (!strcmp(options->name, BLOCK_OPT_PREALLOC)) {
1494             if (!options->value.s || !strcmp(options->value.s, "off")) {
1495                 prealloc = false;
1496             } else if (!strcmp(options->value.s, "full")) {
1497                 prealloc = true;
1498             } else {
1499                 error_report("Invalid preallocation mode: '%s'",
1500                              options->value.s);
1501                 ret = -EINVAL;
1502                 goto out;
1503             }
1504         }
1505         options++;
1506     }
1507
1508     if (vdi_size > SD_MAX_VDI_SIZE) {
1509         error_report("too big image size");
1510         ret = -EINVAL;
1511         goto out;
1512     }
1513
1514     if (backing_file) {
1515         BlockDriverState *bs;
1516         BDRVSheepdogState *s;
1517         BlockDriver *drv;
1518
1519         /* Currently, only Sheepdog backing image is supported. */
1520         drv = bdrv_find_protocol(backing_file, true);
1521         if (!drv || strcmp(drv->protocol_name, "sheepdog") != 0) {
1522             error_report("backing_file must be a sheepdog image");
1523             ret = -EINVAL;
1524             goto out;
1525         }
1526
1527         ret = bdrv_file_open(&bs, backing_file, NULL, 0, &local_err);
1528         if (ret < 0) {
1529             qerror_report_err(local_err);
1530             error_free(local_err);
1531             goto out;
1532         }
1533
1534         s = bs->opaque;
1535
1536         if (!is_snapshot(&s->inode)) {
1537             error_report("cannot clone from a non snapshot vdi");
1538             bdrv_unref(bs);
1539             ret = -EINVAL;
1540             goto out;
1541         }
1542
1543         base_vid = s->inode.vdi_id;
1544         bdrv_unref(bs);
1545     }
1546
1547     /* TODO: allow users to specify copy number */
1548     ret = do_sd_create(s, vdi, vdi_size, base_vid, &vid, 0, 0);
1549     if (!prealloc || ret) {
1550         goto out;
1551     }
1552
1553     ret = sd_prealloc(filename);
1554 out:
1555     g_free(s);
1556     return ret;
1557 }
1558
1559 static void sd_close(BlockDriverState *bs)
1560 {
1561     BDRVSheepdogState *s = bs->opaque;
1562     SheepdogVdiReq hdr;
1563     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1564     unsigned int wlen, rlen = 0;
1565     int fd, ret;
1566
1567     DPRINTF("%s\n", s->name);
1568
1569     fd = connect_to_sdog(s);
1570     if (fd < 0) {
1571         return;
1572     }
1573
1574     memset(&hdr, 0, sizeof(hdr));
1575
1576     hdr.opcode = SD_OP_RELEASE_VDI;
1577     hdr.vdi_id = s->inode.vdi_id;
1578     wlen = strlen(s->name) + 1;
1579     hdr.data_length = wlen;
1580     hdr.flags = SD_FLAG_CMD_WRITE;
1581
1582     ret = do_req(fd, (SheepdogReq *)&hdr, s->name, &wlen, &rlen);
1583
1584     closesocket(fd);
1585
1586     if (!ret && rsp->result != SD_RES_SUCCESS &&
1587         rsp->result != SD_RES_VDI_NOT_LOCKED) {
1588         error_report("%s, %s", sd_strerror(rsp->result), s->name);
1589     }
1590
1591     qemu_aio_set_fd_handler(s->fd, NULL, NULL, NULL);
1592     closesocket(s->fd);
1593     g_free(s->host_spec);
1594 }
1595
1596 static int64_t sd_getlength(BlockDriverState *bs)
1597 {
1598     BDRVSheepdogState *s = bs->opaque;
1599
1600     return s->inode.vdi_size;
1601 }
1602
1603 static int sd_truncate(BlockDriverState *bs, int64_t offset)
1604 {
1605     BDRVSheepdogState *s = bs->opaque;
1606     int ret, fd;
1607     unsigned int datalen;
1608
1609     if (offset < s->inode.vdi_size) {
1610         error_report("shrinking is not supported");
1611         return -EINVAL;
1612     } else if (offset > SD_MAX_VDI_SIZE) {
1613         error_report("too big image size");
1614         return -EINVAL;
1615     }
1616
1617     fd = connect_to_sdog(s);
1618     if (fd < 0) {
1619         return fd;
1620     }
1621
1622     /* we don't need to update entire object */
1623     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
1624     s->inode.vdi_size = offset;
1625     ret = write_object(fd, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id),
1626                        s->inode.nr_copies, datalen, 0, false, s->cache_flags);
1627     close(fd);
1628
1629     if (ret < 0) {
1630         error_report("failed to update an inode.");
1631     }
1632
1633     return ret;
1634 }
1635
1636 /*
1637  * This function is called after writing data objects.  If we need to
1638  * update metadata, this sends a write request to the vdi object.
1639  * Otherwise, this switches back to sd_co_readv/writev.
1640  */
1641 static void coroutine_fn sd_write_done(SheepdogAIOCB *acb)
1642 {
1643     int ret;
1644     BDRVSheepdogState *s = acb->common.bs->opaque;
1645     struct iovec iov;
1646     AIOReq *aio_req;
1647     uint32_t offset, data_len, mn, mx;
1648
1649     mn = s->min_dirty_data_idx;
1650     mx = s->max_dirty_data_idx;
1651     if (mn <= mx) {
1652         /* we need to update the vdi object. */
1653         offset = sizeof(s->inode) - sizeof(s->inode.data_vdi_id) +
1654             mn * sizeof(s->inode.data_vdi_id[0]);
1655         data_len = (mx - mn + 1) * sizeof(s->inode.data_vdi_id[0]);
1656
1657         s->min_dirty_data_idx = UINT32_MAX;
1658         s->max_dirty_data_idx = 0;
1659
1660         iov.iov_base = &s->inode;
1661         iov.iov_len = sizeof(s->inode);
1662         aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
1663                                 data_len, offset, 0, 0, offset);
1664         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1665         ret = add_aio_request(s, aio_req, &iov, 1, false, AIOCB_WRITE_UDATA);
1666         if (ret) {
1667             free_aio_req(s, aio_req);
1668             acb->ret = -EIO;
1669             goto out;
1670         }
1671
1672         acb->aio_done_func = sd_finish_aiocb;
1673         acb->aiocb_type = AIOCB_WRITE_UDATA;
1674         return;
1675     }
1676 out:
1677     sd_finish_aiocb(acb);
1678 }
1679
1680 /* Delete current working VDI on the snapshot chain */
1681 static bool sd_delete(BDRVSheepdogState *s)
1682 {
1683     unsigned int wlen = SD_MAX_VDI_LEN, rlen = 0;
1684     SheepdogVdiReq hdr = {
1685         .opcode = SD_OP_DEL_VDI,
1686         .vdi_id = s->inode.vdi_id,
1687         .data_length = wlen,
1688         .flags = SD_FLAG_CMD_WRITE,
1689     };
1690     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1691     int fd, ret;
1692
1693     fd = connect_to_sdog(s);
1694     if (fd < 0) {
1695         return false;
1696     }
1697
1698     ret = do_req(fd, (SheepdogReq *)&hdr, s->name, &wlen, &rlen);
1699     closesocket(fd);
1700     if (ret) {
1701         return false;
1702     }
1703     switch (rsp->result) {
1704     case SD_RES_NO_VDI:
1705         error_report("%s was already deleted", s->name);
1706         /* fall through */
1707     case SD_RES_SUCCESS:
1708         break;
1709     default:
1710         error_report("%s, %s", sd_strerror(rsp->result), s->name);
1711         return false;
1712     }
1713
1714     return true;
1715 }
1716
1717 /*
1718  * Create a writable VDI from a snapshot
1719  */
1720 static int sd_create_branch(BDRVSheepdogState *s)
1721 {
1722     int ret, fd;
1723     uint32_t vid;
1724     char *buf;
1725     bool deleted;
1726
1727     DPRINTF("%" PRIx32 " is snapshot.\n", s->inode.vdi_id);
1728
1729     buf = g_malloc(SD_INODE_SIZE);
1730
1731     /*
1732      * Even If deletion fails, we will just create extra snapshot based on
1733      * the workding VDI which was supposed to be deleted. So no need to
1734      * false bail out.
1735      */
1736     deleted = sd_delete(s);
1737     ret = do_sd_create(s, s->name, s->inode.vdi_size, s->inode.vdi_id, &vid,
1738                        !deleted, s->inode.copy_policy);
1739     if (ret) {
1740         goto out;
1741     }
1742
1743     DPRINTF("%" PRIx32 " is created.\n", vid);
1744
1745     fd = connect_to_sdog(s);
1746     if (fd < 0) {
1747         ret = fd;
1748         goto out;
1749     }
1750
1751     ret = read_object(fd, buf, vid_to_vdi_oid(vid), s->inode.nr_copies,
1752                       SD_INODE_SIZE, 0, s->cache_flags);
1753
1754     closesocket(fd);
1755
1756     if (ret < 0) {
1757         goto out;
1758     }
1759
1760     memcpy(&s->inode, buf, sizeof(s->inode));
1761
1762     s->is_snapshot = false;
1763     ret = 0;
1764     DPRINTF("%" PRIx32 " was newly created.\n", s->inode.vdi_id);
1765
1766 out:
1767     g_free(buf);
1768
1769     return ret;
1770 }
1771
1772 /*
1773  * Send I/O requests to the server.
1774  *
1775  * This function sends requests to the server, links the requests to
1776  * the inflight_list in BDRVSheepdogState, and exits without
1777  * waiting the response.  The responses are received in the
1778  * `aio_read_response' function which is called from the main loop as
1779  * a fd handler.
1780  *
1781  * Returns 1 when we need to wait a response, 0 when there is no sent
1782  * request and -errno in error cases.
1783  */
1784 static int coroutine_fn sd_co_rw_vector(void *p)
1785 {
1786     SheepdogAIOCB *acb = p;
1787     int ret = 0;
1788     unsigned long len, done = 0, total = acb->nb_sectors * BDRV_SECTOR_SIZE;
1789     unsigned long idx = acb->sector_num * BDRV_SECTOR_SIZE / SD_DATA_OBJ_SIZE;
1790     uint64_t oid;
1791     uint64_t offset = (acb->sector_num * BDRV_SECTOR_SIZE) % SD_DATA_OBJ_SIZE;
1792     BDRVSheepdogState *s = acb->common.bs->opaque;
1793     SheepdogInode *inode = &s->inode;
1794     AIOReq *aio_req;
1795
1796     if (acb->aiocb_type == AIOCB_WRITE_UDATA && s->is_snapshot) {
1797         /*
1798          * In the case we open the snapshot VDI, Sheepdog creates the
1799          * writable VDI when we do a write operation first.
1800          */
1801         ret = sd_create_branch(s);
1802         if (ret) {
1803             acb->ret = -EIO;
1804             goto out;
1805         }
1806     }
1807
1808     /*
1809      * Make sure we don't free the aiocb before we are done with all requests.
1810      * This additional reference is dropped at the end of this function.
1811      */
1812     acb->nr_pending++;
1813
1814     while (done != total) {
1815         uint8_t flags = 0;
1816         uint64_t old_oid = 0;
1817         bool create = false;
1818
1819         oid = vid_to_data_oid(inode->data_vdi_id[idx], idx);
1820
1821         len = MIN(total - done, SD_DATA_OBJ_SIZE - offset);
1822
1823         switch (acb->aiocb_type) {
1824         case AIOCB_READ_UDATA:
1825             if (!inode->data_vdi_id[idx]) {
1826                 qemu_iovec_memset(acb->qiov, done, 0, len);
1827                 goto done;
1828             }
1829             break;
1830         case AIOCB_WRITE_UDATA:
1831             if (!inode->data_vdi_id[idx]) {
1832                 create = true;
1833             } else if (!is_data_obj_writable(inode, idx)) {
1834                 /* Copy-On-Write */
1835                 create = true;
1836                 old_oid = oid;
1837                 flags = SD_FLAG_CMD_COW;
1838             }
1839             break;
1840         case AIOCB_DISCARD_OBJ:
1841             /*
1842              * We discard the object only when the whole object is
1843              * 1) allocated 2) trimmed. Otherwise, simply skip it.
1844              */
1845             if (len != SD_DATA_OBJ_SIZE || inode->data_vdi_id[idx] == 0) {
1846                 goto done;
1847             }
1848             break;
1849         default:
1850             break;
1851         }
1852
1853         if (create) {
1854             DPRINTF("update ino (%" PRIu32 ") %" PRIu64 " %" PRIu64 " %ld\n",
1855                     inode->vdi_id, oid,
1856                     vid_to_data_oid(inode->data_vdi_id[idx], idx), idx);
1857             oid = vid_to_data_oid(inode->vdi_id, idx);
1858             DPRINTF("new oid %" PRIx64 "\n", oid);
1859         }
1860
1861         aio_req = alloc_aio_req(s, acb, oid, len, offset, flags, old_oid, done);
1862
1863         if (create) {
1864             AIOReq *areq;
1865             QLIST_FOREACH(areq, &s->inflight_aio_head, aio_siblings) {
1866                 if (areq->oid == oid) {
1867                     /*
1868                      * Sheepdog cannot handle simultaneous create
1869                      * requests to the same object.  So we cannot send
1870                      * the request until the previous request
1871                      * finishes.
1872                      */
1873                     aio_req->flags = 0;
1874                     aio_req->base_oid = 0;
1875                     QLIST_INSERT_HEAD(&s->pending_aio_head, aio_req,
1876                                       aio_siblings);
1877                     goto done;
1878                 }
1879             }
1880         }
1881
1882         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1883         ret = add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
1884                               create, acb->aiocb_type);
1885         if (ret < 0) {
1886             error_report("add_aio_request is failed");
1887             free_aio_req(s, aio_req);
1888             acb->ret = -EIO;
1889             goto out;
1890         }
1891     done:
1892         offset = 0;
1893         idx++;
1894         done += len;
1895     }
1896 out:
1897     if (!--acb->nr_pending) {
1898         return acb->ret;
1899     }
1900     return 1;
1901 }
1902
1903 static coroutine_fn int sd_co_writev(BlockDriverState *bs, int64_t sector_num,
1904                         int nb_sectors, QEMUIOVector *qiov)
1905 {
1906     SheepdogAIOCB *acb;
1907     int ret;
1908
1909     if (bs->growable && sector_num + nb_sectors > bs->total_sectors) {
1910         ret = sd_truncate(bs, (sector_num + nb_sectors) * BDRV_SECTOR_SIZE);
1911         if (ret < 0) {
1912             return ret;
1913         }
1914         bs->total_sectors = sector_num + nb_sectors;
1915     }
1916
1917     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors);
1918     acb->aio_done_func = sd_write_done;
1919     acb->aiocb_type = AIOCB_WRITE_UDATA;
1920
1921     ret = sd_co_rw_vector(acb);
1922     if (ret <= 0) {
1923         qemu_aio_release(acb);
1924         return ret;
1925     }
1926
1927     qemu_coroutine_yield();
1928
1929     return acb->ret;
1930 }
1931
1932 static coroutine_fn int sd_co_readv(BlockDriverState *bs, int64_t sector_num,
1933                        int nb_sectors, QEMUIOVector *qiov)
1934 {
1935     SheepdogAIOCB *acb;
1936     int ret;
1937
1938     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors);
1939     acb->aiocb_type = AIOCB_READ_UDATA;
1940     acb->aio_done_func = sd_finish_aiocb;
1941
1942     ret = sd_co_rw_vector(acb);
1943     if (ret <= 0) {
1944         qemu_aio_release(acb);
1945         return ret;
1946     }
1947
1948     qemu_coroutine_yield();
1949
1950     return acb->ret;
1951 }
1952
1953 static int coroutine_fn sd_co_flush_to_disk(BlockDriverState *bs)
1954 {
1955     BDRVSheepdogState *s = bs->opaque;
1956     SheepdogAIOCB *acb;
1957     AIOReq *aio_req;
1958     int ret;
1959
1960     if (s->cache_flags != SD_FLAG_CMD_CACHE) {
1961         return 0;
1962     }
1963
1964     acb = sd_aio_setup(bs, NULL, 0, 0);
1965     acb->aiocb_type = AIOCB_FLUSH_CACHE;
1966     acb->aio_done_func = sd_finish_aiocb;
1967
1968     aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
1969                             0, 0, 0, 0, 0);
1970     QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1971     ret = add_aio_request(s, aio_req, NULL, 0, false, acb->aiocb_type);
1972     if (ret < 0) {
1973         error_report("add_aio_request is failed");
1974         free_aio_req(s, aio_req);
1975         qemu_aio_release(acb);
1976         return ret;
1977     }
1978
1979     qemu_coroutine_yield();
1980     return acb->ret;
1981 }
1982
1983 static int sd_snapshot_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info)
1984 {
1985     BDRVSheepdogState *s = bs->opaque;
1986     int ret, fd;
1987     uint32_t new_vid;
1988     SheepdogInode *inode;
1989     unsigned int datalen;
1990
1991     DPRINTF("sn_info: name %s id_str %s s: name %s vm_state_size %" PRId64 " "
1992             "is_snapshot %d\n", sn_info->name, sn_info->id_str,
1993             s->name, sn_info->vm_state_size, s->is_snapshot);
1994
1995     if (s->is_snapshot) {
1996         error_report("You can't create a snapshot of a snapshot VDI, "
1997                      "%s (%" PRIu32 ").", s->name, s->inode.vdi_id);
1998
1999         return -EINVAL;
2000     }
2001
2002     DPRINTF("%s %s\n", sn_info->name, sn_info->id_str);
2003
2004     s->inode.vm_state_size = sn_info->vm_state_size;
2005     s->inode.vm_clock_nsec = sn_info->vm_clock_nsec;
2006     /* It appears that inode.tag does not require a NUL terminator,
2007      * which means this use of strncpy is ok.
2008      */
2009     strncpy(s->inode.tag, sn_info->name, sizeof(s->inode.tag));
2010     /* we don't need to update entire object */
2011     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
2012
2013     /* refresh inode. */
2014     fd = connect_to_sdog(s);
2015     if (fd < 0) {
2016         ret = fd;
2017         goto cleanup;
2018     }
2019
2020     ret = write_object(fd, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id),
2021                        s->inode.nr_copies, datalen, 0, false, s->cache_flags);
2022     if (ret < 0) {
2023         error_report("failed to write snapshot's inode.");
2024         goto cleanup;
2025     }
2026
2027     ret = do_sd_create(s, s->name, s->inode.vdi_size, s->inode.vdi_id, &new_vid,
2028                        1, s->inode.copy_policy);
2029     if (ret < 0) {
2030         error_report("failed to create inode for snapshot. %s",
2031                      strerror(errno));
2032         goto cleanup;
2033     }
2034
2035     inode = (SheepdogInode *)g_malloc(datalen);
2036
2037     ret = read_object(fd, (char *)inode, vid_to_vdi_oid(new_vid),
2038                       s->inode.nr_copies, datalen, 0, s->cache_flags);
2039
2040     if (ret < 0) {
2041         error_report("failed to read new inode info. %s", strerror(errno));
2042         goto cleanup;
2043     }
2044
2045     memcpy(&s->inode, inode, datalen);
2046     DPRINTF("s->inode: name %s snap_id %x oid %x\n",
2047             s->inode.name, s->inode.snap_id, s->inode.vdi_id);
2048
2049 cleanup:
2050     closesocket(fd);
2051     return ret;
2052 }
2053
2054 /*
2055  * We implement rollback(loadvm) operation to the specified snapshot by
2056  * 1) switch to the snapshot
2057  * 2) rely on sd_create_branch to delete working VDI and
2058  * 3) create a new working VDI based on the speicified snapshot
2059  */
2060 static int sd_snapshot_goto(BlockDriverState *bs, const char *snapshot_id)
2061 {
2062     BDRVSheepdogState *s = bs->opaque;
2063     BDRVSheepdogState *old_s;
2064     char tag[SD_MAX_VDI_TAG_LEN];
2065     uint32_t snapid = 0;
2066     int ret = 0;
2067
2068     old_s = g_malloc(sizeof(BDRVSheepdogState));
2069
2070     memcpy(old_s, s, sizeof(BDRVSheepdogState));
2071
2072     snapid = strtoul(snapshot_id, NULL, 10);
2073     if (snapid) {
2074         tag[0] = 0;
2075     } else {
2076         pstrcpy(tag, sizeof(tag), snapshot_id);
2077     }
2078
2079     ret = reload_inode(s, snapid, tag);
2080     if (ret) {
2081         goto out;
2082     }
2083
2084     ret = sd_create_branch(s);
2085     if (ret) {
2086         goto out;
2087     }
2088
2089     g_free(old_s);
2090
2091     return 0;
2092 out:
2093     /* recover bdrv_sd_state */
2094     memcpy(s, old_s, sizeof(BDRVSheepdogState));
2095     g_free(old_s);
2096
2097     error_report("failed to open. recover old bdrv_sd_state.");
2098
2099     return ret;
2100 }
2101
2102 static int sd_snapshot_delete(BlockDriverState *bs,
2103                               const char *snapshot_id,
2104                               const char *name,
2105                               Error **errp)
2106 {
2107     /* FIXME: Delete specified snapshot id.  */
2108     return 0;
2109 }
2110
2111 static int sd_snapshot_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab)
2112 {
2113     BDRVSheepdogState *s = bs->opaque;
2114     SheepdogReq req;
2115     int fd, nr = 1024, ret, max = BITS_TO_LONGS(SD_NR_VDIS) * sizeof(long);
2116     QEMUSnapshotInfo *sn_tab = NULL;
2117     unsigned wlen, rlen;
2118     int found = 0;
2119     static SheepdogInode inode;
2120     unsigned long *vdi_inuse;
2121     unsigned int start_nr;
2122     uint64_t hval;
2123     uint32_t vid;
2124
2125     vdi_inuse = g_malloc(max);
2126
2127     fd = connect_to_sdog(s);
2128     if (fd < 0) {
2129         ret = fd;
2130         goto out;
2131     }
2132
2133     rlen = max;
2134     wlen = 0;
2135
2136     memset(&req, 0, sizeof(req));
2137
2138     req.opcode = SD_OP_READ_VDIS;
2139     req.data_length = max;
2140
2141     ret = do_req(fd, (SheepdogReq *)&req, vdi_inuse, &wlen, &rlen);
2142
2143     closesocket(fd);
2144     if (ret) {
2145         goto out;
2146     }
2147
2148     sn_tab = g_malloc0(nr * sizeof(*sn_tab));
2149
2150     /* calculate a vdi id with hash function */
2151     hval = fnv_64a_buf(s->name, strlen(s->name), FNV1A_64_INIT);
2152     start_nr = hval & (SD_NR_VDIS - 1);
2153
2154     fd = connect_to_sdog(s);
2155     if (fd < 0) {
2156         ret = fd;
2157         goto out;
2158     }
2159
2160     for (vid = start_nr; found < nr; vid = (vid + 1) % SD_NR_VDIS) {
2161         if (!test_bit(vid, vdi_inuse)) {
2162             break;
2163         }
2164
2165         /* we don't need to read entire object */
2166         ret = read_object(fd, (char *)&inode, vid_to_vdi_oid(vid),
2167                           0, SD_INODE_SIZE - sizeof(inode.data_vdi_id), 0,
2168                           s->cache_flags);
2169
2170         if (ret) {
2171             continue;
2172         }
2173
2174         if (!strcmp(inode.name, s->name) && is_snapshot(&inode)) {
2175             sn_tab[found].date_sec = inode.snap_ctime >> 32;
2176             sn_tab[found].date_nsec = inode.snap_ctime & 0xffffffff;
2177             sn_tab[found].vm_state_size = inode.vm_state_size;
2178             sn_tab[found].vm_clock_nsec = inode.vm_clock_nsec;
2179
2180             snprintf(sn_tab[found].id_str, sizeof(sn_tab[found].id_str), "%u",
2181                      inode.snap_id);
2182             pstrcpy(sn_tab[found].name,
2183                     MIN(sizeof(sn_tab[found].name), sizeof(inode.tag)),
2184                     inode.tag);
2185             found++;
2186         }
2187     }
2188
2189     closesocket(fd);
2190 out:
2191     *psn_tab = sn_tab;
2192
2193     g_free(vdi_inuse);
2194
2195     if (ret < 0) {
2196         return ret;
2197     }
2198
2199     return found;
2200 }
2201
2202 static int do_load_save_vmstate(BDRVSheepdogState *s, uint8_t *data,
2203                                 int64_t pos, int size, int load)
2204 {
2205     bool create;
2206     int fd, ret = 0, remaining = size;
2207     unsigned int data_len;
2208     uint64_t vmstate_oid;
2209     uint64_t offset;
2210     uint32_t vdi_index;
2211     uint32_t vdi_id = load ? s->inode.parent_vdi_id : s->inode.vdi_id;
2212
2213     fd = connect_to_sdog(s);
2214     if (fd < 0) {
2215         return fd;
2216     }
2217
2218     while (remaining) {
2219         vdi_index = pos / SD_DATA_OBJ_SIZE;
2220         offset = pos % SD_DATA_OBJ_SIZE;
2221
2222         data_len = MIN(remaining, SD_DATA_OBJ_SIZE - offset);
2223
2224         vmstate_oid = vid_to_vmstate_oid(vdi_id, vdi_index);
2225
2226         create = (offset == 0);
2227         if (load) {
2228             ret = read_object(fd, (char *)data, vmstate_oid,
2229                               s->inode.nr_copies, data_len, offset,
2230                               s->cache_flags);
2231         } else {
2232             ret = write_object(fd, (char *)data, vmstate_oid,
2233                                s->inode.nr_copies, data_len, offset, create,
2234                                s->cache_flags);
2235         }
2236
2237         if (ret < 0) {
2238             error_report("failed to save vmstate %s", strerror(errno));
2239             goto cleanup;
2240         }
2241
2242         pos += data_len;
2243         data += data_len;
2244         remaining -= data_len;
2245     }
2246     ret = size;
2247 cleanup:
2248     closesocket(fd);
2249     return ret;
2250 }
2251
2252 static int sd_save_vmstate(BlockDriverState *bs, QEMUIOVector *qiov,
2253                            int64_t pos)
2254 {
2255     BDRVSheepdogState *s = bs->opaque;
2256     void *buf;
2257     int ret;
2258
2259     buf = qemu_blockalign(bs, qiov->size);
2260     qemu_iovec_to_buf(qiov, 0, buf, qiov->size);
2261     ret = do_load_save_vmstate(s, (uint8_t *) buf, pos, qiov->size, 0);
2262     qemu_vfree(buf);
2263
2264     return ret;
2265 }
2266
2267 static int sd_load_vmstate(BlockDriverState *bs, uint8_t *data,
2268                            int64_t pos, int size)
2269 {
2270     BDRVSheepdogState *s = bs->opaque;
2271
2272     return do_load_save_vmstate(s, data, pos, size, 1);
2273 }
2274
2275
2276 static coroutine_fn int sd_co_discard(BlockDriverState *bs, int64_t sector_num,
2277                                       int nb_sectors)
2278 {
2279     SheepdogAIOCB *acb;
2280     QEMUIOVector dummy;
2281     BDRVSheepdogState *s = bs->opaque;
2282     int ret;
2283
2284     if (!s->discard_supported) {
2285             return 0;
2286     }
2287
2288     acb = sd_aio_setup(bs, &dummy, sector_num, nb_sectors);
2289     acb->aiocb_type = AIOCB_DISCARD_OBJ;
2290     acb->aio_done_func = sd_finish_aiocb;
2291
2292     ret = sd_co_rw_vector(acb);
2293     if (ret <= 0) {
2294         qemu_aio_release(acb);
2295         return ret;
2296     }
2297
2298     qemu_coroutine_yield();
2299
2300     return acb->ret;
2301 }
2302
2303 static coroutine_fn int64_t
2304 sd_co_get_block_status(BlockDriverState *bs, int64_t sector_num, int nb_sectors,
2305                        int *pnum)
2306 {
2307     BDRVSheepdogState *s = bs->opaque;
2308     SheepdogInode *inode = &s->inode;
2309     unsigned long start = sector_num * BDRV_SECTOR_SIZE / SD_DATA_OBJ_SIZE,
2310                   end = DIV_ROUND_UP((sector_num + nb_sectors) *
2311                                      BDRV_SECTOR_SIZE, SD_DATA_OBJ_SIZE);
2312     unsigned long idx;
2313     int64_t ret = BDRV_BLOCK_DATA;
2314
2315     for (idx = start; idx < end; idx++) {
2316         if (inode->data_vdi_id[idx] == 0) {
2317             break;
2318         }
2319     }
2320     if (idx == start) {
2321         /* Get the longest length of unallocated sectors */
2322         ret = 0;
2323         for (idx = start + 1; idx < end; idx++) {
2324             if (inode->data_vdi_id[idx] != 0) {
2325                 break;
2326             }
2327         }
2328     }
2329
2330     *pnum = (idx - start) * SD_DATA_OBJ_SIZE / BDRV_SECTOR_SIZE;
2331     if (*pnum > nb_sectors) {
2332         *pnum = nb_sectors;
2333     }
2334     return ret;
2335 }
2336
2337 static QEMUOptionParameter sd_create_options[] = {
2338     {
2339         .name = BLOCK_OPT_SIZE,
2340         .type = OPT_SIZE,
2341         .help = "Virtual disk size"
2342     },
2343     {
2344         .name = BLOCK_OPT_BACKING_FILE,
2345         .type = OPT_STRING,
2346         .help = "File name of a base image"
2347     },
2348     {
2349         .name = BLOCK_OPT_PREALLOC,
2350         .type = OPT_STRING,
2351         .help = "Preallocation mode (allowed values: off, full)"
2352     },
2353     { NULL }
2354 };
2355
2356 static BlockDriver bdrv_sheepdog = {
2357     .format_name    = "sheepdog",
2358     .protocol_name  = "sheepdog",
2359     .instance_size  = sizeof(BDRVSheepdogState),
2360     .bdrv_needs_filename = true,
2361     .bdrv_file_open = sd_open,
2362     .bdrv_close     = sd_close,
2363     .bdrv_create    = sd_create,
2364     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2365     .bdrv_getlength = sd_getlength,
2366     .bdrv_truncate  = sd_truncate,
2367
2368     .bdrv_co_readv  = sd_co_readv,
2369     .bdrv_co_writev = sd_co_writev,
2370     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2371     .bdrv_co_discard = sd_co_discard,
2372     .bdrv_co_get_block_status = sd_co_get_block_status,
2373
2374     .bdrv_snapshot_create   = sd_snapshot_create,
2375     .bdrv_snapshot_goto     = sd_snapshot_goto,
2376     .bdrv_snapshot_delete   = sd_snapshot_delete,
2377     .bdrv_snapshot_list     = sd_snapshot_list,
2378
2379     .bdrv_save_vmstate  = sd_save_vmstate,
2380     .bdrv_load_vmstate  = sd_load_vmstate,
2381
2382     .create_options = sd_create_options,
2383 };
2384
2385 static BlockDriver bdrv_sheepdog_tcp = {
2386     .format_name    = "sheepdog",
2387     .protocol_name  = "sheepdog+tcp",
2388     .instance_size  = sizeof(BDRVSheepdogState),
2389     .bdrv_needs_filename = true,
2390     .bdrv_file_open = sd_open,
2391     .bdrv_close     = sd_close,
2392     .bdrv_create    = sd_create,
2393     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2394     .bdrv_getlength = sd_getlength,
2395     .bdrv_truncate  = sd_truncate,
2396
2397     .bdrv_co_readv  = sd_co_readv,
2398     .bdrv_co_writev = sd_co_writev,
2399     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2400     .bdrv_co_discard = sd_co_discard,
2401     .bdrv_co_get_block_status = sd_co_get_block_status,
2402
2403     .bdrv_snapshot_create   = sd_snapshot_create,
2404     .bdrv_snapshot_goto     = sd_snapshot_goto,
2405     .bdrv_snapshot_delete   = sd_snapshot_delete,
2406     .bdrv_snapshot_list     = sd_snapshot_list,
2407
2408     .bdrv_save_vmstate  = sd_save_vmstate,
2409     .bdrv_load_vmstate  = sd_load_vmstate,
2410
2411     .create_options = sd_create_options,
2412 };
2413
2414 static BlockDriver bdrv_sheepdog_unix = {
2415     .format_name    = "sheepdog",
2416     .protocol_name  = "sheepdog+unix",
2417     .instance_size  = sizeof(BDRVSheepdogState),
2418     .bdrv_needs_filename = true,
2419     .bdrv_file_open = sd_open,
2420     .bdrv_close     = sd_close,
2421     .bdrv_create    = sd_create,
2422     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2423     .bdrv_getlength = sd_getlength,
2424     .bdrv_truncate  = sd_truncate,
2425
2426     .bdrv_co_readv  = sd_co_readv,
2427     .bdrv_co_writev = sd_co_writev,
2428     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2429     .bdrv_co_discard = sd_co_discard,
2430     .bdrv_co_get_block_status = sd_co_get_block_status,
2431
2432     .bdrv_snapshot_create   = sd_snapshot_create,
2433     .bdrv_snapshot_goto     = sd_snapshot_goto,
2434     .bdrv_snapshot_delete   = sd_snapshot_delete,
2435     .bdrv_snapshot_list     = sd_snapshot_list,
2436
2437     .bdrv_save_vmstate  = sd_save_vmstate,
2438     .bdrv_load_vmstate  = sd_load_vmstate,
2439
2440     .create_options = sd_create_options,
2441 };
2442
2443 static void bdrv_sheepdog_init(void)
2444 {
2445     bdrv_register(&bdrv_sheepdog);
2446     bdrv_register(&bdrv_sheepdog_tcp);
2447     bdrv_register(&bdrv_sheepdog_unix);
2448 }
2449 block_init(bdrv_sheepdog_init);