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  * Lustre MDT Proxy Device
  29  *
  30  * Author: Di Wang <di.wang@intel.com>
  31  */
  32
  33 #define DEBUG_SUBSYSTEM S_MDS
  34
  35 #include <lustre_log.h>
  36 #include "osp_internal.h"
  37
  38 static const char dot[] = ".";
  39 static const char dotdot[] = "..";
  40
  41 static int osp_md_declare_object_create(const struct lu_env *env,
  42                                         struct dt_object *dt,
  43                                         struct lu_attr *attr,
  44                                         struct dt_allocation_hint *hint,
  45                                         struct dt_object_format *dof,
  46                                         struct thandle *th)
  47 {
  48         struct osp_thread_info  *osi = osp_env_info(env);
  49         struct update_request   *update;
  50         struct lu_fid           *fid1;
  51         int                     sizes[2] = {sizeof(struct obdo), 0};
  52         char                    *bufs[2] = {NULL, NULL};
  53         int                     buf_count;
  54         int                     rc;
  55
  56         update = out_find_create_update_loc(th, dt);
  57         if (IS_ERR(update)) {
  58                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
  59                        dt->do_lu.lo_dev->ld_obd->obd_name,
  60                        (int)PTR_ERR(update));
  61                 return PTR_ERR(update);
  62         }
  63
  64         osi->osi_obdo.o_valid = 0;
  65         LASSERT(S_ISDIR(attr->la_mode));
  66         obdo_from_la(&osi->osi_obdo, attr, attr->la_valid);
  67         lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
  68         obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
  69
  70         bufs[0] = (char *)&osi->osi_obdo;
  71         buf_count = 1;
  72         fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
  73         if (hint->dah_parent) {
  74                 struct lu_fid *fid2;
  75                 struct lu_fid *tmp_fid = &osi->osi_fid;
  76
  77                 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
  78                 fid_cpu_to_le(tmp_fid, fid2);
  79                 sizes[1] = sizeof(*tmp_fid);
  80                 bufs[1] = (char *)tmp_fid;
  81                 buf_count++;
  82         }
  83
  84         if (lu_object_exists(&dt->do_lu)) {
  85                 /* If the object already exists, we needs to destroy
  86                  * this orphan object first.
  87                  *
  88                  * The scenario might happen in this case
  89                  *
  90                  * 1. client send remote create to MDT0.
  91                  * 2. MDT0 send create update to MDT1.
  92                  * 3. MDT1 finished create synchronously.
  93                  * 4. MDT0 failed and reboot.
  94                  * 5. client resend remote create to MDT0.
  95                  * 6. MDT0 tries to resend create update to MDT1,
  96                  *    but find the object already exists
  97                  */
  98                 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
  99                        dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1));
 100
 101                 rc = out_insert_update(env, update, OBJ_REF_DEL, fid1, 0,
 102                                        NULL, NULL);
 103                 if (rc != 0)
 104                         GOTO(out, rc);
 105
 106                 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
 107                         /* decrease for ".." */
 108                         rc = out_insert_update(env, update, OBJ_REF_DEL, fid1,
 109                                                0, NULL, NULL);
 110                         if (rc != 0)
 111                                 GOTO(out, rc);
 112                 }
 113
 114                 rc = out_insert_update(env, update, OBJ_DESTROY, fid1, 0, NULL,
 115                                        NULL);
 116                 if (rc != 0)
 117                         GOTO(out, rc);
 118
 119                 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
 120                 /* Increase batchid to add this orphan object deletion
 121                  * to separate transaction */
 122                 update_inc_batchid(update);
 123         }
 124
 125         rc = out_insert_update(env, update, OBJ_CREATE, fid1, buf_count, sizes,
 126                                (const char **)bufs);
 127 out:
 128         if (rc)
 129                 CERROR("%s: Insert update error: rc = %d\n",
 130                        dt->do_lu.lo_dev->ld_obd->obd_name, rc);
 131
 132         return rc;
 133 }
 134
 135 static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
 136                                 struct lu_attr *attr,
 137                                 struct dt_allocation_hint *hint,
 138                                 struct dt_object_format *dof,
 139                                 struct thandle *th)
 140 {
 141         struct osp_object  *obj = dt2osp_obj(dt);
 142
 143         CDEBUG(D_INFO, "create object "DFID"\n",
 144                PFID(&dt->do_lu.lo_header->loh_fid));
 145
 146         /* Because the create update RPC will be sent during declare phase,
 147          * if creation reaches here, it means the object has been created
 148          * successfully */
 149         dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
 150         obj->opo_empty = 1;
 151
 152         return 0;
 153 }
 154
 155 static int osp_md_declare_object_ref_del(const struct lu_env *env,
 156                                          struct dt_object *dt,
 157                                          struct thandle *th)
 158 {
 159         struct update_request   *update;
 160         struct lu_fid           *fid;
 161         int                     rc;
 162
 163         update = out_find_create_update_loc(th, dt);
 164         if (IS_ERR(update)) {
 165                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 166                        dt->do_lu.lo_dev->ld_obd->obd_name,
 167                       (int)PTR_ERR(update));
 168                 return PTR_ERR(update);
 169         }
 170
 171         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 172
 173         rc = out_insert_update(env, update, OBJ_REF_DEL, fid, 0, NULL, NULL);
 174
 175         return rc;
 176 }
 177
 178 static int osp_md_object_ref_del(const struct lu_env *env,
 179                                  struct dt_object *dt,
 180                                  struct thandle *th)
 181 {
 182         CDEBUG(D_INFO, "ref del object "DFID"\n",
 183                PFID(&dt->do_lu.lo_header->loh_fid));
 184
 185         return 0;
 186 }
 187
 188 static int osp_md_declare_ref_add(const struct lu_env *env,
 189                                   struct dt_object *dt, struct thandle *th)
 190 {
 191         struct update_request   *update;
 192         struct lu_fid           *fid;
 193         int                     rc;
 194
 195         update = out_find_create_update_loc(th, dt);
 196         if (IS_ERR(update)) {
 197                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 198                        dt->do_lu.lo_dev->ld_obd->obd_name,
 199                        (int)PTR_ERR(update));
 200                 return PTR_ERR(update);
 201         }
 202
 203         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 204
 205         rc = out_insert_update(env, update, OBJ_REF_ADD, fid, 0, NULL, NULL);
 206
 207         return rc;
 208 }
 209
 210 static int osp_md_object_ref_add(const struct lu_env *env,
 211                                  struct dt_object *dt,
 212                                  struct thandle *th)
 213 {
 214         CDEBUG(D_INFO, "ref add object "DFID"\n",
 215                PFID(&dt->do_lu.lo_header->loh_fid));
 216
 217         return 0;
 218 }
 219
 220 static void osp_md_ah_init(const struct lu_env *env,
 221                            struct dt_allocation_hint *ah,
 222                            struct dt_object *parent,
 223                            struct dt_object *child,
 224                            umode_t child_mode)
 225 {
 226         LASSERT(ah);
 227
 228         memset(ah, 0, sizeof(*ah));
 229         ah->dah_parent = parent;
 230         ah->dah_mode = child_mode;
 231 }
 232
 233 static int osp_md_declare_attr_set(const struct lu_env *env,
 234                                    struct dt_object *dt,
 235                                    const struct lu_attr *attr,
 236                                    struct thandle *th)
 237 {
 238         struct osp_thread_info *osi = osp_env_info(env);
 239         struct update_request  *update;
 240         struct lu_fid          *fid;
 241         int                     size = sizeof(struct obdo);
 242         char                   *buf;
 243         int                     rc;
 244
 245         update = out_find_create_update_loc(th, dt);
 246         if (IS_ERR(update)) {
 247                 CERROR("%s: Get OSP update buf failed: %d\n",
 248                        dt->do_lu.lo_dev->ld_obd->obd_name,
 249                        (int)PTR_ERR(update));
 250                 return PTR_ERR(update);
 251         }
 252
 253         osi->osi_obdo.o_valid = 0;
 254         LASSERT(!(attr->la_valid & (LA_MODE | LA_TYPE)));
 255         obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
 256                      attr->la_valid);
 257         lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
 258         obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
 259
 260         buf = (char *)&osi->osi_obdo;
 261         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 262
 263         rc = out_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size,
 264                                (const char **)&buf);
 265
 266         return rc;
 267 }
 268
 269 static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
 270                            const struct lu_attr *attr, struct thandle *th,
 271                            struct lustre_capa *capa)
 272 {
 273         CDEBUG(D_INFO, "attr set object "DFID"\n",
 274                PFID(&dt->do_lu.lo_header->loh_fid));
 275
 276         RETURN(0);
 277 }
 278
 279 static void osp_md_object_read_lock(const struct lu_env *env,
 280                                     struct dt_object *dt, unsigned role)
 281 {
 282         struct osp_object  *obj = dt2osp_obj(dt);
 283
 284         LASSERT(obj->opo_owner != env);
 285         down_read_nested(&obj->opo_sem, role);
 286
 287         LASSERT(obj->opo_owner == NULL);
 288 }
 289
 290 static void osp_md_object_write_lock(const struct lu_env *env,
 291                                      struct dt_object *dt, unsigned role)
 292 {
 293         struct osp_object *obj = dt2osp_obj(dt);
 294
 295         down_write_nested(&obj->opo_sem, role);
 296
 297         LASSERT(obj->opo_owner == NULL);
 298         obj->opo_owner = env;
 299 }
 300
 301 static void osp_md_object_read_unlock(const struct lu_env *env,
 302                                       struct dt_object *dt)
 303 {
 304         struct osp_object *obj = dt2osp_obj(dt);
 305
 306         up_read(&obj->opo_sem);
 307 }
 308
 309 static void osp_md_object_write_unlock(const struct lu_env *env,
 310                                        struct dt_object *dt)
 311 {
 312         struct osp_object *obj = dt2osp_obj(dt);
 313
 314         LASSERT(obj->opo_owner == env);
 315         obj->opo_owner = NULL;
 316         up_write(&obj->opo_sem);
 317 }
 318
 319 static int osp_md_object_write_locked(const struct lu_env *env,
 320                                       struct dt_object *dt)
 321 {
 322         struct osp_object *obj = dt2osp_obj(dt);
 323
 324         return obj->opo_owner == env;
 325 }
 326
 327 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
 328                                struct dt_rec *rec, const struct dt_key *key,
 329                                struct lustre_capa *capa)
 330 {
 331         struct lu_buf           *lbuf   = &osp_env_info(env)->osi_lb2;
 332         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 333         struct dt_device        *dt_dev = &osp->opd_dt_dev;
 334         struct update_request   *update;
 335         struct ptlrpc_request   *req = NULL;
 336         int                     size = strlen((char *)key) + 1;
 337         int                     rc;
 338         struct update_reply     *reply;
 339         struct lu_fid           *fid;
 340
 341         ENTRY;
 342
 343         /* Because it needs send the update buffer right away,
 344          * just create an update buffer, instead of attaching the
 345          * update_remote list of the thandle.
 346          */
 347         update = out_create_update_req(dt_dev);
 348         if (IS_ERR(update))
 349                 RETURN(PTR_ERR(update));
 350
 351         rc = out_insert_update(env, update, OBJ_INDEX_LOOKUP,
 352                                lu_object_fid(&dt->do_lu), 1, &size,
 353                                (const char **)&key);
 354         if (rc) {
 355                 CERROR("%s: Insert update error: rc = %d\n",
 356                        dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
 357                 GOTO(out, rc);
 358         }
 359
 360         rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
 361         if (rc < 0)
 362                 GOTO(out, rc);
 363
 364         reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
 365                                              UPDATE_BUFFER_SIZE);
 366         if (reply->ur_version != UPDATE_REPLY_V1) {
 367                 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
 368                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 369                        reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
 370                 GOTO(out, rc = -EPROTO);
 371         }
 372
 373         rc = update_get_reply_result(reply, NULL, 0);
 374         if (rc < 0) {
 375                 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
 376                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 377                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
 378                 GOTO(out, rc);
 379         }
 380
 381         rc = update_get_reply_buf(reply, lbuf, 0);
 382         if (rc < 0)
 383                 GOTO(out, rc);
 384
 385         if (lbuf->lb_len != sizeof(*fid)) {
 386                 CERROR("%s: lookup "DFID" %s wrong size %d\n",
 387                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 388                        PFID(lu_object_fid(&dt->do_lu)), (char *)key,
 389                        (int)lbuf->lb_len);
 390                 GOTO(out, rc = -EINVAL);
 391         }
 392
 393         fid = lbuf->lb_buf;
 394         fid_le_to_cpu(fid, fid);
 395         if (!fid_is_sane(fid)) {
 396                 CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
 397                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 398                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid));
 399                 GOTO(out, rc = -EINVAL);
 400         }
 401
 402         memcpy(rec, fid, sizeof(*fid));
 403
 404         GOTO(out, rc = 1);
 405
 406 out:
 407         if (req != NULL)
 408                 ptlrpc_req_finished(req);
 409
 410         out_destroy_update_req(update);
 411
 412         return rc;
 413 }
 414
 415 static int osp_md_declare_insert(const struct lu_env *env,
 416                                  struct dt_object *dt,
 417                                  const struct dt_rec *rec,
 418                                  const struct dt_key *key,
 419                                  struct thandle *th)
 420 {
 421         struct update_request   *update;
 422         struct lu_fid           *fid;
 423         struct lu_fid           *rec_fid = (struct lu_fid *)rec;
 424         int                     size[2] = {strlen((char *)key) + 1,
 425                                                   sizeof(*rec_fid)};
 426         const char              *bufs[2] = {(char *)key, (char *)rec_fid};
 427         int                     rc;
 428
 429         update = out_find_create_update_loc(th, dt);
 430         if (IS_ERR(update)) {
 431                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 432                        dt->do_lu.lo_dev->ld_obd->obd_name,
 433                        (int)PTR_ERR(update));
 434                 return PTR_ERR(update);
 435         }
 436
 437         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 438
 439         CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
 440                dt->do_lu.lo_dev->ld_obd->obd_name,
 441                PFID(fid), (char *)key, PFID(rec_fid));
 442
 443         fid_cpu_to_le(rec_fid, rec_fid);
 444
 445         rc = out_insert_update(env, update, OBJ_INDEX_INSERT, fid,
 446                                ARRAY_SIZE(size), size, bufs);
 447         return rc;
 448 }
 449
 450 static int osp_md_index_insert(const struct lu_env *env,
 451                                struct dt_object *dt,
 452                                const struct dt_rec *rec,
 453                                const struct dt_key *key,
 454                                struct thandle *th,
 455                                struct lustre_capa *capa,
 456                                int ignore_quota)
 457 {
 458         return 0;
 459 }
 460
 461 static int osp_md_declare_delete(const struct lu_env *env,
 462                                  struct dt_object *dt,
 463                                  const struct dt_key *key,
 464                                  struct thandle *th)
 465 {
 466         struct update_request *update;
 467         struct lu_fid *fid;
 468         int size = strlen((char *)key) + 1;
 469         int rc;
 470
 471         update = out_find_create_update_loc(th, dt);
 472         if (IS_ERR(update)) {
 473                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 474                        dt->do_lu.lo_dev->ld_obd->obd_name,
 475                        (int)PTR_ERR(update));
 476                 return PTR_ERR(update);
 477         }
 478
 479         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 480
 481         rc = out_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
 482                                (const char **)&key);
 483
 484         return rc;
 485 }
 486
 487 static int osp_md_index_delete(const struct lu_env *env,
 488                                struct dt_object *dt,
 489                                const struct dt_key *key,
 490                                struct thandle *th,
 491                                struct lustre_capa *capa)
 492 {
 493         CDEBUG(D_INFO, "index delete "DFID" %s\n",
 494                PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
 495
 496         return 0;
 497 }
 498
 499 /**
 500  * Creates or initializes iterator context.
 501  *
 502  * Note: for OSP, these index iterate api is only used to check
 503  * whether the directory is empty now (see mdd_dir_is_empty).
 504  * Since dir_empty will be return by OBJ_ATTR_GET(see osp_attr_get/
 505  * out_attr_get). So the implementation of these iterator is simplied
 506  * to make mdd_dir_is_empty happy. The real iterator should be
 507  * implemented, if we need it one day.
 508  */
 509 static struct dt_it *osp_it_init(const struct lu_env *env,
 510                                  struct dt_object *dt,
 511                                  __u32 attr,
 512                                 struct lustre_capa *capa)
 513 {
 514         lu_object_get(&dt->do_lu);
 515         return (struct dt_it *)dt;
 516 }
 517
 518 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
 519 {
 520         struct dt_object *dt = (struct dt_object *)di;
 521         lu_object_put(env, &dt->do_lu);
 522 }
 523
 524 static int osp_it_get(const struct lu_env *env,
 525                       struct dt_it *di, const struct dt_key *key)
 526 {
 527         return 1;
 528 }
 529
 530 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
 531 {
 532         return;
 533 }
 534
 535 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
 536 {
 537         struct dt_object *dt = (struct dt_object *)di;
 538         struct osp_object *o = dt2osp_obj(dt);
 539
 540         if (o->opo_empty)
 541                 return 1;
 542
 543         return 0;
 544 }
 545
 546 static struct dt_key *osp_it_key(const struct lu_env *env,
 547                                  const struct dt_it *di)
 548 {
 549         LBUG();
 550         return NULL;
 551 }
 552
 553 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
 554 {
 555         LBUG();
 556         return 0;
 557 }
 558
 559 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
 560                       struct dt_rec *lde, __u32 attr)
 561 {
 562         LBUG();
 563         return 0;
 564 }
 565
 566 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
 567 {
 568         LBUG();
 569         return 0;
 570 }
 571
 572 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
 573                        __u64 hash)
 574 {
 575         LBUG();
 576         return 0;
 577 }
 578
 579 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
 580                           void *key_rec)
 581 {
 582         LBUG();
 583         return 0;
 584 }
 585
 586 static const struct dt_index_operations osp_md_index_ops = {
 587         .dio_lookup         = osp_md_index_lookup,
 588         .dio_declare_insert = osp_md_declare_insert,
 589         .dio_insert         = osp_md_index_insert,
 590         .dio_declare_delete = osp_md_declare_delete,
 591         .dio_delete         = osp_md_index_delete,
 592         .dio_it     = {
 593                 .init     = osp_it_init,
 594                 .fini     = osp_it_fini,
 595                 .get      = osp_it_get,
 596                 .put      = osp_it_put,
 597                 .next     = osp_it_next,
 598                 .key      = osp_it_key,
 599                 .key_size = osp_it_key_size,
 600                 .rec      = osp_it_rec,
 601                 .store    = osp_it_store,
 602                 .load     = osp_it_load,
 603                 .key_rec  = osp_it_key_rec,
 604         }
 605 };
 606
 607 static int osp_md_index_try(const struct lu_env *env,
 608                             struct dt_object *dt,
 609                             const struct dt_index_features *feat)
 610 {
 611         dt->do_index_ops = &osp_md_index_ops;
 612         return 0;
 613 }
 614
 615 static int osp_md_object_lock(const struct lu_env *env,
 616                               struct dt_object *dt,
 617                               struct lustre_handle *lh,
 618                               struct ldlm_enqueue_info *einfo,
 619                               void *policy)
 620 {
 621         struct osp_thread_info *info = osp_env_info(env);
 622         struct ldlm_res_id     *res_id = &info->osi_resid;
 623         struct dt_device       *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 624         struct osp_device      *osp = dt2osp_dev(dt_dev);
 625         struct ptlrpc_request  *req = NULL;
 626         int                     rc = 0;
 627         __u64                   flags = 0;
 628         ldlm_mode_t             mode;
 629
 630         fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
 631
 632         mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
 633                                LDLM_FL_BLOCK_GRANTED, res_id,
 634                                einfo->ei_type,
 635                                (ldlm_policy_data_t *)policy,
 636                                einfo->ei_mode, lh, 0);
 637         if (mode > 0)
 638                 return ELDLM_OK;
 639
 640         req = ldlm_enqueue_pack(osp->opd_exp, 0);
 641         if (IS_ERR(req))
 642                 RETURN(PTR_ERR(req));
 643
 644         rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
 645                               (const ldlm_policy_data_t *)policy,
 646                               &flags, NULL, 0, LVB_T_NONE, lh, 0);
 647
 648         ptlrpc_req_finished(req);
 649
 650         return rc == ELDLM_OK ? 0 : -EIO;
 651 }
 652
 653 struct dt_object_operations osp_md_obj_ops = {
 654         .do_read_lock         = osp_md_object_read_lock,
 655         .do_write_lock        = osp_md_object_write_lock,
 656         .do_read_unlock       = osp_md_object_read_unlock,
 657         .do_write_unlock      = osp_md_object_write_unlock,
 658         .do_write_locked      = osp_md_object_write_locked,
 659         .do_declare_create    = osp_md_declare_object_create,
 660         .do_create            = osp_md_object_create,
 661         .do_declare_ref_add   = osp_md_declare_ref_add,
 662         .do_ref_add           = osp_md_object_ref_add,
 663         .do_declare_ref_del   = osp_md_declare_object_ref_del,
 664         .do_ref_del           = osp_md_object_ref_del,
 665         .do_declare_destroy   = osp_declare_object_destroy,
 666         .do_destroy           = osp_object_destroy,
 667         .do_ah_init           = osp_md_ah_init,
 668         .do_attr_get          = osp_attr_get,
 669         .do_declare_attr_set  = osp_md_declare_attr_set,
 670         .do_attr_set          = osp_md_attr_set,
 671         .do_xattr_get         = osp_xattr_get,
 672         .do_declare_xattr_set = osp_declare_xattr_set,
 673         .do_xattr_set         = osp_xattr_set,
 674         .do_index_try         = osp_md_index_try,
 675         .do_object_lock       = osp_md_object_lock,
 676 };