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         /* being here means we reach end of llog but didn't find needed idx
1089          * normally could happen while processing remote llog, return -EBADR
1090          * to indicate access beyond end of file like dt_read() does and to
1091          * distunguish this situation from real IO or network issues.
1092          */
1093         GOTO(out, rc = -EBADR);
1094 out:
1095         dt_read_unlock(env, o);
1096         return rc;
1097 }
1098
1099 /**
1100  * Implementation of the llog_operations::lop_prev_block
1101  *
1102  * This function finds the llog block to return which contains
1103  * record with required index but in reverse order - from end of llog
1104  * to the beginning.
1105  * It is main part of reverse llog processing.
1106  *
1107  * \param[in] env       execution environment
1108  * \param[in] loghandle llog handle of the current llog
1109  * \param[in] prev_idx  target index to find
1110  * \param[in] buf       pointer to data buffer to fill
1111  * \param[in] len       required len to read, it is llog_chunk_size usually.
1112  *
1113  * \retval              0 on successful buffer read
1114  * \retval              negative value on error
1115  */
1116 static int llog_osd_prev_block(const struct lu_env *env,
1117                                struct llog_handle *loghandle,
1118                                int prev_idx, void *buf, int len)
1119 {
1120         struct llog_thread_info *lgi = llog_info(env);
1121         struct dt_object        *o;
1122         struct dt_device        *dt;
1123         loff_t                   cur_offset;
1124         __u32                   chunk_size;
1125         int                      rc;
1126
1127         ENTRY;
1128
1129         chunk_size = loghandle->lgh_hdr->llh_hdr.lrh_len;
1130         if (len == 0 || len & (chunk_size - 1))
1131                 RETURN(-EINVAL);
1132
1133         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
1134
1135         LASSERT(loghandle);
1136         LASSERT(loghandle->lgh_ctxt);
1137
1138         o = loghandle->lgh_obj;
1139         LASSERT(o);
1140         dt_read_lock(env, o, 0);
1141         if (!llog_osd_exist(loghandle))
1142                 GOTO(out, rc = -ESTALE);
1143
1144         dt = lu2dt_dev(o->do_lu.lo_dev);
1145         LASSERT(dt);
1146
1147         /* Let's only use mini record size for previous block read
1148          * for now XXX */
1149         cur_offset = chunk_size;
1150         llog_skip_over(loghandle, &cur_offset, 0, prev_idx,
1151                        chunk_size, true);
1152
1153         rc = dt_attr_get(env, o, &lgi->lgi_attr);
1154         if (rc)
1155                 GOTO(out, rc);
1156
1157         while (cur_offset < lgi->lgi_attr.la_size) {
1158                 struct llog_rec_hdr     *rec, *last_rec;
1159                 struct llog_rec_tail    *tail;
1160
1161                 lgi->lgi_buf.lb_len = len;
1162                 lgi->lgi_buf.lb_buf = buf;
1163                 rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
1164                 if (rc < 0) {
1165                         CERROR("%s: can't read llog block from log "DFID
1166                                " offset %llu: rc = %d\n",
1167                                o->do_lu.lo_dev->ld_obd->obd_name,
1168                                PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
1169                         GOTO(out, rc);
1170                 }
1171
1172                 if (rc == 0) /* end of file, nothing to do */
1173                         GOTO(out, rc);
1174
1175                 if (rc < sizeof(*tail)) {
1176                         CERROR("%s: invalid llog block at log id "DFID":%x "
1177                                "offset %llu\n",
1178                                o->do_lu.lo_dev->ld_obd->obd_name,
1179                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1180                                loghandle->lgh_id.lgl_ogen, cur_offset);
1181                         GOTO(out, rc = -EINVAL);
1182                 }
1183
1184                 rec = buf;
1185                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1186                         lustre_swab_llog_rec(rec);
1187
1188                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1189                                                 sizeof(struct llog_rec_tail));
1190                 /* get the last record in block */
1191                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1192                                                    le32_to_cpu(tail->lrt_len));
1193
1194                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1195                         lustre_swab_llog_rec(last_rec);
1196                 LASSERT(last_rec->lrh_index == tail->lrt_index);
1197
1198                 /* this shouldn't happen */
1199                 if (tail->lrt_index == 0) {
1200                         CERROR("%s: invalid llog tail at log id "DFID":%x "
1201                                "offset %llu\n",
1202                                o->do_lu.lo_dev->ld_obd->obd_name,
1203                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1204                                loghandle->lgh_id.lgl_ogen, cur_offset);
1205                         GOTO(out, rc = -EINVAL);
1206                 }
1207                 if (tail->lrt_index < prev_idx)
1208                         continue;
1209
1210                 /* sanity check that the start of the new buffer is no farther
1211                  * than the record that we wanted.  This shouldn't happen. */
1212                 if (rec->lrh_index > prev_idx) {
1213                         CERROR("%s: missed desired record? %u > %u\n",
1214                                o->do_lu.lo_dev->ld_obd->obd_name,
1215                                rec->lrh_index, prev_idx);
1216                         GOTO(out, rc = -ENOENT);
1217                 }
1218
1219                 /* Trim unsupported extensions for compat w/ older clients */
1220                 changelog_block_trim_ext(rec, last_rec, loghandle);
1221
1222                 GOTO(out, rc = 0);
1223         }
1224         GOTO(out, rc = -EIO);
1225 out:
1226         dt_read_unlock(env, o);
1227         return rc;
1228 }
1229
1230 /**
1231  * This is helper function to get llog directory object. It is used by named
1232  * llog operations to find/insert/delete llog entry from llog directory.
1233  *
1234  * \param[in] env       execution environment
1235  * \param[in] ctxt      llog context
1236  *
1237  * \retval              dt_object of llog directory
1238  * \retval              ERR_PTR of negative value on error
1239  */
1240 static struct dt_object *llog_osd_dir_get(const struct lu_env *env,
1241                                           struct llog_ctxt *ctxt)
1242 {
1243         struct dt_device        *dt;
1244         struct dt_thread_info   *dti = dt_info(env);
1245         struct dt_object        *dir;
1246         int                      rc;
1247
1248         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1249         if (ctxt->loc_dir == NULL) {
1250                 rc = dt_root_get(env, dt, &dti->dti_fid);
1251                 if (rc)
1252                         return ERR_PTR(rc);
1253                 dir = dt_locate(env, dt, &dti->dti_fid);
1254
1255                 if (!IS_ERR(dir) && !dt_try_as_dir(env, dir, true)) {
1256                         dt_object_put(env, dir);
1257                         return ERR_PTR(-ENOTDIR);
1258                 }
1259         } else {
1260                 lu_object_get(&ctxt->loc_dir->do_lu);
1261                 dir = ctxt->loc_dir;
1262         }
1263
1264         return dir;
1265 }
1266
1267 /**
1268  * Implementation of the llog_operations::lop_open
1269  *
1270  * This function opens the llog by its logid or by name, it may open also
1271  * non existent llog and assing then new id to it.
1272  * The llog_open/llog_close pair works similar to lu_object_find/put,
1273  * the object may not exist prior open. The result of open is just dt_object
1274  * in the llog header.
1275  *
1276  * \param[in] env               execution environment
1277  * \param[in] handle            llog handle of the current llog
1278  * \param[in] logid             logid of llog to open (nameless llog)
1279  * \param[in] name              name of llog to open (named llog)
1280  * \param[in] open_param
1281  *                              LLOG_OPEN_NEW - new llog, may not exist
1282  *                              LLOG_OPEN_EXIST - old llog, must exist
1283  *
1284  * \retval                      0 on successful open, llog_handle::lgh_obj
1285  *                              contains the dt_object of the llog.
1286  * \retval                      negative value on error
1287  */
1288 static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
1289                          struct llog_logid *logid, char *name,
1290                          enum llog_open_param open_param)
1291 {
1292         struct llog_thread_info         *lgi = llog_info(env);
1293         struct llog_ctxt                *ctxt = handle->lgh_ctxt;
1294         struct dt_object                *o;
1295         struct dt_device                *dt;
1296         struct ls_device                *ls;
1297         struct local_oid_storage        *los = NULL;
1298         int                              rc = 0;
1299         bool new_id = false;
1300
1301         ENTRY;
1302
1303         LASSERT(env);
1304         LASSERT(ctxt);
1305         LASSERT(ctxt->loc_exp);
1306         LASSERT(ctxt->loc_exp->exp_obd);
1307         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1308         LASSERT(dt);
1309         if (ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1310                 struct lu_object_conf conf = { 0 };
1311                 if (logid != NULL) {
1312                         logid_to_fid(logid, &lgi->lgi_fid);
1313                 } else {
1314                         /* If logid == NULL, then it means the caller needs
1315                          * to allocate new FID (llog_cat_declare_add_rec()). */
1316                         rc = dt_fid_alloc(env, dt, &lgi->lgi_fid, NULL, NULL);
1317                         if (rc < 0)
1318                                 RETURN(rc);
1319                         rc = 0;
1320                         conf.loc_flags = LOC_F_NEW;
1321                 }
1322
1323                 o = dt_locate_at(env, dt, &lgi->lgi_fid,
1324                                  dt->dd_lu_dev.ld_site->ls_top_dev, &conf);
1325                 if (IS_ERR(o))
1326                         RETURN(PTR_ERR(o));
1327
1328                 goto after_open;
1329         }
1330
1331         ls = ls_device_get(dt);
1332         if (IS_ERR(ls))
1333                 RETURN(PTR_ERR(ls));
1334
1335         mutex_lock(&ls->ls_los_mutex);
1336         los = dt_los_find(ls, name != NULL ? FID_SEQ_LLOG_NAME : FID_SEQ_LLOG);
1337         mutex_unlock(&ls->ls_los_mutex);
1338         LASSERT(los);
1339         ls_device_put(env, ls);
1340
1341         LASSERT(handle);
1342
1343         if (logid != NULL) {
1344                 logid_to_fid(logid, &lgi->lgi_fid);
1345         } else if (name) {
1346                 struct dt_object *llog_dir;
1347
1348                 llog_dir = llog_osd_dir_get(env, ctxt);
1349                 if (IS_ERR(llog_dir))
1350                         GOTO(out, rc = PTR_ERR(llog_dir));
1351                 dt_read_lock(env, llog_dir, 0);
1352                 rc = dt_lookup_dir(env, llog_dir, name, &lgi->lgi_fid);
1353                 dt_read_unlock(env, llog_dir);
1354                 dt_object_put(env, llog_dir);
1355                 if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
1356                         /* generate fid for new llog */
1357                         rc = local_object_fid_generate(env, los,
1358                                                        &lgi->lgi_fid);
1359                         new_id = true;
1360                 }
1361                 if (rc < 0)
1362                         GOTO(out, rc);
1363                 OBD_ALLOC(handle->lgh_name, strlen(name) + 1);
1364                 if (handle->lgh_name)
1365                         strcpy(handle->lgh_name, name);
1366                 else
1367                         GOTO(out, rc = -ENOMEM);
1368         } else {
1369                 LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
1370                 /* generate fid for new llog */
1371 generate:
1372                 rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
1373                 if (rc < 0)
1374                         GOTO(out, rc);
1375                 new_id = true;
1376         }
1377         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_LLOG_UMOUNT_RACE) &&
1378             cfs_fail_val == 1) {
1379                 cfs_fail_val = 2;
1380                 OBD_RACE(OBD_FAIL_MDS_LLOG_UMOUNT_RACE);
1381                 msleep(MSEC_PER_SEC);
1382         }
1383         o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
1384         if (IS_ERR(o))
1385                 GOTO(out_name, rc = PTR_ERR(o));
1386
1387         if (dt_object_exists(o) && new_id) {
1388                 /* llog exists with just generated ID, e.g. some old llog file
1389                  * still is in use or is orphan, drop a warn and skip it. */
1390                 CDEBUG(D_INFO, "%s: llog exists with the same FID: "DFID
1391                        ", skipping\n",
1392                        o->do_lu.lo_dev->ld_obd->obd_name,
1393                        PFID(lu_object_fid(&o->do_lu)));
1394                 dt_object_put(env, o);
1395                 /* just skip this llog ID, we shouldn't delete it because we
1396                  * don't know exactly what is its purpose and state. */
1397                 goto generate;
1398         }
1399
1400 after_open:
1401         /* No new llog is expected but doesn't exist */
1402         if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o)) {
1403                 CDEBUG(D_INFO, "%s: llog FID: "DFID" obj %p doesn`t exist\n",
1404                        o->do_lu.lo_dev->ld_obd->obd_name,
1405                        PFID(lu_object_fid(&o->do_lu)), o);
1406                 GOTO(out_put, rc = -ENOENT);
1407         }
1408         fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
1409         handle->lgh_obj = o;
1410         handle->private_data = los;
1411         LASSERT(handle->lgh_ctxt);
1412
1413         RETURN(rc);
1414
1415 out_put:
1416         dt_object_put(env, o);
1417 out_name:
1418         if (handle->lgh_name != NULL)
1419                 OBD_FREE(handle->lgh_name, strlen(name) + 1);
1420 out:
1421         if (los != NULL)
1422                 dt_los_put(los);
1423         RETURN(rc);
1424 }
1425
1426 /**
1427  * Get dir for regular fid log object
1428  *
1429  * Get directory for regular fid log object, and these regular fid log
1430  * object will be inserted under this directory, to satisfy the FS
1431  * consistency check, e2fsck etc.
1432  *
1433  * \param [in] env      execution environment
1434  * \param [in] dto      llog object
1435  *
1436  * \retval              pointer to the directory if it is found.
1437  * \retval              ERR_PTR(negative errno) if it fails.
1438  */
1439 struct dt_object *llog_osd_get_regular_fid_dir(const struct lu_env *env,
1440                                                struct dt_object *dto)
1441 {
1442         struct llog_thread_info *lgi = llog_info(env);
1443         struct seq_server_site *ss = dto->do_lu.lo_dev->ld_site->ld_seq_site;
1444         struct lu_seq_range     *range = &lgi->lgi_range;
1445         struct lu_fid           *dir_fid = &lgi->lgi_fid;
1446         struct dt_object        *dir;
1447         int                     rc;
1448         ENTRY;
1449
1450         fld_range_set_any(range);
1451         LASSERT(ss != NULL);
1452         rc = ss->ss_server_fld->lsf_seq_lookup(env, ss->ss_server_fld,
1453                                    fid_seq(lu_object_fid(&dto->do_lu)), range);
1454         if (rc < 0)
1455                 RETURN(ERR_PTR(rc));
1456
1457         lu_update_log_dir_fid(dir_fid, range->lsr_index);
1458         dir = dt_locate(env, lu2dt_dev(dto->do_lu.lo_dev), dir_fid);
1459         if (IS_ERR(dir))
1460                 RETURN(dir);
1461
1462         if (!dt_try_as_dir(env, dir, true)) {
1463                 dt_object_put(env, dir);
1464                 RETURN(ERR_PTR(-ENOTDIR));
1465         }
1466
1467         RETURN(dir);
1468 }
1469
1470 /**
1471  * Add llog object with regular FID to name entry
1472  *
1473  * Add llog object with regular FID to name space, and each llog
1474  * object on each MDT will be /update_log_dir/[seq:oid:ver],
1475  * so to satisfy the namespace consistency check, e2fsck etc.
1476  *
1477  * \param [in] env      execution environment
1478  * \param [in] dto      llog object
1479  * \param [in] th       thandle
1480  * \param [in] declare  if it is declare or execution
1481  *
1482  * \retval              0 if insertion succeeds.
1483  * \retval              negative errno if insertion fails.
1484  */
1485 static int
1486 llog_osd_regular_fid_add_name_entry(const struct lu_env *env,
1487                                     struct dt_object *dto,
1488                                     struct thandle *th, bool declare)
1489 {
1490         struct llog_thread_info *lgi = llog_info(env);
1491         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1492         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1493         struct dt_object        *dir;
1494         char                    *name = lgi->lgi_name;
1495         int                     rc;
1496         ENTRY;
1497
1498         if (!fid_is_norm(fid))
1499                 RETURN(0);
1500
1501         dir = llog_osd_get_regular_fid_dir(env, dto);
1502         if (IS_ERR(dir))
1503                 RETURN(PTR_ERR(dir));
1504
1505         rec->rec_fid = fid;
1506         rec->rec_type = S_IFREG;
1507         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1508         dt_write_lock(env, dir, 0);
1509         if (declare) {
1510                 rc = dt_declare_insert(env, dir, (struct dt_rec *)rec,
1511                                (struct dt_key *)name, th);
1512         } else {
1513                 rc = dt_insert(env, dir, (struct dt_rec *)rec,
1514                                (struct dt_key *)name, th);
1515         }
1516         dt_write_unlock(env, dir);
1517
1518         dt_object_put(env, dir);
1519         RETURN(rc);
1520 }
1521
1522
1523 /**
1524  * Implementation of the llog_operations::lop_declare_create
1525  *
1526  * This function declares the llog create. It declares also name insert
1527  * into llog directory in case of named llog.
1528  *
1529  * \param[in] env       execution environment
1530  * \param[in] res       llog handle of the current llog
1531  * \param[in] th        current transaction handle
1532  *
1533  * \retval              0 on successful create declaration
1534  * \retval              negative value on error
1535  */
1536 static int llog_osd_declare_create(const struct lu_env *env,
1537                                    struct llog_handle *res, struct thandle *th)
1538 {
1539         struct llog_thread_info         *lgi = llog_info(env);
1540         struct dt_insert_rec            *rec = &lgi->lgi_dt_rec;
1541         struct local_oid_storage        *los;
1542         struct dt_object                *o;
1543         int                              rc;
1544
1545         ENTRY;
1546
1547         LASSERT(res->lgh_obj);
1548         LASSERT(th);
1549
1550         /* object can be created by another thread */
1551         o = res->lgh_obj;
1552         if (dt_object_exists(o))
1553                 RETURN(0);
1554
1555         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1556                 struct llog_thread_info *lgi = llog_info(env);
1557
1558                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE;
1559                 lgi->lgi_attr.la_size = 0;
1560                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1561                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1562
1563                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
1564                                        &lgi->lgi_dof, th);
1565                 if (rc < 0)
1566                         RETURN(rc);
1567
1568
1569                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, true);
1570
1571                 RETURN(rc);
1572         }
1573         los = res->private_data;
1574         LASSERT(los);
1575
1576         rc = llog_osd_declare_new_object(env, los, o, th);
1577         if (rc)
1578                 RETURN(rc);
1579
1580         /* do not declare header initialization here as it's declared
1581          * in llog_osd_declare_write_rec() which is always called */
1582
1583         if (res->lgh_name) {
1584                 struct dt_object *llog_dir;
1585
1586                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1587                 if (IS_ERR(llog_dir))
1588                         RETURN(PTR_ERR(llog_dir));
1589                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1590                 rec->rec_fid = &lgi->lgi_fid;
1591                 rec->rec_type = S_IFREG;
1592                 rc = dt_declare_insert(env, llog_dir,
1593                                        (struct dt_rec *)rec,
1594                                        (struct dt_key *)res->lgh_name, th);
1595                 dt_object_put(env, llog_dir);
1596                 if (rc)
1597                         CERROR("%s: can't declare named llog %s: rc = %d\n",
1598                                o->do_lu.lo_dev->ld_obd->obd_name,
1599                                res->lgh_name, rc);
1600         }
1601         RETURN(rc);
1602 }
1603
1604 /**
1605  * Implementation of the llog_operations::lop_create
1606  *
1607  * This function creates the llog according with llog_handle::lgh_obj
1608  * and llog_handle::lgh_name.
1609  *
1610  * \param[in] env       execution environment
1611  * \param[in] res       llog handle of the current llog
1612  * \param[in] th        current transaction handle
1613  *
1614  * \retval              0 on successful create
1615  * \retval              negative value on error
1616  */
1617 static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
1618                            struct thandle *th)
1619 {
1620         struct llog_thread_info *lgi = llog_info(env);
1621         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1622         struct local_oid_storage *los;
1623         struct dt_object        *o;
1624         int                      rc = 0;
1625
1626         ENTRY;
1627
1628         LASSERT(env);
1629         o = res->lgh_obj;
1630         LASSERT(o);
1631
1632         /* llog can be already created */
1633         if (dt_object_exists(o))
1634                 RETURN(-EEXIST);
1635
1636         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1637                 struct llog_thread_info *lgi = llog_info(env);
1638
1639                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE | LA_TYPE;
1640                 lgi->lgi_attr.la_size = 0;
1641                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1642                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1643
1644                 dt_write_lock(env, o, 0);
1645                 rc = dt_create(env, o, &lgi->lgi_attr, NULL,
1646                                &lgi->lgi_dof, th);
1647                 dt_write_unlock(env, o);
1648                 if (rc < 0)
1649                         RETURN(rc);
1650
1651                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, false);
1652
1653                 RETURN(rc);
1654         }
1655
1656         los = res->private_data;
1657         LASSERT(los);
1658
1659         dt_write_lock(env, o, 0);
1660         if (!dt_object_exists(o))
1661                 rc = llog_osd_create_new_object(env, los, o, th);
1662         else
1663                 rc = -EEXIST;
1664
1665         dt_write_unlock(env, o);
1666         if (rc)
1667                 RETURN(rc);
1668
1669         if (res->lgh_name) {
1670                 struct dt_object *llog_dir;
1671
1672                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1673                 if (IS_ERR(llog_dir))
1674                         RETURN(PTR_ERR(llog_dir));
1675
1676                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1677                 rec->rec_fid = &lgi->lgi_fid;
1678                 rec->rec_type = S_IFREG;
1679                 dt_read_lock(env, llog_dir, 0);
1680                 rc = dt_insert(env, llog_dir, (struct dt_rec *)rec,
1681                                (struct dt_key *)res->lgh_name, th);
1682                 dt_read_unlock(env, llog_dir);
1683                 dt_object_put(env, llog_dir);
1684                 if (rc)
1685                         CERROR("%s: can't create named llog %s: rc = %d\n",
1686                                o->do_lu.lo_dev->ld_obd->obd_name,
1687                                res->lgh_name, rc);
1688         }
1689         RETURN(rc);
1690 }
1691
1692 /**
1693  * Implementation of the llog_operations::lop_close
1694  *
1695  * This function closes the llog. It just put llog object and referenced
1696  * local storage.
1697  *
1698  * \param[in] env       execution environment
1699  * \param[in] handle    llog handle of the current llog
1700  *
1701  * \retval              0 on successful llog close
1702  * \retval              negative value on error
1703  */
1704 static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle)
1705 {
1706         struct local_oid_storage        *los;
1707         int                              rc = 0;
1708
1709         ENTRY;
1710
1711         LASSERT(handle->lgh_obj);
1712
1713         if (handle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1714                 /* Remove the object from the cache, otherwise it may
1715                  * hold LOD being released during cleanup process */
1716                 dt_object_put_nocache(env, handle->lgh_obj);
1717                 LASSERT(handle->private_data == NULL);
1718                 RETURN(rc);
1719         } else {
1720                 dt_object_put(env, handle->lgh_obj);
1721         }
1722         los = handle->private_data;
1723         LASSERT(los);
1724         dt_los_put(los);
1725
1726         if (handle->lgh_name)
1727                 OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
1728
1729         RETURN(rc);
1730 }
1731
1732 /**
1733  * delete llog object name entry
1734  *
1735  * Delete llog object (with regular FID) from name space (under
1736  * update_log_dir).
1737  *
1738  * \param [in] env      execution environment
1739  * \param [in] dto      llog object
1740  * \param [in] th       thandle
1741  * \param [in] declare  if it is declare or execution
1742  *
1743  * \retval              0 if deletion succeeds.
1744  * \retval              negative errno if deletion fails.
1745  */
1746 static int
1747 llog_osd_regular_fid_del_name_entry(const struct lu_env *env,
1748                                     struct dt_object *dto,
1749                                     struct thandle *th, bool declare)
1750 {
1751         struct llog_thread_info *lgi = llog_info(env);
1752         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1753         struct dt_object        *dir;
1754         char                    *name = lgi->lgi_name;
1755         int                     rc;
1756         ENTRY;
1757
1758         if (!fid_is_norm(fid))
1759                 RETURN(0);
1760
1761         dir = llog_osd_get_regular_fid_dir(env, dto);
1762         if (IS_ERR(dir))
1763                 RETURN(PTR_ERR(dir));
1764
1765         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1766         dt_write_lock(env, dir, 0);
1767         if (declare) {
1768                 rc = dt_declare_delete(env, dir, (struct dt_key *)name,
1769                                        th);
1770         } else {
1771                 rc = dt_delete(env, dir, (struct dt_key *)name, th);
1772         }
1773         dt_write_unlock(env, dir);
1774
1775         dt_object_put(env, dir);
1776         RETURN(rc);
1777 }
1778
1779 /**
1780  * Implementation of the llog_operations::lop_declare_destroy
1781  *
1782  * This function declare destroys the llog and deletes also entry in the
1783  * llog directory in case of named llog. Llog should be opened prior that.
1784  *
1785  * \param[in] env               execution environment
1786  * \param[in] loghandle llog handle of the current llog
1787  *
1788  * \retval              0 on successful destroy
1789  * \retval              negative value on error
1790  */
1791 static int llog_osd_declare_destroy(const struct lu_env *env,
1792                                     struct llog_handle *loghandle,
1793                                     struct thandle *th)
1794 {
1795         struct llog_ctxt        *ctxt;
1796         struct dt_object        *o, *llog_dir = NULL;
1797         int                      rc;
1798
1799         ENTRY;
1800
1801         ctxt = loghandle->lgh_ctxt;
1802         LASSERT(ctxt);
1803
1804         o = loghandle->lgh_obj;
1805         LASSERT(o);
1806
1807         if (loghandle->lgh_name) {
1808                 llog_dir = llog_osd_dir_get(env, ctxt);
1809                 if (IS_ERR(llog_dir))
1810                         RETURN(PTR_ERR(llog_dir));
1811
1812                 rc = dt_declare_delete(env, llog_dir,
1813                                        (struct dt_key *)loghandle->lgh_name,
1814                                        th);
1815                 if (rc < 0)
1816                         GOTO(out_put, rc);
1817         }
1818
1819         rc = dt_declare_ref_del(env, o, th);
1820         if (rc < 0)
1821                 GOTO(out_put, rc);
1822
1823         rc = dt_declare_destroy(env, o, th);
1824         if (rc < 0)
1825                 GOTO(out_put, rc);
1826
1827         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1828                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, true);
1829                 if (rc < 0)
1830                         GOTO(out_put, rc);
1831         }
1832
1833 out_put:
1834         if (!(IS_ERR_OR_NULL(llog_dir)))
1835                 dt_object_put(env, llog_dir);
1836
1837         RETURN(rc);
1838 }
1839
1840
1841 /**
1842  * Implementation of the llog_operations::lop_destroy
1843  *
1844  * This function destroys the llog and deletes also entry in the
1845  * llog directory in case of named llog. Llog should be opened prior that.
1846  * Destroy method is not part of external transaction and does everything
1847  * inside.
1848  *
1849  * \param[in] env               execution environment
1850  * \param[in] loghandle llog handle of the current llog
1851  *
1852  * \retval              0 on successful destroy
1853  * \retval              negative value on error
1854  */
1855 static int llog_osd_destroy(const struct lu_env *env,
1856                             struct llog_handle *loghandle, struct thandle *th)
1857 {
1858         struct llog_ctxt        *ctxt;
1859         struct dt_object        *o, *llog_dir = NULL;
1860         int                      rc;
1861
1862         ENTRY;
1863
1864         ctxt = loghandle->lgh_ctxt;
1865         LASSERT(ctxt != NULL);
1866
1867         o = loghandle->lgh_obj;
1868         LASSERT(o != NULL);
1869
1870         dt_write_lock(env, o, 0);
1871         if (!llog_osd_exist(loghandle))
1872                 GOTO(out_unlock, rc = 0);
1873
1874         if (loghandle->lgh_name) {
1875                 llog_dir = llog_osd_dir_get(env, ctxt);
1876                 if (IS_ERR(llog_dir))
1877                         GOTO(out_unlock, rc = PTR_ERR(llog_dir));
1878
1879                 dt_read_lock(env, llog_dir, 0);
1880                 rc = dt_delete(env, llog_dir,
1881                                (struct dt_key *)loghandle->lgh_name,
1882                                th);
1883                 dt_read_unlock(env, llog_dir);
1884                 if (rc) {
1885                         CERROR("%s: can't remove llog %s: rc = %d\n",
1886                                o->do_lu.lo_dev->ld_obd->obd_name,
1887                                loghandle->lgh_name, rc);
1888                         GOTO(out_unlock, rc);
1889                 }
1890         }
1891
1892         dt_ref_del(env, o, th);
1893         rc = dt_destroy(env, o, th);
1894         if (rc < 0)
1895                 GOTO(out_unlock, rc);
1896
1897         loghandle->lgh_destroyed = true;
1898         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1899                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, false);
1900                 if (rc < 0)
1901                         GOTO(out_unlock, rc);
1902         }
1903
1904 out_unlock:
1905         dt_write_unlock(env, o);
1906         if (!(IS_ERR_OR_NULL(llog_dir)))
1907                 dt_object_put(env, llog_dir);
1908         RETURN(rc);
1909 }
1910
1911 /**
1912  * Implementation of the llog_operations::lop_setup
1913  *
1914  * This function setup the llog on local storage.
1915  *
1916  * \param[in] env       execution environment
1917  * \param[in] obd       obd device the llog belongs to
1918  * \param[in] olg       the llog group, it is always zero group now.
1919  * \param[in] ctxt_idx  the llog index, it defines the purpose of this llog.
1920  *                      Every new llog type have to use own index.
1921  * \param[in] disk_obd  the storage obd, where llog is stored.
1922  *
1923  * \retval              0 on successful llog setup
1924  * \retval              negative value on error
1925  */
1926 static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
1927                           struct obd_llog_group *olg, int ctxt_idx,
1928                           struct obd_device *disk_obd)
1929 {
1930         struct llog_thread_info         *lgi = llog_info(env);
1931         struct llog_ctxt                *ctxt;
1932         int                              rc = 0;
1933         ENTRY;
1934
1935         LASSERT(obd);
1936         LASSERT(olg->olg_ctxts[ctxt_idx]);
1937
1938         ctxt = llog_ctxt_get(olg->olg_ctxts[ctxt_idx]);
1939         LASSERT(ctxt);
1940
1941         if (disk_obd == NULL)
1942                 GOTO(out, rc = 0);
1943
1944         /* initialize data allowing to generate new fids,
1945          * literally we need a sequece */
1946         lgi->lgi_fid.f_seq = FID_SEQ_LLOG;
1947         lgi->lgi_fid.f_oid = 1;
1948         lgi->lgi_fid.f_ver = 0;
1949         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1950                                     &lgi->lgi_fid,
1951                                     &ctxt->loc_los_nameless);
1952         if (rc != 0)
1953                 GOTO(out, rc);
1954
1955         lgi->lgi_fid.f_seq = FID_SEQ_LLOG_NAME;
1956         lgi->lgi_fid.f_oid = 1;
1957         lgi->lgi_fid.f_ver = 0;
1958         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1959                                     &lgi->lgi_fid,
1960                                     &ctxt->loc_los_named);
1961         if (rc != 0) {
1962                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1963                 ctxt->loc_los_nameless = NULL;
1964         }
1965
1966         GOTO(out, rc);
1967
1968 out:
1969         llog_ctxt_put(ctxt);
1970         return rc;
1971 }
1972
1973 /**
1974  * Implementation of the llog_operations::lop_cleanup
1975  *
1976  * This function cleanups the llog on local storage.
1977  *
1978  * \param[in] env       execution environment
1979  * \param[in] ctxt      the llog context
1980  *
1981  * \retval              0
1982  */
1983 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
1984 {
1985         if (ctxt->loc_los_nameless != NULL) {
1986                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1987                 ctxt->loc_los_nameless = NULL;
1988         }
1989
1990         if (ctxt->loc_los_named != NULL) {
1991                 local_oid_storage_fini(env, ctxt->loc_los_named);
1992                 ctxt->loc_los_named = NULL;
1993         }
1994
1995         return 0;
1996 }
1997
1998 const struct llog_operations llog_osd_ops = {
1999         .lop_next_block         = llog_osd_next_block,
2000         .lop_prev_block         = llog_osd_prev_block,
2001         .lop_read_header        = llog_osd_read_header,
2002         .lop_declare_destroy    = llog_osd_declare_destroy,
2003         .lop_destroy            = llog_osd_destroy,
2004         .lop_setup              = llog_osd_setup,
2005         .lop_cleanup            = llog_osd_cleanup,
2006         .lop_open               = llog_osd_open,
2007         .lop_exist              = llog_osd_exist,
2008         .lop_declare_create     = llog_osd_declare_create,
2009         .lop_create             = llog_osd_create,
2010         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2011         .lop_write_rec          = llog_osd_write_rec,
2012         .lop_close              = llog_osd_close,
2013 };
2014 EXPORT_SYMBOL(llog_osd_ops);
2015
2016 const struct llog_operations llog_common_cat_ops = {
2017         .lop_next_block         = llog_osd_next_block,
2018         .lop_prev_block         = llog_osd_prev_block,
2019         .lop_read_header        = llog_osd_read_header,
2020         .lop_declare_destroy    = llog_osd_declare_destroy,
2021         .lop_destroy            = llog_osd_destroy,
2022         .lop_setup              = llog_osd_setup,
2023         .lop_cleanup            = llog_osd_cleanup,
2024         .lop_open               = llog_osd_open,
2025         .lop_exist              = llog_osd_exist,
2026         .lop_declare_create     = llog_osd_declare_create,
2027         .lop_create             = llog_osd_create,
2028         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2029         .lop_write_rec          = llog_osd_write_rec,
2030         .lop_close              = llog_osd_close,
2031         .lop_add                = llog_cat_add_rec,
2032         .lop_declare_add        = llog_cat_declare_add_rec,
2033 };
2034 EXPORT_SYMBOL(llog_common_cat_ops);
2035
2036 /**
2037  * Read the special file which contains the list of llog catalogs IDs
2038  *
2039  * This function reads the CATALOGS file which contains the array of llog
2040  * catalogs IDs. The main purpose of this file is to store OSP llogs indexed
2041  * by OST/MDT number.
2042  *
2043  * \param[in]  env              execution environment
2044  * \param[in]  d                corresponding storage device
2045  * \param[in]  idx              position to start from, usually OST/MDT index
2046  * \param[in]  count            how many catalog IDs to read
2047  * \param[out] idarray          the buffer for the data. If it is NULL then
2048  *                              function returns just number of catalog IDs
2049  *                              in the file.
2050  * \param[in]  fid              LLOG_CATALOGS_OID for CATALOG object
2051  *
2052  * \retval                      0 on successful read of catalog IDs
2053  * \retval                      negative value on error
2054  * \retval                      positive value which is number of records in
2055  *                              the file if \a idarray is NULL
2056  */
2057 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
2058                           int idx, int count, struct llog_catid *idarray,
2059                           const struct lu_fid *fid)
2060 {
2061         struct llog_thread_info *lgi = llog_info(env);
2062         struct dt_object        *o = NULL;
2063         struct thandle          *th;
2064         int                      rc, size;
2065
2066         ENTRY;
2067
2068         LASSERT(d);
2069
2070         size = sizeof(*idarray) * count;
2071         lgi->lgi_off = idx *  sizeof(*idarray);
2072
2073         lgi->lgi_fid = *fid;
2074         o = dt_locate(env, d, &lgi->lgi_fid);
2075         if (IS_ERR(o))
2076                 RETURN(PTR_ERR(o));
2077
2078         if (!dt_object_exists(o)) {
2079                 th = dt_trans_create(env, d);
2080                 if (IS_ERR(th))
2081                         GOTO(out, rc = PTR_ERR(th));
2082
2083                 lgi->lgi_attr.la_valid = LA_MODE | LA_TYPE;
2084                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
2085                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
2086
2087                 th->th_wait_submit = 1;
2088                 /* Make the llog object creation synchronization, so
2089                  * it will be reliable to the reference, especially
2090                  * for remote reference */
2091                 th->th_sync = 1;
2092
2093                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
2094                                        &lgi->lgi_dof, th);
2095                 if (rc)
2096                         GOTO(out_trans, rc);
2097
2098                 rc = dt_trans_start_local(env, d, th);
2099                 if (rc)
2100                         GOTO(out_trans, rc);
2101
2102                 dt_write_lock(env, o, 0);
2103                 if (!dt_object_exists(o))
2104                         rc = dt_create(env, o, &lgi->lgi_attr, NULL,
2105                                        &lgi->lgi_dof, th);
2106                 dt_write_unlock(env, o);
2107 out_trans:
2108                 dt_trans_stop(env, d, th);
2109                 if (rc)
2110                         GOTO(out, rc);
2111         }
2112
2113         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2114         if (rc)
2115                 GOTO(out, rc);
2116
2117         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2118                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2119                        o->do_lu.lo_dev->ld_obd->obd_name,
2120                        lgi->lgi_attr.la_mode);
2121                 GOTO(out, rc = -ENOENT);
2122         }
2123
2124         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
2125                (int)lgi->lgi_attr.la_size, size);
2126
2127         /* return just number of llogs */
2128         if (idarray == NULL) {
2129                 rc = lgi->lgi_attr.la_size / sizeof(*idarray);
2130                 GOTO(out, rc);
2131         }
2132
2133         /* read for new ost index or for empty file */
2134         memset(idarray, 0, size);
2135         if (lgi->lgi_attr.la_size <= lgi->lgi_off)
2136                 GOTO(out, rc = 0);
2137         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
2138                 size = lgi->lgi_attr.la_size - lgi->lgi_off;
2139
2140         lgi->lgi_buf.lb_buf = idarray;
2141         lgi->lgi_buf.lb_len = size;
2142         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
2143         /* -EFAULT means the llog is a sparse file. This is not an error
2144          * after arbitrary OST index is supported. */
2145         if (rc < 0 && rc != -EFAULT) {
2146                 CERROR("%s: error reading CATALOGS: rc = %d\n",
2147                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
2148                 GOTO(out, rc);
2149         }
2150
2151         EXIT;
2152 out:
2153         dt_object_put(env, o);
2154         RETURN(rc);
2155 }
2156 EXPORT_SYMBOL(llog_osd_get_cat_list);
2157
2158 /**
2159  * Write the special file which contains the list of llog catalogs IDs
2160  *
2161  * This function writes the CATALOG file which contains the array of llog
2162  * catalogs IDs. It is used mostly to store OSP llogs indexed by OST/MDT
2163  * number.
2164  *
2165  * \param[in]  env      execution environment
2166  * \param[in]  d        corresponding storage device
2167  * \param[in]  idx      position to start from, usually OST/MDT index
2168  * \param[in]  count    how many catalog IDs to write
2169  * \param[out] idarray  the buffer with the data to write.
2170  * \param[in]  fid      LLOG_CATALOGS_OID for CATALOG object
2171  *
2172  * \retval              0 on successful write of catalog IDs
2173  * \retval              negative value on error
2174  */
2175 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
2176                           int idx, int count, struct llog_catid *idarray,
2177                           const struct lu_fid *fid)
2178 {
2179         struct llog_thread_info *lgi = llog_info(env);
2180         struct dt_object        *o = NULL;
2181         struct thandle          *th;
2182         int                      rc, size;
2183
2184         if (count == 0)
2185                 RETURN(0);
2186
2187         LASSERT(d);
2188
2189         size = sizeof(*idarray) * count;
2190         lgi->lgi_off = idx * sizeof(*idarray);
2191         lgi->lgi_fid = *fid;
2192
2193         o = dt_locate(env, d, &lgi->lgi_fid);
2194         if (IS_ERR(o))
2195                 RETURN(PTR_ERR(o));
2196
2197         if (!dt_object_exists(o))
2198                 GOTO(out, rc = -ENOENT);
2199
2200         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2201         if (rc)
2202                 GOTO(out, rc);
2203
2204         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2205                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2206                        o->do_lu.lo_dev->ld_obd->obd_name,
2207                        lgi->lgi_attr.la_mode);
2208                 GOTO(out, rc = -ENOENT);
2209         }
2210
2211         th = dt_trans_create(env, d);
2212         if (IS_ERR(th))
2213                 GOTO(out, rc = PTR_ERR(th));
2214
2215         lgi->lgi_buf.lb_len = size;
2216         lgi->lgi_buf.lb_buf = idarray;
2217         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, lgi->lgi_off, th);
2218         if (rc)
2219                 GOTO(out_trans, rc);
2220
2221         /* For update log, this happens during initialization,
2222          * see lod_sub_prep_llog(), and we need make sure catlog
2223          * file ID is written to catlist file(committed) before
2224          * cross-MDT operation write update records to catlog FILE,
2225          * otherwise, during failover these update records might
2226          * missing */
2227         if (fid_is_update_log(fid))
2228                 th->th_sync = 1;
2229
2230         rc = dt_trans_start_local(env, d, th);
2231         if (rc)
2232                 GOTO(out_trans, rc);
2233
2234         th->th_wait_submit = 1;
2235
2236         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
2237         if (rc)
2238                 CDEBUG(D_INODE, "can't write CATALOGS at index %d: rc = %d\n",
2239                        idx, rc);
2240 out_trans:
2241         dt_trans_stop(env, d, th);
2242 out:
2243         dt_object_put(env, o);
2244         RETURN(rc);
2245 }
2246 EXPORT_SYMBOL(llog_osd_put_cat_list);