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 (obj->opo_ooa == NULL)
 717                         GOTO(out, rc);
 718
 719                 if (oxe == NULL)
 720                         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
 721
 722                 if (oxe == NULL) {
 723                         CWARN("%s: Fail to add xattr (%s) to cache for "
 724                               DFID" (1): rc = %d\n", dname, name,
 725                               PFID(lu_object_fid(&dt->do_lu)), rc);
 726
 727                         GOTO(out, rc);
 728                 }
 729
 730                 spin_lock(&obj->opo_lock);
 731                 if (rc == -ENOENT || rc == -ENODATA) {
 732                         oxe->oxe_exist = 0;
 733                         oxe->oxe_ready = 1;
 734                 } else {
 735                         oxe->oxe_ready = 0;
 736                 }
 737                 spin_unlock(&obj->opo_lock);
 738
 739                 GOTO(out, rc);
 740         }
 741
 742         reply = req_capsule_server_sized_get(&req->rq_pill,
 743                                              &RMF_OUT_UPDATE_REPLY,
 744                                              OUT_UPDATE_REPLY_SIZE);
 745         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
 746                 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
 747                        dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
 748                        PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
 749
 750                 GOTO(out, rc = -EPROTO);
 751         }
 752
 753         rc = object_update_result_data_get(reply, rbuf, 0);
 754         if (rc < 0)
 755                 GOTO(out, rc);
 756
 757         if (buf->lb_buf == NULL)
 758                 GOTO(out, rc = rbuf->lb_len);
 759
 760         if (unlikely(buf->lb_len < rbuf->lb_len))
 761                 GOTO(out, rc = -ERANGE);
 762
 763         memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
 764         rc = rbuf->lb_len;
 765         if (obj->opo_ooa == NULL)
 766                 GOTO(out, rc);
 767
 768         if (oxe == NULL) {
 769                 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
 770                 if (oxe == NULL) {
 771                         CWARN("%s: Fail to add xattr (%s) to "
 772                               "cache for "DFID" (2): rc = %d\n",
 773                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
 774
 775                         GOTO(out, rc);
 776                 }
 777         }
 778
 779         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
 780                 struct osp_xattr_entry *old = oxe;
 781                 struct osp_xattr_entry *tmp;
 782
 783                 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
 784                 osp_oac_xattr_put(oxe);
 785                 oxe = tmp;
 786                 if (tmp == NULL) {
 787                         CWARN("%s: Fail to update xattr (%s) to "
 788                               "cache for "DFID": rc = %d\n",
 789                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
 790                         spin_lock(&obj->opo_lock);
 791                         old->oxe_ready = 0;
 792                         spin_unlock(&obj->opo_lock);
 793
 794                         GOTO(out, rc);
 795                 }
 796
 797                 /* Drop the ref for entry on list. */
 798                 osp_oac_xattr_put(old);
 799         }
 800
 801         spin_lock(&obj->opo_lock);
 802         oxe->oxe_vallen = rbuf->lb_len;
 803         memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
 804         oxe->oxe_exist = 1;
 805         oxe->oxe_ready = 1;
 806         spin_unlock(&obj->opo_lock);
 807
 808         GOTO(out, rc);
 809
 810 out:
 811         if (req != NULL)
 812                 ptlrpc_req_finished(req);
 813
 814         if (update != NULL && !IS_ERR(update))
 815                 dt_update_request_destroy(update);
 816
 817         if (oxe != NULL)
 818                 osp_oac_xattr_put(oxe);
 819
 820         return rc;
 821 }
 822
 823 static int __osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
 824                            const struct lu_buf *buf, const char *name,
 825                            int flag, struct thandle *th)
 826 {
 827         struct osp_object       *o = dt2osp_obj(dt);
 828         struct dt_update_request *update;
 829         struct osp_xattr_entry  *oxe;
 830         int                     rc;
 831         ENTRY;
 832
 833         LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
 834
 835         update = dt_update_request_find_or_create(th, dt);
 836         if (IS_ERR(update)) {
 837                 CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
 838                        dt->do_lu.lo_dev->ld_obd->obd_name,
 839                        PFID(lu_object_fid(&dt->do_lu)),
 840                        (int)PTR_ERR(update));
 841
 842                 RETURN(PTR_ERR(update));
 843         }
 844
 845         rc = out_xattr_set_pack(env, &update->dur_buf,
 846                                 lu_object_fid(&dt->do_lu),
 847                                 buf, name, flag, update->dur_batchid);
 848         if (rc != 0 || o->opo_ooa == NULL)
 849                 RETURN(rc);
 850
 851         oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
 852         if (oxe == NULL) {
 853                 CWARN("%s: Fail to add xattr (%s) to cache for "DFID,
 854                       dt->do_lu.lo_dev->ld_obd->obd_name,
 855                       name, PFID(lu_object_fid(&dt->do_lu)));
 856
 857                 RETURN(0);
 858         }
 859
 860         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
 861                 struct osp_xattr_entry *old = oxe;
 862                 struct osp_xattr_entry *tmp;
 863
 864                 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
 865                 osp_oac_xattr_put(oxe);
 866                 oxe = tmp;
 867                 if (tmp == NULL) {
 868                         CWARN("%s: Fail to update xattr (%s) to cache for "DFID,
 869                               dt->do_lu.lo_dev->ld_obd->obd_name,
 870                               name, PFID(lu_object_fid(&dt->do_lu)));
 871                         spin_lock(&o->opo_lock);
 872                         old->oxe_ready = 0;
 873                         spin_unlock(&o->opo_lock);
 874
 875                         RETURN(0);
 876                 }
 877
 878                 /* Drop the ref for entry on list. */
 879                 osp_oac_xattr_put(old);
 880         }
 881
 882         spin_lock(&o->opo_lock);
 883         oxe->oxe_vallen = buf->lb_len;
 884         memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
 885         oxe->oxe_exist = 1;
 886         oxe->oxe_ready = 1;
 887         spin_unlock(&o->opo_lock);
 888         osp_oac_xattr_put(oxe);
 889
 890         RETURN(0);
 891 }
 892
 893 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 894                           const struct lu_buf *buf, const char *name,
 895                           int flag, struct thandle *th)
 896 {
 897         int rc = 0;
 898
 899         /* Please check the comment in osp_attr_set() for handling
 900          * remote transaction. */
 901         if (!is_only_remote_trans(th))
 902                 rc = __osp_xattr_set(env, dt, buf, name, flag, th);
 903
 904         return rc;
 905 }
 906
 907 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
 908                   const struct lu_buf *buf, const char *name, int fl,
 909                   struct thandle *th, struct lustre_capa *capa)
 910 {
 911         int rc = 0;
 912
 913         CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
 914                PFID(&dt->do_lu.lo_header->loh_fid));
 915
 916         /* Please check the comment in osp_attr_set() for handling
 917          * remote transaction. */
 918         if (is_only_remote_trans(th))
 919                 rc = __osp_xattr_set(env, dt, buf, name, fl, th);
 920
 921         return rc;
 922 }
 923
 924 static int __osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
 925                            const char *name, struct thandle *th)
 926 {
 927         struct dt_update_request *update;
 928         const struct lu_fid      *fid;
 929         struct osp_object        *o     = dt2osp_obj(dt);
 930         struct osp_xattr_entry   *oxe;
 931         int                       rc;
 932
 933         update = dt_update_request_find_or_create(th, dt);
 934         if (IS_ERR(update))
 935                 return PTR_ERR(update);
 936
 937         fid = lu_object_fid(&dt->do_lu);
 938
 939         rc = out_xattr_del_pack(env, &update->dur_buf, fid, name,
 940                                 update->dur_batchid);
 941
 942         if (rc != 0 || o->opo_ooa == NULL)
 943                 return rc;
 944
 945         oxe = osp_oac_xattr_find(o, name, true);
 946         if (oxe != NULL)
 947                 /* Drop the ref for entry on list. */
 948                 osp_oac_xattr_put(oxe);
 949
 950         return 0;
 951 }
 952
 953 int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 954                           const char *name, struct thandle *th)
 955 {
 956         int rc = 0;
 957
 958         /* Please check the comment in osp_attr_set() for handling
 959          * remote transaction. */
 960         if (!is_only_remote_trans(th))
 961                 rc = __osp_xattr_del(env, dt, name, th);
 962
 963         return rc;
 964 }
 965
 966 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
 967                   const char *name, struct thandle *th,
 968                   struct lustre_capa *capa)
 969 {
 970         int rc = 0;
 971
 972         CDEBUG(D_INFO, "xattr %s del object "DFID"\n", name,
 973                PFID(&dt->do_lu.lo_header->loh_fid));
 974
 975         /* Please check the comment in osp_attr_set() for handling
 976          * remote transaction. */
 977         if (is_only_remote_trans(th))
 978                 rc = __osp_xattr_del(env, dt, name, th);
 979
 980         return rc;
 981 }
 982
 983 static int osp_declare_object_create(const struct lu_env *env,
 984                                      struct dt_object *dt,
 985                                      struct lu_attr *attr,
 986                                      struct dt_allocation_hint *hint,
 987                                      struct dt_object_format *dof,
 988                                      struct thandle *th)
 989 {
 990         struct osp_thread_info  *osi = osp_env_info(env);
 991         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 992         struct osp_object       *o = dt2osp_obj(dt);
 993         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 994         int                      rc = 0;
 995
 996         ENTRY;
 997
 998         if (is_only_remote_trans(th)) {
 999                 LASSERT(fid_is_sane(fid));
1000
1001                 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
1002
1003                 RETURN(rc);
1004         }
1005
1006         /* should happen to non-0 OSP only so that at least one object
1007          * has been already declared in the scenario and LOD should
1008          * cleanup that */
1009         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
1010                 RETURN(-ENOSPC);
1011
1012         LASSERT(d->opd_last_used_oid_file);
1013
1014         /*
1015          * There can be gaps in precreated ids and record to unlink llog
1016          * XXX: we do not handle gaps yet, implemented before solution
1017          *      was found to be racy, so we disabled that. there is no
1018          *      point in making useless but expensive llog declaration.
1019          */
1020         /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
1021
1022         if (unlikely(!fid_is_zero(fid))) {
1023                 /* replay case: caller knows fid */
1024                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
1025                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
1026                 osi->osi_lb.lb_buf = NULL;
1027                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1028                                              &osi->osi_lb, osi->osi_off, th);
1029                 RETURN(rc);
1030         }
1031
1032         /*
1033          * in declaration we need to reserve object so that we don't block
1034          * awaiting precreation RPC to complete
1035          */
1036         rc = osp_precreate_reserve(env, d);
1037         /*
1038          * we also need to declare update to local "last used id" file for
1039          * recovery if object isn't used for a reason, we need to release
1040          * reservation, this can be made in osd_object_release()
1041          */
1042         if (rc == 0) {
1043                 /* mark id is reserved: in create we don't want to talk
1044                  * to OST */
1045                 LASSERT(o->opo_reserved == 0);
1046                 o->opo_reserved = 1;
1047
1048                 /* common for all OSPs file hystorically */
1049                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
1050                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
1051                 osi->osi_lb.lb_buf = NULL;
1052                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1053                                              &osi->osi_lb, osi->osi_off, th);
1054         } else {
1055                 /* not needed in the cache anymore */
1056                 set_bit(LU_OBJECT_HEARD_BANSHEE,
1057                             &dt->do_lu.lo_header->loh_flags);
1058         }
1059         RETURN(rc);
1060 }
1061
1062 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
1063                              struct lu_attr *attr,
1064                              struct dt_allocation_hint *hint,
1065                              struct dt_object_format *dof, struct thandle *th)
1066 {
1067         struct osp_thread_info  *osi = osp_env_info(env);
1068         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1069         struct osp_object       *o = dt2osp_obj(dt);
1070         int                     rc = 0;
1071         struct lu_fid           *fid = &osi->osi_fid;
1072         ENTRY;
1073
1074         if (is_only_remote_trans(th)) {
1075                 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
1076
1077                 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
1078                 if (rc == 0)
1079                         o->opo_non_exist = 0;
1080
1081                 RETURN(rc);
1082         }
1083
1084         o->opo_non_exist = 0;
1085         if (o->opo_reserved) {
1086                 /* regular case, fid is assigned holding trunsaction open */
1087                  osp_object_assign_fid(env, d, o);
1088         }
1089
1090         memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
1091
1092         LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
1093                  o, PFID(fid));
1094
1095         if (!o->opo_reserved) {
1096                 /* special case, id was assigned outside of transaction
1097                  * see comments in osp_declare_attr_set */
1098                 LASSERT(d->opd_pre != NULL);
1099                 spin_lock(&d->opd_pre_lock);
1100                 osp_update_last_fid(d, fid);
1101                 spin_unlock(&d->opd_pre_lock);
1102         }
1103
1104         CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
1105
1106         /* If the precreate ends, it means it will be ready to rollover to
1107          * the new sequence soon, all the creation should be synchronized,
1108          * otherwise during replay, the replay fid will be inconsistent with
1109          * last_used/create fid */
1110         if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
1111                 th->th_sync = 1;
1112
1113         /*
1114          * it's OK if the import is inactive by this moment - id was created
1115          * by OST earlier, we just need to maintain it consistently on the disk
1116          * once import is reconnected, OSP will claim this and other objects
1117          * used and OST either keep them, if they exist or recreate
1118          */
1119
1120         /* we might have lost precreated objects */
1121         if (unlikely(d->opd_gap_count) > 0) {
1122                 LASSERT(d->opd_pre != NULL);
1123                 spin_lock(&d->opd_pre_lock);
1124                 if (d->opd_gap_count > 0) {
1125                         int count = d->opd_gap_count;
1126
1127                         ostid_set_id(&osi->osi_oi,
1128                                      fid_oid(&d->opd_gap_start_fid));
1129                         d->opd_gap_count = 0;
1130                         spin_unlock(&d->opd_pre_lock);
1131
1132                         CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
1133                                PFID(&d->opd_gap_start_fid), count);
1134                         /* real gap handling is disabled intil ORI-692 will be
1135                          * fixed, now we only report gaps */
1136                 } else {
1137                         spin_unlock(&d->opd_pre_lock);
1138                 }
1139         }
1140
1141         /* new object, the very first ->attr_set()
1142          * initializing attributes needs no logging */
1143         o->opo_new = 1;
1144
1145         /* Only need update last_used oid file, seq file will only be update
1146          * during seq rollover */
1147         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1148                            &d->opd_last_used_fid.f_oid, d->opd_index);
1149
1150         rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1151                              &osi->osi_off, th);
1152
1153         CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1154                d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1155
1156         RETURN(rc);
1157 }
1158
1159 int osp_declare_object_destroy(const struct lu_env *env,
1160                                struct dt_object *dt, struct thandle *th)
1161 {
1162         struct osp_object       *o = dt2osp_obj(dt);
1163         int                      rc = 0;
1164
1165         ENTRY;
1166
1167         /*
1168          * track objects to be destroyed via llog
1169          */
1170         rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1171
1172         RETURN(rc);
1173 }
1174
1175 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1176                        struct thandle *th)
1177 {
1178         struct osp_object       *o = dt2osp_obj(dt);
1179         int                      rc = 0;
1180
1181         ENTRY;
1182
1183         o->opo_non_exist = 1;
1184         /*
1185          * once transaction is committed put proper command on
1186          * the queue going to our OST
1187          */
1188         rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1189
1190         /* not needed in cache any more */
1191         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1192
1193         RETURN(rc);
1194 }
1195
1196 static int osp_orphan_index_lookup(const struct lu_env *env,
1197                                    struct dt_object *dt,
1198                                    struct dt_rec *rec,
1199                                    const struct dt_key *key,
1200                                    struct lustre_capa *capa)
1201 {
1202         return -EOPNOTSUPP;
1203 }
1204
1205 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1206                                            struct dt_object *dt,
1207                                            const struct dt_rec *rec,
1208                                            const struct dt_key *key,
1209                                            struct thandle *handle)
1210 {
1211         return -EOPNOTSUPP;
1212 }
1213
1214 static int osp_orphan_index_insert(const struct lu_env *env,
1215                                    struct dt_object *dt,
1216                                    const struct dt_rec *rec,
1217                                    const struct dt_key *key,
1218                                    struct thandle *handle,
1219                                    struct lustre_capa *capa,
1220                                    int ignore_quota)
1221 {
1222         return -EOPNOTSUPP;
1223 }
1224
1225 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1226                                            struct dt_object *dt,
1227                                            const struct dt_key *key,
1228                                            struct thandle *handle)
1229 {
1230         return -EOPNOTSUPP;
1231 }
1232
1233 static int osp_orphan_index_delete(const struct lu_env *env,
1234                                    struct dt_object *dt,
1235                                    const struct dt_key *key,
1236                                    struct thandle *handle,
1237                                    struct lustre_capa *capa)
1238 {
1239         return -EOPNOTSUPP;
1240 }
1241
1242 struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
1243                           __u32 attr, struct lustre_capa *capa)
1244 {
1245         struct osp_it *it;
1246
1247         OBD_ALLOC_PTR(it);
1248         if (it == NULL)
1249                 return ERR_PTR(-ENOMEM);
1250
1251         it->ooi_pos_ent = -1;
1252         it->ooi_obj = dt;
1253
1254         return (struct dt_it *)it;
1255 }
1256
1257 void osp_it_fini(const struct lu_env *env, struct dt_it *di)
1258 {
1259         struct osp_it   *it = (struct osp_it *)di;
1260         struct page     **pages = it->ooi_pages;
1261         int             npages = it->ooi_total_npages;
1262         int             i;
1263
1264         if (pages != NULL) {
1265                 for (i = 0; i < npages; i++) {
1266                         if (pages[i] != NULL) {
1267                                 if (pages[i] == it->ooi_cur_page) {
1268                                         kunmap(pages[i]);
1269                                         it->ooi_cur_page = NULL;
1270                                 }
1271                                 __free_page(pages[i]);
1272                         }
1273                 }
1274                 OBD_FREE(pages, npages * sizeof(*pages));
1275         }
1276         OBD_FREE_PTR(it);
1277 }
1278
1279 static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
1280 {
1281         struct lu_device         *dev   = it->ooi_obj->do_lu.lo_dev;
1282         struct osp_device        *osp   = lu2osp_dev(dev);
1283         struct page             **pages;
1284         struct ptlrpc_request    *req   = NULL;
1285         struct ptlrpc_bulk_desc  *desc;
1286         struct idx_info          *ii;
1287         int                       npages;
1288         int                       rc;
1289         int                       i;
1290         ENTRY;
1291
1292         /* 1MB bulk */
1293         npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1294         npages /= PAGE_CACHE_SIZE;
1295
1296         OBD_ALLOC(pages, npages * sizeof(*pages));
1297         if (pages == NULL)
1298                 RETURN(-ENOMEM);
1299
1300         it->ooi_pages = pages;
1301         it->ooi_total_npages = npages;
1302         for (i = 0; i < npages; i++) {
1303                 pages[i] = alloc_page(GFP_IOFS);
1304                 if (pages[i] == NULL)
1305                         RETURN(-ENOMEM);
1306         }
1307
1308         req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1309                                    &RQF_OBD_IDX_READ);
1310         if (req == NULL)
1311                 RETURN(-ENOMEM);
1312
1313         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1314         if (rc != 0) {
1315                 ptlrpc_request_free(req);
1316                 RETURN(rc);
1317         }
1318
1319         req->rq_request_portal = OUT_PORTAL;
1320         ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1321         memset(ii, 0, sizeof(*ii));
1322         if (fid_is_last_id(lu_object_fid(&it->ooi_obj->do_lu))) {
1323                 /* LFSCK will iterate orphan object[FID_SEQ_LAYOUT_BTREE,
1324                  * ost_index, 0] with LAST_ID FID, so it needs to replace
1325                  * the FID with orphan FID here */
1326                 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1327                 ii->ii_fid.f_oid = osp->opd_index;
1328                 ii->ii_fid.f_ver = 0;
1329                 ii->ii_flags = II_FL_NOHASH;
1330         } else {
1331                 ii->ii_fid = *lu_object_fid(&it->ooi_obj->do_lu);
1332                 ii->ii_flags = II_FL_NOHASH | II_FL_NOKEY | II_FL_VARKEY |
1333                                II_FL_VARREC;
1334         }
1335         ii->ii_magic = IDX_INFO_MAGIC;
1336         ii->ii_count = npages * LU_PAGE_COUNT;
1337         ii->ii_hash_start = it->ooi_next;
1338         ii->ii_attrs = osp_dev2node(osp);
1339
1340         ptlrpc_at_set_req_timeout(req);
1341
1342         desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
1343                                     MDS_BULK_PORTAL);
1344         if (desc == NULL) {
1345                 ptlrpc_request_free(req);
1346                 RETURN(-ENOMEM);
1347         }
1348
1349         for (i = 0; i < npages; i++)
1350                 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
1351
1352         ptlrpc_request_set_replen(req);
1353         rc = ptlrpc_queue_wait(req);
1354         if (rc != 0)
1355                 GOTO(out, rc);
1356
1357         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1358                                           req->rq_bulk->bd_nob_transferred);
1359         if (rc < 0)
1360                 GOTO(out, rc);
1361         rc = 0;
1362
1363         ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1364         if (ii->ii_magic != IDX_INFO_MAGIC)
1365                  GOTO(out, rc = -EPROTO);
1366
1367         npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1368                  (PAGE_CACHE_SHIFT - LU_PAGE_SHIFT);
1369         if (npages > it->ooi_total_npages) {
1370                 CERROR("%s: returned more pages than expected, %u > %u\n",
1371                        osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1372                 GOTO(out, rc = -EINVAL);
1373         }
1374
1375         it->ooi_valid_npages = npages;
1376         if (ptlrpc_rep_need_swab(req))
1377                 it->ooi_swab = 1;
1378
1379         it->ooi_next = ii->ii_hash_end;
1380
1381 out:
1382         ptlrpc_req_finished(req);
1383
1384         return rc;
1385 }
1386
1387 int osp_it_next_page(const struct lu_env *env, struct dt_it *di)
1388 {
1389         struct osp_it           *it = (struct osp_it *)di;
1390         struct lu_idxpage       *idxpage;
1391         struct page             **pages;
1392         int                     rc;
1393         int                     i;
1394         ENTRY;
1395
1396 again2:
1397         idxpage = it->ooi_cur_idxpage;
1398         if (idxpage != NULL) {
1399                 if (idxpage->lip_nr == 0)
1400                         RETURN(1);
1401
1402                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1403                         CDEBUG(D_INFO, "ooi_pos %d nr %d\n",
1404                                (int)it->ooi_pos_ent, (int)idxpage->lip_nr);
1405                         RETURN(0);
1406                 }
1407                 it->ooi_cur_idxpage = NULL;
1408                 it->ooi_pos_lu_page++;
1409 again1:
1410                 if (it->ooi_pos_lu_page < LU_PAGE_COUNT) {
1411                         it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1412                                          LU_PAGE_SIZE * it->ooi_pos_lu_page;
1413                         if (it->ooi_swab)
1414                                 lustre_swab_lip_header(it->ooi_cur_idxpage);
1415                         if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1416                                 struct osp_device *osp =
1417                                         lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1418
1419                                 CERROR("%s: invalid magic (%x != %x) for page "
1420                                        "%d/%d while read layout orphan index\n",
1421                                        osp->opd_obd->obd_name,
1422                                        it->ooi_cur_idxpage->lip_magic,
1423                                        LIP_MAGIC, it->ooi_pos_page,
1424                                        it->ooi_pos_lu_page);
1425                                 /* Skip this lu_page next time. */
1426                                 it->ooi_pos_ent = idxpage->lip_nr - 1;
1427                                 RETURN(-EINVAL);
1428                         }
1429                         it->ooi_pos_ent = -1;
1430                         goto again2;
1431                 }
1432
1433                 kunmap(it->ooi_cur_page);
1434                 it->ooi_cur_page = NULL;
1435                 it->ooi_pos_page++;
1436
1437 again0:
1438                 pages = it->ooi_pages;
1439                 if (it->ooi_pos_page < it->ooi_valid_npages) {
1440                         it->ooi_cur_page = kmap(pages[it->ooi_pos_page]);
1441                         it->ooi_pos_lu_page = 0;
1442                         goto again1;
1443                 }
1444
1445                 for (i = 0; i < it->ooi_total_npages; i++) {
1446                         if (pages[i] != NULL)
1447                                 __free_page(pages[i]);
1448                 }
1449                 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1450
1451                 it->ooi_pos_page = 0;
1452                 it->ooi_total_npages = 0;
1453                 it->ooi_valid_npages = 0;
1454                 it->ooi_swab = 0;
1455                 it->ooi_ent = NULL;
1456                 it->ooi_cur_page = NULL;
1457                 it->ooi_cur_idxpage = NULL;
1458                 it->ooi_pages = NULL;
1459         }
1460
1461         if (it->ooi_next == II_END_OFF)
1462                 RETURN(1);
1463
1464         rc = osp_it_fetch(env, it);
1465         if (rc == 0)
1466                 goto again0;
1467
1468         RETURN(rc);
1469 }
1470
1471 int osp_orphan_it_next(const struct lu_env *env, struct dt_it *di)
1472 {
1473         struct osp_it           *it = (struct osp_it *)di;
1474         struct lu_idxpage       *idxpage;
1475         int                     rc;
1476         ENTRY;
1477
1478 again:
1479         idxpage = it->ooi_cur_idxpage;
1480         if (idxpage != NULL) {
1481                 if (idxpage->lip_nr == 0)
1482                         RETURN(1);
1483
1484                 it->ooi_pos_ent++;
1485                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1486                         it->ooi_ent =
1487                                 (struct lu_orphan_ent *)idxpage->lip_entries +
1488                                                         it->ooi_pos_ent;
1489                         if (it->ooi_swab)
1490                                 lustre_swab_orphan_ent(it->ooi_ent);
1491                         RETURN(0);
1492                 }
1493         }
1494
1495         rc = osp_it_next_page(env, di);
1496         if (rc == 0)
1497                 goto again;
1498
1499         RETURN(rc);
1500 }
1501
1502 int osp_it_get(const struct lu_env *env, struct dt_it *di,
1503                const struct dt_key *key)
1504 {
1505         return 1;
1506 }
1507
1508 void osp_it_put(const struct lu_env *env, struct dt_it *di)
1509 {
1510 }
1511
1512 struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1513                                  const struct dt_it *di)
1514 {
1515         struct osp_it   *it  = (struct osp_it *)di;
1516         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
1517
1518         if (likely(ent != NULL))
1519                 return (struct dt_key *)(&ent->loe_key);
1520
1521         return NULL;
1522 }
1523
1524 int osp_orphan_it_key_size(const struct lu_env *env, const struct dt_it *di)
1525 {
1526         return sizeof(struct lu_fid);
1527 }
1528
1529 int osp_orphan_it_rec(const struct lu_env *env, const struct dt_it *di,
1530                       struct dt_rec *rec, __u32 attr)
1531 {
1532         struct osp_it   *it  = (struct osp_it *)di;
1533         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
1534
1535         if (likely(ent != NULL)) {
1536                 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1537                 return 0;
1538         }
1539
1540         return -EINVAL;
1541 }
1542
1543 __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
1544 {
1545         struct osp_it   *it = (struct osp_it *)di;
1546
1547         return it->ooi_next;
1548 }
1549
1550 /**
1551  * \retval       +1: locate to the exactly position
1552  * \retval        0: cannot locate to the exactly position,
1553  *                   call next() to move to a valid position.
1554  * \retval      -ve: on error
1555  */
1556 int osp_orphan_it_load(const struct lu_env *env, const struct dt_it *di,
1557                        __u64 hash)
1558 {
1559         struct osp_it   *it     = (struct osp_it *)di;
1560         int              rc;
1561
1562         it->ooi_next = hash;
1563         rc = osp_orphan_it_next(env, (struct dt_it *)di);
1564         if (rc == 1)
1565                 return 0;
1566
1567         if (rc == 0)
1568                 return 1;
1569
1570         return rc;
1571 }
1572
1573 int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
1574                    void *key_rec)
1575 {
1576         return 0;
1577 }
1578
1579 static const struct dt_index_operations osp_orphan_index_ops = {
1580         .dio_lookup             = osp_orphan_index_lookup,
1581         .dio_declare_insert     = osp_orphan_index_declare_insert,
1582         .dio_insert             = osp_orphan_index_insert,
1583         .dio_declare_delete     = osp_orphan_index_declare_delete,
1584         .dio_delete             = osp_orphan_index_delete,
1585         .dio_it = {
1586                 .init           = osp_it_init,
1587                 .fini           = osp_it_fini,
1588                 .next           = osp_orphan_it_next,
1589                 .get            = osp_it_get,
1590                 .put            = osp_it_put,
1591                 .key            = osp_orphan_it_key,
1592                 .key_size       = osp_orphan_it_key_size,
1593                 .rec            = osp_orphan_it_rec,
1594                 .store          = osp_it_store,
1595                 .load           = osp_orphan_it_load,
1596                 .key_rec        = osp_it_key_rec,
1597         }
1598 };
1599
1600 static int osp_index_try(const struct lu_env *env,
1601                          struct dt_object *dt,
1602                          const struct dt_index_features *feat)
1603 {
1604         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1605
1606         if (fid_is_last_id(fid) && fid_is_idif(fid))
1607                 dt->do_index_ops = &osp_orphan_index_ops;
1608         else
1609                 dt->do_index_ops = &osp_md_index_ops;
1610         return 0;
1611 }
1612
1613 struct dt_object_operations osp_obj_ops = {
1614         .do_declare_attr_get    = osp_declare_attr_get,
1615         .do_attr_get            = osp_attr_get,
1616         .do_declare_attr_set    = osp_declare_attr_set,
1617         .do_attr_set            = osp_attr_set,
1618         .do_declare_xattr_get   = osp_declare_xattr_get,
1619         .do_xattr_get           = osp_xattr_get,
1620         .do_declare_xattr_set   = osp_declare_xattr_set,
1621         .do_xattr_set           = osp_xattr_set,
1622         .do_declare_create      = osp_declare_object_create,
1623         .do_create              = osp_object_create,
1624         .do_declare_destroy     = osp_declare_object_destroy,
1625         .do_destroy             = osp_object_destroy,
1626         .do_index_try           = osp_index_try,
1627 };
1628
1629 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1630                            const struct lu_object_conf *conf)
1631 {
1632         struct osp_object       *po = lu2osp_obj(o);
1633         int                     rc = 0;
1634         ENTRY;
1635
1636         spin_lock_init(&po->opo_lock);
1637         o->lo_header->loh_attr |= LOHA_REMOTE;
1638
1639         if (is_ost_obj(o)) {
1640                 po->opo_obj.do_ops = &osp_obj_ops;
1641         } else {
1642                 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1643
1644                 po->opo_obj.do_ops = &osp_md_obj_ops;
1645                 po->opo_obj.do_body_ops = &osp_md_body_ops;
1646                 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1647                                                      la, NULL);
1648                 if (rc == 0)
1649                         o->lo_header->loh_attr |=
1650                                 LOHA_EXISTS | (la->la_mode & S_IFMT);
1651                 if (rc == -ENOENT) {
1652                         po->opo_non_exist = 1;
1653                         rc = 0;
1654                 }
1655                 init_rwsem(&po->opo_sem);
1656         }
1657         RETURN(rc);
1658 }
1659
1660 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1661 {
1662         struct osp_object       *obj = lu2osp_obj(o);
1663         struct lu_object_header *h = o->lo_header;
1664
1665         dt_object_fini(&obj->opo_obj);
1666         lu_object_header_fini(h);
1667         if (obj->opo_ooa != NULL) {
1668                 struct osp_xattr_entry *oxe;
1669                 struct osp_xattr_entry *tmp;
1670                 int                     count;
1671
1672                 list_for_each_entry_safe(oxe, tmp,
1673                                          &obj->opo_ooa->ooa_xattr_list,
1674                                          oxe_list) {
1675                         list_del(&oxe->oxe_list);
1676                         count = atomic_read(&oxe->oxe_ref);
1677                         LASSERTF(count == 1,
1678                                  "Still has %d users on the xattr entry %.*s\n",
1679                                  count-1, (int)oxe->oxe_namelen, oxe->oxe_buf);
1680
1681                         OBD_FREE(oxe, oxe->oxe_buflen);
1682                 }
1683                 OBD_FREE_PTR(obj->opo_ooa);
1684         }
1685         OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1686 }
1687
1688 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1689 {
1690         struct osp_object       *po = lu2osp_obj(o);
1691         struct osp_device       *d  = lu2osp_dev(o->lo_dev);
1692
1693         ENTRY;
1694
1695         /*
1696          * release reservation if object was declared but not created
1697          * this may require lu_object_put() in LOD
1698          */
1699         if (unlikely(po->opo_reserved)) {
1700                 LASSERT(d->opd_pre != NULL);
1701                 LASSERT(d->opd_pre_reserved > 0);
1702                 spin_lock(&d->opd_pre_lock);
1703                 d->opd_pre_reserved--;
1704                 spin_unlock(&d->opd_pre_lock);
1705
1706                 /* not needed in cache any more */
1707                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1708         }
1709
1710         if (is_ost_obj(o))
1711                 /* XXX: Currently, NOT cache OST-object on MDT because:
1712                  *      1. it is not often accessed on MDT.
1713                  *      2. avoid up layer (such as LFSCK) to load too many
1714                  *         once-used OST-objects. */
1715                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1716
1717         EXIT;
1718 }
1719
1720 static int osp_object_print(const struct lu_env *env, void *cookie,
1721                             lu_printer_t p, const struct lu_object *l)
1722 {
1723         const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1724
1725         return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1726 }
1727
1728 static int osp_object_invariant(const struct lu_object *o)
1729 {
1730         LBUG();
1731 }
1732
1733 struct lu_object_operations osp_lu_obj_ops = {
1734         .loo_object_init        = osp_object_init,
1735         .loo_object_free        = osp_object_free,
1736         .loo_object_release     = osp_object_release,
1737         .loo_object_print       = osp_object_print,
1738         .loo_object_invariant   = osp_object_invariant
1739 };