+ rc = lod_striping_load(env, lo);
+ if (rc != 0)
+ RETURN(rc);
+
+ /* No stripes */
+ if (lo->ldo_dir_stripe_count <= 1)
+ RETURN(0);
+
+ slave_locks_size = offsetof(typeof(*slave_locks),
+ ha_handles[lo->ldo_dir_stripe_count]);
+ /* Freed in lod_object_unlock */
+ OBD_ALLOC(slave_locks, slave_locks_size);
+ if (!slave_locks)
+ RETURN(-ENOMEM);
+ slave_locks->ha_count = lo->ldo_dir_stripe_count;
+
+ /* striped directory lock */
+ for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
+ struct lustre_handle lockh;
+ struct ldlm_res_id *res_id;
+ struct dt_object *stripe;
+
+ stripe = lo->ldo_stripe[i];
+ if (!stripe)
+ continue;
+
+ res_id = &lod_env_info(env)->lti_res_id;
+ fid_build_reg_res_name(lu_object_fid(&stripe->do_lu), res_id);
+ einfo->ei_res_id = res_id;
+
+ if (dt_object_remote(stripe)) {
+ set_bit(i, (void *)slave_locks->ha_map);
+ rc = dt_object_lock(env, stripe, &lockh, einfo, policy);
+ } else {
+ struct ldlm_namespace *ns = einfo->ei_namespace;
+ ldlm_blocking_callback blocking = einfo->ei_cb_local_bl;
+ ldlm_completion_callback completion = einfo->ei_cb_cp;
+ __u64 dlmflags = LDLM_FL_ATOMIC_CB;
+
+ if (einfo->ei_mode == LCK_PW ||
+ einfo->ei_mode == LCK_EX)
+ dlmflags |= LDLM_FL_COS_INCOMPAT;
+
+ LASSERT(ns != NULL);
+ rc = ldlm_cli_enqueue_local(env, ns, res_id, LDLM_IBITS,
+ policy, einfo->ei_mode,
+ &dlmflags, blocking,
+ completion, NULL,
+ NULL, 0, LVB_T_NONE,
+ NULL, &lockh);
+ }
+ if (rc) {
+ while (i--)
+ ldlm_lock_decref_and_cancel(
+ &slave_locks->ha_handles[i],
+ einfo->ei_mode);
+ OBD_FREE(slave_locks, slave_locks_size);
+ RETURN(rc);
+ }
+ slave_locks->ha_handles[i] = lockh;
+ }
+ einfo->ei_cbdata = slave_locks;
+
+ RETURN(0);
+}
+
+/**
+ * Implementation of dt_object_operations::do_invalidate.
+ *
+ * \see dt_object_operations::do_invalidate() in the API description for details
+ */
+static int lod_invalidate(const struct lu_env *env, struct dt_object *dt)
+{
+ return dt_invalidate(env, dt_object_child(dt));
+}
+
+static int lod_declare_instantiate_components(const struct lu_env *env,
+ struct lod_object *lo, struct thandle *th)
+{
+ struct lod_thread_info *info = lod_env_info(env);
+ int i;
+ int rc = 0;
+ ENTRY;
+
+ LASSERT(info->lti_count < lo->ldo_comp_cnt);
+
+ for (i = 0; i < info->lti_count; i++) {
+ rc = lod_qos_prep_create(env, lo, NULL, th,
+ info->lti_comp_idx[i]);
+ if (rc)
+ break;
+ }
+
+ if (!rc) {
+ info->lti_buf.lb_len = lod_comp_md_size(lo, false);
+ rc = lod_sub_declare_xattr_set(env, lod_object_child(lo),
+ &info->lti_buf, XATTR_NAME_LOV, 0, th);
+ }
+
+ RETURN(rc);
+}
+
+/**
+ * Check OSTs for an existing component for further extension
+ *
+ * Checks if OSTs are still healthy and not out of space. Gets free space
+ * on OSTs (relative to allocation watermark rmb_low) and compares to
+ * the proposed new_end for this component.
+ *
+ * Decides whether or not to extend a component on its current OSTs.
+ *
+ * \param[in] env execution environment for this thread
+ * \param[in] lo object we're checking
+ * \param[in] index index of this component
+ * \param[in] extension_size extension size for this component
+ * \param[in] extent layout extent for requested operation
+ * \param[in] comp_extent extension component extent
+ * \param[in] write if this is write operation
+ *
+ * \retval true - OK to extend on current OSTs
+ * \retval false - do not extend on current OSTs
+ */
+static bool lod_sel_osts_allowed(const struct lu_env *env,
+ struct lod_object *lo,
+ int index, __u64 extension_size,
+ struct lu_extent *extent,
+ struct lu_extent *comp_extent, int write)
+{
+ struct lod_layout_component *lod_comp = &lo->ldo_comp_entries[index];
+ struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
+ struct obd_statfs *sfs = &lod_env_info(env)->lti_osfs;
+ __u64 available = 0;
+ __u64 size;
+ bool ret = true;
+ int i, rc;
+
+ ENTRY;
+
+ LASSERT(lod_comp->llc_stripe_count != 0);
+
+ if (write == 0 ||
+ (extent->e_start == 0 && extent->e_end == OBD_OBJECT_EOF)) {
+ /* truncate or append */
+ size = extension_size;
+ } else {
+ /* In case of write op, check the real write extent,
+ * it may be larger than the extension_size */
+ size = roundup(min(extent->e_end, comp_extent->e_end) -
+ max(extent->e_start, comp_extent->e_start),
+ extension_size);
+ }
+ /* extension_size is file level, so we must divide by stripe count to
+ * compare it to available space on a single OST */
+ size /= lod_comp->llc_stripe_count;
+
+ lod_getref(&lod->lod_ost_descs);
+ for (i = 0; i < lod_comp->llc_stripe_count; i++) {
+ int index = lod_comp->llc_ost_indices[i];
+ struct lod_tgt_desc *ost = OST_TGT(lod, index);
+ struct obd_statfs_info info = { 0 };
+ int j, repeated = 0;
+
+ LASSERT(ost);
+
+ /* Get the number of times this OST repeats in this component.
+ * Note: inter-component repeats are not counted as this is
+ * considered as a rare case: we try to not repeat OST in other
+ * components if possible. */
+ for (j = 0; j < lod_comp->llc_stripe_count; j++) {
+ if (index != lod_comp->llc_ost_indices[j])
+ continue;
+
+ /* already handled */
+ if (j < i)
+ break;
+
+ repeated++;
+ }
+ if (j < lod_comp->llc_stripe_count)
+ continue;
+
+ if (!cfs_bitmap_check(lod->lod_ost_bitmap, index)) {
+ CDEBUG(D_LAYOUT, "ost %d no longer present\n", index);
+ ret = false;
+ break;
+ }
+
+ rc = dt_statfs_info(env, ost->ltd_tgt, sfs, &info);
+ if (rc) {
+ CDEBUG(D_LAYOUT, "statfs failed for ost %d, error %d\n",
+ index, rc);
+ ret = false;
+ break;
+ }
+
+ if (sfs->os_state & OS_STATE_ENOSPC ||
+ sfs->os_state & OS_STATE_READONLY ||
+ sfs->os_state & OS_STATE_DEGRADED) {
+ CDEBUG(D_LAYOUT, "ost %d is not availble for SEL "
+ "extension, state %u\n", index, sfs->os_state);
+ ret = false;
+ break;
+ }
+
+ /* In bytes */
+ available = sfs->os_bavail * sfs->os_bsize;
+ /* 'available' is relative to the allocation threshold */
+ available -= (__u64) info.os_reserved_mb_low << 20;
+
+ CDEBUG(D_LAYOUT, "ost %d lowwm: %d highwm: %d, "
+ "%llu %% blocks available, %llu %% blocks free\n",
+ index, info.os_reserved_mb_low, info.os_reserved_mb_high,
+ (100ull * sfs->os_bavail) / sfs->os_blocks,
+ (100ull * sfs->os_bfree) / sfs->os_blocks);
+
+ if (size * repeated > available) {
+ ret = false;
+ CDEBUG(D_LAYOUT, "low space on ost %d, available %llu "
+ "< extension size %llu\n", index, available,
+ extension_size);
+ break;
+ }
+ }
+ lod_putref(lod, &lod->lod_ost_descs);
+
+ RETURN(ret);
+}
+
+/**
+ * Adjust extents after component removal
+ *
+ * When we remove an extension component, we move the start of the next
+ * component to match the start of the extension component, so no space is left
+ * without layout.
+ *
+ * \param[in] env execution environment for this thread
+ * \param[in] lo object
+ * \param[in] max_comp layout component
+ * \param[in] index index of this component
+ *
+ * \retval 0 on success
+ * \retval negative errno on error
+ */
+static void lod_sel_adjust_extents(const struct lu_env *env,
+ struct lod_object *lo,
+ int max_comp, int index)
+{
+ struct lod_layout_component *lod_comp = NULL;
+ struct lod_layout_component *next = NULL;
+ struct lod_layout_component *prev = NULL;
+ __u64 new_start = 0;
+ __u64 start;
+ int i;
+
+ /* Extension space component */
+ lod_comp = &lo->ldo_comp_entries[index];
+ next = &lo->ldo_comp_entries[index + 1];
+ prev = &lo->ldo_comp_entries[index - 1];
+
+ LASSERT(lod_comp != NULL && prev != NULL && next != NULL);
+ LASSERT(lod_comp->llc_flags & LCME_FL_EXTENSION);
+
+ /* Previous is being removed */
+ if (prev && prev->llc_id == LCME_ID_INVAL)
+ new_start = prev->llc_extent.e_start;
+ else
+ new_start = lod_comp->llc_extent.e_start;
+
+ for (i = index + 1; i < max_comp; i++) {
+ lod_comp = &lo->ldo_comp_entries[i];