Whamcloud - gitweb
03e6aac39e830d6f0e52c62ed9120debb860bc04
[fs/lustre-release.git] / lustre / mdc / mdc_reint.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  */
32
33 #define DEBUG_SUBSYSTEM S_MDC
34
35 #include <linux/module.h>
36 #include <linux/kernel.h>
37
38 #include <obd_class.h>
39 #include "mdc_internal.h"
40 #include <lustre_fid.h>
41
42 /* mdc_setattr does its own semaphore handling */
43 static int mdc_reint(struct ptlrpc_request *request, int level)
44 {
45         int rc;
46
47         request->rq_send_state = level;
48
49         mdc_get_mod_rpc_slot(request, NULL);
50         rc = ptlrpc_queue_wait(request);
51         mdc_put_mod_rpc_slot(request, NULL);
52         if (rc)
53                 CDEBUG(D_INFO, "error in handling %d\n", rc);
54         else if (!req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY)) {
55                 rc = -EPROTO;
56         }
57         return rc;
58 }
59
60 /* Find and cancel locally locks matched by inode @bits & @mode in the resource
61  * found by @fid. Found locks are added into @cancel list. Returns the amount of
62  * locks added to @cancels list. */
63 int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
64                             struct list_head *cancels, enum ldlm_mode mode,
65                             __u64 bits)
66 {
67         struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
68         union ldlm_policy_data policy = { {0} };
69         struct ldlm_res_id res_id;
70         struct ldlm_resource *res;
71         int count;
72         ENTRY;
73
74         /* Return, i.e. cancel nothing, only if ELC is supported (flag in
75          * export) but disabled through procfs (flag in NS).
76          *
77          * This distinguishes from a case when ELC is not supported originally,
78          * when we still want to cancel locks in advance and just cancel them
79          * locally, without sending any RPC. */
80         if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
81                 RETURN(0);
82
83         fid_build_reg_res_name(fid, &res_id);
84         res = ldlm_resource_get(exp->exp_obd->obd_namespace,
85                                 NULL, &res_id, 0, 0);
86         if (IS_ERR(res))
87                 RETURN(0);
88         LDLM_RESOURCE_ADDREF(res);
89         /* Initialize ibits lock policy. */
90         policy.l_inodebits.bits = bits;
91         count = ldlm_cancel_resource_local(res, cancels, &policy,
92                                            mode, 0, 0, NULL);
93         LDLM_RESOURCE_DELREF(res);
94         ldlm_resource_putref(res);
95         RETURN(count);
96 }
97
98 int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
99                 void *ea, size_t ealen, struct ptlrpc_request **request)
100 {
101         struct list_head cancels = LIST_HEAD_INIT(cancels);
102         struct ptlrpc_request *req;
103         int count = 0, rc;
104         __u64 bits;
105         ENTRY;
106
107         LASSERT(op_data != NULL);
108
109         bits = MDS_INODELOCK_UPDATE;
110         if (op_data->op_attr.ia_valid & (ATTR_MODE|ATTR_UID|ATTR_GID))
111                 bits |= MDS_INODELOCK_LOOKUP;
112         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
113             (fid_is_sane(&op_data->op_fid1)))
114                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
115                                                 &cancels, LCK_EX, bits);
116         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
117                                    &RQF_MDS_REINT_SETATTR);
118         if (req == NULL) {
119                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
120                 RETURN(-ENOMEM);
121         }
122
123         req_capsule_set_size(&req->rq_pill, &RMF_MDT_EPOCH, RCL_CLIENT, 0);
124         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, ealen);
125         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_CLIENT, 0);
126
127         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
128         if (rc) {
129                 ptlrpc_request_free(req);
130                 RETURN(rc);
131         }
132
133         if (op_data->op_attr.ia_valid & (ATTR_MTIME | ATTR_CTIME))
134                 CDEBUG(D_INODE, "setting mtime %lld, ctime %lld\n",
135                        (s64)op_data->op_attr.ia_mtime.tv_sec,
136                        (s64)op_data->op_attr.ia_ctime.tv_sec);
137         mdc_setattr_pack(req, op_data, ea, ealen);
138
139         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
140                              MIN(req->rq_import->imp_connect_data.ocd_max_easize,
141                                  XATTR_SIZE_MAX));
142         ptlrpc_request_set_replen(req);
143
144         rc = mdc_reint(req, LUSTRE_IMP_FULL);
145         if (rc == -ERESTARTSYS)
146                 rc = 0;
147
148         *request = req;
149
150         RETURN(rc);
151 }
152
153 int mdc_create(struct obd_export *exp, struct md_op_data *op_data,
154                 const void *data, size_t datalen,
155                 umode_t mode, uid_t uid, gid_t gid,
156                 cfs_cap_t cap_effective, __u64 rdev,
157                 struct ptlrpc_request **request)
158 {
159         struct ptlrpc_request *req;
160         int level, rc;
161         int count, resends = 0;
162         struct obd_import *import = exp->exp_obd->u.cli.cl_import;
163         int generation = import->imp_generation;
164         struct list_head cancels = LIST_HEAD_INIT(cancels);
165         ENTRY;
166
167         /* For case if upper layer did not alloc fid, do it now. */
168         if (!fid_is_sane(&op_data->op_fid2)) {
169                 /*
170                  * mdc_fid_alloc() may return errno 1 in case of switch to new
171                  * sequence, handle this.
172                  */
173                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
174                 if (rc < 0)
175                         RETURN(rc);
176         }
177
178 rebuild:
179         count = 0;
180         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
181             (fid_is_sane(&op_data->op_fid1)))
182                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
183                                                 &cancels, LCK_EX,
184                                                 MDS_INODELOCK_UPDATE);
185
186         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
187                                    &RQF_MDS_REINT_CREATE_ACL);
188         if (req == NULL) {
189                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
190                 RETURN(-ENOMEM);
191         }
192
193         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
194                              op_data->op_namelen + 1);
195         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
196                              data && datalen ? datalen : 0);
197
198         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
199                              RCL_CLIENT, op_data->op_file_secctx_name != NULL ?
200                              strlen(op_data->op_file_secctx_name) + 1 : 0);
201
202         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX, RCL_CLIENT,
203                              op_data->op_file_secctx_size);
204
205         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
206         if (rc) {
207                 ptlrpc_request_free(req);
208                 RETURN(rc);
209         }
210
211         /*
212          * mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
213          * tgt, for symlinks or lov MD data.
214          */
215         mdc_create_pack(req, op_data, data, datalen, mode, uid,
216                         gid, cap_effective, rdev);
217
218         ptlrpc_request_set_replen(req);
219
220         /* ask ptlrpc not to resend on EINPROGRESS since we have our own retry
221          * logic here */
222         req->rq_no_retry_einprogress = 1;
223
224         if (resends) {
225                 req->rq_generation_set = 1;
226                 req->rq_import_generation = generation;
227                 req->rq_sent = ktime_get_real_seconds() + resends;
228         }
229         level = LUSTRE_IMP_FULL;
230  resend:
231         rc = mdc_reint(req, level);
232
233         /* Resend if we were told to. */
234         if (rc == -ERESTARTSYS) {
235                 level = LUSTRE_IMP_RECOVER;
236                 goto resend;
237         } else if (rc == -EINPROGRESS) {
238                 /* Retry create infinitely until succeed or get other
239                  * error code or interrupted. */
240                 ptlrpc_req_finished(req);
241                 if (generation == import->imp_generation) {
242                         if (signal_pending(current))
243                                 RETURN(-EINTR);
244
245                         resends++;
246                         CDEBUG(D_HA, "%s: resend:%d create on "DFID"/"DFID"\n",
247                                exp->exp_obd->obd_name, resends,
248                                PFID(&op_data->op_fid1),
249                                PFID(&op_data->op_fid2));
250                         goto rebuild;
251                 } else {
252                         CDEBUG(D_HA, "resend cross eviction\n");
253                         RETURN(-EIO);
254                 }
255         }
256
257         *request = req;
258         RETURN(rc);
259 }
260
261 int mdc_unlink(struct obd_export *exp, struct md_op_data *op_data,
262                struct ptlrpc_request **request)
263 {
264         struct list_head cancels = LIST_HEAD_INIT(cancels);
265         struct obd_device *obd = class_exp2obd(exp);
266         struct ptlrpc_request *req = *request;
267         int count = 0, rc;
268         ENTRY;
269
270         LASSERT(req == NULL);
271
272         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
273             (fid_is_sane(&op_data->op_fid1)))
274                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
275                                                 &cancels, LCK_EX,
276                                                 MDS_INODELOCK_UPDATE);
277         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
278             (fid_is_sane(&op_data->op_fid3)))
279                 /* don't cancel DoM lock which may cause data flush */
280                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
281                                                  &cancels, LCK_EX,
282                                                  MDS_INODELOCK_ELC);
283         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
284                                    &RQF_MDS_REINT_UNLINK);
285         if (req == NULL) {
286                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
287                 RETURN(-ENOMEM);
288         }
289
290         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
291                              op_data->op_namelen + 1);
292
293         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
294         if (rc) {
295                 ptlrpc_request_free(req);
296                 RETURN(rc);
297         }
298
299         mdc_unlink_pack(req, op_data);
300
301         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
302                              obd->u.cli.cl_default_mds_easize);
303         ptlrpc_request_set_replen(req);
304
305         *request = req;
306
307         rc = mdc_reint(req, LUSTRE_IMP_FULL);
308         if (rc == -ERESTARTSYS)
309                 rc = 0;
310         RETURN(rc);
311 }
312
313 int mdc_link(struct obd_export *exp, struct md_op_data *op_data,
314              struct ptlrpc_request **request)
315 {
316         struct list_head cancels = LIST_HEAD_INIT(cancels);
317         struct ptlrpc_request *req;
318         int count = 0, rc;
319         ENTRY;
320
321         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
322             (fid_is_sane(&op_data->op_fid2)))
323                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
324                                                 &cancels, LCK_EX,
325                                                 MDS_INODELOCK_UPDATE);
326         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
327             (fid_is_sane(&op_data->op_fid1)))
328                 count += mdc_resource_get_unused(exp, &op_data->op_fid1,
329                                                  &cancels, LCK_EX,
330                                                  MDS_INODELOCK_UPDATE);
331
332         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
333         if (req == NULL) {
334                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
335                 RETURN(-ENOMEM);
336         }
337
338         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
339                              op_data->op_namelen + 1);
340
341         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
342         if (rc) {
343                 ptlrpc_request_free(req);
344                 RETURN(rc);
345         }
346
347         mdc_link_pack(req, op_data);
348         ptlrpc_request_set_replen(req);
349
350         rc = mdc_reint(req, LUSTRE_IMP_FULL);
351         *request = req;
352         if (rc == -ERESTARTSYS)
353                 rc = 0;
354
355         RETURN(rc);
356 }
357
358 int mdc_rename(struct obd_export *exp, struct md_op_data *op_data,
359                 const char *old, size_t oldlen, const char *new, size_t newlen,
360                 struct ptlrpc_request **request)
361 {
362         struct list_head cancels = LIST_HEAD_INIT(cancels);
363         struct obd_device *obd = exp->exp_obd;
364         struct ptlrpc_request *req;
365         int count = 0, rc;
366
367         ENTRY;
368
369         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
370             (fid_is_sane(&op_data->op_fid1)))
371                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
372                                                 &cancels, LCK_EX,
373                                                 MDS_INODELOCK_UPDATE);
374         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
375             (fid_is_sane(&op_data->op_fid2)))
376                 count += mdc_resource_get_unused(exp, &op_data->op_fid2,
377                                                  &cancels, LCK_EX,
378                                                  MDS_INODELOCK_UPDATE);
379         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
380             (fid_is_sane(&op_data->op_fid3)))
381                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
382                                                  &cancels, LCK_EX,
383                                                  MDS_INODELOCK_LOOKUP);
384         if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
385             (fid_is_sane(&op_data->op_fid4)))
386                 count += mdc_resource_get_unused(exp, &op_data->op_fid4,
387                                                  &cancels, LCK_EX,
388                                                  MDS_INODELOCK_ELC);
389
390         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
391                            op_data->op_cli_flags & CLI_MIGRATE ?
392                            &RQF_MDS_REINT_MIGRATE : &RQF_MDS_REINT_RENAME);
393         if (req == NULL) {
394                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
395                 RETURN(-ENOMEM);
396         }
397
398         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
399         req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT, newlen+1);
400         if (op_data->op_cli_flags & CLI_MIGRATE)
401                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
402                                      op_data->op_data_size);
403
404         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
405         if (rc) {
406                 ptlrpc_request_free(req);
407                 RETURN(rc);
408         }
409
410         if (exp_connect_cancelset(exp) && req)
411                 ldlm_cli_cancel_list(&cancels, count, req, 0);
412
413         if (op_data->op_cli_flags & CLI_MIGRATE)
414                 mdc_migrate_pack(req, op_data, old, oldlen);
415         else
416                 mdc_rename_pack(req, op_data, old, oldlen, new, newlen);
417
418         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
419                              obd->u.cli.cl_default_mds_easize);
420         ptlrpc_request_set_replen(req);
421
422         rc = mdc_reint(req, LUSTRE_IMP_FULL);
423         *request = req;
424         if (rc == -ERESTARTSYS)
425                 rc = 0;
426
427         RETURN(rc);
428 }
429
430 int mdc_file_resync(struct obd_export *exp, struct md_op_data *op_data)
431 {
432         struct list_head cancels = LIST_HEAD_INIT(cancels);
433         struct ptlrpc_request *req;
434         struct ldlm_lock *lock;
435         struct mdt_rec_resync *rec;
436         int count = 0, rc;
437         ENTRY;
438
439         if (op_data->op_flags & MF_MDC_CANCEL_FID1 &&
440             fid_is_sane(&op_data->op_fid1))
441                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
442                                                 &cancels, LCK_EX,
443                                                 MDS_INODELOCK_LAYOUT);
444
445         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
446                                    &RQF_MDS_REINT_RESYNC);
447         if (req == NULL) {
448                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
449                 RETURN(-ENOMEM);
450         }
451
452         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
453         if (rc) {
454                 ptlrpc_request_free(req);
455                 RETURN(rc);
456         }
457
458         CLASSERT(sizeof(*rec) == sizeof(struct mdt_rec_reint));
459         rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
460         rec->rs_opcode  = REINT_RESYNC;
461         rec->rs_fsuid   = op_data->op_fsuid;
462         rec->rs_fsgid   = op_data->op_fsgid;
463         rec->rs_cap     = op_data->op_cap;
464         rec->rs_fid     = op_data->op_fid1;
465         rec->rs_bias    = op_data->op_bias;
466         rec->rs_mirror_id = op_data->op_mirror_id;
467
468         lock = ldlm_handle2lock(&op_data->op_lease_handle);
469         if (lock != NULL) {
470                 rec->rs_lease_handle = lock->l_remote_handle;
471                 LDLM_LOCK_PUT(lock);
472         }
473
474         ptlrpc_request_set_replen(req);
475
476         rc = mdc_reint(req, LUSTRE_IMP_FULL);
477         if (rc == -ERESTARTSYS)
478                 rc = 0;
479
480         ptlrpc_req_finished(req);
481         RETURN(rc);
482 }