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