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