oss capability fully works now.

add setattr capability check for oss (including req from mds).
code cleanup.

diff --git a/lustre/cmm/cmm_device.c b/lustre/cmm/cmm_device.c
index 43fe651..da54150 100644 (file)
--- a/lustre/cmm/cmm_device.c
+++ b/lustre/cmm/cmm_device.c
@@ -87,7 +87,7 @@ static int cmm_maxsize_get(const struct lu_env *env, struct md_device *md,
 }
 
 static int cmm_init_capa_ctxt(const struct lu_env *env, struct md_device *md,
-                              __u32 valid, unsigned long timeout, __u32 alg,
+                              int mode , unsigned long timeout, __u32 alg,
                               struct lustre_capa_key *keys)
 {
         struct cmm_device *cmm_dev = md2cmm_dev(md);
@@ -95,7 +95,7 @@ static int cmm_init_capa_ctxt(const struct lu_env *env, struct md_device *md,
         ENTRY;
         LASSERT(cmm_child_ops(cmm_dev)->mdo_init_capa_ctxt);
         rc = cmm_child_ops(cmm_dev)->mdo_init_capa_ctxt(env, cmm_dev->cmm_child,
-                                                        valid, timeout, alg,
+                                                        mode, timeout, alg,
                                                         keys);
         RETURN(rc);
 }

diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index 95dd462..4c3ae54 100644 (file)
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -100,7 +100,7 @@ struct dt_device_operations {
          */
         int   (*dt_init_capa_ctxt)(const struct lu_env *env,
                                    struct dt_device *dev,
-                                   __u32 valid, unsigned long timeout,
+                                   int mode, unsigned long timeout,
                                    __u32 alg, struct lustre_capa_key *keys);
 
         /*
@@ -252,8 +252,8 @@ struct dt_object_operations {
         int   (*do_readpage)(const struct lu_env *env,
                              struct dt_object *dt, const struct lu_rdpg *rdpg,
                              struct lustre_capa *capa);
-        int   (*do_capa_get)(const struct lu_env *env,
-                             struct dt_object *dt, struct lustre_capa *capa);
+        struct obd_capa *(*do_capa_get)(const struct lu_env *env,
+                                        struct dt_object *dt, __u64 opc);
 };
 
 /*

diff --git a/lustre/include/lustre_capa.h b/lustre/include/lustre_capa.h
index 3ec4ba1..329cebc 100644 (file)
--- a/lustre/include/lustre_capa.h
+++ b/lustre/include/lustre_capa.h
@@ -152,7 +152,7 @@ extern spinlock_t capa_lock;
 extern int capa_count[];
 extern cfs_mem_cache_t *capa_cachep;
 
-void capa_add(struct lustre_capa *capa);
+struct obd_capa *capa_add(struct lustre_capa *capa);
 struct obd_capa *capa_lookup(struct lustre_capa *capa);
 
 int capa_hmac(__u8 *hmac, struct lustre_capa *capa, __u8 *key);
@@ -342,11 +342,4 @@ struct filter_capa_key {
 };
 
 #define BYPASS_CAPA (struct lustre_capa *)ERR_PTR(-ENOENT)
-
-enum {
-        CAPA_CTX_ON           = 1,
-        CAPA_CTX_TIMEOUT      = 1<<1,
-        CAPA_CTX_ALG          = 1<<2,
-        CAPA_CTX_KEYS         = 1<<3,
-};
 #endif /* __LINUX_CAPA_H_ */

diff --git a/lustre/include/lustre_mds.h b/lustre/include/lustre_mds.h
index 5730b3a..c0e20dc 100644 (file)
--- a/lustre/include/lustre_mds.h
+++ b/lustre/include/lustre_mds.h
@@ -81,7 +81,8 @@ int mds_reint_rec(struct mds_update_record *r, int offset,
 
 int mds_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
                           struct lov_mds_md *lmm, int lmm_size,
-                          struct llog_cookie *logcookies, __u64 id, __u32 gen);
+                          struct llog_cookie *logcookies, __u64 id, __u32 gen,
+                          struct obd_capa *oc);
 
 int mds_log_op_unlink(struct obd_device *obd,
                       struct lov_mds_md *lmm, int lmm_size,

diff --git a/lustre/include/md_object.h b/lustre/include/md_object.h
index a138925..4cf1906 100644 (file)
--- a/lustre/include/md_object.h
+++ b/lustre/include/md_object.h
@@ -244,7 +244,7 @@ struct md_device_operations {
                           struct kstatfs *sfs);
 
         int (*mdo_init_capa_ctxt)(const struct lu_env *env, struct md_device *m,
-                                  __u32 valid, unsigned long timeout, __u32 alg,
+                                  int mode, unsigned long timeout, __u32 alg,
                                   struct lustre_capa_key *keys);
 
         int (*mdo_update_capa_key)(const struct lu_env *env,

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 072e819..84c9452 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -1042,11 +1042,12 @@ struct obd_ops {
                          struct lov_stripe_md *mem_tgt);
         int (*o_preallocate)(struct lustre_handle *, obd_count *req,
                              obd_id *ids);
+        /* FIXME: add fid capability support for create & destroy! */
         int (*o_create)(struct obd_export *exp,  struct obdo *oa,
                         struct lov_stripe_md **ea, struct obd_trans_info *oti);
         int (*o_destroy)(struct obd_export *exp, struct obdo *oa,
                          struct lov_stripe_md *ea, struct obd_trans_info *oti,
-                         struct obd_export *md_exp, void *capa);
+                         struct obd_export *md_exp);
         int (*o_setattr)(struct obd_export *exp, struct obd_info *oinfo,
                          struct obd_trans_info *oti);
         int (*o_setattr_async)(struct obd_export *exp, struct obd_info *oinfo,

diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index d4ca3d1..8654050 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -602,7 +602,7 @@ static inline int obd_create(struct obd_export *exp, struct obdo *obdo,
 static inline int obd_destroy(struct obd_export *exp, struct obdo *obdo,
                               struct lov_stripe_md *ea,
                               struct obd_trans_info *oti,
-                              struct obd_export *md_exp, void *capa)
+                              struct obd_export *md_exp)
 {
         int rc;
         ENTRY;
@@ -610,7 +610,7 @@ static inline int obd_destroy(struct obd_export *exp, struct obdo *obdo,
         EXP_CHECK_DT_OP(exp, destroy);
         OBD_COUNTER_INCREMENT(exp->exp_obd, destroy);
 
-        rc = OBP(exp->exp_obd, destroy)(exp, obdo, ea, oti, md_exp, capa);
+        rc = OBP(exp->exp_obd, destroy)(exp, obdo, ea, oti, md_exp);
         RETURN(rc);
 }
 

diff --git a/lustre/liblustre/file.c b/lustre/liblustre/file.c
index fea369a..7055784 100644 (file)
--- a/lustre/liblustre/file.c
+++ b/lustre/liblustre/file.c
@@ -302,7 +302,7 @@ int llu_objects_destroy(struct ptlrpc_request *request, struct inode *dir)
                 }
         }
 
-        rc = obd_destroy(llu_i2obdexp(dir), oa, lsm, &oti, NULL, NULL);
+        rc = obd_destroy(llu_i2obdexp(dir), oa, lsm, &oti, NULL);
         obdo_free(oa);
         if (rc)
                 CERROR("obd destroy objid 0x"LPX64" error %d\n",

diff --git a/lustre/llite/llite_capa.c b/lustre/llite/llite_capa.c
index 7a9f126..729039e 100644 (file)
--- a/lustre/llite/llite_capa.c
+++ b/lustre/llite/llite_capa.c
@@ -325,10 +325,11 @@ struct obd_capa *ll_lookup_oss_capa(struct inode *inode, __u64 opc)
                 if (!obd_capa_is_valid(ocapa))
                         continue;
                 if ((opc & CAPA_OPC_OSS_WRITE) &&
-                    capa_opc_supported(&ocapa->c_capa, opc)) {
+                    capa_opc_supported(&ocapa->c_capa, CAPA_OPC_OSS_WRITE)) {
                         found = 1; break;
                 } else if ((opc & CAPA_OPC_OSS_READ) &&
-                           capa_opc_supported(&ocapa->c_capa, opc)) {
+                           capa_opc_supported(&ocapa->c_capa,
+                                              CAPA_OPC_OSS_READ)) {
                         found = 1; break;
                 } else if ((opc & CAPA_OPC_OSS_TRUNC) &&
                            capa_opc_supported(&ocapa->c_capa, opc)) {

diff --git a/lustre/llite/llite_lib.c b/lustre/llite/llite_lib.c
index 92757a4..ded9328 100644 (file)
--- a/lustre/llite/llite_lib.c
+++ b/lustre/llite/llite_lib.c
@@ -1532,9 +1532,11 @@ int ll_setattr_raw(struct inode *inode, struct iattr *attr)
 
                         oinfo.oi_oa = oa;
                         oinfo.oi_md = lsm;
+                        oinfo.oi_capa = ll_i2mdscapa(inode);
 
                         /* XXX: this looks unnecessary now. */
                         rc = obd_setattr_rqset(sbi->ll_dt_exp, &oinfo, NULL);
+                        capa_put(oinfo.oi_capa);
                         if (rc)
                                 CERROR("obd_setattr_async fails: rc=%d\n", rc);
                         obdo_free(oa);

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index f189c30..f460195 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -1037,7 +1037,6 @@ int ll_objects_destroy(struct ptlrpc_request *request, struct inode *dir)
         struct lov_stripe_md *lsm = NULL;
         struct obd_trans_info oti = { 0 };
         struct obdo *oa;
-        struct obd_capa *oc;
         int rc;
         ENTRY;
 
@@ -1096,10 +1095,7 @@ int ll_objects_destroy(struct ptlrpc_request *request, struct inode *dir)
                 }
         }
 
-        /* FIXME: parent mds capability is the only one can find! */
-        oc = ll_i2mdscapa(dir);
-        rc = obd_destroy(ll_i2dtexp(dir), oa, lsm, &oti, ll_i2mdexp(dir), oc);
-        capa_put(oc);
+        rc = obd_destroy(ll_i2dtexp(dir), oa, lsm, &oti, ll_i2mdexp(dir));
         obdo_free(oa);
         if (rc)
                 CERROR("obd destroy objid "LPX64" error %d\n",

diff --git a/lustre/lov/lov_obd.c b/lustre/lov/lov_obd.c
index 24d88ad..2eafe1c 100644 (file)
--- a/lustre/lov/lov_obd.c
+++ b/lustre/lov/lov_obd.c
@@ -1031,7 +1031,7 @@ do {
 
 static int lov_destroy(struct obd_export *exp, struct obdo *oa,
                        struct lov_stripe_md *lsm, struct obd_trans_info *oti,
-                       struct obd_export *md_exp, void *capa)
+                       struct obd_export *md_exp)
 {
         struct lov_request_set *set;
         struct obd_info oinfo;
@@ -1064,7 +1064,7 @@ static int lov_destroy(struct obd_export *exp, struct obdo *oa,
                         oti->oti_logcookies = set->set_cookies + req->rq_stripe;
 
                 err = obd_destroy(lov->lov_tgts[req->rq_idx]->ltd_exp,
-                                  req->rq_oi.oi_oa, NULL, oti, NULL, capa);
+                                  req->rq_oi.oi_oa, NULL, oti, NULL);
                 err = lov_update_common_set(set, req, err);
                 if (err) {
                         CERROR("error: destroying objid "LPX64" subobj "

diff --git a/lustre/lov/lov_qos.c b/lustre/lov/lov_qos.c
index 5d8a13b..2bc909c 100644 (file)
--- a/lustre/lov/lov_qos.c
+++ b/lustre/lov/lov_qos.c
@@ -822,7 +822,6 @@ int qos_prep_create(struct obd_export *exp, struct lov_request_set *set)
                         CDEBUG(D_INODE, "stripe %d has size "LPU64"/"LPU64"\n",
                                i, req->rq_oi.oi_oa->o_size, src_oa->o_size);
                 }
-
         }
         LASSERT(set->set_count == stripes);
 

diff --git a/lustre/lov/lov_request.c b/lustre/lov/lov_request.c
index 6538c45..95975d6 100644 (file)
--- a/lustre/lov/lov_request.c
+++ b/lustre/lov/lov_request.c
@@ -615,8 +615,7 @@ cleanup:
                         continue;
 
                 sub_exp = lov->lov_tgts[req->rq_idx]->ltd_exp;
-                err = obd_destroy(sub_exp, req->rq_oi.oi_oa, NULL, oti, NULL,
-                                  NULL);
+                err = obd_destroy(sub_exp, req->rq_oi.oi_oa, NULL, oti, NULL);
                 if (err)
                         CERROR("Failed to uncreate objid "LPX64" subobj "
                                LPX64" on OST idx %d: rc = %d\n",
@@ -905,6 +904,8 @@ int lov_prep_brw_set(struct obd_export *exp, struct obd_info *oinfo,
                 /* remember the index for sort brw_page array */
                 info[i].index = req->rq_pgaidx;
 
+                req->rq_oi.oi_capa = oinfo->oi_capa;
+
                 lov_set_add_req(req, set);
         }
         if (!set->set_count)
@@ -1002,6 +1003,7 @@ int lov_prep_getattr_set(struct obd_export *exp, struct obd_info *oinfo,
                        sizeof(*req->rq_oi.oi_oa));
                 req->rq_oi.oi_oa->o_id = loi->loi_id;
                 req->rq_oi.oi_cb_up = cb_getattr_update;
+                req->rq_oi.oi_capa = oinfo->oi_capa;
                 req->rq_rqset = set;
 
                 lov_set_add_req(req, set);
@@ -1196,6 +1198,7 @@ int lov_prep_setattr_set(struct obd_export *exp, struct obd_info *oinfo,
                                 || req->rq_oi.oi_oa->o_gr>0);
                 req->rq_oi.oi_oa->o_stripe_idx = i;
                 req->rq_oi.oi_cb_up = cb_setattr_update;
+                req->rq_oi.oi_capa = oinfo->oi_capa;
                 req->rq_rqset = set;
 
                 if (oinfo->oi_oa->o_valid & OBD_MD_FLSIZE) {
@@ -1310,6 +1313,8 @@ int lov_prep_punch_set(struct obd_export *exp, struct obd_info *oinfo,
                 req->rq_oi.oi_policy.l_extent.end = re;
                 req->rq_oi.oi_policy.l_extent.gid = -1;
 
+                req->rq_oi.oi_capa = oinfo->oi_capa;
+
                 lov_set_add_req(req, set);
         }
         if (!set->set_count)
@@ -1395,6 +1400,8 @@ int lov_prep_sync_set(struct obd_export *exp, struct obd_info *oinfo,
                 req->rq_oi.oi_policy.l_extent.end = re;
                 req->rq_oi.oi_policy.l_extent.gid = -1;
 
+                req->rq_oi.oi_capa = oinfo->oi_capa;
+
                 lov_set_add_req(req, set);
         }
         if (!set->set_count)

diff --git a/lustre/mdc/mdc_reint.c b/lustre/mdc/mdc_reint.c
index d5ae1e9..75f8d5a 100644 (file)
--- a/lustre/mdc/mdc_reint.c
+++ b/lustre/mdc/mdc_reint.c
@@ -81,8 +81,8 @@ int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
 
         LASSERT(op_data != NULL);
 
-        if (op_data->mod_capa1)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = op_data->mod_capa1 ?
+                                        sizeof(struct lustre_capa) : 0;
 
         if (op_data->flags & (MF_SOM_CHANGE | MF_EPOCH_OPEN))
                 size[REQ_REC_OFF + 2] = sizeof(struct mdt_epoch);
@@ -135,9 +135,8 @@ int mdc_create(struct obd_export *exp, struct md_op_data *op_data,
         struct ptlrpc_request *req;
         ENTRY;
 
-        if (op_data->mod_capa1)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
-
+        size[REQ_REC_OFF + 1] = op_data->mod_capa1 ?
+                                        sizeof(struct lustre_capa) : 0;
         if (data && datalen) {
                 size[bufcount] = datalen;
                 bufcount++;
@@ -185,8 +184,8 @@ int mdc_unlink(struct obd_export *exp, struct md_op_data *op_data,
 
         LASSERT(req == NULL);
 
-        if (op_data->mod_capa1)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = op_data->mod_capa1 ?
+                                        sizeof(struct lustre_capa) : 0;
 
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_REINT, 4, size, NULL);
@@ -218,10 +217,10 @@ int mdc_link(struct obd_export *exp, struct md_op_data *op_data,
         int rc;
         ENTRY;
 
-        if (op_data->mod_capa1)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
-        if (op_data->mod_capa2)
-                size[REQ_REC_OFF + 2] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = op_data->mod_capa1 ?
+                                        sizeof(struct lustre_capa) : 0;
+        size[REQ_REC_OFF + 2] = op_data->mod_capa2 ?
+                                        sizeof(struct lustre_capa) : 0;
 
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_REINT, 5, size, NULL);
@@ -253,10 +252,10 @@ int mdc_rename(struct obd_export *exp, struct md_op_data *op_data,
         int rc;
         ENTRY;
 
-        if (op_data->mod_capa1)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
-        if (op_data->mod_capa2)
-                size[REQ_REC_OFF + 2] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = op_data->mod_capa1 ?
+                                        sizeof(struct lustre_capa) : 0;
+        size[REQ_REC_OFF + 2] = op_data->mod_capa2 ?
+                                        sizeof(struct lustre_capa) : 0;
 
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_REINT, 6, size, NULL);

diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c
index 2175b72..b01a7ec 100644 (file)
--- a/lustre/mdc/mdc_request.c
+++ b/lustre/mdc/mdc_request.c
@@ -53,7 +53,7 @@ static int mdc_cleanup(struct obd_device *obd);
 
 extern int mds_queue_req(struct ptlrpc_request *);
 
-static inline struct obd_capa *mdc_unpack_capa(struct ptlrpc_request *req,
+static struct obd_capa *mdc_unpack_capa(struct ptlrpc_request *req,
                                                unsigned int offset)
 {
         struct lustre_capa *capa;
@@ -169,8 +169,7 @@ static int mdc_getattr_common(struct obd_export *exp, unsigned int ea_size,
                        ea_size);
         if (acl_size)
                 CDEBUG(D_INODE, "reserved %u bytes for ACL\n", acl_size);
-        if (mdscapa)
-                size[REPLY_REC_OFF + 2] = sizeof(struct lustre_capa);
+        size[REPLY_REC_OFF + 2] = mdscapa ? sizeof(struct lustre_capa) : 0;
 
         ptlrpc_req_set_repsize(req, 5, size);
 
@@ -233,8 +232,7 @@ int mdc_getattr(struct obd_export *exp, const struct lu_fid *fid,
         int acl_size = 0, rc;
         ENTRY;
 
-        if (oc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
 
         /*
          * XXX do we need to make another request here?  We just did a getattr
@@ -276,8 +274,7 @@ int mdc_getattr_name(struct obd_export *exp, const struct lu_fid *fid,
         int rc;
         ENTRY;
 
-        if (oc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
 
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_GETATTR_NAME, 4, size, NULL);
@@ -314,10 +311,8 @@ int mdc_is_subdir(struct obd_export *exp, const struct lu_fid *pfid,
         int rc;
         ENTRY;
 
-        if (pc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
-        if (cc)
-                size[REQ_REC_OFF + 2] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = pc ? sizeof(struct lustre_capa) : 0;
+        size[REQ_REC_OFF + 2] = cc ? sizeof(struct lustre_capa) : 0;
 
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_IS_SUBDIR, 4, size, NULL);
@@ -357,8 +352,7 @@ int mdc_xattr_common(struct obd_export *exp, const struct lu_fid *fid,
         void *tmp;
         ENTRY;
 
-        if (oc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
         if (xattr_name) {
                 xattr_namelen = strlen(xattr_name) + 1;
                 size[bufcnt++] = xattr_namelen;
@@ -797,8 +791,8 @@ int mdc_close(struct obd_export *exp, struct md_op_data *op_data,
         int rc;
         ENTRY;
 
-        if (op_data->mod_capa1)
-                reqsize[REQ_REC_OFF + 2] = sizeof(struct lustre_capa);
+        reqsize[REQ_REC_OFF + 2] = op_data->mod_capa1 ?
+                                        sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_CLOSE, 4, reqsize, NULL);
         if (req == NULL)
@@ -953,11 +947,9 @@ int mdc_readpage(struct obd_export *exp, const struct lu_fid *fid,
 
         CDEBUG(D_INODE, "object: "DFID"\n", PFID(fid));
 
-        if (oc != NULL)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
-        
-        req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_READPAGE, 3,
-                              size, NULL);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
+        req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, MDS_READPAGE, 3, size,
+                              NULL);
         if (req == NULL)
                 GOTO(out, rc = -ENOMEM);
 
@@ -1208,8 +1200,7 @@ static int mdc_pin(struct obd_export *exp, const struct lu_fid *fid,
         int rc, size[3] = { sizeof(struct ptlrpc_body), sizeof(*body) };
         ENTRY;
 
-        if (oc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_PIN, 3, size, NULL);
         if (req == NULL)
@@ -1286,16 +1277,14 @@ static int mdc_unpin(struct obd_export *exp,
 }
 
 int mdc_sync(struct obd_export *exp, const struct lu_fid *fid,
-             struct obd_capa *oc,
-             struct ptlrpc_request **request)
+             struct obd_capa *oc, struct ptlrpc_request **request)
 {
         struct ptlrpc_request *req;
         int size[3] = { sizeof(struct ptlrpc_body), sizeof(struct mdt_body) };
         int rc;
         ENTRY;
 
-        if (oc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,
                               MDS_SYNC, 3, size, NULL);
         if (!req)
@@ -1583,8 +1572,7 @@ int mdc_get_remote_perm(struct obd_export *exp, const struct lu_fid *fid,
         int rc;
         ENTRY;
 
-        if (oc)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
+        size[REQ_REC_OFF + 1] = oc ? sizeof(struct lustre_capa) : 0;
 
         *request = NULL;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_MDS_VERSION,

diff --git a/lustre/mdd/mdd_handler.c b/lustre/mdd/mdd_handler.c
index 7ca2aff..5c2fdb9 100644 (file)
--- a/lustre/mdd/mdd_handler.c
+++ b/lustre/mdd/mdd_handler.c
@@ -2830,7 +2830,7 @@ static int mdd_maxsize_get(const struct lu_env *env, struct md_device *m,
 }
 
 static int mdd_init_capa_ctxt(const struct lu_env *env, struct md_device *m,
-                              __u32 valid, unsigned long timeout, __u32 alg,
+                              int mode, unsigned long timeout, __u32 alg,
                               struct lustre_capa_key *keys)
 {
        struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
@@ -2838,10 +2838,8 @@ static int mdd_init_capa_ctxt(const struct lu_env *env, struct md_device *m,
         int rc;
         ENTRY;
 
-        if (valid & CAPA_CTX_KEYS)
-                mds->mds_capa_keys = keys;
-
-        rc = mdd_child_ops(mdd)->dt_init_capa_ctxt(env, mdd->mdd_child, valid,
+        mds->mds_capa_keys = keys;
+        rc = mdd_child_ops(mdd)->dt_init_capa_ctxt(env, mdd->mdd_child, mode,
                                                    timeout, alg, keys);
         RETURN(rc);
 }
@@ -3334,15 +3332,19 @@ static int mdd_capa_get(const struct lu_env *env, struct md_object *obj,
 {
         struct dt_object *next;
         struct mdd_object *mdd_obj = md2mdd_obj(obj);
-        int rc;
+        struct obd_capa *oc;
         ENTRY;
 
         LASSERT(lu_object_exists(mdd2lu_obj(mdd_obj)));
         next = mdd_object_child(mdd_obj);
 
-        rc = next->do_ops->do_capa_get(env, next, capa);
+        oc = next->do_ops->do_capa_get(env, next, capa->lc_opc);
+        if (oc) {
+                capa_cpy(capa, oc);
+                capa_put(oc);
+        }
 
-        RETURN(rc);
+        RETURN(0);
 }
 
 struct md_device_operations mdd_ops = {

diff --git a/lustre/mdd/mdd_lov.c b/lustre/mdd/mdd_lov.c
index b2456fa..f0fc2b4 100644 (file)
--- a/lustre/mdd/mdd_lov.c
+++ b/lustre/mdd/mdd_lov.c
@@ -392,6 +392,7 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
         __u32                  create_flags = spec->sp_cr_flags;
         int                    rc = 0;
         struct obd_trans_info *oti = &mdd_env_info(env)->mti_oti;
+        struct dt_object      *next;
         ENTRY;
 
         if (create_flags & MDS_OPEN_DELAY_CREATE ||
@@ -426,6 +427,7 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                 OBD_MD_FLMODE | OBD_MD_FLUID | OBD_MD_FLGID | OBD_MD_FLGROUP;
         oa->o_size = 0;
 
+        next = mdd_object_child(child);
         if (!(create_flags & MDS_OPEN_HAS_OBJS)) {
                 if (create_flags & MDS_OPEN_HAS_EA) {
                         LASSERT(eadata != NULL);
@@ -457,6 +459,7 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                         if (rc)
                                 GOTO(out_oa, rc);
                 }
+
                 rc = obd_create(lov_exp, oa, &lsm, oti);
                 if (rc) {
                         if (rc > 0) {
@@ -504,8 +507,11 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd,
                 oa->o_valid |= OBD_MD_FLFID | OBD_MD_FLGENER;
                 oinfo->oi_oa = oa;
                 oinfo->oi_md = lsm;
+                oinfo->oi_capa = next->do_ops->do_capa_get(env, next,
+                                                          CAPA_OPC_MDS_DEFAULT);
 
                 rc = obd_setattr(lov_exp, oinfo, oti);
+                capa_put(oinfo->oi_capa);
                 if (rc) {
                         CERROR("Error setting attrs for "DFID": rc %d\n",
                                PFID(mdo2fid(child)), rc);
@@ -565,6 +571,7 @@ int mdd_lov_setattr_async(const struct lu_env *env, struct mdd_object *obj,
         struct dt_object        *next = mdd_object_child(obj);
         __u32  seq  = lu_object_fid(mdd2lu_obj(obj))->f_seq;
         __u32  oid  = lu_object_fid(mdd2lu_obj(obj))->f_oid;
+        struct obd_capa *oc;
         int rc = 0;
         ENTRY;
 
@@ -573,8 +580,10 @@ int mdd_lov_setattr_async(const struct lu_env *env, struct mdd_object *obj,
         if (rc)
                 RETURN(rc);
 
+        oc = next->do_ops->do_capa_get(env, next, CAPA_OPC_MDS_DEFAULT);
         rc = mds_osc_setattr_async(obd, tmp_la->la_uid, tmp_la->la_gid, lmm,
-                                   lmm_size, NULL, seq, oid);
+                                   lmm_size, NULL, seq, oid, oc);
+        capa_put(oc);
 
         RETURN(rc);
 }

diff --git a/lustre/mds/mds_fs.c b/lustre/mds/mds_fs.c
index 1442802..497de4c 100644 (file)
--- a/lustre/mds/mds_fs.c
+++ b/lustre/mds/mds_fs.c
@@ -737,7 +737,7 @@ out_pop:
 
 int mds_obd_destroy(struct obd_export *exp, struct obdo *oa,
                     struct lov_stripe_md *ea, struct obd_trans_info *oti,
-                    struct obd_export *md_exp, void *unused)
+                    struct obd_export *md_exp)
 {
         struct mds_obd *mds = &exp->exp_obd->u.mds;
         struct inode *parent_inode = mds->mds_objects_dir->d_inode;

diff --git a/lustre/mds/mds_internal.h b/lustre/mds/mds_internal.h
index 172b40f..fd63280 100644 (file)
--- a/lustre/mds/mds_internal.h
+++ b/lustre/mds/mds_internal.h
@@ -223,7 +223,7 @@ int mds_obd_create(struct obd_export *exp, struct obdo *oa,
                    struct lov_stripe_md **ea, struct obd_trans_info *oti);
 int mds_obd_destroy(struct obd_export *exp, struct obdo *oa,
                     struct lov_stripe_md *ea, struct obd_trans_info *oti,
-                    struct obd_export *md_exp, void *unused);
+                    struct obd_export *md_exp);
 void mds_init_ctxt(struct obd_device *obd, struct vfsmount *mnt);
 
 /* mds/handler.c */

diff --git a/lustre/mds/mds_reint.c b/lustre/mds/mds_reint.c
index 6d13fbe..1fede7a 100644 (file)
--- a/lustre/mds/mds_reint.c
+++ b/lustre/mds/mds_reint.c
@@ -412,7 +412,8 @@ static void reconstruct_reint_setattr(struct mds_update_record *rec,
 
 int mds_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
                           struct lov_mds_md *lmm, int lmm_size,
-                          struct llog_cookie *logcookies, __u64 id, __u32 gen)
+                          struct llog_cookie *logcookies, __u64 id, __u32 gen,
+                          struct obd_capa *oc)
 {
         struct mds_obd *mds = &obd->u.mds;
         struct obd_trans_info oti = { 0 };
@@ -455,6 +456,7 @@ int mds_osc_setattr_async(struct obd_device *obd, __u32 uid, __u32 gid,
         oinfo.oi_oa->o_fid = id;
         oinfo.oi_oa->o_generation = gen;
         oinfo.oi_oa->o_valid |= OBD_MD_FLFID | OBD_MD_FLGENER;
+        oinfo.oi_capa = oc;
 
         /* do async setattr from mds to ost not waiting for responses. */
         rc = obd_setattr_async(mds->mds_osc_exp, &oinfo, &oti, NULL);
@@ -669,7 +671,7 @@ static int mds_reint_setattr(struct mds_update_record *rec, int offset,
         if (!rc && !err && lmm_size)
                 mds_osc_setattr_async(obd, inode->i_ino, inode->i_generation, lmm, 
                                       lmm_size, logcookies, rec->ur_fid1->id,
-                                      rec->ur_fid1->generation);
+                                      rec->ur_fid1->generation, NULL);
 
         switch (cleanup_phase) {
         case 2:

diff --git a/lustre/mds/mds_unlink_open.c b/lustre/mds/mds_unlink_open.c
index 09e5321..bc68bc3 100644 (file)
--- a/lustre/mds/mds_unlink_open.c
+++ b/lustre/mds/mds_unlink_open.c
@@ -86,8 +86,7 @@ static int mds_osc_destroy_orphan(struct obd_device *obd,
                 oa->o_valid |= OBD_MD_FLCOOKIE;
                 oti.oti_logcookies = logcookies;
         }
-        rc = obd_destroy(mds->mds_osc_exp, oa, lsm, &oti, obd->obd_self_export,
-                         NULL);
+        rc = obd_destroy(mds->mds_osc_exp, oa, lsm, &oti, obd->obd_self_export);
         obdo_free(oa);
         if (rc)
                 CDEBUG(D_INODE, "destroy orphan objid 0x"LPX64" on ost error "

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index dc43d2d..efd5394 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -445,6 +445,7 @@ static int mdt_renew_capa(struct mdt_thread_info *info)
         LASSERT(capa);
 
         *capa = *c;
+        capa->lc_expiry = 0;
         rc = mo_capa_get(info->mti_env, mdt_object_child(obj), capa);
 
         body = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
@@ -473,7 +474,8 @@ static int mdt_getattr(struct mdt_thread_info *info)
 
         if (reqbody->valid & OBD_MD_FLOSSCAPA) {
                 rc = mdt_renew_capa(info);
-                GOTO(out, rc);
+                mdt_shrink_reply(info, REPLY_REC_OFF + 1, 0, 0);
+                RETURN(rc);
         }
 
         if (reqbody->valid & OBD_MD_FLRMTPERM) {
@@ -1579,16 +1581,12 @@ static inline void mdt_finish_reply(struct mdt_thread_info *info, int rc)
 static int mdt_init_capa_ctxt(const struct lu_env *env, struct mdt_device *m)
 {
         struct md_device *next = m->mdt_child;
-        __u32 valid = CAPA_CTX_TIMEOUT | CAPA_CTX_ALG | CAPA_CTX_KEYS;
-        int rc;
 
-        if (m->mdt_opts.mo_mds_capa)
-                valid |= CAPA_CTX_ON;
-        rc = next->md_ops->mdo_init_capa_ctxt(env, next, valid,
-                                              m->mdt_capa_timeout,
-                                              m->mdt_capa_alg,
-                                              m->mdt_capa_keys);
-        return rc;
+        return next->md_ops->mdo_init_capa_ctxt(env, next,
+                                                m->mdt_opts.mo_mds_capa,
+                                                m->mdt_capa_timeout,
+                                                m->mdt_capa_alg,
+                                                m->mdt_capa_keys);
 }
 
 /*
@@ -3185,7 +3183,7 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
                 m->mdt_rootsquash_info = NULL;
         }
 
-        next->md_ops->mdo_init_capa_ctxt(env, next, CAPA_CTX_KEYS, 0, 0, NULL);
+        next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL);
         cleanup_capas(CAPA_SITE_SERVER);
         del_timer(&m->mdt_ck_timer);
         mdt_ck_thread_stop(m);

diff --git a/lustre/mdt/mdt_lib.c b/lustre/mdt/mdt_lib.c
index d0464c4..b4495e4 100644 (file)
--- a/lustre/mdt/mdt_lib.c
+++ b/lustre/mdt/mdt_lib.c
@@ -504,7 +504,7 @@ void mdt_shrink_reply(struct mdt_thread_info *info, int offset,
         offset += !!md_size;
         lustre_shrink_reply(req, offset, acl_size, 1);
         offset += !!acl_size;
-        if (mdscapa && !(body->valid & (OBD_MD_FLMDSCAPA | OBD_MD_FLOSSCAPA)))
+        if (mdscapa && !(body->valid & OBD_MD_FLMDSCAPA))
                 lustre_shrink_reply(req, offset, 0, 1);
         offset += mdscapa;
         if (osscapa && !(body->valid & OBD_MD_FLOSSCAPA))

diff --git a/lustre/obdclass/capa.c b/lustre/obdclass/capa.c
index bdc4a76..f814784 100644 (file)
--- a/lustre/obdclass/capa.c
+++ b/lustre/obdclass/capa.c
@@ -174,14 +174,14 @@ static inline void free_capa_lru(struct list_head *head)
 }
 
 /* add or update */
-void capa_add(struct lustre_capa *capa)
+struct obd_capa *capa_add(struct lustre_capa *capa)
 {
         struct hlist_head *head = capa_hash + capa_hashfn(&capa->lc_fid);
         struct obd_capa *ocapa, *old = NULL;
 
         ocapa = alloc_capa(CAPA_SITE_SERVER);
         if (!ocapa)
-                return;
+                return NULL;
 
         spin_lock(&capa_lock);
 
@@ -191,6 +191,7 @@ void capa_add(struct lustre_capa *capa)
                 set_capa_expiry(ocapa);
                 hlist_add_head(&ocapa->u.tgt.c_hash, head);
                 list_add_tail(&ocapa->c_list, &capa_list[CAPA_SITE_SERVER]);
+                capa_get(ocapa);
 
                 if (capa_count[CAPA_SITE_SERVER] > CAPA_HASH_SIZE)
                         free_capa_lru(&capa_list[CAPA_SITE_SERVER]);
@@ -198,21 +199,18 @@ void capa_add(struct lustre_capa *capa)
                 DEBUG_CAPA(D_SEC, &ocapa->c_capa, "new");
                                         
                 spin_unlock(&capa_lock);
-                return;
+                return ocapa;
         }
 
-        spin_lock(&old->c_lock);
-        old->c_capa = *capa;
-        set_capa_expiry(old);
-        spin_unlock(&old->c_lock);
-
         list_move_tail(&old->c_list, &capa_list[CAPA_SITE_SERVER]);
+        capa_get(old);
 
         spin_unlock(&capa_lock);
 
         DEBUG_CAPA(D_SEC, &old->c_capa, "update");
 
         free_capa(ocapa);
+        return old;
 }
 
 struct obd_capa *capa_lookup(struct lustre_capa *capa)

diff --git a/lustre/obdclass/llog_lvfs.c b/lustre/obdclass/llog_lvfs.c
index 34f7707..460cd4b 100644 (file)
--- a/lustre/obdclass/llog_lvfs.c
+++ b/lustre/obdclass/llog_lvfs.c
@@ -694,8 +694,7 @@ static int llog_lvfs_destroy(struct llog_handle *handle)
         if (rc)
                 GOTO(out, rc);
 
-        rc = obd_destroy(handle->lgh_ctxt->loc_exp, oa, NULL, NULL, NULL,
-                         NULL);
+        rc = obd_destroy(handle->lgh_ctxt->loc_exp, oa, NULL, NULL, NULL);
  out:
         obdo_free(oa);
         RETURN(rc);

diff --git a/lustre/obdecho/echo.c b/lustre/obdecho/echo.c
index 5dbe4f3..9dcf202 100644 (file)
--- a/lustre/obdecho/echo.c
+++ b/lustre/obdecho/echo.c
@@ -130,7 +130,7 @@ int echo_create(struct obd_export *exp, struct obdo *oa,
 
 int echo_destroy(struct obd_export *exp, struct obdo *oa,
                  struct lov_stripe_md *ea, struct obd_trans_info *oti,
-                 struct obd_export *md_exp, void *unused)
+                 struct obd_export *md_exp)
 {
         struct obd_device *obd = class_exp2obd(exp);
 

diff --git a/lustre/obdecho/echo_client.c b/lustre/obdecho/echo_client.c
index 375cc95..90c7b3c 100644 (file)
--- a/lustre/obdecho/echo_client.c
+++ b/lustre/obdecho/echo_client.c
@@ -240,7 +240,7 @@ static int echo_create_object(struct obd_device *obd, int on_target,
                         oa->o_id, on_target ? " (undoing create)" : "");
 
                 if (on_target)
-                        obd_destroy(ec->ec_exp, oa, lsm, oti, NULL, NULL);
+                        obd_destroy(ec->ec_exp, oa, lsm, oti, NULL);
 
                 rc = -EEXIST;
                 goto failed;
@@ -1234,7 +1234,7 @@ echo_client_iocontrol(unsigned int cmd, struct obd_export *exp,
                         oa->o_gr = FILTER_GROUP_ECHO;
                         oa->o_valid |= OBD_MD_FLGROUP;
                         rc = obd_destroy(ec->ec_exp, oa, eco->eco_lsm,
-                                         &dummy_oti, NULL, NULL);
+                                         &dummy_oti, NULL);
                         if (rc == 0)
                                 eco->eco_deleted = 1;
                         echo_put_object(eco);

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index 60c5393..740cedd 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -2117,14 +2117,14 @@ struct obd_llogs *filter_grab_llog_for_group(struct obd_device *obd, int group,
         if (export == NULL)
                 RETURN(NULL);
 
-        OBD_ALLOC(fglog, sizeof(*fglog));
+        OBD_ALLOC_PTR(fglog);
         if (fglog == NULL)
                 RETURN(NULL);
         fglog->group = group;
 
-        OBD_ALLOC(fglog->llogs, sizeof(struct obd_llogs));
+        OBD_ALLOC_PTR(fglog->llogs);
         if (fglog->llogs == NULL) {
-                OBD_FREE(fglog, sizeof(*fglog));
+                OBD_FREE_PTR(fglog);
                 RETURN(NULL);
         }
 
@@ -2138,8 +2138,8 @@ struct obd_llogs *filter_grab_llog_for_group(struct obd_device *obd, int group,
 
         rc = llog_cat_initialize(obd, fglog->llogs, 1, NULL);
         if (rc) {
-                OBD_FREE(fglog->llogs, sizeof(*(fglog->llogs)));
-                OBD_FREE(fglog, sizeof(*fglog));
+                OBD_FREE_PTR(fglog->llogs);
+                OBD_FREE_PTR(fglog);
                 RETURN(NULL);
         }
 
@@ -2205,8 +2205,8 @@ static int filter_llog_preclean (struct obd_device *obd)
                 if (rc)
                         CERROR("failed to cleanup llogging subsystem for %u\n",
                                 log->group);
-                OBD_FREE(log->llogs, sizeof(*(log->llogs)));
-                OBD_FREE(log, sizeof(*log));
+                OBD_FREE_PTR(log->llogs);
+                OBD_FREE_PTR(log);
                 spin_lock(&filter->fo_llog_list_lock);
         }
         spin_unlock(&filter->fo_llog_list_lock);
@@ -3068,7 +3068,7 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
                exp->exp_obd->obd_name, oa->o_id + 1, last);
         for (id = last; id > oa->o_id; id--) {
                 doa.o_id = id;
-                rc = filter_destroy(exp, &doa, NULL, NULL, NULL, NULL);
+                rc = filter_destroy(exp, &doa, NULL, NULL, NULL);
                 if (rc && rc != -ENOENT) /* this is pretty fatal... */
                         CEMERG("error destroying precreate objid "LPU64": %d\n",
                                id, rc);
@@ -3440,7 +3440,7 @@ static int filter_create(struct obd_export *exp, struct obdo *oa,
 
 int filter_destroy(struct obd_export *exp, struct obdo *oa,
                    struct lov_stripe_md *md, struct obd_trans_info *oti,
-                   struct obd_export *md_exp, void *capa)
+                   struct obd_export *md_exp)
 {
         unsigned int qcids[MAXQUOTAS] = {0, 0};
         struct obd_device *obd;
@@ -3455,14 +3455,6 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa,
 
         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
 
-#if 0   /* some places don't support capability yet */
-        rc = filter_verify_capa(exp, NULL, obdo_mdsno(oa),
-                                (struct lustre_capa *)capa,
-                                CAPA_OPC_INDEX_LOOKUP);
-        if (rc)
-                RETURN(rc);
-#endif
-
 #if 0
         if (!(oa->o_valid & OBD_MD_FLGROUP))
                 oa->o_gr = 0;

diff --git a/lustre/obdfilter/filter_capa.c b/lustre/obdfilter/filter_capa.c
index cbdca5b..b0bb3c7 100644 (file)
--- a/lustre/obdfilter/filter_capa.c
+++ b/lustre/obdfilter/filter_capa.c
@@ -88,7 +88,7 @@ int filter_verify_capa(struct obd_export *exp, struct lu_fid *fid, __u64 mdsid,
         struct filter_obd *filter = &obd->u.filter;
         struct filter_capa_key *k;
         struct lustre_capa_key key;
-        struct obd_capa *c;
+        struct obd_capa *oc;
         __u8 *hmac;
         int keys_ready = 0, key_found = 0, rc = 0;
         ENTRY;
@@ -104,24 +104,30 @@ int filter_verify_capa(struct obd_export *exp, struct lu_fid *fid, __u64 mdsid,
 
 #warning "enable fid check in filter_verify_capa when fid ready"
 
-        if (!capa_opc_supported(capa, opc)) {
+        if (opc == CAPA_OPC_OSS_READ) {
+                if (!(capa->lc_opc & (CAPA_OPC_OSS_READ | CAPA_OPC_OSS_WRITE)))
+                        rc = -EACCES;
+        } else if (!capa_opc_supported(capa, opc)) {
+                rc = -EACCES;
+        }
+        if (rc) {
                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
-                RETURN(-EACCES);
+                RETURN(rc);
         }
 
-        c = capa_lookup(capa);
-        if (c) {
-                spin_lock(&c->c_lock);
-                if (memcmp(&c->c_capa, capa, sizeof(*capa))) {
+        oc = capa_lookup(capa);
+        if (oc) {
+                spin_lock(&oc->c_lock);
+                if (memcmp(&oc->c_capa, capa, sizeof(*capa))) {
                         DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
                         rc = -EACCES;
-                } else if (capa_is_expired(c)) {
+                } else if (capa_is_expired(oc)) {
                         DEBUG_CAPA(D_ERROR, capa, "expired");
                         rc = -ESTALE;
                 }
-                spin_unlock(&c->c_lock);
+                spin_unlock(&oc->c_lock);
 
-                capa_put(c);
+                capa_put(oc);
                 RETURN(rc);
         }
 
@@ -167,7 +173,8 @@ int filter_verify_capa(struct obd_export *exp, struct lu_fid *fid, __u64 mdsid,
         }
 
         /* store in capa hash */
-        capa_add(capa);
+        oc = capa_add(capa);
+        capa_put(oc);
         RETURN(0);
 }
 

diff --git a/lustre/obdfilter/filter_internal.h b/lustre/obdfilter/filter_internal.h
index 8c4a0de..2ea4d0f 100644 (file)
--- a/lustre/obdfilter/filter_internal.h
+++ b/lustre/obdfilter/filter_internal.h
@@ -107,7 +107,7 @@ int filter_common_setup(struct obd_device *, struct lustre_cfg *lcfg,
                         void *option);
 int filter_destroy(struct obd_export *exp, struct obdo *oa,
                    struct lov_stripe_md *md, struct obd_trans_info *,
-                   struct obd_export *, void *capa);
+                   struct obd_export *);
 int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry,
                             struct obdo *oa, struct obd_trans_info *oti);
 int filter_setattr(struct obd_export *exp, struct obd_info *oinfo,

diff --git a/lustre/obdfilter/filter_io.c b/lustre/obdfilter/filter_io.c
index a6645d4..9aa509d 100644 (file)
--- a/lustre/obdfilter/filter_io.c
+++ b/lustre/obdfilter/filter_io.c
@@ -293,8 +293,12 @@ static int filter_preprw_read(int cmd, struct obd_export *exp, struct obdo *oa,
 
         rc = filter_verify_capa(exp, NULL, obdo_mdsno(oa), capa,
                                 CAPA_OPC_OSS_READ);
-        if (rc)
-                RETURN(rc);
+        if (rc) {
+                rc = filter_verify_capa(exp, NULL, obdo_mdsno(oa), capa,
+                                        CAPA_OPC_OSS_READ);
+                if (rc)
+                        RETURN(rc);
+        }
 
         if (oa && oa->o_valid & OBD_MD_FLGRANT) {
                 spin_lock(&obd->obd_osfs_lock);

diff --git a/lustre/obdfilter/filter_log.c b/lustre/obdfilter/filter_log.c
index 90f4358..d737983 100644 (file)
--- a/lustre/obdfilter/filter_log.c
+++ b/lustre/obdfilter/filter_log.c
@@ -149,7 +149,7 @@ static int filter_recov_log_unlink_cb(struct llog_ctxt *ctxt,
         memcpy(obdo_logcookie(oa), cookie, sizeof(*cookie));
         oid = oa->o_id;
 
-        rc = filter_destroy(exp, oa, NULL, NULL, NULL, NULL);
+        rc = filter_destroy(exp, oa, NULL, NULL, NULL);
         obdo_free(oa);
         if (rc == -ENOENT) {
                 CDEBUG(D_HA, "object already removed, send cookie\n");

diff --git a/lustre/osc/osc_create.c b/lustre/osc/osc_create.c
index 6d7fff7..3b15b58 100644 (file)
--- a/lustre/osc/osc_create.c
+++ b/lustre/osc/osc_create.c
@@ -239,9 +239,9 @@ int osc_create(struct obd_export *exp, struct obdo *oa,
         struct osc_creator *oscc = &exp->exp_obd->u.cli.cl_oscc;
         int try_again = 1, rc = 0;
         ENTRY;
+
         LASSERT(oa);
         LASSERT(ea);
-        
         LASSERT(oa->o_gr > 0);
         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
 

diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index 9928a52..e02419a 100644 (file)
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -209,7 +209,7 @@ static int osc_getattr_async(struct obd_export *exp, struct obd_info *oinfo,
         struct osc_async_args *aa;
         ENTRY;
 
-        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(*oinfo->oi_capa) : 0;
+        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
                               OST_GETATTR, 3, size,NULL);
         if (!req)
@@ -235,7 +235,7 @@ static int osc_getattr(struct obd_export *exp, struct obd_info *oinfo)
         int rc, size[3] = { sizeof(struct ptlrpc_body), sizeof(*body) };
         ENTRY;
 
-        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(*oinfo->oi_capa) : 0;
+        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
                               OST_GETATTR, 3, size, NULL);
         if (!req)
@@ -281,7 +281,7 @@ static int osc_setattr(struct obd_export *exp, struct obd_info *oinfo,
 
         LASSERT(!(oinfo->oi_oa->o_valid & OBD_MD_FLGROUP) || 
                                         oinfo->oi_oa->o_gr > 0);
-        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(*oinfo->oi_capa) : 0;
+        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
                               OST_SETATTR, 3, size, NULL);
         if (!req)
@@ -340,7 +340,7 @@ static int osc_setattr_async(struct obd_export *exp, struct obd_info *oinfo,
         struct osc_async_args *aa;
         ENTRY;
 
-        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(*oinfo->oi_capa) : 0;
+        size[REQ_REC_OFF + 1] = oinfo->oi_capa ? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
                               OST_SETATTR, 3, size, NULL);
         if (!req)
@@ -379,7 +379,7 @@ int osc_real_create(struct obd_export *exp, struct obdo *oa,
         struct ptlrpc_request *req;
         struct ost_body *body;
         struct lov_stripe_md *lsm;
-        int rc, size[3] = { sizeof(struct ptlrpc_body), sizeof(*body) };
+        int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
         ENTRY;
 
         LASSERT(oa);
@@ -392,9 +392,8 @@ int osc_real_create(struct obd_export *exp, struct obdo *oa,
                         RETURN(rc);
         }
 
-        /* FIXME: how to find one OSS WRITE capability? */
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
-                              OST_CREATE, 3, size, NULL);
+                              OST_CREATE, 2, size, NULL);
         if (!req)
                 GOTO(out, rc = -ENOMEM);
 
@@ -487,7 +486,6 @@ static int osc_punch(struct obd_export *exp, struct obd_info *oinfo,
         struct ptlrpc_request *req;
         struct osc_async_args *aa;
         struct ost_body *body;
-        struct lustre_capa *capa = oinfo->oi_capa;
         int size[3] = { sizeof(struct ptlrpc_body), sizeof(*body) };
         ENTRY;
 
@@ -496,7 +494,7 @@ static int osc_punch(struct obd_export *exp, struct obd_info *oinfo,
                 RETURN(-EINVAL);
         }
 
-        size[REQ_REC_OFF + 1] = capa ? sizeof(*capa) : 0;
+        size[REQ_REC_OFF + 1] = oinfo->oi_capa? sizeof(struct lustre_capa) : 0;
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
                               OST_PUNCH, 3, size, NULL);
         if (!req)
@@ -514,15 +512,6 @@ static int osc_punch(struct obd_export *exp, struct obd_info *oinfo,
         body->oa.o_blocks = oinfo->oi_policy.l_extent.end;
         body->oa.o_valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS);
 
-        if (capa) {
-                struct lustre_capa *c;
-
-                c = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 1, sizeof(*c));
-                /* setattr_raw is protected by i_sem, no need to lock here */
-                *c = *capa;
-                body->oa.o_valid |= OBD_MD_FLOSSCAPA;
-        }
-
         ptlrpc_req_set_repsize(req, 2, size);
 
         req->rq_interpret_reply = osc_punch_interpret;
@@ -598,11 +587,11 @@ static int osc_sync(struct obd_export *exp, struct obdo *oa,
  * cookies to the MDS after committing destroy transactions. */
 static int osc_destroy(struct obd_export *exp, struct obdo *oa,
                        struct lov_stripe_md *ea, struct obd_trans_info *oti,
-                       struct obd_export *md_export, void *capa)
+                       struct obd_export *md_export)
 {
         struct ptlrpc_request *req;
         struct ost_body *body;
-        int size[3] = { sizeof(struct ptlrpc_body), sizeof(*body) };
+        int size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
         ENTRY;
 
         if (!oa) {
@@ -610,10 +599,8 @@ static int osc_destroy(struct obd_export *exp, struct obdo *oa,
                 RETURN(-EINVAL);
         }
 
-        if (capa)
-                size[REQ_REC_OFF + 1] = sizeof(struct lustre_capa);
         req = ptlrpc_prep_req(class_exp2cliimp(exp), LUSTRE_OST_VERSION,
-                              OST_DESTROY, 3, size, NULL);
+                              OST_DESTROY, 2, size, NULL);
         if (!req)
                 RETURN(-ENOMEM);
 
@@ -628,8 +615,6 @@ static int osc_destroy(struct obd_export *exp, struct obdo *oa,
                        sizeof(*oti->oti_logcookies));
         body->oa = *oa;
 
-        osc_pack_capa(req, REQ_REC_OFF + 1, body, capa);
-
         ptlrpc_req_set_repsize(req, 2, size);
 
         ptlrpcd_add_req(req);

diff --git a/lustre/osd/osd_handler.c b/lustre/osd/osd_handler.c
index dfc9e1f..5bc2208 100644 (file)
--- a/lustre/osd/osd_handler.c
+++ b/lustre/osd/osd_handler.c
@@ -686,25 +686,16 @@ static void osd_ro(const struct lu_env *env, struct dt_device *d)
 }
 
 static int osd_init_capa_ctxt(const struct lu_env *env, struct dt_device *d,
-                              __u32 valid, unsigned long timeout, __u32 alg,
+                              int mode, unsigned long timeout, __u32 alg,
                               struct lustre_capa_key *keys)
 {
         struct osd_device *dev = osd_dt_dev(d);
         ENTRY;
 
-        if (valid & CAPA_CTX_ON)
-                dev->od_fl_capa = 1;
-        else
-                dev->od_fl_capa = 0;
-       
-        if (valid & CAPA_CTX_TIMEOUT)
-                dev->od_capa_timeout = timeout;
-
-        if (valid & CAPA_CTX_ALG)
-                dev->od_capa_alg = alg;
-
-        if (valid & CAPA_CTX_KEYS)
-                dev->od_capa_keys = keys;
+        dev->od_fl_capa = mode;
+        dev->od_capa_timeout = timeout;
+        dev->od_capa_alg = alg;
+        dev->od_capa_keys = keys;
         RETURN(0);
 }
 
@@ -770,7 +761,7 @@ static struct dt_device_operations osd_dt_ops = {
         .dt_sync           = osd_sync,
         .dt_ro             = osd_ro,
         .dt_credit_get     = osd_credit_get,
-        .dt_init_capa_ctxt = osd_init_capa_ctxt
+        .dt_init_capa_ctxt = osd_init_capa_ctxt,
 };
 
 static void osd_object_read_lock(const struct lu_env *env,
@@ -832,24 +823,23 @@ static int capa_is_sane(const struct lu_env *env,
                         struct lustre_capa *capa,
                         struct lustre_capa_key *keys)
 {
-        struct obd_capa *c;
-        struct osd_thread_info *oti = lu_context_key_get(&env->le_ctx, &osd_key);
+        struct osd_thread_info *oti;
+        struct obd_capa *oc;
         int i, rc = 1;
         ENTRY;
 
-        c = capa_lookup(capa);
-        if (c) {
-                spin_lock(&c->c_lock);
-                if (memcmp(&c->c_capa, capa, sizeof(*capa))) {
+        oti = lu_context_key_get(&env->le_ctx, &osd_key);
+
+        oc = capa_lookup(capa);
+        if (oc) {
+                if (memcmp(&oc->c_capa, capa, sizeof(*capa))) {
                         DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
                         rc = -EACCES;
-                } else if (capa_is_expired(c)) {
+                } else if (capa_is_expired(oc)) {
                         DEBUG_CAPA(D_ERROR, capa, "expired");
                         rc = -ESTALE;
                 }
-                spin_unlock(&c->c_lock);
-
-                capa_put(c);
+                capa_put(oc);
                 RETURN(rc);
         }
 
@@ -875,7 +865,8 @@ static int capa_is_sane(const struct lu_env *env,
                 RETURN(-EACCES);
         }
 
-        capa_add(capa);
+        oc = capa_add(capa);
+        capa_put(oc);
 
         RETURN(1);
 }
@@ -894,7 +885,6 @@ static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
 
         if (!capa) {
                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
-                LBUG();
                 return -EACCES;
         }
 
@@ -1526,8 +1516,9 @@ static int osd_readpage(const struct lu_env *env,
         return rc ? rc : rc1;
 }
 
-static int osd_capa_get(const struct lu_env *env,
-                        struct dt_object *dt, struct lustre_capa *capa)
+static struct obd_capa *osd_capa_get(const struct lu_env *env,
+                                     struct dt_object *dt,
+                                     __u64 opc)
 {
         struct osd_thread_info *info = lu_context_key_get(&env->le_ctx,
                                                           &osd_key);
@@ -1535,40 +1526,44 @@ static int osd_capa_get(const struct lu_env *env,
         struct osd_object *obj = osd_dt_obj(dt);
         struct osd_device *dev = osd_obj2dev(obj);
         struct lustre_capa_key *key = &info->oti_capa_key;
+        struct lustre_capa *capa = &info->oti_capa;
         struct obd_capa *oc;
         int rc;
         ENTRY;
 
+        if (!dev->od_fl_capa)
+                RETURN(NULL);
+
         LASSERT(dt_object_exists(dt));
         LASSERT(osd_invariant(obj));
 
         capa->lc_fid = *fid;
+        capa->lc_opc = opc;
+        capa->lc_flags |= dev->od_capa_alg << 24;
         if (dev->od_capa_timeout < CAPA_TIMEOUT)
                 capa->lc_flags |= CAPA_FL_SHORT_EXPIRY;
 
-        capa->lc_flags = dev->od_capa_alg << 24;
-
-        /* TODO: get right permission here */
         oc = capa_lookup(capa);
         if (oc) {
                 LASSERT(!capa_is_expired(oc));
-                capa_cpy(capa, oc);
-                capa_put(oc);
-                RETURN(0);
+                RETURN(oc);
         }
 
         spin_lock(&capa_lock);
         *key = dev->od_capa_keys[1];
-        capa->lc_expiry = CURRENT_SECONDS + dev->od_capa_timeout;
         spin_unlock(&capa_lock);
 
         capa->lc_keyid = key->lk_keyid;
+        capa->lc_expiry = CURRENT_SECONDS + dev->od_capa_timeout;
+
         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
-        if (rc)
-                RETURN(rc);
+        if (rc) {
+                DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
+                RETURN(NULL);
+        }
 
-        capa_add(capa);
-        RETURN(0);
+        oc = capa_add(capa);
+        RETURN(oc);
 }
 
 static struct dt_object_operations osd_obj_ops = {

diff --git a/lustre/osd/osd_internal.h b/lustre/osd/osd_internal.h
index 9a43fc9..2c9aae9 100644 (file)
--- a/lustre/osd/osd_internal.h
+++ b/lustre/osd/osd_internal.h
@@ -82,6 +82,7 @@ struct osd_thread_info {
          */
         char                   oti_capa_hmac[CAPA_HMAC_KEY_MAX_LEN];
         struct lustre_capa_key oti_capa_key;
+        struct lustre_capa     oti_capa;
 };
 
 #endif /* __KERNEL__ */

diff --git a/lustre/ost/ost_handler.c b/lustre/ost/ost_handler.c
index e55c527..5ffe2d0 100644 (file)
--- a/lustre/ost/ost_handler.c
+++ b/lustre/ost/ost_handler.c
@@ -80,7 +80,6 @@ static int ost_destroy(struct obd_export *exp, struct ptlrpc_request *req,
                        struct obd_trans_info *oti)
 {
         struct ost_body *body, *repbody;
-        struct lustre_capa *capa = NULL;
         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
         ENTRY;
 
@@ -98,9 +97,7 @@ static int ost_destroy(struct obd_export *exp, struct ptlrpc_request *req,
         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
                                  sizeof(*repbody));
         memcpy(&repbody->oa, &body->oa, sizeof(body->oa));
-        if (body->oa.o_valid & OBD_MD_FLOSSCAPA)
-                capa = lustre_unpack_capa(req->rq_reqmsg, REQ_REC_OFF + 1);
-        req->rq_status = obd_destroy(exp, &body->oa, NULL, oti, NULL, capa);
+        req->rq_status = obd_destroy(exp, &body->oa, NULL, oti, NULL);
         RETURN(0);
 }
 
@@ -126,7 +123,7 @@ static int ost_getattr(struct obd_export *exp, struct ptlrpc_request *req)
 
         oinfo.oi_oa = &repbody->oa;
         if (oinfo.oi_oa->o_valid & OBD_MD_FLOSSCAPA)
-                oinfo.oi_capa = lustre_unpack_capa(req->rq_repmsg,
+                oinfo.oi_capa = lustre_unpack_capa(req->rq_reqmsg,
                                                    REQ_REC_OFF + 1);
         req->rq_status = obd_getattr(exp, &oinfo);
         RETURN(0);
@@ -285,7 +282,7 @@ static int ost_punch(struct obd_export *exp, struct ptlrpc_request *req,
                         oinfo.oi_oa->o_valid &= ~OBD_MD_FLFLAGS;
 
                 if (oinfo.oi_oa->o_valid & OBD_MD_FLOSSCAPA)
-                        oinfo.oi_capa = lustre_unpack_capa(req->rq_repmsg,
+                        oinfo.oi_capa = lustre_unpack_capa(req->rq_reqmsg,
                                                            REQ_REC_OFF + 1);
                 req->rq_status = obd_punch(exp, &oinfo, oti, NULL);
                 ost_punch_lock_put(exp, oinfo.oi_oa, &lh);
@@ -343,7 +340,7 @@ static int ost_setattr(struct obd_export *exp, struct ptlrpc_request *req,
 
         oinfo.oi_oa = &repbody->oa;
         if (oinfo.oi_oa->o_valid & OBD_MD_FLOSSCAPA)
-                oinfo.oi_capa = lustre_unpack_capa(req->rq_repmsg,
+                oinfo.oi_capa = lustre_unpack_capa(req->rq_reqmsg,
                                                    REQ_REC_OFF + 1);
         req->rq_status = obd_setattr(exp, &oinfo, oti);
         RETURN(0);

diff --git a/lustre/ptlrpc/pinger.c b/lustre/ptlrpc/pinger.c
index 9b00e1e..cfc9779 100644 (file)
--- a/lustre/ptlrpc/pinger.c
+++ b/lustre/ptlrpc/pinger.c
@@ -261,7 +261,7 @@ int ptlrpc_start_pinger(void)
         if (pinger_thread != NULL)
                 RETURN(-EALREADY);
 
-        OBD_ALLOC(pinger_thread, sizeof(*pinger_thread));
+        OBD_ALLOC_PTR(pinger_thread);
         if (pinger_thread == NULL)
                 RETURN(-ENOMEM);
         cfs_waitq_init(&pinger_thread->t_ctl_waitq);
@@ -303,7 +303,7 @@ int ptlrpc_stop_pinger(void)
         l_wait_event(pinger_thread->t_ctl_waitq,
                      (pinger_thread->t_flags & SVC_STOPPED), &lwi);
 
-        OBD_FREE(pinger_thread, sizeof(*pinger_thread));
+        OBD_FREE_PTR(pinger_thread);
         pinger_thread = NULL;
         RETURN(rc);
 }

diff --git a/lustre/ptlrpc/service.c b/lustre/ptlrpc/service.c
index b35f4a7..d21a54c 100644 (file)
--- a/lustre/ptlrpc/service.c
+++ b/lustre/ptlrpc/service.c
@@ -72,7 +72,7 @@ ptlrpc_alloc_rqbd (struct ptlrpc_service *svc)
 {
         struct ptlrpc_request_buffer_desc *rqbd;
 
-        OBD_ALLOC(rqbd, sizeof (*rqbd));
+        OBD_ALLOC_PTR(rqbd);
         if (rqbd == NULL)
                 return (NULL);
 
@@ -84,7 +84,7 @@ ptlrpc_alloc_rqbd (struct ptlrpc_service *svc)
         rqbd->rqbd_buffer = ptlrpc_alloc_request_buffer(svc->srv_buf_size);
 
         if (rqbd->rqbd_buffer == NULL) {
-                OBD_FREE(rqbd, sizeof (*rqbd));
+                OBD_FREE_PTR(rqbd);
                 return (NULL);
         }
 
@@ -110,7 +110,7 @@ ptlrpc_free_rqbd (struct ptlrpc_request_buffer_desc *rqbd)
         spin_unlock(&svc->srv_lock);
 
         ptlrpc_free_request_buffer (rqbd->rqbd_buffer, svc->srv_buf_size);
-        OBD_FREE (rqbd, sizeof (*rqbd));
+        OBD_FREE_PTR(rqbd);
 }
 
 int
@@ -282,7 +282,7 @@ ptlrpc_init_svc(int nbufs, int bufsize, int max_req_size, int max_reply_size,
         LASSERT (bufsize >= max_req_size + SPTLRPC_MAX_PAYLOAD);
         LASSERT (ctx_tags != 0);
 
-        OBD_ALLOC(service, sizeof(*service));
+        OBD_ALLOC_PTR(service);
         if (service == NULL)
                 RETURN(NULL);
 
@@ -1111,7 +1111,7 @@ static void ptlrpc_stop_thread(struct ptlrpc_service *svc,
         list_del(&thread->t_link);
         spin_unlock(&svc->srv_lock);
 
-        OBD_FREE(thread, sizeof(*thread));
+        OBD_FREE_PTR(thread);
 }
 
 void ptlrpc_stop_all_threads(struct ptlrpc_service *svc)
@@ -1161,7 +1161,7 @@ int ptlrpc_start_thread(struct obd_device *dev, struct ptlrpc_service *svc,
         int rc;
         ENTRY;
 
-        OBD_ALLOC(thread, sizeof(*thread));
+        OBD_ALLOC_PTR(thread);
         if (thread == NULL)
                 RETURN(-ENOMEM);
         cfs_waitq_init(&thread->t_ctl_waitq);
@@ -1317,7 +1317,7 @@ int ptlrpc_unregister_service(struct ptlrpc_service *service)
                 OBD_FREE(rs, service->srv_max_reply_size);
         }
 
-        OBD_FREE(service, sizeof(*service));
+        OBD_FREE_PTR(service);
         return 0;
 }