4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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
23 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24 * Use is subject to license terms.
26 * Copyright (c) 2011, 2017, Intel Corporation.
29 * This file is part of Lustre, http://www.lustre.org/
30 * Lustre is a trademark of Sun Microsystems, Inc.
32 * lustre/fid/fid_request.c
34 * Lustre Sequence Manager
36 * Author: Yury Umanets <umka@clusterfs.com>
39 #define DEBUG_SUBSYSTEM S_FID
41 #include <linux/module.h>
43 #include <obd_class.h>
44 #include <obd_support.h>
45 #include <lustre_fid.h>
47 #include <lustre_mdc.h>
48 #include "fid_internal.h"
50 static int seq_client_rpc(struct lu_client_seq *seq,
51 struct lu_seq_range *output, __u32 opc,
54 struct obd_export *exp = seq->lcs_exp;
55 struct ptlrpc_request *req;
56 struct lu_seq_range *out, *in;
58 unsigned int debug_mask;
62 LASSERT(exp != NULL && !IS_ERR(exp));
63 req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_SEQ_QUERY,
64 LUSTRE_MDS_VERSION, SEQ_QUERY);
68 /* Init operation code */
69 op = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_OPC);
72 /* Zero out input range, this is not recovery yet. */
73 in = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_RANGE);
74 lu_seq_range_init(in);
76 ptlrpc_request_set_replen(req);
78 in->lsr_index = seq->lcs_space.lsr_index;
79 if (seq->lcs_type == LUSTRE_SEQ_METADATA)
80 fld_range_set_mdt(in);
82 fld_range_set_ost(in);
84 if (opc == SEQ_ALLOC_SUPER) {
85 req->rq_request_portal = SEQ_CONTROLLER_PORTAL;
86 req->rq_reply_portal = MDC_REPLY_PORTAL;
87 /* During allocating super sequence for data object,
88 * the current thread might hold the export of MDT0(MDT0
89 * precreating objects on this OST), and it will send the
90 * request to MDT0 here, so we can not keep resending the
91 * request here, otherwise if MDT0 is failed(umounted),
92 * it can not release the export of MDT0 */
93 if (seq->lcs_type == LUSTRE_SEQ_DATA)
94 req->rq_no_delay = req->rq_no_resend = 1;
95 debug_mask = D_CONSOLE;
97 if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
98 req->rq_reply_portal = MDC_REPLY_PORTAL;
99 req->rq_request_portal = SEQ_METADATA_PORTAL;
101 req->rq_reply_portal = OSC_REPLY_PORTAL;
102 req->rq_request_portal = SEQ_DATA_PORTAL;
108 /* Allow seq client RPC during recovery time. */
109 req->rq_allow_replay = 1;
111 ptlrpc_at_set_req_timeout(req);
113 rc = ptlrpc_queue_wait(req);
118 out = req_capsule_server_get(&req->rq_pill, &RMF_SEQ_RANGE);
121 if (!lu_seq_range_is_sane(output)) {
122 CERROR("%s: Invalid range received from server: "
123 DRANGE"\n", seq->lcs_name, PRANGE(output));
124 GOTO(out_req, rc = -EINVAL);
127 if (lu_seq_range_is_exhausted(output)) {
128 CERROR("%s: Range received from server is exhausted: "
129 DRANGE"]\n", seq->lcs_name, PRANGE(output));
130 GOTO(out_req, rc = -EINVAL);
133 CDEBUG_LIMIT(debug_mask, "%s: Allocated %s-sequence "DRANGE"]\n",
134 seq->lcs_name, opcname, PRANGE(output));
138 ptlrpc_req_finished(req);
142 /* Request sequence-controller node to allocate new super-sequence. */
143 int seq_client_alloc_super(struct lu_client_seq *seq,
144 const struct lu_env *env)
149 mutex_lock(&seq->lcs_mutex);
152 #ifdef HAVE_SEQ_SERVER
153 LASSERT(env != NULL);
154 rc = seq_server_alloc_super(seq->lcs_srv, &seq->lcs_space,
160 /* Check whether the connection to seq controller has been
161 * setup (lcs_exp != NULL) */
162 if (seq->lcs_exp == NULL) {
163 mutex_unlock(&seq->lcs_mutex);
164 RETURN(-EINPROGRESS);
167 rc = seq_client_rpc(seq, &seq->lcs_space,
168 SEQ_ALLOC_SUPER, "super");
170 mutex_unlock(&seq->lcs_mutex);
174 /* Request sequence-controller node to allocate new meta-sequence. */
175 static int seq_client_alloc_meta(const struct lu_env *env,
176 struct lu_client_seq *seq)
182 #ifdef HAVE_SEQ_SERVER
183 LASSERT(env != NULL);
184 rc = seq_server_alloc_meta(seq->lcs_srv, &seq->lcs_space, env);
190 /* If meta server return -EINPROGRESS or EAGAIN,
191 * it means meta server might not be ready to
192 * allocate super sequence from sequence controller
194 rc = seq_client_rpc(seq, &seq->lcs_space,
195 SEQ_ALLOC_META, "meta");
196 if (rc == -EINPROGRESS || rc == -EAGAIN) {
197 wait_queue_head_t waitq;
198 struct l_wait_info lwi;
200 /* MDT0 is not ready, let's wait for 2
201 * seconds and retry. */
202 init_waitqueue_head(&waitq);
203 lwi = LWI_TIMEOUT(cfs_time_seconds(2), NULL,
205 l_wait_event(waitq, 0, &lwi);
207 } while (rc == -EINPROGRESS || rc == -EAGAIN);
213 /* Allocate new sequence for client. */
214 static int seq_client_alloc_seq(const struct lu_env *env,
215 struct lu_client_seq *seq, u64 *seqnr)
220 LASSERT(lu_seq_range_is_sane(&seq->lcs_space));
222 if (lu_seq_range_is_exhausted(&seq->lcs_space)) {
223 rc = seq_client_alloc_meta(env, seq);
225 if (rc != -EINPROGRESS)
226 CERROR("%s: Can't allocate new meta-sequence,"
227 "rc = %d\n", seq->lcs_name, rc);
230 CDEBUG(D_INFO, "%s: New range - "DRANGE"\n",
231 seq->lcs_name, PRANGE(&seq->lcs_space));
237 LASSERT(!lu_seq_range_is_exhausted(&seq->lcs_space));
238 *seqnr = seq->lcs_space.lsr_start;
239 seq->lcs_space.lsr_start += 1;
241 CDEBUG(D_INFO, "%s: Allocated sequence [%#llx]\n", seq->lcs_name,
247 static int seq_fid_alloc_prep(struct lu_client_seq *seq,
250 if (seq->lcs_update) {
251 add_wait_queue(&seq->lcs_waitq, link);
252 set_current_state(TASK_UNINTERRUPTIBLE);
253 mutex_unlock(&seq->lcs_mutex);
257 mutex_lock(&seq->lcs_mutex);
258 remove_wait_queue(&seq->lcs_waitq, link);
259 set_current_state(TASK_RUNNING);
264 mutex_unlock(&seq->lcs_mutex);
269 static void seq_fid_alloc_fini(struct lu_client_seq *seq, __u64 seqnr,
272 LASSERT(seq->lcs_update == 1);
274 mutex_lock(&seq->lcs_mutex);
276 CDEBUG(D_INFO, "%s: New sequence [0x%16.16llx]\n",
277 seq->lcs_name, seqnr);
279 seq->lcs_fid.f_seq = seqnr;
281 /* Since the caller require the whole seq,
282 * so marked this seq to be used */
283 if (seq->lcs_type == LUSTRE_SEQ_METADATA)
285 LUSTRE_METADATA_SEQ_MAX_WIDTH;
287 seq->lcs_fid.f_oid = LUSTRE_DATA_SEQ_MAX_WIDTH;
289 seq->lcs_fid.f_oid = LUSTRE_FID_INIT_OID;
291 seq->lcs_fid.f_ver = 0;
295 wake_up_all(&seq->lcs_waitq);
299 * Allocate the whole non-used seq to the caller.
301 * \param[in] env pointer to the thread context
302 * \param[in,out] seq pointer to the client sequence manager
303 * \param[out] seqnr to hold the new allocated sequence
305 * \retval 0 for new sequence allocated.
306 * \retval Negative error number on failure.
308 int seq_client_get_seq(const struct lu_env *env,
309 struct lu_client_seq *seq, u64 *seqnr)
314 LASSERT(seqnr != NULL);
316 mutex_lock(&seq->lcs_mutex);
317 init_waitqueue_entry(&link, current);
319 /* To guarantee that we can get a whole non-used sequence. */
320 while (seq_fid_alloc_prep(seq, &link) != 0);
322 rc = seq_client_alloc_seq(env, seq, seqnr);
323 seq_fid_alloc_fini(seq, rc ? 0 : *seqnr, true);
325 CERROR("%s: Can't allocate new sequence: rc = %d\n",
327 mutex_unlock(&seq->lcs_mutex);
331 EXPORT_SYMBOL(seq_client_get_seq);
334 * Allocate new fid on passed client @seq and save it to @fid.
336 * \param[in] env pointer to the thread context
337 * \param[in,out] seq pointer to the client sequence manager
338 * \param[out] fid to hold the new allocated fid
340 * \retval 1 for notify the caller that sequence switch
341 * is performed to allow it to setup FLD for it.
342 * \retval 0 for new FID allocated in current sequence.
343 * \retval Negative error number on failure.
345 int seq_client_alloc_fid(const struct lu_env *env,
346 struct lu_client_seq *seq, struct lu_fid *fid)
352 LASSERT(seq != NULL);
353 LASSERT(fid != NULL);
355 init_waitqueue_entry(&link, current);
356 mutex_lock(&seq->lcs_mutex);
358 if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_EXHAUST))
359 seq->lcs_fid.f_oid = seq->lcs_width;
364 if (unlikely(!fid_is_zero(&seq->lcs_fid) &&
365 fid_oid(&seq->lcs_fid) < seq->lcs_width)) {
366 /* Just bump last allocated fid and return to caller. */
367 seq->lcs_fid.f_oid++;
372 /* Release seq::lcs_mutex via seq_fid_alloc_prep() to avoid
373 * deadlock during seq_client_alloc_seq(). */
374 rc = seq_fid_alloc_prep(seq, &link);
378 rc = seq_client_alloc_seq(env, seq, &seqnr);
379 /* Re-take seq::lcs_mutex via seq_fid_alloc_fini(). */
380 seq_fid_alloc_fini(seq, rc ? 0 : seqnr, false);
382 if (rc != -EINPROGRESS)
383 CERROR("%s: Can't allocate new sequence: "
384 "rc = %d\n", seq->lcs_name, rc);
385 mutex_unlock(&seq->lcs_mutex);
395 mutex_unlock(&seq->lcs_mutex);
397 CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n", seq->lcs_name, PFID(fid));
401 EXPORT_SYMBOL(seq_client_alloc_fid);
404 * Finish the current sequence due to disconnect.
405 * See mdc_import_event()
407 void seq_client_flush(struct lu_client_seq *seq)
411 LASSERT(seq != NULL);
412 init_waitqueue_entry(&link, current);
413 mutex_lock(&seq->lcs_mutex);
415 while (seq->lcs_update) {
416 add_wait_queue(&seq->lcs_waitq, &link);
417 set_current_state(TASK_UNINTERRUPTIBLE);
418 mutex_unlock(&seq->lcs_mutex);
422 mutex_lock(&seq->lcs_mutex);
423 remove_wait_queue(&seq->lcs_waitq, &link);
424 set_current_state(TASK_RUNNING);
427 fid_zero(&seq->lcs_fid);
429 * this id shld not be used for seq range allocation.
430 * set to -1 for dgb check.
433 seq->lcs_space.lsr_index = -1;
435 lu_seq_range_init(&seq->lcs_space);
436 mutex_unlock(&seq->lcs_mutex);
438 EXPORT_SYMBOL(seq_client_flush);
440 static void seq_client_proc_fini(struct lu_client_seq *seq)
442 #ifdef CONFIG_PROC_FS
444 if (seq->lcs_proc_dir) {
445 if (!IS_ERR(seq->lcs_proc_dir))
446 lprocfs_remove(&seq->lcs_proc_dir);
447 seq->lcs_proc_dir = NULL;
450 #endif /* CONFIG_PROC_FS */
453 static int seq_client_proc_init(struct lu_client_seq *seq)
455 #ifdef CONFIG_PROC_FS
459 seq->lcs_proc_dir = lprocfs_register(seq->lcs_name, seq_type_proc_dir,
461 if (IS_ERR(seq->lcs_proc_dir)) {
462 CERROR("%s: LProcFS failed in seq-init\n",
464 rc = PTR_ERR(seq->lcs_proc_dir);
468 rc = lprocfs_add_vars(seq->lcs_proc_dir, seq_client_proc_list, seq);
470 CERROR("%s: Can't init sequence manager "
471 "proc, rc %d\n", seq->lcs_name, rc);
472 GOTO(out_cleanup, rc);
478 seq_client_proc_fini(seq);
481 #else /* !CONFIG_PROC_FS */
483 #endif /* CONFIG_PROC_FS */
486 int seq_client_init(struct lu_client_seq *seq,
487 struct obd_export *exp,
488 enum lu_cli_type type,
490 struct lu_server_seq *srv)
495 LASSERT(seq != NULL);
496 LASSERT(prefix != NULL);
499 seq->lcs_type = type;
501 mutex_init(&seq->lcs_mutex);
502 if (type == LUSTRE_SEQ_METADATA)
503 seq->lcs_width = LUSTRE_METADATA_SEQ_MAX_WIDTH;
505 seq->lcs_width = LUSTRE_DATA_SEQ_MAX_WIDTH;
507 init_waitqueue_head(&seq->lcs_waitq);
508 /* Make sure that things are clear before work is started. */
509 seq_client_flush(seq);
512 seq->lcs_exp = class_export_get(exp);
514 snprintf(seq->lcs_name, sizeof(seq->lcs_name),
517 rc = seq_client_proc_init(seq);
519 seq_client_fini(seq);
522 EXPORT_SYMBOL(seq_client_init);
524 void seq_client_fini(struct lu_client_seq *seq)
528 seq_client_proc_fini(seq);
530 if (seq->lcs_exp != NULL) {
531 class_export_put(seq->lcs_exp);
538 EXPORT_SYMBOL(seq_client_fini);
540 int client_fid_init(struct obd_device *obd,
541 struct obd_export *exp, enum lu_cli_type type)
543 struct client_obd *cli = &obd->u.cli;
548 down_write(&cli->cl_seq_rwsem);
549 OBD_ALLOC_PTR(cli->cl_seq);
551 GOTO(out, rc = -ENOMEM);
553 OBD_ALLOC(prefix, MAX_OBD_NAME + 5);
555 GOTO(out, rc = -ENOMEM);
557 snprintf(prefix, MAX_OBD_NAME + 5, "cli-%s", obd->obd_name);
559 /* Init client side sequence-manager */
560 rc = seq_client_init(cli->cl_seq, exp, type, prefix, NULL);
561 OBD_FREE(prefix, MAX_OBD_NAME + 5);
566 if (rc && cli->cl_seq) {
567 OBD_FREE_PTR(cli->cl_seq);
570 up_write(&cli->cl_seq_rwsem);
574 EXPORT_SYMBOL(client_fid_init);
576 int client_fid_fini(struct obd_device *obd)
578 struct client_obd *cli = &obd->u.cli;
581 down_write(&cli->cl_seq_rwsem);
583 seq_client_fini(cli->cl_seq);
584 OBD_FREE_PTR(cli->cl_seq);
587 up_write(&cli->cl_seq_rwsem);
591 EXPORT_SYMBOL(client_fid_fini);
593 struct proc_dir_entry *seq_type_proc_dir;
595 static int __init fid_init(void)
597 seq_type_proc_dir = lprocfs_register(LUSTRE_SEQ_NAME,
600 if (IS_ERR(seq_type_proc_dir))
601 return PTR_ERR(seq_type_proc_dir);
603 # ifdef HAVE_SERVER_SUPPORT
604 fid_server_mod_init();
610 static void __exit fid_exit(void)
612 # ifdef HAVE_SERVER_SUPPORT
613 fid_server_mod_exit();
616 if (seq_type_proc_dir != NULL && !IS_ERR(seq_type_proc_dir)) {
617 lprocfs_remove(&seq_type_proc_dir);
618 seq_type_proc_dir = NULL;
622 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
623 MODULE_DESCRIPTION("Lustre File IDentifier");
624 MODULE_VERSION(LUSTRE_VERSION_STRING);
625 MODULE_LICENSE("GPL");
627 module_init(fid_init);
628 module_exit(fid_exit);