Metaslabs y space maps

	ASSERT(MUTEX_HELD(&msp->ms_lock));
	VERIFY(!msp->ms_condensing);
	VERIFY0(msp->ms_disabled);
	VERIFY0(msp->ms_new);

	start = mc->mc_ops->msop_alloc(msp, size, max_size, actual_size);
	if (start != -1ULL) {
		size = *actual_size;
		metaslab_group_t *mg = msp->ms_group;
		vdev_t *vd = mg->mg_vd;

		VERIFY0(P2PHASE(start, 1ULL << vd->vdev_ashift));
		VERIFY0(P2PHASE(size, 1ULL << vd->vdev_ashift));
		VERIFY3U(zfs_range_tree_space(rt) - size, <=, msp->ms_size);
		zfs_range_tree_remove(rt, start, size);
		zfs_range_tree_clear(msp->ms_trim, start, size);

		if (zfs_range_tree_is_empty(msp->ms_allocating[txg & TXG_MASK]))
			vdev_dirty(mg->mg_vd, VDD_METASLAB, msp, txg);

		zfs_range_tree_add(msp->ms_allocating[txg & TXG_MASK], start,
		    size);
		msp->ms_allocating_total += size;

		/* Track the last successful allocation */
		msp->ms_alloc_txg = txg;
		metaslab_verify_space(msp, txg);
	}

	uint64_t align = max_size & -max_size;
	uint64_t *cursor = &msp->ms_lbas[highbit64(align) - 1];
	zfs_range_tree_t *rt = msp->ms_allocatable;
	uint_t free_pct = zfs_range_tree_space(rt) * 100 / msp->ms_size;
	uint64_t offset;

	ASSERT(MUTEX_HELD(&msp->ms_lock));

	/*
	 * If we're running low on space, find a segment based on size,
	 * rather than iterating based on offset.
	 */
	if (metaslab_largest_allocatable(msp) < metaslab_df_alloc_threshold ||
	    free_pct < metaslab_df_free_pct) {
		align = size & -size;
		cursor = &msp->ms_lbas[highbit64(align) - 1];
		offset = -1;
	} else {
		offset = metaslab_block_picker(rt, cursor, size, max_size,
		    metaslab_df_max_search, found_size);
		if (max_size != size && offset == -1) {
			align = size & -size;
			cursor = &msp->ms_lbas[highbit64(align) - 1];
			offset = metaslab_block_picker(rt, cursor, size,
			    max_size, metaslab_df_max_search, found_size);
		}
	}

	/*
	 * We'll be going to disk to sync our space accounting, thus we
	 * drop the ms_lock during that time so allocations coming from
	 * open-context (ZIL) for future TXGs do not block.
	 */
	mutex_exit(&msp->ms_lock);
	space_map_t *log_sm = spa_syncing_log_sm(spa);
	if (log_sm != NULL) {
		ASSERT(spa_feature_is_enabled(spa, SPA_FEATURE_LOG_SPACEMAP));
		if (metaslab_unflushed_txg(msp) == 0)
			metaslab_unflushed_add(msp, tx);
		else if (!metaslab_unflushed_dirty(msp))
			metaslab_unflushed_bump(msp, tx, B_TRUE);

		space_map_write(log_sm, alloctree, SM_ALLOC,
		    vd->vdev_id, tx);
		space_map_write(log_sm, msp->ms_freeing, SM_FREE,
		    vd->vdev_id, tx);
		mutex_enter(&msp->ms_lock);

		ASSERT3U(spa->spa_unflushed_stats.sus_memused, >=,
		    metaslab_unflushed_changes_memused(msp));
		spa->spa_unflushed_stats.sus_memused -=
		    metaslab_unflushed_changes_memused(msp);
		zfs_range_tree_remove_xor_add(alloctree,
		    msp->ms_unflushed_frees, msp->ms_unflushed_allocs);
		zfs_range_tree_remove_xor_add(msp->ms_freeing,
		    msp->ms_unflushed_allocs, msp->ms_unflushed_frees);
		spa->spa_unflushed_stats.sus_memused +=
		    metaslab_unflushed_changes_memused(msp);
	} else {
		ASSERT(!spa_feature_is_enabled(spa, SPA_FEATURE_LOG_SPACEMAP));

		space_map_write(msp->ms_sm, alloctree, SM_ALLOC,
		    SM_NO_VDEVID, tx);
		space_map_write(msp->ms_sm, msp->ms_freeing, SM_FREE,
		    SM_NO_VDEVID, tx);
		mutex_enter(&msp->ms_lock);
	}

	zfs_btree_t *t = &rt->rt_root;
	zfs_btree_index_t where;
	for (zfs_range_seg_t *rs = zfs_btree_first(t, &where); rs != NULL;
	    rs = zfs_btree_next(t, &where, &where)) {
		uint64_t offset = (zfs_rs_get_start(rs, rt) - sm->sm_start) >>
		    sm->sm_shift;
		uint64_t length = (zfs_rs_get_end(rs, rt) -
		    zfs_rs_get_start(rs, rt)) >> sm->sm_shift;
		uint8_t words = 1;

		/*
		 * We only write two-word entries when both of the following
		 * are true:
		 *
		 * [1] The feature is enabled.
		 * [2] The offset or run is too big for a single-word entry,
		 *	or the vdev_id is set (meaning not equal to
		 *	SM_NO_VDEVID).
		 *
		 * Note that for purposes of testing we've added the case that
		 * we write two-word entries occasionally when the feature is
		 * enabled and zfs_force_some_double_word_sm_entries has been
		 * set.
		 */
		if (spa_feature_is_active(spa, SPA_FEATURE_SPACEMAP_V2) &&
		    (offset >= (1ULL << SM_OFFSET_BITS) ||
		    length > SM_RUN_MAX ||
		    vdev_id != SM_NO_VDEVID ||
		    (zfs_force_some_double_word_sm_entries &&
		    random_in_range(100) == 0)))
			words = 2;

		space_map_write_seg(sm, zfs_rs_get_start(rs, rt),
		    zfs_rs_get_end(rs, rt), maptype, vdev_id, words, &db,
		    FTAG, tx);
	}

