Whamcloud - gitweb
b=19486 add server identifier into lu_seq_range.
[fs/lustre-release.git] / lustre / fid / fid_handler.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_handler.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 <lu_target.h>
58 #include <dt_object.h>
59 #include <md_object.h>
60 #include <obd_support.h>
61 #include <lustre_req_layout.h>
62 #include <lustre_fid.h>
63 #include "fid_internal.h"
64
65 #ifdef __KERNEL__
66 /* Assigns client to sequence controller node. */
67 int seq_server_set_cli(struct lu_server_seq *seq,
68                        struct lu_client_seq *cli,
69                        const struct lu_env *env)
70 {
71         int rc = 0;
72         ENTRY;
73
74         /*
75          * Ask client for new range, assign that range to ->seq_space and write
76          * seq state to backing store should be atomic.
77          */
78         cfs_down(&seq->lss_sem);
79
80         if (cli == NULL) {
81                 CDEBUG(D_INFO, "%s: Detached sequence client %s\n",
82                        seq->lss_name, cli->lcs_name);
83                 seq->lss_cli = cli;
84                 GOTO(out_up, rc = 0);
85         }
86
87         if (seq->lss_cli != NULL) {
88                 CERROR("%s: Sequence controller is already "
89                        "assigned\n", seq->lss_name);
90                 GOTO(out_up, rc = -EINVAL);
91         }
92
93         CDEBUG(D_INFO, "%s: Attached sequence controller %s\n",
94                seq->lss_name, cli->lcs_name);
95
96         seq->lss_cli = cli;
97         cli->lcs_space.lsr_index = seq->lss_site->ms_node_id;
98         EXIT;
99 out_up:
100         cfs_up(&seq->lss_sem);
101         return rc;
102 }
103 EXPORT_SYMBOL(seq_server_set_cli);
104 /*
105  * allocate \a w units of sequence from range \a from.
106  */
107 static inline void range_alloc(struct lu_seq_range *to,
108                                struct lu_seq_range *from,
109                                __u64 width)
110 {
111         width = min(range_space(from), width);
112         to->lsr_start = from->lsr_start;
113         to->lsr_end = from->lsr_start + width;
114         from->lsr_start += width;
115 }
116
117 /**
118  * On controller node, allocate new super sequence for regular sequence server.
119  * As this super sequence controller, this node suppose to maintain fld
120  * and update index.
121  * \a out range always has currect mds node number of requester.
122  */
123
124 static int __seq_server_alloc_super(struct lu_server_seq *seq,
125                                     struct lu_seq_range *out,
126                                     const struct lu_env *env)
127 {
128         struct lu_seq_range *space = &seq->lss_space;
129         int rc;
130         ENTRY;
131
132         LASSERT(range_is_sane(space));
133
134         if (range_is_exhausted(space)) {
135                 CERROR("%s: Sequences space is exhausted\n",
136                        seq->lss_name);
137                 RETURN(-ENOSPC);
138         } else {
139                 range_alloc(out, space, seq->lss_width);
140         }
141
142         rc = seq_store_update(env, seq, out, 1 /* sync */);
143
144         CDEBUG(D_INFO, "%s: super-sequence allocation rc = %d "
145                DRANGE"\n", seq->lss_name, rc, PRANGE(out));
146
147         RETURN(rc);
148 }
149
150 int seq_server_alloc_super(struct lu_server_seq *seq,
151                            struct lu_seq_range *out,
152                            const struct lu_env *env)
153 {
154         int rc;
155         ENTRY;
156
157         cfs_down(&seq->lss_sem);
158         rc = __seq_server_alloc_super(seq, out, env);
159         cfs_up(&seq->lss_sem);
160
161         RETURN(rc);
162 }
163
164 static int __seq_set_init(const struct lu_env *env,
165                             struct lu_server_seq *seq)
166 {
167         struct lu_seq_range *space = &seq->lss_space;
168         int rc;
169
170         range_alloc(&seq->lss_lowater_set, space, seq->lss_set_width);
171         range_alloc(&seq->lss_hiwater_set, space, seq->lss_set_width);
172
173         rc = seq_store_update(env, seq, NULL, 1);
174         seq->lss_set_transno = 0;
175
176         return rc;
177 }
178
179 /*
180  * This function implements new seq allocation algorithm using async
181  * updates to seq file on disk. ref bug 18857 for details.
182  * there are four variable to keep track of this process
183  *
184  * lss_space; - available lss_space
185  * lss_lowater_set; - lu_seq_range for all seqs before barrier, i.e. safe to use
186  * lss_hiwater_set; - lu_seq_range after barrier, i.e. allocated but may be
187  *                    not yet committed
188  *
189  * when lss_lowater_set reaches the end it is replaced with hiwater one and
190  * a write operation is initiated to allocate new hiwater range.
191  * if last seq write opearion is still not commited, current operation is
192  * flaged as sync write op.
193  */
194 static int range_alloc_set(const struct lu_env *env,
195                             struct lu_seq_range *out,
196                             struct lu_server_seq *seq)
197 {
198         struct lu_seq_range *space = &seq->lss_space;
199         struct lu_seq_range *loset = &seq->lss_lowater_set;
200         struct lu_seq_range *hiset = &seq->lss_hiwater_set;
201         int rc = 0;
202
203         if (range_is_zero(loset))
204                 __seq_set_init(env, seq);
205
206         if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_ALLOC)) /* exhaust set */
207                 loset->lsr_start = loset->lsr_end;
208
209         if (range_is_exhausted(loset)) {
210                 /* reached high water mark. */
211                 struct lu_device *dev = seq->lss_site->ms_lu.ls_top_dev;
212                 struct lu_target *tg = dev->ld_obd->u.obt.obt_lut;
213                 int obd_num_clients = dev->ld_obd->obd_num_exports;
214                 __u64 set_sz;
215                 int sync = 0;
216
217                 /* calculate new seq width based on number of clients */
218                 set_sz = max(seq->lss_set_width,
219                                obd_num_clients * seq->lss_width);
220                 set_sz = min(range_space(space), set_sz);
221
222                 /* Switch to hiwater range now */
223                 loset = hiset;
224                 /* allocate new hiwater range */
225                 range_alloc(hiset, space, set_sz);
226
227                 if (seq->lss_set_transno > dev->ld_obd->obd_last_committed)
228                         sync = 1;
229
230                 /* update ondisk seq with new *space */
231                 rc = seq_store_update(env, seq, NULL, sync);
232
233                 /* set new hiwater transno */
234                 cfs_spin_lock(&tg->lut_translock);
235                 seq->lss_set_transno = tg->lut_last_transno;
236                 cfs_spin_unlock(&tg->lut_translock);
237         }
238
239         LASSERTF(!range_is_exhausted(loset) || range_is_sane(loset),
240                  DRANGE"\n", PRANGE(loset));
241
242         if (rc == 0)
243                 range_alloc(out, loset, seq->lss_width);
244
245         RETURN(rc);
246 }
247
248 static int __seq_server_alloc_meta(struct lu_server_seq *seq,
249                                    struct lu_seq_range *out,
250                                    const struct lu_env *env)
251 {
252         struct lu_seq_range *space = &seq->lss_space;
253         int rc = 0;
254
255         ENTRY;
256
257         LASSERT(range_is_sane(space));
258
259         /* Check if available space ends and allocate new super seq */
260         if (range_is_exhausted(space)) {
261                 if (!seq->lss_cli) {
262                         CERROR("%s: No sequence controller is attached.\n",
263                                seq->lss_name);
264                         RETURN(-ENODEV);
265                 }
266
267                 rc = seq_client_alloc_super(seq->lss_cli, env);
268                 if (rc) {
269                         CERROR("%s: Can't allocate super-sequence, rc %d\n",
270                                seq->lss_name, rc);
271                         RETURN(rc);
272                 }
273
274                 /* Saving new range to allocation space. */
275                 *space = seq->lss_cli->lcs_space;
276                 LASSERT(range_is_sane(space));
277         }
278
279         rc = range_alloc_set(env, out, seq);
280         if (rc == 0) {
281                 CDEBUG(D_INFO, "%s: Allocated meta-sequence "
282                        DRANGE"\n", seq->lss_name, PRANGE(out));
283         }
284
285         RETURN(rc);
286 }
287
288 int seq_server_alloc_meta(struct lu_server_seq *seq,
289                           struct lu_seq_range *out,
290                           const struct lu_env *env)
291 {
292         int rc;
293         ENTRY;
294
295         cfs_down(&seq->lss_sem);
296         rc = __seq_server_alloc_meta(seq, out, env);
297         cfs_up(&seq->lss_sem);
298
299         RETURN(rc);
300 }
301 EXPORT_SYMBOL(seq_server_alloc_meta);
302
303 static int seq_server_handle(struct lu_site *site,
304                              const struct lu_env *env,
305                              __u32 opc, struct lu_seq_range *out)
306 {
307         int rc;
308         struct md_site *mite;
309         ENTRY;
310
311         mite = lu_site2md(site);
312         switch (opc) {
313         case SEQ_ALLOC_META:
314                 if (!mite->ms_server_seq) {
315                         CERROR("Sequence server is not "
316                                "initialized\n");
317                         RETURN(-EINVAL);
318                 }
319                 rc = seq_server_alloc_meta(mite->ms_server_seq, out, env);
320                 break;
321         case SEQ_ALLOC_SUPER:
322                 if (!mite->ms_control_seq) {
323                         CERROR("Sequence controller is not "
324                                "initialized\n");
325                         RETURN(-EINVAL);
326                 }
327                 rc = seq_server_alloc_super(mite->ms_control_seq, out, env);
328                 break;
329         default:
330                 rc = -EINVAL;
331                 break;
332         }
333
334         RETURN(rc);
335 }
336
337 static int seq_req_handle(struct ptlrpc_request *req,
338                           const struct lu_env *env,
339                           struct seq_thread_info *info)
340 {
341         struct lu_seq_range *out, *tmp;
342         struct lu_site *site;
343         int rc = -EPROTO;
344         __u32 *opc;
345         ENTRY;
346
347         LASSERT(!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY));
348         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
349         LASSERT(site != NULL);
350
351         rc = req_capsule_server_pack(info->sti_pill);
352         if (rc)
353                 RETURN(err_serious(rc));
354
355         opc = req_capsule_client_get(info->sti_pill, &RMF_SEQ_OPC);
356         if (opc != NULL) {
357                 out = req_capsule_server_get(info->sti_pill, &RMF_SEQ_RANGE);
358                 if (out == NULL)
359                         RETURN(err_serious(-EPROTO));
360
361                 tmp = req_capsule_client_get(info->sti_pill, &RMF_SEQ_RANGE);
362
363                 /* seq client passed mdt id, we need to pass that using out
364                  * range parameter */
365
366                 out->lsr_index = tmp->lsr_index;
367                 out->lsr_flags = tmp->lsr_flags;
368                 rc = seq_server_handle(site, env, *opc, out);
369         } else
370                 rc = err_serious(-EPROTO);
371
372         RETURN(rc);
373 }
374
375 /* context key constructor/destructor: seq_key_init, seq_key_fini */
376 LU_KEY_INIT_FINI(seq, struct seq_thread_info);
377
378 /* context key: seq_thread_key */
379 LU_CONTEXT_KEY_DEFINE(seq, LCT_MD_THREAD);
380
381 static void seq_thread_info_init(struct ptlrpc_request *req,
382                                  struct seq_thread_info *info)
383 {
384         info->sti_pill = &req->rq_pill;
385         /* Init request capsule */
386         req_capsule_init(info->sti_pill, req, RCL_SERVER);
387         req_capsule_set(info->sti_pill, &RQF_SEQ_QUERY);
388 }
389
390 static void seq_thread_info_fini(struct seq_thread_info *info)
391 {
392         req_capsule_fini(info->sti_pill);
393 }
394
395 static int seq_handle(struct ptlrpc_request *req)
396 {
397         const struct lu_env *env;
398         struct seq_thread_info *info;
399         int rc;
400
401         env = req->rq_svc_thread->t_env;
402         LASSERT(env != NULL);
403
404         info = lu_context_key_get(&env->le_ctx, &seq_thread_key);
405         LASSERT(info != NULL);
406
407         seq_thread_info_init(req, info);
408         rc = seq_req_handle(req, env, info);
409         /* XXX: we don't need replay but MDT assign transno in any case,
410          * remove it manually before reply*/
411         lustre_msg_set_transno(req->rq_repmsg, 0);
412         seq_thread_info_fini(info);
413
414         return rc;
415 }
416
417 /*
418  * Entry point for handling FLD RPCs called from MDT.
419  */
420 int seq_query(struct com_thread_info *info)
421 {
422         return seq_handle(info->cti_pill->rc_req);
423 }
424 EXPORT_SYMBOL(seq_query);
425
426 static void seq_server_proc_fini(struct lu_server_seq *seq);
427
428 #ifdef LPROCFS
429 static int seq_server_proc_init(struct lu_server_seq *seq)
430 {
431         int rc;
432         ENTRY;
433
434         seq->lss_proc_dir = lprocfs_register(seq->lss_name,
435                                              seq_type_proc_dir,
436                                              NULL, NULL);
437         if (IS_ERR(seq->lss_proc_dir)) {
438                 rc = PTR_ERR(seq->lss_proc_dir);
439                 RETURN(rc);
440         }
441
442         rc = lprocfs_add_vars(seq->lss_proc_dir,
443                               seq_server_proc_list, seq);
444         if (rc) {
445                 CERROR("%s: Can't init sequence manager "
446                        "proc, rc %d\n", seq->lss_name, rc);
447                 GOTO(out_cleanup, rc);
448         }
449
450         RETURN(0);
451
452 out_cleanup:
453         seq_server_proc_fini(seq);
454         return rc;
455 }
456
457 static void seq_server_proc_fini(struct lu_server_seq *seq)
458 {
459         ENTRY;
460         if (seq->lss_proc_dir != NULL) {
461                 if (!IS_ERR(seq->lss_proc_dir))
462                         lprocfs_remove(&seq->lss_proc_dir);
463                 seq->lss_proc_dir = NULL;
464         }
465         EXIT;
466 }
467 #else
468 static int seq_server_proc_init(struct lu_server_seq *seq)
469 {
470         return 0;
471 }
472
473 static void seq_server_proc_fini(struct lu_server_seq *seq)
474 {
475         return;
476 }
477 #endif
478
479
480 int seq_server_init(struct lu_server_seq *seq,
481                     struct dt_device *dev,
482                     const char *prefix,
483                     enum lu_mgr_type type,
484                     struct md_site *ms,
485                     const struct lu_env *env)
486 {
487         int rc, is_srv = (type == LUSTRE_SEQ_SERVER);
488         ENTRY;
489
490         LASSERT(dev != NULL);
491         LASSERT(prefix != NULL);
492
493         seq->lss_cli = NULL;
494         seq->lss_type = type;
495         seq->lss_site = ms;
496         range_init(&seq->lss_space);
497
498         range_init(&seq->lss_lowater_set);
499         range_init(&seq->lss_hiwater_set);
500         seq->lss_set_width = LUSTRE_SEQ_BATCH_WIDTH;
501
502         cfs_sema_init(&seq->lss_sem, 1);
503
504         seq->lss_width = is_srv ?
505                 LUSTRE_SEQ_META_WIDTH : LUSTRE_SEQ_SUPER_WIDTH;
506
507         snprintf(seq->lss_name, sizeof(seq->lss_name),
508                  "%s-%s", (is_srv ? "srv" : "ctl"), prefix);
509
510         rc = seq_store_init(seq, env, dev);
511         if (rc)
512                 GOTO(out, rc);
513         /* Request backing store for saved sequence info. */
514         rc = seq_store_read(seq, env);
515         if (rc == -ENODATA) {
516
517                 /* Nothing is read, init by default value. */
518                 seq->lss_space = is_srv ?
519                         LUSTRE_SEQ_ZERO_RANGE:
520                         LUSTRE_SEQ_SPACE_RANGE;
521
522                 seq->lss_space.lsr_index = ms->ms_node_id;
523                 CDEBUG(D_INFO, "%s: No data found "
524                        "on store. Initialize space\n",
525                        seq->lss_name);
526
527                 rc = seq_store_update(env, seq, NULL, 0);
528                 if (rc) {
529                         CERROR("%s: Can't write space data, "
530                                "rc %d\n", seq->lss_name, rc);
531                 }
532         } else if (rc) {
533                 CERROR("%s: Can't read space data, rc %d\n",
534                        seq->lss_name, rc);
535                 GOTO(out, rc);
536         }
537
538         if (is_srv) {
539                 LASSERT(range_is_sane(&seq->lss_space));
540         } else {
541                 LASSERT(!range_is_zero(&seq->lss_space) &&
542                         range_is_sane(&seq->lss_space));
543         }
544
545         rc  = seq_server_proc_init(seq);
546         if (rc)
547                 GOTO(out, rc);
548
549         EXIT;
550 out:
551         if (rc)
552                 seq_server_fini(seq, env);
553         return rc;
554 }
555 EXPORT_SYMBOL(seq_server_init);
556
557 void seq_server_fini(struct lu_server_seq *seq,
558                      const struct lu_env *env)
559 {
560         ENTRY;
561
562         seq_server_proc_fini(seq);
563         seq_store_fini(seq, env);
564
565         EXIT;
566 }
567 EXPORT_SYMBOL(seq_server_fini);
568
569 cfs_proc_dir_entry_t *seq_type_proc_dir = NULL;
570
571 static struct lu_local_obj_desc llod_seq_srv = {
572         .llod_name      = LUSTRE_SEQ_SRV_NAME,
573         .llod_oid       = FID_SEQ_SRV_OID,
574         .llod_is_index  = 0,
575 };
576
577 static struct lu_local_obj_desc llod_seq_ctl = {
578         .llod_name      = LUSTRE_SEQ_CTL_NAME,
579         .llod_oid       = FID_SEQ_CTL_OID,
580         .llod_is_index  = 0,
581 };
582
583 static int __init fid_mod_init(void)
584 {
585         seq_type_proc_dir = lprocfs_register(LUSTRE_SEQ_NAME,
586                                              proc_lustre_root,
587                                              NULL, NULL);
588         if (IS_ERR(seq_type_proc_dir))
589                 return PTR_ERR(seq_type_proc_dir);
590
591         llo_local_obj_register(&llod_seq_srv);
592         llo_local_obj_register(&llod_seq_ctl);
593
594         LU_CONTEXT_KEY_INIT(&seq_thread_key);
595         lu_context_key_register(&seq_thread_key);
596         return 0;
597 }
598
599 static void __exit fid_mod_exit(void)
600 {
601         llo_local_obj_unregister(&llod_seq_srv);
602         llo_local_obj_unregister(&llod_seq_ctl);
603
604         lu_context_key_degister(&seq_thread_key);
605         if (seq_type_proc_dir != NULL && !IS_ERR(seq_type_proc_dir)) {
606                 lprocfs_remove(&seq_type_proc_dir);
607                 seq_type_proc_dir = NULL;
608         }
609 }
610
611 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
612 MODULE_DESCRIPTION("Lustre FID Module");
613 MODULE_LICENSE("GPL");
614
615 cfs_module(fid, "0.1.0", fid_mod_init, fid_mod_exit);
616 #endif