Version:  2.0.40 2.2.26 2.4.37 2.6.39 3.0 3.1 3.2 3.3 3.4 3.5 3.6 3.7 3.8 3.9 3.10 3.11 3.12 3.13 3.14 3.15

Linux/drivers/staging/lustre/lustre/mdc/mdc_request.c

  1 /*
  2  * GPL HEADER START
  3  *
  4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  5  *
  6  * This program is free software; you can redistribute it and/or modify
  7  * it under the terms of the GNU General Public License version 2 only,
  8  * as published by the Free Software Foundation.
  9  *
 10  * This program is distributed in the hope that it will be useful, but
 11  * WITHOUT ANY WARRANTY; without even the implied warranty of
 12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 13  * General Public License version 2 for more details (a copy is included
 14  * in the LICENSE file that accompanied this code).
 15  *
 16  * You should have received a copy of the GNU General Public License
 17  * version 2 along with this program; If not, see
 18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
 19  *
 20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
 21  * CA 95054 USA or visit www.sun.com if you need additional information or
 22  * have any questions.
 23  *
 24  * GPL HEADER END
 25  */
 26 /*
 27  * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
 28  * Use is subject to license terms.
 29  *
 30  * Copyright (c) 2011, 2012, Intel Corporation.
 31  */
 32 /*
 33  * This file is part of Lustre, http://www.lustre.org/
 34  * Lustre is a trademark of Sun Microsystems, Inc.
 35  */
 36 
 37 #define DEBUG_SUBSYSTEM S_MDC
 38 
 39 # include <linux/module.h>
 40 # include <linux/pagemap.h>
 41 # include <linux/miscdevice.h>
 42 # include <linux/init.h>
 43 # include <linux/utsname.h>
 44 
 45 #include <lustre_acl.h>
 46 #include <obd_class.h>
 47 #include <lustre_fid.h>
 48 #include <lprocfs_status.h>
 49 #include <lustre_param.h>
 50 #include <lustre_log.h>
 51 
 52 #include "mdc_internal.h"
 53 
 54 #define REQUEST_MINOR 244
 55 
 56 struct mdc_renew_capa_args {
 57         struct obd_capa *ra_oc;
 58         renew_capa_cb_t  ra_cb;
 59 };
 60 
 61 static int mdc_cleanup(struct obd_device *obd);
 62 
 63 int mdc_unpack_capa(struct obd_export *exp, struct ptlrpc_request *req,
 64                     const struct req_msg_field *field, struct obd_capa **oc)
 65 {
 66         struct lustre_capa *capa;
 67         struct obd_capa *c;
 68 
 69         /* swabbed already in mdc_enqueue */
 70         capa = req_capsule_server_get(&req->rq_pill, field);
 71         if (capa == NULL)
 72                 return -EPROTO;
 73 
 74         c = alloc_capa(CAPA_SITE_CLIENT);
 75         if (IS_ERR(c)) {
 76                 CDEBUG(D_INFO, "alloc capa failed!\n");
 77                 return PTR_ERR(c);
 78         } else {
 79                 c->c_capa = *capa;
 80                 *oc = c;
 81                 return 0;
 82         }
 83 }
 84 
 85 static inline int mdc_queue_wait(struct ptlrpc_request *req)
 86 {
 87         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
 88         int rc;
 89 
 90         /* mdc_enter_request() ensures that this client has no more
 91          * than cl_max_rpcs_in_flight RPCs simultaneously inf light
 92          * against an MDT. */
 93         rc = mdc_enter_request(cli);
 94         if (rc != 0)
 95                 return rc;
 96 
 97         rc = ptlrpc_queue_wait(req);
 98         mdc_exit_request(cli);
 99 
100         return rc;
101 }
102 
103 /* Helper that implements most of mdc_getstatus and signal_completed_replay. */
104 /* XXX this should become mdc_get_info("key"), sending MDS_GET_INFO RPC */
105 static int send_getstatus(struct obd_import *imp, struct lu_fid *rootfid,
106                           struct obd_capa **pc, int level, int msg_flags)
107 {
108         struct ptlrpc_request *req;
109         struct mdt_body       *body;
110         int                 rc;
111 
112         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_GETSTATUS,
113                                         LUSTRE_MDS_VERSION, MDS_GETSTATUS);
114         if (req == NULL)
115                 return -ENOMEM;
116 
117         mdc_pack_body(req, NULL, NULL, 0, 0, -1, 0);
118         lustre_msg_add_flags(req->rq_reqmsg, msg_flags);
119         req->rq_send_state = level;
120 
121         ptlrpc_request_set_replen(req);
122 
123         rc = ptlrpc_queue_wait(req);
124         if (rc)
125                 GOTO(out, rc);
126 
127         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
128         if (body == NULL)
129                 GOTO(out, rc = -EPROTO);
130 
131         if (body->valid & OBD_MD_FLMDSCAPA) {
132                 rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, pc);
133                 if (rc)
134                         GOTO(out, rc);
135         }
136 
137         *rootfid = body->fid1;
138         CDEBUG(D_NET,
139                "root fid="DFID", last_committed="LPU64"\n",
140                PFID(rootfid),
141                lustre_msg_get_last_committed(req->rq_repmsg));
142 out:
143         ptlrpc_req_finished(req);
144         return rc;
145 }
146 
147 /* This should be mdc_get_info("rootfid") */
148 int mdc_getstatus(struct obd_export *exp, struct lu_fid *rootfid,
149                   struct obd_capa **pc)
150 {
151         return send_getstatus(class_exp2cliimp(exp), rootfid, pc,
152                               LUSTRE_IMP_FULL, 0);
153 }
154 
155 /*
156  * This function now is known to always saying that it will receive 4 buffers
157  * from server. Even for cases when acl_size and md_size is zero, RPC header
158  * will contain 4 fields and RPC itself will contain zero size fields. This is
159  * because mdt_getattr*() _always_ returns 4 fields, but if acl is not needed
160  * and thus zero, it shrinks it, making zero size. The same story about
161  * md_size. And this is course of problem when client waits for smaller number
162  * of fields. This issue will be fixed later when client gets aware of RPC
163  * layouts.  --umka
164  */
165 static int mdc_getattr_common(struct obd_export *exp,
166                               struct ptlrpc_request *req)
167 {
168         struct req_capsule *pill = &req->rq_pill;
169         struct mdt_body    *body;
170         void           *eadata;
171         int              rc;
172 
173         /* Request message already built. */
174         rc = ptlrpc_queue_wait(req);
175         if (rc != 0)
176                 return rc;
177 
178         /* sanity check for the reply */
179         body = req_capsule_server_get(pill, &RMF_MDT_BODY);
180         if (body == NULL)
181                 return -EPROTO;
182 
183         CDEBUG(D_NET, "mode: %o\n", body->mode);
184 
185         if (body->eadatasize != 0) {
186                 mdc_update_max_ea_from_body(exp, body);
187 
188                 eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
189                                                       body->eadatasize);
190                 if (eadata == NULL)
191                         return -EPROTO;
192         }
193 
194         if (body->valid & OBD_MD_FLRMTPERM) {
195                 struct mdt_remote_perm *perm;
196 
197                 LASSERT(client_is_remote(exp));
198                 perm = req_capsule_server_swab_get(pill, &RMF_ACL,
199                                                 lustre_swab_mdt_remote_perm);
200                 if (perm == NULL)
201                         return -EPROTO;
202         }
203 
204         if (body->valid & OBD_MD_FLMDSCAPA) {
205                 struct lustre_capa *capa;
206                 capa = req_capsule_server_get(pill, &RMF_CAPA1);
207                 if (capa == NULL)
208                         return -EPROTO;
209         }
210 
211         return 0;
212 }
213 
214 int mdc_getattr(struct obd_export *exp, struct md_op_data *op_data,
215                 struct ptlrpc_request **request)
216 {
217         struct ptlrpc_request *req;
218         int                 rc;
219 
220         /* Single MDS without an LMV case */
221         if (op_data->op_flags & MF_GET_MDT_IDX) {
222                 op_data->op_mds = 0;
223                 return 0;
224         }
225         *request = NULL;
226         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_GETATTR);
227         if (req == NULL)
228                 return -ENOMEM;
229 
230         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
231 
232         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR);
233         if (rc) {
234                 ptlrpc_request_free(req);
235                 return rc;
236         }
237 
238         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
239                       op_data->op_valid, op_data->op_mode, -1, 0);
240 
241         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
242                              op_data->op_mode);
243         if (op_data->op_valid & OBD_MD_FLRMTPERM) {
244                 LASSERT(client_is_remote(exp));
245                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
246                                      sizeof(struct mdt_remote_perm));
247         }
248         ptlrpc_request_set_replen(req);
249 
250         rc = mdc_getattr_common(exp, req);
251         if (rc)
252                 ptlrpc_req_finished(req);
253         else
254                 *request = req;
255         return rc;
256 }
257 
258 int mdc_getattr_name(struct obd_export *exp, struct md_op_data *op_data,
259                      struct ptlrpc_request **request)
260 {
261         struct ptlrpc_request *req;
262         int                 rc;
263 
264         *request = NULL;
265         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
266                                    &RQF_MDS_GETATTR_NAME);
267         if (req == NULL)
268                 return -ENOMEM;
269 
270         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
271         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
272                              op_data->op_namelen + 1);
273 
274         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR_NAME);
275         if (rc) {
276                 ptlrpc_request_free(req);
277                 return rc;
278         }
279 
280         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
281                       op_data->op_valid, op_data->op_mode,
282                       op_data->op_suppgids[0], 0);
283 
284         if (op_data->op_name) {
285                 char *name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
286                 LASSERT(strnlen(op_data->op_name, op_data->op_namelen) ==
287                                 op_data->op_namelen);
288                 memcpy(name, op_data->op_name, op_data->op_namelen);
289         }
290 
291         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
292                              op_data->op_mode);
293         ptlrpc_request_set_replen(req);
294 
295         rc = mdc_getattr_common(exp, req);
296         if (rc)
297                 ptlrpc_req_finished(req);
298         else
299                 *request = req;
300         return rc;
301 }
302 
303 static int mdc_is_subdir(struct obd_export *exp,
304                          const struct lu_fid *pfid,
305                          const struct lu_fid *cfid,
306                          struct ptlrpc_request **request)
307 {
308         struct ptlrpc_request  *req;
309         int                  rc;
310 
311         *request = NULL;
312         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
313                                         &RQF_MDS_IS_SUBDIR, LUSTRE_MDS_VERSION,
314                                         MDS_IS_SUBDIR);
315         if (req == NULL)
316                 return -ENOMEM;
317 
318         mdc_is_subdir_pack(req, pfid, cfid, 0);
319         ptlrpc_request_set_replen(req);
320 
321         rc = ptlrpc_queue_wait(req);
322         if (rc && rc != -EREMOTE)
323                 ptlrpc_req_finished(req);
324         else
325                 *request = req;
326         return rc;
327 }
328 
329 static int mdc_xattr_common(struct obd_export *exp,const struct req_format *fmt,
330                             const struct lu_fid *fid,
331                             struct obd_capa *oc, int opcode, obd_valid valid,
332                             const char *xattr_name, const char *input,
333                             int input_size, int output_size, int flags,
334                             __u32 suppgid, struct ptlrpc_request **request)
335 {
336         struct ptlrpc_request *req;
337         int   xattr_namelen = 0;
338         char *tmp;
339         int   rc;
340 
341         *request = NULL;
342         req = ptlrpc_request_alloc(class_exp2cliimp(exp), fmt);
343         if (req == NULL)
344                 return -ENOMEM;
345 
346         mdc_set_capa_size(req, &RMF_CAPA1, oc);
347         if (xattr_name) {
348                 xattr_namelen = strlen(xattr_name) + 1;
349                 req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
350                                      xattr_namelen);
351         }
352         if (input_size) {
353                 LASSERT(input);
354                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
355                                      input_size);
356         }
357 
358         /* Flush local XATTR locks to get rid of a possible cancel RPC */
359         if (opcode == MDS_REINT && fid_is_sane(fid) &&
360             exp->exp_connect_data.ocd_ibits_known & MDS_INODELOCK_XATTR) {
361                 LIST_HEAD(cancels);
362                 int count;
363 
364                 /* Without that packing would fail */
365                 if (input_size == 0)
366                         req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
367                                              RCL_CLIENT, 0);
368 
369                 count = mdc_resource_get_unused(exp, fid,
370                                                 &cancels, LCK_EX,
371                                                 MDS_INODELOCK_XATTR);
372 
373                 rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
374                 if (rc) {
375                         ptlrpc_request_free(req);
376                         return rc;
377                 }
378         } else {
379                 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, opcode);
380                 if (rc) {
381                         ptlrpc_request_free(req);
382                         return rc;
383                 }
384         }
385 
386         if (opcode == MDS_REINT) {
387                 struct mdt_rec_setxattr *rec;
388 
389                 CLASSERT(sizeof(struct mdt_rec_setxattr) ==
390                          sizeof(struct mdt_rec_reint));
391                 rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
392                 rec->sx_opcode = REINT_SETXATTR;
393                 rec->sx_fsuid  = from_kuid(&init_user_ns, current_fsuid());
394                 rec->sx_fsgid  = from_kgid(&init_user_ns, current_fsgid());
395                 rec->sx_cap    = cfs_curproc_cap_pack();
396                 rec->sx_suppgid1 = suppgid;
397                 rec->sx_suppgid2 = -1;
398                 rec->sx_fid    = *fid;
399                 rec->sx_valid  = valid | OBD_MD_FLCTIME;
400                 rec->sx_time   = cfs_time_current_sec();
401                 rec->sx_size   = output_size;
402                 rec->sx_flags  = flags;
403 
404                 mdc_pack_capa(req, &RMF_CAPA1, oc);
405         } else {
406                 mdc_pack_body(req, fid, oc, valid, output_size, suppgid, flags);
407         }
408 
409         if (xattr_name) {
410                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
411                 memcpy(tmp, xattr_name, xattr_namelen);
412         }
413         if (input_size) {
414                 tmp = req_capsule_client_get(&req->rq_pill, &RMF_EADATA);
415                 memcpy(tmp, input, input_size);
416         }
417 
418         if (req_capsule_has_field(&req->rq_pill, &RMF_EADATA, RCL_SERVER))
419                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
420                                      RCL_SERVER, output_size);
421         ptlrpc_request_set_replen(req);
422 
423         /* make rpc */
424         if (opcode == MDS_REINT)
425                 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
426 
427         rc = ptlrpc_queue_wait(req);
428 
429         if (opcode == MDS_REINT)
430                 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
431 
432         if (rc)
433                 ptlrpc_req_finished(req);
434         else
435                 *request = req;
436         return rc;
437 }
438 
439 int mdc_setxattr(struct obd_export *exp, const struct lu_fid *fid,
440                  struct obd_capa *oc, obd_valid valid, const char *xattr_name,
441                  const char *input, int input_size, int output_size,
442                  int flags, __u32 suppgid, struct ptlrpc_request **request)
443 {
444         return mdc_xattr_common(exp, &RQF_MDS_REINT_SETXATTR,
445                                 fid, oc, MDS_REINT, valid, xattr_name,
446                                 input, input_size, output_size, flags,
447                                 suppgid, request);
448 }
449 
450 int mdc_getxattr(struct obd_export *exp, const struct lu_fid *fid,
451                  struct obd_capa *oc, obd_valid valid, const char *xattr_name,
452                  const char *input, int input_size, int output_size,
453                  int flags, struct ptlrpc_request **request)
454 {
455         return mdc_xattr_common(exp, &RQF_MDS_GETXATTR,
456                                 fid, oc, MDS_GETXATTR, valid, xattr_name,
457                                 input, input_size, output_size, flags,
458                                 -1, request);
459 }
460 
461 #ifdef CONFIG_FS_POSIX_ACL
462 static int mdc_unpack_acl(struct ptlrpc_request *req, struct lustre_md *md)
463 {
464         struct req_capsule     *pill = &req->rq_pill;
465         struct mdt_body *body = md->body;
466         struct posix_acl       *acl;
467         void               *buf;
468         int                  rc;
469 
470         if (!body->aclsize)
471                 return 0;
472 
473         buf = req_capsule_server_sized_get(pill, &RMF_ACL, body->aclsize);
474 
475         if (!buf)
476                 return -EPROTO;
477 
478         acl = posix_acl_from_xattr(&init_user_ns, buf, body->aclsize);
479         if (IS_ERR(acl)) {
480                 rc = PTR_ERR(acl);
481                 CERROR("convert xattr to acl: %d\n", rc);
482                 return rc;
483         }
484 
485         rc = posix_acl_valid(acl);
486         if (rc) {
487                 CERROR("validate acl: %d\n", rc);
488                 posix_acl_release(acl);
489                 return rc;
490         }
491 
492         md->posix_acl = acl;
493         return 0;
494 }
495 #else
496 #define mdc_unpack_acl(req, md) 0
497 #endif
498 
499 int mdc_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req,
500                       struct obd_export *dt_exp, struct obd_export *md_exp,
501                       struct lustre_md *md)
502 {
503         struct req_capsule *pill = &req->rq_pill;
504         int rc;
505 
506         LASSERT(md);
507         memset(md, 0, sizeof(*md));
508 
509         md->body = req_capsule_server_get(pill, &RMF_MDT_BODY);
510         LASSERT(md->body != NULL);
511 
512         if (md->body->valid & OBD_MD_FLEASIZE) {
513                 int lmmsize;
514                 struct lov_mds_md *lmm;
515 
516                 if (!S_ISREG(md->body->mode)) {
517                         CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, should be a "
518                                "regular file, but is not\n");
519                         GOTO(out, rc = -EPROTO);
520                 }
521 
522                 if (md->body->eadatasize == 0) {
523                         CDEBUG(D_INFO, "OBD_MD_FLEASIZE set, "
524                                "but eadatasize 0\n");
525                         GOTO(out, rc = -EPROTO);
526                 }
527                 lmmsize = md->body->eadatasize;
528                 lmm = req_capsule_server_sized_get(pill, &RMF_MDT_MD, lmmsize);
529                 if (!lmm)
530                         GOTO(out, rc = -EPROTO);
531 
532                 rc = obd_unpackmd(dt_exp, &md->lsm, lmm, lmmsize);
533                 if (rc < 0)
534                         GOTO(out, rc);
535 
536                 if (rc < sizeof(*md->lsm)) {
537                         CDEBUG(D_INFO, "lsm size too small: "
538                                "rc < sizeof (*md->lsm) (%d < %d)\n",
539                                rc, (int)sizeof(*md->lsm));
540                         GOTO(out, rc = -EPROTO);
541                 }
542 
543         } else if (md->body->valid & OBD_MD_FLDIREA) {
544                 int lmvsize;
545                 struct lov_mds_md *lmv;
546 
547                 if(!S_ISDIR(md->body->mode)) {
548                         CDEBUG(D_INFO, "OBD_MD_FLDIREA set, should be a "
549                                "directory, but is not\n");
550                         GOTO(out, rc = -EPROTO);
551                 }
552 
553                 if (md->body->eadatasize == 0) {
554                         CDEBUG(D_INFO, "OBD_MD_FLDIREA is set, "
555                                "but eadatasize 0\n");
556                         return -EPROTO;
557                 }
558                 if (md->body->valid & OBD_MD_MEA) {
559                         lmvsize = md->body->eadatasize;
560                         lmv = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
561                                                            lmvsize);
562                         if (!lmv)
563                                 GOTO(out, rc = -EPROTO);
564 
565                         rc = obd_unpackmd(md_exp, (void *)&md->mea, lmv,
566                                           lmvsize);
567                         if (rc < 0)
568                                 GOTO(out, rc);
569 
570                         if (rc < sizeof(*md->mea)) {
571                                 CDEBUG(D_INFO, "size too small:  "
572                                        "rc < sizeof(*md->mea) (%d < %d)\n",
573                                         rc, (int)sizeof(*md->mea));
574                                 GOTO(out, rc = -EPROTO);
575                         }
576                 }
577         }
578         rc = 0;
579 
580         if (md->body->valid & OBD_MD_FLRMTPERM) {
581                 /* remote permission */
582                 LASSERT(client_is_remote(exp));
583                 md->remote_perm = req_capsule_server_swab_get(pill, &RMF_ACL,
584                                                 lustre_swab_mdt_remote_perm);
585                 if (!md->remote_perm)
586                         GOTO(out, rc = -EPROTO);
587         }
588         else if (md->body->valid & OBD_MD_FLACL) {
589                 /* for ACL, it's possible that FLACL is set but aclsize is zero.
590                  * only when aclsize != 0 there's an actual segment for ACL
591                  * in reply buffer.
592                  */
593                 if (md->body->aclsize) {
594                         rc = mdc_unpack_acl(req, md);
595                         if (rc)
596                                 GOTO(out, rc);
597 #ifdef CONFIG_FS_POSIX_ACL
598                 } else {
599                         md->posix_acl = NULL;
600 #endif
601                 }
602         }
603         if (md->body->valid & OBD_MD_FLMDSCAPA) {
604                 struct obd_capa *oc = NULL;
605 
606                 rc = mdc_unpack_capa(NULL, req, &RMF_CAPA1, &oc);
607                 if (rc)
608                         GOTO(out, rc);
609                 md->mds_capa = oc;
610         }
611 
612         if (md->body->valid & OBD_MD_FLOSSCAPA) {
613                 struct obd_capa *oc = NULL;
614 
615                 rc = mdc_unpack_capa(NULL, req, &RMF_CAPA2, &oc);
616                 if (rc)
617                         GOTO(out, rc);
618                 md->oss_capa = oc;
619         }
620 
621 out:
622         if (rc) {
623                 if (md->oss_capa) {
624                         capa_put(md->oss_capa);
625                         md->oss_capa = NULL;
626                 }
627                 if (md->mds_capa) {
628                         capa_put(md->mds_capa);
629                         md->mds_capa = NULL;
630                 }
631 #ifdef CONFIG_FS_POSIX_ACL
632                 posix_acl_release(md->posix_acl);
633 #endif
634                 if (md->lsm)
635                         obd_free_memmd(dt_exp, &md->lsm);
636         }
637         return rc;
638 }
639 
640 int mdc_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
641 {
642         return 0;
643 }
644 
645 /**
646  * Handles both OPEN and SETATTR RPCs for OPEN-CLOSE and SETATTR-DONE_WRITING
647  * RPC chains.
648  */
649 void mdc_replay_open(struct ptlrpc_request *req)
650 {
651         struct md_open_data *mod = req->rq_cb_data;
652         struct ptlrpc_request *close_req;
653         struct obd_client_handle *och;
654         struct lustre_handle old;
655         struct mdt_body *body;
656 
657         if (mod == NULL) {
658                 DEBUG_REQ(D_ERROR, req,
659                           "Can't properly replay without open data.");
660                 return;
661         }
662 
663         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
664         LASSERT(body != NULL);
665 
666         och = mod->mod_och;
667         if (och != NULL) {
668                 struct lustre_handle *file_fh;
669 
670                 LASSERT(och->och_magic == OBD_CLIENT_HANDLE_MAGIC);
671 
672                 file_fh = &och->och_fh;
673                 CDEBUG(D_HA, "updating handle from "LPX64" to "LPX64"\n",
674                        file_fh->cookie, body->handle.cookie);
675                 old = *file_fh;
676                 *file_fh = body->handle;
677         }
678         close_req = mod->mod_close_req;
679         if (close_req != NULL) {
680                 __u32 opc = lustre_msg_get_opc(close_req->rq_reqmsg);
681                 struct mdt_ioepoch *epoch;
682 
683                 LASSERT(opc == MDS_CLOSE || opc == MDS_DONE_WRITING);
684                 epoch = req_capsule_client_get(&close_req->rq_pill,
685                                                &RMF_MDT_EPOCH);
686                 LASSERT(epoch);
687 
688                 if (och != NULL)
689                         LASSERT(!memcmp(&old, &epoch->handle, sizeof(old)));
690                 DEBUG_REQ(D_HA, close_req, "updating close body with new fh");
691                 epoch->handle = body->handle;
692         }
693 }
694 
695 void mdc_commit_open(struct ptlrpc_request *req)
696 {
697         struct md_open_data *mod = req->rq_cb_data;
698         if (mod == NULL)
699                 return;
700 
701         /**
702          * No need to touch md_open_data::mod_och, it holds a reference on
703          * \var mod and will zero references to each other, \var mod will be
704          * freed after that when md_open_data::mod_och will put the reference.
705          */
706 
707         /**
708          * Do not let open request to disappear as it still may be needed
709          * for close rpc to happen (it may happen on evict only, otherwise
710          * ptlrpc_request::rq_replay does not let mdc_commit_open() to be
711          * called), just mark this rpc as committed to distinguish these 2
712          * cases, see mdc_close() for details. The open request reference will
713          * be put along with freeing \var mod.
714          */
715         ptlrpc_request_addref(req);
716         spin_lock(&req->rq_lock);
717         req->rq_committed = 1;
718         spin_unlock(&req->rq_lock);
719         req->rq_cb_data = NULL;
720         obd_mod_put(mod);
721 }
722 
723 int mdc_set_open_replay_data(struct obd_export *exp,
724                              struct obd_client_handle *och,
725                              struct lookup_intent *it)
726 {
727         struct md_open_data   *mod;
728         struct mdt_rec_create *rec;
729         struct mdt_body       *body;
730         struct ptlrpc_request *open_req = it->d.lustre.it_data;
731         struct obd_import     *imp = open_req->rq_import;
732 
733         if (!open_req->rq_replay)
734                 return 0;
735 
736         rec = req_capsule_client_get(&open_req->rq_pill, &RMF_REC_REINT);
737         body = req_capsule_server_get(&open_req->rq_pill, &RMF_MDT_BODY);
738         LASSERT(rec != NULL);
739         /* Incoming message in my byte order (it's been swabbed). */
740         /* Outgoing messages always in my byte order. */
741         LASSERT(body != NULL);
742 
743         /* Only if the import is replayable, we set replay_open data */
744         if (och && imp->imp_replayable) {
745                 mod = obd_mod_alloc();
746                 if (mod == NULL) {
747                         DEBUG_REQ(D_ERROR, open_req,
748                                   "Can't allocate md_open_data");
749                         return 0;
750                 }
751 
752                 /**
753                  * Take a reference on \var mod, to be freed on mdc_close().
754                  * It protects \var mod from being freed on eviction (commit
755                  * callback is called despite rq_replay flag).
756                  * Another reference for \var och.
757                  */
758                 obd_mod_get(mod);
759                 obd_mod_get(mod);
760 
761                 spin_lock(&open_req->rq_lock);
762                 och->och_mod = mod;
763                 mod->mod_och = och;
764                 mod->mod_is_create = it_disposition(it, DISP_OPEN_CREATE) ||
765                                      it_disposition(it, DISP_OPEN_STRIPE);
766                 mod->mod_open_req = open_req;
767                 open_req->rq_cb_data = mod;
768                 open_req->rq_commit_cb = mdc_commit_open;
769                 spin_unlock(&open_req->rq_lock);
770         }
771 
772         rec->cr_fid2 = body->fid1;
773         rec->cr_ioepoch = body->ioepoch;
774         rec->cr_old_handle.cookie = body->handle.cookie;
775         open_req->rq_replay_cb = mdc_replay_open;
776         if (!fid_is_sane(&body->fid1)) {
777                 DEBUG_REQ(D_ERROR, open_req, "Saving replay request with "
778                           "insane fid");
779                 LBUG();
780         }
781 
782         DEBUG_REQ(D_RPCTRACE, open_req, "Set up open replay data");
783         return 0;
784 }
785 
786 static void mdc_free_open(struct md_open_data *mod)
787 {
788         int committed = 0;
789 
790         if (mod->mod_is_create == 0 &&
791             imp_connect_disp_stripe(mod->mod_open_req->rq_import))
792                 committed = 1;
793 
794         LASSERT(mod->mod_open_req->rq_replay == 0);
795 
796         DEBUG_REQ(D_RPCTRACE, mod->mod_open_req, "free open request\n");
797 
798         ptlrpc_request_committed(mod->mod_open_req, committed);
799         if (mod->mod_close_req)
800                 ptlrpc_request_committed(mod->mod_close_req, committed);
801 }
802 
803 int mdc_clear_open_replay_data(struct obd_export *exp,
804                                struct obd_client_handle *och)
805 {
806         struct md_open_data *mod = och->och_mod;
807 
808         /**
809          * It is possible to not have \var mod in a case of eviction between
810          * lookup and ll_file_open().
811          **/
812         if (mod == NULL)
813                 return 0;
814 
815         LASSERT(mod != LP_POISON);
816         LASSERT(mod->mod_open_req != NULL);
817         mdc_free_open(mod);
818 
819         mod->mod_och = NULL;
820         och->och_mod = NULL;
821         obd_mod_put(mod);
822 
823         return 0;
824 }
825 
826 /* Prepares the request for the replay by the given reply */
827 static void mdc_close_handle_reply(struct ptlrpc_request *req,
828                                    struct md_op_data *op_data, int rc) {
829         struct mdt_body  *repbody;
830         struct mdt_ioepoch *epoch;
831 
832         if (req && rc == -EAGAIN) {
833                 repbody = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
834                 epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH);
835 
836                 epoch->flags |= MF_SOM_AU;
837                 if (repbody->valid & OBD_MD_FLGETATTRLOCK)
838                         op_data->op_flags |= MF_GETATTR_LOCK;
839         }
840 }
841 
842 int mdc_close(struct obd_export *exp, struct md_op_data *op_data,
843               struct md_open_data *mod, struct ptlrpc_request **request)
844 {
845         struct obd_device     *obd = class_exp2obd(exp);
846         struct ptlrpc_request *req;
847         struct req_format     *req_fmt;
848         int                    rc;
849         int                    saved_rc = 0;
850 
851 
852         req_fmt = &RQF_MDS_CLOSE;
853         if (op_data->op_bias & MDS_HSM_RELEASE) {
854                 req_fmt = &RQF_MDS_RELEASE_CLOSE;
855 
856                 /* allocate a FID for volatile file */
857                 rc = mdc_fid_alloc(exp, &op_data->op_fid2, op_data);
858                 if (rc < 0) {
859                         CERROR("%s: "DFID" failed to allocate FID: %d\n",
860                                obd->obd_name, PFID(&op_data->op_fid1), rc);
861                         /* save the errcode and proceed to close */
862                         saved_rc = rc;
863                 }
864         }
865 
866         *request = NULL;
867         req = ptlrpc_request_alloc(class_exp2cliimp(exp), req_fmt);
868         if (req == NULL)
869                 return -ENOMEM;
870 
871         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
872 
873         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_CLOSE);
874         if (rc) {
875                 ptlrpc_request_free(req);
876                 return rc;
877         }
878 
879         /* To avoid a livelock (bug 7034), we need to send CLOSE RPCs to a
880          * portal whose threads are not taking any DLM locks and are therefore
881          * always progressing */
882         req->rq_request_portal = MDS_READPAGE_PORTAL;
883         ptlrpc_at_set_req_timeout(req);
884 
885         /* Ensure that this close's handle is fixed up during replay. */
886         if (likely(mod != NULL)) {
887                 LASSERTF(mod->mod_open_req != NULL &&
888                          mod->mod_open_req->rq_type != LI_POISON,
889                          "POISONED open %p!\n", mod->mod_open_req);
890 
891                 mod->mod_close_req = req;
892 
893                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched open");
894                 /* We no longer want to preserve this open for replay even
895                  * though the open was committed. b=3632, b=3633 */
896                 spin_lock(&mod->mod_open_req->rq_lock);
897                 mod->mod_open_req->rq_replay = 0;
898                 spin_unlock(&mod->mod_open_req->rq_lock);
899         } else {
900                  CDEBUG(D_HA, "couldn't find open req; expecting close error\n");
901         }
902 
903         mdc_close_pack(req, op_data);
904 
905         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
906                              obd->u.cli.cl_max_mds_easize);
907         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
908                              obd->u.cli.cl_max_mds_cookiesize);
909 
910         ptlrpc_request_set_replen(req);
911 
912         mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
913         rc = ptlrpc_queue_wait(req);
914         mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
915 
916         if (req->rq_repmsg == NULL) {
917                 CDEBUG(D_RPCTRACE, "request failed to send: %p, %d\n", req,
918                        req->rq_status);
919                 if (rc == 0)
920                         rc = req->rq_status ?: -EIO;
921         } else if (rc == 0 || rc == -EAGAIN) {
922                 struct mdt_body *body;
923 
924                 rc = lustre_msg_get_status(req->rq_repmsg);
925                 if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
926                         DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR, err "
927                                   "= %d", rc);
928                         if (rc > 0)
929                                 rc = -rc;
930                 }
931                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
932                 if (body == NULL)
933                         rc = -EPROTO;
934         } else if (rc == -ESTALE) {
935                 /**
936                  * it can be allowed error after 3633 if open was committed and
937                  * server failed before close was sent. Let's check if mod
938                  * exists and return no error in that case
939                  */
940                 if (mod) {
941                         DEBUG_REQ(D_HA, req, "Reset ESTALE = %d", rc);
942                         LASSERT(mod->mod_open_req != NULL);
943                         if (mod->mod_open_req->rq_committed)
944                                 rc = 0;
945                 }
946         }
947 
948         if (mod) {
949                 if (rc != 0)
950                         mod->mod_close_req = NULL;
951                 /* Since now, mod is accessed through open_req only,
952                  * thus close req does not keep a reference on mod anymore. */
953                 obd_mod_put(mod);
954         }
955         *request = req;
956         mdc_close_handle_reply(req, op_data, rc);
957         return rc < 0 ? rc : saved_rc;
958 }
959 
960 int mdc_done_writing(struct obd_export *exp, struct md_op_data *op_data,
961                      struct md_open_data *mod)
962 {
963         struct obd_device     *obd = class_exp2obd(exp);
964         struct ptlrpc_request *req;
965         int                 rc;
966 
967         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
968                                    &RQF_MDS_DONE_WRITING);
969         if (req == NULL)
970                 return -ENOMEM;
971 
972         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
973         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_DONE_WRITING);
974         if (rc) {
975                 ptlrpc_request_free(req);
976                 return rc;
977         }
978 
979         if (mod != NULL) {
980                 LASSERTF(mod->mod_open_req != NULL &&
981                          mod->mod_open_req->rq_type != LI_POISON,
982                          "POISONED setattr %p!\n", mod->mod_open_req);
983 
984                 mod->mod_close_req = req;
985                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched setattr");
986                 /* We no longer want to preserve this setattr for replay even
987                  * though the open was committed. b=3632, b=3633 */
988                 spin_lock(&mod->mod_open_req->rq_lock);
989                 mod->mod_open_req->rq_replay = 0;
990                 spin_unlock(&mod->mod_open_req->rq_lock);
991         }
992 
993         mdc_close_pack(req, op_data);
994         ptlrpc_request_set_replen(req);
995 
996         mdc_get_rpc_lock(obd->u.cli.cl_close_lock, NULL);
997         rc = ptlrpc_queue_wait(req);
998         mdc_put_rpc_lock(obd->u.cli.cl_close_lock, NULL);
999 
1000         if (rc == -ESTALE) {
1001                 /**
1002                  * it can be allowed error after 3633 if open or setattr were
1003                  * committed and server failed before close was sent.
1004                  * Let's check if mod exists and return no error in that case
1005                  */
1006                 if (mod) {
1007                         LASSERT(mod->mod_open_req != NULL);
1008                         if (mod->mod_open_req->rq_committed)
1009                                 rc = 0;
1010                 }
1011         }
1012 
1013         if (mod) {
1014                 if (rc != 0)
1015                         mod->mod_close_req = NULL;
1016                 LASSERT(mod->mod_open_req != NULL);
1017                 mdc_free_open(mod);
1018 
1019                 /* Since now, mod is accessed through setattr req only,
1020                  * thus DW req does not keep a reference on mod anymore. */
1021                 obd_mod_put(mod);
1022         }
1023 
1024         mdc_close_handle_reply(req, op_data, rc);
1025         ptlrpc_req_finished(req);
1026         return rc;
1027 }
1028 
1029 
1030 int mdc_readpage(struct obd_export *exp, struct md_op_data *op_data,
1031                  struct page **pages, struct ptlrpc_request **request)
1032 {
1033         struct ptlrpc_request   *req;
1034         struct ptlrpc_bulk_desc *desc;
1035         int                   i;
1036         wait_queue_head_t             waitq;
1037         int                   resends = 0;
1038         struct l_wait_info       lwi;
1039         int                   rc;
1040 
1041         *request = NULL;
1042         init_waitqueue_head(&waitq);
1043 
1044 restart_bulk:
1045         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_READPAGE);
1046         if (req == NULL)
1047                 return -ENOMEM;
1048 
1049         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1050 
1051         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_READPAGE);
1052         if (rc) {
1053                 ptlrpc_request_free(req);
1054                 return rc;
1055         }
1056 
1057         req->rq_request_portal = MDS_READPAGE_PORTAL;
1058         ptlrpc_at_set_req_timeout(req);
1059 
1060         desc = ptlrpc_prep_bulk_imp(req, op_data->op_npages, 1, BULK_PUT_SINK,
1061                                     MDS_BULK_PORTAL);
1062         if (desc == NULL) {
1063                 ptlrpc_request_free(req);
1064                 return -ENOMEM;
1065         }
1066 
1067         /* NB req now owns desc and will free it when it gets freed */
1068         for (i = 0; i < op_data->op_npages; i++)
1069                 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1070 
1071         mdc_readdir_pack(req, op_data->op_offset,
1072                          PAGE_CACHE_SIZE * op_data->op_npages,
1073                          &op_data->op_fid1, op_data->op_capa1);
1074 
1075         ptlrpc_request_set_replen(req);
1076         rc = ptlrpc_queue_wait(req);
1077         if (rc) {
1078                 ptlrpc_req_finished(req);
1079                 if (rc != -ETIMEDOUT)
1080                         return rc;
1081 
1082                 resends++;
1083                 if (!client_should_resend(resends, &exp->exp_obd->u.cli)) {
1084                         CERROR("too many resend retries, returning error\n");
1085                         return -EIO;
1086                 }
1087                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(resends), NULL, NULL, NULL);
1088                 l_wait_event(waitq, 0, &lwi);
1089 
1090                 goto restart_bulk;
1091         }
1092 
1093         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1094                                           req->rq_bulk->bd_nob_transferred);
1095         if (rc < 0) {
1096                 ptlrpc_req_finished(req);
1097                 return rc;
1098         }
1099 
1100         if (req->rq_bulk->bd_nob_transferred & ~LU_PAGE_MASK) {
1101                 CERROR("Unexpected # bytes transferred: %d (%ld expected)\n",
1102                         req->rq_bulk->bd_nob_transferred,
1103                         PAGE_CACHE_SIZE * op_data->op_npages);
1104                 ptlrpc_req_finished(req);
1105                 return -EPROTO;
1106         }
1107 
1108         *request = req;
1109         return 0;
1110 }
1111 
1112 static int mdc_statfs(const struct lu_env *env,
1113                       struct obd_export *exp, struct obd_statfs *osfs,
1114                       __u64 max_age, __u32 flags)
1115 {
1116         struct obd_device     *obd = class_exp2obd(exp);
1117         struct ptlrpc_request *req;
1118         struct obd_statfs     *msfs;
1119         struct obd_import     *imp = NULL;
1120         int                 rc;
1121 
1122         /*
1123          * Since the request might also come from lprocfs, so we need
1124          * sync this with client_disconnect_export Bug15684
1125          */
1126         down_read(&obd->u.cli.cl_sem);
1127         if (obd->u.cli.cl_import)
1128                 imp = class_import_get(obd->u.cli.cl_import);
1129         up_read(&obd->u.cli.cl_sem);
1130         if (!imp)
1131                 return -ENODEV;
1132 
1133         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_STATFS,
1134                                         LUSTRE_MDS_VERSION, MDS_STATFS);
1135         if (req == NULL)
1136                 GOTO(output, rc = -ENOMEM);
1137 
1138         ptlrpc_request_set_replen(req);
1139 
1140         if (flags & OBD_STATFS_NODELAY) {
1141                 /* procfs requests not want stay in wait for avoid deadlock */
1142                 req->rq_no_resend = 1;
1143                 req->rq_no_delay = 1;
1144         }
1145 
1146         rc = ptlrpc_queue_wait(req);
1147         if (rc) {
1148                 /* check connection error first */
1149                 if (imp->imp_connect_error)
1150                         rc = imp->imp_connect_error;
1151                 GOTO(out, rc);
1152         }
1153 
1154         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1155         if (msfs == NULL)
1156                 GOTO(out, rc = -EPROTO);
1157 
1158         *osfs = *msfs;
1159 out:
1160         ptlrpc_req_finished(req);
1161 output:
1162         class_import_put(imp);
1163         return rc;
1164 }
1165 
1166 static int mdc_ioc_fid2path(struct obd_export *exp, struct getinfo_fid2path *gf)
1167 {
1168         __u32 keylen, vallen;
1169         void *key;
1170         int rc;
1171 
1172         if (gf->gf_pathlen > PATH_MAX)
1173                 return -ENAMETOOLONG;
1174         if (gf->gf_pathlen < 2)
1175                 return -EOVERFLOW;
1176 
1177         /* Key is KEY_FID2PATH + getinfo_fid2path description */
1178         keylen = cfs_size_round(sizeof(KEY_FID2PATH)) + sizeof(*gf);
1179         OBD_ALLOC(key, keylen);
1180         if (key == NULL)
1181                 return -ENOMEM;
1182         memcpy(key, KEY_FID2PATH, sizeof(KEY_FID2PATH));
1183         memcpy(key + cfs_size_round(sizeof(KEY_FID2PATH)), gf, sizeof(*gf));
1184 
1185         CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n",
1186                PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno);
1187 
1188         if (!fid_is_sane(&gf->gf_fid))
1189                 GOTO(out, rc = -EINVAL);
1190 
1191         /* Val is struct getinfo_fid2path result plus path */
1192         vallen = sizeof(*gf) + gf->gf_pathlen;
1193 
1194         rc = obd_get_info(NULL, exp, keylen, key, &vallen, gf, NULL);
1195         if (rc != 0 && rc != -EREMOTE)
1196                 GOTO(out, rc);
1197 
1198         if (vallen <= sizeof(*gf))
1199                 GOTO(out, rc = -EPROTO);
1200         else if (vallen > sizeof(*gf) + gf->gf_pathlen)
1201                 GOTO(out, rc = -EOVERFLOW);
1202 
1203         CDEBUG(D_IOCTL, "path get "DFID" from "LPU64" #%d\n%s\n",
1204                PFID(&gf->gf_fid), gf->gf_recno, gf->gf_linkno, gf->gf_path);
1205 
1206 out:
1207         OBD_FREE(key, keylen);
1208         return rc;
1209 }
1210 
1211 static int mdc_ioc_hsm_progress(struct obd_export *exp,
1212                                 struct hsm_progress_kernel *hpk)
1213 {
1214         struct obd_import               *imp = class_exp2cliimp(exp);
1215         struct hsm_progress_kernel      *req_hpk;
1216         struct ptlrpc_request           *req;
1217         int                              rc;
1218 
1219         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_PROGRESS,
1220                                         LUSTRE_MDS_VERSION, MDS_HSM_PROGRESS);
1221         if (req == NULL)
1222                 GOTO(out, rc = -ENOMEM);
1223 
1224         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1225 
1226         /* Copy hsm_progress struct */
1227         req_hpk = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_PROGRESS);
1228         if (req_hpk == NULL)
1229                 GOTO(out, rc = -EPROTO);
1230 
1231         *req_hpk = *hpk;
1232         req_hpk->hpk_errval = lustre_errno_hton(hpk->hpk_errval);
1233 
1234         ptlrpc_request_set_replen(req);
1235 
1236         rc = mdc_queue_wait(req);
1237         GOTO(out, rc);
1238 out:
1239         ptlrpc_req_finished(req);
1240         return rc;
1241 }
1242 
1243 static int mdc_ioc_hsm_ct_register(struct obd_import *imp, __u32 archives)
1244 {
1245         __u32                   *archive_mask;
1246         struct ptlrpc_request   *req;
1247         int                      rc;
1248 
1249         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_REGISTER,
1250                                         LUSTRE_MDS_VERSION,
1251                                         MDS_HSM_CT_REGISTER);
1252         if (req == NULL)
1253                 GOTO(out, rc = -ENOMEM);
1254 
1255         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1256 
1257         /* Copy hsm_progress struct */
1258         archive_mask = req_capsule_client_get(&req->rq_pill,
1259                                               &RMF_MDS_HSM_ARCHIVE);
1260         if (archive_mask == NULL)
1261                 GOTO(out, rc = -EPROTO);
1262 
1263         *archive_mask = archives;
1264 
1265         ptlrpc_request_set_replen(req);
1266 
1267         rc = mdc_queue_wait(req);
1268         GOTO(out, rc);
1269 out:
1270         ptlrpc_req_finished(req);
1271         return rc;
1272 }
1273 
1274 static int mdc_ioc_hsm_current_action(struct obd_export *exp,
1275                                       struct md_op_data *op_data)
1276 {
1277         struct hsm_current_action       *hca = op_data->op_data;
1278         struct hsm_current_action       *req_hca;
1279         struct ptlrpc_request           *req;
1280         int                              rc;
1281 
1282         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1283                                    &RQF_MDS_HSM_ACTION);
1284         if (req == NULL)
1285                 return -ENOMEM;
1286 
1287         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1288 
1289         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_ACTION);
1290         if (rc) {
1291                 ptlrpc_request_free(req);
1292                 return rc;
1293         }
1294 
1295         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
1296                       OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0);
1297 
1298         ptlrpc_request_set_replen(req);
1299 
1300         rc = mdc_queue_wait(req);
1301         if (rc)
1302                 GOTO(out, rc);
1303 
1304         req_hca = req_capsule_server_get(&req->rq_pill,
1305                                          &RMF_MDS_HSM_CURRENT_ACTION);
1306         if (req_hca == NULL)
1307                 GOTO(out, rc = -EPROTO);
1308 
1309         *hca = *req_hca;
1310 
1311 out:
1312         ptlrpc_req_finished(req);
1313         return rc;
1314 }
1315 
1316 static int mdc_ioc_hsm_ct_unregister(struct obd_import *imp)
1317 {
1318         struct ptlrpc_request   *req;
1319         int                      rc;
1320 
1321         req = ptlrpc_request_alloc_pack(imp, &RQF_MDS_HSM_CT_UNREGISTER,
1322                                         LUSTRE_MDS_VERSION,
1323                                         MDS_HSM_CT_UNREGISTER);
1324         if (req == NULL)
1325                 GOTO(out, rc = -ENOMEM);
1326 
1327         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1328 
1329         ptlrpc_request_set_replen(req);
1330 
1331         rc = mdc_queue_wait(req);
1332         GOTO(out, rc);
1333 out:
1334         ptlrpc_req_finished(req);
1335         return rc;
1336 }
1337 
1338 static int mdc_ioc_hsm_state_get(struct obd_export *exp,
1339                                  struct md_op_data *op_data)
1340 {
1341         struct hsm_user_state   *hus = op_data->op_data;
1342         struct hsm_user_state   *req_hus;
1343         struct ptlrpc_request   *req;
1344         int                      rc;
1345 
1346         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1347                                    &RQF_MDS_HSM_STATE_GET);
1348         if (req == NULL)
1349                 return -ENOMEM;
1350 
1351         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1352 
1353         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_STATE_GET);
1354         if (rc != 0) {
1355                 ptlrpc_request_free(req);
1356                 return rc;
1357         }
1358 
1359         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
1360                       OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0);
1361 
1362         ptlrpc_request_set_replen(req);
1363 
1364         rc = mdc_queue_wait(req);
1365         if (rc)
1366                 GOTO(out, rc);
1367 
1368         req_hus = req_capsule_server_get(&req->rq_pill, &RMF_HSM_USER_STATE);
1369         if (req_hus == NULL)
1370                 GOTO(out, rc = -EPROTO);
1371 
1372         *hus = *req_hus;
1373 
1374 out:
1375         ptlrpc_req_finished(req);
1376         return rc;
1377 }
1378 
1379 static int mdc_ioc_hsm_state_set(struct obd_export *exp,
1380                                  struct md_op_data *op_data)
1381 {
1382         struct hsm_state_set    *hss = op_data->op_data;
1383         struct hsm_state_set    *req_hss;
1384         struct ptlrpc_request   *req;
1385         int                      rc;
1386 
1387         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1388                                    &RQF_MDS_HSM_STATE_SET);
1389         if (req == NULL)
1390                 return -ENOMEM;
1391 
1392         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1393 
1394         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_STATE_SET);
1395         if (rc) {
1396                 ptlrpc_request_free(req);
1397                 return rc;
1398         }
1399 
1400         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
1401                       OBD_MD_FLRMTPERM, 0, op_data->op_suppgids[0], 0);
1402 
1403         /* Copy states */
1404         req_hss = req_capsule_client_get(&req->rq_pill, &RMF_HSM_STATE_SET);
1405         if (req_hss == NULL)
1406                 GOTO(out, rc = -EPROTO);
1407         *req_hss = *hss;
1408 
1409         ptlrpc_request_set_replen(req);
1410 
1411         rc = mdc_queue_wait(req);
1412         GOTO(out, rc);
1413 
1414 out:
1415         ptlrpc_req_finished(req);
1416         return rc;
1417 }
1418 
1419 static int mdc_ioc_hsm_request(struct obd_export *exp,
1420                                struct hsm_user_request *hur)
1421 {
1422         struct obd_import       *imp = class_exp2cliimp(exp);
1423         struct ptlrpc_request   *req;
1424         struct hsm_request      *req_hr;
1425         struct hsm_user_item    *req_hui;
1426         char                    *req_opaque;
1427         int                      rc;
1428 
1429         req = ptlrpc_request_alloc(imp, &RQF_MDS_HSM_REQUEST);
1430         if (req == NULL)
1431                 GOTO(out, rc = -ENOMEM);
1432 
1433         req_capsule_set_size(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM, RCL_CLIENT,
1434                              hur->hur_request.hr_itemcount
1435                              * sizeof(struct hsm_user_item));
1436         req_capsule_set_size(&req->rq_pill, &RMF_GENERIC_DATA, RCL_CLIENT,
1437                              hur->hur_request.hr_data_len);
1438 
1439         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_HSM_REQUEST);
1440         if (rc) {
1441                 ptlrpc_request_free(req);
1442                 return rc;
1443         }
1444 
1445         mdc_pack_body(req, NULL, NULL, OBD_MD_FLRMTPERM, 0, 0, 0);
1446 
1447         /* Copy hsm_request struct */
1448         req_hr = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_REQUEST);
1449         if (req_hr == NULL)
1450                 GOTO(out, rc = -EPROTO);
1451         *req_hr = hur->hur_request;
1452 
1453         /* Copy hsm_user_item structs */
1454         req_hui = req_capsule_client_get(&req->rq_pill, &RMF_MDS_HSM_USER_ITEM);
1455         if (req_hui == NULL)
1456                 GOTO(out, rc = -EPROTO);
1457         memcpy(req_hui, hur->hur_user_item,
1458                hur->hur_request.hr_itemcount * sizeof(struct hsm_user_item));
1459 
1460         /* Copy opaque field */
1461         req_opaque = req_capsule_client_get(&req->rq_pill, &RMF_GENERIC_DATA);
1462         if (req_opaque == NULL)
1463                 GOTO(out, rc = -EPROTO);
1464         memcpy(req_opaque, hur_data(hur), hur->hur_request.hr_data_len);
1465 
1466         ptlrpc_request_set_replen(req);
1467 
1468         rc = mdc_queue_wait(req);
1469         GOTO(out, rc);
1470 
1471 out:
1472         ptlrpc_req_finished(req);
1473         return rc;
1474 }
1475 
1476 static struct kuc_hdr *changelog_kuc_hdr(char *buf, int len, int flags)
1477 {
1478         struct kuc_hdr *lh = (struct kuc_hdr *)buf;
1479 
1480         LASSERT(len <= KUC_CHANGELOG_MSG_MAXSIZE);
1481 
1482         lh->kuc_magic = KUC_MAGIC;
1483         lh->kuc_transport = KUC_TRANSPORT_CHANGELOG;
1484         lh->kuc_flags = flags;
1485         lh->kuc_msgtype = CL_RECORD;
1486         lh->kuc_msglen = len;
1487         return lh;
1488 }
1489 
1490 #define D_CHANGELOG 0
1491 
1492 struct changelog_show {
1493         __u64           cs_startrec;
1494         __u32           cs_flags;
1495         struct file     *cs_fp;
1496         char            *cs_buf;
1497         struct obd_device *cs_obd;
1498 };
1499 
1500 static int changelog_kkuc_cb(const struct lu_env *env, struct llog_handle *llh,
1501                              struct llog_rec_hdr *hdr, void *data)
1502 {
1503         struct changelog_show *cs = data;
1504         struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
1505         struct kuc_hdr *lh;
1506         int len, rc;
1507 
1508         if (rec->cr_hdr.lrh_type != CHANGELOG_REC) {
1509                 rc = -EINVAL;
1510                 CERROR("%s: not a changelog rec %x/%d: rc = %d\n",
1511                        cs->cs_obd->obd_name, rec->cr_hdr.lrh_type,
1512                        rec->cr.cr_type, rc);
1513                 return rc;
1514         }
1515 
1516         if (rec->cr.cr_index < cs->cs_startrec) {
1517                 /* Skip entries earlier than what we are interested in */
1518                 CDEBUG(D_CHANGELOG, "rec="LPU64" start="LPU64"\n",
1519                        rec->cr.cr_index, cs->cs_startrec);
1520                 return 0;
1521         }
1522 
1523         CDEBUG(D_CHANGELOG, LPU64" %02d%-5s "LPU64" 0x%x t="DFID" p="DFID
1524                 " %.*s\n", rec->cr.cr_index, rec->cr.cr_type,
1525                 changelog_type2str(rec->cr.cr_type), rec->cr.cr_time,
1526                 rec->cr.cr_flags & CLF_FLAGMASK,
1527                 PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid),
1528                 rec->cr.cr_namelen, changelog_rec_name(&rec->cr));
1529 
1530         len = sizeof(*lh) + changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;
1531 
1532         /* Set up the message */
1533         lh = changelog_kuc_hdr(cs->cs_buf, len, cs->cs_flags);
1534         memcpy(lh + 1, &rec->cr, len - sizeof(*lh));
1535 
1536         rc = libcfs_kkuc_msg_put(cs->cs_fp, lh);
1537         CDEBUG(D_CHANGELOG, "kucmsg fp %p len %d rc %d\n", cs->cs_fp, len,rc);
1538 
1539         return rc;
1540 }
1541 
1542 static int mdc_changelog_send_thread(void *csdata)
1543 {
1544         struct changelog_show *cs = csdata;
1545         struct llog_ctxt *ctxt = NULL;
1546         struct llog_handle *llh = NULL;
1547         struct kuc_hdr *kuch;
1548         int rc;
1549 
1550         CDEBUG(D_CHANGELOG, "changelog to fp=%p start "LPU64"\n",
1551                cs->cs_fp, cs->cs_startrec);
1552 
1553         OBD_ALLOC(cs->cs_buf, KUC_CHANGELOG_MSG_MAXSIZE);
1554         if (cs->cs_buf == NULL)
1555                 GOTO(out, rc = -ENOMEM);
1556 
1557         /* Set up the remote catalog handle */
1558         ctxt = llog_get_context(cs->cs_obd, LLOG_CHANGELOG_REPL_CTXT);
1559         if (ctxt == NULL)
1560                 GOTO(out, rc = -ENOENT);
1561         rc = llog_open(NULL, ctxt, &llh, NULL, CHANGELOG_CATALOG,
1562                        LLOG_OPEN_EXISTS);
1563         if (rc) {
1564                 CERROR("%s: fail to open changelog catalog: rc = %d\n",
1565                        cs->cs_obd->obd_name, rc);
1566                 GOTO(out, rc);
1567         }
1568         rc = llog_init_handle(NULL, llh, LLOG_F_IS_CAT, NULL);
1569         if (rc) {
1570                 CERROR("llog_init_handle failed %d\n", rc);
1571                 GOTO(out, rc);
1572         }
1573 
1574         rc = llog_cat_process(NULL, llh, changelog_kkuc_cb, cs, 0, 0);
1575 
1576         /* Send EOF no matter what our result */
1577         kuch = changelog_kuc_hdr(cs->cs_buf, sizeof(*kuch), cs->cs_flags);
1578         if (kuch) {
1579                 kuch->kuc_msgtype = CL_EOF;
1580                 libcfs_kkuc_msg_put(cs->cs_fp, kuch);
1581         }
1582 
1583 out:
1584         fput(cs->cs_fp);
1585         if (llh)
1586                 llog_cat_close(NULL, llh);
1587         if (ctxt)
1588                 llog_ctxt_put(ctxt);
1589         if (cs->cs_buf)
1590                 OBD_FREE(cs->cs_buf, KUC_CHANGELOG_MSG_MAXSIZE);
1591         OBD_FREE_PTR(cs);
1592         return rc;
1593 }
1594 
1595 static int mdc_ioc_changelog_send(struct obd_device *obd,
1596                                   struct ioc_changelog *icc)
1597 {
1598         struct changelog_show *cs;
1599         int rc;
1600 
1601         /* Freed in mdc_changelog_send_thread */
1602         OBD_ALLOC_PTR(cs);
1603         if (!cs)
1604                 return -ENOMEM;
1605 
1606         cs->cs_obd = obd;
1607         cs->cs_startrec = icc->icc_recno;
1608         /* matching fput in mdc_changelog_send_thread */
1609         cs->cs_fp = fget(icc->icc_id);
1610         cs->cs_flags = icc->icc_flags;
1611 
1612         /*
1613          * New thread because we should return to user app before
1614          * writing into our pipe
1615          */
1616         rc = PTR_ERR(kthread_run(mdc_changelog_send_thread, cs,
1617                                  "mdc_clg_send_thread"));
1618         if (!IS_ERR_VALUE(rc)) {
1619                 CDEBUG(D_CHANGELOG, "start changelog thread\n");
1620                 return 0;
1621         }
1622 
1623         CERROR("Failed to start changelog thread: %d\n", rc);
1624         OBD_FREE_PTR(cs);
1625         return rc;
1626 }
1627 
1628 static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
1629                                 struct lustre_kernelcomm *lk);
1630 
1631 static int mdc_quotacheck(struct obd_device *unused, struct obd_export *exp,
1632                           struct obd_quotactl *oqctl)
1633 {
1634         struct client_obd       *cli = &exp->exp_obd->u.cli;
1635         struct ptlrpc_request   *req;
1636         struct obd_quotactl     *body;
1637         int                   rc;
1638 
1639         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
1640                                         &RQF_MDS_QUOTACHECK, LUSTRE_MDS_VERSION,
1641                                         MDS_QUOTACHECK);
1642         if (req == NULL)
1643                 return -ENOMEM;
1644 
1645         body = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1646         *body = *oqctl;
1647 
1648         ptlrpc_request_set_replen(req);
1649 
1650         /* the next poll will find -ENODATA, that means quotacheck is
1651          * going on */
1652         cli->cl_qchk_stat = -ENODATA;
1653         rc = ptlrpc_queue_wait(req);
1654         if (rc)
1655                 cli->cl_qchk_stat = rc;
1656         ptlrpc_req_finished(req);
1657         return rc;
1658 }
1659 
1660 static int mdc_quota_poll_check(struct obd_export *exp,
1661                                 struct if_quotacheck *qchk)
1662 {
1663         struct client_obd *cli = &exp->exp_obd->u.cli;
1664         int rc;
1665 
1666         qchk->obd_uuid = cli->cl_target_uuid;
1667         memcpy(qchk->obd_type, LUSTRE_MDS_NAME, strlen(LUSTRE_MDS_NAME));
1668 
1669         rc = cli->cl_qchk_stat;
1670         /* the client is not the previous one */
1671         if (rc == CL_NOT_QUOTACHECKED)
1672                 rc = -EINTR;
1673         return rc;
1674 }
1675 
1676 static int mdc_quotactl(struct obd_device *unused, struct obd_export *exp,
1677                         struct obd_quotactl *oqctl)
1678 {
1679         struct ptlrpc_request   *req;
1680         struct obd_quotactl     *oqc;
1681         int                   rc;
1682 
1683         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
1684                                         &RQF_MDS_QUOTACTL, LUSTRE_MDS_VERSION,
1685                                         MDS_QUOTACTL);
1686         if (req == NULL)
1687                 return -ENOMEM;
1688 
1689         oqc = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1690         *oqc = *oqctl;
1691 
1692         ptlrpc_request_set_replen(req);
1693         ptlrpc_at_set_req_timeout(req);
1694         req->rq_no_resend = 1;
1695 
1696         rc = ptlrpc_queue_wait(req);
1697         if (rc)
1698                 CERROR("ptlrpc_queue_wait failed, rc: %d\n", rc);
1699 
1700         if (req->rq_repmsg) {
1701                 oqc = req_capsule_server_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1702                 if (oqc) {
1703                         *oqctl = *oqc;
1704                 } else if (!rc) {
1705                         CERROR ("Can't unpack obd_quotactl\n");
1706                         rc = -EPROTO;
1707                 }
1708         } else if (!rc) {
1709                 CERROR("Can't unpack obd_quotactl\n");
1710                 rc = -EPROTO;
1711         }
1712         ptlrpc_req_finished(req);
1713 
1714         return rc;
1715 }
1716 
1717 static int mdc_ioc_swap_layouts(struct obd_export *exp,
1718                                 struct md_op_data *op_data)
1719 {
1720         LIST_HEAD(cancels);
1721         struct ptlrpc_request   *req;
1722         int                      rc, count;
1723         struct mdc_swap_layouts *msl, *payload;
1724 
1725         msl = op_data->op_data;
1726 
1727         /* When the MDT will get the MDS_SWAP_LAYOUTS RPC the
1728          * first thing it will do is to cancel the 2 layout
1729          * locks hold by this client.
1730          * So the client must cancel its layout locks on the 2 fids
1731          * with the request RPC to avoid extra RPC round trips
1732          */
1733         count = mdc_resource_get_unused(exp, &op_data->op_fid1, &cancels,
1734                                         LCK_CR, MDS_INODELOCK_LAYOUT);
1735         count += mdc_resource_get_unused(exp, &op_data->op_fid2, &cancels,
1736                                          LCK_CR, MDS_INODELOCK_LAYOUT);
1737 
1738         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
1739                                    &RQF_MDS_SWAP_LAYOUTS);
1740         if (req == NULL) {
1741                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
1742                 return -ENOMEM;
1743         }
1744 
1745         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
1746         mdc_set_capa_size(req, &RMF_CAPA2, op_data->op_capa2);
1747 
1748         rc = mdc_prep_elc_req(exp, req, MDS_SWAP_LAYOUTS, &cancels, count);
1749         if (rc) {
1750                 ptlrpc_request_free(req);
1751                 return rc;
1752         }
1753 
1754         mdc_swap_layouts_pack(req, op_data);
1755 
1756         payload = req_capsule_client_get(&req->rq_pill, &RMF_SWAP_LAYOUTS);
1757         LASSERT(payload);
1758 
1759         *payload = *msl;
1760 
1761         ptlrpc_request_set_replen(req);
1762 
1763         rc = ptlrpc_queue_wait(req);
1764         if (rc)
1765                 GOTO(out, rc);
1766 
1767 out:
1768         ptlrpc_req_finished(req);
1769         return rc;
1770 }
1771 
1772 static int mdc_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1773                          void *karg, void *uarg)
1774 {
1775         struct obd_device *obd = exp->exp_obd;
1776         struct obd_ioctl_data *data = karg;
1777         struct obd_import *imp = obd->u.cli.cl_import;
1778         struct llog_ctxt *ctxt;
1779         int rc;
1780 
1781         if (!try_module_get(THIS_MODULE)) {
1782                 CERROR("Can't get module. Is it alive?");
1783                 return -EINVAL;
1784         }
1785         switch (cmd) {
1786         case OBD_IOC_CHANGELOG_SEND:
1787                 rc = mdc_ioc_changelog_send(obd, karg);
1788                 GOTO(out, rc);
1789         case OBD_IOC_CHANGELOG_CLEAR: {
1790                 struct ioc_changelog *icc = karg;
1791                 struct changelog_setinfo cs =
1792                         {.cs_recno = icc->icc_recno, .cs_id = icc->icc_id};
1793                 rc = obd_set_info_async(NULL, exp, strlen(KEY_CHANGELOG_CLEAR),
1794                                         KEY_CHANGELOG_CLEAR, sizeof(cs), &cs,
1795                                         NULL);
1796                 GOTO(out, rc);
1797         }
1798         case OBD_IOC_FID2PATH:
1799                 rc = mdc_ioc_fid2path(exp, karg);
1800                 GOTO(out, rc);
1801         case LL_IOC_HSM_CT_START:
1802                 rc = mdc_ioc_hsm_ct_start(exp, karg);
1803                 /* ignore if it was already registered on this MDS. */
1804                 if (rc == -EEXIST)
1805                         rc = 0;
1806                 GOTO(out, rc);
1807         case LL_IOC_HSM_PROGRESS:
1808                 rc = mdc_ioc_hsm_progress(exp, karg);
1809                 GOTO(out, rc);
1810         case LL_IOC_HSM_STATE_GET:
1811                 rc = mdc_ioc_hsm_state_get(exp, karg);
1812                 GOTO(out, rc);
1813         case LL_IOC_HSM_STATE_SET:
1814                 rc = mdc_ioc_hsm_state_set(exp, karg);
1815                 GOTO(out, rc);
1816         case LL_IOC_HSM_ACTION:
1817                 rc = mdc_ioc_hsm_current_action(exp, karg);
1818                 GOTO(out, rc);
1819         case LL_IOC_HSM_REQUEST:
1820                 rc = mdc_ioc_hsm_request(exp, karg);
1821                 GOTO(out, rc);
1822         case OBD_IOC_CLIENT_RECOVER:
1823                 rc = ptlrpc_recover_import(imp, data->ioc_inlbuf1, 0);
1824                 if (rc < 0)
1825                         GOTO(out, rc);
1826                 GOTO(out, rc = 0);
1827         case IOC_OSC_SET_ACTIVE:
1828                 rc = ptlrpc_set_import_active(imp, data->ioc_offset);
1829                 GOTO(out, rc);
1830         case OBD_IOC_PARSE: {
1831                 ctxt = llog_get_context(exp->exp_obd, LLOG_CONFIG_REPL_CTXT);
1832                 rc = class_config_parse_llog(NULL, ctxt, data->ioc_inlbuf1,
1833                                              NULL);
1834                 llog_ctxt_put(ctxt);
1835                 GOTO(out, rc);
1836         }
1837         case OBD_IOC_LLOG_INFO:
1838         case OBD_IOC_LLOG_PRINT: {
1839                 ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
1840                 rc = llog_ioctl(NULL, ctxt, cmd, data);
1841                 llog_ctxt_put(ctxt);
1842                 GOTO(out, rc);
1843         }
1844         case OBD_IOC_POLL_QUOTACHECK:
1845                 rc = mdc_quota_poll_check(exp, (struct if_quotacheck *)karg);
1846                 GOTO(out, rc);
1847         case OBD_IOC_PING_TARGET:
1848                 rc = ptlrpc_obd_ping(obd);
1849                 GOTO(out, rc);
1850         /*
1851          * Normally IOC_OBD_STATFS, OBD_IOC_QUOTACTL iocontrol are handled by
1852          * LMV instead of MDC. But when the cluster is upgraded from 1.8,
1853          * there'd be no LMV layer thus we might be called here. Eventually
1854          * this code should be removed.
1855          * bz20731, LU-592.
1856          */
1857         case IOC_OBD_STATFS: {
1858                 struct obd_statfs stat_buf = {0};
1859 
1860                 if (*((__u32 *) data->ioc_inlbuf2) != 0)
1861                         GOTO(out, rc = -ENODEV);
1862 
1863                 /* copy UUID */
1864                 if (copy_to_user(data->ioc_pbuf2, obd2cli_tgt(obd),
1865                                      min((int) data->ioc_plen2,
1866                                          (int) sizeof(struct obd_uuid))))
1867                         GOTO(out, rc = -EFAULT);
1868 
1869                 rc = mdc_statfs(NULL, obd->obd_self_export, &stat_buf,
1870                                 cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS),
1871                                 0);
1872                 if (rc != 0)
1873                         GOTO(out, rc);
1874 
1875                 if (copy_to_user(data->ioc_pbuf1, &stat_buf,
1876                                      min((int) data->ioc_plen1,
1877                                          (int) sizeof(stat_buf))))
1878                         GOTO(out, rc = -EFAULT);
1879 
1880                 GOTO(out, rc = 0);
1881         }
1882         case OBD_IOC_QUOTACTL: {
1883                 struct if_quotactl *qctl = karg;
1884                 struct obd_quotactl *oqctl;
1885 
1886                 OBD_ALLOC_PTR(oqctl);
1887                 if (oqctl == NULL)
1888                         GOTO(out, rc = -ENOMEM);
1889 
1890                 QCTL_COPY(oqctl, qctl);
1891                 rc = obd_quotactl(exp, oqctl);
1892                 if (rc == 0) {
1893                         QCTL_COPY(qctl, oqctl);
1894                         qctl->qc_valid = QC_MDTIDX;
1895                         qctl->obd_uuid = obd->u.cli.cl_target_uuid;
1896                 }
1897 
1898                 OBD_FREE_PTR(oqctl);
1899                 GOTO(out, rc);
1900         }
1901         case LL_IOC_GET_CONNECT_FLAGS:
1902                 if (copy_to_user(uarg, exp_connect_flags_ptr(exp),
1903                                  sizeof(*exp_connect_flags_ptr(exp))))
1904                         GOTO(out, rc = -EFAULT);
1905 
1906                 GOTO(out, rc = 0);
1907         case LL_IOC_LOV_SWAP_LAYOUTS:
1908                 rc = mdc_ioc_swap_layouts(exp, karg);
1909                 GOTO(out, rc);
1910         default:
1911                 CERROR("unrecognised ioctl: cmd = %#x\n", cmd);
1912                 GOTO(out, rc = -ENOTTY);
1913         }
1914 out:
1915         module_put(THIS_MODULE);
1916 
1917         return rc;
1918 }
1919 
1920 int mdc_get_info_rpc(struct obd_export *exp,
1921                      obd_count keylen, void *key,
1922                      int vallen, void *val)
1923 {
1924         struct obd_import      *imp = class_exp2cliimp(exp);
1925         struct ptlrpc_request  *req;
1926         char               *tmp;
1927         int                  rc = -EINVAL;
1928 
1929         req = ptlrpc_request_alloc(imp, &RQF_MDS_GET_INFO);
1930         if (req == NULL)
1931                 return -ENOMEM;
1932 
1933         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_KEY,
1934                              RCL_CLIENT, keylen);
1935         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VALLEN,
1936                              RCL_CLIENT, sizeof(__u32));
1937 
1938         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GET_INFO);
1939         if (rc) {
1940                 ptlrpc_request_free(req);
1941                 return rc;
1942         }
1943 
1944         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_KEY);
1945         memcpy(tmp, key, keylen);
1946         tmp = req_capsule_client_get(&req->rq_pill, &RMF_GETINFO_VALLEN);
1947         memcpy(tmp, &vallen, sizeof(__u32));
1948 
1949         req_capsule_set_size(&req->rq_pill, &RMF_GETINFO_VAL,
1950                              RCL_SERVER, vallen);
1951         ptlrpc_request_set_replen(req);
1952 
1953         rc = ptlrpc_queue_wait(req);
1954         /* -EREMOTE means the get_info result is partial, and it needs to
1955          * continue on another MDT, see fid2path part in lmv_iocontrol */
1956         if (rc == 0 || rc == -EREMOTE) {
1957                 tmp = req_capsule_server_get(&req->rq_pill, &RMF_GETINFO_VAL);
1958                 memcpy(val, tmp, vallen);
1959                 if (ptlrpc_rep_need_swab(req)) {
1960                         if (KEY_IS(KEY_FID2PATH))
1961                                 lustre_swab_fid2path(val);
1962                 }
1963         }
1964         ptlrpc_req_finished(req);
1965 
1966         return rc;
1967 }
1968 
1969 static void lustre_swab_hai(struct hsm_action_item *h)
1970 {
1971         __swab32s(&h->hai_len);
1972         __swab32s(&h->hai_action);
1973         lustre_swab_lu_fid(&h->hai_fid);
1974         lustre_swab_lu_fid(&h->hai_dfid);
1975         __swab64s(&h->hai_cookie);
1976         __swab64s(&h->hai_extent.offset);
1977         __swab64s(&h->hai_extent.length);
1978         __swab64s(&h->hai_gid);
1979 }
1980 
1981 static void lustre_swab_hal(struct hsm_action_list *h)
1982 {
1983         struct hsm_action_item  *hai;
1984         int                      i;
1985 
1986         __swab32s(&h->hal_version);
1987         __swab32s(&h->hal_count);
1988         __swab32s(&h->hal_archive_id);
1989         __swab64s(&h->hal_flags);
1990         hai = hai_zero(h);
1991         for (i = 0; i < h->hal_count; i++, hai = hai_next(hai))
1992                 lustre_swab_hai(hai);
1993 }
1994 
1995 static void lustre_swab_kuch(struct kuc_hdr *l)
1996 {
1997         __swab16s(&l->kuc_magic);
1998         /* __u8 l->kuc_transport */
1999         __swab16s(&l->kuc_msgtype);
2000         __swab16s(&l->kuc_msglen);
2001 }
2002 
2003 static int mdc_ioc_hsm_ct_start(struct obd_export *exp,
2004                                 struct lustre_kernelcomm *lk)
2005 {
2006         struct obd_import  *imp = class_exp2cliimp(exp);
2007         __u32               archive = lk->lk_data;
2008         int                 rc = 0;
2009 
2010         if (lk->lk_group != KUC_GRP_HSM) {
2011                 CERROR("Bad copytool group %d\n", lk->lk_group);
2012                 return -EINVAL;
2013         }
2014 
2015         CDEBUG(D_HSM, "CT start r%d w%d u%d g%d f%#x\n", lk->lk_rfd, lk->lk_wfd,
2016                lk->lk_uid, lk->lk_group, lk->lk_flags);
2017 
2018         if (lk->lk_flags & LK_FLG_STOP) {
2019                 /* Unregister with the coordinator */
2020                 rc = mdc_ioc_hsm_ct_unregister(imp);
2021         } else {
2022                 rc = mdc_ioc_hsm_ct_register(imp, archive);
2023         }
2024 
2025         return rc;
2026 }
2027 
2028 /**
2029  * Send a message to any listening copytools
2030  * @param val KUC message (kuc_hdr + hsm_action_list)
2031  * @param len total length of message
2032  */
2033 static int mdc_hsm_copytool_send(int len, void *val)
2034 {
2035         struct kuc_hdr          *lh = (struct kuc_hdr *)val;
2036         struct hsm_action_list  *hal = (struct hsm_action_list *)(lh + 1);
2037         int                      rc;
2038 
2039         if (len < sizeof(*lh) + sizeof(*hal)) {
2040                 CERROR("Short HSM message %d < %d\n", len,
2041                        (int) (sizeof(*lh) + sizeof(*hal)));
2042                 return -EPROTO;
2043         }
2044         if (lh->kuc_magic == __swab16(KUC_MAGIC)) {
2045                 lustre_swab_kuch(lh);
2046                 lustre_swab_hal(hal);
2047         } else if (lh->kuc_magic != KUC_MAGIC) {
2048                 CERROR("Bad magic %x!=%x\n", lh->kuc_magic, KUC_MAGIC);
2049                 return -EPROTO;
2050         }
2051 
2052         CDEBUG(D_HSM, " Received message mg=%x t=%d m=%d l=%d actions=%d "
2053                "on %s\n",
2054                lh->kuc_magic, lh->kuc_transport, lh->kuc_msgtype,
2055                lh->kuc_msglen, hal->hal_count, hal->hal_fsname);
2056 
2057         /* Broadcast to HSM listeners */
2058         rc = libcfs_kkuc_group_put(KUC_GRP_HSM, lh);
2059 
2060         return rc;
2061 }
2062 
2063 /**
2064  * callback function passed to kuc for re-registering each HSM copytool
2065  * running on MDC, after MDT shutdown/recovery.
2066  * @param data archive id served by the copytool
2067  * @param cb_arg callback argument (obd_import)
2068  */
2069 static int mdc_hsm_ct_reregister(__u32 data, void *cb_arg)
2070 {
2071         struct obd_import       *imp = (struct obd_import *)cb_arg;
2072         __u32                    archive = data;
2073         int                      rc;
2074 
2075         CDEBUG(D_HA, "recover copytool registration to MDT (archive=%#x)\n",
2076                archive);
2077         rc = mdc_ioc_hsm_ct_register(imp, archive);
2078 
2079         /* ignore error if the copytool is already registered */
2080         return ((rc != 0) && (rc != -EEXIST)) ? rc : 0;
2081 }
2082 
2083 /**
2084  * Re-establish all kuc contexts with MDT
2085  * after MDT shutdown/recovery.
2086  */
2087 static int mdc_kuc_reregister(struct obd_import *imp)
2088 {
2089         /* re-register HSM agents */
2090         return libcfs_kkuc_group_foreach(KUC_GRP_HSM, mdc_hsm_ct_reregister,
2091                                          (void *)imp);
2092 }
2093 
2094 int mdc_set_info_async(const struct lu_env *env,
2095                        struct obd_export *exp,
2096                        obd_count keylen, void *key,
2097                        obd_count vallen, void *val,
2098                        struct ptlrpc_request_set *set)
2099 {
2100         struct obd_import       *imp = class_exp2cliimp(exp);
2101         int                      rc;
2102 
2103         if (KEY_IS(KEY_READ_ONLY)) {
2104                 if (vallen != sizeof(int))
2105                         return -EINVAL;
2106 
2107                 spin_lock(&imp->imp_lock);
2108                 if (*((int *)val)) {
2109                         imp->imp_connect_flags_orig |= OBD_CONNECT_RDONLY;
2110                         imp->imp_connect_data.ocd_connect_flags |=
2111                                                         OBD_CONNECT_RDONLY;
2112                 } else {
2113                         imp->imp_connect_flags_orig &= ~OBD_CONNECT_RDONLY;
2114                         imp->imp_connect_data.ocd_connect_flags &=
2115                                                         ~OBD_CONNECT_RDONLY;
2116                 }
2117                 spin_unlock(&imp->imp_lock);
2118 
2119                 rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION,
2120                                        keylen, key, vallen, val, set);
2121                 return rc;
2122         }
2123         if (KEY_IS(KEY_SPTLRPC_CONF)) {
2124                 sptlrpc_conf_client_adapt(exp->exp_obd);
2125                 return 0;
2126         }
2127         if (KEY_IS(KEY_FLUSH_CTX)) {
2128                 sptlrpc_import_flush_my_ctx(imp);
2129                 return 0;
2130         }
2131         if (KEY_IS(KEY_CHANGELOG_CLEAR)) {
2132                 rc = do_set_info_async(imp, MDS_SET_INFO, LUSTRE_MDS_VERSION,
2133                                        keylen, key, vallen, val, set);
2134                 return rc;
2135         }
2136         if (KEY_IS(KEY_HSM_COPYTOOL_SEND)) {
2137                 rc = mdc_hsm_copytool_send(vallen, val);
2138                 return rc;
2139         }
2140 
2141         CERROR("Unknown key %s\n", (char *)key);
2142         return -EINVAL;
2143 }
2144 
2145 int mdc_get_info(const struct lu_env *env, struct obd_export *exp,
2146                  __u32 keylen, void *key, __u32 *vallen, void *val,
2147                  struct lov_stripe_md *lsm)
2148 {
2149         int rc = -EINVAL;
2150 
2151         if (KEY_IS(KEY_MAX_EASIZE)) {
2152                 int mdsize, *max_easize;
2153 
2154                 if (*vallen != sizeof(int))
2155                         return -EINVAL;
2156                 mdsize = *(int*)val;
2157                 if (mdsize > exp->exp_obd->u.cli.cl_max_mds_easize)
2158                         exp->exp_obd->u.cli.cl_max_mds_easize = mdsize;
2159                 max_easize = val;
2160                 *max_easize = exp->exp_obd->u.cli.cl_max_mds_easize;
2161                 return 0;
2162         } else if (KEY_IS(KEY_CONN_DATA)) {
2163                 struct obd_import *imp = class_exp2cliimp(exp);
2164                 struct obd_connect_data *data = val;
2165 
2166                 if (*vallen != sizeof(*data))
2167                         return -EINVAL;
2168 
2169                 *data = imp->imp_connect_data;
2170                 return 0;
2171         } else if (KEY_IS(KEY_TGT_COUNT)) {
2172                 *((int *)val) = 1;
2173                 return 0;
2174         }
2175 
2176         rc = mdc_get_info_rpc(exp, keylen, key, *vallen, val);
2177 
2178         return rc;
2179 }
2180 
2181 static int mdc_pin(struct obd_export *exp, const struct lu_fid *fid,
2182                    struct obd_capa *oc, struct obd_client_handle *handle,
2183                    int flags)
2184 {
2185         struct ptlrpc_request *req;
2186         struct mdt_body       *body;
2187         int                 rc;
2188 
2189         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_PIN);
2190         if (req == NULL)
2191                 return -ENOMEM;
2192 
2193         mdc_set_capa_size(req, &RMF_CAPA1, oc);
2194 
2195         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_PIN);
2196         if (rc) {
2197                 ptlrpc_request_free(req);
2198                 return rc;
2199         }
2200 
2201         mdc_pack_body(req, fid, oc, 0, 0, -1, flags);
2202 
2203         ptlrpc_request_set_replen(req);
2204 
2205         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2206         rc = ptlrpc_queue_wait(req);
2207         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2208         if (rc) {
2209                 CERROR("Pin failed: %d\n", rc);
2210                 GOTO(err_out, rc);
2211         }
2212 
2213         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
2214         if (body == NULL)
2215                 GOTO(err_out, rc = -EPROTO);
2216 
2217         handle->och_fh = body->handle;
2218         handle->och_magic = OBD_CLIENT_HANDLE_MAGIC;
2219 
2220         handle->och_mod = obd_mod_alloc();
2221         if (handle->och_mod == NULL) {
2222                 DEBUG_REQ(D_ERROR, req, "can't allocate md_open_data");
2223                 GOTO(err_out, rc = -ENOMEM);
2224         }
2225         handle->och_mod->mod_open_req = req; /* will be dropped by unpin */
2226 
2227         return 0;
2228 
2229 err_out:
2230         ptlrpc_req_finished(req);
2231         return rc;
2232 }
2233 
2234 static int mdc_unpin(struct obd_export *exp, struct obd_client_handle *handle,
2235                      int flag)
2236 {
2237         struct ptlrpc_request *req;
2238         struct mdt_body       *body;
2239         int                 rc;
2240 
2241         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_UNPIN,
2242                                         LUSTRE_MDS_VERSION, MDS_UNPIN);
2243         if (req == NULL)
2244                 return -ENOMEM;
2245 
2246         body = req_capsule_client_get(&req->rq_pill, &RMF_MDT_BODY);
2247         body->handle = handle->och_fh;
2248         body->flags = flag;
2249 
2250         ptlrpc_request_set_replen(req);
2251 
2252         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2253         rc = ptlrpc_queue_wait(req);
2254         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
2255 
2256         if (rc != 0)
2257                 CERROR("Unpin failed: %d\n", rc);
2258 
2259         ptlrpc_req_finished(req);
2260         ptlrpc_req_finished(handle->och_mod->mod_open_req);
2261 
2262         obd_mod_put(handle->och_mod);
2263         return rc;
2264 }
2265 
2266 int mdc_sync(struct obd_export *exp, const struct lu_fid *fid,
2267              struct obd_capa *oc, struct ptlrpc_request **request)
2268 {
2269         struct ptlrpc_request *req;
2270         int                 rc;
2271 
2272         *request = NULL;
2273         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_SYNC);
2274         if (req == NULL)
2275                 return -ENOMEM;
2276 
2277         mdc_set_capa_size(req, &RMF_CAPA1, oc);
2278 
2279         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SYNC);
2280         if (rc) {
2281                 ptlrpc_request_free(req);
2282                 return rc;
2283         }
2284 
2285         mdc_pack_body(req, fid, oc, 0, 0, -1, 0);
2286 
2287         ptlrpc_request_set_replen(req);
2288 
2289         rc = ptlrpc_queue_wait(req);
2290         if (rc)
2291                 ptlrpc_req_finished(req);
2292         else
2293                 *request = req;
2294         return rc;
2295 }
2296 
2297 static int mdc_import_event(struct obd_device *obd, struct obd_import *imp,
2298                             enum obd_import_event event)
2299 {
2300         int rc = 0;
2301 
2302         LASSERT(imp->imp_obd == obd);
2303 
2304         switch (event) {
2305         case IMP_EVENT_DISCON: {
2306 #if 0
2307                 /* XXX Pass event up to OBDs stack. used only for FLD now */
2308                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_DISCON, NULL);
2309 #endif
2310                 break;
2311         }
2312         case IMP_EVENT_INACTIVE: {
2313                 struct client_obd *cli = &obd->u.cli;
2314                 /*
2315                  * Flush current sequence to make client obtain new one
2316                  * from server in case of disconnect/reconnect.
2317                  */
2318                 if (cli->cl_seq != NULL)
2319                         seq_client_flush(cli->cl_seq);
2320 
2321                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_INACTIVE, NULL);
2322                 break;
2323         }
2324         case IMP_EVENT_INVALIDATE: {
2325                 struct ldlm_namespace *ns = obd->obd_namespace;
2326 
2327                 ldlm_namespace_cleanup(ns, LDLM_FL_LOCAL_ONLY);
2328 
2329                 break;
2330         }
2331         case IMP_EVENT_ACTIVE:
2332                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_ACTIVE, NULL);
2333                 /* redo the kuc registration after reconnecting */
2334                 if (rc == 0)
2335                         rc = mdc_kuc_reregister(imp);
2336                 break;
2337         case IMP_EVENT_OCD:
2338                 rc = obd_notify_observer(obd, obd, OBD_NOTIFY_OCD, NULL);
2339                 break;
2340         case IMP_EVENT_DEACTIVATE:
2341         case IMP_EVENT_ACTIVATE:
2342                 break;
2343         default:
2344                 CERROR("Unknown import event %x\n", event);
2345                 LBUG();
2346         }
2347         return rc;
2348 }
2349 
2350 int mdc_fid_alloc(struct obd_export *exp, struct lu_fid *fid,
2351                   struct md_op_data *op_data)
2352 {
2353         struct client_obd *cli = &exp->exp_obd->u.cli;
2354         struct lu_client_seq *seq = cli->cl_seq;
2355 
2356         return seq_client_alloc_fid(NULL, seq, fid);
2357 }
2358 
2359 struct obd_uuid *mdc_get_uuid(struct obd_export *exp) {
2360         struct client_obd *cli = &exp->exp_obd->u.cli;
2361         return &cli->cl_target_uuid;
2362 }
2363 
2364 /**
2365  * Determine whether the lock can be canceled before replaying it during
2366  * recovery, non zero value will be return if the lock can be canceled,
2367  * or zero returned for not
2368  */
2369 static int mdc_cancel_for_recovery(struct ldlm_lock *lock)
2370 {
2371         if (lock->l_resource->lr_type != LDLM_IBITS)
2372                 return 0;
2373 
2374         /* FIXME: if we ever get into a situation where there are too many
2375          * opened files with open locks on a single node, then we really
2376          * should replay these open locks to reget it */
2377         if (lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_OPEN)
2378                 return 0;
2379 
2380         return 1;
2381 }
2382 
2383 static int mdc_resource_inode_free(struct ldlm_resource *res)
2384 {
2385         if (res->lr_lvb_inode)
2386                 res->lr_lvb_inode = NULL;
2387 
2388         return 0;
2389 }
2390 
2391 struct ldlm_valblock_ops inode_lvbo = {
2392         .lvbo_free = mdc_resource_inode_free,
2393 };
2394 
2395 static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg)
2396 {
2397         struct client_obd *cli = &obd->u.cli;
2398         struct lprocfs_static_vars lvars = { 0 };
2399         int rc;
2400 
2401         OBD_ALLOC(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
2402         if (!cli->cl_rpc_lock)
2403                 return -ENOMEM;
2404         mdc_init_rpc_lock(cli->cl_rpc_lock);
2405 
2406         ptlrpcd_addref();
2407 
2408         OBD_ALLOC(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
2409         if (!cli->cl_close_lock)
2410                 GOTO(err_rpc_lock, rc = -ENOMEM);
2411         mdc_init_rpc_lock(cli->cl_close_lock);
2412 
2413         rc = client_obd_setup(obd, cfg);
2414         if (rc)
2415                 GOTO(err_close_lock, rc);
2416         lprocfs_mdc_init_vars(&lvars);
2417         lprocfs_obd_setup(obd, lvars.obd_vars);
2418         sptlrpc_lprocfs_cliobd_attach(obd);
2419         ptlrpc_lprocfs_register_obd(obd);
2420 
2421         ns_register_cancel(obd->obd_namespace, mdc_cancel_for_recovery);
2422 
2423         obd->obd_namespace->ns_lvbo = &inode_lvbo;
2424 
2425         rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL);
2426         if (rc) {
2427                 mdc_cleanup(obd);
2428                 CERROR("failed to setup llogging subsystems\n");
2429         }
2430 
2431         return rc;
2432 
2433 err_close_lock:
2434         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
2435 err_rpc_lock:
2436         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
2437         ptlrpcd_decref();
2438         return rc;
2439 }
2440 
2441 /* Initialize the default and maximum LOV EA and cookie sizes.  This allows
2442  * us to make MDS RPCs with large enough reply buffers to hold the
2443  * maximum-sized (= maximum striped) EA and cookie without having to
2444  * calculate this (via a call into the LOV + OSCs) each time we make an RPC. */
2445 static int mdc_init_ea_size(struct obd_export *exp, int easize,
2446                      int def_easize, int cookiesize)
2447 {
2448         struct obd_device *obd = exp->exp_obd;
2449         struct client_obd *cli = &obd->u.cli;
2450 
2451         if (cli->cl_max_mds_easize < easize)
2452                 cli->cl_max_mds_easize = easize;
2453 
2454         if (cli->cl_default_mds_easize < def_easize)
2455                 cli->cl_default_mds_easize = def_easize;
2456 
2457         if (cli->cl_max_mds_cookiesize < cookiesize)
2458                 cli->cl_max_mds_cookiesize = cookiesize;
2459 
2460         return 0;
2461 }
2462 
2463 static int mdc_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
2464 {
2465         int rc = 0;
2466 
2467         switch (stage) {
2468         case OBD_CLEANUP_EARLY:
2469                 break;
2470         case OBD_CLEANUP_EXPORTS:
2471                 /* Failsafe, ok if racy */
2472                 if (obd->obd_type->typ_refcnt <= 1)
2473                         libcfs_kkuc_group_rem(0, KUC_GRP_HSM);
2474 
2475                 obd_cleanup_client_import(obd);
2476                 ptlrpc_lprocfs_unregister_obd(obd);
2477                 lprocfs_obd_cleanup(obd);
2478 
2479                 rc = obd_llog_finish(obd, 0);
2480                 if (rc != 0)
2481                         CERROR("failed to cleanup llogging subsystems\n");
2482                 break;
2483         }
2484         return rc;
2485 }
2486 
2487 static int mdc_cleanup(struct obd_device *obd)
2488 {
2489         struct client_obd *cli = &obd->u.cli;
2490 
2491         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
2492         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
2493 
2494         ptlrpcd_decref();
2495 
2496         return client_obd_cleanup(obd);
2497 }
2498 
2499 
2500 static int mdc_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
2501                          struct obd_device *tgt, int *index)
2502 {
2503         struct llog_ctxt        *ctxt;
2504         int                      rc;
2505 
2506         LASSERT(olg == &obd->obd_olg);
2507 
2508         rc = llog_setup(NULL, obd, olg, LLOG_CHANGELOG_REPL_CTXT, tgt,
2509                         &llog_client_ops);
2510         if (rc)
2511                 return rc;
2512 
2513         ctxt = llog_group_get_ctxt(olg, LLOG_CHANGELOG_REPL_CTXT);
2514         llog_initiator_connect(ctxt);
2515         llog_ctxt_put(ctxt);
2516 
2517         return 0;
2518 }
2519 
2520 static int mdc_llog_finish(struct obd_device *obd, int count)
2521 {
2522         struct llog_ctxt *ctxt;
2523 
2524         ctxt = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
2525         if (ctxt)
2526                 llog_cleanup(NULL, ctxt);
2527 
2528         return 0;
2529 }
2530 
2531 static int mdc_process_config(struct obd_device *obd, obd_count len, void *buf)
2532 {
2533         struct lustre_cfg *lcfg = buf;
2534         struct lprocfs_static_vars lvars = { 0 };
2535         int rc = 0;
2536 
2537         lprocfs_mdc_init_vars(&lvars);
2538         switch (lcfg->lcfg_command) {
2539         default:
2540                 rc = class_process_proc_param(PARAM_MDC, lvars.obd_vars,
2541                                               lcfg, obd);
2542                 if (rc > 0)
2543                         rc = 0;
2544                 break;
2545         }
2546         return(rc);
2547 }
2548 
2549 
2550 /* get remote permission for current user on fid */
2551 int mdc_get_remote_perm(struct obd_export *exp, const struct lu_fid *fid,
2552                         struct obd_capa *oc, __u32 suppgid,
2553                         struct ptlrpc_request **request)
2554 {
2555         struct ptlrpc_request  *req;
2556         int                 rc;
2557 
2558         LASSERT(client_is_remote(exp));
2559 
2560         *request = NULL;
2561         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_GETATTR);
2562         if (req == NULL)
2563                 return -ENOMEM;
2564 
2565         mdc_set_capa_size(req, &RMF_CAPA1, oc);
2566 
2567         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_GETATTR);
2568         if (rc) {
2569                 ptlrpc_request_free(req);
2570                 return rc;
2571         }
2572 
2573         mdc_pack_body(req, fid, oc, OBD_MD_FLRMTPERM, 0, suppgid, 0);
2574 
2575         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
2576                              sizeof(struct mdt_remote_perm));
2577 
2578         ptlrpc_request_set_replen(req);
2579 
2580         rc = ptlrpc_queue_wait(req);
2581         if (rc)
2582                 ptlrpc_req_finished(req);
2583         else
2584                 *request = req;
2585         return rc;
2586 }
2587 
2588 static int mdc_interpret_renew_capa(const struct lu_env *env,
2589                                     struct ptlrpc_request *req, void *args,
2590                                     int status)
2591 {
2592         struct mdc_renew_capa_args *ra = args;
2593         struct mdt_body *body = NULL;
2594         struct lustre_capa *capa;
2595 
2596         if (status)
2597                 GOTO(out, capa = ERR_PTR(status));
2598 
2599         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
2600         if (body == NULL)
2601                 GOTO(out, capa = ERR_PTR(-EFAULT));
2602 
2603         if ((body->valid & OBD_MD_FLOSSCAPA) == 0)
2604                 GOTO(out, capa = ERR_PTR(-ENOENT));
2605 
2606         capa = req_capsule_server_get(&req->rq_pill, &RMF_CAPA2);
2607         if (!capa)
2608                 GOTO(out, capa = ERR_PTR(-EFAULT));
2609 out:
2610         ra->ra_cb(ra->ra_oc, capa);
2611         return 0;
2612 }
2613 
2614 static int mdc_renew_capa(struct obd_export *exp, struct obd_capa *oc,
2615                           renew_capa_cb_t cb)
2616 {
2617         struct ptlrpc_request *req;
2618         struct mdc_renew_capa_args *ra;
2619 
2620         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_MDS_GETATTR,
2621                                         LUSTRE_MDS_VERSION, MDS_GETATTR);
2622         if (req == NULL)
2623                 return -ENOMEM;
2624 
2625         /* NB, OBD_MD_FLOSSCAPA is set here, but it doesn't necessarily mean the
2626          * capa to renew is oss capa.
2627          */
2628         mdc_pack_body(req, &oc->c_capa.lc_fid, oc, OBD_MD_FLOSSCAPA, 0, -1, 0);
2629         ptlrpc_request_set_replen(req);
2630 
2631         CLASSERT(sizeof(*ra) <= sizeof(req->rq_async_args));
2632         ra = ptlrpc_req_async_args(req);
2633         ra->ra_oc = oc;
2634         ra->ra_cb = cb;
2635         req->rq_interpret_reply = mdc_interpret_renew_capa;
2636         ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
2637         return 0;
2638 }
2639 
2640 struct obd_ops mdc_obd_ops = {
2641         .o_owner            = THIS_MODULE,
2642         .o_setup            = mdc_setup,
2643         .o_precleanup       = mdc_precleanup,
2644         .o_cleanup        = mdc_cleanup,
2645         .o_add_conn      = client_import_add_conn,
2646         .o_del_conn      = client_import_del_conn,
2647         .o_connect          = client_connect_import,
2648         .o_disconnect       = client_disconnect_export,
2649         .o_iocontrol    = mdc_iocontrol,
2650         .o_set_info_async   = mdc_set_info_async,
2651         .o_statfs          = mdc_statfs,
2652         .o_pin        = mdc_pin,
2653         .o_unpin            = mdc_unpin,
2654         .o_fid_init         = client_fid_init,
2655         .o_fid_fini         = client_fid_fini,
2656         .o_fid_alloc    = mdc_fid_alloc,
2657         .o_import_event     = mdc_import_event,
2658         .o_llog_init    = mdc_llog_init,
2659         .o_llog_finish      = mdc_llog_finish,
2660         .o_get_info      = mdc_get_info,
2661         .o_process_config   = mdc_process_config,
2662         .o_get_uuid      = mdc_get_uuid,
2663         .o_quotactl      = mdc_quotactl,
2664         .o_quotacheck       = mdc_quotacheck
2665 };
2666 
2667 struct md_ops mdc_md_ops = {
2668         .m_getstatus    = mdc_getstatus,
2669         .m_null_inode       = mdc_null_inode,
2670         .m_find_cbdata      = mdc_find_cbdata,
2671         .m_close            = mdc_close,
2672         .m_create          = mdc_create,
2673         .m_done_writing     = mdc_done_writing,
2674         .m_enqueue        = mdc_enqueue,
2675         .m_getattr        = mdc_getattr,
2676         .m_getattr_name     = mdc_getattr_name,
2677         .m_intent_lock      = mdc_intent_lock,
2678         .m_link      = mdc_link,
2679         .m_is_subdir    = mdc_is_subdir,
2680         .m_rename          = mdc_rename,
2681         .m_setattr        = mdc_setattr,
2682         .m_setxattr      = mdc_setxattr,
2683         .m_getxattr      = mdc_getxattr,
2684         .m_sync      = mdc_sync,
2685         .m_readpage      = mdc_readpage,
2686         .m_unlink          = mdc_unlink,
2687         .m_cancel_unused    = mdc_cancel_unused,
2688         .m_init_ea_size     = mdc_init_ea_size,
2689         .m_set_lock_data    = mdc_set_lock_data,
2690         .m_lock_match       = mdc_lock_match,
2691         .m_get_lustre_md    = mdc_get_lustre_md,
2692         .m_free_lustre_md   = mdc_free_lustre_md,
2693         .m_set_open_replay_data = mdc_set_open_replay_data,
2694         .m_clear_open_replay_data = mdc_clear_open_replay_data,
2695         .m_renew_capa       = mdc_renew_capa,
2696         .m_unpack_capa      = mdc_unpack_capa,
2697         .m_get_remote_perm  = mdc_get_remote_perm,
2698         .m_intent_getattr_async = mdc_intent_getattr_async,
2699         .m_revalidate_lock      = mdc_revalidate_lock
2700 };
2701 
2702 int __init mdc_init(void)
2703 {
2704         int rc;
2705         struct lprocfs_static_vars lvars = { 0 };
2706         lprocfs_mdc_init_vars(&lvars);
2707 
2708         rc = class_register_type(&mdc_obd_ops, &mdc_md_ops, lvars.module_vars,
2709                                  LUSTRE_MDC_NAME, NULL);
2710         return rc;
2711 }
2712 
2713 static void /*__exit*/ mdc_exit(void)
2714 {
2715         class_unregister_type(LUSTRE_MDC_NAME);
2716 }
2717 
2718 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
2719 MODULE_DESCRIPTION("Lustre Metadata Client");
2720 MODULE_LICENSE("GPL");
2721 
2722 module_init(mdc_init);
2723 module_exit(mdc_exit);
2724 

This page was automatically generated by LXR 0.3.1 (source).  •  Linux is a registered trademark of Linus Torvalds  •  Contact us