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         printk(KERN_INFO "Lustre: Fid Location Database; "
  87                "info@clusterfs.com\n");
  88
  89         fld_type_proc_dir = lprocfs_register(LUSTRE_FLD_NAME,
  90                                              proc_lustre_root,
  91                                              NULL, NULL);
  92         if (IS_ERR(fld_type_proc_dir))
  93                 return PTR_ERR(fld_type_proc_dir);
  94
  95         lu_context_key_register(&fld_thread_key);
  96         return 0;
  97 }
  98
  99 static void __exit fld_mod_exit(void)
 100 {
 101         lu_context_key_degister(&fld_thread_key);
 102         if (fld_type_proc_dir != NULL && !IS_ERR(fld_type_proc_dir)) {
 103                 lprocfs_remove(fld_type_proc_dir);
 104                 fld_type_proc_dir = NULL;
 105         }
 106 }
 107
 108 /* insert index entry and update cache */
 109 int fld_server_create(struct lu_server_fld *fld,
 110                       const struct lu_context *ctx,
 111                       seqno_t seq, mdsno_t mds)
 112 {
 113         return fld_index_create(fld, ctx, seq, mds);
 114 }
 115 EXPORT_SYMBOL(fld_server_create);
 116
 117 /* delete index entry */
 118 int fld_server_delete(struct lu_server_fld *fld,
 119                       const struct lu_context *ctx,
 120                       seqno_t seq)
 121 {
 122         return fld_index_delete(fld, ctx, seq);
 123 }
 124 EXPORT_SYMBOL(fld_server_delete);
 125
 126 /* issue on-disk index lookup */
 127 int fld_server_lookup(struct lu_server_fld *fld,
 128                       const struct lu_context *ctx,
 129                       seqno_t seq, mdsno_t *mds)
 130 {
 131         return fld_index_lookup(fld, ctx, seq, mds);
 132 }
 133 EXPORT_SYMBOL(fld_server_lookup);
 134
 135 static int fld_server_handle(struct lu_server_fld *fld,
 136                              const struct lu_context *ctx,
 137                              __u32 opc, struct md_fld *mf,
 138                              struct fld_thread_info *info)
 139 {
 140         int rc;
 141         ENTRY;
 142
 143         switch (opc) {
 144         case FLD_CREATE:
 145                 rc = fld_server_create(fld, ctx,
 146                                        mf->mf_seq, mf->mf_mds);
 147
 148                 /* do not return -EEXIST error for resent case */
 149                 if ((info->fti_flags & MSG_RESENT) && rc == -EEXIST)
 150                         rc = 0;
 151                 break;
 152         case FLD_DELETE:
 153                 rc = fld_server_delete(fld, ctx, mf->mf_seq);
 154
 155                 /* do not return -ENOENT error for resent case */
 156                 if ((info->fti_flags & MSG_RESENT) && rc == -ENOENT)
 157                         rc = 0;
 158                 break;
 159         case FLD_LOOKUP:
 160                 rc = fld_server_lookup(fld, ctx,
 161                                        mf->mf_seq, &mf->mf_mds);
 162                 break;
 163         default:
 164                 rc = -EINVAL;
 165                 break;
 166         }
 167         RETURN(rc);
 168
 169 }
 170
 171 static int fld_req_handle(struct ptlrpc_request *req,
 172                           struct fld_thread_info *info)
 173 {
 174         struct lu_site *site;
 175         struct md_fld *in;
 176         struct md_fld *out;
 177         int rc = -EPROTO;
 178         __u32 *opc;
 179         ENTRY;
 180
 181         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
 182
 183         rc = req_capsule_pack(&info->fti_pill);
 184         if (rc)
 185                 RETURN(rc);
 186
 187         opc = req_capsule_client_get(&info->fti_pill, &RMF_FLD_OPC);
 188         if (opc != NULL) {
 189                 in = req_capsule_client_get(&info->fti_pill, &RMF_FLD_MDFLD);
 190                 if (in == NULL)
 191                         RETURN(-EPROTO);
 192                 out = req_capsule_server_get(&info->fti_pill, &RMF_FLD_MDFLD);
 193                 if (out == NULL)
 194                         RETURN(-EPROTO);
 195                 *out = *in;
 196
 197                 rc = fld_server_handle(site->ls_server_fld,
 198                                        req->rq_svc_thread->t_ctx,
 199                                        *opc, out, info);
 200         } else
 201                 rc = -EPROTO;
 202
 203         RETURN(rc);
 204 }
 205
 206 static void fld_thread_info_init(struct ptlrpc_request *req,
 207                                  struct fld_thread_info *info)
 208 {
 209         int i;
 210
 211         info->fti_flags = lustre_msg_get_flags(req->rq_reqmsg);
 212
 213         /* mark rep buffer as req-layout stuff expects */
 214         for (i = 0; i < ARRAY_SIZE(info->fti_rep_buf_size); i++)
 215                 info->fti_rep_buf_size[i] = -1;
 216
 217         /* init request capsule */
 218         req_capsule_init(&info->fti_pill, req, RCL_SERVER,
 219                          info->fti_rep_buf_size);
 220
 221         req_capsule_set(&info->fti_pill, &RQF_FLD_QUERY);
 222 }
 223
 224 static void fld_thread_info_fini(struct fld_thread_info *info)
 225 {
 226         req_capsule_fini(&info->fti_pill);
 227 }
 228
 229 static int fld_handle(struct ptlrpc_request *req)
 230 {
 231         const struct lu_context *ctx;
 232         struct fld_thread_info *info;
 233         int rc;
 234
 235         ctx = req->rq_svc_thread->t_ctx;
 236         LASSERT(ctx != NULL);
 237         LASSERT(ctx->lc_thread == req->rq_svc_thread);
 238
 239         info = lu_context_key_get(ctx, &fld_thread_key);
 240         LASSERT(info != NULL);
 241
 242         fld_thread_info_init(req, info);
 243         rc = fld_req_handle(req, info);
 244         fld_thread_info_fini(info);
 245
 246         return rc;
 247 }
 248
 249 /*
 250  * Entry point for handling FLD RPCs called from MDT.
 251  */
 252 int fld_query(struct com_thread_info *info)
 253 {
 254         return fld_handle(info->cti_pill.rc_req);
 255 }
 256 EXPORT_SYMBOL(fld_query);
 257
 258 /*
 259  * Returns true, if fid is local to this server node.
 260  *
 261  * WARNING: this function is *not* guaranteed to return false if fid is
 262  * remote: it makes an educated conservative guess only.
 263  *
 264  * fid_is_local() is supposed to be used in assertion checks only.
 265  */
 266 int fid_is_local(struct lu_site *site, const struct lu_fid *fid)
 267 {
 268         int result;
 269
 270         result = 1; /* conservatively assume fid is local */
 271         if (site->ls_client_fld != NULL) {
 272                 mdsno_t mds;
 273                 int rc;
 274
 275                 rc = fld_cache_lookup(site->ls_client_fld->lcf_cache,
 276                                       fid_seq(fid), &mds);
 277                 if (rc == 0)
 278                         result = (mds == site->ls_node_id);
 279         }
 280         return result;
 281 }
 282 EXPORT_SYMBOL(fid_is_local);
 283
 284 static void fld_server_proc_fini(struct lu_server_fld *fld);
 285
 286 #ifdef LPROCFS
 287 static int fld_server_proc_init(struct lu_server_fld *fld)
 288 {
 289         int rc = 0;
 290         ENTRY;
 291
 292         fld->lsf_proc_dir = lprocfs_register(fld->lsf_name,
 293                                              fld_type_proc_dir,
 294                                              fld_server_proc_list, fld);
 295         if (IS_ERR(fld->lsf_proc_dir)) {
 296                 rc = PTR_ERR(fld->lsf_proc_dir);
 297                 RETURN(rc);
 298         }
 299
 300         RETURN(rc);
 301 }
 302
 303 static void fld_server_proc_fini(struct lu_server_fld *fld)
 304 {
 305         ENTRY;
 306         if (fld->lsf_proc_dir != NULL) {
 307                 if (!IS_ERR(fld->lsf_proc_dir))
 308                         lprocfs_remove(fld->lsf_proc_dir);
 309                 fld->lsf_proc_dir = NULL;
 310         }
 311         EXIT;
 312 }
 313 #else
 314 static int fld_server_proc_init(struct lu_server_fld *fld)
 315 {
 316         return 0;
 317 }
 318
 319 static void fld_server_proc_fini(struct lu_server_fld *fld)
 320 {
 321         return;
 322 }
 323 #endif
 324
 325 int fld_server_init(struct lu_server_fld *fld, struct dt_device *dt,
 326                     const char *prefix, const struct lu_context *ctx)
 327 {
 328         int rc;
 329         ENTRY;
 330
 331         snprintf(fld->lsf_name, sizeof(fld->lsf_name),
 332                  "srv-%s", prefix);
 333
 334         rc = fld_index_init(fld, ctx, dt);
 335         if (rc)
 336                 GOTO(out, rc);
 337
 338         rc = fld_server_proc_init(fld);
 339         if (rc)
 340                 GOTO(out, rc);
 341
 342         EXIT;
 343 out:
 344         if (rc)
 345                 fld_server_fini(fld, ctx);
 346         return rc;
 347 }
 348 EXPORT_SYMBOL(fld_server_init);
 349
 350 void fld_server_fini(struct lu_server_fld *fld,
 351                      const struct lu_context *ctx)
 352 {
 353         ENTRY;
 354
 355         fld_server_proc_fini(fld);
 356         fld_index_fini(fld, ctx);
 357
 358         EXIT;
 359 }
 360 EXPORT_SYMBOL(fld_server_fini);
 361
 362 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 363 MODULE_DESCRIPTION("Lustre FLD");
 364 MODULE_LICENSE("GPL");
 365
 366 cfs_module(mdd, "0.1.0", fld_mod_init, fld_mod_exit);
 367 #endif