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: WangDi <wangdi@clusterfs.com>
   9  *           Yury Umanets <umka@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 <dt_object.h>
  51 #include <md_object.h>
  52 #include <lustre_req_layout.h>
  53 #include <lustre_fld.h>
  54 #include "fld_internal.h"
  55
  56 #ifdef __KERNEL__
  57 static int __init fld_mod_init(void)
  58 {
  59         /* nothing to init seems */
  60         return 0;
  61 }
  62
  63 static void __exit fld_mod_exit(void)
  64 {
  65         /* nothing to fini seems */
  66         return;
  67 }
  68
  69 /* insert index entry and update cache */
  70 int fld_server_create(struct lu_server_fld *fld,
  71                       const struct lu_context *ctx,
  72                       seqno_t seq, mdsno_t mds)
  73 {
  74         ENTRY;
  75         RETURN(fld_index_create(fld, ctx, seq, mds));
  76 }
  77 EXPORT_SYMBOL(fld_server_create);
  78
  79 /* delete index entry */
  80 int fld_server_delete(struct lu_server_fld *fld,
  81                       const struct lu_context *ctx,
  82                       seqno_t seq)
  83 {
  84         ENTRY;
  85         RETURN(fld_index_delete(fld, ctx, seq));
  86 }
  87 EXPORT_SYMBOL(fld_server_delete);
  88
  89 /* issue on-disk index lookup */
  90 int fld_server_lookup(struct lu_server_fld *fld,
  91                       const struct lu_context *ctx,
  92                       seqno_t seq, mdsno_t *mds)
  93 {
  94         ENTRY;
  95         RETURN(fld_index_lookup(fld, ctx, seq, mds));
  96 }
  97 EXPORT_SYMBOL(fld_server_lookup);
  98
  99 static int fld_server_handle(struct lu_server_fld *fld,
 100                              const struct lu_context *ctx,
 101                              __u32 opc, struct md_fld *mf)
 102 {
 103         int rc;
 104         ENTRY;
 105
 106         switch (opc) {
 107         case FLD_CREATE:
 108                 rc = fld_server_create(fld, ctx,
 109                                        mf->mf_seq, mf->mf_mds);
 110                 break;
 111         case FLD_DELETE:
 112                 rc = fld_server_delete(fld, ctx, mf->mf_seq);
 113                 break;
 114         case FLD_LOOKUP:
 115                 rc = fld_server_lookup(fld, ctx,
 116                                        mf->mf_seq, &mf->mf_mds);
 117                 break;
 118         default:
 119                 rc = -EINVAL;
 120                 break;
 121         }
 122         RETURN(rc);
 123
 124 }
 125
 126 static int fld_req_handle0(const struct lu_context *ctx,
 127                            struct lu_server_fld *fld,
 128                            struct ptlrpc_request *req)
 129 {
 130         int rep_buf_size[3] = { 0, };
 131         struct req_capsule pill;
 132         struct md_fld *in;
 133         struct md_fld *out;
 134         int rc = -EPROTO;
 135         __u32 *opc;
 136         ENTRY;
 137
 138         req_capsule_init(&pill, req, RCL_SERVER,
 139                          rep_buf_size);
 140
 141         req_capsule_set(&pill, &RQF_FLD_QUERY);
 142         req_capsule_pack(&pill);
 143
 144         opc = req_capsule_client_get(&pill, &RMF_FLD_OPC);
 145         if (opc != NULL) {
 146                 in = req_capsule_client_get(&pill, &RMF_FLD_MDFLD);
 147                 if (in == NULL) {
 148                         CERROR("cannot unpack fld request\n");
 149                         GOTO(out_pill, rc = -EPROTO);
 150                 }
 151                 out = req_capsule_server_get(&pill, &RMF_FLD_MDFLD);
 152                 if (out == NULL) {
 153                         CERROR("cannot allocate fld response\n");
 154                         GOTO(out_pill, rc = -EPROTO);
 155                 }
 156                 *out = *in;
 157                 rc = fld_server_handle(fld, ctx, *opc, out);
 158         } else {
 159                 CERROR("cannot unpack FLD operation\n");
 160         }
 161
 162 out_pill:
 163         EXIT;
 164         req_capsule_fini(&pill);
 165         return rc;
 166 }
 167
 168 static int fld_req_handle(struct ptlrpc_request *req)
 169 {
 170         int fail = OBD_FAIL_FLD_ALL_REPLY_NET;
 171         const struct lu_context *ctx;
 172         struct lu_site *site;
 173         int rc = -EPROTO;
 174         ENTRY;
 175
 176         OBD_FAIL_RETURN(OBD_FAIL_FLD_ALL_REPLY_NET | OBD_FAIL_ONCE, 0);
 177
 178         ctx = req->rq_svc_thread->t_ctx;
 179         LASSERT(ctx != NULL);
 180         LASSERT(ctx->lc_thread == req->rq_svc_thread);
 181         if (req->rq_reqmsg->opc == FLD_QUERY) {
 182                 if (req->rq_export != NULL) {
 183                         site = req->rq_export->exp_obd->obd_lu_dev->ld_site;
 184                         LASSERT(site != NULL);
 185                         rc = fld_req_handle0(ctx, site->ls_server_fld, req);
 186                 } else {
 187                         CERROR("Unconnected request\n");
 188                         req->rq_status = -ENOTCONN;
 189                         GOTO(out, rc = -ENOTCONN);
 190                 }
 191         } else {
 192                 CERROR("Wrong opcode: %d\n", req->rq_reqmsg->opc);
 193                 req->rq_status = -ENOTSUPP;
 194                 rc = ptlrpc_error(req);
 195                 RETURN(rc);
 196         }
 197
 198         EXIT;
 199 out:
 200         target_send_reply(req, rc, fail);
 201         return 0;
 202 }
 203
 204 /*
 205  * Returns true, if fid is local to this server node.
 206  *
 207  * WARNING: this function is *not* guaranteed to return false if fid is
 208  * remote: it makes an educated conservative guess only.
 209  *
 210  * fid_is_local() is supposed to be used in assertion checks only.
 211  */
 212 int fid_is_local(struct lu_site *site, const struct lu_fid *fid)
 213 {
 214         int result;
 215
 216         result = 1; /* conservatively assume fid is local */
 217         if (site->ls_client_fld != NULL) {
 218                 mdsno_t mds;
 219                 int rc;
 220
 221                 rc = fld_cache_lookup(site->ls_client_fld->fld_cache,
 222                                       fid_seq(fid), &mds);
 223                 if (rc == 0)
 224                         result = (mds == site->ls_node_id);
 225         }
 226         return result;
 227 }
 228 EXPORT_SYMBOL(fid_is_local);
 229
 230 #ifdef LPROCFS
 231 static int fld_server_proc_init(struct lu_server_fld *fld)
 232 {
 233         int rc;
 234         ENTRY;
 235
 236         fld->fld_proc_dir = lprocfs_register(fld->fld_name,
 237                                              proc_lustre_root,
 238                                              NULL, NULL);
 239         if (IS_ERR(fld->fld_proc_dir)) {
 240                 CERROR("LProcFS failed in fld-init\n");
 241                 rc = PTR_ERR(fld->fld_proc_dir);
 242                 GOTO(err, rc);
 243         }
 244
 245         fld->fld_proc_entry = lprocfs_register("services",
 246                                                fld->fld_proc_dir,
 247                                                NULL, NULL);
 248         if (IS_ERR(fld->fld_proc_entry)) {
 249                 CERROR("LProcFS failed in fld-init\n");
 250                 rc = PTR_ERR(fld->fld_proc_entry);
 251                 GOTO(err_type, rc);
 252         }
 253
 254         rc = lprocfs_add_vars(fld->fld_proc_dir,
 255                               fld_server_proc_list, fld);
 256         if (rc) {
 257                 CERROR("can't init FLD proc, rc %d\n", rc);
 258                 GOTO(err_entry, rc);
 259         }
 260
 261         RETURN(0);
 262
 263 err_entry:
 264         lprocfs_remove(fld->fld_proc_entry);
 265 err_type:
 266         lprocfs_remove(fld->fld_proc_dir);
 267 err:
 268         fld->fld_proc_dir = NULL;
 269         fld->fld_proc_entry = NULL;
 270         return rc;
 271 }
 272
 273 static void fld_server_proc_fini(struct lu_server_fld *fld)
 274 {
 275         ENTRY;
 276         if (fld->fld_proc_entry) {
 277                 lprocfs_remove(fld->fld_proc_entry);
 278                 fld->fld_proc_entry = NULL;
 279         }
 280
 281         if (fld->fld_proc_dir) {
 282                 lprocfs_remove(fld->fld_proc_dir);
 283                 fld->fld_proc_dir = NULL;
 284         }
 285         EXIT;
 286 }
 287 #endif
 288
 289 int fld_server_init(struct lu_server_fld *fld,
 290                     const struct lu_context *ctx,
 291                     struct dt_device *dt,
 292                     const char *uuid)
 293 {
 294         int rc;
 295         struct ptlrpc_service_conf fld_conf = {
 296                 .psc_nbufs            = MDS_NBUFS,
 297                 .psc_bufsize          = MDS_BUFSIZE,
 298                 .psc_max_req_size     = FLD_MAXREQSIZE,
 299                 .psc_max_reply_size   = FLD_MAXREPSIZE,
 300                 .psc_req_portal       = FLD_REQUEST_PORTAL,
 301                 .psc_rep_portal       = MDC_REPLY_PORTAL,
 302                 .psc_watchdog_timeout = FLD_SERVICE_WATCHDOG_TIMEOUT,
 303                 .psc_num_threads      = FLD_NUM_THREADS,
 304                 .psc_ctx_tags         = LCT_DT_THREAD|LCT_MD_THREAD
 305         };
 306         ENTRY;
 307
 308         fld->fld_dt = dt;
 309         lu_device_get(&dt->dd_lu_dev);
 310
 311         snprintf(fld->fld_name, sizeof(fld->fld_name),
 312                  "%s-srv-%s", LUSTRE_FLD_NAME, uuid);
 313
 314         rc = fld_index_init(fld, ctx);
 315         if (rc)
 316                 GOTO(out, rc);
 317
 318 #ifdef LPROCFS
 319         rc = fld_server_proc_init(fld);
 320         if (rc)
 321                 GOTO(out, rc);
 322 #endif
 323
 324         fld->fld_service =
 325                 ptlrpc_init_svc_conf(&fld_conf, fld_req_handle,
 326                                      LUSTRE_FLD_NAME,
 327                                      fld->fld_proc_entry, NULL);
 328         if (fld->fld_service != NULL)
 329                 rc = ptlrpc_start_threads(NULL, fld->fld_service,
 330                                           LUSTRE_FLD_NAME);
 331         else
 332                 rc = -ENOMEM;
 333
 334         EXIT;
 335 out:
 336         if (rc)
 337                 fld_server_fini(fld, ctx);
 338         else
 339                 CDEBUG(D_INFO|D_WARNING, "Server FLD\n");
 340         return rc;
 341 }
 342 EXPORT_SYMBOL(fld_server_init);
 343
 344 void fld_server_fini(struct lu_server_fld *fld,
 345                      const struct lu_context *ctx)
 346 {
 347         ENTRY;
 348
 349 #ifdef LPROCFS
 350         fld_server_proc_fini(fld);
 351 #endif
 352
 353         if (fld->fld_service != NULL) {
 354                 ptlrpc_unregister_service(fld->fld_service);
 355                 fld->fld_service = NULL;
 356         }
 357
 358         if (fld->fld_dt != NULL) {
 359                 lu_device_put(&fld->fld_dt->dd_lu_dev);
 360                 fld_index_fini(fld, ctx);
 361                 fld->fld_dt = NULL;
 362         }
 363         EXIT;
 364 }
 365 EXPORT_SYMBOL(fld_server_fini);
 366
 367 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 368 MODULE_DESCRIPTION("Lustre FLD");
 369 MODULE_LICENSE("GPL");
 370
 371 cfs_module(mdd, "0.1.0", fld_mod_init, fld_mod_exit);
 372 #endif