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