mirrors/scoutfs
1
0
Fork 0
mirror of https://github.com/versity/scoutfs.git synced 2026-09-06 08:06:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

scoutfs: stop livelocking in item_next

Browse Source 
scoutfs_item_next() could livelock given the right key and segment key
boundaries.  This was easiest to trigger with an fio command that wrote
a lot of data:

	fio --filesize=100m --nrfiles=25 --name=100m --numjobs=1 \
		--iodepth=1 --ioengine=sync --fallocate=0 \
		--rw=write --openfiles=256 \
		--directory=$TEST_DIR

There were two problems.

First, if it found a cached region that didn't contain a next item it
would try to read the *end* of the existing cached region instead of
trying to populate more items by reading from the key past the existing
cached region.  This is fixed by incrementing the key to read from after
setting it to the end of the cached region.

Second, it got totally confused by non-merged but adjacent cached
regions.  It would find a cached region that contains the search key and
try to read from the key after that region, but that key could also be
cached and just not merged with its previous region.  This is fixed by
(duh) having an allocated pos key that we set as we walk through cached
regions.  It used to always try and read from the search key which was
bonkers.

With these fixes fio now completes.

Signed-off-by: Zach Brown <zab@versity.com>
This commit is contained in:
Zach Brown
2017-09-05 14:22:47 -07:00
committed by Mark Fasheh
parent 82f8daaebf
commit 51e8b614e5
2 changed files with 70 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kmod/src/item.c
+40 -31
View File
@@ -866,7 +866,7 @@ int scoutfs_item_next(struct super_block *sb, struct scoutfs_key_buf *key,
{
struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
struct item_cache *cac = sbi->item_cache;
struct scoutfs_key_buf *read_start = NULL;
struct scoutfs_key_buf *pos = NULL;
struct scoutfs_key_buf *range_end = NULL;
struct cached_item *item;
unsigned long flags;
@@ -883,57 +883,66 @@ int scoutfs_item_next(struct super_block *sb, struct scoutfs_key_buf *key,
goto out;
}
read_start = scoutfs_key_alloc(sb, SCOUTFS_MAX_KEY_SIZE);
pos = scoutfs_key_alloc(sb, SCOUTFS_MAX_KEY_SIZE);
range_end = scoutfs_key_alloc(sb, SCOUTFS_MAX_KEY_SIZE);
if (!read_start || !range_end) {
if (!pos || !range_end) {
ret = -ENOMEM;
goto out;
}
scoutfs_key_copy(pos, key);
spin_lock_irqsave(&cac->lock, flags);
for(;;) {
/* see if we have a usable item in cache and before last */
cached = check_range(sb, &cac->ranges, key, range_end);
/* see if we have cache coverage of our iterator pos */
cached = check_range(sb, &cac->ranges, pos, range_end);
if (cached && (item = item_for_next(&cac->items, key,
range_end, last))) {
scoutfs_key_copy(key, item->key);
if (val) {
item_referenced(cac, item);
ret = scoutfs_kvec_memcpy(val, item->val);
} else {
ret = 0;
}
break;
}
trace_scoutfs_item_next_range_check(sb, !!cached, key,
pos, last, end, range_end);
if (!cached) {
/* missing cache starts at key */
scoutfs_key_copy(read_start, key);
/* populate missing cached range starting at pos */
spin_unlock_irqrestore(&cac->lock, flags);
} else if (scoutfs_key_compare(range_end, last) < 0) {
/* missing cache starts at range_end */
scoutfs_key_copy(read_start, range_end);
ret = scoutfs_manifest_read_items(sb, pos, end);
} else {
/* no items and we have cache between key and last */
spin_lock_irqsave(&cac->lock, flags);
if (ret)
break;
else
continue;
}
/* see if there's an item in the cached range from pos */
item = item_for_next(&cac->items, pos, range_end, last);
if (!item) {
if (scoutfs_key_compare(range_end, last) < 0) {
/* keep searching after empty cached range */
scoutfs_key_copy(pos, range_end);
scoutfs_key_inc(pos);
continue;
}
/* no item and cache covers last, done */
ret = -ENOENT;
break;
}
spin_unlock_irqrestore(&cac->lock, flags);
ret = scoutfs_manifest_read_items(sb, read_start, end);
spin_lock_irqsave(&cac->lock, flags);
if (ret)
break;
/* we have a next item inside the cached range, done */
scoutfs_key_copy(key, item->key);
if (val) {
item_referenced(cac, item);
ret = scoutfs_kvec_memcpy(val, item->val);
} else {
ret = 0;
}
break;
}
spin_unlock_irqrestore(&cac->lock, flags);
out:
scoutfs_key_free(sb, read_start);
scoutfs_key_free(sb, pos);
scoutfs_key_free(sb, range_end);
trace_printk("ret %d\n", ret);
kmod/src/scoutfs_trace.h
+30
View File
@@ -484,6 +484,36 @@ DEFINE_EVENT(scoutfs_net_class, scoutfs_client_recv_reply,
TP_ARGS(sb, name, peer, nh)
);
TRACE_EVENT(scoutfs_item_next_range_check,
TP_PROTO(struct super_block *sb, int cached,
struct scoutfs_key_buf *key, struct scoutfs_key_buf *pos,
struct scoutfs_key_buf *last, struct scoutfs_key_buf *end,
struct scoutfs_key_buf *range_end),
TP_ARGS(sb, cached, key, pos, last, end, range_end),
TP_STRUCT__entry(
__field(void *, sb)
__field(int, cached)
__dynamic_array(char, key, scoutfs_key_str(NULL, key))
__dynamic_array(char, pos, scoutfs_key_str(NULL, pos))
__dynamic_array(char, last, scoutfs_key_str(NULL, last))
__dynamic_array(char, end, scoutfs_key_str(NULL, end))
__dynamic_array(char, range_end,
scoutfs_key_str(NULL, range_end))
),
TP_fast_assign(
__entry->sb = sb;
__entry->cached = cached;
scoutfs_key_str(__get_dynamic_array(key), key);
scoutfs_key_str(__get_dynamic_array(pos), pos);
scoutfs_key_str(__get_dynamic_array(last), last);
scoutfs_key_str(__get_dynamic_array(end), end);
scoutfs_key_str(__get_dynamic_array(range_end), range_end);
),
TP_printk("sb %p cached %d key %s pos %s last %s end %s range_end %s",
__entry->sb, __entry->cached, __get_str(key), __get_str(pos),
__get_str(last), __get_str(end), __get_str(range_end))
);
#endif /* _TRACE_SCOUTFS_H */
/* This part must be outside protection */