Version:  2.0.40 2.2.26 2.4.37 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 3.16 3.17 3.18

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

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