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, 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 <lustre_log.h>
  57 #include "osp_internal.h"
  58
  59 static const char dot[] = ".";
  60 static const char dotdot[] = "..";
  61
  62 /**
  63  * Implementation of dt_object_operations::do_declare_create
  64  *
  65  * Insert object create update into the RPC, which will be sent during
  66  * transaction start. Note: if the object has already been created,
  67  * we must add object destroy updates ahead of create updates, so it will
  68  * destroy then recreate the object.
  69  *
  70  * \param[in] env       execution environment
  71  * \param[in] dt        remote object to be created
  72  * \param[in] attr      attribute of the created object
  73  * \param[in] hint      creation hint
  74  * \param[in] dof       creation format information
  75  * \param[in] th        the transaction handle
  76  *
  77  * \retval              0 if the insertion succeeds.
  78  * \retval              negative errno if the insertion fails.
  79  */
  80 int osp_md_declare_object_create(const struct lu_env *env,
  81                                  struct dt_object *dt,
  82                                  struct lu_attr *attr,
  83                                  struct dt_allocation_hint *hint,
  84                                  struct dt_object_format *dof,
  85                                  struct thandle *th)
  86 {
  87         struct dt_update_request        *update;
  88         int                             rc;
  89
  90         update = dt_update_request_find_or_create(th, dt);
  91         if (IS_ERR(update)) {
  92                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
  93                        dt->do_lu.lo_dev->ld_obd->obd_name,
  94                        (int)PTR_ERR(update));
  95                 return PTR_ERR(update);
  96         }
  97
  98         if (lu_object_exists(&dt->do_lu)) {
  99                 /* If the object already exists, we needs to destroy
 100                  * this orphan object first.
 101                  *
 102                  * The scenario might happen in this case
 103                  *
 104                  * 1. client send remote create to MDT0.
 105                  * 2. MDT0 send create update to MDT1.
 106                  * 3. MDT1 finished create synchronously.
 107                  * 4. MDT0 failed and reboot.
 108                  * 5. client resend remote create to MDT0.
 109                  * 6. MDT0 tries to resend create update to MDT1,
 110                  *    but find the object already exists
 111                  */
 112                 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
 113                        dt->do_lu.lo_dev->ld_obd->obd_name,
 114                        PFID(lu_object_fid(&dt->do_lu)));
 115
 116                 rc = out_ref_del_pack(env, &update->dur_buf,
 117                                       lu_object_fid(&dt->do_lu),
 118                                       update->dur_batchid);
 119                 if (rc != 0)
 120                         GOTO(out, rc);
 121
 122                 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
 123                         /* decrease for ".." */
 124                         rc = out_ref_del_pack(env, &update->dur_buf,
 125                                               lu_object_fid(&dt->do_lu),
 126                                               update->dur_batchid);
 127                         if (rc != 0)
 128                                 GOTO(out, rc);
 129                 }
 130
 131                 rc = out_object_destroy_pack(env, &update->dur_buf,
 132                                              lu_object_fid(&dt->do_lu),
 133                                              update->dur_batchid);
 134                 if (rc != 0)
 135                         GOTO(out, rc);
 136
 137                 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
 138                 /* Increase batchid to add this orphan object deletion
 139                  * to separate transaction */
 140                 update_inc_batchid(update);
 141         }
 142
 143         rc = out_create_pack(env, &update->dur_buf,
 144                              lu_object_fid(&dt->do_lu), attr, hint, dof,
 145                              update->dur_batchid);
 146         if (rc != 0)
 147                 GOTO(out, rc);
 148 out:
 149         if (rc)
 150                 CERROR("%s: Insert update error: rc = %d\n",
 151                        dt->do_lu.lo_dev->ld_obd->obd_name, rc);
 152
 153         return rc;
 154 }
 155
 156 /**
 157  * Implementation of dt_object_operations::do_create
 158  *
 159  * It sets necessary flags for created object. In DNE phase I,
 160  * remote updates are actually executed during transaction start,
 161  * i.e. the object has already been created when calling this method.
 162  *
 163  * \param[in] env       execution environment
 164  * \param[in] dt        object to be created
 165  * \param[in] attr      attribute of the created object
 166  * \param[in] hint      creation hint
 167  * \param[in] dof       creation format information
 168  * \param[in] th        the transaction handle
 169  *
 170  * \retval              only return 0 for now
 171  */
 172 int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
 173                          struct lu_attr *attr, struct dt_allocation_hint *hint,
 174                          struct dt_object_format *dof, struct thandle *th)
 175 {
 176         CDEBUG(D_INFO, "create object "DFID"\n",
 177                PFID(&dt->do_lu.lo_header->loh_fid));
 178
 179         /* Because the create update RPC will be sent during declare phase,
 180          * if creation reaches here, it means the object has been created
 181          * successfully */
 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         return 0;
 186 }
 187
 188 /**
 189  * Implementation of dt_object_operations::do_declare_ref_del
 190  *
 191  * Declare decreasing the reference count of the remote object, i.e. insert
 192  * decreasing object reference count update into the RPC, which will be sent
 193  * during transaction start.
 194  *
 195  * \param[in] env       execution environment
 196  * \param[in] dt        object to decrease the reference count.
 197  * \param[in] th        the transaction handle of refcount decrease.
 198  *
 199  * \retval              0 if the insertion succeeds.
 200  * \retval              negative errno if the insertion fails.
 201  */
 202 static int osp_md_declare_object_ref_del(const struct lu_env *env,
 203                                          struct dt_object *dt,
 204                                          struct thandle *th)
 205 {
 206         struct dt_update_request        *update;
 207         int                             rc;
 208
 209         update = dt_update_request_find_or_create(th, dt);
 210         if (IS_ERR(update)) {
 211                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 212                        dt->do_lu.lo_dev->ld_obd->obd_name,
 213                       (int)PTR_ERR(update));
 214                 return PTR_ERR(update);
 215         }
 216
 217         rc = out_ref_del_pack(env, &update->dur_buf,
 218                               lu_object_fid(&dt->do_lu),
 219                               update->dur_batchid);
 220         return rc;
 221 }
 222
 223 /**
 224  * Implementation of dt_object_operations::do_ref_del
 225  *
 226  * Do nothing in this method for now. In DNE phase I, remote updates are
 227  * actually executed during transaction start, i.e. the object reference
 228  * count has already been decreased when calling this method.
 229  *
 230  * \param[in] env       execution environment
 231  * \param[in] dt        object to decrease the reference count
 232  * \param[in] th        the transaction handle
 233  *
 234  * \retval              only return 0 for now
 235  */
 236 static int osp_md_object_ref_del(const struct lu_env *env,
 237                                  struct dt_object *dt,
 238                                  struct thandle *th)
 239 {
 240         CDEBUG(D_INFO, "ref del object "DFID"\n",
 241                PFID(&dt->do_lu.lo_header->loh_fid));
 242
 243         return 0;
 244 }
 245
 246 /**
 247  * Implementation of dt_object_operations::do_declare_ref_del
 248  *
 249  * Declare increasing the reference count of the remote object,
 250  * i.e. insert increasing object reference count update into RPC.
 251  *
 252  * \param[in] env       execution environment
 253  * \param[in] dt        object on which to increase the reference count.
 254  * \param[in] th        the transaction handle.
 255  *
 256  * \retval              0 if the insertion succeeds.
 257  * \retval              negative errno if the insertion fails.
 258  */
 259 static int osp_md_declare_ref_add(const struct lu_env *env,
 260                                   struct dt_object *dt, struct thandle *th)
 261 {
 262         struct dt_update_request        *update;
 263         int                             rc;
 264
 265         update = dt_update_request_find_or_create(th, dt);
 266         if (IS_ERR(update)) {
 267                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 268                        dt->do_lu.lo_dev->ld_obd->obd_name,
 269                        (int)PTR_ERR(update));
 270                 return PTR_ERR(update);
 271         }
 272
 273         rc = out_ref_add_pack(env, &update->dur_buf,
 274                               lu_object_fid(&dt->do_lu),
 275                               update->dur_batchid);
 276
 277         return rc;
 278 }
 279
 280 /**
 281  * Implementation of dt_object_operations::do_ref_add
 282  *
 283  * Do nothing in this method for now. In DNE phase I, remote updates are
 284  * actually executed during transaction start, i.e. the object reference
 285  * count has already been increased when calling this method.
 286  *
 287  * \param[in] env       execution environment
 288  * \param[in] dt        object on which to increase the reference count
 289  * \param[in] th        the transaction handle
 290  *
 291  * \retval              only return 0 for now
 292  */
 293 static int osp_md_object_ref_add(const struct lu_env *env, struct dt_object *dt,
 294                                  struct thandle *th)
 295 {
 296         CDEBUG(D_INFO, "ref add object "DFID"\n",
 297                PFID(&dt->do_lu.lo_header->loh_fid));
 298
 299         return 0;
 300 }
 301
 302 /**
 303  * Implementation of dt_object_operations::do_ah_init
 304  *
 305  * Initialize the allocation hint for object creation, which is usually called
 306  * before the creation, and these hints (parent and child mode) will be sent to
 307  * the remote Object Update Target (OUT) and used in the object create process,
 308  * same as OSD object creation.
 309  *
 310  * \param[in] env       execution environment
 311  * \param[in] ah        the hint to be initialized
 312  * \param[in] parent    the parent of the object
 313  * \param[in] child     the object to be created
 314  * \param[in] child_mode the mode of the created object
 315  */
 316 static void osp_md_ah_init(const struct lu_env *env,
 317                            struct dt_allocation_hint *ah,
 318                            struct dt_object *parent,
 319                            struct dt_object *child,
 320                            umode_t child_mode)
 321 {
 322         LASSERT(ah);
 323
 324         ah->dah_parent = parent;
 325         ah->dah_mode = child_mode;
 326 }
 327
 328 /**
 329  * Implementation of dt_object_operations::do_declare_attr_get
 330  *
 331  * Declare setting attributes of the remote object, i.e. insert remote
 332  * object attr_set update into RPC.
 333  *
 334  * \param[in] env       execution environment
 335  * \param[in] dt        object on which to set attributes
 336  * \param[in] attr      attributes to be set
 337  * \param[in] th        the transaction handle
 338  *
 339  * \retval              0 if the insertion succeeds.
 340  * \retval              negative errno if the insertion fails.
 341  */
 342 int osp_md_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
 343                             const struct lu_attr *attr, struct thandle *th)
 344 {
 345         struct dt_update_request        *update;
 346         int                             rc;
 347
 348         update = dt_update_request_find_or_create(th, dt);
 349         if (IS_ERR(update)) {
 350                 CERROR("%s: Get OSP update buf failed: %d\n",
 351                        dt->do_lu.lo_dev->ld_obd->obd_name,
 352                        (int)PTR_ERR(update));
 353                 return PTR_ERR(update);
 354         }
 355
 356         rc = out_attr_set_pack(env, &update->dur_buf,
 357                                lu_object_fid(&dt->do_lu), attr,
 358                                update->dur_batchid);
 359
 360         return rc;
 361 }
 362
 363 /**
 364  * Implementation of dt_object_operations::do_attr_set
 365  *
 366  * Do nothing in this method for now. In DNE phase I, remote updates
 367  * are actually executed during transaction start, i.e. object attributes
 368  * have already been set when calling this method.
 369  *
 370  * \param[in] env       execution environment
 371  * \param[in] dt        object to set attributes
 372  * \param[in] attr      attributes to be set
 373  * \param[in] th        the transaction handle
 374  * \param[in] capa      capability of setting attributes (not yet implemented).
 375  *
 376  * \retval              only return 0 for now
 377  */
 378 int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
 379                     const struct lu_attr *attr, struct thandle *th,
 380                     struct lustre_capa *capa)
 381 {
 382         CDEBUG(D_INFO, "attr set object "DFID"\n",
 383                PFID(&dt->do_lu.lo_header->loh_fid));
 384
 385         RETURN(0);
 386 }
 387
 388 /**
 389  * Implementation of dt_object_operations::do_read_lock
 390  *
 391  * osp_md_object_{read,write}_lock() will only lock the remote object in the
 392  * local cache, which uses the semaphore (opo_sem) inside the osp_object to
 393  * lock the object. Note: it will not lock the object in the whole cluster,
 394  * which relies on the LDLM lock.
 395  *
 396  * \param[in] env       execution environment
 397  * \param[in] dt        object to be locked
 398  * \param[in] role      lock role from MDD layer, see mdd_object_role().
 399  */
 400 static void osp_md_object_read_lock(const struct lu_env *env,
 401                                     struct dt_object *dt, unsigned role)
 402 {
 403         struct osp_object  *obj = dt2osp_obj(dt);
 404
 405         LASSERT(obj->opo_owner != env);
 406         down_read_nested(&obj->opo_sem, role);
 407
 408         LASSERT(obj->opo_owner == NULL);
 409 }
 410
 411 /**
 412  * Implementation of dt_object_operations::do_write_lock
 413  *
 414  * Lock the remote object in write mode.
 415  *
 416  * \param[in] env       execution environment
 417  * \param[in] dt        object to be locked
 418  * \param[in] role      lock role from MDD layer, see mdd_object_role().
 419  */
 420 static void osp_md_object_write_lock(const struct lu_env *env,
 421                                      struct dt_object *dt, unsigned role)
 422 {
 423         struct osp_object *obj = dt2osp_obj(dt);
 424
 425         down_write_nested(&obj->opo_sem, role);
 426
 427         LASSERT(obj->opo_owner == NULL);
 428         obj->opo_owner = env;
 429 }
 430
 431 /**
 432  * Implementation of dt_object_operations::do_read_unlock
 433  *
 434  * Unlock the read lock of remote object.
 435  *
 436  * \param[in] env       execution environment
 437  * \param[in] dt        object to be unlocked
 438  */
 439 static void osp_md_object_read_unlock(const struct lu_env *env,
 440                                       struct dt_object *dt)
 441 {
 442         struct osp_object *obj = dt2osp_obj(dt);
 443
 444         up_read(&obj->opo_sem);
 445 }
 446
 447 /**
 448  * Implementation of dt_object_operations::do_write_unlock
 449  *
 450  * Unlock the write lock of remote object.
 451  *
 452  * \param[in] env       execution environment
 453  * \param[in] dt        object to be unlocked
 454  */
 455 static void osp_md_object_write_unlock(const struct lu_env *env,
 456                                        struct dt_object *dt)
 457 {
 458         struct osp_object *obj = dt2osp_obj(dt);
 459
 460         LASSERT(obj->opo_owner == env);
 461         obj->opo_owner = NULL;
 462         up_write(&obj->opo_sem);
 463 }
 464
 465 /**
 466  * Implementation of dt_object_operations::do_write_locked
 467  *
 468  * Test if the object is locked in write mode.
 469  *
 470  * \param[in] env       execution environment
 471  * \param[in] dt        object to be tested
 472  */
 473 static int osp_md_object_write_locked(const struct lu_env *env,
 474                                       struct dt_object *dt)
 475 {
 476         struct osp_object *obj = dt2osp_obj(dt);
 477
 478         return obj->opo_owner == env;
 479 }
 480
 481 /**
 482  * Implementation of dt_index_operations::dio_lookup
 483  *
 484  * Look up record by key under a remote index object. It packs lookup update
 485  * into RPC, sends to the remote OUT and waits for the lookup result.
 486  *
 487  * \param[in] env       execution environment
 488  * \param[in] dt        index object to lookup
 489  * \param[out] rec      record in which to return lookup result
 490  * \param[in] key       key of index which will be looked up
 491  * \param[in] capa      capability of lookup (not yet implemented)
 492  *
 493  * \retval              1 if the lookup succeeds.
 494  * \retval              negative errno if the lookup fails.
 495  */
 496 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
 497                                struct dt_rec *rec, const struct dt_key *key,
 498                                struct lustre_capa *capa)
 499 {
 500         struct lu_buf           *lbuf   = &osp_env_info(env)->osi_lb2;
 501         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 502         struct dt_device        *dt_dev = &osp->opd_dt_dev;
 503         struct dt_update_request   *update;
 504         struct object_update_reply *reply;
 505         struct ptlrpc_request      *req = NULL;
 506         struct lu_fid              *fid;
 507         int                        rc;
 508         ENTRY;
 509
 510         /* Because it needs send the update buffer right away,
 511          * just create an update buffer, instead of attaching the
 512          * update_remote list of the thandle.
 513          */
 514         update = dt_update_request_create(dt_dev);
 515         if (IS_ERR(update))
 516                 RETURN(PTR_ERR(update));
 517
 518         rc = out_index_lookup_pack(env, &update->dur_buf,
 519                                    lu_object_fid(&dt->do_lu), rec, key);
 520         if (rc != 0) {
 521                 CERROR("%s: Insert update error: rc = %d\n",
 522                        dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
 523                 GOTO(out, rc);
 524         }
 525
 526         rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
 527         if (rc < 0)
 528                 GOTO(out, rc);
 529
 530         reply = req_capsule_server_sized_get(&req->rq_pill,
 531                                              &RMF_OUT_UPDATE_REPLY,
 532                                              OUT_UPDATE_REPLY_SIZE);
 533         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
 534                 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
 535                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 536                        reply->ourp_magic, UPDATE_REPLY_MAGIC, -EPROTO);
 537                 GOTO(out, rc = -EPROTO);
 538         }
 539
 540         rc = object_update_result_data_get(reply, lbuf, 0);
 541         if (rc < 0)
 542                 GOTO(out, rc);
 543
 544         if (lbuf->lb_len != sizeof(*fid)) {
 545                 CERROR("%s: lookup "DFID" %s wrong size %d\n",
 546                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 547                        PFID(lu_object_fid(&dt->do_lu)), (char *)key,
 548                        (int)lbuf->lb_len);
 549                 GOTO(out, rc = -EINVAL);
 550         }
 551
 552         fid = lbuf->lb_buf;
 553         if (ptlrpc_rep_need_swab(req))
 554                 lustre_swab_lu_fid(fid);
 555         if (!fid_is_sane(fid)) {
 556                 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
 557                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 558                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
 559                 GOTO(out, rc = -EINVAL);
 560         }
 561
 562         memcpy(rec, fid, sizeof(*fid));
 563
 564         GOTO(out, rc = 1);
 565
 566 out:
 567         if (req != NULL)
 568                 ptlrpc_req_finished(req);
 569
 570         dt_update_request_destroy(update);
 571
 572         return rc;
 573 }
 574
 575 /**
 576  * Implementation of dt_index_operations::dio_declare_insert
 577  *
 578  * Declare the index insert of the remote object, i.e. pack index insert update
 579  * into the RPC, which will be sent during transaction start.
 580  *
 581  * \param[in] env       execution environment
 582  * \param[in] dt        object for which to insert index
 583  * \param[in] rec       record of the index which will be inserted
 584  * \param[in] key       key of the index which will be inserted
 585  * \param[in] th        the transaction handle
 586  *
 587  * \retval              0 if the insertion succeeds.
 588  * \retval              negative errno if the insertion fails.
 589  */
 590 static int osp_md_declare_insert(const struct lu_env *env,
 591                                  struct dt_object *dt,
 592                                  const struct dt_rec *rec,
 593                                  const struct dt_key *key,
 594                                  struct thandle *th)
 595 {
 596         struct dt_update_request *update;
 597         int                      rc;
 598
 599         update = dt_update_request_find_or_create(th, dt);
 600         if (IS_ERR(update)) {
 601                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 602                        dt->do_lu.lo_dev->ld_obd->obd_name,
 603                        (int)PTR_ERR(update));
 604                 return PTR_ERR(update);
 605         }
 606
 607         rc = out_index_insert_pack(env, &update->dur_buf,
 608                                    lu_object_fid(&dt->do_lu), rec, key,
 609                                    update->dur_batchid);
 610         return rc;
 611 }
 612
 613 /**
 614  * Implementation of dt_index_operations::dio_insert
 615  *
 616  * Do nothing in this method for now. In DNE phase I, remote updates
 617  * are actually executed during transaction start, i.e. the index has
 618  * already been inserted when calling this method.
 619  *
 620  * \param[in] env       execution environment
 621  * \param[in] dt        object for which to insert index
 622  * \param[in] rec       record of the index to be inserted
 623  * \param[in] key       key of the index to be inserted
 624  * \param[in] th        the transaction handle
 625  * \param[in] capa      capability of insert (not yet implemented)
 626  * \param[in] ignore_quota quota enforcement for insert
 627  *
 628  * \retval              only return 0 for now
 629  */
 630 static int osp_md_index_insert(const struct lu_env *env,
 631                                struct dt_object *dt,
 632                                const struct dt_rec *rec,
 633                                const struct dt_key *key,
 634                                struct thandle *th,
 635                                struct lustre_capa *capa,
 636                                int ignore_quota)
 637 {
 638         return 0;
 639 }
 640
 641 /**
 642  * Implementation of dt_index_operations::dio_declare_delete
 643  *
 644  * Declare the index delete of the remote object, i.e. insert index delete
 645  * update into the RPC, which will be sent during transaction start.
 646  *
 647  * \param[in] env       execution environment
 648  * \param[in] dt        object for which to delete index
 649  * \param[in] key       key of the index
 650  * \param[in] th        the transaction handle
 651  *
 652  * \retval              0 if the insertion succeeds.
 653  * \retval              negative errno if the insertion fails.
 654  */
 655 static int osp_md_declare_delete(const struct lu_env *env,
 656                                  struct dt_object *dt,
 657                                  const struct dt_key *key,
 658                                  struct thandle *th)
 659 {
 660         struct dt_update_request *update;
 661         int                      rc;
 662
 663         update = dt_update_request_find_or_create(th, dt);
 664         if (IS_ERR(update)) {
 665                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 666                        dt->do_lu.lo_dev->ld_obd->obd_name,
 667                        (int)PTR_ERR(update));
 668                 return PTR_ERR(update);
 669         }
 670
 671         rc = out_index_delete_pack(env, &update->dur_buf,
 672                                    lu_object_fid(&dt->do_lu), key,
 673                                    update->dur_batchid);
 674         return rc;
 675 }
 676
 677 /**
 678  * Implementation of dt_index_operations::dio_delete
 679  *
 680  * Do nothing in this method for now. Because in DNE phase I, remote updates
 681  * are actually executed during transaction start, i.e. the index has already
 682  * been deleted when calling this method.
 683  *
 684  * \param[in] env       execution environment
 685  * \param[in] dt        object for which to delete index
 686  * \param[in] key       key of the index which will be deleted
 687  * \param[in] th        the transaction handle
 688  * \param[in] capa      capability of delete (not yet implemented)
 689  *
 690  * \retval              only return 0 for now
 691  */
 692 static int osp_md_index_delete(const struct lu_env *env,
 693                                struct dt_object *dt,
 694                                const struct dt_key *key,
 695                                struct thandle *th,
 696                                struct lustre_capa *capa)
 697 {
 698         CDEBUG(D_INFO, "index delete "DFID" %s\n",
 699                PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
 700
 701         return 0;
 702 }
 703
 704 /**
 705  * Implementation of dt_index_operations::dio_it.next
 706  *
 707  * Advance the pointer of the iterator to the next entry. It shares a similar
 708  * internal implementation with osp_orphan_it_next(), which is being used for
 709  * remote orphan index object. This method will be used for remote directory.
 710  *
 711  * \param[in] env       execution environment
 712  * \param[in] di        iterator of this iteration
 713  *
 714  * \retval              0 if the pointer is advanced successfuly.
 715  * \retval              1 if it reaches to the end of the index object.
 716  * \retval              negative errno if the pointer cannot be advanced.
 717  */
 718 int osp_md_index_it_next(const struct lu_env *env, struct dt_it *di)
 719 {
 720         struct osp_it           *it = (struct osp_it *)di;
 721         struct lu_idxpage       *idxpage;
 722         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 723         int                     rc;
 724         ENTRY;
 725
 726 again:
 727         idxpage = it->ooi_cur_idxpage;
 728         if (idxpage != NULL) {
 729                 if (idxpage->lip_nr == 0)
 730                         RETURN(1);
 731
 732                 it->ooi_pos_ent++;
 733                 if (ent == NULL) {
 734                         it->ooi_ent =
 735                               (struct lu_dirent *)idxpage->lip_entries;
 736                         RETURN(0);
 737                 } else if (le16_to_cpu(ent->lde_reclen) != 0 &&
 738                            it->ooi_pos_ent < idxpage->lip_nr) {
 739                         ent = (struct lu_dirent *)(((char *)ent) +
 740                                         le16_to_cpu(ent->lde_reclen));
 741                         it->ooi_ent = ent;
 742                         RETURN(0);
 743                 } else {
 744                         it->ooi_ent = NULL;
 745                 }
 746         }
 747
 748         rc = osp_it_next_page(env, di);
 749         if (rc == 0)
 750                 goto again;
 751
 752         RETURN(rc);
 753 }
 754
 755 /**
 756  * Implementation of dt_index_operations::dio_it.key
 757  *
 758  * Get the key at current iterator poisiton. These iteration methods
 759  * (dio_it) will only be used for iterating the remote directory, so
 760  * the key is the name of the directory entry.
 761  *
 762  * \param[in] env       execution environment
 763  * \param[in] di        iterator of this iteration
 764  *
 765  * \retval              name of the current entry
 766  */
 767 static struct dt_key *osp_it_key(const struct lu_env *env,
 768                                  const struct dt_it *di)
 769 {
 770         struct osp_it           *it = (struct osp_it *)di;
 771         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 772
 773         return (struct dt_key *)ent->lde_name;
 774 }
 775
 776 /**
 777  * Implementation of dt_index_operations::dio_it.key_size
 778  *
 779  * Get the key size at current iterator poisiton. These iteration methods
 780  * (dio_it) will only be used for iterating the remote directory, so the key
 781  * size is the name size of the directory entry.
 782  *
 783  * \param[in] env       execution environment
 784  * \param[in] di        iterator of this iteration
 785  *
 786  * \retval              name size of the current entry
 787  */
 788
 789 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
 790 {
 791         struct osp_it           *it = (struct osp_it *)di;
 792         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 793
 794         return (int)le16_to_cpu(ent->lde_namelen);
 795 }
 796
 797 /**
 798  * Implementation of dt_index_operations::dio_it.rec
 799  *
 800  * Get the record at current iterator position. These iteration methods
 801  * (dio_it) will only be used for iterating the remote directory, so it
 802  * uses lu_dirent_calc_size() to calculate the record size.
 803  *
 804  * \param[in] env       execution environment
 805  * \param[in] di        iterator of this iteration
 806  * \param[out] rec      the record to be returned
 807  * \param[in] attr      attributes of the index object, so it knows
 808  *                      how to pack the entry.
 809  *
 810  * \retval              only return 0 for now
 811  */
 812 static int osp_md_index_it_rec(const struct lu_env *env, const struct dt_it *di,
 813                                struct dt_rec *rec, __u32 attr)
 814 {
 815         struct osp_it           *it = (struct osp_it *)di;
 816         struct lu_dirent        *ent = (struct lu_dirent *)it->ooi_ent;
 817         int                     reclen;
 818
 819         reclen = lu_dirent_calc_size(le16_to_cpu(ent->lde_namelen), attr);
 820         memcpy(rec, ent, reclen);
 821         return 0;
 822 }
 823
 824 /**
 825  * Implementation of dt_index_operations::dio_it.load
 826  *
 827  * Locate the iteration cursor to the specified position (cookie).
 828  *
 829  * \param[in] env       pointer to the thread context
 830  * \param[in] di        pointer to the iteration structure
 831  * \param[in] hash      the specified position
 832  *
 833  * \retval              positive number for locating to the exactly position
 834  *                      or the next
 835  * \retval              0 for arriving at the end of the iteration
 836  * \retval              negative error number on failure
 837  */
 838 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
 839                        __u64 hash)
 840 {
 841         struct osp_it   *it     = (struct osp_it *)di;
 842         int              rc;
 843
 844         it->ooi_next = hash;
 845         rc = osp_md_index_it_next(env, (struct dt_it *)di);
 846         if (rc == 1)
 847                 return 0;
 848
 849         if (rc == 0)
 850                 return 1;
 851
 852         return rc;
 853 }
 854
 855 const struct dt_index_operations osp_md_index_ops = {
 856         .dio_lookup         = osp_md_index_lookup,
 857         .dio_declare_insert = osp_md_declare_insert,
 858         .dio_insert         = osp_md_index_insert,
 859         .dio_declare_delete = osp_md_declare_delete,
 860         .dio_delete         = osp_md_index_delete,
 861         .dio_it     = {
 862                 .init     = osp_it_init,
 863                 .fini     = osp_it_fini,
 864                 .get      = osp_it_get,
 865                 .put      = osp_it_put,
 866                 .next     = osp_md_index_it_next,
 867                 .key      = osp_it_key,
 868                 .key_size = osp_it_key_size,
 869                 .rec      = osp_md_index_it_rec,
 870                 .store    = osp_it_store,
 871                 .load     = osp_it_load,
 872                 .key_rec  = osp_it_key_rec,
 873         }
 874 };
 875
 876 /**
 877  * Implementation of dt_object_operations::do_index_try
 878  *
 879  * Try to initialize the index API pointer for the given object. This
 880  * is the entry point of the index API, i.e. we must call this method
 881  * to initialize the index object before calling other index methods.
 882  *
 883  * \param[in] env       execution environment
 884  * \param[in] dt        index object to be initialized
 885  * \param[in] feat      the index feature of the object
 886  *
 887  * \retval              0 if the initialization succeeds.
 888  * \retval              negative errno if the initialization fails.
 889  */
 890 static int osp_md_index_try(const struct lu_env *env,
 891                             struct dt_object *dt,
 892                             const struct dt_index_features *feat)
 893 {
 894         dt->do_index_ops = &osp_md_index_ops;
 895         return 0;
 896 }
 897
 898 /**
 899  * Implementation of dt_object_operations::do_object_lock
 900  *
 901  * Enqueue a lock (by ldlm_cli_enqueue()) of remote object on the remote MDT,
 902  * which will lock the object in the global namespace.
 903  *
 904  * \param[in] env       execution environment
 905  * \param[in] dt        object to be locked
 906  * \param[out] lh       lock handle
 907  * \param[in] einfo     enqueue information
 908  * \param[in] policy    lock policy
 909  *
 910  * \retval              ELDLM_OK if locking the object succeeds.
 911  * \retval              negative errno if locking fails.
 912  */
 913 static int osp_md_object_lock(const struct lu_env *env,
 914                               struct dt_object *dt,
 915                               struct lustre_handle *lh,
 916                               struct ldlm_enqueue_info *einfo,
 917                               ldlm_policy_data_t *policy)
 918 {
 919         struct ldlm_res_id      *res_id;
 920         struct dt_device        *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 921         struct osp_device       *osp = dt2osp_dev(dt_dev);
 922         struct ptlrpc_request   *req;
 923         int                     rc = 0;
 924         __u64                   flags = 0;
 925         ldlm_mode_t             mode;
 926
 927         res_id = einfo->ei_res_id;
 928         LASSERT(res_id != NULL);
 929
 930         mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
 931                                LDLM_FL_BLOCK_GRANTED, res_id,
 932                                einfo->ei_type, policy,
 933                                einfo->ei_mode, lh, 0);
 934         if (mode > 0)
 935                 return ELDLM_OK;
 936
 937         req = ldlm_enqueue_pack(osp->opd_exp, 0);
 938         if (IS_ERR(req))
 939                 RETURN(PTR_ERR(req));
 940
 941         rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
 942                               (const ldlm_policy_data_t *)policy,
 943                               &flags, NULL, 0, LVB_T_NONE, lh, 0);
 944
 945         ptlrpc_req_finished(req);
 946
 947         return rc == ELDLM_OK ? 0 : -EIO;
 948 }
 949
 950 /**
 951  * Implementation of dt_object_operations::do_object_unlock
 952  *
 953  * Cancel a lock of a remote object.
 954  *
 955  * \param[in] env       execution environment
 956  * \param[in] dt        object to be unlocked
 957  * \param[in] einfo     lock enqueue information
 958  * \param[in] policy    lock policy
 959  *
 960  * \retval              Only return 0 for now.
 961  */
 962 static int osp_md_object_unlock(const struct lu_env *env,
 963                                 struct dt_object *dt,
 964                                 struct ldlm_enqueue_info *einfo,
 965                                 ldlm_policy_data_t *policy)
 966 {
 967         struct lustre_handle    *lockh = einfo->ei_cbdata;
 968
 969         /* unlock finally */
 970         ldlm_lock_decref(lockh, einfo->ei_mode);
 971
 972         return 0;
 973 }
 974
 975 struct dt_object_operations osp_md_obj_ops = {
 976         .do_read_lock         = osp_md_object_read_lock,
 977         .do_write_lock        = osp_md_object_write_lock,
 978         .do_read_unlock       = osp_md_object_read_unlock,
 979         .do_write_unlock      = osp_md_object_write_unlock,
 980         .do_write_locked      = osp_md_object_write_locked,
 981         .do_declare_create    = osp_md_declare_object_create,
 982         .do_create            = osp_md_object_create,
 983         .do_declare_ref_add   = osp_md_declare_ref_add,
 984         .do_ref_add           = osp_md_object_ref_add,
 985         .do_declare_ref_del   = osp_md_declare_object_ref_del,
 986         .do_ref_del           = osp_md_object_ref_del,
 987         .do_declare_destroy   = osp_declare_object_destroy,
 988         .do_destroy           = osp_object_destroy,
 989         .do_ah_init           = osp_md_ah_init,
 990         .do_attr_get          = osp_attr_get,
 991         .do_declare_attr_set  = osp_md_declare_attr_set,
 992         .do_attr_set          = osp_md_attr_set,
 993         .do_xattr_get         = osp_xattr_get,
 994         .do_declare_xattr_set = osp_declare_xattr_set,
 995         .do_xattr_set         = osp_xattr_set,
 996         .do_declare_xattr_del = osp_declare_xattr_del,
 997         .do_xattr_del         = osp_xattr_del,
 998         .do_index_try         = osp_md_index_try,
 999         .do_object_lock       = osp_md_object_lock,
1000         .do_object_unlock     = osp_md_object_unlock,
1001 };
1002
1003 /**
1004  * Implementation of dt_body_operations::dbo_declare_write
1005  *
1006  * Declare an object write. In DNE phase I, it will pack the write
1007  * object update into the RPC.
1008  *
1009  * \param[in] env       execution environment
1010  * \param[in] dt        object to be written
1011  * \param[in] buf       buffer to write which includes an embedded size field
1012  * \param[in] pos       offet in the object to start writing at
1013  * \param[in] th        transaction handle
1014  *
1015  * \retval              0 if the insertion succeeds.
1016  * \retval              negative errno if the insertion fails.
1017  */
1018 static ssize_t osp_md_declare_write(const struct lu_env *env,
1019                                     struct dt_object *dt,
1020                                     const struct lu_buf *buf,
1021                                     loff_t pos, struct thandle *th)
1022 {
1023         struct dt_update_request  *update;
1024         ssize_t                   rc;
1025
1026         update = dt_update_request_find_or_create(th, dt);
1027         if (IS_ERR(update)) {
1028                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
1029                        dt->do_lu.lo_dev->ld_obd->obd_name,
1030                        (int)PTR_ERR(update));
1031                 return PTR_ERR(update);
1032         }
1033
1034         rc = out_write_pack(env, &update->dur_buf, lu_object_fid(&dt->do_lu),
1035                             buf, pos, update->dur_batchid);
1036
1037         return rc;
1038
1039 }
1040
1041 /**
1042  * Implementation of dt_body_operations::dbo_write
1043  *
1044  * Return the buffer size. In DNE phase I, remote updates
1045  * are actually executed during transaction start, the buffer has
1046  * already been written when this method is being called.
1047  *
1048  * \param[in] env       execution environment
1049  * \param[in] dt        object to be written
1050  * \param[in] buf       buffer to write which includes an embedded size field
1051  * \param[in] pos       offet in the object to start writing at
1052  * \param[in] th        transaction handle
1053  * \param[in] capa      capability of the write (not yet implemented)
1054  * \param[in] ignore_quota quota enforcement for this write
1055  *
1056  * \retval              the buffer size in bytes.
1057  */
1058 static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
1059                             const struct lu_buf *buf, loff_t *pos,
1060                             struct thandle *handle,
1061                             struct lustre_capa *capa, int ignore_quota)
1062 {
1063         *pos += buf->lb_len;
1064         return buf->lb_len;
1065 }
1066
1067 /* These body operation will be used to write symlinks during migration etc */
1068 struct dt_body_operations osp_md_body_ops = {
1069         .dbo_declare_write      = osp_md_declare_write,
1070         .dbo_write              = osp_md_write,
1071 };