send / receive

static __attribute__((noreturn)) void
send_traverse_thread(void *arg)
{
	struct send_thread_arg *st_arg = arg;
	int err = 0;
	struct send_range *data;
	fstrans_cookie_t cookie = spl_fstrans_mark();

	err = traverse_dataset_resume(st_arg->os->os_dsl_dataset,
	    st_arg->fromtxg, &st_arg->resume,
	    st_arg->flags | TRAVERSE_LOGICAL, send_cb, st_arg);

	if (err != EINTR)
		st_arg->error_code = err;
	data = range_alloc(DATA, 0, 0, 0, B_TRUE);
	bqueue_enqueue_flush(&st_arg->q, data, sizeof (*data));
	spl_fstrans_unmark(cookie);
	thread_exit();
}

	if (BP_GET_LOGICAL_BIRTH(bp) == 0) {
		/*
		 * Since this block has a birth time of 0 it must be one of
		 * two things: a hole created before the
		 * SPA_FEATURE_HOLE_BIRTH feature was enabled, or a hole
		 * which has always been a hole in an object.
		 *
		 * If a file is written sparsely, then the unwritten parts of
		 * the file were "always holes" -- that is, they have been
		 * holes since this object was allocated.  However, we (and
		 * our callers) can not necessarily tell when an object was
		 * allocated.  Therefore, if it's possible that this object
		 * was freed and then its object number reused, we need to
		 * visit all the holes with birth==0.
		 *
		 * If it isn't possible that the object number was reused,
		 * then if SPA_FEATURE_HOLE_BIRTH was enabled before we wrote
		 * all the blocks we will visit as part of this traversal,
		 * then this hole must have always existed, so we can skip
		 * it.  We visit blocks born after (exclusive) td_min_txg.
		 *
		 * Note that the meta-dnode cannot be reallocated.
		 */
		if (!send_holes_without_birth_time &&
		    (!td->td_realloc_possible ||
		    zb->zb_object == DMU_META_DNODE_OBJECT) &&
		    td->td_hole_birth_enabled_txg <= td->td_min_txg)
			return (0);
	} else if (get_birth_time(td, bp) <= td->td_min_txg) {
		return (0);
	}

	if (td->td_flags & TRAVERSE_PRE) {
		SET_BOOKMARK(&czb, objset, object, ZB_DNODE_LEVEL,
		    ZB_DNODE_BLKID);
		err = td->td_func(td->td_spa, NULL, bp, &czb, dnp,
		    td->td_arg);
		if (err == TRAVERSE_VISIT_NO_CHILDREN)
			return (0);
		if (err != 0)
			return (err);
	}

	for (j = 0; j < dnp->dn_nblkptr; j++) {
		SET_BOOKMARK(&czb, objset, object, dnp->dn_nlevels - 1, j);
		err = traverse_visitbp(td, dnp, &dnp->dn_blkptr[j], &czb);
		if (err != 0)
			break;
	}

	if (zb->zb_level == ZB_DNODE_LEVEL) {
		if (zb->zb_object == DMU_META_DNODE_OBJECT)
			return (0);
		record = range_alloc(OBJECT, zb->zb_object, 0, 0, B_FALSE);
		record->sru.object.bp = *bp;
		size_t size  = sizeof (*dnp) * (dnp->dn_extra_slots + 1);
		record->sru.object.dnp = kmem_alloc(size, KM_SLEEP);
		memcpy(record->sru.object.dnp, dnp, size);
		bqueue_enqueue(&sta->q, record, sizeof (*record));
		return (0);
	}
	if (zb->zb_level == 0 && zb->zb_object == DMU_META_DNODE_OBJECT &&
	    !BP_IS_HOLE(bp)) {
		record = range_alloc(OBJECT_RANGE, 0, zb->zb_blkid,
		    zb->zb_blkid + 1, B_FALSE);
		record->sru.object_range.bp = *bp;
		bqueue_enqueue(&sta->q, record, sizeof (*record));
		return (0);
	}
	if (zb->zb_level < 0 || (zb->zb_level > 0 && !BP_IS_HOLE(bp)))
		return (0);
	if (zb->zb_object == DMU_META_DNODE_OBJECT && !BP_IS_HOLE(bp))
		return (0);

	if (!overflow_multiply(span, zb->zb_blkid, &start) || (!(zb->zb_blkid ==
	    DMU_SPILL_BLKID || DMU_OT_IS_METADATA(dnp->dn_type)) &&
	    span * zb->zb_blkid > dnp->dn_maxblkid)) {
		ASSERT(BP_IS_HOLE(bp));
		return (0);
	}

	if (zb->zb_blkid == DMU_SPILL_BLKID)
		ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);

	enum type record_type = DATA;
	if (BP_IS_HOLE(bp))
		record_type = HOLE;
	else if (BP_IS_REDACTED(bp))
		record_type = REDACT;
	else
		record_type = DATA;