/*
 * debug entry
 *
 *     2     2        10                     50
 *  +-----+-----+------------+----------------------------------+
 *  | 1 0 | act |  syncpass  |        txg (lower bits)          |
 *  +-----+-----+------------+----------------------------------+
 *   63 62 61 60 59        50 49                                0
 *
 *
 * one-word entry
 *
 *    1               47                   1           15
 *  +-----------------------------------------------------------+
 *  | 0 |   offset (sm_shift units)    | type |       run       |
 *  +-----------------------------------------------------------+
 *   63  62                          16   15   14               0
 *
 *
 * two-word entry
 *
 *     2     2               36                      24
 *  +-----+-----+---------------------------+-------------------+
 *  | 1 1 | pad |            run            |       vdev        |
 *  +-----+-----+---------------------------+-------------------+
 *   63 62 61 60 59                       24 23                 0
 *
 *     1                            63
 *  +------+----------------------------------------------------+
 *  | type |                      offset                        |
 *  +------+----------------------------------------------------+
 *     63   62                                                  0
 *
 * Note that a two-word entry will not straddle a block boundary.
 * If necessary, the last word of a block will be padded with a
 * debug entry (with act = syncpass = txg = 0).
 */

		uint64_t run_len = MIN(size, run_max);
		switch (words) {
		case 1:
			*block_cursor = SM_OFFSET_ENCODE(start) |
			    SM_TYPE_ENCODE(maptype) |
			    SM_RUN_ENCODE(run_len);
			block_cursor++;
			break;
		case 2:
			/* write the first word of the entry */
			*block_cursor = SM_PREFIX_ENCODE(SM2_PREFIX) |
			    SM2_RUN_ENCODE(run_len) |
			    SM2_VDEV_ENCODE(vdev_id);
			block_cursor++;

			/* move on to the second word of the entry */
			ASSERT3P(block_cursor, <, block_end);
			*block_cursor = SM2_TYPE_ENCODE(maptype) |
			    SM2_OFFSET_ENCODE(start);
			block_cursor++;
			break;
		default:
			panic("%d-word space map entries are not supported",
			    words);
			break;
		}
		sm->sm_phys->smp_length += words * sizeof (uint64_t);

		start += run_len;
		size -= run_len;
	}

	/*
	 * Move the frees from the defer_tree back to the free
	 * range tree (if it's loaded). Swap the freed_tree and
	 * the defer_tree -- this is safe to do because we've
	 * just emptied out the defer_tree.
	 */
	zfs_range_tree_vacate(*defer_tree,
	    msp->ms_loaded ? zfs_range_tree_add : NULL, msp->ms_allocatable);
	if (defer_allowed) {
		zfs_range_tree_swap(&msp->ms_freed, defer_tree);
	} else {
		zfs_range_tree_vacate(msp->ms_freed,
		    msp->ms_loaded ? zfs_range_tree_add : NULL,
		    msp->ms_allocatable);
	}

