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