1 /* 2 * CDDL HEADER START 3 * 4 * The contents of this file are subject to the terms of the 5 * Common Development and Distribution License (the "License"). 6 * You may not use this file except in compliance with the License. 7 * 8 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE 9 * or https://opensource.org/licenses/CDDL-1.0. 10 * See the License for the specific language governing permissions 11 * and limitations under the License. 12 * 13 * When distributing Covered Code, include this CDDL HEADER in each 14 * file and include the License file at usr/src/OPENSOLARIS.LICENSE. 15 * If applicable, add the following below this CDDL HEADER, with the 16 * fields enclosed by brackets "[]" replaced with your own identifying 17 * information: Portions Copyright [yyyy] [name of copyright owner] 18 * 19 * CDDL HEADER END 20 */ 21 /* 22 * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved. 23 * Copyright 2011 Nexenta Systems, Inc. All rights reserved. 24 * Copyright (c) 2011, 2020 by Delphix. All rights reserved. 25 * Copyright (c) 2014, Joyent, Inc. All rights reserved. 26 * Copyright 2014 HybridCluster. All rights reserved. 27 * Copyright (c) 2018, loli10K <ezomori.nozomu@gmail.com>. All rights reserved. 28 * Copyright (c) 2019, 2024, Klara, Inc. 29 * Copyright (c) 2019, Allan Jude 30 * Copyright (c) 2019 Datto Inc. 31 * Copyright (c) 2022 Axcient. 32 */ 33 34 #include <sys/arc.h> 35 #include <sys/spa_impl.h> 36 #include <sys/dmu.h> 37 #include <sys/dmu_impl.h> 38 #include <sys/dmu_send.h> 39 #include <sys/dmu_recv.h> 40 #include <sys/dmu_tx.h> 41 #include <sys/dbuf.h> 42 #include <sys/dnode.h> 43 #include <sys/zfs_context.h> 44 #include <sys/dmu_objset.h> 45 #include <sys/dmu_traverse.h> 46 #include <sys/dsl_dataset.h> 47 #include <sys/dsl_dir.h> 48 #include <sys/dsl_prop.h> 49 #include <sys/dsl_pool.h> 50 #include <sys/dsl_synctask.h> 51 #include <sys/zfs_ioctl.h> 52 #include <sys/zap.h> 53 #include <sys/zvol.h> 54 #include <sys/zio_checksum.h> 55 #include <sys/zfs_znode.h> 56 #include <zfs_fletcher.h> 57 #include <sys/avl.h> 58 #include <sys/ddt.h> 59 #include <sys/zfs_onexit.h> 60 #include <sys/dsl_destroy.h> 61 #include <sys/blkptr.h> 62 #include <sys/dsl_bookmark.h> 63 #include <sys/zfeature.h> 64 #include <sys/bqueue.h> 65 #include <sys/objlist.h> 66 #ifdef _KERNEL 67 #include <sys/zfs_vfsops.h> 68 #endif 69 #include <sys/zfs_file.h> 70 71 static uint_t zfs_recv_queue_length = SPA_MAXBLOCKSIZE; 72 static uint_t zfs_recv_queue_ff = 20; 73 static uint_t zfs_recv_write_batch_size = 1024 * 1024; 74 static int zfs_recv_best_effort_corrective = 0; 75 76 static const void *const dmu_recv_tag = "dmu_recv_tag"; 77 const char *const recv_clone_name = "%recv"; 78 79 typedef enum { 80 ORNS_NO, 81 ORNS_YES, 82 ORNS_MAYBE 83 } or_need_sync_t; 84 85 static int receive_read_payload_and_next_header(dmu_recv_cookie_t *ra, int len, 86 void *buf); 87 88 struct receive_record_arg { 89 dmu_replay_record_t header; 90 void *payload; /* Pointer to a buffer containing the payload */ 91 /* 92 * If the record is a WRITE or SPILL, pointer to the abd containing the 93 * payload. 94 */ 95 abd_t *abd; 96 int payload_size; 97 uint64_t bytes_read; /* bytes read from stream when record created */ 98 boolean_t eos_marker; /* Marks the end of the stream */ 99 bqueue_node_t node; 100 }; 101 102 struct receive_writer_arg { 103 objset_t *os; 104 boolean_t byteswap; 105 bqueue_t q; 106 107 /* 108 * These three members are used to signal to the main thread when 109 * we're done. 110 */ 111 kmutex_t mutex; 112 kcondvar_t cv; 113 boolean_t done; 114 115 int err; 116 const char *tofs; 117 boolean_t heal; 118 boolean_t resumable; 119 boolean_t raw; /* DMU_BACKUP_FEATURE_RAW set */ 120 boolean_t spill; /* DRR_FLAG_SPILL_BLOCK set */ 121 boolean_t full; /* this is a full send stream */ 122 uint64_t last_object; 123 uint64_t last_offset; 124 uint64_t max_object; /* highest object ID referenced in stream */ 125 uint64_t bytes_read; /* bytes read when current record created */ 126 127 list_t write_batch; 128 129 /* Encryption parameters for the last received DRR_OBJECT_RANGE */ 130 boolean_t or_crypt_params_present; 131 uint64_t or_firstobj; 132 uint64_t or_numslots; 133 uint8_t or_salt[ZIO_DATA_SALT_LEN]; 134 uint8_t or_iv[ZIO_DATA_IV_LEN]; 135 uint8_t or_mac[ZIO_DATA_MAC_LEN]; 136 boolean_t or_byteorder; 137 zio_t *heal_pio; 138 139 /* Keep track of DRR_FREEOBJECTS right after DRR_OBJECT_RANGE */ 140 or_need_sync_t or_need_sync; 141 }; 142 143 typedef struct dmu_recv_begin_arg { 144 const char *drba_origin; 145 dmu_recv_cookie_t *drba_cookie; 146 cred_t *drba_cred; 147 proc_t *drba_proc; 148 dsl_crypto_params_t *drba_dcp; 149 } dmu_recv_begin_arg_t; 150 151 static void 152 byteswap_record(dmu_replay_record_t *drr) 153 { 154 #define DO64(X) (drr->drr_u.X = BSWAP_64(drr->drr_u.X)) 155 #define DO32(X) (drr->drr_u.X = BSWAP_32(drr->drr_u.X)) 156 drr->drr_type = BSWAP_32(drr->drr_type); 157 drr->drr_payloadlen = BSWAP_32(drr->drr_payloadlen); 158 159 switch (drr->drr_type) { 160 case DRR_BEGIN: 161 DO64(drr_begin.drr_magic); 162 DO64(drr_begin.drr_versioninfo); 163 DO64(drr_begin.drr_creation_time); 164 DO32(drr_begin.drr_type); 165 DO32(drr_begin.drr_flags); 166 DO64(drr_begin.drr_toguid); 167 DO64(drr_begin.drr_fromguid); 168 break; 169 case DRR_OBJECT: 170 DO64(drr_object.drr_object); 171 DO32(drr_object.drr_type); 172 DO32(drr_object.drr_bonustype); 173 DO32(drr_object.drr_blksz); 174 DO32(drr_object.drr_bonuslen); 175 DO32(drr_object.drr_raw_bonuslen); 176 DO64(drr_object.drr_toguid); 177 DO64(drr_object.drr_maxblkid); 178 break; 179 case DRR_FREEOBJECTS: 180 DO64(drr_freeobjects.drr_firstobj); 181 DO64(drr_freeobjects.drr_numobjs); 182 DO64(drr_freeobjects.drr_toguid); 183 break; 184 case DRR_WRITE: 185 DO64(drr_write.drr_object); 186 DO32(drr_write.drr_type); 187 DO64(drr_write.drr_offset); 188 DO64(drr_write.drr_logical_size); 189 DO64(drr_write.drr_toguid); 190 ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_write.drr_key.ddk_cksum); 191 DO64(drr_write.drr_key.ddk_prop); 192 DO64(drr_write.drr_compressed_size); 193 break; 194 case DRR_WRITE_EMBEDDED: 195 DO64(drr_write_embedded.drr_object); 196 DO64(drr_write_embedded.drr_offset); 197 DO64(drr_write_embedded.drr_length); 198 DO64(drr_write_embedded.drr_toguid); 199 DO32(drr_write_embedded.drr_lsize); 200 DO32(drr_write_embedded.drr_psize); 201 break; 202 case DRR_FREE: 203 DO64(drr_free.drr_object); 204 DO64(drr_free.drr_offset); 205 DO64(drr_free.drr_length); 206 DO64(drr_free.drr_toguid); 207 break; 208 case DRR_SPILL: 209 DO64(drr_spill.drr_object); 210 DO64(drr_spill.drr_length); 211 DO64(drr_spill.drr_toguid); 212 DO64(drr_spill.drr_compressed_size); 213 DO32(drr_spill.drr_type); 214 break; 215 case DRR_OBJECT_RANGE: 216 DO64(drr_object_range.drr_firstobj); 217 DO64(drr_object_range.drr_numslots); 218 DO64(drr_object_range.drr_toguid); 219 break; 220 case DRR_REDACT: 221 DO64(drr_redact.drr_object); 222 DO64(drr_redact.drr_offset); 223 DO64(drr_redact.drr_length); 224 DO64(drr_redact.drr_toguid); 225 break; 226 case DRR_END: 227 DO64(drr_end.drr_toguid); 228 ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_end.drr_checksum); 229 break; 230 default: 231 break; 232 } 233 234 if (drr->drr_type != DRR_BEGIN) { 235 ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_checksum.drr_checksum); 236 } 237 238 #undef DO64 239 #undef DO32 240 } 241 242 static boolean_t 243 redact_snaps_contains(uint64_t *snaps, uint64_t num_snaps, uint64_t guid) 244 { 245 for (int i = 0; i < num_snaps; i++) { 246 if (snaps[i] == guid) 247 return (B_TRUE); 248 } 249 return (B_FALSE); 250 } 251 252 /* 253 * Check that the new stream we're trying to receive is redacted with respect to 254 * a subset of the snapshots that the origin was redacted with respect to. For 255 * the reasons behind this, see the man page on redacted zfs sends and receives. 256 */ 257 static boolean_t 258 compatible_redact_snaps(uint64_t *origin_snaps, uint64_t origin_num_snaps, 259 uint64_t *redact_snaps, uint64_t num_redact_snaps) 260 { 261 /* 262 * Short circuit the comparison; if we are redacted with respect to 263 * more snapshots than the origin, we can't be redacted with respect 264 * to a subset. 265 */ 266 if (num_redact_snaps > origin_num_snaps) { 267 return (B_FALSE); 268 } 269 270 for (int i = 0; i < num_redact_snaps; i++) { 271 if (!redact_snaps_contains(origin_snaps, origin_num_snaps, 272 redact_snaps[i])) { 273 return (B_FALSE); 274 } 275 } 276 return (B_TRUE); 277 } 278 279 static boolean_t 280 redact_check(dmu_recv_begin_arg_t *drba, dsl_dataset_t *origin) 281 { 282 uint64_t *origin_snaps; 283 uint64_t origin_num_snaps; 284 dmu_recv_cookie_t *drc = drba->drba_cookie; 285 struct drr_begin *drrb = drc->drc_drrb; 286 int featureflags = DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo); 287 int err = 0; 288 boolean_t ret = B_TRUE; 289 uint64_t *redact_snaps; 290 uint_t numredactsnaps; 291 292 /* 293 * If this is a full send stream, we're safe no matter what. 294 */ 295 if (drrb->drr_fromguid == 0) 296 return (ret); 297 298 VERIFY(dsl_dataset_get_uint64_array_feature(origin, 299 SPA_FEATURE_REDACTED_DATASETS, &origin_num_snaps, &origin_snaps)); 300 301 if (nvlist_lookup_uint64_array(drc->drc_begin_nvl, 302 BEGINNV_REDACT_FROM_SNAPS, &redact_snaps, &numredactsnaps) == 303 0) { 304 /* 305 * If the send stream was sent from the redaction bookmark or 306 * the redacted version of the dataset, then we're safe. Verify 307 * that this is from the a compatible redaction bookmark or 308 * redacted dataset. 309 */ 310 if (!compatible_redact_snaps(origin_snaps, origin_num_snaps, 311 redact_snaps, numredactsnaps)) { 312 err = EINVAL; 313 } 314 } else if (featureflags & DMU_BACKUP_FEATURE_REDACTED) { 315 /* 316 * If the stream is redacted, it must be redacted with respect 317 * to a subset of what the origin is redacted with respect to. 318 * See case number 2 in the zfs man page section on redacted zfs 319 * send. 320 */ 321 err = nvlist_lookup_uint64_array(drc->drc_begin_nvl, 322 BEGINNV_REDACT_SNAPS, &redact_snaps, &numredactsnaps); 323 324 if (err != 0 || !compatible_redact_snaps(origin_snaps, 325 origin_num_snaps, redact_snaps, numredactsnaps)) { 326 err = EINVAL; 327 } 328 } else if (!redact_snaps_contains(origin_snaps, origin_num_snaps, 329 drrb->drr_toguid)) { 330 /* 331 * If the stream isn't redacted but the origin is, this must be 332 * one of the snapshots the origin is redacted with respect to. 333 * See case number 1 in the zfs man page section on redacted zfs 334 * send. 335 */ 336 err = EINVAL; 337 } 338 339 if (err != 0) 340 ret = B_FALSE; 341 return (ret); 342 } 343 344 /* 345 * If we previously received a stream with --large-block, we don't support 346 * receiving an incremental on top of it without --large-block. This avoids 347 * forcing a read-modify-write or trying to re-aggregate a string of WRITE 348 * records. 349 */ 350 static int 351 recv_check_large_blocks(dsl_dataset_t *ds, uint64_t featureflags) 352 { 353 if (dsl_dataset_feature_is_active(ds, SPA_FEATURE_LARGE_BLOCKS) && 354 !(featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS)) 355 return (SET_ERROR(ZFS_ERR_STREAM_LARGE_BLOCK_MISMATCH)); 356 return (0); 357 } 358 359 static int 360 recv_begin_check_existing_impl(dmu_recv_begin_arg_t *drba, dsl_dataset_t *ds, 361 uint64_t fromguid, uint64_t featureflags) 362 { 363 uint64_t obj; 364 uint64_t children; 365 int error; 366 dsl_dataset_t *snap; 367 dsl_pool_t *dp = ds->ds_dir->dd_pool; 368 boolean_t encrypted = ds->ds_dir->dd_crypto_obj != 0; 369 boolean_t raw = (featureflags & DMU_BACKUP_FEATURE_RAW) != 0; 370 boolean_t embed = (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) != 0; 371 372 /* Temporary clone name must not exist. */ 373 error = zap_lookup(dp->dp_meta_objset, 374 dsl_dir_phys(ds->ds_dir)->dd_child_dir_zapobj, recv_clone_name, 375 8, 1, &obj); 376 if (error != ENOENT) 377 return (error == 0 ? SET_ERROR(EBUSY) : error); 378 379 /* Resume state must not be set. */ 380 if (dsl_dataset_has_resume_receive_state(ds)) 381 return (SET_ERROR(EBUSY)); 382 383 /* New snapshot name must not exist if we're not healing it. */ 384 error = zap_lookup(dp->dp_meta_objset, 385 dsl_dataset_phys(ds)->ds_snapnames_zapobj, 386 drba->drba_cookie->drc_tosnap, 8, 1, &obj); 387 if (drba->drba_cookie->drc_heal) { 388 if (error != 0) 389 return (error); 390 } else if (error != ENOENT) { 391 return (error == 0 ? SET_ERROR(EEXIST) : error); 392 } 393 394 /* Must not have children if receiving a ZVOL. */ 395 error = zap_count(dp->dp_meta_objset, 396 dsl_dir_phys(ds->ds_dir)->dd_child_dir_zapobj, &children); 397 if (error != 0) 398 return (error); 399 if (drba->drba_cookie->drc_drrb->drr_type != DMU_OST_ZFS && 400 children > 0) 401 return (SET_ERROR(ZFS_ERR_WRONG_PARENT)); 402 403 /* 404 * Check snapshot limit before receiving. We'll recheck again at the 405 * end, but might as well abort before receiving if we're already over 406 * the limit. 407 * 408 * Note that we do not check the file system limit with 409 * dsl_dir_fscount_check because the temporary %clones don't count 410 * against that limit. 411 */ 412 error = dsl_fs_ss_limit_check(ds->ds_dir, 1, ZFS_PROP_SNAPSHOT_LIMIT, 413 NULL, drba->drba_cred, drba->drba_proc); 414 if (error != 0) 415 return (error); 416 417 if (drba->drba_cookie->drc_heal) { 418 /* Encryption is incompatible with embedded data. */ 419 if (encrypted && embed) 420 return (SET_ERROR(EINVAL)); 421 422 /* Healing is not supported when in 'force' mode. */ 423 if (drba->drba_cookie->drc_force) 424 return (SET_ERROR(EINVAL)); 425 426 /* Must have keys loaded if doing encrypted non-raw recv. */ 427 if (encrypted && !raw) { 428 if (spa_keystore_lookup_key(dp->dp_spa, ds->ds_object, 429 NULL, NULL) != 0) 430 return (SET_ERROR(EACCES)); 431 } 432 433 error = dsl_dataset_hold_obj(dp, obj, FTAG, &snap); 434 if (error != 0) 435 return (error); 436 437 /* 438 * When not doing best effort corrective recv healing can only 439 * be done if the send stream is for the same snapshot as the 440 * one we are trying to heal. 441 */ 442 if (zfs_recv_best_effort_corrective == 0 && 443 drba->drba_cookie->drc_drrb->drr_toguid != 444 dsl_dataset_phys(snap)->ds_guid) { 445 dsl_dataset_rele(snap, FTAG); 446 return (SET_ERROR(ENOTSUP)); 447 } 448 dsl_dataset_rele(snap, FTAG); 449 } else if (fromguid != 0) { 450 /* Sanity check the incremental recv */ 451 uint64_t obj = dsl_dataset_phys(ds)->ds_prev_snap_obj; 452 453 /* Can't perform a raw receive on top of a non-raw receive */ 454 if (!encrypted && raw) 455 return (SET_ERROR(EINVAL)); 456 457 /* Encryption is incompatible with embedded data */ 458 if (encrypted && embed) 459 return (SET_ERROR(EINVAL)); 460 461 /* Find snapshot in this dir that matches fromguid. */ 462 while (obj != 0) { 463 error = dsl_dataset_hold_obj(dp, obj, FTAG, 464 &snap); 465 if (error != 0) 466 return (SET_ERROR(ENODEV)); 467 if (snap->ds_dir != ds->ds_dir) { 468 dsl_dataset_rele(snap, FTAG); 469 return (SET_ERROR(ENODEV)); 470 } 471 if (dsl_dataset_phys(snap)->ds_guid == fromguid) 472 break; 473 obj = dsl_dataset_phys(snap)->ds_prev_snap_obj; 474 dsl_dataset_rele(snap, FTAG); 475 } 476 if (obj == 0) 477 return (SET_ERROR(ENODEV)); 478 479 if (drba->drba_cookie->drc_force) { 480 drba->drba_cookie->drc_fromsnapobj = obj; 481 } else { 482 /* 483 * If we are not forcing, there must be no 484 * changes since fromsnap. Raw sends have an 485 * additional constraint that requires that 486 * no "noop" snapshots exist between fromsnap 487 * and tosnap for the IVset checking code to 488 * work properly. 489 */ 490 if (dsl_dataset_modified_since_snap(ds, snap) || 491 (raw && 492 dsl_dataset_phys(ds)->ds_prev_snap_obj != 493 snap->ds_object)) { 494 dsl_dataset_rele(snap, FTAG); 495 return (SET_ERROR(ETXTBSY)); 496 } 497 drba->drba_cookie->drc_fromsnapobj = 498 ds->ds_prev->ds_object; 499 } 500 501 if (dsl_dataset_feature_is_active(snap, 502 SPA_FEATURE_REDACTED_DATASETS) && !redact_check(drba, 503 snap)) { 504 dsl_dataset_rele(snap, FTAG); 505 return (SET_ERROR(EINVAL)); 506 } 507 508 error = recv_check_large_blocks(snap, featureflags); 509 if (error != 0) { 510 dsl_dataset_rele(snap, FTAG); 511 return (error); 512 } 513 514 dsl_dataset_rele(snap, FTAG); 515 } else { 516 /* If full and not healing then must be forced. */ 517 if (!drba->drba_cookie->drc_force) 518 return (SET_ERROR(EEXIST)); 519 520 /* 521 * We don't support using zfs recv -F to blow away 522 * encrypted filesystems. This would require the 523 * dsl dir to point to the old encryption key and 524 * the new one at the same time during the receive. 525 */ 526 if ((!encrypted && raw) || encrypted) 527 return (SET_ERROR(EINVAL)); 528 529 /* 530 * Perform the same encryption checks we would if 531 * we were creating a new dataset from scratch. 532 */ 533 if (!raw) { 534 boolean_t will_encrypt; 535 536 error = dmu_objset_create_crypt_check( 537 ds->ds_dir->dd_parent, drba->drba_dcp, 538 &will_encrypt); 539 if (error != 0) 540 return (error); 541 542 if (will_encrypt && embed) 543 return (SET_ERROR(EINVAL)); 544 } 545 } 546 547 return (0); 548 } 549 550 /* 551 * Check that any feature flags used in the data stream we're receiving are 552 * supported by the pool we are receiving into. 553 * 554 * Note that some of the features we explicitly check here have additional 555 * (implicit) features they depend on, but those dependencies are enforced 556 * through the zfeature_register() calls declaring the features that we 557 * explicitly check. 558 */ 559 static int 560 recv_begin_check_feature_flags_impl(uint64_t featureflags, spa_t *spa) 561 { 562 /* 563 * Check if there are any unsupported feature flags. 564 */ 565 if (!DMU_STREAM_SUPPORTED(featureflags)) { 566 return (SET_ERROR(ZFS_ERR_UNKNOWN_SEND_STREAM_FEATURE)); 567 } 568 569 /* Verify pool version supports SA if SA_SPILL feature set */ 570 if ((featureflags & DMU_BACKUP_FEATURE_SA_SPILL) && 571 spa_version(spa) < SPA_VERSION_SA) 572 return (SET_ERROR(ENOTSUP)); 573 574 /* 575 * LZ4 compressed, ZSTD compressed, embedded, mooched, large blocks, 576 * and large_dnodes in the stream can only be used if those pool 577 * features are enabled because we don't attempt to decompress / 578 * un-embed / un-mooch / split up the blocks / dnodes during the 579 * receive process. 580 */ 581 if ((featureflags & DMU_BACKUP_FEATURE_LZ4) && 582 !spa_feature_is_enabled(spa, SPA_FEATURE_LZ4_COMPRESS)) 583 return (SET_ERROR(ENOTSUP)); 584 if ((featureflags & DMU_BACKUP_FEATURE_ZSTD) && 585 !spa_feature_is_enabled(spa, SPA_FEATURE_ZSTD_COMPRESS)) 586 return (SET_ERROR(ENOTSUP)); 587 if ((featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) && 588 !spa_feature_is_enabled(spa, SPA_FEATURE_EMBEDDED_DATA)) 589 return (SET_ERROR(ENOTSUP)); 590 if ((featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) && 591 !spa_feature_is_enabled(spa, SPA_FEATURE_LARGE_BLOCKS)) 592 return (SET_ERROR(ENOTSUP)); 593 if ((featureflags & DMU_BACKUP_FEATURE_LARGE_DNODE) && 594 !spa_feature_is_enabled(spa, SPA_FEATURE_LARGE_DNODE)) 595 return (SET_ERROR(ENOTSUP)); 596 if ((featureflags & DMU_BACKUP_FEATURE_LARGE_MICROZAP) && 597 !spa_feature_is_enabled(spa, SPA_FEATURE_LARGE_MICROZAP)) 598 return (SET_ERROR(ENOTSUP)); 599 600 /* 601 * Receiving redacted streams requires that redacted datasets are 602 * enabled. 603 */ 604 if ((featureflags & DMU_BACKUP_FEATURE_REDACTED) && 605 !spa_feature_is_enabled(spa, SPA_FEATURE_REDACTED_DATASETS)) 606 return (SET_ERROR(ENOTSUP)); 607 608 /* 609 * If the LONGNAME is not enabled on the target, fail that request. 610 */ 611 if ((featureflags & DMU_BACKUP_FEATURE_LONGNAME) && 612 !spa_feature_is_enabled(spa, SPA_FEATURE_LONGNAME)) 613 return (SET_ERROR(ENOTSUP)); 614 615 return (0); 616 } 617 618 static int 619 dmu_recv_begin_check(void *arg, dmu_tx_t *tx) 620 { 621 dmu_recv_begin_arg_t *drba = arg; 622 dsl_pool_t *dp = dmu_tx_pool(tx); 623 struct drr_begin *drrb = drba->drba_cookie->drc_drrb; 624 uint64_t fromguid = drrb->drr_fromguid; 625 int flags = drrb->drr_flags; 626 ds_hold_flags_t dsflags = DS_HOLD_FLAG_NONE; 627 int error; 628 uint64_t featureflags = drba->drba_cookie->drc_featureflags; 629 dsl_dataset_t *ds; 630 const char *tofs = drba->drba_cookie->drc_tofs; 631 632 /* already checked */ 633 ASSERT3U(drrb->drr_magic, ==, DMU_BACKUP_MAGIC); 634 ASSERT(!(featureflags & DMU_BACKUP_FEATURE_RESUMING)); 635 636 if (DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) == 637 DMU_COMPOUNDSTREAM || 638 drrb->drr_type >= DMU_OST_NUMTYPES || 639 ((flags & DRR_FLAG_CLONE) && drba->drba_origin == NULL)) 640 return (SET_ERROR(EINVAL)); 641 642 error = recv_begin_check_feature_flags_impl(featureflags, dp->dp_spa); 643 if (error != 0) 644 return (error); 645 646 /* Resumable receives require extensible datasets */ 647 if (drba->drba_cookie->drc_resumable && 648 !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_EXTENSIBLE_DATASET)) 649 return (SET_ERROR(ENOTSUP)); 650 651 if (featureflags & DMU_BACKUP_FEATURE_RAW) { 652 /* raw receives require the encryption feature */ 653 if (!spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_ENCRYPTION)) 654 return (SET_ERROR(ENOTSUP)); 655 656 /* embedded data is incompatible with encryption and raw recv */ 657 if (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) 658 return (SET_ERROR(EINVAL)); 659 660 /* raw receives require spill block allocation flag */ 661 if (!(flags & DRR_FLAG_SPILL_BLOCK)) 662 return (SET_ERROR(ZFS_ERR_SPILL_BLOCK_FLAG_MISSING)); 663 } else { 664 /* 665 * We support unencrypted datasets below encrypted ones now, 666 * so add the DS_HOLD_FLAG_DECRYPT flag only if we are dealing 667 * with a dataset we may encrypt. 668 */ 669 if (drba->drba_dcp == NULL || 670 drba->drba_dcp->cp_crypt != ZIO_CRYPT_OFF) { 671 dsflags |= DS_HOLD_FLAG_DECRYPT; 672 } 673 } 674 675 error = dsl_dataset_hold_flags(dp, tofs, dsflags, FTAG, &ds); 676 if (error == 0) { 677 /* target fs already exists; recv into temp clone */ 678 679 /* Can't recv a clone into an existing fs */ 680 if (flags & DRR_FLAG_CLONE || drba->drba_origin) { 681 dsl_dataset_rele_flags(ds, dsflags, FTAG); 682 return (SET_ERROR(EINVAL)); 683 } 684 685 error = recv_begin_check_existing_impl(drba, ds, fromguid, 686 featureflags); 687 dsl_dataset_rele_flags(ds, dsflags, FTAG); 688 } else if (error == ENOENT) { 689 /* target fs does not exist; must be a full backup or clone */ 690 char buf[ZFS_MAX_DATASET_NAME_LEN]; 691 objset_t *os; 692 693 /* healing recv must be done "into" an existing snapshot */ 694 if (drba->drba_cookie->drc_heal == B_TRUE) 695 return (SET_ERROR(ENOTSUP)); 696 697 /* 698 * If it's a non-clone incremental, we are missing the 699 * target fs, so fail the recv. 700 */ 701 if (fromguid != 0 && !((flags & DRR_FLAG_CLONE) || 702 drba->drba_origin)) 703 return (SET_ERROR(ENOENT)); 704 705 /* 706 * If we're receiving a full send as a clone, and it doesn't 707 * contain all the necessary free records and freeobject 708 * records, reject it. 709 */ 710 if (fromguid == 0 && drba->drba_origin != NULL && 711 !(flags & DRR_FLAG_FREERECORDS)) 712 return (SET_ERROR(EINVAL)); 713 714 /* Open the parent of tofs */ 715 ASSERT3U(strlen(tofs), <, sizeof (buf)); 716 (void) strlcpy(buf, tofs, strrchr(tofs, '/') - tofs + 1); 717 error = dsl_dataset_hold(dp, buf, FTAG, &ds); 718 if (error != 0) 719 return (error); 720 721 if ((featureflags & DMU_BACKUP_FEATURE_RAW) == 0 && 722 drba->drba_origin == NULL) { 723 boolean_t will_encrypt; 724 725 /* 726 * Check that we aren't breaking any encryption rules 727 * and that we have all the parameters we need to 728 * create an encrypted dataset if necessary. If we are 729 * making an encrypted dataset the stream can't have 730 * embedded data. 731 */ 732 error = dmu_objset_create_crypt_check(ds->ds_dir, 733 drba->drba_dcp, &will_encrypt); 734 if (error != 0) { 735 dsl_dataset_rele(ds, FTAG); 736 return (error); 737 } 738 739 if (will_encrypt && 740 (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)) { 741 dsl_dataset_rele(ds, FTAG); 742 return (SET_ERROR(EINVAL)); 743 } 744 } 745 746 /* 747 * Check filesystem and snapshot limits before receiving. We'll 748 * recheck snapshot limits again at the end (we create the 749 * filesystems and increment those counts during begin_sync). 750 */ 751 error = dsl_fs_ss_limit_check(ds->ds_dir, 1, 752 ZFS_PROP_FILESYSTEM_LIMIT, NULL, 753 drba->drba_cred, drba->drba_proc); 754 if (error != 0) { 755 dsl_dataset_rele(ds, FTAG); 756 return (error); 757 } 758 759 error = dsl_fs_ss_limit_check(ds->ds_dir, 1, 760 ZFS_PROP_SNAPSHOT_LIMIT, NULL, 761 drba->drba_cred, drba->drba_proc); 762 if (error != 0) { 763 dsl_dataset_rele(ds, FTAG); 764 return (error); 765 } 766 767 /* can't recv below anything but filesystems (eg. no ZVOLs) */ 768 error = dmu_objset_from_ds(ds, &os); 769 if (error != 0) { 770 dsl_dataset_rele(ds, FTAG); 771 return (error); 772 } 773 if (dmu_objset_type(os) != DMU_OST_ZFS) { 774 dsl_dataset_rele(ds, FTAG); 775 return (SET_ERROR(ZFS_ERR_WRONG_PARENT)); 776 } 777 778 if (drba->drba_origin != NULL) { 779 dsl_dataset_t *origin; 780 error = dsl_dataset_hold_flags(dp, drba->drba_origin, 781 dsflags, FTAG, &origin); 782 if (error != 0) { 783 dsl_dataset_rele(ds, FTAG); 784 return (error); 785 } 786 if (!origin->ds_is_snapshot) { 787 dsl_dataset_rele_flags(origin, dsflags, FTAG); 788 dsl_dataset_rele(ds, FTAG); 789 return (SET_ERROR(EINVAL)); 790 } 791 if (dsl_dataset_phys(origin)->ds_guid != fromguid && 792 fromguid != 0) { 793 dsl_dataset_rele_flags(origin, dsflags, FTAG); 794 dsl_dataset_rele(ds, FTAG); 795 return (SET_ERROR(ENODEV)); 796 } 797 798 if (origin->ds_dir->dd_crypto_obj != 0 && 799 (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)) { 800 dsl_dataset_rele_flags(origin, dsflags, FTAG); 801 dsl_dataset_rele(ds, FTAG); 802 return (SET_ERROR(EINVAL)); 803 } 804 805 /* 806 * If the origin is redacted we need to verify that this 807 * send stream can safely be received on top of the 808 * origin. 809 */ 810 if (dsl_dataset_feature_is_active(origin, 811 SPA_FEATURE_REDACTED_DATASETS)) { 812 if (!redact_check(drba, origin)) { 813 dsl_dataset_rele_flags(origin, dsflags, 814 FTAG); 815 dsl_dataset_rele_flags(ds, dsflags, 816 FTAG); 817 return (SET_ERROR(EINVAL)); 818 } 819 } 820 821 error = recv_check_large_blocks(ds, featureflags); 822 if (error != 0) { 823 dsl_dataset_rele_flags(origin, dsflags, FTAG); 824 dsl_dataset_rele_flags(ds, dsflags, FTAG); 825 return (error); 826 } 827 828 dsl_dataset_rele_flags(origin, dsflags, FTAG); 829 } 830 831 dsl_dataset_rele(ds, FTAG); 832 error = 0; 833 } 834 return (error); 835 } 836 837 static void 838 dmu_recv_begin_sync(void *arg, dmu_tx_t *tx) 839 { 840 dmu_recv_begin_arg_t *drba = arg; 841 dsl_pool_t *dp = dmu_tx_pool(tx); 842 objset_t *mos = dp->dp_meta_objset; 843 dmu_recv_cookie_t *drc = drba->drba_cookie; 844 struct drr_begin *drrb = drc->drc_drrb; 845 const char *tofs = drc->drc_tofs; 846 uint64_t featureflags = drc->drc_featureflags; 847 dsl_dataset_t *ds, *newds; 848 objset_t *os; 849 uint64_t dsobj; 850 ds_hold_flags_t dsflags = DS_HOLD_FLAG_NONE; 851 int error; 852 uint64_t crflags = 0; 853 dsl_crypto_params_t dummy_dcp = { 0 }; 854 dsl_crypto_params_t *dcp = drba->drba_dcp; 855 856 if (drrb->drr_flags & DRR_FLAG_CI_DATA) 857 crflags |= DS_FLAG_CI_DATASET; 858 859 if ((featureflags & DMU_BACKUP_FEATURE_RAW) == 0) 860 dsflags |= DS_HOLD_FLAG_DECRYPT; 861 862 /* 863 * Raw, non-incremental recvs always use a dummy dcp with 864 * the raw cmd set. Raw incremental recvs do not use a dcp 865 * since the encryption parameters are already set in stone. 866 */ 867 if (dcp == NULL && drrb->drr_fromguid == 0 && 868 drba->drba_origin == NULL) { 869 ASSERT3P(dcp, ==, NULL); 870 dcp = &dummy_dcp; 871 872 if (featureflags & DMU_BACKUP_FEATURE_RAW) 873 dcp->cp_cmd = DCP_CMD_RAW_RECV; 874 } 875 876 error = dsl_dataset_hold_flags(dp, tofs, dsflags, FTAG, &ds); 877 if (error == 0) { 878 /* Create temporary clone unless we're doing corrective recv */ 879 dsl_dataset_t *snap = NULL; 880 881 if (drba->drba_cookie->drc_fromsnapobj != 0) { 882 VERIFY0(dsl_dataset_hold_obj(dp, 883 drba->drba_cookie->drc_fromsnapobj, FTAG, &snap)); 884 ASSERT3P(dcp, ==, NULL); 885 } 886 if (drc->drc_heal) { 887 /* When healing we want to use the provided snapshot */ 888 VERIFY0(dsl_dataset_snap_lookup(ds, drc->drc_tosnap, 889 &dsobj)); 890 } else { 891 dsobj = dsl_dataset_create_sync(ds->ds_dir, 892 recv_clone_name, snap, crflags, drba->drba_cred, 893 dcp, tx); 894 } 895 if (drba->drba_cookie->drc_fromsnapobj != 0) 896 dsl_dataset_rele(snap, FTAG); 897 dsl_dataset_rele_flags(ds, dsflags, FTAG); 898 } else { 899 dsl_dir_t *dd; 900 const char *tail; 901 dsl_dataset_t *origin = NULL; 902 903 VERIFY0(dsl_dir_hold(dp, tofs, FTAG, &dd, &tail)); 904 905 if (drba->drba_origin != NULL) { 906 VERIFY0(dsl_dataset_hold(dp, drba->drba_origin, 907 FTAG, &origin)); 908 ASSERT3P(dcp, ==, NULL); 909 } 910 911 /* Create new dataset. */ 912 dsobj = dsl_dataset_create_sync(dd, strrchr(tofs, '/') + 1, 913 origin, crflags, drba->drba_cred, dcp, tx); 914 if (origin != NULL) 915 dsl_dataset_rele(origin, FTAG); 916 dsl_dir_rele(dd, FTAG); 917 drc->drc_newfs = B_TRUE; 918 } 919 VERIFY0(dsl_dataset_own_obj_force(dp, dsobj, dsflags, dmu_recv_tag, 920 &newds)); 921 if (dsl_dataset_feature_is_active(newds, 922 SPA_FEATURE_REDACTED_DATASETS)) { 923 /* 924 * If the origin dataset is redacted, the child will be redacted 925 * when we create it. We clear the new dataset's 926 * redaction info; if it should be redacted, we'll fill 927 * in its information later. 928 */ 929 dsl_dataset_deactivate_feature(newds, 930 SPA_FEATURE_REDACTED_DATASETS, tx); 931 } 932 VERIFY0(dmu_objset_from_ds(newds, &os)); 933 934 if (drc->drc_resumable) { 935 dsl_dataset_zapify(newds, tx); 936 if (drrb->drr_fromguid != 0) { 937 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_FROMGUID, 938 8, 1, &drrb->drr_fromguid, tx)); 939 } 940 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_TOGUID, 941 8, 1, &drrb->drr_toguid, tx)); 942 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_TONAME, 943 1, strlen(drrb->drr_toname) + 1, drrb->drr_toname, tx)); 944 uint64_t one = 1; 945 uint64_t zero = 0; 946 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_OBJECT, 947 8, 1, &one, tx)); 948 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_OFFSET, 949 8, 1, &zero, tx)); 950 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_BYTES, 951 8, 1, &zero, tx)); 952 if (featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) { 953 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_LARGEBLOCK, 954 8, 1, &one, tx)); 955 } 956 if (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) { 957 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_EMBEDOK, 958 8, 1, &one, tx)); 959 } 960 if (featureflags & DMU_BACKUP_FEATURE_COMPRESSED) { 961 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_COMPRESSOK, 962 8, 1, &one, tx)); 963 } 964 if (featureflags & DMU_BACKUP_FEATURE_RAW) { 965 VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_RAWOK, 966 8, 1, &one, tx)); 967 } 968 969 uint64_t *redact_snaps; 970 uint_t numredactsnaps; 971 if (nvlist_lookup_uint64_array(drc->drc_begin_nvl, 972 BEGINNV_REDACT_FROM_SNAPS, &redact_snaps, 973 &numredactsnaps) == 0) { 974 VERIFY0(zap_add(mos, dsobj, 975 DS_FIELD_RESUME_REDACT_BOOKMARK_SNAPS, 976 sizeof (*redact_snaps), numredactsnaps, 977 redact_snaps, tx)); 978 } 979 } 980 981 /* 982 * Usually the os->os_encrypted value is tied to the presence of a 983 * DSL Crypto Key object in the dd. However, that will not be received 984 * until dmu_recv_stream(), so we set the value manually for now. 985 */ 986 if (featureflags & DMU_BACKUP_FEATURE_RAW) { 987 os->os_encrypted = B_TRUE; 988 drba->drba_cookie->drc_raw = B_TRUE; 989 } 990 991 if (featureflags & DMU_BACKUP_FEATURE_REDACTED) { 992 uint64_t *redact_snaps; 993 uint_t numredactsnaps; 994 VERIFY0(nvlist_lookup_uint64_array(drc->drc_begin_nvl, 995 BEGINNV_REDACT_SNAPS, &redact_snaps, &numredactsnaps)); 996 dsl_dataset_activate_redaction(newds, redact_snaps, 997 numredactsnaps, tx); 998 } 999 1000 if (featureflags & DMU_BACKUP_FEATURE_LARGE_MICROZAP) { 1001 /* 1002 * The source has seen a large microzap at least once in its 1003 * life, so we activate the feature here to match. It's not 1004 * strictly necessary since a large microzap is usable without 1005 * the feature active, but if that object is sent on from here, 1006 * we need this info to know to add the stream feature. 1007 * 1008 * There may be no large microzap in the incoming stream, or 1009 * ever again, but this is a very niche feature and its very 1010 * difficult to spot a large microzap in the stream, so its 1011 * not worth the effort of trying harder to activate the 1012 * feature at first use. 1013 */ 1014 dsl_dataset_activate_feature(dsobj, SPA_FEATURE_LARGE_MICROZAP, 1015 (void *)B_TRUE, tx); 1016 } 1017 1018 dmu_buf_will_dirty(newds->ds_dbuf, tx); 1019 dsl_dataset_phys(newds)->ds_flags |= DS_FLAG_INCONSISTENT; 1020 1021 /* 1022 * Activate longname feature if received 1023 */ 1024 if (featureflags & DMU_BACKUP_FEATURE_LONGNAME && 1025 !dsl_dataset_feature_is_active(newds, SPA_FEATURE_LONGNAME)) { 1026 dsl_dataset_activate_feature(newds->ds_object, 1027 SPA_FEATURE_LONGNAME, (void *)B_TRUE, tx); 1028 newds->ds_feature[SPA_FEATURE_LONGNAME] = (void *)B_TRUE; 1029 } 1030 1031 /* 1032 * If we actually created a non-clone, we need to create the objset 1033 * in our new dataset. If this is a raw send we postpone this until 1034 * dmu_recv_stream() so that we can allocate the metadnode with the 1035 * properties from the DRR_BEGIN payload. 1036 */ 1037 rrw_enter(&newds->ds_bp_rwlock, RW_READER, FTAG); 1038 if (BP_IS_HOLE(dsl_dataset_get_blkptr(newds)) && 1039 (featureflags & DMU_BACKUP_FEATURE_RAW) == 0 && 1040 !drc->drc_heal) { 1041 (void) dmu_objset_create_impl(dp->dp_spa, 1042 newds, dsl_dataset_get_blkptr(newds), drrb->drr_type, tx); 1043 } 1044 rrw_exit(&newds->ds_bp_rwlock, FTAG); 1045 1046 drba->drba_cookie->drc_ds = newds; 1047 drba->drba_cookie->drc_os = os; 1048 1049 spa_history_log_internal_ds(newds, "receive", tx, " "); 1050 } 1051 1052 static int 1053 dmu_recv_resume_begin_check(void *arg, dmu_tx_t *tx) 1054 { 1055 dmu_recv_begin_arg_t *drba = arg; 1056 dmu_recv_cookie_t *drc = drba->drba_cookie; 1057 dsl_pool_t *dp = dmu_tx_pool(tx); 1058 struct drr_begin *drrb = drc->drc_drrb; 1059 int error; 1060 ds_hold_flags_t dsflags = DS_HOLD_FLAG_NONE; 1061 dsl_dataset_t *ds; 1062 const char *tofs = drc->drc_tofs; 1063 1064 /* already checked */ 1065 ASSERT3U(drrb->drr_magic, ==, DMU_BACKUP_MAGIC); 1066 ASSERT(drc->drc_featureflags & DMU_BACKUP_FEATURE_RESUMING); 1067 1068 if (DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) == 1069 DMU_COMPOUNDSTREAM || 1070 drrb->drr_type >= DMU_OST_NUMTYPES) 1071 return (SET_ERROR(EINVAL)); 1072 1073 /* 1074 * This is mostly a sanity check since we should have already done these 1075 * checks during a previous attempt to receive the data. 1076 */ 1077 error = recv_begin_check_feature_flags_impl(drc->drc_featureflags, 1078 dp->dp_spa); 1079 if (error != 0) 1080 return (error); 1081 1082 /* 6 extra bytes for /%recv */ 1083 char recvname[ZFS_MAX_DATASET_NAME_LEN + 6]; 1084 1085 (void) snprintf(recvname, sizeof (recvname), "%s/%s", 1086 tofs, recv_clone_name); 1087 1088 if (drc->drc_featureflags & DMU_BACKUP_FEATURE_RAW) { 1089 /* raw receives require spill block allocation flag */ 1090 if (!(drrb->drr_flags & DRR_FLAG_SPILL_BLOCK)) 1091 return (SET_ERROR(ZFS_ERR_SPILL_BLOCK_FLAG_MISSING)); 1092 } else { 1093 dsflags |= DS_HOLD_FLAG_DECRYPT; 1094 } 1095 1096 boolean_t recvexist = B_TRUE; 1097 if (dsl_dataset_hold_flags(dp, recvname, dsflags, FTAG, &ds) != 0) { 1098 /* %recv does not exist; continue in tofs */ 1099 recvexist = B_FALSE; 1100 error = dsl_dataset_hold_flags(dp, tofs, dsflags, FTAG, &ds); 1101 if (error != 0) 1102 return (error); 1103 } 1104 1105 /* 1106 * Resume of full/newfs recv on existing dataset should be done with 1107 * force flag 1108 */ 1109 if (recvexist && drrb->drr_fromguid == 0 && !drc->drc_force) { 1110 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1111 return (SET_ERROR(ZFS_ERR_RESUME_EXISTS)); 1112 } 1113 1114 /* check that ds is marked inconsistent */ 1115 if (!DS_IS_INCONSISTENT(ds)) { 1116 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1117 return (SET_ERROR(EINVAL)); 1118 } 1119 1120 /* check that there is resuming data, and that the toguid matches */ 1121 if (!dsl_dataset_is_zapified(ds)) { 1122 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1123 return (SET_ERROR(EINVAL)); 1124 } 1125 uint64_t val; 1126 error = zap_lookup(dp->dp_meta_objset, ds->ds_object, 1127 DS_FIELD_RESUME_TOGUID, sizeof (val), 1, &val); 1128 if (error != 0 || drrb->drr_toguid != val) { 1129 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1130 return (SET_ERROR(EINVAL)); 1131 } 1132 1133 /* 1134 * Check if the receive is still running. If so, it will be owned. 1135 * Note that nothing else can own the dataset (e.g. after the receive 1136 * fails) because it will be marked inconsistent. 1137 */ 1138 if (dsl_dataset_has_owner(ds)) { 1139 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1140 return (SET_ERROR(EBUSY)); 1141 } 1142 1143 /* There should not be any snapshots of this fs yet. */ 1144 if (ds->ds_prev != NULL && ds->ds_prev->ds_dir == ds->ds_dir) { 1145 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1146 return (SET_ERROR(EINVAL)); 1147 } 1148 1149 /* 1150 * Note: resume point will be checked when we process the first WRITE 1151 * record. 1152 */ 1153 1154 /* check that the origin matches */ 1155 val = 0; 1156 (void) zap_lookup(dp->dp_meta_objset, ds->ds_object, 1157 DS_FIELD_RESUME_FROMGUID, sizeof (val), 1, &val); 1158 if (drrb->drr_fromguid != val) { 1159 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1160 return (SET_ERROR(EINVAL)); 1161 } 1162 1163 if (ds->ds_prev != NULL && drrb->drr_fromguid != 0) 1164 drc->drc_fromsnapobj = ds->ds_prev->ds_object; 1165 1166 /* 1167 * If we're resuming, and the send is redacted, then the original send 1168 * must have been redacted, and must have been redacted with respect to 1169 * the same snapshots. 1170 */ 1171 if (drc->drc_featureflags & DMU_BACKUP_FEATURE_REDACTED) { 1172 uint64_t num_ds_redact_snaps; 1173 uint64_t *ds_redact_snaps; 1174 1175 uint_t num_stream_redact_snaps; 1176 uint64_t *stream_redact_snaps; 1177 1178 if (nvlist_lookup_uint64_array(drc->drc_begin_nvl, 1179 BEGINNV_REDACT_SNAPS, &stream_redact_snaps, 1180 &num_stream_redact_snaps) != 0) { 1181 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1182 return (SET_ERROR(EINVAL)); 1183 } 1184 1185 if (!dsl_dataset_get_uint64_array_feature(ds, 1186 SPA_FEATURE_REDACTED_DATASETS, &num_ds_redact_snaps, 1187 &ds_redact_snaps)) { 1188 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1189 return (SET_ERROR(EINVAL)); 1190 } 1191 1192 for (int i = 0; i < num_ds_redact_snaps; i++) { 1193 if (!redact_snaps_contains(ds_redact_snaps, 1194 num_ds_redact_snaps, stream_redact_snaps[i])) { 1195 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1196 return (SET_ERROR(EINVAL)); 1197 } 1198 } 1199 } 1200 1201 error = recv_check_large_blocks(ds, drc->drc_featureflags); 1202 if (error != 0) { 1203 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1204 return (error); 1205 } 1206 1207 dsl_dataset_rele_flags(ds, dsflags, FTAG); 1208 return (0); 1209 } 1210 1211 static void 1212 dmu_recv_resume_begin_sync(void *arg, dmu_tx_t *tx) 1213 { 1214 dmu_recv_begin_arg_t *drba = arg; 1215 dsl_pool_t *dp = dmu_tx_pool(tx); 1216 const char *tofs = drba->drba_cookie->drc_tofs; 1217 uint64_t featureflags = drba->drba_cookie->drc_featureflags; 1218 dsl_dataset_t *ds; 1219 ds_hold_flags_t dsflags = DS_HOLD_FLAG_NONE; 1220 /* 6 extra bytes for /%recv */ 1221 char recvname[ZFS_MAX_DATASET_NAME_LEN + 6]; 1222 1223 (void) snprintf(recvname, sizeof (recvname), "%s/%s", tofs, 1224 recv_clone_name); 1225 1226 if (featureflags & DMU_BACKUP_FEATURE_RAW) { 1227 drba->drba_cookie->drc_raw = B_TRUE; 1228 } else { 1229 dsflags |= DS_HOLD_FLAG_DECRYPT; 1230 } 1231 1232 if (dsl_dataset_own_force(dp, recvname, dsflags, dmu_recv_tag, &ds) 1233 != 0) { 1234 /* %recv does not exist; continue in tofs */ 1235 VERIFY0(dsl_dataset_own_force(dp, tofs, dsflags, dmu_recv_tag, 1236 &ds)); 1237 drba->drba_cookie->drc_newfs = B_TRUE; 1238 } 1239 1240 ASSERT(DS_IS_INCONSISTENT(ds)); 1241 rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG); 1242 ASSERT(!BP_IS_HOLE(dsl_dataset_get_blkptr(ds)) || 1243 drba->drba_cookie->drc_raw); 1244 rrw_exit(&ds->ds_bp_rwlock, FTAG); 1245 1246 drba->drba_cookie->drc_ds = ds; 1247 VERIFY0(dmu_objset_from_ds(ds, &drba->drba_cookie->drc_os)); 1248 drba->drba_cookie->drc_should_save = B_TRUE; 1249 1250 spa_history_log_internal_ds(ds, "resume receive", tx, " "); 1251 } 1252 1253 /* 1254 * NB: callers *MUST* call dmu_recv_stream() if dmu_recv_begin() 1255 * succeeds; otherwise we will leak the holds on the datasets. 1256 */ 1257 int 1258 dmu_recv_begin(const char *tofs, const char *tosnap, 1259 dmu_replay_record_t *drr_begin, boolean_t force, boolean_t heal, 1260 boolean_t resumable, nvlist_t *localprops, nvlist_t *hidden_args, 1261 const char *origin, dmu_recv_cookie_t *drc, zfs_file_t *fp, 1262 offset_t *voffp) 1263 { 1264 dmu_recv_begin_arg_t drba = { 0 }; 1265 int err = 0; 1266 1267 memset(drc, 0, sizeof (dmu_recv_cookie_t)); 1268 drc->drc_drr_begin = drr_begin; 1269 drc->drc_drrb = &drr_begin->drr_u.drr_begin; 1270 drc->drc_tosnap = tosnap; 1271 drc->drc_tofs = tofs; 1272 drc->drc_force = force; 1273 drc->drc_heal = heal; 1274 drc->drc_resumable = resumable; 1275 drc->drc_cred = CRED(); 1276 drc->drc_proc = curproc; 1277 drc->drc_clone = (origin != NULL); 1278 1279 if (drc->drc_drrb->drr_magic == BSWAP_64(DMU_BACKUP_MAGIC)) { 1280 drc->drc_byteswap = B_TRUE; 1281 (void) fletcher_4_incremental_byteswap(drr_begin, 1282 sizeof (dmu_replay_record_t), &drc->drc_cksum); 1283 byteswap_record(drr_begin); 1284 } else if (drc->drc_drrb->drr_magic == DMU_BACKUP_MAGIC) { 1285 (void) fletcher_4_incremental_native(drr_begin, 1286 sizeof (dmu_replay_record_t), &drc->drc_cksum); 1287 } else { 1288 return (SET_ERROR(EINVAL)); 1289 } 1290 1291 drc->drc_fp = fp; 1292 drc->drc_voff = *voffp; 1293 drc->drc_featureflags = 1294 DMU_GET_FEATUREFLAGS(drc->drc_drrb->drr_versioninfo); 1295 1296 uint32_t payloadlen = drc->drc_drr_begin->drr_payloadlen; 1297 1298 /* 1299 * Since OpenZFS 2.0.0, we have enforced a 64MB limit in userspace 1300 * configurable via ZFS_SENDRECV_MAX_NVLIST. We enforce 256MB as a hard 1301 * upper limit. Systems with less than 1GB of RAM will see a lower 1302 * limit from `arc_all_memory() / 4`. 1303 */ 1304 if (payloadlen > (MIN((1U << 28), arc_all_memory() / 4))) 1305 return (E2BIG); 1306 1307 1308 if (payloadlen != 0) { 1309 void *payload = vmem_alloc(payloadlen, KM_SLEEP); 1310 /* 1311 * For compatibility with recursive send streams, we don't do 1312 * this here if the stream could be part of a package. Instead, 1313 * we'll do it in dmu_recv_stream. If we pull the next header 1314 * too early, and it's the END record, we break the `recv_skip` 1315 * logic. 1316 */ 1317 1318 err = receive_read_payload_and_next_header(drc, payloadlen, 1319 payload); 1320 if (err != 0) { 1321 vmem_free(payload, payloadlen); 1322 return (err); 1323 } 1324 err = nvlist_unpack(payload, payloadlen, &drc->drc_begin_nvl, 1325 KM_SLEEP); 1326 vmem_free(payload, payloadlen); 1327 if (err != 0) { 1328 kmem_free(drc->drc_next_rrd, 1329 sizeof (*drc->drc_next_rrd)); 1330 return (err); 1331 } 1332 } 1333 1334 if (drc->drc_drrb->drr_flags & DRR_FLAG_SPILL_BLOCK) 1335 drc->drc_spill = B_TRUE; 1336 1337 drba.drba_origin = origin; 1338 drba.drba_cookie = drc; 1339 drba.drba_cred = CRED(); 1340 drba.drba_proc = curproc; 1341 1342 if (drc->drc_featureflags & DMU_BACKUP_FEATURE_RESUMING) { 1343 err = dsl_sync_task(tofs, 1344 dmu_recv_resume_begin_check, dmu_recv_resume_begin_sync, 1345 &drba, 5, ZFS_SPACE_CHECK_NORMAL); 1346 } else { 1347 /* 1348 * For non-raw, non-incremental, non-resuming receives the 1349 * user can specify encryption parameters on the command line 1350 * with "zfs recv -o". For these receives we create a dcp and 1351 * pass it to the sync task. Creating the dcp will implicitly 1352 * remove the encryption params from the localprops nvlist, 1353 * which avoids errors when trying to set these normally 1354 * read-only properties. Any other kind of receive that 1355 * attempts to set these properties will fail as a result. 1356 */ 1357 if ((DMU_GET_FEATUREFLAGS(drc->drc_drrb->drr_versioninfo) & 1358 DMU_BACKUP_FEATURE_RAW) == 0 && 1359 origin == NULL && drc->drc_drrb->drr_fromguid == 0) { 1360 err = dsl_crypto_params_create_nvlist(DCP_CMD_NONE, 1361 localprops, hidden_args, &drba.drba_dcp); 1362 } 1363 1364 if (err == 0) { 1365 err = dsl_sync_task(tofs, 1366 dmu_recv_begin_check, dmu_recv_begin_sync, 1367 &drba, 5, ZFS_SPACE_CHECK_NORMAL); 1368 dsl_crypto_params_free(drba.drba_dcp, !!err); 1369 } 1370 } 1371 1372 if (err != 0) { 1373 kmem_free(drc->drc_next_rrd, sizeof (*drc->drc_next_rrd)); 1374 nvlist_free(drc->drc_begin_nvl); 1375 } 1376 return (err); 1377 } 1378 1379 /* 1380 * Holds data need for corrective recv callback 1381 */ 1382 typedef struct cr_cb_data { 1383 uint64_t size; 1384 zbookmark_phys_t zb; 1385 spa_t *spa; 1386 } cr_cb_data_t; 1387 1388 static void 1389 corrective_read_done(zio_t *zio) 1390 { 1391 cr_cb_data_t *data = zio->io_private; 1392 /* Corruption corrected; update error log if needed */ 1393 if (zio->io_error == 0) { 1394 spa_remove_error(data->spa, &data->zb, 1395 BP_GET_LOGICAL_BIRTH(zio->io_bp)); 1396 } 1397 kmem_free(data, sizeof (cr_cb_data_t)); 1398 abd_free(zio->io_abd); 1399 } 1400 1401 /* 1402 * zio_rewrite the data pointed to by bp with the data from the rrd's abd. 1403 */ 1404 static int 1405 do_corrective_recv(struct receive_writer_arg *rwa, struct drr_write *drrw, 1406 struct receive_record_arg *rrd, blkptr_t *bp) 1407 { 1408 int err; 1409 zio_t *io; 1410 zbookmark_phys_t zb; 1411 dnode_t *dn; 1412 abd_t *abd = rrd->abd; 1413 zio_cksum_t bp_cksum = bp->blk_cksum; 1414 zio_flag_t flags = ZIO_FLAG_SPECULATIVE | ZIO_FLAG_DONT_RETRY | 1415 ZIO_FLAG_CANFAIL; 1416 1417 if (rwa->raw) 1418 flags |= ZIO_FLAG_RAW; 1419 1420 err = dnode_hold(rwa->os, drrw->drr_object, FTAG, &dn); 1421 if (err != 0) 1422 return (err); 1423 SET_BOOKMARK(&zb, dmu_objset_id(rwa->os), drrw->drr_object, 0, 1424 dbuf_whichblock(dn, 0, drrw->drr_offset)); 1425 dnode_rele(dn, FTAG); 1426 1427 if (!rwa->raw && DRR_WRITE_COMPRESSED(drrw)) { 1428 /* Decompress the stream data */ 1429 abd_t *dabd = abd_alloc_linear( 1430 drrw->drr_logical_size, B_FALSE); 1431 err = zio_decompress_data(drrw->drr_compressiontype, 1432 abd, dabd, abd_get_size(abd), 1433 abd_get_size(dabd), NULL); 1434 1435 if (err != 0) { 1436 abd_free(dabd); 1437 return (err); 1438 } 1439 /* Swap in the newly decompressed data into the abd */ 1440 abd_free(abd); 1441 abd = dabd; 1442 } 1443 1444 if (!rwa->raw && BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF) { 1445 /* Recompress the data */ 1446 abd_t *cabd = abd_alloc_linear(BP_GET_PSIZE(bp), 1447 B_FALSE); 1448 uint64_t csize = zio_compress_data(BP_GET_COMPRESS(bp), 1449 abd, &cabd, abd_get_size(abd), BP_GET_PSIZE(bp), 1450 rwa->os->os_complevel); 1451 abd_zero_off(cabd, csize, BP_GET_PSIZE(bp) - csize); 1452 /* Swap in newly compressed data into the abd */ 1453 abd_free(abd); 1454 abd = cabd; 1455 flags |= ZIO_FLAG_RAW_COMPRESS; 1456 } 1457 1458 /* 1459 * The stream is not encrypted but the data on-disk is. 1460 * We need to re-encrypt the buf using the same 1461 * encryption type, salt, iv, and mac that was used to encrypt 1462 * the block previosly. 1463 */ 1464 if (!rwa->raw && BP_USES_CRYPT(bp)) { 1465 dsl_dataset_t *ds; 1466 dsl_crypto_key_t *dck = NULL; 1467 uint8_t salt[ZIO_DATA_SALT_LEN]; 1468 uint8_t iv[ZIO_DATA_IV_LEN]; 1469 uint8_t mac[ZIO_DATA_MAC_LEN]; 1470 boolean_t no_crypt = B_FALSE; 1471 dsl_pool_t *dp = dmu_objset_pool(rwa->os); 1472 abd_t *eabd = abd_alloc_linear(BP_GET_PSIZE(bp), B_FALSE); 1473 1474 zio_crypt_decode_params_bp(bp, salt, iv); 1475 zio_crypt_decode_mac_bp(bp, mac); 1476 1477 dsl_pool_config_enter(dp, FTAG); 1478 err = dsl_dataset_hold_flags(dp, rwa->tofs, 1479 DS_HOLD_FLAG_DECRYPT, FTAG, &ds); 1480 if (err != 0) { 1481 dsl_pool_config_exit(dp, FTAG); 1482 abd_free(eabd); 1483 return (SET_ERROR(EACCES)); 1484 } 1485 1486 /* Look up the key from the spa's keystore */ 1487 err = spa_keystore_lookup_key(rwa->os->os_spa, 1488 zb.zb_objset, FTAG, &dck); 1489 if (err != 0) { 1490 dsl_dataset_rele_flags(ds, DS_HOLD_FLAG_DECRYPT, 1491 FTAG); 1492 dsl_pool_config_exit(dp, FTAG); 1493 abd_free(eabd); 1494 return (SET_ERROR(EACCES)); 1495 } 1496 1497 err = zio_do_crypt_abd(B_TRUE, &dck->dck_key, 1498 BP_GET_TYPE(bp), BP_SHOULD_BYTESWAP(bp), salt, iv, 1499 mac, abd_get_size(abd), abd, eabd, &no_crypt); 1500 1501 spa_keystore_dsl_key_rele(rwa->os->os_spa, dck, FTAG); 1502 dsl_dataset_rele_flags(ds, DS_HOLD_FLAG_DECRYPT, FTAG); 1503 dsl_pool_config_exit(dp, FTAG); 1504 1505 ASSERT0(no_crypt); 1506 if (err != 0) { 1507 abd_free(eabd); 1508 return (err); 1509 } 1510 /* Swap in the newly encrypted data into the abd */ 1511 abd_free(abd); 1512 abd = eabd; 1513 1514 /* 1515 * We want to prevent zio_rewrite() from trying to 1516 * encrypt the data again 1517 */ 1518 flags |= ZIO_FLAG_RAW_ENCRYPT; 1519 } 1520 rrd->abd = abd; 1521 1522 io = zio_rewrite(NULL, rwa->os->os_spa, BP_GET_LOGICAL_BIRTH(bp), bp, 1523 abd, BP_GET_PSIZE(bp), NULL, NULL, ZIO_PRIORITY_SYNC_WRITE, flags, 1524 &zb); 1525 1526 ASSERT(abd_get_size(abd) == BP_GET_LSIZE(bp) || 1527 abd_get_size(abd) == BP_GET_PSIZE(bp)); 1528 1529 /* compute new bp checksum value and make sure it matches the old one */ 1530 zio_checksum_compute(io, BP_GET_CHECKSUM(bp), abd, abd_get_size(abd)); 1531 if (!ZIO_CHECKSUM_EQUAL(bp_cksum, io->io_bp->blk_cksum)) { 1532 zio_destroy(io); 1533 if (zfs_recv_best_effort_corrective != 0) 1534 return (0); 1535 return (SET_ERROR(ECKSUM)); 1536 } 1537 1538 /* Correct the corruption in place */ 1539 err = zio_wait(io); 1540 if (err == 0) { 1541 cr_cb_data_t *cb_data = 1542 kmem_alloc(sizeof (cr_cb_data_t), KM_SLEEP); 1543 cb_data->spa = rwa->os->os_spa; 1544 cb_data->size = drrw->drr_logical_size; 1545 cb_data->zb = zb; 1546 /* Test if healing worked by re-reading the bp */ 1547 err = zio_wait(zio_read(rwa->heal_pio, rwa->os->os_spa, bp, 1548 abd_alloc_for_io(drrw->drr_logical_size, B_FALSE), 1549 drrw->drr_logical_size, corrective_read_done, 1550 cb_data, ZIO_PRIORITY_ASYNC_READ, flags, NULL)); 1551 } 1552 if (err != 0 && zfs_recv_best_effort_corrective != 0) 1553 err = 0; 1554 1555 return (err); 1556 } 1557 1558 static int 1559 receive_read(dmu_recv_cookie_t *drc, int len, void *buf) 1560 { 1561 int done = 0; 1562 1563 /* 1564 * The code doesn't rely on this (lengths being multiples of 8). See 1565 * comment in dump_bytes. 1566 */ 1567 ASSERT(len % 8 == 0 || 1568 (drc->drc_featureflags & DMU_BACKUP_FEATURE_RAW) != 0); 1569 1570 while (done < len) { 1571 ssize_t resid = len - done; 1572 zfs_file_t *fp = drc->drc_fp; 1573 int err = zfs_file_read(fp, (char *)buf + done, 1574 len - done, &resid); 1575 if (err == 0 && resid == len - done) { 1576 /* 1577 * Note: ECKSUM or ZFS_ERR_STREAM_TRUNCATED indicates 1578 * that the receive was interrupted and can 1579 * potentially be resumed. 1580 */ 1581 err = SET_ERROR(ZFS_ERR_STREAM_TRUNCATED); 1582 } 1583 drc->drc_voff += len - done - resid; 1584 done = len - resid; 1585 if (err != 0) 1586 return (err); 1587 } 1588 1589 drc->drc_bytes_read += len; 1590 1591 ASSERT3U(done, ==, len); 1592 return (0); 1593 } 1594 1595 static inline uint8_t 1596 deduce_nblkptr(dmu_object_type_t bonus_type, uint64_t bonus_size) 1597 { 1598 if (bonus_type == DMU_OT_SA) { 1599 return (1); 1600 } else { 1601 return (1 + 1602 ((DN_OLD_MAX_BONUSLEN - 1603 MIN(DN_OLD_MAX_BONUSLEN, bonus_size)) >> SPA_BLKPTRSHIFT)); 1604 } 1605 } 1606 1607 static void 1608 save_resume_state(struct receive_writer_arg *rwa, 1609 uint64_t object, uint64_t offset, dmu_tx_t *tx) 1610 { 1611 int txgoff = dmu_tx_get_txg(tx) & TXG_MASK; 1612 1613 if (!rwa->resumable) 1614 return; 1615 1616 /* 1617 * We use ds_resume_bytes[] != 0 to indicate that we need to 1618 * update this on disk, so it must not be 0. 1619 */ 1620 ASSERT(rwa->bytes_read != 0); 1621 1622 /* 1623 * We only resume from write records, which have a valid 1624 * (non-meta-dnode) object number. 1625 */ 1626 ASSERT(object != 0); 1627 1628 /* 1629 * For resuming to work correctly, we must receive records in order, 1630 * sorted by object,offset. This is checked by the callers, but 1631 * assert it here for good measure. 1632 */ 1633 ASSERT3U(object, >=, rwa->os->os_dsl_dataset->ds_resume_object[txgoff]); 1634 ASSERT(object != rwa->os->os_dsl_dataset->ds_resume_object[txgoff] || 1635 offset >= rwa->os->os_dsl_dataset->ds_resume_offset[txgoff]); 1636 ASSERT3U(rwa->bytes_read, >=, 1637 rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff]); 1638 1639 rwa->os->os_dsl_dataset->ds_resume_object[txgoff] = object; 1640 rwa->os->os_dsl_dataset->ds_resume_offset[txgoff] = offset; 1641 rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff] = rwa->bytes_read; 1642 } 1643 1644 static int 1645 receive_object_is_same_generation(objset_t *os, uint64_t object, 1646 dmu_object_type_t old_bonus_type, dmu_object_type_t new_bonus_type, 1647 const void *new_bonus, boolean_t *samegenp) 1648 { 1649 zfs_file_info_t zoi; 1650 int err; 1651 1652 dmu_buf_t *old_bonus_dbuf; 1653 err = dmu_bonus_hold(os, object, FTAG, &old_bonus_dbuf); 1654 if (err != 0) 1655 return (err); 1656 err = dmu_get_file_info(os, old_bonus_type, old_bonus_dbuf->db_data, 1657 &zoi); 1658 dmu_buf_rele(old_bonus_dbuf, FTAG); 1659 if (err != 0) 1660 return (err); 1661 uint64_t old_gen = zoi.zfi_generation; 1662 1663 err = dmu_get_file_info(os, new_bonus_type, new_bonus, &zoi); 1664 if (err != 0) 1665 return (err); 1666 uint64_t new_gen = zoi.zfi_generation; 1667 1668 *samegenp = (old_gen == new_gen); 1669 return (0); 1670 } 1671 1672 static int 1673 receive_handle_existing_object(const struct receive_writer_arg *rwa, 1674 const struct drr_object *drro, const dmu_object_info_t *doi, 1675 const void *bonus_data, 1676 uint64_t *object_to_hold, uint32_t *new_blksz) 1677 { 1678 uint32_t indblksz = drro->drr_indblkshift ? 1679 1ULL << drro->drr_indblkshift : 0; 1680 int nblkptr = deduce_nblkptr(drro->drr_bonustype, 1681 drro->drr_bonuslen); 1682 uint8_t dn_slots = drro->drr_dn_slots != 0 ? 1683 drro->drr_dn_slots : DNODE_MIN_SLOTS; 1684 boolean_t do_free_range = B_FALSE; 1685 int err; 1686 1687 *object_to_hold = drro->drr_object; 1688 1689 /* nblkptr should be bounded by the bonus size and type */ 1690 if (rwa->raw && nblkptr != drro->drr_nblkptr) 1691 return (SET_ERROR(EINVAL)); 1692 1693 /* 1694 * After the previous send stream, the sending system may 1695 * have freed this object, and then happened to re-allocate 1696 * this object number in a later txg. In this case, we are 1697 * receiving a different logical file, and the block size may 1698 * appear to be different. i.e. we may have a different 1699 * block size for this object than what the send stream says. 1700 * In this case we need to remove the object's contents, 1701 * so that its structure can be changed and then its contents 1702 * entirely replaced by subsequent WRITE records. 1703 * 1704 * If this is a -L (--large-block) incremental stream, and 1705 * the previous stream was not -L, the block size may appear 1706 * to increase. i.e. we may have a smaller block size for 1707 * this object than what the send stream says. In this case 1708 * we need to keep the object's contents and block size 1709 * intact, so that we don't lose parts of the object's 1710 * contents that are not changed by this incremental send 1711 * stream. 1712 * 1713 * We can distinguish between the two above cases by using 1714 * the ZPL's generation number (see 1715 * receive_object_is_same_generation()). However, we only 1716 * want to rely on the generation number when absolutely 1717 * necessary, because with raw receives, the generation is 1718 * encrypted. We also want to minimize dependence on the 1719 * ZPL, so that other types of datasets can also be received 1720 * (e.g. ZVOLs, although note that ZVOLS currently do not 1721 * reallocate their objects or change their structure). 1722 * Therefore, we check a number of different cases where we 1723 * know it is safe to discard the object's contents, before 1724 * using the ZPL's generation number to make the above 1725 * distinction. 1726 */ 1727 if (drro->drr_blksz != doi->doi_data_block_size) { 1728 if (rwa->raw) { 1729 /* 1730 * RAW streams always have large blocks, so 1731 * we are sure that the data is not needed 1732 * due to changing --large-block to be on. 1733 * Which is fortunate since the bonus buffer 1734 * (which contains the ZPL generation) is 1735 * encrypted, and the key might not be 1736 * loaded. 1737 */ 1738 do_free_range = B_TRUE; 1739 } else if (rwa->full) { 1740 /* 1741 * This is a full send stream, so it always 1742 * replaces what we have. Even if the 1743 * generation numbers happen to match, this 1744 * can not actually be the same logical file. 1745 * This is relevant when receiving a full 1746 * send as a clone. 1747 */ 1748 do_free_range = B_TRUE; 1749 } else if (drro->drr_type != 1750 DMU_OT_PLAIN_FILE_CONTENTS || 1751 doi->doi_type != DMU_OT_PLAIN_FILE_CONTENTS) { 1752 /* 1753 * PLAIN_FILE_CONTENTS are the only type of 1754 * objects that have ever been stored with 1755 * large blocks, so we don't need the special 1756 * logic below. ZAP blocks can shrink (when 1757 * there's only one block), so we don't want 1758 * to hit the error below about block size 1759 * only increasing. 1760 */ 1761 do_free_range = B_TRUE; 1762 } else if (doi->doi_max_offset <= 1763 doi->doi_data_block_size) { 1764 /* 1765 * There is only one block. We can free it, 1766 * because its contents will be replaced by a 1767 * WRITE record. This can not be the no-L -> 1768 * -L case, because the no-L case would have 1769 * resulted in multiple blocks. If we 1770 * supported -L -> no-L, it would not be safe 1771 * to free the file's contents. Fortunately, 1772 * that is not allowed (see 1773 * recv_check_large_blocks()). 1774 */ 1775 do_free_range = B_TRUE; 1776 } else { 1777 boolean_t is_same_gen; 1778 err = receive_object_is_same_generation(rwa->os, 1779 drro->drr_object, doi->doi_bonus_type, 1780 drro->drr_bonustype, bonus_data, &is_same_gen); 1781 if (err != 0) 1782 return (SET_ERROR(EINVAL)); 1783 1784 if (is_same_gen) { 1785 /* 1786 * This is the same logical file, and 1787 * the block size must be increasing. 1788 * It could only decrease if 1789 * --large-block was changed to be 1790 * off, which is checked in 1791 * recv_check_large_blocks(). 1792 */ 1793 if (drro->drr_blksz <= 1794 doi->doi_data_block_size) 1795 return (SET_ERROR(EINVAL)); 1796 /* 1797 * We keep the existing blocksize and 1798 * contents. 1799 */ 1800 *new_blksz = 1801 doi->doi_data_block_size; 1802 } else { 1803 do_free_range = B_TRUE; 1804 } 1805 } 1806 } 1807 1808 /* nblkptr can only decrease if the object was reallocated */ 1809 if (nblkptr < doi->doi_nblkptr) 1810 do_free_range = B_TRUE; 1811 1812 /* number of slots can only change on reallocation */ 1813 if (dn_slots != doi->doi_dnodesize >> DNODE_SHIFT) 1814 do_free_range = B_TRUE; 1815 1816 /* 1817 * For raw sends we also check a few other fields to 1818 * ensure we are preserving the objset structure exactly 1819 * as it was on the receive side: 1820 * - A changed indirect block size 1821 * - A smaller nlevels 1822 */ 1823 if (rwa->raw) { 1824 if (indblksz != doi->doi_metadata_block_size) 1825 do_free_range = B_TRUE; 1826 if (drro->drr_nlevels < doi->doi_indirection) 1827 do_free_range = B_TRUE; 1828 } 1829 1830 if (do_free_range) { 1831 err = dmu_free_long_range(rwa->os, drro->drr_object, 1832 0, DMU_OBJECT_END); 1833 if (err != 0) 1834 return (SET_ERROR(EINVAL)); 1835 } 1836 1837 /* 1838 * The dmu does not currently support decreasing nlevels or changing 1839 * indirect block size if there is already one, same as changing the 1840 * number of of dnode slots on an object. For non-raw sends this 1841 * does not matter and the new object can just use the previous one's 1842 * parameters. For raw sends, however, the structure of the received 1843 * dnode (including indirects and dnode slots) must match that of the 1844 * send side. Therefore, instead of using dmu_object_reclaim(), we 1845 * must free the object completely and call dmu_object_claim_dnsize() 1846 * instead. 1847 */ 1848 if ((rwa->raw && ((doi->doi_indirection > 1 && 1849 indblksz != doi->doi_metadata_block_size) || 1850 drro->drr_nlevels < doi->doi_indirection)) || 1851 dn_slots != doi->doi_dnodesize >> DNODE_SHIFT) { 1852 err = dmu_free_long_object(rwa->os, drro->drr_object); 1853 if (err != 0) 1854 return (SET_ERROR(EINVAL)); 1855 1856 txg_wait_synced(dmu_objset_pool(rwa->os), 0); 1857 *object_to_hold = DMU_NEW_OBJECT; 1858 } 1859 1860 /* 1861 * For raw receives, free everything beyond the new incoming 1862 * maxblkid. Normally this would be done with a DRR_FREE 1863 * record that would come after this DRR_OBJECT record is 1864 * processed. However, for raw receives we manually set the 1865 * maxblkid from the drr_maxblkid and so we must first free 1866 * everything above that blkid to ensure the DMU is always 1867 * consistent with itself. We will never free the first block 1868 * of the object here because a maxblkid of 0 could indicate 1869 * an object with a single block or one with no blocks. This 1870 * free may be skipped when dmu_free_long_range() was called 1871 * above since it covers the entire object's contents. 1872 */ 1873 if (rwa->raw && *object_to_hold != DMU_NEW_OBJECT && !do_free_range) { 1874 err = dmu_free_long_range(rwa->os, drro->drr_object, 1875 (drro->drr_maxblkid + 1) * doi->doi_data_block_size, 1876 DMU_OBJECT_END); 1877 if (err != 0) 1878 return (SET_ERROR(EINVAL)); 1879 } 1880 return (0); 1881 } 1882 1883 noinline static int 1884 receive_object(struct receive_writer_arg *rwa, struct drr_object *drro, 1885 void *data) 1886 { 1887 dmu_object_info_t doi; 1888 dmu_tx_t *tx; 1889 int err; 1890 uint32_t new_blksz = drro->drr_blksz; 1891 uint8_t dn_slots = drro->drr_dn_slots != 0 ? 1892 drro->drr_dn_slots : DNODE_MIN_SLOTS; 1893 1894 if (drro->drr_type == DMU_OT_NONE || 1895 !DMU_OT_IS_VALID(drro->drr_type) || 1896 !DMU_OT_IS_VALID(drro->drr_bonustype) || 1897 drro->drr_checksumtype >= ZIO_CHECKSUM_FUNCTIONS || 1898 drro->drr_compress >= ZIO_COMPRESS_FUNCTIONS || 1899 P2PHASE(drro->drr_blksz, SPA_MINBLOCKSIZE) || 1900 drro->drr_blksz < SPA_MINBLOCKSIZE || 1901 drro->drr_blksz > spa_maxblocksize(dmu_objset_spa(rwa->os)) || 1902 drro->drr_bonuslen > 1903 DN_BONUS_SIZE(spa_maxdnodesize(dmu_objset_spa(rwa->os))) || 1904 dn_slots > 1905 (spa_maxdnodesize(dmu_objset_spa(rwa->os)) >> DNODE_SHIFT)) { 1906 return (SET_ERROR(EINVAL)); 1907 } 1908 1909 if (rwa->raw) { 1910 /* 1911 * We should have received a DRR_OBJECT_RANGE record 1912 * containing this block and stored it in rwa. 1913 */ 1914 if (drro->drr_object < rwa->or_firstobj || 1915 drro->drr_object >= rwa->or_firstobj + rwa->or_numslots || 1916 drro->drr_raw_bonuslen < drro->drr_bonuslen || 1917 drro->drr_indblkshift > SPA_MAXBLOCKSHIFT || 1918 drro->drr_nlevels > DN_MAX_LEVELS || 1919 drro->drr_nblkptr > DN_MAX_NBLKPTR || 1920 DN_SLOTS_TO_BONUSLEN(dn_slots) < 1921 drro->drr_raw_bonuslen) 1922 return (SET_ERROR(EINVAL)); 1923 } else { 1924 /* 1925 * The DRR_OBJECT_SPILL flag is valid when the DRR_BEGIN 1926 * record indicates this by setting DRR_FLAG_SPILL_BLOCK. 1927 */ 1928 if (((drro->drr_flags & ~(DRR_OBJECT_SPILL))) || 1929 (!rwa->spill && DRR_OBJECT_HAS_SPILL(drro->drr_flags))) { 1930 return (SET_ERROR(EINVAL)); 1931 } 1932 1933 if (drro->drr_raw_bonuslen != 0 || drro->drr_nblkptr != 0 || 1934 drro->drr_indblkshift != 0 || drro->drr_nlevels != 0) { 1935 return (SET_ERROR(EINVAL)); 1936 } 1937 } 1938 1939 err = dmu_object_info(rwa->os, drro->drr_object, &doi); 1940 1941 if (err != 0 && err != ENOENT && err != EEXIST) 1942 return (SET_ERROR(EINVAL)); 1943 1944 if (drro->drr_object > rwa->max_object) 1945 rwa->max_object = drro->drr_object; 1946 1947 /* 1948 * If we are losing blkptrs or changing the block size this must 1949 * be a new file instance. We must clear out the previous file 1950 * contents before we can change this type of metadata in the dnode. 1951 * Raw receives will also check that the indirect structure of the 1952 * dnode hasn't changed. 1953 */ 1954 uint64_t object_to_hold; 1955 if (err == 0) { 1956 err = receive_handle_existing_object(rwa, drro, &doi, data, 1957 &object_to_hold, &new_blksz); 1958 if (err != 0) 1959 return (err); 1960 } else if (err == EEXIST) { 1961 /* 1962 * The object requested is currently an interior slot of a 1963 * multi-slot dnode. This will be resolved when the next txg 1964 * is synced out, since the send stream will have told us 1965 * to free this slot when we freed the associated dnode 1966 * earlier in the stream. 1967 */ 1968 txg_wait_synced(dmu_objset_pool(rwa->os), 0); 1969 1970 if (dmu_object_info(rwa->os, drro->drr_object, NULL) != ENOENT) 1971 return (SET_ERROR(EINVAL)); 1972 1973 /* object was freed and we are about to allocate a new one */ 1974 object_to_hold = DMU_NEW_OBJECT; 1975 } else { 1976 /* 1977 * If the only record in this range so far was DRR_FREEOBJECTS 1978 * with at least one actually freed object, it's possible that 1979 * the block will now be converted to a hole. We need to wait 1980 * for the txg to sync to prevent races. 1981 */ 1982 if (rwa->or_need_sync == ORNS_YES) 1983 txg_wait_synced(dmu_objset_pool(rwa->os), 0); 1984 1985 /* object is free and we are about to allocate a new one */ 1986 object_to_hold = DMU_NEW_OBJECT; 1987 } 1988 1989 /* Only relevant for the first object in the range */ 1990 rwa->or_need_sync = ORNS_NO; 1991 1992 /* 1993 * If this is a multi-slot dnode there is a chance that this 1994 * object will expand into a slot that is already used by 1995 * another object from the previous snapshot. We must free 1996 * these objects before we attempt to allocate the new dnode. 1997 */ 1998 if (dn_slots > 1) { 1999 boolean_t need_sync = B_FALSE; 2000 2001 for (uint64_t slot = drro->drr_object + 1; 2002 slot < drro->drr_object + dn_slots; 2003 slot++) { 2004 dmu_object_info_t slot_doi; 2005 2006 err = dmu_object_info(rwa->os, slot, &slot_doi); 2007 if (err == ENOENT || err == EEXIST) 2008 continue; 2009 else if (err != 0) 2010 return (err); 2011 2012 err = dmu_free_long_object(rwa->os, slot); 2013 if (err != 0) 2014 return (err); 2015 2016 need_sync = B_TRUE; 2017 } 2018 2019 if (need_sync) 2020 txg_wait_synced(dmu_objset_pool(rwa->os), 0); 2021 } 2022 2023 tx = dmu_tx_create(rwa->os); 2024 dmu_tx_hold_bonus(tx, object_to_hold); 2025 dmu_tx_hold_write(tx, object_to_hold, 0, 0); 2026 err = dmu_tx_assign(tx, TXG_WAIT); 2027 if (err != 0) { 2028 dmu_tx_abort(tx); 2029 return (err); 2030 } 2031 2032 if (object_to_hold == DMU_NEW_OBJECT) { 2033 /* Currently free, wants to be allocated */ 2034 err = dmu_object_claim_dnsize(rwa->os, drro->drr_object, 2035 drro->drr_type, new_blksz, 2036 drro->drr_bonustype, drro->drr_bonuslen, 2037 dn_slots << DNODE_SHIFT, tx); 2038 } else if (drro->drr_type != doi.doi_type || 2039 new_blksz != doi.doi_data_block_size || 2040 drro->drr_bonustype != doi.doi_bonus_type || 2041 drro->drr_bonuslen != doi.doi_bonus_size) { 2042 /* Currently allocated, but with different properties */ 2043 err = dmu_object_reclaim_dnsize(rwa->os, drro->drr_object, 2044 drro->drr_type, new_blksz, 2045 drro->drr_bonustype, drro->drr_bonuslen, 2046 dn_slots << DNODE_SHIFT, rwa->spill ? 2047 DRR_OBJECT_HAS_SPILL(drro->drr_flags) : B_FALSE, tx); 2048 } else if (rwa->spill && !DRR_OBJECT_HAS_SPILL(drro->drr_flags)) { 2049 /* 2050 * Currently allocated, the existing version of this object 2051 * may reference a spill block that is no longer allocated 2052 * at the source and needs to be freed. 2053 */ 2054 err = dmu_object_rm_spill(rwa->os, drro->drr_object, tx); 2055 } 2056 2057 if (err != 0) { 2058 dmu_tx_commit(tx); 2059 return (SET_ERROR(EINVAL)); 2060 } 2061 2062 if (rwa->or_crypt_params_present) { 2063 /* 2064 * Set the crypt params for the buffer associated with this 2065 * range of dnodes. This causes the blkptr_t to have the 2066 * same crypt params (byteorder, salt, iv, mac) as on the 2067 * sending side. 2068 * 2069 * Since we are committing this tx now, it is possible for 2070 * the dnode block to end up on-disk with the incorrect MAC, 2071 * if subsequent objects in this block are received in a 2072 * different txg. However, since the dataset is marked as 2073 * inconsistent, no code paths will do a non-raw read (or 2074 * decrypt the block / verify the MAC). The receive code and 2075 * scrub code can safely do raw reads and verify the 2076 * checksum. They don't need to verify the MAC. 2077 */ 2078 dmu_buf_t *db = NULL; 2079 uint64_t offset = rwa->or_firstobj * DNODE_MIN_SIZE; 2080 2081 err = dmu_buf_hold_by_dnode(DMU_META_DNODE(rwa->os), 2082 offset, FTAG, &db, DMU_READ_PREFETCH | DMU_READ_NO_DECRYPT); 2083 if (err != 0) { 2084 dmu_tx_commit(tx); 2085 return (SET_ERROR(EINVAL)); 2086 } 2087 2088 dmu_buf_set_crypt_params(db, rwa->or_byteorder, 2089 rwa->or_salt, rwa->or_iv, rwa->or_mac, tx); 2090 2091 dmu_buf_rele(db, FTAG); 2092 2093 rwa->or_crypt_params_present = B_FALSE; 2094 } 2095 2096 dmu_object_set_checksum(rwa->os, drro->drr_object, 2097 drro->drr_checksumtype, tx); 2098 dmu_object_set_compress(rwa->os, drro->drr_object, 2099 drro->drr_compress, tx); 2100 2101 /* handle more restrictive dnode structuring for raw recvs */ 2102 if (rwa->raw) { 2103 /* 2104 * Set the indirect block size, block shift, nlevels. 2105 * This will not fail because we ensured all of the 2106 * blocks were freed earlier if this is a new object. 2107 * For non-new objects block size and indirect block 2108 * shift cannot change and nlevels can only increase. 2109 */ 2110 ASSERT3U(new_blksz, ==, drro->drr_blksz); 2111 VERIFY0(dmu_object_set_blocksize(rwa->os, drro->drr_object, 2112 drro->drr_blksz, drro->drr_indblkshift, tx)); 2113 VERIFY0(dmu_object_set_nlevels(rwa->os, drro->drr_object, 2114 drro->drr_nlevels, tx)); 2115 2116 /* 2117 * Set the maxblkid. This will always succeed because 2118 * we freed all blocks beyond the new maxblkid above. 2119 */ 2120 VERIFY0(dmu_object_set_maxblkid(rwa->os, drro->drr_object, 2121 drro->drr_maxblkid, tx)); 2122 } 2123 2124 if (data != NULL) { 2125 dmu_buf_t *db; 2126 dnode_t *dn; 2127 uint32_t flags = DMU_READ_NO_PREFETCH; 2128 2129 if (rwa->raw) 2130 flags |= DMU_READ_NO_DECRYPT; 2131 2132 VERIFY0(dnode_hold(rwa->os, drro->drr_object, FTAG, &dn)); 2133 VERIFY0(dmu_bonus_hold_by_dnode(dn, FTAG, &db, flags)); 2134 2135 dmu_buf_will_dirty(db, tx); 2136 2137 ASSERT3U(db->db_size, >=, drro->drr_bonuslen); 2138 memcpy(db->db_data, data, DRR_OBJECT_PAYLOAD_SIZE(drro)); 2139 2140 /* 2141 * Raw bonus buffers have their byteorder determined by the 2142 * DRR_OBJECT_RANGE record. 2143 */ 2144 if (rwa->byteswap && !rwa->raw) { 2145 dmu_object_byteswap_t byteswap = 2146 DMU_OT_BYTESWAP(drro->drr_bonustype); 2147 dmu_ot_byteswap[byteswap].ob_func(db->db_data, 2148 DRR_OBJECT_PAYLOAD_SIZE(drro)); 2149 } 2150 dmu_buf_rele(db, FTAG); 2151 dnode_rele(dn, FTAG); 2152 } 2153 2154 /* 2155 * If the receive fails, we want the resume stream to start with the 2156 * same record that we last successfully received. There is no way to 2157 * request resume from the object record, but we can benefit from the 2158 * fact that sender always sends object record before anything else, 2159 * after which it will "resend" data at offset 0 and resume normally. 2160 */ 2161 save_resume_state(rwa, drro->drr_object, 0, tx); 2162 2163 dmu_tx_commit(tx); 2164 2165 return (0); 2166 } 2167 2168 noinline static int 2169 receive_freeobjects(struct receive_writer_arg *rwa, 2170 struct drr_freeobjects *drrfo) 2171 { 2172 uint64_t obj; 2173 int next_err = 0; 2174 2175 if (drrfo->drr_firstobj + drrfo->drr_numobjs < drrfo->drr_firstobj) 2176 return (SET_ERROR(EINVAL)); 2177 2178 for (obj = drrfo->drr_firstobj == 0 ? 1 : drrfo->drr_firstobj; 2179 obj < drrfo->drr_firstobj + drrfo->drr_numobjs && 2180 obj < DN_MAX_OBJECT && next_err == 0; 2181 next_err = dmu_object_next(rwa->os, &obj, FALSE, 0)) { 2182 dmu_object_info_t doi; 2183 int err; 2184 2185 err = dmu_object_info(rwa->os, obj, &doi); 2186 if (err == ENOENT) 2187 continue; 2188 else if (err != 0) 2189 return (err); 2190 2191 err = dmu_free_long_object(rwa->os, obj); 2192 2193 if (err != 0) 2194 return (err); 2195 2196 if (rwa->or_need_sync == ORNS_MAYBE) 2197 rwa->or_need_sync = ORNS_YES; 2198 } 2199 if (next_err != ESRCH) 2200 return (next_err); 2201 return (0); 2202 } 2203 2204 /* 2205 * Note: if this fails, the caller will clean up any records left on the 2206 * rwa->write_batch list. 2207 */ 2208 static int 2209 flush_write_batch_impl(struct receive_writer_arg *rwa) 2210 { 2211 dnode_t *dn; 2212 int err; 2213 2214 if (dnode_hold(rwa->os, rwa->last_object, FTAG, &dn) != 0) 2215 return (SET_ERROR(EINVAL)); 2216 2217 struct receive_record_arg *last_rrd = list_tail(&rwa->write_batch); 2218 struct drr_write *last_drrw = &last_rrd->header.drr_u.drr_write; 2219 2220 struct receive_record_arg *first_rrd = list_head(&rwa->write_batch); 2221 struct drr_write *first_drrw = &first_rrd->header.drr_u.drr_write; 2222 2223 ASSERT3U(rwa->last_object, ==, last_drrw->drr_object); 2224 ASSERT3U(rwa->last_offset, ==, last_drrw->drr_offset); 2225 2226 dmu_tx_t *tx = dmu_tx_create(rwa->os); 2227 dmu_tx_hold_write_by_dnode(tx, dn, first_drrw->drr_offset, 2228 last_drrw->drr_offset - first_drrw->drr_offset + 2229 last_drrw->drr_logical_size); 2230 err = dmu_tx_assign(tx, TXG_WAIT); 2231 if (err != 0) { 2232 dmu_tx_abort(tx); 2233 dnode_rele(dn, FTAG); 2234 return (err); 2235 } 2236 2237 struct receive_record_arg *rrd; 2238 while ((rrd = list_head(&rwa->write_batch)) != NULL) { 2239 struct drr_write *drrw = &rrd->header.drr_u.drr_write; 2240 abd_t *abd = rrd->abd; 2241 2242 ASSERT3U(drrw->drr_object, ==, rwa->last_object); 2243 2244 if (drrw->drr_logical_size != dn->dn_datablksz) { 2245 /* 2246 * The WRITE record is larger than the object's block 2247 * size. We must be receiving an incremental 2248 * large-block stream into a dataset that previously did 2249 * a non-large-block receive. Lightweight writes must 2250 * be exactly one block, so we need to decompress the 2251 * data (if compressed) and do a normal dmu_write(). 2252 */ 2253 ASSERT3U(drrw->drr_logical_size, >, dn->dn_datablksz); 2254 if (DRR_WRITE_COMPRESSED(drrw)) { 2255 abd_t *decomp_abd = 2256 abd_alloc_linear(drrw->drr_logical_size, 2257 B_FALSE); 2258 2259 err = zio_decompress_data( 2260 drrw->drr_compressiontype, 2261 abd, decomp_abd, 2262 abd_get_size(abd), 2263 abd_get_size(decomp_abd), NULL); 2264 2265 if (err == 0) { 2266 dmu_write_by_dnode(dn, 2267 drrw->drr_offset, 2268 drrw->drr_logical_size, 2269 abd_to_buf(decomp_abd), tx); 2270 } 2271 abd_free(decomp_abd); 2272 } else { 2273 dmu_write_by_dnode(dn, 2274 drrw->drr_offset, 2275 drrw->drr_logical_size, 2276 abd_to_buf(abd), tx); 2277 } 2278 if (err == 0) 2279 abd_free(abd); 2280 } else { 2281 zio_prop_t zp = {0}; 2282 dmu_write_policy(rwa->os, dn, 0, 0, &zp); 2283 2284 zio_flag_t zio_flags = 0; 2285 2286 if (rwa->raw) { 2287 zp.zp_encrypt = B_TRUE; 2288 zp.zp_compress = drrw->drr_compressiontype; 2289 zp.zp_byteorder = ZFS_HOST_BYTEORDER ^ 2290 !!DRR_IS_RAW_BYTESWAPPED(drrw->drr_flags) ^ 2291 rwa->byteswap; 2292 memcpy(zp.zp_salt, drrw->drr_salt, 2293 ZIO_DATA_SALT_LEN); 2294 memcpy(zp.zp_iv, drrw->drr_iv, 2295 ZIO_DATA_IV_LEN); 2296 memcpy(zp.zp_mac, drrw->drr_mac, 2297 ZIO_DATA_MAC_LEN); 2298 if (DMU_OT_IS_ENCRYPTED(zp.zp_type)) { 2299 zp.zp_nopwrite = B_FALSE; 2300 zp.zp_copies = MIN(zp.zp_copies, 2301 SPA_DVAS_PER_BP - 1); 2302 } 2303 zio_flags |= ZIO_FLAG_RAW; 2304 } else if (DRR_WRITE_COMPRESSED(drrw)) { 2305 ASSERT3U(drrw->drr_compressed_size, >, 0); 2306 ASSERT3U(drrw->drr_logical_size, >=, 2307 drrw->drr_compressed_size); 2308 zp.zp_compress = drrw->drr_compressiontype; 2309 zio_flags |= ZIO_FLAG_RAW_COMPRESS; 2310 } else if (rwa->byteswap) { 2311 /* 2312 * Note: compressed blocks never need to be 2313 * byteswapped, because WRITE records for 2314 * metadata blocks are never compressed. The 2315 * exception is raw streams, which are written 2316 * in the original byteorder, and the byteorder 2317 * bit is preserved in the BP by setting 2318 * zp_byteorder above. 2319 */ 2320 dmu_object_byteswap_t byteswap = 2321 DMU_OT_BYTESWAP(drrw->drr_type); 2322 dmu_ot_byteswap[byteswap].ob_func( 2323 abd_to_buf(abd), 2324 DRR_WRITE_PAYLOAD_SIZE(drrw)); 2325 } 2326 2327 /* 2328 * Since this data can't be read until the receive 2329 * completes, we can do a "lightweight" write for 2330 * improved performance. 2331 */ 2332 err = dmu_lightweight_write_by_dnode(dn, 2333 drrw->drr_offset, abd, &zp, zio_flags, tx); 2334 } 2335 2336 if (err != 0) { 2337 /* 2338 * This rrd is left on the list, so the caller will 2339 * free it (and the abd). 2340 */ 2341 break; 2342 } 2343 2344 /* 2345 * Note: If the receive fails, we want the resume stream to 2346 * start with the same record that we last successfully 2347 * received (as opposed to the next record), so that we can 2348 * verify that we are resuming from the correct location. 2349 */ 2350 save_resume_state(rwa, drrw->drr_object, drrw->drr_offset, tx); 2351 2352 list_remove(&rwa->write_batch, rrd); 2353 kmem_free(rrd, sizeof (*rrd)); 2354 } 2355 2356 dmu_tx_commit(tx); 2357 dnode_rele(dn, FTAG); 2358 return (err); 2359 } 2360 2361 noinline static int 2362 flush_write_batch(struct receive_writer_arg *rwa) 2363 { 2364 if (list_is_empty(&rwa->write_batch)) 2365 return (0); 2366 int err = rwa->err; 2367 if (err == 0) 2368 err = flush_write_batch_impl(rwa); 2369 if (err != 0) { 2370 struct receive_record_arg *rrd; 2371 while ((rrd = list_remove_head(&rwa->write_batch)) != NULL) { 2372 abd_free(rrd->abd); 2373 kmem_free(rrd, sizeof (*rrd)); 2374 } 2375 } 2376 ASSERT(list_is_empty(&rwa->write_batch)); 2377 return (err); 2378 } 2379 2380 noinline static int 2381 receive_process_write_record(struct receive_writer_arg *rwa, 2382 struct receive_record_arg *rrd) 2383 { 2384 int err = 0; 2385 2386 ASSERT3U(rrd->header.drr_type, ==, DRR_WRITE); 2387 struct drr_write *drrw = &rrd->header.drr_u.drr_write; 2388 2389 if (drrw->drr_offset + drrw->drr_logical_size < drrw->drr_offset || 2390 !DMU_OT_IS_VALID(drrw->drr_type)) 2391 return (SET_ERROR(EINVAL)); 2392 2393 if (rwa->heal) { 2394 blkptr_t *bp; 2395 dmu_buf_t *dbp; 2396 int flags = DB_RF_CANFAIL; 2397 2398 if (rwa->raw) 2399 flags |= DB_RF_NO_DECRYPT; 2400 2401 if (rwa->byteswap) { 2402 dmu_object_byteswap_t byteswap = 2403 DMU_OT_BYTESWAP(drrw->drr_type); 2404 dmu_ot_byteswap[byteswap].ob_func(abd_to_buf(rrd->abd), 2405 DRR_WRITE_PAYLOAD_SIZE(drrw)); 2406 } 2407 2408 err = dmu_buf_hold_noread(rwa->os, drrw->drr_object, 2409 drrw->drr_offset, FTAG, &dbp); 2410 if (err != 0) 2411 return (err); 2412 2413 /* Try to read the object to see if it needs healing */ 2414 err = dbuf_read((dmu_buf_impl_t *)dbp, NULL, flags); 2415 /* 2416 * We only try to heal when dbuf_read() returns a ECKSUMs. 2417 * Other errors (even EIO) get returned to caller. 2418 * EIO indicates that the device is not present/accessible, 2419 * so writing to it will likely fail. 2420 * If the block is healthy, we don't want to overwrite it 2421 * unnecessarily. 2422 */ 2423 if (err != ECKSUM) { 2424 dmu_buf_rele(dbp, FTAG); 2425 return (err); 2426 } 2427 /* Make sure the on-disk block and recv record sizes match */ 2428 if (drrw->drr_logical_size != dbp->db_size) { 2429 err = ENOTSUP; 2430 dmu_buf_rele(dbp, FTAG); 2431 return (err); 2432 } 2433 /* Get the block pointer for the corrupted block */ 2434 bp = dmu_buf_get_blkptr(dbp); 2435 err = do_corrective_recv(rwa, drrw, rrd, bp); 2436 dmu_buf_rele(dbp, FTAG); 2437 return (err); 2438 } 2439 2440 /* 2441 * For resuming to work, records must be in increasing order 2442 * by (object, offset). 2443 */ 2444 if (drrw->drr_object < rwa->last_object || 2445 (drrw->drr_object == rwa->last_object && 2446 drrw->drr_offset < rwa->last_offset)) { 2447 return (SET_ERROR(EINVAL)); 2448 } 2449 2450 struct receive_record_arg *first_rrd = list_head(&rwa->write_batch); 2451 struct drr_write *first_drrw = &first_rrd->header.drr_u.drr_write; 2452 uint64_t batch_size = 2453 MIN(zfs_recv_write_batch_size, DMU_MAX_ACCESS / 2); 2454 if (first_rrd != NULL && 2455 (drrw->drr_object != first_drrw->drr_object || 2456 drrw->drr_offset >= first_drrw->drr_offset + batch_size)) { 2457 err = flush_write_batch(rwa); 2458 if (err != 0) 2459 return (err); 2460 } 2461 2462 rwa->last_object = drrw->drr_object; 2463 rwa->last_offset = drrw->drr_offset; 2464 2465 if (rwa->last_object > rwa->max_object) 2466 rwa->max_object = rwa->last_object; 2467 2468 list_insert_tail(&rwa->write_batch, rrd); 2469 /* 2470 * Return EAGAIN to indicate that we will use this rrd again, 2471 * so the caller should not free it 2472 */ 2473 return (EAGAIN); 2474 } 2475 2476 static int 2477 receive_write_embedded(struct receive_writer_arg *rwa, 2478 struct drr_write_embedded *drrwe, void *data) 2479 { 2480 dmu_tx_t *tx; 2481 int err; 2482 2483 if (drrwe->drr_offset + drrwe->drr_length < drrwe->drr_offset) 2484 return (SET_ERROR(EINVAL)); 2485 2486 if (drrwe->drr_psize > BPE_PAYLOAD_SIZE) 2487 return (SET_ERROR(EINVAL)); 2488 2489 if (drrwe->drr_etype >= NUM_BP_EMBEDDED_TYPES) 2490 return (SET_ERROR(EINVAL)); 2491 if (drrwe->drr_compression >= ZIO_COMPRESS_FUNCTIONS) 2492 return (SET_ERROR(EINVAL)); 2493 if (rwa->raw) 2494 return (SET_ERROR(EINVAL)); 2495 2496 if (drrwe->drr_object > rwa->max_object) 2497 rwa->max_object = drrwe->drr_object; 2498 2499 tx = dmu_tx_create(rwa->os); 2500 2501 dmu_tx_hold_write(tx, drrwe->drr_object, 2502 drrwe->drr_offset, drrwe->drr_length); 2503 err = dmu_tx_assign(tx, TXG_WAIT); 2504 if (err != 0) { 2505 dmu_tx_abort(tx); 2506 return (err); 2507 } 2508 2509 dmu_write_embedded(rwa->os, drrwe->drr_object, 2510 drrwe->drr_offset, data, drrwe->drr_etype, 2511 drrwe->drr_compression, drrwe->drr_lsize, drrwe->drr_psize, 2512 rwa->byteswap ^ ZFS_HOST_BYTEORDER, tx); 2513 2514 /* See comment in restore_write. */ 2515 save_resume_state(rwa, drrwe->drr_object, drrwe->drr_offset, tx); 2516 dmu_tx_commit(tx); 2517 return (0); 2518 } 2519 2520 static int 2521 receive_spill(struct receive_writer_arg *rwa, struct drr_spill *drrs, 2522 abd_t *abd) 2523 { 2524 dmu_buf_t *db, *db_spill; 2525 int err; 2526 2527 if (drrs->drr_length < SPA_MINBLOCKSIZE || 2528 drrs->drr_length > spa_maxblocksize(dmu_objset_spa(rwa->os))) 2529 return (SET_ERROR(EINVAL)); 2530 2531 /* 2532 * This is an unmodified spill block which was added to the stream 2533 * to resolve an issue with incorrectly removing spill blocks. It 2534 * should be ignored by current versions of the code which support 2535 * the DRR_FLAG_SPILL_BLOCK flag. 2536 */ 2537 if (rwa->spill && DRR_SPILL_IS_UNMODIFIED(drrs->drr_flags)) { 2538 abd_free(abd); 2539 return (0); 2540 } 2541 2542 if (rwa->raw) { 2543 if (!DMU_OT_IS_VALID(drrs->drr_type) || 2544 drrs->drr_compressiontype >= ZIO_COMPRESS_FUNCTIONS || 2545 drrs->drr_compressed_size == 0) 2546 return (SET_ERROR(EINVAL)); 2547 } 2548 2549 if (dmu_object_info(rwa->os, drrs->drr_object, NULL) != 0) 2550 return (SET_ERROR(EINVAL)); 2551 2552 if (drrs->drr_object > rwa->max_object) 2553 rwa->max_object = drrs->drr_object; 2554 2555 VERIFY0(dmu_bonus_hold(rwa->os, drrs->drr_object, FTAG, &db)); 2556 if ((err = dmu_spill_hold_by_bonus(db, DMU_READ_NO_DECRYPT, FTAG, 2557 &db_spill)) != 0) { 2558 dmu_buf_rele(db, FTAG); 2559 return (err); 2560 } 2561 2562 dmu_tx_t *tx = dmu_tx_create(rwa->os); 2563 2564 dmu_tx_hold_spill(tx, db->db_object); 2565 2566 err = dmu_tx_assign(tx, TXG_WAIT); 2567 if (err != 0) { 2568 dmu_buf_rele(db, FTAG); 2569 dmu_buf_rele(db_spill, FTAG); 2570 dmu_tx_abort(tx); 2571 return (err); 2572 } 2573 2574 /* 2575 * Spill blocks may both grow and shrink. When a change in size 2576 * occurs any existing dbuf must be updated to match the logical 2577 * size of the provided arc_buf_t. 2578 */ 2579 if (db_spill->db_size != drrs->drr_length) { 2580 dmu_buf_will_fill(db_spill, tx, B_FALSE); 2581 VERIFY0(dbuf_spill_set_blksz(db_spill, 2582 drrs->drr_length, tx)); 2583 } 2584 2585 arc_buf_t *abuf; 2586 if (rwa->raw) { 2587 boolean_t byteorder = ZFS_HOST_BYTEORDER ^ 2588 !!DRR_IS_RAW_BYTESWAPPED(drrs->drr_flags) ^ 2589 rwa->byteswap; 2590 2591 abuf = arc_loan_raw_buf(dmu_objset_spa(rwa->os), 2592 drrs->drr_object, byteorder, drrs->drr_salt, 2593 drrs->drr_iv, drrs->drr_mac, drrs->drr_type, 2594 drrs->drr_compressed_size, drrs->drr_length, 2595 drrs->drr_compressiontype, 0); 2596 } else { 2597 abuf = arc_loan_buf(dmu_objset_spa(rwa->os), 2598 DMU_OT_IS_METADATA(drrs->drr_type), 2599 drrs->drr_length); 2600 if (rwa->byteswap) { 2601 dmu_object_byteswap_t byteswap = 2602 DMU_OT_BYTESWAP(drrs->drr_type); 2603 dmu_ot_byteswap[byteswap].ob_func(abd_to_buf(abd), 2604 DRR_SPILL_PAYLOAD_SIZE(drrs)); 2605 } 2606 } 2607 2608 memcpy(abuf->b_data, abd_to_buf(abd), DRR_SPILL_PAYLOAD_SIZE(drrs)); 2609 abd_free(abd); 2610 dbuf_assign_arcbuf((dmu_buf_impl_t *)db_spill, abuf, tx); 2611 2612 dmu_buf_rele(db, FTAG); 2613 dmu_buf_rele(db_spill, FTAG); 2614 2615 dmu_tx_commit(tx); 2616 return (0); 2617 } 2618 2619 noinline static int 2620 receive_free(struct receive_writer_arg *rwa, struct drr_free *drrf) 2621 { 2622 int err; 2623 2624 if (drrf->drr_length != -1ULL && 2625 drrf->drr_offset + drrf->drr_length < drrf->drr_offset) 2626 return (SET_ERROR(EINVAL)); 2627 2628 if (dmu_object_info(rwa->os, drrf->drr_object, NULL) != 0) 2629 return (SET_ERROR(EINVAL)); 2630 2631 if (drrf->drr_object > rwa->max_object) 2632 rwa->max_object = drrf->drr_object; 2633 2634 err = dmu_free_long_range(rwa->os, drrf->drr_object, 2635 drrf->drr_offset, drrf->drr_length); 2636 2637 return (err); 2638 } 2639 2640 static int 2641 receive_object_range(struct receive_writer_arg *rwa, 2642 struct drr_object_range *drror) 2643 { 2644 /* 2645 * By default, we assume this block is in our native format 2646 * (ZFS_HOST_BYTEORDER). We then take into account whether 2647 * the send stream is byteswapped (rwa->byteswap). Finally, 2648 * we need to byteswap again if this particular block was 2649 * in non-native format on the send side. 2650 */ 2651 boolean_t byteorder = ZFS_HOST_BYTEORDER ^ rwa->byteswap ^ 2652 !!DRR_IS_RAW_BYTESWAPPED(drror->drr_flags); 2653 2654 /* 2655 * Since dnode block sizes are constant, we should not need to worry 2656 * about making sure that the dnode block size is the same on the 2657 * sending and receiving sides for the time being. For non-raw sends, 2658 * this does not matter (and in fact we do not send a DRR_OBJECT_RANGE 2659 * record at all). Raw sends require this record type because the 2660 * encryption parameters are used to protect an entire block of bonus 2661 * buffers. If the size of dnode blocks ever becomes variable, 2662 * handling will need to be added to ensure that dnode block sizes 2663 * match on the sending and receiving side. 2664 */ 2665 if (drror->drr_numslots != DNODES_PER_BLOCK || 2666 P2PHASE(drror->drr_firstobj, DNODES_PER_BLOCK) != 0 || 2667 !rwa->raw) 2668 return (SET_ERROR(EINVAL)); 2669 2670 if (drror->drr_firstobj > rwa->max_object) 2671 rwa->max_object = drror->drr_firstobj; 2672 2673 /* 2674 * The DRR_OBJECT_RANGE handling must be deferred to receive_object() 2675 * so that the block of dnodes is not written out when it's empty, 2676 * and converted to a HOLE BP. 2677 */ 2678 rwa->or_crypt_params_present = B_TRUE; 2679 rwa->or_firstobj = drror->drr_firstobj; 2680 rwa->or_numslots = drror->drr_numslots; 2681 memcpy(rwa->or_salt, drror->drr_salt, ZIO_DATA_SALT_LEN); 2682 memcpy(rwa->or_iv, drror->drr_iv, ZIO_DATA_IV_LEN); 2683 memcpy(rwa->or_mac, drror->drr_mac, ZIO_DATA_MAC_LEN); 2684 rwa->or_byteorder = byteorder; 2685 2686 rwa->or_need_sync = ORNS_MAYBE; 2687 2688 return (0); 2689 } 2690 2691 /* 2692 * Until we have the ability to redact large ranges of data efficiently, we 2693 * process these records as frees. 2694 */ 2695 noinline static int 2696 receive_redact(struct receive_writer_arg *rwa, struct drr_redact *drrr) 2697 { 2698 struct drr_free drrf = {0}; 2699 drrf.drr_length = drrr->drr_length; 2700 drrf.drr_object = drrr->drr_object; 2701 drrf.drr_offset = drrr->drr_offset; 2702 drrf.drr_toguid = drrr->drr_toguid; 2703 return (receive_free(rwa, &drrf)); 2704 } 2705 2706 /* used to destroy the drc_ds on error */ 2707 static void 2708 dmu_recv_cleanup_ds(dmu_recv_cookie_t *drc) 2709 { 2710 dsl_dataset_t *ds = drc->drc_ds; 2711 ds_hold_flags_t dsflags; 2712 2713 dsflags = (drc->drc_raw) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT; 2714 /* 2715 * Wait for the txg sync before cleaning up the receive. For 2716 * resumable receives, this ensures that our resume state has 2717 * been written out to disk. For raw receives, this ensures 2718 * that the user accounting code will not attempt to do anything 2719 * after we stopped receiving the dataset. 2720 */ 2721 txg_wait_synced(ds->ds_dir->dd_pool, 0); 2722 ds->ds_objset->os_raw_receive = B_FALSE; 2723 2724 rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG); 2725 if (drc->drc_resumable && drc->drc_should_save && 2726 !BP_IS_HOLE(dsl_dataset_get_blkptr(ds))) { 2727 rrw_exit(&ds->ds_bp_rwlock, FTAG); 2728 dsl_dataset_disown(ds, dsflags, dmu_recv_tag); 2729 } else { 2730 char name[ZFS_MAX_DATASET_NAME_LEN]; 2731 rrw_exit(&ds->ds_bp_rwlock, FTAG); 2732 dsl_dataset_name(ds, name); 2733 dsl_dataset_disown(ds, dsflags, dmu_recv_tag); 2734 if (!drc->drc_heal) 2735 (void) dsl_destroy_head(name); 2736 } 2737 } 2738 2739 static void 2740 receive_cksum(dmu_recv_cookie_t *drc, int len, void *buf) 2741 { 2742 if (drc->drc_byteswap) { 2743 (void) fletcher_4_incremental_byteswap(buf, len, 2744 &drc->drc_cksum); 2745 } else { 2746 (void) fletcher_4_incremental_native(buf, len, &drc->drc_cksum); 2747 } 2748 } 2749 2750 /* 2751 * Read the payload into a buffer of size len, and update the current record's 2752 * payload field. 2753 * Allocate drc->drc_next_rrd and read the next record's header into 2754 * drc->drc_next_rrd->header. 2755 * Verify checksum of payload and next record. 2756 */ 2757 static int 2758 receive_read_payload_and_next_header(dmu_recv_cookie_t *drc, int len, void *buf) 2759 { 2760 int err; 2761 2762 if (len != 0) { 2763 ASSERT3U(len, <=, SPA_MAXBLOCKSIZE); 2764 err = receive_read(drc, len, buf); 2765 if (err != 0) 2766 return (err); 2767 receive_cksum(drc, len, buf); 2768 2769 /* note: rrd is NULL when reading the begin record's payload */ 2770 if (drc->drc_rrd != NULL) { 2771 drc->drc_rrd->payload = buf; 2772 drc->drc_rrd->payload_size = len; 2773 drc->drc_rrd->bytes_read = drc->drc_bytes_read; 2774 } 2775 } else { 2776 ASSERT3P(buf, ==, NULL); 2777 } 2778 2779 drc->drc_prev_cksum = drc->drc_cksum; 2780 2781 drc->drc_next_rrd = kmem_zalloc(sizeof (*drc->drc_next_rrd), KM_SLEEP); 2782 err = receive_read(drc, sizeof (drc->drc_next_rrd->header), 2783 &drc->drc_next_rrd->header); 2784 drc->drc_next_rrd->bytes_read = drc->drc_bytes_read; 2785 2786 if (err != 0) { 2787 kmem_free(drc->drc_next_rrd, sizeof (*drc->drc_next_rrd)); 2788 drc->drc_next_rrd = NULL; 2789 return (err); 2790 } 2791 if (drc->drc_next_rrd->header.drr_type == DRR_BEGIN) { 2792 kmem_free(drc->drc_next_rrd, sizeof (*drc->drc_next_rrd)); 2793 drc->drc_next_rrd = NULL; 2794 return (SET_ERROR(EINVAL)); 2795 } 2796 2797 /* 2798 * Note: checksum is of everything up to but not including the 2799 * checksum itself. 2800 */ 2801 ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum), 2802 ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t)); 2803 receive_cksum(drc, 2804 offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum), 2805 &drc->drc_next_rrd->header); 2806 2807 zio_cksum_t cksum_orig = 2808 drc->drc_next_rrd->header.drr_u.drr_checksum.drr_checksum; 2809 zio_cksum_t *cksump = 2810 &drc->drc_next_rrd->header.drr_u.drr_checksum.drr_checksum; 2811 2812 if (drc->drc_byteswap) 2813 byteswap_record(&drc->drc_next_rrd->header); 2814 2815 if ((!ZIO_CHECKSUM_IS_ZERO(cksump)) && 2816 !ZIO_CHECKSUM_EQUAL(drc->drc_cksum, *cksump)) { 2817 kmem_free(drc->drc_next_rrd, sizeof (*drc->drc_next_rrd)); 2818 drc->drc_next_rrd = NULL; 2819 return (SET_ERROR(ECKSUM)); 2820 } 2821 2822 receive_cksum(drc, sizeof (cksum_orig), &cksum_orig); 2823 2824 return (0); 2825 } 2826 2827 /* 2828 * Issue the prefetch reads for any necessary indirect blocks. 2829 * 2830 * We use the object ignore list to tell us whether or not to issue prefetches 2831 * for a given object. We do this for both correctness (in case the blocksize 2832 * of an object has changed) and performance (if the object doesn't exist, don't 2833 * needlessly try to issue prefetches). We also trim the list as we go through 2834 * the stream to prevent it from growing to an unbounded size. 2835 * 2836 * The object numbers within will always be in sorted order, and any write 2837 * records we see will also be in sorted order, but they're not sorted with 2838 * respect to each other (i.e. we can get several object records before 2839 * receiving each object's write records). As a result, once we've reached a 2840 * given object number, we can safely remove any reference to lower object 2841 * numbers in the ignore list. In practice, we receive up to 32 object records 2842 * before receiving write records, so the list can have up to 32 nodes in it. 2843 */ 2844 static void 2845 receive_read_prefetch(dmu_recv_cookie_t *drc, uint64_t object, uint64_t offset, 2846 uint64_t length) 2847 { 2848 if (!objlist_exists(drc->drc_ignore_objlist, object)) { 2849 dmu_prefetch(drc->drc_os, object, 1, offset, length, 2850 ZIO_PRIORITY_SYNC_READ); 2851 } 2852 } 2853 2854 /* 2855 * Read records off the stream, issuing any necessary prefetches. 2856 */ 2857 static int 2858 receive_read_record(dmu_recv_cookie_t *drc) 2859 { 2860 int err; 2861 2862 switch (drc->drc_rrd->header.drr_type) { 2863 case DRR_OBJECT: 2864 { 2865 struct drr_object *drro = 2866 &drc->drc_rrd->header.drr_u.drr_object; 2867 uint32_t size = DRR_OBJECT_PAYLOAD_SIZE(drro); 2868 void *buf = NULL; 2869 dmu_object_info_t doi; 2870 2871 if (size != 0) 2872 buf = kmem_zalloc(size, KM_SLEEP); 2873 2874 err = receive_read_payload_and_next_header(drc, size, buf); 2875 if (err != 0) { 2876 kmem_free(buf, size); 2877 return (err); 2878 } 2879 err = dmu_object_info(drc->drc_os, drro->drr_object, &doi); 2880 /* 2881 * See receive_read_prefetch for an explanation why we're 2882 * storing this object in the ignore_obj_list. 2883 */ 2884 if (err == ENOENT || err == EEXIST || 2885 (err == 0 && doi.doi_data_block_size != drro->drr_blksz)) { 2886 objlist_insert(drc->drc_ignore_objlist, 2887 drro->drr_object); 2888 err = 0; 2889 } 2890 return (err); 2891 } 2892 case DRR_FREEOBJECTS: 2893 { 2894 err = receive_read_payload_and_next_header(drc, 0, NULL); 2895 return (err); 2896 } 2897 case DRR_WRITE: 2898 { 2899 struct drr_write *drrw = &drc->drc_rrd->header.drr_u.drr_write; 2900 int size = DRR_WRITE_PAYLOAD_SIZE(drrw); 2901 abd_t *abd = abd_alloc_linear(size, B_FALSE); 2902 err = receive_read_payload_and_next_header(drc, size, 2903 abd_to_buf(abd)); 2904 if (err != 0) { 2905 abd_free(abd); 2906 return (err); 2907 } 2908 drc->drc_rrd->abd = abd; 2909 receive_read_prefetch(drc, drrw->drr_object, drrw->drr_offset, 2910 drrw->drr_logical_size); 2911 return (err); 2912 } 2913 case DRR_WRITE_EMBEDDED: 2914 { 2915 struct drr_write_embedded *drrwe = 2916 &drc->drc_rrd->header.drr_u.drr_write_embedded; 2917 uint32_t size = P2ROUNDUP(drrwe->drr_psize, 8); 2918 void *buf = kmem_zalloc(size, KM_SLEEP); 2919 2920 err = receive_read_payload_and_next_header(drc, size, buf); 2921 if (err != 0) { 2922 kmem_free(buf, size); 2923 return (err); 2924 } 2925 2926 receive_read_prefetch(drc, drrwe->drr_object, drrwe->drr_offset, 2927 drrwe->drr_length); 2928 return (err); 2929 } 2930 case DRR_FREE: 2931 case DRR_REDACT: 2932 { 2933 /* 2934 * It might be beneficial to prefetch indirect blocks here, but 2935 * we don't really have the data to decide for sure. 2936 */ 2937 err = receive_read_payload_and_next_header(drc, 0, NULL); 2938 return (err); 2939 } 2940 case DRR_END: 2941 { 2942 struct drr_end *drre = &drc->drc_rrd->header.drr_u.drr_end; 2943 if (!ZIO_CHECKSUM_EQUAL(drc->drc_prev_cksum, 2944 drre->drr_checksum)) 2945 return (SET_ERROR(ECKSUM)); 2946 return (0); 2947 } 2948 case DRR_SPILL: 2949 { 2950 struct drr_spill *drrs = &drc->drc_rrd->header.drr_u.drr_spill; 2951 int size = DRR_SPILL_PAYLOAD_SIZE(drrs); 2952 abd_t *abd = abd_alloc_linear(size, B_FALSE); 2953 err = receive_read_payload_and_next_header(drc, size, 2954 abd_to_buf(abd)); 2955 if (err != 0) 2956 abd_free(abd); 2957 else 2958 drc->drc_rrd->abd = abd; 2959 return (err); 2960 } 2961 case DRR_OBJECT_RANGE: 2962 { 2963 err = receive_read_payload_and_next_header(drc, 0, NULL); 2964 return (err); 2965 2966 } 2967 default: 2968 return (SET_ERROR(EINVAL)); 2969 } 2970 } 2971 2972 2973 2974 static void 2975 dprintf_drr(struct receive_record_arg *rrd, int err) 2976 { 2977 #ifdef ZFS_DEBUG 2978 switch (rrd->header.drr_type) { 2979 case DRR_OBJECT: 2980 { 2981 struct drr_object *drro = &rrd->header.drr_u.drr_object; 2982 dprintf("drr_type = OBJECT obj = %llu type = %u " 2983 "bonustype = %u blksz = %u bonuslen = %u cksumtype = %u " 2984 "compress = %u dn_slots = %u err = %d\n", 2985 (u_longlong_t)drro->drr_object, drro->drr_type, 2986 drro->drr_bonustype, drro->drr_blksz, drro->drr_bonuslen, 2987 drro->drr_checksumtype, drro->drr_compress, 2988 drro->drr_dn_slots, err); 2989 break; 2990 } 2991 case DRR_FREEOBJECTS: 2992 { 2993 struct drr_freeobjects *drrfo = 2994 &rrd->header.drr_u.drr_freeobjects; 2995 dprintf("drr_type = FREEOBJECTS firstobj = %llu " 2996 "numobjs = %llu err = %d\n", 2997 (u_longlong_t)drrfo->drr_firstobj, 2998 (u_longlong_t)drrfo->drr_numobjs, err); 2999 break; 3000 } 3001 case DRR_WRITE: 3002 { 3003 struct drr_write *drrw = &rrd->header.drr_u.drr_write; 3004 dprintf("drr_type = WRITE obj = %llu type = %u offset = %llu " 3005 "lsize = %llu cksumtype = %u flags = %u " 3006 "compress = %u psize = %llu err = %d\n", 3007 (u_longlong_t)drrw->drr_object, drrw->drr_type, 3008 (u_longlong_t)drrw->drr_offset, 3009 (u_longlong_t)drrw->drr_logical_size, 3010 drrw->drr_checksumtype, drrw->drr_flags, 3011 drrw->drr_compressiontype, 3012 (u_longlong_t)drrw->drr_compressed_size, err); 3013 break; 3014 } 3015 case DRR_WRITE_BYREF: 3016 { 3017 struct drr_write_byref *drrwbr = 3018 &rrd->header.drr_u.drr_write_byref; 3019 dprintf("drr_type = WRITE_BYREF obj = %llu offset = %llu " 3020 "length = %llu toguid = %llx refguid = %llx " 3021 "refobject = %llu refoffset = %llu cksumtype = %u " 3022 "flags = %u err = %d\n", 3023 (u_longlong_t)drrwbr->drr_object, 3024 (u_longlong_t)drrwbr->drr_offset, 3025 (u_longlong_t)drrwbr->drr_length, 3026 (u_longlong_t)drrwbr->drr_toguid, 3027 (u_longlong_t)drrwbr->drr_refguid, 3028 (u_longlong_t)drrwbr->drr_refobject, 3029 (u_longlong_t)drrwbr->drr_refoffset, 3030 drrwbr->drr_checksumtype, drrwbr->drr_flags, err); 3031 break; 3032 } 3033 case DRR_WRITE_EMBEDDED: 3034 { 3035 struct drr_write_embedded *drrwe = 3036 &rrd->header.drr_u.drr_write_embedded; 3037 dprintf("drr_type = WRITE_EMBEDDED obj = %llu offset = %llu " 3038 "length = %llu compress = %u etype = %u lsize = %u " 3039 "psize = %u err = %d\n", 3040 (u_longlong_t)drrwe->drr_object, 3041 (u_longlong_t)drrwe->drr_offset, 3042 (u_longlong_t)drrwe->drr_length, 3043 drrwe->drr_compression, drrwe->drr_etype, 3044 drrwe->drr_lsize, drrwe->drr_psize, err); 3045 break; 3046 } 3047 case DRR_FREE: 3048 { 3049 struct drr_free *drrf = &rrd->header.drr_u.drr_free; 3050 dprintf("drr_type = FREE obj = %llu offset = %llu " 3051 "length = %lld err = %d\n", 3052 (u_longlong_t)drrf->drr_object, 3053 (u_longlong_t)drrf->drr_offset, 3054 (longlong_t)drrf->drr_length, 3055 err); 3056 break; 3057 } 3058 case DRR_SPILL: 3059 { 3060 struct drr_spill *drrs = &rrd->header.drr_u.drr_spill; 3061 dprintf("drr_type = SPILL obj = %llu length = %llu " 3062 "err = %d\n", (u_longlong_t)drrs->drr_object, 3063 (u_longlong_t)drrs->drr_length, err); 3064 break; 3065 } 3066 case DRR_OBJECT_RANGE: 3067 { 3068 struct drr_object_range *drror = 3069 &rrd->header.drr_u.drr_object_range; 3070 dprintf("drr_type = OBJECT_RANGE firstobj = %llu " 3071 "numslots = %llu flags = %u err = %d\n", 3072 (u_longlong_t)drror->drr_firstobj, 3073 (u_longlong_t)drror->drr_numslots, 3074 drror->drr_flags, err); 3075 break; 3076 } 3077 default: 3078 return; 3079 } 3080 #endif 3081 } 3082 3083 /* 3084 * Commit the records to the pool. 3085 */ 3086 static int 3087 receive_process_record(struct receive_writer_arg *rwa, 3088 struct receive_record_arg *rrd) 3089 { 3090 int err; 3091 3092 /* Processing in order, therefore bytes_read should be increasing. */ 3093 ASSERT3U(rrd->bytes_read, >=, rwa->bytes_read); 3094 rwa->bytes_read = rrd->bytes_read; 3095 3096 /* We can only heal write records; other ones get ignored */ 3097 if (rwa->heal && rrd->header.drr_type != DRR_WRITE) { 3098 if (rrd->abd != NULL) { 3099 abd_free(rrd->abd); 3100 rrd->abd = NULL; 3101 } else if (rrd->payload != NULL) { 3102 kmem_free(rrd->payload, rrd->payload_size); 3103 rrd->payload = NULL; 3104 } 3105 return (0); 3106 } 3107 3108 if (!rwa->heal && rrd->header.drr_type != DRR_WRITE) { 3109 err = flush_write_batch(rwa); 3110 if (err != 0) { 3111 if (rrd->abd != NULL) { 3112 abd_free(rrd->abd); 3113 rrd->abd = NULL; 3114 rrd->payload = NULL; 3115 } else if (rrd->payload != NULL) { 3116 kmem_free(rrd->payload, rrd->payload_size); 3117 rrd->payload = NULL; 3118 } 3119 3120 return (err); 3121 } 3122 } 3123 3124 switch (rrd->header.drr_type) { 3125 case DRR_OBJECT: 3126 { 3127 struct drr_object *drro = &rrd->header.drr_u.drr_object; 3128 err = receive_object(rwa, drro, rrd->payload); 3129 kmem_free(rrd->payload, rrd->payload_size); 3130 rrd->payload = NULL; 3131 break; 3132 } 3133 case DRR_FREEOBJECTS: 3134 { 3135 struct drr_freeobjects *drrfo = 3136 &rrd->header.drr_u.drr_freeobjects; 3137 err = receive_freeobjects(rwa, drrfo); 3138 break; 3139 } 3140 case DRR_WRITE: 3141 { 3142 err = receive_process_write_record(rwa, rrd); 3143 if (rwa->heal) { 3144 /* 3145 * If healing - always free the abd after processing 3146 */ 3147 abd_free(rrd->abd); 3148 rrd->abd = NULL; 3149 } else if (err != EAGAIN) { 3150 /* 3151 * On success, a non-healing 3152 * receive_process_write_record() returns 3153 * EAGAIN to indicate that we do not want to free 3154 * the rrd or arc_buf. 3155 */ 3156 ASSERT(err != 0); 3157 abd_free(rrd->abd); 3158 rrd->abd = NULL; 3159 } 3160 break; 3161 } 3162 case DRR_WRITE_EMBEDDED: 3163 { 3164 struct drr_write_embedded *drrwe = 3165 &rrd->header.drr_u.drr_write_embedded; 3166 err = receive_write_embedded(rwa, drrwe, rrd->payload); 3167 kmem_free(rrd->payload, rrd->payload_size); 3168 rrd->payload = NULL; 3169 break; 3170 } 3171 case DRR_FREE: 3172 { 3173 struct drr_free *drrf = &rrd->header.drr_u.drr_free; 3174 err = receive_free(rwa, drrf); 3175 break; 3176 } 3177 case DRR_SPILL: 3178 { 3179 struct drr_spill *drrs = &rrd->header.drr_u.drr_spill; 3180 err = receive_spill(rwa, drrs, rrd->abd); 3181 if (err != 0) 3182 abd_free(rrd->abd); 3183 rrd->abd = NULL; 3184 rrd->payload = NULL; 3185 break; 3186 } 3187 case DRR_OBJECT_RANGE: 3188 { 3189 struct drr_object_range *drror = 3190 &rrd->header.drr_u.drr_object_range; 3191 err = receive_object_range(rwa, drror); 3192 break; 3193 } 3194 case DRR_REDACT: 3195 { 3196 struct drr_redact *drrr = &rrd->header.drr_u.drr_redact; 3197 err = receive_redact(rwa, drrr); 3198 break; 3199 } 3200 default: 3201 err = (SET_ERROR(EINVAL)); 3202 } 3203 3204 if (err != 0) 3205 dprintf_drr(rrd, err); 3206 3207 return (err); 3208 } 3209 3210 /* 3211 * dmu_recv_stream's worker thread; pull records off the queue, and then call 3212 * receive_process_record When we're done, signal the main thread and exit. 3213 */ 3214 static __attribute__((noreturn)) void 3215 receive_writer_thread(void *arg) 3216 { 3217 struct receive_writer_arg *rwa = arg; 3218 struct receive_record_arg *rrd; 3219 fstrans_cookie_t cookie = spl_fstrans_mark(); 3220 3221 for (rrd = bqueue_dequeue(&rwa->q); !rrd->eos_marker; 3222 rrd = bqueue_dequeue(&rwa->q)) { 3223 /* 3224 * If there's an error, the main thread will stop putting things 3225 * on the queue, but we need to clear everything in it before we 3226 * can exit. 3227 */ 3228 int err = 0; 3229 if (rwa->err == 0) { 3230 err = receive_process_record(rwa, rrd); 3231 } else if (rrd->abd != NULL) { 3232 abd_free(rrd->abd); 3233 rrd->abd = NULL; 3234 rrd->payload = NULL; 3235 } else if (rrd->payload != NULL) { 3236 kmem_free(rrd->payload, rrd->payload_size); 3237 rrd->payload = NULL; 3238 } 3239 /* 3240 * EAGAIN indicates that this record has been saved (on 3241 * raw->write_batch), and will be used again, so we don't 3242 * free it. 3243 * When healing data we always need to free the record. 3244 */ 3245 if (err != EAGAIN || rwa->heal) { 3246 if (rwa->err == 0) 3247 rwa->err = err; 3248 kmem_free(rrd, sizeof (*rrd)); 3249 } 3250 } 3251 kmem_free(rrd, sizeof (*rrd)); 3252 3253 if (rwa->heal) { 3254 zio_wait(rwa->heal_pio); 3255 } else { 3256 int err = flush_write_batch(rwa); 3257 if (rwa->err == 0) 3258 rwa->err = err; 3259 } 3260 mutex_enter(&rwa->mutex); 3261 rwa->done = B_TRUE; 3262 cv_signal(&rwa->cv); 3263 mutex_exit(&rwa->mutex); 3264 spl_fstrans_unmark(cookie); 3265 thread_exit(); 3266 } 3267 3268 static int 3269 resume_check(dmu_recv_cookie_t *drc, nvlist_t *begin_nvl) 3270 { 3271 uint64_t val; 3272 objset_t *mos = dmu_objset_pool(drc->drc_os)->dp_meta_objset; 3273 uint64_t dsobj = dmu_objset_id(drc->drc_os); 3274 uint64_t resume_obj, resume_off; 3275 3276 if (nvlist_lookup_uint64(begin_nvl, 3277 "resume_object", &resume_obj) != 0 || 3278 nvlist_lookup_uint64(begin_nvl, 3279 "resume_offset", &resume_off) != 0) { 3280 return (SET_ERROR(EINVAL)); 3281 } 3282 VERIFY0(zap_lookup(mos, dsobj, 3283 DS_FIELD_RESUME_OBJECT, sizeof (val), 1, &val)); 3284 if (resume_obj != val) 3285 return (SET_ERROR(EINVAL)); 3286 VERIFY0(zap_lookup(mos, dsobj, 3287 DS_FIELD_RESUME_OFFSET, sizeof (val), 1, &val)); 3288 if (resume_off != val) 3289 return (SET_ERROR(EINVAL)); 3290 3291 return (0); 3292 } 3293 3294 /* 3295 * Read in the stream's records, one by one, and apply them to the pool. There 3296 * are two threads involved; the thread that calls this function will spin up a 3297 * worker thread, read the records off the stream one by one, and issue 3298 * prefetches for any necessary indirect blocks. It will then push the records 3299 * onto an internal blocking queue. The worker thread will pull the records off 3300 * the queue, and actually write the data into the DMU. This way, the worker 3301 * thread doesn't have to wait for reads to complete, since everything it needs 3302 * (the indirect blocks) will be prefetched. 3303 * 3304 * NB: callers *must* call dmu_recv_end() if this succeeds. 3305 */ 3306 int 3307 dmu_recv_stream(dmu_recv_cookie_t *drc, offset_t *voffp) 3308 { 3309 int err = 0; 3310 struct receive_writer_arg *rwa = kmem_zalloc(sizeof (*rwa), KM_SLEEP); 3311 3312 if (dsl_dataset_has_resume_receive_state(drc->drc_ds)) { 3313 uint64_t bytes = 0; 3314 (void) zap_lookup(drc->drc_ds->ds_dir->dd_pool->dp_meta_objset, 3315 drc->drc_ds->ds_object, DS_FIELD_RESUME_BYTES, 3316 sizeof (bytes), 1, &bytes); 3317 drc->drc_bytes_read += bytes; 3318 } 3319 3320 drc->drc_ignore_objlist = objlist_create(); 3321 3322 /* these were verified in dmu_recv_begin */ 3323 ASSERT3U(DMU_GET_STREAM_HDRTYPE(drc->drc_drrb->drr_versioninfo), ==, 3324 DMU_SUBSTREAM); 3325 ASSERT3U(drc->drc_drrb->drr_type, <, DMU_OST_NUMTYPES); 3326 3327 ASSERT(dsl_dataset_phys(drc->drc_ds)->ds_flags & DS_FLAG_INCONSISTENT); 3328 ASSERT0(drc->drc_os->os_encrypted && 3329 (drc->drc_featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)); 3330 3331 /* handle DSL encryption key payload */ 3332 if (drc->drc_featureflags & DMU_BACKUP_FEATURE_RAW) { 3333 nvlist_t *keynvl = NULL; 3334 3335 ASSERT(drc->drc_os->os_encrypted); 3336 ASSERT(drc->drc_raw); 3337 3338 err = nvlist_lookup_nvlist(drc->drc_begin_nvl, "crypt_keydata", 3339 &keynvl); 3340 if (err != 0) 3341 goto out; 3342 3343 if (!drc->drc_heal) { 3344 /* 3345 * If this is a new dataset we set the key immediately. 3346 * Otherwise we don't want to change the key until we 3347 * are sure the rest of the receive succeeded so we 3348 * stash the keynvl away until then. 3349 */ 3350 err = dsl_crypto_recv_raw(spa_name(drc->drc_os->os_spa), 3351 drc->drc_ds->ds_object, drc->drc_fromsnapobj, 3352 drc->drc_drrb->drr_type, keynvl, drc->drc_newfs); 3353 if (err != 0) 3354 goto out; 3355 } 3356 3357 /* see comment in dmu_recv_end_sync() */ 3358 drc->drc_ivset_guid = 0; 3359 (void) nvlist_lookup_uint64(keynvl, "to_ivset_guid", 3360 &drc->drc_ivset_guid); 3361 3362 if (!drc->drc_newfs) 3363 drc->drc_keynvl = fnvlist_dup(keynvl); 3364 } 3365 3366 if (drc->drc_featureflags & DMU_BACKUP_FEATURE_RESUMING) { 3367 err = resume_check(drc, drc->drc_begin_nvl); 3368 if (err != 0) 3369 goto out; 3370 } 3371 3372 /* 3373 * For compatibility with recursive send streams, we do this here, 3374 * rather than in dmu_recv_begin. If we pull the next header too 3375 * early, and it's the END record, we break the `recv_skip` logic. 3376 */ 3377 if (drc->drc_drr_begin->drr_payloadlen == 0) { 3378 err = receive_read_payload_and_next_header(drc, 0, NULL); 3379 if (err != 0) 3380 goto out; 3381 } 3382 3383 /* 3384 * If we failed before this point we will clean up any new resume 3385 * state that was created. Now that we've gotten past the initial 3386 * checks we are ok to retain that resume state. 3387 */ 3388 drc->drc_should_save = B_TRUE; 3389 3390 (void) bqueue_init(&rwa->q, zfs_recv_queue_ff, 3391 MAX(zfs_recv_queue_length, 2 * zfs_max_recordsize), 3392 offsetof(struct receive_record_arg, node)); 3393 cv_init(&rwa->cv, NULL, CV_DEFAULT, NULL); 3394 mutex_init(&rwa->mutex, NULL, MUTEX_DEFAULT, NULL); 3395 rwa->os = drc->drc_os; 3396 rwa->byteswap = drc->drc_byteswap; 3397 rwa->heal = drc->drc_heal; 3398 rwa->tofs = drc->drc_tofs; 3399 rwa->resumable = drc->drc_resumable; 3400 rwa->raw = drc->drc_raw; 3401 rwa->spill = drc->drc_spill; 3402 rwa->full = (drc->drc_drr_begin->drr_u.drr_begin.drr_fromguid == 0); 3403 rwa->os->os_raw_receive = drc->drc_raw; 3404 if (drc->drc_heal) { 3405 rwa->heal_pio = zio_root(drc->drc_os->os_spa, NULL, NULL, 3406 ZIO_FLAG_GODFATHER); 3407 } 3408 list_create(&rwa->write_batch, sizeof (struct receive_record_arg), 3409 offsetof(struct receive_record_arg, node.bqn_node)); 3410 3411 (void) thread_create(NULL, 0, receive_writer_thread, rwa, 0, curproc, 3412 TS_RUN, minclsyspri); 3413 /* 3414 * We're reading rwa->err without locks, which is safe since we are the 3415 * only reader, and the worker thread is the only writer. It's ok if we 3416 * miss a write for an iteration or two of the loop, since the writer 3417 * thread will keep freeing records we send it until we send it an eos 3418 * marker. 3419 * 3420 * We can leave this loop in 3 ways: First, if rwa->err is 3421 * non-zero. In that case, the writer thread will free the rrd we just 3422 * pushed. Second, if we're interrupted; in that case, either it's the 3423 * first loop and drc->drc_rrd was never allocated, or it's later, and 3424 * drc->drc_rrd has been handed off to the writer thread who will free 3425 * it. Finally, if receive_read_record fails or we're at the end of the 3426 * stream, then we free drc->drc_rrd and exit. 3427 */ 3428 while (rwa->err == 0) { 3429 if (issig()) { 3430 err = SET_ERROR(EINTR); 3431 break; 3432 } 3433 3434 ASSERT3P(drc->drc_rrd, ==, NULL); 3435 drc->drc_rrd = drc->drc_next_rrd; 3436 drc->drc_next_rrd = NULL; 3437 /* Allocates and loads header into drc->drc_next_rrd */ 3438 err = receive_read_record(drc); 3439 3440 if (drc->drc_rrd->header.drr_type == DRR_END || err != 0) { 3441 kmem_free(drc->drc_rrd, sizeof (*drc->drc_rrd)); 3442 drc->drc_rrd = NULL; 3443 break; 3444 } 3445 3446 bqueue_enqueue(&rwa->q, drc->drc_rrd, 3447 sizeof (struct receive_record_arg) + 3448 drc->drc_rrd->payload_size); 3449 drc->drc_rrd = NULL; 3450 } 3451 3452 ASSERT3P(drc->drc_rrd, ==, NULL); 3453 drc->drc_rrd = kmem_zalloc(sizeof (*drc->drc_rrd), KM_SLEEP); 3454 drc->drc_rrd->eos_marker = B_TRUE; 3455 bqueue_enqueue_flush(&rwa->q, drc->drc_rrd, 1); 3456 3457 mutex_enter(&rwa->mutex); 3458 while (!rwa->done) { 3459 /* 3460 * We need to use cv_wait_sig() so that any process that may 3461 * be sleeping here can still fork. 3462 */ 3463 (void) cv_wait_sig(&rwa->cv, &rwa->mutex); 3464 } 3465 mutex_exit(&rwa->mutex); 3466 3467 /* 3468 * If we are receiving a full stream as a clone, all object IDs which 3469 * are greater than the maximum ID referenced in the stream are 3470 * by definition unused and must be freed. 3471 */ 3472 if (drc->drc_clone && drc->drc_drrb->drr_fromguid == 0) { 3473 uint64_t obj = rwa->max_object + 1; 3474 int free_err = 0; 3475 int next_err = 0; 3476 3477 while (next_err == 0) { 3478 free_err = dmu_free_long_object(rwa->os, obj); 3479 if (free_err != 0 && free_err != ENOENT) 3480 break; 3481 3482 next_err = dmu_object_next(rwa->os, &obj, FALSE, 0); 3483 } 3484 3485 if (err == 0) { 3486 if (free_err != 0 && free_err != ENOENT) 3487 err = free_err; 3488 else if (next_err != ESRCH) 3489 err = next_err; 3490 } 3491 } 3492 3493 cv_destroy(&rwa->cv); 3494 mutex_destroy(&rwa->mutex); 3495 bqueue_destroy(&rwa->q); 3496 list_destroy(&rwa->write_batch); 3497 if (err == 0) 3498 err = rwa->err; 3499 3500 out: 3501 /* 3502 * If we hit an error before we started the receive_writer_thread 3503 * we need to clean up the next_rrd we create by processing the 3504 * DRR_BEGIN record. 3505 */ 3506 if (drc->drc_next_rrd != NULL) 3507 kmem_free(drc->drc_next_rrd, sizeof (*drc->drc_next_rrd)); 3508 3509 /* 3510 * The objset will be invalidated by dmu_recv_end() when we do 3511 * dsl_dataset_clone_swap_sync_impl(). 3512 */ 3513 drc->drc_os = NULL; 3514 3515 kmem_free(rwa, sizeof (*rwa)); 3516 nvlist_free(drc->drc_begin_nvl); 3517 3518 if (err != 0) { 3519 /* 3520 * Clean up references. If receive is not resumable, 3521 * destroy what we created, so we don't leave it in 3522 * the inconsistent state. 3523 */ 3524 dmu_recv_cleanup_ds(drc); 3525 nvlist_free(drc->drc_keynvl); 3526 } 3527 3528 objlist_destroy(drc->drc_ignore_objlist); 3529 drc->drc_ignore_objlist = NULL; 3530 *voffp = drc->drc_voff; 3531 return (err); 3532 } 3533 3534 static int 3535 dmu_recv_end_check(void *arg, dmu_tx_t *tx) 3536 { 3537 dmu_recv_cookie_t *drc = arg; 3538 dsl_pool_t *dp = dmu_tx_pool(tx); 3539 int error; 3540 3541 ASSERT3P(drc->drc_ds->ds_owner, ==, dmu_recv_tag); 3542 3543 if (drc->drc_heal) { 3544 error = 0; 3545 } else if (!drc->drc_newfs) { 3546 dsl_dataset_t *origin_head; 3547 3548 error = dsl_dataset_hold(dp, drc->drc_tofs, FTAG, &origin_head); 3549 if (error != 0) 3550 return (error); 3551 if (drc->drc_force) { 3552 /* 3553 * We will destroy any snapshots in tofs (i.e. before 3554 * origin_head) that are after the origin (which is 3555 * the snap before drc_ds, because drc_ds can not 3556 * have any snaps of its own). 3557 */ 3558 uint64_t obj; 3559 3560 obj = dsl_dataset_phys(origin_head)->ds_prev_snap_obj; 3561 while (obj != 3562 dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj) { 3563 dsl_dataset_t *snap; 3564 error = dsl_dataset_hold_obj(dp, obj, FTAG, 3565 &snap); 3566 if (error != 0) 3567 break; 3568 if (snap->ds_dir != origin_head->ds_dir) 3569 error = SET_ERROR(EINVAL); 3570 if (error == 0) { 3571 error = dsl_destroy_snapshot_check_impl( 3572 snap, B_FALSE); 3573 } 3574 obj = dsl_dataset_phys(snap)->ds_prev_snap_obj; 3575 dsl_dataset_rele(snap, FTAG); 3576 if (error != 0) 3577 break; 3578 } 3579 if (error != 0) { 3580 dsl_dataset_rele(origin_head, FTAG); 3581 return (error); 3582 } 3583 } 3584 if (drc->drc_keynvl != NULL) { 3585 error = dsl_crypto_recv_raw_key_check(drc->drc_ds, 3586 drc->drc_keynvl, tx); 3587 if (error != 0) { 3588 dsl_dataset_rele(origin_head, FTAG); 3589 return (error); 3590 } 3591 } 3592 3593 error = dsl_dataset_clone_swap_check_impl(drc->drc_ds, 3594 origin_head, drc->drc_force, drc->drc_owner, tx); 3595 if (error != 0) { 3596 dsl_dataset_rele(origin_head, FTAG); 3597 return (error); 3598 } 3599 error = dsl_dataset_snapshot_check_impl(origin_head, 3600 drc->drc_tosnap, tx, B_TRUE, 1, 3601 drc->drc_cred, drc->drc_proc); 3602 dsl_dataset_rele(origin_head, FTAG); 3603 if (error != 0) 3604 return (error); 3605 3606 error = dsl_destroy_head_check_impl(drc->drc_ds, 1); 3607 } else { 3608 error = dsl_dataset_snapshot_check_impl(drc->drc_ds, 3609 drc->drc_tosnap, tx, B_TRUE, 1, 3610 drc->drc_cred, drc->drc_proc); 3611 } 3612 return (error); 3613 } 3614 3615 static void 3616 dmu_recv_end_sync(void *arg, dmu_tx_t *tx) 3617 { 3618 dmu_recv_cookie_t *drc = arg; 3619 dsl_pool_t *dp = dmu_tx_pool(tx); 3620 boolean_t encrypted = drc->drc_ds->ds_dir->dd_crypto_obj != 0; 3621 uint64_t newsnapobj = 0; 3622 3623 spa_history_log_internal_ds(drc->drc_ds, "finish receiving", 3624 tx, "snap=%s", drc->drc_tosnap); 3625 drc->drc_ds->ds_objset->os_raw_receive = B_FALSE; 3626 3627 if (drc->drc_heal) { 3628 if (drc->drc_keynvl != NULL) { 3629 nvlist_free(drc->drc_keynvl); 3630 drc->drc_keynvl = NULL; 3631 } 3632 } else if (!drc->drc_newfs) { 3633 dsl_dataset_t *origin_head; 3634 3635 VERIFY0(dsl_dataset_hold(dp, drc->drc_tofs, FTAG, 3636 &origin_head)); 3637 3638 if (drc->drc_force) { 3639 /* 3640 * Destroy any snapshots of drc_tofs (origin_head) 3641 * after the origin (the snap before drc_ds). 3642 */ 3643 uint64_t obj; 3644 3645 obj = dsl_dataset_phys(origin_head)->ds_prev_snap_obj; 3646 while (obj != 3647 dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj) { 3648 dsl_dataset_t *snap; 3649 VERIFY0(dsl_dataset_hold_obj(dp, obj, FTAG, 3650 &snap)); 3651 ASSERT3P(snap->ds_dir, ==, origin_head->ds_dir); 3652 obj = dsl_dataset_phys(snap)->ds_prev_snap_obj; 3653 dsl_destroy_snapshot_sync_impl(snap, 3654 B_FALSE, tx); 3655 dsl_dataset_rele(snap, FTAG); 3656 } 3657 } 3658 if (drc->drc_keynvl != NULL) { 3659 dsl_crypto_recv_raw_key_sync(drc->drc_ds, 3660 drc->drc_keynvl, tx); 3661 nvlist_free(drc->drc_keynvl); 3662 drc->drc_keynvl = NULL; 3663 } 3664 3665 VERIFY3P(drc->drc_ds->ds_prev, ==, 3666 origin_head->ds_prev); 3667 3668 dsl_dataset_clone_swap_sync_impl(drc->drc_ds, 3669 origin_head, tx); 3670 /* 3671 * The objset was evicted by dsl_dataset_clone_swap_sync_impl, 3672 * so drc_os is no longer valid. 3673 */ 3674 drc->drc_os = NULL; 3675 3676 dsl_dataset_snapshot_sync_impl(origin_head, 3677 drc->drc_tosnap, tx); 3678 3679 /* set snapshot's creation time and guid */ 3680 dmu_buf_will_dirty(origin_head->ds_prev->ds_dbuf, tx); 3681 dsl_dataset_phys(origin_head->ds_prev)->ds_creation_time = 3682 drc->drc_drrb->drr_creation_time; 3683 dsl_dataset_phys(origin_head->ds_prev)->ds_guid = 3684 drc->drc_drrb->drr_toguid; 3685 dsl_dataset_phys(origin_head->ds_prev)->ds_flags &= 3686 ~DS_FLAG_INCONSISTENT; 3687 3688 dmu_buf_will_dirty(origin_head->ds_dbuf, tx); 3689 dsl_dataset_phys(origin_head)->ds_flags &= 3690 ~DS_FLAG_INCONSISTENT; 3691 3692 newsnapobj = 3693 dsl_dataset_phys(origin_head)->ds_prev_snap_obj; 3694 3695 dsl_dataset_rele(origin_head, FTAG); 3696 dsl_destroy_head_sync_impl(drc->drc_ds, tx); 3697 3698 if (drc->drc_owner != NULL) 3699 VERIFY3P(origin_head->ds_owner, ==, drc->drc_owner); 3700 } else { 3701 dsl_dataset_t *ds = drc->drc_ds; 3702 3703 dsl_dataset_snapshot_sync_impl(ds, drc->drc_tosnap, tx); 3704 3705 /* set snapshot's creation time and guid */ 3706 dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx); 3707 dsl_dataset_phys(ds->ds_prev)->ds_creation_time = 3708 drc->drc_drrb->drr_creation_time; 3709 dsl_dataset_phys(ds->ds_prev)->ds_guid = 3710 drc->drc_drrb->drr_toguid; 3711 dsl_dataset_phys(ds->ds_prev)->ds_flags &= 3712 ~DS_FLAG_INCONSISTENT; 3713 3714 dmu_buf_will_dirty(ds->ds_dbuf, tx); 3715 dsl_dataset_phys(ds)->ds_flags &= ~DS_FLAG_INCONSISTENT; 3716 if (dsl_dataset_has_resume_receive_state(ds)) { 3717 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3718 DS_FIELD_RESUME_FROMGUID, tx); 3719 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3720 DS_FIELD_RESUME_OBJECT, tx); 3721 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3722 DS_FIELD_RESUME_OFFSET, tx); 3723 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3724 DS_FIELD_RESUME_BYTES, tx); 3725 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3726 DS_FIELD_RESUME_TOGUID, tx); 3727 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3728 DS_FIELD_RESUME_TONAME, tx); 3729 (void) zap_remove(dp->dp_meta_objset, ds->ds_object, 3730 DS_FIELD_RESUME_REDACT_BOOKMARK_SNAPS, tx); 3731 } 3732 newsnapobj = 3733 dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj; 3734 } 3735 3736 /* 3737 * If this is a raw receive, the crypt_keydata nvlist will include 3738 * a to_ivset_guid for us to set on the new snapshot. This value 3739 * will override the value generated by the snapshot code. However, 3740 * this value may not be present, because older implementations of 3741 * the raw send code did not include this value, and we are still 3742 * allowed to receive them if the zfs_disable_ivset_guid_check 3743 * tunable is set, in which case we will leave the newly-generated 3744 * value. 3745 */ 3746 if (!drc->drc_heal && drc->drc_raw && drc->drc_ivset_guid != 0) { 3747 dmu_object_zapify(dp->dp_meta_objset, newsnapobj, 3748 DMU_OT_DSL_DATASET, tx); 3749 VERIFY0(zap_update(dp->dp_meta_objset, newsnapobj, 3750 DS_FIELD_IVSET_GUID, sizeof (uint64_t), 1, 3751 &drc->drc_ivset_guid, tx)); 3752 } 3753 3754 /* 3755 * Release the hold from dmu_recv_begin. This must be done before 3756 * we return to open context, so that when we free the dataset's dnode 3757 * we can evict its bonus buffer. Since the dataset may be destroyed 3758 * at this point (and therefore won't have a valid pointer to the spa) 3759 * we release the key mapping manually here while we do have a valid 3760 * pointer, if it exists. 3761 */ 3762 if (!drc->drc_raw && encrypted) { 3763 (void) spa_keystore_remove_mapping(dmu_tx_pool(tx)->dp_spa, 3764 drc->drc_ds->ds_object, drc->drc_ds); 3765 } 3766 dsl_dataset_disown(drc->drc_ds, 0, dmu_recv_tag); 3767 drc->drc_ds = NULL; 3768 } 3769 3770 static int dmu_recv_end_modified_blocks = 3; 3771 3772 static int 3773 dmu_recv_existing_end(dmu_recv_cookie_t *drc) 3774 { 3775 #ifdef _KERNEL 3776 /* 3777 * We will be destroying the ds; make sure its origin is unmounted if 3778 * necessary. 3779 */ 3780 char name[ZFS_MAX_DATASET_NAME_LEN]; 3781 dsl_dataset_name(drc->drc_ds, name); 3782 zfs_destroy_unmount_origin(name); 3783 #endif 3784 3785 return (dsl_sync_task(drc->drc_tofs, 3786 dmu_recv_end_check, dmu_recv_end_sync, drc, 3787 dmu_recv_end_modified_blocks, ZFS_SPACE_CHECK_NORMAL)); 3788 } 3789 3790 static int 3791 dmu_recv_new_end(dmu_recv_cookie_t *drc) 3792 { 3793 return (dsl_sync_task(drc->drc_tofs, 3794 dmu_recv_end_check, dmu_recv_end_sync, drc, 3795 dmu_recv_end_modified_blocks, ZFS_SPACE_CHECK_NORMAL)); 3796 } 3797 3798 int 3799 dmu_recv_end(dmu_recv_cookie_t *drc, void *owner) 3800 { 3801 int error; 3802 3803 drc->drc_owner = owner; 3804 3805 if (drc->drc_newfs) 3806 error = dmu_recv_new_end(drc); 3807 else 3808 error = dmu_recv_existing_end(drc); 3809 3810 if (error != 0) { 3811 dmu_recv_cleanup_ds(drc); 3812 nvlist_free(drc->drc_keynvl); 3813 } else if (!drc->drc_heal) { 3814 if (drc->drc_newfs) { 3815 zvol_create_minor(drc->drc_tofs); 3816 } 3817 char *snapname = kmem_asprintf("%s@%s", 3818 drc->drc_tofs, drc->drc_tosnap); 3819 zvol_create_minor(snapname); 3820 kmem_strfree(snapname); 3821 } 3822 return (error); 3823 } 3824 3825 /* 3826 * Return TRUE if this objset is currently being received into. 3827 */ 3828 boolean_t 3829 dmu_objset_is_receiving(objset_t *os) 3830 { 3831 return (os->os_dsl_dataset != NULL && 3832 os->os_dsl_dataset->ds_owner == dmu_recv_tag); 3833 } 3834 3835 ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, queue_length, UINT, ZMOD_RW, 3836 "Maximum receive queue length"); 3837 3838 ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, queue_ff, UINT, ZMOD_RW, 3839 "Receive queue fill fraction"); 3840 3841 ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, write_batch_size, UINT, ZMOD_RW, 3842 "Maximum amount of writes to batch into one transaction"); 3843 3844 ZFS_MODULE_PARAM(zfs_recv, zfs_recv_, best_effort_corrective, INT, ZMOD_RW, 3845 "Ignore errors during corrective receive"); 3846 /* END CSTYLED */ 3847