[Qemu-devel] [PATCH v1 09/18] block/pcache: add AIO readahead

[Pavel Butsykin]

This patch adds readahead data to the cache. Here the readahead is a separate
asynchronous request, which doesn't depend on completion of filtered read
requests. The readahead is done only by the condition, if before the current
request there's sequential read data enough size. This information can give
the request statistics, of course this method of detection is not very reliable,
but in most cases it'll work.

Signed-off-by: Pavel Butsykin <address@hidden>
---
 block/pcache.c | 229 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 227 insertions(+), 2 deletions(-)

diff --git a/block/pcache.c b/block/pcache.c
index dd598f3..3717037 100644
--- a/block/pcache.c
+++ b/block/pcache.c
@@ -17,6 +17,8 @@
 
 #define PCACHE_OPT_STATS_SIZE "pcache-stats-size"
 #define PCACHE_OPT_MAX_AIO_SIZE "pcache-max-aio-size"
+#define PCACHE_OPT_CACHE_SIZE "pcache-full-size"
+#define PCACHE_OPT_READAHEAD_SIZE "pcache-readahead-size"
 
 static QemuOptsList runtime_opts = {
     .name = "pcache",
@@ -37,6 +39,16 @@ static QemuOptsList runtime_opts = {
             .type = QEMU_OPT_SIZE,
             .help = "Maximum size of aio which is handled by pcache",
         },
+        {
+            .name = PCACHE_OPT_CACHE_SIZE,
+            .type = QEMU_OPT_SIZE,
+            .help = "Total cache size",
+        },
+        {
+            .name = PCACHE_OPT_READAHEAD_SIZE,
+            .type = QEMU_OPT_SIZE,
+            .help = "Prefetch cache readahead size",
+        },
         { /* end of list */ }
     },
 };
