lustre/obdclass/llog_osd.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  */
  31 /*
  32  * lustre/obdclass/llog_osd.c
  33  *
  34  * Low level llog routines on top of OSD API
  35  *
  36  * This file provides set of methods for llog operations on top of
  37  * dt_device. It contains all supported llog_operations interfaces and
  38  * supplimental functions.
  39  *
  40  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
  41  * Author: Mikhail Pershin <mike.pershin@intel.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_LOG
  45
  46 #include <linux/delay.h>
  47
  48 #include <dt_object.h>
  49 #include <llog_swab.h>
  50 #include <lustre_fid.h>
  51 #include <obd.h>
  52 #include <obd_class.h>
  53
  54 #include "llog_internal.h"
  55 #include "local_storage.h"
  56
  57 /**
  58  * Implementation of the llog_operations::lop_declare_create
  59  *
  60  * This function is a wrapper over local_storage API function
  61  * local_object_declare_create().
  62  *
  63  * \param[in] env       execution environment
  64  * \param[in] los       local_storage for bottom storage device
  65  * \param[in] o         dt_object to create
  66  * \param[in] th        current transaction handle
  67  *
  68  * \retval              0 on successful declaration of the new object
  69  * \retval              negative error if declaration was failed
  70  */
  71 static int llog_osd_declare_new_object(const struct lu_env *env,
  72                                        struct local_oid_storage *los,
  73                                        struct dt_object *o,
  74                                        struct thandle *th)
  75 {
  76         struct llog_thread_info *lgi = llog_info(env);
  77
  78         lgi->lgi_attr.la_valid = LA_MODE;
  79         lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
  80         lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
  81
  82         return local_object_declare_create(env, los, o, &lgi->lgi_attr,
  83                                            &lgi->lgi_dof, th);
  84 }
  85
  86 /**
  87  * Implementation of the llog_operations::lop_create
  88  *
  89  * This function is a wrapper over local_storage API function
  90  * local_object_create().
  91  *
  92  * \param[in] env       execution environment
  93  * \param[in] los       local_storage for bottom storage device
  94  * \param[in] o         dt_object to create
  95  * \param[in] th        current transaction handle
  96  *
  97  * \retval              0 on successful creation of the new object
  98  * \retval              negative error if creation was failed
  99  */
 100 static int llog_osd_create_new_object(const struct lu_env *env,
 101                                       struct local_oid_storage *los,
 102                                       struct dt_object *o,
 103                                       struct thandle *th)
 104 {
 105         struct llog_thread_info *lgi = llog_info(env);
 106
 107         lgi->lgi_attr.la_valid = LA_MODE;
 108         lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
 109         lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
 110
 111         return local_object_create(env, los, o, &lgi->lgi_attr,
 112                                    &lgi->lgi_dof, th);
 113 }
 114
 115 /**
 116  * Implementation of the llog_operations::lop_exist
 117  *
 118  * This function checks that llog exists on storage.
 119  *
 120  * \param[in] handle    llog handle of the current llog
 121  *
 122  * \retval              true if llog object exists and is not just destroyed
 123  * \retval              false if llog doesn't exist or just destroyed
 124  */
 125 static int llog_osd_exist(struct llog_handle *handle)
 126 {
 127         LASSERT(handle->lgh_obj);
 128         return dt_object_exists(handle->lgh_obj) && !handle->lgh_destroyed;
 129 }
 130
 131 static void *rec_tail(struct llog_rec_hdr *rec)
 132 {
 133         return (void *)((char *)rec + rec->lrh_len -
 134                         sizeof(struct llog_rec_tail));
 135 }
 136
 137 /**
 138  * Write a padding record to the llog
 139  *
 140  * This function writes a padding record to the end of llog. That may
 141  * be needed if llog contains records of variable size, e.g. config logs
 142  * or changelogs.
 143  * The padding record just aligns llog to the llog chunk_size boundary if
 144  * the current record doesn't fit in the remaining space.
 145  *
 146  * It allocates full length to avoid two separate writes for header and tail.
 147  * Such 2-steps scheme needs extra protection and complex error handling.
 148  *
 149  * \param[in]     env   execution environment
 150  * \param[in]     o     dt_object to create
 151  * \param[in,out] off   pointer to the padding start offset
 152  * \param[in]     len   padding length
 153  * \param[in]     index index of the padding record in a llog
 154  * \param[in]     th    current transaction handle
 155  *
 156  * \retval              0 on successful padding write
 157  * \retval              negative error if write failed
 158  */
 159 static int llog_osd_pad(const struct lu_env *env, struct dt_object *o,
 160                         loff_t *off, int len, int index, struct thandle *th)
 161 {
 162         struct llog_thread_info *lgi = llog_info(env);
 163         struct llog_rec_hdr     *rec;
 164         struct llog_rec_tail    *tail;
 165         int                      rc;
 166
 167         ENTRY;
 168
 169         LASSERT(th);
 170         LASSERT(off);
 171         LASSERT(len >= LLOG_MIN_REC_SIZE && (len & 0x7) == 0);
 172
 173         OBD_ALLOC(rec, len);
 174         if (rec == NULL)
 175                 RETURN(-ENOMEM);
 176
 177         rec->lrh_len = len;
 178         rec->lrh_index = index;
 179         rec->lrh_type = LLOG_PAD_MAGIC;
 180
 181         tail = rec_tail(rec);
 182         tail->lrt_len = len;
 183         tail->lrt_index = index;
 184
 185         lgi->lgi_buf.lb_buf = rec;
 186         lgi->lgi_buf.lb_len = len;
 187         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
 188         if (rc)
 189                 CERROR("%s: error writing padding record: rc = %d\n",
 190                        o->do_lu.lo_dev->ld_obd->obd_name, rc);
 191
 192         OBD_FREE(rec, len);
 193         RETURN(rc);
 194 }
 195
 196 /**
 197  * Implementation of the llog_operations::lop_read_header
 198  *
 199  * This function reads the current llog header from the bottom storage
 200  * device.
 201  *
 202  * \param[in] env       execution environment
 203  * \param[in] handle    llog handle of the current llog
 204  *
 205  * \retval              0 on successful header read
 206  * \retval              negative error if read failed
 207  */
 208 static int llog_osd_read_header(const struct lu_env *env,
 209                                 struct llog_handle *handle)
 210 {
 211         struct llog_rec_hdr     *llh_hdr;
 212         struct dt_object        *o;
 213         struct llog_thread_info *lgi;
 214         enum llog_flag           flags;
 215         int                      rc;
 216
 217         ENTRY;
 218
 219         o = handle->lgh_obj;
 220         LASSERT(o);
 221
 222         lgi = llog_info(env);
 223
 224         dt_read_lock(env, o, 0);
 225
 226         rc = dt_attr_get(env, o, &lgi->lgi_attr);
 227         if (rc)
 228                 GOTO(unlock, rc);
 229
 230         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
 231
 232         if (lgi->lgi_attr.la_size == 0) {
 233                 CDEBUG(D_HA, "not reading header from 0-byte log\n");
 234                 GOTO(unlock, rc = LLOG_EEMPTY);
 235         }
 236
 237         flags = handle->lgh_hdr->llh_flags;
 238
 239         lgi->lgi_off = 0;
 240         lgi->lgi_buf.lb_buf = handle->lgh_hdr;
 241         lgi->lgi_buf.lb_len = handle->lgh_hdr_size;
 242         rc = dt_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
 243         llh_hdr = &handle->lgh_hdr->llh_hdr;
 244         if (rc < sizeof(*llh_hdr) || rc < llh_hdr->lrh_len) {
 245                 CERROR("%s: error reading "DFID" log header size %d: rc = %d\n",
 246                        o->do_lu.lo_dev->ld_obd->obd_name,
 247                        PFID(lu_object_fid(&o->do_lu)), rc < 0 ? 0 : rc,
 248                        -EFAULT);
 249
 250                 if (rc >= 0)
 251                         rc = -EFAULT;
 252
 253                 GOTO(unlock, rc);
 254         }
 255
 256         if (LLOG_REC_HDR_NEEDS_SWABBING(llh_hdr))
 257                 lustre_swab_llog_hdr(handle->lgh_hdr);
 258
 259         if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
 260                 CERROR("%s: bad log %s "DFID" header magic: %#x "
 261                        "(expected %#x)\n", o->do_lu.lo_dev->ld_obd->obd_name,
 262                        handle->lgh_name ? handle->lgh_name : "",
 263                        PFID(lu_object_fid(&o->do_lu)),
 264                        llh_hdr->lrh_type, LLOG_HDR_MAGIC);
 265                 GOTO(unlock, rc = -EIO);
 266         } else if (llh_hdr->lrh_len < LLOG_MIN_CHUNK_SIZE ||
 267                    llh_hdr->lrh_len > handle->lgh_hdr_size) {
 268                 CERROR("%s: incorrectly sized log %s "DFID" header: "
 269                        "%#x (expected at least %#x)\n"
 270                        "you may need to re-run lconf --write_conf.\n",
 271                        o->do_lu.lo_dev->ld_obd->obd_name,
 272                        handle->lgh_name ? handle->lgh_name : "",
 273                        PFID(lu_object_fid(&o->do_lu)),
 274                        llh_hdr->lrh_len, LLOG_MIN_CHUNK_SIZE);
 275                 GOTO(unlock, rc = -EIO);
 276         } else if (LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index >
 277                    LLOG_HDR_BITMAP_SIZE(handle->lgh_hdr) ||
 278                    LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_len !=
 279                         llh_hdr->lrh_len) {
 280                 CERROR("%s: incorrectly sized log %s "DFID" tailer: "
 281                        "%#x : rc = %d\n",
 282                        o->do_lu.lo_dev->ld_obd->obd_name,
 283                        handle->lgh_name ? handle->lgh_name : "",
 284                        PFID(lu_object_fid(&o->do_lu)),
 285                        LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_len, -EIO);
 286                 GOTO(unlock, rc = -EIO);
 287         }
 288
 289         handle->lgh_hdr->llh_flags |= (flags & LLOG_F_EXT_MASK);
 290         handle->lgh_last_idx = LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index;
 291         rc = 0;
 292
 293 unlock:
 294         dt_read_unlock(env, o);
 295         RETURN(rc);
 296 }
 297
 298 /**
 299  * Implementation of the llog_operations::lop_declare_write
 300  *
 301  * This function declares the new record write.
 302  *
 303  * \param[in] env       execution environment
 304  * \param[in] loghandle llog handle of the current llog
 305  * \param[in] rec       llog record header. This is a real header of the full
 306  *                      llog record to write. This is the beginning of buffer
 307  *                      to write, the length of buffer is stored in
 308  *                      \a rec::lrh_len
 309  * \param[in] idx       index of the llog record. If \a idx == -1 then this is
 310  *                      append case, otherwise \a idx is the index of record
 311  *                      to modify
 312  * \param[in] th        current transaction handle
 313  *
 314  * \retval              0 on successful declaration
 315  * \retval              negative error if declaration failed
 316  */
 317 static int llog_osd_declare_write_rec(const struct lu_env *env,
 318                                       struct llog_handle *loghandle,
 319                                       struct llog_rec_hdr *rec,
 320                                       int idx, struct thandle *th)
 321 {
 322         struct llog_thread_info *lgi = llog_info(env);
 323         __u32                   chunk_size;
 324         struct dt_object        *o;
 325         int                      rc;
 326
 327         ENTRY;
 328
 329         LASSERT(env);
 330         LASSERT(th);
 331         LASSERT(loghandle);
 332         LASSERT(rec);
 333         LASSERT(rec->lrh_len <= loghandle->lgh_ctxt->loc_chunk_size);
 334
 335         o = loghandle->lgh_obj;
 336         LASSERT(o);
 337
 338         chunk_size = loghandle->lgh_ctxt->loc_chunk_size;
 339         lgi->lgi_buf.lb_len = chunk_size;
 340         lgi->lgi_buf.lb_buf = NULL;
 341         /* each time we update header */
 342         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, 0,
 343                                      th);
 344         if (rc || idx == 0) /* if error or just header */
 345                 RETURN(rc);
 346
 347         /**
 348          * the pad record can be inserted so take into account double
 349          * record size
 350          */
 351         lgi->lgi_buf.lb_len = chunk_size * 2;
 352         lgi->lgi_buf.lb_buf = NULL;
 353         /* XXX: implement declared window or multi-chunks approach */
 354         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, -1, th);
 355
 356         RETURN(rc);
 357 }
 358
 359 /**
 360  * Implementation of the llog_operations::lop_write
 361  *
 362  * This function writes the new record in the llog or modify the existed one.
 363  *
 364  * \param[in]  env              execution environment
 365  * \param[in]  loghandle        llog handle of the current llog
 366  * \param[in]  rec              llog record header. This is a real header of
 367  *                              the full llog record to write. This is
 368  *                              the beginning of buffer to write, the length
 369  *                              of buffer is stored in \a rec::lrh_len
 370  * \param[in,out] reccookie     pointer to the cookie to return back if needed.
 371  *                              It is used for further cancel of this llog
 372  *                              record.
 373  * \param[in]  idx              index of the llog record. If \a idx == -1 then
 374  *                              this is append case, otherwise \a idx is
 375  *                              the index of record to modify
 376  * \param[in]  th               current transaction handle
 377  *
 378  * \retval                      0 on successful write && \a reccookie == NULL
 379  *                              1 on successful write && \a reccookie != NULL
 380  * \retval                      negative error if write failed
 381  */
 382 static int llog_osd_write_rec(const struct lu_env *env,
 383                               struct llog_handle *loghandle,
 384                               struct llog_rec_hdr *rec,
 385                               struct llog_cookie *reccookie,
 386                               int idx, struct thandle *th)
 387 {
 388         struct llog_thread_info *lgi = llog_info(env);
 389         struct llog_log_hdr *llh;
 390         int reclen = rec->lrh_len;
 391         int index, rc;
 392         struct llog_rec_tail *lrt;
 393         struct dt_object *o;
 394         __u32 chunk_size;
 395         size_t left;
 396         __u32 orig_last_idx;
 397         bool pad = false;
 398         ENTRY;
 399
 400         llh = loghandle->lgh_hdr;
 401         o = loghandle->lgh_obj;
 402
 403         chunk_size = llh->llh_hdr.lrh_len;
 404         CDEBUG(D_OTHER, "new record %x to "DFID"\n",
 405                rec->lrh_type, PFID(lu_object_fid(&o->do_lu)));
 406
 407         if (!llog_osd_exist(loghandle))
 408                 RETURN(-ENOENT);
 409
 410         /* record length should not bigger than  */
 411         if (reclen > loghandle->lgh_hdr->llh_hdr.lrh_len)
 412                 RETURN(-E2BIG);
 413
 414         /* sanity check for fixed-records llog */
 415         if (idx != LLOG_HEADER_IDX && (llh->llh_flags & LLOG_F_IS_FIXSIZE)) {
 416                 LASSERT(llh->llh_size != 0);
 417                 LASSERT(llh->llh_size == reclen);
 418         }
 419
 420         /* return error if osp object is stale */
 421         if (idx != LLOG_HEADER_IDX && dt_object_stale(o))
 422                 RETURN(-ESTALE);
 423         rc = dt_attr_get(env, o, &lgi->lgi_attr);
 424         if (rc)
 425                 RETURN(rc);
 426
 427         /**
 428          * The modification case.
 429          * If idx set then the record with that index must be modified.
 430          * There are three cases possible:
 431          * 1) the common case is the llog header update (idx == 0)
 432          * 2) the llog record modification during llog process.
 433          *    This is indicated by the \a loghandle::lgh_cur_idx > 0.
 434          *    In that case the \a loghandle::lgh_cur_offset
 435          * 3) otherwise this is assumed that llog consist of records of
 436          *    fixed size, i.e. catalog. The llog header must has llh_size
 437          *    field equal to record size. The record offset is calculated
 438          *    just by /a idx value
 439          *
 440          * During modification we don't need extra header update because
 441          * the bitmap and record count are not changed. The record header
 442          * and tail remains the same too.
 443          */
 444         if (idx != LLOG_NEXT_IDX) {
 445                 /* llog can be empty only when first record is being written */
 446                 LASSERT(ergo(idx > 0, lgi->lgi_attr.la_size > 0));
 447
 448                 if (!test_bit_le(idx, LLOG_HDR_BITMAP(llh))) {
 449                         CERROR("%s: modify unset record %u\n",
 450                                o->do_lu.lo_dev->ld_obd->obd_name, idx);
 451                         RETURN(-ENOENT);
 452                 }
 453
 454                 if (idx != rec->lrh_index) {
 455                         CERROR("%s: modify index mismatch %d %u\n",
 456                                o->do_lu.lo_dev->ld_obd->obd_name, idx,
 457                                rec->lrh_index);
 458                         RETURN(-EFAULT);
 459                 }
 460
 461                 if (idx == LLOG_HEADER_IDX) {
 462                         /* llog header update */
 463                         __u32   *bitmap = LLOG_HDR_BITMAP(llh);
 464
 465                         lgi->lgi_off = 0;
 466
 467                         /* If it does not indicate the bitmap index
 468                          * (reccookie == NULL), then it means update
 469                          * the whole update header. Otherwise only
 470                          * update header and bits needs to be updated,
 471                          * and in DNE cases, it will signaficantly
 472                          * shrink the RPC size.
 473                          * see distribute_txn_cancel_records()*/
 474                         if (reccookie == NULL) {
 475                                 lgi->lgi_buf.lb_len = reclen;
 476                                 lgi->lgi_buf.lb_buf = rec;
 477                                 rc = dt_record_write(env, o, &lgi->lgi_buf,
 478                                                      &lgi->lgi_off, th);
 479                                 RETURN(rc);
 480                         }
 481
 482                         /* update the header */
 483                         lgi->lgi_buf.lb_len = llh->llh_bitmap_offset;
 484                         lgi->lgi_buf.lb_buf = llh;
 485                         rc = dt_record_write(env, o, &lgi->lgi_buf,
 486                                              &lgi->lgi_off, th);
 487                         if (rc != 0)
 488                                 RETURN(rc);
 489
 490                         /* update the bitmap */
 491                         index = reccookie->lgc_index;
 492                         lgi->lgi_off = llh->llh_bitmap_offset +
 493                                       (index / (sizeof(*bitmap) * 8)) *
 494                                                         sizeof(*bitmap);
 495                         lgi->lgi_buf.lb_len = sizeof(*bitmap);
 496                         lgi->lgi_buf.lb_buf =
 497                                         &bitmap[index/(sizeof(*bitmap)*8)];
 498                         rc = dt_record_write(env, o, &lgi->lgi_buf,
 499                                              &lgi->lgi_off, th);
 500
 501                         RETURN(rc);
 502                 } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
 503                         lgi->lgi_off = llh->llh_hdr.lrh_len +
 504                                        (idx - 1) * reclen;
 505                 } else if (reccookie != NULL && reccookie->lgc_index > 0) {
 506                         /**
 507                          * The lgc_offset can be used only if index is
 508                          * the same.
 509                          */
 510                         if (idx != reccookie->lgc_index) {
 511                                 CERROR("%s: modify index mismatch %d %d\n",
 512                                        o->do_lu.lo_dev->ld_obd->obd_name, idx,
 513                                        reccookie->lgc_index);
 514                                 RETURN(-EFAULT);
 515                         }
 516
 517                         lgi->lgi_off = reccookie->lgc_offset;
 518                         CDEBUG(D_OTHER, "modify record "DFID": idx:%u, "
 519                                "len:%u offset %llu\n",
 520                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid), idx,
 521                                rec->lrh_len, (long long)lgi->lgi_off);
 522                 } else {
 523                         /* This can be result of lgh_cur_idx is not set during
 524                          * llog processing or llh_size is not set to proper
 525                          * record size for fixed records llog. Therefore it is
 526                          * impossible to get record offset. */
 527                         CERROR("%s: can't get record offset, idx:%d, "
 528                                "len:%u.\n", o->do_lu.lo_dev->ld_obd->obd_name,
 529                                idx, rec->lrh_len);
 530                         RETURN(-EFAULT);
 531                 }
 532
 533                 /* update only data, header and tail remain the same */
 534                 lgi->lgi_off += sizeof(struct llog_rec_hdr);
 535                 lgi->lgi_buf.lb_len = REC_DATA_LEN(rec);
 536                 lgi->lgi_buf.lb_buf = REC_DATA(rec);
 537                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 538                 if (rc == 0 && reccookie) {
 539                         reccookie->lgc_lgl = loghandle->lgh_id;
 540                         reccookie->lgc_index = idx;
 541                         rc = 1;
 542                 }
 543                 RETURN(rc);
 544         }
 545
 546         /**
 547          * The append case.
 548          * The most common case of using llog. The new index is assigned to
 549          * the new record, new bit is set in llog bitmap and llog count is
 550          * incremented.
 551          *
 552          * Make sure that records don't cross a chunk boundary, so we can
 553          * process them page-at-a-time if needed.  If it will cross a chunk
 554          * boundary, write in a fake (but referenced) entry to pad the chunk.
 555          */
 556
 557
 558         /* simulate ENOSPC when new plain llog is being added to the
 559          * catalog */
 560         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED2) &&
 561             llh->llh_flags & LLOG_F_IS_CAT)
 562                 RETURN(-ENOSPC);
 563
 564         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
 565         orig_last_idx = loghandle->lgh_last_idx;
 566         lgi->lgi_off = lgi->lgi_attr.la_size;
 567
 568         if (loghandle->lgh_max_size > 0 &&
 569             lgi->lgi_off >= loghandle->lgh_max_size) {
 570                 CDEBUG(D_OTHER, "llog is getting too large (%u > %u) at %u "
 571                        DFID"\n", (unsigned)lgi->lgi_off,
 572                        loghandle->lgh_max_size, (int)loghandle->lgh_last_idx,
 573                        PFID(&loghandle->lgh_id.lgl_oi.oi_fid));
 574                 /* this is to signal that this llog is full */
 575                 loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) - 1;
 576                 RETURN(-ENOSPC);
 577         }
 578
 579         left = chunk_size - (lgi->lgi_off & (chunk_size - 1));
 580         /* NOTE: padding is a record, but no bit is set */
 581         if (left != 0 && left != reclen &&
 582             left < (reclen + LLOG_MIN_REC_SIZE)) {
 583                 index = loghandle->lgh_last_idx + 1;
 584                 rc = llog_osd_pad(env, o, &lgi->lgi_off, left, index, th);
 585                 if (rc)
 586                         RETURN(rc);
 587
 588                 loghandle->lgh_last_idx++; /* for pad rec */
 589                 pad = true;
 590         }
 591         /* if it's the last idx in log file, then return -ENOSPC
 592          * or wrap around if a catalog */
 593         if (llog_is_full(loghandle) ||
 594             unlikely(llh->llh_flags & LLOG_F_IS_CAT &&
 595                      OBD_FAIL_PRECHECK(OBD_FAIL_CAT_RECORDS) &&
 596                      loghandle->lgh_last_idx >= cfs_fail_val)) {
 597                 if (llh->llh_flags & LLOG_F_IS_CAT)
 598                         loghandle->lgh_last_idx = 0;
 599                 else
 600                         RETURN(-ENOSPC);
 601         }
 602
 603         down_write(&loghandle->lgh_last_sem);
 604         /* increment the last_idx along with llh_tail index, they should
 605          * be equal for a llog lifetime */
 606         if (OBD_FAIL_CHECK(OBD_FAIL_LLOG_ADD_GAP) && --cfs_fail_val == 0)
 607                 loghandle->lgh_last_idx++;
 608         loghandle->lgh_last_idx++;
 609         index = loghandle->lgh_last_idx;
 610         LLOG_HDR_TAIL(llh)->lrt_index = index;
 611         /**
 612          * NB: the caller should make sure only 1 process access
 613          * the lgh_last_idx, e.g. append should be exclusive.
 614          * Otherwise it might hit the assert.
 615          */
 616         LASSERT(index < LLOG_HDR_BITMAP_SIZE(llh));
 617         rec->lrh_index = index;
 618         lrt = rec_tail(rec);
 619         lrt->lrt_len = rec->lrh_len;
 620         lrt->lrt_index = rec->lrh_index;
 621
 622         /* the lgh_hdr_mutex protects llog header data from concurrent
 623          * update/cancel, the llh_count and llh_bitmap are protected */
 624         mutex_lock(&loghandle->lgh_hdr_mutex);
 625         if (__test_and_set_bit_le(index, LLOG_HDR_BITMAP(llh))) {
 626                 CERROR("%s: index %u already set in llog bitmap "DFID"\n",
 627                        o->do_lu.lo_dev->ld_obd->obd_name, index,
 628                        PFID(lu_object_fid(&o->do_lu)));
 629                 mutex_unlock(&loghandle->lgh_hdr_mutex);
 630                 LBUG(); /* should never happen */
 631         }
 632         llh->llh_count++;
 633
 634         if (!(llh->llh_flags & LLOG_F_IS_FIXSIZE)) {
 635                 /* Update the minimum size of the llog record */
 636                 if (llh->llh_size == 0)
 637                         llh->llh_size = reclen;
 638                 else if (reclen < llh->llh_size)
 639                         llh->llh_size = reclen;
 640         }
 641
 642         /*
 643          * readers (e.g. llog_osd_read_header()) must not find
 644          * llog updated partially (bitmap/counter claims record,
 645          * but a record hasn't been added yet) as this results
 646          * in EIO.
 647          */
 648         dt_write_lock(env, o, 0);
 649
 650         if (lgi->lgi_attr.la_size == 0) {
 651                 lgi->lgi_off = 0;
 652                 lgi->lgi_buf.lb_len = llh->llh_hdr.lrh_len;
 653                 lgi->lgi_buf.lb_buf = &llh->llh_hdr;
 654                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 655                 if (rc != 0)
 656                         GOTO(out_unlock, rc);
 657         } else {
 658                 __u32   *bitmap = LLOG_HDR_BITMAP(llh);
 659
 660                 /* Note: If this is not initialization (size == 0), then do not
 661                  * write the whole header (8k bytes), only update header/tail
 662                  * and bits needs to be updated. Because this update might be
 663                  * part of cross-MDT operation, which needs to write these
 664                  * updates into the update log(32KB limit) and also pack inside
 665                  * the RPC (1MB limit), if we write 8K for each operation, which
 666                  * will cost a lot space, and keep us adding more updates to one
 667                  * update log.*/
 668                 lgi->lgi_off = 0;
 669                 lgi->lgi_buf.lb_len = llh->llh_bitmap_offset;
 670                 lgi->lgi_buf.lb_buf = &llh->llh_hdr;
 671                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 672                 if (rc != 0)
 673                         GOTO(out_unlock, rc);
 674
 675                 lgi->lgi_off = llh->llh_bitmap_offset +
 676                               (index / (sizeof(*bitmap) * 8)) * sizeof(*bitmap);
 677                 lgi->lgi_buf.lb_len = sizeof(*bitmap);
 678                 lgi->lgi_buf.lb_buf = &bitmap[index/(sizeof(*bitmap)*8)];
 679                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 680                 if (rc != 0)
 681                         GOTO(out_unlock, rc);
 682
 683                 lgi->lgi_off =  (unsigned long)LLOG_HDR_TAIL(llh) -
 684                                 (unsigned long)llh;
 685                 lgi->lgi_buf.lb_len = sizeof(llh->llh_tail);
 686                 lgi->lgi_buf.lb_buf = LLOG_HDR_TAIL(llh);
 687                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 688                 if (rc != 0)
 689                         GOTO(out_unlock, rc);
 690         }
 691         if (OBD_FAIL_PRECHECK(OBD_FAIL_LLOG_PAUSE_AFTER_PAD) && pad) {
 692                 /* a window for concurrent llog reader, see LU-12577 */
 693                 OBD_FAIL_TIMEOUT(OBD_FAIL_LLOG_PAUSE_AFTER_PAD,
 694                                  cfs_fail_val ?: 1);
 695         }
 696
 697 out_unlock:
 698         /* unlock here for remote object */
 699         mutex_unlock(&loghandle->lgh_hdr_mutex);
 700         if (rc) {
 701                 dt_write_unlock(env, o);
 702                 GOTO(out, rc);
 703         }
 704
 705         if (OBD_FAIL_PRECHECK(OBD_FAIL_LLOG_PROCESS_TIMEOUT) &&
 706            cfs_fail_val == (unsigned int)(loghandle->lgh_id.lgl_oi.oi.oi_id &
 707                                           0xFFFFFFFF)) {
 708                 OBD_RACE(OBD_FAIL_LLOG_PROCESS_TIMEOUT);
 709                 msleep(1 * MSEC_PER_SEC);
 710         }
 711         /* computed index can be used to determine offset for fixed-size
 712          * records. This also allows to handle Catalog wrap around case */
 713         if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
 714                 lgi->lgi_off = llh->llh_hdr.lrh_len + (index - 1) * reclen;
 715         } else {
 716                 rc = dt_attr_get(env, o, &lgi->lgi_attr);
 717                 if (rc) {
 718                         dt_write_unlock(env, o);
 719                         GOTO(out, rc);
 720                 }
 721
 722                 LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
 723                 lgi->lgi_off = max_t(__u64, lgi->lgi_attr.la_size,
 724                                      lgi->lgi_off);
 725         }
 726
 727         lgi->lgi_buf.lb_len = reclen;
 728         lgi->lgi_buf.lb_buf = rec;
 729         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 730
 731         dt_write_unlock(env, o);
 732         if (rc < 0)
 733                 GOTO(out, rc);
 734
 735         up_write(&loghandle->lgh_last_sem);
 736
 737         CDEBUG(D_HA, "added record "DFID".%u, %u off%llu\n",
 738                PFID(lu_object_fid(&o->do_lu)), index, rec->lrh_len,
 739                lgi->lgi_off);
 740         if (reccookie != NULL) {
 741                 reccookie->lgc_lgl = loghandle->lgh_id;
 742                 reccookie->lgc_index = index;
 743                 if ((rec->lrh_type == MDS_UNLINK_REC) ||
 744                     (rec->lrh_type == MDS_SETATTR64_REC))
 745                         reccookie->lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
 746                 else if (rec->lrh_type == OST_SZ_REC)
 747                         reccookie->lgc_subsys = LLOG_SIZE_ORIG_CTXT;
 748                 else
 749                         reccookie->lgc_subsys = -1;
 750                 rc = 1;
 751         }
 752         RETURN(rc);
 753 out:
 754         /* cleanup llog for error case */
 755         mutex_lock(&loghandle->lgh_hdr_mutex);
 756         clear_bit_le(index, LLOG_HDR_BITMAP(llh));
 757         llh->llh_count--;
 758         mutex_unlock(&loghandle->lgh_hdr_mutex);
 759
 760         /* restore llog last_idx */
 761         if (dt_object_remote(o)) {
 762                 loghandle->lgh_last_idx = orig_last_idx;
 763         } else if (--loghandle->lgh_last_idx == 0 &&
 764             (llh->llh_flags & LLOG_F_IS_CAT) && llh->llh_cat_idx != 0) {
 765                 /* catalog had just wrap-around case */
 766                 loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) - 1;
 767         }
 768
 769         LLOG_HDR_TAIL(llh)->lrt_index = loghandle->lgh_last_idx;
 770         up_write(&loghandle->lgh_last_sem);
 771
 772         RETURN(rc);
 773 }
 774
 775 /**
 776  * We can skip reading at least as many log blocks as the number of
 777  * minimum sized log records we are skipping.  If it turns out
 778  * that we are not far enough along the log (because the
 779  * actual records are larger than minimum size) we just skip
 780  * some more records.
 781  *
 782  * Note: in llog_process_thread, it will use bitmap offset as
 783  * the index to locate the record, which also includs some pad
 784  * records, whose record size is very small, and it also does not
 785  * consider pad record when recording minimum record size (otherwise
 786  * min_record size might be too small), so in some rare cases,
 787  * it might skip too much record for @goal, see llog_osd_next_block().
 788  *
 789  * When force_mini_rec is true, it means we have to use LLOG_MIN_REC_SIZE
 790  * as the min record size to skip over, usually because in the previous
 791  * try, it skip too much record, see loog_osd_next(prev)_block().
 792  */
 793 static inline void llog_skip_over(struct llog_handle *lgh, __u64 *off,
 794                                   int curr, int goal, __u32 chunk_size,
 795                                   bool force_mini_rec)
 796 {
 797         struct llog_log_hdr *llh = lgh->lgh_hdr;
 798
 799         /* Goal should not bigger than the record count */
 800         if (goal > lgh->lgh_last_idx)
 801                 goal = lgh->lgh_last_idx;
 802
 803         if (goal > curr) {
 804                 if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
 805                         *off = chunk_size + (goal - 1) * llh->llh_size;
 806                 } else {
 807                         __u64 min_rec_size = LLOG_MIN_REC_SIZE;
 808
 809                         if (llh->llh_size > 0 && !force_mini_rec)
 810                                 min_rec_size = llh->llh_size;
 811
 812                         *off = *off + (goal - curr - 1) * min_rec_size;
 813                 }
 814         }
 815         /* always align with lower chunk boundary*/
 816         *off &= ~(chunk_size - 1);
 817 }
 818
 819 /**
 820  * Remove optional fields that the client doesn't expect.
 821  * This is typically in order to ensure compatibility with older clients.
 822  * It is assumed that since we exclusively remove fields, the block will be
 823  * big enough to handle the remapped records. It is also assumed that records
 824  * of a block have the same format (i.e.: the same features enabled).
 825  *
 826  * \param[in,out]    hdr           Header of the block of records to remap.
 827  * \param[in,out]    last_hdr      Last header, don't read past this point.
 828  * \param[in]        flags         Flags describing the fields to keep.
 829  * \param[in]        extra_flags   Flags describing the extra fields to keep.
 830  */
 831 static void changelog_block_trim_ext(struct llog_rec_hdr *hdr,
 832                                      struct llog_rec_hdr *last_hdr,
 833                                      struct llog_handle *loghandle)
 834 {
 835         enum changelog_rec_flags flags = CLF_SUPPORTED;
 836         enum changelog_rec_extra_flags extra_flags = CLFE_SUPPORTED;
 837
 838         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_XATTR))
 839                 extra_flags &= ~CLFE_XATTR;
 840         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_OMODE))
 841                 extra_flags &= ~CLFE_OPEN;
 842         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_NID))
 843                 extra_flags &= ~CLFE_NID;
 844         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_UIDGID))
 845                 extra_flags &= ~CLFE_UIDGID;
 846         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_EXTRA_FLAGS))
 847                 flags &= ~CLF_EXTRA_FLAGS;
 848         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_JOBID))
 849                 flags &= ~CLF_JOBID;
 850
 851         if (flags == CLF_SUPPORTED && extra_flags == CLFE_SUPPORTED)
 852                 return;
 853
 854         if (hdr->lrh_type != CHANGELOG_REC)
 855                 return;
 856
 857         do {
 858                 struct changelog_rec *rec = (struct changelog_rec *)(hdr + 1);
 859                 enum changelog_rec_extra_flags xflag = CLFE_INVALID;
 860
 861                 if (flags & CLF_EXTRA_FLAGS &&
 862                     rec->cr_flags & CLF_EXTRA_FLAGS) {
 863                         xflag = changelog_rec_extra_flags(rec)->cr_extra_flags &
 864                                 extra_flags;
 865                 }
 866
 867                 if (unlikely(hdr->lrh_len == 0)) {
 868                         /* It is corruption case, we cannot know the next rec,
 869                          * jump to the last one directly to avoid dead loop. */
 870                         LCONSOLE(D_WARNING, "Hit invalid llog record: "
 871                                  "idx %u, type %u, id %u\n",
 872                                  hdr->lrh_index, hdr->lrh_type, hdr->lrh_id);
 873                         hdr = llog_rec_hdr_next(last_hdr);
 874                         if (unlikely(hdr == last_hdr))
 875                                 LCONSOLE(D_WARNING, "The last record crashed: "
 876                                          "idx %u, type %u, id %u\n",
 877                                          hdr->lrh_index, hdr->lrh_type,
 878                                          hdr->lrh_id);
 879                         break;
 880                 }
 881
 882                 changelog_remap_rec(rec, rec->cr_flags & flags, xflag);
 883                 hdr = llog_rec_hdr_next(hdr);
 884                 /* Yield CPU to avoid soft-lockup if there are too many records
 885                  * to be handled. */
 886                 cond_resched();
 887         } while ((char *)hdr <= (char *)last_hdr);
 888 }
 889
 890 /**
 891  * Implementation of the llog_operations::lop_next_block
 892  *
 893  * This function finds the the next llog block to return which contains
 894  * record with required index. It is main part of llog processing.
 895  *
 896  * \param[in]     env           execution environment
 897  * \param[in]     loghandle     llog handle of the current llog
 898  * \param[in,out] cur_idx       index preceeding cur_offset
 899  * \param[in]     next_idx      target index to find
 900  * \param[in,out] cur_offset    furtherst point read in the file
 901  * \param[in]     buf           pointer to data buffer to fill
 902  * \param[in]     len           required len to read, it is
 903  *                              usually llog chunk_size.
 904  *
 905  * \retval                      0 on successful buffer read
 906  * \retval                      negative value on error
 907  */
 908 static int llog_osd_next_block(const struct lu_env *env,
 909                                struct llog_handle *loghandle, int *cur_idx,
 910                                int next_idx, __u64 *cur_offset, void *buf,
 911                                int len)
 912 {
 913         struct llog_thread_info *lgi = llog_info(env);
 914         struct dt_object        *o;
 915         struct dt_device        *dt;
 916         int                      rc;
 917         __u32                   chunk_size;
 918         int last_idx = *cur_idx;
 919         __u64 last_offset = *cur_offset;
 920         bool force_mini_rec = false;
 921
 922         ENTRY;
 923
 924         LASSERT(env);
 925         LASSERT(lgi);
 926
 927         chunk_size = loghandle->lgh_hdr->llh_hdr.lrh_len;
 928         if (len == 0 || len & (chunk_size - 1))
 929                 RETURN(-EINVAL);
 930
 931         LASSERT(loghandle);
 932         LASSERT(loghandle->lgh_ctxt);
 933
 934         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_CHANGELOG_DEL) &&
 935             cfs_fail_val == ((unsigned long)loghandle & 0xFFFFFFFF)) {
 936                 OBD_RACE(OBD_FAIL_MDS_CHANGELOG_DEL);
 937                 msleep(MSEC_PER_SEC >> 2);
 938         }
 939
 940         o = loghandle->lgh_obj;
 941         LASSERT(o);
 942         dt_read_lock(env, o, 0);
 943         if (!llog_osd_exist(loghandle))
 944                 GOTO(out, rc = -ESTALE); //object was destroyed
 945
 946         dt = lu2dt_dev(o->do_lu.lo_dev);
 947         LASSERT(dt);
 948
 949         rc = dt_attr_get(env, o, &lgi->lgi_attr);
 950         if (rc)
 951                 GOTO(out, rc);
 952
 953         CDEBUG(D_OTHER,
 954                "looking for log index %u (cur idx %u off %llu), size %llu\n",
 955                next_idx, *cur_idx,
 956                *cur_offset, lgi->lgi_attr.la_size);
 957
 958         while (*cur_offset < lgi->lgi_attr.la_size) {
 959                 struct llog_rec_hdr     *rec, *last_rec;
 960                 struct llog_rec_tail    *tail;
 961
 962                 llog_skip_over(loghandle, cur_offset, *cur_idx,
 963                                next_idx, chunk_size, force_mini_rec);
 964
 965                 /* read up to next llog chunk_size block */
 966                 lgi->lgi_buf.lb_len = chunk_size -
 967                                       (*cur_offset & (chunk_size - 1));
 968                 lgi->lgi_buf.lb_buf = buf;
 969
 970                 rc = dt_read(env, o, &lgi->lgi_buf, cur_offset);
 971                 if (rc < 0) {
 972                         if (rc == -EBADR && !force_mini_rec)
 973                                 goto retry;
 974
 975                         CERROR("%s: can't read llog block from log "DFID
 976                                " offset %llu: rc = %d\n",
 977                                o->do_lu.lo_dev->ld_obd->obd_name,
 978                                PFID(lu_object_fid(&o->do_lu)), *cur_offset,
 979                                rc);
 980                         GOTO(out, rc);
 981                 }
 982
 983                 if (rc < len) {
 984                         /* signal the end of the valid buffer to
 985                          * llog_process */
 986                         memset(buf + rc, 0, len - rc);
 987                 }
 988
 989                 if (rc == 0) { /* end of file, nothing to do */
 990                         if (!force_mini_rec)
 991                                 goto retry;
 992                         GOTO(out, rc);
 993                 }
 994
 995                 if (rc < sizeof(*tail)) {
 996                         if (!force_mini_rec)
 997                                 goto retry;
 998
 999                         CERROR("%s: invalid llog block at log id "DFID":%x "
1000                                "offset %llu\n",
1001                                o->do_lu.lo_dev->ld_obd->obd_name,
1002                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1003                                loghandle->lgh_id.lgl_ogen, *cur_offset);
1004                         GOTO(out, rc = -EINVAL);
1005                 }
1006
1007                 rec = buf;
1008                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1009                         lustre_swab_llog_rec(rec);
1010                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1011                                                 sizeof(struct llog_rec_tail));
1012
1013                 if (llog_verify_record(loghandle, rec)) {
1014                         /*
1015                          * the block seems corrupted. make a pad record so the
1016                          * caller can skip the block and try with the next one
1017                          */
1018                         rec->lrh_len = rc;
1019                         rec->lrh_index = next_idx;
1020                         rec->lrh_type = LLOG_PAD_MAGIC;
1021
1022                         tail = rec_tail(rec);
1023                         tail->lrt_len = rc;
1024                         tail->lrt_index = next_idx;
1025
1026                         GOTO(out, rc = 0);
1027                 }
1028
1029                 /* get the last record in block */
1030                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1031                                                    tail->lrt_len);
1032
1033                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1034                         lustre_swab_llog_rec(last_rec);
1035
1036                 if (last_rec->lrh_index != tail->lrt_index) {
1037                         CERROR("%s: invalid llog tail at log id "DFID":%x offset %llu last_rec idx %u tail idx %u lrt len %u read_size %d\n",
1038                                o->do_lu.lo_dev->ld_obd->obd_name,
1039                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1040                                loghandle->lgh_id.lgl_ogen, *cur_offset,
1041                                last_rec->lrh_index, tail->lrt_index,
1042                                tail->lrt_len, rc);
1043                         GOTO(out, rc = -EINVAL);
1044                 }
1045
1046                 *cur_idx = tail->lrt_index;
1047
1048                 /* this shouldn't happen */
1049                 if (tail->lrt_index == 0) {
1050                         CERROR("%s: invalid llog tail at log id "DFID":%x "
1051                                "offset %llu bytes %d\n",
1052                                o->do_lu.lo_dev->ld_obd->obd_name,
1053                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1054                                loghandle->lgh_id.lgl_ogen, *cur_offset, rc);
1055                         GOTO(out, rc = -EINVAL);
1056                 }
1057                 if (tail->lrt_index < next_idx) {
1058                         last_idx = *cur_idx;
1059                         last_offset = *cur_offset;
1060                         continue;
1061                 }
1062
1063                 /* sanity check that the start of the new buffer is no farther
1064                  * than the record that we wanted.  This shouldn't happen. */
1065                 if (next_idx && rec->lrh_index > next_idx) {
1066                         if (!force_mini_rec && next_idx > last_idx)
1067                                 goto retry;
1068
1069                         CERROR("%s: missed desired record? %u > %u\n",
1070                                o->do_lu.lo_dev->ld_obd->obd_name,
1071                                rec->lrh_index, next_idx);
1072                         GOTO(out, rc = -ENOENT);
1073                 }
1074
1075                 /* Trim unsupported extensions for compat w/ older clients */
1076                 changelog_block_trim_ext(rec, last_rec, loghandle);
1077
1078                 GOTO(out, rc = 0);
1079
1080 retry:
1081                 /* Note: because there are some pad records in the
1082                  * llog, so llog_skip_over() might skip too much
1083                  * records, let's try skip again with minimum record */
1084                 force_mini_rec = true;
1085                 *cur_offset = last_offset;
1086                 *cur_idx = last_idx;
1087         }
1088         GOTO(out, rc = -EIO);
1089 out:
1090         dt_read_unlock(env, o);
1091         return rc;
1092 }
1093
1094 /**
1095  * Implementation of the llog_operations::lop_prev_block
1096  *
1097  * This function finds the llog block to return which contains
1098  * record with required index but in reverse order - from end of llog
1099  * to the beginning.
1100  * It is main part of reverse llog processing.
1101  *
1102  * \param[in] env       execution environment
1103  * \param[in] loghandle llog handle of the current llog
1104  * \param[in] prev_idx  target index to find
1105  * \param[in] buf       pointer to data buffer to fill
1106  * \param[in] len       required len to read, it is llog_chunk_size usually.
1107  *
1108  * \retval              0 on successful buffer read
1109  * \retval              negative value on error
1110  */
1111 static int llog_osd_prev_block(const struct lu_env *env,
1112                                struct llog_handle *loghandle,
1113                                int prev_idx, void *buf, int len)
1114 {
1115         struct llog_thread_info *lgi = llog_info(env);
1116         struct dt_object        *o;
1117         struct dt_device        *dt;
1118         loff_t                   cur_offset;
1119         __u32                   chunk_size;
1120         int                      rc;
1121
1122         ENTRY;
1123
1124         chunk_size = loghandle->lgh_hdr->llh_hdr.lrh_len;
1125         if (len == 0 || len & (chunk_size - 1))
1126                 RETURN(-EINVAL);
1127
1128         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
1129
1130         LASSERT(loghandle);
1131         LASSERT(loghandle->lgh_ctxt);
1132
1133         o = loghandle->lgh_obj;
1134         LASSERT(o);
1135         dt_read_lock(env, o, 0);
1136         if (!llog_osd_exist(loghandle))
1137                 GOTO(out, rc = -ESTALE);
1138
1139         dt = lu2dt_dev(o->do_lu.lo_dev);
1140         LASSERT(dt);
1141
1142         /* Let's only use mini record size for previous block read
1143          * for now XXX */
1144         cur_offset = chunk_size;
1145         llog_skip_over(loghandle, &cur_offset, 0, prev_idx,
1146                        chunk_size, true);
1147
1148         rc = dt_attr_get(env, o, &lgi->lgi_attr);
1149         if (rc)
1150                 GOTO(out, rc);
1151
1152         while (cur_offset < lgi->lgi_attr.la_size) {
1153                 struct llog_rec_hdr     *rec, *last_rec;
1154                 struct llog_rec_tail    *tail;
1155
1156                 lgi->lgi_buf.lb_len = len;
1157                 lgi->lgi_buf.lb_buf = buf;
1158                 rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
1159                 if (rc < 0) {
1160                         CERROR("%s: can't read llog block from log "DFID
1161                                " offset %llu: rc = %d\n",
1162                                o->do_lu.lo_dev->ld_obd->obd_name,
1163                                PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
1164                         GOTO(out, rc);
1165                 }
1166
1167                 if (rc == 0) /* end of file, nothing to do */
1168                         GOTO(out, rc);
1169
1170                 if (rc < sizeof(*tail)) {
1171                         CERROR("%s: invalid llog block at log id "DFID":%x "
1172                                "offset %llu\n",
1173                                o->do_lu.lo_dev->ld_obd->obd_name,
1174                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1175                                loghandle->lgh_id.lgl_ogen, cur_offset);
1176                         GOTO(out, rc = -EINVAL);
1177                 }
1178
1179                 rec = buf;
1180                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1181                         lustre_swab_llog_rec(rec);
1182
1183                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1184                                                 sizeof(struct llog_rec_tail));
1185                 /* get the last record in block */
1186                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1187                                                    le32_to_cpu(tail->lrt_len));
1188
1189                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1190                         lustre_swab_llog_rec(last_rec);
1191                 LASSERT(last_rec->lrh_index == tail->lrt_index);
1192
1193                 /* this shouldn't happen */
1194                 if (tail->lrt_index == 0) {
1195                         CERROR("%s: invalid llog tail at log id "DFID":%x "
1196                                "offset %llu\n",
1197                                o->do_lu.lo_dev->ld_obd->obd_name,
1198                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1199                                loghandle->lgh_id.lgl_ogen, cur_offset);
1200                         GOTO(out, rc = -EINVAL);
1201                 }
1202                 if (tail->lrt_index < prev_idx)
1203                         continue;
1204
1205                 /* sanity check that the start of the new buffer is no farther
1206                  * than the record that we wanted.  This shouldn't happen. */
1207                 if (rec->lrh_index > prev_idx) {
1208                         CERROR("%s: missed desired record? %u > %u\n",
1209                                o->do_lu.lo_dev->ld_obd->obd_name,
1210                                rec->lrh_index, prev_idx);
1211                         GOTO(out, rc = -ENOENT);
1212                 }
1213
1214                 /* Trim unsupported extensions for compat w/ older clients */
1215                 changelog_block_trim_ext(rec, last_rec, loghandle);
1216
1217                 GOTO(out, rc = 0);
1218         }
1219         GOTO(out, rc = -EIO);
1220 out:
1221         dt_read_unlock(env, o);
1222         return rc;
1223 }
1224
1225 /**
1226  * This is helper function to get llog directory object. It is used by named
1227  * llog operations to find/insert/delete llog entry from llog directory.
1228  *
1229  * \param[in] env       execution environment
1230  * \param[in] ctxt      llog context
1231  *
1232  * \retval              dt_object of llog directory
1233  * \retval              ERR_PTR of negative value on error
1234  */
1235 static struct dt_object *llog_osd_dir_get(const struct lu_env *env,
1236                                           struct llog_ctxt *ctxt)
1237 {
1238         struct dt_device        *dt;
1239         struct dt_thread_info   *dti = dt_info(env);
1240         struct dt_object        *dir;
1241         int                      rc;
1242
1243         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1244         if (ctxt->loc_dir == NULL) {
1245                 rc = dt_root_get(env, dt, &dti->dti_fid);
1246                 if (rc)
1247                         return ERR_PTR(rc);
1248                 dir = dt_locate(env, dt, &dti->dti_fid);
1249
1250                 if (!IS_ERR(dir) && !dt_try_as_dir(env, dir)) {
1251                         dt_object_put(env, dir);
1252                         return ERR_PTR(-ENOTDIR);
1253                 }
1254         } else {
1255                 lu_object_get(&ctxt->loc_dir->do_lu);
1256                 dir = ctxt->loc_dir;
1257         }
1258
1259         return dir;
1260 }
1261
1262 /**
1263  * Implementation of the llog_operations::lop_open
1264  *
1265  * This function opens the llog by its logid or by name, it may open also
1266  * non existent llog and assing then new id to it.
1267  * The llog_open/llog_close pair works similar to lu_object_find/put,
1268  * the object may not exist prior open. The result of open is just dt_object
1269  * in the llog header.
1270  *
1271  * \param[in] env               execution environment
1272  * \param[in] handle            llog handle of the current llog
1273  * \param[in] logid             logid of llog to open (nameless llog)
1274  * \param[in] name              name of llog to open (named llog)
1275  * \param[in] open_param
1276  *                              LLOG_OPEN_NEW - new llog, may not exist
1277  *                              LLOG_OPEN_EXIST - old llog, must exist
1278  *
1279  * \retval                      0 on successful open, llog_handle::lgh_obj
1280  *                              contains the dt_object of the llog.
1281  * \retval                      negative value on error
1282  */
1283 static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
1284                          struct llog_logid *logid, char *name,
1285                          enum llog_open_param open_param)
1286 {
1287         struct llog_thread_info         *lgi = llog_info(env);
1288         struct llog_ctxt                *ctxt = handle->lgh_ctxt;
1289         struct dt_object                *o;
1290         struct dt_device                *dt;
1291         struct ls_device                *ls;
1292         struct local_oid_storage        *los = NULL;
1293         int                              rc = 0;
1294         bool new_id = false;
1295
1296         ENTRY;
1297
1298         LASSERT(env);
1299         LASSERT(ctxt);
1300         LASSERT(ctxt->loc_exp);
1301         LASSERT(ctxt->loc_exp->exp_obd);
1302         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1303         LASSERT(dt);
1304         if (ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1305                 struct lu_object_conf conf = { 0 };
1306                 if (logid != NULL) {
1307                         logid_to_fid(logid, &lgi->lgi_fid);
1308                 } else {
1309                         /* If logid == NULL, then it means the caller needs
1310                          * to allocate new FID (llog_cat_declare_add_rec()). */
1311                         rc = dt_fid_alloc(env, dt, &lgi->lgi_fid, NULL, NULL);
1312                         if (rc < 0)
1313                                 RETURN(rc);
1314                         rc = 0;
1315                         conf.loc_flags = LOC_F_NEW;
1316                 }
1317
1318                 o = dt_locate_at(env, dt, &lgi->lgi_fid,
1319                                  dt->dd_lu_dev.ld_site->ls_top_dev, &conf);
1320                 if (IS_ERR(o))
1321                         RETURN(PTR_ERR(o));
1322
1323                 goto after_open;
1324         }
1325
1326         ls = ls_device_get(dt);
1327         if (IS_ERR(ls))
1328                 RETURN(PTR_ERR(ls));
1329
1330         mutex_lock(&ls->ls_los_mutex);
1331         los = dt_los_find(ls, name != NULL ? FID_SEQ_LLOG_NAME : FID_SEQ_LLOG);
1332         mutex_unlock(&ls->ls_los_mutex);
1333         LASSERT(los);
1334         ls_device_put(env, ls);
1335
1336         LASSERT(handle);
1337
1338         if (logid != NULL) {
1339                 logid_to_fid(logid, &lgi->lgi_fid);
1340         } else if (name) {
1341                 struct dt_object *llog_dir;
1342
1343                 llog_dir = llog_osd_dir_get(env, ctxt);
1344                 if (IS_ERR(llog_dir))
1345                         GOTO(out, rc = PTR_ERR(llog_dir));
1346                 dt_read_lock(env, llog_dir, 0);
1347                 rc = dt_lookup_dir(env, llog_dir, name, &lgi->lgi_fid);
1348                 dt_read_unlock(env, llog_dir);
1349                 dt_object_put(env, llog_dir);
1350                 if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
1351                         /* generate fid for new llog */
1352                         rc = local_object_fid_generate(env, los,
1353                                                        &lgi->lgi_fid);
1354                         new_id = true;
1355                 }
1356                 if (rc < 0)
1357                         GOTO(out, rc);
1358                 OBD_ALLOC(handle->lgh_name, strlen(name) + 1);
1359                 if (handle->lgh_name)
1360                         strcpy(handle->lgh_name, name);
1361                 else
1362                         GOTO(out, rc = -ENOMEM);
1363         } else {
1364                 LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
1365                 /* generate fid for new llog */
1366 generate:
1367                 rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
1368                 if (rc < 0)
1369                         GOTO(out, rc);
1370                 new_id = true;
1371         }
1372         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_LLOG_UMOUNT_RACE) &&
1373             cfs_fail_val == 1) {
1374                 cfs_fail_val = 2;
1375                 OBD_RACE(OBD_FAIL_MDS_LLOG_UMOUNT_RACE);
1376                 msleep(MSEC_PER_SEC);
1377         }
1378         o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
1379         if (IS_ERR(o))
1380                 GOTO(out_name, rc = PTR_ERR(o));
1381
1382         if (dt_object_exists(o) && new_id) {
1383                 /* llog exists with just generated ID, e.g. some old llog file
1384                  * still is in use or is orphan, drop a warn and skip it. */
1385                 CDEBUG(D_INFO, "%s: llog exists with the same FID: "DFID
1386                        ", skipping\n",
1387                        o->do_lu.lo_dev->ld_obd->obd_name,
1388                        PFID(lu_object_fid(&o->do_lu)));
1389                 dt_object_put(env, o);
1390                 /* just skip this llog ID, we shouldn't delete it because we
1391                  * don't know exactly what is its purpose and state. */
1392                 goto generate;
1393         }
1394
1395 after_open:
1396         /* No new llog is expected but doesn't exist */
1397         if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o)) {
1398                 CDEBUG(D_INFO, "%s: llog FID: "DFID" obj %p doesn`t exist\n",
1399                        o->do_lu.lo_dev->ld_obd->obd_name,
1400                        PFID(lu_object_fid(&o->do_lu)), o);
1401                 GOTO(out_put, rc = -ENOENT);
1402         }
1403         fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
1404         handle->lgh_obj = o;
1405         handle->private_data = los;
1406         LASSERT(handle->lgh_ctxt);
1407
1408         RETURN(rc);
1409
1410 out_put:
1411         dt_object_put(env, o);
1412 out_name:
1413         if (handle->lgh_name != NULL)
1414                 OBD_FREE(handle->lgh_name, strlen(name) + 1);
1415 out:
1416         if (los != NULL)
1417                 dt_los_put(los);
1418         RETURN(rc);
1419 }
1420
1421 /**
1422  * Get dir for regular fid log object
1423  *
1424  * Get directory for regular fid log object, and these regular fid log
1425  * object will be inserted under this directory, to satisfy the FS
1426  * consistency check, e2fsck etc.
1427  *
1428  * \param [in] env      execution environment
1429  * \param [in] dto      llog object
1430  *
1431  * \retval              pointer to the directory if it is found.
1432  * \retval              ERR_PTR(negative errno) if it fails.
1433  */
1434 struct dt_object *llog_osd_get_regular_fid_dir(const struct lu_env *env,
1435                                                struct dt_object *dto)
1436 {
1437         struct llog_thread_info *lgi = llog_info(env);
1438         struct seq_server_site *ss = dto->do_lu.lo_dev->ld_site->ld_seq_site;
1439         struct lu_seq_range     *range = &lgi->lgi_range;
1440         struct lu_fid           *dir_fid = &lgi->lgi_fid;
1441         struct dt_object        *dir;
1442         int                     rc;
1443         ENTRY;
1444
1445         fld_range_set_any(range);
1446         LASSERT(ss != NULL);
1447         rc = ss->ss_server_fld->lsf_seq_lookup(env, ss->ss_server_fld,
1448                                    fid_seq(lu_object_fid(&dto->do_lu)), range);
1449         if (rc < 0)
1450                 RETURN(ERR_PTR(rc));
1451
1452         lu_update_log_dir_fid(dir_fid, range->lsr_index);
1453         dir = dt_locate(env, lu2dt_dev(dto->do_lu.lo_dev), dir_fid);
1454         if (IS_ERR(dir))
1455                 RETURN(dir);
1456
1457         if (!dt_try_as_dir(env, dir)) {
1458                 dt_object_put(env, dir);
1459                 RETURN(ERR_PTR(-ENOTDIR));
1460         }
1461
1462         RETURN(dir);
1463 }
1464
1465 /**
1466  * Add llog object with regular FID to name entry
1467  *
1468  * Add llog object with regular FID to name space, and each llog
1469  * object on each MDT will be /update_log_dir/[seq:oid:ver],
1470  * so to satisfy the namespace consistency check, e2fsck etc.
1471  *
1472  * \param [in] env      execution environment
1473  * \param [in] dto      llog object
1474  * \param [in] th       thandle
1475  * \param [in] declare  if it is declare or execution
1476  *
1477  * \retval              0 if insertion succeeds.
1478  * \retval              negative errno if insertion fails.
1479  */
1480 static int
1481 llog_osd_regular_fid_add_name_entry(const struct lu_env *env,
1482                                     struct dt_object *dto,
1483                                     struct thandle *th, bool declare)
1484 {
1485         struct llog_thread_info *lgi = llog_info(env);
1486         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1487         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1488         struct dt_object        *dir;
1489         char                    *name = lgi->lgi_name;
1490         int                     rc;
1491         ENTRY;
1492
1493         if (!fid_is_norm(fid))
1494                 RETURN(0);
1495
1496         dir = llog_osd_get_regular_fid_dir(env, dto);
1497         if (IS_ERR(dir))
1498                 RETURN(PTR_ERR(dir));
1499
1500         rec->rec_fid = fid;
1501         rec->rec_type = S_IFREG;
1502         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1503         dt_write_lock(env, dir, 0);
1504         if (declare) {
1505                 rc = dt_declare_insert(env, dir, (struct dt_rec *)rec,
1506                                (struct dt_key *)name, th);
1507         } else {
1508                 rc = dt_insert(env, dir, (struct dt_rec *)rec,
1509                                (struct dt_key *)name, th);
1510         }
1511         dt_write_unlock(env, dir);
1512
1513         dt_object_put(env, dir);
1514         RETURN(rc);
1515 }
1516
1517
1518 /**
1519  * Implementation of the llog_operations::lop_declare_create
1520  *
1521  * This function declares the llog create. It declares also name insert
1522  * into llog directory in case of named llog.
1523  *
1524  * \param[in] env       execution environment
1525  * \param[in] res       llog handle of the current llog
1526  * \param[in] th        current transaction handle
1527  *
1528  * \retval              0 on successful create declaration
1529  * \retval              negative value on error
1530  */
1531 static int llog_osd_declare_create(const struct lu_env *env,
1532                                    struct llog_handle *res, struct thandle *th)
1533 {
1534         struct llog_thread_info         *lgi = llog_info(env);
1535         struct dt_insert_rec            *rec = &lgi->lgi_dt_rec;
1536         struct local_oid_storage        *los;
1537         struct dt_object                *o;
1538         int                              rc;
1539
1540         ENTRY;
1541
1542         LASSERT(res->lgh_obj);
1543         LASSERT(th);
1544
1545         /* object can be created by another thread */
1546         o = res->lgh_obj;
1547         if (dt_object_exists(o))
1548                 RETURN(0);
1549
1550         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1551                 struct llog_thread_info *lgi = llog_info(env);
1552
1553                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE;
1554                 lgi->lgi_attr.la_size = 0;
1555                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1556                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1557
1558                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
1559                                        &lgi->lgi_dof, th);
1560                 if (rc < 0)
1561                         RETURN(rc);
1562
1563
1564                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, true);
1565
1566                 RETURN(rc);
1567         }
1568         los = res->private_data;
1569         LASSERT(los);
1570
1571         rc = llog_osd_declare_new_object(env, los, o, th);
1572         if (rc)
1573                 RETURN(rc);
1574
1575         /* do not declare header initialization here as it's declared
1576          * in llog_osd_declare_write_rec() which is always called */
1577
1578         if (res->lgh_name) {
1579                 struct dt_object *llog_dir;
1580
1581                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1582                 if (IS_ERR(llog_dir))
1583                         RETURN(PTR_ERR(llog_dir));
1584                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1585                 rec->rec_fid = &lgi->lgi_fid;
1586                 rec->rec_type = S_IFREG;
1587                 rc = dt_declare_insert(env, llog_dir,
1588                                        (struct dt_rec *)rec,
1589                                        (struct dt_key *)res->lgh_name, th);
1590                 dt_object_put(env, llog_dir);
1591                 if (rc)
1592                         CERROR("%s: can't declare named llog %s: rc = %d\n",
1593                                o->do_lu.lo_dev->ld_obd->obd_name,
1594                                res->lgh_name, rc);
1595         }
1596         RETURN(rc);
1597 }
1598
1599 /**
1600  * Implementation of the llog_operations::lop_create
1601  *
1602  * This function creates the llog according with llog_handle::lgh_obj
1603  * and llog_handle::lgh_name.
1604  *
1605  * \param[in] env       execution environment
1606  * \param[in] res       llog handle of the current llog
1607  * \param[in] th        current transaction handle
1608  *
1609  * \retval              0 on successful create
1610  * \retval              negative value on error
1611  */
1612 static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
1613                            struct thandle *th)
1614 {
1615         struct llog_thread_info *lgi = llog_info(env);
1616         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1617         struct local_oid_storage *los;
1618         struct dt_object        *o;
1619         int                      rc = 0;
1620
1621         ENTRY;
1622
1623         LASSERT(env);
1624         o = res->lgh_obj;
1625         LASSERT(o);
1626
1627         /* llog can be already created */
1628         if (dt_object_exists(o))
1629                 RETURN(-EEXIST);
1630
1631         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1632                 struct llog_thread_info *lgi = llog_info(env);
1633
1634                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE | LA_TYPE;
1635                 lgi->lgi_attr.la_size = 0;
1636                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1637                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1638
1639                 dt_write_lock(env, o, 0);
1640                 rc = dt_create(env, o, &lgi->lgi_attr, NULL,
1641                                &lgi->lgi_dof, th);
1642                 dt_write_unlock(env, o);
1643                 if (rc < 0)
1644                         RETURN(rc);
1645
1646                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, false);
1647
1648                 RETURN(rc);
1649         }
1650
1651         los = res->private_data;
1652         LASSERT(los);
1653
1654         dt_write_lock(env, o, 0);
1655         if (!dt_object_exists(o))
1656                 rc = llog_osd_create_new_object(env, los, o, th);
1657         else
1658                 rc = -EEXIST;
1659
1660         dt_write_unlock(env, o);
1661         if (rc)
1662                 RETURN(rc);
1663
1664         if (res->lgh_name) {
1665                 struct dt_object *llog_dir;
1666
1667                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1668                 if (IS_ERR(llog_dir))
1669                         RETURN(PTR_ERR(llog_dir));
1670
1671                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1672                 rec->rec_fid = &lgi->lgi_fid;
1673                 rec->rec_type = S_IFREG;
1674                 dt_read_lock(env, llog_dir, 0);
1675                 rc = dt_insert(env, llog_dir, (struct dt_rec *)rec,
1676                                (struct dt_key *)res->lgh_name, th);
1677                 dt_read_unlock(env, llog_dir);
1678                 dt_object_put(env, llog_dir);
1679                 if (rc)
1680                         CERROR("%s: can't create named llog %s: rc = %d\n",
1681                                o->do_lu.lo_dev->ld_obd->obd_name,
1682                                res->lgh_name, rc);
1683         }
1684         RETURN(rc);
1685 }
1686
1687 /**
1688  * Implementation of the llog_operations::lop_close
1689  *
1690  * This function closes the llog. It just put llog object and referenced
1691  * local storage.
1692  *
1693  * \param[in] env       execution environment
1694  * \param[in] handle    llog handle of the current llog
1695  *
1696  * \retval              0 on successful llog close
1697  * \retval              negative value on error
1698  */
1699 static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle)
1700 {
1701         struct local_oid_storage        *los;
1702         int                              rc = 0;
1703
1704         ENTRY;
1705
1706         LASSERT(handle->lgh_obj);
1707
1708         if (handle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1709                 /* Remove the object from the cache, otherwise it may
1710                  * hold LOD being released during cleanup process */
1711                 dt_object_put_nocache(env, handle->lgh_obj);
1712                 LASSERT(handle->private_data == NULL);
1713                 RETURN(rc);
1714         } else {
1715                 dt_object_put(env, handle->lgh_obj);
1716         }
1717         los = handle->private_data;
1718         LASSERT(los);
1719         dt_los_put(los);
1720
1721         if (handle->lgh_name)
1722                 OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
1723
1724         RETURN(rc);
1725 }
1726
1727 /**
1728  * delete llog object name entry
1729  *
1730  * Delete llog object (with regular FID) from name space (under
1731  * update_log_dir).
1732  *
1733  * \param [in] env      execution environment
1734  * \param [in] dto      llog object
1735  * \param [in] th       thandle
1736  * \param [in] declare  if it is declare or execution
1737  *
1738  * \retval              0 if deletion succeeds.
1739  * \retval              negative errno if deletion fails.
1740  */
1741 static int
1742 llog_osd_regular_fid_del_name_entry(const struct lu_env *env,
1743                                     struct dt_object *dto,
1744                                     struct thandle *th, bool declare)
1745 {
1746         struct llog_thread_info *lgi = llog_info(env);
1747         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1748         struct dt_object        *dir;
1749         char                    *name = lgi->lgi_name;
1750         int                     rc;
1751         ENTRY;
1752
1753         if (!fid_is_norm(fid))
1754                 RETURN(0);
1755
1756         dir = llog_osd_get_regular_fid_dir(env, dto);
1757         if (IS_ERR(dir))
1758                 RETURN(PTR_ERR(dir));
1759
1760         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1761         dt_write_lock(env, dir, 0);
1762         if (declare) {
1763                 rc = dt_declare_delete(env, dir, (struct dt_key *)name,
1764                                        th);
1765         } else {
1766                 rc = dt_delete(env, dir, (struct dt_key *)name, th);
1767         }
1768         dt_write_unlock(env, dir);
1769
1770         dt_object_put(env, dir);
1771         RETURN(rc);
1772 }
1773
1774 /**
1775  * Implementation of the llog_operations::lop_declare_destroy
1776  *
1777  * This function declare destroys the llog and deletes also entry in the
1778  * llog directory in case of named llog. Llog should be opened prior that.
1779  *
1780  * \param[in] env               execution environment
1781  * \param[in] loghandle llog handle of the current llog
1782  *
1783  * \retval              0 on successful destroy
1784  * \retval              negative value on error
1785  */
1786 static int llog_osd_declare_destroy(const struct lu_env *env,
1787                                     struct llog_handle *loghandle,
1788                                     struct thandle *th)
1789 {
1790         struct llog_ctxt        *ctxt;
1791         struct dt_object        *o, *llog_dir = NULL;
1792         int                      rc;
1793
1794         ENTRY;
1795
1796         ctxt = loghandle->lgh_ctxt;
1797         LASSERT(ctxt);
1798
1799         o = loghandle->lgh_obj;
1800         LASSERT(o);
1801
1802         if (loghandle->lgh_name) {
1803                 llog_dir = llog_osd_dir_get(env, ctxt);
1804                 if (IS_ERR(llog_dir))
1805                         RETURN(PTR_ERR(llog_dir));
1806
1807                 rc = dt_declare_delete(env, llog_dir,
1808                                        (struct dt_key *)loghandle->lgh_name,
1809                                        th);
1810                 if (rc < 0)
1811                         GOTO(out_put, rc);
1812         }
1813
1814         rc = dt_declare_ref_del(env, o, th);
1815         if (rc < 0)
1816                 GOTO(out_put, rc);
1817
1818         rc = dt_declare_destroy(env, o, th);
1819         if (rc < 0)
1820                 GOTO(out_put, rc);
1821
1822         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1823                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, true);
1824                 if (rc < 0)
1825                         GOTO(out_put, rc);
1826         }
1827
1828 out_put:
1829         if (!(IS_ERR_OR_NULL(llog_dir)))
1830                 dt_object_put(env, llog_dir);
1831
1832         RETURN(rc);
1833 }
1834
1835
1836 /**
1837  * Implementation of the llog_operations::lop_destroy
1838  *
1839  * This function destroys the llog and deletes also entry in the
1840  * llog directory in case of named llog. Llog should be opened prior that.
1841  * Destroy method is not part of external transaction and does everything
1842  * inside.
1843  *
1844  * \param[in] env               execution environment
1845  * \param[in] loghandle llog handle of the current llog
1846  *
1847  * \retval              0 on successful destroy
1848  * \retval              negative value on error
1849  */
1850 static int llog_osd_destroy(const struct lu_env *env,
1851                             struct llog_handle *loghandle, struct thandle *th)
1852 {
1853         struct llog_ctxt        *ctxt;
1854         struct dt_object        *o, *llog_dir = NULL;
1855         int                      rc;
1856
1857         ENTRY;
1858
1859         ctxt = loghandle->lgh_ctxt;
1860         LASSERT(ctxt != NULL);
1861
1862         o = loghandle->lgh_obj;
1863         LASSERT(o != NULL);
1864
1865         dt_write_lock(env, o, 0);
1866         if (!llog_osd_exist(loghandle))
1867                 GOTO(out_unlock, rc = 0);
1868
1869         if (loghandle->lgh_name) {
1870                 llog_dir = llog_osd_dir_get(env, ctxt);
1871                 if (IS_ERR(llog_dir))
1872                         GOTO(out_unlock, rc = PTR_ERR(llog_dir));
1873
1874                 dt_read_lock(env, llog_dir, 0);
1875                 rc = dt_delete(env, llog_dir,
1876                                (struct dt_key *)loghandle->lgh_name,
1877                                th);
1878                 dt_read_unlock(env, llog_dir);
1879                 if (rc) {
1880                         CERROR("%s: can't remove llog %s: rc = %d\n",
1881                                o->do_lu.lo_dev->ld_obd->obd_name,
1882                                loghandle->lgh_name, rc);
1883                         GOTO(out_unlock, rc);
1884                 }
1885         }
1886
1887         dt_ref_del(env, o, th);
1888         rc = dt_destroy(env, o, th);
1889         if (rc < 0)
1890                 GOTO(out_unlock, rc);
1891
1892         loghandle->lgh_destroyed = true;
1893         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1894                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, false);
1895                 if (rc < 0)
1896                         GOTO(out_unlock, rc);
1897         }
1898
1899 out_unlock:
1900         dt_write_unlock(env, o);
1901         if (!(IS_ERR_OR_NULL(llog_dir)))
1902                 dt_object_put(env, llog_dir);
1903         RETURN(rc);
1904 }
1905
1906 /**
1907  * Implementation of the llog_operations::lop_setup
1908  *
1909  * This function setup the llog on local storage.
1910  *
1911  * \param[in] env       execution environment
1912  * \param[in] obd       obd device the llog belongs to
1913  * \param[in] olg       the llog group, it is always zero group now.
1914  * \param[in] ctxt_idx  the llog index, it defines the purpose of this llog.
1915  *                      Every new llog type have to use own index.
1916  * \param[in] disk_obd  the storage obd, where llog is stored.
1917  *
1918  * \retval              0 on successful llog setup
1919  * \retval              negative value on error
1920  */
1921 static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
1922                           struct obd_llog_group *olg, int ctxt_idx,
1923                           struct obd_device *disk_obd)
1924 {
1925         struct llog_thread_info         *lgi = llog_info(env);
1926         struct llog_ctxt                *ctxt;
1927         int                              rc = 0;
1928         ENTRY;
1929
1930         LASSERT(obd);
1931         LASSERT(olg->olg_ctxts[ctxt_idx]);
1932
1933         ctxt = llog_ctxt_get(olg->olg_ctxts[ctxt_idx]);
1934         LASSERT(ctxt);
1935
1936         if (disk_obd == NULL)
1937                 GOTO(out, rc = 0);
1938
1939         /* initialize data allowing to generate new fids,
1940          * literally we need a sequece */
1941         lgi->lgi_fid.f_seq = FID_SEQ_LLOG;
1942         lgi->lgi_fid.f_oid = 1;
1943         lgi->lgi_fid.f_ver = 0;
1944         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1945                                     &lgi->lgi_fid,
1946                                     &ctxt->loc_los_nameless);
1947         if (rc != 0)
1948                 GOTO(out, rc);
1949
1950         lgi->lgi_fid.f_seq = FID_SEQ_LLOG_NAME;
1951         lgi->lgi_fid.f_oid = 1;
1952         lgi->lgi_fid.f_ver = 0;
1953         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1954                                     &lgi->lgi_fid,
1955                                     &ctxt->loc_los_named);
1956         if (rc != 0) {
1957                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1958                 ctxt->loc_los_nameless = NULL;
1959         }
1960
1961         GOTO(out, rc);
1962
1963 out:
1964         llog_ctxt_put(ctxt);
1965         return rc;
1966 }
1967
1968 /**
1969  * Implementation of the llog_operations::lop_cleanup
1970  *
1971  * This function cleanups the llog on local storage.
1972  *
1973  * \param[in] env       execution environment
1974  * \param[in] ctxt      the llog context
1975  *
1976  * \retval              0
1977  */
1978 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
1979 {
1980         if (ctxt->loc_los_nameless != NULL) {
1981                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1982                 ctxt->loc_los_nameless = NULL;
1983         }
1984
1985         if (ctxt->loc_los_named != NULL) {
1986                 local_oid_storage_fini(env, ctxt->loc_los_named);
1987                 ctxt->loc_los_named = NULL;
1988         }
1989
1990         return 0;
1991 }
1992
1993 const struct llog_operations llog_osd_ops = {
1994         .lop_next_block         = llog_osd_next_block,
1995         .lop_prev_block         = llog_osd_prev_block,
1996         .lop_read_header        = llog_osd_read_header,
1997         .lop_declare_destroy    = llog_osd_declare_destroy,
1998         .lop_destroy            = llog_osd_destroy,
1999         .lop_setup              = llog_osd_setup,
2000         .lop_cleanup            = llog_osd_cleanup,
2001         .lop_open               = llog_osd_open,
2002         .lop_exist              = llog_osd_exist,
2003         .lop_declare_create     = llog_osd_declare_create,
2004         .lop_create             = llog_osd_create,
2005         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2006         .lop_write_rec          = llog_osd_write_rec,
2007         .lop_close              = llog_osd_close,
2008 };
2009 EXPORT_SYMBOL(llog_osd_ops);
2010
2011 const struct llog_operations llog_common_cat_ops = {
2012         .lop_next_block         = llog_osd_next_block,
2013         .lop_prev_block         = llog_osd_prev_block,
2014         .lop_read_header        = llog_osd_read_header,
2015         .lop_declare_destroy    = llog_osd_declare_destroy,
2016         .lop_destroy            = llog_osd_destroy,
2017         .lop_setup              = llog_osd_setup,
2018         .lop_cleanup            = llog_osd_cleanup,
2019         .lop_open               = llog_osd_open,
2020         .lop_exist              = llog_osd_exist,
2021         .lop_declare_create     = llog_osd_declare_create,
2022         .lop_create             = llog_osd_create,
2023         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2024         .lop_write_rec          = llog_osd_write_rec,
2025         .lop_close              = llog_osd_close,
2026         .lop_add                = llog_cat_add_rec,
2027         .lop_declare_add        = llog_cat_declare_add_rec,
2028 };
2029 EXPORT_SYMBOL(llog_common_cat_ops);
2030
2031 /**
2032  * Read the special file which contains the list of llog catalogs IDs
2033  *
2034  * This function reads the CATALOGS file which contains the array of llog
2035  * catalogs IDs. The main purpose of this file is to store OSP llogs indexed
2036  * by OST/MDT number.
2037  *
2038  * \param[in]  env              execution environment
2039  * \param[in]  d                corresponding storage device
2040  * \param[in]  idx              position to start from, usually OST/MDT index
2041  * \param[in]  count            how many catalog IDs to read
2042  * \param[out] idarray          the buffer for the data. If it is NULL then
2043  *                              function returns just number of catalog IDs
2044  *                              in the file.
2045  * \param[in]  fid              LLOG_CATALOGS_OID for CATALOG object
2046  *
2047  * \retval                      0 on successful read of catalog IDs
2048  * \retval                      negative value on error
2049  * \retval                      positive value which is number of records in
2050  *                              the file if \a idarray is NULL
2051  */
2052 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
2053                           int idx, int count, struct llog_catid *idarray,
2054                           const struct lu_fid *fid)
2055 {
2056         struct llog_thread_info *lgi = llog_info(env);
2057         struct dt_object        *o = NULL;
2058         struct thandle          *th;
2059         int                      rc, size;
2060
2061         ENTRY;
2062
2063         LASSERT(d);
2064
2065         size = sizeof(*idarray) * count;
2066         lgi->lgi_off = idx *  sizeof(*idarray);
2067
2068         lgi->lgi_fid = *fid;
2069         o = dt_locate(env, d, &lgi->lgi_fid);
2070         if (IS_ERR(o))
2071                 RETURN(PTR_ERR(o));
2072
2073         if (!dt_object_exists(o)) {
2074                 th = dt_trans_create(env, d);
2075                 if (IS_ERR(th))
2076                         GOTO(out, rc = PTR_ERR(th));
2077
2078                 lgi->lgi_attr.la_valid = LA_MODE | LA_TYPE;
2079                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
2080                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
2081
2082                 th->th_wait_submit = 1;
2083                 /* Make the llog object creation synchronization, so
2084                  * it will be reliable to the reference, especially
2085                  * for remote reference */
2086                 th->th_sync = 1;
2087
2088                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
2089                                        &lgi->lgi_dof, th);
2090                 if (rc)
2091                         GOTO(out_trans, rc);
2092
2093                 rc = dt_trans_start_local(env, d, th);
2094                 if (rc)
2095                         GOTO(out_trans, rc);
2096
2097                 dt_write_lock(env, o, 0);
2098                 if (!dt_object_exists(o))
2099                         rc = dt_create(env, o, &lgi->lgi_attr, NULL,
2100                                        &lgi->lgi_dof, th);
2101                 dt_write_unlock(env, o);
2102 out_trans:
2103                 dt_trans_stop(env, d, th);
2104                 if (rc)
2105                         GOTO(out, rc);
2106         }
2107
2108         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2109         if (rc)
2110                 GOTO(out, rc);
2111
2112         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2113                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2114                        o->do_lu.lo_dev->ld_obd->obd_name,
2115                        lgi->lgi_attr.la_mode);
2116                 GOTO(out, rc = -ENOENT);
2117         }
2118
2119         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
2120                (int)lgi->lgi_attr.la_size, size);
2121
2122         /* return just number of llogs */
2123         if (idarray == NULL) {
2124                 rc = lgi->lgi_attr.la_size / sizeof(*idarray);
2125                 GOTO(out, rc);
2126         }
2127
2128         /* read for new ost index or for empty file */
2129         memset(idarray, 0, size);
2130         if (lgi->lgi_attr.la_size <= lgi->lgi_off)
2131                 GOTO(out, rc = 0);
2132         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
2133                 size = lgi->lgi_attr.la_size - lgi->lgi_off;
2134
2135         lgi->lgi_buf.lb_buf = idarray;
2136         lgi->lgi_buf.lb_len = size;
2137         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
2138         /* -EFAULT means the llog is a sparse file. This is not an error
2139          * after arbitrary OST index is supported. */
2140         if (rc < 0 && rc != -EFAULT) {
2141                 CERROR("%s: error reading CATALOGS: rc = %d\n",
2142                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
2143                 GOTO(out, rc);
2144         }
2145
2146         EXIT;
2147 out:
2148         dt_object_put(env, o);
2149         RETURN(rc);
2150 }
2151 EXPORT_SYMBOL(llog_osd_get_cat_list);
2152
2153 /**
2154  * Write the special file which contains the list of llog catalogs IDs
2155  *
2156  * This function writes the CATALOG file which contains the array of llog
2157  * catalogs IDs. It is used mostly to store OSP llogs indexed by OST/MDT
2158  * number.
2159  *
2160  * \param[in]  env      execution environment
2161  * \param[in]  d        corresponding storage device
2162  * \param[in]  idx      position to start from, usually OST/MDT index
2163  * \param[in]  count    how many catalog IDs to write
2164  * \param[out] idarray  the buffer with the data to write.
2165  * \param[in]  fid      LLOG_CATALOGS_OID for CATALOG object
2166  *
2167  * \retval              0 on successful write of catalog IDs
2168  * \retval              negative value on error
2169  */
2170 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
2171                           int idx, int count, struct llog_catid *idarray,
2172                           const struct lu_fid *fid)
2173 {
2174         struct llog_thread_info *lgi = llog_info(env);
2175         struct dt_object        *o = NULL;
2176         struct thandle          *th;
2177         int                      rc, size;
2178
2179         if (count == 0)
2180                 RETURN(0);
2181
2182         LASSERT(d);
2183
2184         size = sizeof(*idarray) * count;
2185         lgi->lgi_off = idx * sizeof(*idarray);
2186         lgi->lgi_fid = *fid;
2187
2188         o = dt_locate(env, d, &lgi->lgi_fid);
2189         if (IS_ERR(o))
2190                 RETURN(PTR_ERR(o));
2191
2192         if (!dt_object_exists(o))
2193                 GOTO(out, rc = -ENOENT);
2194
2195         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2196         if (rc)
2197                 GOTO(out, rc);
2198
2199         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2200                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2201                        o->do_lu.lo_dev->ld_obd->obd_name,
2202                        lgi->lgi_attr.la_mode);
2203                 GOTO(out, rc = -ENOENT);
2204         }
2205
2206         th = dt_trans_create(env, d);
2207         if (IS_ERR(th))
2208                 GOTO(out, rc = PTR_ERR(th));
2209
2210         lgi->lgi_buf.lb_len = size;
2211         lgi->lgi_buf.lb_buf = idarray;
2212         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, lgi->lgi_off, th);
2213         if (rc)
2214                 GOTO(out_trans, rc);
2215
2216         /* For update log, this happens during initialization,
2217          * see lod_sub_prep_llog(), and we need make sure catlog
2218          * file ID is written to catlist file(committed) before
2219          * cross-MDT operation write update records to catlog FILE,
2220          * otherwise, during failover these update records might
2221          * missing */
2222         if (fid_is_update_log(fid))
2223                 th->th_sync = 1;
2224
2225         rc = dt_trans_start_local(env, d, th);
2226         if (rc)
2227                 GOTO(out_trans, rc);
2228
2229         th->th_wait_submit = 1;
2230
2231         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
2232         if (rc)
2233                 CDEBUG(D_INODE, "can't write CATALOGS at index %d: rc = %d\n",
2234                        idx, rc);
2235 out_trans:
2236         dt_trans_stop(env, d, th);
2237 out:
2238         dt_object_put(env, o);
2239         RETURN(rc);
2240 }
2241 EXPORT_SYMBOL(llog_osd_put_cat_list);