]> git.sesse.net Git - bcachefs-tools-debian/blobdiff - libbcachefs/movinggc.c
Update bcachefs sources to 0d63ed13ea3d closures: Fix race in closure_sync()
[bcachefs-tools-debian] / libbcachefs / movinggc.c
index f0ab65ffab7373dca455b4bec9528c31d6d0fbe3..f73b9b7f4bf7ee111e586754d868e13c5fc1d0bb 100644 (file)
 #include "alloc_foreground.h"
 #include "btree_iter.h"
 #include "btree_update.h"
+#include "btree_write_buffer.h"
 #include "buckets.h"
 #include "clock.h"
-#include "disk_groups.h"
 #include "errcode.h"
 #include "error.h"
-#include "extents.h"
-#include "eytzinger.h"
-#include "io.h"
-#include "keylist.h"
+#include "lru.h"
 #include "move.h"
 #include "movinggc.h"
-#include "super-io.h"
+#include "trace.h"
 
-#include <trace/events/bcachefs.h>
 #include <linux/freezer.h>
 #include <linux/kthread.h>
 #include <linux/math64.h>
 #include <linux/sched/task.h>
-#include <linux/sort.h>
 #include <linux/wait.h>
 
-static inline int fragmentation_cmp(copygc_heap *heap,
-                                  struct copygc_heap_entry l,
-                                  struct copygc_heap_entry r)
+struct buckets_in_flight {
+       struct rhashtable               table;
+       struct move_bucket_in_flight    *first;
+       struct move_bucket_in_flight    *last;
+       size_t                          nr;
+       size_t                          sectors;
+};
+
+static const struct rhashtable_params bch_move_bucket_params = {
+       .head_offset    = offsetof(struct move_bucket_in_flight, hash),
+       .key_offset     = offsetof(struct move_bucket_in_flight, bucket.k),
+       .key_len        = sizeof(struct move_bucket_key),
+};
+
+static struct move_bucket_in_flight *
+move_bucket_in_flight_add(struct buckets_in_flight *list, struct move_bucket b)
 {
-       return cmp_int(l.fragmentation, r.fragmentation);
+       struct move_bucket_in_flight *new = kzalloc(sizeof(*new), GFP_KERNEL);
+       int ret;
+
+       if (!new)
+               return ERR_PTR(-ENOMEM);
+
+       new->bucket = b;
+
+       ret = rhashtable_lookup_insert_fast(&list->table, &new->hash,
+                                           bch_move_bucket_params);
+       if (ret) {
+               kfree(new);
+               return ERR_PTR(ret);
+       }
+
+       if (!list->first)
+               list->first = new;
+       else
+               list->last->next = new;
+
+       list->last = new;
+       list->nr++;
+       list->sectors += b.sectors;
+       return new;
 }
 
-static int find_buckets_to_copygc(struct bch_fs *c)
+static int bch2_bucket_is_movable(struct btree_trans *trans,
+                                 struct move_bucket *b, u64 time)
 {
-       copygc_heap *h = &c->copygc_heap;
-       struct btree_trans trans;
        struct btree_iter iter;
        struct bkey_s_c k;
+       struct bch_alloc_v4 _a;
+       const struct bch_alloc_v4 *a;
        int ret;
 
-       bch2_trans_init(&trans, c, 0, 0);
-
-       /*
-        * Find buckets with lowest sector counts, skipping completely
-        * empty buckets, by building a maxheap sorted by sector count,
-        * and repeatedly replacing the maximum element until all
-        * buckets have been visited.
-        */
-       h->used = 0;
-
-       for_each_btree_key(&trans, iter, BTREE_ID_alloc, POS_MIN,
-                          BTREE_ITER_PREFETCH, k, ret) {
-               struct bch_dev *ca = bch_dev_bkey_exists(c, iter.pos.inode);
-               struct copygc_heap_entry e;
-               struct bch_alloc_v4 a_convert;
-               const struct bch_alloc_v4 *a;
-
-               a = bch2_alloc_to_v4(k, &a_convert);
-
-               if ((a->data_type != BCH_DATA_btree &&
-                    a->data_type != BCH_DATA_user) ||
-                   a->dirty_sectors >= ca->mi.bucket_size ||
-                   bch2_bucket_is_open(c, iter.pos.inode, iter.pos.offset))
-                       continue;
+       if (bch2_bucket_is_open(trans->c,
+                               b->k.bucket.inode,
+                               b->k.bucket.offset))
+               return 0;
 
-               e = (struct copygc_heap_entry) {
-                       .dev            = iter.pos.inode,
-                       .gen            = a->gen,
-                       .replicas       = 1 + a->stripe_redundancy,
-                       .fragmentation  = div_u64((u64) a->dirty_sectors * (1ULL << 31),
-                                                 ca->mi.bucket_size),
-                       .sectors        = a->dirty_sectors,
-                       .bucket         = iter.pos.offset,
-               };
-               heap_add_or_replace(h, e, -fragmentation_cmp, NULL);
+       k = bch2_bkey_get_iter(trans, &iter, BTREE_ID_alloc,
+                              b->k.bucket, BTREE_ITER_CACHED);
+       ret = bkey_err(k);
+       if (ret)
+               return ret;
 
-       }
-       bch2_trans_iter_exit(&trans, &iter);
+       a = bch2_alloc_to_v4(k, &_a);
+       b->k.gen        = a->gen;
+       b->sectors      = a->dirty_sectors;
+
+       ret = data_type_movable(a->data_type) &&
+               a->fragmentation_lru &&
+               a->fragmentation_lru <= time;
 
-       bch2_trans_exit(&trans);
+       bch2_trans_iter_exit(trans, &iter);
        return ret;
 }
 
