Whamcloud - gitweb
LU-4423 libcfs: remove CFS_TIME_T definition
[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, 2016, 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 %ld, ctime %ld\n",
135                        LTIME_S(op_data->op_attr.ia_mtime),
136                        LTIME_S(op_data->op_attr.ia_ctime));
137         mdc_setattr_pack(req, op_data, ea, ealen);
138
139         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
140                              req->rq_import->imp_connect_data.ocd_max_easize);
141         ptlrpc_request_set_replen(req);
142
143         rc = mdc_reint(req, LUSTRE_IMP_FULL);
144         if (rc == -ERESTARTSYS)
145                 rc = 0;
146
147         *request = req;
148
149         RETURN(rc);
150 }
151
152 int mdc_create(struct obd_export *exp, struct md_op_data *op_data,
153                 const void *data, size_t datalen,
154                 umode_t mode, uid_t uid, gid_t gid,
155                 cfs_cap_t cap_effective, __u64 rdev,
156                 struct ptlrpc_request **request)
157 {
158         struct ptlrpc_request *req;
159         int level, rc;
160         int count, resends = 0;
161         struct obd_import *import = exp->exp_obd->u.cli.cl_import;
162         int generation = import->imp_generation;
163         struct list_head cancels = LIST_HEAD_INIT(cancels);
164         ENTRY;
165
166         /* For case if upper layer did not alloc fid, do it now. */
167         if (!fid_is_sane(&op_data->op_fid2)) {
168                 /*
169                  * mdc_fid_alloc() may return errno 1 in case of switch to new
170                  * sequence, handle this.
171                  */
172                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
173                 if (rc < 0)
174                         RETURN(rc);
175         }
176
177 rebuild:
178         count = 0;
179         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
180             (fid_is_sane(&op_data->op_fid1)))
181                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
182                                                 &cancels, LCK_EX,
183                                                 MDS_INODELOCK_UPDATE);
184
185         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
186                                    &RQF_MDS_REINT_CREATE_ACL);
187         if (req == NULL) {
188                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
189                 RETURN(-ENOMEM);
190         }
191
192         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
193                              op_data->op_namelen + 1);
194         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
195                              data && datalen ? datalen : 0);
196
197         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
198                              RCL_CLIENT, op_data->op_file_secctx_name != NULL ?
199                              strlen(op_data->op_file_secctx_name) + 1 : 0);
200
201         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX, RCL_CLIENT,
202                              op_data->op_file_secctx_size);
203
204         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
205         if (rc) {
206                 ptlrpc_request_free(req);
207                 RETURN(rc);
208         }
209
210         /*
211          * mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
212          * tgt, for symlinks or lov MD data.
213          */
214         mdc_create_pack(req, op_data, data, datalen, mode, uid,
215                         gid, cap_effective, rdev);
216
217         ptlrpc_request_set_replen(req);
218
219         /* ask ptlrpc not to resend on EINPROGRESS since we have our own retry
220          * logic here */
221         req->rq_no_retry_einprogress = 1;
222
223         if (resends) {
224                 req->rq_generation_set = 1;
225                 req->rq_import_generation = generation;
226                 req->rq_sent = ktime_get_real_seconds() + resends;
227         }
228         level = LUSTRE_IMP_FULL;
229  resend:
230         rc = mdc_reint(req, level);
231
232         /* Resend if we were told to. */
233         if (rc == -ERESTARTSYS) {
234                 level = LUSTRE_IMP_RECOVER;
235                 goto resend;
236         } else if (rc == -EINPROGRESS) {
237                 /* Retry create infinitely until succeed or get other
238                  * error code. */
239                 ptlrpc_req_finished(req);
240                 resends++;
241
242                 CDEBUG(D_HA, "%s: resend:%d create on "DFID"/"DFID"\n",
243                        exp->exp_obd->obd_name, resends,
244                        PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));
245
246                 if (generation == import->imp_generation) {
247                         goto rebuild;
248                 } else {
249                         CDEBUG(D_HA, "resend cross eviction\n");
250                         RETURN(-EIO);
251                 }
252         }
253
254         *request = req;
255         RETURN(rc);
256 }
257
258 int mdc_unlink(struct obd_export *exp, struct md_op_data *op_data,
259                struct ptlrpc_request **request)
260 {
261         struct list_head cancels = LIST_HEAD_INIT(cancels);
262         struct obd_device *obd = class_exp2obd(exp);
263         struct ptlrpc_request *req = *request;
264         int count = 0, rc;
265         ENTRY;
266
267         LASSERT(req == NULL);
268
269         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
270             (fid_is_sane(&op_data->op_fid1)))
271                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
272                                                 &cancels, LCK_EX,
273                                                 MDS_INODELOCK_UPDATE);
274         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
275             (fid_is_sane(&op_data->op_fid3)))
276                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
277                                                  &cancels, LCK_EX,
278                                                  MDS_INODELOCK_FULL);
279         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
280                                    &RQF_MDS_REINT_UNLINK);
281         if (req == NULL) {
282                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
283                 RETURN(-ENOMEM);
284         }
285
286         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
287                              op_data->op_namelen + 1);
288
289         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
290         if (rc) {
291                 ptlrpc_request_free(req);
292                 RETURN(rc);
293         }
294
295         mdc_unlink_pack(req, op_data);
296
297         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
298                              obd->u.cli.cl_default_mds_easize);
299         ptlrpc_request_set_replen(req);
300
301         *request = req;
302
303         rc = mdc_reint(req, LUSTRE_IMP_FULL);
304         if (rc == -ERESTARTSYS)
305                 rc = 0;
306         RETURN(rc);
307 }
308
309 int mdc_link(struct obd_export *exp, struct md_op_data *op_data,
310              struct ptlrpc_request **request)
311 {
312         struct list_head cancels = LIST_HEAD_INIT(cancels);
313         struct ptlrpc_request *req;
314         int count = 0, rc;
315         ENTRY;
316
317         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
318             (fid_is_sane(&op_data->op_fid2)))
319                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
320                                                 &cancels, LCK_EX,
321                                                 MDS_INODELOCK_UPDATE);
322         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
323             (fid_is_sane(&op_data->op_fid1)))
324                 count += mdc_resource_get_unused(exp, &op_data->op_fid1,
325                                                  &cancels, LCK_EX,
326                                                  MDS_INODELOCK_UPDATE);
327
328         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
329         if (req == NULL) {
330                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
331                 RETURN(-ENOMEM);
332         }
333
334         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
335                              op_data->op_namelen + 1);
336
337         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
338         if (rc) {
339                 ptlrpc_request_free(req);
340                 RETURN(rc);
341         }
342
343         mdc_link_pack(req, op_data);
344         ptlrpc_request_set_replen(req);
345
346         rc = mdc_reint(req, LUSTRE_IMP_FULL);
347         *request = req;
348         if (rc == -ERESTARTSYS)
349                 rc = 0;
350
351         RETURN(rc);
352 }
353
354 int mdc_rename(struct obd_export *exp, struct md_op_data *op_data,
355                 const char *old, size_t oldlen, const char *new, size_t newlen,
356                 struct ptlrpc_request **request)
357 {
358         struct list_head cancels = LIST_HEAD_INIT(cancels);
359         struct obd_device *obd = exp->exp_obd;
360         struct ptlrpc_request *req;
361         int count = 0, rc;
362         ENTRY;
363
364         if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
365             (fid_is_sane(&op_data->op_fid1)))
366                 count = mdc_resource_get_unused(exp, &op_data->op_fid1,
367                                                 &cancels, LCK_EX,
368                                                 MDS_INODELOCK_UPDATE);
369         if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
370             (fid_is_sane(&op_data->op_fid2)))
371                 count += mdc_resource_get_unused(exp, &op_data->op_fid2,
372                                                  &cancels, LCK_EX,
373                                                  MDS_INODELOCK_UPDATE);
374         if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
375             (fid_is_sane(&op_data->op_fid3)))
376                 count += mdc_resource_get_unused(exp, &op_data->op_fid3,
377                                                  &cancels, LCK_EX,
378                                                  MDS_INODELOCK_LOOKUP);
379         if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
380              (fid_is_sane(&op_data->op_fid4)))
381                 count += mdc_resource_get_unused(exp, &op_data->op_fid4,
382                                                  &cancels, LCK_EX,
383                                                  MDS_INODELOCK_FULL);
384
385         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
386                            op_data->op_cli_flags & CLI_MIGRATE ?
387                            &RQF_MDS_REINT_MIGRATE : &RQF_MDS_REINT_RENAME);
388         if (req == NULL) {
389                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
390                 RETURN(-ENOMEM);
391         }
392
393         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
394         req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT, newlen+1);
395
396         rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
397         if (rc) {
398                 ptlrpc_request_free(req);
399                 RETURN(rc);
400         }
401
402         if (op_data->op_cli_flags & CLI_MIGRATE && op_data->op_data != NULL) {
403                 struct md_open_data *mod = op_data->op_data;
404
405                 LASSERTF(mod->mod_open_req != NULL &&
406                          mod->mod_open_req->rq_type != LI_POISON,
407                          "POISONED open %p!\n", mod->mod_open_req);
408
409                 DEBUG_REQ(D_HA, mod->mod_open_req, "matched open");
410                 /* We no longer want to preserve this open for replay even
411                  * though the open was committed. b=3632, b=3633 */
412                 spin_lock(&mod->mod_open_req->rq_lock);
413                 mod->mod_open_req->rq_replay = 0;
414                 spin_unlock(&mod->mod_open_req->rq_lock);
415         }
416
417         if (exp_connect_cancelset(exp) && req)
418                 ldlm_cli_cancel_list(&cancels, count, req, 0);
419
420         mdc_rename_pack(req, op_data, old, oldlen, new, newlen);
421
422         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
423                              obd->u.cli.cl_default_mds_easize);
424         ptlrpc_request_set_replen(req);
425
426         rc = mdc_reint(req, LUSTRE_IMP_FULL);
427         *request = req;
428         if (rc == -ERESTARTSYS)
429                 rc = 0;
430
431         RETURN(rc);
432 }