lustre/ofd/ofd_obd.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2014 Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/ofd/ofd_obd.c
  33  *
  34  * This file contains OBD API methods for OBD Filter Device (OFD) which are
  35  * used for export handling, configuration purposes and recovery.
  36  * Several methods are used by ECHO client only since it still uses OBD API.
  37  * Such methods have _echo_ prefix in name.
  38  *
  39  * Author: Andreas Dilger <andreas.dilger@intel.com>
  40  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
  41  * Author: Mikhail Pershin <mike.pershin@intel.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_FILTER
  45
  46 #include <lustre/lustre_idl.h>
  47 #include "ofd_internal.h"
  48 #include <obd_cksum.h>
  49 #include <lustre_ioctl.h>
  50 #include <lustre_quota.h>
  51 #include <lustre_lfsck.h>
  52
  53 /**
  54  * Initialize OFD per-export statistics.
  55  *
  56  * This function sets up procfs entries for various OFD export counters. These
  57  * counters are for per-client statistics tracked on the server.
  58  *
  59  * \param[in] ofd        OFD device
  60  * \param[in] exp        OBD export
  61  * \param[in] client_nid NID of client
  62  *
  63  * \retval              0 if successful
  64  * \retval              negative value on error
  65  */
  66 static int ofd_export_stats_init(struct ofd_device *ofd,
  67                                  struct obd_export *exp, void *client_nid)
  68 {
  69         struct obd_device       *obd = ofd_obd(ofd);
  70         struct nid_stat         *stats;
  71         int                      num_stats;
  72         int                      rc, newnid = 0;
  73
  74         ENTRY;
  75
  76         LASSERT(obd->obd_uses_nid_stats);
  77
  78         if (obd_uuid_equals(&exp->exp_client_uuid, &obd->obd_uuid))
  79                 /* Self-export gets no proc entry */
  80                 RETURN(0);
  81
  82         rc = lprocfs_exp_setup(exp, client_nid, &newnid);
  83         if (rc) {
  84                 /* Mask error for already created
  85                  * /proc entries */
  86                 if (rc == -EALREADY)
  87                         rc = 0;
  88                 RETURN(rc);
  89         }
  90
  91         if (newnid == 0)
  92                 RETURN(0);
  93
  94         stats = exp->exp_nid_stats;
  95         LASSERT(stats != NULL);
  96
  97         num_stats = NUM_OBD_STATS + LPROC_OFD_STATS_LAST;
  98
  99         stats->nid_stats = lprocfs_alloc_stats(num_stats,
 100                                                LPROCFS_STATS_FLAG_NOPERCPU);
 101         if (stats->nid_stats == NULL)
 102                 return -ENOMEM;
 103
 104         lprocfs_init_ops_stats(LPROC_OFD_STATS_LAST, stats->nid_stats);
 105         ofd_stats_counter_init(stats->nid_stats);
 106         rc = lprocfs_register_stats(stats->nid_proc, "stats",
 107                                     stats->nid_stats);
 108         if (rc)
 109                 GOTO(clean, rc);
 110
 111         rc = lprocfs_nid_ldlm_stats_init(stats);
 112         if (rc) {
 113                 lprocfs_free_stats(&stats->nid_stats);
 114                 GOTO(clean, rc);
 115         }
 116
 117         RETURN(0);
 118 clean:
 119         return rc;
 120 }
 121
 122 /**
 123  * Match client and OST server connection feature flags.
 124  *
 125  * Compute the compatibility flags for a connection request based on
 126  * features mutually supported by client and server.
 127  *
 128  * The obd_export::exp_connect_data.ocd_connect_flags field in \a exp
 129  * must not be updated here, otherwise a partially initialized value may
 130  * be exposed. After the connection request is successfully processed,
 131  * the top-level tgt_connect() request handler atomically updates the export
 132  * connect flags from the obd_connect_data::ocd_connect_flags field of the
 133  * reply. \see tgt_connect().
 134  *
 135  * \param[in] env               execution environment
 136  * \param[in] exp               the obd_export associated with this
 137  *                              client/target pair
 138  * \param[in] data              stores data for this connect request
 139  * \param[in] new_connection    is this connection new or not
 140  *
 141  * \retval              0 if success
 142  * \retval              -EPROTO client and server feature requirements are
 143  *                      incompatible
 144  * \retval              -EBADF  OST index in connect request doesn't match
 145  *                      real OST index
 146  */
 147 static int ofd_parse_connect_data(const struct lu_env *env,
 148                                   struct obd_export *exp,
 149                                   struct obd_connect_data *data,
 150                                   bool new_connection)
 151 {
 152         struct ofd_device                *ofd = ofd_exp(exp);
 153         struct filter_export_data        *fed = &exp->exp_filter_data;
 154
 155         if (!data)
 156                 RETURN(0);
 157
 158         CDEBUG(D_RPCTRACE, "%s: cli %s/%p ocd_connect_flags: "LPX64
 159                " ocd_version: %x ocd_grant: %d ocd_index: %u"
 160                " ocd_group %u\n",
 161                exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp,
 162                data->ocd_connect_flags, data->ocd_version,
 163                data->ocd_grant, data->ocd_index, data->ocd_group);
 164
 165         if (fed->fed_group != 0 && fed->fed_group != data->ocd_group) {
 166                 CWARN("!!! This export (nid %s) used object group %d "
 167                       "earlier; now it's trying to use group %d!  This could "
 168                       "be a bug in the MDS. Please report to "
 169                       "https://jira.hpdd.intel.com/\n",
 170                       obd_export_nid2str(exp), fed->fed_group,
 171                       data->ocd_group);
 172                 RETURN(-EPROTO);
 173         }
 174         fed->fed_group = data->ocd_group;
 175
 176         data->ocd_connect_flags &= OST_CONNECT_SUPPORTED;
 177         data->ocd_version = LUSTRE_VERSION_CODE;
 178
 179         /* Kindly make sure the SKIP_ORPHAN flag is from MDS. */
 180         if (data->ocd_connect_flags & OBD_CONNECT_MDS)
 181                 CDEBUG(D_HA, "%s: Received MDS connection for group %u\n",
 182                        exp->exp_obd->obd_name, data->ocd_group);
 183         else if (data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN)
 184                 RETURN(-EPROTO);
 185
 186         if (ofd_grant_param_supp(exp)) {
 187                 exp->exp_filter_data.fed_pagesize = data->ocd_blocksize;
 188                 /* ocd_{blocksize,inodespace} are log2 values */
 189                 data->ocd_blocksize  = ofd->ofd_blockbits;
 190                 data->ocd_inodespace = ofd->ofd_dt_conf.ddp_inodespace;
 191                 /* ocd_grant_extent is in 1K blocks */
 192                 data->ocd_grant_extent = ofd->ofd_dt_conf.ddp_grant_frag >> 10;
 193         }
 194
 195         if (data->ocd_connect_flags & OBD_CONNECT_GRANT)
 196                 data->ocd_grant = ofd_grant_connect(env, exp, data->ocd_grant,
 197                                                     new_connection);
 198
 199         if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
 200                 struct lr_server_data *lsd = &ofd->ofd_lut.lut_lsd;
 201                 int                    index = lsd->lsd_osd_index;
 202
 203                 if (index != data->ocd_index) {
 204                         LCONSOLE_ERROR_MSG(0x136, "Connection from %s to index"
 205                                            " %u doesn't match actual OST index"
 206                                            " %u in last_rcvd file, bad "
 207                                            "configuration?\n",
 208                                            obd_export_nid2str(exp), index,
 209                                            data->ocd_index);
 210                         RETURN(-EBADF);
 211                 }
 212                 if (!(lsd->lsd_feature_compat & OBD_COMPAT_OST)) {
 213                         /* this will only happen on the first connect */
 214                         lsd->lsd_feature_compat |= OBD_COMPAT_OST;
 215                         /* sync is not needed here as lut_client_add will
 216                          * set exp_need_sync flag */
 217                         tgt_server_data_update(env, &ofd->ofd_lut, 0);
 218                 }
 219         }
 220         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_SIZE)) {
 221                 data->ocd_brw_size = 65536;
 222         } else if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
 223                 data->ocd_brw_size = min(data->ocd_brw_size,
 224                                          (__u32)DT_MAX_BRW_SIZE);
 225                 if (data->ocd_brw_size == 0) {
 226                         CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64
 227                                " ocd_version: %x ocd_grant: %d ocd_index: %u "
 228                                "ocd_brw_size is unexpectedly zero, "
 229                                "network data corruption?"
 230                                "Refusing connection of this client\n",
 231                                exp->exp_obd->obd_name,
 232                                exp->exp_client_uuid.uuid,
 233                                exp, data->ocd_connect_flags, data->ocd_version,
 234                                data->ocd_grant, data->ocd_index);
 235                         RETURN(-EPROTO);
 236                 }
 237         }
 238
 239         if (data->ocd_connect_flags & OBD_CONNECT_CKSUM) {
 240                 __u32 cksum_types = data->ocd_cksum_types;
 241
 242                 /* The client set in ocd_cksum_types the checksum types it
 243                  * supports. We have to mask off the algorithms that we don't
 244                  * support */
 245                 data->ocd_cksum_types &= cksum_types_supported_server();
 246
 247                 if (unlikely(data->ocd_cksum_types == 0)) {
 248                         CERROR("%s: Connect with checksum support but no "
 249                                "ocd_cksum_types is set\n",
 250                                exp->exp_obd->obd_name);
 251                         RETURN(-EPROTO);
 252                 }
 253
 254                 CDEBUG(D_RPCTRACE, "%s: cli %s supports cksum type %x, return "
 255                        "%x\n", exp->exp_obd->obd_name, obd_export_nid2str(exp),
 256                        cksum_types, data->ocd_cksum_types);
 257         } else {
 258                 /* This client does not support OBD_CONNECT_CKSUM
 259                  * fall back to CRC32 */
 260                 CDEBUG(D_RPCTRACE, "%s: cli %s does not support "
 261                        "OBD_CONNECT_CKSUM, CRC32 will be used\n",
 262                        exp->exp_obd->obd_name, obd_export_nid2str(exp));
 263         }
 264
 265         if (data->ocd_connect_flags & OBD_CONNECT_MAXBYTES)
 266                 data->ocd_maxbytes = ofd->ofd_dt_conf.ddp_maxbytes;
 267
 268         if (OCD_HAS_FLAG(data, PINGLESS)) {
 269                 if (ptlrpc_pinger_suppress_pings()) {
 270                         spin_lock(&exp->exp_obd->obd_dev_lock);
 271                         list_del_init(&exp->exp_obd_chain_timed);
 272                         spin_unlock(&exp->exp_obd->obd_dev_lock);
 273                 } else {
 274                         data->ocd_connect_flags &= ~OBD_CONNECT_PINGLESS;
 275                 }
 276         }
 277
 278         RETURN(0);
 279 }
 280
 281 /**
 282  * Re-initialize export upon client reconnection.
 283  *
 284  * This function parses connection data from reconnect and resets
 285  * export statistics.
 286  *
 287  * \param[in] env       execution environment
 288  * \param[in] exp       OBD export
 289  * \param[in] obd       OFD device
 290  * \param[in] cluuid    NID of client
 291  * \param[in] data      connection data from request
 292  * \param[in] localdata client NID
 293  *
 294  * \retval              0 if successful
 295  * \retval              negative value on error
 296  */
 297 static int ofd_obd_reconnect(const struct lu_env *env, struct obd_export *exp,
 298                              struct obd_device *obd, struct obd_uuid *cluuid,
 299                              struct obd_connect_data *data, void *localdata)
 300 {
 301         struct ofd_device       *ofd;
 302         int                      rc;
 303
 304         ENTRY;
 305
 306         if (exp == NULL || obd == NULL || cluuid == NULL)
 307                 RETURN(-EINVAL);
 308
 309         ofd = ofd_dev(obd->obd_lu_dev);
 310
 311         rc = ofd_parse_connect_data(env, exp, data, false);
 312         if (rc == 0)
 313                 ofd_export_stats_init(ofd, exp, localdata);
 314
 315         RETURN(rc);
 316 }
 317
 318 /**
 319  * Initialize new client connection.
 320  *
 321  * This function handles new connection to the OFD. The new export is
 322  * created (in context of class_connect()) and persistent client data is
 323  * initialized on storage.
 324  *
 325  * \param[in] env       execution environment
 326  * \param[out] _exp     stores pointer to new export
 327  * \param[in] obd       OFD device
 328  * \param[in] cluuid    client UUID
 329  * \param[in] data      connection data from request
 330  * \param[in] localdata client NID
 331  *
 332  * \retval              0 if successful
 333  * \retval              negative value on error
 334  */
 335 static int ofd_obd_connect(const struct lu_env *env, struct obd_export **_exp,
 336                            struct obd_device *obd, struct obd_uuid *cluuid,
 337                            struct obd_connect_data *data, void *localdata)
 338 {
 339         struct obd_export       *exp;
 340         struct ofd_device       *ofd;
 341         struct lustre_handle     conn = { 0 };
 342         int                      rc;
 343         ENTRY;
 344
 345         if (_exp == NULL || obd == NULL || cluuid == NULL)
 346                 RETURN(-EINVAL);
 347
 348         ofd = ofd_dev(obd->obd_lu_dev);
 349
 350         rc = class_connect(&conn, obd, cluuid);
 351         if (rc)
 352                 RETURN(rc);
 353
 354         exp = class_conn2export(&conn);
 355         LASSERT(exp != NULL);
 356
 357         rc = ofd_parse_connect_data(env, exp, data, true);
 358         if (rc)
 359                 GOTO(out, rc);
 360
 361         if (obd->obd_replayable) {
 362                 struct tg_export_data *ted = &exp->exp_target_data;
 363
 364                 memcpy(ted->ted_lcd->lcd_uuid, cluuid,
 365                        sizeof(ted->ted_lcd->lcd_uuid));
 366                 rc = tgt_client_new(env, exp);
 367                 if (rc != 0)
 368                         GOTO(out, rc);
 369                 ofd_export_stats_init(ofd, exp, localdata);
 370         }
 371
 372         CDEBUG(D_HA, "%s: get connection from MDS %d\n", obd->obd_name,
 373                data ? data->ocd_group : -1);
 374
 375 out:
 376         if (rc != 0) {
 377                 class_disconnect(exp);
 378                 *_exp = NULL;
 379         } else {
 380                 *_exp = exp;
 381         }
 382         RETURN(rc);
 383 }
 384
 385 /**
 386  * Disconnect a connected client.
 387  *
 388  * This function terminates the client connection. The client export is
 389  * disconnected (cleaned up) and client data on persistent storage is removed.
 390  *
 391  * \param[in] exp       OBD export
 392  *
 393  * \retval              0 if successful
 394  * \retval              negative value on error
 395  */
 396 int ofd_obd_disconnect(struct obd_export *exp)
 397 {
 398         struct ofd_device       *ofd = ofd_exp(exp);
 399         struct lu_env            env;
 400         int                      rc;
 401
 402         ENTRY;
 403
 404         LASSERT(exp);
 405         class_export_get(exp);
 406
 407         if (!(exp->exp_flags & OBD_OPT_FORCE))
 408                 ofd_grant_sanity_check(ofd_obd(ofd), __FUNCTION__);
 409
 410         rc = server_disconnect_export(exp);
 411
 412         ofd_grant_discard(exp);
 413
 414         /* Do not erase record for recoverable client. */
 415         if (exp->exp_obd->obd_replayable &&
 416             (!exp->exp_obd->obd_fail || exp->exp_failed)) {
 417                 rc = lu_env_init(&env, LCT_DT_THREAD);
 418                 if (rc)
 419                         GOTO(out, rc);
 420
 421                 tgt_client_del(&env, exp);
 422                 lu_env_fini(&env);
 423         }
 424 out:
 425         class_export_put(exp);
 426         RETURN(rc);
 427 }
 428
 429 /**
 430  * Implementation of obd_ops::o_init_export.
 431  *
 432  * This function is called from class_new_export() and initializes
 433  * the OFD-specific data for new export.
 434  *
 435  * \param[in] exp       OBD export
 436  *
 437  * \retval              0 if successful
 438  * \retval              negative value on error
 439  */
 440 static int ofd_init_export(struct obd_export *exp)
 441 {
 442         int rc;
 443
 444         spin_lock_init(&exp->exp_filter_data.fed_lock);
 445         INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list);
 446         atomic_set(&exp->exp_filter_data.fed_soft_sync_count, 0);
 447         spin_lock(&exp->exp_lock);
 448         exp->exp_connecting = 1;
 449         spin_unlock(&exp->exp_lock);
 450
 451         /* self-export doesn't need client data and ldlm initialization */
 452         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
 453                                      &exp->exp_client_uuid)))
 454                 return 0;
 455
 456         rc = tgt_client_alloc(exp);
 457         if (rc == 0)
 458                 ldlm_init_export(exp);
 459         if (rc)
 460                 CERROR("%s: Can't initialize export: rc %d\n",
 461                        exp->exp_obd->obd_name, rc);
 462         return rc;
 463 }
 464
 465 /**
 466  * Implementation of obd_ops::o_destroy_export.
 467  *
 468  * This function is called from class_export_destroy() to cleanup
 469  * the OFD-specific data for export being destroyed.
 470  *
 471  * \param[in] exp       OBD export
 472  *
 473  * \retval              0 if successful
 474  * \retval              negative value on error
 475  */
 476 static int ofd_destroy_export(struct obd_export *exp)
 477 {
 478         struct ofd_device *ofd = ofd_exp(exp);
 479
 480         if (exp->exp_filter_data.fed_pending)
 481                 CERROR("%s: cli %s/%p has %lu pending on destroyed export"
 482                        "\n", exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
 483                        exp, exp->exp_filter_data.fed_pending);
 484
 485         target_destroy_export(exp);
 486
 487         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
 488                                      &exp->exp_client_uuid)))
 489                 return 0;
 490
 491         ldlm_destroy_export(exp);
 492         tgt_client_free(exp);
 493
 494         ofd_fmd_cleanup(exp);
 495
 496         /*
 497          * discard grants once we're sure no more
 498          * interaction with the client is possible
 499          */
 500         ofd_grant_discard(exp);
 501         ofd_fmd_cleanup(exp);
 502
 503         if (exp_connect_flags(exp) & OBD_CONNECT_GRANT_SHRINK) {
 504                 if (ofd->ofd_tot_granted_clients > 0)
 505                         ofd->ofd_tot_granted_clients --;
 506         }
 507
 508         if (!(exp->exp_flags & OBD_OPT_FORCE))
 509                 ofd_grant_sanity_check(exp->exp_obd, __FUNCTION__);
 510
 511         LASSERT(list_empty(&exp->exp_filter_data.fed_mod_list));
 512         return 0;
 513 }
 514
 515 /**
 516  * Notify all devices in server stack about recovery completion.
 517  *
 518  * This function calls ldo_recovery_complete() for all lower devices in the
 519  * server stack so they will be prepared for normal operations.
 520  *
 521  * \param[in] env       execution environment
 522  * \param[in] ofd       OFD device
 523  *
 524  * \retval              0 if successful
 525  * \retval              negative value on error
 526  */
 527 int ofd_postrecov(const struct lu_env *env, struct ofd_device *ofd)
 528 {
 529         struct lu_device *ldev = &ofd->ofd_dt_dev.dd_lu_dev;
 530         struct lfsck_start_param lsp;
 531         int rc;
 532
 533         CDEBUG(D_HA, "%s: recovery is over\n", ofd_name(ofd));
 534
 535         lsp.lsp_start = NULL;
 536         lsp.lsp_index_valid = 0;
 537         rc = lfsck_start(env, ofd->ofd_osd, &lsp);
 538         if (rc != 0 && rc != -EALREADY)
 539                 CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n",
 540                       ofd_name(ofd), rc);
 541
 542         return ldev->ld_ops->ldo_recovery_complete(env, ldev);
 543 }
 544
 545 /**
 546  * Implementation of obd_ops::o_postrecov.
 547  *
 548  * This function is called from target_finish_recovery() upon recovery
 549  * completion.
 550  *
 551  * \param[in] obd       OBD device of OFD
 552  *
 553  * \retval              0 if successful
 554  * \retval              negative value on error
 555  */
 556 int ofd_obd_postrecov(struct obd_device *obd)
 557 {
 558         struct lu_env            env;
 559         struct lu_device        *ldev = obd->obd_lu_dev;
 560         int                      rc;
 561
 562         ENTRY;
 563
 564         rc = lu_env_init(&env, LCT_DT_THREAD);
 565         if (rc)
 566                 RETURN(rc);
 567         ofd_info_init(&env, obd->obd_self_export);
 568
 569         rc = ofd_postrecov(&env, ofd_dev(ldev));
 570
 571         lu_env_fini(&env);
 572         RETURN(rc);
 573 }
 574
 575 /**
 576  * Implementation of obd_ops::o_set_info_async.
 577  *
 578  * This function is not called from request handler, it is only used by
 579  * class_notify_sptlrpc_conf() locally by direct obd_set_info_async() call.
 580  * \see  ofd_set_info_hdl() for request handler function.
 581  *
 582  * \param[in] env       execution environment
 583  * \param[in] exp       OBD export of OFD device
 584  * \param[in] keylen    length of \a key
 585  * \param[in] key       key name
 586  * \param[in] vallen    length of \a val
 587  * \param[in] val       the \a key value
 588  * \param[in] set       not used in OFD
 589  *
 590  * \retval              0 if successful
 591  * \retval              negative value on error
 592  */
 593 static int ofd_set_info_async(const struct lu_env *env, struct obd_export *exp,
 594                               __u32 keylen, void *key, __u32 vallen, void *val,
 595                               struct ptlrpc_request_set *set)
 596 {
 597         int rc = 0;
 598
 599         ENTRY;
 600
 601         if (exp->exp_obd == NULL) {
 602                 CDEBUG(D_IOCTL, "invalid export %p\n", exp);
 603                 RETURN(-EINVAL);
 604         }
 605
 606         if (KEY_IS(KEY_SPTLRPC_CONF)) {
 607                 rc = tgt_adapt_sptlrpc_conf(class_exp2tgt(exp), 0);
 608         } else {
 609                 CERROR("%s: Unsupported key %s\n",
 610                        exp->exp_obd->obd_name, (char*)key);
 611                 rc = -EOPNOTSUPP;
 612         }
 613         RETURN(rc);
 614 }
 615
 616 /**
 617  * Implementation of obd_ops::o_get_info.
 618  *
 619  * This function is not called from request handler, it is only used by
 620  * direct call from nrs_orr_range_fill_physical() in ptlrpc, see LU-3239.
 621  *
 622  * \see  ofd_get_info_hdl() for request handler function.
 623  *
 624  * \param[in]  env      execution environment
 625  * \param[in]  exp      OBD export of OFD device
 626  * \param[in]  keylen   length of \a key
 627  * \param[in]  key      key name
 628  * \param[out] vallen   length of key value
 629  * \param[out] val      the key value to return
 630  * \param[in]  lsm      not used in OFD
 631  *
 632  * \retval              0 if successful
 633  * \retval              negative value on error
 634  */
 635 static int ofd_get_info(const struct lu_env *env, struct obd_export *exp,
 636                         __u32 keylen, void *key, __u32 *vallen, void *val,
 637                         struct lov_stripe_md *lsm)
 638 {
 639         struct ofd_thread_info          *info;
 640         struct ofd_device               *ofd;
 641         struct ll_fiemap_info_key       *fm_key = key;
 642         struct ll_user_fiemap           *fiemap = val;
 643         int                              rc = 0;
 644
 645         ENTRY;
 646
 647         if (exp->exp_obd == NULL) {
 648                 CDEBUG(D_IOCTL, "invalid client export %p\n", exp);
 649                 RETURN(-EINVAL);
 650         }
 651
 652         ofd = ofd_exp(exp);
 653
 654         if (KEY_IS(KEY_FIEMAP)) {
 655                 info = ofd_info_init(env, exp);
 656
 657                 rc = ostid_to_fid(&info->fti_fid, &fm_key->oa.o_oi,
 658                                   ofd->ofd_lut.lut_lsd.lsd_osd_index);
 659                 if (rc != 0)
 660                         RETURN(rc);
 661
 662                 rc = ofd_fiemap_get(env, ofd, &info->fti_fid, fiemap);
 663         } else {
 664                 CERROR("%s: not supported key %s\n", ofd_name(ofd), (char*)key);
 665                 rc = -EOPNOTSUPP;
 666         }
 667
 668         RETURN(rc);
 669 }
 670
 671 /**
 672  * Get file system statistics of OST server.
 673  *
 674  * Helper function for ofd_statfs(), also used by grant code.
 675  * Implements caching for statistics to avoid calling OSD device each time.
 676  *
 677  * \param[in]  env        execution environment
 678  * \param[in]  ofd        OFD device
 679  * \param[out] osfs       statistic data to return
 680  * \param[in]  max_age    maximum age for cached data
 681  * \param[in]  from_cache show that data was get from cache or not
 682  *
 683  * \retval              0 if successful
 684  * \retval              negative value on error
 685  */
 686 int ofd_statfs_internal(const struct lu_env *env, struct ofd_device *ofd,
 687                         struct obd_statfs *osfs, __u64 max_age, int *from_cache)
 688 {
 689         int rc = 0;
 690         ENTRY;
 691
 692         down_read(&ofd->ofd_lastid_rwsem);
 693         /* Currently, for safe, we do not distinguish which LAST_ID is broken,
 694          * we may do that in the future.
 695          * Return -ENOSPC until the LAST_ID rebuilt. */
 696         if (unlikely(ofd->ofd_lastid_rebuilding))
 697                 GOTO(out, rc = -ENOSPC);
 698
 699         spin_lock(&ofd->ofd_osfs_lock);
 700         if (cfs_time_before_64(ofd->ofd_osfs_age, max_age) || max_age == 0) {
 701                 obd_size unstable;
 702
 703                 /* statfs data are too old, get up-to-date one.
 704                  * we must be cautious here since multiple threads might be
 705                  * willing to update statfs data concurrently and we must
 706                  * grant that cached statfs data are always consistent */
 707
 708                 if (ofd->ofd_statfs_inflight == 0)
 709                         /* clear inflight counter if no users, although it would
 710                          * take a while to overflow this 64-bit counter ... */
 711                         ofd->ofd_osfs_inflight = 0;
 712                 /* notify ofd_grant_commit() that we want to track writes
 713                  * completed as of now */
 714                 ofd->ofd_statfs_inflight++;
 715                 /* record value of inflight counter before running statfs to
 716                  * compute the diff once statfs is completed */
 717                 unstable = ofd->ofd_osfs_inflight;
 718                 spin_unlock(&ofd->ofd_osfs_lock);
 719
 720                 /* statfs can sleep ... hopefully not for too long since we can
 721                  * call it fairly often as space fills up */
 722                 rc = dt_statfs(env, ofd->ofd_osd, osfs);
 723                 if (unlikely(rc))
 724                         GOTO(out, rc);
 725
 726                 spin_lock(&ofd->ofd_grant_lock);
 727                 spin_lock(&ofd->ofd_osfs_lock);
 728                 /* calculate how much space was written while we released the
 729                  * ofd_osfs_lock */
 730                 unstable = ofd->ofd_osfs_inflight - unstable;
 731                 ofd->ofd_osfs_unstable = 0;
 732                 if (unstable) {
 733                         /* some writes completed while we were running statfs
 734                          * w/o the ofd_osfs_lock. Those ones got added to
 735                          * the cached statfs data that we are about to crunch.
 736                          * Take them into account in the new statfs data */
 737                         osfs->os_bavail -= min_t(obd_size, osfs->os_bavail,
 738                                                unstable >> ofd->ofd_blockbits);
 739                         /* However, we don't really know if those writes got
 740                          * accounted in the statfs call, so tell
 741                          * ofd_grant_space_left() there is some uncertainty
 742                          * on the accounting of those writes.
 743                          * The purpose is to prevent spurious error messages in
 744                          * ofd_grant_space_left() since those writes might be
 745                          * accounted twice. */
 746                         ofd->ofd_osfs_unstable += unstable;
 747                 }
 748                 /* similarly, there is some uncertainty on write requests
 749                  * between prepare & commit */
 750                 ofd->ofd_osfs_unstable += ofd->ofd_tot_pending;
 751                 spin_unlock(&ofd->ofd_grant_lock);
 752
 753                 /* finally udpate cached statfs data */
 754                 ofd->ofd_osfs = *osfs;
 755                 ofd->ofd_osfs_age = cfs_time_current_64();
 756
 757                 ofd->ofd_statfs_inflight--; /* stop tracking */
 758                 if (ofd->ofd_statfs_inflight == 0)
 759                         ofd->ofd_osfs_inflight = 0;
 760                 spin_unlock(&ofd->ofd_osfs_lock);
 761
 762                 if (from_cache)
 763                         *from_cache = 0;
 764         } else {
 765                 /* use cached statfs data */
 766                 *osfs = ofd->ofd_osfs;
 767                 spin_unlock(&ofd->ofd_osfs_lock);
 768                 if (from_cache)
 769                         *from_cache = 1;
 770         }
 771
 772         GOTO(out, rc);
 773
 774 out:
 775         up_read(&ofd->ofd_lastid_rwsem);
 776
 777         return rc;
 778 }
 779
 780 /**
 781  * Implementation of obd_ops::o_statfs.
 782  *
 783  * This function returns information about a storage file system.
 784  * It is called from several places by using the OBD API as well as
 785  * by direct call, e.g. from request handler.
 786  *
 787  * \see  ofd_statfs_hdl() for request handler function.
 788  *
 789  * Report also the state of the OST to the caller in osfs->os_state
 790  * (OS_STATE_READONLY, OS_STATE_DEGRADED).
 791  *
 792  * \param[in]  env      execution environment
 793  * \param[in]  exp      OBD export of OFD device
 794  * \param[out] osfs     statistic data to return
 795  * \param[in]  max_age  maximum age for cached data
 796  * \param[in]  flags    not used in OFD
 797  *
 798  * \retval              0 if successful
 799  * \retval              negative value on error
 800  */
 801 int ofd_statfs(const struct lu_env *env,  struct obd_export *exp,
 802                struct obd_statfs *osfs, __u64 max_age, __u32 flags)
 803 {
 804         struct obd_device       *obd = class_exp2obd(exp);
 805         struct ofd_device       *ofd = ofd_exp(exp);
 806         int                      rc;
 807
 808         ENTRY;
 809
 810         rc = ofd_statfs_internal(env, ofd, osfs, max_age, NULL);
 811         if (unlikely(rc))
 812                 GOTO(out, rc);
 813
 814         /* at least try to account for cached pages.  its still racy and
 815          * might be under-reporting if clients haven't announced their
 816          * caches with brw recently */
 817
 818         CDEBUG(D_SUPER | D_CACHE, "blocks cached "LPU64" granted "LPU64
 819                " pending "LPU64" free "LPU64" avail "LPU64"\n",
 820                ofd->ofd_tot_dirty, ofd->ofd_tot_granted, ofd->ofd_tot_pending,
 821                osfs->os_bfree << ofd->ofd_blockbits,
 822                osfs->os_bavail << ofd->ofd_blockbits);
 823
 824         osfs->os_bavail -= min_t(obd_size, osfs->os_bavail,
 825                                  ((ofd->ofd_tot_dirty + ofd->ofd_tot_pending +
 826                                    osfs->os_bsize - 1) >> ofd->ofd_blockbits));
 827
 828         /* The QoS code on the MDS does not care about space reserved for
 829          * precreate, so take it out. */
 830         if (exp_connect_flags(exp) & OBD_CONNECT_MDS) {
 831                 struct filter_export_data *fed;
 832
 833                 fed = &obd->obd_self_export->exp_filter_data;
 834                 osfs->os_bavail -= min_t(obd_size, osfs->os_bavail,
 835                                          fed->fed_grant >> ofd->ofd_blockbits);
 836         }
 837
 838         ofd_grant_sanity_check(obd, __FUNCTION__);
 839         CDEBUG(D_CACHE, LPU64" blocks: "LPU64" free, "LPU64" avail; "
 840                LPU64" objects: "LPU64" free; state %x\n",
 841                osfs->os_blocks, osfs->os_bfree, osfs->os_bavail,
 842                osfs->os_files, osfs->os_ffree, osfs->os_state);
 843
 844         if (OBD_FAIL_CHECK_VALUE(OBD_FAIL_OST_ENOINO,
 845                                  ofd->ofd_lut.lut_lsd.lsd_osd_index))
 846                 osfs->os_ffree = 0;
 847
 848         /* OS_STATE_READONLY can be set by OSD already */
 849         if (ofd->ofd_raid_degraded)
 850                 osfs->os_state |= OS_STATE_DEGRADED;
 851
 852         if (obd->obd_self_export != exp && ofd_grant_compat(exp, ofd)) {
 853                 /* clients which don't support OBD_CONNECT_GRANT_PARAM
 854                  * should not see a block size > page size, otherwise
 855                  * cl_lost_grant goes mad. Therefore, we emulate a 4KB (=2^12)
 856                  * block size which is the biggest block size known to work
 857                  * with all client's page size. */
 858                 osfs->os_blocks <<= ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT;
 859                 osfs->os_bfree  <<= ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT;
 860                 osfs->os_bavail <<= ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT;
 861                 osfs->os_bsize    = 1 << COMPAT_BSIZE_SHIFT;
 862         }
 863
 864         if (OBD_FAIL_CHECK_VALUE(OBD_FAIL_OST_ENOSPC,
 865                                  ofd->ofd_lut.lut_lsd.lsd_osd_index))
 866                 osfs->os_bfree = osfs->os_bavail = 2;
 867
 868         EXIT;
 869 out:
 870         return rc;
 871 }
 872
 873 /**
 874  * Implementation of obd_ops::o_setattr.
 875  *
 876  * This function is only used by ECHO client when it is run on top of OFD,
 877  * \see  ofd_setattr_hdl() for request handler function.
 878
 879  * \param[in] env       execution environment
 880  * \param[in] exp       OBD export of OFD device
 881  * \param[in] oinfo     obd_info with setattr parameters
 882  * \param[in] oti       not used in OFD
 883  *
 884  * \retval              0 if successful
 885  * \retval              negative value on error
 886  */
 887 int ofd_echo_setattr(const struct lu_env *env, struct obd_export *exp,
 888                      struct obd_info *oinfo, struct obd_trans_info *oti)
 889 {
 890         struct ofd_thread_info  *info;
 891         struct ofd_device       *ofd = ofd_exp(exp);
 892         struct ldlm_namespace   *ns = ofd->ofd_namespace;
 893         struct ldlm_resource    *res;
 894         struct ofd_object       *fo;
 895         struct obdo             *oa = oinfo->oi_oa;
 896         struct lu_fid           *fid = &oa->o_oi.oi_fid;
 897         struct filter_fid       *ff = NULL;
 898         int                      rc = 0;
 899
 900         ENTRY;
 901
 902         info = ofd_info_init(env, exp);
 903
 904         ost_fid_build_resid(fid, &info->fti_resid);
 905
 906         /* This would be very bad - accidentally truncating a file when
 907          * changing the time or similar - bug 12203. */
 908         if (oa->o_valid & OBD_MD_FLSIZE &&
 909             oinfo->oi_policy.l_extent.end != OBD_OBJECT_EOF) {
 910                 static char mdsinum[48];
 911
 912                 if (oa->o_valid & OBD_MD_FLFID)
 913                         snprintf(mdsinum, sizeof(mdsinum) - 1,
 914                                  "of parent "DFID, oa->o_parent_seq,
 915                                  oa->o_parent_oid, 0);
 916                 else
 917                         mdsinum[0] = '\0';
 918
 919                 CERROR("%s: setattr from %s trying to truncate object "DFID
 920                        " %s\n", ofd_name(ofd), obd_export_nid2str(exp),
 921                        PFID(fid), mdsinum);
 922                 GOTO(out, rc = -EPERM);
 923         }
 924
 925         fo = ofd_object_find_exists(env, ofd, fid);
 926         if (IS_ERR(fo)) {
 927                 CERROR("%s: can't find object "DFID"\n",
 928                        ofd_name(ofd), PFID(fid));
 929                 GOTO(out, rc = PTR_ERR(fo));
 930         }
 931
 932         la_from_obdo(&info->fti_attr, oa, oa->o_valid);
 933         info->fti_attr.la_valid &= ~LA_TYPE;
 934
 935         if (oa->o_valid & OBD_MD_FLFID) {
 936                 ff = &info->fti_mds_fid;
 937                 ofd_prepare_fidea(ff, oa);
 938         }
 939
 940         /* setting objects attributes (including owner/group) */
 941         rc = ofd_attr_set(env, fo, &info->fti_attr, ff);
 942         if (rc)
 943                 GOTO(out_unlock, rc);
 944
 945         ofd_counter_incr(exp, LPROC_OFD_STATS_SETATTR, NULL, 1);
 946         EXIT;
 947 out_unlock:
 948         ofd_object_put(env, fo);
 949 out:
 950         if (rc == 0) {
 951                 /* we do not call this before to avoid lu_object_find() in
 952                  *  ->lvbo_update() holding another reference on the object.
 953                  * otherwise concurrent destroy can make the object unavailable
 954                  * for 2nd lu_object_find() waiting for the first reference
 955                  * to go... deadlock! */
 956                 res = ldlm_resource_get(ns, NULL, &info->fti_resid, LDLM_EXTENT, 0);
 957                 if (!IS_ERR(res)) {
 958                         ldlm_res_lvbo_update(res, NULL, 0);
 959                         ldlm_resource_putref(res);
 960                 }
 961         }
 962
 963         return rc;
 964 }
 965
 966 /**
 967  * Destroy OFD object by its FID.
 968  *
 969  * Supplemental function to destroy object by FID, it is used by request
 970  * handler and by ofd_echo_destroy() below to find object by FID, lock it
 971  * and call ofd_object_destroy() finally.
 972  *
 973  * \param[in] env       execution environment
 974  * \param[in] ofd       OFD device
 975  * \param[in] fid       FID of object
 976  * \param[in] orphan    set if object being destroyed is an orphan
 977  *
 978  * \retval              0 if successful
 979  * \retval              negative value on error
 980  */
 981 int ofd_destroy_by_fid(const struct lu_env *env, struct ofd_device *ofd,
 982                        const struct lu_fid *fid, int orphan)
 983 {
 984         struct ofd_thread_info  *info = ofd_info(env);
 985         struct lustre_handle     lockh;
 986         __u64                    flags = LDLM_FL_AST_DISCARD_DATA;
 987         __u64                    rc = 0;
 988         ldlm_policy_data_t       policy = {
 989                                         .l_extent = { 0, OBD_OBJECT_EOF }
 990                                  };
 991         struct ofd_object       *fo;
 992
 993         ENTRY;
 994
 995         fo = ofd_object_find_exists(env, ofd, fid);
 996         if (IS_ERR(fo))
 997                 RETURN(PTR_ERR(fo));
 998
 999         /* Tell the clients that the object is gone now and that they should
1000          * throw away any cached pages. */
1001         ost_fid_build_resid(fid, &info->fti_resid);
1002         rc = ldlm_cli_enqueue_local(ofd->ofd_namespace, &info->fti_resid,
1003                                     LDLM_EXTENT, &policy, LCK_PW, &flags,
1004                                     ldlm_blocking_ast, ldlm_completion_ast,
1005                                     NULL, NULL, 0, LVB_T_NONE, NULL, &lockh);
1006
1007         /* We only care about the side-effects, just drop the lock. */
1008         if (rc == ELDLM_OK)
1009                 ldlm_lock_decref(&lockh, LCK_PW);
1010
1011         LASSERT(fo != NULL);
1012
1013         rc = ofd_object_destroy(env, fo, orphan);
1014         EXIT;
1015
1016         ofd_object_put(env, fo);
1017         RETURN(rc);
1018 }
1019
1020 /**
1021  * Implementation of obd_ops::o_destroy.
1022  *
1023  * This function is only used by ECHO client when it is run on top of OFD,
1024  * \see  ofd_destroy_hdl() for request handler function.
1025
1026  * \param[in] env       execution environment
1027  * \param[in] exp       OBD export of OFD device
1028  * \param[in] oa        obdo structure with FID
1029  * \param[in] md        not used in OFD
1030  * \param[in] oti       not used in OFD
1031  * \param[in] md_exp    not used in OFD
1032  * \param[in] capa      not used in OFD
1033  *
1034  * Note: this is OBD API method which is common API for server OBDs and
1035  * client OBDs. Thus some parameters used in client OBDs may not be used
1036  * on server OBDs and vice versa.
1037  *
1038  * \retval              0 if successful
1039  * \retval              negative value on error
1040  */
1041 int ofd_echo_destroy(const struct lu_env *env, struct obd_export *exp,
1042                      struct obdo *oa, struct lov_stripe_md *md,
1043                      struct obd_trans_info *oti, struct obd_export *md_exp,
1044                      void *capa)
1045 {
1046         struct ofd_device       *ofd = ofd_exp(exp);
1047         struct lu_fid           *fid = &oa->o_oi.oi_fid;
1048         int                      rc = 0;
1049
1050         ENTRY;
1051
1052         ofd_info_init(env, exp);
1053
1054         CDEBUG(D_HA, "%s: Destroy object "DFID"\n", ofd_name(ofd), PFID(fid));
1055
1056         rc = ofd_destroy_by_fid(env, ofd, fid, 0);
1057         if (rc == -ENOENT) {
1058                 CDEBUG(D_INODE, "%s: destroying non-existent object "DFID"\n",
1059                        ofd_name(ofd), PFID(fid));
1060                 GOTO(out, rc);
1061         } else if (rc != 0) {
1062                 CERROR("%s: error destroying object "DFID": %d\n",
1063                        ofd_name(ofd), PFID(fid), rc);
1064                 GOTO(out, rc);
1065         }
1066         EXIT;
1067 out:
1068         return rc;
1069 }
1070
1071 /**
1072  * Implementation of obd_ops::o_create.
1073  *
1074  * This function is only used by ECHO client when it is run on top of OFD
1075  * and just creates an object.
1076  * \see  ofd_create_hdl() for request handler function.
1077  *
1078  * \param[in]  env      execution environment
1079  * \param[in]  exp      OBD export of OFD device
1080  * \param[in]  oa       obdo structure with FID sequence to use
1081  * \param[out] ea       contains object ID/SEQ to return
1082  * \param[in]  oti      not used in OFD
1083  *
1084  * Note: this is OBD API method which is common API for server OBDs and
1085  * client OBDs. Thus some parameters used in client OBDs may not be used
1086  * on server OBDs and vice versa.
1087  *
1088  * \retval              0 if successful
1089  * \retval              negative value on error
1090  */
1091 int ofd_echo_create(const struct lu_env *env, struct obd_export *exp,
1092                     struct obdo *oa, struct lov_stripe_md **ea,
1093                     struct obd_trans_info *oti)
1094 {
1095         struct ofd_device       *ofd = ofd_exp(exp);
1096         struct ofd_thread_info  *info;
1097         obd_seq                  seq = ostid_seq(&oa->o_oi);
1098         struct ofd_seq          *oseq;
1099         int                      rc = 0, diff = 1;
1100         obd_id                   next_id;
1101         int                      count;
1102
1103         ENTRY;
1104
1105         info = ofd_info_init(env, exp);
1106
1107         LASSERT(seq == FID_SEQ_ECHO);
1108         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
1109
1110         CDEBUG(D_INFO, "ofd_create("DOSTID")\n", POSTID(&oa->o_oi));
1111
1112         down_read(&ofd->ofd_lastid_rwsem);
1113         /* Currently, for safe, we do not distinguish which LAST_ID is broken,
1114          * we may do that in the future.
1115          * Return -ENOSPC until the LAST_ID rebuilt. */
1116         if (unlikely(ofd->ofd_lastid_rebuilding))
1117                 GOTO(out_sem, rc = -ENOSPC);
1118
1119         oseq = ofd_seq_load(env, ofd, seq);
1120         if (IS_ERR(oseq)) {
1121                 CERROR("%s: Can't find FID Sequence "LPX64": rc = %ld\n",
1122                        ofd_name(ofd), seq, PTR_ERR(oseq));
1123                 GOTO(out_sem, rc = -EINVAL);
1124         }
1125
1126         mutex_lock(&oseq->os_create_lock);
1127         rc = ofd_grant_create(env, ofd_obd(ofd)->obd_self_export, &diff);
1128         if (rc < 0) {
1129                 CDEBUG(D_HA, "%s: failed to acquire grant space for "
1130                        "precreate (%d): rc = %d\n", ofd_name(ofd), diff, rc);
1131                 diff = 0;
1132                 GOTO(out, rc);
1133         }
1134
1135         next_id = ofd_seq_last_oid(oseq) + 1;
1136         count = ofd_precreate_batch(ofd, diff);
1137
1138         rc = ofd_precreate_objects(env, ofd, next_id, oseq, count, 0);
1139         if (rc < 0) {
1140                 CERROR("%s: unable to precreate: rc = %d\n",
1141                        ofd_name(ofd), rc);
1142         } else {
1143                 ostid_set_id(&oa->o_oi, ofd_seq_last_oid(oseq));
1144                 oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
1145                 rc = 0;
1146         }
1147
1148         ofd_grant_commit(env, ofd_obd(ofd)->obd_self_export, rc);
1149 out:
1150         mutex_unlock(&oseq->os_create_lock);
1151         if (rc == 0 && ea != NULL) {
1152                 struct lov_stripe_md *lsm = *ea;
1153
1154                 lsm->lsm_oi = oa->o_oi;
1155         }
1156         ofd_seq_put(env, oseq);
1157
1158 out_sem:
1159         up_read(&ofd->ofd_lastid_rwsem);
1160         RETURN(rc);
1161 }
1162
1163 /**
1164  * Implementation of obd_ops::o_getattr.
1165  *
1166  * This function is only used by ECHO client when it is run on top of OFD
1167  * and returns attributes of object.
1168  * \see  ofd_getattr_hdl() for request handler function.
1169  *
1170  * \param[in]     env   execution environment
1171  * \param[in]     exp   OBD export of OFD device
1172  * \param[in,out] oinfo contains FID of object to get attributes from and
1173  *                      is used to return attributes back
1174  *
1175  * \retval              0 if successful
1176  * \retval              negative value on error
1177  */
1178 int ofd_echo_getattr(const struct lu_env *env, struct obd_export *exp,
1179                      struct obd_info *oinfo)
1180 {
1181         struct ofd_device       *ofd = ofd_exp(exp);
1182         struct ofd_thread_info  *info;
1183         struct lu_fid           *fid = &oinfo->oi_oa->o_oi.oi_fid;
1184         struct ofd_object       *fo;
1185         int                      rc = 0;
1186
1187         ENTRY;
1188
1189         info = ofd_info_init(env, exp);
1190
1191         fo = ofd_object_find_exists(env, ofd, fid);
1192         if (IS_ERR(fo))
1193                 GOTO(out, rc = PTR_ERR(fo));
1194
1195         LASSERT(fo != NULL);
1196         rc = ofd_attr_get(env, fo, &info->fti_attr);
1197         oinfo->oi_oa->o_valid = OBD_MD_FLID;
1198         if (rc == 0) {
1199                 __u64 curr_version;
1200
1201                 obdo_from_la(oinfo->oi_oa, &info->fti_attr,
1202                              OFD_VALID_FLAGS | LA_UID | LA_GID);
1203
1204                 /* Store object version in reply */
1205                 curr_version = dt_version_get(env, ofd_object_child(fo));
1206                 if ((__s64)curr_version != -EOPNOTSUPP) {
1207                         oinfo->oi_oa->o_valid |= OBD_MD_FLDATAVERSION;
1208                         oinfo->oi_oa->o_data_version = curr_version;
1209                 }
1210         }
1211
1212         ofd_object_put(env, fo);
1213 out:
1214         RETURN(rc);
1215 }
1216
1217 /**
1218  * Get object version for OBD_IOC_GET_OBJ_VERSION ioctl.
1219  *
1220  * This is supplemental function for ofd_iocontrol() to return object
1221  * version for lctl tool.
1222  *
1223  * \param[in]  env      execution environment
1224  * \param[in]  ofd      OFD device
1225  * \param[out] karg     ioctl data
1226  *
1227  * \retval              0 if successful
1228  * \retval              negative value on error
1229  */
1230 static int ofd_ioc_get_obj_version(const struct lu_env *env,
1231                                    struct ofd_device *ofd, void *karg)
1232 {
1233         struct obd_ioctl_data *data = karg;
1234         struct lu_fid          fid;
1235         struct ofd_object     *fo;
1236         dt_obj_version_t       version;
1237         int                    rc = 0;
1238
1239         ENTRY;
1240
1241         if (data->ioc_inlbuf2 == NULL || data->ioc_inllen2 != sizeof(version))
1242                 GOTO(out, rc = -EINVAL);
1243
1244         if (data->ioc_inlbuf1 != NULL && data->ioc_inllen1 == sizeof(fid)) {
1245                 fid = *(struct lu_fid *)data->ioc_inlbuf1;
1246         } else if (data->ioc_inlbuf3 != NULL &&
1247                    data->ioc_inllen3 == sizeof(__u64) &&
1248                    data->ioc_inlbuf4 != NULL &&
1249                    data->ioc_inllen4 == sizeof(__u64)) {
1250                 struct ost_id ostid;
1251
1252                 ostid_set_seq(&ostid, *(__u64 *)data->ioc_inlbuf4);
1253                 ostid_set_id(&ostid, *(__u64 *)data->ioc_inlbuf3);
1254                 rc = ostid_to_fid(&fid, &ostid,
1255                                   ofd->ofd_lut.lut_lsd.lsd_osd_index);
1256                 if (rc != 0)
1257                         GOTO(out, rc);
1258         } else {
1259                 GOTO(out, rc = -EINVAL);
1260         }
1261
1262         if (!fid_is_sane(&fid))
1263                 GOTO(out, rc = -EINVAL);
1264
1265         fo = ofd_object_find(env, ofd, &fid);
1266         if (IS_ERR(fo))
1267                 GOTO(out, rc = PTR_ERR(fo));
1268
1269         if (!ofd_object_exists(fo))
1270                 GOTO(out_fo, rc = -ENOENT);
1271
1272         if (lu_object_remote(&fo->ofo_obj.do_lu))
1273                 GOTO(out_fo, rc = -EREMOTE);
1274
1275         version = dt_version_get(env, ofd_object_child(fo));
1276         if (version == 0)
1277                 GOTO(out_fo, rc = -EIO);
1278
1279         *(dt_obj_version_t *)data->ioc_inlbuf2 = version;
1280
1281         EXIT;
1282 out_fo:
1283         ofd_object_put(env, fo);
1284 out:
1285         return rc;
1286 }
1287
1288 /**
1289  * Implementation of obd_ops::o_iocontrol.
1290  *
1291  * This is OFD ioctl handling function which is primary interface for
1292  * Lustre tools like lfs, lctl and lfsck.
1293  *
1294  * \param[in]     cmd   ioctl command
1295  * \param[in]     exp   OBD export of OFD
1296  * \param[in]     len   not used
1297  * \param[in,out] karg  buffer with data
1298  * \param[in]     uarg  not used
1299  *
1300  * \retval              0 if successful
1301  * \retval              negative value on error
1302  */
1303 int ofd_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1304                   void *karg, void *uarg)
1305 {
1306         struct lu_env            env;
1307         struct ofd_device       *ofd = ofd_exp(exp);
1308         struct obd_device       *obd = ofd_obd(ofd);
1309         int                      rc;
1310
1311         ENTRY;
1312
1313         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
1314         rc = lu_env_init(&env, LCT_DT_THREAD);
1315         if (rc)
1316                 RETURN(rc);
1317
1318         switch (cmd) {
1319         case OBD_IOC_ABORT_RECOVERY:
1320                 CERROR("%s: aborting recovery\n", obd->obd_name);
1321                 target_stop_recovery_thread(obd);
1322                 break;
1323         case OBD_IOC_SYNC:
1324                 CDEBUG(D_RPCTRACE, "syncing ost %s\n", obd->obd_name);
1325                 rc = dt_sync(&env, ofd->ofd_osd);
1326                 break;
1327         case OBD_IOC_SET_READONLY:
1328                 rc = dt_sync(&env, ofd->ofd_osd);
1329                 if (rc == 0)
1330                         rc = dt_ro(&env, ofd->ofd_osd);
1331                 break;
1332         case OBD_IOC_START_LFSCK: {
1333                 struct obd_ioctl_data *data = karg;
1334                 struct lfsck_start_param lsp;
1335
1336                 if (unlikely(data == NULL)) {
1337                         rc = -EINVAL;
1338                         break;
1339                 }
1340
1341                 lsp.lsp_start = (struct lfsck_start *)(data->ioc_inlbuf1);
1342                 lsp.lsp_index_valid = 0;
1343                 rc = lfsck_start(&env, ofd->ofd_osd, &lsp);
1344                 break;
1345         }
1346         case OBD_IOC_STOP_LFSCK: {
1347                 struct obd_ioctl_data *data = karg;
1348                 struct lfsck_stop      stop;
1349
1350                 stop.ls_status = LS_STOPPED;
1351                 /* Old lfsck utils may pass NULL @stop. */
1352                 if (data->ioc_inlbuf1 == NULL)
1353                         stop.ls_flags = 0;
1354                 else
1355                         stop.ls_flags =
1356                         ((struct lfsck_stop *)(data->ioc_inlbuf1))->ls_flags;
1357
1358                 rc = lfsck_stop(&env, ofd->ofd_osd, &stop);
1359                 break;
1360         }
1361         case OBD_IOC_GET_OBJ_VERSION:
1362                 rc = ofd_ioc_get_obj_version(&env, ofd, karg);
1363                 break;
1364         default:
1365                 CERROR("%s: not supported cmd = %d\n", obd->obd_name, cmd);
1366                 rc = -ENOTTY;
1367         }
1368
1369         lu_env_fini(&env);
1370         RETURN(rc);
1371 }
1372
1373 /**
1374  * Implementation of obd_ops::o_precleanup.
1375  *
1376  * This function stops device activity before shutting it down. It is called
1377  * from a cleanup function upon forceful device cleanup. For OFD there are no
1378  * special actions, it just invokes target_recovery_cleanup().
1379  *
1380  * \param[in] obd       OBD device of OFD
1381  * \param[in] stage     cleanup stage
1382  *
1383  * \retval              0
1384  */
1385 static int ofd_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
1386 {
1387         int rc = 0;
1388
1389         ENTRY;
1390
1391         switch(stage) {
1392         case OBD_CLEANUP_EARLY:
1393                 break;
1394         case OBD_CLEANUP_EXPORTS:
1395                 target_cleanup_recovery(obd);
1396                 break;
1397         }
1398         RETURN(rc);
1399 }
1400
1401 /**
1402  * Implementation of obd_ops::o_ping.
1403  *
1404  * This is OFD-specific part of OBD_PING request handling.
1405  * It controls Filter Modification Data (FMD) expiration each time PING is
1406  * received.
1407  *
1408  * \see  ofd_fmd_expire() and ofd_fmd.c for details
1409  *
1410  * \param[in] env       execution environment
1411  * \param[in] exp       OBD export of client
1412  *
1413  * \retval              0
1414  */
1415 static int ofd_ping(const struct lu_env *env, struct obd_export *exp)
1416 {
1417         ofd_fmd_expire(exp);
1418         return 0;
1419 }
1420
1421 /**
1422  * Implementation of obd_ops::o_health_check.
1423  *
1424  * This function checks the OFD device health - ability to respond on
1425  * incoming requests. There are two health_check methods:
1426  * - get statfs from the OSD. It checks just responsiveness of
1427  *   bottom device
1428  * - do write attempt on bottom device to check it is fully operational and
1429  *   is not stuck. This is expensive method and requires special configuration
1430  *   option --enable-health-write while building Lustre, it is turned off
1431  *   by default.
1432  *
1433  * \param[in] nul       not used
1434  * \param[in] obd       OBD device of OFD
1435  *
1436  * \retval              0 if successful
1437  * \retval              negative value in case of error
1438  */
1439 static int ofd_health_check(const struct lu_env *nul, struct obd_device *obd)
1440 {
1441         struct ofd_device       *ofd = ofd_dev(obd->obd_lu_dev);
1442         struct ofd_thread_info  *info;
1443         struct lu_env            env;
1444 #ifdef USE_HEALTH_CHECK_WRITE
1445         struct thandle          *th;
1446 #endif
1447         int                      rc = 0;
1448
1449         /* obd_proc_read_health pass NULL env, we need real one */
1450         rc = lu_env_init(&env, LCT_DT_THREAD);
1451         if (rc)
1452                 RETURN(rc);
1453
1454         info = ofd_info_init(&env, NULL);
1455         rc = dt_statfs(&env, ofd->ofd_osd, &info->fti_u.osfs);
1456         if (unlikely(rc))
1457                 GOTO(out, rc);
1458
1459         if (info->fti_u.osfs.os_state == OS_STATE_READONLY)
1460                 GOTO(out, rc = -EROFS);
1461
1462 #ifdef USE_HEALTH_CHECK_WRITE
1463         OBD_ALLOC(info->fti_buf.lb_buf, PAGE_CACHE_SIZE);
1464         if (info->fti_buf.lb_buf == NULL)
1465                 GOTO(out, rc = -ENOMEM);
1466
1467         info->fti_buf.lb_len = PAGE_CACHE_SIZE;
1468         info->fti_off = 0;
1469
1470         th = dt_trans_create(&env, ofd->ofd_osd);
1471         if (IS_ERR(th))
1472                 GOTO(out, rc = PTR_ERR(th));
1473
1474         rc = dt_declare_record_write(&env, ofd->ofd_health_check_file,
1475                                      &info->fti_buf, info->fti_off, th);
1476         if (rc == 0) {
1477                 th->th_sync = 1; /* sync IO is needed */
1478                 rc = dt_trans_start_local(&env, ofd->ofd_osd, th);
1479                 if (rc == 0)
1480                         rc = dt_record_write(&env, ofd->ofd_health_check_file,
1481                                              &info->fti_buf, &info->fti_off,
1482                                              th);
1483         }
1484         dt_trans_stop(&env, ofd->ofd_osd, th);
1485
1486         OBD_FREE(info->fti_buf.lb_buf, PAGE_CACHE_SIZE);
1487
1488         CDEBUG(D_INFO, "write 1 page synchronously for checking io rc %d\n",rc);
1489 #endif
1490 out:
1491         lu_env_fini(&env);
1492         return !!rc;
1493 }
1494
1495 struct obd_ops ofd_obd_ops = {
1496         .o_owner                = THIS_MODULE,
1497         .o_connect              = ofd_obd_connect,
1498         .o_reconnect            = ofd_obd_reconnect,
1499         .o_disconnect           = ofd_obd_disconnect,
1500         .o_create               = ofd_echo_create,
1501         .o_statfs               = ofd_statfs,
1502         .o_setattr              = ofd_echo_setattr,
1503         .o_preprw               = ofd_preprw,
1504         .o_commitrw             = ofd_commitrw,
1505         .o_destroy              = ofd_echo_destroy,
1506         .o_init_export          = ofd_init_export,
1507         .o_destroy_export       = ofd_destroy_export,
1508         .o_postrecov            = ofd_obd_postrecov,
1509         .o_getattr              = ofd_echo_getattr,
1510         .o_iocontrol            = ofd_iocontrol,
1511         .o_precleanup           = ofd_precleanup,
1512         .o_ping                 = ofd_ping,
1513         .o_health_check         = ofd_health_check,
1514         .o_set_info_async       = ofd_set_info_async,
1515         .o_get_info             = ofd_get_info,
1516 };