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