]>
git.proxmox.com Git - mirror_ubuntu-jammy-kernel.git/blob - fs/ceph/snap.c
1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
4 #include <linux/sort.h>
5 #include <linux/slab.h>
6 #include <linux/iversion.h>
8 #include "mds_client.h"
9 #include <linux/ceph/decode.h>
11 /* unused map expires after 5 minutes */
12 #define CEPH_SNAPID_MAP_TIMEOUT (5 * 60 * HZ)
15 * Snapshots in ceph are driven in large part by cooperation from the
16 * client. In contrast to local file systems or file servers that
17 * implement snapshots at a single point in the system, ceph's
18 * distributed access to storage requires clients to help decide
19 * whether a write logically occurs before or after a recently created
22 * This provides a perfect instantanous client-wide snapshot. Between
23 * clients, however, snapshots may appear to be applied at slightly
24 * different points in time, depending on delays in delivering the
25 * snapshot notification.
27 * Snapshots are _not_ file system-wide. Instead, each snapshot
28 * applies to the subdirectory nested beneath some directory. This
29 * effectively divides the hierarchy into multiple "realms," where all
30 * of the files contained by each realm share the same set of
31 * snapshots. An individual realm's snap set contains snapshots
32 * explicitly created on that realm, as well as any snaps in its
33 * parent's snap set _after_ the point at which the parent became it's
34 * parent (due to, say, a rename). Similarly, snaps from prior parents
35 * during the time intervals during which they were the parent are included.
37 * The client is spared most of this detail, fortunately... it must only
38 * maintains a hierarchy of realms reflecting the current parent/child
39 * realm relationship, and for each realm has an explicit list of snaps
40 * inherited from prior parents.
42 * A snap_realm struct is maintained for realms containing every inode
43 * with an open cap in the system. (The needed snap realm information is
44 * provided by the MDS whenever a cap is issued, i.e., on open.) A 'seq'
45 * version number is used to ensure that as realm parameters change (new
46 * snapshot, new parent, etc.) the client's realm hierarchy is updated.
48 * The realm hierarchy drives the generation of a 'snap context' for each
49 * realm, which simply lists the resulting set of snaps for the realm. This
50 * is attached to any writes sent to OSDs.
53 * Unfortunately error handling is a bit mixed here. If we get a snap
54 * update, but don't have enough memory to update our realm hierarchy,
55 * it's not clear what we can do about it (besides complaining to the
61 * increase ref count for the realm
63 * caller must hold snap_rwsem.
65 void ceph_get_snap_realm(struct ceph_mds_client
*mdsc
,
66 struct ceph_snap_realm
*realm
)
68 lockdep_assert_held(&mdsc
->snap_rwsem
);
71 * The 0->1 and 1->0 transitions must take the snap_empty_lock
72 * atomically with the refcount change. Go ahead and bump the
73 * nref here, unless it's 0, in which case we take the spinlock
74 * and then do the increment and remove it from the list.
76 if (atomic_inc_not_zero(&realm
->nref
))
79 spin_lock(&mdsc
->snap_empty_lock
);
80 if (atomic_inc_return(&realm
->nref
) == 1)
81 list_del_init(&realm
->empty_item
);
82 spin_unlock(&mdsc
->snap_empty_lock
);
85 static void __insert_snap_realm(struct rb_root
*root
,
86 struct ceph_snap_realm
*new)
88 struct rb_node
**p
= &root
->rb_node
;
89 struct rb_node
*parent
= NULL
;
90 struct ceph_snap_realm
*r
= NULL
;
94 r
= rb_entry(parent
, struct ceph_snap_realm
, node
);
95 if (new->ino
< r
->ino
)
97 else if (new->ino
> r
->ino
)
103 rb_link_node(&new->node
, parent
, p
);
104 rb_insert_color(&new->node
, root
);
108 * create and get the realm rooted at @ino and bump its ref count.
110 * caller must hold snap_rwsem for write.
112 static struct ceph_snap_realm
*ceph_create_snap_realm(
113 struct ceph_mds_client
*mdsc
,
116 struct ceph_snap_realm
*realm
;
118 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
120 realm
= kzalloc(sizeof(*realm
), GFP_NOFS
);
122 return ERR_PTR(-ENOMEM
);
124 atomic_set(&realm
->nref
, 1); /* for caller */
126 INIT_LIST_HEAD(&realm
->children
);
127 INIT_LIST_HEAD(&realm
->child_item
);
128 INIT_LIST_HEAD(&realm
->empty_item
);
129 INIT_LIST_HEAD(&realm
->dirty_item
);
130 INIT_LIST_HEAD(&realm
->inodes_with_caps
);
131 spin_lock_init(&realm
->inodes_with_caps_lock
);
132 __insert_snap_realm(&mdsc
->snap_realms
, realm
);
133 mdsc
->num_snap_realms
++;
135 dout("create_snap_realm %llx %p\n", realm
->ino
, realm
);
140 * lookup the realm rooted at @ino.
142 * caller must hold snap_rwsem.
144 static struct ceph_snap_realm
*__lookup_snap_realm(struct ceph_mds_client
*mdsc
,
147 struct rb_node
*n
= mdsc
->snap_realms
.rb_node
;
148 struct ceph_snap_realm
*r
;
150 lockdep_assert_held(&mdsc
->snap_rwsem
);
153 r
= rb_entry(n
, struct ceph_snap_realm
, node
);
156 else if (ino
> r
->ino
)
159 dout("lookup_snap_realm %llx %p\n", r
->ino
, r
);
166 struct ceph_snap_realm
*ceph_lookup_snap_realm(struct ceph_mds_client
*mdsc
,
169 struct ceph_snap_realm
*r
;
170 r
= __lookup_snap_realm(mdsc
, ino
);
172 ceph_get_snap_realm(mdsc
, r
);
176 static void __put_snap_realm(struct ceph_mds_client
*mdsc
,
177 struct ceph_snap_realm
*realm
);
180 * called with snap_rwsem (write)
182 static void __destroy_snap_realm(struct ceph_mds_client
*mdsc
,
183 struct ceph_snap_realm
*realm
)
185 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
187 dout("__destroy_snap_realm %p %llx\n", realm
, realm
->ino
);
189 rb_erase(&realm
->node
, &mdsc
->snap_realms
);
190 mdsc
->num_snap_realms
--;
193 list_del_init(&realm
->child_item
);
194 __put_snap_realm(mdsc
, realm
->parent
);
197 kfree(realm
->prior_parent_snaps
);
199 ceph_put_snap_context(realm
->cached_context
);
204 * caller holds snap_rwsem (write)
206 static void __put_snap_realm(struct ceph_mds_client
*mdsc
,
207 struct ceph_snap_realm
*realm
)
209 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
212 * We do not require the snap_empty_lock here, as any caller that
213 * increments the value must hold the snap_rwsem.
215 if (atomic_dec_and_test(&realm
->nref
))
216 __destroy_snap_realm(mdsc
, realm
);
220 * See comments in ceph_get_snap_realm. Caller needn't hold any locks.
222 void ceph_put_snap_realm(struct ceph_mds_client
*mdsc
,
223 struct ceph_snap_realm
*realm
)
225 if (!atomic_dec_and_lock(&realm
->nref
, &mdsc
->snap_empty_lock
))
228 if (down_write_trylock(&mdsc
->snap_rwsem
)) {
229 spin_unlock(&mdsc
->snap_empty_lock
);
230 __destroy_snap_realm(mdsc
, realm
);
231 up_write(&mdsc
->snap_rwsem
);
233 list_add(&realm
->empty_item
, &mdsc
->snap_empty
);
234 spin_unlock(&mdsc
->snap_empty_lock
);
239 * Clean up any realms whose ref counts have dropped to zero. Note
240 * that this does not include realms who were created but not yet
243 * Called under snap_rwsem (write)
245 static void __cleanup_empty_realms(struct ceph_mds_client
*mdsc
)
247 struct ceph_snap_realm
*realm
;
249 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
251 spin_lock(&mdsc
->snap_empty_lock
);
252 while (!list_empty(&mdsc
->snap_empty
)) {
253 realm
= list_first_entry(&mdsc
->snap_empty
,
254 struct ceph_snap_realm
, empty_item
);
255 list_del(&realm
->empty_item
);
256 spin_unlock(&mdsc
->snap_empty_lock
);
257 __destroy_snap_realm(mdsc
, realm
);
258 spin_lock(&mdsc
->snap_empty_lock
);
260 spin_unlock(&mdsc
->snap_empty_lock
);
263 void ceph_cleanup_empty_realms(struct ceph_mds_client
*mdsc
)
265 down_write(&mdsc
->snap_rwsem
);
266 __cleanup_empty_realms(mdsc
);
267 up_write(&mdsc
->snap_rwsem
);
271 * adjust the parent realm of a given @realm. adjust child list, and parent
272 * pointers, and ref counts appropriately.
274 * return true if parent was changed, 0 if unchanged, <0 on error.
276 * caller must hold snap_rwsem for write.
278 static int adjust_snap_realm_parent(struct ceph_mds_client
*mdsc
,
279 struct ceph_snap_realm
*realm
,
282 struct ceph_snap_realm
*parent
;
284 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
286 if (realm
->parent_ino
== parentino
)
289 parent
= ceph_lookup_snap_realm(mdsc
, parentino
);
291 parent
= ceph_create_snap_realm(mdsc
, parentino
);
293 return PTR_ERR(parent
);
295 dout("adjust_snap_realm_parent %llx %p: %llx %p -> %llx %p\n",
296 realm
->ino
, realm
, realm
->parent_ino
, realm
->parent
,
299 list_del_init(&realm
->child_item
);
300 ceph_put_snap_realm(mdsc
, realm
->parent
);
302 realm
->parent_ino
= parentino
;
303 realm
->parent
= parent
;
304 list_add(&realm
->child_item
, &parent
->children
);
309 static int cmpu64_rev(const void *a
, const void *b
)
311 if (*(u64
*)a
< *(u64
*)b
)
313 if (*(u64
*)a
> *(u64
*)b
)
320 * build the snap context for a given realm.
322 static int build_snap_context(struct ceph_snap_realm
*realm
,
323 struct list_head
* dirty_realms
)
325 struct ceph_snap_realm
*parent
= realm
->parent
;
326 struct ceph_snap_context
*snapc
;
328 u32 num
= realm
->num_prior_parent_snaps
+ realm
->num_snaps
;
331 * build parent context, if it hasn't been built.
332 * conservatively estimate that all parent snaps might be
336 if (!parent
->cached_context
) {
337 err
= build_snap_context(parent
, dirty_realms
);
341 num
+= parent
->cached_context
->num_snaps
;
344 /* do i actually need to update? not if my context seq
345 matches realm seq, and my parents' does to. (this works
346 because we rebuild_snap_realms() works _downward_ in
347 hierarchy after each update.) */
348 if (realm
->cached_context
&&
349 realm
->cached_context
->seq
== realm
->seq
&&
351 realm
->cached_context
->seq
>= parent
->cached_context
->seq
)) {
352 dout("build_snap_context %llx %p: %p seq %lld (%u snaps)"
354 realm
->ino
, realm
, realm
->cached_context
,
355 realm
->cached_context
->seq
,
356 (unsigned int)realm
->cached_context
->num_snaps
);
360 /* alloc new snap context */
362 if (num
> (SIZE_MAX
- sizeof(*snapc
)) / sizeof(u64
))
364 snapc
= ceph_create_snap_context(num
, GFP_NOFS
);
368 /* build (reverse sorted) snap vector */
370 snapc
->seq
= realm
->seq
;
374 /* include any of parent's snaps occurring _after_ my
375 parent became my parent */
376 for (i
= 0; i
< parent
->cached_context
->num_snaps
; i
++)
377 if (parent
->cached_context
->snaps
[i
] >=
379 snapc
->snaps
[num
++] =
380 parent
->cached_context
->snaps
[i
];
381 if (parent
->cached_context
->seq
> snapc
->seq
)
382 snapc
->seq
= parent
->cached_context
->seq
;
384 memcpy(snapc
->snaps
+ num
, realm
->snaps
,
385 sizeof(u64
)*realm
->num_snaps
);
386 num
+= realm
->num_snaps
;
387 memcpy(snapc
->snaps
+ num
, realm
->prior_parent_snaps
,
388 sizeof(u64
)*realm
->num_prior_parent_snaps
);
389 num
+= realm
->num_prior_parent_snaps
;
391 sort(snapc
->snaps
, num
, sizeof(u64
), cmpu64_rev
, NULL
);
392 snapc
->num_snaps
= num
;
393 dout("build_snap_context %llx %p: %p seq %lld (%u snaps)\n",
394 realm
->ino
, realm
, snapc
, snapc
->seq
,
395 (unsigned int) snapc
->num_snaps
);
397 ceph_put_snap_context(realm
->cached_context
);
398 realm
->cached_context
= snapc
;
399 /* queue realm for cap_snap creation */
400 list_add_tail(&realm
->dirty_item
, dirty_realms
);
405 * if we fail, clear old (incorrect) cached_context... hopefully
406 * we'll have better luck building it later
408 if (realm
->cached_context
) {
409 ceph_put_snap_context(realm
->cached_context
);
410 realm
->cached_context
= NULL
;
412 pr_err("build_snap_context %llx %p fail %d\n", realm
->ino
,
418 * rebuild snap context for the given realm and all of its children.
420 static void rebuild_snap_realms(struct ceph_snap_realm
*realm
,
421 struct list_head
*dirty_realms
)
423 struct ceph_snap_realm
*child
;
425 dout("rebuild_snap_realms %llx %p\n", realm
->ino
, realm
);
426 build_snap_context(realm
, dirty_realms
);
428 list_for_each_entry(child
, &realm
->children
, child_item
)
429 rebuild_snap_realms(child
, dirty_realms
);
434 * helper to allocate and decode an array of snapids. free prior
437 static int dup_array(u64
**dst
, __le64
*src
, u32 num
)
443 *dst
= kcalloc(num
, sizeof(u64
), GFP_NOFS
);
446 for (i
= 0; i
< num
; i
++)
447 (*dst
)[i
] = get_unaligned_le64(src
+ i
);
454 static bool has_new_snaps(struct ceph_snap_context
*o
,
455 struct ceph_snap_context
*n
)
457 if (n
->num_snaps
== 0)
459 /* snaps are in descending order */
460 return n
->snaps
[0] > o
->seq
;
464 * When a snapshot is applied, the size/mtime inode metadata is queued
465 * in a ceph_cap_snap (one for each snapshot) until writeback
466 * completes and the metadata can be flushed back to the MDS.
468 * However, if a (sync) write is currently in-progress when we apply
469 * the snapshot, we have to wait until the write succeeds or fails
470 * (and a final size/mtime is known). In this case the
471 * cap_snap->writing = 1, and is said to be "pending." When the write
472 * finishes, we __ceph_finish_cap_snap().
474 * Caller must hold snap_rwsem for read (i.e., the realm topology won't
477 static void ceph_queue_cap_snap(struct ceph_inode_info
*ci
)
479 struct inode
*inode
= &ci
->vfs_inode
;
480 struct ceph_cap_snap
*capsnap
;
481 struct ceph_snap_context
*old_snapc
, *new_snapc
;
482 struct ceph_buffer
*old_blob
= NULL
;
485 capsnap
= kzalloc(sizeof(*capsnap
), GFP_NOFS
);
487 pr_err("ENOMEM allocating ceph_cap_snap on %p\n", inode
);
490 capsnap
->cap_flush
.is_capsnap
= true;
491 INIT_LIST_HEAD(&capsnap
->cap_flush
.i_list
);
492 INIT_LIST_HEAD(&capsnap
->cap_flush
.g_list
);
494 spin_lock(&ci
->i_ceph_lock
);
495 used
= __ceph_caps_used(ci
);
496 dirty
= __ceph_caps_dirty(ci
);
498 old_snapc
= ci
->i_head_snapc
;
499 new_snapc
= ci
->i_snap_realm
->cached_context
;
502 * If there is a write in progress, treat that as a dirty Fw,
503 * even though it hasn't completed yet; by the time we finish
504 * up this capsnap it will be.
506 if (used
& CEPH_CAP_FILE_WR
)
507 dirty
|= CEPH_CAP_FILE_WR
;
509 if (__ceph_have_pending_cap_snap(ci
)) {
510 /* there is no point in queuing multiple "pending" cap_snaps,
511 as no new writes are allowed to start when pending, so any
512 writes in progress now were started before the previous
513 cap_snap. lucky us. */
514 dout("queue_cap_snap %p already pending\n", inode
);
517 if (ci
->i_wrbuffer_ref_head
== 0 &&
518 !(dirty
& (CEPH_CAP_ANY_EXCL
|CEPH_CAP_FILE_WR
))) {
519 dout("queue_cap_snap %p nothing dirty|writing\n", inode
);
526 * There is no need to send FLUSHSNAP message to MDS if there is
527 * no new snapshot. But when there is dirty pages or on-going
528 * writes, we still need to create cap_snap. cap_snap is needed
529 * by the write path and page writeback path.
531 * also see ceph_try_drop_cap_snap()
533 if (has_new_snaps(old_snapc
, new_snapc
)) {
534 if (dirty
& (CEPH_CAP_ANY_EXCL
|CEPH_CAP_FILE_WR
))
535 capsnap
->need_flush
= true;
537 if (!(used
& CEPH_CAP_FILE_WR
) &&
538 ci
->i_wrbuffer_ref_head
== 0) {
539 dout("queue_cap_snap %p "
540 "no new_snap|dirty_page|writing\n", inode
);
545 dout("queue_cap_snap %p cap_snap %p queuing under %p %s %s\n",
546 inode
, capsnap
, old_snapc
, ceph_cap_string(dirty
),
547 capsnap
->need_flush
? "" : "no_flush");
550 refcount_set(&capsnap
->nref
, 1);
551 INIT_LIST_HEAD(&capsnap
->ci_item
);
553 capsnap
->follows
= old_snapc
->seq
;
554 capsnap
->issued
= __ceph_caps_issued(ci
, NULL
);
555 capsnap
->dirty
= dirty
;
557 capsnap
->mode
= inode
->i_mode
;
558 capsnap
->uid
= inode
->i_uid
;
559 capsnap
->gid
= inode
->i_gid
;
561 if (dirty
& CEPH_CAP_XATTR_EXCL
) {
562 old_blob
= __ceph_build_xattrs_blob(ci
);
563 capsnap
->xattr_blob
=
564 ceph_buffer_get(ci
->i_xattrs
.blob
);
565 capsnap
->xattr_version
= ci
->i_xattrs
.version
;
567 capsnap
->xattr_blob
= NULL
;
568 capsnap
->xattr_version
= 0;
571 capsnap
->inline_data
= ci
->i_inline_version
!= CEPH_INLINE_NONE
;
573 /* dirty page count moved from _head to this cap_snap;
574 all subsequent writes page dirties occur _after_ this
576 capsnap
->dirty_pages
= ci
->i_wrbuffer_ref_head
;
577 ci
->i_wrbuffer_ref_head
= 0;
578 capsnap
->context
= old_snapc
;
579 list_add_tail(&capsnap
->ci_item
, &ci
->i_cap_snaps
);
581 if (used
& CEPH_CAP_FILE_WR
) {
582 dout("queue_cap_snap %p cap_snap %p snapc %p"
583 " seq %llu used WR, now pending\n", inode
,
584 capsnap
, old_snapc
, old_snapc
->seq
);
585 capsnap
->writing
= 1;
587 /* note mtime, size NOW. */
588 __ceph_finish_cap_snap(ci
, capsnap
);
594 if (ci
->i_wrbuffer_ref_head
== 0 &&
596 ci
->i_dirty_caps
== 0 &&
597 ci
->i_flushing_caps
== 0) {
598 ci
->i_head_snapc
= NULL
;
600 ci
->i_head_snapc
= ceph_get_snap_context(new_snapc
);
601 dout(" new snapc is %p\n", new_snapc
);
603 spin_unlock(&ci
->i_ceph_lock
);
605 ceph_buffer_put(old_blob
);
607 ceph_put_snap_context(old_snapc
);
611 * Finalize the size, mtime for a cap_snap.. that is, settle on final values
612 * to be used for the snapshot, to be flushed back to the mds.
614 * If capsnap can now be flushed, add to snap_flush list, and return 1.
616 * Caller must hold i_ceph_lock.
618 int __ceph_finish_cap_snap(struct ceph_inode_info
*ci
,
619 struct ceph_cap_snap
*capsnap
)
621 struct inode
*inode
= &ci
->vfs_inode
;
622 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(inode
->i_sb
);
624 BUG_ON(capsnap
->writing
);
625 capsnap
->size
= i_size_read(inode
);
626 capsnap
->mtime
= inode
->i_mtime
;
627 capsnap
->atime
= inode
->i_atime
;
628 capsnap
->ctime
= inode
->i_ctime
;
629 capsnap
->btime
= ci
->i_btime
;
630 capsnap
->change_attr
= inode_peek_iversion_raw(inode
);
631 capsnap
->time_warp_seq
= ci
->i_time_warp_seq
;
632 capsnap
->truncate_size
= ci
->i_truncate_size
;
633 capsnap
->truncate_seq
= ci
->i_truncate_seq
;
634 if (capsnap
->dirty_pages
) {
635 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu "
636 "still has %d dirty pages\n", inode
, capsnap
,
637 capsnap
->context
, capsnap
->context
->seq
,
638 ceph_cap_string(capsnap
->dirty
), capsnap
->size
,
639 capsnap
->dirty_pages
);
643 /* Fb cap still in use, delay it */
645 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu "
646 "used WRBUFFER, delaying\n", inode
, capsnap
,
647 capsnap
->context
, capsnap
->context
->seq
,
648 ceph_cap_string(capsnap
->dirty
), capsnap
->size
);
649 capsnap
->writing
= 1;
653 ci
->i_ceph_flags
|= CEPH_I_FLUSH_SNAPS
;
654 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu\n",
655 inode
, capsnap
, capsnap
->context
,
656 capsnap
->context
->seq
, ceph_cap_string(capsnap
->dirty
),
659 spin_lock(&mdsc
->snap_flush_lock
);
660 if (list_empty(&ci
->i_snap_flush_item
))
661 list_add_tail(&ci
->i_snap_flush_item
, &mdsc
->snap_flush_list
);
662 spin_unlock(&mdsc
->snap_flush_lock
);
663 return 1; /* caller may want to ceph_flush_snaps */
667 * Queue cap_snaps for snap writeback for this realm and its children.
668 * Called under snap_rwsem, so realm topology won't change.
670 static void queue_realm_cap_snaps(struct ceph_snap_realm
*realm
)
672 struct ceph_inode_info
*ci
;
673 struct inode
*lastinode
= NULL
;
675 dout("queue_realm_cap_snaps %p %llx inodes\n", realm
, realm
->ino
);
677 spin_lock(&realm
->inodes_with_caps_lock
);
678 list_for_each_entry(ci
, &realm
->inodes_with_caps
, i_snap_realm_item
) {
679 struct inode
*inode
= igrab(&ci
->vfs_inode
);
682 spin_unlock(&realm
->inodes_with_caps_lock
);
685 ceph_queue_cap_snap(ci
);
686 spin_lock(&realm
->inodes_with_caps_lock
);
688 spin_unlock(&realm
->inodes_with_caps_lock
);
691 dout("queue_realm_cap_snaps %p %llx done\n", realm
, realm
->ino
);
695 * Parse and apply a snapblob "snap trace" from the MDS. This specifies
696 * the snap realm parameters from a given realm and all of its ancestors,
699 * Caller must hold snap_rwsem for write.
701 int ceph_update_snap_trace(struct ceph_mds_client
*mdsc
,
702 void *p
, void *e
, bool deletion
,
703 struct ceph_snap_realm
**realm_ret
)
705 struct ceph_mds_snap_realm
*ri
; /* encoded */
706 __le64
*snaps
; /* encoded */
707 __le64
*prior_parent_snaps
; /* encoded */
708 struct ceph_snap_realm
*realm
= NULL
;
709 struct ceph_snap_realm
*first_realm
= NULL
;
712 LIST_HEAD(dirty_realms
);
714 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
716 dout("update_snap_trace deletion=%d\n", deletion
);
718 ceph_decode_need(&p
, e
, sizeof(*ri
), bad
);
721 ceph_decode_need(&p
, e
, sizeof(u64
)*(le32_to_cpu(ri
->num_snaps
) +
722 le32_to_cpu(ri
->num_prior_parent_snaps
)), bad
);
724 p
+= sizeof(u64
) * le32_to_cpu(ri
->num_snaps
);
725 prior_parent_snaps
= p
;
726 p
+= sizeof(u64
) * le32_to_cpu(ri
->num_prior_parent_snaps
);
728 realm
= ceph_lookup_snap_realm(mdsc
, le64_to_cpu(ri
->ino
));
730 realm
= ceph_create_snap_realm(mdsc
, le64_to_cpu(ri
->ino
));
732 err
= PTR_ERR(realm
);
737 /* ensure the parent is correct */
738 err
= adjust_snap_realm_parent(mdsc
, realm
, le64_to_cpu(ri
->parent
));
743 if (le64_to_cpu(ri
->seq
) > realm
->seq
) {
744 dout("update_snap_trace updating %llx %p %lld -> %lld\n",
745 realm
->ino
, realm
, realm
->seq
, le64_to_cpu(ri
->seq
));
746 /* update realm parameters, snap lists */
747 realm
->seq
= le64_to_cpu(ri
->seq
);
748 realm
->created
= le64_to_cpu(ri
->created
);
749 realm
->parent_since
= le64_to_cpu(ri
->parent_since
);
751 realm
->num_snaps
= le32_to_cpu(ri
->num_snaps
);
752 err
= dup_array(&realm
->snaps
, snaps
, realm
->num_snaps
);
756 realm
->num_prior_parent_snaps
=
757 le32_to_cpu(ri
->num_prior_parent_snaps
);
758 err
= dup_array(&realm
->prior_parent_snaps
, prior_parent_snaps
,
759 realm
->num_prior_parent_snaps
);
763 if (realm
->seq
> mdsc
->last_snap_seq
)
764 mdsc
->last_snap_seq
= realm
->seq
;
767 } else if (!realm
->cached_context
) {
768 dout("update_snap_trace %llx %p seq %lld new\n",
769 realm
->ino
, realm
, realm
->seq
);
772 dout("update_snap_trace %llx %p seq %lld unchanged\n",
773 realm
->ino
, realm
, realm
->seq
);
776 dout("done with %llx %p, invalidated=%d, %p %p\n", realm
->ino
,
777 realm
, invalidate
, p
, e
);
779 /* invalidate when we reach the _end_ (root) of the trace */
780 if (invalidate
&& p
>= e
)
781 rebuild_snap_realms(realm
, &dirty_realms
);
786 ceph_put_snap_realm(mdsc
, realm
);
792 * queue cap snaps _after_ we've built the new snap contexts,
793 * so that i_head_snapc can be set appropriately.
795 while (!list_empty(&dirty_realms
)) {
796 realm
= list_first_entry(&dirty_realms
, struct ceph_snap_realm
,
798 list_del_init(&realm
->dirty_item
);
799 queue_realm_cap_snaps(realm
);
803 *realm_ret
= first_realm
;
805 ceph_put_snap_realm(mdsc
, first_realm
);
807 __cleanup_empty_realms(mdsc
);
813 if (realm
&& !IS_ERR(realm
))
814 ceph_put_snap_realm(mdsc
, realm
);
816 ceph_put_snap_realm(mdsc
, first_realm
);
817 pr_err("update_snap_trace error %d\n", err
);
823 * Send any cap_snaps that are queued for flush. Try to carry
824 * s_mutex across multiple snap flushes to avoid locking overhead.
826 * Caller holds no locks.
828 static void flush_snaps(struct ceph_mds_client
*mdsc
)
830 struct ceph_inode_info
*ci
;
832 struct ceph_mds_session
*session
= NULL
;
834 dout("flush_snaps\n");
835 spin_lock(&mdsc
->snap_flush_lock
);
836 while (!list_empty(&mdsc
->snap_flush_list
)) {
837 ci
= list_first_entry(&mdsc
->snap_flush_list
,
838 struct ceph_inode_info
, i_snap_flush_item
);
839 inode
= &ci
->vfs_inode
;
841 spin_unlock(&mdsc
->snap_flush_lock
);
842 ceph_flush_snaps(ci
, &session
);
844 spin_lock(&mdsc
->snap_flush_lock
);
846 spin_unlock(&mdsc
->snap_flush_lock
);
848 ceph_put_mds_session(session
);
849 dout("flush_snaps done\n");
853 * ceph_change_snap_realm - change the snap_realm for an inode
854 * @inode: inode to move to new snap realm
855 * @realm: new realm to move inode into (may be NULL)
857 * Detach an inode from its old snaprealm (if any) and attach it to
858 * the new snaprealm (if any). The old snap realm reference held by
859 * the inode is put. If realm is non-NULL, then the caller's reference
860 * to it is taken over by the inode.
862 void ceph_change_snap_realm(struct inode
*inode
, struct ceph_snap_realm
*realm
)
864 struct ceph_inode_info
*ci
= ceph_inode(inode
);
865 struct ceph_mds_client
*mdsc
= ceph_inode_to_client(inode
)->mdsc
;
866 struct ceph_snap_realm
*oldrealm
= ci
->i_snap_realm
;
868 lockdep_assert_held(&ci
->i_ceph_lock
);
871 spin_lock(&oldrealm
->inodes_with_caps_lock
);
872 list_del_init(&ci
->i_snap_realm_item
);
873 if (oldrealm
->ino
== ci
->i_vino
.ino
)
874 oldrealm
->inode
= NULL
;
875 spin_unlock(&oldrealm
->inodes_with_caps_lock
);
876 ceph_put_snap_realm(mdsc
, oldrealm
);
879 ci
->i_snap_realm
= realm
;
882 spin_lock(&realm
->inodes_with_caps_lock
);
883 list_add(&ci
->i_snap_realm_item
, &realm
->inodes_with_caps
);
884 if (realm
->ino
== ci
->i_vino
.ino
)
885 realm
->inode
= inode
;
886 spin_unlock(&realm
->inodes_with_caps_lock
);
891 * Handle a snap notification from the MDS.
893 * This can take two basic forms: the simplest is just a snap creation
894 * or deletion notification on an existing realm. This should update the
895 * realm and its children.
897 * The more difficult case is realm creation, due to snap creation at a
898 * new point in the file hierarchy, or due to a rename that moves a file or
899 * directory into another realm.
901 void ceph_handle_snap(struct ceph_mds_client
*mdsc
,
902 struct ceph_mds_session
*session
,
903 struct ceph_msg
*msg
)
905 struct super_block
*sb
= mdsc
->fsc
->sb
;
906 int mds
= session
->s_mds
;
910 struct ceph_snap_realm
*realm
= NULL
;
911 void *p
= msg
->front
.iov_base
;
912 void *e
= p
+ msg
->front
.iov_len
;
913 struct ceph_mds_snap_head
*h
;
914 int num_split_inos
, num_split_realms
;
915 __le64
*split_inos
= NULL
, *split_realms
= NULL
;
917 int locked_rwsem
= 0;
920 if (msg
->front
.iov_len
< sizeof(*h
))
923 op
= le32_to_cpu(h
->op
);
924 split
= le64_to_cpu(h
->split
); /* non-zero if we are splitting an
926 num_split_inos
= le32_to_cpu(h
->num_split_inos
);
927 num_split_realms
= le32_to_cpu(h
->num_split_realms
);
928 trace_len
= le32_to_cpu(h
->trace_len
);
931 dout("handle_snap from mds%d op %s split %llx tracelen %d\n", mds
,
932 ceph_snap_op_name(op
), split
, trace_len
);
934 mutex_lock(&session
->s_mutex
);
935 inc_session_sequence(session
);
936 mutex_unlock(&session
->s_mutex
);
938 down_write(&mdsc
->snap_rwsem
);
941 if (op
== CEPH_SNAP_OP_SPLIT
) {
942 struct ceph_mds_snap_realm
*ri
;
945 * A "split" breaks part of an existing realm off into
946 * a new realm. The MDS provides a list of inodes
947 * (with caps) and child realms that belong to the new
951 p
+= sizeof(u64
) * num_split_inos
;
953 p
+= sizeof(u64
) * num_split_realms
;
954 ceph_decode_need(&p
, e
, sizeof(*ri
), bad
);
955 /* we will peek at realm info here, but will _not_
956 * advance p, as the realm update will occur below in
957 * ceph_update_snap_trace. */
960 realm
= ceph_lookup_snap_realm(mdsc
, split
);
962 realm
= ceph_create_snap_realm(mdsc
, split
);
967 dout("splitting snap_realm %llx %p\n", realm
->ino
, realm
);
968 for (i
= 0; i
< num_split_inos
; i
++) {
969 struct ceph_vino vino
= {
970 .ino
= le64_to_cpu(split_inos
[i
]),
973 struct inode
*inode
= ceph_find_inode(sb
, vino
);
974 struct ceph_inode_info
*ci
;
978 ci
= ceph_inode(inode
);
980 spin_lock(&ci
->i_ceph_lock
);
981 if (!ci
->i_snap_realm
)
984 * If this inode belongs to a realm that was
985 * created after our new realm, we experienced
986 * a race (due to another split notifications
987 * arriving from a different MDS). So skip
990 if (ci
->i_snap_realm
->created
>
991 le64_to_cpu(ri
->created
)) {
992 dout(" leaving %p in newer realm %llx %p\n",
993 inode
, ci
->i_snap_realm
->ino
,
997 dout(" will move %p to split realm %llx %p\n",
998 inode
, realm
->ino
, realm
);
1000 ceph_get_snap_realm(mdsc
, realm
);
1001 ceph_change_snap_realm(inode
, realm
);
1002 spin_unlock(&ci
->i_ceph_lock
);
1007 spin_unlock(&ci
->i_ceph_lock
);
1011 /* we may have taken some of the old realm's children. */
1012 for (i
= 0; i
< num_split_realms
; i
++) {
1013 struct ceph_snap_realm
*child
=
1014 __lookup_snap_realm(mdsc
,
1015 le64_to_cpu(split_realms
[i
]));
1018 adjust_snap_realm_parent(mdsc
, child
, realm
->ino
);
1023 * update using the provided snap trace. if we are deleting a
1024 * snap, we can avoid queueing cap_snaps.
1026 ceph_update_snap_trace(mdsc
, p
, e
,
1027 op
== CEPH_SNAP_OP_DESTROY
, NULL
);
1029 if (op
== CEPH_SNAP_OP_SPLIT
)
1030 /* we took a reference when we created the realm, above */
1031 ceph_put_snap_realm(mdsc
, realm
);
1033 __cleanup_empty_realms(mdsc
);
1035 up_write(&mdsc
->snap_rwsem
);
1041 pr_err("corrupt snap message from mds%d\n", mds
);
1045 up_write(&mdsc
->snap_rwsem
);
1049 struct ceph_snapid_map
* ceph_get_snapid_map(struct ceph_mds_client
*mdsc
,
1052 struct ceph_snapid_map
*sm
, *exist
;
1053 struct rb_node
**p
, *parent
;
1057 spin_lock(&mdsc
->snapid_map_lock
);
1058 p
= &mdsc
->snapid_map_tree
.rb_node
;
1060 exist
= rb_entry(*p
, struct ceph_snapid_map
, node
);
1061 if (snap
> exist
->snap
) {
1063 } else if (snap
< exist
->snap
) {
1064 p
= &(*p
)->rb_right
;
1066 if (atomic_inc_return(&exist
->ref
) == 1)
1067 list_del_init(&exist
->lru
);
1072 spin_unlock(&mdsc
->snapid_map_lock
);
1074 dout("found snapid map %llx -> %x\n", exist
->snap
, exist
->dev
);
1078 sm
= kmalloc(sizeof(*sm
), GFP_NOFS
);
1082 ret
= get_anon_bdev(&sm
->dev
);
1088 INIT_LIST_HEAD(&sm
->lru
);
1089 atomic_set(&sm
->ref
, 1);
1094 p
= &mdsc
->snapid_map_tree
.rb_node
;
1095 spin_lock(&mdsc
->snapid_map_lock
);
1098 exist
= rb_entry(*p
, struct ceph_snapid_map
, node
);
1099 if (snap
> exist
->snap
)
1101 else if (snap
< exist
->snap
)
1102 p
= &(*p
)->rb_right
;
1108 if (atomic_inc_return(&exist
->ref
) == 1)
1109 list_del_init(&exist
->lru
);
1111 rb_link_node(&sm
->node
, parent
, p
);
1112 rb_insert_color(&sm
->node
, &mdsc
->snapid_map_tree
);
1114 spin_unlock(&mdsc
->snapid_map_lock
);
1116 free_anon_bdev(sm
->dev
);
1118 dout("found snapid map %llx -> %x\n", exist
->snap
, exist
->dev
);
1122 dout("create snapid map %llx -> %x\n", sm
->snap
, sm
->dev
);
1126 void ceph_put_snapid_map(struct ceph_mds_client
* mdsc
,
1127 struct ceph_snapid_map
*sm
)
1131 if (atomic_dec_and_lock(&sm
->ref
, &mdsc
->snapid_map_lock
)) {
1132 if (!RB_EMPTY_NODE(&sm
->node
)) {
1133 sm
->last_used
= jiffies
;
1134 list_add_tail(&sm
->lru
, &mdsc
->snapid_map_lru
);
1135 spin_unlock(&mdsc
->snapid_map_lock
);
1137 /* already cleaned up by
1138 * ceph_cleanup_snapid_map() */
1139 spin_unlock(&mdsc
->snapid_map_lock
);
1145 void ceph_trim_snapid_map(struct ceph_mds_client
*mdsc
)
1147 struct ceph_snapid_map
*sm
;
1151 spin_lock(&mdsc
->snapid_map_lock
);
1154 while (!list_empty(&mdsc
->snapid_map_lru
)) {
1155 sm
= list_first_entry(&mdsc
->snapid_map_lru
,
1156 struct ceph_snapid_map
, lru
);
1157 if (time_after(sm
->last_used
+ CEPH_SNAPID_MAP_TIMEOUT
, now
))
1160 rb_erase(&sm
->node
, &mdsc
->snapid_map_tree
);
1161 list_move(&sm
->lru
, &to_free
);
1163 spin_unlock(&mdsc
->snapid_map_lock
);
1165 while (!list_empty(&to_free
)) {
1166 sm
= list_first_entry(&to_free
, struct ceph_snapid_map
, lru
);
1168 dout("trim snapid map %llx -> %x\n", sm
->snap
, sm
->dev
);
1169 free_anon_bdev(sm
->dev
);
1174 void ceph_cleanup_snapid_map(struct ceph_mds_client
*mdsc
)
1176 struct ceph_snapid_map
*sm
;
1180 spin_lock(&mdsc
->snapid_map_lock
);
1181 while ((p
= rb_first(&mdsc
->snapid_map_tree
))) {
1182 sm
= rb_entry(p
, struct ceph_snapid_map
, node
);
1183 rb_erase(p
, &mdsc
->snapid_map_tree
);
1185 list_move(&sm
->lru
, &to_free
);
1187 spin_unlock(&mdsc
->snapid_map_lock
);
1189 while (!list_empty(&to_free
)) {
1190 sm
= list_first_entry(&to_free
, struct ceph_snapid_map
, lru
);
1192 free_anon_bdev(sm
->dev
);
1193 if (WARN_ON_ONCE(atomic_read(&sm
->ref
))) {
1194 pr_err("snapid map %llx -> %x still in use\n",