Whamcloud - gitweb
436cba2a634b68ce313405673b4def51b6480aac
[fs/lustre-release.git] / lustre / fid / fid_request.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
30  * Use is subject to license terms.
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 #ifndef EXPORT_SYMTAB
44 # define EXPORT_SYMTAB
45 #endif
46 #define DEBUG_SUBSYSTEM S_FID
47
48 #ifdef __KERNEL__
49 # include <libcfs/libcfs.h>
50 # include <linux/module.h>
51 #else /* __KERNEL__ */
52 # include <liblustre.h>
53 #endif
54
55 #include <obd.h>
56 #include <obd_class.h>
57 #include <dt_object.h>
58 #include <md_object.h>
59 #include <obd_support.h>
60 #include <lustre_req_layout.h>
61 #include <lustre_fid.h>
62 /* mdc RPC locks */
63 #include <lustre_mdc.h>
64 #include "fid_internal.h"
65
66 static int seq_client_rpc(struct lu_client_seq *seq,
67                           struct lu_seq_range *output, __u32 opc,
68                           const char *opcname)
69 {
70         struct obd_export     *exp = seq->lcs_exp;
71         struct ptlrpc_request *req;
72         struct lu_seq_range   *out, *in;
73         __u32                 *op;
74         int                    rc;
75         ENTRY;
76
77         req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_SEQ_QUERY,
78                                         LUSTRE_MDS_VERSION, SEQ_QUERY);
79         if (req == NULL)
80                 RETURN(-ENOMEM);
81
82         /* Init operation code */
83         op = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_OPC);
84         *op = opc;
85
86         /* Zero out input range, this is not recovery yet. */
87         in = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_RANGE);
88         range_init(in);
89
90         ptlrpc_request_set_replen(req);
91
92         if (seq->lcs_type == LUSTRE_SEQ_METADATA) {
93                 req->rq_request_portal = (opc == SEQ_ALLOC_SUPER) ?
94                         SEQ_CONTROLLER_PORTAL : SEQ_METADATA_PORTAL;
95                 /* update mdt field of *in, it is required for fld update
96                  * on super sequence allocator node. */
97                 if (opc == SEQ_ALLOC_SUPER)
98                         in->lsr_mdt = seq->lcs_space.lsr_mdt;
99         } else {
100                 LASSERT(opc == SEQ_ALLOC_META);
101                 req->rq_request_portal = SEQ_DATA_PORTAL;
102         }
103         ptlrpc_at_set_req_timeout(req);
104
105         mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
106         rc = ptlrpc_queue_wait(req);
107         mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
108
109         if (rc)
110                 GOTO(out_req, rc);
111
112         out = req_capsule_server_get(&req->rq_pill, &RMF_SEQ_RANGE);
113         *output = *out;
114
115         if (!range_is_sane(output)) {
116                 CERROR("%s: Invalid range received from server: "
117                        DRANGE"\n", seq->lcs_name, PRANGE(output));
118                 GOTO(out_req, rc = -EINVAL);
119         }
120
121         if (range_is_exhausted(output)) {
122                 CERROR("%s: Range received from server is exhausted: "
123                        DRANGE"]\n", seq->lcs_name, PRANGE(output));
124                 GOTO(out_req, rc = -EINVAL);
125         }
126
127         CDEBUG(D_INFO, "%s: Allocated %s-sequence "DRANGE"]\n",
128                seq->lcs_name, opcname, PRANGE(output));
129
130         EXIT;
131 out_req:
132         ptlrpc_req_finished(req);
133         return rc;
134 }
135
136 /* Request sequence-controller node to allocate new super-sequence. */
137 int seq_client_alloc_super(struct lu_client_seq *seq,
138                            const struct lu_env *env)
139 {
140         int rc;
141         ENTRY;
142
143         cfs_down(&seq->lcs_sem);
144
145 #ifdef __KERNEL__
146         if (seq->lcs_srv) {
147                 LASSERT(env != NULL);
148                 rc = seq_server_alloc_super(seq->lcs_srv, &seq->lcs_space,
149                                             env);
150         } else {
151 #endif
152                 rc = seq_client_rpc(seq, &seq->lcs_space,
153                                     SEQ_ALLOC_SUPER, "super");
154 #ifdef __KERNEL__
155         }
156 #endif
157         cfs_up(&seq->lcs_sem);
158         RETURN(rc);
159 }
160
161 /* Request sequence-controller node to allocate new meta-sequence. */
162 static int seq_client_alloc_meta(struct lu_client_seq *seq,
163                                  const struct lu_env *env)
164 {
165         int rc;
166         ENTRY;
167
168 #ifdef __KERNEL__
169         if (seq->lcs_srv) {
170                 LASSERT(env != NULL);
171                 rc = seq_server_alloc_meta(seq->lcs_srv, &seq->lcs_space, env);
172         } else {
173 #endif
174                 rc = seq_client_rpc(seq, &seq->lcs_space,
175                                     SEQ_ALLOC_META, "meta");
176 #ifdef __KERNEL__
177         }
178 #endif
179         RETURN(rc);
180 }
181
182 /* Allocate new sequence for client. */
183 static int seq_client_alloc_seq(struct lu_client_seq *seq, seqno_t *seqnr)
184 {
185         int rc;
186         ENTRY;
187
188         LASSERT(range_is_sane(&seq->lcs_space));
189
190         if (range_is_exhausted(&seq->lcs_space)) {
191                 rc = seq_client_alloc_meta(seq, NULL);
192                 if (rc) {
193                         CERROR("%s: Can't allocate new meta-sequence, "
194                                "rc %d\n", seq->lcs_name, rc);
195                         RETURN(rc);
196                 } else {
197                         CDEBUG(D_INFO, "%s: New range - "DRANGE"\n",
198                                seq->lcs_name, PRANGE(&seq->lcs_space));
199                 }
200         } else {
201                 rc = 0;
202         }
203
204         LASSERT(!range_is_exhausted(&seq->lcs_space));
205         *seqnr = seq->lcs_space.lsr_start;
206         seq->lcs_space.lsr_start += 1;
207
208         CDEBUG(D_INFO, "%s: Allocated sequence ["LPX64"]\n", seq->lcs_name,
209                *seqnr);
210
211         RETURN(rc);
212 }
213
214 /* Allocate new fid on passed client @seq and save it to @fid. */
215 int seq_client_alloc_fid(struct lu_client_seq *seq, struct lu_fid *fid)
216 {
217         int rc;
218         ENTRY;
219
220         LASSERT(seq != NULL);
221         LASSERT(fid != NULL);
222
223         cfs_down(&seq->lcs_sem);
224
225         if (fid_is_zero(&seq->lcs_fid) ||
226             fid_oid(&seq->lcs_fid) >= seq->lcs_width)
227         {
228                 seqno_t seqnr;
229
230                 rc = seq_client_alloc_seq(seq, &seqnr);
231                 if (rc) {
232                         CERROR("%s: Can't allocate new sequence, "
233                                "rc %d\n", seq->lcs_name, rc);
234                         cfs_up(&seq->lcs_sem);
235                         RETURN(rc);
236                 }
237
238                 CDEBUG(D_INFO, "%s: Switch to sequence "
239                        "[0x%16.16"LPF64"x]\n", seq->lcs_name, seqnr);
240
241                 seq->lcs_fid.f_oid = LUSTRE_FID_INIT_OID;
242                 seq->lcs_fid.f_seq = seqnr;
243                 seq->lcs_fid.f_ver = 0;
244
245                 /*
246                  * Inform caller that sequence switch is performed to allow it
247                  * to setup FLD for it.
248                  */
249                 rc = 1;
250         } else {
251                 /* Just bump last allocated fid and return to caller. */
252                 seq->lcs_fid.f_oid += 1;
253                 rc = 0;
254         }
255
256         *fid = seq->lcs_fid;
257         cfs_up(&seq->lcs_sem);
258
259         CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n", seq->lcs_name,  PFID(fid));
260         RETURN(rc);
261 }
262 EXPORT_SYMBOL(seq_client_alloc_fid);
263
264 /*
265  * Finish the current sequence due to disconnect.
266  * See mdc_import_event()
267  */
268 void seq_client_flush(struct lu_client_seq *seq)
269 {
270         LASSERT(seq != NULL);
271         cfs_down(&seq->lcs_sem);
272         fid_zero(&seq->lcs_fid);
273         /**
274          * this id shld not be used for seq range allocation.
275          * set to -1 for dgb check.
276          */
277
278         seq->lcs_space.lsr_mdt = -1;
279
280         range_init(&seq->lcs_space);
281         cfs_up(&seq->lcs_sem);
282 }
283 EXPORT_SYMBOL(seq_client_flush);
284
285 static void seq_client_proc_fini(struct lu_client_seq *seq);
286
287 #ifdef LPROCFS
288 static int seq_client_proc_init(struct lu_client_seq *seq)
289 {
290         int rc;
291         ENTRY;
292
293         seq->lcs_proc_dir = lprocfs_register(seq->lcs_name,
294                                              seq_type_proc_dir,
295                                              NULL, NULL);
296
297         if (IS_ERR(seq->lcs_proc_dir)) {
298                 CERROR("%s: LProcFS failed in seq-init\n",
299                        seq->lcs_name);
300                 rc = PTR_ERR(seq->lcs_proc_dir);
301                 RETURN(rc);
302         }
303
304         rc = lprocfs_add_vars(seq->lcs_proc_dir,
305                               seq_client_proc_list, seq);
306         if (rc) {
307                 CERROR("%s: Can't init sequence manager "
308                        "proc, rc %d\n", seq->lcs_name, rc);
309                 GOTO(out_cleanup, rc);
310         }
311
312         RETURN(0);
313
314 out_cleanup:
315         seq_client_proc_fini(seq);
316         return rc;
317 }
318
319 static void seq_client_proc_fini(struct lu_client_seq *seq)
320 {
321         ENTRY;
322         if (seq->lcs_proc_dir) {
323                 if (!IS_ERR(seq->lcs_proc_dir))
324                         lprocfs_remove(&seq->lcs_proc_dir);
325                 seq->lcs_proc_dir = NULL;
326         }
327         EXIT;
328 }
329 #else
330 static int seq_client_proc_init(struct lu_client_seq *seq)
331 {
332         return 0;
333 }
334
335 static void seq_client_proc_fini(struct lu_client_seq *seq)
336 {
337         return;
338 }
339 #endif
340
341 int seq_client_init(struct lu_client_seq *seq,
342                     struct obd_export *exp,
343                     enum lu_cli_type type,
344                     const char *prefix,
345                     struct lu_server_seq *srv)
346 {
347         int rc;
348         ENTRY;
349
350         LASSERT(seq != NULL);
351         LASSERT(prefix != NULL);
352
353         seq->lcs_exp = exp;
354         seq->lcs_srv = srv;
355         seq->lcs_type = type;
356         cfs_sema_init(&seq->lcs_sem, 1);
357         seq->lcs_width = LUSTRE_SEQ_MAX_WIDTH;
358
359         /* Make sure that things are clear before work is started. */
360         seq_client_flush(seq);
361
362         if (exp == NULL) {
363                 LASSERT(seq->lcs_srv != NULL);
364         } else {
365                 LASSERT(seq->lcs_exp != NULL);
366                 seq->lcs_exp = class_export_get(seq->lcs_exp);
367         }
368
369         snprintf(seq->lcs_name, sizeof(seq->lcs_name),
370                  "cli-%s", prefix);
371
372         rc = seq_client_proc_init(seq);
373         if (rc)
374                 seq_client_fini(seq);
375         RETURN(rc);
376 }
377 EXPORT_SYMBOL(seq_client_init);
378
379 void seq_client_fini(struct lu_client_seq *seq)
380 {
381         ENTRY;
382
383         seq_client_proc_fini(seq);
384
385         if (seq->lcs_exp != NULL) {
386                 class_export_put(seq->lcs_exp);
387                 seq->lcs_exp = NULL;
388         }
389
390         seq->lcs_srv = NULL;
391         EXIT;
392 }
393 EXPORT_SYMBOL(seq_client_fini);