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 static void *fld_key_init(const struct lu_context *ctx,
  56                           struct lu_context_key *key)
  57 {
  58         struct fld_thread_info *info;
  59         ENTRY;
  60
  61         OBD_ALLOC_PTR(info);
  62         if (info == NULL)
  63                 info = ERR_PTR(-ENOMEM);
  64         RETURN(info);
  65 }
  66
  67 static void fld_key_fini(const struct lu_context *ctx,
  68                          struct lu_context_key *key, void *data)
  69 {
  70         struct fld_thread_info *info = data;
  71         ENTRY;
  72         OBD_FREE_PTR(info);
  73         EXIT;
  74 }
  75
  76 struct lu_context_key fld_thread_key = {
  77         .lct_tags = LCT_MD_THREAD|LCT_DT_THREAD,
  78         .lct_init = fld_key_init,
  79         .lct_fini = fld_key_fini
  80 };
  81
  82 cfs_proc_dir_entry_t *fld_type_proc_dir = NULL;
  83
  84 static int __init fld_mod_init(void)
  85 {
  86         fld_type_proc_dir = lprocfs_register(LUSTRE_FLD_NAME,
  87                                              proc_lustre_root,
  88                                              NULL, NULL);
  89         if (IS_ERR(fld_type_proc_dir))
  90                 return PTR_ERR(fld_type_proc_dir);
  91
  92         LU_CONTEXT_KEY_INIT(&fld_thread_key);
  93         lu_context_key_register(&fld_thread_key);
  94         return 0;
  95 }
  96
  97 static void __exit fld_mod_exit(void)
  98 {
  99         lu_context_key_degister(&fld_thread_key);
 100         if (fld_type_proc_dir != NULL && !IS_ERR(fld_type_proc_dir)) {
 101                 lprocfs_remove(&fld_type_proc_dir);
 102                 fld_type_proc_dir = NULL;
 103         }
 104 }
 105
 106 /* Insert index entry and update cache. */
 107 int fld_server_create(struct lu_server_fld *fld,
 108                       const struct lu_env *env,
 109                       seqno_t seq, mdsno_t mds)
 110 {
 111         int rc;
 112         ENTRY;
 113
 114         rc = fld_index_create(fld, env, seq, mds);
 115
 116         if (rc == 0) {
 117                 /*
 118                  * Do not return result of calling fld_cache_insert()
 119                  * here. First of all because it may return -EEXISTS. Another
 120                  * reason is that, we do not want to stop proceeding even after
 121                  * cache errors.
 122                  */
 123                 fld_cache_insert(fld->lsf_cache, seq, mds);
 124         }
 125
 126         RETURN(rc);
 127 }
 128 EXPORT_SYMBOL(fld_server_create);
 129
 130 /* Delete index entry. */
 131 int fld_server_delete(struct lu_server_fld *fld,
 132                       const struct lu_env *env,
 133                       seqno_t seq)
 134 {
 135         int rc;
 136         ENTRY;
 137
 138         fld_cache_delete(fld->lsf_cache, seq);
 139         rc = fld_index_delete(fld, env, seq);
 140
 141         RETURN(rc);
 142 }
 143 EXPORT_SYMBOL(fld_server_delete);
 144
 145 /* Lookup mds by seq. */
 146 int fld_server_lookup(struct lu_server_fld *fld,
 147                       const struct lu_env *env,
 148                       seqno_t seq, mdsno_t *mds)
 149 {
 150         int rc;
 151         ENTRY;
 152
 153         /* Lookup it in the cache. */
 154         rc = fld_cache_lookup(fld->lsf_cache, seq, mds);
 155         if (rc == 0)
 156                 RETURN(0);
 157
 158         rc = fld_index_lookup(fld, env, seq, mds);
 159         if (rc == 0) {
 160                 /*
 161                  * Do not return error here as well. See previous comment in
 162                  * same situation in function fld_server_create().
 163                  */
 164                 fld_cache_insert(fld->lsf_cache, seq, *mds);
 165         }
 166         RETURN(rc);
 167 }
 168 EXPORT_SYMBOL(fld_server_lookup);
 169
 170 static int fld_server_handle(struct lu_server_fld *fld,
 171                              const struct lu_env *env,
 172                              __u32 opc, struct md_fld *mf,
 173                              struct fld_thread_info *info)
 174 {
 175         int rc;
 176         ENTRY;
 177
 178         switch (opc) {
 179         case FLD_CREATE:
 180                 rc = fld_server_create(fld, env,
 181                                        mf->mf_seq, mf->mf_mds);
 182
 183                 /* Do not return -EEXIST error for resent case */
 184                 if ((info->fti_flags & MSG_RESENT) && rc == -EEXIST)
 185                         rc = 0;
 186                 break;
 187         case FLD_DELETE:
 188                 rc = fld_server_delete(fld, env, mf->mf_seq);
 189
 190                 /* Do not return -ENOENT error for resent case */
 191                 if ((info->fti_flags & MSG_RESENT) && rc == -ENOENT)
 192                         rc = 0;
 193                 break;
 194         case FLD_LOOKUP:
 195                 rc = fld_server_lookup(fld, env,
 196                                        mf->mf_seq, &mf->mf_mds);
 197                 break;
 198         default:
 199                 rc = -EINVAL;
 200                 break;
 201         }
 202
 203         CDEBUG(D_INFO, "%s: FLD req handle: error %d (opc: %d, seq: "
 204                LPX64", mds: "LPU64")\n", fld->lsf_name, rc, opc,
 205                mf->mf_seq, mf->mf_mds);
 206
 207         RETURN(rc);
 208
 209 }
 210
 211 static int fld_req_handle(struct ptlrpc_request *req,
 212                           struct fld_thread_info *info)
 213 {
 214         struct lu_site *site;
 215         struct md_fld *in;
 216         struct md_fld *out;
 217         int rc;
 218         __u32 *opc;
 219         ENTRY;
 220
 221         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
 222
 223         rc = req_capsule_pack(&info->fti_pill);
 224         if (rc)
 225                 RETURN(err_serious(rc));
 226
 227         opc = req_capsule_client_get(&info->fti_pill, &RMF_FLD_OPC);
 228         if (opc != NULL) {
 229                 in = req_capsule_client_get(&info->fti_pill, &RMF_FLD_MDFLD);
 230                 if (in == NULL)
 231                         RETURN(err_serious(-EPROTO));
 232                 out = req_capsule_server_get(&info->fti_pill, &RMF_FLD_MDFLD);
 233                 if (out == NULL)
 234                         RETURN(err_serious(-EPROTO));
 235                 *out = *in;
 236
 237                 rc = fld_server_handle(site->ls_server_fld,
 238                                        req->rq_svc_thread->t_env,
 239                                        *opc, out, info);
 240         } else
 241                 rc = err_serious(-EPROTO);
 242
 243         RETURN(rc);
 244 }
 245
 246 static void fld_thread_info_init(struct ptlrpc_request *req,
 247                                  struct fld_thread_info *info)
 248 {
 249         int i;
 250
 251         info->fti_flags = lustre_msg_get_flags(req->rq_reqmsg);
 252
 253         /* Mark rep buffer as req-layout stuff expects. */
 254         for (i = 0; i < ARRAY_SIZE(info->fti_rep_buf_size); i++)
 255                 info->fti_rep_buf_size[i] = -1;
 256
 257         /* Init request capsule. */
 258         req_capsule_init(&info->fti_pill, req, RCL_SERVER,
 259                          info->fti_rep_buf_size);
 260
 261         req_capsule_set(&info->fti_pill, &RQF_FLD_QUERY);
 262 }
 263
 264 static void fld_thread_info_fini(struct fld_thread_info *info)
 265 {
 266         req_capsule_fini(&info->fti_pill);
 267 }
 268
 269 static int fld_handle(struct ptlrpc_request *req)
 270 {
 271         struct fld_thread_info *info;
 272         const struct lu_env *env;
 273         int rc;
 274
 275         env = req->rq_svc_thread->t_env;
 276         LASSERT(env != NULL);
 277
 278         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 279         LASSERT(info != NULL);
 280
 281         fld_thread_info_init(req, info);
 282         rc = fld_req_handle(req, info);
 283         fld_thread_info_fini(info);
 284
 285         return rc;
 286 }
 287
 288 /*
 289  * Entry point for handling FLD RPCs called from MDT.
 290  */
 291 int fld_query(struct com_thread_info *info)
 292 {
 293         return fld_handle(info->cti_pill.rc_req);
 294 }
 295 EXPORT_SYMBOL(fld_query);
 296
 297 /*
 298  * Returns true, if fid is local to this server node.
 299  *
 300  * WARNING: this function is *not* guaranteed to return false if fid is
 301  * remote: it makes an educated conservative guess only.
 302  *
 303  * fid_is_local() is supposed to be used in assertion checks only.
 304  */
 305 int fid_is_local(struct lu_site *site, const struct lu_fid *fid)
 306 {
 307         int result;
 308
 309         result = 1; /* conservatively assume fid is local */
 310         if (site->ls_client_fld != NULL) {
 311                 mdsno_t mds;
 312                 int rc;
 313
 314                 rc = fld_cache_lookup(site->ls_client_fld->lcf_cache,
 315                                       fid_seq(fid), &mds);
 316                 if (rc == 0)
 317                         result = (mds == site->ls_node_id);
 318         }
 319         return result;
 320 }
 321 EXPORT_SYMBOL(fid_is_local);
 322
 323 static void fld_server_proc_fini(struct lu_server_fld *fld);
 324
 325 #ifdef LPROCFS
 326 static int fld_server_proc_init(struct lu_server_fld *fld)
 327 {
 328         int rc = 0;
 329         ENTRY;
 330
 331         fld->lsf_proc_dir = lprocfs_register(fld->lsf_name,
 332                                              fld_type_proc_dir,
 333                                              fld_server_proc_list, fld);
 334         if (IS_ERR(fld->lsf_proc_dir)) {
 335                 rc = PTR_ERR(fld->lsf_proc_dir);
 336                 RETURN(rc);
 337         }
 338
 339         RETURN(rc);
 340 }
 341
 342 static void fld_server_proc_fini(struct lu_server_fld *fld)
 343 {
 344         ENTRY;
 345         if (fld->lsf_proc_dir != NULL) {
 346                 if (!IS_ERR(fld->lsf_proc_dir))
 347                         lprocfs_remove(&fld->lsf_proc_dir);
 348                 fld->lsf_proc_dir = NULL;
 349         }
 350         EXIT;
 351 }
 352 #else
 353 static int fld_server_proc_init(struct lu_server_fld *fld)
 354 {
 355         return 0;
 356 }
 357
 358 static void fld_server_proc_fini(struct lu_server_fld *fld)
 359 {
 360         return;
 361 }
 362 #endif
 363
 364 int fld_server_init(struct lu_server_fld *fld, struct dt_device *dt,
 365                     const char *prefix, const struct lu_env *env)
 366 {
 367         int cache_size, cache_threshold;
 368         int rc;
 369         ENTRY;
 370
 371         snprintf(fld->lsf_name, sizeof(fld->lsf_name),
 372                  "srv-%s", prefix);
 373
 374         cache_size = FLD_SERVER_CACHE_SIZE /
 375                 sizeof(struct fld_cache_entry);
 376
 377         cache_threshold = cache_size *
 378                 FLD_SERVER_CACHE_THRESHOLD / 100;
 379
 380         fld->lsf_cache = fld_cache_init(fld->lsf_name,
 381                                         FLD_SERVER_HTABLE_SIZE,
 382                                         cache_size, cache_threshold);
 383         if (IS_ERR(fld->lsf_cache)) {
 384                 rc = PTR_ERR(fld->lsf_cache);
 385                 fld->lsf_cache = NULL;
 386                 GOTO(out, rc);
 387         }
 388
 389         rc = fld_index_init(fld, env, dt);
 390         if (rc)
 391                 GOTO(out, rc);
 392
 393         rc = fld_server_proc_init(fld);
 394         if (rc)
 395                 GOTO(out, rc);
 396
 397         EXIT;
 398 out:
 399         if (rc)
 400                 fld_server_fini(fld, env);
 401         return rc;
 402 }
 403 EXPORT_SYMBOL(fld_server_init);
 404
 405 void fld_server_fini(struct lu_server_fld *fld,
 406                      const struct lu_env *env)
 407 {
 408         ENTRY;
 409
 410         fld_server_proc_fini(fld);
 411         fld_index_fini(fld, env);
 412
 413         if (fld->lsf_cache != NULL) {
 414                 if (!IS_ERR(fld->lsf_cache))
 415                         fld_cache_fini(fld->lsf_cache);
 416                 fld->lsf_cache = NULL;
 417         }
 418
 419         EXIT;
 420 }
 421 EXPORT_SYMBOL(fld_server_fini);
 422
 423 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 424 MODULE_DESCRIPTION("Lustre FLD");
 425 MODULE_LICENSE("GPL");
 426
 427 cfs_module(mdd, "0.1.0", fld_mod_init, fld_mod_exit);
 428 #endif