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 <lustre_update.h>
  37 #include "osp_internal.h"
  38 static const char dot[] = ".";
  39 static const char dotdot[] = "..";
  40
  41 static int osp_prep_update_req(const struct lu_env *env,
  42                                struct osp_device *osp,
  43                                struct update_buf *ubuf, int ubuf_len,
  44                                struct ptlrpc_request **reqp)
  45 {
  46         struct obd_import      *imp;
  47         struct ptlrpc_request  *req;
  48         struct update_buf      *tmp;
  49         int                     rc;
  50         ENTRY;
  51
  52         imp = osp->opd_obd->u.cli.cl_import;
  53         LASSERT(imp);
  54
  55         req = ptlrpc_request_alloc(imp, &RQF_UPDATE_OBJ);
  56         if (req == NULL)
  57                 RETURN(-ENOMEM);
  58
  59         req_capsule_set_size(&req->rq_pill, &RMF_UPDATE, RCL_CLIENT,
  60                              UPDATE_BUFFER_SIZE);
  61
  62         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, UPDATE_OBJ);
  63         if (rc != 0) {
  64                 ptlrpc_req_finished(req);
  65                 RETURN(rc);
  66         }
  67
  68         req_capsule_set_size(&req->rq_pill, &RMF_UPDATE_REPLY, RCL_SERVER,
  69                              UPDATE_BUFFER_SIZE);
  70
  71         tmp = req_capsule_client_get(&req->rq_pill, &RMF_UPDATE);
  72         memcpy(tmp, ubuf, ubuf_len);
  73
  74         ptlrpc_request_set_replen(req);
  75
  76         *reqp = req;
  77
  78         RETURN(rc);
  79 }
  80
  81 static int osp_remote_sync(const struct lu_env *env, struct dt_device *dt,
  82                            struct update_request *update,
  83                            struct ptlrpc_request **reqp)
  84 {
  85         struct osp_device       *osp = dt2osp_dev(dt);
  86         struct ptlrpc_request   *req = NULL;
  87         int                     rc;
  88         ENTRY;
  89
  90         rc = osp_prep_update_req(env, osp, update->ur_buf, UPDATE_BUFFER_SIZE,
  91                                  &req);
  92         if (rc)
  93                 RETURN(rc);
  94
  95         /* Note: some dt index api might return non-zero result here, like
  96          * osd_index_ea_lookup, so we should only check rc < 0 here */
  97         rc = ptlrpc_queue_wait(req);
  98         if (rc < 0) {
  99                 ptlrpc_req_finished(req);
 100                 update->ur_rc = rc;
 101                 RETURN(rc);
 102         }
 103
 104         if (reqp != NULL) {
 105                 *reqp = req;
 106                 RETURN(rc);
 107         }
 108
 109         update->ur_rc = rc;
 110
 111         ptlrpc_req_finished(req);
 112
 113         RETURN(rc);
 114 }
 115
 116 /**
 117  * Create a new update request for the device.
 118  */
 119 static struct update_request
 120 *osp_create_update_req(struct dt_device *dt)
 121 {
 122         struct update_request *update;
 123
 124         OBD_ALLOC_PTR(update);
 125         if (!update)
 126                 return ERR_PTR(-ENOMEM);
 127
 128         OBD_ALLOC_LARGE(update->ur_buf, UPDATE_BUFFER_SIZE);
 129         if (update->ur_buf == NULL) {
 130                 OBD_FREE_PTR(update);
 131                 return ERR_PTR(-ENOMEM);
 132         }
 133
 134         CFS_INIT_LIST_HEAD(&update->ur_list);
 135         update->ur_dt = dt;
 136         update->ur_batchid = 0;
 137         update->ur_buf->ub_magic = UPDATE_BUFFER_MAGIC;
 138         update->ur_buf->ub_count = 0;
 139
 140         return update;
 141 }
 142
 143 static void osp_destroy_update_req(struct update_request *update)
 144 {
 145         if (update == NULL)
 146                 return;
 147
 148         cfs_list_del(&update->ur_list);
 149         if (update->ur_buf != NULL)
 150                 OBD_FREE_LARGE(update->ur_buf, UPDATE_BUFFER_SIZE);
 151
 152         OBD_FREE_PTR(update);
 153         return;
 154 }
 155
 156 int osp_trans_stop(const struct lu_env *env, struct thandle *th)
 157 {
 158         int rc = 0;
 159
 160         rc = th->th_current_request->ur_rc;
 161         osp_destroy_update_req(th->th_current_request);
 162         th->th_current_request = NULL;
 163
 164         return rc;
 165 }
 166
 167 /**
 168  * In DNE phase I, all remote updates will be packed into RPC (the format
 169  * description is in lustre_idl.h) during declare phase, all of updates
 170  * are attached to the transaction, one entry per OSP. Then in trans start,
 171  * LOD will walk through these entries and send these UPDATEs to the remote
 172  * MDT to be executed synchronously.
 173  */
 174 int osp_trans_start(const struct lu_env *env, struct dt_device *dt,
 175                     struct thandle *th)
 176 {
 177         struct update_request *update;
 178         int rc = 0;
 179
 180         /* In phase I, if the transaction includes remote updates, the local
 181          * update should be synchronized, so it will set th_sync = 1 */
 182         update = th->th_current_request;
 183         LASSERT(update != NULL && update->ur_dt == dt);
 184         if (update->ur_buf->ub_count > 0) {
 185                 rc = osp_remote_sync(env, dt, update, NULL);
 186                 th->th_sync = 1;
 187         }
 188
 189         RETURN(rc);
 190 }
 191
 192 /**
 193  * Insert the update into the th_bufs for the device.
 194  */
 195 static int osp_insert_update(const struct lu_env *env,
 196                              struct update_request *update, int op,
 197                              struct lu_fid *fid, int count,
 198                              int *lens, char **bufs)
 199 {
 200         struct update_buf    *ubuf = update->ur_buf;
 201         struct update        *obj_update;
 202         char                 *ptr;
 203         int                   i;
 204         int                   update_length;
 205         int                   rc = 0;
 206         ENTRY;
 207
 208         obj_update = (struct update *)((char *)ubuf +
 209                       cfs_size_round(update_buf_size(ubuf)));
 210
 211         /* Check update size to make sure it can fit into the buffer */
 212         update_length = cfs_size_round(offsetof(struct update,
 213                                        u_bufs[0]));
 214         for (i = 0; i < count; i++)
 215                 update_length += cfs_size_round(lens[i]);
 216
 217         if (cfs_size_round(update_buf_size(ubuf)) + update_length >
 218             UPDATE_BUFFER_SIZE || ubuf->ub_count >= UPDATE_MAX_OPS) {
 219                 CERROR("%s: insert up %p, idx %d cnt %d len %lu: rc = %d\n",
 220                         update->ur_dt->dd_lu_dev.ld_obd->obd_name, ubuf,
 221                         update_length, ubuf->ub_count, update_buf_size(ubuf),
 222                         -E2BIG);
 223                 RETURN(-E2BIG);
 224         }
 225
 226         if (count > UPDATE_BUF_COUNT) {
 227                 CERROR("%s: Insert too much params %d "DFID" op %d: rc = %d\n",
 228                         update->ur_dt->dd_lu_dev.ld_obd->obd_name, count,
 229                         PFID(fid), op, -E2BIG);
 230                 RETURN(-E2BIG);
 231         }
 232
 233         /* fill the update into the update buffer */
 234         fid_cpu_to_le(&obj_update->u_fid, fid);
 235         obj_update->u_type = cpu_to_le32(op);
 236         obj_update->u_batchid = update->ur_batchid;
 237         for (i = 0; i < count; i++)
 238                 obj_update->u_lens[i] = cpu_to_le32(lens[i]);
 239
 240         ptr = (char *)obj_update +
 241                         cfs_size_round(offsetof(struct update, u_bufs[0]));
 242         for (i = 0; i < count; i++)
 243                 LOGL(bufs[i], lens[i], ptr);
 244
 245         ubuf->ub_count++;
 246
 247         CDEBUG(D_INFO, "%s: %p "DFID" idx %d: op %d params %d:%lu\n",
 248                update->ur_dt->dd_lu_dev.ld_obd->obd_name, ubuf, PFID(fid),
 249                ubuf->ub_count, op, count, update_buf_size(ubuf));
 250
 251         RETURN(rc);
 252 }
 253
 254 static struct update_request
 255 *osp_find_update(struct thandle *th, struct dt_device *dt_dev)
 256 {
 257         struct update_request   *update;
 258
 259         /* Because transaction api does not proivde the interface
 260          * to transfer the update from LOD to OSP,  we need walk
 261          * remote update list to find the update, this probably
 262          * should move to LOD layer, when update can be part of
 263          * the trancation api parameter. XXX */
 264         cfs_list_for_each_entry(update, &th->th_remote_update_list, ur_list) {
 265                 if (update->ur_dt == dt_dev)
 266                         return update;
 267         }
 268         return NULL;
 269 }
 270
 271 static inline void osp_md_add_update_batchid(struct update_request *update)
 272 {
 273         update->ur_batchid++;
 274 }
 275
 276 /**
 277  * Find one loc in th_dev/dev_obj_update for the update,
 278  * Because only one thread can access this thandle, no need
 279  * lock now.
 280  */
 281 static struct update_request
 282 *osp_find_create_update_loc(struct thandle *th, struct dt_object *dt)
 283 {
 284         struct dt_device        *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 285         struct update_request   *update;
 286         ENTRY;
 287
 288         update = osp_find_update(th, dt_dev);
 289         if (update != NULL)
 290                 RETURN(update);
 291
 292         update = osp_create_update_req(dt_dev);
 293         if (IS_ERR(update))
 294                 RETURN(update);
 295
 296         cfs_list_add_tail(&update->ur_list, &th->th_remote_update_list);
 297
 298         RETURN(update);
 299 }
 300
 301 static int osp_get_attr_from_req(const struct lu_env *env,
 302                                  struct ptlrpc_request *req,
 303                                  struct lu_attr *attr, int index)
 304 {
 305         struct update_reply     *reply;
 306         struct obdo             *lobdo = &osp_env_info(env)->osi_obdo;
 307         struct obdo             *wobdo;
 308         int                     size;
 309
 310         LASSERT(attr != NULL);
 311
 312         reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
 313                                              UPDATE_BUFFER_SIZE);
 314         if (reply->ur_version != UPDATE_REPLY_V1)
 315                 return -EPROTO;
 316
 317         size = update_get_reply_buf(reply, (void **)&wobdo, index);
 318         if (size != sizeof(struct obdo))
 319                 return -EPROTO;
 320
 321         obdo_le_to_cpu(wobdo, wobdo);
 322         lustre_get_wire_obdo(NULL, lobdo, wobdo);
 323         la_from_obdo(attr, lobdo, lobdo->o_valid);
 324
 325         return 0;
 326 }
 327
 328 static int osp_md_declare_object_create(const struct lu_env *env,
 329                                         struct dt_object *dt,
 330                                         struct lu_attr *attr,
 331                                         struct dt_allocation_hint *hint,
 332                                         struct dt_object_format *dof,
 333                                         struct thandle *th)
 334 {
 335         struct osp_thread_info  *osi = osp_env_info(env);
 336         struct update_request   *update;
 337         struct lu_fid           *fid1;
 338         int                     sizes[2] = {sizeof(struct obdo), 0};
 339         char                    *bufs[2] = {NULL, NULL};
 340         int                     buf_count;
 341         int                     rc;
 342
 343
 344         update = osp_find_create_update_loc(th, dt);
 345         if (IS_ERR(update)) {
 346                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 347                        dt->do_lu.lo_dev->ld_obd->obd_name,
 348                        (int)PTR_ERR(update));
 349                 return PTR_ERR(update);
 350         }
 351
 352         osi->osi_obdo.o_valid = 0;
 353         LASSERT(S_ISDIR(attr->la_mode));
 354         obdo_from_la(&osi->osi_obdo, attr, attr->la_valid);
 355         lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
 356         obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
 357
 358         bufs[0] = (char *)&osi->osi_obdo;
 359         buf_count = 1;
 360         fid1 = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 361         if (hint->dah_parent) {
 362                 struct lu_fid *fid2;
 363                 struct lu_fid *tmp_fid = &osi->osi_fid;
 364
 365                 fid2 = (struct lu_fid *)lu_object_fid(&hint->dah_parent->do_lu);
 366                 fid_cpu_to_le(tmp_fid, fid2);
 367                 sizes[1] = sizeof(*tmp_fid);
 368                 bufs[1] = (char *)tmp_fid;
 369                 buf_count++;
 370         }
 371
 372         if (lu_object_exists(&dt->do_lu)) {
 373                 /* If the object already exists, we needs to destroy
 374                  * this orphan object first.
 375                  *
 376                  * The scenario might happen in this case
 377                  *
 378                  * 1. client send remote create to MDT0.
 379                  * 2. MDT0 send create update to MDT1.
 380                  * 3. MDT1 finished create synchronously.
 381                  * 4. MDT0 failed and reboot.
 382                  * 5. client resend remote create to MDT0.
 383                  * 6. MDT0 tries to resend create update to MDT1,
 384                  *    but find the object already exists
 385                  */
 386                 CDEBUG(D_HA, "%s: object "DFID" exists, destroy this orphan\n",
 387                        dt->do_lu.lo_dev->ld_obd->obd_name, PFID(fid1));
 388
 389                 rc = osp_insert_update(env, update, OBJ_REF_DEL, fid1, 0,
 390                                        NULL, NULL);
 391                 if (rc != 0)
 392                         GOTO(out, rc);
 393
 394                 if (S_ISDIR(lu_object_attr(&dt->do_lu))) {
 395                         /* decrease for ".." */
 396                         rc = osp_insert_update(env, update, OBJ_REF_DEL, fid1,
 397                                                0, NULL, NULL);
 398                         if (rc != 0)
 399                                 GOTO(out, rc);
 400                 }
 401
 402                 rc = osp_insert_update(env, update, OBJ_DESTROY, fid1, 0, NULL,
 403                                        NULL);
 404                 if (rc != 0)
 405                         GOTO(out, rc);
 406
 407                 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
 408                 /* Increase batchid to add this orphan object deletion
 409                  * to separate transaction */
 410                 osp_md_add_update_batchid(update);
 411         }
 412
 413         rc = osp_insert_update(env, update, OBJ_CREATE, fid1, buf_count, sizes,
 414                                bufs);
 415 out:
 416         if (rc)
 417                 CERROR("%s: Insert update error: rc = %d\n",
 418                        dt->do_lu.lo_dev->ld_obd->obd_name, rc);
 419
 420         return rc;
 421 }
 422
 423 static int osp_md_object_create(const struct lu_env *env, struct dt_object *dt,
 424                                 struct lu_attr *attr,
 425                                 struct dt_allocation_hint *hint,
 426                                 struct dt_object_format *dof,
 427                                 struct thandle *th)
 428 {
 429         struct osp_object  *obj = dt2osp_obj(dt);
 430
 431         CDEBUG(D_INFO, "create object "DFID"\n",
 432                PFID(&dt->do_lu.lo_header->loh_fid));
 433
 434         /* Because the create update RPC will be sent during declare phase,
 435          * if creation reaches here, it means the object has been created
 436          * successfully */
 437         dt->do_lu.lo_header->loh_attr |= LOHA_EXISTS | (attr->la_mode & S_IFMT);
 438         obj->opo_empty = 1;
 439
 440         return 0;
 441 }
 442
 443 static int osp_md_declare_object_ref_del(const struct lu_env *env,
 444                                          struct dt_object *dt,
 445                                          struct thandle *th)
 446 {
 447         struct update_request   *update;
 448         struct lu_fid           *fid;
 449         int                     rc;
 450
 451         update = osp_find_create_update_loc(th, dt);
 452         if (IS_ERR(update)) {
 453                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 454                        dt->do_lu.lo_dev->ld_obd->obd_name,
 455                       (int)PTR_ERR(update));
 456                 return PTR_ERR(update);
 457         }
 458
 459         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 460
 461         rc = osp_insert_update(env, update, OBJ_REF_DEL, fid, 0, NULL, NULL);
 462
 463         return rc;
 464 }
 465
 466 static int osp_md_object_ref_del(const struct lu_env *env,
 467                                  struct dt_object *dt,
 468                                  struct thandle *th)
 469 {
 470         CDEBUG(D_INFO, "ref del object "DFID"\n",
 471                PFID(&dt->do_lu.lo_header->loh_fid));
 472
 473         return 0;
 474 }
 475
 476 static int osp_md_declare_ref_add(const struct lu_env *env,
 477                                   struct dt_object *dt, struct thandle *th)
 478 {
 479         struct update_request   *update;
 480         struct lu_fid           *fid;
 481         int                     rc;
 482
 483         update = osp_find_create_update_loc(th, dt);
 484         if (IS_ERR(update)) {
 485                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 486                        dt->do_lu.lo_dev->ld_obd->obd_name,
 487                        (int)PTR_ERR(update));
 488                 return PTR_ERR(update);
 489         }
 490
 491         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 492
 493         rc = osp_insert_update(env, update, OBJ_REF_ADD, fid, 0, NULL, NULL);
 494
 495         return rc;
 496 }
 497
 498 static int osp_md_object_ref_add(const struct lu_env *env,
 499                                  struct dt_object *dt,
 500                                  struct thandle *th)
 501 {
 502         CDEBUG(D_INFO, "ref add object "DFID"\n",
 503                PFID(&dt->do_lu.lo_header->loh_fid));
 504
 505         return 0;
 506 }
 507
 508 static void osp_md_ah_init(const struct lu_env *env,
 509                            struct dt_allocation_hint *ah,
 510                            struct dt_object *parent,
 511                            struct dt_object *child,
 512                            umode_t child_mode)
 513 {
 514         LASSERT(ah);
 515
 516         memset(ah, 0, sizeof(*ah));
 517         ah->dah_parent = parent;
 518         ah->dah_mode = child_mode;
 519 }
 520
 521 static int osp_md_declare_attr_set(const struct lu_env *env,
 522                                    struct dt_object *dt,
 523                                    const struct lu_attr *attr,
 524                                    struct thandle *th)
 525 {
 526         struct osp_thread_info *osi = osp_env_info(env);
 527         struct update_request  *update;
 528         struct lu_fid          *fid;
 529         int                     size = sizeof(struct obdo);
 530         char                   *buf;
 531         int                     rc;
 532
 533         update = osp_find_create_update_loc(th, dt);
 534         if (IS_ERR(update)) {
 535                 CERROR("%s: Get OSP update buf failed: %d\n",
 536                        dt->do_lu.lo_dev->ld_obd->obd_name,
 537                        (int)PTR_ERR(update));
 538                 return PTR_ERR(update);
 539         }
 540
 541         osi->osi_obdo.o_valid = 0;
 542         LASSERT(!(attr->la_valid & (LA_MODE | LA_TYPE)));
 543         obdo_from_la(&osi->osi_obdo, (struct lu_attr *)attr,
 544                      attr->la_valid);
 545         lustre_set_wire_obdo(NULL, &osi->osi_obdo, &osi->osi_obdo);
 546         obdo_cpu_to_le(&osi->osi_obdo, &osi->osi_obdo);
 547
 548         buf = (char *)&osi->osi_obdo;
 549         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 550
 551         rc = osp_insert_update(env, update, OBJ_ATTR_SET, fid, 1, &size, &buf);
 552
 553         return rc;
 554 }
 555
 556 static int osp_md_attr_set(const struct lu_env *env, struct dt_object *dt,
 557                            const struct lu_attr *attr, struct thandle *th,
 558                            struct lustre_capa *capa)
 559 {
 560         CDEBUG(D_INFO, "attr set object "DFID"\n",
 561                PFID(&dt->do_lu.lo_header->loh_fid));
 562
 563         RETURN(0);
 564 }
 565
 566 static int osp_md_declare_xattr_set(const struct lu_env *env,
 567                                     struct dt_object *dt,
 568                                     const struct lu_buf *buf,
 569                                     const char *name, int flag,
 570                                     struct thandle *th)
 571 {
 572         struct update_request   *update;
 573         struct lu_fid           *fid;
 574         int                     sizes[3] = {strlen(name), buf->lb_len,
 575                                             sizeof(int)};
 576         char                    *bufs[3] = {(char *)name, (char *)buf->lb_buf };
 577         int                     rc;
 578
 579         LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
 580         update = osp_find_create_update_loc(th, dt);
 581         if (IS_ERR(update)) {
 582                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 583                        dt->do_lu.lo_dev->ld_obd->obd_name,
 584                        (int)PTR_ERR(update));
 585                 return PTR_ERR(update);
 586         }
 587
 588         flag = cpu_to_le32(flag);
 589         bufs[2] = (char *)&flag;
 590
 591         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 592         rc = osp_insert_update(env, update, OBJ_XATTR_SET, fid,
 593                                ARRAY_SIZE(sizes), sizes, bufs);
 594
 595         return rc;
 596 }
 597
 598 static int osp_md_xattr_set(const struct lu_env *env, struct dt_object *dt,
 599                             const struct lu_buf *buf, const char *name, int fl,
 600                             struct thandle *th, struct lustre_capa *capa)
 601 {
 602         CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
 603                PFID(&dt->do_lu.lo_header->loh_fid));
 604
 605         return 0;
 606 }
 607
 608 static int osp_md_xattr_get(const struct lu_env *env, struct dt_object *dt,
 609                             struct lu_buf *buf, const char *name,
 610                             struct lustre_capa *capa)
 611 {
 612         struct dt_device        *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 613         struct update_request   *update = NULL;
 614         struct ptlrpc_request   *req = NULL;
 615         int                     rc;
 616         int                     buf_len;
 617         int                     size;
 618         struct update_reply     *reply;
 619         void                    *ea_buf;
 620         ENTRY;
 621
 622         /* Because it needs send the update buffer right away,
 623          * just create an update buffer, instead of attaching the
 624          * update_remote list of the thandle.
 625          */
 626         update = osp_create_update_req(dt_dev);
 627         if (IS_ERR(update))
 628                 RETURN(PTR_ERR(update));
 629
 630         LASSERT(name != NULL);
 631         buf_len = strlen(name);
 632         rc = osp_insert_update(env, update, OBJ_XATTR_GET,
 633                                (struct lu_fid *)lu_object_fid(&dt->do_lu),
 634                                1, &buf_len, (char **)&name);
 635         if (rc != 0) {
 636                 CERROR("%s: Insert update error: rc = %d\n",
 637                        dt->do_lu.lo_dev->ld_obd->obd_name, rc);
 638                 GOTO(out, rc);
 639         }
 640         dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 641
 642         rc = osp_remote_sync(env, dt_dev, update, &req);
 643         if (rc != 0)
 644                 GOTO(out, rc);
 645
 646         reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
 647                                             UPDATE_BUFFER_SIZE);
 648         if (reply->ur_version != UPDATE_REPLY_V1) {
 649                 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
 650                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 651                        reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
 652                 GOTO(out, rc = -EPROTO);
 653         }
 654
 655         size = update_get_reply_buf(reply, &ea_buf, 0);
 656         if (size < 0)
 657                 GOTO(out, rc = size);
 658
 659         LASSERT(size > 0 && size < PAGE_CACHE_SIZE);
 660         LASSERT(ea_buf != NULL);
 661
 662         rc = size;
 663         if (buf->lb_buf != NULL)
 664                 memcpy(buf->lb_buf, ea_buf, size);
 665 out:
 666         if (req != NULL)
 667                 ptlrpc_req_finished(req);
 668
 669         if (update != NULL)
 670                 osp_destroy_update_req(update);
 671
 672         RETURN(rc);
 673 }
 674
 675 static void osp_md_object_read_lock(const struct lu_env *env,
 676                                     struct dt_object *dt, unsigned role)
 677 {
 678         struct osp_object  *obj = dt2osp_obj(dt);
 679
 680         LASSERT(obj->opo_owner != env);
 681         down_read_nested(&obj->opo_sem, role);
 682
 683         LASSERT(obj->opo_owner == NULL);
 684 }
 685
 686 static void osp_md_object_write_lock(const struct lu_env *env,
 687                                      struct dt_object *dt, unsigned role)
 688 {
 689         struct osp_object *obj = dt2osp_obj(dt);
 690
 691         down_write_nested(&obj->opo_sem, role);
 692
 693         LASSERT(obj->opo_owner == NULL);
 694         obj->opo_owner = env;
 695 }
 696
 697 static void osp_md_object_read_unlock(const struct lu_env *env,
 698                                       struct dt_object *dt)
 699 {
 700         struct osp_object *obj = dt2osp_obj(dt);
 701
 702         up_read(&obj->opo_sem);
 703 }
 704
 705 static void osp_md_object_write_unlock(const struct lu_env *env,
 706                                        struct dt_object *dt)
 707 {
 708         struct osp_object *obj = dt2osp_obj(dt);
 709
 710         LASSERT(obj->opo_owner == env);
 711         obj->opo_owner = NULL;
 712         up_write(&obj->opo_sem);
 713 }
 714
 715 static int osp_md_object_write_locked(const struct lu_env *env,
 716                                       struct dt_object *dt)
 717 {
 718         struct osp_object *obj = dt2osp_obj(dt);
 719
 720         return obj->opo_owner == env;
 721 }
 722
 723 static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
 724                                struct dt_rec *rec, const struct dt_key *key,
 725                                struct lustre_capa *capa)
 726 {
 727         struct dt_device        *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
 728         struct update_request   *update;
 729         struct ptlrpc_request   *req = NULL;
 730         int                     size = strlen((char *)key) + 1;
 731         char                    *name = (char *)key;
 732         int                     rc;
 733         struct update_reply     *reply;
 734         struct lu_fid           *fid;
 735
 736         ENTRY;
 737
 738         /* Because it needs send the update buffer right away,
 739          * just create an update buffer, instead of attaching the
 740          * update_remote list of the thandle.
 741          */
 742         update = osp_create_update_req(dt_dev);
 743         if (IS_ERR(update))
 744                 RETURN(PTR_ERR(update));
 745
 746         rc = osp_insert_update(env, update, OBJ_INDEX_LOOKUP,
 747                                (struct lu_fid *)lu_object_fid(&dt->do_lu),
 748                                1, &size, (char **)&name);
 749         if (rc) {
 750                 CERROR("%s: Insert update error: rc = %d\n",
 751                        dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
 752                 GOTO(out, rc);
 753         }
 754
 755         rc = osp_remote_sync(env, dt_dev, update, &req);
 756         if (rc < 0) {
 757                 CERROR("%s: lookup "DFID" %s failed: rc = %d\n",
 758                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 759                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
 760                 GOTO(out, rc);
 761         }
 762
 763         reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
 764                                              UPDATE_BUFFER_SIZE);
 765         if (reply->ur_version != UPDATE_REPLY_V1) {
 766                 CERROR("%s: Wrong version %x expected %x: rc = %d\n",
 767                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 768                        reply->ur_version, UPDATE_REPLY_V1, -EPROTO);
 769                 GOTO(out, rc = -EPROTO);
 770         }
 771
 772         rc = update_get_reply_result(reply, NULL, 0);
 773         if (rc < 0) {
 774                 CERROR("%s: wrong version lookup "DFID" %s: rc = %d\n",
 775                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 776                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, rc);
 777                 GOTO(out, rc);
 778         }
 779
 780         size = update_get_reply_buf(reply, (void **)&fid, 0);
 781         if (size < 0)
 782                 GOTO(out, rc = size);
 783
 784         if (size != sizeof(struct lu_fid)) {
 785                 CERROR("%s: lookup "DFID" %s wrong size %d: rc = %d\n",
 786                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 787                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, size, rc);
 788                 GOTO(out, rc = -EINVAL);
 789         }
 790
 791         fid_le_to_cpu(fid, fid);
 792         if (!fid_is_sane(fid)) {
 793                 CERROR("%s: lookup "DFID" %s invalid fid "DFID": rc = %d\n",
 794                        dt_dev->dd_lu_dev.ld_obd->obd_name,
 795                        PFID(lu_object_fid(&dt->do_lu)), (char *)key, PFID(fid),
 796                        rc);
 797                 GOTO(out, rc = -EINVAL);
 798         }
 799         memcpy(rec, fid, sizeof(*fid));
 800 out:
 801         if (req != NULL)
 802                 ptlrpc_req_finished(req);
 803
 804         if (update != NULL)
 805                 osp_destroy_update_req(update);
 806
 807         RETURN(rc);
 808 }
 809
 810 static int osp_md_declare_insert(const struct lu_env *env,
 811                                  struct dt_object *dt,
 812                                  const struct dt_rec *rec,
 813                                  const struct dt_key *key,
 814                                  struct thandle *th)
 815 {
 816         struct update_request   *update;
 817         struct lu_fid           *fid;
 818         struct lu_fid           *rec_fid = (struct lu_fid *)rec;
 819         int                     size[2] = {strlen((char *)key) + 1,
 820                                                   sizeof(*rec_fid)};
 821         char                    *bufs[2] = {(char *)key, (char *)rec_fid};
 822         int                     rc;
 823
 824         update = osp_find_create_update_loc(th, dt);
 825         if (IS_ERR(update)) {
 826                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 827                        dt->do_lu.lo_dev->ld_obd->obd_name,
 828                        (int)PTR_ERR(update));
 829                 return PTR_ERR(update);
 830         }
 831
 832         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 833
 834         CDEBUG(D_INFO, "%s: insert index of "DFID" %s: "DFID"\n",
 835                dt->do_lu.lo_dev->ld_obd->obd_name,
 836                PFID(fid), (char *)key, PFID(rec_fid));
 837
 838         fid_cpu_to_le(rec_fid, rec_fid);
 839
 840         rc = osp_insert_update(env, update, OBJ_INDEX_INSERT, fid,
 841                                ARRAY_SIZE(size), size, bufs);
 842         return rc;
 843 }
 844
 845 static int osp_md_index_insert(const struct lu_env *env,
 846                                struct dt_object *dt,
 847                                const struct dt_rec *rec,
 848                                const struct dt_key *key,
 849                                struct thandle *th,
 850                                struct lustre_capa *capa,
 851                                int ignore_quota)
 852 {
 853         return 0;
 854 }
 855
 856 static int osp_md_declare_delete(const struct lu_env *env,
 857                                  struct dt_object *dt,
 858                                  const struct dt_key *key,
 859                                  struct thandle *th)
 860 {
 861         struct update_request *update;
 862         struct lu_fid *fid;
 863         int size = strlen((char *)key) + 1;
 864         char *buf = (char *)key;
 865         int rc;
 866
 867         update = osp_find_create_update_loc(th, dt);
 868         if (IS_ERR(update)) {
 869                 CERROR("%s: Get OSP update buf failed: rc = %d\n",
 870                        dt->do_lu.lo_dev->ld_obd->obd_name,
 871                        (int)PTR_ERR(update));
 872                 return PTR_ERR(update);
 873         }
 874
 875         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 876
 877         rc = osp_insert_update(env, update, OBJ_INDEX_DELETE, fid, 1, &size,
 878                                &buf);
 879
 880         return rc;
 881 }
 882
 883 static int osp_md_index_delete(const struct lu_env *env,
 884                                struct dt_object *dt,
 885                                const struct dt_key *key,
 886                                struct thandle *th,
 887                                struct lustre_capa *capa)
 888 {
 889         CDEBUG(D_INFO, "index delete "DFID" %s\n",
 890                PFID(&dt->do_lu.lo_header->loh_fid), (char *)key);
 891
 892         return 0;
 893 }
 894
 895 /**
 896  * Creates or initializes iterator context.
 897  *
 898  * Note: for OSP, these index iterate api is only used to check
 899  * whether the directory is empty now (see mdd_dir_is_empty).
 900  * Since dir_empty will be return by OBJ_ATTR_GET(see osp_md_attr_get/
 901  * out_attr_get). So the implementation of these iterator is simplied
 902  * to make mdd_dir_is_empty happy. The real iterator should be
 903  * implemented, if we need it one day.
 904  */
 905 static struct dt_it *osp_it_init(const struct lu_env *env,
 906                                  struct dt_object *dt,
 907                                  __u32 attr,
 908                                 struct lustre_capa *capa)
 909 {
 910         lu_object_get(&dt->do_lu);
 911         return (struct dt_it *)dt;
 912 }
 913
 914 static void osp_it_fini(const struct lu_env *env, struct dt_it *di)
 915 {
 916         struct dt_object *dt = (struct dt_object *)di;
 917         lu_object_put(env, &dt->do_lu);
 918 }
 919
 920 static int osp_it_get(const struct lu_env *env,
 921                       struct dt_it *di, const struct dt_key *key)
 922 {
 923         return 1;
 924 }
 925
 926 static void osp_it_put(const struct lu_env *env, struct dt_it *di)
 927 {
 928         return;
 929 }
 930
 931 static int osp_it_next(const struct lu_env *env, struct dt_it *di)
 932 {
 933         struct dt_object *dt = (struct dt_object *)di;
 934         struct osp_object *o = dt2osp_obj(dt);
 935
 936         if (o->opo_empty)
 937                 return 1;
 938
 939         return 0;
 940 }
 941
 942 static struct dt_key *osp_it_key(const struct lu_env *env,
 943                                  const struct dt_it *di)
 944 {
 945         LBUG();
 946         return NULL;
 947 }
 948
 949 static int osp_it_key_size(const struct lu_env *env, const struct dt_it *di)
 950 {
 951         LBUG();
 952         return 0;
 953 }
 954
 955 static int osp_it_rec(const struct lu_env *env, const struct dt_it *di,
 956                       struct dt_rec *lde, __u32 attr)
 957 {
 958         LBUG();
 959         return 0;
 960 }
 961
 962 static __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
 963 {
 964         LBUG();
 965         return 0;
 966 }
 967
 968 static int osp_it_load(const struct lu_env *env, const struct dt_it *di,
 969                        __u64 hash)
 970 {
 971         LBUG();
 972         return 0;
 973 }
 974
 975 static int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
 976                           void *key_rec)
 977 {
 978         LBUG();
 979         return 0;
 980 }
 981
 982 static const struct dt_index_operations osp_md_index_ops = {
 983         .dio_lookup         = osp_md_index_lookup,
 984         .dio_declare_insert = osp_md_declare_insert,
 985         .dio_insert         = osp_md_index_insert,
 986         .dio_declare_delete = osp_md_declare_delete,
 987         .dio_delete         = osp_md_index_delete,
 988         .dio_it     = {
 989                 .init     = osp_it_init,
 990                 .fini     = osp_it_fini,
 991                 .get      = osp_it_get,
 992                 .put      = osp_it_put,
 993                 .next     = osp_it_next,
 994                 .key      = osp_it_key,
 995                 .key_size = osp_it_key_size,
 996                 .rec      = osp_it_rec,
 997                 .store    = osp_it_store,
 998                 .load     = osp_it_load,
 999                 .key_rec  = osp_it_key_rec,
1000         }
1001 };
1002
1003 static int osp_md_index_try(const struct lu_env *env,
1004                             struct dt_object *dt,
1005                             const struct dt_index_features *feat)
1006 {
1007         dt->do_index_ops = &osp_md_index_ops;
1008         return 0;
1009 }
1010
1011 static int osp_md_attr_get(const struct lu_env *env,
1012                            struct dt_object *dt, struct lu_attr *attr,
1013                            struct lustre_capa *capa)
1014 {
1015         struct osp_object     *obj = dt2osp_obj(dt);
1016         struct dt_device      *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1017         struct update_request *update = NULL;
1018         struct ptlrpc_request *req = NULL;
1019         int rc;
1020         ENTRY;
1021
1022         /* Because it needs send the update buffer right away,
1023          * just create an update buffer, instead of attaching the
1024          * update_remote list of the thandle.
1025          */
1026         update = osp_create_update_req(dt_dev);
1027         if (IS_ERR(update))
1028                 RETURN(PTR_ERR(update));
1029
1030         rc = osp_insert_update(env, update, OBJ_ATTR_GET,
1031                                (struct lu_fid *)lu_object_fid(&dt->do_lu),
1032                                0, NULL, NULL);
1033         if (rc) {
1034                 CERROR("%s: Insert update error: rc = %d\n",
1035                        dt_dev->dd_lu_dev.ld_obd->obd_name, rc);
1036                 GOTO(out, rc);
1037         }
1038         dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1039
1040         rc = osp_remote_sync(env, dt_dev, update, &req);
1041         if (rc < 0)
1042                 GOTO(out, rc);
1043
1044         rc = osp_get_attr_from_req(env, req, attr, 0);
1045         if (rc)
1046                 GOTO(out, rc);
1047
1048         if (attr->la_flags == 1)
1049                 obj->opo_empty = 0;
1050         else
1051                 obj->opo_empty = 1;
1052 out:
1053         if (req != NULL)
1054                 ptlrpc_req_finished(req);
1055
1056         if (update != NULL)
1057                 osp_destroy_update_req(update);
1058
1059         RETURN(rc);
1060 }
1061
1062 static int osp_md_declare_object_destroy(const struct lu_env *env,
1063                                          struct dt_object *dt,
1064                                          struct thandle *th)
1065 {
1066         struct osp_object  *o = dt2osp_obj(dt);
1067         int                 rc = 0;
1068         ENTRY;
1069
1070         /*
1071          * track objects to be destroyed via llog
1072          */
1073         rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1074
1075         RETURN(rc);
1076 }
1077
1078 static int osp_md_object_destroy(const struct lu_env *env,
1079                                  struct dt_object *dt, struct thandle *th)
1080 {
1081         struct osp_object  *o = dt2osp_obj(dt);
1082         int                 rc = 0;
1083         ENTRY;
1084
1085         /*
1086          * once transaction is committed put proper command on
1087          * the queue going to our OST
1088          */
1089         rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1090
1091         /* not needed in cache any more */
1092         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1093
1094         RETURN(rc);
1095 }
1096
1097 static int osp_md_object_lock(const struct lu_env *env,
1098                               struct dt_object *dt,
1099                               struct lustre_handle *lh,
1100                               struct ldlm_enqueue_info *einfo,
1101                               void *policy)
1102 {
1103         struct osp_thread_info *info = osp_env_info(env);
1104         struct ldlm_res_id     *res_id = &info->osi_resid;
1105         struct dt_device       *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
1106         struct osp_device      *osp = dt2osp_dev(dt_dev);
1107         struct ptlrpc_request  *req = NULL;
1108         int                     rc = 0;
1109         __u64                   flags = 0;
1110         ldlm_mode_t             mode;
1111
1112         fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
1113
1114         mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
1115                                LDLM_FL_BLOCK_GRANTED, res_id,
1116                                einfo->ei_type,
1117                                (ldlm_policy_data_t *)policy,
1118                                einfo->ei_mode, lh, 0);
1119         if (mode > 0)
1120                 return ELDLM_OK;
1121
1122         req = ldlm_enqueue_pack(osp->opd_exp, 0);
1123         if (IS_ERR(req))
1124                 RETURN(PTR_ERR(req));
1125
1126         rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
1127                               (const ldlm_policy_data_t *)policy,
1128                               &flags, NULL, 0, LVB_T_NONE, lh, 0);
1129
1130         ptlrpc_req_finished(req);
1131
1132         return rc == ELDLM_OK ? 0 : -EIO;
1133 }
1134
1135 struct dt_object_operations osp_md_obj_ops = {
1136         .do_read_lock         = osp_md_object_read_lock,
1137         .do_write_lock        = osp_md_object_write_lock,
1138         .do_read_unlock       = osp_md_object_read_unlock,
1139         .do_write_unlock      = osp_md_object_write_unlock,
1140         .do_write_locked      = osp_md_object_write_locked,
1141         .do_declare_create    = osp_md_declare_object_create,
1142         .do_create            = osp_md_object_create,
1143         .do_declare_ref_add   = osp_md_declare_ref_add,
1144         .do_ref_add           = osp_md_object_ref_add,
1145         .do_declare_ref_del   = osp_md_declare_object_ref_del,
1146         .do_ref_del           = osp_md_object_ref_del,
1147         .do_declare_destroy   = osp_md_declare_object_destroy,
1148         .do_destroy           = osp_md_object_destroy,
1149         .do_ah_init           = osp_md_ah_init,
1150         .do_attr_get          = osp_md_attr_get,
1151         .do_declare_attr_set  = osp_md_declare_attr_set,
1152         .do_attr_set          = osp_md_attr_set,
1153         .do_declare_xattr_set = osp_md_declare_xattr_set,
1154         .do_xattr_set         = osp_md_xattr_set,
1155         .do_xattr_get         = osp_md_xattr_get,
1156         .do_index_try         = osp_md_index_try,
1157         .do_object_lock       = osp_md_object_lock,
1158 };