lustre/osp/osp_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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/osp/osp_object.c
  37  *
  38  * Lustre OST Proxy Device
  39  *
  40  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  41  * Author: Mikhail Pershin <mike.tappro@intel.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_MDS
  45
  46 #include "osp_internal.h"
  47
  48 static inline bool is_ost_obj(struct lu_object *lo)
  49 {
  50         return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
  51 }
  52
  53 static void osp_object_assign_fid(const struct lu_env *env,
  54                                  struct osp_device *d, struct osp_object *o)
  55 {
  56         struct osp_thread_info *osi = osp_env_info(env);
  57
  58         LASSERT(fid_is_zero(lu_object_fid(&o->opo_obj.do_lu)));
  59         LASSERT(o->opo_reserved);
  60         o->opo_reserved = 0;
  61
  62         osp_precreate_get_fid(env, d, &osi->osi_fid);
  63
  64         lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
  65 }
  66
  67 static int osp_oac_init(struct osp_object *obj)
  68 {
  69         struct osp_object_attr *ooa;
  70
  71         OBD_ALLOC_PTR(ooa);
  72         if (ooa == NULL)
  73                 return -ENOMEM;
  74
  75         INIT_LIST_HEAD(&ooa->ooa_xattr_list);
  76         spin_lock(&obj->opo_lock);
  77         if (likely(obj->opo_ooa == NULL)) {
  78                 obj->opo_ooa = ooa;
  79                 spin_unlock(&obj->opo_lock);
  80         } else {
  81                 spin_unlock(&obj->opo_lock);
  82                 OBD_FREE_PTR(ooa);
  83         }
  84
  85         return 0;
  86 }
  87
  88 static struct osp_xattr_entry *
  89 osp_oac_xattr_find_locked(struct osp_object_attr *ooa,
  90                           const char *name, int namelen, bool unlink)
  91 {
  92         struct osp_xattr_entry *oxe;
  93
  94         list_for_each_entry(oxe, &ooa->ooa_xattr_list, oxe_list) {
  95                 if (namelen == oxe->oxe_namelen &&
  96                     strncmp(name, oxe->oxe_buf, namelen) == 0) {
  97                         if (unlink)
  98                                 list_del_init(&oxe->oxe_list);
  99                         else
 100                                 atomic_inc(&oxe->oxe_ref);
 101
 102                         return oxe;
 103                 }
 104         }
 105
 106         return NULL;
 107 }
 108
 109 static struct osp_xattr_entry *osp_oac_xattr_find(struct osp_object *obj,
 110                                                   const char *name)
 111 {
 112         struct osp_xattr_entry *oxe = NULL;
 113
 114         spin_lock(&obj->opo_lock);
 115         if (obj->opo_ooa != NULL)
 116                 oxe = osp_oac_xattr_find_locked(obj->opo_ooa, name,
 117                                                 strlen(name), false);
 118         spin_unlock(&obj->opo_lock);
 119
 120         return oxe;
 121 }
 122
 123 static struct osp_xattr_entry *
 124 osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, int len)
 125 {
 126         struct osp_object_attr *ooa     = obj->opo_ooa;
 127         struct osp_xattr_entry *oxe;
 128         struct osp_xattr_entry *tmp     = NULL;
 129         int                     namelen = strlen(name);
 130         int                     size    = sizeof(*oxe) + namelen + 1 + len;
 131
 132         LASSERT(ooa != NULL);
 133
 134         oxe = osp_oac_xattr_find(obj, name);
 135         if (oxe != NULL)
 136                 return oxe;
 137
 138         OBD_ALLOC(oxe, size);
 139         if (unlikely(oxe == NULL))
 140                 return NULL;
 141
 142         INIT_LIST_HEAD(&oxe->oxe_list);
 143         oxe->oxe_buflen = size;
 144         oxe->oxe_namelen = namelen;
 145         memcpy(oxe->oxe_buf, name, namelen);
 146         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
 147         /* One ref is for the caller, the other is for the entry on the list. */
 148         atomic_set(&oxe->oxe_ref, 2);
 149
 150         spin_lock(&obj->opo_lock);
 151         tmp = osp_oac_xattr_find_locked(ooa, name, namelen, false);
 152         if (tmp == NULL)
 153                 list_add_tail(&oxe->oxe_list, &ooa->ooa_xattr_list);
 154         spin_unlock(&obj->opo_lock);
 155
 156         if (tmp != NULL) {
 157                 OBD_FREE(oxe, size);
 158                 oxe = tmp;
 159         }
 160
 161         return oxe;
 162 }
 163
 164 static struct osp_xattr_entry *
 165 osp_oac_xattr_replace(struct osp_object *obj,
 166                       struct osp_xattr_entry **poxe, int len)
 167 {
 168         struct osp_object_attr *ooa     = obj->opo_ooa;
 169         struct osp_xattr_entry *old     = *poxe;
 170         struct osp_xattr_entry *oxe;
 171         struct osp_xattr_entry *tmp     = NULL;
 172         int                     namelen = old->oxe_namelen;
 173         int                     size    = sizeof(*oxe) + namelen + 1 + len;
 174
 175         LASSERT(ooa != NULL);
 176
 177         OBD_ALLOC(oxe, size);
 178         if (unlikely(oxe == NULL))
 179                 return NULL;
 180
 181         INIT_LIST_HEAD(&oxe->oxe_list);
 182         oxe->oxe_buflen = size;
 183         oxe->oxe_namelen = namelen;
 184         memcpy(oxe->oxe_buf, old->oxe_buf, namelen);
 185         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
 186         /* One ref is for the caller, the other is for the entry on the list. */
 187         atomic_set(&oxe->oxe_ref, 2);
 188
 189         spin_lock(&obj->opo_lock);
 190         tmp = osp_oac_xattr_find_locked(ooa, oxe->oxe_buf, namelen, true);
 191         list_add_tail(&oxe->oxe_list, &ooa->ooa_xattr_list);
 192         spin_unlock(&obj->opo_lock);
 193
 194         *poxe = tmp;
 195         LASSERT(tmp != NULL);
 196
 197         return oxe;
 198 }
 199
 200 static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
 201 {
 202         if (atomic_dec_and_test(&oxe->oxe_ref)) {
 203                 LASSERT(list_empty(&oxe->oxe_list));
 204
 205                 OBD_FREE(oxe, oxe->oxe_buflen);
 206         }
 207 }
 208
 209 static int osp_get_attr_from_reply(const struct lu_env *env,
 210                                    struct object_update_reply *reply,
 211                                    struct lu_attr *attr,
 212                                    struct osp_object *obj, int index)
 213 {
 214         struct osp_thread_info  *osi    = osp_env_info(env);
 215         struct lu_buf           *rbuf   = &osi->osi_lb2;
 216         struct obdo             *lobdo  = &osi->osi_obdo;
 217         struct obdo             *wobdo;
 218         int                     rc;
 219
 220         rc = object_update_result_data_get(reply, rbuf, index);
 221         if (rc < 0)
 222                 return rc;
 223
 224         wobdo = rbuf->lb_buf;
 225         if (rbuf->lb_len != sizeof(*wobdo))
 226                 return -EPROTO;
 227
 228         obdo_le_to_cpu(wobdo, wobdo);
 229         lustre_get_wire_obdo(NULL, lobdo, wobdo);
 230         spin_lock(&obj->opo_lock);
 231         if (obj->opo_ooa != NULL) {
 232                 la_from_obdo(&obj->opo_ooa->ooa_attr, lobdo, lobdo->o_valid);
 233                 if (attr != NULL)
 234                         *attr = obj->opo_ooa->ooa_attr;
 235         } else {
 236                 LASSERT(attr != NULL);
 237
 238                 la_from_obdo(attr, lobdo, lobdo->o_valid);
 239         }
 240         spin_unlock(&obj->opo_lock);
 241
 242         return 0;
 243 }
 244
 245 static int osp_attr_get_interpterer(const struct lu_env *env,
 246                                     struct object_update_reply *reply,
 247                                     struct osp_object *obj,
 248                                     void *data, int index, int rc)
 249 {
 250         struct lu_attr *attr = data;
 251
 252         LASSERT(obj->opo_ooa != NULL);
 253
 254         if (rc == 0) {
 255                 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
 256                 obj->opo_non_exist = 0;
 257
 258                 return osp_get_attr_from_reply(env, reply, NULL, obj, index);
 259         } else {
 260                 if (rc == -ENOENT) {
 261                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
 262                         obj->opo_non_exist = 1;
 263                 }
 264
 265                 spin_lock(&obj->opo_lock);
 266                 attr->la_valid = 0;
 267                 spin_unlock(&obj->opo_lock);
 268         }
 269
 270         return 0;
 271 }
 272
 273 static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
 274                                 struct lustre_capa *capa)
 275 {
 276         struct osp_object       *obj    = dt2osp_obj(dt);
 277         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 278         struct dt_update_request *update;
 279         int                      rc     = 0;
 280
 281         if (obj->opo_ooa == NULL) {
 282                 rc = osp_oac_init(obj);
 283                 if (rc != 0)
 284                         return rc;
 285         }
 286
 287         mutex_lock(&osp->opd_async_requests_mutex);
 288         update = osp_find_or_create_async_update_request(osp);
 289         if (IS_ERR(update))
 290                 rc = PTR_ERR(update);
 291         else
 292                 rc = osp_insert_async_update(env, update, OUT_ATTR_GET, obj, 0,
 293                                              NULL, NULL,
 294                                              &obj->opo_ooa->ooa_attr,
 295                                              osp_attr_get_interpterer);
 296         mutex_unlock(&osp->opd_async_requests_mutex);
 297
 298         return rc;
 299 }
 300
 301 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
 302                  struct lu_attr *attr, struct lustre_capa *capa)
 303 {
 304         struct osp_device               *osp = lu2osp_dev(dt->do_lu.lo_dev);
 305         struct osp_object               *obj = dt2osp_obj(dt);
 306         struct dt_device                *dev = &osp->opd_dt_dev;
 307         struct dt_update_request        *update;
 308         struct object_update_reply      *reply;
 309         struct ptlrpc_request           *req = NULL;
 310         int                             rc = 0;
 311         ENTRY;
 312
 313         if (is_ost_obj(&dt->do_lu) && obj->opo_non_exist)
 314                 RETURN(-ENOENT);
 315
 316         if (obj->opo_ooa != NULL) {
 317                 spin_lock(&obj->opo_lock);
 318                 if (obj->opo_ooa->ooa_attr.la_valid != 0) {
 319                         *attr = obj->opo_ooa->ooa_attr;
 320                         spin_unlock(&obj->opo_lock);
 321
 322                         RETURN(0);
 323                 }
 324                 spin_unlock(&obj->opo_lock);
 325         }
 326
 327         update = out_create_update_req(dev);
 328         if (IS_ERR(update))
 329                 RETURN(PTR_ERR(update));
 330
 331         rc = out_insert_update(env, update, OUT_ATTR_GET,
 332                                lu_object_fid(&dt->do_lu), 0, NULL, NULL);
 333         if (rc != 0) {
 334                 CERROR("%s: Insert update error "DFID": rc = %d\n",
 335                        dev->dd_lu_dev.ld_obd->obd_name,
 336                        PFID(lu_object_fid(&dt->do_lu)), rc);
 337
 338                 GOTO(out, rc);
 339         }
 340
 341         rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
 342         if (rc != 0) {
 343                 if (rc == -ENOENT) {
 344                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
 345                         obj->opo_non_exist = 1;
 346                 } else {
 347                         CERROR("%s:osp_attr_get update error "DFID": rc = %d\n",
 348                                dev->dd_lu_dev.ld_obd->obd_name,
 349                                PFID(lu_object_fid(&dt->do_lu)), rc);
 350                 }
 351
 352                 GOTO(out, rc);
 353         }
 354
 355         osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
 356         obj->opo_non_exist = 0;
 357         reply = req_capsule_server_sized_get(&req->rq_pill,
 358                                              &RMF_OUT_UPDATE_REPLY,
 359                                              OUT_UPDATE_REPLY_SIZE);
 360         if (reply == NULL || reply->ourp_magic != UPDATE_REPLY_MAGIC)
 361                 GOTO(out, rc = -EPROTO);
 362
 363         rc = osp_get_attr_from_reply(env, reply, attr, obj, 0);
 364         if (rc != 0)
 365                 GOTO(out, rc);
 366
 367         if (!is_ost_obj(&dt->do_lu)) {
 368                 if (attr->la_flags == 1)
 369                         obj->opo_empty = 0;
 370                 else
 371                         obj->opo_empty = 1;
 372         }
 373
 374         GOTO(out, rc = 0);
 375
 376 out:
 377         if (req != NULL)
 378                 ptlrpc_req_finished(req);
 379
 380         out_destroy_update_req(update);
 381
 382         return rc;
 383 }
 384
 385 static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
 386                                 const struct lu_attr *attr, struct thandle *th)
 387 {
 388         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 389         struct osp_object       *o = dt2osp_obj(dt);
 390         struct lu_attr          *la;
 391         int                      rc = 0;
 392
 393         ENTRY;
 394
 395         /*
 396          * Usually we don't allow server stack to manipulate size
 397          * but there is a special case when striping is created
 398          * late, after stripless file got truncated to non-zero.
 399          *
 400          * In this case we do the following:
 401          *
 402          * 1) grab id in declare - this can lead to leaked OST objects
 403          *    but we don't currently have proper mechanism and the only
 404          *    options we have are to do truncate RPC holding transaction
 405          *    open (very bad) or to grab id in declare at cost of leaked
 406          *    OST object in same very rare unfortunate case (just bad)
 407          *    notice 1.6-2.0 do assignment outside of running transaction
 408          *    all the time, meaning many more chances for leaked objects.
 409          *
 410          * 2) send synchronous truncate RPC with just assigned id
 411          */
 412
 413         /* there are few places in MDD code still passing NULL
 414          * XXX: to be fixed soon */
 415         if (attr == NULL)
 416                 RETURN(0);
 417
 418         if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
 419             fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
 420                 LASSERT(!dt_object_exists(dt));
 421                 osp_object_assign_fid(env, d, o);
 422                 rc = osp_object_truncate(env, dt, attr->la_size);
 423                 if (rc)
 424                         RETURN(rc);
 425         }
 426
 427         if (o->opo_new) {
 428                 /* no need in logging for new objects being created */
 429                 RETURN(0);
 430         }
 431
 432         if (!(attr->la_valid & (LA_UID | LA_GID)))
 433                 RETURN(0);
 434
 435         if (!is_only_remote_trans(th))
 436                 /*
 437                  * track all UID/GID changes via llog
 438                  */
 439                 rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
 440         else
 441                 /* It is for OST-object attr_set directly without updating
 442                  * local MDT-object attribute. It is usually used by LFSCK. */
 443                 rc = osp_md_declare_attr_set(env, dt, attr, th);
 444
 445         if (rc != 0 || o->opo_ooa == NULL)
 446                 RETURN(rc);
 447
 448         la = &o->opo_ooa->ooa_attr;
 449         spin_lock(&o->opo_lock);
 450         if (attr->la_valid & LA_UID) {
 451                 la->la_uid = attr->la_uid;
 452                 la->la_valid |= LA_UID;
 453         }
 454
 455         if (attr->la_valid & LA_GID) {
 456                 la->la_gid = attr->la_gid;
 457                 la->la_valid |= LA_GID;
 458         }
 459         spin_unlock(&o->opo_lock);
 460
 461         RETURN(0);
 462 }
 463
 464 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
 465                         const struct lu_attr *attr, struct thandle *th,
 466                         struct lustre_capa *capa)
 467 {
 468         struct osp_object       *o = dt2osp_obj(dt);
 469         int                      rc = 0;
 470
 471         ENTRY;
 472
 473         /* we're interested in uid/gid changes only */
 474         if (!(attr->la_valid & (LA_UID | LA_GID)))
 475                 RETURN(0);
 476
 477         /* new object, the very first ->attr_set()
 478          * initializing attributes needs no logging
 479          * all subsequent one are subject to the
 480          * logging and synchronization with OST */
 481         if (o->opo_new) {
 482                 o->opo_new = 0;
 483                 RETURN(0);
 484         }
 485
 486         if (!is_only_remote_trans(th))
 487                 /*
 488                  * once transaction is committed put proper command on
 489                  * the queue going to our OST
 490                  */
 491                 rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
 492                 /* XXX: send new uid/gid to OST ASAP? */
 493         else
 494                 /* It is for OST-object attr_set directly without updating
 495                  * local MDT-object attribute. It is usually used by LFSCK. */
 496                 rc = osp_md_attr_set(env, dt, attr, th, capa);
 497
 498         RETURN(rc);
 499 }
 500
 501 static int osp_xattr_get_interpterer(const struct lu_env *env,
 502                                      struct object_update_reply *reply,
 503                                      struct osp_object *obj,
 504                                      void *data, int index, int rc)
 505 {
 506         struct osp_object_attr  *ooa  = obj->opo_ooa;
 507         struct osp_xattr_entry  *oxe  = data;
 508         struct lu_buf           *rbuf = &osp_env_info(env)->osi_lb2;
 509
 510         LASSERT(ooa != NULL);
 511
 512         if (rc == 0) {
 513                 int len = sizeof(*oxe) + oxe->oxe_namelen + 1;
 514
 515                 rc = object_update_result_data_get(reply, rbuf, index);
 516                 if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
 517                         spin_lock(&obj->opo_lock);
 518                         oxe->oxe_ready = 0;
 519                         spin_unlock(&obj->opo_lock);
 520                         osp_oac_xattr_put(oxe);
 521
 522                         return rc < 0 ? rc : -ERANGE;
 523                 }
 524
 525                 spin_lock(&obj->opo_lock);
 526                 oxe->oxe_vallen = rbuf->lb_len;
 527                 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
 528                 oxe->oxe_exist = 1;
 529                 oxe->oxe_ready = 1;
 530                 spin_unlock(&obj->opo_lock);
 531         } else if (rc == -ENOENT || rc == -ENODATA) {
 532                 spin_lock(&obj->opo_lock);
 533                 oxe->oxe_exist = 0;
 534                 oxe->oxe_ready = 1;
 535                 spin_unlock(&obj->opo_lock);
 536         } else {
 537                 spin_lock(&obj->opo_lock);
 538                 oxe->oxe_ready = 0;
 539                 spin_unlock(&obj->opo_lock);
 540         }
 541
 542         osp_oac_xattr_put(oxe);
 543
 544         return 0;
 545 }
 546
 547 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
 548                                  struct lu_buf *buf, const char *name,
 549                                  struct lustre_capa *capa)
 550 {
 551         struct osp_object       *obj     = dt2osp_obj(dt);
 552         struct osp_device       *osp     = lu2osp_dev(dt->do_lu.lo_dev);
 553         struct dt_update_request *update;
 554         struct osp_xattr_entry  *oxe;
 555         int                      namelen = strlen(name);
 556         int                      rc      = 0;
 557
 558         LASSERT(buf != NULL);
 559         LASSERT(name != NULL);
 560
 561         /* If only for xattr size, return directly. */
 562         if (unlikely(buf->lb_len == 0))
 563                 return 0;
 564
 565         if (obj->opo_ooa == NULL) {
 566                 rc = osp_oac_init(obj);
 567                 if (rc != 0)
 568                         return rc;
 569         }
 570
 571         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
 572         if (oxe == NULL)
 573                 return -ENOMEM;
 574
 575         mutex_lock(&osp->opd_async_requests_mutex);
 576         update = osp_find_or_create_async_update_request(osp);
 577         if (IS_ERR(update)) {
 578                 rc = PTR_ERR(update);
 579                 mutex_unlock(&osp->opd_async_requests_mutex);
 580                 osp_oac_xattr_put(oxe);
 581         } else {
 582                 rc = osp_insert_async_update(env, update, OUT_XATTR_GET, obj,
 583                                              1, &namelen, &name, oxe,
 584                                              osp_xattr_get_interpterer);
 585                 if (rc != 0) {
 586                         mutex_unlock(&osp->opd_async_requests_mutex);
 587                         osp_oac_xattr_put(oxe);
 588                 } else {
 589                         /* XXX: Currently, we trigger the batched async OUT
 590                          *      RPC via dt_declare_xattr_get(). It is not
 591                          *      perfect solution, but works well now.
 592                          *
 593                          *      We will improve it in the future. */
 594                         update = osp->opd_async_requests;
 595                         if (update != NULL && update->dur_req != NULL &&
 596                             update->dur_req->ourq_count > 0) {
 597                                 osp->opd_async_requests = NULL;
 598                                 mutex_unlock(&osp->opd_async_requests_mutex);
 599                                 rc = osp_unplug_async_update(env, osp, update);
 600                         } else {
 601                                 mutex_unlock(&osp->opd_async_requests_mutex);
 602                         }
 603                 }
 604         }
 605
 606         return rc;
 607 }
 608
 609 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
 610                   struct lu_buf *buf, const char *name,
 611                   struct lustre_capa *capa)
 612 {
 613         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 614         struct osp_object       *obj    = dt2osp_obj(dt);
 615         struct dt_device        *dev    = &osp->opd_dt_dev;
 616         struct lu_buf           *rbuf   = &osp_env_info(env)->osi_lb2;
 617         struct dt_update_request *update = NULL;
 618         struct ptlrpc_request   *req    = NULL;
 619         struct object_update_reply *reply;
 620         struct osp_xattr_entry  *oxe    = NULL;
 621         const char              *dname  = dt->do_lu.lo_dev->ld_obd->obd_name;
 622         int                      namelen;
 623         int                      rc     = 0;
 624         ENTRY;
 625
 626         LASSERT(buf != NULL);
 627         LASSERT(name != NULL);
 628
 629         if (unlikely(obj->opo_non_exist))
 630                 RETURN(-ENOENT);
 631
 632         oxe = osp_oac_xattr_find(obj, name);
 633         if (oxe != NULL) {
 634                 spin_lock(&obj->opo_lock);
 635                 if (oxe->oxe_ready) {
 636                         if (!oxe->oxe_exist)
 637                                 GOTO(unlock, rc = -ENODATA);
 638
 639                         if (buf->lb_buf == NULL)
 640                                 GOTO(unlock, rc = oxe->oxe_vallen);
 641
 642                         if (buf->lb_len < oxe->oxe_vallen)
 643                                 GOTO(unlock, rc = -ERANGE);
 644
 645                         memcpy(buf->lb_buf, oxe->oxe_value, oxe->oxe_vallen);
 646
 647                         GOTO(unlock, rc = oxe->oxe_vallen);
 648
 649 unlock:
 650                         spin_unlock(&obj->opo_lock);
 651                         osp_oac_xattr_put(oxe);
 652
 653                         return rc;
 654                 }
 655                 spin_unlock(&obj->opo_lock);
 656         }
 657
 658         update = out_create_update_req(dev);
 659         if (IS_ERR(update))
 660                 GOTO(out, rc = PTR_ERR(update));
 661
 662         namelen = strlen(name) + 1;
 663         rc = out_insert_update(env, update, OUT_XATTR_GET,
 664                                lu_object_fid(&dt->do_lu), 1, &namelen, &name);
 665         if (rc != 0) {
 666                 CERROR("%s: Insert update error "DFID": rc = %d\n",
 667                        dname, PFID(lu_object_fid(&dt->do_lu)), rc);
 668
 669                 GOTO(out, rc);
 670         }
 671
 672         rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
 673         if (rc != 0) {
 674                 if (obj->opo_ooa == NULL)
 675                         GOTO(out, rc);
 676
 677                 if (oxe == NULL)
 678                         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
 679
 680                 if (oxe == NULL) {
 681                         CWARN("%s: Fail to add xattr (%s) to cache for "
 682                               DFID" (1): rc = %d\n", dname, name,
 683                               PFID(lu_object_fid(&dt->do_lu)), rc);
 684
 685                         GOTO(out, rc);
 686                 }
 687
 688                 spin_lock(&obj->opo_lock);
 689                 if (rc == -ENOENT || rc == -ENODATA) {
 690                         oxe->oxe_exist = 0;
 691                         oxe->oxe_ready = 1;
 692                 } else {
 693                         oxe->oxe_ready = 0;
 694                 }
 695                 spin_unlock(&obj->opo_lock);
 696
 697                 GOTO(out, rc);
 698         }
 699
 700         reply = req_capsule_server_sized_get(&req->rq_pill,
 701                                              &RMF_OUT_UPDATE_REPLY,
 702                                              OUT_UPDATE_REPLY_SIZE);
 703         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
 704                 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
 705                        dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
 706                        PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
 707
 708                 GOTO(out, rc = -EPROTO);
 709         }
 710
 711         rc = object_update_result_data_get(reply, rbuf, 0);
 712         if (rc < 0)
 713                 GOTO(out, rc);
 714
 715         if (buf->lb_buf == NULL)
 716                 GOTO(out, rc = rbuf->lb_len);
 717
 718         if (unlikely(buf->lb_len < rbuf->lb_len))
 719                 GOTO(out, rc = -ERANGE);
 720
 721         memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
 722         rc = rbuf->lb_len;
 723         if (obj->opo_ooa == NULL)
 724                 GOTO(out, rc);
 725
 726         if (oxe == NULL) {
 727                 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
 728                 if (oxe == NULL) {
 729                         CWARN("%s: Fail to add xattr (%s) to "
 730                               "cache for "DFID" (2): rc = %d\n",
 731                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
 732
 733                         GOTO(out, rc);
 734                 }
 735         }
 736
 737         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
 738                 struct osp_xattr_entry *old = oxe;
 739                 struct osp_xattr_entry *tmp;
 740
 741                 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
 742                 osp_oac_xattr_put(oxe);
 743                 oxe = tmp;
 744                 if (tmp == NULL) {
 745                         CWARN("%s: Fail to update xattr (%s) to "
 746                               "cache for "DFID": rc = %d\n",
 747                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
 748                         spin_lock(&obj->opo_lock);
 749                         oxe->oxe_ready = 0;
 750                         spin_unlock(&obj->opo_lock);
 751
 752                         GOTO(out, rc);
 753                 }
 754
 755                 /* Drop the ref for entry on list. */
 756                 osp_oac_xattr_put(old);
 757         }
 758
 759         spin_lock(&obj->opo_lock);
 760         oxe->oxe_vallen = rbuf->lb_len;
 761         memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
 762         oxe->oxe_exist = 1;
 763         oxe->oxe_ready = 1;
 764         spin_unlock(&obj->opo_lock);
 765
 766         GOTO(out, rc);
 767
 768 out:
 769         if (req != NULL)
 770                 ptlrpc_req_finished(req);
 771
 772         if (update != NULL && !IS_ERR(update))
 773                 out_destroy_update_req(update);
 774
 775         if (oxe != NULL)
 776                 osp_oac_xattr_put(oxe);
 777
 778         return rc;
 779 }
 780
 781 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 782                           const struct lu_buf *buf, const char *name,
 783                           int flag, struct thandle *th)
 784 {
 785         struct osp_object       *o       = dt2osp_obj(dt);
 786         struct dt_update_request *update;
 787         struct lu_fid           *fid;
 788         struct osp_xattr_entry  *oxe;
 789         int                     sizes[3] = {strlen(name), buf->lb_len,
 790                                             sizeof(int)};
 791         char                    *bufs[3] = {(char *)name, (char *)buf->lb_buf };
 792         int                     rc;
 793
 794         LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
 795
 796         update = out_find_create_update_loc(th, dt);
 797         if (IS_ERR(update)) {
 798                 CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
 799                        dt->do_lu.lo_dev->ld_obd->obd_name,
 800                        PFID(lu_object_fid(&dt->do_lu)),
 801                        (int)PTR_ERR(update));
 802
 803                 return PTR_ERR(update);
 804         }
 805
 806         flag = cpu_to_le32(flag);
 807         bufs[2] = (char *)&flag;
 808
 809         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 810         rc = out_insert_update(env, update, OUT_XATTR_SET, fid,
 811                                ARRAY_SIZE(sizes), sizes, (const char **)bufs);
 812         if (rc != 0 || o->opo_ooa == NULL)
 813                 return rc;
 814
 815         oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
 816         if (oxe == NULL) {
 817                 CWARN("%s: Fail to add xattr (%s) to cache for "DFID
 818                       ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
 819                       name, PFID(lu_object_fid(&dt->do_lu)), rc);
 820
 821                 return 0;
 822         }
 823
 824         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
 825                 struct osp_xattr_entry *old = oxe;
 826                 struct osp_xattr_entry *tmp;
 827
 828                 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
 829                 osp_oac_xattr_put(oxe);
 830                 oxe = tmp;
 831                 if (tmp == NULL) {
 832                         CWARN("%s: Fail to update xattr (%s) to cache for "DFID
 833                               ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
 834                               name, PFID(lu_object_fid(&dt->do_lu)), rc);
 835                         spin_lock(&o->opo_lock);
 836                         oxe->oxe_ready = 0;
 837                         spin_unlock(&o->opo_lock);
 838
 839                         return 0;
 840                 }
 841
 842                 /* Drop the ref for entry on list. */
 843                 osp_oac_xattr_put(old);
 844         }
 845
 846         spin_lock(&o->opo_lock);
 847         oxe->oxe_vallen = buf->lb_len;
 848         memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
 849         oxe->oxe_exist = 1;
 850         oxe->oxe_ready = 1;
 851         spin_unlock(&o->opo_lock);
 852         osp_oac_xattr_put(oxe);
 853
 854         return 0;
 855 }
 856
 857 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
 858                   const struct lu_buf *buf, const char *name, int fl,
 859                   struct thandle *th, struct lustre_capa *capa)
 860 {
 861         CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
 862                PFID(&dt->do_lu.lo_header->loh_fid));
 863
 864         return 0;
 865 }
 866
 867 static int osp_declare_object_create(const struct lu_env *env,
 868                                      struct dt_object *dt,
 869                                      struct lu_attr *attr,
 870                                      struct dt_allocation_hint *hint,
 871                                      struct dt_object_format *dof,
 872                                      struct thandle *th)
 873 {
 874         struct osp_thread_info  *osi = osp_env_info(env);
 875         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 876         struct osp_object       *o = dt2osp_obj(dt);
 877         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 878         int                      rc = 0;
 879
 880         ENTRY;
 881
 882         if (is_only_remote_trans(th)) {
 883                 LASSERT(fid_is_sane(fid));
 884
 885                 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
 886
 887                 RETURN(rc);
 888         }
 889
 890         /* should happen to non-0 OSP only so that at least one object
 891          * has been already declared in the scenario and LOD should
 892          * cleanup that */
 893         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
 894                 RETURN(-ENOSPC);
 895
 896         LASSERT(d->opd_last_used_oid_file);
 897
 898         /*
 899          * There can be gaps in precreated ids and record to unlink llog
 900          * XXX: we do not handle gaps yet, implemented before solution
 901          *      was found to be racy, so we disabled that. there is no
 902          *      point in making useless but expensive llog declaration.
 903          */
 904         /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
 905
 906         if (unlikely(!fid_is_zero(fid))) {
 907                 /* replay case: caller knows fid */
 908                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
 909                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
 910                 osi->osi_lb.lb_buf = NULL;
 911                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
 912                                              &osi->osi_lb, osi->osi_off, th);
 913                 RETURN(rc);
 914         }
 915
 916         /*
 917          * in declaration we need to reserve object so that we don't block
 918          * awaiting precreation RPC to complete
 919          */
 920         rc = osp_precreate_reserve(env, d);
 921         /*
 922          * we also need to declare update to local "last used id" file for
 923          * recovery if object isn't used for a reason, we need to release
 924          * reservation, this can be made in osd_object_release()
 925          */
 926         if (rc == 0) {
 927                 /* mark id is reserved: in create we don't want to talk
 928                  * to OST */
 929                 LASSERT(o->opo_reserved == 0);
 930                 o->opo_reserved = 1;
 931
 932                 /* common for all OSPs file hystorically */
 933                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
 934                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
 935                 osi->osi_lb.lb_buf = NULL;
 936                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
 937                                              &osi->osi_lb, osi->osi_off, th);
 938         } else {
 939                 /* not needed in the cache anymore */
 940                 set_bit(LU_OBJECT_HEARD_BANSHEE,
 941                             &dt->do_lu.lo_header->loh_flags);
 942         }
 943         RETURN(rc);
 944 }
 945
 946 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
 947                              struct lu_attr *attr,
 948                              struct dt_allocation_hint *hint,
 949                              struct dt_object_format *dof, struct thandle *th)
 950 {
 951         struct osp_thread_info  *osi = osp_env_info(env);
 952         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 953         struct osp_object       *o = dt2osp_obj(dt);
 954         int                     rc = 0;
 955         struct lu_fid           *fid = &osi->osi_fid;
 956         ENTRY;
 957
 958         if (is_only_remote_trans(th)) {
 959                 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
 960
 961                 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
 962                 if (rc == 0)
 963                         o->opo_non_exist = 0;
 964
 965                 RETURN(rc);
 966         }
 967
 968         o->opo_non_exist = 0;
 969         if (o->opo_reserved) {
 970                 /* regular case, fid is assigned holding trunsaction open */
 971                  osp_object_assign_fid(env, d, o);
 972         }
 973
 974         memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
 975
 976         LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
 977                  o, PFID(fid));
 978
 979         if (!o->opo_reserved) {
 980                 /* special case, id was assigned outside of transaction
 981                  * see comments in osp_declare_attr_set */
 982                 LASSERT(d->opd_pre != NULL);
 983                 spin_lock(&d->opd_pre_lock);
 984                 osp_update_last_fid(d, fid);
 985                 spin_unlock(&d->opd_pre_lock);
 986         }
 987
 988         CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
 989
 990         /* If the precreate ends, it means it will be ready to rollover to
 991          * the new sequence soon, all the creation should be synchronized,
 992          * otherwise during replay, the replay fid will be inconsistent with
 993          * last_used/create fid */
 994         if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
 995                 th->th_sync = 1;
 996
 997         /*
 998          * it's OK if the import is inactive by this moment - id was created
 999          * by OST earlier, we just need to maintain it consistently on the disk
1000          * once import is reconnected, OSP will claim this and other objects
1001          * used and OST either keep them, if they exist or recreate
1002          */
1003
1004         /* we might have lost precreated objects */
1005         if (unlikely(d->opd_gap_count) > 0) {
1006                 LASSERT(d->opd_pre != NULL);
1007                 spin_lock(&d->opd_pre_lock);
1008                 if (d->opd_gap_count > 0) {
1009                         int count = d->opd_gap_count;
1010
1011                         ostid_set_id(&osi->osi_oi,
1012                                      fid_oid(&d->opd_gap_start_fid));
1013                         d->opd_gap_count = 0;
1014                         spin_unlock(&d->opd_pre_lock);
1015
1016                         CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
1017                                PFID(&d->opd_gap_start_fid), count);
1018                         /* real gap handling is disabled intil ORI-692 will be
1019                          * fixed, now we only report gaps */
1020                 } else {
1021                         spin_unlock(&d->opd_pre_lock);
1022                 }
1023         }
1024
1025         /* new object, the very first ->attr_set()
1026          * initializing attributes needs no logging */
1027         o->opo_new = 1;
1028
1029         /* Only need update last_used oid file, seq file will only be update
1030          * during seq rollover */
1031         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1032                            &d->opd_last_used_fid.f_oid, d->opd_index);
1033
1034         rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1035                              &osi->osi_off, th);
1036
1037         CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1038                d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1039
1040         RETURN(rc);
1041 }
1042
1043 int osp_declare_object_destroy(const struct lu_env *env,
1044                                struct dt_object *dt, struct thandle *th)
1045 {
1046         struct osp_object       *o = dt2osp_obj(dt);
1047         int                      rc = 0;
1048
1049         ENTRY;
1050
1051         /*
1052          * track objects to be destroyed via llog
1053          */
1054         rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1055
1056         RETURN(rc);
1057 }
1058
1059 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1060                        struct thandle *th)
1061 {
1062         struct osp_object       *o = dt2osp_obj(dt);
1063         int                      rc = 0;
1064
1065         ENTRY;
1066
1067         o->opo_non_exist = 1;
1068         /*
1069          * once transaction is committed put proper command on
1070          * the queue going to our OST
1071          */
1072         rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1073
1074         /* not needed in cache any more */
1075         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1076
1077         RETURN(rc);
1078 }
1079
1080 struct osp_orphan_it {
1081         int                       ooi_pos0;
1082         int                       ooi_pos1;
1083         int                       ooi_pos2;
1084         int                       ooi_total_npages;
1085         int                       ooi_valid_npages;
1086         unsigned int              ooi_swab:1;
1087         __u64                     ooi_next;
1088         struct dt_object         *ooi_obj;
1089         struct lu_orphan_ent     *ooi_ent;
1090         struct page              *ooi_cur_page;
1091         struct lu_idxpage        *ooi_cur_idxpage;
1092         struct page             **ooi_pages;
1093 };
1094
1095 static int osp_orphan_index_lookup(const struct lu_env *env,
1096                                    struct dt_object *dt,
1097                                    struct dt_rec *rec,
1098                                    const struct dt_key *key,
1099                                    struct lustre_capa *capa)
1100 {
1101         return -EOPNOTSUPP;
1102 }
1103
1104 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1105                                            struct dt_object *dt,
1106                                            const struct dt_rec *rec,
1107                                            const struct dt_key *key,
1108                                            struct thandle *handle)
1109 {
1110         return -EOPNOTSUPP;
1111 }
1112
1113 static int osp_orphan_index_insert(const struct lu_env *env,
1114                                    struct dt_object *dt,
1115                                    const struct dt_rec *rec,
1116                                    const struct dt_key *key,
1117                                    struct thandle *handle,
1118                                    struct lustre_capa *capa,
1119                                    int ignore_quota)
1120 {
1121         return -EOPNOTSUPP;
1122 }
1123
1124 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1125                                            struct dt_object *dt,
1126                                            const struct dt_key *key,
1127                                            struct thandle *handle)
1128 {
1129         return -EOPNOTSUPP;
1130 }
1131
1132 static int osp_orphan_index_delete(const struct lu_env *env,
1133                                    struct dt_object *dt,
1134                                    const struct dt_key *key,
1135                                    struct thandle *handle,
1136                                    struct lustre_capa *capa)
1137 {
1138         return -EOPNOTSUPP;
1139 }
1140
1141 static struct dt_it *osp_orphan_it_init(const struct lu_env *env,
1142                                         struct dt_object *dt,
1143                                         __u32 attr,
1144                                         struct lustre_capa *capa)
1145 {
1146         struct osp_orphan_it *it;
1147
1148         OBD_ALLOC_PTR(it);
1149         if (it == NULL)
1150                 return ERR_PTR(-ENOMEM);
1151
1152         it->ooi_pos2 = -1;
1153         it->ooi_obj = dt;
1154
1155         return (struct dt_it *)it;
1156 }
1157
1158 static void osp_orphan_it_fini(const struct lu_env *env,
1159                                struct dt_it *di)
1160 {
1161         struct osp_orphan_it     *it            = (struct osp_orphan_it *)di;
1162         struct page             **pages         = it->ooi_pages;
1163         int                       npages        = it->ooi_total_npages;
1164         int                       i;
1165
1166         if (pages != NULL) {
1167                 for (i = 0; i < npages; i++) {
1168                         if (pages[i] != NULL) {
1169                                 if (pages[i] == it->ooi_cur_page) {
1170                                         kunmap(pages[i]);
1171                                         it->ooi_cur_page = NULL;
1172                                 }
1173                                 __free_page(pages[i]);
1174                         }
1175                 }
1176                 OBD_FREE(pages, npages * sizeof(*pages));
1177         }
1178         OBD_FREE_PTR(it);
1179 }
1180
1181 static int osp_orphan_it_fetch(const struct lu_env *env,
1182                                struct osp_orphan_it *it)
1183 {
1184         struct lu_device         *dev   = it->ooi_obj->do_lu.lo_dev;
1185         struct osp_device        *osp   = lu2osp_dev(dev);
1186         struct page             **pages;
1187         struct ptlrpc_request    *req   = NULL;
1188         struct ptlrpc_bulk_desc  *desc;
1189         struct idx_info          *ii;
1190         int                       npages;
1191         int                       rc;
1192         int                       i;
1193         ENTRY;
1194
1195         /* 1MB bulk */
1196         npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1197         npages /= PAGE_CACHE_SIZE;
1198
1199         OBD_ALLOC(pages, npages * sizeof(*pages));
1200         if (pages == NULL)
1201                 RETURN(-ENOMEM);
1202
1203         it->ooi_pages = pages;
1204         it->ooi_total_npages = npages;
1205         for (i = 0; i < npages; i++) {
1206                 pages[i] = alloc_page(GFP_IOFS);
1207                 if (pages[i] == NULL)
1208                         RETURN(-ENOMEM);
1209         }
1210
1211         req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1212                                    &RQF_OBD_IDX_READ);
1213         if (req == NULL)
1214                 RETURN(-ENOMEM);
1215
1216         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1217         if (rc != 0) {
1218                 ptlrpc_request_free(req);
1219                 RETURN(rc);
1220         }
1221
1222         req->rq_request_portal = OST_IDX_PORTAL;
1223         ptlrpc_at_set_req_timeout(req);
1224
1225         desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
1226                                     MDS_BULK_PORTAL);
1227         if (desc == NULL) {
1228                 ptlrpc_request_free(req);
1229                 RETURN(-ENOMEM);
1230         }
1231
1232         for (i = 0; i < npages; i++)
1233                 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1234
1235         ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1236         memset(ii, 0, sizeof(*ii));
1237         ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1238         ii->ii_fid.f_oid = osp->opd_index;
1239         ii->ii_fid.f_ver = 0;
1240         ii->ii_magic = IDX_INFO_MAGIC;
1241         ii->ii_flags = II_FL_NOHASH;
1242         ii->ii_count = npages * LU_PAGE_COUNT;
1243         ii->ii_hash_start = it->ooi_next;
1244         ii->ii_attrs =
1245                 osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
1246
1247         ptlrpc_request_set_replen(req);
1248         rc = ptlrpc_queue_wait(req);
1249         if (rc != 0)
1250                 GOTO(out, rc);
1251
1252         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1253                                           req->rq_bulk->bd_nob_transferred);
1254         if (rc < 0)
1255                 GOTO(out, rc);
1256
1257         ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1258         if (ii->ii_magic != IDX_INFO_MAGIC)
1259                  GOTO(out, rc = -EPROTO);
1260
1261         npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1262                  (PAGE_CACHE_SHIFT - LU_PAGE_SHIFT);
1263         if (npages > it->ooi_total_npages) {
1264                 CERROR("%s: returned more pages than expected, %u > %u\n",
1265                        osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1266                 GOTO(out, rc = -EINVAL);
1267         }
1268
1269         it->ooi_valid_npages = npages;
1270         if (ptlrpc_rep_need_swab(req))
1271                 it->ooi_swab = 1;
1272
1273         it->ooi_next = ii->ii_hash_end;
1274
1275         GOTO(out, rc = 0);
1276
1277 out:
1278         ptlrpc_req_finished(req);
1279
1280         return rc;
1281 }
1282
1283 static int osp_orphan_it_next(const struct lu_env *env,
1284                               struct dt_it *di)
1285 {
1286         struct osp_orphan_it     *it            = (struct osp_orphan_it *)di;
1287         struct lu_idxpage        *idxpage;
1288         struct page             **pages;
1289         int                       rc;
1290         int                       i;
1291         ENTRY;
1292
1293 again2:
1294         idxpage = it->ooi_cur_idxpage;
1295         if (idxpage != NULL) {
1296                 if (idxpage->lip_nr == 0)
1297                         RETURN(1);
1298
1299                 it->ooi_pos2++;
1300                 if (it->ooi_pos2 < idxpage->lip_nr) {
1301                         it->ooi_ent =
1302                                 (struct lu_orphan_ent *)idxpage->lip_entries +
1303                                 it->ooi_pos2;
1304                         if (it->ooi_swab)
1305                                 lustre_swab_orphan_ent(it->ooi_ent);
1306                         RETURN(0);
1307                 }
1308
1309                 it->ooi_cur_idxpage = NULL;
1310                 it->ooi_pos1++;
1311
1312 again1:
1313                 if (it->ooi_pos1 < LU_PAGE_COUNT) {
1314                         it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1315                                               LU_PAGE_SIZE * it->ooi_pos1;
1316                         if (it->ooi_swab)
1317                                 lustre_swab_lip_header(it->ooi_cur_idxpage);
1318                         if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1319                                 struct osp_device *osp =
1320                                         lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1321
1322                                 CERROR("%s: invalid magic (%x != %x) for page "
1323                                        "%d/%d while read layout orphan index\n",
1324                                        osp->opd_obd->obd_name,
1325                                        it->ooi_cur_idxpage->lip_magic,
1326                                        LIP_MAGIC, it->ooi_pos0, it->ooi_pos1);
1327                                 /* Skip this lu_page next time. */
1328                                 it->ooi_pos2 = idxpage->lip_nr - 1;
1329                                 RETURN(-EINVAL);
1330                         }
1331                         it->ooi_pos2 = -1;
1332                         goto again2;
1333                 }
1334
1335                 kunmap(it->ooi_cur_page);
1336                 it->ooi_cur_page = NULL;
1337                 it->ooi_pos0++;
1338
1339 again0:
1340                 pages = it->ooi_pages;
1341                 if (it->ooi_pos0 < it->ooi_valid_npages) {
1342                         it->ooi_cur_page = kmap(pages[it->ooi_pos0]);
1343                         it->ooi_pos1 = 0;
1344                         goto again1;
1345                 }
1346
1347                 for (i = 0; i < it->ooi_total_npages; i++) {
1348                         if (pages[i] != NULL)
1349                                 __free_page(pages[i]);
1350                 }
1351                 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1352
1353                 it->ooi_pos0 = 0;
1354                 it->ooi_total_npages = 0;
1355                 it->ooi_valid_npages = 0;
1356                 it->ooi_swab = 0;
1357                 it->ooi_ent = NULL;
1358                 it->ooi_cur_page = NULL;
1359                 it->ooi_cur_idxpage = NULL;
1360                 it->ooi_pages = NULL;
1361         }
1362
1363         if (it->ooi_next == II_END_OFF)
1364                 RETURN(1);
1365
1366         rc = osp_orphan_it_fetch(env, it);
1367         if (rc == 0)
1368                 goto again0;
1369
1370         RETURN(rc);
1371 }
1372
1373 static int osp_orphan_it_get(const struct lu_env *env,
1374                              struct dt_it *di,
1375                              const struct dt_key *key)
1376 {
1377         return -ENOSYS;
1378 }
1379
1380 static void osp_orphan_it_put(const struct lu_env *env,
1381                               struct dt_it *di)
1382 {
1383 }
1384
1385 static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1386                                         const struct dt_it *di)
1387 {
1388         struct osp_orphan_it    *it  = (struct osp_orphan_it *)di;
1389         struct lu_orphan_ent    *ent = it->ooi_ent;
1390
1391         if (likely(ent != NULL))
1392                 return (struct dt_key *)(&ent->loe_key);
1393
1394         return NULL;
1395 }
1396
1397 static int osp_orphan_it_key_size(const struct lu_env *env,
1398                                   const struct dt_it *di)
1399 {
1400         return sizeof(struct lu_fid);
1401 }
1402
1403 static int osp_orphan_it_rec(const struct lu_env *env,
1404                              const struct dt_it *di,
1405                              struct dt_rec *rec,
1406                              __u32 attr)
1407 {
1408         struct osp_orphan_it    *it  = (struct osp_orphan_it *)di;
1409         struct lu_orphan_ent    *ent = it->ooi_ent;
1410
1411         if (likely(ent != NULL)) {
1412                 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1413                 return 0;
1414         }
1415
1416         return -EINVAL;
1417 }
1418
1419 static __u64 osp_orphan_it_store(const struct lu_env *env,
1420                                  const struct dt_it *di)
1421 {
1422         struct osp_orphan_it    *it     = (struct osp_orphan_it *)di;
1423
1424         return it->ooi_next;
1425 }
1426
1427 /**
1428  * \retval       +1: locate to the exactly position
1429  * \retval        0: cannot locate to the exactly position,
1430  *                   call next() to move to a valid position.
1431  * \retval      -ve: on error
1432  */
1433 static int osp_orphan_it_load(const struct lu_env *env,
1434                               const struct dt_it *di,
1435                               __u64 hash)
1436 {
1437         struct osp_orphan_it    *it     = (struct osp_orphan_it *)di;
1438         int                      rc;
1439
1440         it->ooi_next = hash;
1441         rc = osp_orphan_it_next(env, (struct dt_it *)di);
1442         if (rc == 1)
1443                 return 0;
1444
1445         if (rc == 0)
1446                 return 1;
1447
1448         return rc;
1449 }
1450
1451 static int osp_orphan_it_key_rec(const struct lu_env *env,
1452                                 const struct dt_it *di,
1453                                 void *key_rec)
1454 {
1455         return 0;
1456 }
1457
1458 static const struct dt_index_operations osp_orphan_index_ops = {
1459         .dio_lookup             = osp_orphan_index_lookup,
1460         .dio_declare_insert     = osp_orphan_index_declare_insert,
1461         .dio_insert             = osp_orphan_index_insert,
1462         .dio_declare_delete     = osp_orphan_index_declare_delete,
1463         .dio_delete             = osp_orphan_index_delete,
1464         .dio_it = {
1465                 .init           = osp_orphan_it_init,
1466                 .fini           = osp_orphan_it_fini,
1467                 .next           = osp_orphan_it_next,
1468                 .get            = osp_orphan_it_get,
1469                 .put            = osp_orphan_it_put,
1470                 .key            = osp_orphan_it_key,
1471                 .key_size       = osp_orphan_it_key_size,
1472                 .rec            = osp_orphan_it_rec,
1473                 .store          = osp_orphan_it_store,
1474                 .load           = osp_orphan_it_load,
1475                 .key_rec        = osp_orphan_it_key_rec,
1476         }
1477 };
1478
1479 static int osp_index_try(const struct lu_env *env,
1480                          struct dt_object *dt,
1481                          const struct dt_index_features *feat)
1482 {
1483         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1484
1485         if (fid_is_last_id(fid) && fid_is_idif(fid)) {
1486                 dt->do_index_ops = &osp_orphan_index_ops;
1487
1488                 return 0;
1489         }
1490
1491         return -EINVAL;
1492 }
1493
1494 struct dt_object_operations osp_obj_ops = {
1495         .do_declare_attr_get    = osp_declare_attr_get,
1496         .do_attr_get            = osp_attr_get,
1497         .do_declare_attr_set    = osp_declare_attr_set,
1498         .do_attr_set            = osp_attr_set,
1499         .do_declare_xattr_get   = osp_declare_xattr_get,
1500         .do_xattr_get           = osp_xattr_get,
1501         .do_declare_xattr_set   = osp_declare_xattr_set,
1502         .do_xattr_set           = osp_xattr_set,
1503         .do_declare_create      = osp_declare_object_create,
1504         .do_create              = osp_object_create,
1505         .do_declare_destroy     = osp_declare_object_destroy,
1506         .do_destroy             = osp_object_destroy,
1507         .do_index_try           = osp_index_try,
1508 };
1509
1510 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1511                            const struct lu_object_conf *conf)
1512 {
1513         struct osp_object       *po = lu2osp_obj(o);
1514         int                     rc = 0;
1515         ENTRY;
1516
1517         spin_lock_init(&po->opo_lock);
1518         o->lo_header->loh_attr |= LOHA_REMOTE;
1519
1520         if (is_ost_obj(o)) {
1521                 po->opo_obj.do_ops = &osp_obj_ops;
1522         } else {
1523                 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1524
1525                 po->opo_obj.do_ops = &osp_md_obj_ops;
1526                 po->opo_obj.do_body_ops = &osp_md_body_ops;
1527                 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1528                                                      la, NULL);
1529                 if (rc == 0)
1530                         o->lo_header->loh_attr |=
1531                                 LOHA_EXISTS | (la->la_mode & S_IFMT);
1532                 if (rc == -ENOENT) {
1533                         po->opo_non_exist = 1;
1534                         rc = 0;
1535                 }
1536                 init_rwsem(&po->opo_sem);
1537         }
1538         RETURN(rc);
1539 }
1540
1541 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1542 {
1543         struct osp_object       *obj = lu2osp_obj(o);
1544         struct lu_object_header *h = o->lo_header;
1545
1546         dt_object_fini(&obj->opo_obj);
1547         lu_object_header_fini(h);
1548         if (obj->opo_ooa != NULL) {
1549                 struct osp_xattr_entry *oxe;
1550                 struct osp_xattr_entry *tmp;
1551                 int                     count;
1552
1553                 list_for_each_entry_safe(oxe, tmp,
1554                                          &obj->opo_ooa->ooa_xattr_list,
1555                                          oxe_list) {
1556                         list_del(&oxe->oxe_list);
1557                         count = atomic_read(&oxe->oxe_ref);
1558                         LASSERTF(count == 1,
1559                                  "Still has %d users on the xattr entry %.*s\n",
1560                                  count - 1, oxe->oxe_namelen, oxe->oxe_buf);
1561
1562                         OBD_FREE(oxe, oxe->oxe_buflen);
1563                 }
1564                 OBD_FREE_PTR(obj->opo_ooa);
1565         }
1566         OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1567 }
1568
1569 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1570 {
1571         struct osp_object       *po = lu2osp_obj(o);
1572         struct osp_device       *d  = lu2osp_dev(o->lo_dev);
1573
1574         ENTRY;
1575
1576         /*
1577          * release reservation if object was declared but not created
1578          * this may require lu_object_put() in LOD
1579          */
1580         if (unlikely(po->opo_reserved)) {
1581                 LASSERT(d->opd_pre != NULL);
1582                 LASSERT(d->opd_pre_reserved > 0);
1583                 spin_lock(&d->opd_pre_lock);
1584                 d->opd_pre_reserved--;
1585                 spin_unlock(&d->opd_pre_lock);
1586
1587                 /* not needed in cache any more */
1588                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1589         }
1590
1591         if (is_ost_obj(o))
1592                 /* XXX: Currently, NOT cache OST-object on MDT because:
1593                  *      1. it is not often accessed on MDT.
1594                  *      2. avoid up layer (such as LFSCK) to load too many
1595                  *         once-used OST-objects. */
1596                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1597
1598         EXIT;
1599 }
1600
1601 static int osp_object_print(const struct lu_env *env, void *cookie,
1602                             lu_printer_t p, const struct lu_object *l)
1603 {
1604         const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1605
1606         return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1607 }
1608
1609 static int osp_object_invariant(const struct lu_object *o)
1610 {
1611         LBUG();
1612 }
1613
1614 struct lu_object_operations osp_lu_obj_ops = {
1615         .loo_object_init        = osp_object_init,
1616         .loo_object_free        = osp_object_free,
1617         .loo_object_release     = osp_object_release,
1618         .loo_object_print       = osp_object_print,
1619         .loo_object_invariant   = osp_object_invariant
1620 };