lustre/osp/osp_md_object.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) 2013, 2015, Intel Corporation.
  24  */
  25 /*
  26  * lustre/osp/osp_md_object.c
  27  *
  28  * OST/MDT proxy device (OSP) Metadata methods
  29  *
  30  * This file implements methods for remote MD object, which include
  31  * dt_object_operations, dt_index_operations and dt_body_operations.
  32  *
  33  * If there are multiple MDTs in one filesystem, one operation might
  34  * include modifications in several MDTs. In such cases, clients
  35  * send the RPC to the master MDT, then the operation is decomposed into
  36  * object updates which will be dispatched to OSD or OSP. The local updates
  37  * go to local OSD and the remote updates go to OSP. In OSP, these remote
  38  * object updates will be packed into an update RPC, sent to the remote MDT
  39  * and handled by Object Update Target (OUT).
  40  *
  41  * In DNE phase I, because of missing complete recovery solution, updates
  42  * will be executed in order and synchronously.
  43  *     1. The transaction is created.
  44  *     2. In transaction declare, it collects and packs remote
  45  *        updates (in osp_md_declare_xxx()).
  46  *     3. In transaction start, it sends these remote updates
  47  *        to remote MDTs, which will execute these updates synchronously.
  48  *     4. In transaction execute phase, the local updates will be executed
  49  *        synchronously.
  50  *
  51  * Author: Di Wang <di.wang@intel.com>
  52  */
  53
  54 #define DEBUG_SUBSYSTEM S_MDS
  55
  56 #include <llog_swab.h>
  57 #include <lustre_log.h>
  58 #include "osp_internal.h"
  59
  60 #define OUT_UPDATE_BUFFER_SIZE_ADD      4096
  61 #define OUT_UPDATE_BUFFER_SIZE_MAX      (256 * 4096)  /*  1M update size now */
  62
  63 /**
  64  * Interpreter call for object creation
  65  *
  66  * Object creation interpreter, which will be called after creating
  67  * the remote object to set flags and status.
  68  *
  69  * \param[in] env       execution environment
  70  * \param[in] reply     update reply
  71  * \param[in] req       ptlrpc update request for creating object
  72  * \param[in] obj       object to be created
  73  * \param[in] data      data used in this function.
  74  * \param[in] index     index(position) of create update in the whole
  75  *                      updates
  76  * \param[in] rc        update result on the remote MDT.
  77  *
  78  * \retval              only return 0 for now
  79  */
  80 static int osp_object_create_interpreter(const struct lu_env *env,
  81                                          struct object_update_reply *reply,
  82                                          struct ptlrpc_request *req,
  83                                          struct osp_object *obj,
  84                                          void *data, int index, int rc)
  85 {
  86         if (rc != 0 && rc != -EEXIST) {
  87                 obj->opo_obj.do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
  88                 obj->opo_non_exist = 1;
  89         }
  90
  91         /*
  92          * invalidate opo cache for the object after the object is created, so
  93          * attr_get will try to get attr from remote object.
  94          */
  95         osp_invalidate(env, &obj->opo_obj);
  96
  97         return 0;
  98 }
  99
 100 /**
 101  * Implementation of dt_object_operations::do_declare_create
 102  *
 103  * Create the osp_update_request to track the update for this OSP
 104  * in the transaction.
 105  *
 106  * \param[in] env       execution environment
 107  * \param[in] dt        remote object to be created
 108  * \param[in] attr      attribute of the created object
 109  * \param[in] hint      creation hint
 110  * \param[in] dof       creation format information
 111  * \param[in] th        the transaction handle
 112  *
 113  * \retval              0 if preparation succeeds.
 114  * \retval              negative errno if preparation fails.
 115  */
 116 int osp_md_declare_object_create(const struct lu_env *env,
 117                                  struct dt_object *dt,
 118                                  struct lu_attr *attr,
 119                                  struct dt_allocation_hint *hint,
 120                                  struct dt_object_format *dof,
 121                                  struct thandle *th)
 122 {
 123         return osp_trans_update_request_create(th);
 124 }
 125
 126 struct object_update *
 127 update_buffer_get_update(struct object_update_request *request,
 128                          unsigned int index)
 129 {
 130         void    *ptr;
 131         int     i;
 132
 133         if (index > request->ourq_count)
 134                 return NULL;
 135
 136         ptr = &request->ourq_updates[0];
 137         for (i = 0; i < index; i++)
 138                 ptr += object_update_size(ptr);
 139
 140         return ptr;
 141 }
 142
 143 /**
 144  * Implementation of dt_object_operations::do_create
 145  *
 146  * It adds an OUT_CREATE sub-request into the OUT RPC that will be flushed
 147  * when the transaction stop, and sets necessary flags for created object.
 148  *
 149  * \param[in] env       execution environment
 150  * \param[in] dt        object to be created
 151  * \param[in] attr      attribute of the created object
 152  * \param[in] hint      creation hint
 153  * \param[in] dof       creation format information
 154  * \param[in] th        the transaction handle
 155  *
 156  * \retval              0 if packing creation succeeds.
 157  * \retval              negative errno if packing creation fails.
 158  */
 159 int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
 160                          struct lu_attr *attr, struct dt_allocation_hint *hint,
 161                          struct dt_object_format *dof, struct thandle *th)
 162 {
 163         struct osp_update_request       *update;
 164         struct osp_object               *obj = dt2osp_obj(dt);
 165         int                             rc;
 166
 167         update = thandle_to_osp_update_request(th);
 168         LASSERT(update != NULL);
 169
 170         LASSERT(attr->la_valid & LA_TYPE);
 171         rc = osp_update_rpc_pack(env, create, update, OUT_CREATE,
 172                                  lu_object_fid(&dt->do_lu), attr, hint, dof);
 173         if (rc != 0)
 174                 GOTO(out, rc);
 175
 176         rc = osp_insert_update_callback(env, update, dt2osp_obj(dt), NULL,
 177                                         osp_object_create_interpreter);
 178
 179         if (rc < 0)
 180                 GOTO(out, rc);
 181
 182         dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
 183         dt2osp_obj(dt)->opo_non_exist = 0;
 184
 185         obj->opo_attr = *attr;
 186 out:
 187         return rc;
 188 }
 189
 190 /**
 191  * Implementation of dt_object_operations::do_declare_ref_del
 192  *
 193  * Create the osp_update_request to track the update for this OSP
 194  * in the transaction.
 195  *
 196  * \param[in] env       execution environment
 197  * \param[in] dt        object to decrease the reference count.
 198  * \param[in] th        the transaction handle of refcount decrease.
 199  *
 200  * \retval              0 if preparation succeeds.
 201  * \retval              negative errno if preparation fails.
 202  */
 203 static int osp_md_declare_ref_del(const struct lu_env *env,
 204                                   struct dt_object *dt, struct thandle *th)
 205 {
 206         return osp_trans_update_request_create(th);
 207 }
 208
 209 /**
 210  * Implementation of dt_object_operations::do_ref_del
 211  *
 212  * Add an OUT_REF_DEL sub-request into the OUT RPC that will be
 213  * flushed when the transaction stop.
 214  *
 215  * \param[in] env       execution environment
 216  * \param[in] dt        object to decrease the reference count
 217  * \param[in] th        the transaction handle
 218  *
 219  * \retval              0 if packing ref_del succeeds.
 220  * \retval              negative errno if packing fails.
 221  */
 222 static int osp_md_ref_del(const struct lu_env *env, struct dt_object *dt,
 223                           struct thandle *th)
 224 {
 225         struct osp_update_request       *update;
 226         int                             rc;
 227
 228         update = thandle_to_osp_update_request(th);
 229         LASSERT(update != NULL);
 230
 231         rc = osp_update_rpc_pack(env, ref_del, update, OUT_REF_DEL,
 232                                  lu_object_fid(&dt->do_lu));
 233         return rc;
 234 }
 235
 236 /**
 237  * Implementation of dt_object_operations::do_declare_ref_del
 238  *
 239  * Create the osp_update_request to track the update for this OSP
 240  * in the transaction.
 241  *
 242  * \param[in] env       execution environment
 243  * \param[in] dt        object on which to increase the reference count.
 244  * \param[in] th        the transaction handle.
 245  *
 246  * \retval              0 if preparation succeeds.
 247  * \retval              negative errno if preparation fails.
 248  */
 249 static int osp_md_declare_ref_add(const struct lu_env *env,
 250                                   struct dt_object *dt, struct thandle *th)
 251 {
 252         return osp_trans_update_request_create(th);
 253 }
 254
 255 /**
 256  * Implementation of dt_object_operations::do_ref_add
 257  *
 258  * Add an OUT_REF_ADD sub-request into the OUT RPC that will be flushed
 259  * when the transaction stop.
 260  *
 261  * \param[in] env       execution environment
 262  * \param[in] dt        object on which to increase the reference count
 263  * \param[in] th        the transaction handle
 264  *
 265  * \retval              0 if packing ref_add succeeds.
 266  * \retval              negative errno if packing fails.
 267  */
 268 static int osp_md_ref_add(const struct lu_env *env, struct dt_object *dt,
 269                           struct thandle *th)
 270 {
 271         struct osp_update_request       *update;
 272         int                             rc;
 273
 274         update = thandle_to_osp_update_request(th);
 275         LASSERT(update != NULL);
 276
 277         rc = osp_update_rpc_pack(env, ref_add, update, OUT_REF_ADD,
 278                                  lu_object_fid(&dt->do_lu));
 279         return rc;
 280 }
 281
 282 /**
 283  * Implementation of dt_object_operations::do_ah_init
 284  *
 285  * Initialize the allocation hint for object creation, which is usually called
 286  * before the creation, and these hints (parent and child mode) will be sent to
 287  * the remote Object Update Target (OUT) and used in the object create process,
 288  * same as OSD object creation.
 289  *
 290  * \param[in] env       execution environment
 291  * \param[in] ah        the hint to be initialized
 292  * \param[in] parent    the parent of the object
 293  * \param[in] child     the object to be created
 294  * \param[in] child_mode the mode of the created object
 295  */
 296 static void osp_md_ah_init(const struct lu_env *env,
 297                            struct dt_allocation_hint *ah,
 298                            struct dt_object *parent,
 299                            struct dt_object *child,
 300                            umode_t child_mode)
 301 {
 302         LASSERT(ah);
 303
 304         ah->dah_parent = parent;
 305         ah->dah_mode = child_mode;
 306 }
 307
 308 /**
 309  * Implementation of dt_object_operations::do_declare_attr_get
 310  *
 311  * Create the osp_update_request to track the update for this OSP
 312  * in the transaction.
 313  *
 314  * \param[in] env       execution environment
 315  * \param[in] dt        object on which to set attributes
 316  * \param[in] attr      attributes to be set
 317  * \param[in] th        the transaction handle
 318  *
 319  * \retval              0 if preparation succeeds.
 320  * \retval              negative errno if preparation fails.
 321  */
 322 int osp_md_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
 323                             const struct lu_attr *attr, struct thandle *th)
 324 {
 325         return osp_trans_update_request_create(th);
 326 }
 327
 328 /**
 329  * Implementation of dt_object_operations::do_attr_set
 330  *
 331  * Set attributes to the specified remote object.
 332  *
 333  * Add the OUT_ATTR_SET sub-request into the OUT RPC that will be flushed
 334  * when the transaction stop.
 335  *
 336  * \param[in] env       execution environment
 337  * \param[in] dt        object to set attributes
 338  * \param[in] attr      attributes to be set
 339  * \param[in] th        the transaction handle
 340  *
 341  * \retval              0 if packing attr_set succeeds.
 342  * \retval              negative errno if packing fails.
 343  */
 344 int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
 345                     const struct lu_attr *attr, struct thandle *th)
 346 {
 347         struct osp_update_request       *update;
 348         int                             rc;
 349
 350         update = thandle_to_osp_update_request(th);
 351         LASSERT(update != NULL);
 352
 353         rc = osp_update_rpc_pack(env, attr_set, update, OUT_ATTR_SET,
 354                                  lu_object_fid(&dt->do_lu), attr);
 355         return rc;
 356 }
 357
 358 /**
 359  * Implementation of dt_object_operations::do_read_lock
 360  *
 361  * osp_md_object_{read,write}_lock() will only lock the remote object in the
 362  * local cache, which uses the semaphore (opo_sem) inside the osp_object to
 363  * lock the object. Note: it will not lock the object in the whole cluster,
 364  * which relies on the LDLM lock.
 365  *
 366  * \param[in] env       execution environment
 367  * \param[in] dt        object to be locked
 368  * \param[in] role      lock role from MDD layer, see mdd_object_role().
 369  */
 370 static void osp_md_object_read_lock(const struct lu_env *env,
 371                                     struct dt_object *dt, unsigned role)
 372 {
 373         struct osp_object  *obj = dt2osp_obj(dt);
 374
 375         LASSERT(obj->opo_owner != env);
 376         down_read_nested(&obj->opo_sem, role);
 377
 378         LASSERT(obj->opo_owner == NULL);
 379 }
 380
 381 /**
 382  * Implementation of dt_object_operations::do_write_lock
 383  *
 384  * Lock the remote object in write mode.
 385  *
 386  * \param[in] env       execution environment
 387  * \param[in] dt        object to be locked
 388  * \param[in] role      lock role from MDD layer, see mdd_object_role().
 389  */
 390 static void osp_md_object_write_lock(const struct lu_env *env,
 391                                      struct dt_object *dt, unsigned role)
 392 {
 393         struct osp_object *obj = dt2osp_obj(dt);
 394
 395         down_write_nested(&obj->opo_sem, role);
 396
 397         LASSERT(obj->opo_owner == NULL);
 398         obj->opo_owner = env;
 399 }
 400
 401 /**
 402  * Implementation of dt_object_operations::do_read_unlock
 403  *
 404  * Unlock the read lock of remote object.
 405  *
 406  * \param[in] env       execution environment
 407  * \param[in] dt        object to be unlocked
 408  */
 409 static void osp_md_object_read_unlock(const struct lu_env *env,
 410                                       struct dt_object *dt)
 411 {
 412         struct osp_object *obj = dt2osp_obj(dt);
 413
 414         up_read(&obj->opo_sem);
 415 }
 416
 417 /**
 418  * Implementation of dt_object_operations::do_write_unlock
 419  *
 420  * Unlock the write lock of remote object.
 421  *
 422  * \param[in] env       execution environment
 423  * \param[in] dt        object to be unlocked
 424  */
 425 static void osp_md_object_write_unlock(const struct lu_env *env,
 426                                        struct dt_object *dt)
 427 {
 428         struct osp_object *obj = dt2osp_obj(dt);
 429
 430         LASSERT(obj->opo_owner == env);
 431         obj->opo_owner = NULL;
 432         up_write(&obj->opo_sem);
 433 }
 434
 435 /**
 436  * Implementation of dt_object_operations::do_write_locked
 437  *
 438  * Test if the object is locked in write mode.
 439  *
 440  * \param[in] env       execution environment
 441  * \param[in] dt        object to be tested
 442  */
 443 static int osp_md_object_write_locked(const struct lu_env *env,
 444                                       struct dt_object *dt)
 445 {
 446         struct osp_object *obj = dt2osp_obj(dt);
 447
 448         return obj->opo_owner == env;
 449 }
 450
 451 /**
 452  * Implementation of dt_index_operations::dio_lookup
 453  *
 454  * Look up record by key under a remote index object. It packs lookup update
 455  * into RPC, sends to the remote OUT and waits for the lookup result.
 456  *
 457  * \param[in] env       execution environment
 458  * \param[in] dt        index object to lookup
 459  * \param[out] rec      record in which to return lookup result
 460  * \param[in] key       key of index which will be looked up
 461  *
 462  * \retval              1 if the lookup succeeds.
 463  * \retval              negative errno if the lookup fails.
 464  */
 465 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
 466                                struct dt_rec *rec, const struct dt_key *key)
 467 {
 468         struct lu_buf           *lbuf   = &osp_env_info(env)->osi_lb2;
 469         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 470         struct dt_device        *dt_dev = &osp->opd_dt_dev;
 471         struct osp_update_request   *update;
 472         struct object_update_reply *reply;
 473         struct ptlrpc_request      *req = NULL;
 474         struct lu_fid              *fid;
 475         int                        rc;
 476         ENTRY;
 477
 478         /* Because it needs send the update buffer right away,
 479          * just create an update buffer, instead of attaching the
 480          * update_remote list of the thandle.
 481          */
 482         update = osp_update_request_create(dt_dev);
 483         if (IS_ERR(update))
 484                 RETURN(PTR_ERR(update));
 485
 486         rc = osp_update_rpc_pack(env, index_lookup, update, OUT_INDEX_LOOKUP,
 487                                  lu_object_fid(&dt->do_lu), rec, key);
 488         if (rc != 0) {
 489                 CERROR("%s: Insert update error: rc = %d\n",
 490                        dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
 491                 GOTO(out, rc);
 492         }
 493
 494         rc = osp_remote_sync(env, osp, update, &req);
 495         if (rc < 0)
 496                 GOTO(out, rc);
 497
 498         reply = req_capsule_server_sized_get(&req->rq_pill,
 499                                              &RMF_OUT_UPDATE_REPLY,
 500                                              OUT_UPDATE_REPLY_SIZE);
 501         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
 502                 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
 503                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 504                        reply->ourp_magic, UPDATE_REPLY_MAGIC, -EPROTO);
 505                 GOTO(out, rc = -EPROTO);
 506         }
 507
 508         rc = object_update_result_data_get(reply, lbuf, 0);
 509         if (rc < 0)
 510                 GOTO(out, rc);
 511
 512         if (lbuf->lb_len != sizeof(*fid)) {
 513                 CERROR("%s: lookup "DFID" %s wrong size %d\n",
 514                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 515                        PFID(lu_object_fid(&dt->do_lu)), (char *)key,
 516                        (int)lbuf->lb_len);
 517                 GOTO(out, rc = -EINVAL);
 518         }
 519
 520         fid = lbuf->lb_buf;
 521         if (ptlrpc_rep_need_swab(req))
 522                 lustre_swab_lu_fid(fid);
 523         if (!fid_is_sane(fid)) {
 524                 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
 525                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 526                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
 527                 GOTO(out, rc = -EINVAL);
 528         }
 529
 530         memcpy(rec, fid, sizeof(*fid));
 531
 532         GOTO(out, rc = 1);
 533
 534 out:
 535         if (req != NULL)
 536                 ptlrpc_req_finished(req);
 537
 538         osp_update_request_destroy(update);
 539
 540         return rc;
 541 }
 542
 543 /**
 544  * Implementation of dt_index_operations::dio_declare_insert
 545  *
 546  * Create the osp_update_request to track the update for this OSP
 547  * in the transaction.
 548  *
 549  * \param[in] env       execution environment
 550  * \param[in] dt        object for which to insert index
 551  * \param[in] rec       record of the index which will be inserted
 552  * \param[in] key       key of the index which will be inserted
 553  * \param[in] th        the transaction handle
 554  *
 555  * \retval              0 if preparation succeeds.
 556  * \retval              negative errno if preparation fails.
 557  */
 558 static int osp_md_declare_index_insert(const struct lu_env *env,
 559                                        struct dt_object *dt,
 560                                        const struct dt_rec *rec,
 561                                        const struct dt_key *key,
 562                                        struct thandle *th)
 563 {
 564         return osp_trans_update_request_create(th);
 565 }
 566
 567 /**
 568  * Implementation of dt_index_operations::dio_insert
 569  *
 570  * Add an OUT_INDEX_INSERT sub-request into the OUT RPC that will
 571  * be flushed when the transaction stop.
 572  *
 573  * \param[in] env       execution environment
 574  * \param[in] dt        object for which to insert index
 575  * \param[in] rec       record of the index to be inserted
 576  * \param[in] key       key of the index to be inserted
 577  * \param[in] th        the transaction handle
 578  * \param[in] ignore_quota quota enforcement for insert
 579  *
 580  * \retval              0 if packing index insert succeeds.
 581  * \retval              negative errno if packing fails.
 582  */
 583 static int osp_md_index_insert(const struct lu_env *env,
 584                                struct dt_object *dt,
 585                                const struct dt_rec *rec,
 586                                const struct dt_key *key,
 587                                struct thandle *th,
 588                                int ignore_quota)
 589 {
 590         struct osp_update_request *update;
 591         int                      rc;
 592
 593         update = thandle_to_osp_update_request(th);
 594         LASSERT(update != NULL);
 595
 596         rc = osp_update_rpc_pack(env, index_insert, update, OUT_INDEX_INSERT,
 597                                  lu_object_fid(&dt->do_lu), rec, key);
 598         return rc;
 599 }
 600
 601 /**
 602  * Implementation of dt_index_operations::dio_declare_delete
 603  *
 604  * Create the osp_update_request to track the update for this OSP
 605  * in the transaction.
 606  *
 607  * \param[in] env       execution environment
 608  * \param[in] dt        object for which to delete index
 609  * \param[in] key       key of the index
 610  * \param[in] th        the transaction handle
 611  *
 612  * \retval              0 if preparation succeeds.
 613  * \retval              negative errno if preparation fails.
 614  */
 615 static int osp_md_declare_index_delete(const struct lu_env *env,
 616                                        struct dt_object *dt,
 617                                        const struct dt_key *key,
 618                                        struct thandle *th)
 619 {
 620         return osp_trans_update_request_create(th);
 621 }
 622
 623 /**
 624  * Implementation of dt_index_operations::dio_delete
 625  *
 626  * Add an OUT_INDEX_DELETE sub-request into the OUT RPC that will
 627  * be flushed when the transaction stop.
 628  *
 629  * \param[in] env       execution environment
 630  * \param[in] dt        object for which to delete index
 631  * \param[in] key       key of the index which will be deleted
 632  * \param[in] th        the transaction handle
 633  *
 634  * \retval              0 if packing index delete succeeds.
 635  * \retval              negative errno if packing fails.
 636  */
 637 static int osp_md_index_delete(const struct lu_env *env,
 638                                struct dt_object *dt,
 639                                const struct dt_key *key,
 640                                struct thandle *th)
 641 {
 642         struct osp_update_request *update;
 643         int                      rc;
 644
 645         update = thandle_to_osp_update_request(th);
 646         LASSERT(update != NULL);
 647
 648         rc = osp_update_rpc_pack(env, index_delete, update, OUT_INDEX_DELETE,
 649                                  lu_object_fid(&dt->do_lu), key);
 650
 651         return rc;
 652 }
 653
 654 /**
 655  * Implementation of dt_index_operations::dio_it.next
 656  *
 657  * Advance the pointer of the iterator to the next entry. It shares a similar
 658  * internal implementation with osp_orphan_it_next(), which is being used for
 659  * remote orphan index object. This method will be used for remote directory.
 660  *
 661  * \param[in] env       execution environment
 662  * \param[in] di        iterator of this iteration
 663  *
 664  * \retval              0 if the pointer is advanced successfully.
 665  * \retval              1 if it reaches to the end of the index object.
 666  * \retval              negative errno if the pointer cannot be advanced.
 667  */
 668 static int osp_md_index_it_next(const struct lu_env *env, struct dt_it *di)
 669 {
 670         struct osp_it           *it = (struct osp_it *)di;
 671         struct lu_idxpage       *idxpage;
 672         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 673         int                     rc;
 674         ENTRY;
 675
 676 again:
 677         idxpage = it->ooi_cur_idxpage;
 678         if (idxpage != NULL) {
 679                 if (idxpage->lip_nr == 0)
 680                         RETURN(1);
 681
 682                 it->ooi_pos_ent++;
 683                 if (ent == NULL) {
 684                         it->ooi_ent =
 685                               (struct lu_dirent *)idxpage->lip_entries;
 686                         RETURN(0);
 687                 } else if (le16_to_cpu(ent->lde_reclen) != 0 &&
 688                            it->ooi_pos_ent < idxpage->lip_nr) {
 689                         ent = (struct lu_dirent *)(((char *)ent) +
 690                                         le16_to_cpu(ent->lde_reclen));
 691                         it->ooi_ent = ent;
 692                         RETURN(0);
 693                 } else {
 694                         it->ooi_ent = NULL;
 695                 }
 696         }
 697
 698         rc = osp_it_next_page(env, di);
 699         if (rc == 0)
 700                 goto again;
 701
 702         RETURN(rc);
 703 }
 704
 705 /**
 706  * Implementation of dt_index_operations::dio_it.key
 707  *
 708  * Get the key at current iterator poisiton. These iteration methods
 709  * (dio_it) will only be used for iterating the remote directory, so
 710  * the key is the name of the directory entry.
 711  *
 712  * \param[in] env       execution environment
 713  * \param[in] di        iterator of this iteration
 714  *
 715  * \retval              name of the current entry
 716  */
 717 static struct dt_key *osp_it_key(const struct lu_env *env,
 718                                  const struct dt_it *di)
 719 {
 720         struct osp_it           *it = (struct osp_it *)di;
 721         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 722
 723         return (struct dt_key *)ent->lde_name;
 724 }
 725
 726 /**
 727  * Implementation of dt_index_operations::dio_it.key_size
 728  *
 729  * Get the key size at current iterator poisiton. These iteration methods
 730  * (dio_it) will only be used for iterating the remote directory, so the key
 731  * size is the name size of the directory entry.
 732  *
 733  * \param[in] env       execution environment
 734  * \param[in] di        iterator of this iteration
 735  *
 736  * \retval              name size of the current entry
 737  */
 738
 739 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
 740 {
 741         struct osp_it           *it = (struct osp_it *)di;
 742         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 743
 744         return (int)le16_to_cpu(ent->lde_namelen);
 745 }
 746
 747 /**
 748  * Implementation of dt_index_operations::dio_it.rec
 749  *
 750  * Get the record at current iterator position. These iteration methods
 751  * (dio_it) will only be used for iterating the remote directory, so it
 752  * uses lu_dirent_calc_size() to calculate the record size.
 753  *
 754  * \param[in] env       execution environment
 755  * \param[in] di        iterator of this iteration
 756  * \param[out] rec      the record to be returned
 757  * \param[in] attr      attributes of the index object, so it knows
 758  *                      how to pack the entry.
 759  *
 760  * \retval              only return 0 for now
 761  */
 762 static int osp_md_index_it_rec(const struct lu_env *env, const struct dt_it *di,
 763                                struct dt_rec *rec, __u32 attr)
 764 {
 765         struct osp_it           *it = (struct osp_it *)di;
 766         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 767         size_t                  reclen;
 768
 769         reclen = lu_dirent_calc_size(le16_to_cpu(ent->lde_namelen), attr);
 770         memcpy(rec, ent, reclen);
 771         return 0;
 772 }
 773
 774 /**
 775  * Implementation of dt_index_operations::dio_it.load
 776  *
 777  * Locate the iteration cursor to the specified position (cookie).
 778  *
 779  * \param[in] env       pointer to the thread context
 780  * \param[in] di        pointer to the iteration structure
 781  * \param[in] hash      the specified position
 782  *
 783  * \retval              positive number for locating to the exactly position
 784  *                      or the next
 785  * \retval              0 for arriving at the end of the iteration
 786  * \retval              negative error number on failure
 787  */
 788 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
 789                        __u64 hash)
 790 {
 791         struct osp_it   *it     = (struct osp_it *)di;
 792         int              rc;
 793
 794         it->ooi_next = hash;
 795         rc = osp_md_index_it_next(env, (struct dt_it *)di);
 796         if (rc == 1)
 797                 return 0;
 798
 799         if (rc == 0)
 800                 return 1;
 801
 802         return rc;
 803 }
 804
 805 const struct dt_index_operations osp_md_index_ops = {
 806         .dio_lookup         = osp_md_index_lookup,
 807         .dio_declare_insert = osp_md_declare_index_insert,
 808         .dio_insert         = osp_md_index_insert,
 809         .dio_declare_delete = osp_md_declare_index_delete,
 810         .dio_delete         = osp_md_index_delete,
 811         .dio_it     = {
 812                 .init     = osp_it_init,
 813                 .fini     = osp_it_fini,
 814                 .get      = osp_it_get,
 815                 .put      = osp_it_put,
 816                 .next     = osp_md_index_it_next,
 817                 .key      = osp_it_key,
 818                 .key_size = osp_it_key_size,
 819                 .rec      = osp_md_index_it_rec,
 820                 .store    = osp_it_store,
 821                 .load     = osp_it_load,
 822                 .key_rec  = osp_it_key_rec,
 823         }
 824 };
 825
 826 /**
 827  * Implementation of dt_object_operations::do_index_try
 828  *
 829  * Try to initialize the index API pointer for the given object. This
 830  * is the entry point of the index API, i.e. we must call this method
 831  * to initialize the index object before calling other index methods.
 832  *
 833  * \param[in] env       execution environment
 834  * \param[in] dt        index object to be initialized
 835  * \param[in] feat      the index feature of the object
 836  *
 837  * \retval              0 if the initialization succeeds.
 838  * \retval              negative errno if the initialization fails.
 839  */
 840 static int osp_md_index_try(const struct lu_env *env,
 841                             struct dt_object *dt,
 842                             const struct dt_index_features *feat)
 843 {
 844         dt->do_index_ops = &osp_md_index_ops;
 845         return 0;
 846 }
 847
 848 /**
 849  * Implementation of dt_object_operations::do_object_lock
 850  *
 851  * Enqueue a lock (by ldlm_cli_enqueue()) of remote object on the remote MDT,
 852  * which will lock the object in the global namespace. And because the
 853  * cross-MDT locks are relatively rare compared with normal local MDT operation,
 854  * let's release it right away, instead of putting it into the LRU list.
 855  *
 856  * \param[in] env       execution environment
 857  * \param[in] dt        object to be locked
 858  * \param[out] lh       lock handle
 859  * \param[in] einfo     enqueue information
 860  * \param[in] policy    lock policy
 861  *
 862  * \retval              ELDLM_OK if locking the object succeeds.
 863  * \retval              negative errno if locking fails.
 864  */
 865 static int osp_md_object_lock(const struct lu_env *env,
 866                               struct dt_object *dt,
 867                               struct lustre_handle *lh,
 868                               struct ldlm_enqueue_info *einfo,
 869                               union ldlm_policy_data *policy)
 870 {
 871         struct ldlm_res_id      *res_id;
 872         struct dt_device        *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 873         struct osp_device       *osp = dt2osp_dev(dt_dev);
 874         struct lu_device        *top_device;
 875         struct ptlrpc_request   *req;
 876         int                     rc = 0;
 877         __u64                   flags = LDLM_FL_NO_LRU;
 878
 879         res_id = einfo->ei_res_id;
 880         LASSERT(res_id != NULL);
 881
 882         if (einfo->ei_nonblock)
 883                 flags |= LDLM_FL_BLOCK_NOWAIT;
 884         if (einfo->ei_mode & (LCK_EX | LCK_PW))
 885                 flags |= LDLM_FL_COS_INCOMPAT;
 886
 887         req = ldlm_enqueue_pack(osp->opd_exp, 0);
 888         if (IS_ERR(req))
 889                 RETURN(PTR_ERR(req));
 890
 891         /* During recovery, it needs to let OSP send enqueue
 892          * without checking recoverying status, in case the
 893          * other target is being recovered at the same time,
 894          * and if we wait here for the import to be recovered,
 895          * it might cause deadlock */
 896         top_device = dt_dev->dd_lu_dev.ld_site->ls_top_dev;
 897         if (top_device->ld_obd->obd_recovering)
 898                 req->rq_allow_replay = 1;
 899
 900         rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
 901                               (const union ldlm_policy_data *)policy,
 902                               &flags, NULL, 0, LVB_T_NONE, lh, 0);
 903
 904         ptlrpc_req_finished(req);
 905
 906         return rc == ELDLM_OK ? 0 : -EIO;
 907 }
 908
 909 /**
 910  * Implementation of dt_object_operations::do_object_unlock
 911  *
 912  * Cancel a lock of a remote object.
 913  *
 914  * \param[in] env       execution environment
 915  * \param[in] dt        object to be unlocked
 916  * \param[in] einfo     lock enqueue information
 917  * \param[in] policy    lock policy
 918  *
 919  * \retval              Only return 0 for now.
 920  */
 921 static int osp_md_object_unlock(const struct lu_env *env,
 922                                 struct dt_object *dt,
 923                                 struct ldlm_enqueue_info *einfo,
 924                                 union ldlm_policy_data *policy)
 925 {
 926         struct lustre_handle    *lockh = einfo->ei_cbdata;
 927
 928         /* unlock finally */
 929         ldlm_lock_decref(lockh, einfo->ei_mode);
 930
 931         return 0;
 932 }
 933
 934 /**
 935  * Implement OSP layer dt_object_operations::do_declare_destroy() interface.
 936  *
 937  * Create the dt_update_request to track the update for this OSP
 938  * in the transaction.
 939  *
 940  * \param[in] env       pointer to the thread context
 941  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
 942  * \param[in] th        pointer to the transaction handler
 943  *
 944  * \retval              0 for success
 945  * \retval              negative error number on failure
 946  */
 947 int osp_md_declare_object_destroy(const struct lu_env *env,
 948                                struct dt_object *dt, struct thandle *th)
 949 {
 950         return osp_trans_update_request_create(th);
 951 }
 952
 953 /**
 954  * Implement OSP layer dt_object_operations::do_destroy() interface.
 955  *
 956  * Pack the destroy update into the RPC buffer, which will be sent
 957  * to the remote MDT during transaction stop.
 958  *
 959  * It also marks the object as non-cached.
 960  *
 961  * \param[in] env       pointer to the thread context
 962  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
 963  * \param[in] th        pointer to the transaction handler
 964  *
 965  * \retval              0 for success
 966  * \retval              negative error number on failure
 967  */
 968 int osp_md_object_destroy(const struct lu_env *env, struct dt_object *dt,
 969                           struct thandle *th)
 970 {
 971         struct osp_object               *o = dt2osp_obj(dt);
 972         struct osp_device               *osp = lu2osp_dev(dt->do_lu.lo_dev);
 973         struct osp_update_request       *update;
 974         int                             rc = 0;
 975
 976         ENTRY;
 977         o->opo_non_exist = 1;
 978
 979         LASSERT(osp->opd_connect_mdt);
 980         update = thandle_to_osp_update_request(th);
 981         LASSERT(update != NULL);
 982
 983         rc = osp_update_rpc_pack(env, object_destroy, update, OUT_DESTROY,
 984                                  lu_object_fid(&dt->do_lu));
 985         if (rc != 0)
 986                 RETURN(rc);
 987
 988         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
 989         rc = osp_insert_update_callback(env, update, dt2osp_obj(dt), NULL,
 990                                         NULL);
 991
 992         RETURN(rc);
 993 }
 994
 995 struct dt_object_operations osp_md_obj_ops = {
 996         .do_read_lock         = osp_md_object_read_lock,
 997         .do_write_lock        = osp_md_object_write_lock,
 998         .do_read_unlock       = osp_md_object_read_unlock,
 999         .do_write_unlock      = osp_md_object_write_unlock,
1000         .do_write_locked      = osp_md_object_write_locked,
1001         .do_declare_create    = osp_md_declare_object_create,
1002         .do_create            = osp_md_object_create,
1003         .do_declare_ref_add   = osp_md_declare_ref_add,
1004         .do_ref_add           = osp_md_ref_add,
1005         .do_declare_ref_del   = osp_md_declare_ref_del,
1006         .do_ref_del           = osp_md_ref_del,
1007         .do_declare_destroy   = osp_md_declare_object_destroy,
1008         .do_destroy           = osp_md_object_destroy,
1009         .do_ah_init           = osp_md_ah_init,
1010         .do_attr_get          = osp_attr_get,
1011         .do_declare_attr_set  = osp_md_declare_attr_set,
1012         .do_attr_set          = osp_md_attr_set,
1013         .do_xattr_get         = osp_xattr_get,
1014         .do_declare_xattr_set = osp_declare_xattr_set,
1015         .do_xattr_set         = osp_xattr_set,
1016         .do_declare_xattr_del = osp_declare_xattr_del,
1017         .do_xattr_del         = osp_xattr_del,
1018         .do_index_try         = osp_md_index_try,
1019         .do_object_lock       = osp_md_object_lock,
1020         .do_object_unlock     = osp_md_object_unlock,
1021         .do_invalidate        = osp_invalidate,
1022 };
1023
1024 /**
1025  * Implementation of dt_body_operations::dbo_declare_write
1026  *
1027  * Create the osp_update_request to track the update for this OSP
1028  * in the transaction.
1029   *
1030  * \param[in] env       execution environment
1031  * \param[in] dt        object to be written
1032  * \param[in] buf       buffer to write which includes an embedded size field
1033  * \param[in] pos       offet in the object to start writing at
1034  * \param[in] th        transaction handle
1035  *
1036  * \retval              0 if preparation succeeds.
1037  * \retval              negative errno if preparation fails.
1038  */
1039 static ssize_t osp_md_declare_write(const struct lu_env *env,
1040                                     struct dt_object *dt,
1041                                     const struct lu_buf *buf,
1042                                     loff_t pos, struct thandle *th)
1043 {
1044         struct osp_device *osp = dt2osp_dev(th->th_dev);
1045         int rc;
1046
1047         rc = osp_trans_update_request_create(th);
1048         if (rc != 0)
1049                 return rc;
1050
1051         if (osp->opd_update == NULL)
1052                 return 0;
1053
1054         if (dt2osp_obj(dt)->opo_stale)
1055                 return -ESTALE;
1056
1057         return 0;
1058 }
1059
1060 /**
1061  * Implementation of dt_body_operations::dbo_write
1062  *
1063  * Pack the write object update into the RPC buffer, which will be sent
1064  * to the remote MDT during transaction stop.
1065  *
1066  * \param[in] env       execution environment
1067  * \param[in] dt        object to be written
1068  * \param[in] buf       buffer to write which includes an embedded size field
1069  * \param[in] pos       offet in the object to start writing at
1070  * \param[in] th        transaction handle
1071  * \param[in] ignore_quota quota enforcement for this write
1072  *
1073  * \retval              the buffer size in bytes if packing succeeds.
1074  * \retval              negative errno if packing fails.
1075  */
1076 static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
1077                             const struct lu_buf *buf, loff_t *pos,
1078                             struct thandle *th, int ignore_quota)
1079 {
1080         struct osp_object         *obj = dt2osp_obj(dt);
1081         struct osp_update_request  *update;
1082         struct osp_thandle        *oth = thandle_to_osp_thandle(th);
1083         ssize_t                   rc;
1084         ENTRY;
1085
1086         update = thandle_to_osp_update_request(th);
1087         LASSERT(update != NULL);
1088
1089         CDEBUG(D_INFO, "write "DFID" offset = "LPU64" length = %zu\n",
1090                PFID(lu_object_fid(&dt->do_lu)), *pos, buf->lb_len);
1091
1092         rc = osp_update_rpc_pack(env, write, update, OUT_WRITE,
1093                                  lu_object_fid(&dt->do_lu), buf, *pos);
1094         if (rc < 0)
1095                 RETURN(rc);
1096
1097         rc = osp_check_and_set_rpc_version(oth, obj);
1098         if (rc < 0)
1099                 RETURN(rc);
1100
1101         /* XXX: how about the write error happened later? */
1102         *pos += buf->lb_len;
1103
1104         if (obj->opo_attr.la_valid & LA_SIZE && obj->opo_attr.la_size < *pos)
1105                 obj->opo_attr.la_size = *pos;
1106
1107         RETURN(buf->lb_len);
1108 }
1109
1110 static ssize_t osp_md_read(const struct lu_env *env, struct dt_object *dt,
1111                            struct lu_buf *rbuf, loff_t *pos)
1112 {
1113         struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
1114         struct dt_device *dt_dev        = &osp->opd_dt_dev;
1115         struct lu_buf *lbuf = &osp_env_info(env)->osi_lb2;
1116         char *ptr = rbuf->lb_buf;
1117         struct osp_update_request *update;
1118         struct ptlrpc_request *req = NULL;
1119         struct out_read_reply *orr;
1120         struct ptlrpc_bulk_desc *desc;
1121         struct object_update_reply *reply;
1122         __u32 left_size;
1123         int nbufs;
1124         int i;
1125         int rc;
1126         ENTRY;
1127
1128         /* Because it needs send the update buffer right away,
1129          * just create an update buffer, instead of attaching the
1130          * update_remote list of the thandle.  */
1131         update = osp_update_request_create(dt_dev);
1132         if (IS_ERR(update))
1133                 RETURN(PTR_ERR(update));
1134
1135         rc = osp_update_rpc_pack(env, read, update, OUT_READ,
1136                                  lu_object_fid(&dt->do_lu),
1137                                  rbuf->lb_len, *pos);
1138         if (rc != 0) {
1139                 CERROR("%s: cannot insert update: rc = %d\n",
1140                        dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
1141                 GOTO(out_update, rc);
1142         }
1143
1144         CDEBUG(D_INFO, "%s "DFID" read offset %llu size %zu\n",
1145                dt_dev->dd_lu_dev.ld_obd->obd_name,
1146                PFID(lu_object_fid(&dt->do_lu)), *pos, rbuf->lb_len);
1147         rc = osp_prep_update_req(env, osp->opd_obd->u.cli.cl_import, update,
1148                                  &req);
1149         if (rc != 0)
1150                 GOTO(out_update, rc);
1151
1152         nbufs = (rbuf->lb_len + OUT_BULK_BUFFER_SIZE - 1) /
1153                                         OUT_BULK_BUFFER_SIZE;
1154         /* allocate bulk descriptor */
1155         desc = ptlrpc_prep_bulk_imp(req, nbufs, 1,
1156                                     PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KVEC,
1157                                     MDS_BULK_PORTAL, &ptlrpc_bulk_kvec_ops);
1158         if (desc == NULL)
1159                 GOTO(out, rc = -ENOMEM);
1160
1161         /* split the buffer into small chunk size */
1162         left_size = rbuf->lb_len;
1163         for (i = 0; i < nbufs; i++) {
1164                 int read_size;
1165
1166                 read_size = left_size > OUT_BULK_BUFFER_SIZE ?
1167                                 OUT_BULK_BUFFER_SIZE : left_size;
1168                 desc->bd_frag_ops->add_iov_frag(desc, ptr, read_size);
1169
1170                 ptr += read_size;
1171         }
1172
1173         /* This will only be called with read-only update, and these updates
1174          * might be used to retrieve update log during recovery process, so
1175          * it will be allowed to send during recovery process */
1176         req->rq_allow_replay = 1;
1177         req->rq_bulk_read = 1;
1178         /* send request to master and wait for RPC to complete */
1179         rc = ptlrpc_queue_wait(req);
1180         if (rc != 0)
1181                 GOTO(out, rc);
1182
1183         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1184                                           req->rq_bulk->bd_nob_transferred);
1185         if (rc < 0)
1186                 GOTO(out, rc);
1187
1188         reply = req_capsule_server_sized_get(&req->rq_pill,
1189                                              &RMF_OUT_UPDATE_REPLY,
1190                                              OUT_UPDATE_REPLY_SIZE);
1191
1192         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
1193                 CERROR("%s: invalid update reply magic %x expected %x:"
1194                        " rc = %d\n", dt_dev->dd_lu_dev.ld_obd->obd_name,
1195                        reply->ourp_magic, UPDATE_REPLY_MAGIC, -EPROTO);
1196                 GOTO(out, rc = -EPROTO);
1197         }
1198
1199         rc = object_update_result_data_get(reply, lbuf, 0);
1200         if (rc < 0)
1201                 GOTO(out, rc);
1202
1203         if (lbuf->lb_len < sizeof(*orr))
1204                 GOTO(out, rc = -EPROTO);
1205
1206         orr = lbuf->lb_buf;
1207         orr_le_to_cpu(orr, orr);
1208         rc = orr->orr_size;
1209         *pos = orr->orr_offset;
1210 out:
1211         ptlrpc_req_finished(req);
1212
1213 out_update:
1214         osp_update_request_destroy(update);
1215
1216         RETURN(rc);
1217 }
1218
1219 /* These body operation will be used to write symlinks during migration etc */
1220 struct dt_body_operations osp_md_body_ops = {
1221         .dbo_declare_write      = osp_md_declare_write,
1222         .dbo_write              = osp_md_write,
1223         .dbo_read               = osp_md_read,
1224 };