| Message ID | 20210129165030.640169-3-vsementsov@virtuozzo.com |
|---|---|
| State | New |
| Headers | show |
| Series | qcow2: compressed write cache | expand |
On 29.01.21 17:50, Vladimir Sementsov-Ogievskiy wrote: > Compressed writes and O_DIRECT are not friends: they works too slow, > because compressed writes does many small unaligned to 512 writes. > > Let's introduce an internal cache, so that compressed writes may work > well when O_DIRECT is on. > > Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com> > --- > block/qcow2.h | 29 + > block/qcow2-compressed-write-cache.c | 770 +++++++++++++++++++++++++++ > block/meson.build | 1 + > 3 files changed, 800 insertions(+) > create mode 100644 block/qcow2-compressed-write-cache.c > > diff --git a/block/qcow2.h b/block/qcow2.h > index 0678073b74..fbdedf89fa 100644 > --- a/block/qcow2.h > +++ b/block/qcow2.h > @@ -322,6 +322,8 @@ typedef struct Qcow2BitmapHeaderExt { > uint64_t bitmap_directory_offset; > } QEMU_PACKED Qcow2BitmapHeaderExt; > > +typedef struct Qcow2CompressedWriteCache Qcow2CompressedWriteCache; > + > #define QCOW2_MAX_THREADS 4 > > typedef struct BDRVQcow2State { > @@ -1010,4 +1012,31 @@ int coroutine_fn > qcow2_co_decrypt(BlockDriverState *bs, uint64_t host_offset, > uint64_t guest_offset, void *buf, size_t len); > > +Qcow2CompressedWriteCache *qcow2_compressed_cache_new(BdrvChild *data_file, > + int64_t cluster_size, > + int64_t cache_size); > +void qcow2_compressed_cache_free(Qcow2CompressedWriteCache *s); > +int coroutine_fn > +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t offset, > + int64_t bytes, void *buf); > +int coroutine_fn > +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, int64_t offset, > + int64_t bytes, void *buf); > +void coroutine_fn > +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, > + int64_t cluster_data_end); > +int coroutine_fn > +qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s); > +int qcow2_compressed_cache_flush(BlockDriverState *bs, > + Qcow2CompressedWriteCache *state); > +int coroutine_fn > +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s); > +int qcow2_compressed_cache_stop_flush(BlockDriverState *bs, > + Qcow2CompressedWriteCache *s); > +void qcow2_compressed_cache_set_size(Qcow2CompressedWriteCache *s, > + int64_t size); > +void coroutine_fn > +qcow2_compressed_cache_co_discard(Qcow2CompressedWriteCache *s, > + int64_t cluster_offset); > + It would be nice if these functions had their interface documented somewhere. > #endif > diff --git a/block/qcow2-compressed-write-cache.c b/block/qcow2-compressed-write-cache.c > new file mode 100644 > index 0000000000..7bb92cb550 > --- /dev/null > +++ b/block/qcow2-compressed-write-cache.c > @@ -0,0 +1,770 @@ > +/* > + * Write cache for qcow2 compressed writes > + * > + * Copyright (c) 2021 Virtuozzo International GmbH. > + * > + * Permission is hereby granted, free of charge, to any person obtaining a copy > + * of this software and associated documentation files (the "Software"), to deal > + * in the Software without restriction, including without limitation the rights > + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell > + * copies of the Software, and to permit persons to whom the Software is > + * furnished to do so, subject to the following conditions: > + * > + * The above copyright notice and this permission notice shall be included in > + * all copies or substantial portions of the Software. > + * > + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR > + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, > + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL > + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER > + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, > + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN > + * THE SOFTWARE. > + */ > + > +#include "qemu/osdep.h" > + > +#include "block/block_int.h" > +#include "block/block-gen.h" > +#include "qemu/coroutine.h" > +#include "qapi/qapi-events-block-core.h" > +#include "qcow2.h" > + > +typedef struct CacheExtent { > + int64_t offset; > + int64_t bytes; > + void *buf; > + QLIST_ENTRY(CacheExtent) next; > +} CacheExtent; > + > +typedef struct CacheCluster { It isn’t immediately clear what these two structures mean by just their name, because “extent” has no meaning in the context of qcow2. I understand CacheExtent to basically be a compressed cluster, and CacheCluster to be a host cluster. Perhaps their names should reflect that. (OTOH, the Cache* prefix seems unnecessary to me, because these are just local structs.) > + int64_t cluster_offset; > + int64_t n_bytes; /* sum of extents lengths */ > + > + /* > + * data_end: cluster is full if data_end reached and ready to be flushed. > + * data_end is absoluted offset, not relative. *absolute > + */ > + int64_t data_end; > + > + bool in_flight; /* cluster is being flushed now */ > + > + /* > + * waiters: coroutines to wake after flush. No, once in_flight is reset to false, which may happen even if the cluster hasn’t been flushed. As a note that doesn’t really belong anywhere, I don’t like the current use of this queue perfectly well. This is how it is used right now: Some context A sets in_flight and flushes the cluster. Some context B wants to evict the cluster from the cache. It sees in_flight, so it gets into the queue. (If in_flight were not set, B would remove the cluster from the cache.) A finishes flushing, resets in_flight, queues B. Then if flushing succeeded, it will evict the cluster from the cache. If it failed, the cluster remains in the cache. Then B is executed. If the cluster isn’t in the cache anymore, it’s done. If still is, it will remove it, and then be done. My problem is that if B were executed right where it’s enqueued (with queue_restart_all()), we would crash because B would see the cluster in the cache, remove it, and then A would have a dangling reference and remove the (freed) cluster again. So, well, it works, but I personally find it strange to “invoke“ B where you don’t actually want it to run. I think the @waiters queue should only be restarted once A has had the chance to remove the cluster from the cache, or to be more general, once A no longer has a reference to the cluster. tl;dr: I think the @waiters queue should only be restarted once the reference to the cluster is no longer going to be used. > Must be empty when in_flight is > + * false > + */ > + CoQueue waiters; > + > + QTAILQ_ENTRY(CacheCluster) next; > + QLIST_HEAD(, CacheExtent) extents; /* sorted by offset */ > +} CacheCluster; > + > +typedef QTAILQ_HEAD(, CacheCluster) CacheClusterQueue; Why the typedef? It’s used one time. > + > +struct Qcow2CompressedWriteCache { > + /* > + * @data_file and @cluster_size are copied from qcow2 state. Not huge > + * duplications, seems better to avoid accessing the whole qcow2 state > + * instead. Why? What if in the future the data_file child can be changed with reopen? What’s the argument against accessing the whole qcow2 state? That it’s a layering violation, that we shouldn’t access it while the rest uses it? If that’s a concern, perhaps *data_file should be an indirect pointer to the pointer in the BDRVQcow2State. > + */ > + BdrvChild *data_file; > + int64_t cluster_size; > + > + CoQueue waiters; /* coroutines, waiting for free place in the cache */ > + > + /* > + * Cache limits not total number of clusters but total number of active > + * clusters. Active clusters are clusters with non-zero @n_bytes (and > + * therefor non-empty @extents). This is done so we don't need to wait for > + * cache flush qcow2_compressed_cache_set_cluster_end() (which may create > + * cluster with @data_end set but zero @n_bytes), as > + * qcow2_compressed_cache_set_cluster_end() is intended to be called from > + * qcow2 mutex critical section. > + */ > + int nb_active_clusters; > + > + /* > + * If max_active_clusters is 0 it means that cache is inactive: all new > + * writes should fallthrough to data_file immediately. > + */ > + int max_active_clusters; > + > + CacheClusterQueue clusters; > +}; > + > +/* New extent takes ownership of @buf */ > +static CacheExtent *cache_extent_new(int64_t offset, int64_t bytes, void *buf) > +{ > + CacheExtent *e = g_new(CacheExtent, 1); > + > + *e = (CacheExtent) { > + .offset = offset, > + .bytes = bytes, > + .buf = buf, > + }; > + > + return e; > +} > + > +static void cache_extent_free(CacheExtent *e) > +{ > + if (e) { > + g_free(e->buf); > + g_free(e); > + } > +} > + > +static CacheCluster *find_cluster(Qcow2CompressedWriteCache *s, > + int64_t cluster_offset) > +{ > + CacheCluster *c; > + > + assert(QEMU_IS_ALIGNED(cluster_offset, s->cluster_size)); > + > + QTAILQ_FOREACH(c, &s->clusters, next) { > + if (cluster_offset == c->cluster_offset) { > + return c; > + } > + } This reads like in any language but C we’d opt for a hash table. On that thought, why don’t we opt for a hash table even in C? > + > + return NULL; > +} > + > +/* Creates "inactive" cluster, which doesn't influence s->nb_active_clusters */ > +static CacheCluster *cache_cluster_new(Qcow2CompressedWriteCache *s, > + int64_t cluster_offset) > +{ > + CacheCluster *c; > + > + assert(!find_cluster(s, cluster_offset)); > + > + c = g_new(CacheCluster, 1); > + *c = (CacheCluster) { > + .cluster_offset = cluster_offset, > + .data_end = cluster_offset + s->cluster_size > + }; > + > + qemu_co_queue_init(&c->waiters); > + QTAILQ_INSERT_TAIL(&s->clusters, c, next); > + > + return c; > +} > + > +static void cache_cluster_free(CacheCluster *cluster) > +{ > + assert(!cluster->in_flight); > + assert(qemu_co_queue_empty(&cluster->waiters)); > + QLIST_FOREACH_FUNC_SAFE(&cluster->extents, next, cache_extent_free); > + g_free(cluster); > +} > + > +static bool cache_cluster_is_full(CacheCluster *cluster) > +{ > + assert(cluster->cluster_offset + cluster->n_bytes <= cluster->data_end); > + return cluster->cluster_offset + cluster->n_bytes == cluster->data_end; > +} > + > +static void cache_cluster_remove(Qcow2CompressedWriteCache *s, > + CacheCluster *cluster) > +{ > + if (cluster->n_bytes) { > + s->nb_active_clusters--; > + } > + QTAILQ_REMOVE(&s->clusters, cluster, next); > + cache_cluster_free(cluster); > +} > + > +/* > + * Return number of consequtive clusters starting from @first. Store next after %s/consequi\?tive/consecutive/ > + * last extent pointer into @next, store end offset of last extent into > + * @end_off. > + */ > +static int count_consequitive_extents(CacheExtent *first, > + CacheExtent **next, > + int64_t *end_off) > +{ > + CacheExtent *e; > + int64_t off = first->offset; > + int nb = 0; > + > + for (e = first; e; e = QLIST_NEXT(e, next)) { > + assert(e->offset >= off); > + if (e->offset > off) { > + break; > + } > + off += e->bytes; > + nb++; > + } > + > + if (next) { > + *next = e; > + } > + > + if (end_off) { > + *end_off = off; > + } > + > + return nb; > +} > + > +/* > + * Write consequtive extents, starting from @firest. Store next after last *first > + * extent pointer into @next. If align > 1, align end of the whole write by > + * zero. > + */ > +static int coroutine_fn flush_consequitive_extents(Qcow2CompressedWriteCache *s, > + CacheExtent *first, > + CacheExtent **next, > + int64_t align) > +{ > + CacheExtent *e, *end_extent; > + int64_t end; > + int nb_extents = count_consequitive_extents(first, &end_extent, &end); I’d prefer nb_bufs, because it will include a tail if present. > + int64_t aligned_end = QEMU_ALIGN_UP(end, align); > + int64_t tail = aligned_end - end; > + int64_t len = aligned_end - first->offset; > + > + /* > + * Alignment if for flushing full cluster, first extent offset is always s/.*,/The alignment will not exceed the cluster size, so the/? > + * aligned. > + */ > + assert(QEMU_IS_ALIGNED(first->offset, align)); > + > + if (next) { > + *next = end_extent; > + } > + > + if (tail) { > + nb_extents++; > + } > + > + if (nb_extents > IOV_MAX) { > + g_autofree void *buf = g_malloc(len); I wonder if this should be blk_blockalign() if first->offset is aligned to the cluster size. (Don’t know how common that case is.) > + char *p = buf; > + > + for (e = first; e != end_extent; e = QLIST_NEXT(e, next)) { > + memcpy(p, e->buf, e->bytes); > + p += e->bytes; > + } > + > + if (tail) { > + memset(p, 0, tail); > + } > + > + BLKDBG_EVENT(s->data_file, BLKDBG_WRITE_COMPRESSED); > + return bdrv_co_pwrite(s->data_file, first->offset, len, buf, 0); > + } else { > + int ret; > + QEMUIOVector qiov; > + g_autofree void *tail_buf = NULL; > + > + qemu_iovec_init(&qiov, nb_extents); > + for (e = first; e != end_extent; e = QLIST_NEXT(e, next)) { > + qemu_iovec_add(&qiov, e->buf, e->bytes); > + } > + > + if (tail) { > + tail_buf = g_malloc0(tail); > + qemu_iovec_add(&qiov, tail_buf, tail); > + } > + > + BLKDBG_EVENT(s->data_file, BLKDBG_WRITE_COMPRESSED); > + ret = bdrv_co_pwritev(s->data_file, first->offset, len, &qiov, 0); > + qemu_iovec_destroy(&qiov); > + return ret; > + } The write functions are missing overlap checks. It can be argued that this is done by qcow2_co_pwritev_compressed_task() already, but the idea was to pair all actual writes with overlap checks (i.e., cached writes in qcow2_co_pwritev_compressed_task() wouldn’t need to do a check). > +} > + > +static int coroutine_fn cache_cluster_flush_full(Qcow2CompressedWriteCache *s, > + CacheCluster *cluster) > +{ > + int ret; > + CacheExtent *end_extent; > + int64_t align = MIN(s->cluster_size, > + MAX(s->data_file->bs->bl.request_alignment, 4 * 1024)); > + > + assert(cache_cluster_is_full(cluster)); > + > + ret = flush_consequitive_extents(s, QLIST_FIRST(&cluster->extents), > + &end_extent, align); > + > + assert(end_extent == NULL); /* all extents flushed */ > + > + return ret; > +} > + > +static int coroutine_fn cache_cluster_flush(Qcow2CompressedWriteCache *s, > + CacheCluster *c) > +{ > + int ret; > + CacheExtent *e = QLIST_FIRST(&c->extents); > + > + if (cache_cluster_is_full(c)) { > + return cache_cluster_flush_full(s, c); > + } > + > + while (e) { > + ret = flush_consequitive_extents(s, e, &e, 1); > + if (ret < 0) { > + return ret; > + } > + } > + > + return 0; > +} > + > +int coroutine_fn qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s) > +{ > + int ret = 0; > + CacheCluster *c; > + GList *local_clusters = NULL, *p; > + > + /* > + * Make a snapshot of current state: we will not flush clusters created in > + * parallel with flush operations and don't allow adding more extents to > + * staged clusters. We are also protected from parallel flush operations > + * flushing the same clusters. > + */ > + QTAILQ_FOREACH(c, &s->clusters, next) { > + if (!c->in_flight && c->n_bytes) { > + c->in_flight = true; > + local_clusters = g_list_append(local_clusters, c); > + } > + } > + > + for (p = local_clusters; p; p = p->next) { > + CacheCluster *c = p->data; > + > + if (ret == 0) { > + ret = cache_cluster_flush(s, c); > + } > + > + c->in_flight = false; > + qemu_co_queue_restart_all(&c->waiters); > + > + if (ret == 0) { > + cache_cluster_remove(s, c); > + } > + } > + > + g_list_free(local_clusters); > + > + return ret; > +} > + > +int coroutine_fn > +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s) > +{ > + int ret, save; > + > + save = s->max_active_clusters; > + s->max_active_clusters = 0; /* No more extents */ Perhaps better “Don’t cache any more extents” / “Don’t cache any more compressed clusters”? > + > + ret = qcow2_compressed_cache_co_flush(s); > + if (ret < 0) { > + s->max_active_clusters = save; > + return ret; > + } > + > + assert(QTAILQ_EMPTY(&s->clusters)); > + return 0; > +} > + > +/* @cluster takes ownership of @extent */ > +static void cluster_add_extent(Qcow2CompressedWriteCache *s, > + CacheCluster *cluster, CacheExtent *extent) > +{ > + CacheExtent *e; > + > + assert(extent->bytes); > + assert(extent->offset >= cluster->cluster_offset); > + assert(extent->offset + extent->bytes <= cluster->data_end); > + assert(!cluster->in_flight); > + > + e = QLIST_FIRST(&cluster->extents); > + if (!e) { > + /* inactive cluster */ > + assert(!cluster->n_bytes); > + s->nb_active_clusters++; > + assert(s->nb_active_clusters <= s->max_active_clusters); > + QLIST_INSERT_HEAD(&cluster->extents, extent, next); > + } else if (e->offset > extent->offset) { > + assert(extent->offset + extent->bytes <= e->offset); > + QLIST_INSERT_HEAD(&cluster->extents, extent, next); > + } else { > + while (QLIST_NEXT(e, next) && > + QLIST_NEXT(e, next)->offset < extent->offset) { > + e = QLIST_NEXT(e, next); > + } > + > + /* Now e is last element with offset < extent->offset */ > + assert(e->offset + e->bytes <= extent->offset); > + > + QLIST_INSERT_AFTER(e, extent, next); > + > + e = QLIST_NEXT(extent, next); > + if (e) { > + assert(extent->offset + extent->bytes <= e->offset); > + } > + } I like how in C one always tends to implement everything from scratch. I think adding an element into an already-sorted list is a standard problem, so it’s a shame that it’s implemented from scratch here. Also that it doesn’t use binary search, which would require a random-access list, so... Well. Not sure if it’s necessary. Probably not. Although I find it weird to start the search from the front when I’d imagine it’s more likely that new compressed clusters (extents) are added to the back. > + > + cluster->n_bytes += extent->bytes; > +} > + > +static CacheCluster *find_cluster_to_flush(Qcow2CompressedWriteCache *s) > +{ > + CacheCluster *c; > + > + QTAILQ_FOREACH(c, &s->clusters, next) { > + if (!c->in_flight && cache_cluster_is_full(c)) { > + return c; I don’t like it very much that this cache is built as an R/W cache, i.e. that entries are retained until they need to be flushed because a new entry needs space. It was my impression this was meant as a write cache, that should speed up specifically O_DIRECT operation. To me, that implies that entries are flushed once they are done, which is precisely why it works for compressed clusters, because we know when those are done. I suppose nobody cares in practice because compression is basically only used for writing out whole images (so displacing clusters happens all the time, and once you’re done, the image is flushed and closed), but it just looks strange to me. > + } > + } > + > + return NULL; > +} > + > +/* Cache an extent if there is a place */ > +static bool coroutine_fn > +try_cache_extent(Qcow2CompressedWriteCache *s, CacheExtent *extent, > + bool *cluster_in_flight) Like with other functions, but here especially, I would have liked some documentation on the interface. I suppose the return value reflects whether the “try” worked (so a good guess is possible to me, but it’s still a guess at this point). I have no idea what *cluster_in_flight means. I would have guessed it means whether the host cluster touched by the compressed cluster is in flight, but it’s also set when the cache is disabled. The caller seems to interpret it as “write this immediately”, which to me actually seems wrong, but more on that in said caller function below. Perhaps this function should just have an enum return value that tells the caller precisely what to do with some expressively named values. > +{ > + CacheCluster *c; > + int64_t cluster_offset = QEMU_ALIGN_DOWN(extent->offset, s->cluster_size); > + > + assert(extent->bytes); > + > + if (s->max_active_clusters == 0) { > + *cluster_in_flight = true; > + return false; > + } > + > + *cluster_in_flight = false; > + > + c = find_cluster(s, cluster_offset); > + if (c && c->in_flight) { > + *cluster_in_flight = true; > + return false; > + } > + if (s->nb_active_clusters >= s->max_active_clusters && > + (!c || !c->n_bytes)) > + { > + /* > + * Cache is full, we can't allocate a new cluster and can't activate > + * existing inactive cluster > + */ > + return false; > + } > + > + if (!c) { > + c = cache_cluster_new(s, cluster_offset); > + } > + > + cluster_add_extent(s, c, extent); > + > + if (cache_cluster_is_full(c)) { > + qemu_co_queue_restart_all(&s->waiters); This interface is unclear. The documentation of s->waiters says to wake them up once there is a free space in the cache, but that isn’t the case here. It’s true that it’s easy to make a free space (by flushing the full cluster), but there is no free space. This also ties in to my uneasiness about how full clusters aren’t flushed immediately. > + } > + > + return true; > +} > + > +/* Takes ownership of @buf, don't free it after call! */ > +int coroutine_fn > +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, int64_t offset, > + int64_t bytes, void *buf) > +{ > + int ret; > + int i; > + CacheExtent *extents[] = {NULL, NULL}; > + int nb_extents = 0; /* number of non-NULL elements in @extents */ > + int64_t e0_len; > + > + assert(bytes > 0); > + assert(bytes < s->cluster_size); > + > + e0_len = MIN(bytes, QEMU_ALIGN_UP(offset + 1, s->cluster_size) - offset); > + extents[0] = cache_extent_new(offset, e0_len, buf); > + nb_extents = 1; > + > + if (bytes > e0_len) { > + int64_t e1_len = bytes - e0_len; > + /* > + * We have to allocate a new buffer, so that clusters are independent > + * and can free their extents when needed. > + */ > + void *e1_buf = g_memdup(((const char *)buf) + e0_len, e1_len); > + > + extents[1] = cache_extent_new(offset + e0_len, e1_len, e1_buf); > + nb_extents = 2; > + } > + > + while (nb_extents) { > + bool do_write = false; > + > + for (i = 0; i < 2; i++) { > + CacheExtent *e = extents[i]; > + > + do_write = false; > + > + if (!e) { > + continue; > + } > + > + if (try_cache_extent(s, e, &do_write)) { > + extents[i] = NULL; > + nb_extents--; > + continue; > + } > + > + if (do_write) { > + ret = bdrv_co_pwrite(s->data_file, e->offset, e->bytes, > + e->buf, 0); Is this safe? do_write is set if the host cluster touched by this extent is in flight. Because it can’t be full, it must currently be in the process of being flushed by cache_cluster_flush(), which flushes all consecutive areas with flush_consequitive_extents(). If the underlying file has some request alignment (which it tends to do with O_DIRECT), then those writes are likely transformed into RMW operations spanning more than their precise byte range. Isn’t it possible that one of those RMWs then intersects with this write here? If the host cluster is in flight, shouldn’t we just wait until it’s flushed? > + > + cache_extent_free(e); > + extents[i] = NULL; > + nb_extents--; > + > + if (ret < 0) { > + goto out; > + } > + } > + } > + > + if (do_write) { > + /* > + * We yielded during second extent writing. Probably the cache is > + * already free and we can now cache the first extent. Where do we yield? Do you mean it probably takes time to do the write, so it’s possible that if do_write is true because some cluster was in the process of being flushed, it is now removed from the cache, so we have a free entry? Why not just check for whether we have a free entry? Also, it’s possible that do_write is true just because the cache is disabled, which I guess is handled fine because then both extents will have been written to disk, so that nb_extents is now 0... But that behavior isn’t documented here. > + */ > + continue; > + } > + > + if (nb_extents) { > + CacheCluster *cluster = find_cluster_to_flush(s); > + > + if (cluster) { > + cluster->in_flight = true; > + ret = cache_cluster_flush_full(s, cluster); > + cluster->in_flight = false; > + qemu_co_queue_restart_all(&cluster->waiters); > + qemu_co_queue_restart_all(&s->waiters); Looks weird to wake up other waiters here, because we are very likely going to take that spot, so it’s likely that nobody will be able to make use of it after we took another iteration. It’s also weird because it’s different than the two other places which wake up s->waiters not if there’s a free space, but if a free space can be made by flushing find_cluster_to_flush(). As in, this is the only place where the queue is used as it is documented. I think this again ties in with the question whether we shouldn’t flush clusters as soon as they are full instead of waiting until we need another free entry. > + if (ret < 0) { > + goto out; > + } > + cache_cluster_remove(s, cluster); > + continue; > + } > + > + qemu_co_queue_wait(&s->waiters, NULL); > + } > + } > + > + ret = 0; > + > +out: > + for (i = 0; i < 2; i++) { > + cache_extent_free(extents[i]); > + } > + > + return 0; return ret, I think. > +} > + > +int coroutine_fn > +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t offset, > + int64_t bytes, void *buf) > +{ > + CacheCluster *c; > + CacheExtent *e; > + int64_t cluster_offset = QEMU_ALIGN_DOWN(offset, s->cluster_size); > + > + c = find_cluster(s, cluster_offset); > + if (c) { > + QLIST_FOREACH(e, &c->extents, next) { > + if (e->offset == offset && e->bytes <= bytes) { > + memcpy(buf, e->buf, e->bytes); > + return 0; > + } Again, reads like it should be a binary search. I suppose e->bytes < bytes is OK because the only caller of this function is qcow2_co_preadv_compressed(), which passes a maximum length, effectively. But then I’d call the parameter accordingly (i.e., max_bytes) and zero out the tail. > + } > + } > + > + return bdrv_co_pread(s->data_file, offset, bytes, buf, 0); > +} > + > +/* > + * Caller states, that there would be no writes to this cluster beyond > + * specified @cluster_data_end. So, it's OK to flush cluster when it is filled > + * up to @cluster_data_end and it's OK to align flushing write operation up to > + * some alignment (not greater than cluster_size of course). > + */ > +void coroutine_fn > +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, > + int64_t cluster_data_end) > +{ > + int64_t cluster_offset = QEMU_ALIGN_DOWN(cluster_data_end, s->cluster_size); If cluster_data_end is aligned to a cluster boundary (I don’t think the caller does that, but...) this will do nothing, so cluster_offset is likely to be one cluster too high. I wonder whether we should assert that cluster_data_end is not aligned to the cluster size, or whether we should use cluster_data_end - 1 here, or whether the caller should pass cluster_offset. > + CacheExtent *e; > + CacheCluster *c; > + > + c = find_cluster(s, cluster_offset); > + if (!c) { > + c = cache_cluster_new(s, cluster_offset); Is this case reachable? How? > + } > + > + QLIST_FOREACH(e, &c->extents, next) { > + assert(e->offset + e->bytes <= cluster_data_end); > + } > + > + /* Shouldn't set data_end several times */ > + assert(c->data_end == c->cluster_offset + s->cluster_size); > + > + c->data_end = cluster_data_end; > + if (cache_cluster_is_full(c)) { > + qemu_co_queue_restart_all(&s->waiters); Like in try_cache_extent(), this wakes up the queue while technically there is no free space. I get the intention here, because as documented somewhere, this function is called from a mutexed section and we probably don’t want to write data to the data_file here. But it would make sense for me to flush the cluster later as soon as possible, and then wake up the waiters. > + } > +} > + > +Qcow2CompressedWriteCache *qcow2_compressed_cache_new(BdrvChild *data_file, > + int64_t cluster_size, > + int64_t cache_size) > +{ > + Qcow2CompressedWriteCache *s = g_new(Qcow2CompressedWriteCache, 1); > + > + assert(cache_size >= cluster_size); > + > + *s = (Qcow2CompressedWriteCache) { > + .data_file = data_file, > + .cluster_size = cluster_size, > + .max_active_clusters = cache_size / cluster_size, > + }; > + > + qemu_co_queue_init(&s->waiters); > + QTAILQ_INIT(&s->clusters); > + > + return s; > +} > + > +void qcow2_compressed_cache_free(Qcow2CompressedWriteCache *s) > +{ > + if (!s) { > + return; > + } > + > + QTAILQ_FOREACH_FUNC_SAFE(&s->clusters, next, cache_cluster_free); It seems strange to me to call this function if there are still clusters in the cache. I suppose you can’t do anything else if the flush function failed to evict them, so I suppose it’s correct to do this, but it makes me flinch a bit. > + g_free(s); > +} > + > +void qcow2_compressed_cache_set_size(Qcow2CompressedWriteCache *s, > + int64_t size) > +{ > + /* > + * We don't do flush here. Don't care too much, it's safe to have cache > + * larger than maximum, it will only decrease until it reaches the new > + * maximum. > + */ > + assert(size >= s->cluster_size); > + s->max_active_clusters = size / s->cluster_size; > +} Simple enough, but I don’t see why we need it. More on that in the next patch, though. > + > +void coroutine_fn > +qcow2_compressed_cache_co_discard(Qcow2CompressedWriteCache *s, > + int64_t cluster_offset) I don’t like this function’s name because I associate “discard” with discarding data. I think “evict” would fit better, or perhaps “drop”. (I think “drop” better conveys that the data isn’t supposed to be written back.) Max > +{ > + CacheCluster *c; > + > + while (true) { > + c = find_cluster(s, cluster_offset); > + if (!c) { > + return; > + } > + if (!c->in_flight) { > + cache_cluster_remove(s, c); > + return; > + } > + qemu_co_queue_wait(&c->waiters, NULL); > + } > +} > + > + > +/* > + * Wrappers for qcow2_compressed_cache_co_flush > + * > + * TODO: update scripts/block-coroutine-wrapper to generate this too > + */ > + > +typedef struct Qcow2CompressedCacheCoFlush { > + BdrvPollCo poll_state; > + Qcow2CompressedWriteCache *state; > +} Qcow2CompressedCacheCoFlush; > + > +static void coroutine_fn > +qcow2_compressed_cache_co_flush_entry(void *opaque) > +{ > + Qcow2CompressedCacheCoFlush *s = opaque; > + > + s->poll_state.ret = qcow2_compressed_cache_co_flush(s->state); > + s->poll_state.in_progress = false; > + > + aio_wait_kick(); > +} > + > +int qcow2_compressed_cache_flush(BlockDriverState *bs, > + Qcow2CompressedWriteCache *state) > +{ > + if (qemu_in_coroutine()) { > + return qcow2_compressed_cache_co_flush(state); > + } else { > + Qcow2CompressedCacheCoFlush s = { > + .poll_state.bs = bs, > + .poll_state.in_progress = true, > + > + .state = state, > + }; > + > + s.poll_state.co = > + qemu_coroutine_create(qcow2_compressed_cache_co_flush_entry, &s); > + > + return bdrv_poll_co(&s.poll_state); > + } > +} > + > +/* > + * Wrappers for qcow2_compressed_cache_co_stop_flush > + * > + * TODO: update scripts/block-coroutine-wrapper to generate this too > + */ > + > +typedef struct Qcow2CompressedCacheCoStopFlush { > + BdrvPollCo poll_state; > + Qcow2CompressedWriteCache *state; > +} Qcow2CompressedCacheCoStopFlush; > + > +static void coroutine_fn > +qcow2_compressed_cache_co_stop_flush_entry(void *opaque) > +{ > + Qcow2CompressedCacheCoStopFlush *s = opaque; > + > + s->poll_state.ret = qcow2_compressed_cache_co_stop_flush(s->state); > + s->poll_state.in_progress = false; > + > + aio_wait_kick(); > +} > + > +int qcow2_compressed_cache_stop_flush(BlockDriverState *bs, > + Qcow2CompressedWriteCache *state) > +{ > + if (qemu_in_coroutine()) { > + return qcow2_compressed_cache_co_stop_flush(state); > + } else { > + Qcow2CompressedCacheCoStopFlush s = { > + .poll_state.bs = bs, > + .poll_state.in_progress = true, > + > + .state = state, > + }; > + > + s.poll_state.co = > + qemu_coroutine_create(qcow2_compressed_cache_co_stop_flush_entry, > + &s); > + > + return bdrv_poll_co(&s.poll_state); > + } > +} > diff --git a/block/meson.build b/block/meson.build > index eeaefe5809..7b94794c28 100644 > --- a/block/meson.build > +++ b/block/meson.build > @@ -25,6 +25,7 @@ block_ss.add(files( > 'qcow2-bitmap.c', > 'qcow2-cache.c', > 'qcow2-cluster.c', > + 'qcow2-compressed-write-cache.c', > 'qcow2-refcount.c', > 'qcow2-snapshot.c', > 'qcow2-threads.c', >
you may jump first to my last inline answer 10.02.2021 20:07, Max Reitz wrote: > On 29.01.21 17:50, Vladimir Sementsov-Ogievskiy wrote: >> Compressed writes and O_DIRECT are not friends: they works too slow, >> because compressed writes does many small unaligned to 512 writes. >> >> Let's introduce an internal cache, so that compressed writes may work >> well when O_DIRECT is on. >> >> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com> >> --- >> block/qcow2.h | 29 + >> block/qcow2-compressed-write-cache.c | 770 +++++++++++++++++++++++++++ >> block/meson.build | 1 + >> 3 files changed, 800 insertions(+) >> create mode 100644 block/qcow2-compressed-write-cache.c >> >> diff --git a/block/qcow2.h b/block/qcow2.h >> index 0678073b74..fbdedf89fa 100644 >> --- a/block/qcow2.h >> +++ b/block/qcow2.h >> @@ -322,6 +322,8 @@ typedef struct Qcow2BitmapHeaderExt { >> uint64_t bitmap_directory_offset; >> } QEMU_PACKED Qcow2BitmapHeaderExt; >> +typedef struct Qcow2CompressedWriteCache Qcow2CompressedWriteCache; >> + >> #define QCOW2_MAX_THREADS 4 >> typedef struct BDRVQcow2State { >> @@ -1010,4 +1012,31 @@ int coroutine_fn >> qcow2_co_decrypt(BlockDriverState *bs, uint64_t host_offset, >> uint64_t guest_offset, void *buf, size_t len); >> +Qcow2CompressedWriteCache *qcow2_compressed_cache_new(BdrvChild *data_file, >> + int64_t cluster_size, >> + int64_t cache_size); >> +void qcow2_compressed_cache_free(Qcow2CompressedWriteCache *s); >> +int coroutine_fn >> +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t offset, >> + int64_t bytes, void *buf); >> +int coroutine_fn >> +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, int64_t offset, >> + int64_t bytes, void *buf); >> +void coroutine_fn >> +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, >> + int64_t cluster_data_end); >> +int coroutine_fn >> +qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s); >> +int qcow2_compressed_cache_flush(BlockDriverState *bs, >> + Qcow2CompressedWriteCache *state); >> +int coroutine_fn >> +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s); >> +int qcow2_compressed_cache_stop_flush(BlockDriverState *bs, >> + Qcow2CompressedWriteCache *s); >> +void qcow2_compressed_cache_set_size(Qcow2CompressedWriteCache *s, >> + int64_t size); >> +void coroutine_fn >> +qcow2_compressed_cache_co_discard(Qcow2CompressedWriteCache *s, >> + int64_t cluster_offset); >> + > > It would be nice if these functions had their interface documented somewhere. I tried to comment dificult things in .c... Is there a prefernce, where to document how and what function does, in .h or in .c ? > >> #endif >> diff --git a/block/qcow2-compressed-write-cache.c b/block/qcow2-compressed-write-cache.c >> new file mode 100644 >> index 0000000000..7bb92cb550 >> --- /dev/null >> +++ b/block/qcow2-compressed-write-cache.c >> @@ -0,0 +1,770 @@ >> +/* >> + * Write cache for qcow2 compressed writes >> + * >> + * Copyright (c) 2021 Virtuozzo International GmbH. >> + * >> + * Permission is hereby granted, free of charge, to any person obtaining a copy >> + * of this software and associated documentation files (the "Software"), to deal >> + * in the Software without restriction, including without limitation the rights >> + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell >> + * copies of the Software, and to permit persons to whom the Software is >> + * furnished to do so, subject to the following conditions: >> + * >> + * The above copyright notice and this permission notice shall be included in >> + * all copies or substantial portions of the Software. >> + * >> + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR >> + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, >> + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL >> + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER >> + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, >> + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN >> + * THE SOFTWARE. >> + */ >> + >> +#include "qemu/osdep.h" >> + >> +#include "block/block_int.h" >> +#include "block/block-gen.h" >> +#include "qemu/coroutine.h" >> +#include "qapi/qapi-events-block-core.h" >> +#include "qcow2.h" >> + >> +typedef struct CacheExtent { >> + int64_t offset; >> + int64_t bytes; >> + void *buf; >> + QLIST_ENTRY(CacheExtent) next; >> +} CacheExtent; >> + >> +typedef struct CacheCluster { > > It isn’t immediately clear what these two structures mean by just their name, because “extent” has no meaning in the context of qcow2. It's not important for the cache are extents compressed clusters or not.. So I'd keep more generic name > > I understand CacheExtent to basically be a compressed cluster, and CacheCluster to be a host cluster. Perhaps their names should reflect that. > > (OTOH, the Cache* prefix seems unnecessary to me, because these are just local structs.) > >> + int64_t cluster_offset; >> + int64_t n_bytes; /* sum of extents lengths */ >> + >> + /* >> + * data_end: cluster is full if data_end reached and ready to be flushed. >> + * data_end is absoluted offset, not relative. > > *absolute > >> + */ >> + int64_t data_end; >> + >> + bool in_flight; /* cluster is being flushed now */ >> + >> + /* >> + * waiters: coroutines to wake after flush. > > No, once in_flight is reset to false, which may happen even if the cluster hasn’t been flushed. > > > As a note that doesn’t really belong anywhere, I don’t like the current use of this queue perfectly well. > > This is how it is used right now: > > Some context A sets in_flight and flushes the cluster. > > Some context B wants to evict the cluster from the cache. It sees in_flight, so it gets into the queue. (If in_flight were not set, B would remove the cluster from the cache.) > > A finishes flushing, resets in_flight, queues B. Then if flushing succeeded, it will evict the cluster from the cache. If it failed, the cluster remains in the cache. > > Then B is executed. If the cluster isn’t in the cache anymore, it’s done. If still is, it will remove it, and then be done. all correct > > My problem is that if B were executed right where it’s enqueued (with queue_restart_all()), we would crash because B would see the cluster in the cache, remove it, and then A would have a dangling reference and remove the (freed) cluster again. But queue_restart_all() doesn't execute B immediately, but only when A yield. And I think a lot of code rely on this. > > So, well, it works, but I personally find it strange to “invoke“ B where you don’t actually want it to run. I think the @waiters queue should only be restarted once A has had the chance to remove the cluster from the cache, or to be more general, once A no longer has a reference to the cluster. You are right it looks strange. Why I've written it this way? I dont remember :\ .. Will try to reorder. > > tl;dr: I think the @waiters queue should only be restarted once the reference to the cluster is no longer going to be used. yes, that will look better. > >> Must be empty when in_flight is >> + * false >> + */ >> + CoQueue waiters; >> + >> + QTAILQ_ENTRY(CacheCluster) next; >> + QLIST_HEAD(, CacheExtent) extents; /* sorted by offset */ >> +} CacheCluster; >> + >> +typedef QTAILQ_HEAD(, CacheCluster) CacheClusterQueue; > > Why the typedef? It’s used one time. Haha, good question. At some stage of developepment I had two queue, separate for complete clusters, and passed these queues to functions.. But I decided that this makes me to go through two queues in same places instead of one and this doesn't make things simpler and safer. So I've dropped it. But typedef was left. Will drop it as well. > >> + >> +struct Qcow2CompressedWriteCache { >> + /* >> + * @data_file and @cluster_size are copied from qcow2 state. Not huge >> + * duplications, seems better to avoid accessing the whole qcow2 state >> + * instead. > > Why? What if in the future the data_file child can be changed with reopen? I didn't considered this.. Still it's impossible now, yes? But would be possible in future of course. > > What’s the argument against accessing the whole qcow2 state? That it’s a layering violation, that we shouldn’t access it while the rest uses it? To keep it simple to reuse the cache for something other. > If that’s a concern, perhaps *data_file should be an indirect pointer to the pointer in the BDRVQcow2State. Or we should update it on reopen (like cache_size) > >> + */ >> + BdrvChild *data_file; >> + int64_t cluster_size; >> + >> + CoQueue waiters; /* coroutines, waiting for free place in the cache */ >> + >> + /* >> + * Cache limits not total number of clusters but total number of active >> + * clusters. Active clusters are clusters with non-zero @n_bytes (and >> + * therefor non-empty @extents). This is done so we don't need to wait for >> + * cache flush qcow2_compressed_cache_set_cluster_end() (which may create >> + * cluster with @data_end set but zero @n_bytes), as >> + * qcow2_compressed_cache_set_cluster_end() is intended to be called from >> + * qcow2 mutex critical section. >> + */ >> + int nb_active_clusters; >> + >> + /* >> + * If max_active_clusters is 0 it means that cache is inactive: all new >> + * writes should fallthrough to data_file immediately. >> + */ >> + int max_active_clusters; >> + >> + CacheClusterQueue clusters; >> +}; >> + >> +/* New extent takes ownership of @buf */ >> +static CacheExtent *cache_extent_new(int64_t offset, int64_t bytes, void *buf) >> +{ >> + CacheExtent *e = g_new(CacheExtent, 1); >> + >> + *e = (CacheExtent) { >> + .offset = offset, >> + .bytes = bytes, >> + .buf = buf, >> + }; >> + >> + return e; >> +} >> + >> +static void cache_extent_free(CacheExtent *e) >> +{ >> + if (e) { >> + g_free(e->buf); >> + g_free(e); >> + } >> +} >> + >> +static CacheCluster *find_cluster(Qcow2CompressedWriteCache *s, >> + int64_t cluster_offset) >> +{ >> + CacheCluster *c; >> + >> + assert(QEMU_IS_ALIGNED(cluster_offset, s->cluster_size)); >> + >> + QTAILQ_FOREACH(c, &s->clusters, next) { >> + if (cluster_offset == c->cluster_offset) { >> + return c; >> + } >> + } > > This reads like in any language but C we’d opt for a hash table. On that thought, why don’t we opt for a hash table even in C? Agree why not. I just tried to avoid difficulties that can be avoided for the first run. > >> + >> + return NULL; >> +} >> + >> +/* Creates "inactive" cluster, which doesn't influence s->nb_active_clusters */ >> +static CacheCluster *cache_cluster_new(Qcow2CompressedWriteCache *s, >> + int64_t cluster_offset) >> +{ >> + CacheCluster *c; >> + >> + assert(!find_cluster(s, cluster_offset)); >> + >> + c = g_new(CacheCluster, 1); >> + *c = (CacheCluster) { >> + .cluster_offset = cluster_offset, >> + .data_end = cluster_offset + s->cluster_size >> + }; >> + >> + qemu_co_queue_init(&c->waiters); >> + QTAILQ_INSERT_TAIL(&s->clusters, c, next); >> + >> + return c; >> +} >> + >> +static void cache_cluster_free(CacheCluster *cluster) >> +{ >> + assert(!cluster->in_flight); >> + assert(qemu_co_queue_empty(&cluster->waiters)); >> + QLIST_FOREACH_FUNC_SAFE(&cluster->extents, next, cache_extent_free); >> + g_free(cluster); >> +} >> + >> +static bool cache_cluster_is_full(CacheCluster *cluster) >> +{ >> + assert(cluster->cluster_offset + cluster->n_bytes <= cluster->data_end); >> + return cluster->cluster_offset + cluster->n_bytes == cluster->data_end; >> +} >> + >> +static void cache_cluster_remove(Qcow2CompressedWriteCache *s, >> + CacheCluster *cluster) >> +{ >> + if (cluster->n_bytes) { >> + s->nb_active_clusters--; >> + } >> + QTAILQ_REMOVE(&s->clusters, cluster, next); >> + cache_cluster_free(cluster); >> +} >> + >> +/* >> + * Return number of consequtive clusters starting from @first. Store next after > > %s/consequi\?tive/consecutive/ > >> + * last extent pointer into @next, store end offset of last extent into >> + * @end_off. >> + */ >> +static int count_consequitive_extents(CacheExtent *first, >> + CacheExtent **next, >> + int64_t *end_off) >> +{ >> + CacheExtent *e; >> + int64_t off = first->offset; >> + int nb = 0; >> + >> + for (e = first; e; e = QLIST_NEXT(e, next)) { >> + assert(e->offset >= off); >> + if (e->offset > off) { >> + break; >> + } >> + off += e->bytes; >> + nb++; >> + } >> + >> + if (next) { >> + *next = e; >> + } >> + >> + if (end_off) { >> + *end_off = off; >> + } >> + >> + return nb; >> +} >> + >> +/* >> + * Write consequtive extents, starting from @firest. Store next after last > > *first > >> + * extent pointer into @next. If align > 1, align end of the whole write by >> + * zero. >> + */ >> +static int coroutine_fn flush_consequitive_extents(Qcow2CompressedWriteCache *s, >> + CacheExtent *first, >> + CacheExtent **next, >> + int64_t align) >> +{ >> + CacheExtent *e, *end_extent; >> + int64_t end; >> + int nb_extents = count_consequitive_extents(first, &end_extent, &end); > > I’d prefer nb_bufs, because it will include a tail if present. > >> + int64_t aligned_end = QEMU_ALIGN_UP(end, align); >> + int64_t tail = aligned_end - end; >> + int64_t len = aligned_end - first->offset; >> + >> + /* >> + * Alignment if for flushing full cluster, first extent offset is always > > s/.*,/The alignment will not exceed the cluster size, so the/? s/if/is/. When cluster is "full", which means it is filled up to its data_end, the remaining tail is unsed, and we can align it up. > >> + * aligned. >> + */ >> + assert(QEMU_IS_ALIGNED(first->offset, align)); >> + >> + if (next) { >> + *next = end_extent; >> + } >> + >> + if (tail) { >> + nb_extents++; >> + } >> + >> + if (nb_extents > IOV_MAX) { >> + g_autofree void *buf = g_malloc(len); > > I wonder if this should be blk_blockalign() if first->offset is aligned to the cluster size. (Don’t know how common that case is.) blockalign may be better, yes > >> + char *p = buf; >> + >> + for (e = first; e != end_extent; e = QLIST_NEXT(e, next)) { >> + memcpy(p, e->buf, e->bytes); >> + p += e->bytes; >> + } >> + >> + if (tail) { >> + memset(p, 0, tail); >> + } >> + >> + BLKDBG_EVENT(s->data_file, BLKDBG_WRITE_COMPRESSED); >> + return bdrv_co_pwrite(s->data_file, first->offset, len, buf, 0); >> + } else { >> + int ret; >> + QEMUIOVector qiov; >> + g_autofree void *tail_buf = NULL; >> + >> + qemu_iovec_init(&qiov, nb_extents); >> + for (e = first; e != end_extent; e = QLIST_NEXT(e, next)) { >> + qemu_iovec_add(&qiov, e->buf, e->bytes); >> + } >> + >> + if (tail) { >> + tail_buf = g_malloc0(tail); >> + qemu_iovec_add(&qiov, tail_buf, tail); >> + } >> + >> + BLKDBG_EVENT(s->data_file, BLKDBG_WRITE_COMPRESSED); >> + ret = bdrv_co_pwritev(s->data_file, first->offset, len, &qiov, 0); >> + qemu_iovec_destroy(&qiov); >> + return ret; >> + } > > The write functions are missing overlap checks. It can be argued that this is done by qcow2_co_pwritev_compressed_task() already, but the idea was to pair all actual writes with overlap checks (i.e., cached writes in qcow2_co_pwritev_compressed_task() wouldn’t need to do a check). Some more checks will not hurt anyway > >> +} >> + >> +static int coroutine_fn cache_cluster_flush_full(Qcow2CompressedWriteCache *s, >> + CacheCluster *cluster) >> +{ >> + int ret; >> + CacheExtent *end_extent; >> + int64_t align = MIN(s->cluster_size, >> + MAX(s->data_file->bs->bl.request_alignment, 4 * 1024)); >> + >> + assert(cache_cluster_is_full(cluster)); >> + >> + ret = flush_consequitive_extents(s, QLIST_FIRST(&cluster->extents), >> + &end_extent, align); >> + >> + assert(end_extent == NULL); /* all extents flushed */ >> + >> + return ret; >> +} >> + >> +static int coroutine_fn cache_cluster_flush(Qcow2CompressedWriteCache *s, >> + CacheCluster *c) >> +{ >> + int ret; >> + CacheExtent *e = QLIST_FIRST(&c->extents); >> + >> + if (cache_cluster_is_full(c)) { >> + return cache_cluster_flush_full(s, c); >> + } >> + >> + while (e) { >> + ret = flush_consequitive_extents(s, e, &e, 1); >> + if (ret < 0) { >> + return ret; >> + } >> + } >> + >> + return 0; >> +} >> + >> +int coroutine_fn qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s) >> +{ >> + int ret = 0; >> + CacheCluster *c; >> + GList *local_clusters = NULL, *p; >> + >> + /* >> + * Make a snapshot of current state: we will not flush clusters created in >> + * parallel with flush operations and don't allow adding more extents to >> + * staged clusters. We are also protected from parallel flush operations >> + * flushing the same clusters. >> + */ >> + QTAILQ_FOREACH(c, &s->clusters, next) { >> + if (!c->in_flight && c->n_bytes) { >> + c->in_flight = true; >> + local_clusters = g_list_append(local_clusters, c); >> + } >> + } >> + >> + for (p = local_clusters; p; p = p->next) { >> + CacheCluster *c = p->data; >> + >> + if (ret == 0) { >> + ret = cache_cluster_flush(s, c); >> + } >> + >> + c->in_flight = false; >> + qemu_co_queue_restart_all(&c->waiters); >> + >> + if (ret == 0) { >> + cache_cluster_remove(s, c); >> + } >> + } >> + >> + g_list_free(local_clusters); >> + >> + return ret; >> +} >> + >> +int coroutine_fn >> +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s) >> +{ >> + int ret, save; >> + >> + save = s->max_active_clusters; >> + s->max_active_clusters = 0; /* No more extents */ > > Perhaps better “Don’t cache any more extents” / “Don’t cache any more compressed clusters”? right. or just "clusters", as compressed cluster actually maps to extent. > >> + >> + ret = qcow2_compressed_cache_co_flush(s); >> + if (ret < 0) { >> + s->max_active_clusters = save; >> + return ret; >> + } >> + >> + assert(QTAILQ_EMPTY(&s->clusters)); >> + return 0; >> +} >> + >> +/* @cluster takes ownership of @extent */ >> +static void cluster_add_extent(Qcow2CompressedWriteCache *s, >> + CacheCluster *cluster, CacheExtent *extent) >> +{ >> + CacheExtent *e; >> + >> + assert(extent->bytes); >> + assert(extent->offset >= cluster->cluster_offset); >> + assert(extent->offset + extent->bytes <= cluster->data_end); >> + assert(!cluster->in_flight); >> + >> + e = QLIST_FIRST(&cluster->extents); >> + if (!e) { >> + /* inactive cluster */ >> + assert(!cluster->n_bytes); >> + s->nb_active_clusters++; >> + assert(s->nb_active_clusters <= s->max_active_clusters); >> + QLIST_INSERT_HEAD(&cluster->extents, extent, next); >> + } else if (e->offset > extent->offset) { >> + assert(extent->offset + extent->bytes <= e->offset); >> + QLIST_INSERT_HEAD(&cluster->extents, extent, next); >> + } else { >> + while (QLIST_NEXT(e, next) && >> + QLIST_NEXT(e, next)->offset < extent->offset) { >> + e = QLIST_NEXT(e, next); >> + } >> + >> + /* Now e is last element with offset < extent->offset */ >> + assert(e->offset + e->bytes <= extent->offset); >> + >> + QLIST_INSERT_AFTER(e, extent, next); >> + >> + e = QLIST_NEXT(extent, next); >> + if (e) { >> + assert(extent->offset + extent->bytes <= e->offset); >> + } >> + } > > I like how in C one always tends to implement everything from scratch.I think adding an element into an already-sorted list is a standard problem, so it’s a shame that it’s implemented from scratch here. Also that it doesn’t use binary search, which would require a random-access list, so... Yes, we have g_list_insert_sorted(), but it doesn't use binary search anyway. So I decided to use typed lists as Kevin said they are better.. Still I'm not fun of all these macros. > > Well. Not sure if it’s necessary. Probably not. Although I find it weird to start the search from the front when I’d imagine it’s more likely that new compressed clusters (extents) are added to the back. Thought about this too, but decided to not care. 1. I think, it's all doesn't really matter, as io write operation is a lot longer than all these opeartions with a list. 2. Still it's interesting :) When adding extent we want to check that it doesn't intersect with other extents. And we want to have sorted list when flush the cluster. In GLib there is GTree. It has g_tree_foreach in sorted order. But it doesn't have an option to get neighbours of inserted element (to check intersection), neither it have something like lower_bound and upper_bound :\ Hmm, implementing a tree from stratch is not what I want to do in context of my cache :\ Any ideas? wow, I found this: https://gitlab.gnome.org/GNOME/glib/-/merge_requests/1509 So, 3 months ago all we need was merged into GLib master.. Aha and it's in GLib 2.67.0.. and centos7 has 2.56.. I can add a comment: "refactor using GTree node-based API when it become available (it will in GLib 2.67)" > >> + >> + cluster->n_bytes += extent->bytes; >> +} >> + >> +static CacheCluster *find_cluster_to_flush(Qcow2CompressedWriteCache *s) >> +{ >> + CacheCluster *c; >> + >> + QTAILQ_FOREACH(c, &s->clusters, next) { >> + if (!c->in_flight && cache_cluster_is_full(c)) { >> + return c; > > I don’t like it very much that this cache is built as an R/W cache, i.e. that entries are retained until they need to be flushed because a new entry needs space. > > It was my impression this was meant as a write cache, that should speed up specifically O_DIRECT operation. To me, that implies that entries are flushed once they are done, which is precisely why it works for compressed clusters, because we know when those are done. We learn that cluster is finished in two cases: 1. when data_end is set and we see that cluster is full. That's a bad place for flushing, as we are under qcow2 mutex. 2. when we add an extent. That's occures during some write operation.. And I don't see real difference, flushing cluster here or when we want to create one more cluster. And because we don't want to flush at [1], we'll probably have some lost full cluster if flush only on [2]. So, we anyway should flush when want to create new cluster but cache is full. And then no real reason to flush on [2] > > I suppose nobody cares in practice because compression is basically only used for writing out whole images (so displacing clusters happens all the time, and once you’re done, the image is flushed and closed), but it just looks strange to me. > >> + } >> + } >> + >> + return NULL; >> +} >> + >> +/* Cache an extent if there is a place */ >> +static bool coroutine_fn >> +try_cache_extent(Qcow2CompressedWriteCache *s, CacheExtent *extent, >> + bool *cluster_in_flight) > > Like with other functions, but here especially, I would have liked some documentation on the interface. will add > I suppose the return value reflects whether the “try” worked (so a good guess is possible to me, but it’s still a guess at this point). > > I have no idea what *cluster_in_flight means. Neither me at this point, I don't remember:) But everything is so obvious when you are writing the code.. Ok, looking forward, *cluster_in_flight means that cluster is now flushing, so please don't even retry to cache this extent! > I would have guessed it means whether the host cluster touched by the compressed cluster is in flight, but it’s also set when the cache is disabled. The caller seems to interpret it as “write this immediately”, which to me actually seems wrong, but more on that in said caller function below. > > Perhaps this function should just have an enum return value that tells the caller precisely what to do with some expressively named values. > >> +{ >> + CacheCluster *c; >> + int64_t cluster_offset = QEMU_ALIGN_DOWN(extent->offset, s->cluster_size); >> + >> + assert(extent->bytes); >> + >> + if (s->max_active_clusters == 0) { >> + *cluster_in_flight = true; >> + return false; >> + } >> + >> + *cluster_in_flight = false; >> + >> + c = find_cluster(s, cluster_offset); >> + if (c && c->in_flight) { >> + *cluster_in_flight = true; >> + return false; >> + } >> + if (s->nb_active_clusters >= s->max_active_clusters && >> + (!c || !c->n_bytes)) >> + { >> + /* >> + * Cache is full, we can't allocate a new cluster and can't activate >> + * existing inactive cluster >> + */ >> + return false; >> + } >> + >> + if (!c) { >> + c = cache_cluster_new(s, cluster_offset); >> + } >> + >> + cluster_add_extent(s, c, extent); >> + >> + if (cache_cluster_is_full(c)) { >> + qemu_co_queue_restart_all(&s->waiters); > > This interface is unclear. The documentation of s->waiters says to wake them up once there is a free space in the cache, but that isn’t the case here. It’s true that it’s easy to make a free space (by flushing the full cluster), but there is no free space. Problem is in documentation, will fix. > > This also ties in to my uneasiness about how full clusters aren’t flushed immediately. > >> + } >> + >> + return true; >> +} >> + >> +/* Takes ownership of @buf, don't free it after call! */ >> +int coroutine_fn >> +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, int64_t offset, >> + int64_t bytes, void *buf) >> +{ >> + int ret; >> + int i; >> + CacheExtent *extents[] = {NULL, NULL}; >> + int nb_extents = 0; /* number of non-NULL elements in @extents */ >> + int64_t e0_len; >> + >> + assert(bytes > 0); >> + assert(bytes < s->cluster_size); >> + >> + e0_len = MIN(bytes, QEMU_ALIGN_UP(offset + 1, s->cluster_size) - offset); >> + extents[0] = cache_extent_new(offset, e0_len, buf); >> + nb_extents = 1; >> + >> + if (bytes > e0_len) { >> + int64_t e1_len = bytes - e0_len; >> + /* >> + * We have to allocate a new buffer, so that clusters are independent >> + * and can free their extents when needed. >> + */ >> + void *e1_buf = g_memdup(((const char *)buf) + e0_len, e1_len); >> + >> + extents[1] = cache_extent_new(offset + e0_len, e1_len, e1_buf); >> + nb_extents = 2; >> + } >> + >> + while (nb_extents) { >> + bool do_write = false; >> + >> + for (i = 0; i < 2; i++) { >> + CacheExtent *e = extents[i]; >> + >> + do_write = false; >> + >> + if (!e) { >> + continue; >> + } >> + >> + if (try_cache_extent(s, e, &do_write)) { >> + extents[i] = NULL; >> + nb_extents--; >> + continue; >> + } >> + >> + if (do_write) { >> + ret = bdrv_co_pwrite(s->data_file, e->offset, e->bytes, >> + e->buf, 0); > > Is this safe? do_write is set if the host cluster touched by this extent is in flight. Because it can’t be full, it must currently be in the process of being flushed by cache_cluster_flush(), which flushes all consecutive areas with flush_consequitive_extents(). If the underlying file has some request alignment (which it tends to do with O_DIRECT), then those writes are likely transformed into RMW operations spanning more than their precise byte range. Isn’t it possible that one of those RMWs then intersects with this write here? Hmm. I thought that if user writes to in_fligth cluster, it's wrong anyway, so we should not care.. But it may be not-full in-fligth cluster because of previous "flush" operation.. And in this case we should wait until it flushed. > > If the host cluster is in flight, shouldn’t we just wait until it’s flushed? agree > >> + >> + cache_extent_free(e); >> + extents[i] = NULL; >> + nb_extents--; >> + >> + if (ret < 0) { >> + goto out; >> + } >> + } >> + } >> + >> + if (do_write) { >> + /* >> + * We yielded during second extent writing. Probably the cache is >> + * already free and we can now cache the first extent. > > Where do we yield? Do you mean it probably takes time to do the write, so it’s possible that if do_write is true because some cluster was in the process of being flushed, it is now removed from the cache, so we have a free entry? > > Why not just check for whether we have a free entry? and we do it, moving to try_cache_extent at start of loop.. > > Also, it’s possible that do_write is true just because the cache is disabled, which I guess is handled fine because then both extents will have been written to disk, so that nb_extents is now 0... But that behavior isn’t documented here. > >> + */ >> + continue; >> + } >> + >> + if (nb_extents) { >> + CacheCluster *cluster = find_cluster_to_flush(s); >> + >> + if (cluster) { >> + cluster->in_flight = true; >> + ret = cache_cluster_flush_full(s, cluster); >> + cluster->in_flight = false; >> + qemu_co_queue_restart_all(&cluster->waiters); >> + qemu_co_queue_restart_all(&s->waiters); > > Looks weird to wake up other waiters here, because we are very likely going to take that spot, so it’s likely that nobody will be able to make use of it after we took another iteration. > > It’s also weird because it’s different than the two other places which wake up s->waiters not if there’s a free space, but if a free space can be made by flushing find_cluster_to_flush(). As in, this is the only place where the queue is used as it is documented. Sorry me. I'll come back with better documentation for v2 > > I think this again ties in with the question whether we shouldn’t flush clusters as soon as they are full instead of waiting until we need another free entry. The only reason is what I've said above about qcow2 mutext.. We should not do extra things, when data_end of cluster is set. And that's why we'll have postponed clusters anyway. > >> + if (ret < 0) { >> + goto out; >> + } >> + cache_cluster_remove(s, cluster); >> + continue; >> + } >> + >> + qemu_co_queue_wait(&s->waiters, NULL); >> + } >> + } >> + >> + ret = 0; >> + >> +out: >> + for (i = 0; i < 2; i++) { >> + cache_extent_free(extents[i]); >> + } >> + >> + return 0; > > return ret, I think. > >> +} >> + >> +int coroutine_fn >> +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t offset, >> + int64_t bytes, void *buf) >> +{ >> + CacheCluster *c; >> + CacheExtent *e; >> + int64_t cluster_offset = QEMU_ALIGN_DOWN(offset, s->cluster_size); >> + >> + c = find_cluster(s, cluster_offset); >> + if (c) { >> + QLIST_FOREACH(e, &c->extents, next) { >> + if (e->offset == offset && e->bytes <= bytes) { >> + memcpy(buf, e->buf, e->bytes); >> + return 0; >> + } > > Again, reads like it should be a binary search. > > I suppose e->bytes < bytes is OK because the only caller of this function is qcow2_co_preadv_compressed(), which passes a maximum length, effectively. But then I’d call the parameter accordingly (i.e., max_bytes) and zero out the tail. OK > >> + } >> + } >> + >> + return bdrv_co_pread(s->data_file, offset, bytes, buf, 0); >> +} >> + >> +/* >> + * Caller states, that there would be no writes to this cluster beyond >> + * specified @cluster_data_end. So, it's OK to flush cluster when it is filled >> + * up to @cluster_data_end and it's OK to align flushing write operation up to >> + * some alignment (not greater than cluster_size of course). >> + */ >> +void coroutine_fn >> +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, >> + int64_t cluster_data_end) >> +{ >> + int64_t cluster_offset = QEMU_ALIGN_DOWN(cluster_data_end, s->cluster_size); > > If cluster_data_end is aligned to a cluster boundary (I don’t think the caller does that, but...) this will do nothing, so cluster_offset is likely to be one cluster too high. well add an assertion > > I wonder whether we should assert that cluster_data_end is not aligned to the cluster size, or whether we should use cluster_data_end - 1 here, or whether the caller should pass cluster_offset. > >> + CacheExtent *e; >> + CacheCluster *c; >> + >> + c = find_cluster(s, cluster_offset); >> + if (!c) { >> + c = cache_cluster_new(s, cluster_offset); > > Is this case reachable? How? Ooops... I feel stupid. Great thing that you asked this question! Somehow I thought that already allocated clusters may be in progress of... what? compression? writing?.. But actually allocation is of course done after compression. So we may compress clusters in parallel, but than they allocated and written sequentially. Of courese without the cache the writes themselves may go in parallel. But write to cache is immediate and doesn't yield. We even can do it before unlocking qcow2 mutex to be sure that we don't switch to another coroutine to write next extent first. Seems I was wrong, and all this can be a lot simpler. Very sorry for your time. But it helped me to open my eyes :\
On 11.02.21 13:49, Vladimir Sementsov-Ogievskiy wrote: > you may jump first to my last inline answer > > 10.02.2021 20:07, Max Reitz wrote: >> On 29.01.21 17:50, Vladimir Sementsov-Ogievskiy wrote: >>> Compressed writes and O_DIRECT are not friends: they works too slow, >>> because compressed writes does many small unaligned to 512 writes. >>> >>> Let's introduce an internal cache, so that compressed writes may work >>> well when O_DIRECT is on. >>> >>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com> >>> --- >>> block/qcow2.h | 29 + >>> block/qcow2-compressed-write-cache.c | 770 +++++++++++++++++++++++++++ >>> block/meson.build | 1 + >>> 3 files changed, 800 insertions(+) >>> create mode 100644 block/qcow2-compressed-write-cache.c >>> >>> diff --git a/block/qcow2.h b/block/qcow2.h >>> index 0678073b74..fbdedf89fa 100644 >>> --- a/block/qcow2.h >>> +++ b/block/qcow2.h >>> @@ -322,6 +322,8 @@ typedef struct Qcow2BitmapHeaderExt { >>> uint64_t bitmap_directory_offset; >>> } QEMU_PACKED Qcow2BitmapHeaderExt; >>> +typedef struct Qcow2CompressedWriteCache Qcow2CompressedWriteCache; >>> + >>> #define QCOW2_MAX_THREADS 4 >>> typedef struct BDRVQcow2State { >>> @@ -1010,4 +1012,31 @@ int coroutine_fn >>> qcow2_co_decrypt(BlockDriverState *bs, uint64_t host_offset, >>> uint64_t guest_offset, void *buf, size_t len); >>> +Qcow2CompressedWriteCache *qcow2_compressed_cache_new(BdrvChild >>> *data_file, >>> + int64_t >>> cluster_size, >>> + int64_t >>> cache_size); >>> +void qcow2_compressed_cache_free(Qcow2CompressedWriteCache *s); >>> +int coroutine_fn >>> +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t >>> offset, >>> + int64_t bytes, void *buf); >>> +int coroutine_fn >>> +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, >>> int64_t offset, >>> + int64_t bytes, void *buf); >>> +void coroutine_fn >>> +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, >>> + int64_t cluster_data_end); >>> +int coroutine_fn >>> +qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s); >>> +int qcow2_compressed_cache_flush(BlockDriverState *bs, >>> + Qcow2CompressedWriteCache *state); >>> +int coroutine_fn >>> +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s); >>> +int qcow2_compressed_cache_stop_flush(BlockDriverState *bs, >>> + Qcow2CompressedWriteCache *s); >>> +void qcow2_compressed_cache_set_size(Qcow2CompressedWriteCache *s, >>> + int64_t size); >>> +void coroutine_fn >>> +qcow2_compressed_cache_co_discard(Qcow2CompressedWriteCache *s, >>> + int64_t cluster_offset); >>> + >> >> It would be nice if these functions had their interface documented >> somewhere. > > I tried to comment dificult things in .c... Is there a prefernce, where > to document > how and what function does, in .h or in .c ? No, but my problem was that I found even the things you probably didn’t consider difficult not completely obvious; i.e., I would’ve liked a full documentation on the function interface. (And I don’t think there is such documentation in the .c file.) >>> #endif >>> diff --git a/block/qcow2-compressed-write-cache.c >>> b/block/qcow2-compressed-write-cache.c >>> new file mode 100644 >>> index 0000000000..7bb92cb550 >>> --- /dev/null >>> +++ b/block/qcow2-compressed-write-cache.c >>> @@ -0,0 +1,770 @@ >>> +/* >>> + * Write cache for qcow2 compressed writes >>> + * >>> + * Copyright (c) 2021 Virtuozzo International GmbH. >>> + * >>> + * Permission is hereby granted, free of charge, to any person >>> obtaining a copy >>> + * of this software and associated documentation files (the >>> "Software"), to deal >>> + * in the Software without restriction, including without limitation >>> the rights >>> + * to use, copy, modify, merge, publish, distribute, sublicense, >>> and/or sell >>> + * copies of the Software, and to permit persons to whom the >>> Software is >>> + * furnished to do so, subject to the following conditions: >>> + * >>> + * The above copyright notice and this permission notice shall be >>> included in >>> + * all copies or substantial portions of the Software. >>> + * >>> + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, >>> EXPRESS OR >>> + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF >>> MERCHANTABILITY, >>> + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT >>> SHALL >>> + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES >>> OR OTHER >>> + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, >>> ARISING FROM, >>> + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER >>> DEALINGS IN >>> + * THE SOFTWARE. >>> + */ >>> + >>> +#include "qemu/osdep.h" >>> + >>> +#include "block/block_int.h" >>> +#include "block/block-gen.h" >>> +#include "qemu/coroutine.h" >>> +#include "qapi/qapi-events-block-core.h" >>> +#include "qcow2.h" >>> + >>> +typedef struct CacheExtent { >>> + int64_t offset; >>> + int64_t bytes; >>> + void *buf; >>> + QLIST_ENTRY(CacheExtent) next; >>> +} CacheExtent; >>> + >>> +typedef struct CacheCluster { >> >> It isn’t immediately clear what these two structures mean by just >> their name, because “extent” has no meaning in the context of qcow2. > > It's not important for the cache are extents compressed clusters or > not.. So I'd keep more generic name OK, but then my problem was that I had no idea what it’s supposed to represent. I had to read the code, which I don’t think is ideal (i.e., I’d like to know what the structures are so I can understand what the code does, not having to figure out both at the same time). >> I understand CacheExtent to basically be a compressed cluster, and >> CacheCluster to be a host cluster. Perhaps their names should reflect >> that. >> >> (OTOH, the Cache* prefix seems unnecessary to me, because these are >> just local structs.) [...] >>> + >>> +struct Qcow2CompressedWriteCache { >>> + /* >>> + * @data_file and @cluster_size are copied from qcow2 state. Not >>> huge >>> + * duplications, seems better to avoid accessing the whole qcow2 >>> state >>> + * instead. >> >> Why? What if in the future the data_file child can be changed with >> reopen? > > I didn't considered this.. Still it's impossible now, yes? But would be > possible in future of course. > >> >> What’s the argument against accessing the whole qcow2 state? That >> it’s a layering violation, that we shouldn’t access it while the rest >> uses it? > > To keep it simple to reuse the cache for something other. I could give the argument of “isn’t that impossible right now?” right back. But anyhow, that is a reason for “why”, but then it should be noted in the comment (i.e., not saying it’s better, but saying why it’s necessary). >> If that’s a concern, perhaps *data_file should be an indirect pointer >> to the pointer in the BDRVQcow2State. > > Or we should update it on reopen (like cache_size) Yes, if we keep it duplicated here, then we’ll need to do it. (The code didn’t note why it needs to be duplicated here, so if there had been no reason, it would’ve been easier not to duplicate it.) [...] >>> +static int coroutine_fn >>> flush_consequitive_extents(Qcow2CompressedWriteCache *s, >>> + CacheExtent *first, >>> + CacheExtent **next, >>> + int64_t align) >>> +{ >>> + CacheExtent *e, *end_extent; >>> + int64_t end; >>> + int nb_extents = count_consequitive_extents(first, &end_extent, >>> &end); >> >> I’d prefer nb_bufs, because it will include a tail if present. >> >>> + int64_t aligned_end = QEMU_ALIGN_UP(end, align); >>> + int64_t tail = aligned_end - end; >>> + int64_t len = aligned_end - first->offset; >>> + >>> + /* >>> + * Alignment if for flushing full cluster, first extent offset >>> is always >> >> s/.*,/The alignment will not exceed the cluster size, so the/? > > s/if/is/. When cluster is "full", which means it is filled up to its > data_end, the remaining tail is unsed, and we can align it up. That doesn’t really help me understand. What does “Alignment is for flushing full cluster” mean? That the alignment is equal to the cluster size? I think there should still be an "if" in there somewhere. I think the full explanation for why this assertion is here and what it means is “If a full cluster is flushed and so a block alignment has been given*, the first extent's offset will be at the start of the cluster. Otherwise, the alignment is 1. Therefore, if an alignment greater than 1 is given, at most we need to add a tail, but not a head (which is what this assertion shows).”, but perhaps I’m still wrong. *I think this statement then requires a comment on what the alignment is even for. I’m not sure myself, because, well, it isn’t commented on anywhere. I think it’s because we want to avoid RMW cycles for O_DIRECT, so the alignment most of the time will fit the host device block size (though at least 4k, I suppose to include devices with a logical block size of 512, but a physical block size of 4k). But we can’t necessarily do that for non-full clusters, because those might need a padding head to accomplish that, but that might overwrite existing data that we don’t want to overwrite, so to keep it simple, we just align=1 there. Is that about right? [...] >>> +/* @cluster takes ownership of @extent */ >>> +static void cluster_add_extent(Qcow2CompressedWriteCache *s, >>> + CacheCluster *cluster, CacheExtent >>> *extent) >>> +{ >>> + CacheExtent *e; >>> + >>> + assert(extent->bytes); >>> + assert(extent->offset >= cluster->cluster_offset); >>> + assert(extent->offset + extent->bytes <= cluster->data_end); >>> + assert(!cluster->in_flight); >>> + >>> + e = QLIST_FIRST(&cluster->extents); >>> + if (!e) { >>> + /* inactive cluster */ >>> + assert(!cluster->n_bytes); >>> + s->nb_active_clusters++; >>> + assert(s->nb_active_clusters <= s->max_active_clusters); >>> + QLIST_INSERT_HEAD(&cluster->extents, extent, next); >>> + } else if (e->offset > extent->offset) { >>> + assert(extent->offset + extent->bytes <= e->offset); >>> + QLIST_INSERT_HEAD(&cluster->extents, extent, next); >>> + } else { >>> + while (QLIST_NEXT(e, next) && >>> + QLIST_NEXT(e, next)->offset < extent->offset) { >>> + e = QLIST_NEXT(e, next); >>> + } >>> + >>> + /* Now e is last element with offset < extent->offset */ >>> + assert(e->offset + e->bytes <= extent->offset); >>> + >>> + QLIST_INSERT_AFTER(e, extent, next); >>> + >>> + e = QLIST_NEXT(extent, next); >>> + if (e) { >>> + assert(extent->offset + extent->bytes <= e->offset); >>> + } >>> + } >> >> I like how in C one always tends to implement everything from >> scratch.I think adding an element into an already-sorted list is a >> standard problem, so it’s a shame that it’s implemented from scratch >> here. Also that it doesn’t use binary search, which would require a >> random-access list, so... > > Yes, we have g_list_insert_sorted(), but it doesn't use binary search > anyway. lol (I mean, it’s impossible to use binary search on a GList, but still.) > So I decided to use typed lists as Kevin said they are better.. > Still I'm not fun of all these macros. > >> >> Well. Not sure if it’s necessary. Probably not. Although I find it >> weird to start the search from the front when I’d imagine it’s more >> likely that new compressed clusters (extents) are added to the back. > > Thought about this too, but decided to not care. > > 1. I think, it's all doesn't really matter, as io write operation is a > lot longer than all these opeartions with a list. > > 2. Still it's interesting :) Yes, basically my thinking, too. I just couldn’t help but notice it seems like a standard problem that should have a pre-implemented simple solution, but I don’t think there’s a reason to spend too much time and effort here. > When adding extent we want to check that it doesn't intersect with other > extents. And we want to have sorted list when flush the cluster. > > In GLib there is GTree. It has g_tree_foreach in sorted order. But it > doesn't have an option to get neighbours of inserted element (to check > intersection), neither it have something like lower_bound and > upper_bound :\ > > Hmm, implementing a tree from stratch is not what I want to do in > context of my cache :\ Any ideas? > > wow, I found this: > https://gitlab.gnome.org/GNOME/glib/-/merge_requests/1509 > > So, 3 months ago all we need was merged into GLib master.. Aha and it's > in GLib 2.67.0.. and centos7 has 2.56.. > > I can add a comment: "refactor using GTree node-based API when it become > available (it will in GLib 2.67)" Why not. Might be cool. >>> + >>> + cluster->n_bytes += extent->bytes; >>> +} >>> + >>> +static CacheCluster *find_cluster_to_flush(Qcow2CompressedWriteCache >>> *s) >>> +{ >>> + CacheCluster *c; >>> + >>> + QTAILQ_FOREACH(c, &s->clusters, next) { >>> + if (!c->in_flight && cache_cluster_is_full(c)) { >>> + return c; >> >> I don’t like it very much that this cache is built as an R/W cache, >> i.e. that entries are retained until they need to be flushed because a >> new entry needs space. >> >> It was my impression this was meant as a write cache, that should >> speed up specifically O_DIRECT operation. To me, that implies that >> entries are flushed once they are done, which is precisely why it >> works for compressed clusters, because we know when those are done. > > We learn that cluster is finished in two cases: > > 1. when data_end is set and we see that cluster is full. That's a bad > place for flushing, as we are under qcow2 mutex. > 2. when we add an extent. That's occures during some write operation.. > And I don't see real difference, flushing cluster here or when we want > to create one more cluster. > > And because we don't want to flush at [1], we'll probably have some lost > full cluster if flush only on [2]. So, we anyway should flush when want > to create new cluster but cache is full. And then no real reason to > flush on [2] Yes, I acknowledged that for qcow2_compressed_cache_co_set_cluster_end() (“I get the intention here, because as documented somewhere, this function is called from a mutexed section and we probably don’t want to write data to the data_file here.”). But I also raised the question of whether we might delay the set_cluster_end() or the flush until the mutex is no longer taken, so we’d still flush as soon as possible. I suppose if you want a more versatile cache, delaying the flush until the cache is needed will make sense, so it also serves as a read cache, but in its advertised form for caching compressed clusters specifically for the O_DIRECT case, I’d prefer to flush as soon as possible. >> I suppose nobody cares in practice because compression is basically >> only used for writing out whole images (so displacing clusters happens >> all the time, and once you’re done, the image is flushed and closed), >> but it just looks strange to me. >> >>> + } >>> + } >>> + >>> + return NULL; >>> +} >>> + >>> +/* Cache an extent if there is a place */ >>> +static bool coroutine_fn >>> +try_cache_extent(Qcow2CompressedWriteCache *s, CacheExtent *extent, >>> + bool *cluster_in_flight) >> >> Like with other functions, but here especially, I would have liked >> some documentation on the interface. > > will add > >> I suppose the return value reflects whether the “try” worked (so a >> good guess is possible to me, but it’s still a guess at this point). >> >> I have no idea what *cluster_in_flight means. > > Neither me at this point, I don't remember:) But everything is so > obvious when you are writing the code.. > > Ok, looking forward, *cluster_in_flight means that cluster is now > flushing, so please don't even retry to cache this extent! But that doesn’t apply when the cache is disabled, in which case *cluster_in_flight is also set. >> I would have guessed it means whether the host cluster touched by the >> compressed cluster is in flight, but it’s also set when the cache is >> disabled. The caller seems to interpret it as “write this >> immediately”, which to me actually seems wrong, but more on that in >> said caller function below. >> >> Perhaps this function should just have an enum return value that tells >> the caller precisely what to do with some expressively named values. So this is still something I think might make sense. [...] >>> + >>> + cache_extent_free(e); >>> + extents[i] = NULL; >>> + nb_extents--; >>> + >>> + if (ret < 0) { >>> + goto out; >>> + } >>> + } >>> + } >>> + >>> + if (do_write) { >>> + /* >>> + * We yielded during second extent writing. Probably the >>> cache is >>> + * already free and we can now cache the first extent. >> >> Where do we yield? Do you mean it probably takes time to do the >> write, so it’s possible that if do_write is true because some cluster >> was in the process of being flushed, it is now removed from the cache, >> so we have a free entry? >> >> Why not just check for whether we have a free entry? > > and we do it, moving to try_cache_extent at start of loop.. Yes. But why not check here, and only continue if there is a free entry? (The problem with “continue” is that it’s a goto in disguise, so I don’t like using it just to avoid code duplication. I think “continue” should only be used if you really want to do the next iteration, but in this case it seems like you just want to avoid a duplicated check.) [...] >>> + CacheExtent *e; >>> + CacheCluster *c; >>> + >>> + c = find_cluster(s, cluster_offset); >>> + if (!c) { >>> + c = cache_cluster_new(s, cluster_offset); >> >> Is this case reachable? How? > > Ooops... I feel stupid. Great thing that you asked this question! > > Somehow I thought that already allocated clusters may be in progress > of... what? compression? writing?.. But actually allocation is of course > done after compression. So we may compress clusters in parallel, but > than they allocated and written sequentially. Of courese without the > cache the writes themselves may go in parallel. But write to cache is > immediate and doesn't yield. We even can do it before unlocking qcow2 > mutex to be sure that we don't switch to another coroutine to write next > extent first. > > Seems I was wrong, and all this can be a lot simpler. Very sorry for > your time. But it helped me to open my eyes :\ Well, looking forward to v2 then. :) Max
diff --git a/block/qcow2.h b/block/qcow2.h index 0678073b74..fbdedf89fa 100644 --- a/block/qcow2.h +++ b/block/qcow2.h @@ -322,6 +322,8 @@ typedef struct Qcow2BitmapHeaderExt { uint64_t bitmap_directory_offset; } QEMU_PACKED Qcow2BitmapHeaderExt; +typedef struct Qcow2CompressedWriteCache Qcow2CompressedWriteCache; + #define QCOW2_MAX_THREADS 4 typedef struct BDRVQcow2State { @@ -1010,4 +1012,31 @@ int coroutine_fn qcow2_co_decrypt(BlockDriverState *bs, uint64_t host_offset, uint64_t guest_offset, void *buf, size_t len); +Qcow2CompressedWriteCache *qcow2_compressed_cache_new(BdrvChild *data_file, + int64_t cluster_size, + int64_t cache_size); +void qcow2_compressed_cache_free(Qcow2CompressedWriteCache *s); +int coroutine_fn +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t offset, + int64_t bytes, void *buf); +int coroutine_fn +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, int64_t offset, + int64_t bytes, void *buf); +void coroutine_fn +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, + int64_t cluster_data_end); +int coroutine_fn +qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s); +int qcow2_compressed_cache_flush(BlockDriverState *bs, + Qcow2CompressedWriteCache *state); +int coroutine_fn +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s); +int qcow2_compressed_cache_stop_flush(BlockDriverState *bs, + Qcow2CompressedWriteCache *s); +void qcow2_compressed_cache_set_size(Qcow2CompressedWriteCache *s, + int64_t size); +void coroutine_fn +qcow2_compressed_cache_co_discard(Qcow2CompressedWriteCache *s, + int64_t cluster_offset); + #endif diff --git a/block/qcow2-compressed-write-cache.c b/block/qcow2-compressed-write-cache.c new file mode 100644 index 0000000000..7bb92cb550 --- /dev/null +++ b/block/qcow2-compressed-write-cache.c @@ -0,0 +1,770 @@ +/* + * Write cache for qcow2 compressed writes + * + * Copyright (c) 2021 Virtuozzo International GmbH. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +#include "qemu/osdep.h" + +#include "block/block_int.h" +#include "block/block-gen.h" +#include "qemu/coroutine.h" +#include "qapi/qapi-events-block-core.h" +#include "qcow2.h" + +typedef struct CacheExtent { + int64_t offset; + int64_t bytes; + void *buf; + QLIST_ENTRY(CacheExtent) next; +} CacheExtent; + +typedef struct CacheCluster { + int64_t cluster_offset; + int64_t n_bytes; /* sum of extents lengths */ + + /* + * data_end: cluster is full if data_end reached and ready to be flushed. + * data_end is absoluted offset, not relative. + */ + int64_t data_end; + + bool in_flight; /* cluster is being flushed now */ + + /* + * waiters: coroutines to wake after flush. Must be empty when in_flight is + * false + */ + CoQueue waiters; + + QTAILQ_ENTRY(CacheCluster) next; + QLIST_HEAD(, CacheExtent) extents; /* sorted by offset */ +} CacheCluster; + +typedef QTAILQ_HEAD(, CacheCluster) CacheClusterQueue; + +struct Qcow2CompressedWriteCache { + /* + * @data_file and @cluster_size are copied from qcow2 state. Not huge + * duplications, seems better to avoid accessing the whole qcow2 state + * instead. + */ + BdrvChild *data_file; + int64_t cluster_size; + + CoQueue waiters; /* coroutines, waiting for free place in the cache */ + + /* + * Cache limits not total number of clusters but total number of active + * clusters. Active clusters are clusters with non-zero @n_bytes (and + * therefor non-empty @extents). This is done so we don't need to wait for + * cache flush qcow2_compressed_cache_set_cluster_end() (which may create + * cluster with @data_end set but zero @n_bytes), as + * qcow2_compressed_cache_set_cluster_end() is intended to be called from + * qcow2 mutex critical section. + */ + int nb_active_clusters; + + /* + * If max_active_clusters is 0 it means that cache is inactive: all new + * writes should fallthrough to data_file immediately. + */ + int max_active_clusters; + + CacheClusterQueue clusters; +}; + +/* New extent takes ownership of @buf */ +static CacheExtent *cache_extent_new(int64_t offset, int64_t bytes, void *buf) +{ + CacheExtent *e = g_new(CacheExtent, 1); + + *e = (CacheExtent) { + .offset = offset, + .bytes = bytes, + .buf = buf, + }; + + return e; +} + +static void cache_extent_free(CacheExtent *e) +{ + if (e) { + g_free(e->buf); + g_free(e); + } +} + +static CacheCluster *find_cluster(Qcow2CompressedWriteCache *s, + int64_t cluster_offset) +{ + CacheCluster *c; + + assert(QEMU_IS_ALIGNED(cluster_offset, s->cluster_size)); + + QTAILQ_FOREACH(c, &s->clusters, next) { + if (cluster_offset == c->cluster_offset) { + return c; + } + } + + return NULL; +} + +/* Creates "inactive" cluster, which doesn't influence s->nb_active_clusters */ +static CacheCluster *cache_cluster_new(Qcow2CompressedWriteCache *s, + int64_t cluster_offset) +{ + CacheCluster *c; + + assert(!find_cluster(s, cluster_offset)); + + c = g_new(CacheCluster, 1); + *c = (CacheCluster) { + .cluster_offset = cluster_offset, + .data_end = cluster_offset + s->cluster_size + }; + + qemu_co_queue_init(&c->waiters); + QTAILQ_INSERT_TAIL(&s->clusters, c, next); + + return c; +} + +static void cache_cluster_free(CacheCluster *cluster) +{ + assert(!cluster->in_flight); + assert(qemu_co_queue_empty(&cluster->waiters)); + QLIST_FOREACH_FUNC_SAFE(&cluster->extents, next, cache_extent_free); + g_free(cluster); +} + +static bool cache_cluster_is_full(CacheCluster *cluster) +{ + assert(cluster->cluster_offset + cluster->n_bytes <= cluster->data_end); + return cluster->cluster_offset + cluster->n_bytes == cluster->data_end; +} + +static void cache_cluster_remove(Qcow2CompressedWriteCache *s, + CacheCluster *cluster) +{ + if (cluster->n_bytes) { + s->nb_active_clusters--; + } + QTAILQ_REMOVE(&s->clusters, cluster, next); + cache_cluster_free(cluster); +} + +/* + * Return number of consequtive clusters starting from @first. Store next after + * last extent pointer into @next, store end offset of last extent into + * @end_off. + */ +static int count_consequitive_extents(CacheExtent *first, + CacheExtent **next, + int64_t *end_off) +{ + CacheExtent *e; + int64_t off = first->offset; + int nb = 0; + + for (e = first; e; e = QLIST_NEXT(e, next)) { + assert(e->offset >= off); + if (e->offset > off) { + break; + } + off += e->bytes; + nb++; + } + + if (next) { + *next = e; + } + + if (end_off) { + *end_off = off; + } + + return nb; +} + +/* + * Write consequtive extents, starting from @firest. Store next after last + * extent pointer into @next. If align > 1, align end of the whole write by + * zero. + */ +static int coroutine_fn flush_consequitive_extents(Qcow2CompressedWriteCache *s, + CacheExtent *first, + CacheExtent **next, + int64_t align) +{ + CacheExtent *e, *end_extent; + int64_t end; + int nb_extents = count_consequitive_extents(first, &end_extent, &end); + int64_t aligned_end = QEMU_ALIGN_UP(end, align); + int64_t tail = aligned_end - end; + int64_t len = aligned_end - first->offset; + + /* + * Alignment if for flushing full cluster, first extent offset is always + * aligned. + */ + assert(QEMU_IS_ALIGNED(first->offset, align)); + + if (next) { + *next = end_extent; + } + + if (tail) { + nb_extents++; + } + + if (nb_extents > IOV_MAX) { + g_autofree void *buf = g_malloc(len); + char *p = buf; + + for (e = first; e != end_extent; e = QLIST_NEXT(e, next)) { + memcpy(p, e->buf, e->bytes); + p += e->bytes; + } + + if (tail) { + memset(p, 0, tail); + } + + BLKDBG_EVENT(s->data_file, BLKDBG_WRITE_COMPRESSED); + return bdrv_co_pwrite(s->data_file, first->offset, len, buf, 0); + } else { + int ret; + QEMUIOVector qiov; + g_autofree void *tail_buf = NULL; + + qemu_iovec_init(&qiov, nb_extents); + for (e = first; e != end_extent; e = QLIST_NEXT(e, next)) { + qemu_iovec_add(&qiov, e->buf, e->bytes); + } + + if (tail) { + tail_buf = g_malloc0(tail); + qemu_iovec_add(&qiov, tail_buf, tail); + } + + BLKDBG_EVENT(s->data_file, BLKDBG_WRITE_COMPRESSED); + ret = bdrv_co_pwritev(s->data_file, first->offset, len, &qiov, 0); + qemu_iovec_destroy(&qiov); + return ret; + } +} + +static int coroutine_fn cache_cluster_flush_full(Qcow2CompressedWriteCache *s, + CacheCluster *cluster) +{ + int ret; + CacheExtent *end_extent; + int64_t align = MIN(s->cluster_size, + MAX(s->data_file->bs->bl.request_alignment, 4 * 1024)); + + assert(cache_cluster_is_full(cluster)); + + ret = flush_consequitive_extents(s, QLIST_FIRST(&cluster->extents), + &end_extent, align); + + assert(end_extent == NULL); /* all extents flushed */ + + return ret; +} + +static int coroutine_fn cache_cluster_flush(Qcow2CompressedWriteCache *s, + CacheCluster *c) +{ + int ret; + CacheExtent *e = QLIST_FIRST(&c->extents); + + if (cache_cluster_is_full(c)) { + return cache_cluster_flush_full(s, c); + } + + while (e) { + ret = flush_consequitive_extents(s, e, &e, 1); + if (ret < 0) { + return ret; + } + } + + return 0; +} + +int coroutine_fn qcow2_compressed_cache_co_flush(Qcow2CompressedWriteCache *s) +{ + int ret = 0; + CacheCluster *c; + GList *local_clusters = NULL, *p; + + /* + * Make a snapshot of current state: we will not flush clusters created in + * parallel with flush operations and don't allow adding more extents to + * staged clusters. We are also protected from parallel flush operations + * flushing the same clusters. + */ + QTAILQ_FOREACH(c, &s->clusters, next) { + if (!c->in_flight && c->n_bytes) { + c->in_flight = true; + local_clusters = g_list_append(local_clusters, c); + } + } + + for (p = local_clusters; p; p = p->next) { + CacheCluster *c = p->data; + + if (ret == 0) { + ret = cache_cluster_flush(s, c); + } + + c->in_flight = false; + qemu_co_queue_restart_all(&c->waiters); + + if (ret == 0) { + cache_cluster_remove(s, c); + } + } + + g_list_free(local_clusters); + + return ret; +} + +int coroutine_fn +qcow2_compressed_cache_co_stop_flush(Qcow2CompressedWriteCache *s) +{ + int ret, save; + + save = s->max_active_clusters; + s->max_active_clusters = 0; /* No more extents */ + + ret = qcow2_compressed_cache_co_flush(s); + if (ret < 0) { + s->max_active_clusters = save; + return ret; + } + + assert(QTAILQ_EMPTY(&s->clusters)); + return 0; +} + +/* @cluster takes ownership of @extent */ +static void cluster_add_extent(Qcow2CompressedWriteCache *s, + CacheCluster *cluster, CacheExtent *extent) +{ + CacheExtent *e; + + assert(extent->bytes); + assert(extent->offset >= cluster->cluster_offset); + assert(extent->offset + extent->bytes <= cluster->data_end); + assert(!cluster->in_flight); + + e = QLIST_FIRST(&cluster->extents); + if (!e) { + /* inactive cluster */ + assert(!cluster->n_bytes); + s->nb_active_clusters++; + assert(s->nb_active_clusters <= s->max_active_clusters); + QLIST_INSERT_HEAD(&cluster->extents, extent, next); + } else if (e->offset > extent->offset) { + assert(extent->offset + extent->bytes <= e->offset); + QLIST_INSERT_HEAD(&cluster->extents, extent, next); + } else { + while (QLIST_NEXT(e, next) && + QLIST_NEXT(e, next)->offset < extent->offset) { + e = QLIST_NEXT(e, next); + } + + /* Now e is last element with offset < extent->offset */ + assert(e->offset + e->bytes <= extent->offset); + + QLIST_INSERT_AFTER(e, extent, next); + + e = QLIST_NEXT(extent, next); + if (e) { + assert(extent->offset + extent->bytes <= e->offset); + } + } + + cluster->n_bytes += extent->bytes; +} + +static CacheCluster *find_cluster_to_flush(Qcow2CompressedWriteCache *s) +{ + CacheCluster *c; + + QTAILQ_FOREACH(c, &s->clusters, next) { + if (!c->in_flight && cache_cluster_is_full(c)) { + return c; + } + } + + return NULL; +} + +/* Cache an extent if there is a place */ +static bool coroutine_fn +try_cache_extent(Qcow2CompressedWriteCache *s, CacheExtent *extent, + bool *cluster_in_flight) +{ + CacheCluster *c; + int64_t cluster_offset = QEMU_ALIGN_DOWN(extent->offset, s->cluster_size); + + assert(extent->bytes); + + if (s->max_active_clusters == 0) { + *cluster_in_flight = true; + return false; + } + + *cluster_in_flight = false; + + c = find_cluster(s, cluster_offset); + if (c && c->in_flight) { + *cluster_in_flight = true; + return false; + } + if (s->nb_active_clusters >= s->max_active_clusters && + (!c || !c->n_bytes)) + { + /* + * Cache is full, we can't allocate a new cluster and can't activate + * existing inactive cluster + */ + return false; + } + + if (!c) { + c = cache_cluster_new(s, cluster_offset); + } + + cluster_add_extent(s, c, extent); + + if (cache_cluster_is_full(c)) { + qemu_co_queue_restart_all(&s->waiters); + } + + return true; +} + +/* Takes ownership of @buf, don't free it after call! */ +int coroutine_fn +qcow2_compressed_cache_co_write(Qcow2CompressedWriteCache *s, int64_t offset, + int64_t bytes, void *buf) +{ + int ret; + int i; + CacheExtent *extents[] = {NULL, NULL}; + int nb_extents = 0; /* number of non-NULL elements in @extents */ + int64_t e0_len; + + assert(bytes > 0); + assert(bytes < s->cluster_size); + + e0_len = MIN(bytes, QEMU_ALIGN_UP(offset + 1, s->cluster_size) - offset); + extents[0] = cache_extent_new(offset, e0_len, buf); + nb_extents = 1; + + if (bytes > e0_len) { + int64_t e1_len = bytes - e0_len; + /* + * We have to allocate a new buffer, so that clusters are independent + * and can free their extents when needed. + */ + void *e1_buf = g_memdup(((const char *)buf) + e0_len, e1_len); + + extents[1] = cache_extent_new(offset + e0_len, e1_len, e1_buf); + nb_extents = 2; + } + + while (nb_extents) { + bool do_write = false; + + for (i = 0; i < 2; i++) { + CacheExtent *e = extents[i]; + + do_write = false; + + if (!e) { + continue; + } + + if (try_cache_extent(s, e, &do_write)) { + extents[i] = NULL; + nb_extents--; + continue; + } + + if (do_write) { + ret = bdrv_co_pwrite(s->data_file, e->offset, e->bytes, + e->buf, 0); + + cache_extent_free(e); + extents[i] = NULL; + nb_extents--; + + if (ret < 0) { + goto out; + } + } + } + + if (do_write) { + /* + * We yielded during second extent writing. Probably the cache is + * already free and we can now cache the first extent. + */ + continue; + } + + if (nb_extents) { + CacheCluster *cluster = find_cluster_to_flush(s); + + if (cluster) { + cluster->in_flight = true; + ret = cache_cluster_flush_full(s, cluster); + cluster->in_flight = false; + qemu_co_queue_restart_all(&cluster->waiters); + qemu_co_queue_restart_all(&s->waiters); + if (ret < 0) { + goto out; + } + cache_cluster_remove(s, cluster); + continue; + } + + qemu_co_queue_wait(&s->waiters, NULL); + } + } + + ret = 0; + +out: + for (i = 0; i < 2; i++) { + cache_extent_free(extents[i]); + } + + return 0; +} + +int coroutine_fn +qcow2_compressed_cache_co_read(Qcow2CompressedWriteCache *s, int64_t offset, + int64_t bytes, void *buf) +{ + CacheCluster *c; + CacheExtent *e; + int64_t cluster_offset = QEMU_ALIGN_DOWN(offset, s->cluster_size); + + c = find_cluster(s, cluster_offset); + if (c) { + QLIST_FOREACH(e, &c->extents, next) { + if (e->offset == offset && e->bytes <= bytes) { + memcpy(buf, e->buf, e->bytes); + return 0; + } + } + } + + return bdrv_co_pread(s->data_file, offset, bytes, buf, 0); +} + +/* + * Caller states, that there would be no writes to this cluster beyond + * specified @cluster_data_end. So, it's OK to flush cluster when it is filled + * up to @cluster_data_end and it's OK to align flushing write operation up to + * some alignment (not greater than cluster_size of course). + */ +void coroutine_fn +qcow2_compressed_cache_co_set_cluster_end(Qcow2CompressedWriteCache *s, + int64_t cluster_data_end) +{ + int64_t cluster_offset = QEMU_ALIGN_DOWN(cluster_data_end, s->cluster_size); + CacheExtent *e; + CacheCluster *c; + + c = find_cluster(s, cluster_offset); + if (!c) { + c = cache_cluster_new(s, cluster_offset); + } + + QLIST_FOREACH(e, &c->extents, next) { + assert(e->offset + e->bytes <= cluster_data_end); + } + + /* Shouldn't set data_end several times */ + assert(c->data_end == c->cluster_offset + s->cluster_size); + + c->data_end = cluster_data_end; + if (cache_cluster_is_full(c)) { + qemu_co_queue_restart_all(&s->waiters); + } +} + +Qcow2CompressedWriteCache *qcow2_compressed_cache_new(BdrvChild *data_file, + int64_t cluster_size, + int64_t cache_size) +{ + Qcow2CompressedWriteCache *s = g_new(Qcow2CompressedWriteCache, 1); + + assert(cache_size >= cluster_size); + + *s = (Qcow2CompressedWriteCache) { + .data_file = data_file, + .cluster_size = cluster_size, + .max_active_clusters = cache_size / cluster_size, + }; + + qemu_co_queue_init(&s->waiters); + QTAILQ_INIT(&s->clusters); + + return s; +} + +void qcow2_compressed_cache_free(Qcow2CompressedWriteCache *s) +{ + if (!s) { + return; + } + + QTAILQ_FOREACH_FUNC_SAFE(&s->clusters, next, cache_cluster_free); + g_free(s); +} + +void qcow2_compressed_cache_set_size(Qcow2CompressedWriteCache *s, + int64_t size) +{ + /* + * We don't do flush here. Don't care too much, it's safe to have cache + * larger than maximum, it will only decrease until it reaches the new + * maximum. + */ + assert(size >= s->cluster_size); + s->max_active_clusters = size / s->cluster_size; +} + +void coroutine_fn +qcow2_compressed_cache_co_discard(Qcow2CompressedWriteCache *s, + int64_t cluster_offset) +{ + CacheCluster *c; + + while (true) { + c = find_cluster(s, cluster_offset); + if (!c) { + return; + } + if (!c->in_flight) { + cache_cluster_remove(s, c); + return; + } + qemu_co_queue_wait(&c->waiters, NULL); + } +} + + +/* + * Wrappers for qcow2_compressed_cache_co_flush + * + * TODO: update scripts/block-coroutine-wrapper to generate this too + */ + +typedef struct Qcow2CompressedCacheCoFlush { + BdrvPollCo poll_state; + Qcow2CompressedWriteCache *state; +} Qcow2CompressedCacheCoFlush; + +static void coroutine_fn +qcow2_compressed_cache_co_flush_entry(void *opaque) +{ + Qcow2CompressedCacheCoFlush *s = opaque; + + s->poll_state.ret = qcow2_compressed_cache_co_flush(s->state); + s->poll_state.in_progress = false; + + aio_wait_kick(); +} + +int qcow2_compressed_cache_flush(BlockDriverState *bs, + Qcow2CompressedWriteCache *state) +{ + if (qemu_in_coroutine()) { + return qcow2_compressed_cache_co_flush(state); + } else { + Qcow2CompressedCacheCoFlush s = { + .poll_state.bs = bs, + .poll_state.in_progress = true, + + .state = state, + }; + + s.poll_state.co = + qemu_coroutine_create(qcow2_compressed_cache_co_flush_entry, &s); + + return bdrv_poll_co(&s.poll_state); + } +} + +/* + * Wrappers for qcow2_compressed_cache_co_stop_flush + * + * TODO: update scripts/block-coroutine-wrapper to generate this too + */ + +typedef struct Qcow2CompressedCacheCoStopFlush { + BdrvPollCo poll_state; + Qcow2CompressedWriteCache *state; +} Qcow2CompressedCacheCoStopFlush; + +static void coroutine_fn +qcow2_compressed_cache_co_stop_flush_entry(void *opaque) +{ + Qcow2CompressedCacheCoStopFlush *s = opaque; + + s->poll_state.ret = qcow2_compressed_cache_co_stop_flush(s->state); + s->poll_state.in_progress = false; + + aio_wait_kick(); +} + +int qcow2_compressed_cache_stop_flush(BlockDriverState *bs, + Qcow2CompressedWriteCache *state) +{ + if (qemu_in_coroutine()) { + return qcow2_compressed_cache_co_stop_flush(state); + } else { + Qcow2CompressedCacheCoStopFlush s = { + .poll_state.bs = bs, + .poll_state.in_progress = true, + + .state = state, + }; + + s.poll_state.co = + qemu_coroutine_create(qcow2_compressed_cache_co_stop_flush_entry, + &s); + + return bdrv_poll_co(&s.poll_state); + } +} diff --git a/block/meson.build b/block/meson.build index eeaefe5809..7b94794c28 100644 --- a/block/meson.build +++ b/block/meson.build @@ -25,6 +25,7 @@ block_ss.add(files( 'qcow2-bitmap.c', 'qcow2-cache.c', 'qcow2-cluster.c', + 'qcow2-compressed-write-cache.c', 'qcow2-refcount.c', 'qcow2-snapshot.c', 'qcow2-threads.c',
Compressed writes and O_DIRECT are not friends: they works too slow, because compressed writes does many small unaligned to 512 writes. Let's introduce an internal cache, so that compressed writes may work well when O_DIRECT is on. Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com> --- block/qcow2.h | 29 + block/qcow2-compressed-write-cache.c | 770 +++++++++++++++++++++++++++ block/meson.build | 1 + 3 files changed, 800 insertions(+) create mode 100644 block/qcow2-compressed-write-cache.c