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