@@ -45,17 +57,61 @@ static QemuOptsList runtime_opts = {
 #define MB_BITS 20
 #define PCACHE_DEFAULT_STATS_SIZE (3 << MB_BITS)
 #define PCACHE_DEFAULT_MAX_AIO_SIZE (64 << KB_BITS)
+#define PCACHE_DEFAULT_CACHE_SIZE (4 << MB_BITS)
+#define PCACHE_DEFAULT_READAHEAD_SIZE (128 << KB_BITS)
 
 typedef struct BDRVPCacheState {
     RBCache *req_stats;
+    RBCache *cache;
     uint64_t max_aio_size;
+    uint64_t readahead_size;
 } BDRVPCacheState;
 
+typedef struct PCacheNode {
+    RBCacheNode common;
+    uint8_t *data;
+    enum {
+        NODE_STATUS_NEW       = 0,
+        NODE_STATUS_INFLIGHT  = 1,
+        NODE_STATUS_COMPLETED = 2,
+        NODE_STATUS_REMOVE    = 3,
+        NODE_STATUS_DELETED   = 4, /* only for debugging */
+    } status;
+    int ref;
+} PCacheNode;
+
 typedef struct PCacheAIOCB {
+    BlockDriverState *bs;
     Coroutine *co;
+    uint64_t offset;
+    uint64_t bytes;
     int ret;
 } PCacheAIOCB;
 
+typedef struct PCacheAIOCBReadahead {
+    BlockDriverState *bs;
+    Coroutine *co;
+    QEMUIOVector qiov;
+    PCacheNode *node;
+} PCacheAIOCBReadahead;
+
+static inline void pcache_node_ref(PCacheNode *node)
+{
+    node->ref++;
+}
+
+static void pcache_node_unref(PCacheNode *node)
+{
+    assert(node->ref > 0);
+    if (--node->ref == 0) {
+        assert(node->status == NODE_STATUS_REMOVE);
+        node->status = NODE_STATUS_DELETED;
+
+        g_free(node->data);
+        g_free(node);
+    }
+}
+
 static void pcache_aio_cb(void *opaque, int ret)
 {
     PCacheAIOCB *acb = opaque;
@@ -64,6 +120,27 @@ static void pcache_aio_cb(void *opaque, int ret)
     qemu_coroutine_enter(acb->co);
 }
 
+static void pcache_aio_readahead_cb(void *opaque, int ret)
+{
+    PCacheAIOCBReadahead *acb = opaque;
+    PCacheNode *node = acb->node;
+
+    assert(node->status == NODE_STATUS_INFLIGHT ||
+           node->status == NODE_STATUS_REMOVE);
+
+    if (node->status == NODE_STATUS_INFLIGHT) {
+        if (ret == 0) {
+            node->status = NODE_STATUS_COMPLETED;
+        } else {
+            BDRVPCacheState *s = acb->bs->opaque;
+            rbcache_remove(s->cache, &node->common);
+        }
+    }
+    pcache_node_unref(node);
+
+    qemu_coroutine_enter(acb->co);
+}
+
 static void update_req_stats(RBCache *rbcache, uint64_t offset, uint64_t bytes)
 {
     do {
@@ -98,6 +175,138 @@ static void update_req_stats(RBCache *rbcache, uint64_t 
offset, uint64_t bytes)
     } while (true);
 }
 
+static bool check_request_sequence(BDRVPCacheState *s, uint64_t offset)
+{
+    uint64_t cache_line_size = s->readahead_size;
+    uint64_t check_offset;
+
+    if (offset <= cache_line_size) {
+        return false;
+    }
+    check_offset = offset - cache_line_size;
+
+    do {
+        RBCacheNode *node = rbcache_search(s->req_stats, check_offset,
+                                           offset - check_offset);
+        if (node == NULL) {
+            return false;
+        }
+        if (node->offset > check_offset) {
+            return false;
+        }
+        check_offset = node->offset + node->bytes;
+    } while (check_offset < offset);
+
+    return true;
+}
+
+static void pcache_node_free(RBCacheNode *rbnode, void *opaque)
+{
+    PCacheNode *node = container_of(rbnode, PCacheNode, common);
+
+    assert(node->status == NODE_STATUS_INFLIGHT ||
+           node->status == NODE_STATUS_COMPLETED);
+
+    node->status = NODE_STATUS_REMOVE;
+    pcache_node_unref(node);
+}
+
+static RBCacheNode *pcache_node_alloc(uint64_t offset, uint64_t bytes,
+                                      void *opaque)
+{
+    PCacheNode *node = g_malloc(sizeof(*node));
+
+    node->data = g_malloc(bytes);
+    node->status = NODE_STATUS_NEW;
+    node->ref = 1;
+
+    return &node->common;
+}
+
+#define PCACHE_STEPS_FORWARD 2
+
+static PCacheNode *get_readahead_node(BlockDriverState *bs, RBCache *rbcache,
+                                      uint64_t offset, uint64_t bytes)
+{
+    uint32_t count = PCACHE_STEPS_FORWARD;
+
+    int64_t total_bytes = bdrv_getlength(bs);
+    if (total_bytes < 0) {
+        return NULL;
+    }
+
+    while (count--) {
+        PCacheNode *node;
+
+        if (total_bytes <= offset + bytes) {
+            break;
+        }
+
+        node = rbcache_search_and_insert(rbcache, offset, bytes);
+        if (node->status == NODE_STATUS_NEW) {
+            return node;
+        }
+         /* The range less than the readahead size is not cached to reduce
+          * fragmentation of the cache. If the data is already cached, then we
+          * just step over it.
+          */
+        if (offset <= node->common.offset && !count--) {
+            break;
+        }
+        offset = node->common.offset + node->common.bytes;
+    };
+
+    return NULL;
+}
+
+static void coroutine_fn pcache_co_readahead(void *opaque)
+{
+    PCacheAIOCB *acb = g_memdup(opaque, sizeof(*acb));
+    BlockDriverState *bs = acb->bs;
+    BDRVPCacheState *s = bs->opaque;
+    uint64_t offset;
+    uint64_t bytes;
+    PCacheAIOCBReadahead readahead_acb;
+    PCacheNode *node;
+
+    if (!check_request_sequence(s, acb->offset)) {
+        goto out;
+    }
+
+    offset = acb->offset + acb->bytes;
+    bytes = s->readahead_size;
+
+    node = get_readahead_node(bs, s->cache, offset, bytes);
+    if (node == NULL) {
+        goto out;
+    }
+
+    readahead_acb = (PCacheAIOCBReadahead) {
+        .co = qemu_coroutine_self(),
+        .bs = bs,
+        .node = node,
+    };
+
+    node->status = NODE_STATUS_INFLIGHT;
+    qemu_iovec_init(&readahead_acb.qiov, 1);
+    qemu_iovec_add(&readahead_acb.qiov, node->data, node->common.bytes);
+
+    pcache_node_ref(node);
+
+    bdrv_aio_preadv(bs->file, node->common.offset, &readahead_acb.qiov,
+                    node->common.bytes, pcache_aio_readahead_cb,
+                    &readahead_acb);
+    qemu_coroutine_yield();
+out:
+    free(acb);
+}
+
+static void pcache_readahead_request(PCacheAIOCB *acb)
+{
+    Coroutine *co = qemu_coroutine_create(pcache_co_readahead, acb);
+    qemu_coroutine_enter(co);
+}
+
 static coroutine_fn int pcache_co_preadv(BlockDriverState *bs, uint64_t offset,
                                          uint64_t bytes, QEMUIOVector *qiov,
                                          int flags)
@@ -105,14 +314,23 @@ static coroutine_fn int pcache_co_preadv(BlockDriverState 
*bs, uint64_t offset,
     BDRVPCacheState *s = bs->opaque;
     PCacheAIOCB acb = {
         .co = qemu_coroutine_self(),
+        .bs = bs,
+        .offset = offset,
+        .bytes = bytes,
     };
 
-    if (s->max_aio_size >= bytes) {
-        update_req_stats(s->req_stats, offset, bytes);
+    if (bytes > s->max_aio_size) {
+        bdrv_aio_preadv(bs->file, offset, qiov, bytes, pcache_aio_cb, &acb);
+        goto out;
     }
 
+    update_req_stats(s->req_stats, offset, bytes);
+
     bdrv_aio_preadv(bs->file, offset, qiov, bytes, pcache_aio_cb, &acb);
 
+    pcache_readahead_request(&acb);
+
+out:
     qemu_coroutine_yield();
 
     return acb.ret;
@@ -137,10 +355,16 @@ static void pcache_state_init(QemuOpts *opts, 
BDRVPCacheState *s)
 {
     uint64_t stats_size = qemu_opt_get_size(opts, PCACHE_OPT_STATS_SIZE,
                                             PCACHE_DEFAULT_STATS_SIZE);
+    uint64_t cache_size = qemu_opt_get_size(opts, PCACHE_OPT_CACHE_SIZE,
+                                            PCACHE_DEFAULT_CACHE_SIZE);
     s->req_stats = rbcache_create(NULL, NULL, stats_size, RBCACHE_FIFO, s);
 
     s->max_aio_size = qemu_opt_get_size(opts, PCACHE_OPT_MAX_AIO_SIZE,
                                         PCACHE_DEFAULT_MAX_AIO_SIZE);
+    s->cache = rbcache_create(pcache_node_alloc, pcache_node_free, cache_size,
+                              RBCACHE_LRU, s);
+    s->readahead_size = qemu_opt_get_size(opts, PCACHE_OPT_READAHEAD_SIZE,
+                                          PCACHE_DEFAULT_READAHEAD_SIZE);
 }
 
 static int pcache_file_open(BlockDriverState *bs, QDict *options, int flags,
@@ -177,6 +401,7 @@ static void pcache_close(BlockDriverState *bs)
     BDRVPCacheState *s = bs->opaque;
 
     rbcache_destroy(s->req_stats);
+    rbcache_destroy(s->cache);
 }
 
 static void pcache_parse_filename(const char *filename, QDict *options,
-- 
2.10.1