]>
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
);
491 spin_lock(&ci
->i_ceph_lock
);
492 used
= __ceph_caps_used(ci
);
493 dirty
= __ceph_caps_dirty(ci
);
495 old_snapc
= ci
->i_head_snapc
;
496 new_snapc
= ci
->i_snap_realm
->cached_context
;
499 * If there is a write in progress, treat that as a dirty Fw,
500 * even though it hasn't completed yet; by the time we finish
501 * up this capsnap it will be.
503 if (used
& CEPH_CAP_FILE_WR
)
504 dirty
|= CEPH_CAP_FILE_WR
;
506 if (__ceph_have_pending_cap_snap(ci
)) {
507 /* there is no point in queuing multiple "pending" cap_snaps,
508 as no new writes are allowed to start when pending, so any
509 writes in progress now were started before the previous
510 cap_snap. lucky us. */
511 dout("queue_cap_snap %p already pending\n", inode
);
514 if (ci
->i_wrbuffer_ref_head
== 0 &&
515 !(dirty
& (CEPH_CAP_ANY_EXCL
|CEPH_CAP_FILE_WR
))) {
516 dout("queue_cap_snap %p nothing dirty|writing\n", inode
);
523 * There is no need to send FLUSHSNAP message to MDS if there is
524 * no new snapshot. But when there is dirty pages or on-going
525 * writes, we still need to create cap_snap. cap_snap is needed
526 * by the write path and page writeback path.
528 * also see ceph_try_drop_cap_snap()
530 if (has_new_snaps(old_snapc
, new_snapc
)) {
531 if (dirty
& (CEPH_CAP_ANY_EXCL
|CEPH_CAP_FILE_WR
))
532 capsnap
->need_flush
= true;
534 if (!(used
& CEPH_CAP_FILE_WR
) &&
535 ci
->i_wrbuffer_ref_head
== 0) {
536 dout("queue_cap_snap %p "
537 "no new_snap|dirty_page|writing\n", inode
);
542 dout("queue_cap_snap %p cap_snap %p queuing under %p %s %s\n",
543 inode
, capsnap
, old_snapc
, ceph_cap_string(dirty
),
544 capsnap
->need_flush
? "" : "no_flush");
547 refcount_set(&capsnap
->nref
, 1);
548 INIT_LIST_HEAD(&capsnap
->ci_item
);
550 capsnap
->follows
= old_snapc
->seq
;
551 capsnap
->issued
= __ceph_caps_issued(ci
, NULL
);
552 capsnap
->dirty
= dirty
;
554 capsnap
->mode
= inode
->i_mode
;
555 capsnap
->uid
= inode
->i_uid
;
556 capsnap
->gid
= inode
->i_gid
;
558 if (dirty
& CEPH_CAP_XATTR_EXCL
) {
559 old_blob
= __ceph_build_xattrs_blob(ci
);
560 capsnap
->xattr_blob
=
561 ceph_buffer_get(ci
->i_xattrs
.blob
);
562 capsnap
->xattr_version
= ci
->i_xattrs
.version
;
564 capsnap
->xattr_blob
= NULL
;
565 capsnap
->xattr_version
= 0;
568 capsnap
->inline_data
= ci
->i_inline_version
!= CEPH_INLINE_NONE
;
570 /* dirty page count moved from _head to this cap_snap;
571 all subsequent writes page dirties occur _after_ this
573 capsnap
->dirty_pages
= ci
->i_wrbuffer_ref_head
;
574 ci
->i_wrbuffer_ref_head
= 0;
575 capsnap
->context
= old_snapc
;
576 list_add_tail(&capsnap
->ci_item
, &ci
->i_cap_snaps
);
578 if (used
& CEPH_CAP_FILE_WR
) {
579 dout("queue_cap_snap %p cap_snap %p snapc %p"
580 " seq %llu used WR, now pending\n", inode
,
581 capsnap
, old_snapc
, old_snapc
->seq
);
582 capsnap
->writing
= 1;
584 /* note mtime, size NOW. */
585 __ceph_finish_cap_snap(ci
, capsnap
);
591 if (ci
->i_wrbuffer_ref_head
== 0 &&
593 ci
->i_dirty_caps
== 0 &&
594 ci
->i_flushing_caps
== 0) {
595 ci
->i_head_snapc
= NULL
;
597 ci
->i_head_snapc
= ceph_get_snap_context(new_snapc
);
598 dout(" new snapc is %p\n", new_snapc
);
600 spin_unlock(&ci
->i_ceph_lock
);
602 ceph_buffer_put(old_blob
);
604 ceph_put_snap_context(old_snapc
);
608 * Finalize the size, mtime for a cap_snap.. that is, settle on final values
609 * to be used for the snapshot, to be flushed back to the mds.
611 * If capsnap can now be flushed, add to snap_flush list, and return 1.
613 * Caller must hold i_ceph_lock.
615 int __ceph_finish_cap_snap(struct ceph_inode_info
*ci
,
616 struct ceph_cap_snap
*capsnap
)
618 struct inode
*inode
= &ci
->vfs_inode
;
619 struct ceph_mds_client
*mdsc
= ceph_sb_to_mdsc(inode
->i_sb
);
621 BUG_ON(capsnap
->writing
);
622 capsnap
->size
= i_size_read(inode
);
623 capsnap
->mtime
= inode
->i_mtime
;
624 capsnap
->atime
= inode
->i_atime
;
625 capsnap
->ctime
= inode
->i_ctime
;
626 capsnap
->btime
= ci
->i_btime
;
627 capsnap
->change_attr
= inode_peek_iversion_raw(inode
);
628 capsnap
->time_warp_seq
= ci
->i_time_warp_seq
;
629 capsnap
->truncate_size
= ci
->i_truncate_size
;
630 capsnap
->truncate_seq
= ci
->i_truncate_seq
;
631 if (capsnap
->dirty_pages
) {
632 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu "
633 "still has %d dirty pages\n", inode
, capsnap
,
634 capsnap
->context
, capsnap
->context
->seq
,
635 ceph_cap_string(capsnap
->dirty
), capsnap
->size
,
636 capsnap
->dirty_pages
);
640 /* Fb cap still in use, delay it */
642 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu "
643 "used WRBUFFER, delaying\n", inode
, capsnap
,
644 capsnap
->context
, capsnap
->context
->seq
,
645 ceph_cap_string(capsnap
->dirty
), capsnap
->size
);
646 capsnap
->writing
= 1;
650 ci
->i_ceph_flags
|= CEPH_I_FLUSH_SNAPS
;
651 dout("finish_cap_snap %p cap_snap %p snapc %p %llu %s s=%llu\n",
652 inode
, capsnap
, capsnap
->context
,
653 capsnap
->context
->seq
, ceph_cap_string(capsnap
->dirty
),
656 spin_lock(&mdsc
->snap_flush_lock
);
657 if (list_empty(&ci
->i_snap_flush_item
))
658 list_add_tail(&ci
->i_snap_flush_item
, &mdsc
->snap_flush_list
);
659 spin_unlock(&mdsc
->snap_flush_lock
);
660 return 1; /* caller may want to ceph_flush_snaps */
664 * Queue cap_snaps for snap writeback for this realm and its children.
665 * Called under snap_rwsem, so realm topology won't change.
667 static void queue_realm_cap_snaps(struct ceph_snap_realm
*realm
)
669 struct ceph_inode_info
*ci
;
670 struct inode
*lastinode
= NULL
;
672 dout("queue_realm_cap_snaps %p %llx inodes\n", realm
, realm
->ino
);
674 spin_lock(&realm
->inodes_with_caps_lock
);
675 list_for_each_entry(ci
, &realm
->inodes_with_caps
, i_snap_realm_item
) {
676 struct inode
*inode
= igrab(&ci
->vfs_inode
);
679 spin_unlock(&realm
->inodes_with_caps_lock
);
682 ceph_queue_cap_snap(ci
);
683 spin_lock(&realm
->inodes_with_caps_lock
);
685 spin_unlock(&realm
->inodes_with_caps_lock
);
688 dout("queue_realm_cap_snaps %p %llx done\n", realm
, realm
->ino
);
692 * Parse and apply a snapblob "snap trace" from the MDS. This specifies
693 * the snap realm parameters from a given realm and all of its ancestors,
696 * Caller must hold snap_rwsem for write.
698 int ceph_update_snap_trace(struct ceph_mds_client
*mdsc
,
699 void *p
, void *e
, bool deletion
,
700 struct ceph_snap_realm
**realm_ret
)
702 struct ceph_mds_snap_realm
*ri
; /* encoded */
703 __le64
*snaps
; /* encoded */
704 __le64
*prior_parent_snaps
; /* encoded */
705 struct ceph_snap_realm
*realm
= NULL
;
706 struct ceph_snap_realm
*first_realm
= NULL
;
709 LIST_HEAD(dirty_realms
);
711 lockdep_assert_held_write(&mdsc
->snap_rwsem
);
713 dout("update_snap_trace deletion=%d\n", deletion
);
715 ceph_decode_need(&p
, e
, sizeof(*ri
), bad
);
718 ceph_decode_need(&p
, e
, sizeof(u64
)*(le32_to_cpu(ri
->num_snaps
) +
719 le32_to_cpu(ri
->num_prior_parent_snaps
)), bad
);
721 p
+= sizeof(u64
) * le32_to_cpu(ri
->num_snaps
);
722 prior_parent_snaps
= p
;
723 p
+= sizeof(u64
) * le32_to_cpu(ri
->num_prior_parent_snaps
);
725 realm
= ceph_lookup_snap_realm(mdsc
, le64_to_cpu(ri
->ino
));
727 realm
= ceph_create_snap_realm(mdsc
, le64_to_cpu(ri
->ino
));
729 err
= PTR_ERR(realm
);
734 /* ensure the parent is correct */
735 err
= adjust_snap_realm_parent(mdsc
, realm
, le64_to_cpu(ri
->parent
));
740 if (le64_to_cpu(ri
->seq
) > realm
->seq
) {
741 dout("update_snap_trace updating %llx %p %lld -> %lld\n",
742 realm
->ino
, realm
, realm
->seq
, le64_to_cpu(ri
->seq
));
743 /* update realm parameters, snap lists */
744 realm
->seq
= le64_to_cpu(ri
->seq
);
745 realm
->created
= le64_to_cpu(ri
->created
);
746 realm
->parent_since
= le64_to_cpu(ri
->parent_since
);
748 realm
->num_snaps
= le32_to_cpu(ri
->num_snaps
);
749 err
= dup_array(&realm
->snaps
, snaps
, realm
->num_snaps
);
753 realm
->num_prior_parent_snaps
=
754 le32_to_cpu(ri
->num_prior_parent_snaps
);
755 err
= dup_array(&realm
->prior_parent_snaps
, prior_parent_snaps
,
756 realm
->num_prior_parent_snaps
);
760 if (realm
->seq
> mdsc
->last_snap_seq
)
761 mdsc
->last_snap_seq
= realm
->seq
;
764 } else if (!realm
->cached_context
) {
765 dout("update_snap_trace %llx %p seq %lld new\n",
766 realm
->ino
, realm
, realm
->seq
);
769 dout("update_snap_trace %llx %p seq %lld unchanged\n",
770 realm
->ino
, realm
, realm
->seq
);
773 dout("done with %llx %p, invalidated=%d, %p %p\n", realm
->ino
,
774 realm
, invalidate
, p
, e
);
776 /* invalidate when we reach the _end_ (root) of the trace */
777 if (invalidate
&& p
>= e
)
778 rebuild_snap_realms(realm
, &dirty_realms
);
783 ceph_put_snap_realm(mdsc
, realm
);
789 * queue cap snaps _after_ we've built the new snap contexts,
790 * so that i_head_snapc can be set appropriately.
792 while (!list_empty(&dirty_realms
)) {
793 realm
= list_first_entry(&dirty_realms
, struct ceph_snap_realm
,
795 list_del_init(&realm
->dirty_item
);
796 queue_realm_cap_snaps(realm
);
800 *realm_ret
= first_realm
;
802 ceph_put_snap_realm(mdsc
, first_realm
);
804 __cleanup_empty_realms(mdsc
);
810 if (realm
&& !IS_ERR(realm
))
811 ceph_put_snap_realm(mdsc
, realm
);
813 ceph_put_snap_realm(mdsc
, first_realm
);
814 pr_err("update_snap_trace error %d\n", err
);
820 * Send any cap_snaps that are queued for flush. Try to carry
821 * s_mutex across multiple snap flushes to avoid locking overhead.
823 * Caller holds no locks.
825 static void flush_snaps(struct ceph_mds_client
*mdsc
)
827 struct ceph_inode_info
*ci
;
829 struct ceph_mds_session
*session
= NULL
;
831 dout("flush_snaps\n");
832 spin_lock(&mdsc
->snap_flush_lock
);
833 while (!list_empty(&mdsc
->snap_flush_list
)) {
834 ci
= list_first_entry(&mdsc
->snap_flush_list
,
835 struct ceph_inode_info
, i_snap_flush_item
);
836 inode
= &ci
->vfs_inode
;
838 spin_unlock(&mdsc
->snap_flush_lock
);
839 ceph_flush_snaps(ci
, &session
);
841 spin_lock(&mdsc
->snap_flush_lock
);
843 spin_unlock(&mdsc
->snap_flush_lock
);
845 ceph_put_mds_session(session
);
846 dout("flush_snaps done\n");
851 * Handle a snap notification from the MDS.
853 * This can take two basic forms: the simplest is just a snap creation
854 * or deletion notification on an existing realm. This should update the
855 * realm and its children.
857 * The more difficult case is realm creation, due to snap creation at a
858 * new point in the file hierarchy, or due to a rename that moves a file or
859 * directory into another realm.
861 void ceph_handle_snap(struct ceph_mds_client
*mdsc
,
862 struct ceph_mds_session
*session
,
863 struct ceph_msg
*msg
)
865 struct super_block
*sb
= mdsc
->fsc
->sb
;
866 int mds
= session
->s_mds
;
870 struct ceph_snap_realm
*realm
= NULL
;
871 void *p
= msg
->front
.iov_base
;
872 void *e
= p
+ msg
->front
.iov_len
;
873 struct ceph_mds_snap_head
*h
;
874 int num_split_inos
, num_split_realms
;
875 __le64
*split_inos
= NULL
, *split_realms
= NULL
;
877 int locked_rwsem
= 0;
880 if (msg
->front
.iov_len
< sizeof(*h
))
883 op
= le32_to_cpu(h
->op
);
884 split
= le64_to_cpu(h
->split
); /* non-zero if we are splitting an
886 num_split_inos
= le32_to_cpu(h
->num_split_inos
);
887 num_split_realms
= le32_to_cpu(h
->num_split_realms
);
888 trace_len
= le32_to_cpu(h
->trace_len
);
891 dout("handle_snap from mds%d op %s split %llx tracelen %d\n", mds
,
892 ceph_snap_op_name(op
), split
, trace_len
);
894 mutex_lock(&session
->s_mutex
);
895 inc_session_sequence(session
);
896 mutex_unlock(&session
->s_mutex
);
898 down_write(&mdsc
->snap_rwsem
);
901 if (op
== CEPH_SNAP_OP_SPLIT
) {
902 struct ceph_mds_snap_realm
*ri
;
905 * A "split" breaks part of an existing realm off into
906 * a new realm. The MDS provides a list of inodes
907 * (with caps) and child realms that belong to the new
911 p
+= sizeof(u64
) * num_split_inos
;
913 p
+= sizeof(u64
) * num_split_realms
;
914 ceph_decode_need(&p
, e
, sizeof(*ri
), bad
);
915 /* we will peek at realm info here, but will _not_
916 * advance p, as the realm update will occur below in
917 * ceph_update_snap_trace. */
920 realm
= ceph_lookup_snap_realm(mdsc
, split
);
922 realm
= ceph_create_snap_realm(mdsc
, split
);
927 dout("splitting snap_realm %llx %p\n", realm
->ino
, realm
);
928 for (i
= 0; i
< num_split_inos
; i
++) {
929 struct ceph_vino vino
= {
930 .ino
= le64_to_cpu(split_inos
[i
]),
933 struct inode
*inode
= ceph_find_inode(sb
, vino
);
934 struct ceph_inode_info
*ci
;
935 struct ceph_snap_realm
*oldrealm
;
939 ci
= ceph_inode(inode
);
941 spin_lock(&ci
->i_ceph_lock
);
942 if (!ci
->i_snap_realm
)
945 * If this inode belongs to a realm that was
946 * created after our new realm, we experienced
947 * a race (due to another split notifications
948 * arriving from a different MDS). So skip
951 if (ci
->i_snap_realm
->created
>
952 le64_to_cpu(ri
->created
)) {
953 dout(" leaving %p in newer realm %llx %p\n",
954 inode
, ci
->i_snap_realm
->ino
,
958 dout(" will move %p to split realm %llx %p\n",
959 inode
, realm
->ino
, realm
);
961 * Move the inode to the new realm
963 oldrealm
= ci
->i_snap_realm
;
964 spin_lock(&oldrealm
->inodes_with_caps_lock
);
965 list_del_init(&ci
->i_snap_realm_item
);
966 spin_unlock(&oldrealm
->inodes_with_caps_lock
);
968 spin_lock(&realm
->inodes_with_caps_lock
);
969 list_add(&ci
->i_snap_realm_item
,
970 &realm
->inodes_with_caps
);
971 ci
->i_snap_realm
= realm
;
972 if (realm
->ino
== ci
->i_vino
.ino
)
973 realm
->inode
= inode
;
974 spin_unlock(&realm
->inodes_with_caps_lock
);
976 spin_unlock(&ci
->i_ceph_lock
);
978 ceph_get_snap_realm(mdsc
, realm
);
979 ceph_put_snap_realm(mdsc
, oldrealm
);
985 spin_unlock(&ci
->i_ceph_lock
);
989 /* we may have taken some of the old realm's children. */
990 for (i
= 0; i
< num_split_realms
; i
++) {
991 struct ceph_snap_realm
*child
=
992 __lookup_snap_realm(mdsc
,
993 le64_to_cpu(split_realms
[i
]));
996 adjust_snap_realm_parent(mdsc
, child
, realm
->ino
);
1001 * update using the provided snap trace. if we are deleting a
1002 * snap, we can avoid queueing cap_snaps.
1004 ceph_update_snap_trace(mdsc
, p
, e
,
1005 op
== CEPH_SNAP_OP_DESTROY
, NULL
);
1007 if (op
== CEPH_SNAP_OP_SPLIT
)
1008 /* we took a reference when we created the realm, above */
1009 ceph_put_snap_realm(mdsc
, realm
);
1011 __cleanup_empty_realms(mdsc
);
1013 up_write(&mdsc
->snap_rwsem
);
1019 pr_err("corrupt snap message from mds%d\n", mds
);
1023 up_write(&mdsc
->snap_rwsem
);
1027 struct ceph_snapid_map
* ceph_get_snapid_map(struct ceph_mds_client
*mdsc
,
1030 struct ceph_snapid_map
*sm
, *exist
;
1031 struct rb_node
**p
, *parent
;
1035 spin_lock(&mdsc
->snapid_map_lock
);
1036 p
= &mdsc
->snapid_map_tree
.rb_node
;
1038 exist
= rb_entry(*p
, struct ceph_snapid_map
, node
);
1039 if (snap
> exist
->snap
) {
1041 } else if (snap
< exist
->snap
) {
1042 p
= &(*p
)->rb_right
;
1044 if (atomic_inc_return(&exist
->ref
) == 1)
1045 list_del_init(&exist
->lru
);
1050 spin_unlock(&mdsc
->snapid_map_lock
);
1052 dout("found snapid map %llx -> %x\n", exist
->snap
, exist
->dev
);
1056 sm
= kmalloc(sizeof(*sm
), GFP_NOFS
);
1060 ret
= get_anon_bdev(&sm
->dev
);
1066 INIT_LIST_HEAD(&sm
->lru
);
1067 atomic_set(&sm
->ref
, 1);
1072 p
= &mdsc
->snapid_map_tree
.rb_node
;
1073 spin_lock(&mdsc
->snapid_map_lock
);
1076 exist
= rb_entry(*p
, struct ceph_snapid_map
, node
);
1077 if (snap
> exist
->snap
)
1079 else if (snap
< exist
->snap
)
1080 p
= &(*p
)->rb_right
;
1086 if (atomic_inc_return(&exist
->ref
) == 1)
1087 list_del_init(&exist
->lru
);
1089 rb_link_node(&sm
->node
, parent
, p
);
1090 rb_insert_color(&sm
->node
, &mdsc
->snapid_map_tree
);
1092 spin_unlock(&mdsc
->snapid_map_lock
);
1094 free_anon_bdev(sm
->dev
);
1096 dout("found snapid map %llx -> %x\n", exist
->snap
, exist
->dev
);
1100 dout("create snapid map %llx -> %x\n", sm
->snap
, sm
->dev
);
1104 void ceph_put_snapid_map(struct ceph_mds_client
* mdsc
,
1105 struct ceph_snapid_map
*sm
)
1109 if (atomic_dec_and_lock(&sm
->ref
, &mdsc
->snapid_map_lock
)) {
1110 if (!RB_EMPTY_NODE(&sm
->node
)) {
1111 sm
->last_used
= jiffies
;
1112 list_add_tail(&sm
->lru
, &mdsc
->snapid_map_lru
);
1113 spin_unlock(&mdsc
->snapid_map_lock
);
1115 /* already cleaned up by
1116 * ceph_cleanup_snapid_map() */
1117 spin_unlock(&mdsc
->snapid_map_lock
);
1123 void ceph_trim_snapid_map(struct ceph_mds_client
*mdsc
)
1125 struct ceph_snapid_map
*sm
;
1129 spin_lock(&mdsc
->snapid_map_lock
);
1132 while (!list_empty(&mdsc
->snapid_map_lru
)) {
1133 sm
= list_first_entry(&mdsc
->snapid_map_lru
,
1134 struct ceph_snapid_map
, lru
);
1135 if (time_after(sm
->last_used
+ CEPH_SNAPID_MAP_TIMEOUT
, now
))
1138 rb_erase(&sm
->node
, &mdsc
->snapid_map_tree
);
1139 list_move(&sm
->lru
, &to_free
);
1141 spin_unlock(&mdsc
->snapid_map_lock
);
1143 while (!list_empty(&to_free
)) {
1144 sm
= list_first_entry(&to_free
, struct ceph_snapid_map
, lru
);
1146 dout("trim snapid map %llx -> %x\n", sm
->snap
, sm
->dev
);
1147 free_anon_bdev(sm
->dev
);
1152 void ceph_cleanup_snapid_map(struct ceph_mds_client
*mdsc
)
1154 struct ceph_snapid_map
*sm
;
1158 spin_lock(&mdsc
->snapid_map_lock
);
1159 while ((p
= rb_first(&mdsc
->snapid_map_tree
))) {
1160 sm
= rb_entry(p
, struct ceph_snapid_map
, node
);
1161 rb_erase(p
, &mdsc
->snapid_map_tree
);
1163 list_move(&sm
->lru
, &to_free
);
1165 spin_unlock(&mdsc
->snapid_map_lock
);
1167 while (!list_empty(&to_free
)) {
1168 sm
= list_first_entry(&to_free
, struct ceph_snapid_map
, lru
);
1170 free_anon_bdev(sm
->dev
);
1171 if (WARN_ON_ONCE(atomic_read(&sm
->ref
))) {
1172 pr_err("snapid map %llx -> %x still in use\n",