Whamcloud - gitweb
LU-6158 mdt: always shrink_capsule in getxattr_all
[fs/lustre-release.git] / lustre / fld / fld_handler.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
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.
9  *
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).
15  *
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.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2014, Intel Corporation.
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/fld/fld_handler.c
37  *
38  * FLD (Fids Location Database)
39  *
40  * Author: Yury Umanets <umka@clusterfs.com>
41  * Author: WangDi <wangdi@clusterfs.com>
42  * Author: Pravin Shelar <pravin.shelar@sun.com>
43  */
44
45 #define DEBUG_SUBSYSTEM S_FLD
46
47 #include <libcfs/libcfs.h>
48 #include <linux/module.h>
49
50 #include <obd.h>
51 #include <obd_support.h>
52 #include <lustre_fid.h>
53 #include <lustre_fld.h>
54 #include <lustre_req_layout.h>
55 #include <lprocfs_status.h>
56 #include "fld_internal.h"
57
58 /* context key constructor/destructor: fld_key_init, fld_key_fini */
59 LU_KEY_INIT_FINI(fld, struct fld_thread_info);
60
61 /* context key: fld_thread_key */
62 /* MGS thread may create llog file causing FLD lookup */
63 LU_CONTEXT_KEY_DEFINE(fld, LCT_MD_THREAD | LCT_DT_THREAD | LCT_MG_THREAD);
64
65 int fld_server_mod_init(void)
66 {
67         LU_CONTEXT_KEY_INIT(&fld_thread_key);
68         return lu_context_key_register(&fld_thread_key);
69 }
70
71 void fld_server_mod_exit(void)
72 {
73         lu_context_key_degister(&fld_thread_key);
74 }
75
76 int fld_declare_server_create(const struct lu_env *env,
77                               struct lu_server_fld *fld,
78                               const struct lu_seq_range *range,
79                               struct thandle *th)
80 {
81         int rc;
82
83         rc = fld_declare_index_create(env, fld, range, th);
84         RETURN(rc);
85 }
86 EXPORT_SYMBOL(fld_declare_server_create);
87
88 /**
89  * Insert FLD index entry and update FLD cache.
90  *
91  * This function is called from the sequence allocator when a super-sequence
92  * is granted to a server.
93  */
94 int fld_server_create(const struct lu_env *env, struct lu_server_fld *fld,
95                       const struct lu_seq_range *range, struct thandle *th)
96 {
97         int rc;
98
99         mutex_lock(&fld->lsf_lock);
100         rc = fld_index_create(env, fld, range, th);
101         mutex_unlock(&fld->lsf_lock);
102
103         RETURN(rc);
104 }
105 EXPORT_SYMBOL(fld_server_create);
106
107 /**
108  * Extract index information from fld name like srv-fsname-MDT0000
109  **/
110 int fld_name_to_index(const char *name, __u32 *index)
111 {
112         char *dash;
113         int rc;
114         ENTRY;
115
116         CDEBUG(D_INFO, "get index from %s\n", name);
117         dash = strrchr(name, '-');
118         if (dash == NULL)
119                 RETURN(-EINVAL);
120         dash++;
121         rc = target_name2index(dash, index, NULL);
122         RETURN(rc);
123 }
124
125 /**
126  * Retrieve fldb entry from MDT0 and add to local FLDB and cache.
127  **/
128 int fld_update_from_controller(const struct lu_env *env,
129                                struct lu_server_fld *fld)
130 {
131         struct fld_thread_info    *info;
132         struct lu_seq_range       *range;
133         struct lu_seq_range_array *lsra;
134         __u32                     index;
135         struct ptlrpc_request     *req;
136         int                       rc;
137         int                       i;
138         ENTRY;
139
140         /* Update only happens during initalization, i.e. local FLDB
141          * does not exist yet */
142         if (!fld->lsf_new)
143                 RETURN(0);
144
145         rc = fld_name_to_index(fld->lsf_name, &index);
146         if (rc < 0)
147                 RETURN(rc);
148
149         /* No need update fldb for MDT0 */
150         if (index == 0)
151                 RETURN(0);
152
153         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
154         LASSERT(info != NULL);
155         range = &info->fti_lrange;
156         memset(range, 0, sizeof(*range));
157         range->lsr_index = index;
158         fld_range_set_mdt(range);
159
160         do {
161                 rc = fld_client_rpc(fld->lsf_control_exp, range, FLD_READ,
162                                     &req);
163                 if (rc != 0 && rc != -EAGAIN)
164                         GOTO(out, rc);
165
166                 LASSERT(req != NULL);
167                 lsra = (struct lu_seq_range_array *)req_capsule_server_get(
168                                           &req->rq_pill, &RMF_GENERIC_DATA);
169                 if (lsra == NULL)
170                         GOTO(out, rc = -EPROTO);
171
172                 range_array_le_to_cpu(lsra, lsra);
173                 for (i = 0; i < lsra->lsra_count; i++) {
174                         int rc1;
175
176                         if (lsra->lsra_lsr[i].lsr_flags != LU_SEQ_RANGE_MDT)
177                                 GOTO(out, rc = -EINVAL);
178
179                         if (lsra->lsra_lsr[i].lsr_index != index)
180                                 GOTO(out, rc = -EINVAL);
181
182                         mutex_lock(&fld->lsf_lock);
183                         rc1 = fld_insert_entry(env, fld, &lsra->lsra_lsr[i]);
184                         mutex_unlock(&fld->lsf_lock);
185
186                         if (rc1 != 0)
187                                 GOTO(out, rc = rc1);
188                 }
189                 if (rc == -EAGAIN)
190                         *range = lsra->lsra_lsr[lsra->lsra_count - 1];
191         } while (rc == -EAGAIN);
192
193         fld->lsf_new = 1;
194 out:
195         if (req != NULL)
196                 ptlrpc_req_finished(req);
197
198         RETURN(rc);
199 }
200 EXPORT_SYMBOL(fld_update_from_controller);
201
202 /**
203  * Lookup sequece in local cache/fldb.
204  **/
205 int fld_local_lookup(const struct lu_env *env, struct lu_server_fld *fld,
206                      u64 seq, struct lu_seq_range *range)
207 {
208         struct lu_seq_range *erange;
209         struct fld_thread_info *info;
210         int rc;
211         ENTRY;
212
213         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
214         LASSERT(info != NULL);
215         erange = &info->fti_lrange;
216
217         /* Lookup it in the cache. */
218         rc = fld_cache_lookup(fld->lsf_cache, seq, erange);
219         if (rc == 0) {
220                 if (unlikely(fld_range_type(erange) != fld_range_type(range) &&
221                              !fld_range_is_any(range))) {
222                         CERROR("%s: FLD cache range "DRANGE" does not match"
223                                "requested flag %x: rc = %d\n", fld->lsf_name,
224                                PRANGE(erange), range->lsr_flags, -EIO);
225                         RETURN(-EIO);
226                 }
227                 *range = *erange;
228                 RETURN(0);
229         }
230         RETURN(rc);
231 }
232 EXPORT_SYMBOL(fld_local_lookup);
233
234 /**
235  *  Lookup MDT/OST by seq, returns a range for given seq.
236  *
237  *  If that entry is not cached in fld cache, request is sent to super
238  *  sequence controller node (MDT0). All other MDT[1...N] and client
239  *  cache fld entries, but this cache is not persistent.
240  */
241 int fld_server_lookup(const struct lu_env *env, struct lu_server_fld *fld,
242                       u64 seq, struct lu_seq_range *range)
243 {
244         __u32 index;
245         int rc;
246         ENTRY;
247
248         rc = fld_local_lookup(env, fld, seq, range);
249         if (likely(rc == 0))
250                 RETURN(rc);
251
252         rc = fld_name_to_index(fld->lsf_name, &index);
253         if (rc < 0)
254                 RETURN(rc);
255
256         if (index == 0 && rc == LDD_F_SV_TYPE_MDT) {
257                 /* On server side, all entries should be in cache.
258                  * If we can not find it in cache, just return error */
259                 CERROR("%s: Cannot find sequence "LPX64": rc = %d\n",
260                        fld->lsf_name, seq, -ENOENT);
261                 RETURN(-ENOENT);
262         } else {
263                 if (fld->lsf_control_exp == NULL) {
264                         CERROR("%s: lookup "LPX64", but not connects to MDT0"
265                                "yet: rc = %d.\n", fld->lsf_name, seq, -EIO);
266                         RETURN(-EIO);
267                 }
268                 /* send request to mdt0 i.e. super seq. controller.
269                  * This is temporary solution, long term solution is fld
270                  * replication on all mdt servers.
271                  */
272                 range->lsr_start = seq;
273                 rc = fld_client_rpc(fld->lsf_control_exp,
274                                     range, FLD_QUERY, NULL);
275                 if (rc == 0)
276                         fld_cache_insert(fld->lsf_cache, range);
277         }
278         RETURN(rc);
279 }
280 EXPORT_SYMBOL(fld_server_lookup);
281
282 /**
283  * All MDT server handle fld lookup operation. But only MDT0 has fld index.
284  * if entry is not found in cache we need to forward lookup request to MDT0
285  */
286 static int fld_handle_lookup(struct tgt_session_info *tsi)
287 {
288         struct obd_export       *exp = tsi->tsi_exp;
289         struct lu_site          *site = exp->exp_obd->obd_lu_dev->ld_site;
290         struct lu_server_fld    *fld;
291         struct lu_seq_range     *in;
292         struct lu_seq_range     *out;
293         int                     rc;
294
295         ENTRY;
296
297         in = req_capsule_client_get(tsi->tsi_pill, &RMF_FLD_MDFLD);
298         if (in == NULL)
299                 RETURN(err_serious(-EPROTO));
300
301         rc = req_capsule_server_pack(tsi->tsi_pill);
302         if (unlikely(rc != 0))
303                 RETURN(err_serious(rc));
304
305         out = req_capsule_server_get(tsi->tsi_pill, &RMF_FLD_MDFLD);
306         if (out == NULL)
307                 RETURN(err_serious(-EPROTO));
308         *out = *in;
309
310         fld = lu_site2seq(site)->ss_server_fld;
311
312         rc = fld_server_lookup(tsi->tsi_env, fld, in->lsr_start, out);
313
314         CDEBUG(D_INFO, "%s: FLD req handle: error %d (range: "DRANGE")\n",
315                fld->lsf_name, rc, PRANGE(out));
316
317         RETURN(rc);
318 }
319
320 static int fld_handle_read(struct tgt_session_info *tsi)
321 {
322         struct obd_export       *exp = tsi->tsi_exp;
323         struct lu_site          *site = exp->exp_obd->obd_lu_dev->ld_site;
324         struct lu_seq_range     *in;
325         void                    *data;
326         int                     rc;
327
328         ENTRY;
329
330         req_capsule_set(tsi->tsi_pill, &RQF_FLD_READ);
331
332         in = req_capsule_client_get(tsi->tsi_pill, &RMF_FLD_MDFLD);
333         if (in == NULL)
334                 RETURN(err_serious(-EPROTO));
335
336         req_capsule_set_size(tsi->tsi_pill, &RMF_GENERIC_DATA, RCL_SERVER,
337                              PAGE_CACHE_SIZE);
338
339         rc = req_capsule_server_pack(tsi->tsi_pill);
340         if (unlikely(rc != 0))
341                 RETURN(err_serious(rc));
342
343         data = req_capsule_server_get(tsi->tsi_pill, &RMF_GENERIC_DATA);
344
345         rc = fld_server_read(tsi->tsi_env, lu_site2seq(site)->ss_server_fld,
346                              in, data, PAGE_CACHE_SIZE);
347         RETURN(rc);
348 }
349
350 static int fld_handle_query(struct tgt_session_info *tsi)
351 {
352         int     rc;
353
354         ENTRY;
355
356         req_capsule_set(tsi->tsi_pill, &RQF_FLD_QUERY);
357
358         rc = fld_handle_lookup(tsi);
359
360         RETURN(rc);
361 }
362
363 /*
364  * Returns true, if fid is local to this server node.
365  *
366  * WARNING: this function is *not* guaranteed to return false if fid is
367  * remote: it makes an educated conservative guess only.
368  *
369  * fid_is_local() is supposed to be used in assertion checks only.
370  */
371 int fid_is_local(const struct lu_env *env,
372                  struct lu_site *site, const struct lu_fid *fid)
373 {
374         int result;
375         struct seq_server_site *ss_site;
376         struct lu_seq_range *range;
377         struct fld_thread_info *info;
378         ENTRY;
379
380         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
381         range = &info->fti_lrange;
382
383         result = 1; /* conservatively assume fid is local */
384         ss_site = lu_site2seq(site);
385         if (ss_site->ss_client_fld != NULL) {
386                 int rc;
387
388                 rc = fld_cache_lookup(ss_site->ss_client_fld->lcf_cache,
389                                       fid_seq(fid), range);
390                 if (rc == 0)
391                         result = (range->lsr_index == ss_site->ss_node_id);
392         }
393         return result;
394 }
395
396 static void fld_server_proc_fini(struct lu_server_fld *fld);
397
398 #ifdef CONFIG_PROC_FS
399 static int fld_server_proc_init(struct lu_server_fld *fld)
400 {
401         int rc = 0;
402         ENTRY;
403
404         fld->lsf_proc_dir = lprocfs_register(fld->lsf_name, fld_type_proc_dir,
405                                              fld_server_proc_list, fld);
406         if (IS_ERR(fld->lsf_proc_dir)) {
407                 rc = PTR_ERR(fld->lsf_proc_dir);
408                 RETURN(rc);
409         }
410
411         rc = lprocfs_seq_create(fld->lsf_proc_dir, "fldb", 0444,
412                                 &fld_proc_seq_fops, fld);
413         if (rc) {
414                 lprocfs_remove(&fld->lsf_proc_dir);
415                 fld->lsf_proc_dir = NULL;
416         }
417
418         RETURN(rc);
419 }
420
421 static void fld_server_proc_fini(struct lu_server_fld *fld)
422 {
423         ENTRY;
424         if (fld->lsf_proc_dir != NULL) {
425                 if (!IS_ERR(fld->lsf_proc_dir))
426                         lprocfs_remove(&fld->lsf_proc_dir);
427                 fld->lsf_proc_dir = NULL;
428         }
429         EXIT;
430 }
431 #else
432 static int fld_server_proc_init(struct lu_server_fld *fld)
433 {
434         return 0;
435 }
436
437 static void fld_server_proc_fini(struct lu_server_fld *fld)
438 {
439         return;
440 }
441 #endif
442
443 int fld_server_init(const struct lu_env *env, struct lu_server_fld *fld,
444                     struct dt_device *dt, const char *prefix, int type)
445 {
446         int cache_size, cache_threshold;
447         int rc;
448
449         ENTRY;
450
451         snprintf(fld->lsf_name, sizeof(fld->lsf_name), "srv-%s", prefix);
452
453         cache_size = FLD_SERVER_CACHE_SIZE / sizeof(struct fld_cache_entry);
454
455         cache_threshold = cache_size * FLD_SERVER_CACHE_THRESHOLD / 100;
456
457         mutex_init(&fld->lsf_lock);
458         fld->lsf_cache = fld_cache_init(fld->lsf_name, cache_size,
459                                         cache_threshold);
460         if (IS_ERR(fld->lsf_cache)) {
461                 rc = PTR_ERR(fld->lsf_cache);
462                 fld->lsf_cache = NULL;
463                 RETURN(rc);
464         }
465
466         rc = fld_index_init(env, fld, dt, type);
467         if (rc)
468                 GOTO(out_cache, rc);
469
470         rc = fld_server_proc_init(fld);
471         if (rc)
472                 GOTO(out_index, rc);
473
474         fld->lsf_control_exp = NULL;
475         fld->lsf_seq_lookup = fld_server_lookup;
476
477         fld->lsf_seq_lookup = fld_server_lookup;
478         RETURN(0);
479 out_index:
480         fld_index_fini(env, fld);
481 out_cache:
482         fld_cache_fini(fld->lsf_cache);
483         return rc;
484 }
485 EXPORT_SYMBOL(fld_server_init);
486
487 void fld_server_fini(const struct lu_env *env, struct lu_server_fld *fld)
488 {
489         ENTRY;
490
491         fld_server_proc_fini(fld);
492         fld_index_fini(env, fld);
493
494         if (fld->lsf_cache != NULL) {
495                 if (!IS_ERR(fld->lsf_cache))
496                         fld_cache_fini(fld->lsf_cache);
497                 fld->lsf_cache = NULL;
498         }
499
500         EXIT;
501 }
502 EXPORT_SYMBOL(fld_server_fini);
503
504 struct tgt_handler fld_handlers[] = {
505 TGT_FLD_HDL_VAR(0,      FLD_QUERY,      fld_handle_query),
506 TGT_FLD_HDL_VAR(0,      FLD_READ,       fld_handle_read),
507 };
508 EXPORT_SYMBOL(fld_handlers);