#include <linux/ceph/ceph_debug.h> #include <linux/sort.h>#include <linux/slab.h> #include "super.h"#include "mds_client.h" #include <linux/ceph/decode.h> /* * 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 for write. */void ceph_get_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm){ dout("get_realm %p %d -> %d\n", realm, atomic_read(&realm->nref), atomic_read(&realm->nref)+1); /* * since we _only_ increment realm refs or empty the empty * list with snap_rwsem held, adjusting the empty list here is * safe. we do need to protect against concurrent empty list * additions, however. */ if (atomic_read(&realm->nref) == 0) { spin_lock(&mdsc->snap_empty_lock); list_del_init(&realm->empty_item); spin_unlock(&mdsc->snap_empty_lock); } atomic_inc(&realm->nref);} 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; realm = kzalloc(sizeof(*realm), GFP_NOFS); if (!realm) return ERR_PTR(-ENOMEM); atomic_set(&realm->nref, 0); /* tree does not take a ref */ 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->inodes_with_caps); spin_lock_init(&realm->inodes_with_caps_lock); __insert_snap_realm(&mdsc->snap_realms, realm); dout("create_snap_realm %llx %p\n", realm->ino, realm); return realm;} /* * lookup the realm rooted at @ino. * * caller must hold snap_rwsem for write. */struct ceph_snap_realm *ceph_lookup_snap_realm(struct ceph_mds_client *mdsc, u64 ino){ struct rb_node *n = mdsc->snap_realms.rb_node; struct ceph_snap_realm *r; 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 { dout("lookup_snap_realm %llx %p\n", r->ino, r); return r; } } return NULL;} 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){ dout("__destroy_snap_realm %p %llx\n", realm, realm->ino); rb_erase(&realm->node, &mdsc->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){ dout("__put_snap_realm %llx %p %d -> %d\n", realm->ino, realm, atomic_read(&realm->nref), atomic_read(&realm->nref)-1); if (atomic_dec_and_test(&realm->nref)) __destroy_snap_realm(mdsc, realm);} /* * caller needn't hold any locks */void ceph_put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm){ dout("put_snap_realm %llx %p %d -> %d\n", realm->ino, realm, atomic_read(&realm->nref), atomic_read(&realm->nref)-1); if (!atomic_dec_and_test(&realm->nref)) return; if (down_write_trylock(&mdsc->snap_rwsem)) { __destroy_snap_realm(mdsc, realm); up_write(&mdsc->snap_rwsem); } else { spin_lock(&mdsc->snap_empty_lock); 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; 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_empty_realms(struct ceph_mds_client *mdsc){ down_write(&mdsc->snap_rwsem); __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_snap_realm *parent; 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); } dout("adjust_snap_realm_parent %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; ceph_get_snap_realm(mdsc, 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_snap_realm *realm){ 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) { err = build_snap_context(parent); if (err) goto fail; } 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)) { dout("build_snap_context %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 (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; dout("build_snap_context %llx %p: %p seq %lld (%u snaps)\n", realm->ino, realm, snapc, snapc->seq, (unsigned int) snapc->num_snaps); if (realm->cached_context) ceph_put_snap_context(realm->cached_context); realm->cached_context = snapc; 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("build_snap_context %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_snap_realm *realm){ struct ceph_snap_realm *child; dout("rebuild_snap_realms %llx %p\n", realm->ino, realm); build_snap_context(realm); list_for_each_entry(child, &realm->children, child_item) rebuild_snap_realms(child);} /* * 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); } else { *dst = NULL; } return 0;} /* * When a snapshot is applied, the size/mtime inode metadata is queued * in a ceph_cap_snap (one for each snapshot) until writeback * completes and the metadata can be flushed back to the MDS. * * However, if a (sync) write is currently in-progress when we apply * the snapshot, we have to wait until the write succeeds or fails * (and a final size/mtime is known). In this case the * cap_snap->writing = 1, and is said to be "pending." When the write * finishes, we __ceph_finish_cap_snap(). * * Caller must hold snap_rwsem for read (i.e., the realm topology won't * change). */void ceph_queue_cap_snap(struct ceph_inode_info *ci){ struct inode *inode = &ci->vfs_inode; struct ceph_cap_snap *capsnap; int used, dirty; capsnap = kzalloc(sizeof(*capsnap), GFP_NOFS); if (!capsnap) { pr_err("ENOMEM allocating ceph_cap_snap on %p\n", inode); return; } spin_lock(&ci->i_ceph_lock); used = __ceph_caps_used(ci); dirty = __ceph_caps_dirty(ci); /* * If there is a write in progress, treat that as a dirty Fw, * even though it hasn't completed yet; by the time we finish * up this capsnap it will be. */ if (used & CEPH_CAP_FILE_WR) dirty |= CEPH_CAP_FILE_WR; if (__ceph_have_pending_cap_snap(ci)) { /* there is no point in queuing multiple "pending" cap_snaps, as no new writes are allowed to start when pending, so any writes in progress now were started before the previous cap_snap. lucky us. */ dout("queue_cap_snap %p already pending\n", inode); kfree(capsnap); } else if (dirty & (CEPH_CAP_AUTH_EXCL|CEPH_CAP_XATTR_EXCL| CEPH_CAP_FILE_EXCL|CEPH_CAP_FILE_WR)) { struct ceph_snap_context *snapc = ci->i_head_snapc; /* * if we are a sync write, we may need to go to the snaprealm * to get the current snapc. */ if (!snapc) snapc = ci->i_snap_realm->cached_context; dout("queue_cap_snap %p cap_snap %p queuing under %p %s\n", inode, capsnap, snapc, ceph_cap_string(dirty)); ihold(inode); atomic_set(&capsnap->nref, 1); capsnap->ci = ci; INIT_LIST_HEAD(&capsnap->ci_item); INIT_LIST_HEAD(&capsnap->flushing_item); capsnap->follows = snapc->seq; capsnap->issued = __ceph_caps_issued(ci, NULL); capsnap->dirty = dirty; capsnap->mode = inode->i_mode; capsnap->uid = inode->i_uid; capsnap->gid = inode->i_gid; if (dirty & CEPH_CAP_XATTR_EXCL) { __ceph_build_xattrs_blob(ci); capsnap->xattr_blob = ceph_buffer_get(ci->i_xattrs.blob);Lines 1–500 of 932. Load more
#include <linux/ceph/ceph_debug.h> #include <linux/sort.h> #include <linux/slab.h> #include "super.h" #include "mds_client.h" #include <linux/ceph/decode.h> /* * 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 for write. */ void ceph_get_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm) { dout("get_realm %p %d -> %d\n", realm, atomic_read(&realm->nref), atomic_read(&realm->nref)+1); /* * since we _only_ increment realm refs or empty the empty * list with snap_rwsem held, adjusting the empty list here is * safe. we do need to protect against concurrent empty list * additions, however. */ if (atomic_read(&realm->nref) == 0) { spin_lock(&mdsc->snap_empty_lock); list_del_init(&realm->empty_item); spin_unlock(&mdsc->snap_empty_lock); } atomic_inc(&realm->nref); } 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; realm = kzalloc(sizeof(*realm), GFP_NOFS); if (!realm) return ERR_PTR(-ENOMEM); atomic_set(&realm->nref, 0); /* tree does not take a ref */ 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->inodes_with_caps); spin_lock_init(&realm->inodes_with_caps_lock); __insert_snap_realm(&mdsc->snap_realms, realm); dout("create_snap_realm %llx %p\n", realm->ino, realm); return realm; } /* * lookup the realm rooted at @ino. * * caller must hold snap_rwsem for write. */ struct ceph_snap_realm *ceph_lookup_snap_realm(struct ceph_mds_client *mdsc, u64 ino) { struct rb_node *n = mdsc->snap_realms.rb_node; struct ceph_snap_realm *r; 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 { dout("lookup_snap_realm %llx %p\n", r->ino, r); return r; } } return NULL; } 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) { dout("__destroy_snap_realm %p %llx\n", realm, realm->ino); rb_erase(&realm->node, &mdsc->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) { dout("__put_snap_realm %llx %p %d -> %d\n", realm->ino, realm, atomic_read(&realm->nref), atomic_read(&realm->nref)-1); if (atomic_dec_and_test(&realm->nref)) __destroy_snap_realm(mdsc, realm); } /* * caller needn't hold any locks */ void ceph_put_snap_realm(struct ceph_mds_client *mdsc, struct ceph_snap_realm *realm) { dout("put_snap_realm %llx %p %d -> %d\n", realm->ino, realm, atomic_read(&realm->nref), atomic_read(&realm->nref)-1); if (!atomic_dec_and_test(&realm->nref)) return; if (down_write_trylock(&mdsc->snap_rwsem)) { __destroy_snap_realm(mdsc, realm); up_write(&mdsc->snap_rwsem); } else { spin_lock(&mdsc->snap_empty_lock); 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; 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_empty_realms(struct ceph_mds_client *mdsc) { down_write(&mdsc->snap_rwsem); __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_snap_realm *parent; 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); } dout("adjust_snap_realm_parent %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; ceph_get_snap_realm(mdsc, 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_snap_realm *realm) { 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) { err = build_snap_context(parent); if (err) goto fail; } 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)) { dout("build_snap_context %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 (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; dout("build_snap_context %llx %p: %p seq %lld (%u snaps)\n", realm->ino, realm, snapc, snapc->seq, (unsigned int) snapc->num_snaps); if (realm->cached_context) ceph_put_snap_context(realm->cached_context); realm->cached_context = snapc; 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("build_snap_context %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_snap_realm *realm) { struct ceph_snap_realm *child; dout("rebuild_snap_realms %llx %p\n", realm->ino, realm); build_snap_context(realm); list_for_each_entry(child, &realm->children, child_item) rebuild_snap_realms(child); } /* * 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); } else { *dst = NULL; } return 0; } /* * When a snapshot is applied, the size/mtime inode metadata is queued * in a ceph_cap_snap (one for each snapshot) until writeback * completes and the metadata can be flushed back to the MDS. * * However, if a (sync) write is currently in-progress when we apply * the snapshot, we have to wait until the write succeeds or fails * (and a final size/mtime is known). In this case the * cap_snap->writing = 1, and is said to be "pending." When the write * finishes, we __ceph_finish_cap_snap(). * * Caller must hold snap_rwsem for read (i.e., the realm topology won't * change). */ void ceph_queue_cap_snap(struct ceph_inode_info *ci) { struct inode *inode = &ci->vfs_inode; struct ceph_cap_snap *capsnap; int used, dirty; capsnap = kzalloc(sizeof(*capsnap), GFP_NOFS); if (!capsnap) { pr_err("ENOMEM allocating ceph_cap_snap on %p\n", inode); return; } spin_lock(&ci->i_ceph_lock); used = __ceph_caps_used(ci); dirty = __ceph_caps_dirty(ci); /* * If there is a write in progress, treat that as a dirty Fw, * even though it hasn't completed yet; by the time we finish * up this capsnap it will be. */ if (used & CEPH_CAP_FILE_WR) dirty |= CEPH_CAP_FILE_WR; if (__ceph_have_pending_cap_snap(ci)) { /* there is no point in queuing multiple "pending" cap_snaps, as no new writes are allowed to start when pending, so any writes in progress now were started before the previous cap_snap. lucky us. */ dout("queue_cap_snap %p already pending\n", inode); kfree(capsnap); } else if (dirty & (CEPH_CAP_AUTH_EXCL|CEPH_CAP_XATTR_EXCL| CEPH_CAP_FILE_EXCL|CEPH_CAP_FILE_WR)) { struct ceph_snap_context *snapc = ci->i_head_snapc; /* * if we are a sync write, we may need to go to the snaprealm * to get the current snapc. */ if (!snapc) snapc = ci->i_snap_realm->cached_context; dout("queue_cap_snap %p cap_snap %p queuing under %p %s\n", inode, capsnap, snapc, ceph_cap_string(dirty)); ihold(inode); atomic_set(&capsnap->nref, 1); capsnap->ci = ci; INIT_LIST_HEAD(&capsnap->ci_item); INIT_LIST_HEAD(&capsnap->flushing_item); capsnap->follows = snapc->seq; capsnap->issued = __ceph_caps_issued(ci, NULL); capsnap->dirty = dirty; capsnap->mode = inode->i_mode; capsnap->uid = inode->i_uid; capsnap->gid = inode->i_gid; if (dirty & CEPH_CAP_XATTR_EXCL) { __ceph_build_xattrs_blob(ci); capsnap->xattr_blob = ceph_buffer_get(ci->i_xattrs.blob);