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     aio_req->oid = vid_to_data_oid(s->inode.vdi_id,
1201                                    data_oid_to_idx(aio_req->oid));
1202
1203     /* check whether this request becomes a CoW one */
1204     if (acb->aiocb_type == AIOCB_WRITE_UDATA) {
1205         int idx = data_oid_to_idx(aio_req->oid);
1206         AIOReq *areq;
1207
1208         if (s->inode.data_vdi_id[idx] == 0) {
1209             create = true;
1210             goto out;
1211         }
1212         if (is_data_obj_writable(&s->inode, idx)) {
1213             goto out;
1214         }
1215
1216         /* link to the pending list if there is another CoW request to
1217          * the same object */
1218         QLIST_FOREACH(areq, &s->inflight_aio_head, aio_siblings) {
1219             if (areq != aio_req && areq->oid == aio_req->oid) {
1220                 DPRINTF("simultaneous CoW to %" PRIx64 "\n", aio_req->oid);
1221                 QLIST_REMOVE(aio_req, aio_siblings);
1222                 QLIST_INSERT_HEAD(&s->pending_aio_head, aio_req, aio_siblings);
1223                 return SD_RES_SUCCESS;
1224             }
1225         }
1226
1227         aio_req->base_oid = vid_to_data_oid(s->inode.data_vdi_id[idx], idx);
1228         aio_req->flags |= SD_FLAG_CMD_COW;
1229         create = true;
1230     }
1231 out:
1232     return add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
1233                            create, acb->aiocb_type);
1234 }
1235
1236 /* TODO Convert to fine grained options */
1237 static QemuOptsList runtime_opts = {
1238     .name = "sheepdog",
1239     .head = QTAILQ_HEAD_INITIALIZER(runtime_opts.head),
1240     .desc = {
1241         {
1242             .name = "filename",
1243             .type = QEMU_OPT_STRING,
1244             .help = "URL to the sheepdog image",
1245         },
1246         { /* end of list */ }
1247     },
1248 };
1249
1250 static int sd_open(BlockDriverState *bs, QDict *options, int flags,
1251                    Error **errp)
1252 {
1253     int ret, fd;
1254     uint32_t vid = 0;
1255     BDRVSheepdogState *s = bs->opaque;
1256     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1257     uint32_t snapid;
1258     char *buf = NULL;
1259     QemuOpts *opts;
1260     Error *local_err = NULL;
1261     const char *filename;
1262
1263     opts = qemu_opts_create_nofail(&runtime_opts);
1264     qemu_opts_absorb_qdict(opts, options, &local_err);
1265     if (error_is_set(&local_err)) {
1266         qerror_report_err(local_err);
1267         error_free(local_err);
1268         ret = -EINVAL;
1269         goto out;
1270     }
1271
1272     filename = qemu_opt_get(opts, "filename");
1273
1274     QLIST_INIT(&s->inflight_aio_head);
1275     QLIST_INIT(&s->pending_aio_head);
1276     s->fd = -1;
1277
1278     memset(vdi, 0, sizeof(vdi));
1279     memset(tag, 0, sizeof(tag));
1280
1281     if (strstr(filename, "://")) {
1282         ret = sd_parse_uri(s, filename, vdi, &snapid, tag);
1283     } else {
1284         ret = parse_vdiname(s, filename, vdi, &snapid, tag);
1285     }
1286     if (ret < 0) {
1287         goto out;
1288     }
1289     s->fd = get_sheep_fd(s);
1290     if (s->fd < 0) {
1291         ret = s->fd;
1292         goto out;
1293     }
1294
1295     ret = find_vdi_name(s, vdi, snapid, tag, &vid, true);
1296     if (ret) {
1297         goto out;
1298     }
1299
1300     /*
1301      * QEMU block layer emulates writethrough cache as 'writeback + flush', so
1302      * we always set SD_FLAG_CMD_CACHE (writeback cache) as default.
1303      */
1304     s->cache_flags = SD_FLAG_CMD_CACHE;
1305     if (flags & BDRV_O_NOCACHE) {
1306         s->cache_flags = SD_FLAG_CMD_DIRECT;
1307     }
1308     s->discard_supported = true;
1309
1310     if (snapid || tag[0] != '\0') {
1311         DPRINTF("%" PRIx32 " snapshot inode was open.\n", vid);
1312         s->is_snapshot = true;
1313     }
1314
1315     fd = connect_to_sdog(s);
1316     if (fd < 0) {
1317         ret = fd;
1318         goto out;
1319     }
1320
1321     buf = g_malloc(SD_INODE_SIZE);
1322     ret = read_object(fd, buf, vid_to_vdi_oid(vid), 0, SD_INODE_SIZE, 0,
1323                       s->cache_flags);
1324
1325     closesocket(fd);
1326
1327     if (ret) {
1328         goto out;
1329     }
1330
1331     memcpy(&s->inode, buf, sizeof(s->inode));
1332     s->min_dirty_data_idx = UINT32_MAX;
1333     s->max_dirty_data_idx = 0;
1334
1335     bs->total_sectors = s->inode.vdi_size / BDRV_SECTOR_SIZE;
1336     pstrcpy(s->name, sizeof(s->name), vdi);
1337     qemu_co_mutex_init(&s->lock);
1338     qemu_opts_del(opts);
1339     g_free(buf);
1340     return 0;
1341 out:
1342     qemu_aio_set_fd_handler(s->fd, NULL, NULL, NULL);
1343     if (s->fd >= 0) {
1344         closesocket(s->fd);
1345     }
1346     qemu_opts_del(opts);
1347     g_free(buf);
1348     return ret;
1349 }
1350
1351 static int do_sd_create(BDRVSheepdogState *s, char *filename, int64_t vdi_size,
1352                         uint32_t base_vid, uint32_t *vdi_id, int snapshot,
1353                         uint8_t copy_policy)
1354 {
1355     SheepdogVdiReq hdr;
1356     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1357     int fd, ret;
1358     unsigned int wlen, rlen = 0;
1359     char buf[SD_MAX_VDI_LEN];
1360
1361     fd = connect_to_sdog(s);
1362     if (fd < 0) {
1363         return fd;
1364     }
1365
1366     /* FIXME: would it be better to fail (e.g., return -EIO) when filename
1367      * does not fit in buf?  For now, just truncate and avoid buffer overrun.
1368      */
1369     memset(buf, 0, sizeof(buf));
1370     pstrcpy(buf, sizeof(buf), filename);
1371
1372     memset(&hdr, 0, sizeof(hdr));
1373     hdr.opcode = SD_OP_NEW_VDI;
1374     hdr.vdi_id = base_vid;
1375
1376     wlen = SD_MAX_VDI_LEN;
1377
1378     hdr.flags = SD_FLAG_CMD_WRITE;
1379     hdr.snapid = snapshot;
1380
1381     hdr.data_length = wlen;
1382     hdr.vdi_size = vdi_size;
1383     hdr.copy_policy = copy_policy;
1384
1385     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1386
1387     closesocket(fd);
1388
1389     if (ret) {
1390         return ret;
1391     }
1392
1393     if (rsp->result != SD_RES_SUCCESS) {
1394         error_report("%s, %s", sd_strerror(rsp->result), filename);
1395         return -EIO;
1396     }
1397
1398     if (vdi_id) {
1399         *vdi_id = rsp->vdi_id;
1400     }
1401
1402     return 0;
1403 }
1404
1405 static int sd_prealloc(const char *filename)
1406 {
1407     BlockDriverState *bs = NULL;
1408     uint32_t idx, max_idx;
1409     int64_t vdi_size;
1410     void *buf = g_malloc0(SD_DATA_OBJ_SIZE);
1411     Error *local_err = NULL;
1412     int ret;
1413
1414     ret = bdrv_file_open(&bs, filename, NULL, BDRV_O_RDWR, &local_err);
1415     if (ret < 0) {
1416         qerror_report_err(local_err);
1417         error_free(local_err);
1418         goto out;
1419     }
1420
1421     vdi_size = bdrv_getlength(bs);
1422     if (vdi_size < 0) {
1423         ret = vdi_size;
1424         goto out;
1425     }
1426     max_idx = DIV_ROUND_UP(vdi_size, SD_DATA_OBJ_SIZE);
1427
1428     for (idx = 0; idx < max_idx; idx++) {
1429         /*
1430          * The created image can be a cloned image, so we need to read
1431          * a data from the source image.
1432          */
1433         ret = bdrv_pread(bs, idx * SD_DATA_OBJ_SIZE, buf, SD_DATA_OBJ_SIZE);
1434         if (ret < 0) {
1435             goto out;
1436         }
1437         ret = bdrv_pwrite(bs, idx * SD_DATA_OBJ_SIZE, buf, SD_DATA_OBJ_SIZE);
1438         if (ret < 0) {
1439             goto out;
1440         }
1441     }
1442 out:
1443     if (bs) {
1444         bdrv_unref(bs);
1445     }
1446     g_free(buf);
1447
1448     return ret;
1449 }
1450
1451 static int sd_create(const char *filename, QEMUOptionParameter *options,
1452                      Error **errp)
1453 {
1454     int ret = 0;
1455     uint32_t vid = 0, base_vid = 0;
1456     int64_t vdi_size = 0;
1457     char *backing_file = NULL;
1458     BDRVSheepdogState *s;
1459     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1460     uint32_t snapid;
1461     bool prealloc = false;
1462     Error *local_err = NULL;
1463
1464     s = g_malloc0(sizeof(BDRVSheepdogState));
1465
1466     memset(vdi, 0, sizeof(vdi));
1467     memset(tag, 0, sizeof(tag));
1468     if (strstr(filename, "://")) {
1469         ret = sd_parse_uri(s, filename, vdi, &snapid, tag);
1470     } else {
1471         ret = parse_vdiname(s, filename, vdi, &snapid, tag);
1472     }
1473     if (ret < 0) {
1474         goto out;
1475     }
1476
1477     while (options && options->name) {
1478         if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
1479             vdi_size = options->value.n;
1480         } else if (!strcmp(options->name, BLOCK_OPT_BACKING_FILE)) {
1481             backing_file = options->value.s;
1482         } else if (!strcmp(options->name, BLOCK_OPT_PREALLOC)) {
1483             if (!options->value.s || !strcmp(options->value.s, "off")) {
1484                 prealloc = false;
1485             } else if (!strcmp(options->value.s, "full")) {
1486                 prealloc = true;
1487             } else {
1488                 error_report("Invalid preallocation mode: '%s'",
1489                              options->value.s);
1490                 ret = -EINVAL;
1491                 goto out;
1492             }
1493         }
1494         options++;
1495     }
1496
1497     if (vdi_size > SD_MAX_VDI_SIZE) {
1498         error_report("too big image size");
1499         ret = -EINVAL;
1500         goto out;
1501     }
1502
1503     if (backing_file) {
1504         BlockDriverState *bs;
1505         BDRVSheepdogState *s;
1506         BlockDriver *drv;
1507
1508         /* Currently, only Sheepdog backing image is supported. */
1509         drv = bdrv_find_protocol(backing_file, true);
1510         if (!drv || strcmp(drv->protocol_name, "sheepdog") != 0) {
1511             error_report("backing_file must be a sheepdog image");
1512             ret = -EINVAL;
1513             goto out;
1514         }
1515
1516         ret = bdrv_file_open(&bs, backing_file, NULL, 0, &local_err);
1517         if (ret < 0) {
1518             qerror_report_err(local_err);
1519             error_free(local_err);
1520             goto out;
1521         }
1522
1523         s = bs->opaque;
1524
1525         if (!is_snapshot(&s->inode)) {
1526             error_report("cannot clone from a non snapshot vdi");
1527             bdrv_unref(bs);
1528             ret = -EINVAL;
1529             goto out;
1530         }
1531
1532         base_vid = s->inode.vdi_id;
1533         bdrv_unref(bs);
1534     }
1535
1536     /* TODO: allow users to specify copy number */
1537     ret = do_sd_create(s, vdi, vdi_size, base_vid, &vid, 0, 0);
1538     if (!prealloc || ret) {
1539         goto out;
1540     }
1541
1542     ret = sd_prealloc(filename);
1543 out:
1544     g_free(s);
1545     return ret;
1546 }
1547
1548 static void sd_close(BlockDriverState *bs)
1549 {
1550     BDRVSheepdogState *s = bs->opaque;
1551     SheepdogVdiReq hdr;
1552     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1553     unsigned int wlen, rlen = 0;
1554     int fd, ret;
1555
1556     DPRINTF("%s\n", s->name);
1557
1558     fd = connect_to_sdog(s);
1559     if (fd < 0) {
1560         return;
1561     }
1562
1563     memset(&hdr, 0, sizeof(hdr));
1564
1565     hdr.opcode = SD_OP_RELEASE_VDI;
1566     hdr.vdi_id = s->inode.vdi_id;
1567     wlen = strlen(s->name) + 1;
1568     hdr.data_length = wlen;
1569     hdr.flags = SD_FLAG_CMD_WRITE;
1570
1571     ret = do_req(fd, (SheepdogReq *)&hdr, s->name, &wlen, &rlen);
1572
1573     closesocket(fd);
1574
1575     if (!ret && rsp->result != SD_RES_SUCCESS &&
1576         rsp->result != SD_RES_VDI_NOT_LOCKED) {
1577         error_report("%s, %s", sd_strerror(rsp->result), s->name);
1578     }
1579
1580     qemu_aio_set_fd_handler(s->fd, NULL, NULL, NULL);
1581     closesocket(s->fd);
1582     g_free(s->host_spec);
1583 }
1584
1585 static int64_t sd_getlength(BlockDriverState *bs)
1586 {
1587     BDRVSheepdogState *s = bs->opaque;
1588
1589     return s->inode.vdi_size;
1590 }
1591
1592 static int sd_truncate(BlockDriverState *bs, int64_t offset)
1593 {
1594     BDRVSheepdogState *s = bs->opaque;
1595     int ret, fd;
1596     unsigned int datalen;
1597
1598     if (offset < s->inode.vdi_size) {
1599         error_report("shrinking is not supported");
1600         return -EINVAL;
1601     } else if (offset > SD_MAX_VDI_SIZE) {
1602         error_report("too big image size");
1603         return -EINVAL;
1604     }
1605
1606     fd = connect_to_sdog(s);
1607     if (fd < 0) {
1608         return fd;
1609     }
1610
1611     /* we don't need to update entire object */
1612     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
1613     s->inode.vdi_size = offset;
1614     ret = write_object(fd, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id),
1615                        s->inode.nr_copies, datalen, 0, false, s->cache_flags);
1616     close(fd);
1617
1618     if (ret < 0) {
1619         error_report("failed to update an inode.");
1620     }
1621
1622     return ret;
1623 }
1624
1625 /*
1626  * This function is called after writing data objects.  If we need to
1627  * update metadata, this sends a write request to the vdi object.
1628  * Otherwise, this switches back to sd_co_readv/writev.
1629  */
1630 static void coroutine_fn sd_write_done(SheepdogAIOCB *acb)
1631 {
1632     int ret;
1633     BDRVSheepdogState *s = acb->common.bs->opaque;
1634     struct iovec iov;
1635     AIOReq *aio_req;
1636     uint32_t offset, data_len, mn, mx;
1637
1638     mn = s->min_dirty_data_idx;
1639     mx = s->max_dirty_data_idx;
1640     if (mn <= mx) {
1641         /* we need to update the vdi object. */
1642         offset = sizeof(s->inode) - sizeof(s->inode.data_vdi_id) +
1643             mn * sizeof(s->inode.data_vdi_id[0]);
1644         data_len = (mx - mn + 1) * sizeof(s->inode.data_vdi_id[0]);
1645
1646         s->min_dirty_data_idx = UINT32_MAX;
1647         s->max_dirty_data_idx = 0;
1648
1649         iov.iov_base = &s->inode;
1650         iov.iov_len = sizeof(s->inode);
1651         aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
1652                                 data_len, offset, 0, 0, offset);
1653         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1654         ret = add_aio_request(s, aio_req, &iov, 1, false, AIOCB_WRITE_UDATA);
1655         if (ret) {
1656             free_aio_req(s, aio_req);
1657             acb->ret = -EIO;
1658             goto out;
1659         }
1660
1661         acb->aio_done_func = sd_finish_aiocb;
1662         acb->aiocb_type = AIOCB_WRITE_UDATA;
1663         return;
1664     }
1665 out:
1666     sd_finish_aiocb(acb);
1667 }
1668
1669 /* Delete current working VDI on the snapshot chain */
1670 static bool sd_delete(BDRVSheepdogState *s)
1671 {
1672     unsigned int wlen = SD_MAX_VDI_LEN, rlen = 0;
1673     SheepdogVdiReq hdr = {
1674         .opcode = SD_OP_DEL_VDI,
1675         .vdi_id = s->inode.vdi_id,
1676         .data_length = wlen,
1677         .flags = SD_FLAG_CMD_WRITE,
1678     };
1679     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1680     int fd, ret;
1681
1682     fd = connect_to_sdog(s);
1683     if (fd < 0) {
1684         return false;
1685     }
1686
1687     ret = do_req(fd, (SheepdogReq *)&hdr, s->name, &wlen, &rlen);
1688     closesocket(fd);
1689     if (ret) {
1690         return false;
1691     }
1692     switch (rsp->result) {
1693     case SD_RES_NO_VDI:
1694         error_report("%s was already deleted", s->name);
1695         /* fall through */
1696     case SD_RES_SUCCESS:
1697         break;
1698     default:
1699         error_report("%s, %s", sd_strerror(rsp->result), s->name);
1700         return false;
1701     }
1702
1703     return true;
1704 }
1705
1706 /*
1707  * Create a writable VDI from a snapshot
1708  */
1709 static int sd_create_branch(BDRVSheepdogState *s)
1710 {
1711     int ret, fd;
1712     uint32_t vid;
1713     char *buf;
1714     bool deleted;
1715
1716     DPRINTF("%" PRIx32 " is snapshot.\n", s->inode.vdi_id);
1717
1718     buf = g_malloc(SD_INODE_SIZE);
1719
1720     /*
1721      * Even If deletion fails, we will just create extra snapshot based on
1722      * the workding VDI which was supposed to be deleted. So no need to
1723      * false bail out.
1724      */
1725     deleted = sd_delete(s);
1726     ret = do_sd_create(s, s->name, s->inode.vdi_size, s->inode.vdi_id, &vid,
1727                        !deleted, s->inode.copy_policy);
1728     if (ret) {
1729         goto out;
1730     }
1731
1732     DPRINTF("%" PRIx32 " is created.\n", vid);
1733
1734     fd = connect_to_sdog(s);
1735     if (fd < 0) {
1736         ret = fd;
1737         goto out;
1738     }
1739
1740     ret = read_object(fd, buf, vid_to_vdi_oid(vid), s->inode.nr_copies,
1741                       SD_INODE_SIZE, 0, s->cache_flags);
1742
1743     closesocket(fd);
1744
1745     if (ret < 0) {
1746         goto out;
1747     }
1748
1749     memcpy(&s->inode, buf, sizeof(s->inode));
1750
1751     s->is_snapshot = false;
1752     ret = 0;
1753     DPRINTF("%" PRIx32 " was newly created.\n", s->inode.vdi_id);
1754
1755 out:
1756     g_free(buf);
1757
1758     return ret;
1759 }
1760
1761 /*
1762  * Send I/O requests to the server.
1763  *
1764  * This function sends requests to the server, links the requests to
1765  * the inflight_list in BDRVSheepdogState, and exits without
1766  * waiting the response.  The responses are received in the
1767  * `aio_read_response' function which is called from the main loop as
1768  * a fd handler.
1769  *
1770  * Returns 1 when we need to wait a response, 0 when there is no sent
1771  * request and -errno in error cases.
1772  */
1773 static int coroutine_fn sd_co_rw_vector(void *p)
1774 {
1775     SheepdogAIOCB *acb = p;
1776     int ret = 0;
1777     unsigned long len, done = 0, total = acb->nb_sectors * BDRV_SECTOR_SIZE;
1778     unsigned long idx = acb->sector_num * BDRV_SECTOR_SIZE / SD_DATA_OBJ_SIZE;
1779     uint64_t oid;
1780     uint64_t offset = (acb->sector_num * BDRV_SECTOR_SIZE) % SD_DATA_OBJ_SIZE;
1781     BDRVSheepdogState *s = acb->common.bs->opaque;
1782     SheepdogInode *inode = &s->inode;
1783     AIOReq *aio_req;
1784
1785     if (acb->aiocb_type == AIOCB_WRITE_UDATA && s->is_snapshot) {
1786         /*
1787          * In the case we open the snapshot VDI, Sheepdog creates the
1788          * writable VDI when we do a write operation first.
1789          */
1790         ret = sd_create_branch(s);
1791         if (ret) {
1792             acb->ret = -EIO;
1793             goto out;
1794         }
1795     }
1796
1797     /*
1798      * Make sure we don't free the aiocb before we are done with all requests.
1799      * This additional reference is dropped at the end of this function.
1800      */
1801     acb->nr_pending++;
1802
1803     while (done != total) {
1804         uint8_t flags = 0;
1805         uint64_t old_oid = 0;
1806         bool create = false;
1807
1808         oid = vid_to_data_oid(inode->data_vdi_id[idx], idx);
1809
1810         len = MIN(total - done, SD_DATA_OBJ_SIZE - offset);
1811
1812         switch (acb->aiocb_type) {
1813         case AIOCB_READ_UDATA:
1814             if (!inode->data_vdi_id[idx]) {
1815                 qemu_iovec_memset(acb->qiov, done, 0, len);
1816                 goto done;
1817             }
1818             break;
1819         case AIOCB_WRITE_UDATA:
1820             if (!inode->data_vdi_id[idx]) {
1821                 create = true;
1822             } else if (!is_data_obj_writable(inode, idx)) {
1823                 /* Copy-On-Write */
1824                 create = true;
1825                 old_oid = oid;
1826                 flags = SD_FLAG_CMD_COW;
1827             }
1828             break;
1829         case AIOCB_DISCARD_OBJ:
1830             /*
1831              * We discard the object only when the whole object is
1832              * 1) allocated 2) trimmed. Otherwise, simply skip it.
1833              */
1834             if (len != SD_DATA_OBJ_SIZE || inode->data_vdi_id[idx] == 0) {
1835                 goto done;
1836             }
1837             break;
1838         default:
1839             break;
1840         }
1841
1842         if (create) {
1843             DPRINTF("update ino (%" PRIu32 ") %" PRIu64 " %" PRIu64 " %ld\n",
1844                     inode->vdi_id, oid,
1845                     vid_to_data_oid(inode->data_vdi_id[idx], idx), idx);
1846             oid = vid_to_data_oid(inode->vdi_id, idx);
1847             DPRINTF("new oid %" PRIx64 "\n", oid);
1848         }
1849
1850         aio_req = alloc_aio_req(s, acb, oid, len, offset, flags, old_oid, done);
1851
1852         if (create) {
1853             AIOReq *areq;
1854             QLIST_FOREACH(areq, &s->inflight_aio_head, aio_siblings) {
1855                 if (areq->oid == oid) {
1856                     /*
1857                      * Sheepdog cannot handle simultaneous create
1858                      * requests to the same object.  So we cannot send
1859                      * the request until the previous request
1860                      * finishes.
1861                      */
1862                     aio_req->flags = 0;
1863                     aio_req->base_oid = 0;
1864                     QLIST_INSERT_HEAD(&s->pending_aio_head, aio_req,
1865                                       aio_siblings);
1866                     goto done;
1867                 }
1868             }
1869         }
1870
1871         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1872         ret = add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
1873                               create, acb->aiocb_type);
1874         if (ret < 0) {
1875             error_report("add_aio_request is failed");
1876             free_aio_req(s, aio_req);
1877             acb->ret = -EIO;
1878             goto out;
1879         }
1880     done:
1881         offset = 0;
1882         idx++;
1883         done += len;
1884     }
1885 out:
1886     if (!--acb->nr_pending) {
1887         return acb->ret;
1888     }
1889     return 1;
1890 }
1891
1892 static coroutine_fn int sd_co_writev(BlockDriverState *bs, int64_t sector_num,
1893                         int nb_sectors, QEMUIOVector *qiov)
1894 {
1895     SheepdogAIOCB *acb;
1896     int ret;
1897
1898     if (bs->growable && sector_num + nb_sectors > bs->total_sectors) {
1899         ret = sd_truncate(bs, (sector_num + nb_sectors) * BDRV_SECTOR_SIZE);
1900         if (ret < 0) {
1901             return ret;
1902         }
1903         bs->total_sectors = sector_num + nb_sectors;
1904     }
1905
1906     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors);
1907     acb->aio_done_func = sd_write_done;
1908     acb->aiocb_type = AIOCB_WRITE_UDATA;
1909
1910     ret = sd_co_rw_vector(acb);
1911     if (ret <= 0) {
1912         qemu_aio_release(acb);
1913         return ret;
1914     }
1915
1916     qemu_coroutine_yield();
1917
1918     return acb->ret;
1919 }
1920
1921 static coroutine_fn int sd_co_readv(BlockDriverState *bs, int64_t sector_num,
1922                        int nb_sectors, QEMUIOVector *qiov)
1923 {
1924     SheepdogAIOCB *acb;
1925     int ret;
1926
1927     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors);
1928     acb->aiocb_type = AIOCB_READ_UDATA;
1929     acb->aio_done_func = sd_finish_aiocb;
1930
1931     ret = sd_co_rw_vector(acb);
1932     if (ret <= 0) {
1933         qemu_aio_release(acb);
1934         return ret;
1935     }
1936
1937     qemu_coroutine_yield();
1938
1939     return acb->ret;
1940 }
1941
1942 static int coroutine_fn sd_co_flush_to_disk(BlockDriverState *bs)
1943 {
1944     BDRVSheepdogState *s = bs->opaque;
1945     SheepdogAIOCB *acb;
1946     AIOReq *aio_req;
1947     int ret;
1948
1949     if (s->cache_flags != SD_FLAG_CMD_CACHE) {
1950         return 0;
1951     }
1952
1953     acb = sd_aio_setup(bs, NULL, 0, 0);
1954     acb->aiocb_type = AIOCB_FLUSH_CACHE;
1955     acb->aio_done_func = sd_finish_aiocb;
1956
1957     aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
1958                             0, 0, 0, 0, 0);
1959     QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1960     ret = add_aio_request(s, aio_req, NULL, 0, false, acb->aiocb_type);
1961     if (ret < 0) {
1962         error_report("add_aio_request is failed");
1963         free_aio_req(s, aio_req);
1964         qemu_aio_release(acb);
1965         return ret;
1966     }
1967
1968     qemu_coroutine_yield();
1969     return acb->ret;
1970 }
1971
1972 static int sd_snapshot_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info)
1973 {
1974     BDRVSheepdogState *s = bs->opaque;
1975     int ret, fd;
1976     uint32_t new_vid;
1977     SheepdogInode *inode;
1978     unsigned int datalen;
1979
1980     DPRINTF("sn_info: name %s id_str %s s: name %s vm_state_size %" PRId64 " "
1981             "is_snapshot %d\n", sn_info->name, sn_info->id_str,
1982             s->name, sn_info->vm_state_size, s->is_snapshot);
1983
1984     if (s->is_snapshot) {
1985         error_report("You can't create a snapshot of a snapshot VDI, "
1986                      "%s (%" PRIu32 ").", s->name, s->inode.vdi_id);
1987
1988         return -EINVAL;
1989     }
1990
1991     DPRINTF("%s %s\n", sn_info->name, sn_info->id_str);
1992
1993     s->inode.vm_state_size = sn_info->vm_state_size;
1994     s->inode.vm_clock_nsec = sn_info->vm_clock_nsec;
1995     /* It appears that inode.tag does not require a NUL terminator,
1996      * which means this use of strncpy is ok.
1997      */
1998     strncpy(s->inode.tag, sn_info->name, sizeof(s->inode.tag));
1999     /* we don't need to update entire object */
2000     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
2001
2002     /* refresh inode. */
2003     fd = connect_to_sdog(s);
2004     if (fd < 0) {
2005         ret = fd;
2006         goto cleanup;
2007     }
2008
2009     ret = write_object(fd, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id),
2010                        s->inode.nr_copies, datalen, 0, false, s->cache_flags);
2011     if (ret < 0) {
2012         error_report("failed to write snapshot's inode.");
2013         goto cleanup;
2014     }
2015
2016     ret = do_sd_create(s, s->name, s->inode.vdi_size, s->inode.vdi_id, &new_vid,
2017                        1, s->inode.copy_policy);
2018     if (ret < 0) {
2019         error_report("failed to create inode for snapshot. %s",
2020                      strerror(errno));
2021         goto cleanup;
2022     }
2023
2024     inode = (SheepdogInode *)g_malloc(datalen);
2025
2026     ret = read_object(fd, (char *)inode, vid_to_vdi_oid(new_vid),
2027                       s->inode.nr_copies, datalen, 0, s->cache_flags);
2028
2029     if (ret < 0) {
2030         error_report("failed to read new inode info. %s", strerror(errno));
2031         goto cleanup;
2032     }
2033
2034     memcpy(&s->inode, inode, datalen);
2035     DPRINTF("s->inode: name %s snap_id %x oid %x\n",
2036             s->inode.name, s->inode.snap_id, s->inode.vdi_id);
2037
2038 cleanup:
2039     closesocket(fd);
2040     return ret;
2041 }
2042
2043 /*
2044  * We implement rollback(loadvm) operation to the specified snapshot by
2045  * 1) switch to the snapshot
2046  * 2) rely on sd_create_branch to delete working VDI and
2047  * 3) create a new working VDI based on the speicified snapshot
2048  */
2049 static int sd_snapshot_goto(BlockDriverState *bs, const char *snapshot_id)
2050 {
2051     BDRVSheepdogState *s = bs->opaque;
2052     BDRVSheepdogState *old_s;
2053     char tag[SD_MAX_VDI_TAG_LEN];
2054     uint32_t snapid = 0;
2055     int ret = 0;
2056
2057     old_s = g_malloc(sizeof(BDRVSheepdogState));
2058
2059     memcpy(old_s, s, sizeof(BDRVSheepdogState));
2060
2061     snapid = strtoul(snapshot_id, NULL, 10);
2062     if (snapid) {
2063         tag[0] = 0;
2064     } else {
2065         pstrcpy(tag, sizeof(tag), snapshot_id);
2066     }
2067
2068     ret = reload_inode(s, snapid, tag);
2069     if (ret) {
2070         goto out;
2071     }
2072
2073     ret = sd_create_branch(s);
2074     if (ret) {
2075         goto out;
2076     }
2077
2078     g_free(old_s);
2079
2080     return 0;
2081 out:
2082     /* recover bdrv_sd_state */
2083     memcpy(s, old_s, sizeof(BDRVSheepdogState));
2084     g_free(old_s);
2085
2086     error_report("failed to open. recover old bdrv_sd_state.");
2087
2088     return ret;
2089 }
2090
2091 static int sd_snapshot_delete(BlockDriverState *bs,
2092                               const char *snapshot_id,
2093                               const char *name,
2094                               Error **errp)
2095 {
2096     /* FIXME: Delete specified snapshot id.  */
2097     return 0;
2098 }
2099
2100 static int sd_snapshot_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab)
2101 {
2102     BDRVSheepdogState *s = bs->opaque;
2103     SheepdogReq req;
2104     int fd, nr = 1024, ret, max = BITS_TO_LONGS(SD_NR_VDIS) * sizeof(long);
2105     QEMUSnapshotInfo *sn_tab = NULL;
2106     unsigned wlen, rlen;
2107     int found = 0;
2108     static SheepdogInode inode;
2109     unsigned long *vdi_inuse;
2110     unsigned int start_nr;
2111     uint64_t hval;
2112     uint32_t vid;
2113
2114     vdi_inuse = g_malloc(max);
2115
2116     fd = connect_to_sdog(s);
2117     if (fd < 0) {
2118         ret = fd;
2119         goto out;
2120     }
2121
2122     rlen = max;
2123     wlen = 0;
2124
2125     memset(&req, 0, sizeof(req));
2126
2127     req.opcode = SD_OP_READ_VDIS;
2128     req.data_length = max;
2129
2130     ret = do_req(fd, (SheepdogReq *)&req, vdi_inuse, &wlen, &rlen);
2131
2132     closesocket(fd);
2133     if (ret) {
2134         goto out;
2135     }
2136
2137     sn_tab = g_malloc0(nr * sizeof(*sn_tab));
2138
2139     /* calculate a vdi id with hash function */
2140     hval = fnv_64a_buf(s->name, strlen(s->name), FNV1A_64_INIT);
2141     start_nr = hval & (SD_NR_VDIS - 1);
2142
2143     fd = connect_to_sdog(s);
2144     if (fd < 0) {
2145         ret = fd;
2146         goto out;
2147     }
2148
2149     for (vid = start_nr; found < nr; vid = (vid + 1) % SD_NR_VDIS) {
2150         if (!test_bit(vid, vdi_inuse)) {
2151             break;
2152         }
2153
2154         /* we don't need to read entire object */
2155         ret = read_object(fd, (char *)&inode, vid_to_vdi_oid(vid),
2156                           0, SD_INODE_SIZE - sizeof(inode.data_vdi_id), 0,
2157                           s->cache_flags);
2158
2159         if (ret) {
2160             continue;
2161         }
2162
2163         if (!strcmp(inode.name, s->name) && is_snapshot(&inode)) {
2164             sn_tab[found].date_sec = inode.snap_ctime >> 32;
2165             sn_tab[found].date_nsec = inode.snap_ctime & 0xffffffff;
2166             sn_tab[found].vm_state_size = inode.vm_state_size;
2167             sn_tab[found].vm_clock_nsec = inode.vm_clock_nsec;
2168
2169             snprintf(sn_tab[found].id_str, sizeof(sn_tab[found].id_str), "%u",
2170                      inode.snap_id);
2171             pstrcpy(sn_tab[found].name,
2172                     MIN(sizeof(sn_tab[found].name), sizeof(inode.tag)),
2173                     inode.tag);
2174             found++;
2175         }
2176     }
2177
2178     closesocket(fd);
2179 out:
2180     *psn_tab = sn_tab;
2181
2182     g_free(vdi_inuse);
2183
2184     if (ret < 0) {
2185         return ret;
2186     }
2187
2188     return found;
2189 }
2190
2191 static int do_load_save_vmstate(BDRVSheepdogState *s, uint8_t *data,
2192                                 int64_t pos, int size, int load)
2193 {
2194     bool create;
2195     int fd, ret = 0, remaining = size;
2196     unsigned int data_len;
2197     uint64_t vmstate_oid;
2198     uint64_t offset;
2199     uint32_t vdi_index;
2200     uint32_t vdi_id = load ? s->inode.parent_vdi_id : s->inode.vdi_id;
2201
2202     fd = connect_to_sdog(s);
2203     if (fd < 0) {
2204         return fd;
2205     }
2206
2207     while (remaining) {
2208         vdi_index = pos / SD_DATA_OBJ_SIZE;
2209         offset = pos % SD_DATA_OBJ_SIZE;
2210
2211         data_len = MIN(remaining, SD_DATA_OBJ_SIZE - offset);
2212
2213         vmstate_oid = vid_to_vmstate_oid(vdi_id, vdi_index);
2214
2215         create = (offset == 0);
2216         if (load) {
2217             ret = read_object(fd, (char *)data, vmstate_oid,
2218                               s->inode.nr_copies, data_len, offset,
2219                               s->cache_flags);
2220         } else {
2221             ret = write_object(fd, (char *)data, vmstate_oid,
2222                                s->inode.nr_copies, data_len, offset, create,
2223                                s->cache_flags);
2224         }
2225
2226         if (ret < 0) {
2227             error_report("failed to save vmstate %s", strerror(errno));
2228             goto cleanup;
2229         }
2230
2231         pos += data_len;
2232         data += data_len;
2233         remaining -= data_len;
2234     }
2235     ret = size;
2236 cleanup:
2237     closesocket(fd);
2238     return ret;
2239 }
2240
2241 static int sd_save_vmstate(BlockDriverState *bs, QEMUIOVector *qiov,
2242                            int64_t pos)
2243 {
2244     BDRVSheepdogState *s = bs->opaque;
2245     void *buf;
2246     int ret;
2247
2248     buf = qemu_blockalign(bs, qiov->size);
2249     qemu_iovec_to_buf(qiov, 0, buf, qiov->size);
2250     ret = do_load_save_vmstate(s, (uint8_t *) buf, pos, qiov->size, 0);
2251     qemu_vfree(buf);
2252
2253     return ret;
2254 }
2255
2256 static int sd_load_vmstate(BlockDriverState *bs, uint8_t *data,
2257                            int64_t pos, int size)
2258 {
2259     BDRVSheepdogState *s = bs->opaque;
2260
2261     return do_load_save_vmstate(s, data, pos, size, 1);
2262 }
2263
2264
2265 static coroutine_fn int sd_co_discard(BlockDriverState *bs, int64_t sector_num,
2266                                       int nb_sectors)
2267 {
2268     SheepdogAIOCB *acb;
2269     QEMUIOVector dummy;
2270     BDRVSheepdogState *s = bs->opaque;
2271     int ret;
2272
2273     if (!s->discard_supported) {
2274             return 0;
2275     }
2276
2277     acb = sd_aio_setup(bs, &dummy, sector_num, nb_sectors);
2278     acb->aiocb_type = AIOCB_DISCARD_OBJ;
2279     acb->aio_done_func = sd_finish_aiocb;
2280
2281     ret = sd_co_rw_vector(acb);
2282     if (ret <= 0) {
2283         qemu_aio_release(acb);
2284         return ret;
2285     }
2286
2287     qemu_coroutine_yield();
2288
2289     return acb->ret;
2290 }
2291
2292 static coroutine_fn int64_t
2293 sd_co_get_block_status(BlockDriverState *bs, int64_t sector_num, int nb_sectors,
2294                        int *pnum)
2295 {
2296     BDRVSheepdogState *s = bs->opaque;
2297     SheepdogInode *inode = &s->inode;
2298     unsigned long start = sector_num * BDRV_SECTOR_SIZE / SD_DATA_OBJ_SIZE,
2299                   end = DIV_ROUND_UP((sector_num + nb_sectors) *
2300                                      BDRV_SECTOR_SIZE, SD_DATA_OBJ_SIZE);
2301     unsigned long idx;
2302     int64_t ret = BDRV_BLOCK_DATA;
2303
2304     for (idx = start; idx < end; idx++) {
2305         if (inode->data_vdi_id[idx] == 0) {
2306             break;
2307         }
2308     }
2309     if (idx == start) {
2310         /* Get the longest length of unallocated sectors */
2311         ret = 0;
2312         for (idx = start + 1; idx < end; idx++) {
2313             if (inode->data_vdi_id[idx] != 0) {
2314                 break;
2315             }
2316         }
2317     }
2318
2319     *pnum = (idx - start) * SD_DATA_OBJ_SIZE / BDRV_SECTOR_SIZE;
2320     if (*pnum > nb_sectors) {
2321         *pnum = nb_sectors;
2322     }
2323     return ret;
2324 }
2325
2326 static QEMUOptionParameter sd_create_options[] = {
2327     {
2328         .name = BLOCK_OPT_SIZE,
2329         .type = OPT_SIZE,
2330         .help = "Virtual disk size"
2331     },
2332     {
2333         .name = BLOCK_OPT_BACKING_FILE,
2334         .type = OPT_STRING,
2335         .help = "File name of a base image"
2336     },
2337     {
2338         .name = BLOCK_OPT_PREALLOC,
2339         .type = OPT_STRING,
2340         .help = "Preallocation mode (allowed values: off, full)"
2341     },
2342     { NULL }
2343 };
2344
2345 static BlockDriver bdrv_sheepdog = {
2346     .format_name    = "sheepdog",
2347     .protocol_name  = "sheepdog",
2348     .instance_size  = sizeof(BDRVSheepdogState),
2349     .bdrv_needs_filename = true,
2350     .bdrv_file_open = sd_open,
2351     .bdrv_close     = sd_close,
2352     .bdrv_create    = sd_create,
2353     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2354     .bdrv_getlength = sd_getlength,
2355     .bdrv_truncate  = sd_truncate,
2356
2357     .bdrv_co_readv  = sd_co_readv,
2358     .bdrv_co_writev = sd_co_writev,
2359     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2360     .bdrv_co_discard = sd_co_discard,
2361     .bdrv_co_get_block_status = sd_co_get_block_status,
2362
2363     .bdrv_snapshot_create   = sd_snapshot_create,
2364     .bdrv_snapshot_goto     = sd_snapshot_goto,
2365     .bdrv_snapshot_delete   = sd_snapshot_delete,
2366     .bdrv_snapshot_list     = sd_snapshot_list,
2367
2368     .bdrv_save_vmstate  = sd_save_vmstate,
2369     .bdrv_load_vmstate  = sd_load_vmstate,
2370
2371     .create_options = sd_create_options,
2372 };
2373
2374 static BlockDriver bdrv_sheepdog_tcp = {
2375     .format_name    = "sheepdog",
2376     .protocol_name  = "sheepdog+tcp",
2377     .instance_size  = sizeof(BDRVSheepdogState),
2378     .bdrv_needs_filename = true,
2379     .bdrv_file_open = sd_open,
2380     .bdrv_close     = sd_close,
2381     .bdrv_create    = sd_create,
2382     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2383     .bdrv_getlength = sd_getlength,
2384     .bdrv_truncate  = sd_truncate,
2385
2386     .bdrv_co_readv  = sd_co_readv,
2387     .bdrv_co_writev = sd_co_writev,
2388     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2389     .bdrv_co_discard = sd_co_discard,
2390     .bdrv_co_get_block_status = sd_co_get_block_status,
2391
2392     .bdrv_snapshot_create   = sd_snapshot_create,
2393     .bdrv_snapshot_goto     = sd_snapshot_goto,
2394     .bdrv_snapshot_delete   = sd_snapshot_delete,
2395     .bdrv_snapshot_list     = sd_snapshot_list,
2396
2397     .bdrv_save_vmstate  = sd_save_vmstate,
2398     .bdrv_load_vmstate  = sd_load_vmstate,
2399
2400     .create_options = sd_create_options,
2401 };
2402
2403 static BlockDriver bdrv_sheepdog_unix = {
2404     .format_name    = "sheepdog",
2405     .protocol_name  = "sheepdog+unix",
2406     .instance_size  = sizeof(BDRVSheepdogState),
2407     .bdrv_needs_filename = true,
2408     .bdrv_file_open = sd_open,
2409     .bdrv_close     = sd_close,
2410     .bdrv_create    = sd_create,
2411     .bdrv_has_zero_init = bdrv_has_zero_init_1,
2412     .bdrv_getlength = sd_getlength,
2413     .bdrv_truncate  = sd_truncate,
2414
2415     .bdrv_co_readv  = sd_co_readv,
2416     .bdrv_co_writev = sd_co_writev,
2417     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2418     .bdrv_co_discard = sd_co_discard,
2419     .bdrv_co_get_block_status = sd_co_get_block_status,
2420
2421     .bdrv_snapshot_create   = sd_snapshot_create,
2422     .bdrv_snapshot_goto     = sd_snapshot_goto,
2423     .bdrv_snapshot_delete   = sd_snapshot_delete,
2424     .bdrv_snapshot_list     = sd_snapshot_list,
2425
2426     .bdrv_save_vmstate  = sd_save_vmstate,
2427     .bdrv_load_vmstate  = sd_load_vmstate,
2428
2429     .create_options = sd_create_options,
2430 };
2431
2432 static void bdrv_sheepdog_init(void)
2433 {
2434     bdrv_register(&bdrv_sheepdog);
2435     bdrv_register(&bdrv_sheepdog_tcp);
2436     bdrv_register(&bdrv_sheepdog_unix);
2437 }
2438 block_init(bdrv_sheepdog_init);