Whamcloud - gitweb
Branch HEAD
[fs/lustre-release.git] / lustre / fld / fld_request.c
1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  lustre/fld/fld_request.c
5  *  FLD (Fids Location Database)
6  *
7  *  Copyright (C) 2006 Cluster File Systems, Inc.
8  *   Author: Yury Umanets <umka@clusterfs.com>
9  *
10  *   This file is part of the Lustre file system, http://www.lustre.org
11  *   Lustre is a trademark of Cluster File Systems, Inc.
12  *
13  *   You may have signed or agreed to another license before downloading
14  *   this software.  If so, you are bound by the terms and conditions
15  *   of that agreement, and the following does not apply to you.  See the
16  *   LICENSE file included with this distribution for more information.
17  *
18  *   If you did not agree to a different license, then this copy of Lustre
19  *   is open source software; you can redistribute it and/or modify it
20  *   under the terms of version 2 of the GNU General Public License as
21  *   published by the Free Software Foundation.
22  *
23  *   In either case, Lustre is distributed in the hope that it will be
24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
26  *   license text for more details.
27  */
28 #ifndef EXPORT_SYMTAB
29 # define EXPORT_SYMTAB
30 #endif
31 #define DEBUG_SUBSYSTEM S_FLD
32
33 #ifdef __KERNEL__
34 # include <libcfs/libcfs.h>
35 # include <linux/module.h>
36 # include <linux/jbd.h>
37 # include <asm/div64.h>
38 #else /* __KERNEL__ */
39 # include <liblustre.h>
40 # include <libcfs/list.h>
41 #endif
42
43 #include <obd.h>
44 #include <obd_class.h>
45 #include <lustre_ver.h>
46 #include <obd_support.h>
47 #include <lprocfs_status.h>
48
49 #include <dt_object.h>
50 #include <md_object.h>
51 #include <lustre_req_layout.h>
52 #include <lustre_fld.h>
53 #include <lustre_mdc.h>
54 #include "fld_internal.h"
55
56 /* TODO: these 3 functions are copies of flow-control code from mdc_lib.c 
57  * It should be common thing. The same about mdc RPC lock */
58 static int fld_req_avail(struct client_obd *cli, struct mdc_cache_waiter *mcw)
59 {
60         int rc;
61         ENTRY;
62         spin_lock(&cli->cl_loi_list_lock);
63         rc = list_empty(&mcw->mcw_entry);
64         spin_unlock(&cli->cl_loi_list_lock);
65         RETURN(rc);
66 };
67
68 static void fld_enter_request(struct client_obd *cli)
69 {
70         struct mdc_cache_waiter mcw;
71         struct l_wait_info lwi = { 0 };
72
73         spin_lock(&cli->cl_loi_list_lock);
74         if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
75                 list_add_tail(&mcw.mcw_entry, &cli->cl_cache_waiters);
76                 init_waitqueue_head(&mcw.mcw_waitq);
77                 spin_unlock(&cli->cl_loi_list_lock);
78                 l_wait_event(mcw.mcw_waitq, fld_req_avail(cli, &mcw), &lwi);
79         } else {
80                 cli->cl_r_in_flight++;
81                 spin_unlock(&cli->cl_loi_list_lock);
82         }
83 }
84
85 static void fld_exit_request(struct client_obd *cli)
86 {
87         struct list_head *l, *tmp;
88         struct mdc_cache_waiter *mcw;
89
90         spin_lock(&cli->cl_loi_list_lock);
91         cli->cl_r_in_flight--;
92         list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
93                 
94                 if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
95                         /* No free request slots anymore */
96                         break;
97                 }
98
99                 mcw = list_entry(l, struct mdc_cache_waiter, mcw_entry);
100                 list_del_init(&mcw->mcw_entry);
101                 cli->cl_r_in_flight++;
102                 wake_up(&mcw->mcw_waitq);
103         }
104         spin_unlock(&cli->cl_loi_list_lock);
105 }
106
107 static int fld_rrb_hash(struct lu_client_fld *fld,
108                         seqno_t seq)
109 {
110         LASSERT(fld->lcf_count > 0);
111         return do_div(seq, fld->lcf_count);
112 }
113
114 static struct lu_fld_target *
115 fld_rrb_scan(struct lu_client_fld *fld, seqno_t seq)
116 {
117         struct lu_fld_target *target;
118         int hash;
119         ENTRY;
120
121         hash = fld_rrb_hash(fld, seq);
122
123         list_for_each_entry(target, &fld->lcf_targets, ft_chain) {
124                 if (target->ft_idx == hash)
125                         RETURN(target);
126         }
127
128         CERROR("%s: Can't find target by hash %d (seq "LPX64"). "
129                "Targets (%d):\n", fld->lcf_name, hash, seq,
130                fld->lcf_count);
131
132         list_for_each_entry(target, &fld->lcf_targets, ft_chain) {
133                 const char *srv_name = target->ft_srv != NULL  ?
134                         target->ft_srv->lsf_name : "<null>";
135                 const char *exp_name = target->ft_exp != NULL ?
136                         (char *)target->ft_exp->exp_obd->obd_uuid.uuid :
137                         "<null>";
138
139                 CERROR("  exp: 0x%p (%s), srv: 0x%p (%s), idx: "LPU64"\n",
140                        target->ft_exp, exp_name, target->ft_srv,
141                        srv_name, target->ft_idx);
142         }
143
144         /*
145          * If target is not found, there is logical error anyway, so here is
146          * LBUG() to catch this situation.
147          */
148         LBUG();
149         RETURN(NULL);
150 }
151
152 static int fld_dht_hash(struct lu_client_fld *fld,
153                         seqno_t seq)
154 {
155         /* XXX: here should be DHT hash */
156         return fld_rrb_hash(fld, seq);
157 }
158
159 static struct lu_fld_target *
160 fld_dht_scan(struct lu_client_fld *fld, seqno_t seq)
161 {
162         /* XXX: here should be DHT scan code */
163         return fld_rrb_scan(fld, seq);
164 }
165
166 struct lu_fld_hash fld_hash[3] = {
167         {
168                 .fh_name = "DHT",
169                 .fh_hash_func = fld_dht_hash,
170                 .fh_scan_func = fld_dht_scan
171         },
172         {
173                 .fh_name = "RRB",
174                 .fh_hash_func = fld_rrb_hash,
175                 .fh_scan_func = fld_rrb_scan
176         },
177         {
178                 0,
179         }
180 };
181
182 static struct lu_fld_target *
183 fld_client_get_target(struct lu_client_fld *fld,
184                       seqno_t seq)
185 {
186         struct lu_fld_target *target;
187         ENTRY;
188
189         LASSERT(fld->lcf_hash != NULL);
190
191         spin_lock(&fld->lcf_lock);
192         target = fld->lcf_hash->fh_scan_func(fld, seq);
193         spin_unlock(&fld->lcf_lock);
194
195         if (target != NULL) {
196                 CDEBUG(D_INFO, "%s: Found target (idx "LPU64
197                        ") by seq "LPX64"\n", fld->lcf_name,
198                        target->ft_idx, seq);
199         }
200
201         RETURN(target);
202 }
203
204 /*
205  * Add export to FLD. This is usually done by CMM and LMV as they are main users
206  * of FLD module.
207  */
208 int fld_client_add_target(struct lu_client_fld *fld,
209                           struct lu_fld_target *tar)
210 {
211         const char *name = fld_target_name(tar);
212         struct lu_fld_target *target, *tmp;
213         ENTRY;
214
215         LASSERT(tar != NULL);
216         LASSERT(name != NULL);
217         LASSERT(tar->ft_srv != NULL || tar->ft_exp != NULL);
218
219         if (fld->lcf_flags != LUSTRE_FLD_INIT) {
220                 CERROR("%s: Attempt to add target %s (idx "LPU64") "
221                        "on fly - skip it\n", fld->lcf_name, name,
222                        tar->ft_idx);
223                 RETURN(0);
224         } else {
225                 CDEBUG(D_INFO, "%s: Adding target %s (idx "
226                        LPU64")\n", fld->lcf_name, name, tar->ft_idx);
227         }
228
229         OBD_ALLOC_PTR(target);
230         if (target == NULL)
231                 RETURN(-ENOMEM);
232
233         spin_lock(&fld->lcf_lock);
234         list_for_each_entry(tmp, &fld->lcf_targets, ft_chain) {
235                 if (tmp->ft_idx == tar->ft_idx) {
236                         spin_unlock(&fld->lcf_lock);
237                         OBD_FREE_PTR(target);
238                         CERROR("Target %s exists in FLD and known as %s:#"LPU64"\n",
239                                name, fld_target_name(tmp), tmp->ft_idx);
240                         RETURN(-EEXIST);
241                 }
242         }
243
244         target->ft_exp = tar->ft_exp;
245         if (target->ft_exp != NULL)
246                 class_export_get(target->ft_exp);
247         target->ft_srv = tar->ft_srv;
248         target->ft_idx = tar->ft_idx;
249
250         list_add_tail(&target->ft_chain,
251                       &fld->lcf_targets);
252
253         fld->lcf_count++;
254         spin_unlock(&fld->lcf_lock);
255
256         RETURN(0);
257 }
258 EXPORT_SYMBOL(fld_client_add_target);
259
260 /* Remove export from FLD */
261 int fld_client_del_target(struct lu_client_fld *fld,
262                           __u64 idx)
263 {
264         struct lu_fld_target *target, *tmp;
265         ENTRY;
266
267         spin_lock(&fld->lcf_lock);
268         list_for_each_entry_safe(target, tmp,
269                                  &fld->lcf_targets, ft_chain) {
270                 if (target->ft_idx == idx) {
271                         fld->lcf_count--;
272                         list_del(&target->ft_chain);
273                         spin_unlock(&fld->lcf_lock);
274
275                         if (target->ft_exp != NULL)
276                                 class_export_put(target->ft_exp);
277
278                         OBD_FREE_PTR(target);
279                         RETURN(0);
280                 }
281         }
282         spin_unlock(&fld->lcf_lock);
283         RETURN(-ENOENT);
284 }
285 EXPORT_SYMBOL(fld_client_del_target);
286
287 static void fld_client_proc_fini(struct lu_client_fld *fld);
288
289 #ifdef LPROCFS
290 static int fld_client_proc_init(struct lu_client_fld *fld)
291 {
292         int rc;
293         ENTRY;
294
295         fld->lcf_proc_dir = lprocfs_register(fld->lcf_name,
296                                              fld_type_proc_dir,
297                                              NULL, NULL);
298
299         if (IS_ERR(fld->lcf_proc_dir)) {
300                 CERROR("%s: LProcFS failed in fld-init\n",
301                        fld->lcf_name);
302                 rc = PTR_ERR(fld->lcf_proc_dir);
303                 RETURN(rc);
304         }
305
306         rc = lprocfs_add_vars(fld->lcf_proc_dir,
307                               fld_client_proc_list, fld);
308         if (rc) {
309                 CERROR("%s: Can't init FLD proc, rc %d\n",
310                        fld->lcf_name, rc);
311                 GOTO(out_cleanup, rc);
312         }
313
314         RETURN(0);
315
316 out_cleanup:
317         fld_client_proc_fini(fld);
318         return rc;
319 }
320
321 static void fld_client_proc_fini(struct lu_client_fld *fld)
322 {
323         ENTRY;
324         if (fld->lcf_proc_dir) {
325                 if (!IS_ERR(fld->lcf_proc_dir))
326                         lprocfs_remove(&fld->lcf_proc_dir);
327                 fld->lcf_proc_dir = NULL;
328         }
329         EXIT;
330 }
331 #else
332 static int fld_client_proc_init(struct lu_client_fld *fld)
333 {
334         return 0;
335 }
336
337 static void fld_client_proc_fini(struct lu_client_fld *fld)
338 {
339         return;
340 }
341 #endif
342
343 static inline int hash_is_sane(int hash)
344 {
345         return (hash >= 0 && hash < ARRAY_SIZE(fld_hash));
346 }
347
348 int fld_client_init(struct lu_client_fld *fld,
349                     const char *prefix, int hash)
350 {
351 #ifdef __KERNEL__
352         int cache_size, cache_threshold;
353 #endif
354         int rc;
355         ENTRY;
356
357         LASSERT(fld != NULL);
358
359         snprintf(fld->lcf_name, sizeof(fld->lcf_name),
360                  "cli-%s", prefix);
361
362         if (!hash_is_sane(hash)) {
363                 CERROR("%s: Wrong hash function %#x\n",
364                        fld->lcf_name, hash);
365                 RETURN(-EINVAL);
366         }
367
368         fld->lcf_count = 0;
369         spin_lock_init(&fld->lcf_lock);
370         fld->lcf_hash = &fld_hash[hash];
371         fld->lcf_flags = LUSTRE_FLD_INIT;
372         INIT_LIST_HEAD(&fld->lcf_targets);
373
374 #ifdef __KERNEL__
375         cache_size = FLD_CLIENT_CACHE_SIZE /
376                 sizeof(struct fld_cache_entry);
377
378         cache_threshold = cache_size *
379                 FLD_CLIENT_CACHE_THRESHOLD / 100;
380
381         fld->lcf_cache = fld_cache_init(fld->lcf_name,
382                                         FLD_CLIENT_HTABLE_SIZE,
383                                         cache_size, cache_threshold);
384         if (IS_ERR(fld->lcf_cache)) {
385                 rc = PTR_ERR(fld->lcf_cache);
386                 fld->lcf_cache = NULL;
387                 GOTO(out, rc);
388         }
389 #endif
390
391         rc = fld_client_proc_init(fld);
392         if (rc)
393                 GOTO(out, rc);
394         EXIT;
395 out:
396         if (rc)
397                 fld_client_fini(fld);
398         else
399                 CDEBUG(D_INFO, "%s: Using \"%s\" hash\n",
400                        fld->lcf_name, fld->lcf_hash->fh_name);
401         return rc;
402 }
403 EXPORT_SYMBOL(fld_client_init);
404
405 void fld_client_fini(struct lu_client_fld *fld)
406 {
407         struct lu_fld_target *target, *tmp;
408         ENTRY;
409
410         fld_client_proc_fini(fld);
411
412         spin_lock(&fld->lcf_lock);
413         list_for_each_entry_safe(target, tmp,
414                                  &fld->lcf_targets, ft_chain) {
415                 fld->lcf_count--;
416                 list_del(&target->ft_chain);
417                 if (target->ft_exp != NULL)
418                         class_export_put(target->ft_exp);
419                 OBD_FREE_PTR(target);
420         }
421         spin_unlock(&fld->lcf_lock);
422
423 #ifdef __KERNEL__
424         if (fld->lcf_cache != NULL) {
425                 if (!IS_ERR(fld->lcf_cache))
426                         fld_cache_fini(fld->lcf_cache);
427                 fld->lcf_cache = NULL;
428         }
429 #endif
430
431         EXIT;
432 }
433 EXPORT_SYMBOL(fld_client_fini);
434
435 static int fld_client_rpc(struct obd_export *exp,
436                           struct md_fld *mf, __u32 fld_op)
437 {
438         int size[3] = { sizeof(struct ptlrpc_body),
439                         sizeof(__u32),
440                         sizeof(struct md_fld) };
441         struct ptlrpc_request *req;
442         struct req_capsule pill;
443         struct md_fld *pmf;
444         __u32 *op;
445         int rc;
446         ENTRY;
447
448         LASSERT(exp != NULL);
449
450         req = ptlrpc_prep_req(class_exp2cliimp(exp),
451                               LUSTRE_MDS_VERSION,
452                               FLD_QUERY, 3, size,
453                               NULL);
454         if (req == NULL)
455                 RETURN(-ENOMEM);
456
457         req_capsule_init(&pill, req, RCL_CLIENT, NULL);
458         req_capsule_set(&pill, &RQF_FLD_QUERY);
459
460         op = req_capsule_client_get(&pill, &RMF_FLD_OPC);
461         *op = fld_op;
462
463         pmf = req_capsule_client_get(&pill, &RMF_FLD_MDFLD);
464         *pmf = *mf;
465
466         size[1] = sizeof(struct md_fld);
467         ptlrpc_req_set_repsize(req, 2, size);
468         req->rq_request_portal = FLD_REQUEST_PORTAL;
469
470         if (fld_op != FLD_LOOKUP)
471                 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
472         fld_enter_request(&exp->exp_obd->u.cli);
473         rc = ptlrpc_queue_wait(req);
474         fld_exit_request(&exp->exp_obd->u.cli);
475         if (fld_op != FLD_LOOKUP)
476                 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL);
477         if (rc)
478                 GOTO(out_req, rc);
479
480         pmf = req_capsule_server_get(&pill, &RMF_FLD_MDFLD);
481         if (pmf == NULL)
482                 GOTO(out_req, rc = -EFAULT);
483         *mf = *pmf;
484         EXIT;
485 out_req:
486         req_capsule_fini(&pill);
487         ptlrpc_req_finished(req);
488         return rc;
489 }
490
491 int fld_client_create(struct lu_client_fld *fld,
492                       seqno_t seq, mdsno_t mds,
493                       const struct lu_env *env)
494 {
495         struct md_fld md_fld = { .mf_seq = seq, .mf_mds = mds };
496         struct lu_fld_target *target;
497         int rc;
498         ENTRY;
499
500         fld->lcf_flags |= LUSTRE_FLD_RUN;
501         target = fld_client_get_target(fld, seq);
502         LASSERT(target != NULL);
503
504         CDEBUG(D_INFO, "%s: Create fld entry (seq: "LPX64"; mds: "
505                LPU64") on target %s (idx "LPU64")\n", fld->lcf_name,
506                seq, mds, fld_target_name(target), target->ft_idx);
507
508 #ifdef __KERNEL__
509         if (target->ft_srv != NULL) {
510                 LASSERT(env != NULL);
511                 rc = fld_server_create(target->ft_srv, env, seq, mds);
512         } else {
513 #endif
514                 rc = fld_client_rpc(target->ft_exp, &md_fld, FLD_CREATE);
515 #ifdef __KERNEL__
516         }
517 #endif
518
519         if (rc == 0) {
520                 /*
521                  * Do not return result of calling fld_cache_insert()
522                  * here. First of all because it may return -EEXISTS. Another
523                  * reason is that, we do not want to stop proceeding because of
524                  * cache errors.
525                  */
526                 fld_cache_insert(fld->lcf_cache, seq, mds);
527         } else {
528                 CERROR("%s: Can't create FLD entry, rc %d\n",
529                        fld->lcf_name, rc);
530         }
531
532         RETURN(rc);
533 }
534 EXPORT_SYMBOL(fld_client_create);
535
536 int fld_client_delete(struct lu_client_fld *fld, seqno_t seq,
537                       const struct lu_env *env)
538 {
539         struct md_fld md_fld = { .mf_seq = seq, .mf_mds = 0 };
540         struct lu_fld_target *target;
541         int rc;
542         ENTRY;
543
544         fld->lcf_flags |= LUSTRE_FLD_RUN;
545         fld_cache_delete(fld->lcf_cache, seq);
546
547         target = fld_client_get_target(fld, seq);
548         LASSERT(target != NULL);
549
550         CDEBUG(D_INFO, "%s: Delete fld entry (seq: "LPX64") on "
551                "target %s (idx "LPU64")\n", fld->lcf_name, seq,
552                fld_target_name(target), target->ft_idx);
553
554 #ifdef __KERNEL__
555         if (target->ft_srv != NULL) {
556                 LASSERT(env != NULL);
557                 rc = fld_server_delete(target->ft_srv,
558                                        env, seq);
559         } else {
560 #endif
561                 rc = fld_client_rpc(target->ft_exp,
562                                     &md_fld, FLD_DELETE);
563 #ifdef __KERNEL__
564         }
565 #endif
566
567         RETURN(rc);
568 }
569 EXPORT_SYMBOL(fld_client_delete);
570
571 int fld_client_lookup(struct lu_client_fld *fld,
572                       seqno_t seq, mdsno_t *mds,
573                       const struct lu_env *env)
574 {
575         struct md_fld md_fld = { .mf_seq = seq, .mf_mds = 0 };
576         struct lu_fld_target *target;
577         int rc;
578         ENTRY;
579
580         fld->lcf_flags |= LUSTRE_FLD_RUN;
581
582         rc = fld_cache_lookup(fld->lcf_cache, seq, mds);
583         if (rc == 0)
584                 RETURN(0);
585
586         /* Can not find it in the cache */
587         target = fld_client_get_target(fld, seq);
588         LASSERT(target != NULL);
589
590         CDEBUG(D_INFO, "%s: Lookup fld entry (seq: "LPX64") on "
591                "target %s (idx "LPU64")\n", fld->lcf_name, seq,
592                fld_target_name(target), target->ft_idx);
593
594 #ifdef __KERNEL__
595         if (target->ft_srv != NULL) {
596                 LASSERT(env != NULL);
597                 rc = fld_server_lookup(target->ft_srv,
598                                        env, seq, &md_fld.mf_mds);
599         } else {
600 #endif
601                 /*
602                  * insert the 'inflight' sequence. No need to protect that,
603                  * we are trying to reduce numbers of RPC but not restrict
604                  * to them exactly one 
605                  */
606                 fld_cache_insert_inflight(fld->lcf_cache, seq);
607                 rc = fld_client_rpc(target->ft_exp,
608                                     &md_fld, FLD_LOOKUP);
609 #ifdef __KERNEL__
610         }
611 #endif
612         if (rc == 0) {
613                 *mds = md_fld.mf_mds;
614
615                 /*
616                  * Do not return error here as well. See previous comment in
617                  * same situation in function fld_client_create().
618                  */
619                 fld_cache_insert(fld->lcf_cache, seq, *mds);
620         } else {
621                 /* remove 'inflight' seq if it exists */
622                 fld_cache_delete(fld->lcf_cache, seq);
623         }
624         RETURN(rc);
625 }
626 EXPORT_SYMBOL(fld_client_lookup);
627
628 void fld_client_flush(struct lu_client_fld *fld)
629 {
630 #ifdef __KERNEL__
631         fld_cache_flush(fld->lcf_cache);
632 #endif
633 }
634 EXPORT_SYMBOL(fld_client_flush);
635