1 /*
2 * CDDL HEADER START
3 *
4 * The contents of this file are subject to the terms of the
5 * Common Development and Distribution License (the "License").
6 * You may not use this file except in compliance with the License.
7 *
8 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9 * or https://opensource.org/licenses/CDDL-1.0.
10 * See the License for the specific language governing permissions
11 * and limitations under the License.
12 *
13 * When distributing Covered Code, include this CDDL HEADER in each
14 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15 * If applicable, add the following below this CDDL HEADER, with the
16 * fields enclosed by brackets "[]" replaced with your own identifying
17 * information: Portions Copyright [yyyy] [name of copyright owner]
18 *
19 * CDDL HEADER END
20 */
21 /*
22 * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
23 * Copyright 2011 Nexenta Systems, Inc. All rights reserved.
24 * Copyright (c) 2011, 2018 by Delphix. All rights reserved.
25 * Copyright (c) 2014, Joyent, Inc. All rights reserved.
26 * Copyright 2014 HybridCluster. All rights reserved.
27 * Copyright 2016 RackTop Systems.
28 * Copyright (c) 2016 Actifio, Inc. All rights reserved.
29 * Copyright (c) 2019, Klara Inc.
30 * Copyright (c) 2019, Allan Jude
31 */
32
33 #include <sys/dmu.h>
34 #include <sys/dmu_impl.h>
35 #include <sys/dmu_tx.h>
36 #include <sys/dbuf.h>
37 #include <sys/dnode.h>
38 #include <sys/zfs_context.h>
39 #include <sys/dmu_objset.h>
40 #include <sys/dmu_traverse.h>
41 #include <sys/dsl_dataset.h>
42 #include <sys/dsl_dir.h>
43 #include <sys/dsl_prop.h>
44 #include <sys/dsl_pool.h>
45 #include <sys/dsl_synctask.h>
46 #include <sys/spa_impl.h>
47 #include <sys/zfs_ioctl.h>
48 #include <sys/zap.h>
49 #include <sys/zio_checksum.h>
50 #include <sys/zfs_znode.h>
51 #include <zfs_fletcher.h>
52 #include <sys/avl.h>
53 #include <sys/ddt.h>
54 #include <sys/zfs_onexit.h>
55 #include <sys/dmu_send.h>
56 #include <sys/dmu_recv.h>
57 #include <sys/dsl_destroy.h>
58 #include <sys/blkptr.h>
59 #include <sys/dsl_bookmark.h>
60 #include <sys/zfeature.h>
61 #include <sys/bqueue.h>
62 #include <sys/zvol.h>
63 #include <sys/policy.h>
64 #include <sys/objlist.h>
65 #ifdef _KERNEL
66 #include <sys/zfs_vfsops.h>
67 #endif
68
69 /* Set this tunable to TRUE to replace corrupt data with 0x2f5baddb10c */
70 static int zfs_send_corrupt_data = B_FALSE;
71 /*
72 * This tunable controls the amount of data (measured in bytes) that will be
73 * prefetched by zfs send. If the main thread is blocking on reads that haven't
74 * completed, this variable might need to be increased. If instead the main
75 * thread is issuing new reads because the prefetches have fallen out of the
76 * cache, this may need to be decreased.
77 */
78 static uint_t zfs_send_queue_length = SPA_MAXBLOCKSIZE;
79 /*
80 * This tunable controls the length of the queues that zfs send worker threads
81 * use to communicate. If the send_main_thread is blocking on these queues,
82 * this variable may need to be increased. If there is a significant slowdown
83 * at the start of a send as these threads consume all the available IO
84 * resources, this variable may need to be decreased.
85 */
86 static uint_t zfs_send_no_prefetch_queue_length = 1024 * 1024;
87 /*
88 * These tunables control the fill fraction of the queues by zfs send. The fill
89 * fraction controls the frequency with which threads have to be cv_signaled.
90 * If a lot of cpu time is being spent on cv_signal, then these should be tuned
91 * down. If the queues empty before the signalled thread can catch up, then
92 * these should be tuned up.
93 */
94 static uint_t zfs_send_queue_ff = 20;
95 static uint_t zfs_send_no_prefetch_queue_ff = 20;
96
97 /*
98 * Use this to override the recordsize calculation for fast zfs send estimates.
99 */
100 static uint_t zfs_override_estimate_recordsize = 0;
101
102 /* Set this tunable to FALSE to disable setting of DRR_FLAG_FREERECORDS */
103 static const boolean_t zfs_send_set_freerecords_bit = B_TRUE;
104
105 /* Set this tunable to FALSE is disable sending unmodified spill blocks. */
106 static int zfs_send_unmodified_spill_blocks = B_TRUE;
107
108 static inline boolean_t
109 overflow_multiply(uint64_t a, uint64_t b, uint64_t *c)
110 {
111 uint64_t temp = a * b;
112 if (b != 0 && temp / b != a)
113 return (B_FALSE);
114 *c = temp;
115 return (B_TRUE);
116 }
117
118 struct send_thread_arg {
119 bqueue_t q;
120 objset_t *os; /* Objset to traverse */
121 uint64_t fromtxg; /* Traverse from this txg */
122 int flags; /* flags to pass to traverse_dataset */
123 int error_code;
124 boolean_t cancel;
125 zbookmark_phys_t resume;
126 uint64_t *num_blocks_visited;
127 };
128
129 struct redact_list_thread_arg {
130 boolean_t cancel;
131 bqueue_t q;
132 zbookmark_phys_t resume;
133 redaction_list_t *rl;
134 boolean_t mark_redact;
135 int error_code;
136 uint64_t *num_blocks_visited;
137 };
138
139 struct send_merge_thread_arg {
140 bqueue_t q;
141 objset_t *os;
142 struct redact_list_thread_arg *from_arg;
143 struct send_thread_arg *to_arg;
144 struct redact_list_thread_arg *redact_arg;
145 int error;
146 boolean_t cancel;
147 };
148
149 struct send_range {
150 boolean_t eos_marker; /* Marks the end of the stream */
151 uint64_t object;
152 uint64_t start_blkid;
153 uint64_t end_blkid;
154 bqueue_node_t ln;
155 enum type {DATA, HOLE, OBJECT, OBJECT_RANGE, REDACT,
156 PREVIOUSLY_REDACTED} type;
157 union {
158 struct srd {
159 dmu_object_type_t obj_type;
160 uint32_t datablksz; // logical size
161 uint32_t datasz; // payload size
162 blkptr_t bp;
163 arc_buf_t *abuf;
164 abd_t *abd;
165 kmutex_t lock;
166 kcondvar_t cv;
167 boolean_t io_outstanding;
168 boolean_t io_compressed;
169 int io_err;
170 } data;
171 struct srh {
172 uint32_t datablksz;
173 } hole;
174 struct sro {
175 /*
176 * This is a pointer because embedding it in the
177 * struct causes these structures to be massively larger
178 * for all range types; this makes the code much less
179 * memory efficient.
180 */
181 dnode_phys_t *dnp;
182 blkptr_t bp;
183 } object;
184 struct srr {
185 uint32_t datablksz;
186 } redact;
187 struct sror {
188 blkptr_t bp;
189 } object_range;
190 } sru;
191 };
192
193 /*
194 * The list of data whose inclusion in a send stream can be pending from
195 * one call to backup_cb to another. Multiple calls to dump_free(),
196 * dump_freeobjects(), and dump_redact() can be aggregated into a single
197 * DRR_FREE, DRR_FREEOBJECTS, or DRR_REDACT replay record.
198 */
199 typedef enum {
200 PENDING_NONE,
201 PENDING_FREE,
202 PENDING_FREEOBJECTS,
203 PENDING_REDACT
204 } dmu_pendop_t;
205
206 typedef struct dmu_send_cookie {
207 dmu_replay_record_t *dsc_drr;
208 dmu_send_outparams_t *dsc_dso;
209 offset_t *dsc_off;
210 objset_t *dsc_os;
211 zio_cksum_t dsc_zc;
212 uint64_t dsc_toguid;
213 uint64_t dsc_fromtxg;
214 int dsc_err;
215 dmu_pendop_t dsc_pending_op;
216 uint64_t dsc_featureflags;
217 uint64_t dsc_last_data_object;
218 uint64_t dsc_last_data_offset;
219 uint64_t dsc_resume_object;
220 uint64_t dsc_resume_offset;
221 boolean_t dsc_sent_begin;
222 boolean_t dsc_sent_end;
223 } dmu_send_cookie_t;
224
225 static int do_dump(dmu_send_cookie_t *dscp, struct send_range *range);
226
227 static void
228 range_free(struct send_range *range)
229 {
230 if (range->type == OBJECT) {
231 size_t size = sizeof (dnode_phys_t) *
232 (range->sru.object.dnp->dn_extra_slots + 1);
233 kmem_free(range->sru.object.dnp, size);
234 } else if (range->type == DATA) {
235 mutex_enter(&range->sru.data.lock);
236 while (range->sru.data.io_outstanding)
237 cv_wait(&range->sru.data.cv, &range->sru.data.lock);
238 if (range->sru.data.abd != NULL)
239 abd_free(range->sru.data.abd);
240 if (range->sru.data.abuf != NULL) {
241 arc_buf_destroy(range->sru.data.abuf,
242 &range->sru.data.abuf);
243 }
244 mutex_exit(&range->sru.data.lock);
245
246 cv_destroy(&range->sru.data.cv);
247 mutex_destroy(&range->sru.data.lock);
248 }
249 kmem_free(range, sizeof (*range));
250 }
251
252 /*
253 * For all record types except BEGIN, fill in the checksum (overlaid in
254 * drr_u.drr_checksum.drr_checksum). The checksum verifies everything
255 * up to the start of the checksum itself.
256 */
257 static int
258 dump_record(dmu_send_cookie_t *dscp, void *payload, int payload_len)
259 {
260 dmu_send_outparams_t *dso = dscp->dsc_dso;
261 ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
262 ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
263 (void) fletcher_4_incremental_native(dscp->dsc_drr,
264 offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
265 &dscp->dsc_zc);
266 if (dscp->dsc_drr->drr_type == DRR_BEGIN) {
267 dscp->dsc_sent_begin = B_TRUE;
268 } else {
269 ASSERT(ZIO_CHECKSUM_IS_ZERO(&dscp->dsc_drr->drr_u.
270 drr_checksum.drr_checksum));
271 dscp->dsc_drr->drr_u.drr_checksum.drr_checksum = dscp->dsc_zc;
272 }
273 if (dscp->dsc_drr->drr_type == DRR_END) {
274 dscp->dsc_sent_end = B_TRUE;
275 }
276 (void) fletcher_4_incremental_native(&dscp->dsc_drr->
277 drr_u.drr_checksum.drr_checksum,
278 sizeof (zio_cksum_t), &dscp->dsc_zc);
279 *dscp->dsc_off += sizeof (dmu_replay_record_t);
280 dscp->dsc_err = dso->dso_outfunc(dscp->dsc_os, dscp->dsc_drr,
281 sizeof (dmu_replay_record_t), dso->dso_arg);
282 if (dscp->dsc_err != 0)
283 return (SET_ERROR(EINTR));
284 if (payload_len != 0) {
285 *dscp->dsc_off += payload_len;
286 /*
287 * payload is null when dso_dryrun == B_TRUE (i.e. when we're
288 * doing a send size calculation)
289 */
290 if (payload != NULL) {
291 (void) fletcher_4_incremental_native(
292 payload, payload_len, &dscp->dsc_zc);
293 }
294
295 /*
296 * The code does not rely on this (len being a multiple of 8).
297 * We keep this assertion because of the corresponding assertion
298 * in receive_read(). Keeping this assertion ensures that we do
299 * not inadvertently break backwards compatibility (causing the
300 * assertion in receive_read() to trigger on old software).
301 *
302 * Raw sends cannot be received on old software, and so can
303 * bypass this assertion.
304 */
305
306 ASSERT((payload_len % 8 == 0) ||
307 (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW));
308
309 dscp->dsc_err = dso->dso_outfunc(dscp->dsc_os, payload,
310 payload_len, dso->dso_arg);
311 if (dscp->dsc_err != 0)
312 return (SET_ERROR(EINTR));
313 }
314 return (0);
315 }
316
317 /*
318 * Fill in the drr_free struct, or perform aggregation if the previous record is
319 * also a free record, and the two are adjacent.
320 *
321 * Note that we send free records even for a full send, because we want to be
322 * able to receive a full send as a clone, which requires a list of all the free
323 * and freeobject records that were generated on the source.
324 */
325 static int
326 dump_free(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
327 uint64_t length)
328 {
329 struct drr_free *drrf = &(dscp->dsc_drr->drr_u.drr_free);
330
331 /*
332 * When we receive a free record, dbuf_free_range() assumes
333 * that the receiving system doesn't have any dbufs in the range
334 * being freed. This is always true because there is a one-record
335 * constraint: we only send one WRITE record for any given
336 * object,offset. We know that the one-record constraint is
337 * true because we always send data in increasing order by
338 * object,offset.
339 *
340 * If the increasing-order constraint ever changes, we should find
341 * another way to assert that the one-record constraint is still
342 * satisfied.
343 */
344 ASSERT(object > dscp->dsc_last_data_object ||
345 (object == dscp->dsc_last_data_object &&
346 offset > dscp->dsc_last_data_offset));
347
348 /*
349 * If there is a pending op, but it's not PENDING_FREE, push it out,
350 * since free block aggregation can only be done for blocks of the
351 * same type (i.e., DRR_FREE records can only be aggregated with
352 * other DRR_FREE records. DRR_FREEOBJECTS records can only be
353 * aggregated with other DRR_FREEOBJECTS records).
354 */
355 if (dscp->dsc_pending_op != PENDING_NONE &&
356 dscp->dsc_pending_op != PENDING_FREE) {
357 if (dump_record(dscp, NULL, 0) != 0)
358 return (SET_ERROR(EINTR));
359 dscp->dsc_pending_op = PENDING_NONE;
360 }
361
362 if (dscp->dsc_pending_op == PENDING_FREE) {
363 /*
364 * Check to see whether this free block can be aggregated
365 * with pending one.
366 */
367 if (drrf->drr_object == object && drrf->drr_offset +
368 drrf->drr_length == offset) {
369 if (offset + length < offset || length == UINT64_MAX)
370 drrf->drr_length = UINT64_MAX;
371 else
372 drrf->drr_length += length;
373 return (0);
374 } else {
375 /* not a continuation. Push out pending record */
376 if (dump_record(dscp, NULL, 0) != 0)
377 return (SET_ERROR(EINTR));
378 dscp->dsc_pending_op = PENDING_NONE;
379 }
380 }
381 /* create a FREE record and make it pending */
382 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
383 dscp->dsc_drr->drr_type = DRR_FREE;
384 drrf->drr_object = object;
385 drrf->drr_offset = offset;
386 if (offset + length < offset)
387 drrf->drr_length = DMU_OBJECT_END;
388 else
389 drrf->drr_length = length;
390 drrf->drr_toguid = dscp->dsc_toguid;
391 if (length == DMU_OBJECT_END) {
392 if (dump_record(dscp, NULL, 0) != 0)
393 return (SET_ERROR(EINTR));
394 } else {
395 dscp->dsc_pending_op = PENDING_FREE;
396 }
397
398 return (0);
399 }
400
401 /*
402 * Fill in the drr_redact struct, or perform aggregation if the previous record
403 * is also a redaction record, and the two are adjacent.
404 */
405 static int
406 dump_redact(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
407 uint64_t length)
408 {
409 struct drr_redact *drrr = &dscp->dsc_drr->drr_u.drr_redact;
410
411 /*
412 * If there is a pending op, but it's not PENDING_REDACT, push it out,
413 * since free block aggregation can only be done for blocks of the
414 * same type (i.e., DRR_REDACT records can only be aggregated with
415 * other DRR_REDACT records).
416 */
417 if (dscp->dsc_pending_op != PENDING_NONE &&
418 dscp->dsc_pending_op != PENDING_REDACT) {
419 if (dump_record(dscp, NULL, 0) != 0)
420 return (SET_ERROR(EINTR));
421 dscp->dsc_pending_op = PENDING_NONE;
422 }
423
424 if (dscp->dsc_pending_op == PENDING_REDACT) {
425 /*
426 * Check to see whether this redacted block can be aggregated
427 * with pending one.
428 */
429 if (drrr->drr_object == object && drrr->drr_offset +
430 drrr->drr_length == offset) {
431 drrr->drr_length += length;
432 return (0);
433 } else {
434 /* not a continuation. Push out pending record */
435 if (dump_record(dscp, NULL, 0) != 0)
436 return (SET_ERROR(EINTR));
437 dscp->dsc_pending_op = PENDING_NONE;
438 }
439 }
440 /* create a REDACT record and make it pending */
441 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
442 dscp->dsc_drr->drr_type = DRR_REDACT;
443 drrr->drr_object = object;
444 drrr->drr_offset = offset;
445 drrr->drr_length = length;
446 drrr->drr_toguid = dscp->dsc_toguid;
447 dscp->dsc_pending_op = PENDING_REDACT;
448
449 return (0);
450 }
451
452 static int
453 dmu_dump_write(dmu_send_cookie_t *dscp, dmu_object_type_t type, uint64_t object,
454 uint64_t offset, int lsize, int psize, const blkptr_t *bp,
455 boolean_t io_compressed, void *data)
456 {
457 uint64_t payload_size;
458 boolean_t raw = (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW);
459 struct drr_write *drrw = &(dscp->dsc_drr->drr_u.drr_write);
460
461 /*
462 * We send data in increasing object, offset order.
463 * See comment in dump_free() for details.
464 */
465 ASSERT(object > dscp->dsc_last_data_object ||
466 (object == dscp->dsc_last_data_object &&
467 offset > dscp->dsc_last_data_offset));
468 dscp->dsc_last_data_object = object;
469 dscp->dsc_last_data_offset = offset + lsize - 1;
470
471 /*
472 * If there is any kind of pending aggregation (currently either
473 * a grouping of free objects or free blocks), push it out to
474 * the stream, since aggregation can't be done across operations
475 * of different types.
476 */
477 if (dscp->dsc_pending_op != PENDING_NONE) {
478 if (dump_record(dscp, NULL, 0) != 0)
479 return (SET_ERROR(EINTR));
480 dscp->dsc_pending_op = PENDING_NONE;
481 }
482 /* write a WRITE record */
483 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
484 dscp->dsc_drr->drr_type = DRR_WRITE;
485 drrw->drr_object = object;
486 drrw->drr_type = type;
487 drrw->drr_offset = offset;
488 drrw->drr_toguid = dscp->dsc_toguid;
489 drrw->drr_logical_size = lsize;
490
491 /* only set the compression fields if the buf is compressed or raw */
492 boolean_t compressed =
493 (bp != NULL ? BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF &&
494 io_compressed : lsize != psize);
495 if (raw || compressed) {
496 ASSERT(raw || dscp->dsc_featureflags &
497 DMU_BACKUP_FEATURE_COMPRESSED);
498 ASSERT(!BP_IS_EMBEDDED(bp));
499 ASSERT3S(psize, >, 0);
500
501 if (raw) {
502 ASSERT(BP_IS_PROTECTED(bp));
503
504 /*
505 * This is a raw protected block so we need to pass
506 * along everything the receiving side will need to
507 * interpret this block, including the byteswap, salt,
508 * IV, and MAC.
509 */
510 if (BP_SHOULD_BYTESWAP(bp))
511 drrw->drr_flags |= DRR_RAW_BYTESWAP;
512 zio_crypt_decode_params_bp(bp, drrw->drr_salt,
513 drrw->drr_iv);
514 zio_crypt_decode_mac_bp(bp, drrw->drr_mac);
515 } else {
516 /* this is a compressed block */
517 ASSERT(dscp->dsc_featureflags &
518 DMU_BACKUP_FEATURE_COMPRESSED);
519 ASSERT(!BP_SHOULD_BYTESWAP(bp));
520 ASSERT(!DMU_OT_IS_METADATA(BP_GET_TYPE(bp)));
521 ASSERT3U(BP_GET_COMPRESS(bp), !=, ZIO_COMPRESS_OFF);
522 ASSERT3S(lsize, >=, psize);
523 }
524
525 /* set fields common to compressed and raw sends */
526 drrw->drr_compressiontype = BP_GET_COMPRESS(bp);
527 drrw->drr_compressed_size = psize;
528 payload_size = drrw->drr_compressed_size;
529 } else {
530 payload_size = drrw->drr_logical_size;
531 }
532
533 if (bp == NULL || BP_IS_EMBEDDED(bp) || (BP_IS_PROTECTED(bp) && !raw)) {
534 /*
535 * There's no pre-computed checksum for partial-block writes,
536 * embedded BP's, or encrypted BP's that are being sent as
537 * plaintext, so (like fletcher4-checksummed blocks) userland
538 * will have to compute a dedup-capable checksum itself.
539 */
540 drrw->drr_checksumtype = ZIO_CHECKSUM_OFF;
541 } else {
542 drrw->drr_checksumtype = BP_GET_CHECKSUM(bp);
543 if (zio_checksum_table[drrw->drr_checksumtype].ci_flags &
544 ZCHECKSUM_FLAG_DEDUP)
545 drrw->drr_flags |= DRR_CHECKSUM_DEDUP;
546 DDK_SET_LSIZE(&drrw->drr_key, BP_GET_LSIZE(bp));
547 DDK_SET_PSIZE(&drrw->drr_key, BP_GET_PSIZE(bp));
548 DDK_SET_COMPRESS(&drrw->drr_key, BP_GET_COMPRESS(bp));
549 DDK_SET_CRYPT(&drrw->drr_key, BP_IS_PROTECTED(bp));
550 drrw->drr_key.ddk_cksum = bp->blk_cksum;
551 }
552
553 if (dump_record(dscp, data, payload_size) != 0)
554 return (SET_ERROR(EINTR));
555 return (0);
556 }
557
558 static int
559 dump_write_embedded(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
560 int blksz, const blkptr_t *bp)
561 {
562 char buf[BPE_PAYLOAD_SIZE];
563 struct drr_write_embedded *drrw =
564 &(dscp->dsc_drr->drr_u.drr_write_embedded);
565
566 if (dscp->dsc_pending_op != PENDING_NONE) {
567 if (dump_record(dscp, NULL, 0) != 0)
568 return (SET_ERROR(EINTR));
569 dscp->dsc_pending_op = PENDING_NONE;
570 }
571
572 ASSERT(BP_IS_EMBEDDED(bp));
573
574 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
575 dscp->dsc_drr->drr_type = DRR_WRITE_EMBEDDED;
576 drrw->drr_object = object;
577 drrw->drr_offset = offset;
578 drrw->drr_length = blksz;
579 drrw->drr_toguid = dscp->dsc_toguid;
580 drrw->drr_compression = BP_GET_COMPRESS(bp);
581 drrw->drr_etype = BPE_GET_ETYPE(bp);
582 drrw->drr_lsize = BPE_GET_LSIZE(bp);
583 drrw->drr_psize = BPE_GET_PSIZE(bp);
584
585 decode_embedded_bp_compressed(bp, buf);
586
587 uint32_t psize = drrw->drr_psize;
588 uint32_t rsize = P2ROUNDUP(psize, 8);
589
590 if (psize != rsize)
591 memset(buf + psize, 0, rsize - psize);
592
593 if (dump_record(dscp, buf, rsize) != 0)
594 return (SET_ERROR(EINTR));
595 return (0);
596 }
597
598 static int
599 dump_spill(dmu_send_cookie_t *dscp, const blkptr_t *bp, uint64_t object,
600 void *data)
601 {
602 struct drr_spill *drrs = &(dscp->dsc_drr->drr_u.drr_spill);
603 uint64_t blksz = BP_GET_LSIZE(bp);
604 uint64_t payload_size = blksz;
605
606 if (dscp->dsc_pending_op != PENDING_NONE) {
607 if (dump_record(dscp, NULL, 0) != 0)
608 return (SET_ERROR(EINTR));
609 dscp->dsc_pending_op = PENDING_NONE;
610 }
611
612 /* write a SPILL record */
613 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
614 dscp->dsc_drr->drr_type = DRR_SPILL;
615 drrs->drr_object = object;
616 drrs->drr_length = blksz;
617 drrs->drr_toguid = dscp->dsc_toguid;
618
619 /* See comment in dump_dnode() for full details */
620 if (zfs_send_unmodified_spill_blocks &&
621 (bp->blk_birth <= dscp->dsc_fromtxg)) {
622 drrs->drr_flags |= DRR_SPILL_UNMODIFIED;
623 }
624
625 /* handle raw send fields */
626 if (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW) {
627 ASSERT(BP_IS_PROTECTED(bp));
628
629 if (BP_SHOULD_BYTESWAP(bp))
630 drrs->drr_flags |= DRR_RAW_BYTESWAP;
631 drrs->drr_compressiontype = BP_GET_COMPRESS(bp);
632 drrs->drr_compressed_size = BP_GET_PSIZE(bp);
633 zio_crypt_decode_params_bp(bp, drrs->drr_salt, drrs->drr_iv);
634 zio_crypt_decode_mac_bp(bp, drrs->drr_mac);
635 payload_size = drrs->drr_compressed_size;
636 }
637
638 if (dump_record(dscp, data, payload_size) != 0)
639 return (SET_ERROR(EINTR));
640 return (0);
641 }
642
643 static int
644 dump_freeobjects(dmu_send_cookie_t *dscp, uint64_t firstobj, uint64_t numobjs)
645 {
646 struct drr_freeobjects *drrfo = &(dscp->dsc_drr->drr_u.drr_freeobjects);
647 uint64_t maxobj = DNODES_PER_BLOCK *
648 (DMU_META_DNODE(dscp->dsc_os)->dn_maxblkid + 1);
649
650 /*
651 * ZoL < 0.7 does not handle large FREEOBJECTS records correctly,
652 * leading to zfs recv never completing. to avoid this issue, don't
653 * send FREEOBJECTS records for object IDs which cannot exist on the
654 * receiving side.
655 */
656 if (maxobj > 0) {
657 if (maxobj <= firstobj)
658 return (0);
659
660 if (maxobj < firstobj + numobjs)
661 numobjs = maxobj - firstobj;
662 }
663
664 /*
665 * If there is a pending op, but it's not PENDING_FREEOBJECTS,
666 * push it out, since free block aggregation can only be done for
667 * blocks of the same type (i.e., DRR_FREE records can only be
668 * aggregated with other DRR_FREE records. DRR_FREEOBJECTS records
669 * can only be aggregated with other DRR_FREEOBJECTS records).
670 */
671 if (dscp->dsc_pending_op != PENDING_NONE &&
672 dscp->dsc_pending_op != PENDING_FREEOBJECTS) {
673 if (dump_record(dscp, NULL, 0) != 0)
674 return (SET_ERROR(EINTR));
675 dscp->dsc_pending_op = PENDING_NONE;
676 }
677
678 if (dscp->dsc_pending_op == PENDING_FREEOBJECTS) {
679 /*
680 * See whether this free object array can be aggregated
681 * with pending one
682 */
683 if (drrfo->drr_firstobj + drrfo->drr_numobjs == firstobj) {
684 drrfo->drr_numobjs += numobjs;
685 return (0);
686 } else {
687 /* can't be aggregated. Push out pending record */
688 if (dump_record(dscp, NULL, 0) != 0)
689 return (SET_ERROR(EINTR));
690 dscp->dsc_pending_op = PENDING_NONE;
691 }
692 }
693
694 /* write a FREEOBJECTS record */
695 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
696 dscp->dsc_drr->drr_type = DRR_FREEOBJECTS;
697 drrfo->drr_firstobj = firstobj;
698 drrfo->drr_numobjs = numobjs;
699 drrfo->drr_toguid = dscp->dsc_toguid;
700
701 dscp->dsc_pending_op = PENDING_FREEOBJECTS;
702
703 return (0);
704 }
705
706 static int
707 dump_dnode(dmu_send_cookie_t *dscp, const blkptr_t *bp, uint64_t object,
708 dnode_phys_t *dnp)
709 {
710 struct drr_object *drro = &(dscp->dsc_drr->drr_u.drr_object);
711 int bonuslen;
712
713 if (object < dscp->dsc_resume_object) {
714 /*
715 * Note: when resuming, we will visit all the dnodes in
716 * the block of dnodes that we are resuming from. In
717 * this case it's unnecessary to send the dnodes prior to
718 * the one we are resuming from. We should be at most one
719 * block's worth of dnodes behind the resume point.
720 */
721 ASSERT3U(dscp->dsc_resume_object - object, <,
722 1 << (DNODE_BLOCK_SHIFT - DNODE_SHIFT));
723 return (0);
724 }
725
726 if (dnp == NULL || dnp->dn_type == DMU_OT_NONE)
727 return (dump_freeobjects(dscp, object, 1));
728
729 if (dscp->dsc_pending_op != PENDING_NONE) {
730 if (dump_record(dscp, NULL, 0) != 0)
731 return (SET_ERROR(EINTR));
732 dscp->dsc_pending_op = PENDING_NONE;
733 }
734
735 /* write an OBJECT record */
736 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
737 dscp->dsc_drr->drr_type = DRR_OBJECT;
738 drro->drr_object = object;
739 drro->drr_type = dnp->dn_type;
740 drro->drr_bonustype = dnp->dn_bonustype;
741 drro->drr_blksz = dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT;
742 drro->drr_bonuslen = dnp->dn_bonuslen;
743 drro->drr_dn_slots = dnp->dn_extra_slots + 1;
744 drro->drr_checksumtype = dnp->dn_checksum;
745 drro->drr_compress = dnp->dn_compress;
746 drro->drr_toguid = dscp->dsc_toguid;
747
748 if (!(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) &&
749 drro->drr_blksz > SPA_OLD_MAXBLOCKSIZE)
750 drro->drr_blksz = SPA_OLD_MAXBLOCKSIZE;
751
752 bonuslen = P2ROUNDUP(dnp->dn_bonuslen, 8);
753
754 if ((dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW)) {
755 ASSERT(BP_IS_ENCRYPTED(bp));
756
757 if (BP_SHOULD_BYTESWAP(bp))
758 drro->drr_flags |= DRR_RAW_BYTESWAP;
759
760 /* needed for reconstructing dnp on recv side */
761 drro->drr_maxblkid = dnp->dn_maxblkid;
762 drro->drr_indblkshift = dnp->dn_indblkshift;
763 drro->drr_nlevels = dnp->dn_nlevels;
764 drro->drr_nblkptr = dnp->dn_nblkptr;
765
766 /*
767 * Since we encrypt the entire bonus area, the (raw) part
768 * beyond the bonuslen is actually nonzero, so we need
769 * to send it.
770 */
771 if (bonuslen != 0) {
772 if (drro->drr_bonuslen > DN_MAX_BONUS_LEN(dnp))
773 return (SET_ERROR(EINVAL));
774 drro->drr_raw_bonuslen = DN_MAX_BONUS_LEN(dnp);
775 bonuslen = drro->drr_raw_bonuslen;
776 }
777 }
778
779 /*
780 * DRR_OBJECT_SPILL is set for every dnode which references a
781 * spill block. This allows the receiving pool to definitively
782 * determine when a spill block should be kept or freed.
783 */
784 if (dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR)
785 drro->drr_flags |= DRR_OBJECT_SPILL;
786
787 if (dump_record(dscp, DN_BONUS(dnp), bonuslen) != 0)
788 return (SET_ERROR(EINTR));
789
790 /* Free anything past the end of the file. */
791 if (dump_free(dscp, object, (dnp->dn_maxblkid + 1) *
792 (dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT), DMU_OBJECT_END) != 0)
793 return (SET_ERROR(EINTR));
794
795 /*
796 * Send DRR_SPILL records for unmodified spill blocks. This is useful
797 * because changing certain attributes of the object (e.g. blocksize)
798 * can cause old versions of ZFS to incorrectly remove a spill block.
799 * Including these records in the stream forces an up to date version
800 * to always be written ensuring they're never lost. Current versions
801 * of the code which understand the DRR_FLAG_SPILL_BLOCK feature can
802 * ignore these unmodified spill blocks.
803 */
804 if (zfs_send_unmodified_spill_blocks &&
805 (dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR) &&
806 (DN_SPILL_BLKPTR(dnp)->blk_birth <= dscp->dsc_fromtxg)) {
807 struct send_range record;
808 blkptr_t *bp = DN_SPILL_BLKPTR(dnp);
809
810 memset(&record, 0, sizeof (struct send_range));
811 record.type = DATA;
812 record.object = object;
813 record.eos_marker = B_FALSE;
814 record.start_blkid = DMU_SPILL_BLKID;
815 record.end_blkid = record.start_blkid + 1;
816 record.sru.data.bp = *bp;
817 record.sru.data.obj_type = dnp->dn_type;
818 record.sru.data.datablksz = BP_GET_LSIZE(bp);
819
820 if (do_dump(dscp, &record) != 0)
821 return (SET_ERROR(EINTR));
822 }
823
824 if (dscp->dsc_err != 0)
825 return (SET_ERROR(EINTR));
826
827 return (0);
828 }
829
830 static int
831 dump_object_range(dmu_send_cookie_t *dscp, const blkptr_t *bp,
832 uint64_t firstobj, uint64_t numslots)
833 {
834 struct drr_object_range *drror =
835 &(dscp->dsc_drr->drr_u.drr_object_range);
836
837 /* we only use this record type for raw sends */
838 ASSERT(BP_IS_PROTECTED(bp));
839 ASSERT(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW);
840 ASSERT3U(BP_GET_COMPRESS(bp), ==, ZIO_COMPRESS_OFF);
841 ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_DNODE);
842 ASSERT0(BP_GET_LEVEL(bp));
843
844 if (dscp->dsc_pending_op != PENDING_NONE) {
845 if (dump_record(dscp, NULL, 0) != 0)
846 return (SET_ERROR(EINTR));
847 dscp->dsc_pending_op = PENDING_NONE;
848 }
849
850 memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
851 dscp->dsc_drr->drr_type = DRR_OBJECT_RANGE;
852 drror->drr_firstobj = firstobj;
853 drror->drr_numslots = numslots;
854 drror->drr_toguid = dscp->dsc_toguid;
855 if (BP_SHOULD_BYTESWAP(bp))
856 drror->drr_flags |= DRR_RAW_BYTESWAP;
857 zio_crypt_decode_params_bp(bp, drror->drr_salt, drror->drr_iv);
858 zio_crypt_decode_mac_bp(bp, drror->drr_mac);
859
860 if (dump_record(dscp, NULL, 0) != 0)
861 return (SET_ERROR(EINTR));
862 return (0);
863 }
864
865 static boolean_t
866 send_do_embed(const blkptr_t *bp, uint64_t featureflags)
867 {
868 if (!BP_IS_EMBEDDED(bp))
869 return (B_FALSE);
870
871 /*
872 * Compression function must be legacy, or explicitly enabled.
873 */
874 if ((BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_LEGACY_FUNCTIONS &&
875 !(featureflags & DMU_BACKUP_FEATURE_LZ4)))
876 return (B_FALSE);
877
878 /*
879 * If we have not set the ZSTD feature flag, we can't send ZSTD
880 * compressed embedded blocks, as the receiver may not support them.
881 */
882 if ((BP_GET_COMPRESS(bp) == ZIO_COMPRESS_ZSTD &&
883 !(featureflags & DMU_BACKUP_FEATURE_ZSTD)))
884 return (B_FALSE);
885
886 /*
887 * Embed type must be explicitly enabled.
888 */
889 switch (BPE_GET_ETYPE(bp)) {
890 case BP_EMBEDDED_TYPE_DATA:
891 if (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)
892 return (B_TRUE);
893 break;
894 default:
895 return (B_FALSE);
896 }
897 return (B_FALSE);
898 }
899
900 /*
901 * This function actually handles figuring out what kind of record needs to be
902 * dumped, and calling the appropriate helper function. In most cases,
903 * the data has already been read by send_reader_thread().
904 */
905 static int
906 do_dump(dmu_send_cookie_t *dscp, struct send_range *range)
907 {
908 int err = 0;
909 switch (range->type) {
910 case OBJECT:
911 err = dump_dnode(dscp, &range->sru.object.bp, range->object,
912 range->sru.object.dnp);
913 return (err);
914 case OBJECT_RANGE: {
915 ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
916 if (!(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW)) {
917 return (0);
918 }
919 uint64_t epb = BP_GET_LSIZE(&range->sru.object_range.bp) >>
920 DNODE_SHIFT;
921 uint64_t firstobj = range->start_blkid * epb;
922 err = dump_object_range(dscp, &range->sru.object_range.bp,
923 firstobj, epb);
924 break;
925 }
926 case REDACT: {
927 struct srr *srrp = &range->sru.redact;
928 err = dump_redact(dscp, range->object, range->start_blkid *
929 srrp->datablksz, (range->end_blkid - range->start_blkid) *
930 srrp->datablksz);
931 return (err);
932 }
933 case DATA: {
934 struct srd *srdp = &range->sru.data;
935 blkptr_t *bp = &srdp->bp;
936 spa_t *spa =
937 dmu_objset_spa(dscp->dsc_os);
938
939 ASSERT3U(srdp->datablksz, ==, BP_GET_LSIZE(bp));
940 ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
941 if (BP_GET_TYPE(bp) == DMU_OT_SA) {
942 arc_flags_t aflags = ARC_FLAG_WAIT;
943 zio_flag_t zioflags = ZIO_FLAG_CANFAIL;
944
945 if (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW) {
946 ASSERT(BP_IS_PROTECTED(bp));
947 zioflags |= ZIO_FLAG_RAW;
948 }
949
950 zbookmark_phys_t zb;
951 ASSERT3U(range->start_blkid, ==, DMU_SPILL_BLKID);
952 zb.zb_objset = dmu_objset_id(dscp->dsc_os);
953 zb.zb_object = range->object;
954 zb.zb_level = 0;
955 zb.zb_blkid = range->start_blkid;
956
957 arc_buf_t *abuf = NULL;
958 if (!dscp->dsc_dso->dso_dryrun && arc_read(NULL, spa,
959 bp, arc_getbuf_func, &abuf, ZIO_PRIORITY_ASYNC_READ,
960 zioflags, &aflags, &zb) != 0)
961 return (SET_ERROR(EIO));
962
963 err = dump_spill(dscp, bp, zb.zb_object,
964 (abuf == NULL ? NULL : abuf->b_data));
965 if (abuf != NULL)
966 arc_buf_destroy(abuf, &abuf);
967 return (err);
968 }
969 if (send_do_embed(bp, dscp->dsc_featureflags)) {
970 err = dump_write_embedded(dscp, range->object,
971 range->start_blkid * srdp->datablksz,
972 srdp->datablksz, bp);
973 return (err);
974 }
975 ASSERT(range->object > dscp->dsc_resume_object ||
976 (range->object == dscp->dsc_resume_object &&
977 range->start_blkid * srdp->datablksz >=
978 dscp->dsc_resume_offset));
979 /* it's a level-0 block of a regular object */
980
981 mutex_enter(&srdp->lock);
982 while (srdp->io_outstanding)
983 cv_wait(&srdp->cv, &srdp->lock);
984 err = srdp->io_err;
985 mutex_exit(&srdp->lock);
986
987 if (err != 0) {
988 if (zfs_send_corrupt_data &&
989 !dscp->dsc_dso->dso_dryrun) {
990 /*
991 * Send a block filled with 0x"zfs badd bloc"
992 */
993 srdp->abuf = arc_alloc_buf(spa, &srdp->abuf,
994 ARC_BUFC_DATA, srdp->datablksz);
995 uint64_t *ptr;
996 for (ptr = srdp->abuf->b_data;
997 (char *)ptr < (char *)srdp->abuf->b_data +
998 srdp->datablksz; ptr++)
999 *ptr = 0x2f5baddb10cULL;
1000 } else {
1001 return (SET_ERROR(EIO));
1002 }
1003 }
1004
1005 ASSERT(dscp->dsc_dso->dso_dryrun ||
1006 srdp->abuf != NULL || srdp->abd != NULL);
1007
1008 uint64_t offset = range->start_blkid * srdp->datablksz;
1009
1010 char *data = NULL;
1011 if (srdp->abd != NULL) {
1012 data = abd_to_buf(srdp->abd);
1013 ASSERT3P(srdp->abuf, ==, NULL);
1014 } else if (srdp->abuf != NULL) {
1015 data = srdp->abuf->b_data;
1016 }
1017
1018 /*
1019 * If we have large blocks stored on disk but the send flags
1020 * don't allow us to send large blocks, we split the data from
1021 * the arc buf into chunks.
1022 */
1023 if (srdp->datablksz > SPA_OLD_MAXBLOCKSIZE &&
1024 !(dscp->dsc_featureflags &
1025 DMU_BACKUP_FEATURE_LARGE_BLOCKS)) {
1026 while (srdp->datablksz > 0 && err == 0) {
1027 int n = MIN(srdp->datablksz,
1028 SPA_OLD_MAXBLOCKSIZE);
1029 err = dmu_dump_write(dscp, srdp->obj_type,
1030 range->object, offset, n, n, NULL, B_FALSE,
1031 data);
1032 offset += n;
1033 /*
1034 * When doing dry run, data==NULL is used as a
1035 * sentinel value by
1036 * dmu_dump_write()->dump_record().
1037 */
1038 if (data != NULL)
1039 data += n;
1040 srdp->datablksz -= n;
1041 }
1042 } else {
1043 err = dmu_dump_write(dscp, srdp->obj_type,
1044 range->object, offset,
1045 srdp->datablksz, srdp->datasz, bp,
1046 srdp->io_compressed, data);
1047 }
1048 return (err);
1049 }
1050 case HOLE: {
1051 struct srh *srhp = &range->sru.hole;
1052 if (range->object == DMU_META_DNODE_OBJECT) {
1053 uint32_t span = srhp->datablksz >> DNODE_SHIFT;
1054 uint64_t first_obj = range->start_blkid * span;
1055 uint64_t numobj = range->end_blkid * span - first_obj;
1056 return (dump_freeobjects(dscp, first_obj, numobj));
1057 }
1058 uint64_t offset = 0;
1059
1060 /*
1061 * If this multiply overflows, we don't need to send this block.
1062 * Even if it has a birth time, it can never not be a hole, so
1063 * we don't need to send records for it.
1064 */
1065 if (!overflow_multiply(range->start_blkid, srhp->datablksz,
1066 &offset)) {
1067 return (0);
1068 }
1069 uint64_t len = 0;
1070
1071 if (!overflow_multiply(range->end_blkid, srhp->datablksz, &len))
1072 len = UINT64_MAX;
1073 len = len - offset;
1074 return (dump_free(dscp, range->object, offset, len));
1075 }
1076 default:
1077 panic("Invalid range type in do_dump: %d", range->type);
1078 }
1079 return (err);
1080 }
1081
1082 static struct send_range *
1083 range_alloc(enum type type, uint64_t object, uint64_t start_blkid,
1084 uint64_t end_blkid, boolean_t eos)
1085 {
1086 struct send_range *range = kmem_alloc(sizeof (*range), KM_SLEEP);
1087 range->type = type;
1088 range->object = object;
1089 range->start_blkid = start_blkid;
1090 range->end_blkid = end_blkid;
1091 range->eos_marker = eos;
1092 if (type == DATA) {
1093 range->sru.data.abd = NULL;
1094 range->sru.data.abuf = NULL;
1095 mutex_init(&range->sru.data.lock, NULL, MUTEX_DEFAULT, NULL);
1096 cv_init(&range->sru.data.cv, NULL, CV_DEFAULT, NULL);
1097 range->sru.data.io_outstanding = 0;
1098 range->sru.data.io_err = 0;
1099 range->sru.data.io_compressed = B_FALSE;
1100 }
1101 return (range);
1102 }
1103
1104 /*
1105 * This is the callback function to traverse_dataset that acts as a worker
1106 * thread for dmu_send_impl.
1107 */
1108 static int
1109 send_cb(spa_t *spa, zilog_t *zilog, const blkptr_t *bp,
1110 const zbookmark_phys_t *zb, const struct dnode_phys *dnp, void *arg)
1111 {
1112 (void) zilog;
1113 struct send_thread_arg *sta = arg;
1114 struct send_range *record;
1115
1116 ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
1117 zb->zb_object >= sta->resume.zb_object);
1118
1119 /*
1120 * All bps of an encrypted os should have the encryption bit set.
1121 * If this is not true it indicates tampering and we report an error.
1122 */
1123 if (sta->os->os_encrypted &&
1124 !BP_IS_HOLE(bp) && !BP_USES_CRYPT(bp)) {
1125 spa_log_error(spa, zb);
1126 zfs_panic_recover("unencrypted block in encrypted "
1127 "object set %llu", dmu_objset_id(sta->os));
1128 return (SET_ERROR(EIO));
1129 }
1130
1131 if (sta->cancel)
1132 return (SET_ERROR(EINTR));
1133 if (zb->zb_object != DMU_META_DNODE_OBJECT &&
1134 DMU_OBJECT_IS_SPECIAL(zb->zb_object))
1135 return (0);
1136 atomic_inc_64(sta->num_blocks_visited);
1137
1138 if (zb->zb_level == ZB_DNODE_LEVEL) {
1139 if (zb->zb_object == DMU_META_DNODE_OBJECT)
1140 return (0);
1141 record = range_alloc(OBJECT, zb->zb_object, 0, 0, B_FALSE);
1142 record->sru.object.bp = *bp;
1143 size_t size = sizeof (*dnp) * (dnp->dn_extra_slots + 1);
1144 record->sru.object.dnp = kmem_alloc(size, KM_SLEEP);
1145 memcpy(record->sru.object.dnp, dnp, size);
1146 bqueue_enqueue(&sta->q, record, sizeof (*record));
1147 return (0);
1148 }
1149 if (zb->zb_level == 0 && zb->zb_object == DMU_META_DNODE_OBJECT &&
1150 !BP_IS_HOLE(bp)) {
1151 record = range_alloc(OBJECT_RANGE, 0, zb->zb_blkid,
1152 zb->zb_blkid + 1, B_FALSE);
1153 record->sru.object_range.bp = *bp;
1154 bqueue_enqueue(&sta->q, record, sizeof (*record));
1155 return (0);
1156 }
1157 if (zb->zb_level < 0 || (zb->zb_level > 0 && !BP_IS_HOLE(bp)))
1158 return (0);
1159 if (zb->zb_object == DMU_META_DNODE_OBJECT && !BP_IS_HOLE(bp))
1160 return (0);
1161
1162 uint64_t span = bp_span_in_blocks(dnp->dn_indblkshift, zb->zb_level);
1163 uint64_t start;
1164
1165 /*
1166 * If this multiply overflows, we don't need to send this block.
1167 * Even if it has a birth time, it can never not be a hole, so
1168 * we don't need to send records for it.
1169 */
1170 if (!overflow_multiply(span, zb->zb_blkid, &start) || (!(zb->zb_blkid ==
1171 DMU_SPILL_BLKID || DMU_OT_IS_METADATA(dnp->dn_type)) &&
1172 span * zb->zb_blkid > dnp->dn_maxblkid)) {
1173 ASSERT(BP_IS_HOLE(bp));
1174 return (0);
1175 }
1176
1177 if (zb->zb_blkid == DMU_SPILL_BLKID)
1178 ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);
1179
1180 enum type record_type = DATA;
1181 if (BP_IS_HOLE(bp))
1182 record_type = HOLE;
1183 else if (BP_IS_REDACTED(bp))
1184 record_type = REDACT;
1185 else
1186 record_type = DATA;
1187
1188 record = range_alloc(record_type, zb->zb_object, start,
1189 (start + span < start ? 0 : start + span), B_FALSE);
1190
1191 uint64_t datablksz = (zb->zb_blkid == DMU_SPILL_BLKID ?
1192 BP_GET_LSIZE(bp) : dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT);
1193
1194 if (BP_IS_HOLE(bp)) {
1195 record->sru.hole.datablksz = datablksz;
1196 } else if (BP_IS_REDACTED(bp)) {
1197 record->sru.redact.datablksz = datablksz;
1198 } else {
1199 record->sru.data.datablksz = datablksz;
1200 record->sru.data.obj_type = dnp->dn_type;
1201 record->sru.data.bp = *bp;
1202 }
1203
1204 bqueue_enqueue(&sta->q, record, sizeof (*record));
1205 return (0);
1206 }
1207
1208 struct redact_list_cb_arg {
1209 uint64_t *num_blocks_visited;
1210 bqueue_t *q;
1211 boolean_t *cancel;
1212 boolean_t mark_redact;
1213 };
1214
1215 static int
1216 redact_list_cb(redact_block_phys_t *rb, void *arg)
1217 {
1218 struct redact_list_cb_arg *rlcap = arg;
1219
1220 atomic_inc_64(rlcap->num_blocks_visited);
1221 if (*rlcap->cancel)
1222 return (-1);
1223
1224 struct send_range *data = range_alloc(REDACT, rb->rbp_object,
1225 rb->rbp_blkid, rb->rbp_blkid + redact_block_get_count(rb), B_FALSE);
1226 ASSERT3U(data->end_blkid, >, rb->rbp_blkid);
1227 if (rlcap->mark_redact) {
1228 data->type = REDACT;
1229 data->sru.redact.datablksz = redact_block_get_size(rb);
1230 } else {
1231 data->type = PREVIOUSLY_REDACTED;
1232 }
1233 bqueue_enqueue(rlcap->q, data, sizeof (*data));
1234
1235 return (0);
1236 }
1237
1238 /*
1239 * This function kicks off the traverse_dataset. It also handles setting the
1240 * error code of the thread in case something goes wrong, and pushes the End of
1241 * Stream record when the traverse_dataset call has finished.
1242 */
1243 static __attribute__((noreturn)) void
1244 send_traverse_thread(void *arg)
1245 {
1246 struct send_thread_arg *st_arg = arg;
1247 int err = 0;
1248 struct send_range *data;
1249 fstrans_cookie_t cookie = spl_fstrans_mark();
1250
1251 err = traverse_dataset_resume(st_arg->os->os_dsl_dataset,
1252 st_arg->fromtxg, &st_arg->resume,
1253 st_arg->flags, send_cb, st_arg);
1254
1255 if (err != EINTR)
1256 st_arg->error_code = err;
1257 data = range_alloc(DATA, 0, 0, 0, B_TRUE);
1258 bqueue_enqueue_flush(&st_arg->q, data, sizeof (*data));
1259 spl_fstrans_unmark(cookie);
1260 thread_exit();
1261 }
1262
1263 /*
1264 * Utility function that causes End of Stream records to compare after of all
1265 * others, so that other threads' comparison logic can stay simple.
1266 */
1267 static int __attribute__((unused))
1268 send_range_after(const struct send_range *from, const struct send_range *to)
1269 {
1270 if (from->eos_marker == B_TRUE)
1271 return (1);
1272 if (to->eos_marker == B_TRUE)
1273 return (-1);
1274
1275 uint64_t from_obj = from->object;
1276 uint64_t from_end_obj = from->object + 1;
1277 uint64_t to_obj = to->object;
1278 uint64_t to_end_obj = to->object + 1;
1279 if (from_obj == 0) {
1280 ASSERT(from->type == HOLE || from->type == OBJECT_RANGE);
1281 from_obj = from->start_blkid << DNODES_PER_BLOCK_SHIFT;
1282 from_end_obj = from->end_blkid << DNODES_PER_BLOCK_SHIFT;
1283 }
1284 if (to_obj == 0) {
1285 ASSERT(to->type == HOLE || to->type == OBJECT_RANGE);
1286 to_obj = to->start_blkid << DNODES_PER_BLOCK_SHIFT;
1287 to_end_obj = to->end_blkid << DNODES_PER_BLOCK_SHIFT;
1288 }
1289
1290 if (from_end_obj <= to_obj)
1291 return (-1);
1292 if (from_obj >= to_end_obj)
1293 return (1);
1294 int64_t cmp = TREE_CMP(to->type == OBJECT_RANGE, from->type ==
1295 OBJECT_RANGE);
1296 if (unlikely(cmp))
1297 return (cmp);
1298 cmp = TREE_CMP(to->type == OBJECT, from->type == OBJECT);
1299 if (unlikely(cmp))
1300 return (cmp);
1301 if (from->end_blkid <= to->start_blkid)
1302 return (-1);
1303 if (from->start_blkid >= to->end_blkid)
1304 return (1);
1305 return (0);
1306 }
1307
1308 /*
1309 * Pop the new data off the queue, check that the records we receive are in
1310 * the right order, but do not free the old data. This is used so that the
1311 * records can be sent on to the main thread without copying the data.
1312 */
1313 static struct send_range *
1314 get_next_range_nofree(bqueue_t *bq, struct send_range *prev)
1315 {
1316 struct send_range *next = bqueue_dequeue(bq);
1317 ASSERT3S(send_range_after(prev, next), ==, -1);
1318 return (next);
1319 }
1320
1321 /*
1322 * Pop the new data off the queue, check that the records we receive are in
1323 * the right order, and free the old data.
1324 */
1325 static struct send_range *
1326 get_next_range(bqueue_t *bq, struct send_range *prev)
1327 {
1328 struct send_range *next = get_next_range_nofree(bq, prev);
1329 range_free(prev);
1330 return (next);
1331 }
1332
1333 static __attribute__((noreturn)) void
1334 redact_list_thread(void *arg)
1335 {
1336 struct redact_list_thread_arg *rlt_arg = arg;
1337 struct send_range *record;
1338 fstrans_cookie_t cookie = spl_fstrans_mark();
1339 if (rlt_arg->rl != NULL) {
1340 struct redact_list_cb_arg rlcba = {0};
1341 rlcba.cancel = &rlt_arg->cancel;
1342 rlcba.q = &rlt_arg->q;
1343 rlcba.num_blocks_visited = rlt_arg->num_blocks_visited;
1344 rlcba.mark_redact = rlt_arg->mark_redact;
1345 int err = dsl_redaction_list_traverse(rlt_arg->rl,
1346 &rlt_arg->resume, redact_list_cb, &rlcba);
1347 if (err != EINTR)
1348 rlt_arg->error_code = err;
1349 }
1350 record = range_alloc(DATA, 0, 0, 0, B_TRUE);
1351 bqueue_enqueue_flush(&rlt_arg->q, record, sizeof (*record));
1352 spl_fstrans_unmark(cookie);
1353
1354 thread_exit();
1355 }
1356
1357 /*
1358 * Compare the start point of the two provided ranges. End of stream ranges
1359 * compare last, objects compare before any data or hole inside that object and
1360 * multi-object holes that start at the same object.
1361 */
1362 static int
1363 send_range_start_compare(struct send_range *r1, struct send_range *r2)
1364 {
1365 uint64_t r1_objequiv = r1->object;
1366 uint64_t r1_l0equiv = r1->start_blkid;
1367 uint64_t r2_objequiv = r2->object;
1368 uint64_t r2_l0equiv = r2->start_blkid;
1369 int64_t cmp = TREE_CMP(r1->eos_marker, r2->eos_marker);
1370 if (unlikely(cmp))
1371 return (cmp);
1372 if (r1->object == 0) {
1373 r1_objequiv = r1->start_blkid * DNODES_PER_BLOCK;
1374 r1_l0equiv = 0;
1375 }
1376 if (r2->object == 0) {
1377 r2_objequiv = r2->start_blkid * DNODES_PER_BLOCK;
1378 r2_l0equiv = 0;
1379 }
1380
1381 cmp = TREE_CMP(r1_objequiv, r2_objequiv);
1382 if (likely(cmp))
1383 return (cmp);
1384 cmp = TREE_CMP(r2->type == OBJECT_RANGE, r1->type == OBJECT_RANGE);
1385 if (unlikely(cmp))
1386 return (cmp);
1387 cmp = TREE_CMP(r2->type == OBJECT, r1->type == OBJECT);
1388 if (unlikely(cmp))
1389 return (cmp);
1390
1391 return (TREE_CMP(r1_l0equiv, r2_l0equiv));
1392 }
1393
1394 enum q_idx {
1395 REDACT_IDX = 0,
1396 TO_IDX,
1397 FROM_IDX,
1398 NUM_THREADS
1399 };
1400
1401 /*
1402 * This function returns the next range the send_merge_thread should operate on.
1403 * The inputs are two arrays; the first one stores the range at the front of the
1404 * queues stored in the second one. The ranges are sorted in descending
1405 * priority order; the metadata from earlier ranges overrules metadata from
1406 * later ranges. out_mask is used to return which threads the ranges came from;
1407 * bit i is set if ranges[i] started at the same place as the returned range.
1408 *
1409 * This code is not hardcoded to compare a specific number of threads; it could
1410 * be used with any number, just by changing the q_idx enum.
1411 *
1412 * The "next range" is the one with the earliest start; if two starts are equal,
1413 * the highest-priority range is the next to operate on. If a higher-priority
1414 * range starts in the middle of the first range, then the first range will be
1415 * truncated to end where the higher-priority range starts, and we will operate
1416 * on that one next time. In this way, we make sure that each block covered by
1417 * some range gets covered by a returned range, and each block covered is
1418 * returned using the metadata of the highest-priority range it appears in.
1419 *
1420 * For example, if the three ranges at the front of the queues were [2,4),
1421 * [3,5), and [1,3), then the ranges returned would be [1,2) with the metadata
1422 * from the third range, [2,4) with the metadata from the first range, and then
1423 * [4,5) with the metadata from the second.
1424 */
1425 static struct send_range *
1426 find_next_range(struct send_range **ranges, bqueue_t **qs, uint64_t *out_mask)
1427 {
1428 int idx = 0; // index of the range with the earliest start
1429 int i;
1430 uint64_t bmask = 0;
1431 for (i = 1; i < NUM_THREADS; i++) {
1432 if (send_range_start_compare(ranges[i], ranges[idx]) < 0)
1433 idx = i;
1434 }
1435 if (ranges[idx]->eos_marker) {
1436 struct send_range *ret = range_alloc(DATA, 0, 0, 0, B_TRUE);
1437 *out_mask = 0;
1438 return (ret);
1439 }
1440 /*
1441 * Find all the ranges that start at that same point.
1442 */
1443 for (i = 0; i < NUM_THREADS; i++) {
1444 if (send_range_start_compare(ranges[i], ranges[idx]) == 0)
1445 bmask |= 1 << i;
1446 }
1447 *out_mask = bmask;
1448 /*
1449 * OBJECT_RANGE records only come from the TO thread, and should always
1450 * be treated as overlapping with nothing and sent on immediately. They
1451 * are only used in raw sends, and are never redacted.
1452 */
1453 if (ranges[idx]->type == OBJECT_RANGE) {
1454 ASSERT3U(idx, ==, TO_IDX);
1455 ASSERT3U(*out_mask, ==, 1 << TO_IDX);
1456 struct send_range *ret = ranges[idx];
1457 ranges[idx] = get_next_range_nofree(qs[idx], ranges[idx]);
1458 return (ret);
1459 }
1460 /*
1461 * Find the first start or end point after the start of the first range.
1462 */
1463 uint64_t first_change = ranges[idx]->end_blkid;
1464 for (i = 0; i < NUM_THREADS; i++) {
1465 if (i == idx || ranges[i]->eos_marker ||
1466 ranges[i]->object > ranges[idx]->object ||
1467 ranges[i]->object == DMU_META_DNODE_OBJECT)
1468 continue;
1469 ASSERT3U(ranges[i]->object, ==, ranges[idx]->object);
1470 if (first_change > ranges[i]->start_blkid &&
1471 (bmask & (1 << i)) == 0)
1472 first_change = ranges[i]->start_blkid;
1473 else if (first_change > ranges[i]->end_blkid)
1474 first_change = ranges[i]->end_blkid;
1475 }
1476 /*
1477 * Update all ranges to no longer overlap with the range we're
1478 * returning. All such ranges must start at the same place as the range
1479 * being returned, and end at or after first_change. Thus we update
1480 * their start to first_change. If that makes them size 0, then free
1481 * them and pull a new range from that thread.
1482 */
1483 for (i = 0; i < NUM_THREADS; i++) {
1484 if (i == idx || (bmask & (1 << i)) == 0)
1485 continue;
1486 ASSERT3U(first_change, >, ranges[i]->start_blkid);
1487 ranges[i]->start_blkid = first_change;
1488 ASSERT3U(ranges[i]->start_blkid, <=, ranges[i]->end_blkid);
1489 if (ranges[i]->start_blkid == ranges[i]->end_blkid)
1490 ranges[i] = get_next_range(qs[i], ranges[i]);
1491 }
1492 /*
1493 * Short-circuit the simple case; if the range doesn't overlap with
1494 * anything else, or it only overlaps with things that start at the same
1495 * place and are longer, send it on.
1496 */
1497 if (first_change == ranges[idx]->end_blkid) {
1498 struct send_range *ret = ranges[idx];
1499 ranges[idx] = get_next_range_nofree(qs[idx], ranges[idx]);
1500 return (ret);
1501 }
1502
1503 /*
1504 * Otherwise, return a truncated copy of ranges[idx] and move the start
1505 * of ranges[idx] back to first_change.
1506 */
1507 struct send_range *ret = kmem_alloc(sizeof (*ret), KM_SLEEP);
1508 *ret = *ranges[idx];
1509 ret->end_blkid = first_change;
1510 ranges[idx]->start_blkid = first_change;
1511 return (ret);
1512 }
1513
1514 #define FROM_AND_REDACT_BITS ((1 << REDACT_IDX) | (1 << FROM_IDX))
1515
1516 /*
1517 * Merge the results from the from thread and the to thread, and then hand the
1518 * records off to send_prefetch_thread to prefetch them. If this is not a
1519 * send from a redaction bookmark, the from thread will push an end of stream
1520 * record and stop, and we'll just send everything that was changed in the
1521 * to_ds since the ancestor's creation txg. If it is, then since
1522 * traverse_dataset has a canonical order, we can compare each change as
1523 * they're pulled off the queues. That will give us a stream that is
1524 * appropriately sorted, and covers all records. In addition, we pull the
1525 * data from the redact_list_thread and use that to determine which blocks
1526 * should be redacted.
1527 */
1528 static __attribute__((noreturn)) void
1529 send_merge_thread(void *arg)
1530 {
1531 struct send_merge_thread_arg *smt_arg = arg;
1532 struct send_range *front_ranges[NUM_THREADS];
1533 bqueue_t *queues[NUM_THREADS];
1534 int err = 0;
1535 fstrans_cookie_t cookie = spl_fstrans_mark();
1536
1537 if (smt_arg->redact_arg == NULL) {
1538 front_ranges[REDACT_IDX] =
1539 kmem_zalloc(sizeof (struct send_range), KM_SLEEP);
1540 front_ranges[REDACT_IDX]->eos_marker = B_TRUE;
1541 front_ranges[REDACT_IDX]->type = REDACT;
1542 queues[REDACT_IDX] = NULL;
1543 } else {
1544 front_ranges[REDACT_IDX] =
1545 bqueue_dequeue(&smt_arg->redact_arg->q);
1546 queues[REDACT_IDX] = &smt_arg->redact_arg->q;
1547 }
1548 front_ranges[TO_IDX] = bqueue_dequeue(&smt_arg->to_arg->q);
1549 queues[TO_IDX] = &smt_arg->to_arg->q;
1550 front_ranges[FROM_IDX] = bqueue_dequeue(&smt_arg->from_arg->q);
1551 queues[FROM_IDX] = &smt_arg->from_arg->q;
1552 uint64_t mask = 0;
1553 struct send_range *range;
1554 for (range = find_next_range(front_ranges, queues, &mask);
1555 !range->eos_marker && err == 0 && !smt_arg->cancel;
1556 range = find_next_range(front_ranges, queues, &mask)) {
1557 /*
1558 * If the range in question was in both the from redact bookmark
1559 * and the bookmark we're using to redact, then don't send it.
1560 * It's already redacted on the receiving system, so a redaction
1561 * record would be redundant.
1562 */
1563 if ((mask & FROM_AND_REDACT_BITS) == FROM_AND_REDACT_BITS) {
1564 ASSERT3U(range->type, ==, REDACT);
1565 range_free(range);
1566 continue;
1567 }
1568 bqueue_enqueue(&smt_arg->q, range, sizeof (*range));
1569
1570 if (smt_arg->to_arg->error_code != 0) {
1571 err = smt_arg->to_arg->error_code;
1572 } else if (smt_arg->from_arg->error_code != 0) {
1573 err = smt_arg->from_arg->error_code;
1574 } else if (smt_arg->redact_arg != NULL &&
1575 smt_arg->redact_arg->error_code != 0) {
1576 err = smt_arg->redact_arg->error_code;
1577 }
1578 }
1579 if (smt_arg->cancel && err == 0)
1580 err = SET_ERROR(EINTR);
1581 smt_arg->error = err;
1582 if (smt_arg->error != 0) {
1583 smt_arg->to_arg->cancel = B_TRUE;
1584 smt_arg->from_arg->cancel = B_TRUE;
1585 if (smt_arg->redact_arg != NULL)
1586 smt_arg->redact_arg->cancel = B_TRUE;
1587 }
1588 for (int i = 0; i < NUM_THREADS; i++) {
1589 while (!front_ranges[i]->eos_marker) {
1590 front_ranges[i] = get_next_range(queues[i],
1591 front_ranges[i]);
1592 }
1593 range_free(front_ranges[i]);
1594 }
1595 range->eos_marker = B_TRUE;
1596 bqueue_enqueue_flush(&smt_arg->q, range, 1);
1597 spl_fstrans_unmark(cookie);
1598 thread_exit();
1599 }
1600
1601 struct send_reader_thread_arg {
1602 struct send_merge_thread_arg *smta;
1603 bqueue_t q;
1604 boolean_t cancel;
1605 boolean_t issue_reads;
1606 uint64_t featureflags;
1607 int error;
1608 };
1609
1610 static void
1611 dmu_send_read_done(zio_t *zio)
1612 {
1613 struct send_range *range = zio->io_private;
1614
1615 mutex_enter(&range->sru.data.lock);
1616 if (zio->io_error != 0) {
1617 abd_free(range->sru.data.abd);
1618 range->sru.data.abd = NULL;
1619 range->sru.data.io_err = zio->io_error;
1620 }
1621
1622 ASSERT(range->sru.data.io_outstanding);
1623 range->sru.data.io_outstanding = B_FALSE;
1624 cv_broadcast(&range->sru.data.cv);
1625 mutex_exit(&range->sru.data.lock);
1626 }
1627
1628 static void
1629 issue_data_read(struct send_reader_thread_arg *srta, struct send_range *range)
1630 {
1631 struct srd *srdp = &range->sru.data;
1632 blkptr_t *bp = &srdp->bp;
1633 objset_t *os = srta->smta->os;
1634
1635 ASSERT3U(range->type, ==, DATA);
1636 ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
1637 /*
1638 * If we have large blocks stored on disk but
1639 * the send flags don't allow us to send large
1640 * blocks, we split the data from the arc buf
1641 * into chunks.
1642 */
1643 boolean_t split_large_blocks =
1644 srdp->datablksz > SPA_OLD_MAXBLOCKSIZE &&
1645 !(srta->featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS);
1646 /*
1647 * We should only request compressed data from the ARC if all
1648 * the following are true:
1649 * - stream compression was requested
1650 * - we aren't splitting large blocks into smaller chunks
1651 * - the data won't need to be byteswapped before sending
1652 * - this isn't an embedded block
1653 * - this isn't metadata (if receiving on a different endian
1654 * system it can be byteswapped more easily)
1655 */
1656 boolean_t request_compressed =
1657 (srta->featureflags & DMU_BACKUP_FEATURE_COMPRESSED) &&
1658 !split_large_blocks && !BP_SHOULD_BYTESWAP(bp) &&
1659 !BP_IS_EMBEDDED(bp) && !DMU_OT_IS_METADATA(BP_GET_TYPE(bp));
1660
1661 zio_flag_t zioflags = ZIO_FLAG_CANFAIL;
1662
1663 if (srta->featureflags & DMU_BACKUP_FEATURE_RAW) {
1664 zioflags |= ZIO_FLAG_RAW;
1665 srdp->io_compressed = B_TRUE;
1666 } else if (request_compressed) {
1667 zioflags |= ZIO_FLAG_RAW_COMPRESS;
1668 srdp->io_compressed = B_TRUE;
1669 }
1670
1671 srdp->datasz = (zioflags & ZIO_FLAG_RAW_COMPRESS) ?
1672 BP_GET_PSIZE(bp) : BP_GET_LSIZE(bp);
1673
1674 if (!srta->issue_reads)
1675 return;
1676 if (BP_IS_REDACTED(bp))
1677 return;
1678 if (send_do_embed(bp, srta->featureflags))
1679 return;
1680
1681 zbookmark_phys_t zb = {
1682 .zb_objset = dmu_objset_id(os),
1683 .zb_object = range->object,
1684 .zb_level = 0,
1685 .zb_blkid = range->start_blkid,
1686 };
1687
1688 arc_flags_t aflags = ARC_FLAG_CACHED_ONLY;
1689
1690 int arc_err = arc_read(NULL, os->os_spa, bp,
1691 arc_getbuf_func, &srdp->abuf, ZIO_PRIORITY_ASYNC_READ,
1692 zioflags, &aflags, &zb);
1693 /*
1694 * If the data is not already cached in the ARC, we read directly
1695 * from zio. This avoids the performance overhead of adding a new
1696 * entry to the ARC, and we also avoid polluting the ARC cache with
1697 * data that is not likely to be used in the future.
1698 */
1699 if (arc_err != 0) {
1700 srdp->abd = abd_alloc_linear(srdp->datasz, B_FALSE);
1701 srdp->io_outstanding = B_TRUE;
1702 zio_nowait(zio_read(NULL, os->os_spa, bp, srdp->abd,
1703 srdp->datasz, dmu_send_read_done, range,
1704 ZIO_PRIORITY_ASYNC_READ, zioflags, &zb));
1705 }
1706 }
1707
1708 /*
1709 * Create a new record with the given values.
1710 */
1711 static void
1712 enqueue_range(struct send_reader_thread_arg *srta, bqueue_t *q, dnode_t *dn,
1713 uint64_t blkid, uint64_t count, const blkptr_t *bp, uint32_t datablksz)
1714 {
1715 enum type range_type = (bp == NULL || BP_IS_HOLE(bp) ? HOLE :
1716 (BP_IS_REDACTED(bp) ? REDACT : DATA));
1717
1718 struct send_range *range = range_alloc(range_type, dn->dn_object,
1719 blkid, blkid + count, B_FALSE);
1720
1721 if (blkid == DMU_SPILL_BLKID) {
1722 ASSERT3P(bp, !=, NULL);
1723 ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);
1724 }
1725
1726 switch (range_type) {
1727 case HOLE:
1728 range->sru.hole.datablksz = datablksz;
1729 break;
1730 case DATA:
1731 ASSERT3U(count, ==, 1);
1732 range->sru.data.datablksz = datablksz;
1733 range->sru.data.obj_type = dn->dn_type;
1734 range->sru.data.bp = *bp;
1735 issue_data_read(srta, range);
1736 break;
1737 case REDACT:
1738 range->sru.redact.datablksz = datablksz;
1739 break;
1740 default:
1741 break;
1742 }
1743 bqueue_enqueue(q, range, datablksz);
1744 }
1745
1746 /*
1747 * This thread is responsible for two things: First, it retrieves the correct
1748 * blkptr in the to ds if we need to send the data because of something from
1749 * the from thread. As a result of this, we're the first ones to discover that
1750 * some indirect blocks can be discarded because they're not holes. Second,
1751 * it issues prefetches for the data we need to send.
1752 */
1753 static __attribute__((noreturn)) void
1754 send_reader_thread(void *arg)
1755 {
1756 struct send_reader_thread_arg *srta = arg;
1757 struct send_merge_thread_arg *smta = srta->smta;
1758 bqueue_t *inq = &smta->q;
1759 bqueue_t *outq = &srta->q;
1760 objset_t *os = smta->os;
1761 fstrans_cookie_t cookie = spl_fstrans_mark();
1762 struct send_range *range = bqueue_dequeue(inq);
1763 int err = 0;
1764
1765 /*
1766 * If the record we're analyzing is from a redaction bookmark from the
1767 * fromds, then we need to know whether or not it exists in the tods so
1768 * we know whether to create records for it or not. If it does, we need
1769 * the datablksz so we can generate an appropriate record for it.
1770 * Finally, if it isn't redacted, we need the blkptr so that we can send
1771 * a WRITE record containing the actual data.
1772 */
1773 uint64_t last_obj = UINT64_MAX;
1774 uint64_t last_obj_exists = B_TRUE;
1775 while (!range->eos_marker && !srta->cancel && smta->error == 0 &&
1776 err == 0) {
1777 switch (range->type) {
1778 case DATA:
1779 issue_data_read(srta, range);
1780 bqueue_enqueue(outq, range, range->sru.data.datablksz);
1781 range = get_next_range_nofree(inq, range);
1782 break;
1783 case HOLE:
1784 case OBJECT:
1785 case OBJECT_RANGE:
1786 case REDACT: // Redacted blocks must exist
1787 bqueue_enqueue(outq, range, sizeof (*range));
1788 range = get_next_range_nofree(inq, range);
1789 break;
1790 case PREVIOUSLY_REDACTED: {
1791 /*
1792 * This entry came from the "from bookmark" when
1793 * sending from a bookmark that has a redaction
1794 * list. We need to check if this object/blkid
1795 * exists in the target ("to") dataset, and if
1796 * not then we drop this entry. We also need
1797 * to fill in the block pointer so that we know
1798 * what to prefetch.
1799 *
1800 * To accomplish the above, we first cache whether or
1801 * not the last object we examined exists. If it
1802 * doesn't, we can drop this record. If it does, we hold
1803 * the dnode and use it to call dbuf_dnode_findbp. We do
1804 * this instead of dbuf_bookmark_findbp because we will
1805 * often operate on large ranges, and holding the dnode
1806 * once is more efficient.
1807 */
1808 boolean_t object_exists = B_TRUE;
1809 /*
1810 * If the data is redacted, we only care if it exists,
1811 * so that we don't send records for objects that have
1812 * been deleted.
1813 */
1814 dnode_t *dn;
1815 if (range->object == last_obj && !last_obj_exists) {
1816 /*
1817 * If we're still examining the same object as
1818 * previously, and it doesn't exist, we don't
1819 * need to call dbuf_bookmark_findbp.
1820 */
1821 object_exists = B_FALSE;
1822 } else {
1823 err = dnode_hold(os, range->object, FTAG, &dn);
1824 if (err == ENOENT) {
1825 object_exists = B_FALSE;
1826 err = 0;
1827 }
1828 last_obj = range->object;
1829 last_obj_exists = object_exists;
1830 }
1831
1832 if (err != 0) {
1833 break;
1834 } else if (!object_exists) {
1835 /*
1836 * The block was modified, but doesn't
1837 * exist in the to dataset; if it was
1838 * deleted in the to dataset, then we'll
1839 * visit the hole bp for it at some point.
1840 */
1841 range = get_next_range(inq, range);
1842 continue;
1843 }
1844 uint64_t file_max =
1845 MIN(dn->dn_maxblkid, range->end_blkid);
1846 /*
1847 * The object exists, so we need to try to find the
1848 * blkptr for each block in the range we're processing.
1849 */
1850 rw_enter(&dn->dn_struct_rwlock, RW_READER);
1851 for (uint64_t blkid = range->start_blkid;
1852 blkid < file_max; blkid++) {
1853 blkptr_t bp;
1854 uint32_t datablksz =
1855 dn->dn_phys->dn_datablkszsec <<
1856 SPA_MINBLOCKSHIFT;
1857 uint64_t offset = blkid * datablksz;
1858 /*
1859 * This call finds the next non-hole block in
1860 * the object. This is to prevent a
1861 * performance problem where we're unredacting
1862 * a large hole. Using dnode_next_offset to
1863 * skip over the large hole avoids iterating
1864 * over every block in it.
1865 */
1866 err = dnode_next_offset(dn, DNODE_FIND_HAVELOCK,
1867 &offset, 1, 1, 0);
1868 if (err == ESRCH) {
1869 offset = UINT64_MAX;
1870 err = 0;
1871 } else if (err != 0) {
1872 break;
1873 }
1874 if (offset != blkid * datablksz) {
1875 /*
1876 * if there is a hole from here
1877 * (blkid) to offset
1878 */
1879 offset = MIN(offset, file_max *
1880 datablksz);
1881 uint64_t nblks = (offset / datablksz) -
1882 blkid;
1883 enqueue_range(srta, outq, dn, blkid,
1884 nblks, NULL, datablksz);
1885 blkid += nblks;
1886 }
1887 if (blkid >= file_max)
1888 break;
1889 err = dbuf_dnode_findbp(dn, 0, blkid, &bp,
1890 NULL, NULL);
1891 if (err != 0)
1892 break;
1893 ASSERT(!BP_IS_HOLE(&bp));
1894 enqueue_range(srta, outq, dn, blkid, 1, &bp,
1895 datablksz);
1896 }
1897 rw_exit(&dn->dn_struct_rwlock);
1898 dnode_rele(dn, FTAG);
1899 range = get_next_range(inq, range);
1900 }
1901 }
1902 }
1903 if (srta->cancel || err != 0) {
1904 smta->cancel = B_TRUE;
1905 srta->error = err;
1906 } else if (smta->error != 0) {
1907 srta->error = smta->error;
1908 }
1909 while (!range->eos_marker)
1910 range = get_next_range(inq, range);
1911
1912 bqueue_enqueue_flush(outq, range, 1);
1913 spl_fstrans_unmark(cookie);
1914 thread_exit();
1915 }
1916
1917 #define NUM_SNAPS_NOT_REDACTED UINT64_MAX
1918
1919 struct dmu_send_params {
1920 /* Pool args */
1921 const void *tag; // Tag dp was held with, will be used to release dp.
1922 dsl_pool_t *dp;
1923 /* To snapshot args */
1924 const char *tosnap;
1925 dsl_dataset_t *to_ds;
1926 /* From snapshot args */
1927 zfs_bookmark_phys_t ancestor_zb;
1928 uint64_t *fromredactsnaps;
1929 /* NUM_SNAPS_NOT_REDACTED if not sending from redaction bookmark */
1930 uint64_t numfromredactsnaps;
1931 /* Stream params */
1932 boolean_t is_clone;
1933 boolean_t embedok;
1934 boolean_t large_block_ok;
1935 boolean_t compressok;
1936 boolean_t rawok;
1937 boolean_t savedok;
1938 uint64_t resumeobj;
1939 uint64_t resumeoff;
1940 uint64_t saved_guid;
1941 zfs_bookmark_phys_t *redactbook;
1942 /* Stream output params */
1943 dmu_send_outparams_t *dso;
1944
1945 /* Stream progress params */
1946 offset_t *off;
1947 int outfd;
1948 char saved_toname[MAXNAMELEN];
1949 };
1950
1951 static int
1952 setup_featureflags(struct dmu_send_params *dspp, objset_t *os,
1953 uint64_t *featureflags)
1954 {
1955 dsl_dataset_t *to_ds = dspp->to_ds;
1956 dsl_pool_t *dp = dspp->dp;
1957 #ifdef _KERNEL
1958 if (dmu_objset_type(os) == DMU_OST_ZFS) {
1959 uint64_t version;
1960 if (zfs_get_zplprop(os, ZFS_PROP_VERSION, &version) != 0)
1961 return (SET_ERROR(EINVAL));
1962
1963 if (version >= ZPL_VERSION_SA)
1964 *featureflags |= DMU_BACKUP_FEATURE_SA_SPILL;
1965 }
1966 #endif
1967
1968 /* raw sends imply large_block_ok */
1969 if ((dspp->rawok || dspp->large_block_ok) &&
1970 dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_BLOCKS)) {
1971 *featureflags |= DMU_BACKUP_FEATURE_LARGE_BLOCKS;
1972 }
1973
1974 /* encrypted datasets will not have embedded blocks */
1975 if ((dspp->embedok || dspp->rawok) && !os->os_encrypted &&
1976 spa_feature_is_active(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA)) {
1977 *featureflags |= DMU_BACKUP_FEATURE_EMBED_DATA;
1978 }
1979
1980 /* raw send implies compressok */
1981 if (dspp->compressok || dspp->rawok)
1982 *featureflags |= DMU_BACKUP_FEATURE_COMPRESSED;
1983
1984 if (dspp->rawok && os->os_encrypted)
1985 *featureflags |= DMU_BACKUP_FEATURE_RAW;
1986
1987 if ((*featureflags &
1988 (DMU_BACKUP_FEATURE_EMBED_DATA | DMU_BACKUP_FEATURE_COMPRESSED |
1989 DMU_BACKUP_FEATURE_RAW)) != 0 &&
1990 spa_feature_is_active(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS)) {
1991 *featureflags |= DMU_BACKUP_FEATURE_LZ4;
1992 }
1993
1994 /*
1995 * We specifically do not include DMU_BACKUP_FEATURE_EMBED_DATA here to
1996 * allow sending ZSTD compressed datasets to a receiver that does not
1997 * support ZSTD
1998 */
1999 if ((*featureflags &
2000 (DMU_BACKUP_FEATURE_COMPRESSED | DMU_BACKUP_FEATURE_RAW)) != 0 &&
2001 dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_ZSTD_COMPRESS)) {
2002 *featureflags |= DMU_BACKUP_FEATURE_ZSTD;
2003 }
2004
2005 if (dspp->resumeobj != 0 || dspp->resumeoff != 0) {
2006 *featureflags |= DMU_BACKUP_FEATURE_RESUMING;
2007 }
2008
2009 if (dspp->redactbook != NULL) {
2010 *featureflags |= DMU_BACKUP_FEATURE_REDACTED;
2011 }
2012
2013 if (dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_DNODE)) {
2014 *featureflags |= DMU_BACKUP_FEATURE_LARGE_DNODE;
2015 }
2016 return (0);
2017 }
2018
2019 static dmu_replay_record_t *
2020 create_begin_record(struct dmu_send_params *dspp, objset_t *os,
2021 uint64_t featureflags)
2022 {
2023 dmu_replay_record_t *drr = kmem_zalloc(sizeof (dmu_replay_record_t),
2024 KM_SLEEP);
2025 drr->drr_type = DRR_BEGIN;
2026
2027 struct drr_begin *drrb = &drr->drr_u.drr_begin;
2028 dsl_dataset_t *to_ds = dspp->to_ds;
2029
2030 drrb->drr_magic = DMU_BACKUP_MAGIC;
2031 drrb->drr_creation_time = dsl_dataset_phys(to_ds)->ds_creation_time;
2032 drrb->drr_type = dmu_objset_type(os);
2033 drrb->drr_toguid = dsl_dataset_phys(to_ds)->ds_guid;
2034 drrb->drr_fromguid = dspp->ancestor_zb.zbm_guid;
2035
2036 DMU_SET_STREAM_HDRTYPE(drrb->drr_versioninfo, DMU_SUBSTREAM);
2037 DMU_SET_FEATUREFLAGS(drrb->drr_versioninfo, featureflags);
2038
2039 if (dspp->is_clone)
2040 drrb->drr_flags |= DRR_FLAG_CLONE;
2041 if (dsl_dataset_phys(dspp->to_ds)->ds_flags & DS_FLAG_CI_DATASET)
2042 drrb->drr_flags |= DRR_FLAG_CI_DATA;
2043 if (zfs_send_set_freerecords_bit)
2044 drrb->drr_flags |= DRR_FLAG_FREERECORDS;
2045 drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_SPILL_BLOCK;
2046
2047 if (dspp->savedok) {
2048 drrb->drr_toguid = dspp->saved_guid;
2049 strlcpy(drrb->drr_toname, dspp->saved_toname,
2050 sizeof (drrb->drr_toname));
2051 } else {
2052 dsl_dataset_name(to_ds, drrb->drr_toname);
2053 if (!to_ds->ds_is_snapshot) {
2054 (void) strlcat(drrb->drr_toname, "@--head--",
2055 sizeof (drrb->drr_toname));
2056 }
2057 }
2058 return (drr);
2059 }
2060
2061 static void
2062 setup_to_thread(struct send_thread_arg *to_arg, objset_t *to_os,
2063 dmu_sendstatus_t *dssp, uint64_t fromtxg, boolean_t rawok)
2064 {
2065 VERIFY0(bqueue_init(&to_arg->q, zfs_send_no_prefetch_queue_ff,
2066 MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2067 offsetof(struct send_range, ln)));
2068 to_arg->error_code = 0;
2069 to_arg->cancel = B_FALSE;
2070 to_arg->os = to_os;
2071 to_arg->fromtxg = fromtxg;
2072 to_arg->flags = TRAVERSE_PRE | TRAVERSE_PREFETCH_METADATA;
2073 if (rawok)
2074 to_arg->flags |= TRAVERSE_NO_DECRYPT;
2075 if (zfs_send_corrupt_data)
2076 to_arg->flags |= TRAVERSE_HARD;
2077 to_arg->num_blocks_visited = &dssp->dss_blocks;
2078 (void) thread_create(NULL, 0, send_traverse_thread, to_arg, 0,
2079 curproc, TS_RUN, minclsyspri);
2080 }
2081
2082 static void
2083 setup_from_thread(struct redact_list_thread_arg *from_arg,
2084 redaction_list_t *from_rl, dmu_sendstatus_t *dssp)
2085 {
2086 VERIFY0(bqueue_init(&from_arg->q, zfs_send_no_prefetch_queue_ff,
2087 MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2088 offsetof(struct send_range, ln)));
2089 from_arg->error_code = 0;
2090 from_arg->cancel = B_FALSE;
2091 from_arg->rl = from_rl;
2092 from_arg->mark_redact = B_FALSE;
2093 from_arg->num_blocks_visited = &dssp->dss_blocks;
2094 /*
2095 * If from_ds is null, send_traverse_thread just returns success and
2096 * enqueues an eos marker.
2097 */
2098 (void) thread_create(NULL, 0, redact_list_thread, from_arg, 0,
2099 curproc, TS_RUN, minclsyspri);
2100 }
2101
2102 static void
2103 setup_redact_list_thread(struct redact_list_thread_arg *rlt_arg,
2104 struct dmu_send_params *dspp, redaction_list_t *rl, dmu_sendstatus_t *dssp)
2105 {
2106 if (dspp->redactbook == NULL)
2107 return;
2108
2109 rlt_arg->cancel = B_FALSE;
2110 VERIFY0(bqueue_init(&rlt_arg->q, zfs_send_no_prefetch_queue_ff,
2111 MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2112 offsetof(struct send_range, ln)));
2113 rlt_arg->error_code = 0;
2114 rlt_arg->mark_redact = B_TRUE;
2115 rlt_arg->rl = rl;
2116 rlt_arg->num_blocks_visited = &dssp->dss_blocks;
2117
2118 (void) thread_create(NULL, 0, redact_list_thread, rlt_arg, 0,
2119 curproc, TS_RUN, minclsyspri);
2120 }
2121
2122 static void
2123 setup_merge_thread(struct send_merge_thread_arg *smt_arg,
2124 struct dmu_send_params *dspp, struct redact_list_thread_arg *from_arg,
2125 struct send_thread_arg *to_arg, struct redact_list_thread_arg *rlt_arg,
2126 objset_t *os)
2127 {
2128 VERIFY0(bqueue_init(&smt_arg->q, zfs_send_no_prefetch_queue_ff,
2129 MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2130 offsetof(struct send_range, ln)));
2131 smt_arg->cancel = B_FALSE;
2132 smt_arg->error = 0;
2133 smt_arg->from_arg = from_arg;
2134 smt_arg->to_arg = to_arg;
2135 if (dspp->redactbook != NULL)
2136 smt_arg->redact_arg = rlt_arg;
2137
2138 smt_arg->os = os;
2139 (void) thread_create(NULL, 0, send_merge_thread, smt_arg, 0, curproc,
2140 TS_RUN, minclsyspri);
2141 }
2142
2143 static void
2144 setup_reader_thread(struct send_reader_thread_arg *srt_arg,
2145 struct dmu_send_params *dspp, struct send_merge_thread_arg *smt_arg,
2146 uint64_t featureflags)
2147 {
2148 VERIFY0(bqueue_init(&srt_arg->q, zfs_send_queue_ff,
2149 MAX(zfs_send_queue_length, 2 * zfs_max_recordsize),
2150 offsetof(struct send_range, ln)));
2151 srt_arg->smta = smt_arg;
2152 srt_arg->issue_reads = !dspp->dso->dso_dryrun;
2153 srt_arg->featureflags = featureflags;
2154 (void) thread_create(NULL, 0, send_reader_thread, srt_arg, 0,
2155 curproc, TS_RUN, minclsyspri);
2156 }
2157
2158 static int
2159 setup_resume_points(struct dmu_send_params *dspp,
2160 struct send_thread_arg *to_arg, struct redact_list_thread_arg *from_arg,
2161 struct redact_list_thread_arg *rlt_arg,
2162 struct send_merge_thread_arg *smt_arg, boolean_t resuming, objset_t *os,
2163 redaction_list_t *redact_rl, nvlist_t *nvl)
2164 {
2165 (void) smt_arg;
2166 dsl_dataset_t *to_ds = dspp->to_ds;
2167 int err = 0;
2168
2169 uint64_t obj = 0;
2170 uint64_t blkid = 0;
2171 if (resuming) {
2172 obj = dspp->resumeobj;
2173 dmu_object_info_t to_doi;
2174 err = dmu_object_info(os, obj, &to_doi);
2175 if (err != 0)
2176 return (err);
2177
2178 blkid = dspp->resumeoff / to_doi.doi_data_block_size;
2179 }
2180 /*
2181 * If we're resuming a redacted send, we can skip to the appropriate
2182 * point in the redaction bookmark by binary searching through it.
2183 */
2184 if (redact_rl != NULL) {
2185 SET_BOOKMARK(&rlt_arg->resume, to_ds->ds_object, obj, 0, blkid);
2186 }
2187
2188 SET_BOOKMARK(&to_arg->resume, to_ds->ds_object, obj, 0, blkid);
2189 if (nvlist_exists(nvl, BEGINNV_REDACT_FROM_SNAPS)) {
2190 uint64_t objset = dspp->ancestor_zb.zbm_redaction_obj;
2191 /*
2192 * Note: If the resume point is in an object whose
2193 * blocksize is different in the from vs to snapshots,
2194 * we will have divided by the "wrong" blocksize.
2195 * However, in this case fromsnap's send_cb() will
2196 * detect that the blocksize has changed and therefore
2197 * ignore this object.
2198 *
2199 * If we're resuming a send from a redaction bookmark,
2200 * we still cannot accidentally suggest blocks behind
2201 * the to_ds. In addition, we know that any blocks in
2202 * the object in the to_ds will have to be sent, since
2203 * the size changed. Therefore, we can't cause any harm
2204 * this way either.
2205 */
2206 SET_BOOKMARK(&from_arg->resume, objset, obj, 0, blkid);
2207 }
2208 if (resuming) {
2209 fnvlist_add_uint64(nvl, BEGINNV_RESUME_OBJECT, dspp->resumeobj);
2210 fnvlist_add_uint64(nvl, BEGINNV_RESUME_OFFSET, dspp->resumeoff);
2211 }
2212 return (0);
2213 }
2214
2215 static dmu_sendstatus_t *
2216 setup_send_progress(struct dmu_send_params *dspp)
2217 {
2218 dmu_sendstatus_t *dssp = kmem_zalloc(sizeof (*dssp), KM_SLEEP);
2219 dssp->dss_outfd = dspp->outfd;
2220 dssp->dss_off = dspp->off;
2221 dssp->dss_proc = curproc;
2222 mutex_enter(&dspp->to_ds->ds_sendstream_lock);
2223 list_insert_head(&dspp->to_ds->ds_sendstreams, dssp);
2224 mutex_exit(&dspp->to_ds->ds_sendstream_lock);
2225 return (dssp);
2226 }
2227
2228 /*
2229 * Actually do the bulk of the work in a zfs send.
2230 *
2231 * The idea is that we want to do a send from ancestor_zb to to_ds. We also
2232 * want to not send any data that has been modified by all the datasets in
2233 * redactsnaparr, and store the list of blocks that are redacted in this way in
2234 * a bookmark named redactbook, created on the to_ds. We do this by creating
2235 * several worker threads, whose function is described below.
2236 *
2237 * There are three cases.
2238 * The first case is a redacted zfs send. In this case there are 5 threads.
2239 * The first thread is the to_ds traversal thread: it calls dataset_traverse on
2240 * the to_ds and finds all the blocks that have changed since ancestor_zb (if
2241 * it's a full send, that's all blocks in the dataset). It then sends those
2242 * blocks on to the send merge thread. The redact list thread takes the data
2243 * from the redaction bookmark and sends those blocks on to the send merge
2244 * thread. The send merge thread takes the data from the to_ds traversal
2245 * thread, and combines it with the redaction records from the redact list
2246 * thread. If a block appears in both the to_ds's data and the redaction data,
2247 * the send merge thread will mark it as redacted and send it on to the prefetch
2248 * thread. Otherwise, the send merge thread will send the block on to the
2249 * prefetch thread unchanged. The prefetch thread will issue prefetch reads for
2250 * any data that isn't redacted, and then send the data on to the main thread.
2251 * The main thread behaves the same as in a normal send case, issuing demand
2252 * reads for data blocks and sending out records over the network
2253 *
2254 * The graphic below diagrams the flow of data in the case of a redacted zfs
2255 * send. Each box represents a thread, and each line represents the flow of
2256 * data.
2257 *
2258 * Records from the |
2259 * redaction bookmark |
2260 * +--------------------+ | +---------------------------+
2261 * | | v | Send Merge Thread |
2262 * | Redact List Thread +----------> Apply redaction marks to |
2263 * | | | records as specified by |
2264 * +--------------------+ | redaction ranges |
2265 * +----^---------------+------+
2266 * | | Merged data
2267 * | |
2268 * | +------------v--------+
2269 * | | Prefetch Thread |
2270 * +--------------------+ | | Issues prefetch |
2271 * | to_ds Traversal | | | reads of data blocks|
2272 * | Thread (finds +---------------+ +------------+--------+
2273 * | candidate blocks) | Blocks modified | Prefetched data
2274 * +--------------------+ by to_ds since |
2275 * ancestor_zb +------------v----+
2276 * | Main Thread | File Descriptor
2277 * | Sends data over +->(to zfs receive)
2278 * | wire |
2279 * +-----------------+
2280 *
2281 * The second case is an incremental send from a redaction bookmark. The to_ds
2282 * traversal thread and the main thread behave the same as in the redacted
2283 * send case. The new thread is the from bookmark traversal thread. It
2284 * iterates over the redaction list in the redaction bookmark, and enqueues
2285 * records for each block that was redacted in the original send. The send
2286 * merge thread now has to merge the data from the two threads. For details
2287 * about that process, see the header comment of send_merge_thread(). Any data
2288 * it decides to send on will be prefetched by the prefetch thread. Note that
2289 * you can perform a redacted send from a redaction bookmark; in that case,
2290 * the data flow behaves very similarly to the flow in the redacted send case,
2291 * except with the addition of the bookmark traversal thread iterating over the
2292 * redaction bookmark. The send_merge_thread also has to take on the
2293 * responsibility of merging the redact list thread's records, the bookmark
2294 * traversal thread's records, and the to_ds records.
2295 *
2296 * +---------------------+
2297 * | |
2298 * | Redact List Thread +--------------+
2299 * | | |
2300 * +---------------------+ |
2301 * Blocks in redaction list | Ranges modified by every secure snap
2302 * of from bookmark | (or EOS if not readcted)
2303 * |
2304 * +---------------------+ | +----v----------------------+
2305 * | bookmark Traversal | v | Send Merge Thread |
2306 * | Thread (finds +---------> Merges bookmark, rlt, and |
2307 * | candidate blocks) | | to_ds send records |
2308 * +---------------------+ +----^---------------+------+
2309 * | | Merged data
2310 * | +------------v--------+
2311 * | | Prefetch Thread |
2312 * +--------------------+ | | Issues prefetch |
2313 * | to_ds Traversal | | | reads of data blocks|
2314 * | Thread (finds +---------------+ +------------+--------+
2315 * | candidate blocks) | Blocks modified | Prefetched data
2316 * +--------------------+ by to_ds since +------------v----+
2317 * ancestor_zb | Main Thread | File Descriptor
2318 * | Sends data over +->(to zfs receive)
2319 * | wire |
2320 * +-----------------+
2321 *
2322 * The final case is a simple zfs full or incremental send. The to_ds traversal
2323 * thread behaves the same as always. The redact list thread is never started.
2324 * The send merge thread takes all the blocks that the to_ds traversal thread
2325 * sends it, prefetches the data, and sends the blocks on to the main thread.
2326 * The main thread sends the data over the wire.
2327 *
2328 * To keep performance acceptable, we want to prefetch the data in the worker
2329 * threads. While the to_ds thread could simply use the TRAVERSE_PREFETCH
2330 * feature built into traverse_dataset, the combining and deletion of records
2331 * due to redaction and sends from redaction bookmarks mean that we could
2332 * issue many unnecessary prefetches. As a result, we only prefetch data
2333 * after we've determined that the record is not going to be redacted. To
2334 * prevent the prefetching from getting too far ahead of the main thread, the
2335 * blocking queues that are used for communication are capped not by the
2336 * number of entries in the queue, but by the sum of the size of the
2337 * prefetches associated with them. The limit on the amount of data that the
2338 * thread can prefetch beyond what the main thread has reached is controlled
2339 * by the global variable zfs_send_queue_length. In addition, to prevent poor
2340 * performance in the beginning of a send, we also limit the distance ahead
2341 * that the traversal threads can be. That distance is controlled by the
2342 * zfs_send_no_prefetch_queue_length tunable.
2343 *
2344 * Note: Releases dp using the specified tag.
2345 */
2346 static int
2347 dmu_send_impl(struct dmu_send_params *dspp)
2348 {
2349 objset_t *os;
2350 dmu_replay_record_t *drr;
2351 dmu_sendstatus_t *dssp;
2352 dmu_send_cookie_t dsc = {0};
2353 int err;
2354 uint64_t fromtxg = dspp->ancestor_zb.zbm_creation_txg;
2355 uint64_t featureflags = 0;
2356 struct redact_list_thread_arg *from_arg;
2357 struct send_thread_arg *to_arg;
2358 struct redact_list_thread_arg *rlt_arg;
2359 struct send_merge_thread_arg *smt_arg;
2360 struct send_reader_thread_arg *srt_arg;
2361 struct send_range *range;
2362 redaction_list_t *from_rl = NULL;
2363 redaction_list_t *redact_rl = NULL;
2364 boolean_t resuming = (dspp->resumeobj != 0 || dspp->resumeoff != 0);
2365 boolean_t book_resuming = resuming;
2366
2367 dsl_dataset_t *to_ds = dspp->to_ds;
2368 zfs_bookmark_phys_t *ancestor_zb = &dspp->ancestor_zb;
2369 dsl_pool_t *dp = dspp->dp;
2370 const void *tag = dspp->tag;
2371
2372 err = dmu_objset_from_ds(to_ds, &os);
2373 if (err != 0) {
2374 dsl_pool_rele(dp, tag);
2375 return (err);
2376 }
2377
2378 /*
2379 * If this is a non-raw send of an encrypted ds, we can ensure that
2380 * the objset_phys_t is authenticated. This is safe because this is
2381 * either a snapshot or we have owned the dataset, ensuring that
2382 * it can't be modified.
2383 */
2384 if (!dspp->rawok && os->os_encrypted &&
2385 arc_is_unauthenticated(os->os_phys_buf)) {
2386 zbookmark_phys_t zb;
2387
2388 SET_BOOKMARK(&zb, to_ds->ds_object, ZB_ROOT_OBJECT,
2389 ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
2390 err = arc_untransform(os->os_phys_buf, os->os_spa,
2391 &zb, B_FALSE);
2392 if (err != 0) {
2393 dsl_pool_rele(dp, tag);
2394 return (err);
2395 }
2396
2397 ASSERT0(arc_is_unauthenticated(os->os_phys_buf));
2398 }
2399
2400 if ((err = setup_featureflags(dspp, os, &featureflags)) != 0) {
2401 dsl_pool_rele(dp, tag);
2402 return (err);
2403 }
2404
2405 /*
2406 * If we're doing a redacted send, hold the bookmark's redaction list.
2407 */
2408 if (dspp->redactbook != NULL) {
2409 err = dsl_redaction_list_hold_obj(dp,
2410 dspp->redactbook->zbm_redaction_obj, FTAG,
2411 &redact_rl);
2412 if (err != 0) {
2413 dsl_pool_rele(dp, tag);
2414 return (SET_ERROR(EINVAL));
2415 }
2416 dsl_redaction_list_long_hold(dp, redact_rl, FTAG);
2417 }
2418
2419 /*
2420 * If we're sending from a redaction bookmark, hold the redaction list
2421 * so that we can consider sending the redacted blocks.
2422 */
2423 if (ancestor_zb->zbm_redaction_obj != 0) {
2424 err = dsl_redaction_list_hold_obj(dp,
2425 ancestor_zb->zbm_redaction_obj, FTAG, &from_rl);
2426 if (err != 0) {
2427 if (redact_rl != NULL) {
2428 dsl_redaction_list_long_rele(redact_rl, FTAG);
2429 dsl_redaction_list_rele(redact_rl, FTAG);
2430 }
2431 dsl_pool_rele(dp, tag);
2432 return (SET_ERROR(EINVAL));
2433 }
2434 dsl_redaction_list_long_hold(dp, from_rl, FTAG);
2435 }
2436
2437 dsl_dataset_long_hold(to_ds, FTAG);
2438
2439 from_arg = kmem_zalloc(sizeof (*from_arg), KM_SLEEP);
2440 to_arg = kmem_zalloc(sizeof (*to_arg), KM_SLEEP);
2441 rlt_arg = kmem_zalloc(sizeof (*rlt_arg), KM_SLEEP);
2442 smt_arg = kmem_zalloc(sizeof (*smt_arg), KM_SLEEP);
2443 srt_arg = kmem_zalloc(sizeof (*srt_arg), KM_SLEEP);
2444
2445 drr = create_begin_record(dspp, os, featureflags);
2446 dssp = setup_send_progress(dspp);
2447
2448 dsc.dsc_drr = drr;
2449 dsc.dsc_dso = dspp->dso;
2450 dsc.dsc_os = os;
2451 dsc.dsc_off = dspp->off;
2452 dsc.dsc_toguid = dsl_dataset_phys(to_ds)->ds_guid;
2453 dsc.dsc_fromtxg = fromtxg;
2454 dsc.dsc_pending_op = PENDING_NONE;
2455 dsc.dsc_featureflags = featureflags;
2456 dsc.dsc_resume_object = dspp->resumeobj;
2457 dsc.dsc_resume_offset = dspp->resumeoff;
2458
2459 dsl_pool_rele(dp, tag);
2460
2461 void *payload = NULL;
2462 size_t payload_len = 0;
2463 nvlist_t *nvl = fnvlist_alloc();
2464
2465 /*
2466 * If we're doing a redacted send, we include the snapshots we're
2467 * redacted with respect to so that the target system knows what send
2468 * streams can be correctly received on top of this dataset. If we're
2469 * instead sending a redacted dataset, we include the snapshots that the
2470 * dataset was created with respect to.
2471 */
2472 if (dspp->redactbook != NULL) {
2473 fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_SNAPS,
2474 redact_rl->rl_phys->rlp_snaps,
2475 redact_rl->rl_phys->rlp_num_snaps);
2476 } else if (dsl_dataset_feature_is_active(to_ds,
2477 SPA_FEATURE_REDACTED_DATASETS)) {
2478 uint64_t *tods_guids;
2479 uint64_t length;
2480 VERIFY(dsl_dataset_get_uint64_array_feature(to_ds,
2481 SPA_FEATURE_REDACTED_DATASETS, &length, &tods_guids));
2482 fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_SNAPS, tods_guids,
2483 length);
2484 }
2485
2486 /*
2487 * If we're sending from a redaction bookmark, then we should retrieve
2488 * the guids of that bookmark so we can send them over the wire.
2489 */
2490 if (from_rl != NULL) {
2491 fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_FROM_SNAPS,
2492 from_rl->rl_phys->rlp_snaps,
2493 from_rl->rl_phys->rlp_num_snaps);
2494 }
2495
2496 /*
2497 * If the snapshot we're sending from is redacted, include the redaction
2498 * list in the stream.
2499 */
2500 if (dspp->numfromredactsnaps != NUM_SNAPS_NOT_REDACTED) {
2501 ASSERT3P(from_rl, ==, NULL);
2502 fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_FROM_SNAPS,
2503 dspp->fromredactsnaps, (uint_t)dspp->numfromredactsnaps);
2504 if (dspp->numfromredactsnaps > 0) {
2505 kmem_free(dspp->fromredactsnaps,
2506 dspp->numfromredactsnaps * sizeof (uint64_t));
2507 dspp->fromredactsnaps = NULL;
2508 }
2509 }
2510
2511 if (resuming || book_resuming) {
2512 err = setup_resume_points(dspp, to_arg, from_arg,
2513 rlt_arg, smt_arg, resuming, os, redact_rl, nvl);
2514 if (err != 0)
2515 goto out;
2516 }
2517
2518 if (featureflags & DMU_BACKUP_FEATURE_RAW) {
2519 uint64_t ivset_guid = ancestor_zb->zbm_ivset_guid;
2520 nvlist_t *keynvl = NULL;
2521 ASSERT(os->os_encrypted);
2522
2523 err = dsl_crypto_populate_key_nvlist(os, ivset_guid,
2524 &keynvl);
2525 if (err != 0) {
2526 fnvlist_free(nvl);
2527 goto out;
2528 }
2529
2530 fnvlist_add_nvlist(nvl, "crypt_keydata", keynvl);
2531 fnvlist_free(keynvl);
2532 }
2533
2534 if (!nvlist_empty(nvl)) {
2535 payload = fnvlist_pack(nvl, &payload_len);
2536 drr->drr_payloadlen = payload_len;
2537 }
2538
2539 fnvlist_free(nvl);
2540 err = dump_record(&dsc, payload, payload_len);
2541 fnvlist_pack_free(payload, payload_len);
2542 if (err != 0) {
2543 err = dsc.dsc_err;
2544 goto out;
2545 }
2546
2547 setup_to_thread(to_arg, os, dssp, fromtxg, dspp->rawok);
2548 setup_from_thread(from_arg, from_rl, dssp);
2549 setup_redact_list_thread(rlt_arg, dspp, redact_rl, dssp);
2550 setup_merge_thread(smt_arg, dspp, from_arg, to_arg, rlt_arg, os);
2551 setup_reader_thread(srt_arg, dspp, smt_arg, featureflags);
2552
2553 range = bqueue_dequeue(&srt_arg->q);
2554 while (err == 0 && !range->eos_marker) {
2555 err = do_dump(&dsc, range);
2556 range = get_next_range(&srt_arg->q, range);
2557 if (issig(JUSTLOOKING) && issig(FORREAL))
2558 err = SET_ERROR(EINTR);
2559 }
2560
2561 /*
2562 * If we hit an error or are interrupted, cancel our worker threads and
2563 * clear the queue of any pending records. The threads will pass the
2564 * cancel up the tree of worker threads, and each one will clean up any
2565 * pending records before exiting.
2566 */
2567 if (err != 0) {
2568 srt_arg->cancel = B_TRUE;
2569 while (!range->eos_marker) {
2570 range = get_next_range(&srt_arg->q, range);
2571 }
2572 }
2573 range_free(range);
2574
2575 bqueue_destroy(&srt_arg->q);
2576 bqueue_destroy(&smt_arg->q);
2577 if (dspp->redactbook != NULL)
2578 bqueue_destroy(&rlt_arg->q);
2579 bqueue_destroy(&to_arg->q);
2580 bqueue_destroy(&from_arg->q);
2581
2582 if (err == 0 && srt_arg->error != 0)
2583 err = srt_arg->error;
2584
2585 if (err != 0)
2586 goto out;
2587
2588 if (dsc.dsc_pending_op != PENDING_NONE)
2589 if (dump_record(&dsc, NULL, 0) != 0)
2590 err = SET_ERROR(EINTR);
2591
2592 if (err != 0) {
2593 if (err == EINTR && dsc.dsc_err != 0)
2594 err = dsc.dsc_err;
2595 goto out;
2596 }
2597
2598 /*
2599 * Send the DRR_END record if this is not a saved stream.
2600 * Otherwise, the omitted DRR_END record will signal to
2601 * the receive side that the stream is incomplete.
2602 */
2603 if (!dspp->savedok) {
2604 memset(drr, 0, sizeof (dmu_replay_record_t));
2605 drr->drr_type = DRR_END;
2606 drr->drr_u.drr_end.drr_checksum = dsc.dsc_zc;
2607 drr->drr_u.drr_end.drr_toguid = dsc.dsc_toguid;
2608
2609 if (dump_record(&dsc, NULL, 0) != 0)
2610 err = dsc.dsc_err;
2611 }
2612 out:
2613 mutex_enter(&to_ds->ds_sendstream_lock);
2614 list_remove(&to_ds->ds_sendstreams, dssp);
2615 mutex_exit(&to_ds->ds_sendstream_lock);
2616
2617 VERIFY(err != 0 || (dsc.dsc_sent_begin &&
2618 (dsc.dsc_sent_end || dspp->savedok)));
2619
2620 kmem_free(drr, sizeof (dmu_replay_record_t));
2621 kmem_free(dssp, sizeof (dmu_sendstatus_t));
2622 kmem_free(from_arg, sizeof (*from_arg));
2623 kmem_free(to_arg, sizeof (*to_arg));
2624 kmem_free(rlt_arg, sizeof (*rlt_arg));
2625 kmem_free(smt_arg, sizeof (*smt_arg));
2626 kmem_free(srt_arg, sizeof (*srt_arg));
2627
2628 dsl_dataset_long_rele(to_ds, FTAG);
2629 if (from_rl != NULL) {
2630 dsl_redaction_list_long_rele(from_rl, FTAG);
2631 dsl_redaction_list_rele(from_rl, FTAG);
2632 }
2633 if (redact_rl != NULL) {
2634 dsl_redaction_list_long_rele(redact_rl, FTAG);
2635 dsl_redaction_list_rele(redact_rl, FTAG);
2636 }
2637
2638 return (err);
2639 }
2640
2641 int
2642 dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
2643 boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
2644 boolean_t rawok, boolean_t savedok, int outfd, offset_t *off,
2645 dmu_send_outparams_t *dsop)
2646 {
2647 int err;
2648 dsl_dataset_t *fromds;
2649 ds_hold_flags_t dsflags;
2650 struct dmu_send_params dspp = {0};
2651 dspp.embedok = embedok;
2652 dspp.large_block_ok = large_block_ok;
2653 dspp.compressok = compressok;
2654 dspp.outfd = outfd;
2655 dspp.off = off;
2656 dspp.dso = dsop;
2657 dspp.tag = FTAG;
2658 dspp.rawok = rawok;
2659 dspp.savedok = savedok;
2660
2661 dsflags = (rawok) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT;
2662 err = dsl_pool_hold(pool, FTAG, &dspp.dp);
2663 if (err != 0)
2664 return (err);
2665
2666 err = dsl_dataset_hold_obj_flags(dspp.dp, tosnap, dsflags, FTAG,
2667 &dspp.to_ds);
2668 if (err != 0) {
2669 dsl_pool_rele(dspp.dp, FTAG);
2670 return (err);
2671 }
2672
2673 if (fromsnap != 0) {
2674 err = dsl_dataset_hold_obj_flags(dspp.dp, fromsnap, dsflags,
2675 FTAG, &fromds);
2676 if (err != 0) {
2677 dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
2678 dsl_pool_rele(dspp.dp, FTAG);
2679 return (err);
2680 }
2681 dspp.ancestor_zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
2682 dspp.ancestor_zb.zbm_creation_txg =
2683 dsl_dataset_phys(fromds)->ds_creation_txg;
2684 dspp.ancestor_zb.zbm_creation_time =
2685 dsl_dataset_phys(fromds)->ds_creation_time;
2686
2687 if (dsl_dataset_is_zapified(fromds)) {
2688 (void) zap_lookup(dspp.dp->dp_meta_objset,
2689 fromds->ds_object, DS_FIELD_IVSET_GUID, 8, 1,
2690 &dspp.ancestor_zb.zbm_ivset_guid);
2691 }
2692
2693 /* See dmu_send for the reasons behind this. */
2694 uint64_t *fromredact;
2695
2696 if (!dsl_dataset_get_uint64_array_feature(fromds,
2697 SPA_FEATURE_REDACTED_DATASETS,
2698 &dspp.numfromredactsnaps,
2699 &fromredact)) {
2700 dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2701 } else if (dspp.numfromredactsnaps > 0) {
2702 uint64_t size = dspp.numfromredactsnaps *
2703 sizeof (uint64_t);
2704 dspp.fromredactsnaps = kmem_zalloc(size, KM_SLEEP);
2705 memcpy(dspp.fromredactsnaps, fromredact, size);
2706 }
2707
2708 boolean_t is_before =
2709 dsl_dataset_is_before(dspp.to_ds, fromds, 0);
2710 dspp.is_clone = (dspp.to_ds->ds_dir !=
2711 fromds->ds_dir);
2712 dsl_dataset_rele(fromds, FTAG);
2713 if (!is_before) {
2714 dsl_pool_rele(dspp.dp, FTAG);
2715 err = SET_ERROR(EXDEV);
2716 } else {
2717 err = dmu_send_impl(&dspp);
2718 }
2719 } else {
2720 dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2721 err = dmu_send_impl(&dspp);
2722 }
2723 if (dspp.fromredactsnaps)
2724 kmem_free(dspp.fromredactsnaps,
2725 dspp.numfromredactsnaps * sizeof (uint64_t));
2726
2727 dsl_dataset_rele(dspp.to_ds, FTAG);
2728 return (err);
2729 }
2730
2731 int
2732 dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
2733 boolean_t large_block_ok, boolean_t compressok, boolean_t rawok,
2734 boolean_t savedok, uint64_t resumeobj, uint64_t resumeoff,
2735 const char *redactbook, int outfd, offset_t *off,
2736 dmu_send_outparams_t *dsop)
2737 {
2738 int err = 0;
2739 ds_hold_flags_t dsflags;
2740 boolean_t owned = B_FALSE;
2741 dsl_dataset_t *fromds = NULL;
2742 zfs_bookmark_phys_t book = {0};
2743 struct dmu_send_params dspp = {0};
2744
2745 dsflags = (rawok) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT;
2746 dspp.tosnap = tosnap;
2747 dspp.embedok = embedok;
2748 dspp.large_block_ok = large_block_ok;
2749 dspp.compressok = compressok;
2750 dspp.outfd = outfd;
2751 dspp.off = off;
2752 dspp.dso = dsop;
2753 dspp.tag = FTAG;
2754 dspp.resumeobj = resumeobj;
2755 dspp.resumeoff = resumeoff;
2756 dspp.rawok = rawok;
2757 dspp.savedok = savedok;
2758
2759 if (fromsnap != NULL && strpbrk(fromsnap, "@#") == NULL)
2760 return (SET_ERROR(EINVAL));
2761
2762 err = dsl_pool_hold(tosnap, FTAG, &dspp.dp);
2763 if (err != 0)
2764 return (err);
2765
2766 if (strchr(tosnap, '@') == NULL && spa_writeable(dspp.dp->dp_spa)) {
2767 /*
2768 * We are sending a filesystem or volume. Ensure
2769 * that it doesn't change by owning the dataset.
2770 */
2771
2772 if (savedok) {
2773 /*
2774 * We are looking for the dataset that represents the
2775 * partially received send stream. If this stream was
2776 * received as a new snapshot of an existing dataset,
2777 * this will be saved in a hidden clone named
2778 * "<pool>/<dataset>/%recv". Otherwise, the stream
2779 * will be saved in the live dataset itself. In
2780 * either case we need to use dsl_dataset_own_force()
2781 * because the stream is marked as inconsistent,
2782 * which would normally make it unavailable to be
2783 * owned.
2784 */
2785 char *name = kmem_asprintf("%s/%s", tosnap,
2786 recv_clone_name);
2787 err = dsl_dataset_own_force(dspp.dp, name, dsflags,
2788 FTAG, &dspp.to_ds);
2789 if (err == ENOENT) {
2790 err = dsl_dataset_own_force(dspp.dp, tosnap,
2791 dsflags, FTAG, &dspp.to_ds);
2792 }
2793
2794 if (err == 0) {
2795 err = zap_lookup(dspp.dp->dp_meta_objset,
2796 dspp.to_ds->ds_object,
2797 DS_FIELD_RESUME_TOGUID, 8, 1,
2798 &dspp.saved_guid);
2799 }
2800
2801 if (err == 0) {
2802 err = zap_lookup(dspp.dp->dp_meta_objset,
2803 dspp.to_ds->ds_object,
2804 DS_FIELD_RESUME_TONAME, 1,
2805 sizeof (dspp.saved_toname),
2806 dspp.saved_toname);
2807 }
2808 if (err != 0)
2809 dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
2810
2811 kmem_strfree(name);
2812 } else {
2813 err = dsl_dataset_own(dspp.dp, tosnap, dsflags,
2814 FTAG, &dspp.to_ds);
2815 }
2816 owned = B_TRUE;
2817 } else {
2818 err = dsl_dataset_hold_flags(dspp.dp, tosnap, dsflags, FTAG,
2819 &dspp.to_ds);
2820 }
2821
2822 if (err != 0) {
2823 dsl_pool_rele(dspp.dp, FTAG);
2824 return (err);
2825 }
2826
2827 if (redactbook != NULL) {
2828 char path[ZFS_MAX_DATASET_NAME_LEN];
2829 (void) strlcpy(path, tosnap, sizeof (path));
2830 char *at = strchr(path, '@');
2831 if (at == NULL) {
2832 err = EINVAL;
2833 } else {
2834 (void) snprintf(at, sizeof (path) - (at - path), "#%s",
2835 redactbook);
2836 err = dsl_bookmark_lookup(dspp.dp, path,
2837 NULL, &book);
2838 dspp.redactbook = &book;
2839 }
2840 }
2841
2842 if (err != 0) {
2843 dsl_pool_rele(dspp.dp, FTAG);
2844 if (owned)
2845 dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
2846 else
2847 dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
2848 return (err);
2849 }
2850
2851 if (fromsnap != NULL) {
2852 zfs_bookmark_phys_t *zb = &dspp.ancestor_zb;
2853 int fsnamelen;
2854 if (strpbrk(tosnap, "@#") != NULL)
2855 fsnamelen = strpbrk(tosnap, "@#") - tosnap;
2856 else
2857 fsnamelen = strlen(tosnap);
2858
2859 /*
2860 * If the fromsnap is in a different filesystem, then
2861 * mark the send stream as a clone.
2862 */
2863 if (strncmp(tosnap, fromsnap, fsnamelen) != 0 ||
2864 (fromsnap[fsnamelen] != '@' &&
2865 fromsnap[fsnamelen] != '#')) {
2866 dspp.is_clone = B_TRUE;
2867 }
2868
2869 if (strchr(fromsnap, '@') != NULL) {
2870 err = dsl_dataset_hold(dspp.dp, fromsnap, FTAG,
2871 &fromds);
2872
2873 if (err != 0) {
2874 ASSERT3P(fromds, ==, NULL);
2875 } else {
2876 /*
2877 * We need to make a deep copy of the redact
2878 * snapshots of the from snapshot, because the
2879 * array will be freed when we evict from_ds.
2880 */
2881 uint64_t *fromredact;
2882 if (!dsl_dataset_get_uint64_array_feature(
2883 fromds, SPA_FEATURE_REDACTED_DATASETS,
2884 &dspp.numfromredactsnaps,
2885 &fromredact)) {
2886 dspp.numfromredactsnaps =
2887 NUM_SNAPS_NOT_REDACTED;
2888 } else if (dspp.numfromredactsnaps > 0) {
2889 uint64_t size =
2890 dspp.numfromredactsnaps *
2891 sizeof (uint64_t);
2892 dspp.fromredactsnaps = kmem_zalloc(size,
2893 KM_SLEEP);
2894 memcpy(dspp.fromredactsnaps, fromredact,
2895 size);
2896 }
2897 if (!dsl_dataset_is_before(dspp.to_ds, fromds,
2898 0)) {
2899 err = SET_ERROR(EXDEV);
2900 } else {
2901 zb->zbm_creation_txg =
2902 dsl_dataset_phys(fromds)->
2903 ds_creation_txg;
2904 zb->zbm_creation_time =
2905 dsl_dataset_phys(fromds)->
2906 ds_creation_time;
2907 zb->zbm_guid =
2908 dsl_dataset_phys(fromds)->ds_guid;
2909 zb->zbm_redaction_obj = 0;
2910
2911 if (dsl_dataset_is_zapified(fromds)) {
2912 (void) zap_lookup(
2913 dspp.dp->dp_meta_objset,
2914 fromds->ds_object,
2915 DS_FIELD_IVSET_GUID, 8, 1,
2916 &zb->zbm_ivset_guid);
2917 }
2918 }
2919 dsl_dataset_rele(fromds, FTAG);
2920 }
2921 } else {
2922 dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2923 err = dsl_bookmark_lookup(dspp.dp, fromsnap, dspp.to_ds,
2924 zb);
2925 if (err == EXDEV && zb->zbm_redaction_obj != 0 &&
2926 zb->zbm_guid ==
2927 dsl_dataset_phys(dspp.to_ds)->ds_guid)
2928 err = 0;
2929 }
2930
2931 if (err == 0) {
2932 /* dmu_send_impl will call dsl_pool_rele for us. */
2933 err = dmu_send_impl(&dspp);
2934 } else {
2935 if (dspp.fromredactsnaps)
2936 kmem_free(dspp.fromredactsnaps,
2937 dspp.numfromredactsnaps *
2938 sizeof (uint64_t));
2939 dsl_pool_rele(dspp.dp, FTAG);
2940 }
2941 } else {
2942 dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2943 err = dmu_send_impl(&dspp);
2944 }
2945 if (owned)
2946 dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
2947 else
2948 dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
2949 return (err);
2950 }
2951
2952 static int
2953 dmu_adjust_send_estimate_for_indirects(dsl_dataset_t *ds, uint64_t uncompressed,
2954 uint64_t compressed, boolean_t stream_compressed, uint64_t *sizep)
2955 {
2956 int err = 0;
2957 uint64_t size;
2958 /*
2959 * Assume that space (both on-disk and in-stream) is dominated by
2960 * data. We will adjust for indirect blocks and the copies property,
2961 * but ignore per-object space used (eg, dnodes and DRR_OBJECT records).
2962 */
2963
2964 uint64_t recordsize;
2965 uint64_t record_count;
2966 objset_t *os;
2967 VERIFY0(dmu_objset_from_ds(ds, &os));
2968
2969 /* Assume all (uncompressed) blocks are recordsize. */
2970 if (zfs_override_estimate_recordsize != 0) {
2971 recordsize = zfs_override_estimate_recordsize;
2972 } else if (os->os_phys->os_type == DMU_OST_ZVOL) {
2973 err = dsl_prop_get_int_ds(ds,
2974 zfs_prop_to_name(ZFS_PROP_VOLBLOCKSIZE), &recordsize);
2975 } else {
2976 err = dsl_prop_get_int_ds(ds,
2977 zfs_prop_to_name(ZFS_PROP_RECORDSIZE), &recordsize);
2978 }
2979 if (err != 0)
2980 return (err);
2981 record_count = uncompressed / recordsize;
2982
2983 /*
2984 * If we're estimating a send size for a compressed stream, use the
2985 * compressed data size to estimate the stream size. Otherwise, use the
2986 * uncompressed data size.
2987 */
2988 size = stream_compressed ? compressed : uncompressed;
2989
2990 /*
2991 * Subtract out approximate space used by indirect blocks.
2992 * Assume most space is used by data blocks (non-indirect, non-dnode).
2993 * Assume no ditto blocks or internal fragmentation.
2994 *
2995 * Therefore, space used by indirect blocks is sizeof(blkptr_t) per
2996 * block.
2997 */
2998 size -= record_count * sizeof (blkptr_t);
2999
3000 /* Add in the space for the record associated with each block. */
3001 size += record_count * sizeof (dmu_replay_record_t);
3002
3003 *sizep = size;
3004
3005 return (0);
3006 }
3007
3008 int
3009 dmu_send_estimate_fast(dsl_dataset_t *origds, dsl_dataset_t *fromds,
3010 zfs_bookmark_phys_t *frombook, boolean_t stream_compressed,
3011 boolean_t saved, uint64_t *sizep)
3012 {
3013 int err;
3014 dsl_dataset_t *ds = origds;
3015 uint64_t uncomp, comp;
3016
3017 ASSERT(dsl_pool_config_held(origds->ds_dir->dd_pool));
3018 ASSERT(fromds == NULL || frombook == NULL);
3019
3020 /*
3021 * If this is a saved send we may actually be sending
3022 * from the %recv clone used for resuming.
3023 */
3024 if (saved) {
3025 objset_t *mos = origds->ds_dir->dd_pool->dp_meta_objset;
3026 uint64_t guid;
3027 char dsname[ZFS_MAX_DATASET_NAME_LEN + 6];
3028
3029 dsl_dataset_name(origds, dsname);
3030 (void) strcat(dsname, "/");
3031 (void) strlcat(dsname, recv_clone_name,
3032 sizeof (dsname) - strlen(dsname));
3033
3034 err = dsl_dataset_hold(origds->ds_dir->dd_pool,
3035 dsname, FTAG, &ds);
3036 if (err != ENOENT && err != 0) {
3037 return (err);
3038 } else if (err == ENOENT) {
3039 ds = origds;
3040 }
3041
3042 /* check that this dataset has partially received data */
3043 err = zap_lookup(mos, ds->ds_object,
3044 DS_FIELD_RESUME_TOGUID, 8, 1, &guid);
3045 if (err != 0) {
3046 err = SET_ERROR(err == ENOENT ? EINVAL : err);
3047 goto out;
3048 }
3049
3050 err = zap_lookup(mos, ds->ds_object,
3051 DS_FIELD_RESUME_TONAME, 1, sizeof (dsname), dsname);
3052 if (err != 0) {
3053 err = SET_ERROR(err == ENOENT ? EINVAL : err);
3054 goto out;
3055 }
3056 }
3057
3058 /* tosnap must be a snapshot or the target of a saved send */
3059 if (!ds->ds_is_snapshot && ds == origds)
3060 return (SET_ERROR(EINVAL));
3061
3062 if (fromds != NULL) {
3063 uint64_t used;
3064 if (!fromds->ds_is_snapshot) {
3065 err = SET_ERROR(EINVAL);
3066 goto out;
3067 }
3068
3069 if (!dsl_dataset_is_before(ds, fromds, 0)) {
3070 err = SET_ERROR(EXDEV);
3071 goto out;
3072 }
3073
3074 err = dsl_dataset_space_written(fromds, ds, &used, &comp,
3075 &uncomp);
3076 if (err != 0)
3077 goto out;
3078 } else if (frombook != NULL) {
3079 uint64_t used;
3080 err = dsl_dataset_space_written_bookmark(frombook, ds, &used,
3081 &comp, &uncomp);
3082 if (err != 0)
3083 goto out;
3084 } else {
3085 uncomp = dsl_dataset_phys(ds)->ds_uncompressed_bytes;
3086 comp = dsl_dataset_phys(ds)->ds_compressed_bytes;
3087 }
3088
3089 err = dmu_adjust_send_estimate_for_indirects(ds, uncomp, comp,
3090 stream_compressed, sizep);
3091 /*
3092 * Add the size of the BEGIN and END records to the estimate.
3093 */
3094 *sizep += 2 * sizeof (dmu_replay_record_t);
3095
3096 out:
3097 if (ds != origds)
3098 dsl_dataset_rele(ds, FTAG);
3099 return (err);
3100 }
3101
3102 ZFS_MODULE_PARAM(zfs_send, zfs_send_, corrupt_data, INT, ZMOD_RW,
3103 "Allow sending corrupt data");
3104
3105 ZFS_MODULE_PARAM(zfs_send, zfs_send_, queue_length, UINT, ZMOD_RW,
3106 "Maximum send queue length");
3107
3108 ZFS_MODULE_PARAM(zfs_send, zfs_send_, unmodified_spill_blocks, INT, ZMOD_RW,
3109 "Send unmodified spill blocks");
3110
3111 ZFS_MODULE_PARAM(zfs_send, zfs_send_, no_prefetch_queue_length, UINT, ZMOD_RW,
3112 "Maximum send queue length for non-prefetch queues");
3113
3114 ZFS_MODULE_PARAM(zfs_send, zfs_send_, queue_ff, UINT, ZMOD_RW,
3115 "Send queue fill fraction");
3116
3117 ZFS_MODULE_PARAM(zfs_send, zfs_send_, no_prefetch_queue_ff, UINT, ZMOD_RW,
3118 "Send queue fill fraction for non-prefetch queues");
3119
3120 ZFS_MODULE_PARAM(zfs_send, zfs_, override_estimate_recordsize, UINT, ZMOD_RW,
3121 "Override block size estimate with fixed size");
Cache object: cd986ae553e59c60b394dd228164cac3
|