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_server_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         info->fti_flags = lustre_msg_get_flags(req->rq_reqmsg);
 230
 231         info->fti_pill = &req->rq_pill;
 232         /* Init request capsule. */
 233         req_capsule_init(info->fti_pill, req, RCL_SERVER);
 234         req_capsule_set(info->fti_pill, &RQF_FLD_QUERY);
 235 }
 236
 237 static void fld_thread_info_fini(struct fld_thread_info *info)
 238 {
 239         req_capsule_fini(info->fti_pill);
 240 }
 241
 242 static int fld_handle(struct ptlrpc_request *req)
 243 {
 244         struct fld_thread_info *info;
 245         const struct lu_env *env;
 246         int rc;
 247
 248         env = req->rq_svc_thread->t_env;
 249         LASSERT(env != NULL);
 250
 251         info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
 252         LASSERT(info != NULL);
 253
 254         fld_thread_info_init(req, info);
 255         rc = fld_req_handle(req, info);
 256         fld_thread_info_fini(info);
 257
 258         return rc;
 259 }
 260
 261 /*
 262  * Entry point for handling FLD RPCs called from MDT.
 263  */
 264 int fld_query(struct com_thread_info *info)
 265 {
 266         return fld_handle(info->cti_pill->rc_req);
 267 }
 268 EXPORT_SYMBOL(fld_query);
 269
 270 /*
 271  * Returns true, if fid is local to this server node.
 272  *
 273  * WARNING: this function is *not* guaranteed to return false if fid is
 274  * remote: it makes an educated conservative guess only.
 275  *
 276  * fid_is_local() is supposed to be used in assertion checks only.
 277  */
 278 int fid_is_local(struct lu_site *site, const struct lu_fid *fid)
 279 {
 280         int result;
 281
 282         result = 1; /* conservatively assume fid is local */
 283         if (site->ls_client_fld != NULL) {
 284                 mdsno_t mds;
 285                 int rc;
 286
 287                 rc = fld_cache_lookup(site->ls_client_fld->lcf_cache,
 288                                       fid_seq(fid), &mds);
 289                 if (rc == 0)
 290                         result = (mds == site->ls_node_id);
 291         }
 292         return result;
 293 }
 294 EXPORT_SYMBOL(fid_is_local);
 295
 296 static void fld_server_proc_fini(struct lu_server_fld *fld);
 297
 298 #ifdef LPROCFS
 299 static int fld_server_proc_init(struct lu_server_fld *fld)
 300 {
 301         int rc = 0;
 302         ENTRY;
 303
 304         fld->lsf_proc_dir = lprocfs_register(fld->lsf_name,
 305                                              fld_type_proc_dir,
 306                                              fld_server_proc_list, fld);
 307         if (IS_ERR(fld->lsf_proc_dir)) {
 308                 rc = PTR_ERR(fld->lsf_proc_dir);
 309                 RETURN(rc);
 310         }
 311
 312         RETURN(rc);
 313 }
 314
 315 static void fld_server_proc_fini(struct lu_server_fld *fld)
 316 {
 317         ENTRY;
 318         if (fld->lsf_proc_dir != NULL) {
 319                 if (!IS_ERR(fld->lsf_proc_dir))
 320                         lprocfs_remove(&fld->lsf_proc_dir);
 321                 fld->lsf_proc_dir = NULL;
 322         }
 323         EXIT;
 324 }
 325 #else
 326 static int fld_server_proc_init(struct lu_server_fld *fld)
 327 {
 328         return 0;
 329 }
 330
 331 static void fld_server_proc_fini(struct lu_server_fld *fld)
 332 {
 333         return;
 334 }
 335 #endif
 336
 337 int fld_server_init(struct lu_server_fld *fld, struct dt_device *dt,
 338                     const char *prefix, const struct lu_env *env)
 339 {
 340         int cache_size, cache_threshold;
 341         int rc;
 342         ENTRY;
 343
 344         snprintf(fld->lsf_name, sizeof(fld->lsf_name),
 345                  "srv-%s", prefix);
 346
 347         cache_size = FLD_SERVER_CACHE_SIZE /
 348                 sizeof(struct fld_cache_entry);
 349
 350         cache_threshold = cache_size *
 351                 FLD_SERVER_CACHE_THRESHOLD / 100;
 352
 353         fld->lsf_cache = fld_cache_init(fld->lsf_name,
 354                                         FLD_SERVER_HTABLE_SIZE,
 355                                         cache_size, cache_threshold);
 356         if (IS_ERR(fld->lsf_cache)) {
 357                 rc = PTR_ERR(fld->lsf_cache);
 358                 fld->lsf_cache = NULL;
 359                 GOTO(out, rc);
 360         }
 361
 362         rc = fld_index_init(fld, env, dt);
 363         if (rc)
 364                 GOTO(out, rc);
 365
 366         rc = fld_server_proc_init(fld);
 367         if (rc)
 368                 GOTO(out, rc);
 369
 370         EXIT;
 371 out:
 372         if (rc)
 373                 fld_server_fini(fld, env);
 374         return rc;
 375 }
 376 EXPORT_SYMBOL(fld_server_init);
 377
 378 void fld_server_fini(struct lu_server_fld *fld,
 379                      const struct lu_env *env)
 380 {
 381         ENTRY;
 382
 383         fld_server_proc_fini(fld);
 384         fld_index_fini(fld, env);
 385
 386         if (fld->lsf_cache != NULL) {
 387                 if (!IS_ERR(fld->lsf_cache))
 388                         fld_cache_fini(fld->lsf_cache);
 389                 fld->lsf_cache = NULL;
 390         }
 391
 392         EXIT;
 393 }
 394 EXPORT_SYMBOL(fld_server_fini);
 395
 396 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 397 MODULE_DESCRIPTION("Lustre FLD");
 398 MODULE_LICENSE("GPL");
 399
 400 cfs_module(mdd, "0.1.0", fld_mod_init, fld_mod_exit);
 401 #endif