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