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