// SPDX-License-Identifier: GPL-2.0#include <linux/ceph/ceph_debug.h> #include <linux/fs.h>#include <linux/sort.h>#include <linux/slab.h>#include <linux/iversion.h>#include "super.h"#include "mds_client.h"#include <linux/ceph/decode.h> /* unused map expires after 5 minutes */#define CEPH_SNAPID_MAP_TIMEOUT (5 * 60 * HZ) /* * Snapshots in ceph are driven in large part by cooperation from the * client. In contrast to local file systems or file servers that * implement snapshots at a single point in the system, ceph's * distributed access to storage requires clients to help decide * whether a write logically occurs before or after a recently created * snapshot. * * This provides a perfect instantanous client-wide snapshot. Between * clients, however, snapshots may appear to be applied at slightly * different points in time, depending on delays in delivering the * snapshot notification. * * Snapshots are _not_ file system-wide. Instead, each snapshot * applies to the subdirectory nested beneath some directory. This * effectively divides the hierarchy into multiple "realms," where all * of the files contained by each realm share the same set of * snapshots. An individual realm's snap set contains snapshots * explicitly created on that realm, as well as any snaps in its * parent's snap set _after_ the point at which the parent became it's * parent (due to, say, a rename). Similarly, snaps from prior parents * during the time intervals during which they were the parent are included. * * The client is spared most of this detail, fortunately... it must only * maintains a hierarchy of realms reflecting the current parent/child * realm relationship, and for each realm has an explicit list of snaps * inherited from prior parents. * * A snap_realm struct is maintained for realms containing every inode * with an open cap in the system. (The needed snap realm information is * provided by the MDS whenever a cap is issued, i.e., on open.) A 'seq' * version number is used to ensure that as realm parameters change (new * snapshot, new parent, etc.) the client's realm hierarchy is updated. * * The realm hierarchy drives the generation of a 'snap context' for each * realm, which simply lists the resulting set of snaps for the realm. This * is attached to any writes sent to OSDs. *//* * Unfortunately error handling is a bit mixed here. If we get a snap * update, but don't have enough memory to update our realm hierarchy, * it's not clear what we can do about it (besides complaining to the * console). */ /* * increase ref count for the realm * * caller must hold snap_rwsem. */void ceph_get_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm){ lockdep_assert_held(&mdsc->snap_rwsem); /* * The 0->1 and 1->0 transitions must take the snap_empty_lock * atomically with the refcount change. Go ahead and bump the * nref here, unless it's 0, in which case we take the spinlock * and then do the increment and remove it from the list. */ if (atomic_inc_not_zero(&realm->nref)) return; spin_lock(&mdsc->snap_empty_lock); if (atomic_inc_return(&realm->nref) == 1) list_del_init(&realm->empty_item); spin_unlock(&mdsc->snap_empty_lock);} static void __insert_snap_realm(struct rb_root *root, struct ceph_snap_realm *new){ struct rb_node **p = &root->rb_node; struct rb_node *parent = NULL; struct ceph_snap_realm *r = NULL; while (*p) { parent = *p; r = rb_entry(parent, struct ceph_snap_realm, node); if (new->ino < r->ino) p = &(*p)->rb_left; else if (new->ino > r->ino) p = &(*p)->rb_right; else BUG(); } rb_link_node(&new->node, parent, p); rb_insert_color(&new->node, root);} /* * create and get the realm rooted at @ino and bump its ref count. * * caller must hold snap_rwsem for write. */static struct ceph_snap_realm *ceph_create_snap_realm( struct ceph_mds_client *mdsc, u64 ino){ struct ceph_snap_realm *realm; lockdep_assert_held_write(&mdsc->snap_rwsem); realm = kzalloc_obj(*realm, GFP_NOFS); if (!realm) return ERR_PTR(-ENOMEM); /* Do not release the global dummy snaprealm until unmouting */ if (ino == CEPH_INO_GLOBAL_SNAPREALM) atomic_set(&realm->nref, 2); else atomic_set(&realm->nref, 1); realm->ino = ino; INIT_LIST_HEAD(&realm->children); INIT_LIST_HEAD(&realm->child_item); INIT_LIST_HEAD(&realm->empty_item); INIT_LIST_HEAD(&realm->dirty_item); INIT_LIST_HEAD(&realm->rebuild_item); INIT_LIST_HEAD(&realm->inodes_with_caps); spin_lock_init(&realm->inodes_with_caps_lock); __insert_snap_realm(&mdsc->snap_realms, realm); mdsc->num_snap_realms++; doutc(mdsc->fsc->client, "%llx %p\n", realm->ino, realm); return realm;} /* * lookup the realm rooted at @ino. * * caller must hold snap_rwsem. */static struct ceph_snap_realm *__lookup_snap_realm(struct ceph_mds_client *mdsc, u64 ino){ struct ceph_client *cl = mdsc->fsc->client; struct rb_node *n = mdsc->snap_realms.rb_node; struct ceph_snap_realm *r; lockdep_assert_held(&mdsc->snap_rwsem); while (n) { r = rb_entry(n, struct ceph_snap_realm, node); if (ino < r->ino) n = n->rb_left; else if (ino > r->ino) n = n->rb_right; else { doutc(cl, "%llx %p\n", r->ino, r); return r; } } return NULL;} struct ceph_snap_realm *ceph_lookup_snap_realm(struct ceph_mds_client *mdsc, u64 ino){ struct ceph_snap_realm *r; r = __lookup_snap_realm(mdsc, ino); if (r) ceph_get_snap_realm(mdsc, r); return r;} static void __put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm); /* * called with snap_rwsem (write) */static void __destroy_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm){ struct ceph_client *cl = mdsc->fsc->client; lockdep_assert_held_write(&mdsc->snap_rwsem); doutc(cl, "%p %llx\n", realm, realm->ino); rb_erase(&realm->node, &mdsc->snap_realms); mdsc->num_snap_realms--; if (realm->parent) { list_del_init(&realm->child_item); __put_snap_realm(mdsc, realm->parent); } kfree(realm->prior_parent_snaps); kfree(realm->snaps); ceph_put_snap_context(realm->cached_context); kfree(realm);} /* * caller holds snap_rwsem (write) */static void __put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm){ lockdep_assert_held_write(&mdsc->snap_rwsem); /* * We do not require the snap_empty_lock here, as any caller that * increments the value must hold the snap_rwsem. */ if (atomic_dec_and_test(&realm->nref)) __destroy_snap_realm(mdsc, realm);} /* * See comments in ceph_get_snap_realm. Caller needn't hold any locks. */void ceph_put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm){ if (!atomic_dec_and_lock(&realm->nref, &mdsc->snap_empty_lock)) return; if (down_write_trylock(&mdsc->snap_rwsem)) { spin_unlock(&mdsc->snap_empty_lock); __destroy_snap_realm(mdsc, realm); up_write(&mdsc->snap_rwsem); } else { list_add(&realm->empty_item, &mdsc->snap_empty); spin_unlock(&mdsc->snap_empty_lock); }} /* * Clean up any realms whose ref counts have dropped to zero. Note * that this does not include realms who were created but not yet * used. * * Called under snap_rwsem (write) */static void __cleanup_empty_realms(struct ceph_mds_client *mdsc){ struct ceph_snap_realm *realm; lockdep_assert_held_write(&mdsc->snap_rwsem); spin_lock(&mdsc->snap_empty_lock); while (!list_empty(&mdsc->snap_empty)) { realm = list_first_entry(&mdsc->snap_empty, struct ceph_snap_realm, empty_item); list_del(&realm->empty_item); spin_unlock(&mdsc->snap_empty_lock); __destroy_snap_realm(mdsc, realm); spin_lock(&mdsc->snap_empty_lock); } spin_unlock(&mdsc->snap_empty_lock);} void ceph_cleanup_global_and_empty_realms(struct ceph_mds_client *mdsc){ struct ceph_snap_realm *global_realm; down_write(&mdsc->snap_rwsem); global_realm = __lookup_snap_realm(mdsc, CEPH_INO_GLOBAL_SNAPREALM); if (global_realm) ceph_put_snap_realm(mdsc, global_realm); __cleanup_empty_realms(mdsc); up_write(&mdsc->snap_rwsem);} /* * adjust the parent realm of a given @realm. adjust child list, and parent * pointers, and ref counts appropriately. * * return true if parent was changed, 0 if unchanged, <0 on error. * * caller must hold snap_rwsem for write. */static int adjust_snap_realm_parent(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm, u64 parentino){ struct ceph_client *cl = mdsc->fsc->client; struct ceph_snap_realm *parent; lockdep_assert_held_write(&mdsc->snap_rwsem); if (realm->parent_ino == parentino) return 0; parent = ceph_lookup_snap_realm(mdsc, parentino); if (!parent) { parent = ceph_create_snap_realm(mdsc, parentino); if (IS_ERR(parent)) return PTR_ERR(parent); } doutc(cl, "%llx %p: %llx %p -> %llx %p\n", realm->ino, realm, realm->parent_ino, realm->parent, parentino, parent); if (realm->parent) { list_del_init(&realm->child_item); ceph_put_snap_realm(mdsc, realm->parent); } realm->parent_ino = parentino; realm->parent = parent; list_add(&realm->child_item, &parent->children); return 1;} static int cmpu64_rev(const void *a, const void *b){ if (*(u64 *)a < *(u64 *)b) return 1; if (*(u64 *)a > *(u64 *)b) return -1; return 0;} /* * build the snap context for a given realm. */static int build_snap_context(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm, struct list_head *realm_queue, struct list_head *dirty_realms){ struct ceph_client *cl = mdsc->fsc->client; struct ceph_snap_realm *parent = realm->parent; struct ceph_snap_context *snapc; int err = 0; u32 num = realm->num_prior_parent_snaps + realm->num_snaps; /* * build parent context, if it hasn't been built. * conservatively estimate that all parent snaps might be * included by us. */ if (parent) { if (!parent->cached_context) { /* add to the queue head */ list_add(&parent->rebuild_item, realm_queue); return 1; } num += parent->cached_context->num_snaps; } /* do i actually need to update? not if my context seq matches realm seq, and my parents' does to. (this works because we rebuild_snap_realms() works _downward_ in hierarchy after each update.) */ if (realm->cached_context && realm->cached_context->seq == realm->seq && (!parent || realm->cached_context->seq >= parent->cached_context->seq)) { doutc(cl, "%llx %p: %p seq %lld (%u snaps) (unchanged)\n", realm->ino, realm, realm->cached_context, realm->cached_context->seq, (unsigned int)realm->cached_context->num_snaps); return 0; } /* alloc new snap context */ err = -ENOMEM; if ((size_t)num > (SIZE_MAX - sizeof(*snapc)) / sizeof(u64)) goto fail; snapc = ceph_create_snap_context(num, GFP_NOFS); if (!snapc) goto fail; /* build (reverse sorted) snap vector */ num = 0; snapc->seq = realm->seq; if (parent) { u32 i; /* include any of parent's snaps occurring _after_ my parent became my parent */ for (i = 0; i < parent->cached_context->num_snaps; i++) if (parent->cached_context->snaps[i] >= realm->parent_since) snapc->snaps[num++] = parent->cached_context->snaps[i]; if (parent->cached_context->seq > snapc->seq) snapc->seq = parent->cached_context->seq; } memcpy(snapc->snaps + num, realm->snaps, sizeof(u64)*realm->num_snaps); num += realm->num_snaps; memcpy(snapc->snaps + num, realm->prior_parent_snaps, sizeof(u64)*realm->num_prior_parent_snaps); num += realm->num_prior_parent_snaps; sort(snapc->snaps, num, sizeof(u64), cmpu64_rev, NULL); snapc->num_snaps = num; doutc(cl, "%llx %p: %p seq %lld (%u snaps)\n", realm->ino, realm, snapc, snapc->seq, (unsigned int) snapc->num_snaps); ceph_put_snap_context(realm->cached_context); realm->cached_context = snapc; /* queue realm for cap_snap creation */ list_add_tail(&realm->dirty_item, dirty_realms); return 0; fail: /* * if we fail, clear old (incorrect) cached_context... hopefully * we'll have better luck building it later */ if (realm->cached_context) { ceph_put_snap_context(realm->cached_context); realm->cached_context = NULL; } pr_err_client(cl, "%llx %p fail %d\n", realm->ino, realm, err); return err;} /* * rebuild snap context for the given realm and all of its children. */static void rebuild_snap_realms(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm, struct list_head *dirty_realms){ struct ceph_client *cl = mdsc->fsc->client; LIST_HEAD(realm_queue); int last = 0; bool skip = false; list_add_tail(&realm->rebuild_item, &realm_queue); while (!list_empty(&realm_queue)) { struct ceph_snap_realm *_realm, *child; _realm = list_first_entry(&realm_queue, struct ceph_snap_realm, rebuild_item); /* * If the last building failed dues to memory * issue, just empty the realm_queue and return * to avoid infinite loop. */ if (last < 0) { list_del_init(&_realm->rebuild_item); continue; } last = build_snap_context(mdsc, _realm, &realm_queue, dirty_realms); doutc(cl, "%llx %p, %s\n", realm->ino, realm, last > 0 ? "is deferred" : !last ? "succeeded" : "failed"); /* is any child in the list ? */ list_for_each_entry(child, &_realm->children, child_item) { if (!list_empty(&child->rebuild_item)) { skip = true; break; } } if (!skip) { list_for_each_entry(child, &_realm->children, child_item) list_add_tail(&child->rebuild_item, &realm_queue); } /* last == 1 means need to build parent first */ if (last <= 0) list_del_init(&_realm->rebuild_item); }} /* * helper to allocate and decode an array of snapids. free prior * instance, if any. */static int dup_array(u64 **dst, __le64 *src, u32 num){ u32 i; kfree(*dst); if (num) { *dst = kcalloc(num, sizeof(u64), GFP_NOFS); if (!*dst) return -ENOMEM; for (i = 0; i < num; i++) (*dst)[i] = get_unaligned_le64(src + i);Lines 1–500 of 1343. Load more
// SPDX-License-Identifier: GPL-2.0 #include <linux/ceph/ceph_debug.h> #include <linux/fs.h> #include <linux/sort.h> #include <linux/slab.h> #include <linux/iversion.h> #include "super.h" #include "mds_client.h" #include <linux/ceph/decode.h> /* unused map expires after 5 minutes */ #define CEPH_SNAPID_MAP_TIMEOUT (5 * 60 * HZ) /* * Snapshots in ceph are driven in large part by cooperation from the * client. In contrast to local file systems or file servers that * implement snapshots at a single point in the system, ceph's * distributed access to storage requires clients to help decide * whether a write logically occurs before or after a recently created * snapshot. * * This provides a perfect instantanous client-wide snapshot. Between * clients, however, snapshots may appear to be applied at slightly * different points in time, depending on delays in delivering the * snapshot notification. * * Snapshots are _not_ file system-wide. Instead, each snapshot * applies to the subdirectory nested beneath some directory. This * effectively divides the hierarchy into multiple "realms," where all * of the files contained by each realm share the same set of * snapshots. An individual realm's snap set contains snapshots * explicitly created on that realm, as well as any snaps in its * parent's snap set _after_ the point at which the parent became it's * parent (due to, say, a rename). Similarly, snaps from prior parents * during the time intervals during which they were the parent are included. * * The client is spared most of this detail, fortunately... it must only * maintains a hierarchy of realms reflecting the current parent/child * realm relationship, and for each realm has an explicit list of snaps * inherited from prior parents. * * A snap_realm struct is maintained for realms containing every inode * with an open cap in the system. (The needed snap realm information is * provided by the MDS whenever a cap is issued, i.e., on open.) A 'seq' * version number is used to ensure that as realm parameters change (new * snapshot, new parent, etc.) the client's realm hierarchy is updated. * * The realm hierarchy drives the generation of a 'snap context' for each * realm, which simply lists the resulting set of snaps for the realm. This * is attached to any writes sent to OSDs. */ /* * Unfortunately error handling is a bit mixed here. If we get a snap * update, but don't have enough memory to update our realm hierarchy, * it's not clear what we can do about it (besides complaining to the * console). */ /* * increase ref count for the realm * * caller must hold snap_rwsem. */ void ceph_get_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm) { lockdep_assert_held(&mdsc->snap_rwsem); /* * The 0->1 and 1->0 transitions must take the snap_empty_lock * atomically with the refcount change. Go ahead and bump the * nref here, unless it's 0, in which case we take the spinlock * and then do the increment and remove it from the list. */ if (atomic_inc_not_zero(&realm->nref)) return; spin_lock(&mdsc->snap_empty_lock); if (atomic_inc_return(&realm->nref) == 1) list_del_init(&realm->empty_item); spin_unlock(&mdsc->snap_empty_lock); } static void __insert_snap_realm(struct rb_root *root, struct ceph_snap_realm *new) { struct rb_node **p = &root->rb_node; struct rb_node *parent = NULL; struct ceph_snap_realm *r = NULL; while (*p) { parent = *p; r = rb_entry(parent, struct ceph_snap_realm, node); if (new->ino < r->ino) p = &(*p)->rb_left; else if (new->ino > r->ino) p = &(*p)->rb_right; else BUG(); } rb_link_node(&new->node, parent, p); rb_insert_color(&new->node, root); } /* * create and get the realm rooted at @ino and bump its ref count. * * caller must hold snap_rwsem for write. */ static struct ceph_snap_realm *ceph_create_snap_realm( struct ceph_mds_client *mdsc, u64 ino) { struct ceph_snap_realm *realm; lockdep_assert_held_write(&mdsc->snap_rwsem); realm = kzalloc_obj(*realm, GFP_NOFS); if (!realm) return ERR_PTR(-ENOMEM); /* Do not release the global dummy snaprealm until unmouting */ if (ino == CEPH_INO_GLOBAL_SNAPREALM) atomic_set(&realm->nref, 2); else atomic_set(&realm->nref, 1); realm->ino = ino; INIT_LIST_HEAD(&realm->children); INIT_LIST_HEAD(&realm->child_item); INIT_LIST_HEAD(&realm->empty_item); INIT_LIST_HEAD(&realm->dirty_item); INIT_LIST_HEAD(&realm->rebuild_item); INIT_LIST_HEAD(&realm->inodes_with_caps); spin_lock_init(&realm->inodes_with_caps_lock); __insert_snap_realm(&mdsc->snap_realms, realm); mdsc->num_snap_realms++; doutc(mdsc->fsc->client, "%llx %p\n", realm->ino, realm); return realm; } /* * lookup the realm rooted at @ino. * * caller must hold snap_rwsem. */ static struct ceph_snap_realm *__lookup_snap_realm(struct ceph_mds_client *mdsc, u64 ino) { struct ceph_client *cl = mdsc->fsc->client; struct rb_node *n = mdsc->snap_realms.rb_node; struct ceph_snap_realm *r; lockdep_assert_held(&mdsc->snap_rwsem); while (n) { r = rb_entry(n, struct ceph_snap_realm, node); if (ino < r->ino) n = n->rb_left; else if (ino > r->ino) n = n->rb_right; else { doutc(cl, "%llx %p\n", r->ino, r); return r; } } return NULL; } struct ceph_snap_realm *ceph_lookup_snap_realm(struct ceph_mds_client *mdsc, u64 ino) { struct ceph_snap_realm *r; r = __lookup_snap_realm(mdsc, ino); if (r) ceph_get_snap_realm(mdsc, r); return r; } static void __put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm); /* * called with snap_rwsem (write) */ static void __destroy_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm) { struct ceph_client *cl = mdsc->fsc->client; lockdep_assert_held_write(&mdsc->snap_rwsem); doutc(cl, "%p %llx\n", realm, realm->ino); rb_erase(&realm->node, &mdsc->snap_realms); mdsc->num_snap_realms--; if (realm->parent) { list_del_init(&realm->child_item); __put_snap_realm(mdsc, realm->parent); } kfree(realm->prior_parent_snaps); kfree(realm->snaps); ceph_put_snap_context(realm->cached_context); kfree(realm); } /* * caller holds snap_rwsem (write) */ static void __put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm) { lockdep_assert_held_write(&mdsc->snap_rwsem); /* * We do not require the snap_empty_lock here, as any caller that * increments the value must hold the snap_rwsem. */ if (atomic_dec_and_test(&realm->nref)) __destroy_snap_realm(mdsc, realm); } /* * See comments in ceph_get_snap_realm. Caller needn't hold any locks. */ void ceph_put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm) { if (!atomic_dec_and_lock(&realm->nref, &mdsc->snap_empty_lock)) return; if (down_write_trylock(&mdsc->snap_rwsem)) { spin_unlock(&mdsc->snap_empty_lock); __destroy_snap_realm(mdsc, realm); up_write(&mdsc->snap_rwsem); } else { list_add(&realm->empty_item, &mdsc->snap_empty); spin_unlock(&mdsc->snap_empty_lock); } } /* * Clean up any realms whose ref counts have dropped to zero. Note * that this does not include realms who were created but not yet * used. * * Called under snap_rwsem (write) */ static void __cleanup_empty_realms(struct ceph_mds_client *mdsc) { struct ceph_snap_realm *realm; lockdep_assert_held_write(&mdsc->snap_rwsem); spin_lock(&mdsc->snap_empty_lock); while (!list_empty(&mdsc->snap_empty)) { realm = list_first_entry(&mdsc->snap_empty, struct ceph_snap_realm, empty_item); list_del(&realm->empty_item); spin_unlock(&mdsc->snap_empty_lock); __destroy_snap_realm(mdsc, realm); spin_lock(&mdsc->snap_empty_lock); } spin_unlock(&mdsc->snap_empty_lock); } void ceph_cleanup_global_and_empty_realms(struct ceph_mds_client *mdsc) { struct ceph_snap_realm *global_realm; down_write(&mdsc->snap_rwsem); global_realm = __lookup_snap_realm(mdsc, CEPH_INO_GLOBAL_SNAPREALM); if (global_realm) ceph_put_snap_realm(mdsc, global_realm); __cleanup_empty_realms(mdsc); up_write(&mdsc->snap_rwsem); } /* * adjust the parent realm of a given @realm. adjust child list, and parent * pointers, and ref counts appropriately. * * return true if parent was changed, 0 if unchanged, <0 on error. * * caller must hold snap_rwsem for write. */ static int adjust_snap_realm_parent(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm, u64 parentino) { struct ceph_client *cl = mdsc->fsc->client; struct ceph_snap_realm *parent; lockdep_assert_held_write(&mdsc->snap_rwsem); if (realm->parent_ino == parentino) return 0; parent = ceph_lookup_snap_realm(mdsc, parentino); if (!parent) { parent = ceph_create_snap_realm(mdsc, parentino); if (IS_ERR(parent)) return PTR_ERR(parent); } doutc(cl, "%llx %p: %llx %p -> %llx %p\n", realm->ino, realm, realm->parent_ino, realm->parent, parentino, parent); if (realm->parent) { list_del_init(&realm->child_item); ceph_put_snap_realm(mdsc, realm->parent); } realm->parent_ino = parentino; realm->parent = parent; list_add(&realm->child_item, &parent->children); return 1; } static int cmpu64_rev(const void *a, const void *b) { if (*(u64 *)a < *(u64 *)b) return 1; if (*(u64 *)a > *(u64 *)b) return -1; return 0; } /* * build the snap context for a given realm. */ static int build_snap_context(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm, struct list_head *realm_queue, struct list_head *dirty_realms) { struct ceph_client *cl = mdsc->fsc->client; struct ceph_snap_realm *parent = realm->parent; struct ceph_snap_context *snapc; int err = 0; u32 num = realm->num_prior_parent_snaps + realm->num_snaps; /* * build parent context, if it hasn't been built. * conservatively estimate that all parent snaps might be * included by us. */ if (parent) { if (!parent->cached_context) { /* add to the queue head */ list_add(&parent->rebuild_item, realm_queue); return 1; } num += parent->cached_context->num_snaps; } /* do i actually need to update? not if my context seq matches realm seq, and my parents' does to. (this works because we rebuild_snap_realms() works _downward_ in hierarchy after each update.) */ if (realm->cached_context && realm->cached_context->seq == realm->seq && (!parent || realm->cached_context->seq >= parent->cached_context->seq)) { doutc(cl, "%llx %p: %p seq %lld (%u snaps) (unchanged)\n", realm->ino, realm, realm->cached_context, realm->cached_context->seq, (unsigned int)realm->cached_context->num_snaps); return 0; } /* alloc new snap context */ err = -ENOMEM; if ((size_t)num > (SIZE_MAX - sizeof(*snapc)) / sizeof(u64)) goto fail; snapc = ceph_create_snap_context(num, GFP_NOFS); if (!snapc) goto fail; /* build (reverse sorted) snap vector */ num = 0; snapc->seq = realm->seq; if (parent) { u32 i; /* include any of parent's snaps occurring _after_ my parent became my parent */ for (i = 0; i < parent->cached_context->num_snaps; i++) if (parent->cached_context->snaps[i] >= realm->parent_since) snapc->snaps[num++] = parent->cached_context->snaps[i]; if (parent->cached_context->seq > snapc->seq) snapc->seq = parent->cached_context->seq; } memcpy(snapc->snaps + num, realm->snaps, sizeof(u64)*realm->num_snaps); num += realm->num_snaps; memcpy(snapc->snaps + num, realm->prior_parent_snaps, sizeof(u64)*realm->num_prior_parent_snaps); num += realm->num_prior_parent_snaps; sort(snapc->snaps, num, sizeof(u64), cmpu64_rev, NULL); snapc->num_snaps = num; doutc(cl, "%llx %p: %p seq %lld (%u snaps)\n", realm->ino, realm, snapc, snapc->seq, (unsigned int) snapc->num_snaps); ceph_put_snap_context(realm->cached_context); realm->cached_context = snapc; /* queue realm for cap_snap creation */ list_add_tail(&realm->dirty_item, dirty_realms); return 0; fail: /* * if we fail, clear old (incorrect) cached_context... hopefully * we'll have better luck building it later */ if (realm->cached_context) { ceph_put_snap_context(realm->cached_context); realm->cached_context = NULL; } pr_err_client(cl, "%llx %p fail %d\n", realm->ino, realm, err); return err; } /* * rebuild snap context for the given realm and all of its children. */ static void rebuild_snap_realms(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm, struct list_head *dirty_realms) { struct ceph_client *cl = mdsc->fsc->client; LIST_HEAD(realm_queue); int last = 0; bool skip = false; list_add_tail(&realm->rebuild_item, &realm_queue); while (!list_empty(&realm_queue)) { struct ceph_snap_realm *_realm, *child; _realm = list_first_entry(&realm_queue, struct ceph_snap_realm, rebuild_item); /* * If the last building failed dues to memory * issue, just empty the realm_queue and return * to avoid infinite loop. */ if (last < 0) { list_del_init(&_realm->rebuild_item); continue; } last = build_snap_context(mdsc, _realm, &realm_queue, dirty_realms); doutc(cl, "%llx %p, %s\n", realm->ino, realm, last > 0 ? "is deferred" : !last ? "succeeded" : "failed"); /* is any child in the list ? */ list_for_each_entry(child, &_realm->children, child_item) { if (!list_empty(&child->rebuild_item)) { skip = true; break; } } if (!skip) { list_for_each_entry(child, &_realm->children, child_item) list_add_tail(&child->rebuild_item, &realm_queue); } /* last == 1 means need to build parent first */ if (last <= 0) list_del_init(&_realm->rebuild_item); } } /* * helper to allocate and decode an array of snapids. free prior * instance, if any. */ static int dup_array(u64 **dst, __le64 *src, u32 num) { u32 i; kfree(*dst); if (num) { *dst = kcalloc(num, sizeof(u64), GFP_NOFS); if (!*dst) return -ENOMEM; for (i = 0; i < num; i++) (*dst)[i] = get_unaligned_le64(src + i);