-static int lov_update_enqueue_lov(struct obd_export *exp,
- struct lustre_handle *lov_lockhp,
- struct lov_oinfo *loi, __u64 flags, int idx,
- struct ost_id *oi, int rc)
-{
- struct lov_obd *lov = &exp->exp_obd->u.lov;
-
- if (rc != ELDLM_OK &&
- !(rc == ELDLM_LOCK_ABORTED && (flags & LDLM_FL_HAS_INTENT))) {
- memset(lov_lockhp, 0, sizeof(*lov_lockhp));
- if (lov->lov_tgts[idx] && lov->lov_tgts[idx]->ltd_active) {
- /* -EUSERS used by OST to report file contention */
- if (rc != -EINTR && rc != -EUSERS)
- CERROR("%s: enqueue objid "DOSTID" subobj"
- DOSTID" on OST idx %d: rc %d\n",
- exp->exp_obd->obd_name,
- POSTID(oi), POSTID(&loi->loi_oi),
- loi->loi_ost_idx, rc);
- } else
- rc = ELDLM_OK;
- }
- return rc;
-}
-
-int lov_update_enqueue_set(struct lov_request *req, __u32 mode, int rc)
-{
- struct lov_request_set *set = req->rq_rqset;
- struct lustre_handle *lov_lockhp;
- struct obd_info *oi = set->set_oi;
- struct lov_oinfo *loi;
- ENTRY;
-
- LASSERT(oi != NULL);
-
- lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
- loi = oi->oi_md->lsm_oinfo[req->rq_stripe];
-
- /* XXX LOV STACKING: OSC gets a copy, created in lov_prep_enqueue_set
- * and that copy can be arbitrarily out of date.
- *
- * The LOV API is due for a serious rewriting anyways, and this
- * can be addressed then. */
-
- lov_stripe_lock(oi->oi_md);
- osc_update_enqueue(lov_lockhp, loi, oi->oi_flags,
- &req->rq_oi.oi_md->lsm_oinfo[0]->loi_lvb, mode, rc);
- if (rc == ELDLM_LOCK_ABORTED && (oi->oi_flags & LDLM_FL_HAS_INTENT))
- memset(lov_lockhp, 0, sizeof *lov_lockhp);
- rc = lov_update_enqueue_lov(set->set_exp, lov_lockhp, loi, oi->oi_flags,
- req->rq_idx, &oi->oi_md->lsm_oi, rc);
- lov_stripe_unlock(oi->oi_md);
- lov_update_set(set, req, rc);
- RETURN(rc);
-}
-
-/* The callback for osc_enqueue that updates lov info for every OSC request. */
-static int cb_update_enqueue(void *cookie, int rc)
-{
- struct obd_info *oinfo = cookie;
- struct ldlm_enqueue_info *einfo;
- struct lov_request *lovreq;
-
- lovreq = container_of(oinfo, struct lov_request, rq_oi);
- einfo = lovreq->rq_rqset->set_ei;
- return lov_update_enqueue_set(lovreq, einfo->ei_mode, rc);
-}
-
-static int enqueue_done(struct lov_request_set *set, __u32 mode)
-{
- struct lov_request *req;
- struct lov_obd *lov = &set->set_exp->exp_obd->u.lov;
- int completes = cfs_atomic_read(&set->set_completes);
- int rc = 0;
- ENTRY;
-
- /* enqueue/match success, just return */
- if (completes && completes == cfs_atomic_read(&set->set_success))
- RETURN(0);
-
- /* cancel enqueued/matched locks */
- cfs_list_for_each_entry(req, &set->set_list, rq_link) {
- struct lustre_handle *lov_lockhp;
-
- if (!req->rq_complete || req->rq_rc)
- continue;
-
- lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
- LASSERT(lov_lockhp);
- if (!lustre_handle_is_used(lov_lockhp))
- continue;
-
- rc = obd_cancel(lov->lov_tgts[req->rq_idx]->ltd_exp,
- req->rq_oi.oi_md, mode, lov_lockhp);
- if (rc && lov->lov_tgts[req->rq_idx] &&
- lov->lov_tgts[req->rq_idx]->ltd_active)
- CERROR("%s: cancelling obdjid "DOSTID" on OST"
- "idx %d error: rc = %d\n",
- set->set_exp->exp_obd->obd_name,
- POSTID(&req->rq_oi.oi_md->lsm_oi),
- req->rq_idx, rc);
- }
- if (set->set_lockh)
- lov_llh_put(set->set_lockh);
- RETURN(rc);
-}
-
-int lov_fini_enqueue_set(struct lov_request_set *set, __u32 mode, int rc,
- struct ptlrpc_request_set *rqset)
-{
- int ret = 0;
- ENTRY;
-
- if (set == NULL)
- RETURN(0);
- LASSERT(set->set_exp);
- /* Do enqueue_done only for sync requests and if any request
- * succeeded. */
- if (!rqset) {
- if (rc)
- cfs_atomic_set(&set->set_completes, 0);
- ret = enqueue_done(set, mode);
- } else if (set->set_lockh)
- lov_llh_put(set->set_lockh);
-
- lov_put_reqset(set);
-
- RETURN(rc ? rc : ret);
-}
-
-static void lov_llh_addref(void *llhp)
-{
- struct lov_lock_handles *llh = llhp;
-
- cfs_atomic_inc(&llh->llh_refcount);
- CDEBUG(D_INFO, "GETting llh %p : new refcount %d\n", llh,
- cfs_atomic_read(&llh->llh_refcount));
-}
-
-static struct portals_handle_ops lov_handle_ops = {
- .hop_addref = lov_llh_addref,
- .hop_free = NULL,
-};
-
-static struct lov_lock_handles *lov_llh_new(struct lov_stripe_md *lsm)
-{
- struct lov_lock_handles *llh;
-
- OBD_ALLOC(llh, sizeof *llh +
- sizeof(*llh->llh_handles) * lsm->lsm_stripe_count);
- if (llh == NULL)
- return NULL;
-
- cfs_atomic_set(&llh->llh_refcount, 2);
- llh->llh_stripe_count = lsm->lsm_stripe_count;
- CFS_INIT_LIST_HEAD(&llh->llh_handle.h_link);
- class_handle_hash(&llh->llh_handle, &lov_handle_ops);
-
- return llh;
-}
-
-int lov_prep_enqueue_set(struct obd_export *exp, struct obd_info *oinfo,
- struct ldlm_enqueue_info *einfo,
- struct lov_request_set **reqset)
-{
- struct lov_obd *lov = &exp->exp_obd->u.lov;
- struct lov_request_set *set;
- int i, rc = 0;
- ENTRY;
-
- OBD_ALLOC(set, sizeof(*set));
- if (set == NULL)
- RETURN(-ENOMEM);
- lov_init_set(set);
-
- set->set_exp = exp;
- set->set_oi = oinfo;
- set->set_ei = einfo;
- set->set_lockh = lov_llh_new(oinfo->oi_md);
- if (set->set_lockh == NULL)
- GOTO(out_set, rc = -ENOMEM);
- oinfo->oi_lockh->cookie = set->set_lockh->llh_handle.h_cookie;
-
- for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++) {
- struct lov_oinfo *loi;
- struct lov_request *req;
- obd_off start, end;
-
- loi = oinfo->oi_md->lsm_oinfo[i];
- if (!lov_stripe_intersects(oinfo->oi_md, i,
- oinfo->oi_policy.l_extent.start,
- oinfo->oi_policy.l_extent.end,
- &start, &end))
- continue;
-
- if (!lov_check_and_wait_active(lov, loi->loi_ost_idx)) {
- CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
- continue;
- }
-
- OBD_ALLOC(req, sizeof(*req));
- if (req == NULL)
- GOTO(out_set, rc = -ENOMEM);
-
- req->rq_buflen = sizeof(*req->rq_oi.oi_md) +
- sizeof(struct lov_oinfo *) +
- sizeof(struct lov_oinfo);
- OBD_ALLOC_LARGE(req->rq_oi.oi_md, req->rq_buflen);
- if (req->rq_oi.oi_md == NULL) {
- OBD_FREE(req, sizeof(*req));
- GOTO(out_set, rc = -ENOMEM);
- }
- req->rq_oi.oi_md->lsm_oinfo[0] =
- ((void *)req->rq_oi.oi_md) + sizeof(*req->rq_oi.oi_md) +
- sizeof(struct lov_oinfo *);
-
- /* Set lov request specific parameters. */
- req->rq_oi.oi_lockh = set->set_lockh->llh_handles + i;
- req->rq_oi.oi_cb_up = cb_update_enqueue;
- req->rq_oi.oi_flags = oinfo->oi_flags;
-
- LASSERT(req->rq_oi.oi_lockh);
-
- req->rq_oi.oi_policy.l_extent.gid =
- oinfo->oi_policy.l_extent.gid;
- req->rq_oi.oi_policy.l_extent.start = start;
- req->rq_oi.oi_policy.l_extent.end = end;
-
- req->rq_idx = loi->loi_ost_idx;
- req->rq_stripe = i;
-
- /* XXX LOV STACKING: submd should be from the subobj */
- req->rq_oi.oi_md->lsm_oi = loi->loi_oi;
- req->rq_oi.oi_md->lsm_stripe_count = 0;
- req->rq_oi.oi_md->lsm_oinfo[0]->loi_kms_valid =
- loi->loi_kms_valid;
- req->rq_oi.oi_md->lsm_oinfo[0]->loi_kms = loi->loi_kms;
- req->rq_oi.oi_md->lsm_oinfo[0]->loi_lvb = loi->loi_lvb;
-
- lov_set_add_req(req, set);
- }
- if (!set->set_count)
- GOTO(out_set, rc = -EIO);
- *reqset = set;
- RETURN(0);
-out_set:
- lov_fini_enqueue_set(set, einfo->ei_mode, rc, NULL);
- RETURN(rc);
-}
-
-int lov_fini_match_set(struct lov_request_set *set, __u32 mode, __u64 flags)
-{
- int rc = 0;
- ENTRY;
-
- if (set == NULL)
- RETURN(0);
- LASSERT(set->set_exp);
- rc = enqueue_done(set, mode);
- if ((set->set_count == cfs_atomic_read(&set->set_success)) &&
- (flags & LDLM_FL_TEST_LOCK))
- lov_llh_put(set->set_lockh);
-
- lov_put_reqset(set);
-
- RETURN(rc);
-}
-
-int lov_prep_match_set(struct obd_export *exp, struct obd_info *oinfo,
- struct lov_stripe_md *lsm, ldlm_policy_data_t *policy,
- __u32 mode, struct lustre_handle *lockh,
- struct lov_request_set **reqset)
-{
- struct lov_obd *lov = &exp->exp_obd->u.lov;
- struct lov_request_set *set;
- int i, rc = 0;
- ENTRY;
-
- OBD_ALLOC(set, sizeof(*set));
- if (set == NULL)
- RETURN(-ENOMEM);
- lov_init_set(set);
-
- set->set_exp = exp;
- set->set_oi = oinfo;
- set->set_oi->oi_md = lsm;
- set->set_lockh = lov_llh_new(lsm);
- if (set->set_lockh == NULL)
- GOTO(out_set, rc = -ENOMEM);
- lockh->cookie = set->set_lockh->llh_handle.h_cookie;
-
- for (i = 0; i < lsm->lsm_stripe_count; i++){
- struct lov_oinfo *loi;
- struct lov_request *req;
- obd_off start, end;
-
- loi = lsm->lsm_oinfo[i];
- if (!lov_stripe_intersects(lsm, i, policy->l_extent.start,
- policy->l_extent.end, &start, &end))
- continue;
-
- /* FIXME raid1 should grace this error */
- if (!lov_check_and_wait_active(lov, loi->loi_ost_idx)) {
- CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
- GOTO(out_set, rc = -EIO);
- }
-
- OBD_ALLOC(req, sizeof(*req));
- if (req == NULL)
- GOTO(out_set, rc = -ENOMEM);
-
- req->rq_buflen = sizeof(*req->rq_oi.oi_md);
- OBD_ALLOC_LARGE(req->rq_oi.oi_md, req->rq_buflen);
- if (req->rq_oi.oi_md == NULL) {
- OBD_FREE(req, sizeof(*req));
- GOTO(out_set, rc = -ENOMEM);
- }
-
- req->rq_oi.oi_policy.l_extent.start = start;
- req->rq_oi.oi_policy.l_extent.end = end;
- req->rq_oi.oi_policy.l_extent.gid = policy->l_extent.gid;
-
- req->rq_idx = loi->loi_ost_idx;
- req->rq_stripe = i;
-
- /* XXX LOV STACKING: submd should be from the subobj */
- req->rq_oi.oi_md->lsm_oi = loi->loi_oi;
- req->rq_oi.oi_md->lsm_stripe_count = 0;
-
- lov_set_add_req(req, set);
- }
- if (!set->set_count)
- GOTO(out_set, rc = -EIO);
- *reqset = set;
- RETURN(rc);
-out_set:
- lov_fini_match_set(set, mode, 0);
- RETURN(rc);
-}
-
-int lov_fini_cancel_set(struct lov_request_set *set)
-{
- int rc = 0;
- ENTRY;
-
- if (set == NULL)
- RETURN(0);
-
- LASSERT(set->set_exp);
- if (set->set_lockh)
- lov_llh_put(set->set_lockh);
-
- lov_put_reqset(set);
-
- RETURN(rc);
-}
-
-int lov_prep_cancel_set(struct obd_export *exp, struct obd_info *oinfo,
- struct lov_stripe_md *lsm, __u32 mode,
- struct lustre_handle *lockh,
- struct lov_request_set **reqset)
-{
- struct lov_request_set *set;
- int i, rc = 0;
- ENTRY;
-
- OBD_ALLOC(set, sizeof(*set));
- if (set == NULL)
- RETURN(-ENOMEM);
- lov_init_set(set);
-
- set->set_exp = exp;
- set->set_oi = oinfo;
- set->set_oi->oi_md = lsm;
- set->set_lockh = lov_handle2llh(lockh);
- if (set->set_lockh == NULL) {
- CERROR("LOV: invalid lov lock handle %p\n", lockh);
- GOTO(out_set, rc = -EINVAL);
- }
- lockh->cookie = set->set_lockh->llh_handle.h_cookie;
-
- for (i = 0; i < lsm->lsm_stripe_count; i++){
- struct lov_request *req;
- struct lustre_handle *lov_lockhp;
- struct lov_oinfo *loi = lsm->lsm_oinfo[i];
-
- lov_lockhp = set->set_lockh->llh_handles + i;
- if (!lustre_handle_is_used(lov_lockhp)) {
- CDEBUG(D_INFO, "lov idx %d subobj "DOSTID" no lock\n",
- loi->loi_ost_idx, POSTID(&loi->loi_oi));
- continue;
- }
-
- OBD_ALLOC(req, sizeof(*req));
- if (req == NULL)
- GOTO(out_set, rc = -ENOMEM);
-
- req->rq_buflen = sizeof(*req->rq_oi.oi_md);
- OBD_ALLOC_LARGE(req->rq_oi.oi_md, req->rq_buflen);
- if (req->rq_oi.oi_md == NULL) {
- OBD_FREE(req, sizeof(*req));
- GOTO(out_set, rc = -ENOMEM);
- }
-
- req->rq_idx = loi->loi_ost_idx;
- req->rq_stripe = i;
-
- /* XXX LOV STACKING: submd should be from the subobj */
- req->rq_oi.oi_md->lsm_oi = loi->loi_oi;
- req->rq_oi.oi_md->lsm_stripe_count = 0;
-
- lov_set_add_req(req, set);
- }
- if (!set->set_count)
- GOTO(out_set, rc = -EIO);
- *reqset = set;
- RETURN(rc);
-out_set:
- lov_fini_cancel_set(set);
- RETURN(rc);
-}