-static int bch2_copygc(struct bch_fs *c)
+static void move_buckets_wait(struct moving_context *ctxt,
+                             struct buckets_in_flight *list,
+                             bool flush)
 {
-       copygc_heap *h = &c->copygc_heap;
-       struct copygc_heap_entry e;
-       struct bch_move_stats move_stats;
-       struct bch_dev *ca;
-       unsigned dev_idx;
-       size_t heap_size = 0;
-       struct moving_context ctxt;
-       struct data_update_opts data_opts = {
-               .btree_insert_flags = BTREE_INSERT_USE_RESERVE|JOURNAL_WATERMARK_copygc,
-       };
-       int ret = 0;
+       struct move_bucket_in_flight *i;
+       int ret;
 
-       bch2_move_stats_init(&move_stats, "copygc");
+       while ((i = list->first)) {
+               if (flush)
+                       move_ctxt_wait_event(ctxt, !atomic_read(&i->count));
 
-       for_each_rw_member(ca, c, dev_idx)
-               heap_size += ca->mi.nbuckets >> 7;
+               if (atomic_read(&i->count))
+                       break;
 
-       if (h->size < heap_size) {
-               free_heap(&c->copygc_heap);
-               if (!init_heap(&c->copygc_heap, heap_size, GFP_KERNEL)) {
-                       bch_err(c, "error allocating copygc heap");
-                       return 0;
-               }
+               list->first = i->next;
+               if (!list->first)
+                       list->last = NULL;
+
+               list->nr--;
+               list->sectors -= i->bucket.sectors;
+
+               ret = rhashtable_remove_fast(&list->table, &i->hash,
+                                            bch_move_bucket_params);
+               BUG_ON(ret);
+               kfree(i);
        }
 
-       ret = find_buckets_to_copygc(c);
-       if (ret) {
-               bch2_fs_fatal_error(c, "error walking buckets to copygc!");
+       bch2_trans_unlock(ctxt->trans);
+}
+
+static bool bucket_in_flight(struct buckets_in_flight *list,
+                            struct move_bucket_key k)
+{
+       return rhashtable_lookup_fast(&list->table, &k, bch_move_bucket_params);
+}
+
+typedef DARRAY(struct move_bucket) move_buckets;
+
+static int bch2_copygc_get_buckets(struct moving_context *ctxt,
+                       struct buckets_in_flight *buckets_in_flight,
+                       move_buckets *buckets)
+{
+       struct btree_trans *trans = ctxt->trans;
+       struct bch_fs *c = trans->c;
+       struct btree_iter iter;
+       struct bkey_s_c k;
+       size_t nr_to_get = max_t(size_t, 16U, buckets_in_flight->nr / 4);
+       size_t saw = 0, in_flight = 0, not_movable = 0, sectors = 0;
+       int ret;
+
+       move_buckets_wait(ctxt, buckets_in_flight, false);
+
+       ret = bch2_btree_write_buffer_flush(trans);
+       if (bch2_fs_fatal_err_on(ret, c, "%s: error %s from bch2_btree_write_buffer_flush()",
+                                __func__, bch2_err_str(ret)))
                return ret;
-       }
 
-       if (!h->used) {
-               s64 wait = S64_MAX, dev_wait;
-               u64 dev_min_wait_fragmented = 0;
-               u64 dev_min_wait_allowed = 0;
-               int dev_min_wait = -1;
-
-               for_each_rw_member(ca, c, dev_idx) {
-                       struct bch_dev_usage usage = bch2_dev_usage_read(ca);
-                       s64 allowed = ((__dev_buckets_available(ca, usage, RESERVE_none) *
-                                              ca->mi.bucket_size) >> 1);
-                       s64 fragmented = usage.d[BCH_DATA_user].fragmented;
-
-                       dev_wait = max(0LL, allowed - fragmented);
-
-                       if (dev_min_wait < 0 || dev_wait < wait) {
-                               dev_min_wait = dev_idx;
-                               dev_min_wait_fragmented = fragmented;
-                               dev_min_wait_allowed    = allowed;
-                       }
+       ret = for_each_btree_key2_upto(trans, iter, BTREE_ID_lru,
+                                 lru_pos(BCH_LRU_FRAGMENTATION_START, 0, 0),
+                                 lru_pos(BCH_LRU_FRAGMENTATION_START, U64_MAX, LRU_TIME_MAX),
+                                 0, k, ({
+               struct move_bucket b = { .k.bucket = u64_to_bucket(k.k->p.offset) };
+               int ret2 = 0;
+
+               saw++;
+
+               if (!bch2_bucket_is_movable(trans, &b, lru_pos_time(k.k->p)))
+                       not_movable++;
+               else if (bucket_in_flight(buckets_in_flight, b.k))
+                       in_flight++;
+               else {
+                       ret2 = darray_push(buckets, b) ?: buckets->nr >= nr_to_get;
+                       if (ret2 >= 0)
+                               sectors += b.sectors;
                }
+               ret2;
+       }));
 
-               bch_err_ratelimited(c, "copygc requested to run but found no buckets to move! dev %u fragmented %llu allowed %llu",
-                                   dev_min_wait, dev_min_wait_fragmented, dev_min_wait_allowed);
-               return 0;
-       }
+       pr_debug("have: %zu (%zu) saw %zu in flight %zu not movable %zu got %zu (%zu)/%zu buckets ret %i",
+                buckets_in_flight->nr, buckets_in_flight->sectors,
+                saw, in_flight, not_movable, buckets->nr, sectors, nr_to_get, ret);
 
-       heap_resort(h, fragmentation_cmp, NULL);
+       return ret < 0 ? ret : 0;
+}
 
-       bch2_moving_ctxt_init(&ctxt, c, NULL, &move_stats,
-                             writepoint_ptr(&c->copygc_write_point),
-                             false);
+noinline
+static int bch2_copygc(struct moving_context *ctxt,
+                      struct buckets_in_flight *buckets_in_flight)
+{
+       struct btree_trans *trans = ctxt->trans;
+       struct bch_fs *c = trans->c;
+       struct data_update_opts data_opts = {
+               .btree_insert_flags = BCH_WATERMARK_copygc,
+       };
+       move_buckets buckets = { 0 };
+       struct move_bucket_in_flight *f;
+       struct move_bucket *i;
+       u64 moved = atomic64_read(&ctxt->stats->sectors_moved);
+       int ret = 0;
+
+       ret = bch2_copygc_get_buckets(ctxt, buckets_in_flight, &buckets);
+       if (ret)
+               goto err;
+
+       darray_for_each(buckets, i) {
+               if (unlikely(freezing(current)))
+                       break;
 
-       /* not correct w.r.t. device removal */
-       while (h->used && !ret) {
-               BUG_ON(!heap_pop(h, e, -fragmentation_cmp, NULL));
-               ret = __bch2_evacuate_bucket(&ctxt, POS(e.dev, e.bucket), e.gen,
-                                            data_opts);
+               f = move_bucket_in_flight_add(buckets_in_flight, *i);
+               ret = PTR_ERR_OR_ZERO(f);
+               if (ret == -EEXIST) { /* rare race: copygc_get_buckets returned same bucket more than once */
+                       ret = 0;
+                       continue;
+               }
+               if (ret == -ENOMEM) { /* flush IO, continue later */
+                       ret = 0;
+                       break;
+               }
+
+               ret = __bch2_evacuate_bucket(ctxt, f, f->bucket.k.bucket,
+                                            f->bucket.k.gen, data_opts);
+               if (ret)
+                       goto err;
        }
+err:
+       darray_exit(&buckets);
 
-       bch2_moving_ctxt_exit(&ctxt);
+       /* no entries in LRU btree found, or got to end: */
+       if (bch2_err_matches(ret, ENOENT))
+               ret = 0;
 
        if (ret < 0 && !bch2_err_matches(ret, EROFS))
-               bch_err(c, "error from bch2_move_data() in copygc: %s", bch2_err_str(ret));
+               bch_err_msg(c, ret, "from bch2_move_data()");
 
-       trace_and_count(c, copygc, c, atomic64_read(&move_stats.sectors_moved), 0, 0, 0);
+       moved = atomic64_read(&ctxt->stats->sectors_moved) - moved;
+       trace_and_count(c, copygc, c, moved, 0, 0, 0);
        return ret;
 }
 
@@ -189,13 +259,18 @@ unsigned long bch2_copygc_wait_amount(struct bch_fs *c)
        struct bch_dev *ca;
        unsigned dev_idx;
        s64 wait = S64_MAX, fragmented_allowed, fragmented;
+       unsigned i;
 
        for_each_rw_member(ca, c, dev_idx) {
                struct bch_dev_usage usage = bch2_dev_usage_read(ca);
 
-               fragmented_allowed = ((__dev_buckets_available(ca, usage, RESERVE_none) *
+               fragmented_allowed = ((__dev_buckets_available(ca, usage, BCH_WATERMARK_stripe) *
                                       ca->mi.bucket_size) >> 1);
-               fragmented = usage.d[BCH_DATA_user].fragmented;
+               fragmented = 0;
+
+               for (i = 0; i < BCH_DATA_NR; i++)
+                       if (data_type_movable(i))
+                               fragmented += usage.d[i].fragmented;
 
                wait = min(wait, max(0LL, fragmented_allowed - fragmented));
        }
@@ -203,27 +278,72 @@ unsigned long bch2_copygc_wait_amount(struct bch_fs *c)
        return wait;
 }
 
+void bch2_copygc_wait_to_text(struct printbuf *out, struct bch_fs *c)
+{
+       prt_printf(out, "Currently waiting for:     ");
+       prt_human_readable_u64(out, max(0LL, c->copygc_wait -
+                                       atomic64_read(&c->io_clock[WRITE].now)) << 9);
+       prt_newline(out);
+
+       prt_printf(out, "Currently waiting since:   ");
+       prt_human_readable_u64(out, max(0LL,
+                                       atomic64_read(&c->io_clock[WRITE].now) -
+                                       c->copygc_wait_at) << 9);
+       prt_newline(out);
+
+       prt_printf(out, "Currently calculated wait: ");
+       prt_human_readable_u64(out, bch2_copygc_wait_amount(c));
+       prt_newline(out);
+}
+
 static int bch2_copygc_thread(void *arg)
 {
        struct bch_fs *c = arg;
+       struct moving_context ctxt;
+       struct bch_move_stats move_stats;
        struct io_clock *clock = &c->io_clock[WRITE];
+       struct buckets_in_flight buckets;
        u64 last, wait;
        int ret = 0;
 
+       memset(&buckets, 0, sizeof(buckets));
+
+       ret = rhashtable_init(&buckets.table, &bch_move_bucket_params);
+       if (ret) {
+               bch_err_msg(c, ret, "allocating copygc buckets in flight");
+               return ret;
+       }
+
        set_freezable();
 
+       bch2_move_stats_init(&move_stats, "copygc");
+       bch2_moving_ctxt_init(&ctxt, c, NULL, &move_stats,
+                             writepoint_ptr(&c->copygc_write_point),
+                             false);
+
        while (!ret && !kthread_should_stop()) {
+               bch2_trans_unlock(ctxt.trans);
                cond_resched();
 
-               if (kthread_wait_freezable(c->copy_gc_enabled))
-                       break;
+               if (!c->copy_gc_enabled) {
+                       move_buckets_wait(&ctxt, &buckets, true);
+                       kthread_wait_freezable(c->copy_gc_enabled);
+               }
+
+               if (unlikely(freezing(current))) {
+                       move_buckets_wait(&ctxt, &buckets, true);
+                       __refrigerator(false);
+                       continue;
+               }
 
                last = atomic64_read(&clock->now);
                wait = bch2_copygc_wait_amount(c);
 
                if (wait > clock->max_slop) {
-                       trace_and_count(c, copygc_wait, c, wait, last + wait);
+                       c->copygc_wait_at = last;
                        c->copygc_wait = last + wait;
+                       move_buckets_wait(&ctxt, &buckets, true);
+                       trace_and_count(c, copygc_wait, c, wait, last + wait);
                        bch2_kthread_io_clock_wait(clock, last + wait,
                                        MAX_SCHEDULE_TIMEOUT);
                        continue;
@@ -232,12 +352,17 @@ static int bch2_copygc_thread(void *arg)
                c->copygc_wait = 0;
 
                c->copygc_running = true;
-               ret = bch2_copygc(c);
+               ret = bch2_copygc(&ctxt, &buckets);
                c->copygc_running = false;
 
                wake_up(&c->copygc_running_wq);
        }
 
+       move_buckets_wait(&ctxt, &buckets, true);
+       rhashtable_destroy(&buckets.table);
+       bch2_moving_ctxt_exit(&ctxt);
+       bch2_move_stats_exit(&move_stats, c);
+
        return 0;
 }
 
@@ -267,7 +392,7 @@ int bch2_copygc_start(struct bch_fs *c)
        t = kthread_create(bch2_copygc_thread, c, "bch-copygc/%s", c->name);
        ret = PTR_ERR_OR_ZERO(t);
        if (ret) {
-               bch_err(c, "error creating copygc thread: %s", bch2_err_str(ret));
+               bch_err_msg(c, ret, "creating copygc thread");
                return ret;
        }