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_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 int 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 int 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 int 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     int 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     int is_snapshot;
 296     uint8_t 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 = 1;
 421 }
 422
 423 static AIOPool sd_aio_pool = {
 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_aio_pool, 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 = 0;
 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             break;
 489         }
 490
 491         dprintf("connected to %s:%s\n", addr, port);
 492         goto success;
 493     }
 494     fd = -errno;
 495     error_report("failed connect to %s:%s", addr, port);
 496 success:
 497     freeaddrinfo(res0);
 498     return fd;
 499 }
 500
 501 static coroutine_fn int send_co_req(int sockfd, SheepdogReq *hdr, void *data,
 502                                     unsigned int *wlen)
 503 {
 504     int ret;
 505
 506     ret = qemu_co_send(sockfd, hdr, sizeof(*hdr));
 507     if (ret < sizeof(*hdr)) {
 508         error_report("failed to send a req, %s", strerror(errno));
 509         return ret;
 510     }
 511
 512     ret = qemu_co_send(sockfd, data, *wlen);
 513     if (ret < *wlen) {
 514         error_report("failed to send a req, %s", strerror(errno));
 515     }
 516
 517     return ret;
 518 }
 519
 520 static void restart_co_req(void *opaque)
 521 {
 522     Coroutine *co = opaque;
 523
 524     qemu_coroutine_enter(co, NULL);
 525 }
 526
 527 typedef struct SheepdogReqCo {
 528     int sockfd;
 529     SheepdogReq *hdr;
 530     void *data;
 531     unsigned int *wlen;
 532     unsigned int *rlen;
 533     int ret;
 534     bool finished;
 535 } SheepdogReqCo;
 536
 537 static coroutine_fn void do_co_req(void *opaque)
 538 {
 539     int ret;
 540     Coroutine *co;
 541     SheepdogReqCo *srco = opaque;
 542     int sockfd = srco->sockfd;
 543     SheepdogReq *hdr = srco->hdr;
 544     void *data = srco->data;
 545     unsigned int *wlen = srco->wlen;
 546     unsigned int *rlen = srco->rlen;
 547
 548     co = qemu_coroutine_self();
 549     qemu_aio_set_fd_handler(sockfd, NULL, restart_co_req, NULL, co);
 550
 551     socket_set_block(sockfd);
 552     ret = send_co_req(sockfd, hdr, data, wlen);
 553     if (ret < 0) {
 554         goto out;
 555     }
 556
 557     qemu_aio_set_fd_handler(sockfd, restart_co_req, NULL, NULL, co);
 558
 559     ret = qemu_co_recv(sockfd, hdr, sizeof(*hdr));
 560     if (ret < sizeof(*hdr)) {
 561         error_report("failed to get a rsp, %s", strerror(errno));
 562         ret = -errno;
 563         goto out;
 564     }
 565
 566     if (*rlen > hdr->data_length) {
 567         *rlen = hdr->data_length;
 568     }
 569
 570     if (*rlen) {
 571         ret = qemu_co_recv(sockfd, data, *rlen);
 572         if (ret < *rlen) {
 573             error_report("failed to get the data, %s", strerror(errno));
 574             ret = -errno;
 575             goto out;
 576         }
 577     }
 578     ret = 0;
 579 out:
 580     qemu_aio_set_fd_handler(sockfd, NULL, NULL, NULL, NULL);
 581     socket_set_nonblock(sockfd);
 582
 583     srco->ret = ret;
 584     srco->finished = true;
 585 }
 586
 587 static int do_req(int sockfd, SheepdogReq *hdr, void *data,
 588                   unsigned int *wlen, unsigned int *rlen)
 589 {
 590     Coroutine *co;
 591     SheepdogReqCo srco = {
 592         .sockfd = sockfd,
 593         .hdr = hdr,
 594         .data = data,
 595         .wlen = wlen,
 596         .rlen = rlen,
 597         .ret = 0,
 598         .finished = false,
 599     };
 600
 601     if (qemu_in_coroutine()) {
 602         do_co_req(&srco);
 603     } else {
 604         co = qemu_coroutine_create(do_co_req);
 605         qemu_coroutine_enter(co, &srco);
 606         while (!srco.finished) {
 607             qemu_aio_wait();
 608         }
 609     }
 610
 611     return srco.ret;
 612 }
 613
 614 static int coroutine_fn add_aio_request(BDRVSheepdogState *s, AIOReq *aio_req,
 615                            struct iovec *iov, int niov, int create,
 616                            enum AIOCBState aiocb_type);
 617
 618
 619 static AIOReq *find_pending_req(BDRVSheepdogState *s, uint64_t oid)
 620 {
 621     AIOReq *aio_req;
 622
 623     QLIST_FOREACH(aio_req, &s->pending_aio_head, aio_siblings) {
 624         if (aio_req->oid == oid) {
 625             return aio_req;
 626         }
 627     }
 628
 629     return NULL;
 630 }
 631
 632 /*
 633  * This function searchs pending requests to the object `oid', and
 634  * sends them.
 635  */
 636 static void coroutine_fn send_pending_req(BDRVSheepdogState *s, uint64_t oid)
 637 {
 638     AIOReq *aio_req;
 639     SheepdogAIOCB *acb;
 640     int ret;
 641
 642     while ((aio_req = find_pending_req(s, oid)) != NULL) {
 643         acb = aio_req->aiocb;
 644         /* move aio_req from pending list to inflight one */
 645         QLIST_REMOVE(aio_req, aio_siblings);
 646         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
 647         ret = add_aio_request(s, aio_req, acb->qiov->iov,
 648                               acb->qiov->niov, 0, acb->aiocb_type);
 649         if (ret < 0) {
 650             error_report("add_aio_request is failed");
 651             free_aio_req(s, aio_req);
 652             if (!acb->nr_pending) {
 653                 sd_finish_aiocb(acb);
 654             }
 655         }
 656     }
 657 }
 658
 659 /*
 660  * Receive responses of the I/O requests.
 661  *
 662  * This function is registered as a fd handler, and called from the
 663  * main loop when s->fd is ready for reading responses.
 664  */
 665 static void coroutine_fn aio_read_response(void *opaque)
 666 {
 667     SheepdogObjRsp rsp;
 668     BDRVSheepdogState *s = opaque;
 669     int fd = s->fd;
 670     int ret;
 671     AIOReq *aio_req = NULL;
 672     SheepdogAIOCB *acb;
 673     unsigned long idx;
 674
 675     if (QLIST_EMPTY(&s->inflight_aio_head)) {
 676         goto out;
 677     }
 678
 679     /* read a header */
 680     ret = qemu_co_recv(fd, &rsp, sizeof(rsp));
 681     if (ret < 0) {
 682         error_report("failed to get the header, %s", strerror(errno));
 683         goto out;
 684     }
 685
 686     /* find the right aio_req from the inflight aio list */
 687     QLIST_FOREACH(aio_req, &s->inflight_aio_head, aio_siblings) {
 688         if (aio_req->id == rsp.id) {
 689             break;
 690         }
 691     }
 692     if (!aio_req) {
 693         error_report("cannot find aio_req %x", rsp.id);
 694         goto out;
 695     }
 696
 697     acb = aio_req->aiocb;
 698
 699     switch (acb->aiocb_type) {
 700     case AIOCB_WRITE_UDATA:
 701         /* this coroutine context is no longer suitable for co_recv
 702          * because we may send data to update vdi objects */
 703         s->co_recv = NULL;
 704         if (!is_data_obj(aio_req->oid)) {
 705             break;
 706         }
 707         idx = data_oid_to_idx(aio_req->oid);
 708
 709         if (s->inode.data_vdi_id[idx] != s->inode.vdi_id) {
 710             /*
 711              * If the object is newly created one, we need to update
 712              * the vdi object (metadata object).  min_dirty_data_idx
 713              * and max_dirty_data_idx are changed to include updated
 714              * index between them.
 715              */
 716             s->inode.data_vdi_id[idx] = s->inode.vdi_id;
 717             s->max_dirty_data_idx = MAX(idx, s->max_dirty_data_idx);
 718             s->min_dirty_data_idx = MIN(idx, s->min_dirty_data_idx);
 719
 720             /*
 721              * Some requests may be blocked because simultaneous
 722              * create requests are not allowed, so we search the
 723              * pending requests here.
 724              */
 725             send_pending_req(s, vid_to_data_oid(s->inode.vdi_id, idx));
 726         }
 727         break;
 728     case AIOCB_READ_UDATA:
 729         ret = qemu_co_recvv(fd, acb->qiov->iov, acb->qiov->niov,
 730                             aio_req->iov_offset, rsp.data_length);
 731         if (ret < 0) {
 732             error_report("failed to get the data, %s", strerror(errno));
 733             goto out;
 734         }
 735         break;
 736     }
 737
 738     if (rsp.result != SD_RES_SUCCESS) {
 739         acb->ret = -EIO;
 740         error_report("%s", sd_strerror(rsp.result));
 741     }
 742
 743     free_aio_req(s, aio_req);
 744     if (!acb->nr_pending) {
 745         /*
 746          * We've finished all requests which belong to the AIOCB, so
 747          * we can switch back to sd_co_readv/writev now.
 748          */
 749         acb->aio_done_func(acb);
 750     }
 751 out:
 752     s->co_recv = NULL;
 753 }
 754
 755 static void co_read_response(void *opaque)
 756 {
 757     BDRVSheepdogState *s = opaque;
 758
 759     if (!s->co_recv) {
 760         s->co_recv = qemu_coroutine_create(aio_read_response);
 761     }
 762
 763     qemu_coroutine_enter(s->co_recv, opaque);
 764 }
 765
 766 static void co_write_request(void *opaque)
 767 {
 768     BDRVSheepdogState *s = opaque;
 769
 770     qemu_coroutine_enter(s->co_send, NULL);
 771 }
 772
 773 static int aio_flush_request(void *opaque)
 774 {
 775     BDRVSheepdogState *s = opaque;
 776
 777     return !QLIST_EMPTY(&s->inflight_aio_head) ||
 778         !QLIST_EMPTY(&s->pending_aio_head);
 779 }
 780
 781 static int set_nodelay(int fd)
 782 {
 783     int ret, opt;
 784
 785     opt = 1;
 786     ret = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char *)&opt, sizeof(opt));
 787     return ret;
 788 }
 789
 790 /*
 791  * Return a socket discriptor to read/write objects.
 792  *
 793  * We cannot use this discriptor for other operations because
 794  * the block driver may be on waiting response from the server.
 795  */
 796 static int get_sheep_fd(BDRVSheepdogState *s)
 797 {
 798     int ret, fd;
 799
 800     fd = connect_to_sdog(s->addr, s->port);
 801     if (fd < 0) {
 802         error_report("%s", strerror(errno));
 803         return fd;
 804     }
 805
 806     socket_set_nonblock(fd);
 807
 808     ret = set_nodelay(fd);
 809     if (ret) {
 810         error_report("%s", strerror(errno));
 811         closesocket(fd);
 812         return -errno;
 813     }
 814
 815     qemu_aio_set_fd_handler(fd, co_read_response, NULL, aio_flush_request, s);
 816     return fd;
 817 }
 818
 819 /*
 820  * Parse a filename
 821  *
 822  * filename must be one of the following formats:
 823  *   1. [vdiname]
 824  *   2. [vdiname]:[snapid]
 825  *   3. [vdiname]:[tag]
 826  *   4. [hostname]:[port]:[vdiname]
 827  *   5. [hostname]:[port]:[vdiname]:[snapid]
 828  *   6. [hostname]:[port]:[vdiname]:[tag]
 829  *
 830  * You can boot from the snapshot images by specifying `snapid` or
 831  * `tag'.
 832  *
 833  * You can run VMs outside the Sheepdog cluster by specifying
 834  * `hostname' and `port' (experimental).
 835  */
 836 static int parse_vdiname(BDRVSheepdogState *s, const char *filename,
 837                          char *vdi, uint32_t *snapid, char *tag)
 838 {
 839     char *p, *q;
 840     int nr_sep;
 841
 842     p = q = g_strdup(filename);
 843
 844     /* count the number of separators */
 845     nr_sep = 0;
 846     while (*p) {
 847         if (*p == ':') {
 848             nr_sep++;
 849         }
 850         p++;
 851     }
 852     p = q;
 853
 854     /* use the first two tokens as hostname and port number. */
 855     if (nr_sep >= 2) {
 856         s->addr = p;
 857         p = strchr(p, ':');
 858         *p++ = '\0';
 859
 860         s->port = p;
 861         p = strchr(p, ':');
 862         *p++ = '\0';
 863     } else {
 864         s->addr = NULL;
 865         s->port = 0;
 866     }
 867
 868     strncpy(vdi, p, SD_MAX_VDI_LEN);
 869
 870     p = strchr(vdi, ':');
 871     if (p) {
 872         *p++ = '\0';
 873         *snapid = strtoul(p, NULL, 10);
 874         if (*snapid == 0) {
 875             strncpy(tag, p, SD_MAX_VDI_TAG_LEN);
 876         }
 877     } else {
 878         *snapid = CURRENT_VDI_ID; /* search current vdi */
 879     }
 880
 881     if (s->addr == NULL) {
 882         g_free(q);
 883     }
 884
 885     return 0;
 886 }
 887
 888 static int find_vdi_name(BDRVSheepdogState *s, char *filename, uint32_t snapid,
 889                          char *tag, uint32_t *vid, int for_snapshot)
 890 {
 891     int ret, fd;
 892     SheepdogVdiReq hdr;
 893     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
 894     unsigned int wlen, rlen = 0;
 895     char buf[SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN];
 896
 897     fd = connect_to_sdog(s->addr, s->port);
 898     if (fd < 0) {
 899         return fd;
 900     }
 901
 902     memset(buf, 0, sizeof(buf));
 903     strncpy(buf, filename, SD_MAX_VDI_LEN);
 904     strncpy(buf + SD_MAX_VDI_LEN, tag, SD_MAX_VDI_TAG_LEN);
 905
 906     memset(&hdr, 0, sizeof(hdr));
 907     if (for_snapshot) {
 908         hdr.opcode = SD_OP_GET_VDI_INFO;
 909     } else {
 910         hdr.opcode = SD_OP_LOCK_VDI;
 911     }
 912     wlen = SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN;
 913     hdr.proto_ver = SD_PROTO_VER;
 914     hdr.data_length = wlen;
 915     hdr.snapid = snapid;
 916     hdr.flags = SD_FLAG_CMD_WRITE;
 917
 918     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
 919     if (ret) {
 920         goto out;
 921     }
 922
 923     if (rsp->result != SD_RES_SUCCESS) {
 924         error_report("cannot get vdi info, %s, %s %d %s",
 925                      sd_strerror(rsp->result), filename, snapid, tag);
 926         if (rsp->result == SD_RES_NO_VDI) {
 927             ret = -ENOENT;
 928         } else {
 929             ret = -EIO;
 930         }
 931         goto out;
 932     }
 933     *vid = rsp->vdi_id;
 934
 935     ret = 0;
 936 out:
 937     closesocket(fd);
 938     return ret;
 939 }
 940
 941 static int coroutine_fn add_aio_request(BDRVSheepdogState *s, AIOReq *aio_req,
 942                            struct iovec *iov, int niov, int create,
 943                            enum AIOCBState aiocb_type)
 944 {
 945     int nr_copies = s->inode.nr_copies;
 946     SheepdogObjReq hdr;
 947     unsigned int wlen;
 948     int ret;
 949     uint64_t oid = aio_req->oid;
 950     unsigned int datalen = aio_req->data_len;
 951     uint64_t offset = aio_req->offset;
 952     uint8_t flags = aio_req->flags;
 953     uint64_t old_oid = aio_req->base_oid;
 954
 955     if (!nr_copies) {
 956         error_report("bug");
 957     }
 958
 959     memset(&hdr, 0, sizeof(hdr));
 960
 961     if (aiocb_type == AIOCB_READ_UDATA) {
 962         wlen = 0;
 963         hdr.opcode = SD_OP_READ_OBJ;
 964         hdr.flags = flags;
 965     } else if (create) {
 966         wlen = datalen;
 967         hdr.opcode = SD_OP_CREATE_AND_WRITE_OBJ;
 968         hdr.flags = SD_FLAG_CMD_WRITE | flags;
 969     } else {
 970         wlen = datalen;
 971         hdr.opcode = SD_OP_WRITE_OBJ;
 972         hdr.flags = SD_FLAG_CMD_WRITE | flags;
 973     }
 974
 975     if (s->cache_enabled) {
 976         hdr.flags |= SD_FLAG_CMD_CACHE;
 977     }
 978
 979     hdr.oid = oid;
 980     hdr.cow_oid = old_oid;
 981     hdr.copies = s->inode.nr_copies;
 982
 983     hdr.data_length = datalen;
 984     hdr.offset = offset;
 985
 986     hdr.id = aio_req->id;
 987
 988     qemu_co_mutex_lock(&s->lock);
 989     s->co_send = qemu_coroutine_self();
 990     qemu_aio_set_fd_handler(s->fd, co_read_response, co_write_request,
 991                             aio_flush_request, s);
 992     socket_set_cork(s->fd, 1);
 993
 994     /* send a header */
 995     ret = qemu_co_send(s->fd, &hdr, sizeof(hdr));
 996     if (ret < 0) {
 997         qemu_co_mutex_unlock(&s->lock);
 998         error_report("failed to send a req, %s", strerror(errno));
 999         return -errno;
1000     }
1001
1002     if (wlen) {
1003         ret = qemu_co_sendv(s->fd, iov, niov, aio_req->iov_offset, wlen);
1004         if (ret < 0) {
1005             qemu_co_mutex_unlock(&s->lock);
1006             error_report("failed to send a data, %s", strerror(errno));
1007             return -errno;
1008         }
1009     }
1010
1011     socket_set_cork(s->fd, 0);
1012     qemu_aio_set_fd_handler(s->fd, co_read_response, NULL,
1013                             aio_flush_request, s);
1014     qemu_co_mutex_unlock(&s->lock);
1015
1016     return 0;
1017 }
1018
1019 static int read_write_object(int fd, char *buf, uint64_t oid, int copies,
1020                              unsigned int datalen, uint64_t offset,
1021                              int write, int create, uint8_t cache)
1022 {
1023     SheepdogObjReq hdr;
1024     SheepdogObjRsp *rsp = (SheepdogObjRsp *)&hdr;
1025     unsigned int wlen, rlen;
1026     int ret;
1027
1028     memset(&hdr, 0, sizeof(hdr));
1029
1030     if (write) {
1031         wlen = datalen;
1032         rlen = 0;
1033         hdr.flags = SD_FLAG_CMD_WRITE;
1034         if (create) {
1035             hdr.opcode = SD_OP_CREATE_AND_WRITE_OBJ;
1036         } else {
1037             hdr.opcode = SD_OP_WRITE_OBJ;
1038         }
1039     } else {
1040         wlen = 0;
1041         rlen = datalen;
1042         hdr.opcode = SD_OP_READ_OBJ;
1043     }
1044
1045     if (cache) {
1046         hdr.flags |= SD_FLAG_CMD_CACHE;
1047     }
1048
1049     hdr.oid = oid;
1050     hdr.data_length = datalen;
1051     hdr.offset = offset;
1052     hdr.copies = copies;
1053
1054     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1055     if (ret) {
1056         error_report("failed to send a request to the sheep");
1057         return ret;
1058     }
1059
1060     switch (rsp->result) {
1061     case SD_RES_SUCCESS:
1062         return 0;
1063     default:
1064         error_report("%s", sd_strerror(rsp->result));
1065         return -EIO;
1066     }
1067 }
1068
1069 static int read_object(int fd, char *buf, uint64_t oid, int copies,
1070                        unsigned int datalen, uint64_t offset, uint8_t cache)
1071 {
1072     return read_write_object(fd, buf, oid, copies, datalen, offset, 0, 0,
1073                              cache);
1074 }
1075
1076 static int write_object(int fd, char *buf, uint64_t oid, int copies,
1077                         unsigned int datalen, uint64_t offset, int create,
1078                         uint8_t cache)
1079 {
1080     return read_write_object(fd, buf, oid, copies, datalen, offset, 1, create,
1081                              cache);
1082 }
1083
1084 static int sd_open(BlockDriverState *bs, const char *filename, int flags)
1085 {
1086     int ret, fd;
1087     uint32_t vid = 0;
1088     BDRVSheepdogState *s = bs->opaque;
1089     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1090     uint32_t snapid;
1091     char *buf = NULL;
1092
1093     strstart(filename, "sheepdog:", (const char **)&filename);
1094
1095     QLIST_INIT(&s->inflight_aio_head);
1096     QLIST_INIT(&s->pending_aio_head);
1097     s->fd = -1;
1098
1099     memset(vdi, 0, sizeof(vdi));
1100     memset(tag, 0, sizeof(tag));
1101     if (parse_vdiname(s, filename, vdi, &snapid, tag) < 0) {
1102         ret = -EINVAL;
1103         goto out;
1104     }
1105     s->fd = get_sheep_fd(s);
1106     if (s->fd < 0) {
1107         ret = s->fd;
1108         goto out;
1109     }
1110
1111     ret = find_vdi_name(s, vdi, snapid, tag, &vid, 0);
1112     if (ret) {
1113         goto out;
1114     }
1115
1116     if (flags & BDRV_O_CACHE_WB) {
1117         s->cache_enabled = 1;
1118         s->flush_fd = connect_to_sdog(s->addr, s->port);
1119         if (s->flush_fd < 0) {
1120             error_report("failed to connect");
1121             ret = s->flush_fd;
1122             goto out;
1123         }
1124     }
1125
1126     if (snapid || tag[0] != '\0') {
1127         dprintf("%" PRIx32 " snapshot inode was open.\n", vid);
1128         s->is_snapshot = 1;
1129     }
1130
1131     fd = connect_to_sdog(s->addr, s->port);
1132     if (fd < 0) {
1133         error_report("failed to connect");
1134         ret = fd;
1135         goto out;
1136     }
1137
1138     buf = g_malloc(SD_INODE_SIZE);
1139     ret = read_object(fd, buf, vid_to_vdi_oid(vid), 0, SD_INODE_SIZE, 0,
1140                       s->cache_enabled);
1141
1142     closesocket(fd);
1143
1144     if (ret) {
1145         goto out;
1146     }
1147
1148     memcpy(&s->inode, buf, sizeof(s->inode));
1149     s->min_dirty_data_idx = UINT32_MAX;
1150     s->max_dirty_data_idx = 0;
1151
1152     bs->total_sectors = s->inode.vdi_size / SECTOR_SIZE;
1153     strncpy(s->name, vdi, sizeof(s->name));
1154     qemu_co_mutex_init(&s->lock);
1155     g_free(buf);
1156     return 0;
1157 out:
1158     qemu_aio_set_fd_handler(s->fd, NULL, NULL, NULL, NULL);
1159     if (s->fd >= 0) {
1160         closesocket(s->fd);
1161     }
1162     g_free(buf);
1163     return ret;
1164 }
1165
1166 static int do_sd_create(char *filename, int64_t vdi_size,
1167                         uint32_t base_vid, uint32_t *vdi_id, int snapshot,
1168                         const char *addr, const char *port)
1169 {
1170     SheepdogVdiReq hdr;
1171     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1172     int fd, ret;
1173     unsigned int wlen, rlen = 0;
1174     char buf[SD_MAX_VDI_LEN];
1175
1176     fd = connect_to_sdog(addr, port);
1177     if (fd < 0) {
1178         return fd;
1179     }
1180
1181     memset(buf, 0, sizeof(buf));
1182     strncpy(buf, filename, SD_MAX_VDI_LEN);
1183
1184     memset(&hdr, 0, sizeof(hdr));
1185     hdr.opcode = SD_OP_NEW_VDI;
1186     hdr.base_vdi_id = base_vid;
1187
1188     wlen = SD_MAX_VDI_LEN;
1189
1190     hdr.flags = SD_FLAG_CMD_WRITE;
1191     hdr.snapid = snapshot;
1192
1193     hdr.data_length = wlen;
1194     hdr.vdi_size = vdi_size;
1195
1196     ret = do_req(fd, (SheepdogReq *)&hdr, buf, &wlen, &rlen);
1197
1198     closesocket(fd);
1199
1200     if (ret) {
1201         return ret;
1202     }
1203
1204     if (rsp->result != SD_RES_SUCCESS) {
1205         error_report("%s, %s", sd_strerror(rsp->result), filename);
1206         return -EIO;
1207     }
1208
1209     if (vdi_id) {
1210         *vdi_id = rsp->vdi_id;
1211     }
1212
1213     return 0;
1214 }
1215
1216 static int sd_prealloc(const char *filename)
1217 {
1218     BlockDriverState *bs = NULL;
1219     uint32_t idx, max_idx;
1220     int64_t vdi_size;
1221     void *buf = g_malloc0(SD_DATA_OBJ_SIZE);
1222     int ret;
1223
1224     ret = bdrv_file_open(&bs, filename, BDRV_O_RDWR);
1225     if (ret < 0) {
1226         goto out;
1227     }
1228
1229     vdi_size = bdrv_getlength(bs);
1230     if (vdi_size < 0) {
1231         ret = vdi_size;
1232         goto out;
1233     }
1234     max_idx = DIV_ROUND_UP(vdi_size, SD_DATA_OBJ_SIZE);
1235
1236     for (idx = 0; idx < max_idx; idx++) {
1237         /*
1238          * The created image can be a cloned image, so we need to read
1239          * a data from the source image.
1240          */
1241         ret = bdrv_pread(bs, idx * SD_DATA_OBJ_SIZE, buf, SD_DATA_OBJ_SIZE);
1242         if (ret < 0) {
1243             goto out;
1244         }
1245         ret = bdrv_pwrite(bs, idx * SD_DATA_OBJ_SIZE, buf, SD_DATA_OBJ_SIZE);
1246         if (ret < 0) {
1247             goto out;
1248         }
1249     }
1250 out:
1251     if (bs) {
1252         bdrv_delete(bs);
1253     }
1254     g_free(buf);
1255
1256     return ret;
1257 }
1258
1259 static int sd_create(const char *filename, QEMUOptionParameter *options)
1260 {
1261     int ret = 0;
1262     uint32_t vid = 0, base_vid = 0;
1263     int64_t vdi_size = 0;
1264     char *backing_file = NULL;
1265     BDRVSheepdogState *s;
1266     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1267     uint32_t snapid;
1268     int prealloc = 0;
1269     const char *vdiname;
1270
1271     s = g_malloc0(sizeof(BDRVSheepdogState));
1272
1273     strstart(filename, "sheepdog:", &vdiname);
1274
1275     memset(vdi, 0, sizeof(vdi));
1276     memset(tag, 0, sizeof(tag));
1277     if (parse_vdiname(s, vdiname, vdi, &snapid, tag) < 0) {
1278         error_report("invalid filename");
1279         ret = -EINVAL;
1280         goto out;
1281     }
1282
1283     while (options && options->name) {
1284         if (!strcmp(options->name, BLOCK_OPT_SIZE)) {
1285             vdi_size = options->value.n;
1286         } else if (!strcmp(options->name, BLOCK_OPT_BACKING_FILE)) {
1287             backing_file = options->value.s;
1288         } else if (!strcmp(options->name, BLOCK_OPT_PREALLOC)) {
1289             if (!options->value.s || !strcmp(options->value.s, "off")) {
1290                 prealloc = 0;
1291             } else if (!strcmp(options->value.s, "full")) {
1292                 prealloc = 1;
1293             } else {
1294                 error_report("Invalid preallocation mode: '%s'",
1295                              options->value.s);
1296                 ret = -EINVAL;
1297                 goto out;
1298             }
1299         }
1300         options++;
1301     }
1302
1303     if (vdi_size > SD_MAX_VDI_SIZE) {
1304         error_report("too big image size");
1305         ret = -EINVAL;
1306         goto out;
1307     }
1308
1309     if (backing_file) {
1310         BlockDriverState *bs;
1311         BDRVSheepdogState *s;
1312         BlockDriver *drv;
1313
1314         /* Currently, only Sheepdog backing image is supported. */
1315         drv = bdrv_find_protocol(backing_file);
1316         if (!drv || strcmp(drv->protocol_name, "sheepdog") != 0) {
1317             error_report("backing_file must be a sheepdog image");
1318             ret = -EINVAL;
1319             goto out;
1320         }
1321
1322         ret = bdrv_file_open(&bs, backing_file, 0);
1323         if (ret < 0) {
1324             goto out;
1325         }
1326
1327         s = bs->opaque;
1328
1329         if (!is_snapshot(&s->inode)) {
1330             error_report("cannot clone from a non snapshot vdi");
1331             bdrv_delete(bs);
1332             ret = -EINVAL;
1333             goto out;
1334         }
1335
1336         base_vid = s->inode.vdi_id;
1337         bdrv_delete(bs);
1338     }
1339
1340     ret = do_sd_create(vdi, vdi_size, base_vid, &vid, 0, s->addr, s->port);
1341     if (!prealloc || ret) {
1342         goto out;
1343     }
1344
1345     ret = sd_prealloc(filename);
1346 out:
1347     g_free(s);
1348     return ret;
1349 }
1350
1351 static void sd_close(BlockDriverState *bs)
1352 {
1353     BDRVSheepdogState *s = bs->opaque;
1354     SheepdogVdiReq hdr;
1355     SheepdogVdiRsp *rsp = (SheepdogVdiRsp *)&hdr;
1356     unsigned int wlen, rlen = 0;
1357     int fd, ret;
1358
1359     dprintf("%s\n", s->name);
1360
1361     fd = connect_to_sdog(s->addr, s->port);
1362     if (fd < 0) {
1363         return;
1364     }
1365
1366     memset(&hdr, 0, sizeof(hdr));
1367
1368     hdr.opcode = SD_OP_RELEASE_VDI;
1369     wlen = strlen(s->name) + 1;
1370     hdr.data_length = wlen;
1371     hdr.flags = SD_FLAG_CMD_WRITE;
1372
1373     ret = do_req(fd, (SheepdogReq *)&hdr, s->name, &wlen, &rlen);
1374
1375     closesocket(fd);
1376
1377     if (!ret && rsp->result != SD_RES_SUCCESS &&
1378         rsp->result != SD_RES_VDI_NOT_LOCKED) {
1379         error_report("%s, %s", sd_strerror(rsp->result), s->name);
1380     }
1381
1382     qemu_aio_set_fd_handler(s->fd, NULL, NULL, NULL, NULL);
1383     closesocket(s->fd);
1384     if (s->cache_enabled) {
1385         closesocket(s->flush_fd);
1386     }
1387     g_free(s->addr);
1388 }
1389
1390 static int64_t sd_getlength(BlockDriverState *bs)
1391 {
1392     BDRVSheepdogState *s = bs->opaque;
1393
1394     return s->inode.vdi_size;
1395 }
1396
1397 static int sd_truncate(BlockDriverState *bs, int64_t offset)
1398 {
1399     BDRVSheepdogState *s = bs->opaque;
1400     int ret, fd;
1401     unsigned int datalen;
1402
1403     if (offset < s->inode.vdi_size) {
1404         error_report("shrinking is not supported");
1405         return -EINVAL;
1406     } else if (offset > SD_MAX_VDI_SIZE) {
1407         error_report("too big image size");
1408         return -EINVAL;
1409     }
1410
1411     fd = connect_to_sdog(s->addr, s->port);
1412     if (fd < 0) {
1413         return fd;
1414     }
1415
1416     /* we don't need to update entire object */
1417     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
1418     s->inode.vdi_size = offset;
1419     ret = write_object(fd, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id),
1420                        s->inode.nr_copies, datalen, 0, 0, s->cache_enabled);
1421     close(fd);
1422
1423     if (ret < 0) {
1424         error_report("failed to update an inode.");
1425     }
1426
1427     return ret;
1428 }
1429
1430 /*
1431  * This function is called after writing data objects.  If we need to
1432  * update metadata, this sends a write request to the vdi object.
1433  * Otherwise, this switches back to sd_co_readv/writev.
1434  */
1435 static void coroutine_fn sd_write_done(SheepdogAIOCB *acb)
1436 {
1437     int ret;
1438     BDRVSheepdogState *s = acb->common.bs->opaque;
1439     struct iovec iov;
1440     AIOReq *aio_req;
1441     uint32_t offset, data_len, mn, mx;
1442
1443     mn = s->min_dirty_data_idx;
1444     mx = s->max_dirty_data_idx;
1445     if (mn <= mx) {
1446         /* we need to update the vdi object. */
1447         offset = sizeof(s->inode) - sizeof(s->inode.data_vdi_id) +
1448             mn * sizeof(s->inode.data_vdi_id[0]);
1449         data_len = (mx - mn + 1) * sizeof(s->inode.data_vdi_id[0]);
1450
1451         s->min_dirty_data_idx = UINT32_MAX;
1452         s->max_dirty_data_idx = 0;
1453
1454         iov.iov_base = &s->inode;
1455         iov.iov_len = sizeof(s->inode);
1456         aio_req = alloc_aio_req(s, acb, vid_to_vdi_oid(s->inode.vdi_id),
1457                                 data_len, offset, 0, 0, offset);
1458         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1459         ret = add_aio_request(s, aio_req, &iov, 1, 0, AIOCB_WRITE_UDATA);
1460         if (ret) {
1461             free_aio_req(s, aio_req);
1462             acb->ret = -EIO;
1463             goto out;
1464         }
1465
1466         acb->aio_done_func = sd_finish_aiocb;
1467         acb->aiocb_type = AIOCB_WRITE_UDATA;
1468         return;
1469     }
1470 out:
1471     sd_finish_aiocb(acb);
1472 }
1473
1474 /*
1475  * Create a writable VDI from a snapshot
1476  */
1477 static int sd_create_branch(BDRVSheepdogState *s)
1478 {
1479     int ret, fd;
1480     uint32_t vid;
1481     char *buf;
1482
1483     dprintf("%" PRIx32 " is snapshot.\n", s->inode.vdi_id);
1484
1485     buf = g_malloc(SD_INODE_SIZE);
1486
1487     ret = do_sd_create(s->name, s->inode.vdi_size, s->inode.vdi_id, &vid, 1,
1488                        s->addr, s->port);
1489     if (ret) {
1490         goto out;
1491     }
1492
1493     dprintf("%" PRIx32 " is created.\n", vid);
1494
1495     fd = connect_to_sdog(s->addr, s->port);
1496     if (fd < 0) {
1497         error_report("failed to connect");
1498         ret = fd;
1499         goto out;
1500     }
1501
1502     ret = read_object(fd, buf, vid_to_vdi_oid(vid), s->inode.nr_copies,
1503                       SD_INODE_SIZE, 0, s->cache_enabled);
1504
1505     closesocket(fd);
1506
1507     if (ret < 0) {
1508         goto out;
1509     }
1510
1511     memcpy(&s->inode, buf, sizeof(s->inode));
1512
1513     s->is_snapshot = 0;
1514     ret = 0;
1515     dprintf("%" PRIx32 " was newly created.\n", s->inode.vdi_id);
1516
1517 out:
1518     g_free(buf);
1519
1520     return ret;
1521 }
1522
1523 /*
1524  * Send I/O requests to the server.
1525  *
1526  * This function sends requests to the server, links the requests to
1527  * the inflight_list in BDRVSheepdogState, and exits without
1528  * waiting the response.  The responses are received in the
1529  * `aio_read_response' function which is called from the main loop as
1530  * a fd handler.
1531  *
1532  * Returns 1 when we need to wait a response, 0 when there is no sent
1533  * request and -errno in error cases.
1534  */
1535 static int coroutine_fn sd_co_rw_vector(void *p)
1536 {
1537     SheepdogAIOCB *acb = p;
1538     int ret = 0;
1539     unsigned long len, done = 0, total = acb->nb_sectors * SECTOR_SIZE;
1540     unsigned long idx = acb->sector_num * SECTOR_SIZE / SD_DATA_OBJ_SIZE;
1541     uint64_t oid;
1542     uint64_t offset = (acb->sector_num * SECTOR_SIZE) % SD_DATA_OBJ_SIZE;
1543     BDRVSheepdogState *s = acb->common.bs->opaque;
1544     SheepdogInode *inode = &s->inode;
1545     AIOReq *aio_req;
1546
1547     if (acb->aiocb_type == AIOCB_WRITE_UDATA && s->is_snapshot) {
1548         /*
1549          * In the case we open the snapshot VDI, Sheepdog creates the
1550          * writable VDI when we do a write operation first.
1551          */
1552         ret = sd_create_branch(s);
1553         if (ret) {
1554             acb->ret = -EIO;
1555             goto out;
1556         }
1557     }
1558
1559     /*
1560      * Make sure we don't free the aiocb before we are done with all requests.
1561      * This additional reference is dropped at the end of this function.
1562      */
1563     acb->nr_pending++;
1564
1565     while (done != total) {
1566         uint8_t flags = 0;
1567         uint64_t old_oid = 0;
1568         int create = 0;
1569
1570         oid = vid_to_data_oid(inode->data_vdi_id[idx], idx);
1571
1572         len = MIN(total - done, SD_DATA_OBJ_SIZE - offset);
1573
1574         switch (acb->aiocb_type) {
1575         case AIOCB_READ_UDATA:
1576             if (!inode->data_vdi_id[idx]) {
1577                 qemu_iovec_memset(acb->qiov, done, 0, len);
1578                 goto done;
1579             }
1580             break;
1581         case AIOCB_WRITE_UDATA:
1582             if (!inode->data_vdi_id[idx]) {
1583                 create = 1;
1584             } else if (!is_data_obj_writable(inode, idx)) {
1585                 /* Copy-On-Write */
1586                 create = 1;
1587                 old_oid = oid;
1588                 flags = SD_FLAG_CMD_COW;
1589             }
1590             break;
1591         default:
1592             break;
1593         }
1594
1595         if (create) {
1596             dprintf("update ino (%" PRIu32 ") %" PRIu64 " %" PRIu64 " %ld\n",
1597                     inode->vdi_id, oid,
1598                     vid_to_data_oid(inode->data_vdi_id[idx], idx), idx);
1599             oid = vid_to_data_oid(inode->vdi_id, idx);
1600             dprintf("new oid %" PRIx64 "\n", oid);
1601         }
1602
1603         aio_req = alloc_aio_req(s, acb, oid, len, offset, flags, old_oid, done);
1604
1605         if (create) {
1606             AIOReq *areq;
1607             QLIST_FOREACH(areq, &s->inflight_aio_head, aio_siblings) {
1608                 if (areq->oid == oid) {
1609                     /*
1610                      * Sheepdog cannot handle simultaneous create
1611                      * requests to the same object.  So we cannot send
1612                      * the request until the previous request
1613                      * finishes.
1614                      */
1615                     aio_req->flags = 0;
1616                     aio_req->base_oid = 0;
1617                     QLIST_INSERT_HEAD(&s->pending_aio_head, aio_req,
1618                                       aio_siblings);
1619                     goto done;
1620                 }
1621             }
1622         }
1623
1624         QLIST_INSERT_HEAD(&s->inflight_aio_head, aio_req, aio_siblings);
1625         ret = add_aio_request(s, aio_req, acb->qiov->iov, acb->qiov->niov,
1626                               create, acb->aiocb_type);
1627         if (ret < 0) {
1628             error_report("add_aio_request is failed");
1629             free_aio_req(s, aio_req);
1630             acb->ret = -EIO;
1631             goto out;
1632         }
1633     done:
1634         offset = 0;
1635         idx++;
1636         done += len;
1637     }
1638 out:
1639     if (!--acb->nr_pending) {
1640         return acb->ret;
1641     }
1642     return 1;
1643 }
1644
1645 static coroutine_fn int sd_co_writev(BlockDriverState *bs, int64_t sector_num,
1646                         int nb_sectors, QEMUIOVector *qiov)
1647 {
1648     SheepdogAIOCB *acb;
1649     int ret;
1650
1651     if (bs->growable && sector_num + nb_sectors > bs->total_sectors) {
1652         ret = sd_truncate(bs, (sector_num + nb_sectors) * SECTOR_SIZE);
1653         if (ret < 0) {
1654             return ret;
1655         }
1656         bs->total_sectors = sector_num + nb_sectors;
1657     }
1658
1659     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors, NULL, NULL);
1660     acb->aio_done_func = sd_write_done;
1661     acb->aiocb_type = AIOCB_WRITE_UDATA;
1662
1663     ret = sd_co_rw_vector(acb);
1664     if (ret <= 0) {
1665         qemu_aio_release(acb);
1666         return ret;
1667     }
1668
1669     qemu_coroutine_yield();
1670
1671     return acb->ret;
1672 }
1673
1674 static coroutine_fn int sd_co_readv(BlockDriverState *bs, int64_t sector_num,
1675                        int nb_sectors, QEMUIOVector *qiov)
1676 {
1677     SheepdogAIOCB *acb;
1678     int ret;
1679
1680     acb = sd_aio_setup(bs, qiov, sector_num, nb_sectors, NULL, NULL);
1681     acb->aiocb_type = AIOCB_READ_UDATA;
1682     acb->aio_done_func = sd_finish_aiocb;
1683
1684     ret = sd_co_rw_vector(acb);
1685     if (ret <= 0) {
1686         qemu_aio_release(acb);
1687         return ret;
1688     }
1689
1690     qemu_coroutine_yield();
1691
1692     return acb->ret;
1693 }
1694
1695 static int coroutine_fn sd_co_flush_to_disk(BlockDriverState *bs)
1696 {
1697     BDRVSheepdogState *s = bs->opaque;
1698     SheepdogObjReq hdr = { 0 };
1699     SheepdogObjRsp *rsp = (SheepdogObjRsp *)&hdr;
1700     SheepdogInode *inode = &s->inode;
1701     int ret;
1702     unsigned int wlen = 0, rlen = 0;
1703
1704     if (!s->cache_enabled) {
1705         return 0;
1706     }
1707
1708     hdr.opcode = SD_OP_FLUSH_VDI;
1709     hdr.oid = vid_to_vdi_oid(inode->vdi_id);
1710
1711     ret = do_req(s->flush_fd, (SheepdogReq *)&hdr, NULL, &wlen, &rlen);
1712     if (ret) {
1713         error_report("failed to send a request to the sheep");
1714         return ret;
1715     }
1716
1717     if (rsp->result == SD_RES_INVALID_PARMS) {
1718         dprintf("disable write cache since the server doesn't support it\n");
1719
1720         s->cache_enabled = 0;
1721         closesocket(s->flush_fd);
1722         return 0;
1723     }
1724
1725     if (rsp->result != SD_RES_SUCCESS) {
1726         error_report("%s", sd_strerror(rsp->result));
1727         return -EIO;
1728     }
1729
1730     return 0;
1731 }
1732
1733 static int sd_snapshot_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info)
1734 {
1735     BDRVSheepdogState *s = bs->opaque;
1736     int ret, fd;
1737     uint32_t new_vid;
1738     SheepdogInode *inode;
1739     unsigned int datalen;
1740
1741     dprintf("sn_info: name %s id_str %s s: name %s vm_state_size %" PRId64 " "
1742             "is_snapshot %d\n", sn_info->name, sn_info->id_str,
1743             s->name, sn_info->vm_state_size, s->is_snapshot);
1744
1745     if (s->is_snapshot) {
1746         error_report("You can't create a snapshot of a snapshot VDI, "
1747                      "%s (%" PRIu32 ").", s->name, s->inode.vdi_id);
1748
1749         return -EINVAL;
1750     }
1751
1752     dprintf("%s %s\n", sn_info->name, sn_info->id_str);
1753
1754     s->inode.vm_state_size = sn_info->vm_state_size;
1755     s->inode.vm_clock_nsec = sn_info->vm_clock_nsec;
1756     strncpy(s->inode.tag, sn_info->name, sizeof(s->inode.tag));
1757     /* we don't need to update entire object */
1758     datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id);
1759
1760     /* refresh inode. */
1761     fd = connect_to_sdog(s->addr, s->port);
1762     if (fd < 0) {
1763         ret = fd;
1764         goto cleanup;
1765     }
1766
1767     ret = write_object(fd, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id),
1768                        s->inode.nr_copies, datalen, 0, 0, s->cache_enabled);
1769     if (ret < 0) {
1770         error_report("failed to write snapshot's inode.");
1771         goto cleanup;
1772     }
1773
1774     ret = do_sd_create(s->name, s->inode.vdi_size, s->inode.vdi_id, &new_vid, 1,
1775                        s->addr, s->port);
1776     if (ret < 0) {
1777         error_report("failed to create inode for snapshot. %s",
1778                      strerror(errno));
1779         goto cleanup;
1780     }
1781
1782     inode = (SheepdogInode *)g_malloc(datalen);
1783
1784     ret = read_object(fd, (char *)inode, vid_to_vdi_oid(new_vid),
1785                       s->inode.nr_copies, datalen, 0, s->cache_enabled);
1786
1787     if (ret < 0) {
1788         error_report("failed to read new inode info. %s", strerror(errno));
1789         goto cleanup;
1790     }
1791
1792     memcpy(&s->inode, inode, datalen);
1793     dprintf("s->inode: name %s snap_id %x oid %x\n",
1794             s->inode.name, s->inode.snap_id, s->inode.vdi_id);
1795
1796 cleanup:
1797     closesocket(fd);
1798     return ret;
1799 }
1800
1801 static int sd_snapshot_goto(BlockDriverState *bs, const char *snapshot_id)
1802 {
1803     BDRVSheepdogState *s = bs->opaque;
1804     BDRVSheepdogState *old_s;
1805     char vdi[SD_MAX_VDI_LEN], tag[SD_MAX_VDI_TAG_LEN];
1806     char *buf = NULL;
1807     uint32_t vid;
1808     uint32_t snapid = 0;
1809     int ret = 0, fd;
1810
1811     old_s = g_malloc(sizeof(BDRVSheepdogState));
1812
1813     memcpy(old_s, s, sizeof(BDRVSheepdogState));
1814
1815     memset(vdi, 0, sizeof(vdi));
1816     strncpy(vdi, s->name, sizeof(vdi));
1817
1818     memset(tag, 0, sizeof(tag));
1819     snapid = strtoul(snapshot_id, NULL, 10);
1820     if (!snapid) {
1821         strncpy(tag, s->name, sizeof(tag));
1822     }
1823
1824     ret = find_vdi_name(s, vdi, snapid, tag, &vid, 1);
1825     if (ret) {
1826         error_report("Failed to find_vdi_name");
1827         goto out;
1828     }
1829
1830     fd = connect_to_sdog(s->addr, s->port);
1831     if (fd < 0) {
1832         error_report("failed to connect");
1833         ret = fd;
1834         goto out;
1835     }
1836
1837     buf = g_malloc(SD_INODE_SIZE);
1838     ret = read_object(fd, buf, vid_to_vdi_oid(vid), s->inode.nr_copies,
1839                       SD_INODE_SIZE, 0, s->cache_enabled);
1840
1841     closesocket(fd);
1842
1843     if (ret) {
1844         goto out;
1845     }
1846
1847     memcpy(&s->inode, buf, sizeof(s->inode));
1848
1849     if (!s->inode.vm_state_size) {
1850         error_report("Invalid snapshot");
1851         ret = -ENOENT;
1852         goto out;
1853     }
1854
1855     s->is_snapshot = 1;
1856
1857     g_free(buf);
1858     g_free(old_s);
1859
1860     return 0;
1861 out:
1862     /* recover bdrv_sd_state */
1863     memcpy(s, old_s, sizeof(BDRVSheepdogState));
1864     g_free(buf);
1865     g_free(old_s);
1866
1867     error_report("failed to open. recover old bdrv_sd_state.");
1868
1869     return ret;
1870 }
1871
1872 static int sd_snapshot_delete(BlockDriverState *bs, const char *snapshot_id)
1873 {
1874     /* FIXME: Delete specified snapshot id.  */
1875     return 0;
1876 }
1877
1878 static int sd_snapshot_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab)
1879 {
1880     BDRVSheepdogState *s = bs->opaque;
1881     SheepdogReq req;
1882     int fd, nr = 1024, ret, max = BITS_TO_LONGS(SD_NR_VDIS) * sizeof(long);
1883     QEMUSnapshotInfo *sn_tab = NULL;
1884     unsigned wlen, rlen;
1885     int found = 0;
1886     static SheepdogInode inode;
1887     unsigned long *vdi_inuse;
1888     unsigned int start_nr;
1889     uint64_t hval;
1890     uint32_t vid;
1891
1892     vdi_inuse = g_malloc(max);
1893
1894     fd = connect_to_sdog(s->addr, s->port);
1895     if (fd < 0) {
1896         ret = fd;
1897         goto out;
1898     }
1899
1900     rlen = max;
1901     wlen = 0;
1902
1903     memset(&req, 0, sizeof(req));
1904
1905     req.opcode = SD_OP_READ_VDIS;
1906     req.data_length = max;
1907
1908     ret = do_req(fd, (SheepdogReq *)&req, vdi_inuse, &wlen, &rlen);
1909
1910     closesocket(fd);
1911     if (ret) {
1912         goto out;
1913     }
1914
1915     sn_tab = g_malloc0(nr * sizeof(*sn_tab));
1916
1917     /* calculate a vdi id with hash function */
1918     hval = fnv_64a_buf(s->name, strlen(s->name), FNV1A_64_INIT);
1919     start_nr = hval & (SD_NR_VDIS - 1);
1920
1921     fd = connect_to_sdog(s->addr, s->port);
1922     if (fd < 0) {
1923         error_report("failed to connect");
1924         ret = fd;
1925         goto out;
1926     }
1927
1928     for (vid = start_nr; found < nr; vid = (vid + 1) % SD_NR_VDIS) {
1929         if (!test_bit(vid, vdi_inuse)) {
1930             break;
1931         }
1932
1933         /* we don't need to read entire object */
1934         ret = read_object(fd, (char *)&inode, vid_to_vdi_oid(vid),
1935                           0, SD_INODE_SIZE - sizeof(inode.data_vdi_id), 0,
1936                           s->cache_enabled);
1937
1938         if (ret) {
1939             continue;
1940         }
1941
1942         if (!strcmp(inode.name, s->name) && is_snapshot(&inode)) {
1943             sn_tab[found].date_sec = inode.snap_ctime >> 32;
1944             sn_tab[found].date_nsec = inode.snap_ctime & 0xffffffff;
1945             sn_tab[found].vm_state_size = inode.vm_state_size;
1946             sn_tab[found].vm_clock_nsec = inode.vm_clock_nsec;
1947
1948             snprintf(sn_tab[found].id_str, sizeof(sn_tab[found].id_str), "%u",
1949                      inode.snap_id);
1950             strncpy(sn_tab[found].name, inode.tag,
1951                     MIN(sizeof(sn_tab[found].name), sizeof(inode.tag)));
1952             found++;
1953         }
1954     }
1955
1956     closesocket(fd);
1957 out:
1958     *psn_tab = sn_tab;
1959
1960     g_free(vdi_inuse);
1961
1962     if (ret < 0) {
1963         return ret;
1964     }
1965
1966     return found;
1967 }
1968
1969 static int do_load_save_vmstate(BDRVSheepdogState *s, uint8_t *data,
1970                                 int64_t pos, int size, int load)
1971 {
1972     int fd, create;
1973     int ret = 0, remaining = size;
1974     unsigned int data_len;
1975     uint64_t vmstate_oid;
1976     uint32_t vdi_index;
1977     uint64_t offset;
1978
1979     fd = connect_to_sdog(s->addr, s->port);
1980     if (fd < 0) {
1981         return fd;
1982     }
1983
1984     while (remaining) {
1985         vdi_index = pos / SD_DATA_OBJ_SIZE;
1986         offset = pos % SD_DATA_OBJ_SIZE;
1987
1988         data_len = MIN(remaining, SD_DATA_OBJ_SIZE);
1989
1990         vmstate_oid = vid_to_vmstate_oid(s->inode.vdi_id, vdi_index);
1991
1992         create = (offset == 0);
1993         if (load) {
1994             ret = read_object(fd, (char *)data, vmstate_oid,
1995                               s->inode.nr_copies, data_len, offset,
1996                               s->cache_enabled);
1997         } else {
1998             ret = write_object(fd, (char *)data, vmstate_oid,
1999                                s->inode.nr_copies, data_len, offset, create,
2000                                s->cache_enabled);
2001         }
2002
2003         if (ret < 0) {
2004             error_report("failed to save vmstate %s", strerror(errno));
2005             goto cleanup;
2006         }
2007
2008         pos += data_len;
2009         remaining -= data_len;
2010     }
2011     ret = size;
2012 cleanup:
2013     closesocket(fd);
2014     return ret;
2015 }
2016
2017 static int sd_save_vmstate(BlockDriverState *bs, const uint8_t *data,
2018                            int64_t pos, int size)
2019 {
2020     BDRVSheepdogState *s = bs->opaque;
2021
2022     return do_load_save_vmstate(s, (uint8_t *)data, pos, size, 0);
2023 }
2024
2025 static int sd_load_vmstate(BlockDriverState *bs, uint8_t *data,
2026                            int64_t pos, int size)
2027 {
2028     BDRVSheepdogState *s = bs->opaque;
2029
2030     return do_load_save_vmstate(s, data, pos, size, 1);
2031 }
2032
2033
2034 static QEMUOptionParameter sd_create_options[] = {
2035     {
2036         .name = BLOCK_OPT_SIZE,
2037         .type = OPT_SIZE,
2038         .help = "Virtual disk size"
2039     },
2040     {
2041         .name = BLOCK_OPT_BACKING_FILE,
2042         .type = OPT_STRING,
2043         .help = "File name of a base image"
2044     },
2045     {
2046         .name = BLOCK_OPT_PREALLOC,
2047         .type = OPT_STRING,
2048         .help = "Preallocation mode (allowed values: off, full)"
2049     },
2050     { NULL }
2051 };
2052
2053 BlockDriver bdrv_sheepdog = {
2054     .format_name    = "sheepdog",
2055     .protocol_name  = "sheepdog",
2056     .instance_size  = sizeof(BDRVSheepdogState),
2057     .bdrv_file_open = sd_open,
2058     .bdrv_close     = sd_close,
2059     .bdrv_create    = sd_create,
2060     .bdrv_getlength = sd_getlength,
2061     .bdrv_truncate  = sd_truncate,
2062
2063     .bdrv_co_readv  = sd_co_readv,
2064     .bdrv_co_writev = sd_co_writev,
2065     .bdrv_co_flush_to_disk  = sd_co_flush_to_disk,
2066
2067     .bdrv_snapshot_create   = sd_snapshot_create,
2068     .bdrv_snapshot_goto     = sd_snapshot_goto,
2069     .bdrv_snapshot_delete   = sd_snapshot_delete,
2070     .bdrv_snapshot_list     = sd_snapshot_list,
2071
2072     .bdrv_save_vmstate  = sd_save_vmstate,
2073     .bdrv_load_vmstate  = sd_load_vmstate,
2074
2075     .create_options = sd_create_options,
2076 };
2077
2078 static void bdrv_sheepdog_init(void)
2079 {
2080     bdrv_register(&bdrv_sheepdog);
2081 }
2082 block_init(bdrv_sheepdog_init);