- struct obd_device *obd = exp->exp_obd;
- struct lmv_obd *lmv = &obd->u.lmv;
- struct lmv_tgt_desc *tgt = NULL;
- struct lmv_object *obj;
- int rc;
- int sidx;
- int loop = 0;
- ENTRY;
-
- rc = lmv_check_connect(obd);
- if (rc)
- RETURN(rc);
-
-repeat:
- ++loop;
- LASSERT(loop <= 2);
- LASSERT(op_data->op_namelen != 0);
-
- obj = lmv_object_find(obd, &op_data->op_fid1);
- if (obj) {
- sidx = raw_name2idx(obj->lo_hashtype,
- obj->lo_objcount,
- op_data->op_name,
- op_data->op_namelen);
- op_data->op_bias &= ~MDS_CHECK_SPLIT;
- op_data->op_fid1 = obj->lo_stripes[sidx].ls_fid;
- tgt = lmv_get_target(lmv,
- obj->lo_stripes[sidx].ls_mds);
- lmv_object_put(obj);
- CDEBUG(D_INODE, "UNLINK '%*s' in "DFID" -> %u\n",
- op_data->op_namelen, op_data->op_name,
- PFID(&op_data->op_fid1), sidx);
- }
-
- if (tgt == NULL) {
- tgt = lmv_find_target(lmv, &op_data->op_fid1);
- if (IS_ERR(tgt))
- RETURN(PTR_ERR(tgt));
- op_data->op_bias |= MDS_CHECK_SPLIT;
- }
-
- op_data->op_fsuid = cfs_curproc_fsuid();
- op_data->op_fsgid = cfs_curproc_fsgid();
- op_data->op_cap = cfs_curproc_cap_pack();
-
- /*
- * If child's fid is given, cancel unused locks for it if it is from
- * another export than parent.
- *
- * LOOKUP lock for child (fid3) should also be cancelled on parent
- * tgt_tgt in mdc_unlink().
- */
- op_data->op_flags |= MF_MDC_CANCEL_FID1 | MF_MDC_CANCEL_FID3;
-
- /*
- * Cancel FULL locks on child (fid3).
- */
- rc = lmv_early_cancel(exp, op_data, tgt->ltd_idx, LCK_EX,
- MDS_INODELOCK_FULL, MF_MDC_CANCEL_FID3);
-
- if (rc == 0)
- rc = md_unlink(tgt->ltd_exp, op_data, request);
-
- if (rc == -ERESTART) {
- LASSERT(*request != NULL);
- DEBUG_REQ(D_WARNING|D_RPCTRACE, *request,
- "Got -ERESTART during unlink!\n");
- ptlrpc_req_finished(*request);
- *request = NULL;
-
- /*
- * Directory got split. Time to update local object and repeat
- * the request with proper MDS.
- */
- rc = lmv_handle_split(exp, &op_data->op_fid1);
- if (rc == 0)
- goto repeat;
- }
- RETURN(rc);
+ struct obd_device *obd = exp->exp_obd;
+ struct lmv_obd *lmv = &obd->u.lmv;
+ struct lmv_tgt_desc *tgt = NULL;
+ struct mdt_body *body;
+ int rc;
+ ENTRY;
+
+ rc = lmv_check_connect(obd);
+ if (rc)
+ RETURN(rc);
+retry:
+ /* Send unlink requests to the MDT where the child is located */
+ if (likely(!fid_is_zero(&op_data->op_fid2)))
+ tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid2);
+ else
+ tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
+ if (IS_ERR(tgt))
+ RETURN(PTR_ERR(tgt));
+
+ op_data->op_fsuid = cfs_curproc_fsuid();
+ op_data->op_fsgid = cfs_curproc_fsgid();
+ op_data->op_cap = cfs_curproc_cap_pack();
+
+ /*
+ * If child's fid is given, cancel unused locks for it if it is from
+ * another export than parent.
+ *
+ * LOOKUP lock for child (fid3) should also be cancelled on parent
+ * tgt_tgt in mdc_unlink().
+ */
+ op_data->op_flags |= MF_MDC_CANCEL_FID1 | MF_MDC_CANCEL_FID3;
+
+ /*
+ * Cancel FULL locks on child (fid3).
+ */
+ rc = lmv_early_cancel(exp, op_data, tgt->ltd_idx, LCK_EX,
+ MDS_INODELOCK_FULL, MF_MDC_CANCEL_FID3);
+
+ if (rc != 0)
+ RETURN(rc);
+
+ CDEBUG(D_INODE, "unlink with fid="DFID"/"DFID" -> mds #%d\n",
+ PFID(&op_data->op_fid1), PFID(&op_data->op_fid2), tgt->ltd_idx);
+
+ rc = md_unlink(tgt->ltd_exp, op_data, request);
+ if (rc != 0 && rc != -EREMOTE)
+ RETURN(rc);
+
+ body = req_capsule_server_get(&(*request)->rq_pill, &RMF_MDT_BODY);
+ if (body == NULL)
+ RETURN(-EPROTO);
+
+ /* Not cross-ref case, just get out of here. */
+ if (likely(!(body->valid & OBD_MD_MDS)))
+ RETURN(0);
+
+ CDEBUG(D_INODE, "%s: try unlink to another MDT for "DFID"\n",
+ exp->exp_obd->obd_name, PFID(&body->fid1));
+
+ /* This is a remote object, try remote MDT, Note: it may
+ * try more than 1 time here, Considering following case
+ * /mnt/lustre is root on MDT0, remote1 is on MDT1
+ * 1. Initially A does not know where remote1 is, it send
+ * unlink RPC to MDT0, MDT0 return -EREMOTE, it will
+ * resend unlink RPC to MDT1 (retry 1st time).
+ *
+ * 2. During the unlink RPC in flight,
+ * client B mv /mnt/lustre/remote1 /mnt/lustre/remote2
+ * and create new remote1, but on MDT0
+ *
+ * 3. MDT1 get unlink RPC(from A), then do remote lock on
+ * /mnt/lustre, then lookup get fid of remote1, and find
+ * it is remote dir again, and replay -EREMOTE again.
+ *
+ * 4. Then A will resend unlink RPC to MDT0. (retry 2nd times).
+ *
+ * In theory, it might try unlimited time here, but it should
+ * be very rare case. */
+ op_data->op_fid2 = body->fid1;
+ ptlrpc_req_finished(*request);
+ *request = NULL;
+
+ goto retry;