lustre/osp/osp_object.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/osp/osp_object.c
  37  *
  38  * Lustre OST Proxy Device
  39  *
  40  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  41  * Author: Mikhail Pershin <mike.tappro@intel.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_MDS
  45
  46 #include "osp_internal.h"
  47
  48 static inline bool is_ost_obj(struct lu_object *lo)
  49 {
  50         return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
  51 }
  52
  53 static void osp_object_assign_fid(const struct lu_env *env,
  54                                  struct osp_device *d, struct osp_object *o)
  55 {
  56         struct osp_thread_info *osi = osp_env_info(env);
  57
  58         LASSERT(fid_is_zero(lu_object_fid(&o->opo_obj.do_lu)));
  59         LASSERT(o->opo_reserved);
  60         o->opo_reserved = 0;
  61
  62         osp_precreate_get_fid(env, d, &osi->osi_fid);
  63
  64         lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
  65 }
  66
  67 static int osp_oac_init(struct osp_object *obj)
  68 {
  69         struct osp_object_attr *ooa;
  70
  71         OBD_ALLOC_PTR(ooa);
  72         if (ooa == NULL)
  73                 return -ENOMEM;
  74
  75         INIT_LIST_HEAD(&ooa->ooa_xattr_list);
  76         spin_lock(&obj->opo_lock);
  77         if (likely(obj->opo_ooa == NULL)) {
  78                 obj->opo_ooa = ooa;
  79                 spin_unlock(&obj->opo_lock);
  80         } else {
  81                 spin_unlock(&obj->opo_lock);
  82                 OBD_FREE_PTR(ooa);
  83         }
  84
  85         return 0;
  86 }
  87
  88 static struct osp_xattr_entry *
  89 osp_oac_xattr_find_locked(struct osp_object_attr *ooa,
  90                           const char *name, int namelen, bool unlink)
  91 {
  92         struct osp_xattr_entry *oxe;
  93
  94         list_for_each_entry(oxe, &ooa->ooa_xattr_list, oxe_list) {
  95                 if (namelen == oxe->oxe_namelen &&
  96                     strncmp(name, oxe->oxe_buf, namelen) == 0) {
  97                         if (unlink)
  98                                 list_del_init(&oxe->oxe_list);
  99                         else
 100                                 atomic_inc(&oxe->oxe_ref);
 101
 102                         return oxe;
 103                 }
 104         }
 105
 106         return NULL;
 107 }
 108
 109 static struct osp_xattr_entry *osp_oac_xattr_find(struct osp_object *obj,
 110                                                   const char *name)
 111 {
 112         struct osp_xattr_entry *oxe = NULL;
 113
 114         spin_lock(&obj->opo_lock);
 115         if (obj->opo_ooa != NULL)
 116                 oxe = osp_oac_xattr_find_locked(obj->opo_ooa, name,
 117                                                 strlen(name), false);
 118         spin_unlock(&obj->opo_lock);
 119
 120         return oxe;
 121 }
 122
 123 static struct osp_xattr_entry *
 124 osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, int len)
 125 {
 126         struct osp_object_attr *ooa     = obj->opo_ooa;
 127         struct osp_xattr_entry *oxe;
 128         struct osp_xattr_entry *tmp     = NULL;
 129         int                     namelen = strlen(name);
 130         int                     size    = sizeof(*oxe) + namelen + 1 + len;
 131
 132         LASSERT(ooa != NULL);
 133
 134         oxe = osp_oac_xattr_find(obj, name);
 135         if (oxe != NULL)
 136                 return oxe;
 137
 138         OBD_ALLOC(oxe, size);
 139         if (unlikely(oxe == NULL))
 140                 return NULL;
 141
 142         INIT_LIST_HEAD(&oxe->oxe_list);
 143         oxe->oxe_buflen = size;
 144         oxe->oxe_namelen = namelen;
 145         memcpy(oxe->oxe_buf, name, namelen);
 146         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
 147         /* One ref is for the caller, the other is for the entry on the list. */
 148         atomic_set(&oxe->oxe_ref, 2);
 149
 150         spin_lock(&obj->opo_lock);
 151         tmp = osp_oac_xattr_find_locked(ooa, name, namelen, false);
 152         if (tmp == NULL)
 153                 list_add_tail(&oxe->oxe_list, &ooa->ooa_xattr_list);
 154         spin_unlock(&obj->opo_lock);
 155
 156         if (tmp != NULL) {
 157                 OBD_FREE(oxe, size);
 158                 oxe = tmp;
 159         }
 160
 161         return oxe;
 162 }
 163
 164 static struct osp_xattr_entry *
 165 osp_oac_xattr_replace(struct osp_object *obj,
 166                       struct osp_xattr_entry **poxe, int len)
 167 {
 168         struct osp_object_attr *ooa     = obj->opo_ooa;
 169         struct osp_xattr_entry *old     = *poxe;
 170         struct osp_xattr_entry *oxe;
 171         struct osp_xattr_entry *tmp     = NULL;
 172         int                     namelen = old->oxe_namelen;
 173         int                     size    = sizeof(*oxe) + namelen + 1 + len;
 174
 175         LASSERT(ooa != NULL);
 176
 177         OBD_ALLOC(oxe, size);
 178         if (unlikely(oxe == NULL))
 179                 return NULL;
 180
 181         INIT_LIST_HEAD(&oxe->oxe_list);
 182         oxe->oxe_buflen = size;
 183         oxe->oxe_namelen = namelen;
 184         memcpy(oxe->oxe_buf, old->oxe_buf, namelen);
 185         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
 186         /* One ref is for the caller, the other is for the entry on the list. */
 187         atomic_set(&oxe->oxe_ref, 2);
 188
 189         spin_lock(&obj->opo_lock);
 190         tmp = osp_oac_xattr_find_locked(ooa, oxe->oxe_buf, namelen, true);
 191         list_add_tail(&oxe->oxe_list, &ooa->ooa_xattr_list);
 192         spin_unlock(&obj->opo_lock);
 193
 194         *poxe = tmp;
 195         LASSERT(tmp != NULL);
 196
 197         return oxe;
 198 }
 199
 200 static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
 201 {
 202         if (atomic_dec_and_test(&oxe->oxe_ref)) {
 203                 LASSERT(list_empty(&oxe->oxe_list));
 204
 205                 OBD_FREE(oxe, oxe->oxe_buflen);
 206         }
 207 }
 208
 209 static int osp_get_attr_from_reply(const struct lu_env *env,
 210                                    struct update_reply *reply,
 211                                    struct lu_attr *attr,
 212                                    struct osp_object *obj, int index)
 213 {
 214         struct osp_thread_info  *osi    = osp_env_info(env);
 215         struct lu_buf           *rbuf   = &osi->osi_lb2;
 216         struct obdo             *lobdo  = &osi->osi_obdo;
 217         struct obdo             *wobdo;
 218         int                      rc;
 219
 220         rc = update_get_reply_buf(reply, rbuf, index);
 221         if (rc < 0)
 222                 return rc;
 223
 224         wobdo = rbuf->lb_buf;
 225         if (rbuf->lb_len != sizeof(*wobdo))
 226                 return -EPROTO;
 227
 228         obdo_le_to_cpu(wobdo, wobdo);
 229         lustre_get_wire_obdo(NULL, lobdo, wobdo);
 230         spin_lock(&obj->opo_lock);
 231         if (obj->opo_ooa != NULL) {
 232                 la_from_obdo(&obj->opo_ooa->ooa_attr, lobdo, lobdo->o_valid);
 233                 if (attr != NULL)
 234                         *attr = obj->opo_ooa->ooa_attr;
 235         } else {
 236                 LASSERT(attr != NULL);
 237
 238                 la_from_obdo(attr, lobdo, lobdo->o_valid);
 239         }
 240         spin_unlock(&obj->opo_lock);
 241
 242         return 0;
 243 }
 244
 245 static int osp_attr_get_interpterer(const struct lu_env *env,
 246                                     struct update_reply *reply,
 247                                     struct osp_object *obj,
 248                                     void *data, int index, int rc)
 249 {
 250         struct lu_attr *attr = data;
 251
 252         LASSERT(obj->opo_ooa != NULL);
 253
 254         if (rc == 0) {
 255                 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
 256                 obj->opo_non_exist = 0;
 257
 258                 return osp_get_attr_from_reply(env, reply, NULL, obj, index);
 259         } else {
 260                 if (rc == -ENOENT) {
 261                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
 262                         obj->opo_non_exist = 1;
 263                 }
 264
 265                 spin_lock(&obj->opo_lock);
 266                 attr->la_valid = 0;
 267                 spin_unlock(&obj->opo_lock);
 268         }
 269
 270         return 0;
 271 }
 272
 273 static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
 274                                 struct lustre_capa *capa)
 275 {
 276         struct osp_object       *obj    = dt2osp_obj(dt);
 277         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 278         struct update_request   *update;
 279         int                      rc     = 0;
 280
 281         if (obj->opo_ooa == NULL) {
 282                 rc = osp_oac_init(obj);
 283                 if (rc != 0)
 284                         return rc;
 285         }
 286
 287         mutex_lock(&osp->opd_async_requests_mutex);
 288         update = osp_find_or_create_async_update_request(osp);
 289         if (IS_ERR(update))
 290                 rc = PTR_ERR(update);
 291         else
 292                 rc = osp_insert_async_update(env, update, OBJ_ATTR_GET, obj, 0,
 293                                              NULL, NULL,
 294                                              &obj->opo_ooa->ooa_attr,
 295                                              osp_attr_get_interpterer);
 296         mutex_unlock(&osp->opd_async_requests_mutex);
 297
 298         return rc;
 299 }
 300
 301 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
 302                  struct lu_attr *attr, struct lustre_capa *capa)
 303 {
 304         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 305         struct osp_object       *obj    = dt2osp_obj(dt);
 306         struct dt_device        *dev    = &osp->opd_dt_dev;
 307         struct update_request   *update;
 308         struct update_reply     *reply;
 309         struct ptlrpc_request   *req    = NULL;
 310         int                      rc     = 0;
 311         ENTRY;
 312
 313         if (is_ost_obj(&dt->do_lu) && obj->opo_non_exist)
 314                 RETURN(-ENOENT);
 315
 316         if (obj->opo_ooa != NULL) {
 317                 spin_lock(&obj->opo_lock);
 318                 if (obj->opo_ooa->ooa_attr.la_valid != 0) {
 319                         *attr = obj->opo_ooa->ooa_attr;
 320                         spin_unlock(&obj->opo_lock);
 321
 322                         RETURN(0);
 323                 }
 324                 spin_unlock(&obj->opo_lock);
 325         }
 326
 327         update = out_create_update_req(dev);
 328         if (IS_ERR(update))
 329                 RETURN(PTR_ERR(update));
 330
 331         rc = out_insert_update(env, update, OBJ_ATTR_GET,
 332                                lu_object_fid(&dt->do_lu), 0, NULL, NULL);
 333         if (rc != 0) {
 334                 CERROR("%s: Insert update error "DFID": rc = %d\n",
 335                        dev->dd_lu_dev.ld_obd->obd_name,
 336                        PFID(lu_object_fid(&dt->do_lu)), rc);
 337
 338                 GOTO(out, rc);
 339         }
 340
 341         rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
 342         if (rc != 0) {
 343                 if (rc == -ENOENT) {
 344                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
 345                         obj->opo_non_exist = 1;
 346                 } else {
 347                         CERROR("%s:osp_attr_get update error "DFID": rc = %d\n",
 348                                dev->dd_lu_dev.ld_obd->obd_name,
 349                                PFID(lu_object_fid(&dt->do_lu)), rc);
 350                 }
 351
 352                 GOTO(out, rc);
 353         }
 354
 355         osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
 356         obj->opo_non_exist = 0;
 357         reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
 358                                              UPDATE_BUFFER_SIZE);
 359         if (reply == NULL || reply->ur_version != UPDATE_REPLY_V1)
 360                 GOTO(out, rc = -EPROTO);
 361
 362         rc = osp_get_attr_from_reply(env, reply, attr, obj, 0);
 363         if (rc != 0)
 364                 GOTO(out, rc);
 365
 366         if (!is_ost_obj(&dt->do_lu)) {
 367                 if (attr->la_flags == 1)
 368                         obj->opo_empty = 0;
 369                 else
 370                         obj->opo_empty = 1;
 371         }
 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_declare_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
 392         ENTRY;
 393
 394         /*
 395          * Usually we don't allow server stack to manipulate size
 396          * but there is a special case when striping is created
 397          * late, after stripless file got truncated to non-zero.
 398          *
 399          * In this case we do the following:
 400          *
 401          * 1) grab id in declare - this can lead to leaked OST objects
 402          *    but we don't currently have proper mechanism and the only
 403          *    options we have are to do truncate RPC holding transaction
 404          *    open (very bad) or to grab id in declare at cost of leaked
 405          *    OST object in same very rare unfortunate case (just bad)
 406          *    notice 1.6-2.0 do assignment outside of running transaction
 407          *    all the time, meaning many more chances for leaked objects.
 408          *
 409          * 2) send synchronous truncate RPC with just assigned id
 410          */
 411
 412         /* there are few places in MDD code still passing NULL
 413          * XXX: to be fixed soon */
 414         if (attr == NULL)
 415                 RETURN(0);
 416
 417         if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
 418             fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
 419                 LASSERT(!dt_object_exists(dt));
 420                 osp_object_assign_fid(env, d, o);
 421                 rc = osp_object_truncate(env, dt, attr->la_size);
 422                 if (rc)
 423                         RETURN(rc);
 424         }
 425
 426         if (o->opo_new) {
 427                 /* no need in logging for new objects being created */
 428                 RETURN(0);
 429         }
 430
 431         if (!(attr->la_valid & (LA_UID | LA_GID)))
 432                 RETURN(0);
 433
 434         /*
 435          * track all UID/GID changes via llog
 436          */
 437         rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
 438         if (rc != 0 || o->opo_ooa == NULL)
 439                 RETURN(rc);
 440
 441         la = &o->opo_ooa->ooa_attr;
 442         spin_lock(&o->opo_lock);
 443         if (attr->la_valid & LA_UID) {
 444                 la->la_uid = attr->la_uid;
 445                 la->la_valid |= LA_UID;
 446         }
 447
 448         if (attr->la_valid & LA_GID) {
 449                 la->la_gid = attr->la_gid;
 450                 la->la_valid |= LA_GID;
 451         }
 452         spin_unlock(&o->opo_lock);
 453
 454         RETURN(0);
 455 }
 456
 457 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
 458                         const struct lu_attr *attr, struct thandle *th,
 459                         struct lustre_capa *capa)
 460 {
 461         struct osp_object       *o = dt2osp_obj(dt);
 462         int                      rc = 0;
 463
 464         ENTRY;
 465
 466         /* we're interested in uid/gid changes only */
 467         if (!(attr->la_valid & (LA_UID | LA_GID)))
 468                 RETURN(0);
 469
 470         /* new object, the very first ->attr_set()
 471          * initializing attributes needs no logging
 472          * all subsequent one are subject to the
 473          * logging and synchronization with OST */
 474         if (o->opo_new) {
 475                 o->opo_new = 0;
 476                 RETURN(0);
 477         }
 478
 479         /*
 480          * once transaction is committed put proper command on
 481          * the queue going to our OST
 482          */
 483         rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
 484
 485         /* XXX: send new uid/gid to OST ASAP? */
 486
 487         RETURN(rc);
 488 }
 489
 490 static int osp_xattr_get_interpterer(const struct lu_env *env,
 491                                      struct update_reply *reply,
 492                                      struct osp_object *obj,
 493                                      void *data, int index, int rc)
 494 {
 495         struct osp_object_attr  *ooa  = obj->opo_ooa;
 496         struct osp_xattr_entry  *oxe  = data;
 497         struct lu_buf           *rbuf = &osp_env_info(env)->osi_lb2;
 498
 499         LASSERT(ooa != NULL);
 500
 501         if (rc == 0) {
 502                 int len = sizeof(*oxe) + oxe->oxe_namelen + 1;
 503
 504                 rc = update_get_reply_buf(reply, rbuf, index);
 505                 if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
 506                         spin_lock(&obj->opo_lock);
 507                         oxe->oxe_ready = 0;
 508                         spin_unlock(&obj->opo_lock);
 509                         osp_oac_xattr_put(oxe);
 510
 511                         return rc < 0 ? rc : -ERANGE;
 512                 }
 513
 514                 spin_lock(&obj->opo_lock);
 515                 oxe->oxe_vallen = rbuf->lb_len;
 516                 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
 517                 oxe->oxe_exist = 1;
 518                 oxe->oxe_ready = 1;
 519                 spin_unlock(&obj->opo_lock);
 520         } else if (rc == -ENOENT || rc == -ENODATA) {
 521                 spin_lock(&obj->opo_lock);
 522                 oxe->oxe_exist = 0;
 523                 oxe->oxe_ready = 1;
 524                 spin_unlock(&obj->opo_lock);
 525         } else {
 526                 spin_lock(&obj->opo_lock);
 527                 oxe->oxe_ready = 0;
 528                 spin_unlock(&obj->opo_lock);
 529         }
 530
 531         osp_oac_xattr_put(oxe);
 532
 533         return 0;
 534 }
 535
 536 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
 537                                  struct lu_buf *buf, const char *name,
 538                                  struct lustre_capa *capa)
 539 {
 540         struct osp_object       *obj     = dt2osp_obj(dt);
 541         struct osp_device       *osp     = lu2osp_dev(dt->do_lu.lo_dev);
 542         struct update_request   *update;
 543         struct osp_xattr_entry  *oxe;
 544         int                      namelen = strlen(name);
 545         int                      rc      = 0;
 546
 547         LASSERT(buf != NULL);
 548         LASSERT(name != NULL);
 549
 550         /* If only for xattr size, return directly. */
 551         if (unlikely(buf->lb_len == 0))
 552                 return 0;
 553
 554         if (obj->opo_ooa == NULL) {
 555                 rc = osp_oac_init(obj);
 556                 if (rc != 0)
 557                         return rc;
 558         }
 559
 560         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
 561         if (oxe == NULL)
 562                 return -ENOMEM;
 563
 564         mutex_lock(&osp->opd_async_requests_mutex);
 565         update = osp_find_or_create_async_update_request(osp);
 566         if (IS_ERR(update)) {
 567                 rc = PTR_ERR(update);
 568                 mutex_unlock(&osp->opd_async_requests_mutex);
 569                 osp_oac_xattr_put(oxe);
 570         } else {
 571                 rc = osp_insert_async_update(env, update, OBJ_XATTR_GET, obj,
 572                                              1, &namelen, &name, oxe,
 573                                              osp_xattr_get_interpterer);
 574                 if (rc != 0) {
 575                         mutex_unlock(&osp->opd_async_requests_mutex);
 576                         osp_oac_xattr_put(oxe);
 577                 } else {
 578                         /* XXX: Currently, we trigger the batched async OUT
 579                          *      RPC via dt_declare_xattr_get(). It is not
 580                          *      perfect solution, but works well now.
 581                          *
 582                          *      We will improve it in the future. */
 583                         update = osp->opd_async_requests;
 584                         if (update != NULL && update->ur_buf != NULL &&
 585                             update->ur_buf->ub_count > 0) {
 586                                 osp->opd_async_requests = NULL;
 587                                 mutex_unlock(&osp->opd_async_requests_mutex);
 588                                 rc = osp_unplug_async_update(env, osp, update);
 589                         } else {
 590                                 mutex_unlock(&osp->opd_async_requests_mutex);
 591                         }
 592                 }
 593         }
 594
 595         return rc;
 596 }
 597
 598 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
 599                   struct lu_buf *buf, const char *name,
 600                   struct lustre_capa *capa)
 601 {
 602         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
 603         struct osp_object       *obj    = dt2osp_obj(dt);
 604         struct dt_device        *dev    = &osp->opd_dt_dev;
 605         struct lu_buf           *rbuf   = &osp_env_info(env)->osi_lb2;
 606         struct update_request   *update = NULL;
 607         struct ptlrpc_request   *req    = NULL;
 608         struct update_reply     *reply;
 609         struct osp_xattr_entry  *oxe    = NULL;
 610         const char              *dname  = dt->do_lu.lo_dev->ld_obd->obd_name;
 611         int                      namelen;
 612         int                      rc     = 0;
 613         ENTRY;
 614
 615         LASSERT(buf != NULL);
 616         LASSERT(name != NULL);
 617
 618         if (unlikely(obj->opo_non_exist))
 619                 RETURN(-ENOENT);
 620
 621         oxe = osp_oac_xattr_find(obj, name);
 622         if (oxe != NULL) {
 623                 spin_lock(&obj->opo_lock);
 624                 if (oxe->oxe_ready) {
 625                         if (!oxe->oxe_exist)
 626                                 GOTO(unlock, rc = -ENODATA);
 627
 628                         if (buf->lb_buf == NULL)
 629                                 GOTO(unlock, rc = oxe->oxe_vallen);
 630
 631                         if (buf->lb_len < oxe->oxe_vallen)
 632                                 GOTO(unlock, rc = -ERANGE);
 633
 634                         memcpy(buf->lb_buf, oxe->oxe_value, oxe->oxe_vallen);
 635
 636                         GOTO(unlock, rc = oxe->oxe_vallen);
 637
 638 unlock:
 639                         spin_unlock(&obj->opo_lock);
 640                         osp_oac_xattr_put(oxe);
 641
 642                         return rc;
 643                 }
 644                 spin_unlock(&obj->opo_lock);
 645         }
 646
 647         update = out_create_update_req(dev);
 648         if (IS_ERR(update))
 649                 GOTO(out, rc = PTR_ERR(update));
 650
 651         namelen = strlen(name);
 652         rc = out_insert_update(env, update, OBJ_XATTR_GET,
 653                                lu_object_fid(&dt->do_lu), 1, &namelen, &name);
 654         if (rc != 0) {
 655                 CERROR("%s: Insert update error "DFID": rc = %d\n",
 656                        dname, PFID(lu_object_fid(&dt->do_lu)), rc);
 657
 658                 GOTO(out, rc);
 659         }
 660
 661         rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
 662         if (rc != 0) {
 663                 if (obj->opo_ooa == NULL)
 664                         GOTO(out, rc);
 665
 666                 if (oxe == NULL)
 667                         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
 668
 669                 if (oxe == NULL) {
 670                         CWARN("%s: Fail to add xattr (%s) to cache for "
 671                               DFID" (1): rc = %d\n", dname, name,
 672                               PFID(lu_object_fid(&dt->do_lu)), rc);
 673
 674                         GOTO(out, rc);
 675                 }
 676
 677                 spin_lock(&obj->opo_lock);
 678                 if (rc == -ENOENT || rc == -ENODATA) {
 679                         oxe->oxe_exist = 0;
 680                         oxe->oxe_ready = 1;
 681                 } else {
 682                         oxe->oxe_ready = 0;
 683                 }
 684                 spin_unlock(&obj->opo_lock);
 685
 686                 GOTO(out, rc);
 687         }
 688
 689         reply = req_capsule_server_sized_get(&req->rq_pill, &RMF_UPDATE_REPLY,
 690                                             UPDATE_BUFFER_SIZE);
 691         if (reply->ur_version != UPDATE_REPLY_V1) {
 692                 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
 693                        dname, reply->ur_version, UPDATE_REPLY_V1,
 694                        PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
 695
 696                 GOTO(out, rc = -EPROTO);
 697         }
 698
 699         rc = update_get_reply_buf(reply, rbuf, 0);
 700         if (rc < 0)
 701                 GOTO(out, rc);
 702
 703         LASSERT(rbuf->lb_len > 0 && rbuf->lb_len < PAGE_CACHE_SIZE);
 704
 705         if (buf->lb_buf == NULL)
 706                 GOTO(out, rc = rbuf->lb_len);
 707
 708         if (unlikely(buf->lb_len < rbuf->lb_len))
 709                 GOTO(out, rc = -ERANGE);
 710
 711         memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
 712         rc = rbuf->lb_len;
 713         if (obj->opo_ooa == NULL)
 714                 GOTO(out, rc);
 715
 716         if (oxe == NULL) {
 717                 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
 718                 if (oxe == NULL) {
 719                         CWARN("%s: Fail to add xattr (%s) to "
 720                               "cache for "DFID" (2): rc = %d\n",
 721                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
 722
 723                         GOTO(out, rc);
 724                 }
 725         }
 726
 727         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
 728                 struct osp_xattr_entry *old = oxe;
 729                 struct osp_xattr_entry *tmp;
 730
 731                 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
 732                 osp_oac_xattr_put(oxe);
 733                 oxe = tmp;
 734                 if (tmp == NULL) {
 735                         CWARN("%s: Fail to update xattr (%s) to "
 736                               "cache for "DFID": rc = %d\n",
 737                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
 738                         spin_lock(&obj->opo_lock);
 739                         oxe->oxe_ready = 0;
 740                         spin_unlock(&obj->opo_lock);
 741
 742                         GOTO(out, rc);
 743                 }
 744
 745                 /* Drop the ref for entry on list. */
 746                 osp_oac_xattr_put(old);
 747         }
 748
 749         spin_lock(&obj->opo_lock);
 750         oxe->oxe_vallen = rbuf->lb_len;
 751         memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
 752         oxe->oxe_exist = 1;
 753         oxe->oxe_ready = 1;
 754         spin_unlock(&obj->opo_lock);
 755
 756         GOTO(out, rc);
 757
 758 out:
 759         if (req != NULL)
 760                 ptlrpc_req_finished(req);
 761
 762         if (update != NULL && !IS_ERR(update))
 763                 out_destroy_update_req(update);
 764
 765         if (oxe != NULL)
 766                 osp_oac_xattr_put(oxe);
 767
 768         return rc;
 769 }
 770
 771 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
 772                           const struct lu_buf *buf, const char *name,
 773                           int flag, struct thandle *th)
 774 {
 775         struct osp_object       *o       = dt2osp_obj(dt);
 776         struct update_request   *update;
 777         struct lu_fid           *fid;
 778         struct osp_xattr_entry  *oxe;
 779         int                     sizes[3] = {strlen(name), buf->lb_len,
 780                                             sizeof(int)};
 781         char                    *bufs[3] = {(char *)name, (char *)buf->lb_buf };
 782         int                     rc;
 783
 784         LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
 785
 786         update = out_find_create_update_loc(th, dt);
 787         if (IS_ERR(update)) {
 788                 CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
 789                        dt->do_lu.lo_dev->ld_obd->obd_name,
 790                        PFID(lu_object_fid(&dt->do_lu)),
 791                        (int)PTR_ERR(update));
 792
 793                 return PTR_ERR(update);
 794         }
 795
 796         flag = cpu_to_le32(flag);
 797         bufs[2] = (char *)&flag;
 798
 799         fid = (struct lu_fid *)lu_object_fid(&dt->do_lu);
 800         rc = out_insert_update(env, update, OBJ_XATTR_SET, fid,
 801                                ARRAY_SIZE(sizes), sizes, (const char **)bufs);
 802         if (rc != 0 || o->opo_ooa == NULL)
 803                 return rc;
 804
 805         oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
 806         if (oxe == NULL) {
 807                 CWARN("%s: Fail to add xattr (%s) to cache for "DFID
 808                       ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
 809                       name, PFID(lu_object_fid(&dt->do_lu)), rc);
 810
 811                 return 0;
 812         }
 813
 814         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
 815                 struct osp_xattr_entry *old = oxe;
 816                 struct osp_xattr_entry *tmp;
 817
 818                 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
 819                 osp_oac_xattr_put(oxe);
 820                 oxe = tmp;
 821                 if (tmp == NULL) {
 822                         CWARN("%s: Fail to update xattr (%s) to cache for "DFID
 823                               ": rc = %d\n", dt->do_lu.lo_dev->ld_obd->obd_name,
 824                               name, PFID(lu_object_fid(&dt->do_lu)), rc);
 825                         spin_lock(&o->opo_lock);
 826                         oxe->oxe_ready = 0;
 827                         spin_unlock(&o->opo_lock);
 828
 829                         return 0;
 830                 }
 831
 832                 /* Drop the ref for entry on list. */
 833                 osp_oac_xattr_put(old);
 834         }
 835
 836         spin_lock(&o->opo_lock);
 837         oxe->oxe_vallen = buf->lb_len;
 838         memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
 839         oxe->oxe_exist = 1;
 840         oxe->oxe_ready = 1;
 841         spin_unlock(&o->opo_lock);
 842         osp_oac_xattr_put(oxe);
 843
 844         return 0;
 845 }
 846
 847 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
 848                   const struct lu_buf *buf, const char *name, int fl,
 849                   struct thandle *th, struct lustre_capa *capa)
 850 {
 851         CDEBUG(D_INFO, "xattr %s set object "DFID"\n", name,
 852                PFID(&dt->do_lu.lo_header->loh_fid));
 853
 854         return 0;
 855 }
 856
 857 static int osp_declare_object_create(const struct lu_env *env,
 858                                      struct dt_object *dt,
 859                                      struct lu_attr *attr,
 860                                      struct dt_allocation_hint *hint,
 861                                      struct dt_object_format *dof,
 862                                      struct thandle *th)
 863 {
 864         struct osp_thread_info  *osi = osp_env_info(env);
 865         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 866         struct osp_object       *o = dt2osp_obj(dt);
 867         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 868         int                      rc = 0;
 869
 870         ENTRY;
 871
 872         if (is_remote_trans(th)) {
 873                 LASSERT(fid_is_sane(fid));
 874
 875                 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
 876
 877                 RETURN(rc);
 878         }
 879
 880         /* should happen to non-0 OSP only so that at least one object
 881          * has been already declared in the scenario and LOD should
 882          * cleanup that */
 883         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
 884                 RETURN(-ENOSPC);
 885
 886         LASSERT(d->opd_last_used_oid_file);
 887
 888         /*
 889          * There can be gaps in precreated ids and record to unlink llog
 890          * XXX: we do not handle gaps yet, implemented before solution
 891          *      was found to be racy, so we disabled that. there is no
 892          *      point in making useless but expensive llog declaration.
 893          */
 894         /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
 895
 896         if (unlikely(!fid_is_zero(fid))) {
 897                 /* replay case: caller knows fid */
 898                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
 899                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
 900                                              sizeof(osi->osi_id), osi->osi_off,
 901                                              th);
 902                 RETURN(rc);
 903         }
 904
 905         /*
 906          * in declaration we need to reserve object so that we don't block
 907          * awaiting precreation RPC to complete
 908          */
 909         rc = osp_precreate_reserve(env, d);
 910         /*
 911          * we also need to declare update to local "last used id" file for
 912          * recovery if object isn't used for a reason, we need to release
 913          * reservation, this can be made in osd_object_release()
 914          */
 915         if (rc == 0) {
 916                 /* mark id is reserved: in create we don't want to talk
 917                  * to OST */
 918                 LASSERT(o->opo_reserved == 0);
 919                 o->opo_reserved = 1;
 920
 921                 /* common for all OSPs file hystorically */
 922                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
 923                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
 924                                              sizeof(osi->osi_id), osi->osi_off,
 925                                              th);
 926         } else {
 927                 /* not needed in the cache anymore */
 928                 set_bit(LU_OBJECT_HEARD_BANSHEE,
 929                             &dt->do_lu.lo_header->loh_flags);
 930         }
 931         RETURN(rc);
 932 }
 933
 934 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
 935                              struct lu_attr *attr,
 936                              struct dt_allocation_hint *hint,
 937                              struct dt_object_format *dof, struct thandle *th)
 938 {
 939         struct osp_thread_info  *osi = osp_env_info(env);
 940         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
 941         struct osp_object       *o = dt2osp_obj(dt);
 942         int                     rc = 0;
 943         struct lu_fid           *fid = &osi->osi_fid;
 944         ENTRY;
 945
 946         if (is_remote_trans(th)) {
 947                 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
 948
 949                 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
 950                 if (rc == 0)
 951                         o->opo_non_exist = 0;
 952
 953                 RETURN(rc);
 954         }
 955
 956         o->opo_non_exist = 0;
 957         if (o->opo_reserved) {
 958                 /* regular case, fid is assigned holding trunsaction open */
 959                  osp_object_assign_fid(env, d, o);
 960         }
 961
 962         memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
 963
 964         LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
 965                  o, PFID(fid));
 966
 967         if (!o->opo_reserved) {
 968                 /* special case, id was assigned outside of transaction
 969                  * see comments in osp_declare_attr_set */
 970                 LASSERT(d->opd_pre != NULL);
 971                 spin_lock(&d->opd_pre_lock);
 972                 osp_update_last_fid(d, fid);
 973                 spin_unlock(&d->opd_pre_lock);
 974         }
 975
 976         CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
 977
 978         /* If the precreate ends, it means it will be ready to rollover to
 979          * the new sequence soon, all the creation should be synchronized,
 980          * otherwise during replay, the replay fid will be inconsistent with
 981          * last_used/create fid */
 982         if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
 983                 th->th_sync = 1;
 984
 985         /*
 986          * it's OK if the import is inactive by this moment - id was created
 987          * by OST earlier, we just need to maintain it consistently on the disk
 988          * once import is reconnected, OSP will claim this and other objects
 989          * used and OST either keep them, if they exist or recreate
 990          */
 991
 992         /* we might have lost precreated objects */
 993         if (unlikely(d->opd_gap_count) > 0) {
 994                 LASSERT(d->opd_pre != NULL);
 995                 spin_lock(&d->opd_pre_lock);
 996                 if (d->opd_gap_count > 0) {
 997                         int count = d->opd_gap_count;
 998
 999                         ostid_set_id(&osi->osi_oi,
1000                                      fid_oid(&d->opd_gap_start_fid));
1001                         d->opd_gap_count = 0;
1002                         spin_unlock(&d->opd_pre_lock);
1003
1004                         CDEBUG(D_HA, "Writting gap "DFID"+%d in llog\n",
1005                                PFID(&d->opd_gap_start_fid), count);
1006                         /* real gap handling is disabled intil ORI-692 will be
1007                          * fixed, now we only report gaps */
1008                 } else {
1009                         spin_unlock(&d->opd_pre_lock);
1010                 }
1011         }
1012
1013         /* new object, the very first ->attr_set()
1014          * initializing attributes needs no logging */
1015         o->opo_new = 1;
1016
1017         /* Only need update last_used oid file, seq file will only be update
1018          * during seq rollover */
1019         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1020                            &d->opd_last_used_fid.f_oid, d->opd_index);
1021
1022         rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1023                              &osi->osi_off, th);
1024
1025         CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1026                d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1027
1028         RETURN(rc);
1029 }
1030
1031 int osp_declare_object_destroy(const struct lu_env *env,
1032                                struct dt_object *dt, struct thandle *th)
1033 {
1034         struct osp_object       *o = dt2osp_obj(dt);
1035         int                      rc = 0;
1036
1037         ENTRY;
1038
1039         /*
1040          * track objects to be destroyed via llog
1041          */
1042         rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1043
1044         RETURN(rc);
1045 }
1046
1047 int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1048                        struct thandle *th)
1049 {
1050         struct osp_object       *o = dt2osp_obj(dt);
1051         int                      rc = 0;
1052
1053         ENTRY;
1054
1055         o->opo_non_exist = 1;
1056         /*
1057          * once transaction is committed put proper command on
1058          * the queue going to our OST
1059          */
1060         rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1061
1062         /* not needed in cache any more */
1063         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1064
1065         RETURN(rc);
1066 }
1067
1068 struct dt_object_operations osp_obj_ops = {
1069         .do_declare_attr_get    = osp_declare_attr_get,
1070         .do_attr_get            = osp_attr_get,
1071         .do_declare_attr_set    = osp_declare_attr_set,
1072         .do_attr_set            = osp_attr_set,
1073         .do_declare_xattr_get   = osp_declare_xattr_get,
1074         .do_xattr_get           = osp_xattr_get,
1075         .do_declare_xattr_set   = osp_declare_xattr_set,
1076         .do_xattr_set           = osp_xattr_set,
1077         .do_declare_create      = osp_declare_object_create,
1078         .do_create              = osp_object_create,
1079         .do_declare_destroy     = osp_declare_object_destroy,
1080         .do_destroy             = osp_object_destroy,
1081 };
1082
1083 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
1084                            const struct lu_object_conf *conf)
1085 {
1086         struct osp_object       *po = lu2osp_obj(o);
1087         int                     rc = 0;
1088         ENTRY;
1089
1090         spin_lock_init(&po->opo_lock);
1091         o->lo_header->loh_attr |= LOHA_REMOTE;
1092
1093         if (is_ost_obj(o)) {
1094                 po->opo_obj.do_ops = &osp_obj_ops;
1095         } else {
1096                 struct lu_attr *la = &osp_env_info(env)->osi_attr;
1097
1098                 po->opo_obj.do_ops = &osp_md_obj_ops;
1099                 rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
1100                                                      la, NULL);
1101                 if (rc == 0)
1102                         o->lo_header->loh_attr |=
1103                                 LOHA_EXISTS | (la->la_mode & S_IFMT);
1104                 if (rc == -ENOENT) {
1105                         po->opo_non_exist = 1;
1106                         rc = 0;
1107                 }
1108                 init_rwsem(&po->opo_sem);
1109         }
1110         RETURN(rc);
1111 }
1112
1113 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
1114 {
1115         struct osp_object       *obj = lu2osp_obj(o);
1116         struct lu_object_header *h = o->lo_header;
1117
1118         dt_object_fini(&obj->opo_obj);
1119         lu_object_header_fini(h);
1120         if (obj->opo_ooa != NULL) {
1121                 struct osp_xattr_entry *oxe;
1122                 struct osp_xattr_entry *tmp;
1123                 int                     count;
1124
1125                 list_for_each_entry_safe(oxe, tmp,
1126                                          &obj->opo_ooa->ooa_xattr_list,
1127                                          oxe_list) {
1128                         list_del(&oxe->oxe_list);
1129                         count = atomic_read(&oxe->oxe_ref);
1130                         LASSERTF(count == 1,
1131                                  "Still has %d users on the xattr entry %.*s\n",
1132                                  count - 1, oxe->oxe_namelen, oxe->oxe_buf);
1133
1134                         OBD_FREE(oxe, oxe->oxe_buflen);
1135                 }
1136                 OBD_FREE_PTR(obj->opo_ooa);
1137         }
1138         OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
1139 }
1140
1141 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
1142 {
1143         struct osp_object       *po = lu2osp_obj(o);
1144         struct osp_device       *d  = lu2osp_dev(o->lo_dev);
1145
1146         ENTRY;
1147
1148         /*
1149          * release reservation if object was declared but not created
1150          * this may require lu_object_put() in LOD
1151          */
1152         if (unlikely(po->opo_reserved)) {
1153                 LASSERT(d->opd_pre != NULL);
1154                 LASSERT(d->opd_pre_reserved > 0);
1155                 spin_lock(&d->opd_pre_lock);
1156                 d->opd_pre_reserved--;
1157                 spin_unlock(&d->opd_pre_lock);
1158
1159                 /* not needed in cache any more */
1160                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1161         }
1162
1163         if (is_ost_obj(o))
1164                 /* XXX: Currently, NOT cache OST-object on MDT because:
1165                  *      1. it is not often accessed on MDT.
1166                  *      2. avoid up layer (such as LFSCK) to load too many
1167                  *         once-used OST-objects. */
1168                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
1169
1170         EXIT;
1171 }
1172
1173 static int osp_object_print(const struct lu_env *env, void *cookie,
1174                             lu_printer_t p, const struct lu_object *l)
1175 {
1176         const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
1177
1178         return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
1179 }
1180
1181 static int osp_object_invariant(const struct lu_object *o)
1182 {
1183         LBUG();
1184 }
1185
1186 struct lu_object_operations osp_lu_obj_ops = {
1187         .loo_object_init        = osp_object_init,
1188         .loo_object_free        = osp_object_free,
1189         .loo_object_release     = osp_object_release,
1190         .loo_object_print       = osp_object_print,
1191         .loo_object_invariant   = osp_object_invariant
1192 };