req-layout: handle 0-length fields: use (-1) as variable field marker

diff --git a/lustre/fid/fid_handler.c b/lustre/fid/fid_handler.c
index d165019..52cb51f 100644 (file)
--- a/lustre/fid/fid_handler.c
+++ b/lustre/fid/fid_handler.c
@@ -247,7 +247,7 @@ static int seq_server_alloc_meta(struct lu_server_seq *seq,
 static int seq_req_handle0(const struct lu_context *ctx,
                            struct ptlrpc_request *req)
 {
-        int rep_buf_size[2] = { 0, };
+        int rep_buf_size[2] = { -1, -1 };
         struct req_capsule pill;
         struct lu_site *site;
         struct lu_range *out;
@@ -258,8 +258,7 @@ static int seq_req_handle0(const struct lu_context *ctx,
         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
         LASSERT(site != NULL);
                        
-        req_capsule_init(&pill, req, RCL_SERVER,
-                         rep_buf_size);
+        req_capsule_init(&pill, req, RCL_SERVER, rep_buf_size);
 
         req_capsule_set(&pill, &RQF_SEQ_QUERY);
         req_capsule_pack(&pill);

diff --git a/lustre/fid/fid_request.c b/lustre/fid/fid_request.c
index 88674e8..c6e2b4e 100644 (file)
--- a/lustre/fid/fid_request.c
+++ b/lustre/fid/fid_request.c
@@ -47,7 +47,7 @@
 #include <lustre_fid.h>
 #include "fid_internal.h"
 
-static int seq_client_rpc(struct lu_client_seq *seq, 
+static int seq_client_rpc(struct lu_client_seq *seq,
                           struct lu_range *range,
                           __u32 opc)
 {
@@ -60,15 +60,14 @@ static int seq_client_rpc(struct lu_client_seq *seq,
         __u32 *op;
         ENTRY;
 
-        req = ptlrpc_prep_req(class_exp2cliimp(exp), 
+        req = ptlrpc_prep_req(class_exp2cliimp(exp),
                              LUSTRE_MDS_VERSION,
                               SEQ_QUERY, 1, &reqsize,
                               NULL);
         if (req == NULL)
                 RETURN(-ENOMEM);
 
-        req_capsule_init(&pill, req, RCL_CLIENT,
-                         &repsize);
+        req_capsule_init(&pill, req, RCL_CLIENT, NULL);
 
         req_capsule_set(&pill, &RQF_SEQ_QUERY);
 
@@ -76,7 +75,7 @@ static int seq_client_rpc(struct lu_client_seq *seq,
         *op = opc;
 
         req->rq_replen = lustre_msg_size(1, &repsize);
-        
+
         req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
                 SEQ_CTLR_PORTAL : SEQ_SRV_PORTAL;
 
@@ -90,14 +89,14 @@ static int seq_client_rpc(struct lu_client_seq *seq,
                 GOTO(out_req, rc = -EPROTO);
         }
         *range = *ran;
-        
+
         LASSERT(range_is_sane(range));
         LASSERT(!range_is_exhausted(range));
-        
+
         EXIT;
 out_req:
         req_capsule_fini(&pill);
-        ptlrpc_req_finished(req); 
+        ptlrpc_req_finished(req);
         return rc;
 }
 
@@ -127,7 +126,7 @@ int seq_client_alloc_super(struct lu_client_seq *seq)
 {
         int rc;
         ENTRY;
-        
+
         down(&seq->seq_sem);
         rc = __seq_client_alloc_super(seq);
         up(&seq->seq_sem);
@@ -174,10 +173,10 @@ static int __seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
                         RETURN(rc);
                 }
         }
-        
+
         *seqnr = seq->seq_range.lr_start;
         seq->seq_range.lr_start++;
-        
+
         CDEBUG(D_INFO, "%s: allocated sequence ["LPX64"]\n",
                seq->seq_name, *seqnr);
         RETURN(rc);
@@ -233,7 +232,7 @@ int seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
 
         *fid = seq->seq_fid;
         LASSERT(fid_is_sane(fid));
-        
+
         CDEBUG(D_INFO, "%s: allocated FID "DFID3"\n",
                seq->seq_name, PFID3(fid));
 
@@ -253,7 +252,7 @@ static int seq_client_proc_init(struct lu_client_seq *seq)
         seq->seq_proc_dir = lprocfs_register(seq->seq_name,
                                              proc_lustre_root,
                                              NULL, NULL);
-        
+
         if (IS_ERR(seq->seq_proc_dir)) {
                 CERROR("LProcFS failed in seq-init\n");
                 rc = PTR_ERR(seq->seq_proc_dir);
@@ -296,7 +295,7 @@ int seq_client_init(struct lu_client_seq *seq,
         ENTRY;
 
         LASSERT(exp != NULL);
-        
+
         fid_zero(&seq->seq_fid);
         range_zero(&seq->seq_range);
         sema_init(&seq->seq_sem, 1);
@@ -326,14 +325,14 @@ void seq_client_fini(struct lu_client_seq *seq)
 #ifdef LPROCFS
         seq_client_proc_fini(seq);
 #endif
-        
+
         if (seq->seq_exp != NULL) {
                 class_export_put(seq->seq_exp);
                 seq->seq_exp = NULL;
         }
-        
+
         CDEBUG(D_INFO|D_WARNING, "Client Sequence Manager\n");
-        
+
         EXIT;
 }
 EXPORT_SYMBOL(seq_client_fini);

diff --git a/lustre/fld/fld_handler.c b/lustre/fld/fld_handler.c
index 9d890b8..b05f74a 100644 (file)
--- a/lustre/fld/fld_handler.c
+++ b/lustre/fld/fld_handler.c
@@ -127,7 +127,7 @@ static int fld_req_handle0(const struct lu_context *ctx,
                            struct lu_server_fld *fld,
                            struct ptlrpc_request *req)
 {
-        int rep_buf_size[3] = { 0, };
+        int rep_buf_size[3] = { -1, -1 };
         struct req_capsule pill;
         struct md_fld *in;
         struct md_fld *out;
@@ -135,8 +135,7 @@ static int fld_req_handle0(const struct lu_context *ctx,
         __u32 *opc;
         ENTRY;
 
-        req_capsule_init(&pill, req, RCL_SERVER,
-                         rep_buf_size);
+        req_capsule_init(&pill, req, RCL_SERVER, rep_buf_size);
 
         req_capsule_set(&pill, &RQF_FLD_QUERY);
         req_capsule_pack(&pill);

diff --git a/lustre/fld/fld_request.c b/lustre/fld/fld_request.c
index b8ef8c1..d9acd29 100644 (file)
--- a/lustre/fld/fld_request.c
+++ b/lustre/fld/fld_request.c
@@ -336,8 +336,7 @@ static int fld_client_rpc(struct obd_export *exp,
         if (req == NULL)
                 RETURN(-ENOMEM);
 
-        req_capsule_init(&pill, req, RCL_CLIENT,
-                         &mf_size);
+        req_capsule_init(&pill, req, RCL_CLIENT, NULL);
 
         req_capsule_set(&pill, &RQF_FLD_QUERY);
 

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 6351a24..bbf9c9f 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -53,7 +53,7 @@
 #include <obd.h>
 /* lu2dt_dev() */
 #include <dt_object.h>
-#include <lustre_mds.h> 
+#include <lustre_mds.h>
 #include "mdt_internal.h"
 
 /*
@@ -263,7 +263,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
                         repbody->valid |= OBD_MD_LINKNAME;
                         repbody->eadatasize = rc + 1;
                         ((char*)ma->ma_lmm)[rc] = 0;        /* NULL terminate */
-                        CDEBUG(D_INODE, "symlink dest %s, len = %d\n", 
+                        CDEBUG(D_INODE, "symlink dest %s, len = %d\n",
                                         (char*)buffer, rc);
                         rc = 0;
                 }
@@ -305,11 +305,11 @@ shrink:
 /*
         if (req_capsule_has_field(pill, &RMF_DLM_REP)) {
                 offset = 2;
-        } else 
+        } else
                 offset = 1;
 */
         lustre_shrink_reply(req, offset, repbody->eadatasize, 1);
-        if (repbody->eadatasize) 
+        if (repbody->eadatasize)
                 offset ++;
         lustre_shrink_reply(req, offset, repbody->aclsize, 0);
         RETURN(rc);
@@ -323,11 +323,11 @@ static int mdt_getattr(struct mdt_thread_info *info)
         LASSERT(lu_object_assert_exists(info->mti_ctxt,
                                         &info->mti_object->mot_obj.mo_lu));
         ENTRY;
-        
+
 
         req_capsule_set_size(&info->mti_pill, &RMF_EADATA,
                              RCL_SERVER, LUSTRE_POSIX_ACL_MAX_SIZE);
-       
+
         result = req_capsule_pack(&info->mti_pill);
         if (result)
                 RETURN(result);
@@ -382,7 +382,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
                 result = mdt_object_lock(info, child, lhc, child_bits);
                 if (result != 0) {
                         /* finally, we can get attr for child. */
-                        result = mdt_getattr_internal(info, child, 
+                        result = mdt_getattr_internal(info, child,
                                                       ldlm_rep ? 2 : 1);
                         if (result != 0)
                                 mdt_object_unlock(info, child, lhc, 1);
@@ -459,7 +459,7 @@ static int mdt_getattr_name(struct mdt_thread_info *info)
                              RCL_SERVER, info->mti_mdt->mdt_max_mdsize);
         req_capsule_set_size(&info->mti_pill, &RMF_EADATA,
                              RCL_SERVER, LUSTRE_POSIX_ACL_MAX_SIZE);
-        
+
         rc = req_capsule_pack(&info->mti_pill);
         if (rc)
                 RETURN(rc);
@@ -676,7 +676,7 @@ static int mdt_reint(struct mdt_thread_info *info)
                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
 
                 rc = mdt_reint_internal(info, opc);
-        
+
         } else
                 rc = opc;
         RETURN(rc);
@@ -1209,6 +1209,9 @@ static void mdt_thread_info_init(struct ptlrpc_request *req,
         memset(info, 0, sizeof(*info));
 
         info->mti_rep_buf_nr = ARRAY_SIZE(info->mti_rep_buf_size);
+        for (i = 0; i < ARRAY_SIZE(info->mti_rep_buf_size); i++)
+                info->mti_rep_buf_size[i] = -1;
+
         for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++)
                 mdt_lock_handle_init(&info->mti_lh[i]);
 
@@ -1218,7 +1221,8 @@ static void mdt_thread_info_init(struct ptlrpc_request *req,
         /* it can be NULL while CONNECT */
         if (req->rq_export)
                 info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
-        req_capsule_init(&info->mti_pill, req, RCL_SERVER, info->mti_rep_buf_size);
+        req_capsule_init(&info->mti_pill, req, RCL_SERVER,
+                         info->mti_rep_buf_size);
 }
 
 static void mdt_thread_info_fini(struct mdt_thread_info *info)
@@ -1620,12 +1624,12 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
         }
 
         rc = mdt_reint_internal(info, opc);
-        
+
         rep = req_capsule_server_get(&info->mti_pill, &RMF_DLM_REP);
         if (rep == NULL)
                 RETURN(-EFAULT);
         rep->lock_policy_res2 = rc;
-        
+
         intent_set_disposition(rep, DISP_IT_EXECD);
 
         mdt_update_last_transno(info, rep->lock_policy_res2);
@@ -2062,8 +2066,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m)
                 .psc_ctx_tags      = LCT_MD_THREAD
         };
 
-        m->mdt_setattr_service = 
-                ptlrpc_init_svc_conf(&conf, mdt_handle, 
+        m->mdt_setattr_service =
+                ptlrpc_init_svc_conf(&conf, mdt_handle,
                                      LUSTRE_MDT0_NAME "_setattr",
                                      m->mdt_md_dev.md_lu_dev.ld_proc_entry,
                                      NULL);
@@ -2150,7 +2154,7 @@ static struct lu_device *mdt_layer_setup(const struct lu_context *ctx,
                 GOTO(out_alloc, rc);
         }
         lu_device_get(d);
-      
+
         RETURN(d);
 out_alloc:
         ldt->ldt_ops->ldto_device_free(ctx, d);
@@ -2183,7 +2187,7 @@ static int mdt_stack_init(const struct lu_context *ctx,
         }
         d = tmp;
         md = lu2md_dev(d);
-        
+
         tmp = mdt_layer_setup(ctx, LUSTRE_CMM0_NAME, d, cfg);
         if (IS_ERR(tmp)) {
                 GOTO(out, rc = PTR_ERR(tmp));
@@ -2191,7 +2195,7 @@ static int mdt_stack_init(const struct lu_context *ctx,
         d = tmp;
         /*set mdd upcall device*/
         md->md_upcall.mu_upcall_dev = lu2md_dev(d);
-       
+
         md = lu2md_dev(d);
         /*set cmm upcall device*/
         md->md_upcall.mu_upcall_dev = &m->mdt_md_dev;
@@ -2637,10 +2641,10 @@ static int mdt_upcall(const struct lu_context *ctx, struct md_device *md,
         struct md_device  *next  = m->mdt_child;
         int rc = 0;
         ENTRY;
-        
+
         switch (ev) {
                 case MD_LOV_SYNC:
-                        rc = next->md_ops->mdo_get_maxsize(ctx, next, 
+                        rc = next->md_ops->mdo_get_maxsize(ctx, next,
                                     &m->mdt_max_mdsize, &m->mdt_max_cookiesize);
                         CDEBUG(D_INFO, "get max mdsize %d max cookiesize %d \n",
                                      m->mdt_max_mdsize, m->mdt_max_cookiesize);

diff --git a/lustre/ptlrpc/layout.c b/lustre/ptlrpc/layout.c
index 4f1a402..40dee66 100644 (file)
--- a/lustre/ptlrpc/layout.c
+++ b/lustre/ptlrpc/layout.c
@@ -252,6 +252,9 @@ static const struct req_format *req_formats[] = {
 struct req_msg_field {
         __u32       rmf_flags;
         const char *rmf_name;
+        /*
+         * Field length. (-1) means "variable length".
+         */
         int         rmf_size;
         void      (*rmf_swabber)(void *);
         int         rmf_offset[ARRAY_SIZE(req_formats)][RCL_NR];
@@ -304,11 +307,11 @@ const struct req_msg_field RMF_OBD_STATFS =
 EXPORT_SYMBOL(RMF_OBD_STATFS);
 
 const struct req_msg_field RMF_NAME =
-        DEFINE_MSGF("name", RMF_F_STRING, 0, NULL);
+        DEFINE_MSGF("name", RMF_F_STRING, -1, NULL);
 EXPORT_SYMBOL(RMF_NAME);
 
 const struct req_msg_field RMF_SYMTGT =
-        DEFINE_MSGF("symtgt", RMF_F_STRING, 0, NULL);
+        DEFINE_MSGF("symtgt", RMF_F_STRING, -1, NULL);
 EXPORT_SYMBOL(RMF_SYMTGT);
 
 const struct req_msg_field RMF_TGTUUID =
@@ -383,8 +386,7 @@ const struct req_msg_field RMF_REC_SETATTR =
 EXPORT_SYMBOL(RMF_REC_SETATTR);
 
 /* FIXME: this length should be defined as a macro*/
-const struct req_msg_field RMF_EADATA = DEFINE_MSGF("eadata", 0, 
-                                0, NULL);
+const struct req_msg_field RMF_EADATA = DEFINE_MSGF("eadata", 0, -1, NULL);
 EXPORT_SYMBOL(RMF_EADATA);
 
 const struct req_msg_field RMF_LOGCOOKIES =
@@ -641,7 +643,7 @@ int req_capsule_pack(struct req_capsule *pill)
         int nr;
         int result;
         int total;
-        
+
         const struct req_format *fmt;
 
         LASSERT(pill->rc_loc == RCL_SERVER);
@@ -653,9 +655,9 @@ int req_capsule_pack(struct req_capsule *pill)
                 int *size;
 
                 size = &pill->rc_area[i];
-                if (*size == 0) {
+                if (*size == -1) {
                         *size = fmt->rf_fields[RCL_SERVER].d[i]->rmf_size;
-                        LASSERT(*size != 0);
+                        LASSERT(*size != -1);
                 }
                 total += *size;
         }
@@ -689,6 +691,7 @@ static void *__req_capsule_get(struct req_capsule *pill,
         const struct req_format *fmt;
         struct lustre_msg       *msg;
         void                    *value;
+        int                      len;
         int                      offset;
 
         void *(*getter)(struct lustre_msg *m, int n, int minlen);
@@ -709,7 +712,8 @@ static void *__req_capsule_get(struct req_capsule *pill,
         getter = (field->rmf_flags & RMF_F_STRING) ?
                 (typeof(getter))lustre_msg_string : lustre_msg_buf;
 
-        value = getter(msg, offset, field->rmf_size);
+        len = max(field->rmf_size, 0);
+        value = getter(msg, offset, len);
 
         if (!(pill->rc_swabbed & (1 << offset)) && loc != pill->rc_loc &&
             field->rmf_swabber != NULL && value != NULL &&