Whamcloud - gitweb
LU-6322 lfsck: show start/complete time directly
[fs/lustre-release.git] / lustre / fid / fid_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) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2014, 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  * lustre/fid/fid_request.c
37  *
38  * Lustre Sequence Manager
39  *
40  * Author: Yury Umanets <umka@clusterfs.com>
41  */
42
43 #define DEBUG_SUBSYSTEM S_FID
44
45 #include <linux/module.h>
46 #include <libcfs/libcfs.h>
47 #include <obd.h>
48 #include <obd_class.h>
49 #include <obd_support.h>
50 #include <lustre_fid.h>
51 /* mdc RPC locks */
52 #include <lustre_mdc.h>
53 #include "fid_internal.h"
54
55 static int seq_client_rpc(struct lu_client_seq *seq,
56                           struct lu_seq_range *output, __u32 opc,
57                           const char *opcname)
58 {
59         struct obd_export     *exp = seq->lcs_exp;
60         struct ptlrpc_request *req;
61         struct lu_seq_range   *out, *in;
62         __u32                 *op;
63         unsigned int           debug_mask;
64         int                    rc;
65         ENTRY;
66
67         LASSERT(exp != NULL && !IS_ERR(exp));
68         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_SEQ_QUERY,
69                                         LUSTRE_MDS_VERSION, SEQ_QUERY);
70         if (req == NULL)
71                 RETURN(-ENOMEM);
72
73         /* Init operation code */
74         op = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_OPC);
75         *op = opc;
76
77         /* Zero out input range, this is not recovery yet. */
78         in = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_RANGE);
79         range_init(in);
80
81         ptlrpc_request_set_replen(req);
82
83         in->lsr_index = seq->lcs_space.lsr_index;
84         if (seq->lcs_type == LUSTRE_SEQ_METADATA)
85                 fld_range_set_mdt(in);
86         else
87                 fld_range_set_ost(in);
88
89         if (opc == SEQ_ALLOC_SUPER) {
90                 req->rq_request_portal = SEQ_CONTROLLER_PORTAL;
91                 req->rq_reply_portal = MDC_REPLY_PORTAL;
92                 /* During allocating super sequence for data object,
93                  * the current thread might hold the export of MDT0(MDT0
94                  * precreating objects on this OST), and it will send the
95                  * request to MDT0 here, so we can not keep resending the
96                  * request here, otherwise if MDT0 is failed(umounted),
97                  * it can not release the export of MDT0 */
98                 if (seq->lcs_type == LUSTRE_SEQ_DATA)
99                         req->rq_no_delay = req->rq_no_resend = 1;
100                 debug_mask = D_CONSOLE;
101         } else {
102                 if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
103                         req->rq_reply_portal = MDC_REPLY_PORTAL;
104                         req->rq_request_portal = SEQ_METADATA_PORTAL;
105                 } else {
106                         req->rq_reply_portal = OSC_REPLY_PORTAL;
107                         req->rq_request_portal = SEQ_DATA_PORTAL;
108                 }
109
110                 debug_mask = D_INFO;
111         }
112
113         ptlrpc_at_set_req_timeout(req);
114
115         if (opc != SEQ_ALLOC_SUPER && seq->lcs_type == LUSTRE_SEQ_METADATA)
116                 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
117
118         rc = ptlrpc_queue_wait(req);
119
120         if (opc != SEQ_ALLOC_SUPER && seq->lcs_type == LUSTRE_SEQ_METADATA)
121                 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
122         if (rc)
123                 GOTO(out_req, rc);
124
125         out = req_capsule_server_get(&req->rq_pill, &RMF_SEQ_RANGE);
126         *output = *out;
127
128         if (!range_is_sane(output)) {
129                 CERROR("%s: Invalid range received from server: "
130                        DRANGE"\n", seq->lcs_name, PRANGE(output));
131                 GOTO(out_req, rc = -EINVAL);
132         }
133
134         if (range_is_exhausted(output)) {
135                 CERROR("%s: Range received from server is exhausted: "
136                        DRANGE"]\n", seq->lcs_name, PRANGE(output));
137                 GOTO(out_req, rc = -EINVAL);
138         }
139
140         CDEBUG_LIMIT(debug_mask, "%s: Allocated %s-sequence "DRANGE"]\n",
141                      seq->lcs_name, opcname, PRANGE(output));
142
143         EXIT;
144 out_req:
145         ptlrpc_req_finished(req);
146         return rc;
147 }
148
149 /* Request sequence-controller node to allocate new super-sequence. */
150 int seq_client_alloc_super(struct lu_client_seq *seq,
151                            const struct lu_env *env)
152 {
153         int rc;
154         ENTRY;
155
156         mutex_lock(&seq->lcs_mutex);
157
158         if (seq->lcs_srv) {
159 #ifdef HAVE_SEQ_SERVER
160                 LASSERT(env != NULL);
161                 rc = seq_server_alloc_super(seq->lcs_srv, &seq->lcs_space,
162                                             env);
163 #else
164                 rc = 0;
165 #endif
166         } else {
167                 /* Check whether the connection to seq controller has been
168                  * setup (lcs_exp != NULL) */
169                 if (seq->lcs_exp == NULL) {
170                         mutex_unlock(&seq->lcs_mutex);
171                         RETURN(-EINPROGRESS);
172                 }
173
174                 rc = seq_client_rpc(seq, &seq->lcs_space,
175                                     SEQ_ALLOC_SUPER, "super");
176         }
177         mutex_unlock(&seq->lcs_mutex);
178         RETURN(rc);
179 }
180
181 /* Request sequence-controller node to allocate new meta-sequence. */
182 static int seq_client_alloc_meta(const struct lu_env *env,
183                                  struct lu_client_seq *seq)
184 {
185         int rc;
186         ENTRY;
187
188         if (seq->lcs_srv) {
189 #ifdef HAVE_SEQ_SERVER
190                 LASSERT(env != NULL);
191                 rc = seq_server_alloc_meta(seq->lcs_srv, &seq->lcs_space, env);
192 #else
193                 rc = 0;
194 #endif
195         } else {
196                 do {
197                         /* If meta server return -EINPROGRESS or EAGAIN,
198                          * it means meta server might not be ready to
199                          * allocate super sequence from sequence controller
200                          * (MDT0)yet */
201                         rc = seq_client_rpc(seq, &seq->lcs_space,
202                                             SEQ_ALLOC_META, "meta");
203                 } while (rc == -EINPROGRESS || rc == -EAGAIN);
204         }
205
206         RETURN(rc);
207 }
208
209 /* Allocate new sequence for client. */
210 static int seq_client_alloc_seq(const struct lu_env *env,
211                                 struct lu_client_seq *seq, u64 *seqnr)
212 {
213         int rc;
214         ENTRY;
215
216         LASSERT(range_is_sane(&seq->lcs_space));
217
218         if (range_is_exhausted(&seq->lcs_space)) {
219                 rc = seq_client_alloc_meta(env, seq);
220                 if (rc) {
221                         CERROR("%s: Can't allocate new meta-sequence,"
222                                "rc %d\n", seq->lcs_name, rc);
223                         RETURN(rc);
224                 } else {
225                         CDEBUG(D_INFO, "%s: New range - "DRANGE"\n",
226                                seq->lcs_name, PRANGE(&seq->lcs_space));
227                 }
228         } else {
229                 rc = 0;
230         }
231
232         LASSERT(!range_is_exhausted(&seq->lcs_space));
233         *seqnr = seq->lcs_space.lsr_start;
234         seq->lcs_space.lsr_start += 1;
235
236         CDEBUG(D_INFO, "%s: Allocated sequence ["LPX64"]\n", seq->lcs_name,
237                *seqnr);
238
239         RETURN(rc);
240 }
241
242 static int seq_fid_alloc_prep(struct lu_client_seq *seq,
243                               wait_queue_t *link)
244 {
245         if (seq->lcs_update) {
246                 add_wait_queue(&seq->lcs_waitq, link);
247                 set_current_state(TASK_UNINTERRUPTIBLE);
248                 mutex_unlock(&seq->lcs_mutex);
249
250                 waitq_wait(link, TASK_UNINTERRUPTIBLE);
251
252                 mutex_lock(&seq->lcs_mutex);
253                 remove_wait_queue(&seq->lcs_waitq, link);
254                 set_current_state(TASK_RUNNING);
255                 return -EAGAIN;
256         }
257         ++seq->lcs_update;
258         mutex_unlock(&seq->lcs_mutex);
259         return 0;
260 }
261
262 static void seq_fid_alloc_fini(struct lu_client_seq *seq)
263 {
264         LASSERT(seq->lcs_update == 1);
265         mutex_lock(&seq->lcs_mutex);
266         --seq->lcs_update;
267         wake_up(&seq->lcs_waitq);
268 }
269
270 /**
271  * Allocate the whole seq to the caller.
272  **/
273 int seq_client_get_seq(const struct lu_env *env,
274                        struct lu_client_seq *seq, u64 *seqnr)
275 {
276         wait_queue_t link;
277         int rc;
278
279         LASSERT(seqnr != NULL);
280         mutex_lock(&seq->lcs_mutex);
281         init_waitqueue_entry_current(&link);
282
283         while (1) {
284                 rc = seq_fid_alloc_prep(seq, &link);
285                 if (rc == 0)
286                         break;
287         }
288
289         rc = seq_client_alloc_seq(env, seq, seqnr);
290         if (rc) {
291                 CERROR("%s: Can't allocate new sequence, "
292                        "rc %d\n", seq->lcs_name, rc);
293                 seq_fid_alloc_fini(seq);
294                 mutex_unlock(&seq->lcs_mutex);
295                 return rc;
296         }
297
298         CDEBUG(D_INFO, "%s: allocate sequence "
299                "[0x%16.16"LPF64"x]\n", seq->lcs_name, *seqnr);
300
301         /* Since the caller require the whole seq,
302          * so marked this seq to be used */
303         if (seq->lcs_type == LUSTRE_SEQ_METADATA)
304                 seq->lcs_fid.f_oid = LUSTRE_METADATA_SEQ_MAX_WIDTH;
305         else
306                 seq->lcs_fid.f_oid = LUSTRE_DATA_SEQ_MAX_WIDTH;
307
308         seq->lcs_fid.f_seq = *seqnr;
309         seq->lcs_fid.f_ver = 0;
310         /*
311          * Inform caller that sequence switch is performed to allow it
312          * to setup FLD for it.
313          */
314         seq_fid_alloc_fini(seq);
315         mutex_unlock(&seq->lcs_mutex);
316
317         return rc;
318 }
319 EXPORT_SYMBOL(seq_client_get_seq);
320
321 /* Allocate new fid on passed client @seq and save it to @fid. */
322 int seq_client_alloc_fid(const struct lu_env *env,
323                          struct lu_client_seq *seq, struct lu_fid *fid)
324 {
325         wait_queue_t link;
326         int rc;
327         ENTRY;
328
329         LASSERT(seq != NULL);
330         LASSERT(fid != NULL);
331
332         init_waitqueue_entry_current(&link);
333         mutex_lock(&seq->lcs_mutex);
334
335         if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_EXHAUST))
336                 seq->lcs_fid.f_oid = seq->lcs_width;
337
338         while (1) {
339                 u64 seqnr;
340
341                 if (!fid_is_zero(&seq->lcs_fid) &&
342                     fid_oid(&seq->lcs_fid) < seq->lcs_width) {
343                         /* Just bump last allocated fid and return to caller. */
344                         seq->lcs_fid.f_oid += 1;
345                         rc = 0;
346                         break;
347                 }
348
349                 rc = seq_fid_alloc_prep(seq, &link);
350                 if (rc)
351                         continue;
352
353                 rc = seq_client_alloc_seq(env, seq, &seqnr);
354                 if (rc) {
355                         CERROR("%s: Can't allocate new sequence, "
356                                "rc %d\n", seq->lcs_name, rc);
357                         seq_fid_alloc_fini(seq);
358                         mutex_unlock(&seq->lcs_mutex);
359                         RETURN(rc);
360                 }
361
362                 CDEBUG(D_INFO, "%s: Switch to sequence "
363                        "[0x%16.16"LPF64"x]\n", seq->lcs_name, seqnr);
364
365                 seq->lcs_fid.f_oid = LUSTRE_FID_INIT_OID;
366                 seq->lcs_fid.f_seq = seqnr;
367                 seq->lcs_fid.f_ver = 0;
368
369                 /*
370                  * Inform caller that sequence switch is performed to allow it
371                  * to setup FLD for it.
372                  */
373                 rc = 1;
374
375                 seq_fid_alloc_fini(seq);
376                 break;
377         }
378
379         *fid = seq->lcs_fid;
380         mutex_unlock(&seq->lcs_mutex);
381
382         CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n", seq->lcs_name,  PFID(fid));
383         RETURN(rc);
384 }
385 EXPORT_SYMBOL(seq_client_alloc_fid);
386
387 /*
388  * Finish the current sequence due to disconnect.
389  * See mdc_import_event()
390  */
391 void seq_client_flush(struct lu_client_seq *seq)
392 {
393         wait_queue_t link;
394
395         LASSERT(seq != NULL);
396         init_waitqueue_entry_current(&link);
397         mutex_lock(&seq->lcs_mutex);
398
399         while (seq->lcs_update) {
400                 add_wait_queue(&seq->lcs_waitq, &link);
401                 set_current_state(TASK_UNINTERRUPTIBLE);
402                 mutex_unlock(&seq->lcs_mutex);
403
404                 waitq_wait(&link, TASK_UNINTERRUPTIBLE);
405
406                 mutex_lock(&seq->lcs_mutex);
407                 remove_wait_queue(&seq->lcs_waitq, &link);
408                 set_current_state(TASK_RUNNING);
409         }
410
411         fid_zero(&seq->lcs_fid);
412         /**
413          * this id shld not be used for seq range allocation.
414          * set to -1 for dgb check.
415          */
416
417         seq->lcs_space.lsr_index = -1;
418
419         range_init(&seq->lcs_space);
420         mutex_unlock(&seq->lcs_mutex);
421 }
422 EXPORT_SYMBOL(seq_client_flush);
423
424 static void seq_client_proc_fini(struct lu_client_seq *seq)
425 {
426 #ifdef CONFIG_PROC_FS
427         ENTRY;
428         if (seq->lcs_proc_dir) {
429                 if (!IS_ERR(seq->lcs_proc_dir))
430                         lprocfs_remove(&seq->lcs_proc_dir);
431                 seq->lcs_proc_dir = NULL;
432         }
433         EXIT;
434 #endif /* CONFIG_PROC_FS */
435 }
436
437 static int seq_client_proc_init(struct lu_client_seq *seq)
438 {
439 #ifdef CONFIG_PROC_FS
440         int rc;
441         ENTRY;
442
443         seq->lcs_proc_dir = lprocfs_register(seq->lcs_name, seq_type_proc_dir,
444                                              NULL, NULL);
445         if (IS_ERR(seq->lcs_proc_dir)) {
446                 CERROR("%s: LProcFS failed in seq-init\n",
447                        seq->lcs_name);
448                 rc = PTR_ERR(seq->lcs_proc_dir);
449                 RETURN(rc);
450         }
451
452         rc = lprocfs_add_vars(seq->lcs_proc_dir, seq_client_proc_list, seq);
453         if (rc) {
454                 CERROR("%s: Can't init sequence manager "
455                        "proc, rc %d\n", seq->lcs_name, rc);
456                 GOTO(out_cleanup, rc);
457         }
458
459         RETURN(0);
460
461 out_cleanup:
462         seq_client_proc_fini(seq);
463         return rc;
464
465 #else /* !CONFIG_PROC_FS */
466         return 0;
467 #endif /* CONFIG_PROC_FS */
468 }
469
470 int seq_client_init(struct lu_client_seq *seq,
471                     struct obd_export *exp,
472                     enum lu_cli_type type,
473                     const char *prefix,
474                     struct lu_server_seq *srv)
475 {
476         int rc;
477         ENTRY;
478
479         LASSERT(seq != NULL);
480         LASSERT(prefix != NULL);
481
482         seq->lcs_srv = srv;
483         seq->lcs_type = type;
484
485         mutex_init(&seq->lcs_mutex);
486         if (type == LUSTRE_SEQ_METADATA)
487                 seq->lcs_width = LUSTRE_METADATA_SEQ_MAX_WIDTH;
488         else
489                 seq->lcs_width = LUSTRE_DATA_SEQ_MAX_WIDTH;
490
491         init_waitqueue_head(&seq->lcs_waitq);
492         /* Make sure that things are clear before work is started. */
493         seq_client_flush(seq);
494
495         if (exp != NULL)
496                 seq->lcs_exp = class_export_get(exp);
497
498         snprintf(seq->lcs_name, sizeof(seq->lcs_name),
499                  "cli-%s", prefix);
500
501         rc = seq_client_proc_init(seq);
502         if (rc)
503                 seq_client_fini(seq);
504         RETURN(rc);
505 }
506 EXPORT_SYMBOL(seq_client_init);
507
508 void seq_client_fini(struct lu_client_seq *seq)
509 {
510         ENTRY;
511
512         seq_client_proc_fini(seq);
513
514         if (seq->lcs_exp != NULL) {
515                 class_export_put(seq->lcs_exp);
516                 seq->lcs_exp = NULL;
517         }
518
519         seq->lcs_srv = NULL;
520         EXIT;
521 }
522 EXPORT_SYMBOL(seq_client_fini);
523
524 int client_fid_init(struct obd_device *obd,
525                     struct obd_export *exp, enum lu_cli_type type)
526 {
527         struct client_obd *cli = &obd->u.cli;
528         char *prefix;
529         int rc;
530         ENTRY;
531
532         OBD_ALLOC_PTR(cli->cl_seq);
533         if (cli->cl_seq == NULL)
534                 RETURN(-ENOMEM);
535
536         OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
537         if (prefix == NULL)
538                 GOTO(out_free_seq, rc = -ENOMEM);
539
540         snprintf(prefix, MAX_OBD_NAME + 5, "cli-%s", obd->obd_name);
541
542         /* Init client side sequence-manager */
543         rc = seq_client_init(cli->cl_seq, exp, type, prefix, NULL);
544         OBD_FREE(prefix, MAX_OBD_NAME + 5);
545         if (rc)
546                 GOTO(out_free_seq, rc);
547
548         RETURN(rc);
549 out_free_seq:
550         OBD_FREE_PTR(cli->cl_seq);
551         cli->cl_seq = NULL;
552         return rc;
553 }
554 EXPORT_SYMBOL(client_fid_init);
555
556 int client_fid_fini(struct obd_device *obd)
557 {
558         struct client_obd *cli = &obd->u.cli;
559         ENTRY;
560
561         if (cli->cl_seq != NULL) {
562                 seq_client_fini(cli->cl_seq);
563                 OBD_FREE_PTR(cli->cl_seq);
564                 cli->cl_seq = NULL;
565         }
566
567         RETURN(0);
568 }
569 EXPORT_SYMBOL(client_fid_fini);
570
571 struct proc_dir_entry *seq_type_proc_dir;
572
573 static int __init fid_mod_init(void)
574 {
575         seq_type_proc_dir = lprocfs_register(LUSTRE_SEQ_NAME,
576                                              proc_lustre_root,
577                                              NULL, NULL);
578         if (IS_ERR(seq_type_proc_dir))
579                 return PTR_ERR(seq_type_proc_dir);
580
581 # ifdef HAVE_SERVER_SUPPORT
582         fid_server_mod_init();
583 # endif
584
585         return 0;
586 }
587
588 static void __exit fid_mod_exit(void)
589 {
590 # ifdef HAVE_SERVER_SUPPORT
591         fid_server_mod_exit();
592 # endif
593
594         if (seq_type_proc_dir != NULL && !IS_ERR(seq_type_proc_dir)) {
595                 lprocfs_remove(&seq_type_proc_dir);
596                 seq_type_proc_dir = NULL;
597         }
598 }
599
600 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
601 MODULE_DESCRIPTION("Lustre FID Module");
602 MODULE_LICENSE("GPL");
603
604 cfs_module(fid, "0.1.0", fid_mod_init, fid_mod_exit);