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