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