xref: /freebsd/sys/contrib/openzfs/module/zfs/dsl_bookmark.c (revision a0b956f5ac5e0941f9e74e24c1c53e05ad061a38)
1 /*
2  * CDDL HEADER START
3  *
4  * This file and its contents are supplied under the terms of the
5  * Common Development and Distribution License ("CDDL"), version 1.0.
6  * You may only use this file in accordance with the terms of version
7  * 1.0 of the CDDL.
8  *
9  * A full copy of the text of the CDDL should have accompanied this
10  * source.  A copy of the CDDL is also available via the Internet at
11  * http://www.illumos.org/license/CDDL.
12  *
13  * CDDL HEADER END
14  */
15 
16 /*
17  * Copyright (c) 2013, 2018 by Delphix. All rights reserved.
18  * Copyright 2017 Nexenta Systems, Inc.
19  * Copyright 2019, 2020 by Christian Schwarz. All rights reserved.
20  */
21 
22 #include <sys/zfs_context.h>
23 #include <sys/dsl_dataset.h>
24 #include <sys/dsl_dir.h>
25 #include <sys/dsl_prop.h>
26 #include <sys/dsl_synctask.h>
27 #include <sys/dsl_destroy.h>
28 #include <sys/dmu_impl.h>
29 #include <sys/dmu_tx.h>
30 #include <sys/arc.h>
31 #include <sys/zap.h>
32 #include <sys/zfeature.h>
33 #include <sys/spa.h>
34 #include <sys/dsl_bookmark.h>
35 #include <zfs_namecheck.h>
36 #include <sys/dmu_send.h>
37 
38 static int
39 dsl_bookmark_hold_ds(dsl_pool_t *dp, const char *fullname,
40     dsl_dataset_t **dsp, const void *tag, char **shortnamep)
41 {
42 	char buf[ZFS_MAX_DATASET_NAME_LEN];
43 	char *hashp;
44 
45 	if (strlen(fullname) >= ZFS_MAX_DATASET_NAME_LEN)
46 		return (SET_ERROR(ENAMETOOLONG));
47 	hashp = strchr(fullname, '#');
48 	if (hashp == NULL)
49 		return (SET_ERROR(EINVAL));
50 
51 	*shortnamep = hashp + 1;
52 	if (zfs_component_namecheck(*shortnamep, NULL, NULL))
53 		return (SET_ERROR(EINVAL));
54 	(void) strlcpy(buf, fullname, hashp - fullname + 1);
55 	return (dsl_dataset_hold(dp, buf, tag, dsp));
56 }
57 
58 /*
59  * When reading BOOKMARK_V1 bookmarks, the BOOKMARK_V2 fields are guaranteed
60  * to be zeroed.
61  *
62  * Returns ESRCH if bookmark is not found.
63  * Note, we need to use the ZAP rather than the AVL to look up bookmarks
64  * by name, because only the ZAP honors the casesensitivity setting.
65  */
66 int
67 dsl_bookmark_lookup_impl(dsl_dataset_t *ds, const char *shortname,
68     zfs_bookmark_phys_t *bmark_phys)
69 {
70 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
71 	uint64_t bmark_zapobj = ds->ds_bookmarks_obj;
72 	matchtype_t mt = 0;
73 	int err;
74 
75 	if (bmark_zapobj == 0)
76 		return (SET_ERROR(ESRCH));
77 
78 	if (dsl_dataset_phys(ds)->ds_flags & DS_FLAG_CI_DATASET)
79 		mt = MT_NORMALIZE;
80 
81 	/*
82 	 * Zero out the bookmark in case the one stored on disk
83 	 * is in an older, shorter format.
84 	 */
85 	memset(bmark_phys, 0, sizeof (*bmark_phys));
86 
87 	err = zap_lookup_norm(mos, bmark_zapobj, shortname, sizeof (uint64_t),
88 	    sizeof (*bmark_phys) / sizeof (uint64_t), bmark_phys, mt, NULL, 0,
89 	    NULL);
90 
91 	return (err == ENOENT ? SET_ERROR(ESRCH) : err);
92 }
93 
94 /*
95  * If later_ds is non-NULL, this will return EXDEV if the specified bookmark
96  * does not represents an earlier point in later_ds's timeline.  However,
97  * bmp will still be filled in if we return EXDEV.
98  *
99  * Returns ENOENT if the dataset containing the bookmark does not exist.
100  * Returns ESRCH if the dataset exists but the bookmark was not found in it.
101  */
102 int
103 dsl_bookmark_lookup(dsl_pool_t *dp, const char *fullname,
104     dsl_dataset_t *later_ds, zfs_bookmark_phys_t *bmp)
105 {
106 	char *shortname;
107 	dsl_dataset_t *ds;
108 	int error;
109 
110 	error = dsl_bookmark_hold_ds(dp, fullname, &ds, FTAG, &shortname);
111 	if (error != 0)
112 		return (error);
113 
114 	error = dsl_bookmark_lookup_impl(ds, shortname, bmp);
115 	if (error == 0 && later_ds != NULL) {
116 		if (!dsl_dataset_is_before(later_ds, ds, bmp->zbm_creation_txg))
117 			error = SET_ERROR(EXDEV);
118 	}
119 	dsl_dataset_rele(ds, FTAG);
120 	return (error);
121 }
122 
123 /*
124  * Validates that
125  * - bmark is a full dataset path of a bookmark (bookmark_namecheck)
126  * - source is a full path of a snapshot or bookmark
127  *   ({bookmark,snapshot}_namecheck)
128  *
129  * Returns 0 if valid, -1 otherwise.
130  */
131 static int
132 dsl_bookmark_create_nvl_validate_pair(const char *bmark, const char *source)
133 {
134 	if (bookmark_namecheck(bmark, NULL, NULL) != 0)
135 		return (-1);
136 
137 	int is_bmark, is_snap;
138 	is_bmark = bookmark_namecheck(source, NULL, NULL) == 0;
139 	is_snap = snapshot_namecheck(source, NULL, NULL) == 0;
140 	if (!is_bmark && !is_snap)
141 		return (-1);
142 
143 	return (0);
144 }
145 
146 /*
147  * Check that the given nvlist corresponds to the following schema:
148  *  { newbookmark -> source, ... }
149  * where
150  * - each pair passes dsl_bookmark_create_nvl_validate_pair
151  * - all newbookmarks are in the same pool
152  * - all newbookmarks have unique names
153  *
154  * Note that this function is only validates above schema. Callers must ensure
155  * that the bookmarks can be created, e.g. that sources exist.
156  *
157  * Returns 0 if the nvlist adheres to above schema.
158  * Returns -1 if it doesn't.
159  */
160 int
161 dsl_bookmark_create_nvl_validate(nvlist_t *bmarks)
162 {
163 	char *first = NULL;
164 	size_t first_len = 0;
165 
166 	for (nvpair_t *pair = nvlist_next_nvpair(bmarks, NULL);
167 	    pair != NULL; pair = nvlist_next_nvpair(bmarks, pair)) {
168 
169 		char *bmark = nvpair_name(pair);
170 		char *source;
171 
172 		/* list structure: values must be snapshots XOR bookmarks */
173 		if (nvpair_value_string(pair, &source) != 0)
174 			return (-1);
175 		if (dsl_bookmark_create_nvl_validate_pair(bmark, source) != 0)
176 			return (-1);
177 
178 		/* same pool check */
179 		if (first == NULL) {
180 			char *cp = strpbrk(bmark, "/#");
181 			if (cp == NULL)
182 				return (-1);
183 			first = bmark;
184 			first_len = cp - bmark;
185 		}
186 		if (strncmp(first, bmark, first_len) != 0)
187 			return (-1);
188 		switch (*(bmark + first_len)) {
189 			case '/': /* fallthrough */
190 			case '#':
191 				break;
192 			default:
193 				return (-1);
194 		}
195 
196 		/* unique newbookmark names; todo: O(n^2) */
197 		for (nvpair_t *pair2 = nvlist_next_nvpair(bmarks, pair);
198 		    pair2 != NULL; pair2 = nvlist_next_nvpair(bmarks, pair2)) {
199 			if (strcmp(nvpair_name(pair), nvpair_name(pair2)) == 0)
200 				return (-1);
201 		}
202 
203 	}
204 	return (0);
205 }
206 
207 /*
208  * expects that newbm and source have been validated using
209  * dsl_bookmark_create_nvl_validate_pair
210  */
211 static int
212 dsl_bookmark_create_check_impl(dsl_pool_t *dp,
213     const char *newbm, const char *source)
214 {
215 	ASSERT0(dsl_bookmark_create_nvl_validate_pair(newbm, source));
216 	/* defer source namecheck until we know it's a snapshot or bookmark */
217 
218 	int error;
219 	dsl_dataset_t *newbm_ds;
220 	char *newbm_short;
221 	zfs_bookmark_phys_t bmark_phys;
222 
223 	error = dsl_bookmark_hold_ds(dp, newbm, &newbm_ds, FTAG, &newbm_short);
224 	if (error != 0)
225 		return (error);
226 
227 	/* Verify that the new bookmark does not already exist */
228 	error = dsl_bookmark_lookup_impl(newbm_ds, newbm_short, &bmark_phys);
229 	switch (error) {
230 	case ESRCH:
231 		/* happy path: new bmark doesn't exist, proceed after switch */
232 		error = 0;
233 		break;
234 	case 0:
235 		error = SET_ERROR(EEXIST);
236 		goto eholdnewbmds;
237 	default:
238 		/* dsl_bookmark_lookup_impl already did SET_ERROR */
239 		goto eholdnewbmds;
240 	}
241 
242 	/* error is retval of the following if-cascade */
243 	if (strchr(source, '@') != NULL) {
244 		dsl_dataset_t *source_snap_ds;
245 		ASSERT3S(snapshot_namecheck(source, NULL, NULL), ==, 0);
246 		error = dsl_dataset_hold(dp, source, FTAG, &source_snap_ds);
247 		if (error == 0) {
248 			VERIFY(source_snap_ds->ds_is_snapshot);
249 			/*
250 			 * Verify that source snapshot is an earlier point in
251 			 * newbm_ds's timeline (source may be newbm_ds's origin)
252 			 */
253 			if (!dsl_dataset_is_before(newbm_ds, source_snap_ds, 0))
254 				error = SET_ERROR(
255 				    ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR);
256 			dsl_dataset_rele(source_snap_ds, FTAG);
257 		}
258 	} else if (strchr(source, '#') != NULL) {
259 		zfs_bookmark_phys_t source_phys;
260 		ASSERT3S(bookmark_namecheck(source, NULL, NULL), ==, 0);
261 		/*
262 		 * Source must exists and be an earlier point in newbm_ds's
263 		 * timeline (newbm_ds's origin may be a snap of source's ds)
264 		 */
265 		error = dsl_bookmark_lookup(dp, source, newbm_ds, &source_phys);
266 		switch (error) {
267 		case 0:
268 			break; /* happy path */
269 		case EXDEV:
270 			error = SET_ERROR(ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR);
271 			break;
272 		default:
273 			/* dsl_bookmark_lookup already did SET_ERROR */
274 			break;
275 		}
276 	} else {
277 		/*
278 		 * dsl_bookmark_create_nvl_validate validates that source is
279 		 * either snapshot or bookmark
280 		 */
281 		panic("unreachable code: %s", source);
282 	}
283 
284 eholdnewbmds:
285 	dsl_dataset_rele(newbm_ds, FTAG);
286 	return (error);
287 }
288 
289 int
290 dsl_bookmark_create_check(void *arg, dmu_tx_t *tx)
291 {
292 	dsl_bookmark_create_arg_t *dbca = arg;
293 	int rv = 0;
294 	int schema_err = 0;
295 	ASSERT3P(dbca, !=, NULL);
296 	ASSERT3P(dbca->dbca_bmarks, !=, NULL);
297 	/* dbca->dbca_errors is allowed to be NULL */
298 
299 	dsl_pool_t *dp = dmu_tx_pool(tx);
300 
301 	if (!spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_BOOKMARKS))
302 		return (SET_ERROR(ENOTSUP));
303 
304 	if (dsl_bookmark_create_nvl_validate(dbca->dbca_bmarks) != 0)
305 		rv = schema_err = SET_ERROR(EINVAL);
306 
307 	for (nvpair_t *pair = nvlist_next_nvpair(dbca->dbca_bmarks, NULL);
308 	    pair != NULL; pair = nvlist_next_nvpair(dbca->dbca_bmarks, pair)) {
309 		char *new = nvpair_name(pair);
310 
311 		int error = schema_err;
312 		if (error == 0) {
313 			char *source = fnvpair_value_string(pair);
314 			error = dsl_bookmark_create_check_impl(dp, new, source);
315 			if (error != 0)
316 				error = SET_ERROR(error);
317 		}
318 
319 		if (error != 0) {
320 			rv = error;
321 			if (dbca->dbca_errors != NULL)
322 				fnvlist_add_int32(dbca->dbca_errors,
323 				    new, error);
324 		}
325 	}
326 
327 	return (rv);
328 }
329 
330 static dsl_bookmark_node_t *
331 dsl_bookmark_node_alloc(char *shortname)
332 {
333 	dsl_bookmark_node_t *dbn = kmem_alloc(sizeof (*dbn), KM_SLEEP);
334 	dbn->dbn_name = spa_strdup(shortname);
335 	dbn->dbn_dirty = B_FALSE;
336 	mutex_init(&dbn->dbn_lock, NULL, MUTEX_DEFAULT, NULL);
337 	return (dbn);
338 }
339 
340 /*
341  * Set the fields in the zfs_bookmark_phys_t based on the specified snapshot.
342  */
343 static void
344 dsl_bookmark_set_phys(zfs_bookmark_phys_t *zbm, dsl_dataset_t *snap)
345 {
346 	spa_t *spa = dsl_dataset_get_spa(snap);
347 	objset_t *mos = spa_get_dsl(spa)->dp_meta_objset;
348 	dsl_dataset_phys_t *dsp = dsl_dataset_phys(snap);
349 
350 	memset(zbm, 0, sizeof (zfs_bookmark_phys_t));
351 	zbm->zbm_guid = dsp->ds_guid;
352 	zbm->zbm_creation_txg = dsp->ds_creation_txg;
353 	zbm->zbm_creation_time = dsp->ds_creation_time;
354 	zbm->zbm_redaction_obj = 0;
355 
356 	/*
357 	 * If the dataset is encrypted create a larger bookmark to
358 	 * accommodate the IVset guid. The IVset guid was added
359 	 * after the encryption feature to prevent a problem with
360 	 * raw sends. If we encounter an encrypted dataset without
361 	 * an IVset guid we fall back to a normal bookmark.
362 	 */
363 	if (snap->ds_dir->dd_crypto_obj != 0 &&
364 	    spa_feature_is_enabled(spa, SPA_FEATURE_BOOKMARK_V2)) {
365 		(void) zap_lookup(mos, snap->ds_object,
366 		    DS_FIELD_IVSET_GUID, sizeof (uint64_t), 1,
367 		    &zbm->zbm_ivset_guid);
368 	}
369 
370 	if (spa_feature_is_enabled(spa, SPA_FEATURE_BOOKMARK_WRITTEN)) {
371 		zbm->zbm_flags = ZBM_FLAG_SNAPSHOT_EXISTS | ZBM_FLAG_HAS_FBN;
372 		zbm->zbm_referenced_bytes_refd = dsp->ds_referenced_bytes;
373 		zbm->zbm_compressed_bytes_refd = dsp->ds_compressed_bytes;
374 		zbm->zbm_uncompressed_bytes_refd = dsp->ds_uncompressed_bytes;
375 
376 		dsl_dataset_t *nextds;
377 		VERIFY0(dsl_dataset_hold_obj(snap->ds_dir->dd_pool,
378 		    dsp->ds_next_snap_obj, FTAG, &nextds));
379 		dsl_deadlist_space(&nextds->ds_deadlist,
380 		    &zbm->zbm_referenced_freed_before_next_snap,
381 		    &zbm->zbm_compressed_freed_before_next_snap,
382 		    &zbm->zbm_uncompressed_freed_before_next_snap);
383 		dsl_dataset_rele(nextds, FTAG);
384 	}
385 }
386 
387 /*
388  * Add dsl_bookmark_node_t `dbn` to the given dataset and increment appropriate
389  * SPA feature counters.
390  */
391 void
392 dsl_bookmark_node_add(dsl_dataset_t *hds, dsl_bookmark_node_t *dbn,
393     dmu_tx_t *tx)
394 {
395 	dsl_pool_t *dp = dmu_tx_pool(tx);
396 	objset_t *mos = dp->dp_meta_objset;
397 
398 	if (hds->ds_bookmarks_obj == 0) {
399 		hds->ds_bookmarks_obj = zap_create_norm(mos,
400 		    U8_TEXTPREP_TOUPPER, DMU_OTN_ZAP_METADATA, DMU_OT_NONE, 0,
401 		    tx);
402 		spa_feature_incr(dp->dp_spa, SPA_FEATURE_BOOKMARKS, tx);
403 
404 		dsl_dataset_zapify(hds, tx);
405 		VERIFY0(zap_add(mos, hds->ds_object,
406 		    DS_FIELD_BOOKMARK_NAMES,
407 		    sizeof (hds->ds_bookmarks_obj), 1,
408 		    &hds->ds_bookmarks_obj, tx));
409 	}
410 
411 	avl_add(&hds->ds_bookmarks, dbn);
412 
413 	/*
414 	 * To maintain backwards compatibility with software that doesn't
415 	 * understand SPA_FEATURE_BOOKMARK_V2, we need to use the smallest
416 	 * possible bookmark size.
417 	 */
418 	uint64_t bookmark_phys_size = BOOKMARK_PHYS_SIZE_V1;
419 	if (spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_BOOKMARK_V2) &&
420 	    (dbn->dbn_phys.zbm_ivset_guid != 0 || dbn->dbn_phys.zbm_flags &
421 	    ZBM_FLAG_HAS_FBN || dbn->dbn_phys.zbm_redaction_obj != 0)) {
422 		bookmark_phys_size = BOOKMARK_PHYS_SIZE_V2;
423 		spa_feature_incr(dp->dp_spa, SPA_FEATURE_BOOKMARK_V2, tx);
424 	}
425 
426 	zfs_bookmark_phys_t zero_phys = { 0 };
427 	ASSERT0(memcmp(((char *)&dbn->dbn_phys) + bookmark_phys_size,
428 	    &zero_phys, sizeof (zfs_bookmark_phys_t) - bookmark_phys_size));
429 
430 	VERIFY0(zap_add(mos, hds->ds_bookmarks_obj, dbn->dbn_name,
431 	    sizeof (uint64_t), bookmark_phys_size / sizeof (uint64_t),
432 	    &dbn->dbn_phys, tx));
433 }
434 
435 /*
436  * If redaction_list is non-null, we create a redacted bookmark and redaction
437  * list, and store the object number of the redaction list in redact_obj.
438  */
439 static void
440 dsl_bookmark_create_sync_impl_snap(const char *bookmark, const char *snapshot,
441     dmu_tx_t *tx, uint64_t num_redact_snaps, uint64_t *redact_snaps,
442     const void *tag, redaction_list_t **redaction_list)
443 {
444 	dsl_pool_t *dp = dmu_tx_pool(tx);
445 	objset_t *mos = dp->dp_meta_objset;
446 	dsl_dataset_t *snapds, *bmark_fs;
447 	char *shortname;
448 	boolean_t bookmark_redacted;
449 	uint64_t *dsredactsnaps;
450 	uint64_t dsnumsnaps;
451 
452 	VERIFY0(dsl_dataset_hold(dp, snapshot, FTAG, &snapds));
453 	VERIFY0(dsl_bookmark_hold_ds(dp, bookmark, &bmark_fs, FTAG,
454 	    &shortname));
455 
456 	dsl_bookmark_node_t *dbn = dsl_bookmark_node_alloc(shortname);
457 	dsl_bookmark_set_phys(&dbn->dbn_phys, snapds);
458 
459 	bookmark_redacted = dsl_dataset_get_uint64_array_feature(snapds,
460 	    SPA_FEATURE_REDACTED_DATASETS, &dsnumsnaps, &dsredactsnaps);
461 	if (redaction_list != NULL || bookmark_redacted) {
462 		redaction_list_t *local_rl;
463 		if (bookmark_redacted) {
464 			redact_snaps = dsredactsnaps;
465 			num_redact_snaps = dsnumsnaps;
466 		}
467 		dbn->dbn_phys.zbm_redaction_obj = dmu_object_alloc(mos,
468 		    DMU_OTN_UINT64_METADATA, SPA_OLD_MAXBLOCKSIZE,
469 		    DMU_OTN_UINT64_METADATA, sizeof (redaction_list_phys_t) +
470 		    num_redact_snaps * sizeof (uint64_t), tx);
471 		spa_feature_incr(dp->dp_spa,
472 		    SPA_FEATURE_REDACTION_BOOKMARKS, tx);
473 
474 		VERIFY0(dsl_redaction_list_hold_obj(dp,
475 		    dbn->dbn_phys.zbm_redaction_obj, tag, &local_rl));
476 		dsl_redaction_list_long_hold(dp, local_rl, tag);
477 
478 		ASSERT3U((local_rl)->rl_dbuf->db_size, >=,
479 		    sizeof (redaction_list_phys_t) + num_redact_snaps *
480 		    sizeof (uint64_t));
481 		dmu_buf_will_dirty(local_rl->rl_dbuf, tx);
482 		memcpy(local_rl->rl_phys->rlp_snaps, redact_snaps,
483 		    sizeof (uint64_t) * num_redact_snaps);
484 		local_rl->rl_phys->rlp_num_snaps = num_redact_snaps;
485 		if (bookmark_redacted) {
486 			ASSERT3P(redaction_list, ==, NULL);
487 			local_rl->rl_phys->rlp_last_blkid = UINT64_MAX;
488 			local_rl->rl_phys->rlp_last_object = UINT64_MAX;
489 			dsl_redaction_list_long_rele(local_rl, tag);
490 			dsl_redaction_list_rele(local_rl, tag);
491 		} else {
492 			*redaction_list = local_rl;
493 		}
494 	}
495 
496 	if (dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN) {
497 		spa_feature_incr(dp->dp_spa,
498 		    SPA_FEATURE_BOOKMARK_WRITTEN, tx);
499 	}
500 
501 	dsl_bookmark_node_add(bmark_fs, dbn, tx);
502 
503 	spa_history_log_internal_ds(bmark_fs, "bookmark", tx,
504 	    "name=%s creation_txg=%llu target_snap=%llu redact_obj=%llu",
505 	    shortname, (longlong_t)dbn->dbn_phys.zbm_creation_txg,
506 	    (longlong_t)snapds->ds_object,
507 	    (longlong_t)dbn->dbn_phys.zbm_redaction_obj);
508 
509 	dsl_dataset_rele(bmark_fs, FTAG);
510 	dsl_dataset_rele(snapds, FTAG);
511 }
512 
513 
514 static void
515 dsl_bookmark_create_sync_impl_book(
516     const char *new_name, const char *source_name, dmu_tx_t *tx)
517 {
518 	dsl_pool_t *dp = dmu_tx_pool(tx);
519 	dsl_dataset_t *bmark_fs_source, *bmark_fs_new;
520 	char *source_shortname, *new_shortname;
521 	zfs_bookmark_phys_t source_phys;
522 
523 	VERIFY0(dsl_bookmark_hold_ds(dp, source_name, &bmark_fs_source, FTAG,
524 	    &source_shortname));
525 	VERIFY0(dsl_bookmark_hold_ds(dp, new_name, &bmark_fs_new, FTAG,
526 	    &new_shortname));
527 
528 	/*
529 	 * create a copy of the source bookmark by copying most of its members
530 	 *
531 	 * Caveat: bookmarking a redaction bookmark yields a normal bookmark
532 	 * -----------------------------------------------------------------
533 	 * Reasoning:
534 	 * - The zbm_redaction_obj would be referred to by both source and new
535 	 *   bookmark, but would be destroyed once either source or new is
536 	 *   destroyed, resulting in use-after-free of the referred object.
537 	 * - User expectation when issuing the `zfs bookmark` command is that
538 	 *   a normal bookmark of the source is created
539 	 *
540 	 * Design Alternatives For Full Redaction Bookmark Copying:
541 	 * - reference-count the redaction object => would require on-disk
542 	 *   format change for existing redaction objects
543 	 * - Copy the redaction object => cannot be done in syncing context
544 	 *   because the redaction object might be too large
545 	 */
546 
547 	VERIFY0(dsl_bookmark_lookup_impl(bmark_fs_source, source_shortname,
548 	    &source_phys));
549 	dsl_bookmark_node_t *new_dbn = dsl_bookmark_node_alloc(new_shortname);
550 
551 	memcpy(&new_dbn->dbn_phys, &source_phys, sizeof (source_phys));
552 	new_dbn->dbn_phys.zbm_redaction_obj = 0;
553 
554 	/* update feature counters */
555 	if (new_dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN) {
556 		spa_feature_incr(dp->dp_spa,
557 		    SPA_FEATURE_BOOKMARK_WRITTEN, tx);
558 	}
559 	/* no need for redaction bookmark counter; nulled zbm_redaction_obj */
560 	/* dsl_bookmark_node_add bumps bookmarks and v2-bookmarks counter */
561 
562 	/*
563 	 * write new bookmark
564 	 *
565 	 * Note that dsl_bookmark_lookup_impl guarantees that, if source is a
566 	 * v1 bookmark, the v2-only fields are zeroed.
567 	 * And dsl_bookmark_node_add writes back a v1-sized bookmark if
568 	 * v2 bookmarks are disabled and/or v2-only fields are zeroed.
569 	 * => bookmark copying works on pre-bookmark-v2 pools
570 	 */
571 	dsl_bookmark_node_add(bmark_fs_new, new_dbn, tx);
572 
573 	spa_history_log_internal_ds(bmark_fs_source, "bookmark", tx,
574 	    "name=%s creation_txg=%llu source_guid=%llu",
575 	    new_shortname, (longlong_t)new_dbn->dbn_phys.zbm_creation_txg,
576 	    (longlong_t)source_phys.zbm_guid);
577 
578 	dsl_dataset_rele(bmark_fs_source, FTAG);
579 	dsl_dataset_rele(bmark_fs_new, FTAG);
580 }
581 
582 void
583 dsl_bookmark_create_sync(void *arg, dmu_tx_t *tx)
584 {
585 	dsl_bookmark_create_arg_t *dbca = arg;
586 
587 	ASSERT(spa_feature_is_enabled(dmu_tx_pool(tx)->dp_spa,
588 	    SPA_FEATURE_BOOKMARKS));
589 
590 	for (nvpair_t *pair = nvlist_next_nvpair(dbca->dbca_bmarks, NULL);
591 	    pair != NULL; pair = nvlist_next_nvpair(dbca->dbca_bmarks, pair)) {
592 
593 		char *new = nvpair_name(pair);
594 		char *source = fnvpair_value_string(pair);
595 
596 		if (strchr(source, '@') != NULL) {
597 			dsl_bookmark_create_sync_impl_snap(new, source, tx,
598 			    0, NULL, NULL, NULL);
599 		} else if (strchr(source, '#') != NULL) {
600 			dsl_bookmark_create_sync_impl_book(new, source, tx);
601 		} else {
602 			panic("unreachable code");
603 		}
604 
605 	}
606 }
607 
608 /*
609  * The bookmarks must all be in the same pool.
610  */
611 int
612 dsl_bookmark_create(nvlist_t *bmarks, nvlist_t *errors)
613 {
614 	nvpair_t *pair;
615 	dsl_bookmark_create_arg_t dbca;
616 
617 	pair = nvlist_next_nvpair(bmarks, NULL);
618 	if (pair == NULL)
619 		return (0);
620 
621 	dbca.dbca_bmarks = bmarks;
622 	dbca.dbca_errors = errors;
623 
624 	return (dsl_sync_task(nvpair_name(pair), dsl_bookmark_create_check,
625 	    dsl_bookmark_create_sync, &dbca,
626 	    fnvlist_num_pairs(bmarks), ZFS_SPACE_CHECK_NORMAL));
627 }
628 
629 static int
630 dsl_bookmark_create_redacted_check(void *arg, dmu_tx_t *tx)
631 {
632 	dsl_bookmark_create_redacted_arg_t *dbcra = arg;
633 	dsl_pool_t *dp = dmu_tx_pool(tx);
634 	int rv = 0;
635 
636 	if (!spa_feature_is_enabled(dp->dp_spa,
637 	    SPA_FEATURE_REDACTION_BOOKMARKS))
638 		return (SET_ERROR(ENOTSUP));
639 	/*
640 	 * If the list of redact snaps will not fit in the bonus buffer with
641 	 * the furthest reached object and offset, fail.
642 	 */
643 	if (dbcra->dbcra_numsnaps > (dmu_bonus_max() -
644 	    sizeof (redaction_list_phys_t)) / sizeof (uint64_t))
645 		return (SET_ERROR(E2BIG));
646 
647 	if (dsl_bookmark_create_nvl_validate_pair(
648 	    dbcra->dbcra_bmark, dbcra->dbcra_snap) != 0)
649 		return (SET_ERROR(EINVAL));
650 
651 	rv = dsl_bookmark_create_check_impl(dp,
652 	    dbcra->dbcra_bmark, dbcra->dbcra_snap);
653 	return (rv);
654 }
655 
656 static void
657 dsl_bookmark_create_redacted_sync(void *arg, dmu_tx_t *tx)
658 {
659 	dsl_bookmark_create_redacted_arg_t *dbcra = arg;
660 	dsl_bookmark_create_sync_impl_snap(dbcra->dbcra_bmark,
661 	    dbcra->dbcra_snap, tx, dbcra->dbcra_numsnaps, dbcra->dbcra_snaps,
662 	    dbcra->dbcra_tag, dbcra->dbcra_rl);
663 }
664 
665 int
666 dsl_bookmark_create_redacted(const char *bookmark, const char *snapshot,
667     uint64_t numsnaps, uint64_t *snapguids, const void *tag,
668     redaction_list_t **rl)
669 {
670 	dsl_bookmark_create_redacted_arg_t dbcra;
671 
672 	dbcra.dbcra_bmark = bookmark;
673 	dbcra.dbcra_snap = snapshot;
674 	dbcra.dbcra_rl = rl;
675 	dbcra.dbcra_numsnaps = numsnaps;
676 	dbcra.dbcra_snaps = snapguids;
677 	dbcra.dbcra_tag = tag;
678 
679 	return (dsl_sync_task(bookmark, dsl_bookmark_create_redacted_check,
680 	    dsl_bookmark_create_redacted_sync, &dbcra, 5,
681 	    ZFS_SPACE_CHECK_NORMAL));
682 }
683 
684 /*
685  * Retrieve the list of properties given in the 'props' nvlist for a bookmark.
686  * If 'props' is NULL, retrieves all properties.
687  */
688 static void
689 dsl_bookmark_fetch_props(dsl_pool_t *dp, zfs_bookmark_phys_t *bmark_phys,
690     nvlist_t *props, nvlist_t *out_props)
691 {
692 	ASSERT3P(dp, !=, NULL);
693 	ASSERT3P(bmark_phys, !=, NULL);
694 	ASSERT3P(out_props, !=, NULL);
695 	ASSERT(RRW_LOCK_HELD(&dp->dp_config_rwlock));
696 
697 	if (props == NULL || nvlist_exists(props,
698 	    zfs_prop_to_name(ZFS_PROP_GUID))) {
699 		dsl_prop_nvlist_add_uint64(out_props,
700 		    ZFS_PROP_GUID, bmark_phys->zbm_guid);
701 	}
702 	if (props == NULL || nvlist_exists(props,
703 	    zfs_prop_to_name(ZFS_PROP_CREATETXG))) {
704 		dsl_prop_nvlist_add_uint64(out_props,
705 		    ZFS_PROP_CREATETXG, bmark_phys->zbm_creation_txg);
706 	}
707 	if (props == NULL || nvlist_exists(props,
708 	    zfs_prop_to_name(ZFS_PROP_CREATION))) {
709 		dsl_prop_nvlist_add_uint64(out_props,
710 		    ZFS_PROP_CREATION, bmark_phys->zbm_creation_time);
711 	}
712 	if (props == NULL || nvlist_exists(props,
713 	    zfs_prop_to_name(ZFS_PROP_IVSET_GUID))) {
714 		dsl_prop_nvlist_add_uint64(out_props,
715 		    ZFS_PROP_IVSET_GUID, bmark_phys->zbm_ivset_guid);
716 	}
717 	if (bmark_phys->zbm_flags & ZBM_FLAG_HAS_FBN) {
718 		if (props == NULL || nvlist_exists(props,
719 		    zfs_prop_to_name(ZFS_PROP_REFERENCED))) {
720 			dsl_prop_nvlist_add_uint64(out_props,
721 			    ZFS_PROP_REFERENCED,
722 			    bmark_phys->zbm_referenced_bytes_refd);
723 		}
724 		if (props == NULL || nvlist_exists(props,
725 		    zfs_prop_to_name(ZFS_PROP_LOGICALREFERENCED))) {
726 			dsl_prop_nvlist_add_uint64(out_props,
727 			    ZFS_PROP_LOGICALREFERENCED,
728 			    bmark_phys->zbm_uncompressed_bytes_refd);
729 		}
730 		if (props == NULL || nvlist_exists(props,
731 		    zfs_prop_to_name(ZFS_PROP_REFRATIO))) {
732 			uint64_t ratio =
733 			    bmark_phys->zbm_compressed_bytes_refd == 0 ? 100 :
734 			    bmark_phys->zbm_uncompressed_bytes_refd * 100 /
735 			    bmark_phys->zbm_compressed_bytes_refd;
736 			dsl_prop_nvlist_add_uint64(out_props,
737 			    ZFS_PROP_REFRATIO, ratio);
738 		}
739 	}
740 
741 	if ((props == NULL || nvlist_exists(props, "redact_snaps") ||
742 	    nvlist_exists(props, "redact_complete")) &&
743 	    bmark_phys->zbm_redaction_obj != 0) {
744 		redaction_list_t *rl;
745 		int err = dsl_redaction_list_hold_obj(dp,
746 		    bmark_phys->zbm_redaction_obj, FTAG, &rl);
747 		if (err == 0) {
748 			if (nvlist_exists(props, "redact_snaps")) {
749 				nvlist_t *nvl;
750 				nvl = fnvlist_alloc();
751 				fnvlist_add_uint64_array(nvl, ZPROP_VALUE,
752 				    rl->rl_phys->rlp_snaps,
753 				    rl->rl_phys->rlp_num_snaps);
754 				fnvlist_add_nvlist(out_props, "redact_snaps",
755 				    nvl);
756 				nvlist_free(nvl);
757 			}
758 			if (nvlist_exists(props, "redact_complete")) {
759 				nvlist_t *nvl;
760 				nvl = fnvlist_alloc();
761 				fnvlist_add_boolean_value(nvl, ZPROP_VALUE,
762 				    rl->rl_phys->rlp_last_blkid == UINT64_MAX &&
763 				    rl->rl_phys->rlp_last_object == UINT64_MAX);
764 				fnvlist_add_nvlist(out_props, "redact_complete",
765 				    nvl);
766 				nvlist_free(nvl);
767 			}
768 			dsl_redaction_list_rele(rl, FTAG);
769 		}
770 	}
771 }
772 
773 int
774 dsl_get_bookmarks_impl(dsl_dataset_t *ds, nvlist_t *props, nvlist_t *outnvl)
775 {
776 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
777 
778 	ASSERT(dsl_pool_config_held(dp));
779 
780 	if (dsl_dataset_is_snapshot(ds))
781 		return (SET_ERROR(EINVAL));
782 
783 	for (dsl_bookmark_node_t *dbn = avl_first(&ds->ds_bookmarks);
784 	    dbn != NULL; dbn = AVL_NEXT(&ds->ds_bookmarks, dbn)) {
785 		nvlist_t *out_props = fnvlist_alloc();
786 
787 		dsl_bookmark_fetch_props(dp, &dbn->dbn_phys, props, out_props);
788 
789 		fnvlist_add_nvlist(outnvl, dbn->dbn_name, out_props);
790 		fnvlist_free(out_props);
791 	}
792 	return (0);
793 }
794 
795 /*
796  * Comparison func for ds_bookmarks AVL tree.  We sort the bookmarks by
797  * their TXG, then by their FBN-ness.  The "FBN-ness" component ensures
798  * that all bookmarks at the same TXG that HAS_FBN are adjacent, which
799  * dsl_bookmark_destroy_sync_impl() depends on.  Note that there may be
800  * multiple bookmarks at the same TXG (with the same FBN-ness).  In this
801  * case we differentiate them by an arbitrary metric (in this case,
802  * their names).
803  */
804 static int
805 dsl_bookmark_compare(const void *l, const void *r)
806 {
807 	const dsl_bookmark_node_t *ldbn = l;
808 	const dsl_bookmark_node_t *rdbn = r;
809 
810 	int64_t cmp = TREE_CMP(ldbn->dbn_phys.zbm_creation_txg,
811 	    rdbn->dbn_phys.zbm_creation_txg);
812 	if (likely(cmp))
813 		return (cmp);
814 	cmp = TREE_CMP((ldbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN),
815 	    (rdbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN));
816 	if (likely(cmp))
817 		return (cmp);
818 	cmp = strcmp(ldbn->dbn_name, rdbn->dbn_name);
819 	return (TREE_ISIGN(cmp));
820 }
821 
822 /*
823  * Cache this (head) dataset's bookmarks in the ds_bookmarks AVL tree.
824  */
825 int
826 dsl_bookmark_init_ds(dsl_dataset_t *ds)
827 {
828 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
829 	objset_t *mos = dp->dp_meta_objset;
830 
831 	ASSERT(!ds->ds_is_snapshot);
832 
833 	avl_create(&ds->ds_bookmarks, dsl_bookmark_compare,
834 	    sizeof (dsl_bookmark_node_t),
835 	    offsetof(dsl_bookmark_node_t, dbn_node));
836 
837 	if (!dsl_dataset_is_zapified(ds))
838 		return (0);
839 
840 	int zaperr = zap_lookup(mos, ds->ds_object, DS_FIELD_BOOKMARK_NAMES,
841 	    sizeof (ds->ds_bookmarks_obj), 1, &ds->ds_bookmarks_obj);
842 	if (zaperr == ENOENT)
843 		return (0);
844 	if (zaperr != 0)
845 		return (zaperr);
846 
847 	if (ds->ds_bookmarks_obj == 0)
848 		return (0);
849 
850 	int err = 0;
851 	zap_cursor_t zc;
852 	zap_attribute_t attr;
853 
854 	for (zap_cursor_init(&zc, mos, ds->ds_bookmarks_obj);
855 	    (err = zap_cursor_retrieve(&zc, &attr)) == 0;
856 	    zap_cursor_advance(&zc)) {
857 		dsl_bookmark_node_t *dbn =
858 		    dsl_bookmark_node_alloc(attr.za_name);
859 
860 		err = dsl_bookmark_lookup_impl(ds,
861 		    dbn->dbn_name, &dbn->dbn_phys);
862 		ASSERT3U(err, !=, ENOENT);
863 		if (err != 0) {
864 			kmem_free(dbn, sizeof (*dbn));
865 			break;
866 		}
867 		avl_add(&ds->ds_bookmarks, dbn);
868 	}
869 	zap_cursor_fini(&zc);
870 	if (err == ENOENT)
871 		err = 0;
872 	return (err);
873 }
874 
875 void
876 dsl_bookmark_fini_ds(dsl_dataset_t *ds)
877 {
878 	void *cookie = NULL;
879 	dsl_bookmark_node_t *dbn;
880 
881 	if (ds->ds_is_snapshot)
882 		return;
883 
884 	while ((dbn = avl_destroy_nodes(&ds->ds_bookmarks, &cookie)) != NULL) {
885 		spa_strfree(dbn->dbn_name);
886 		mutex_destroy(&dbn->dbn_lock);
887 		kmem_free(dbn, sizeof (*dbn));
888 	}
889 	avl_destroy(&ds->ds_bookmarks);
890 }
891 
892 /*
893  * Retrieve the bookmarks that exist in the specified dataset, and the
894  * requested properties of each bookmark.
895  *
896  * The "props" nvlist specifies which properties are requested.
897  * See lzc_get_bookmarks() for the list of valid properties.
898  */
899 int
900 dsl_get_bookmarks(const char *dsname, nvlist_t *props, nvlist_t *outnvl)
901 {
902 	dsl_pool_t *dp;
903 	dsl_dataset_t *ds;
904 	int err;
905 
906 	err = dsl_pool_hold(dsname, FTAG, &dp);
907 	if (err != 0)
908 		return (err);
909 	err = dsl_dataset_hold(dp, dsname, FTAG, &ds);
910 	if (err != 0) {
911 		dsl_pool_rele(dp, FTAG);
912 		return (err);
913 	}
914 
915 	err = dsl_get_bookmarks_impl(ds, props, outnvl);
916 
917 	dsl_dataset_rele(ds, FTAG);
918 	dsl_pool_rele(dp, FTAG);
919 	return (err);
920 }
921 
922 /*
923  * Retrieve all properties for a single bookmark in the given dataset.
924  */
925 int
926 dsl_get_bookmark_props(const char *dsname, const char *bmname, nvlist_t *props)
927 {
928 	dsl_pool_t *dp;
929 	dsl_dataset_t *ds;
930 	zfs_bookmark_phys_t bmark_phys = { 0 };
931 	int err;
932 
933 	err = dsl_pool_hold(dsname, FTAG, &dp);
934 	if (err != 0)
935 		return (err);
936 	err = dsl_dataset_hold(dp, dsname, FTAG, &ds);
937 	if (err != 0) {
938 		dsl_pool_rele(dp, FTAG);
939 		return (err);
940 	}
941 
942 	err = dsl_bookmark_lookup_impl(ds, bmname, &bmark_phys);
943 	if (err != 0)
944 		goto out;
945 
946 	dsl_bookmark_fetch_props(dp, &bmark_phys, NULL, props);
947 out:
948 	dsl_dataset_rele(ds, FTAG);
949 	dsl_pool_rele(dp, FTAG);
950 	return (err);
951 }
952 
953 typedef struct dsl_bookmark_destroy_arg {
954 	nvlist_t *dbda_bmarks;
955 	nvlist_t *dbda_success;
956 	nvlist_t *dbda_errors;
957 } dsl_bookmark_destroy_arg_t;
958 
959 static void
960 dsl_bookmark_destroy_sync_impl(dsl_dataset_t *ds, const char *name,
961     dmu_tx_t *tx)
962 {
963 	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;
964 	uint64_t bmark_zapobj = ds->ds_bookmarks_obj;
965 	matchtype_t mt = 0;
966 	uint64_t int_size, num_ints;
967 	/*
968 	 * 'search' must be zeroed so that dbn_flags (which is used in
969 	 * dsl_bookmark_compare()) will be zeroed even if the on-disk
970 	 * (in ZAP) bookmark is shorter than offsetof(dbn_flags).
971 	 */
972 	dsl_bookmark_node_t search = { 0 };
973 	char realname[ZFS_MAX_DATASET_NAME_LEN];
974 
975 	/*
976 	 * Find the real name of this bookmark, which may be different
977 	 * from the given name if the dataset is case-insensitive.  Then
978 	 * use the real name to find the node in the ds_bookmarks AVL tree.
979 	 */
980 
981 	if (dsl_dataset_phys(ds)->ds_flags & DS_FLAG_CI_DATASET)
982 		mt = MT_NORMALIZE;
983 
984 	VERIFY0(zap_length(mos, bmark_zapobj, name, &int_size, &num_ints));
985 
986 	ASSERT3U(int_size, ==, sizeof (uint64_t));
987 
988 	if (num_ints * int_size > BOOKMARK_PHYS_SIZE_V1) {
989 		spa_feature_decr(dmu_objset_spa(mos),
990 		    SPA_FEATURE_BOOKMARK_V2, tx);
991 	}
992 	VERIFY0(zap_lookup_norm(mos, bmark_zapobj, name, sizeof (uint64_t),
993 	    num_ints, &search.dbn_phys, mt, realname, sizeof (realname), NULL));
994 
995 	search.dbn_name = realname;
996 	dsl_bookmark_node_t *dbn = avl_find(&ds->ds_bookmarks, &search, NULL);
997 	ASSERT(dbn != NULL);
998 
999 	if (dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN) {
1000 		/*
1001 		 * If this bookmark HAS_FBN, and it is before the most
1002 		 * recent snapshot, then its TXG is a key in the head's
1003 		 * deadlist (and all clones' heads' deadlists).  If this is
1004 		 * the last thing keeping the key (i.e. there are no more
1005 		 * bookmarks with HAS_FBN at this TXG, and there is no
1006 		 * snapshot at this TXG), then remove the key.
1007 		 *
1008 		 * Note that this algorithm depends on ds_bookmarks being
1009 		 * sorted such that all bookmarks at the same TXG with
1010 		 * HAS_FBN are adjacent (with no non-HAS_FBN bookmarks
1011 		 * at the same TXG in between them).  If this were not
1012 		 * the case, we would need to examine *all* bookmarks
1013 		 * at this TXG, rather than just the adjacent ones.
1014 		 */
1015 
1016 		dsl_bookmark_node_t *dbn_prev =
1017 		    AVL_PREV(&ds->ds_bookmarks, dbn);
1018 		dsl_bookmark_node_t *dbn_next =
1019 		    AVL_NEXT(&ds->ds_bookmarks, dbn);
1020 
1021 		boolean_t more_bookmarks_at_this_txg =
1022 		    (dbn_prev != NULL && dbn_prev->dbn_phys.zbm_creation_txg ==
1023 		    dbn->dbn_phys.zbm_creation_txg &&
1024 		    (dbn_prev->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN)) ||
1025 		    (dbn_next != NULL && dbn_next->dbn_phys.zbm_creation_txg ==
1026 		    dbn->dbn_phys.zbm_creation_txg &&
1027 		    (dbn_next->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN));
1028 
1029 		if (!(dbn->dbn_phys.zbm_flags & ZBM_FLAG_SNAPSHOT_EXISTS) &&
1030 		    !more_bookmarks_at_this_txg &&
1031 		    dbn->dbn_phys.zbm_creation_txg <
1032 		    dsl_dataset_phys(ds)->ds_prev_snap_txg) {
1033 			dsl_dir_remove_clones_key(ds->ds_dir,
1034 			    dbn->dbn_phys.zbm_creation_txg, tx);
1035 			dsl_deadlist_remove_key(&ds->ds_deadlist,
1036 			    dbn->dbn_phys.zbm_creation_txg, tx);
1037 		}
1038 
1039 		spa_feature_decr(dmu_objset_spa(mos),
1040 		    SPA_FEATURE_BOOKMARK_WRITTEN, tx);
1041 	}
1042 
1043 	if (dbn->dbn_phys.zbm_redaction_obj != 0) {
1044 		VERIFY0(dmu_object_free(mos,
1045 		    dbn->dbn_phys.zbm_redaction_obj, tx));
1046 		spa_feature_decr(dmu_objset_spa(mos),
1047 		    SPA_FEATURE_REDACTION_BOOKMARKS, tx);
1048 	}
1049 
1050 	avl_remove(&ds->ds_bookmarks, dbn);
1051 	spa_strfree(dbn->dbn_name);
1052 	mutex_destroy(&dbn->dbn_lock);
1053 	kmem_free(dbn, sizeof (*dbn));
1054 
1055 	VERIFY0(zap_remove_norm(mos, bmark_zapobj, name, mt, tx));
1056 }
1057 
1058 static int
1059 dsl_bookmark_destroy_check(void *arg, dmu_tx_t *tx)
1060 {
1061 	dsl_bookmark_destroy_arg_t *dbda = arg;
1062 	dsl_pool_t *dp = dmu_tx_pool(tx);
1063 	int rv = 0;
1064 
1065 	ASSERT(nvlist_empty(dbda->dbda_success));
1066 	ASSERT(nvlist_empty(dbda->dbda_errors));
1067 
1068 	if (!spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_BOOKMARKS))
1069 		return (0);
1070 
1071 	for (nvpair_t *pair = nvlist_next_nvpair(dbda->dbda_bmarks, NULL);
1072 	    pair != NULL; pair = nvlist_next_nvpair(dbda->dbda_bmarks, pair)) {
1073 		const char *fullname = nvpair_name(pair);
1074 		dsl_dataset_t *ds;
1075 		zfs_bookmark_phys_t bm;
1076 		int error;
1077 		char *shortname;
1078 
1079 		error = dsl_bookmark_hold_ds(dp, fullname, &ds,
1080 		    FTAG, &shortname);
1081 		if (error == ENOENT) {
1082 			/* ignore it; the bookmark is "already destroyed" */
1083 			continue;
1084 		}
1085 		if (error == 0) {
1086 			error = dsl_bookmark_lookup_impl(ds, shortname, &bm);
1087 			dsl_dataset_rele(ds, FTAG);
1088 			if (error == ESRCH) {
1089 				/*
1090 				 * ignore it; the bookmark is
1091 				 * "already destroyed"
1092 				 */
1093 				continue;
1094 			}
1095 			if (error == 0 && bm.zbm_redaction_obj != 0) {
1096 				redaction_list_t *rl = NULL;
1097 				error = dsl_redaction_list_hold_obj(tx->tx_pool,
1098 				    bm.zbm_redaction_obj, FTAG, &rl);
1099 				if (error == ENOENT) {
1100 					error = 0;
1101 				} else if (error == 0 &&
1102 				    dsl_redaction_list_long_held(rl)) {
1103 					error = SET_ERROR(EBUSY);
1104 				}
1105 				if (rl != NULL) {
1106 					dsl_redaction_list_rele(rl, FTAG);
1107 				}
1108 			}
1109 		}
1110 		if (error == 0) {
1111 			if (dmu_tx_is_syncing(tx)) {
1112 				fnvlist_add_boolean(dbda->dbda_success,
1113 				    fullname);
1114 			}
1115 		} else {
1116 			fnvlist_add_int32(dbda->dbda_errors, fullname, error);
1117 			rv = error;
1118 		}
1119 	}
1120 	return (rv);
1121 }
1122 
1123 static void
1124 dsl_bookmark_destroy_sync(void *arg, dmu_tx_t *tx)
1125 {
1126 	dsl_bookmark_destroy_arg_t *dbda = arg;
1127 	dsl_pool_t *dp = dmu_tx_pool(tx);
1128 	objset_t *mos = dp->dp_meta_objset;
1129 
1130 	for (nvpair_t *pair = nvlist_next_nvpair(dbda->dbda_success, NULL);
1131 	    pair != NULL; pair = nvlist_next_nvpair(dbda->dbda_success, pair)) {
1132 		dsl_dataset_t *ds;
1133 		char *shortname;
1134 		uint64_t zap_cnt;
1135 
1136 		VERIFY0(dsl_bookmark_hold_ds(dp, nvpair_name(pair),
1137 		    &ds, FTAG, &shortname));
1138 		dsl_bookmark_destroy_sync_impl(ds, shortname, tx);
1139 
1140 		/*
1141 		 * If all of this dataset's bookmarks have been destroyed,
1142 		 * free the zap object and decrement the feature's use count.
1143 		 */
1144 		VERIFY0(zap_count(mos, ds->ds_bookmarks_obj, &zap_cnt));
1145 		if (zap_cnt == 0) {
1146 			dmu_buf_will_dirty(ds->ds_dbuf, tx);
1147 			VERIFY0(zap_destroy(mos, ds->ds_bookmarks_obj, tx));
1148 			ds->ds_bookmarks_obj = 0;
1149 			spa_feature_decr(dp->dp_spa, SPA_FEATURE_BOOKMARKS, tx);
1150 			VERIFY0(zap_remove(mos, ds->ds_object,
1151 			    DS_FIELD_BOOKMARK_NAMES, tx));
1152 		}
1153 
1154 		spa_history_log_internal_ds(ds, "remove bookmark", tx,
1155 		    "name=%s", shortname);
1156 
1157 		dsl_dataset_rele(ds, FTAG);
1158 	}
1159 }
1160 
1161 /*
1162  * The bookmarks must all be in the same pool.
1163  */
1164 int
1165 dsl_bookmark_destroy(nvlist_t *bmarks, nvlist_t *errors)
1166 {
1167 	int rv;
1168 	dsl_bookmark_destroy_arg_t dbda;
1169 	nvpair_t *pair = nvlist_next_nvpair(bmarks, NULL);
1170 	if (pair == NULL)
1171 		return (0);
1172 
1173 	dbda.dbda_bmarks = bmarks;
1174 	dbda.dbda_errors = errors;
1175 	dbda.dbda_success = fnvlist_alloc();
1176 
1177 	rv = dsl_sync_task(nvpair_name(pair), dsl_bookmark_destroy_check,
1178 	    dsl_bookmark_destroy_sync, &dbda, fnvlist_num_pairs(bmarks),
1179 	    ZFS_SPACE_CHECK_RESERVED);
1180 	fnvlist_free(dbda.dbda_success);
1181 	return (rv);
1182 }
1183 
1184 /* Return B_TRUE if there are any long holds on this dataset. */
1185 boolean_t
1186 dsl_redaction_list_long_held(redaction_list_t *rl)
1187 {
1188 	return (!zfs_refcount_is_zero(&rl->rl_longholds));
1189 }
1190 
1191 void
1192 dsl_redaction_list_long_hold(dsl_pool_t *dp, redaction_list_t *rl,
1193     const void *tag)
1194 {
1195 	ASSERT(dsl_pool_config_held(dp));
1196 	(void) zfs_refcount_add(&rl->rl_longholds, tag);
1197 }
1198 
1199 void
1200 dsl_redaction_list_long_rele(redaction_list_t *rl, const void *tag)
1201 {
1202 	(void) zfs_refcount_remove(&rl->rl_longholds, tag);
1203 }
1204 
1205 static void
1206 redaction_list_evict_sync(void *rlu)
1207 {
1208 	redaction_list_t *rl = rlu;
1209 	zfs_refcount_destroy(&rl->rl_longholds);
1210 
1211 	kmem_free(rl, sizeof (redaction_list_t));
1212 }
1213 
1214 void
1215 dsl_redaction_list_rele(redaction_list_t *rl, const void *tag)
1216 {
1217 	dmu_buf_rele(rl->rl_dbuf, tag);
1218 }
1219 
1220 int
1221 dsl_redaction_list_hold_obj(dsl_pool_t *dp, uint64_t rlobj, const void *tag,
1222     redaction_list_t **rlp)
1223 {
1224 	objset_t *mos = dp->dp_meta_objset;
1225 	dmu_buf_t *dbuf;
1226 	redaction_list_t *rl;
1227 	int err;
1228 
1229 	ASSERT(dsl_pool_config_held(dp));
1230 
1231 	err = dmu_bonus_hold(mos, rlobj, tag, &dbuf);
1232 	if (err != 0)
1233 		return (err);
1234 
1235 	rl = dmu_buf_get_user(dbuf);
1236 	if (rl == NULL) {
1237 		redaction_list_t *winner = NULL;
1238 
1239 		rl = kmem_zalloc(sizeof (redaction_list_t), KM_SLEEP);
1240 		rl->rl_dbuf = dbuf;
1241 		rl->rl_object = rlobj;
1242 		rl->rl_phys = dbuf->db_data;
1243 		rl->rl_mos = dp->dp_meta_objset;
1244 		zfs_refcount_create(&rl->rl_longholds);
1245 		dmu_buf_init_user(&rl->rl_dbu, redaction_list_evict_sync, NULL,
1246 		    &rl->rl_dbuf);
1247 		if ((winner = dmu_buf_set_user_ie(dbuf, &rl->rl_dbu)) != NULL) {
1248 			kmem_free(rl, sizeof (*rl));
1249 			rl = winner;
1250 		}
1251 	}
1252 	*rlp = rl;
1253 	return (0);
1254 }
1255 
1256 /*
1257  * Snapshot ds is being destroyed.
1258  *
1259  * Adjust the "freed_before_next" of any bookmarks between this snap
1260  * and the previous snapshot, because their "next snapshot" is changing.
1261  *
1262  * If there are any bookmarks with HAS_FBN at this snapshot, remove
1263  * their HAS_SNAP flag (note: there can be at most one snapshot of
1264  * each filesystem at a given txg), and return B_TRUE.  In this case
1265  * the caller can not remove the key in the deadlist at this TXG, because
1266  * the HAS_FBN bookmarks require the key be there.
1267  *
1268  * Returns B_FALSE if there are no bookmarks with HAS_FBN at this
1269  * snapshot's TXG.  In this case the caller can remove the key in the
1270  * deadlist at this TXG.
1271  */
1272 boolean_t
1273 dsl_bookmark_ds_destroyed(dsl_dataset_t *ds, dmu_tx_t *tx)
1274 {
1275 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1276 
1277 	dsl_dataset_t *head, *next;
1278 	VERIFY0(dsl_dataset_hold_obj(dp,
1279 	    dsl_dir_phys(ds->ds_dir)->dd_head_dataset_obj, FTAG, &head));
1280 	VERIFY0(dsl_dataset_hold_obj(dp,
1281 	    dsl_dataset_phys(ds)->ds_next_snap_obj, FTAG, &next));
1282 
1283 	/*
1284 	 * Find the first bookmark that HAS_FBN at or after the
1285 	 * previous snapshot.
1286 	 */
1287 	dsl_bookmark_node_t search = { 0 };
1288 	avl_index_t idx;
1289 	search.dbn_phys.zbm_creation_txg =
1290 	    dsl_dataset_phys(ds)->ds_prev_snap_txg;
1291 	search.dbn_phys.zbm_flags = ZBM_FLAG_HAS_FBN;
1292 	/*
1293 	 * The empty-string name can't be in the AVL, and it compares
1294 	 * before any entries with this TXG.
1295 	 */
1296 	search.dbn_name = (char *)"";
1297 	VERIFY3P(avl_find(&head->ds_bookmarks, &search, &idx), ==, NULL);
1298 	dsl_bookmark_node_t *dbn =
1299 	    avl_nearest(&head->ds_bookmarks, idx, AVL_AFTER);
1300 
1301 	/*
1302 	 * Iterate over all bookmarks that are at or after the previous
1303 	 * snapshot, and before this (being deleted) snapshot.  Adjust
1304 	 * their FBN based on their new next snapshot.
1305 	 */
1306 	for (; dbn != NULL && dbn->dbn_phys.zbm_creation_txg <
1307 	    dsl_dataset_phys(ds)->ds_creation_txg;
1308 	    dbn = AVL_NEXT(&head->ds_bookmarks, dbn)) {
1309 		if (!(dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN))
1310 			continue;
1311 		/*
1312 		 * Increase our FBN by the amount of space that was live
1313 		 * (referenced) at the time of this bookmark (i.e.
1314 		 * birth <= zbm_creation_txg), and killed between this
1315 		 * (being deleted) snapshot and the next snapshot (i.e.
1316 		 * on the next snapshot's deadlist).  (Space killed before
1317 		 * this are already on our FBN.)
1318 		 */
1319 		uint64_t referenced, compressed, uncompressed;
1320 		dsl_deadlist_space_range(&next->ds_deadlist,
1321 		    0, dbn->dbn_phys.zbm_creation_txg,
1322 		    &referenced, &compressed, &uncompressed);
1323 		dbn->dbn_phys.zbm_referenced_freed_before_next_snap +=
1324 		    referenced;
1325 		dbn->dbn_phys.zbm_compressed_freed_before_next_snap +=
1326 		    compressed;
1327 		dbn->dbn_phys.zbm_uncompressed_freed_before_next_snap +=
1328 		    uncompressed;
1329 		VERIFY0(zap_update(dp->dp_meta_objset, head->ds_bookmarks_obj,
1330 		    dbn->dbn_name, sizeof (uint64_t),
1331 		    sizeof (zfs_bookmark_phys_t) / sizeof (uint64_t),
1332 		    &dbn->dbn_phys, tx));
1333 	}
1334 	dsl_dataset_rele(next, FTAG);
1335 
1336 	/*
1337 	 * There may be several bookmarks at this txg (the TXG of the
1338 	 * snapshot being deleted).  We need to clear the SNAPSHOT_EXISTS
1339 	 * flag on all of them, and return TRUE if there is at least 1
1340 	 * bookmark here with HAS_FBN (thus preventing the deadlist
1341 	 * key from being removed).
1342 	 */
1343 	boolean_t rv = B_FALSE;
1344 	for (; dbn != NULL && dbn->dbn_phys.zbm_creation_txg ==
1345 	    dsl_dataset_phys(ds)->ds_creation_txg;
1346 	    dbn = AVL_NEXT(&head->ds_bookmarks, dbn)) {
1347 		if (!(dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN)) {
1348 			ASSERT(!(dbn->dbn_phys.zbm_flags &
1349 			    ZBM_FLAG_SNAPSHOT_EXISTS));
1350 			continue;
1351 		}
1352 		ASSERT(dbn->dbn_phys.zbm_flags & ZBM_FLAG_SNAPSHOT_EXISTS);
1353 		dbn->dbn_phys.zbm_flags &= ~ZBM_FLAG_SNAPSHOT_EXISTS;
1354 		VERIFY0(zap_update(dp->dp_meta_objset, head->ds_bookmarks_obj,
1355 		    dbn->dbn_name, sizeof (uint64_t),
1356 		    sizeof (zfs_bookmark_phys_t) / sizeof (uint64_t),
1357 		    &dbn->dbn_phys, tx));
1358 		rv = B_TRUE;
1359 	}
1360 	dsl_dataset_rele(head, FTAG);
1361 	return (rv);
1362 }
1363 
1364 /*
1365  * A snapshot is being created of this (head) dataset.
1366  *
1367  * We don't keep keys in the deadlist for the most recent snapshot, or any
1368  * bookmarks at or after it, because there can't be any blocks on the
1369  * deadlist in this range.  Now that the most recent snapshot is after
1370  * all bookmarks, we need to add these keys.  Note that the caller always
1371  * adds a key at the previous snapshot, so we only add keys for bookmarks
1372  * after that.
1373  */
1374 void
1375 dsl_bookmark_snapshotted(dsl_dataset_t *ds, dmu_tx_t *tx)
1376 {
1377 	uint64_t last_key_added = UINT64_MAX;
1378 	for (dsl_bookmark_node_t *dbn = avl_last(&ds->ds_bookmarks);
1379 	    dbn != NULL && dbn->dbn_phys.zbm_creation_txg >
1380 	    dsl_dataset_phys(ds)->ds_prev_snap_txg;
1381 	    dbn = AVL_PREV(&ds->ds_bookmarks, dbn)) {
1382 		uint64_t creation_txg = dbn->dbn_phys.zbm_creation_txg;
1383 		ASSERT3U(creation_txg, <=, last_key_added);
1384 		/*
1385 		 * Note, there may be multiple bookmarks at this TXG,
1386 		 * and we only want to add the key for this TXG once.
1387 		 * The ds_bookmarks AVL is sorted by TXG, so we will visit
1388 		 * these bookmarks in sequence.
1389 		 */
1390 		if ((dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN) &&
1391 		    creation_txg != last_key_added) {
1392 			dsl_deadlist_add_key(&ds->ds_deadlist,
1393 			    creation_txg, tx);
1394 			last_key_added = creation_txg;
1395 		}
1396 	}
1397 }
1398 
1399 /*
1400  * The next snapshot of the origin dataset has changed, due to
1401  * promote or clone swap.  If there are any bookmarks at this dataset,
1402  * we need to update their zbm_*_freed_before_next_snap to reflect this.
1403  * The head dataset has the relevant bookmarks in ds_bookmarks.
1404  */
1405 void
1406 dsl_bookmark_next_changed(dsl_dataset_t *head, dsl_dataset_t *origin,
1407     dmu_tx_t *tx)
1408 {
1409 	dsl_pool_t *dp = dmu_tx_pool(tx);
1410 
1411 	/*
1412 	 * Find the first bookmark that HAS_FBN at the origin snapshot.
1413 	 */
1414 	dsl_bookmark_node_t search = { 0 };
1415 	avl_index_t idx;
1416 	search.dbn_phys.zbm_creation_txg =
1417 	    dsl_dataset_phys(origin)->ds_creation_txg;
1418 	search.dbn_phys.zbm_flags = ZBM_FLAG_HAS_FBN;
1419 	/*
1420 	 * The empty-string name can't be in the AVL, and it compares
1421 	 * before any entries with this TXG.
1422 	 */
1423 	search.dbn_name = (char *)"";
1424 	VERIFY3P(avl_find(&head->ds_bookmarks, &search, &idx), ==, NULL);
1425 	dsl_bookmark_node_t *dbn =
1426 	    avl_nearest(&head->ds_bookmarks, idx, AVL_AFTER);
1427 
1428 	/*
1429 	 * Iterate over all bookmarks that are at the origin txg.
1430 	 * Adjust their FBN based on their new next snapshot.
1431 	 */
1432 	for (; dbn != NULL && dbn->dbn_phys.zbm_creation_txg ==
1433 	    dsl_dataset_phys(origin)->ds_creation_txg &&
1434 	    (dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN);
1435 	    dbn = AVL_NEXT(&head->ds_bookmarks, dbn)) {
1436 
1437 		/*
1438 		 * Bookmark is at the origin, therefore its
1439 		 * "next dataset" is changing, so we need
1440 		 * to reset its FBN by recomputing it in
1441 		 * dsl_bookmark_set_phys().
1442 		 */
1443 		ASSERT3U(dbn->dbn_phys.zbm_guid, ==,
1444 		    dsl_dataset_phys(origin)->ds_guid);
1445 		ASSERT3U(dbn->dbn_phys.zbm_referenced_bytes_refd, ==,
1446 		    dsl_dataset_phys(origin)->ds_referenced_bytes);
1447 		ASSERT(dbn->dbn_phys.zbm_flags &
1448 		    ZBM_FLAG_SNAPSHOT_EXISTS);
1449 		/*
1450 		 * Save and restore the zbm_redaction_obj, which
1451 		 * is zeroed by dsl_bookmark_set_phys().
1452 		 */
1453 		uint64_t redaction_obj =
1454 		    dbn->dbn_phys.zbm_redaction_obj;
1455 		dsl_bookmark_set_phys(&dbn->dbn_phys, origin);
1456 		dbn->dbn_phys.zbm_redaction_obj = redaction_obj;
1457 
1458 		VERIFY0(zap_update(dp->dp_meta_objset, head->ds_bookmarks_obj,
1459 		    dbn->dbn_name, sizeof (uint64_t),
1460 		    sizeof (zfs_bookmark_phys_t) / sizeof (uint64_t),
1461 		    &dbn->dbn_phys, tx));
1462 	}
1463 }
1464 
1465 /*
1466  * This block is no longer referenced by this (head) dataset.
1467  *
1468  * Adjust the FBN of any bookmarks that reference this block, whose "next"
1469  * is the head dataset.
1470  */
1471 void
1472 dsl_bookmark_block_killed(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx)
1473 {
1474 	(void) tx;
1475 
1476 	/*
1477 	 * Iterate over bookmarks whose "next" is the head dataset.
1478 	 */
1479 	for (dsl_bookmark_node_t *dbn = avl_last(&ds->ds_bookmarks);
1480 	    dbn != NULL && dbn->dbn_phys.zbm_creation_txg >=
1481 	    dsl_dataset_phys(ds)->ds_prev_snap_txg;
1482 	    dbn = AVL_PREV(&ds->ds_bookmarks, dbn)) {
1483 		/*
1484 		 * If the block was live (referenced) at the time of this
1485 		 * bookmark, add its space to the bookmark's FBN.
1486 		 */
1487 		if (bp->blk_birth <= dbn->dbn_phys.zbm_creation_txg &&
1488 		    (dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN)) {
1489 			mutex_enter(&dbn->dbn_lock);
1490 			dbn->dbn_phys.zbm_referenced_freed_before_next_snap +=
1491 			    bp_get_dsize_sync(dsl_dataset_get_spa(ds), bp);
1492 			dbn->dbn_phys.zbm_compressed_freed_before_next_snap +=
1493 			    BP_GET_PSIZE(bp);
1494 			dbn->dbn_phys.zbm_uncompressed_freed_before_next_snap +=
1495 			    BP_GET_UCSIZE(bp);
1496 			/*
1497 			 * Changing the ZAP object here would be too
1498 			 * expensive.  Also, we may be called from the zio
1499 			 * interrupt thread, which can't block on i/o.
1500 			 * Therefore, we mark this bookmark as dirty and
1501 			 * modify the ZAP once per txg, in
1502 			 * dsl_bookmark_sync_done().
1503 			 */
1504 			dbn->dbn_dirty = B_TRUE;
1505 			mutex_exit(&dbn->dbn_lock);
1506 		}
1507 	}
1508 }
1509 
1510 void
1511 dsl_bookmark_sync_done(dsl_dataset_t *ds, dmu_tx_t *tx)
1512 {
1513 	dsl_pool_t *dp = dmu_tx_pool(tx);
1514 
1515 	if (dsl_dataset_is_snapshot(ds))
1516 		return;
1517 
1518 	/*
1519 	 * We only dirty bookmarks that are at or after the most recent
1520 	 * snapshot.  We can't create snapshots between
1521 	 * dsl_bookmark_block_killed() and dsl_bookmark_sync_done(), so we
1522 	 * don't need to look at any bookmarks before ds_prev_snap_txg.
1523 	 */
1524 	for (dsl_bookmark_node_t *dbn = avl_last(&ds->ds_bookmarks);
1525 	    dbn != NULL && dbn->dbn_phys.zbm_creation_txg >=
1526 	    dsl_dataset_phys(ds)->ds_prev_snap_txg;
1527 	    dbn = AVL_PREV(&ds->ds_bookmarks, dbn)) {
1528 		if (dbn->dbn_dirty) {
1529 			/*
1530 			 * We only dirty nodes with HAS_FBN, therefore
1531 			 * we can always use the current bookmark struct size.
1532 			 */
1533 			ASSERT(dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN);
1534 			VERIFY0(zap_update(dp->dp_meta_objset,
1535 			    ds->ds_bookmarks_obj,
1536 			    dbn->dbn_name, sizeof (uint64_t),
1537 			    sizeof (zfs_bookmark_phys_t) / sizeof (uint64_t),
1538 			    &dbn->dbn_phys, tx));
1539 			dbn->dbn_dirty = B_FALSE;
1540 		}
1541 	}
1542 #ifdef ZFS_DEBUG
1543 	for (dsl_bookmark_node_t *dbn = avl_first(&ds->ds_bookmarks);
1544 	    dbn != NULL; dbn = AVL_NEXT(&ds->ds_bookmarks, dbn)) {
1545 		ASSERT(!dbn->dbn_dirty);
1546 	}
1547 #endif
1548 }
1549 
1550 /*
1551  * Return the TXG of the most recent bookmark (or 0 if there are no bookmarks).
1552  */
1553 uint64_t
1554 dsl_bookmark_latest_txg(dsl_dataset_t *ds)
1555 {
1556 	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
1557 	dsl_bookmark_node_t *dbn = avl_last(&ds->ds_bookmarks);
1558 	if (dbn == NULL)
1559 		return (0);
1560 	return (dbn->dbn_phys.zbm_creation_txg);
1561 }
1562 
1563 /*
1564  * Compare the redact_block_phys_t to the bookmark. If the last block in the
1565  * redact_block_phys_t is before the bookmark, return -1.  If the first block in
1566  * the redact_block_phys_t is after the bookmark, return 1.  Otherwise, the
1567  * bookmark is inside the range of the redact_block_phys_t, and we return 0.
1568  */
1569 static int
1570 redact_block_zb_compare(redact_block_phys_t *first,
1571     zbookmark_phys_t *second)
1572 {
1573 	/*
1574 	 * If the block_phys is for a previous object, or the last block in the
1575 	 * block_phys is strictly before the block in the bookmark, the
1576 	 * block_phys is earlier.
1577 	 */
1578 	if (first->rbp_object < second->zb_object ||
1579 	    (first->rbp_object == second->zb_object &&
1580 	    first->rbp_blkid + (redact_block_get_count(first) - 1) <
1581 	    second->zb_blkid)) {
1582 		return (-1);
1583 	}
1584 
1585 	/*
1586 	 * If the bookmark is for a previous object, or the block in the
1587 	 * bookmark is strictly before the first block in the block_phys, the
1588 	 * bookmark is earlier.
1589 	 */
1590 	if (first->rbp_object > second->zb_object ||
1591 	    (first->rbp_object == second->zb_object &&
1592 	    first->rbp_blkid > second->zb_blkid)) {
1593 		return (1);
1594 	}
1595 
1596 	return (0);
1597 }
1598 
1599 /*
1600  * Traverse the redaction list in the provided object, and call the callback for
1601  * each entry we find. Don't call the callback for any records before resume.
1602  */
1603 int
1604 dsl_redaction_list_traverse(redaction_list_t *rl, zbookmark_phys_t *resume,
1605     rl_traverse_callback_t cb, void *arg)
1606 {
1607 	objset_t *mos = rl->rl_mos;
1608 	int err = 0;
1609 
1610 	if (rl->rl_phys->rlp_last_object != UINT64_MAX ||
1611 	    rl->rl_phys->rlp_last_blkid != UINT64_MAX) {
1612 		/*
1613 		 * When we finish a send, we update the last object and offset
1614 		 * to UINT64_MAX.  If a send fails partway through, the last
1615 		 * object and offset will have some other value, indicating how
1616 		 * far the send got. The redaction list must be complete before
1617 		 * it can be traversed, so return EINVAL if the last object and
1618 		 * blkid are not set to UINT64_MAX.
1619 		 */
1620 		return (SET_ERROR(EINVAL));
1621 	}
1622 
1623 	/*
1624 	 * This allows us to skip the binary search and resume checking logic
1625 	 * below, if we're not resuming a redacted send.
1626 	 */
1627 	if (ZB_IS_ZERO(resume))
1628 		resume = NULL;
1629 
1630 	/*
1631 	 * Binary search for the point to resume from.
1632 	 */
1633 	uint64_t maxidx = rl->rl_phys->rlp_num_entries - 1;
1634 	uint64_t minidx = 0;
1635 	while (resume != NULL && maxidx > minidx) {
1636 		redact_block_phys_t rbp = { 0 };
1637 		ASSERT3U(maxidx, >, minidx);
1638 		uint64_t mididx = minidx + ((maxidx - minidx) / 2);
1639 		err = dmu_read(mos, rl->rl_object, mididx * sizeof (rbp),
1640 		    sizeof (rbp), &rbp, DMU_READ_NO_PREFETCH);
1641 		if (err != 0)
1642 			break;
1643 
1644 		int cmp = redact_block_zb_compare(&rbp, resume);
1645 
1646 		if (cmp == 0) {
1647 			minidx = mididx;
1648 			break;
1649 		} else if (cmp > 0) {
1650 			maxidx =
1651 			    (mididx == minidx ? minidx : mididx - 1);
1652 		} else {
1653 			minidx = mididx + 1;
1654 		}
1655 	}
1656 
1657 	unsigned int bufsize = SPA_OLD_MAXBLOCKSIZE;
1658 	redact_block_phys_t *buf = zio_data_buf_alloc(bufsize);
1659 
1660 	unsigned int entries_per_buf = bufsize / sizeof (redact_block_phys_t);
1661 	uint64_t start_block = minidx / entries_per_buf;
1662 	err = dmu_read(mos, rl->rl_object, start_block * bufsize, bufsize, buf,
1663 	    DMU_READ_PREFETCH);
1664 
1665 	for (uint64_t curidx = minidx;
1666 	    err == 0 && curidx < rl->rl_phys->rlp_num_entries;
1667 	    curidx++) {
1668 		/*
1669 		 * We read in the redaction list one block at a time.  Once we
1670 		 * finish with all the entries in a given block, we read in a
1671 		 * new one.  The predictive prefetcher will take care of any
1672 		 * prefetching, and this code shouldn't be the bottleneck, so we
1673 		 * don't need to do manual prefetching.
1674 		 */
1675 		if (curidx % entries_per_buf == 0) {
1676 			err = dmu_read(mos, rl->rl_object, curidx *
1677 			    sizeof (*buf), bufsize, buf,
1678 			    DMU_READ_PREFETCH);
1679 			if (err != 0)
1680 				break;
1681 		}
1682 		redact_block_phys_t *rb = &buf[curidx % entries_per_buf];
1683 		/*
1684 		 * If resume is non-null, we should either not send the data, or
1685 		 * null out resume so we don't have to keep doing these
1686 		 * comparisons.
1687 		 */
1688 		if (resume != NULL) {
1689 			/*
1690 			 * It is possible that after the binary search we got
1691 			 * a record before the resume point. There's two cases
1692 			 * where this can occur. If the record is the last
1693 			 * redaction record, and the resume point is after the
1694 			 * end of the redacted data, curidx will be the last
1695 			 * redaction record. In that case, the loop will end
1696 			 * after this iteration. The second case is if the
1697 			 * resume point is between two redaction records, the
1698 			 * binary search can return either the record before
1699 			 * or after the resume point. In that case, the next
1700 			 * iteration will be greater than the resume point.
1701 			 */
1702 			if (redact_block_zb_compare(rb, resume) < 0) {
1703 				ASSERT3U(curidx, ==, minidx);
1704 				continue;
1705 			} else {
1706 				/*
1707 				 * If the place to resume is in the middle of
1708 				 * the range described by this
1709 				 * redact_block_phys, then modify the
1710 				 * redact_block_phys in memory so we generate
1711 				 * the right records.
1712 				 */
1713 				if (resume->zb_object == rb->rbp_object &&
1714 				    resume->zb_blkid > rb->rbp_blkid) {
1715 					uint64_t diff = resume->zb_blkid -
1716 					    rb->rbp_blkid;
1717 					rb->rbp_blkid = resume->zb_blkid;
1718 					redact_block_set_count(rb,
1719 					    redact_block_get_count(rb) - diff);
1720 				}
1721 				resume = NULL;
1722 			}
1723 		}
1724 
1725 		if (cb(rb, arg) != 0) {
1726 			err = EINTR;
1727 			break;
1728 		}
1729 	}
1730 
1731 	zio_data_buf_free(buf, bufsize);
1732 	return (err);
1733 }
1734