Whamcloud - gitweb
Branch HEAD
[fs/lustre-release.git] / lustre / fld / fld_handler.c
1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  lustre/fld/fld_handler.c
5  *  FLD (Fids Location Database)
6  *
7  *  Copyright (C) 2006 Cluster File Systems, Inc.
8  *   Author: Yury Umanets <umka@clusterfs.com>
9  *           WangDi <wangdi@clusterfs.com>
10  *
11  *   This file is part of the Lustre file system, http://www.lustre.org
12  *   Lustre is a trademark of Cluster File Systems, Inc.
13  *
14  *   You may have signed or agreed to another license before downloading
15  *   this software.  If so, you are bound by the terms and conditions
16  *   of that agreement, and the following does not apply to you.  See the
17  *   LICENSE file included with this distribution for more information.
18  *
19  *   If you did not agree to a different license, then this copy of Lustre
20  *   is open source software; you can redistribute it and/or modify it
21  *   under the terms of version 2 of the GNU General Public License as
22  *   published by the Free Software Foundation.
23  *
24  *   In either case, Lustre is distributed in the hope that it will be
25  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
26  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
27  *   license text for more details.
28  */
29 #ifndef EXPORT_SYMTAB
30 # define EXPORT_SYMTAB
31 #endif
32 #define DEBUG_SUBSYSTEM S_FLD
33
34 #ifdef __KERNEL__
35 # include <libcfs/libcfs.h>
36 # include <linux/module.h>
37 # include <linux/jbd.h>
38 # include <asm/div64.h>
39 #else /* __KERNEL__ */
40 # include <liblustre.h>
41 # include <libcfs/list.h>
42 #endif
43
44 #include <obd.h>
45 #include <obd_class.h>
46 #include <lustre_ver.h>
47 #include <obd_support.h>
48 #include <lprocfs_status.h>
49
50 #include <md_object.h>
51 #include <lustre_req_layout.h>
52 #include "fld_internal.h"
53
54 #ifdef __KERNEL__
55
56 /* context key constructor/destructor: fld_key_init, fld_key_fini */
57 LU_KEY_INIT_FINI(fld, struct fld_thread_info);
58
59 /* context key: fld_thread_key */
60 LU_CONTEXT_KEY_DEFINE(fld, LCT_MD_THREAD|LCT_DT_THREAD);
61
62 cfs_proc_dir_entry_t *fld_type_proc_dir = NULL;
63
64 static int __init fld_mod_init(void)
65 {
66         fld_type_proc_dir = lprocfs_register(LUSTRE_FLD_NAME,
67                                              proc_lustre_root,
68                                              NULL, NULL);
69         if (IS_ERR(fld_type_proc_dir))
70                 return PTR_ERR(fld_type_proc_dir);
71
72         LU_CONTEXT_KEY_INIT(&fld_thread_key);
73         lu_context_key_register(&fld_thread_key);
74         return 0;
75 }
76
77 static void __exit fld_mod_exit(void)
78 {
79         lu_context_key_degister(&fld_thread_key);
80         if (fld_type_proc_dir != NULL && !IS_ERR(fld_type_proc_dir)) {
81                 lprocfs_remove(&fld_type_proc_dir);
82                 fld_type_proc_dir = NULL;
83         }
84 }
85
86 /* Insert index entry and update cache. */
87 int fld_server_create(struct lu_server_fld *fld,
88                       const struct lu_env *env,
89                       seqno_t seq, mdsno_t mds)
90 {
91         int rc;
92         ENTRY;
93         
94         rc = fld_index_create(fld, env, seq, mds);
95         
96         if (rc == 0) {
97                 /*
98                  * Do not return result of calling fld_cache_insert()
99                  * here. First of all because it may return -EEXISTS. Another
100                  * reason is that, we do not want to stop proceeding even after
101                  * cache errors.
102                  */
103                 fld_cache_insert(fld->lsf_cache, seq, mds);
104         }
105
106         RETURN(rc);
107 }
108 EXPORT_SYMBOL(fld_server_create);
109
110 /* Delete index entry. */
111 int fld_server_delete(struct lu_server_fld *fld,
112                       const struct lu_env *env,
113                       seqno_t seq)
114 {
115         int rc;
116         ENTRY;
117
118         fld_cache_delete(fld->lsf_cache, seq);
119         rc = fld_index_delete(fld, env, seq);
120         
121         RETURN(rc);
122 }
123 EXPORT_SYMBOL(fld_server_delete);
124
125 /* Lookup mds by seq. */
126 int fld_server_lookup(struct lu_server_fld *fld,
127                       const struct lu_env *env,
128                       seqno_t seq, mdsno_t *mds)
129 {
130         int rc;
131         ENTRY;
132         
133         /* Lookup it in the cache. */
134         rc = fld_cache_lookup(fld->lsf_cache, seq, mds);
135         if (rc == 0)
136                 RETURN(0);
137
138         rc = fld_index_lookup(fld, env, seq, mds);
139         if (rc == 0) {
140                 /*
141                  * Do not return error here as well. See previous comment in
142                  * same situation in function fld_server_create().
143                  */
144                 fld_cache_insert(fld->lsf_cache, seq, *mds);
145         }
146         RETURN(rc);
147 }
148 EXPORT_SYMBOL(fld_server_lookup);
149
150 static int fld_server_handle(struct lu_server_fld *fld,
151                              const struct lu_env *env,
152                              __u32 opc, struct md_fld *mf,
153                              struct fld_thread_info *info)
154 {
155         int rc;
156         ENTRY;
157
158         switch (opc) {
159         case FLD_CREATE:
160                 rc = fld_server_create(fld, env,
161                                        mf->mf_seq, mf->mf_mds);
162
163                 /* Do not return -EEXIST error for resent case */
164                 if ((info->fti_flags & MSG_RESENT) && rc == -EEXIST)
165                         rc = 0;
166                 break;
167         case FLD_DELETE:
168                 rc = fld_server_delete(fld, env, mf->mf_seq);
169
170                 /* Do not return -ENOENT error for resent case */
171                 if ((info->fti_flags & MSG_RESENT) && rc == -ENOENT)
172                         rc = 0;
173                 break;
174         case FLD_LOOKUP:
175                 rc = fld_server_lookup(fld, env,
176                                        mf->mf_seq, &mf->mf_mds);
177                 break;
178         default:
179                 rc = -EINVAL;
180                 break;
181         }
182
183         CDEBUG(D_INFO, "%s: FLD req handle: error %d (opc: %d, seq: "
184                LPX64", mds: "LPU64")\n", fld->lsf_name, rc, opc,
185                mf->mf_seq, mf->mf_mds);
186         
187         RETURN(rc);
188
189 }
190
191 static int fld_req_handle(struct ptlrpc_request *req,
192                           struct fld_thread_info *info)
193 {
194         struct lu_site *site;
195         struct md_fld *in;
196         struct md_fld *out;
197         int rc;
198         __u32 *opc;
199         ENTRY;
200
201         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
202
203         rc = req_capsule_pack(&info->fti_pill);
204         if (rc)
205                 RETURN(err_serious(rc));
206
207         opc = req_capsule_client_get(&info->fti_pill, &RMF_FLD_OPC);
208         if (opc != NULL) {
209                 in = req_capsule_client_get(&info->fti_pill, &RMF_FLD_MDFLD);
210                 if (in == NULL)
211                         RETURN(err_serious(-EPROTO));
212                 out = req_capsule_server_get(&info->fti_pill, &RMF_FLD_MDFLD);
213                 if (out == NULL)
214                         RETURN(err_serious(-EPROTO));
215                 *out = *in;
216
217                 rc = fld_server_handle(site->ls_server_fld,
218                                        req->rq_svc_thread->t_env,
219                                        *opc, out, info);
220         } else
221                 rc = err_serious(-EPROTO);
222
223         RETURN(rc);
224 }
225
226 static void fld_thread_info_init(struct ptlrpc_request *req,
227                                  struct fld_thread_info *info)
228 {
229         int i;
230
231         info->fti_flags = lustre_msg_get_flags(req->rq_reqmsg);
232
233         /* Mark rep buffer as req-layout stuff expects. */
234         for (i = 0; i < ARRAY_SIZE(info->fti_rep_buf_size); i++)
235                 info->fti_rep_buf_size[i] = -1;
236
237         /* Init request capsule. */
238         req_capsule_init(&info->fti_pill, req, RCL_SERVER,
239                          info->fti_rep_buf_size);
240
241         req_capsule_set(&info->fti_pill, &RQF_FLD_QUERY);
242 }
243
244 static void fld_thread_info_fini(struct fld_thread_info *info)
245 {
246         req_capsule_fini(&info->fti_pill);
247 }
248
249 static int fld_handle(struct ptlrpc_request *req)
250 {
251         struct fld_thread_info *info;
252         const struct lu_env *env;
253         int rc;
254
255         env = req->rq_svc_thread->t_env;
256         LASSERT(env != NULL);
257
258         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
259         LASSERT(info != NULL);
260
261         fld_thread_info_init(req, info);
262         rc = fld_req_handle(req, info);
263         fld_thread_info_fini(info);
264
265         return rc;
266 }
267
268 /*
269  * Entry point for handling FLD RPCs called from MDT.
270  */
271 int fld_query(struct com_thread_info *info)
272 {
273         return fld_handle(info->cti_pill.rc_req);
274 }
275 EXPORT_SYMBOL(fld_query);
276
277 /*
278  * Returns true, if fid is local to this server node.
279  *
280  * WARNING: this function is *not* guaranteed to return false if fid is
281  * remote: it makes an educated conservative guess only.
282  *
283  * fid_is_local() is supposed to be used in assertion checks only.
284  */
285 int fid_is_local(struct lu_site *site, const struct lu_fid *fid)
286 {
287         int result;
288
289         result = 1; /* conservatively assume fid is local */
290         if (site->ls_client_fld != NULL) {
291                 mdsno_t mds;
292                 int rc;
293
294                 rc = fld_cache_lookup(site->ls_client_fld->lcf_cache,
295                                       fid_seq(fid), &mds);
296                 if (rc == 0)
297                         result = (mds == site->ls_node_id);
298         }
299         return result;
300 }
301 EXPORT_SYMBOL(fid_is_local);
302
303 static void fld_server_proc_fini(struct lu_server_fld *fld);
304
305 #ifdef LPROCFS
306 static int fld_server_proc_init(struct lu_server_fld *fld)
307 {
308         int rc = 0;
309         ENTRY;
310
311         fld->lsf_proc_dir = lprocfs_register(fld->lsf_name,
312                                              fld_type_proc_dir,
313                                              fld_server_proc_list, fld);
314         if (IS_ERR(fld->lsf_proc_dir)) {
315                 rc = PTR_ERR(fld->lsf_proc_dir);
316                 RETURN(rc);
317         }
318
319         RETURN(rc);
320 }
321
322 static void fld_server_proc_fini(struct lu_server_fld *fld)
323 {
324         ENTRY;
325         if (fld->lsf_proc_dir != NULL) {
326                 if (!IS_ERR(fld->lsf_proc_dir))
327                         lprocfs_remove(&fld->lsf_proc_dir);
328                 fld->lsf_proc_dir = NULL;
329         }
330         EXIT;
331 }
332 #else
333 static int fld_server_proc_init(struct lu_server_fld *fld)
334 {
335         return 0;
336 }
337
338 static void fld_server_proc_fini(struct lu_server_fld *fld)
339 {
340         return;
341 }
342 #endif
343
344 int fld_server_init(struct lu_server_fld *fld, struct dt_device *dt,
345                     const char *prefix, const struct lu_env *env)
346 {
347         int cache_size, cache_threshold;
348         int rc;
349         ENTRY;
350
351         snprintf(fld->lsf_name, sizeof(fld->lsf_name),
352                  "srv-%s", prefix);
353
354         cache_size = FLD_SERVER_CACHE_SIZE /
355                 sizeof(struct fld_cache_entry);
356
357         cache_threshold = cache_size *
358                 FLD_SERVER_CACHE_THRESHOLD / 100;
359
360         fld->lsf_cache = fld_cache_init(fld->lsf_name,
361                                         FLD_SERVER_HTABLE_SIZE,
362                                         cache_size, cache_threshold);
363         if (IS_ERR(fld->lsf_cache)) {
364                 rc = PTR_ERR(fld->lsf_cache);
365                 fld->lsf_cache = NULL;
366                 GOTO(out, rc);
367         }
368
369         rc = fld_index_init(fld, env, dt);
370         if (rc)
371                 GOTO(out, rc);
372
373         rc = fld_server_proc_init(fld);
374         if (rc)
375                 GOTO(out, rc);
376
377         EXIT;
378 out:
379         if (rc)
380                 fld_server_fini(fld, env);
381         return rc;
382 }
383 EXPORT_SYMBOL(fld_server_init);
384
385 void fld_server_fini(struct lu_server_fld *fld,
386                      const struct lu_env *env)
387 {
388         ENTRY;
389
390         fld_server_proc_fini(fld);
391         fld_index_fini(fld, env);
392
393         if (fld->lsf_cache != NULL) {
394                 if (!IS_ERR(fld->lsf_cache))
395                         fld_cache_fini(fld->lsf_cache);
396                 fld->lsf_cache = NULL;
397         }
398
399         EXIT;
400 }
401 EXPORT_SYMBOL(fld_server_fini);
402
403 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
404 MODULE_DESCRIPTION("Lustre FLD");
405 MODULE_LICENSE("GPL");
406
407 cfs_module(mdd, "0.1.0", fld_mod_init, fld_mod_exit);
408 #endif