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