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 = -EIO);
 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 < 0) {
 245                 CERROR("%s: can't read llog "DFID" header: rc = %d\n",
 246                        o->do_lu.lo_dev->ld_obd->obd_name,
 247                        PFID(lu_object_fid(&o->do_lu)), rc);
 248                 GOTO(unlock, rc = -EIO);
 249         }
 250         if (rc < sizeof(*llh_hdr) || rc < LLOG_MIN_CHUNK_SIZE) {
 251                 /* consider short header as non-initialized llog */
 252                 CERROR("%s: llog "DFID" header too small: rc = %d\n",
 253                        o->do_lu.lo_dev->ld_obd->obd_name,
 254                        PFID(lu_object_fid(&o->do_lu)), rc);
 255                 /* caller flags to be initialized */
 256                 handle->lgh_hdr->llh_flags = flags;
 257                 GOTO(unlock, rc = LLOG_EEMPTY);
 258         }
 259
 260         if (LLOG_REC_HDR_NEEDS_SWABBING(llh_hdr))
 261                 lustre_swab_llog_hdr(handle->lgh_hdr);
 262
 263         if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
 264                 CERROR("%s: bad log %s "DFID" header magic: %#x "
 265                        "(expected %#x)\n", o->do_lu.lo_dev->ld_obd->obd_name,
 266                        handle->lgh_name ? handle->lgh_name : "",
 267                        PFID(lu_object_fid(&o->do_lu)),
 268                        llh_hdr->lrh_type, LLOG_HDR_MAGIC);
 269                 GOTO(unlock, rc = -EINVAL);
 270         } else if (llh_hdr->lrh_len < LLOG_MIN_CHUNK_SIZE ||
 271                    llh_hdr->lrh_len > handle->lgh_hdr_size) {
 272                 CERROR("%s: incorrectly sized log %s "DFID" header: "
 273                        "%#x (expected at least %#x)\n"
 274                        "you may need to re-run lconf --write_conf.\n",
 275                        o->do_lu.lo_dev->ld_obd->obd_name,
 276                        handle->lgh_name ? handle->lgh_name : "",
 277                        PFID(lu_object_fid(&o->do_lu)),
 278                        llh_hdr->lrh_len, LLOG_MIN_CHUNK_SIZE);
 279                 GOTO(unlock, rc = -EINVAL);
 280         } else if (LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index >
 281                    LLOG_HDR_BITMAP_SIZE(handle->lgh_hdr) ||
 282                    LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_len !=
 283                         llh_hdr->lrh_len) {
 284                 CERROR("%s: incorrectly sized log %s "DFID" tailer: "
 285                        "%#x : rc = %d\n",
 286                        o->do_lu.lo_dev->ld_obd->obd_name,
 287                        handle->lgh_name ? handle->lgh_name : "",
 288                        PFID(lu_object_fid(&o->do_lu)),
 289                        LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_len, -EIO);
 290                 GOTO(unlock, rc = -EINVAL);
 291         }
 292
 293         handle->lgh_hdr->llh_flags |= (flags & LLOG_F_EXT_MASK);
 294         handle->lgh_last_idx = LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index;
 295         rc = 0;
 296
 297 unlock:
 298         dt_read_unlock(env, o);
 299         RETURN(rc);
 300 }
 301
 302 /**
 303  * Implementation of the llog_operations::lop_declare_write
 304  *
 305  * This function declares the new record write.
 306  *
 307  * \param[in] env       execution environment
 308  * \param[in] loghandle llog handle of the current llog
 309  * \param[in] rec       llog record header. This is a real header of the full
 310  *                      llog record to write. This is the beginning of buffer
 311  *                      to write, the length of buffer is stored in
 312  *                      \a rec::lrh_len
 313  * \param[in] idx       index of the llog record. If \a idx == -1 then this is
 314  *                      append case, otherwise \a idx is the index of record
 315  *                      to modify
 316  * \param[in] th        current transaction handle
 317  *
 318  * \retval              0 on successful declaration
 319  * \retval              negative error if declaration failed
 320  */
 321 static int llog_osd_declare_write_rec(const struct lu_env *env,
 322                                       struct llog_handle *loghandle,
 323                                       struct llog_rec_hdr *rec,
 324                                       int idx, struct thandle *th)
 325 {
 326         struct llog_thread_info *lgi = llog_info(env);
 327         __u32                   chunk_size;
 328         struct dt_object        *o;
 329         int                      rc;
 330
 331         ENTRY;
 332
 333         LASSERT(env);
 334         LASSERT(th);
 335         LASSERT(loghandle);
 336         LASSERT(rec);
 337         LASSERT(rec->lrh_len <= loghandle->lgh_ctxt->loc_chunk_size);
 338
 339         o = loghandle->lgh_obj;
 340         LASSERT(o);
 341
 342         chunk_size = loghandle->lgh_ctxt->loc_chunk_size;
 343         lgi->lgi_buf.lb_len = chunk_size;
 344         lgi->lgi_buf.lb_buf = NULL;
 345         /* each time we update header */
 346         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, 0, th);
 347         if (rc || idx == 0) /* if error or just header */
 348                 RETURN(rc);
 349
 350         /**
 351          * the pad record can be inserted so take into account double
 352          * record size: pad and the actual record into a new block
 353          */
 354         lgi->lgi_buf.lb_len = rec->lrh_len * 2;
 355         lgi->lgi_buf.lb_buf = NULL;
 356         /* XXX: implement declared window or multi-chunks approach */
 357         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, -1, th);
 358
 359         RETURN(rc);
 360 }
 361
 362 /**
 363  * Implementation of the llog_operations::lop_write
 364  *
 365  * This function writes the new record in the llog or modify the existed one.
 366  *
 367  * \param[in]  env              execution environment
 368  * \param[in]  loghandle        llog handle of the current llog
 369  * \param[in]  rec              llog record header. This is a real header of
 370  *                              the full llog record to write. This is
 371  *                              the beginning of buffer to write, the length
 372  *                              of buffer is stored in \a rec::lrh_len
 373  * \param[in,out] reccookie     pointer to the cookie to return back if needed.
 374  *                              It is used for further cancel of this llog
 375  *                              record.
 376  * \param[in]  idx              index of the llog record. If \a idx == -1 then
 377  *                              this is append case, otherwise \a idx is
 378  *                              the index of record to modify
 379  * \param[in]  th               current transaction handle
 380  *
 381  * \retval                      0 on successful write && \a reccookie == NULL
 382  *                              1 on successful write && \a reccookie != NULL
 383  * \retval                      negative error if write failed
 384  */
 385 static int llog_osd_write_rec(const struct lu_env *env,
 386                               struct llog_handle *loghandle,
 387                               struct llog_rec_hdr *rec,
 388                               struct llog_cookie *reccookie,
 389                               int idx, struct thandle *th)
 390 {
 391         struct llog_thread_info *lgi = llog_info(env);
 392         struct llog_log_hdr *llh;
 393         int reclen = rec->lrh_len;
 394         int index, rc;
 395         struct llog_rec_tail *lrt;
 396         struct dt_object *o;
 397         __u32 chunk_size;
 398         size_t left;
 399         __u32 orig_last_idx;
 400         bool pad = false;
 401         ENTRY;
 402
 403         llh = loghandle->lgh_hdr;
 404         o = loghandle->lgh_obj;
 405
 406         chunk_size = llh->llh_hdr.lrh_len;
 407         CDEBUG(D_OTHER, "new record %x to "DFID"\n",
 408                rec->lrh_type, PFID(lu_object_fid(&o->do_lu)));
 409
 410         if (!llog_osd_exist(loghandle))
 411                 RETURN(-ENOENT);
 412
 413         /* record length should not bigger than  */
 414         if (reclen > loghandle->lgh_hdr->llh_hdr.lrh_len)
 415                 RETURN(-E2BIG);
 416
 417         /* sanity check for fixed-records llog */
 418         if (idx != LLOG_HEADER_IDX && (llh->llh_flags & LLOG_F_IS_FIXSIZE)) {
 419                 LASSERT(llh->llh_size != 0);
 420                 LASSERT(llh->llh_size == reclen);
 421         }
 422
 423         /* return error if osp object is stale */
 424         if (idx != LLOG_HEADER_IDX && dt_object_stale(o))
 425                 RETURN(-ESTALE);
 426         rc = dt_attr_get(env, o, &lgi->lgi_attr);
 427         if (rc)
 428                 RETURN(rc);
 429
 430         /**
 431          * The modification case.
 432          * If idx set then the record with that index must be modified.
 433          * There are three cases possible:
 434          * 1) the common case is the llog header update (idx == 0)
 435          * 2) the llog record modification during llog process.
 436          *    This is indicated by the \a loghandle::lgh_cur_idx > 0.
 437          *    In that case the \a loghandle::lgh_cur_offset
 438          * 3) otherwise this is assumed that llog consist of records of
 439          *    fixed size, i.e. catalog. The llog header must has llh_size
 440          *    field equal to record size. The record offset is calculated
 441          *    just by /a idx value
 442          *
 443          * During modification we don't need extra header update because
 444          * the bitmap and record count are not changed. The record header
 445          * and tail remains the same too.
 446          */
 447         if (idx != LLOG_NEXT_IDX) {
 448                 /* llog can be empty only when first record is being written */
 449                 LASSERT(ergo(idx > 0, lgi->lgi_attr.la_size > 0));
 450
 451                 if (!test_bit_le(idx, LLOG_HDR_BITMAP(llh))) {
 452                         CERROR("%s: modify unset record %u\n",
 453                                o->do_lu.lo_dev->ld_obd->obd_name, idx);
 454                         RETURN(-ENOENT);
 455                 }
 456
 457                 if (idx != rec->lrh_index) {
 458                         CERROR("%s: modify index mismatch %d %u\n",
 459                                o->do_lu.lo_dev->ld_obd->obd_name, idx,
 460                                rec->lrh_index);
 461                         RETURN(-EFAULT);
 462                 }
 463
 464                 if (idx == LLOG_HEADER_IDX) {
 465                         /* llog header update */
 466                         __u32   *bitmap = LLOG_HDR_BITMAP(llh);
 467
 468                         lgi->lgi_off = 0;
 469
 470                         /* If it does not indicate the bitmap index
 471                          * (reccookie == NULL), then it means update
 472                          * the whole update header. Otherwise only
 473                          * update header and bits needs to be updated,
 474                          * and in DNE cases, it will signaficantly
 475                          * shrink the RPC size.
 476                          * see distribute_txn_cancel_records()*/
 477                         if (reccookie == NULL) {
 478                                 lgi->lgi_buf.lb_len = reclen;
 479                                 lgi->lgi_buf.lb_buf = rec;
 480                                 rc = dt_record_write(env, o, &lgi->lgi_buf,
 481                                                      &lgi->lgi_off, th);
 482                                 RETURN(rc);
 483                         }
 484
 485                         /* update the header */
 486                         lgi->lgi_buf.lb_len = llh->llh_bitmap_offset;
 487                         lgi->lgi_buf.lb_buf = llh;
 488                         rc = dt_record_write(env, o, &lgi->lgi_buf,
 489                                              &lgi->lgi_off, th);
 490                         if (rc != 0)
 491                                 RETURN(rc);
 492
 493                         /* update the bitmap */
 494                         index = reccookie->lgc_index;
 495                         lgi->lgi_off = llh->llh_bitmap_offset +
 496                                       (index / (sizeof(*bitmap) * 8)) *
 497                                                         sizeof(*bitmap);
 498                         lgi->lgi_buf.lb_len = sizeof(*bitmap);
 499                         lgi->lgi_buf.lb_buf =
 500                                         &bitmap[index/(sizeof(*bitmap)*8)];
 501                         rc = dt_record_write(env, o, &lgi->lgi_buf,
 502                                              &lgi->lgi_off, th);
 503
 504                         RETURN(rc);
 505                 } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
 506                         lgi->lgi_off = llh->llh_hdr.lrh_len +
 507                                        (idx - 1) * reclen;
 508                 } else if (reccookie != NULL && reccookie->lgc_index > 0) {
 509                         /**
 510                          * The lgc_offset can be used only if index is
 511                          * the same.
 512                          */
 513                         if (idx != reccookie->lgc_index) {
 514                                 CERROR("%s: modify index mismatch %d %d\n",
 515                                        o->do_lu.lo_dev->ld_obd->obd_name, idx,
 516                                        reccookie->lgc_index);
 517                                 RETURN(-EFAULT);
 518                         }
 519
 520                         lgi->lgi_off = reccookie->lgc_offset;
 521                         CDEBUG(D_OTHER, "modify record "DFID": idx:%u, "
 522                                "len:%u offset %llu\n",
 523                                PLOGID(&loghandle->lgh_id), idx,
 524                                rec->lrh_len, (long long)lgi->lgi_off);
 525                 } else {
 526                         /* This can be result of lgh_cur_idx is not set during
 527                          * llog processing or llh_size is not set to proper
 528                          * record size for fixed records llog. Therefore it is
 529                          * impossible to get record offset. */
 530                         CERROR("%s: can't get record offset, idx:%d, "
 531                                "len:%u.\n", o->do_lu.lo_dev->ld_obd->obd_name,
 532                                idx, rec->lrh_len);
 533                         RETURN(-EFAULT);
 534                 }
 535
 536                 /* update only data, header and tail remain the same */
 537                 lgi->lgi_off += sizeof(struct llog_rec_hdr);
 538                 lgi->lgi_buf.lb_len = REC_DATA_LEN(rec);
 539                 lgi->lgi_buf.lb_buf = REC_DATA(rec);
 540                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
 541                 if (rc == 0 && reccookie) {
 542                         reccookie->lgc_lgl = loghandle->lgh_id;
 543                         reccookie->lgc_index = idx;
 544                         rc = 1;
 545                 }
 546                 RETURN(rc);
 547         }
 548
 549         /**
 550          * The append case.
 551          * The most common case of using llog. The new index is assigned to
 552          * the new record, new bit is set in llog bitmap and llog count is
 553          * incremented.
 554          *
 555          * Make sure that records don't cross a chunk boundary, so we can
 556          * process them page-at-a-time if needed.  If it will cross a chunk
 557          * boundary, write in a fake (but referenced) entry to pad the chunk.
 558          */
 559
 560
 561         /* simulate ENOSPC when new plain llog is being added to the
 562          * catalog */
 563         if (CFS_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED2) &&
 564             llh->llh_flags & LLOG_F_IS_CAT)
 565                 RETURN(-ENOSPC);
 566
 567         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
 568         orig_last_idx = loghandle->lgh_last_idx;
 569         lgi->lgi_off = lgi->lgi_attr.la_size;
 570
 571         if (loghandle->lgh_max_size > 0 &&
 572             lgi->lgi_off >= loghandle->lgh_max_size) {
 573                 CDEBUG(D_OTHER, "llog is getting too large (%u > %u) at %u "
 574                        DFID"\n", (unsigned)lgi->lgi_off,
 575                        loghandle->lgh_max_size, (int)loghandle->lgh_last_idx,
 576                        PLOGID(&loghandle->lgh_id));
 577                 /* this is to signal that this llog is full */
 578                 loghandle->lgh_last_idx = llog_max_idx(llh);
 579                 RETURN(-ENOSPC);
 580         }
 581
 582         left = chunk_size - (lgi->lgi_off & (chunk_size - 1));
 583         /* NOTE: padding is a record, but no bit is set */
 584         if (left != 0 && left != reclen &&
 585             left < (reclen + LLOG_MIN_REC_SIZE)) {
 586                 index = loghandle->lgh_last_idx + 1;
 587                 rc = llog_osd_pad(env, o, &lgi->lgi_off, left, index, th);
 588                 if (rc)
 589                         RETURN(rc);
 590
 591                 loghandle->lgh_last_idx++; /* for pad rec */
 592                 pad = true;
 593         }
 594         /* if it's the last idx in log file, then return -ENOSPC
 595          * or wrap around if a catalog */
 596         if (llog_is_full(loghandle)) {
 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 (CFS_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 (CFS_FAIL_PRECHECK(OBD_FAIL_LLOG_PAUSE_AFTER_PAD) && pad) {
 692                 /* a window for concurrent llog reader, see LU-12577 */
 693                 CFS_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 (CFS_FAIL_PRECHECK(OBD_FAIL_LLOG_PROCESS_TIMEOUT) &&
 706            cfs_fail_val == (unsigned int)(loghandle->lgh_id.lgl_oi.oi.oi_id &
 707                                           0xFFFFFFFF)) {
 708                 CFS_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_max_idx(llh);
 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 = !next_idx;
 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 (CFS_FAIL_PRECHECK(OBD_FAIL_MDS_CHANGELOG_DEL) &&
 935             cfs_fail_val == ((unsigned long)loghandle & 0xFFFFFFFF)) {
 936                 CFS_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) {
 973                                 /* no goal is valid case */
 974                                 if (!next_idx)
 975                                         GOTO(out, rc);
 976                                 if (!force_mini_rec)
 977                                         goto retry;
 978                         }
 979                         CERROR("%s: can't read llog block from log "DFID
 980                                " offset %llu: rc = %d\n",
 981                                o->do_lu.lo_dev->ld_obd->obd_name,
 982                                PFID(lu_object_fid(&o->do_lu)), *cur_offset,
 983                                rc);
 984                         GOTO(out, rc);
 985                 }
 986
 987                 if (rc < len) {
 988                         /* signal the end of the valid buffer to
 989                          * llog_process */
 990                         memset(buf + rc, 0, len - rc);
 991                 }
 992
 993                 if (rc == 0) { /* end of file, nothing to do */
 994                         if (!force_mini_rec)
 995                                 goto retry;
 996                         GOTO(out, rc);
 997                 }
 998
 999                 if (rc < sizeof(*tail)) {
1000                         if (!force_mini_rec)
1001                                 goto retry;
1002
1003                         CERROR("%s: invalid llog block at log id "DFID" offset %llu\n",
1004                                o->do_lu.lo_dev->ld_obd->obd_name,
1005                                PLOGID(&loghandle->lgh_id), *cur_offset);
1006                         GOTO(out, rc = -EINVAL);
1007                 }
1008
1009                 rec = buf;
1010                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1011                         lustre_swab_llog_rec(rec);
1012                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1013                                                 sizeof(struct llog_rec_tail));
1014
1015                 /* caller handles bad records if any */
1016                 if (llog_verify_record(loghandle, rec))
1017                         GOTO(out, rc = 0);
1018
1019                 /* get the last record in block */
1020                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1021                                                    tail->lrt_len);
1022
1023                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1024                         lustre_swab_llog_rec(last_rec);
1025
1026                 if (last_rec->lrh_index != tail->lrt_index) {
1027                         CERROR("%s: invalid llog tail at log id "DFID" offset %llu last_rec idx %u tail idx %u lrt len %u read_size %d\n",
1028                                o->do_lu.lo_dev->ld_obd->obd_name,
1029                                PLOGID(&loghandle->lgh_id), *cur_offset,
1030                                last_rec->lrh_index, tail->lrt_index,
1031                                tail->lrt_len, rc);
1032                         GOTO(out, rc = -EINVAL);
1033                 }
1034
1035                 *cur_idx = tail->lrt_index;
1036
1037                 /* this shouldn't happen */
1038                 if (tail->lrt_index == 0) {
1039                         CERROR("%s: invalid llog tail at log id "DFID"offset %llu bytes %d\n",
1040                                o->do_lu.lo_dev->ld_obd->obd_name,
1041                                PLOGID(&loghandle->lgh_id), *cur_offset, rc);
1042                         GOTO(out, rc = -EINVAL);
1043                 }
1044                 if (tail->lrt_index < next_idx) {
1045                         last_idx = *cur_idx;
1046                         last_offset = *cur_offset;
1047                         continue;
1048                 }
1049
1050                 /* sanity check that the start of the new buffer is no farther
1051                  * than the record that we wanted.  This shouldn't happen. */
1052                 if (next_idx && rec->lrh_index > next_idx) {
1053                         if (!force_mini_rec && next_idx > last_idx)
1054                                 goto retry;
1055
1056                         CERROR("%s: missed desired record? %u > %u\n",
1057                                o->do_lu.lo_dev->ld_obd->obd_name,
1058                                rec->lrh_index, next_idx);
1059                         GOTO(out, rc = -ENOENT);
1060                 }
1061
1062                 /* Trim unsupported extensions for compat w/ older clients */
1063                 changelog_block_trim_ext(rec, last_rec, loghandle);
1064
1065                 GOTO(out, rc = 0);
1066
1067 retry:
1068                 /* Note: because there are some pad records in the
1069                  * llog, so llog_skip_over() might skip too much
1070                  * records, let's try skip again with minimum record */
1071                 force_mini_rec = true;
1072                 *cur_offset = last_offset;
1073                 *cur_idx = last_idx;
1074         }
1075         /* being here means we reach end of llog but didn't find needed idx
1076          * normally could happen while processing remote llog, return -EBADR
1077          * to indicate access beyond end of file like dt_read() does and to
1078          * distunguish this situation from real IO or network issues.
1079          */
1080         GOTO(out, rc = -EBADR);
1081 out:
1082         dt_read_unlock(env, o);
1083         return rc;
1084 }
1085
1086 /**
1087  * Implementation of the llog_operations::lop_prev_block
1088  *
1089  * This function finds the llog block to return which contains
1090  * record with required index but in reverse order - from end of llog
1091  * to the beginning.
1092  * It is main part of reverse llog processing.
1093  *
1094  * \param[in] env       execution environment
1095  * \param[in] loghandle llog handle of the current llog
1096  * \param[in] prev_idx  target index to find
1097  * \param[in] buf       pointer to data buffer to fill
1098  * \param[in] len       required len to read, it is llog_chunk_size usually.
1099  *
1100  * \retval              0 on successful buffer read
1101  * \retval              negative value on error
1102  */
1103 static int llog_osd_prev_block(const struct lu_env *env,
1104                                struct llog_handle *loghandle,
1105                                int prev_idx, void *buf, int len)
1106 {
1107         struct llog_thread_info *lgi = llog_info(env);
1108         struct dt_object        *o;
1109         struct dt_device        *dt;
1110         loff_t                   cur_offset;
1111         __u32                   chunk_size;
1112         int                      rc;
1113
1114         ENTRY;
1115
1116         chunk_size = loghandle->lgh_hdr->llh_hdr.lrh_len;
1117         if (len == 0 || len & (chunk_size - 1))
1118                 RETURN(-EINVAL);
1119
1120         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
1121
1122         LASSERT(loghandle);
1123         LASSERT(loghandle->lgh_ctxt);
1124
1125         o = loghandle->lgh_obj;
1126         LASSERT(o);
1127         dt_read_lock(env, o, 0);
1128         if (!llog_osd_exist(loghandle))
1129                 GOTO(out, rc = -ESTALE);
1130
1131         dt = lu2dt_dev(o->do_lu.lo_dev);
1132         LASSERT(dt);
1133
1134         /* Let's only use mini record size for previous block read
1135          * for now XXX */
1136         cur_offset = chunk_size;
1137         llog_skip_over(loghandle, &cur_offset, 0, prev_idx,
1138                        chunk_size, true);
1139
1140         rc = dt_attr_get(env, o, &lgi->lgi_attr);
1141         if (rc)
1142                 GOTO(out, rc);
1143
1144         while (cur_offset < lgi->lgi_attr.la_size) {
1145                 struct llog_rec_hdr     *rec, *last_rec;
1146                 struct llog_rec_tail    *tail;
1147
1148                 lgi->lgi_buf.lb_len = len;
1149                 lgi->lgi_buf.lb_buf = buf;
1150                 rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
1151                 if (rc < 0) {
1152                         CERROR("%s: can't read llog block from log "DFID
1153                                " offset %llu: rc = %d\n",
1154                                o->do_lu.lo_dev->ld_obd->obd_name,
1155                                PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
1156                         GOTO(out, rc);
1157                 }
1158
1159                 if (rc == 0) /* end of file, nothing to do */
1160                         GOTO(out, rc);
1161
1162                 if (rc < sizeof(*tail)) {
1163                         CERROR("%s: invalid llog block at log id "DFID" offset %llu\n",
1164                                o->do_lu.lo_dev->ld_obd->obd_name,
1165                                PLOGID(&loghandle->lgh_id), cur_offset);
1166                         GOTO(out, rc = -EINVAL);
1167                 }
1168
1169                 rec = buf;
1170                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1171                         lustre_swab_llog_rec(rec);
1172
1173                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1174                                                 sizeof(struct llog_rec_tail));
1175                 /* get the last record in block */
1176                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1177                                                    le32_to_cpu(tail->lrt_len));
1178
1179                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1180                         lustre_swab_llog_rec(last_rec);
1181                 LASSERT(last_rec->lrh_index == tail->lrt_index);
1182
1183                 /* this shouldn't happen */
1184                 if (tail->lrt_index == 0) {
1185                         CERROR("%s: invalid llog tail at log id "DFID" offset %llu\n",
1186                                o->do_lu.lo_dev->ld_obd->obd_name,
1187                                PLOGID(&loghandle->lgh_id), cur_offset);
1188                         GOTO(out, rc = -EINVAL);
1189                 }
1190                 if (tail->lrt_index < prev_idx)
1191                         continue;
1192
1193                 /* sanity check that the start of the new buffer is no farther
1194                  * than the record that we wanted.  This shouldn't happen. */
1195                 if (rec->lrh_index > prev_idx) {
1196                         CERROR("%s: missed desired record? %u > %u\n",
1197                                o->do_lu.lo_dev->ld_obd->obd_name,
1198                                rec->lrh_index, prev_idx);
1199                         GOTO(out, rc = -ENOENT);
1200                 }
1201
1202                 /* Trim unsupported extensions for compat w/ older clients */
1203                 changelog_block_trim_ext(rec, last_rec, loghandle);
1204
1205                 GOTO(out, rc = 0);
1206         }
1207         GOTO(out, rc = -EIO);
1208 out:
1209         dt_read_unlock(env, o);
1210         return rc;
1211 }
1212
1213 /**
1214  * This is helper function to get llog directory object. It is used by named
1215  * llog operations to find/insert/delete llog entry from llog directory.
1216  *
1217  * \param[in] env       execution environment
1218  * \param[in] ctxt      llog context
1219  *
1220  * \retval              dt_object of llog directory
1221  * \retval              ERR_PTR of negative value on error
1222  */
1223 static struct dt_object *llog_osd_dir_get(const struct lu_env *env,
1224                                           struct llog_ctxt *ctxt)
1225 {
1226         struct dt_device        *dt;
1227         struct dt_thread_info   *dti = dt_info(env);
1228         struct dt_object        *dir;
1229         int                      rc;
1230
1231         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1232         if (ctxt->loc_dir == NULL) {
1233                 rc = dt_root_get(env, dt, &dti->dti_fid);
1234                 if (rc)
1235                         return ERR_PTR(rc);
1236                 dir = dt_locate(env, dt, &dti->dti_fid);
1237
1238                 if (!IS_ERR(dir) && !dt_try_as_dir(env, dir, true)) {
1239                         dt_object_put(env, dir);
1240                         return ERR_PTR(-ENOTDIR);
1241                 }
1242         } else {
1243                 lu_object_get(&ctxt->loc_dir->do_lu);
1244                 dir = ctxt->loc_dir;
1245         }
1246
1247         return dir;
1248 }
1249
1250 /**
1251  * Implementation of the llog_operations::lop_open
1252  *
1253  * This function opens the llog by its logid or by name, it may open also
1254  * non existent llog and assing then new id to it.
1255  * The llog_open/llog_close pair works similar to lu_object_find/put,
1256  * the object may not exist prior open. The result of open is just dt_object
1257  * in the llog header.
1258  *
1259  * \param[in] env               execution environment
1260  * \param[in] handle            llog handle of the current llog
1261  * \param[in] logid             logid of llog to open (nameless llog)
1262  * \param[in] name              name of llog to open (named llog)
1263  * \param[in] open_param
1264  *                              LLOG_OPEN_NEW - new llog, may not exist
1265  *                              LLOG_OPEN_EXIST - old llog, must exist
1266  *
1267  * \retval                      0 on successful open, llog_handle::lgh_obj
1268  *                              contains the dt_object of the llog.
1269  * \retval                      negative value on error
1270  */
1271 static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
1272                          struct llog_logid *logid, char *name,
1273                          enum llog_open_param open_param)
1274 {
1275         struct llog_thread_info         *lgi = llog_info(env);
1276         struct llog_ctxt                *ctxt = handle->lgh_ctxt;
1277         struct dt_object                *o;
1278         struct dt_device                *dt;
1279         struct ls_device                *ls;
1280         struct local_oid_storage        *los = NULL;
1281         int                              rc = 0;
1282         bool new_id = false;
1283
1284         ENTRY;
1285
1286         LASSERT(env);
1287         LASSERT(ctxt);
1288         LASSERT(ctxt->loc_exp);
1289         LASSERT(ctxt->loc_exp->exp_obd);
1290         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1291         LASSERT(dt);
1292         if (ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1293                 struct lu_object_conf conf = { 0 };
1294                 if (logid != NULL) {
1295                         logid_to_fid(logid, &lgi->lgi_fid);
1296                 } else {
1297                         /* If logid == NULL, then it means the caller needs
1298                          * to allocate new FID (llog_cat_declare_add_rec()). */
1299                         rc = dt_fid_alloc(env, dt, &lgi->lgi_fid, NULL, NULL);
1300                         if (rc < 0)
1301                                 RETURN(rc);
1302                         rc = 0;
1303                         conf.loc_flags = LOC_F_NEW;
1304                 }
1305
1306                 o = dt_locate_at(env, dt, &lgi->lgi_fid,
1307                                  dt->dd_lu_dev.ld_site->ls_top_dev, &conf);
1308                 if (IS_ERR(o))
1309                         RETURN(PTR_ERR(o));
1310
1311                 goto after_open;
1312         }
1313
1314         ls = ls_device_find_or_init(dt);
1315         if (IS_ERR(ls))
1316                 RETURN(PTR_ERR(ls));
1317
1318         mutex_lock(&ls->ls_los_mutex);
1319         los = dt_los_find(ls, name != NULL ? FID_SEQ_LLOG_NAME : FID_SEQ_LLOG);
1320         mutex_unlock(&ls->ls_los_mutex);
1321         LASSERT(los);
1322         ls_device_put(env, ls);
1323
1324         LASSERT(handle);
1325
1326         if (logid != NULL) {
1327                 logid_to_fid(logid, &lgi->lgi_fid);
1328         } else if (name) {
1329                 struct dt_object *llog_dir;
1330
1331                 llog_dir = llog_osd_dir_get(env, ctxt);
1332                 if (IS_ERR(llog_dir))
1333                         GOTO(out, rc = PTR_ERR(llog_dir));
1334                 dt_read_lock(env, llog_dir, 0);
1335                 rc = dt_lookup_dir(env, llog_dir, name, &lgi->lgi_fid);
1336                 dt_read_unlock(env, llog_dir);
1337                 dt_object_put(env, llog_dir);
1338                 if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
1339                         /* generate fid for new llog */
1340                         rc = local_object_fid_generate(env, los,
1341                                                        &lgi->lgi_fid);
1342                         new_id = true;
1343                 }
1344                 if (rc < 0)
1345                         GOTO(out, rc);
1346                 OBD_ALLOC(handle->lgh_name, strlen(name) + 1);
1347                 if (handle->lgh_name)
1348                         strcpy(handle->lgh_name, name);
1349                 else
1350                         GOTO(out, rc = -ENOMEM);
1351         } else {
1352                 LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
1353                 /* generate fid for new llog */
1354 generate:
1355                 rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
1356                 if (rc < 0)
1357                         GOTO(out, rc);
1358                 new_id = true;
1359         }
1360         if (CFS_FAIL_PRECHECK(OBD_FAIL_MDS_LLOG_UMOUNT_RACE) &&
1361             cfs_fail_val == 1) {
1362                 cfs_fail_val = 2;
1363                 CFS_RACE(OBD_FAIL_MDS_LLOG_UMOUNT_RACE);
1364                 msleep(MSEC_PER_SEC);
1365         }
1366         o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
1367         if (IS_ERR(o))
1368                 GOTO(out_name, rc = PTR_ERR(o));
1369
1370         if (dt_object_exists(o) && new_id) {
1371                 /* llog exists with just generated ID, e.g. some old llog file
1372                  * still is in use or is orphan, drop a warn and skip it. */
1373                 CDEBUG(D_INFO, "%s: llog exists with the same FID: "DFID
1374                        ", skipping\n",
1375                        o->do_lu.lo_dev->ld_obd->obd_name,
1376                        PFID(lu_object_fid(&o->do_lu)));
1377                 dt_object_put(env, o);
1378                 /* just skip this llog ID, we shouldn't delete it because we
1379                  * don't know exactly what is its purpose and state. */
1380                 goto generate;
1381         }
1382
1383 after_open:
1384         /* No new llog is expected but doesn't exist */
1385         if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o)) {
1386                 CDEBUG(D_INFO, "%s: llog FID: "DFID" obj %p doesn`t exist\n",
1387                        o->do_lu.lo_dev->ld_obd->obd_name,
1388                        PFID(lu_object_fid(&o->do_lu)), o);
1389                 GOTO(out_put, rc = -ENOENT);
1390         }
1391         fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
1392         handle->lgh_obj = o;
1393         handle->private_data = los;
1394         LASSERT(handle->lgh_ctxt);
1395
1396         RETURN(rc);
1397
1398 out_put:
1399         dt_object_put(env, o);
1400 out_name:
1401         if (handle->lgh_name != NULL)
1402                 OBD_FREE(handle->lgh_name, strlen(name) + 1);
1403 out:
1404         if (los != NULL)
1405                 dt_los_put(los);
1406         RETURN(rc);
1407 }
1408
1409 /**
1410  * Get dir for regular fid log object
1411  *
1412  * Get directory for regular fid log object, and these regular fid log
1413  * object will be inserted under this directory, to satisfy the FS
1414  * consistency check, e2fsck etc.
1415  *
1416  * \param [in] env      execution environment
1417  * \param [in] dto      llog object
1418  *
1419  * \retval              pointer to the directory if it is found.
1420  * \retval              ERR_PTR(negative errno) if it fails.
1421  */
1422 static struct dt_object *llog_osd_get_regular_fid_dir(const struct lu_env *env,
1423                                                       struct dt_object *dto)
1424 {
1425         struct llog_thread_info *lgi = llog_info(env);
1426         struct seq_server_site *ss = dto->do_lu.lo_dev->ld_site->ld_seq_site;
1427         struct lu_seq_range     *range = &lgi->lgi_range;
1428         struct lu_fid           *dir_fid = &lgi->lgi_fid;
1429         struct dt_object        *dir;
1430         int                     rc;
1431         ENTRY;
1432
1433         fld_range_set_any(range);
1434         LASSERT(ss != NULL);
1435         rc = ss->ss_server_fld->lsf_seq_lookup(env, ss->ss_server_fld,
1436                                    fid_seq(lu_object_fid(&dto->do_lu)), range);
1437         if (rc < 0)
1438                 RETURN(ERR_PTR(rc));
1439
1440         lu_update_log_dir_fid(dir_fid, range->lsr_index);
1441         dir = dt_locate(env, lu2dt_dev(dto->do_lu.lo_dev), dir_fid);
1442         if (IS_ERR(dir))
1443                 RETURN(dir);
1444
1445         if (!dt_try_as_dir(env, dir, true)) {
1446                 dt_object_put(env, dir);
1447                 RETURN(ERR_PTR(-ENOTDIR));
1448         }
1449
1450         RETURN(dir);
1451 }
1452
1453 /**
1454  * Add llog object with regular FID to name entry
1455  *
1456  * Add llog object with regular FID to name space, and each llog
1457  * object on each MDT will be /update_log_dir/[seq:oid:ver],
1458  * so to satisfy the namespace consistency check, e2fsck etc.
1459  *
1460  * \param [in] env      execution environment
1461  * \param [in] dto      llog object
1462  * \param [in] th       thandle
1463  * \param [in] declare  if it is declare or execution
1464  *
1465  * \retval              0 if insertion succeeds.
1466  * \retval              negative errno if insertion fails.
1467  */
1468 static int
1469 llog_osd_regular_fid_add_name_entry(const struct lu_env *env,
1470                                     struct dt_object *dto,
1471                                     struct thandle *th, bool declare)
1472 {
1473         struct llog_thread_info *lgi = llog_info(env);
1474         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1475         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1476         struct dt_object        *dir;
1477         char                    *name = lgi->lgi_name;
1478         int                     rc;
1479         ENTRY;
1480
1481         if (!fid_is_norm(fid))
1482                 RETURN(0);
1483
1484         dir = llog_osd_get_regular_fid_dir(env, dto);
1485         if (IS_ERR(dir))
1486                 RETURN(PTR_ERR(dir));
1487
1488         rec->rec_fid = fid;
1489         rec->rec_type = S_IFREG;
1490         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1491         dt_write_lock(env, dir, 0);
1492         if (declare) {
1493                 rc = dt_declare_insert(env, dir, (struct dt_rec *)rec,
1494                                (struct dt_key *)name, th);
1495         } else {
1496                 rc = dt_insert(env, dir, (struct dt_rec *)rec,
1497                                (struct dt_key *)name, th);
1498         }
1499         dt_write_unlock(env, dir);
1500
1501         dt_object_put(env, dir);
1502         RETURN(rc);
1503 }
1504
1505
1506 /**
1507  * Implementation of the llog_operations::lop_declare_create
1508  *
1509  * This function declares the llog create. It declares also name insert
1510  * into llog directory in case of named llog.
1511  *
1512  * \param[in] env       execution environment
1513  * \param[in] res       llog handle of the current llog
1514  * \param[in] th        current transaction handle
1515  *
1516  * \retval              0 on successful create declaration
1517  * \retval              negative value on error
1518  */
1519 static int llog_osd_declare_create(const struct lu_env *env,
1520                                    struct llog_handle *res, struct thandle *th)
1521 {
1522         struct llog_thread_info         *lgi = llog_info(env);
1523         struct dt_insert_rec            *rec = &lgi->lgi_dt_rec;
1524         struct local_oid_storage        *los;
1525         struct dt_object                *o;
1526         int                              rc;
1527
1528         ENTRY;
1529
1530         LASSERT(res->lgh_obj);
1531         LASSERT(th);
1532
1533         /* object can be created by another thread */
1534         o = res->lgh_obj;
1535         if (dt_object_exists(o))
1536                 RETURN(0);
1537
1538         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1539                 struct llog_thread_info *lgi = llog_info(env);
1540
1541                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE;
1542                 lgi->lgi_attr.la_size = 0;
1543                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1544                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1545
1546                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
1547                                        &lgi->lgi_dof, th);
1548                 if (rc < 0)
1549                         RETURN(rc);
1550
1551
1552                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, true);
1553
1554                 RETURN(rc);
1555         }
1556         los = res->private_data;
1557         LASSERT(los);
1558
1559         rc = llog_osd_declare_new_object(env, los, o, th);
1560         if (rc)
1561                 RETURN(rc);
1562
1563         /* do not declare header initialization here as it's declared
1564          * in llog_osd_declare_write_rec() which is always called */
1565
1566         if (res->lgh_name) {
1567                 struct dt_object *llog_dir;
1568
1569                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1570                 if (IS_ERR(llog_dir))
1571                         RETURN(PTR_ERR(llog_dir));
1572                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1573                 rec->rec_fid = &lgi->lgi_fid;
1574                 rec->rec_type = S_IFREG;
1575                 rc = dt_declare_insert(env, llog_dir,
1576                                        (struct dt_rec *)rec,
1577                                        (struct dt_key *)res->lgh_name, th);
1578                 dt_object_put(env, llog_dir);
1579                 if (rc)
1580                         CERROR("%s: can't declare named llog %s: rc = %d\n",
1581                                o->do_lu.lo_dev->ld_obd->obd_name,
1582                                res->lgh_name, rc);
1583         }
1584         RETURN(rc);
1585 }
1586
1587 /**
1588  * Implementation of the llog_operations::lop_create
1589  *
1590  * This function creates the llog according with llog_handle::lgh_obj
1591  * and llog_handle::lgh_name.
1592  *
1593  * \param[in] env       execution environment
1594  * \param[in] res       llog handle of the current llog
1595  * \param[in] th        current transaction handle
1596  *
1597  * \retval              0 on successful create
1598  * \retval              negative value on error
1599  */
1600 static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
1601                            struct thandle *th)
1602 {
1603         struct llog_thread_info *lgi = llog_info(env);
1604         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1605         struct local_oid_storage *los;
1606         struct dt_object        *o;
1607         int                      rc = 0;
1608
1609         ENTRY;
1610
1611         LASSERT(env);
1612         o = res->lgh_obj;
1613         LASSERT(o);
1614
1615         /* llog can be already created */
1616         if (dt_object_exists(o))
1617                 RETURN(-EEXIST);
1618
1619         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1620                 struct llog_thread_info *lgi = llog_info(env);
1621
1622                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE | LA_TYPE;
1623                 lgi->lgi_attr.la_size = 0;
1624                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1625                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1626
1627                 dt_write_lock(env, o, 0);
1628                 rc = dt_create(env, o, &lgi->lgi_attr, NULL,
1629                                &lgi->lgi_dof, th);
1630                 dt_write_unlock(env, o);
1631                 if (rc < 0)
1632                         RETURN(rc);
1633
1634                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, false);
1635
1636                 RETURN(rc);
1637         }
1638
1639         los = res->private_data;
1640         LASSERT(los);
1641
1642         dt_write_lock(env, o, 0);
1643         if (!dt_object_exists(o))
1644                 rc = llog_osd_create_new_object(env, los, o, th);
1645         else
1646                 rc = -EEXIST;
1647
1648         dt_write_unlock(env, o);
1649         if (rc)
1650                 RETURN(rc);
1651
1652         if (res->lgh_name) {
1653                 struct dt_object *llog_dir;
1654
1655                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1656                 if (IS_ERR(llog_dir))
1657                         RETURN(PTR_ERR(llog_dir));
1658
1659                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1660                 rec->rec_fid = &lgi->lgi_fid;
1661                 rec->rec_type = S_IFREG;
1662                 dt_read_lock(env, llog_dir, 0);
1663                 rc = dt_insert(env, llog_dir, (struct dt_rec *)rec,
1664                                (struct dt_key *)res->lgh_name, th);
1665                 dt_read_unlock(env, llog_dir);
1666                 dt_object_put(env, llog_dir);
1667                 if (rc)
1668                         CERROR("%s: can't create named llog %s: rc = %d\n",
1669                                o->do_lu.lo_dev->ld_obd->obd_name,
1670                                res->lgh_name, rc);
1671         }
1672         RETURN(rc);
1673 }
1674
1675 /**
1676  * Implementation of the llog_operations::lop_close
1677  *
1678  * This function closes the llog. It just put llog object and referenced
1679  * local storage.
1680  *
1681  * \param[in] env       execution environment
1682  * \param[in] handle    llog handle of the current llog
1683  *
1684  * \retval              0 on successful llog close
1685  * \retval              negative value on error
1686  */
1687 static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle)
1688 {
1689         struct local_oid_storage        *los;
1690         int                              rc = 0;
1691
1692         ENTRY;
1693
1694         LASSERT(handle->lgh_obj);
1695
1696         if (handle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1697                 /* Remove the object from the cache, otherwise it may
1698                  * hold LOD being released during cleanup process */
1699                 dt_object_put_nocache(env, handle->lgh_obj);
1700                 LASSERT(handle->private_data == NULL);
1701                 RETURN(rc);
1702         } else {
1703                 dt_object_put(env, handle->lgh_obj);
1704         }
1705         los = handle->private_data;
1706         LASSERT(los);
1707         dt_los_put(los);
1708
1709         if (handle->lgh_name)
1710                 OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
1711
1712         RETURN(rc);
1713 }
1714
1715 /**
1716  * delete llog object name entry
1717  *
1718  * Delete llog object (with regular FID) from name space (under
1719  * update_log_dir).
1720  *
1721  * \param [in] env      execution environment
1722  * \param [in] dto      llog object
1723  * \param [in] th       thandle
1724  * \param [in] declare  if it is declare or execution
1725  *
1726  * \retval              0 if deletion succeeds.
1727  * \retval              negative errno if deletion fails.
1728  */
1729 static int
1730 llog_osd_regular_fid_del_name_entry(const struct lu_env *env,
1731                                     struct dt_object *dto,
1732                                     struct thandle *th, bool declare)
1733 {
1734         struct llog_thread_info *lgi = llog_info(env);
1735         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1736         struct dt_object        *dir;
1737         char                    *name = lgi->lgi_name;
1738         int                     rc;
1739         ENTRY;
1740
1741         if (!fid_is_norm(fid))
1742                 RETURN(0);
1743
1744         dir = llog_osd_get_regular_fid_dir(env, dto);
1745         if (IS_ERR(dir))
1746                 RETURN(PTR_ERR(dir));
1747
1748         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1749         dt_write_lock(env, dir, 0);
1750         if (declare) {
1751                 rc = dt_declare_delete(env, dir, (struct dt_key *)name,
1752                                        th);
1753         } else {
1754                 rc = dt_delete(env, dir, (struct dt_key *)name, th);
1755         }
1756         dt_write_unlock(env, dir);
1757
1758         dt_object_put(env, dir);
1759         RETURN(rc);
1760 }
1761
1762 /**
1763  * Implementation of the llog_operations::lop_declare_destroy
1764  *
1765  * This function declare destroys the llog and deletes also entry in the
1766  * llog directory in case of named llog. Llog should be opened prior that.
1767  *
1768  * \param[in] env               execution environment
1769  * \param[in] loghandle llog handle of the current llog
1770  *
1771  * \retval              0 on successful destroy
1772  * \retval              negative value on error
1773  */
1774 static int llog_osd_declare_destroy(const struct lu_env *env,
1775                                     struct llog_handle *loghandle,
1776                                     struct thandle *th)
1777 {
1778         struct llog_ctxt        *ctxt;
1779         struct dt_object        *o, *llog_dir = NULL;
1780         int                      rc;
1781
1782         ENTRY;
1783
1784         ctxt = loghandle->lgh_ctxt;
1785         LASSERT(ctxt);
1786
1787         o = loghandle->lgh_obj;
1788         LASSERT(o);
1789
1790         if (loghandle->lgh_name) {
1791                 llog_dir = llog_osd_dir_get(env, ctxt);
1792                 if (IS_ERR(llog_dir))
1793                         RETURN(PTR_ERR(llog_dir));
1794
1795                 rc = dt_declare_delete(env, llog_dir,
1796                                        (struct dt_key *)loghandle->lgh_name,
1797                                        th);
1798                 if (rc < 0)
1799                         GOTO(out_put, rc);
1800         }
1801
1802         rc = dt_declare_ref_del(env, o, th);
1803         if (rc < 0)
1804                 GOTO(out_put, rc);
1805
1806         rc = dt_declare_destroy(env, o, th);
1807         if (rc < 0)
1808                 GOTO(out_put, rc);
1809
1810         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1811                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, true);
1812                 if (rc < 0)
1813                         GOTO(out_put, rc);
1814         }
1815
1816 out_put:
1817         if (!(IS_ERR_OR_NULL(llog_dir)))
1818                 dt_object_put(env, llog_dir);
1819
1820         RETURN(rc);
1821 }
1822
1823
1824 /**
1825  * Implementation of the llog_operations::lop_destroy
1826  *
1827  * This function destroys the llog and deletes also entry in the
1828  * llog directory in case of named llog. Llog should be opened prior that.
1829  * Destroy method is not part of external transaction and does everything
1830  * inside.
1831  *
1832  * \param[in] env               execution environment
1833  * \param[in] loghandle llog handle of the current llog
1834  *
1835  * \retval              0 on successful destroy
1836  * \retval              negative value on error
1837  */
1838 static int llog_osd_destroy(const struct lu_env *env,
1839                             struct llog_handle *loghandle, struct thandle *th)
1840 {
1841         struct llog_ctxt        *ctxt;
1842         struct dt_object        *o, *llog_dir = NULL;
1843         int                      rc;
1844
1845         ENTRY;
1846
1847         ctxt = loghandle->lgh_ctxt;
1848         LASSERT(ctxt != NULL);
1849
1850         o = loghandle->lgh_obj;
1851         LASSERT(o != NULL);
1852
1853         dt_write_lock(env, o, 0);
1854         if (!llog_osd_exist(loghandle))
1855                 GOTO(out_unlock, rc = 0);
1856
1857         if (loghandle->lgh_name) {
1858                 llog_dir = llog_osd_dir_get(env, ctxt);
1859                 if (IS_ERR(llog_dir))
1860                         GOTO(out_unlock, rc = PTR_ERR(llog_dir));
1861
1862                 dt_read_lock(env, llog_dir, 0);
1863                 rc = dt_delete(env, llog_dir,
1864                                (struct dt_key *)loghandle->lgh_name,
1865                                th);
1866                 dt_read_unlock(env, llog_dir);
1867                 if (rc) {
1868                         CERROR("%s: can't remove llog %s: rc = %d\n",
1869                                o->do_lu.lo_dev->ld_obd->obd_name,
1870                                loghandle->lgh_name, rc);
1871                         GOTO(out_unlock, rc);
1872                 }
1873         }
1874
1875         dt_ref_del(env, o, th);
1876         rc = dt_destroy(env, o, th);
1877         if (rc < 0)
1878                 GOTO(out_unlock, rc);
1879
1880         loghandle->lgh_destroyed = true;
1881         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1882                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, false);
1883                 if (rc < 0)
1884                         GOTO(out_unlock, rc);
1885         }
1886
1887 out_unlock:
1888         dt_write_unlock(env, o);
1889         if (!(IS_ERR_OR_NULL(llog_dir)))
1890                 dt_object_put(env, llog_dir);
1891         RETURN(rc);
1892 }
1893
1894 /**
1895  * Implementation of the llog_operations::lop_setup
1896  *
1897  * This function setup the llog on local storage.
1898  *
1899  * \param[in] env       execution environment
1900  * \param[in] obd       obd device the llog belongs to
1901  * \param[in] olg       the llog group, it is always zero group now.
1902  * \param[in] ctxt_idx  the llog index, it defines the purpose of this llog.
1903  *                      Every new llog type have to use own index.
1904  * \param[in] disk_obd  the storage obd, where llog is stored.
1905  *
1906  * \retval              0 on successful llog setup
1907  * \retval              negative value on error
1908  */
1909 static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
1910                           struct obd_llog_group *olg, int ctxt_idx,
1911                           struct obd_device *disk_obd)
1912 {
1913         struct llog_thread_info         *lgi = llog_info(env);
1914         struct llog_ctxt                *ctxt;
1915         int                              rc = 0;
1916         ENTRY;
1917
1918         LASSERT(obd);
1919         LASSERT(olg->olg_ctxts[ctxt_idx]);
1920
1921         ctxt = llog_ctxt_get(olg->olg_ctxts[ctxt_idx]);
1922         LASSERT(ctxt);
1923
1924         if (disk_obd == NULL)
1925                 GOTO(out, rc = 0);
1926
1927         /* initialize data allowing to generate new fids,
1928          * literally we need a sequece */
1929         lgi->lgi_fid.f_seq = FID_SEQ_LLOG;
1930         lgi->lgi_fid.f_oid = 1;
1931         lgi->lgi_fid.f_ver = 0;
1932         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1933                                     &lgi->lgi_fid,
1934                                     &ctxt->loc_los_nameless);
1935         if (rc != 0)
1936                 GOTO(out, rc);
1937
1938         lgi->lgi_fid.f_seq = FID_SEQ_LLOG_NAME;
1939         lgi->lgi_fid.f_oid = 1;
1940         lgi->lgi_fid.f_ver = 0;
1941         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1942                                     &lgi->lgi_fid,
1943                                     &ctxt->loc_los_named);
1944         if (rc != 0) {
1945                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1946                 ctxt->loc_los_nameless = NULL;
1947         }
1948
1949         GOTO(out, rc);
1950
1951 out:
1952         llog_ctxt_put(ctxt);
1953         return rc;
1954 }
1955
1956 /**
1957  * Implementation of the llog_operations::lop_cleanup
1958  *
1959  * This function cleanups the llog on local storage.
1960  *
1961  * \param[in] env       execution environment
1962  * \param[in] ctxt      the llog context
1963  *
1964  * \retval              0
1965  */
1966 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
1967 {
1968         if (ctxt->loc_los_nameless != NULL) {
1969                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1970                 ctxt->loc_los_nameless = NULL;
1971         }
1972
1973         if (ctxt->loc_los_named != NULL) {
1974                 local_oid_storage_fini(env, ctxt->loc_los_named);
1975                 ctxt->loc_los_named = NULL;
1976         }
1977
1978         return 0;
1979 }
1980
1981 const struct llog_operations llog_osd_ops = {
1982         .lop_next_block         = llog_osd_next_block,
1983         .lop_prev_block         = llog_osd_prev_block,
1984         .lop_read_header        = llog_osd_read_header,
1985         .lop_declare_destroy    = llog_osd_declare_destroy,
1986         .lop_destroy            = llog_osd_destroy,
1987         .lop_setup              = llog_osd_setup,
1988         .lop_cleanup            = llog_osd_cleanup,
1989         .lop_open               = llog_osd_open,
1990         .lop_exist              = llog_osd_exist,
1991         .lop_declare_create     = llog_osd_declare_create,
1992         .lop_create             = llog_osd_create,
1993         .lop_declare_write_rec  = llog_osd_declare_write_rec,
1994         .lop_write_rec          = llog_osd_write_rec,
1995         .lop_close              = llog_osd_close,
1996 };
1997 EXPORT_SYMBOL(llog_osd_ops);
1998
1999 const struct llog_operations llog_common_cat_ops = {
2000         .lop_next_block         = llog_osd_next_block,
2001         .lop_prev_block         = llog_osd_prev_block,
2002         .lop_read_header        = llog_osd_read_header,
2003         .lop_declare_destroy    = llog_osd_declare_destroy,
2004         .lop_destroy            = llog_osd_destroy,
2005         .lop_setup              = llog_osd_setup,
2006         .lop_cleanup            = llog_osd_cleanup,
2007         .lop_open               = llog_osd_open,
2008         .lop_exist              = llog_osd_exist,
2009         .lop_declare_create     = llog_osd_declare_create,
2010         .lop_create             = llog_osd_create,
2011         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2012         .lop_write_rec          = llog_osd_write_rec,
2013         .lop_close              = llog_osd_close,
2014         .lop_add                = llog_cat_add_rec,
2015         .lop_declare_add        = llog_cat_declare_add_rec,
2016 };
2017 EXPORT_SYMBOL(llog_common_cat_ops);
2018
2019 /**
2020  * Read the special file which contains the list of llog catalogs IDs
2021  *
2022  * This function reads the CATALOGS file which contains the array of llog
2023  * catalogs IDs. The main purpose of this file is to store OSP llogs indexed
2024  * by OST/MDT number.
2025  *
2026  * \param[in]  env              execution environment
2027  * \param[in]  d                corresponding storage device
2028  * \param[in]  idx              position to start from, usually OST/MDT index
2029  * \param[in]  count            how many catalog IDs to read
2030  * \param[out] idarray          the buffer for the data. If it is NULL then
2031  *                              function returns just number of catalog IDs
2032  *                              in the file.
2033  * \param[in]  fid              LLOG_CATALOGS_OID for CATALOG object
2034  *
2035  * \retval                      0 on successful read of catalog IDs
2036  * \retval                      negative value on error
2037  * \retval                      positive value which is number of records in
2038  *                              the file if \a idarray is NULL
2039  */
2040 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
2041                           int idx, int count, struct llog_catid *idarray,
2042                           const struct lu_fid *fid)
2043 {
2044         struct llog_thread_info *lgi = llog_info(env);
2045         struct dt_object        *o = NULL;
2046         struct thandle          *th;
2047         int                      rc, size;
2048
2049         ENTRY;
2050
2051         LASSERT(d);
2052
2053         size = sizeof(*idarray) * count;
2054         lgi->lgi_off = idx *  sizeof(*idarray);
2055
2056         lgi->lgi_fid = *fid;
2057         o = dt_locate(env, d, &lgi->lgi_fid);
2058         if (IS_ERR(o))
2059                 RETURN(PTR_ERR(o));
2060
2061         if (!dt_object_exists(o)) {
2062                 th = dt_trans_create(env, d);
2063                 if (IS_ERR(th))
2064                         GOTO(out, rc = PTR_ERR(th));
2065
2066                 lgi->lgi_attr.la_valid = LA_MODE | LA_TYPE;
2067                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
2068                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
2069
2070                 th->th_wait_submit = 1;
2071                 /* Make the llog object creation synchronization, so
2072                  * it will be reliable to the reference, especially
2073                  * for remote reference */
2074                 th->th_sync = 1;
2075
2076                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
2077                                        &lgi->lgi_dof, th);
2078                 if (rc)
2079                         GOTO(out_trans, rc);
2080
2081                 rc = dt_trans_start_local(env, d, th);
2082                 if (rc)
2083                         GOTO(out_trans, rc);
2084
2085                 dt_write_lock(env, o, 0);
2086                 if (!dt_object_exists(o))
2087                         rc = dt_create(env, o, &lgi->lgi_attr, NULL,
2088                                        &lgi->lgi_dof, th);
2089                 dt_write_unlock(env, o);
2090 out_trans:
2091                 dt_trans_stop(env, d, th);
2092                 if (rc)
2093                         GOTO(out, rc);
2094         }
2095
2096         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2097         if (rc)
2098                 GOTO(out, rc);
2099
2100         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2101                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2102                        o->do_lu.lo_dev->ld_obd->obd_name,
2103                        lgi->lgi_attr.la_mode);
2104                 GOTO(out, rc = -ENOENT);
2105         }
2106
2107         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
2108                (int)lgi->lgi_attr.la_size, size);
2109
2110         /* return just number of llogs */
2111         if (idarray == NULL) {
2112                 rc = lgi->lgi_attr.la_size / sizeof(*idarray);
2113                 GOTO(out, rc);
2114         }
2115
2116         /* read for new ost index or for empty file */
2117         memset(idarray, 0, size);
2118         if (lgi->lgi_attr.la_size <= lgi->lgi_off)
2119                 GOTO(out, rc = 0);
2120         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
2121                 size = lgi->lgi_attr.la_size - lgi->lgi_off;
2122
2123         lgi->lgi_buf.lb_buf = idarray;
2124         lgi->lgi_buf.lb_len = size;
2125         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
2126         /* -EFAULT means the llog is a sparse file. This is not an error
2127          * after arbitrary OST index is supported. */
2128         if (rc < 0 && rc != -EFAULT) {
2129                 CERROR("%s: error reading CATALOGS: rc = %d\n",
2130                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
2131                 GOTO(out, rc);
2132         }
2133
2134         EXIT;
2135 out:
2136         dt_object_put(env, o);
2137         RETURN(rc);
2138 }
2139 EXPORT_SYMBOL(llog_osd_get_cat_list);
2140
2141 /**
2142  * Write the special file which contains the list of llog catalogs IDs
2143  *
2144  * This function writes the CATALOG file which contains the array of llog
2145  * catalogs IDs. It is used mostly to store OSP llogs indexed by OST/MDT
2146  * number.
2147  *
2148  * \param[in]  env      execution environment
2149  * \param[in]  d        corresponding storage device
2150  * \param[in]  idx      position to start from, usually OST/MDT index
2151  * \param[in]  count    how many catalog IDs to write
2152  * \param[out] idarray  the buffer with the data to write.
2153  * \param[in]  fid      LLOG_CATALOGS_OID for CATALOG object
2154  *
2155  * \retval              0 on successful write of catalog IDs
2156  * \retval              negative value on error
2157  */
2158 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
2159                           int idx, int count, struct llog_catid *idarray,
2160                           const struct lu_fid *fid)
2161 {
2162         struct llog_thread_info *lgi = llog_info(env);
2163         struct dt_object        *o = NULL;
2164         struct thandle          *th;
2165         int                      rc, size;
2166
2167         if (count == 0)
2168                 RETURN(0);
2169
2170         LASSERT(d);
2171
2172         size = sizeof(*idarray) * count;
2173         lgi->lgi_off = idx * sizeof(*idarray);
2174         lgi->lgi_fid = *fid;
2175
2176         o = dt_locate(env, d, &lgi->lgi_fid);
2177         if (IS_ERR(o))
2178                 RETURN(PTR_ERR(o));
2179
2180         if (!dt_object_exists(o))
2181                 GOTO(out, rc = -ENOENT);
2182
2183         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2184         if (rc)
2185                 GOTO(out, rc);
2186
2187         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2188                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2189                        o->do_lu.lo_dev->ld_obd->obd_name,
2190                        lgi->lgi_attr.la_mode);
2191                 GOTO(out, rc = -ENOENT);
2192         }
2193
2194         th = dt_trans_create(env, d);
2195         if (IS_ERR(th))
2196                 GOTO(out, rc = PTR_ERR(th));
2197
2198         lgi->lgi_buf.lb_len = size;
2199         lgi->lgi_buf.lb_buf = idarray;
2200         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, lgi->lgi_off, th);
2201         if (rc)
2202                 GOTO(out_trans, rc);
2203
2204         /* For update log, this happens during initialization,
2205          * see lod_sub_prep_llog(), and we need make sure catlog
2206          * file ID is written to catlist file(committed) before
2207          * cross-MDT operation write update records to catlog FILE,
2208          * otherwise, during failover these update records might
2209          * missing */
2210         if (fid_is_update_log(fid))
2211                 th->th_sync = 1;
2212
2213         rc = dt_trans_start_local(env, d, th);
2214         if (rc)
2215                 GOTO(out_trans, rc);
2216
2217         th->th_wait_submit = 1;
2218
2219         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
2220         if (rc)
2221                 CDEBUG(D_INODE, "can't write CATALOGS at index %d: rc = %d\n",
2222                        idx, rc);
2223 out_trans:
2224         dt_trans_stop(env, d, th);
2225 out:
2226         dt_object_put(env, o);
2227         RETURN(rc);
2228 }
2229 EXPORT_SYMBOL(llog_osd_put_cat_list);