static int
space_map_load_callback(space_map_entry_t *sme, void *arg)
{
	space_map_load_arg_t *smla = arg;
	if (sme->sme_type == smla->smla_type) {
		VERIFY3U(zfs_range_tree_space(smla->smla_rt) + sme->sme_run, <=,
		    smla->smla_sm->sm_size);
		zfs_range_tree_add(smla->smla_rt, sme->sme_offset,
		    sme->sme_run);
	} else {
		zfs_range_tree_remove(smla->smla_rt, sme->sme_offset,
		    sme->sme_run);
	}

	return (0);
}

static boolean_t
metaslab_should_condense(metaslab_t *msp)
{
	space_map_t *sm = msp->ms_sm;
	vdev_t *vd = msp->ms_group->mg_vd;
	uint64_t vdev_blocksize = 1ULL << vd->vdev_ashift;

	ASSERT(MUTEX_HELD(&msp->ms_lock));
	ASSERT(msp->ms_loaded);
	ASSERT(sm != NULL);
	ASSERT3U(spa_sync_pass(vd->vdev_spa), ==, 1);

	/*
	 * We always condense metaslabs that are empty and metaslabs for
	 * which a condense request has been made.
	 */
	if (zfs_range_tree_numsegs(msp->ms_allocatable) == 0 ||
	    msp->ms_condense_wanted)
		return (B_TRUE);

	uint64_t record_size = MAX(sm->sm_blksz, vdev_blocksize);
	uint64_t object_size = space_map_length(sm);
	uint64_t optimal_size = space_map_estimate_optimal_size(sm,
	    msp->ms_allocatable, SM_NO_VDEVID);

	return (object_size >= (optimal_size * zfs_condense_pct / 100) &&
	    object_size > zfs_metaslab_condense_block_threshold * record_size);

	/*
	 * Note:
	 * When the log space map feature is enabled, each space map will
	 * always have ALLOCS followed by FREES for each sync pass. This is
	 * typically true even when the log space map feature is disabled,
	 * except from the case where a metaslab goes through metaslab_sync()
	 * and gets condensed. In that case the metaslab's space map will have
	 * ALLOCS followed by FREES (due to condensing) followed by ALLOCS
	 * followed by FREES (due to space_map_write() in metaslab_sync()) for
	 * sync pass 1.
	 */
	zfs_range_tree_t *tmp_tree = zfs_range_tree_create_flags(
	    NULL, type, NULL, start, shift,
	    ZFS_RT_F_DYN_NAME,
	    metaslab_rt_name(msp->ms_group, msp, "tmp_tree"));
	zfs_range_tree_add(tmp_tree, msp->ms_start, msp->ms_size);
	space_map_write(sm, tmp_tree, SM_ALLOC, SM_NO_VDEVID, tx);
	space_map_write(sm, msp->ms_allocatable, SM_FREE, SM_NO_VDEVID, tx);
	space_map_write(sm, condense_tree, SM_FREE, SM_NO_VDEVID, tx);