/*
 * For all record types except BEGIN, fill in the checksum (overlaid in
 * drr_u.drr_checksum.drr_checksum).  The checksum verifies everything
 * up to the start of the checksum itself.
 */
static int
dump_record(dmu_send_cookie_t *dscp, void *payload, int payload_len)
{
	dmu_send_outparams_t *dso = dscp->dsc_dso;
	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
	(void) fletcher_4_incremental_native(dscp->dsc_drr,
	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
	    &dscp->dsc_zc);
	if (dscp->dsc_drr->drr_type == DRR_BEGIN) {
		dscp->dsc_sent_begin = B_TRUE;
	} else {
		ASSERT(ZIO_CHECKSUM_IS_ZERO(&dscp->dsc_drr->drr_u.
		    drr_checksum.drr_checksum));
		dscp->dsc_drr->drr_u.drr_checksum.drr_checksum = dscp->dsc_zc;
	}
	if (dscp->dsc_drr->drr_type == DRR_END) {
		dscp->dsc_sent_end = B_TRUE;
	}
	(void) fletcher_4_incremental_native(&dscp->dsc_drr->
	    drr_u.drr_checksum.drr_checksum,
	    sizeof (zio_cksum_t), &dscp->dsc_zc);

	/*
	 * Note: checksum is of everything up to but not including the
	 * checksum itself.
	 */
	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
	receive_cksum(drc,
	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
	    &drc->drc_next_rrd->header);

	zio_cksum_t cksum_orig =
	    drc->drc_next_rrd->header.drr_u.drr_checksum.drr_checksum;
	zio_cksum_t *cksump =
	    &drc->drc_next_rrd->header.drr_u.drr_checksum.drr_checksum;

	if (drc->drc_byteswap)
		byteswap_record(&drc->drc_next_rrd->header);

	if ((!ZIO_CHECKSUM_IS_ZERO(cksump)) &&
	    !ZIO_CHECKSUM_EQUAL(drc->drc_cksum, *cksump)) {
		kmem_free(drc->drc_next_rrd, sizeof (*drc->drc_next_rrd));
		drc->drc_next_rrd = NULL;
		return (SET_ERROR(ECKSUM));
	}

	receive_cksum(drc, sizeof (cksum_orig), &cksum_orig);

	switch (rrd->header.drr_type) {
	case DRR_OBJECT:
	{
		struct drr_object *drro = &rrd->header.drr_u.drr_object;
		err = receive_object(rwa, drro, rrd->payload);
		vmem_free(rrd->payload, rrd->payload_size);
		rrd->payload = NULL;
		break;
	}
	case DRR_FREEOBJECTS:
	{
		struct drr_freeobjects *drrfo =
		    &rrd->header.drr_u.drr_freeobjects;
		err = receive_freeobjects(rwa, drrfo);
		break;
	}
	case DRR_WRITE:
	{
		err = receive_process_write_record(rwa, rrd);
		if (rwa->heal) {
			/*
			 * If healing - always free the abd after processing
			 */
			abd_free(rrd->abd);
			rrd->abd = NULL;
		} else if (err != EAGAIN) {