lustre/obdclass/llog.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/obdclass/llog.c
  33  *
  34  * OST<->MDS recovery logging infrastructure.
  35  * Invariants in implementation:
  36  * - we do not share logs among different OST<->MDS connections, so that
  37  *   if an OST or MDS fails it need only look at log(s) relevant to itself
  38  *
  39  * Author: Andreas Dilger <adilger@clusterfs.com>
  40  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  41  * Author: Mikhail Pershin <tappro@whamcloud.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_LOG
  45
  46 #include <linux/pid_namespace.h>
  47 #include <linux/kthread.h>
  48 #include <llog_swab.h>
  49 #include <lustre_log.h>
  50 #include <obd_class.h>
  51 #include "llog_internal.h"
  52 /*
  53  * Allocate a new log or catalog handle
  54  * Used inside llog_open().
  55  */
  56 static struct llog_handle *llog_alloc_handle(void)
  57 {
  58         struct llog_handle *loghandle;
  59
  60         OBD_ALLOC_PTR(loghandle);
  61         if (loghandle == NULL)
  62                 return NULL;
  63
  64         init_rwsem(&loghandle->lgh_lock);
  65         mutex_init(&loghandle->lgh_hdr_mutex);
  66         INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);
  67         atomic_set(&loghandle->lgh_refcount, 1);
  68
  69         return loghandle;
  70 }
  71
  72 /*
  73  * Free llog handle and header data if exists. Used in llog_close() only
  74  */
  75 static void llog_free_handle(struct llog_handle *loghandle)
  76 {
  77         LASSERT(loghandle != NULL);
  78
  79         /* failed llog_init_handle */
  80         if (loghandle->lgh_hdr == NULL)
  81                 goto out;
  82
  83         if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)
  84                 LASSERT(list_empty(&loghandle->u.phd.phd_entry));
  85         else if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
  86                 LASSERT(list_empty(&loghandle->u.chd.chd_head));
  87         OBD_FREE_LARGE(loghandle->lgh_hdr, loghandle->lgh_hdr_size);
  88 out:
  89         OBD_FREE_PTR(loghandle);
  90 }
  91
  92 void llog_handle_get(struct llog_handle *loghandle)
  93 {
  94         atomic_inc(&loghandle->lgh_refcount);
  95 }
  96
  97 void llog_handle_put(struct llog_handle *loghandle)
  98 {
  99         LASSERT(atomic_read(&loghandle->lgh_refcount) > 0);
 100         if (atomic_dec_and_test(&loghandle->lgh_refcount))
 101                 llog_free_handle(loghandle);
 102 }
 103
 104 static int llog_declare_destroy(const struct lu_env *env,
 105                                 struct llog_handle *handle,
 106                                 struct thandle *th)
 107 {
 108         struct llog_operations *lop;
 109         int rc;
 110
 111         ENTRY;
 112
 113         rc = llog_handle2ops(handle, &lop);
 114         if (rc)
 115                 RETURN(rc);
 116         if (lop->lop_declare_destroy == NULL)
 117                 RETURN(-EOPNOTSUPP);
 118
 119         rc = lop->lop_declare_destroy(env, handle, th);
 120
 121         RETURN(rc);
 122 }
 123
 124 int llog_trans_destroy(const struct lu_env *env, struct llog_handle *handle,
 125                        struct thandle *th)
 126 {
 127         struct llog_operations  *lop;
 128         int rc;
 129         ENTRY;
 130
 131         rc = llog_handle2ops(handle, &lop);
 132         if (rc < 0)
 133                 RETURN(rc);
 134         if (lop->lop_destroy == NULL)
 135                 RETURN(-EOPNOTSUPP);
 136
 137         LASSERT(handle->lgh_obj != NULL);
 138         if (!dt_object_exists(handle->lgh_obj))
 139                 RETURN(0);
 140
 141         rc = lop->lop_destroy(env, handle, th);
 142
 143         RETURN(rc);
 144 }
 145
 146 int llog_destroy(const struct lu_env *env, struct llog_handle *handle)
 147 {
 148         struct llog_operations  *lop;
 149         struct dt_device        *dt;
 150         struct thandle          *th;
 151         int rc;
 152
 153         ENTRY;
 154
 155         rc = llog_handle2ops(handle, &lop);
 156         if (rc < 0)
 157                 RETURN(rc);
 158         if (lop->lop_destroy == NULL)
 159                 RETURN(-EOPNOTSUPP);
 160
 161         if (handle->lgh_obj == NULL) {
 162                 /* if lgh_obj == NULL, then it is from client side destroy */
 163                 rc = lop->lop_destroy(env, handle, NULL);
 164                 RETURN(rc);
 165         }
 166
 167         if (!dt_object_exists(handle->lgh_obj))
 168                 RETURN(0);
 169
 170         dt = lu2dt_dev(handle->lgh_obj->do_lu.lo_dev);
 171
 172         if (unlikely(unlikely(dt->dd_rdonly)))
 173                 RETURN(-EROFS);
 174
 175         th = dt_trans_create(env, dt);
 176         if (IS_ERR(th))
 177                 RETURN(PTR_ERR(th));
 178
 179         rc = llog_declare_destroy(env, handle, th);
 180         if (rc != 0)
 181                 GOTO(out_trans, rc);
 182
 183         rc = dt_trans_start_local(env, dt, th);
 184         if (rc < 0)
 185                 GOTO(out_trans, rc);
 186
 187         rc = lop->lop_destroy(env, handle, th);
 188
 189 out_trans:
 190         dt_trans_stop(env, dt, th);
 191
 192         RETURN(rc);
 193 }
 194 EXPORT_SYMBOL(llog_destroy);
 195
 196 /* returns negative on error; 0 if success; 1 if success & log destroyed */
 197 int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
 198                     int index)
 199 {
 200         struct llog_thread_info *lgi = llog_info(env);
 201         struct dt_device        *dt;
 202         struct llog_log_hdr     *llh;
 203         struct thandle          *th;
 204         __u32                    tmp_lgc_index;
 205         int                      rc;
 206         int rc1;
 207         bool subtract_count = false;
 208
 209         ENTRY;
 210
 211         LASSERT(loghandle != NULL);
 212         LASSERT(loghandle->lgh_ctxt != NULL);
 213         LASSERT(loghandle->lgh_obj != NULL);
 214
 215         llh = loghandle->lgh_hdr;
 216
 217         CDEBUG(D_RPCTRACE, "Canceling %d in log "DFID"\n", index,
 218                PFID(&loghandle->lgh_id.lgl_oi.oi_fid));
 219
 220         if (index == 0) {
 221                 CERROR("Can't cancel index 0 which is header\n");
 222                 RETURN(-EINVAL);
 223         }
 224
 225         dt = lu2dt_dev(loghandle->lgh_obj->do_lu.lo_dev);
 226
 227         if (unlikely(unlikely(dt->dd_rdonly)))
 228                 RETURN(0);
 229
 230         th = dt_trans_create(env, dt);
 231         if (IS_ERR(th))
 232                 RETURN(PTR_ERR(th));
 233
 234         rc = llog_declare_write_rec(env, loghandle, &llh->llh_hdr, index, th);
 235         if (rc < 0)
 236                 GOTO(out_trans, rc);
 237
 238         if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY)) {
 239                 rc = llog_declare_destroy(env, loghandle, th);
 240                 if (rc < 0)
 241                         GOTO(out_trans, rc);
 242         }
 243
 244         th->th_wait_submit = 1;
 245         rc = dt_trans_start_local(env, dt, th);
 246         if (rc < 0)
 247                 GOTO(out_trans, rc);
 248
 249         down_write(&loghandle->lgh_lock);
 250         /* clear bitmap */
 251         mutex_lock(&loghandle->lgh_hdr_mutex);
 252         if (!ext2_clear_bit(index, LLOG_HDR_BITMAP(llh))) {
 253                 CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
 254                 GOTO(out_unlock, rc);
 255         }
 256
 257         loghandle->lgh_hdr->llh_count--;
 258         subtract_count = true;
 259
 260         /* Since llog_process_thread use lgi_cookie, it`s better to save them
 261          * and restore after using
 262          */
 263         tmp_lgc_index = lgi->lgi_cookie.lgc_index;
 264         /* Pass this index to llog_osd_write_rec(), which will use the index
 265          * to only update the necesary bitmap. */
 266         lgi->lgi_cookie.lgc_index = index;
 267         /* update header */
 268         rc = llog_write_rec(env, loghandle, &llh->llh_hdr, &lgi->lgi_cookie,
 269                             LLOG_HEADER_IDX, th);
 270         lgi->lgi_cookie.lgc_index = tmp_lgc_index;
 271
 272         if (rc != 0)
 273                 GOTO(out_unlock, rc);
 274
 275         if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
 276             (llh->llh_count == 1) &&
 277             ((loghandle->lgh_last_idx == LLOG_HDR_BITMAP_SIZE(llh) - 1) ||
 278              (loghandle->u.phd.phd_cat_handle != NULL &&
 279               loghandle->u.phd.phd_cat_handle->u.chd.chd_current_log !=
 280                 loghandle))) {
 281                 /* never try to destroy it again */
 282                 llh->llh_flags &= ~LLOG_F_ZAP_WHEN_EMPTY;
 283                 rc = llog_trans_destroy(env, loghandle, th);
 284                 if (rc < 0) {
 285                         /* Sigh, can not destroy the final plain llog, but
 286                          * the bitmap has been clearly, so the record can not
 287                          * be accessed anymore, let's return 0 for now, and
 288                          * the orphan will be handled by LFSCK. */
 289                         CERROR("%s: can't destroy empty llog "DFID": rc = %d\n",
 290                                loghandle->lgh_ctxt->loc_obd->obd_name,
 291                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid), rc);
 292                         GOTO(out_unlock, rc = 0);
 293                 }
 294                 rc = LLOG_DEL_PLAIN;
 295         }
 296
 297 out_unlock:
 298         mutex_unlock(&loghandle->lgh_hdr_mutex);
 299         up_write(&loghandle->lgh_lock);
 300 out_trans:
 301         rc1 = dt_trans_stop(env, dt, th);
 302         if (rc == 0)
 303                 rc = rc1;
 304         if (rc < 0 && subtract_count) {
 305                 mutex_lock(&loghandle->lgh_hdr_mutex);
 306                 loghandle->lgh_hdr->llh_count++;
 307                 ext2_set_bit(index, LLOG_HDR_BITMAP(llh));
 308                 mutex_unlock(&loghandle->lgh_hdr_mutex);
 309         }
 310         RETURN(rc);
 311 }
 312
 313 int llog_read_header(const struct lu_env *env, struct llog_handle *handle,
 314                      const struct obd_uuid *uuid)
 315 {
 316         struct llog_operations *lop;
 317         int rc;
 318         ENTRY;
 319
 320         rc = llog_handle2ops(handle, &lop);
 321         if (rc)
 322                 RETURN(rc);
 323
 324         if (lop->lop_read_header == NULL)
 325                 RETURN(-EOPNOTSUPP);
 326
 327         rc = lop->lop_read_header(env, handle);
 328         if (rc == LLOG_EEMPTY) {
 329                 struct llog_log_hdr *llh = handle->lgh_hdr;
 330
 331                 /* lrh_len should be initialized in llog_init_handle */
 332                 handle->lgh_last_idx = 0; /* header is record with index 0 */
 333                 llh->llh_count = 1;         /* for the header record */
 334                 llh->llh_hdr.lrh_type = LLOG_HDR_MAGIC;
 335                 LASSERT(handle->lgh_ctxt->loc_chunk_size >=
 336                                                 LLOG_MIN_CHUNK_SIZE);
 337                 llh->llh_hdr.lrh_len = handle->lgh_ctxt->loc_chunk_size;
 338                 llh->llh_hdr.lrh_index = 0;
 339                 llh->llh_timestamp = ktime_get_real_seconds();
 340                 if (uuid)
 341                         memcpy(&llh->llh_tgtuuid, uuid,
 342                                sizeof(llh->llh_tgtuuid));
 343                 llh->llh_bitmap_offset = offsetof(typeof(*llh), llh_bitmap);
 344                 /* Since update llog header might also call this function,
 345                  * let's reset the bitmap to 0 here */
 346                 memset(LLOG_HDR_BITMAP(llh), 0, llh->llh_hdr.lrh_len -
 347                                                 llh->llh_bitmap_offset -
 348                                                 sizeof(llh->llh_tail));
 349                 ext2_set_bit(0, LLOG_HDR_BITMAP(llh));
 350                 LLOG_HDR_TAIL(llh)->lrt_len = llh->llh_hdr.lrh_len;
 351                 LLOG_HDR_TAIL(llh)->lrt_index = llh->llh_hdr.lrh_index;
 352                 rc = 0;
 353         }
 354         RETURN(rc);
 355 }
 356 EXPORT_SYMBOL(llog_read_header);
 357
 358 int llog_init_handle(const struct lu_env *env, struct llog_handle *handle,
 359                      int flags, struct obd_uuid *uuid)
 360 {
 361         struct llog_log_hdr     *llh;
 362         enum llog_flag           fmt = flags & LLOG_F_EXT_MASK;
 363         int                      rc;
 364         int                     chunk_size = handle->lgh_ctxt->loc_chunk_size;
 365         ENTRY;
 366
 367         LASSERT(handle->lgh_hdr == NULL);
 368
 369         LASSERT(chunk_size >= LLOG_MIN_CHUNK_SIZE);
 370         OBD_ALLOC_LARGE(llh, chunk_size);
 371         if (llh == NULL)
 372                 RETURN(-ENOMEM);
 373
 374         handle->lgh_hdr = llh;
 375         handle->lgh_hdr_size = chunk_size;
 376         /* first assign flags to use llog_client_ops */
 377         llh->llh_flags = flags;
 378         rc = llog_read_header(env, handle, uuid);
 379         if (rc == 0) {
 380                 if (unlikely((llh->llh_flags & LLOG_F_IS_PLAIN &&
 381                               flags & LLOG_F_IS_CAT) ||
 382                              (llh->llh_flags & LLOG_F_IS_CAT &&
 383                               flags & LLOG_F_IS_PLAIN))) {
 384                         CERROR("%s: llog type is %s but initializing %s\n",
 385                                handle->lgh_ctxt->loc_obd->obd_name,
 386                                llh->llh_flags & LLOG_F_IS_CAT ?
 387                                "catalog" : "plain",
 388                                flags & LLOG_F_IS_CAT ? "catalog" : "plain");
 389                         GOTO(out, rc = -EINVAL);
 390                 } else if (llh->llh_flags &
 391                            (LLOG_F_IS_PLAIN | LLOG_F_IS_CAT)) {
 392                         /*
 393                          * it is possible to open llog without specifying llog
 394                          * type so it is taken from llh_flags
 395                          */
 396                         flags = llh->llh_flags;
 397                 } else {
 398                         /* for some reason the llh_flags has no type set */
 399                         CERROR("llog type is not specified!\n");
 400                         GOTO(out, rc = -EINVAL);
 401                 }
 402                 if (unlikely(uuid &&
 403                              !obd_uuid_equals(uuid, &llh->llh_tgtuuid))) {
 404                         CERROR("%s: llog uuid mismatch: %s/%s\n",
 405                                handle->lgh_ctxt->loc_obd->obd_name,
 406                                (char *)uuid->uuid,
 407                                (char *)llh->llh_tgtuuid.uuid);
 408                         GOTO(out, rc = -EEXIST);
 409                 }
 410         }
 411         if (flags & LLOG_F_IS_CAT) {
 412                 LASSERT(list_empty(&handle->u.chd.chd_head));
 413                 INIT_LIST_HEAD(&handle->u.chd.chd_head);
 414                 llh->llh_size = sizeof(struct llog_logid_rec);
 415                 llh->llh_flags |= LLOG_F_IS_FIXSIZE;
 416         } else if (!(flags & LLOG_F_IS_PLAIN)) {
 417                 CERROR("%s: unknown flags: %#x (expected %#x or %#x)\n",
 418                        handle->lgh_ctxt->loc_obd->obd_name,
 419                        flags, LLOG_F_IS_CAT, LLOG_F_IS_PLAIN);
 420                 rc = -EINVAL;
 421         }
 422         llh->llh_flags |= fmt;
 423 out:
 424         if (rc) {
 425                 OBD_FREE_LARGE(llh, chunk_size);
 426                 handle->lgh_hdr = NULL;
 427         }
 428         RETURN(rc);
 429 }
 430 EXPORT_SYMBOL(llog_init_handle);
 431
 432 static int llog_process_thread(void *arg)
 433 {
 434         struct llog_process_info        *lpi = arg;
 435         struct llog_handle              *loghandle = lpi->lpi_loghandle;
 436         struct llog_log_hdr             *llh = loghandle->lgh_hdr;
 437         struct llog_process_cat_data    *cd  = lpi->lpi_catdata;
 438         struct llog_thread_info         *lti;
 439         char                            *buf;
 440         size_t                           chunk_size;
 441         __u64                            cur_offset;
 442         int                              rc = 0, index = 1, last_index;
 443         int                              saved_index = 0;
 444         int                              last_called_index = 0;
 445         bool                             repeated = false;
 446
 447         ENTRY;
 448
 449         if (llh == NULL)
 450                 RETURN(-EINVAL);
 451
 452         lti = lpi->lpi_env == NULL ? NULL : llog_info(lpi->lpi_env);
 453
 454         cur_offset = chunk_size = llh->llh_hdr.lrh_len;
 455         /* expect chunk_size to be power of two */
 456         LASSERT(is_power_of_2(chunk_size));
 457
 458         OBD_ALLOC_LARGE(buf, chunk_size);
 459         if (buf == NULL) {
 460                 lpi->lpi_rc = -ENOMEM;
 461                 RETURN(0);
 462         }
 463
 464         if (cd != NULL) {
 465                 last_called_index = cd->lpcd_first_idx;
 466                 index = cd->lpcd_first_idx + 1;
 467         }
 468         if (cd != NULL && cd->lpcd_last_idx)
 469                 last_index = cd->lpcd_last_idx;
 470         else
 471                 last_index = LLOG_HDR_BITMAP_SIZE(llh) - 1;
 472
 473         while (rc == 0) {
 474                 struct llog_rec_hdr *rec;
 475                 off_t chunk_offset = 0;
 476                 unsigned int buf_offset = 0;
 477                 bool partial_chunk;
 478                 int     lh_last_idx;
 479
 480                 /* skip records not set in bitmap */
 481                 while (index <= last_index &&
 482                        !ext2_test_bit(index, LLOG_HDR_BITMAP(llh)))
 483                         ++index;
 484
 485                 /* There are no indices prior the last_index */
 486                 if (index > last_index)
 487                         break;
 488
 489                 CDEBUG(D_OTHER, "index: %d last_index %d\n", index,
 490                        last_index);
 491
 492 repeat:
 493                 /* get the buf with our target record; avoid old garbage */
 494                 memset(buf, 0, chunk_size);
 495                 /* the record index for outdated chunk data */
 496                 lh_last_idx = loghandle->lgh_last_idx + 1;
 497                 rc = llog_next_block(lpi->lpi_env, loghandle, &saved_index,
 498                                      index, &cur_offset, buf, chunk_size);
 499                 if (repeated && rc)
 500                         CDEBUG(D_OTHER, "cur_offset %llu, chunk_offset %llu,"
 501                                " buf_offset %u, rc = %d\n", cur_offset,
 502                                (__u64)chunk_offset, buf_offset, rc);
 503                 /* we`ve tried to reread the chunk, but there is no
 504                  * new records */
 505                 if (rc == -EIO && repeated && (chunk_offset + buf_offset) ==
 506                     cur_offset)
 507                         GOTO(out, rc = 0);
 508                 if (rc != 0)
 509                         GOTO(out, rc);
 510
 511                 /* NB: after llog_next_block() call the cur_offset is the
 512                  * offset of the next block after read one.
 513                  * The absolute offset of the current chunk is calculated
 514                  * from cur_offset value and stored in chunk_offset variable.
 515                  */
 516                 if ((cur_offset & (chunk_size - 1)) != 0) {
 517                         partial_chunk = true;
 518                         chunk_offset = cur_offset & ~(chunk_size - 1);
 519                 } else {
 520                         partial_chunk = false;
 521                         chunk_offset = cur_offset - chunk_size;
 522                 }
 523
 524                 /* NB: when rec->lrh_len is accessed it is already swabbed
 525                  * since it is used at the "end" of the loop and the rec
 526                  * swabbing is done at the beginning of the loop. */
 527                 for (rec = (struct llog_rec_hdr *)(buf + buf_offset);
 528                      (char *)rec < buf + chunk_size;
 529                      rec = llog_rec_hdr_next(rec)) {
 530
 531                         CDEBUG(D_OTHER, "processing rec 0x%p type %#x\n",
 532                                rec, rec->lrh_type);
 533
 534                         if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
 535                                 lustre_swab_llog_rec(rec);
 536
 537                         CDEBUG(D_OTHER, "after swabbing, type=%#x idx=%d\n",
 538                                rec->lrh_type, rec->lrh_index);
 539
 540                         /* for partial chunk the end of it is zeroed, check
 541                          * for index 0 to distinguish it. */
 542                         if (partial_chunk && rec->lrh_index == 0) {
 543                                 /* concurrent llog_add() might add new records
 544                                  * while llog_processing, check this is not
 545                                  * the case and re-read the current chunk
 546                                  * otherwise. */
 547                                 int records;
 548                                 /* lgh_last_idx could be less then index
 549                                  * for catalog, if catalog is wrapped */
 550                                 if ((index > loghandle->lgh_last_idx &&
 551                                     !(loghandle->lgh_hdr->llh_flags &
 552                                       LLOG_F_IS_CAT)) || repeated ||
 553                                     (loghandle->lgh_obj != NULL &&
 554                                      dt_object_remote(loghandle->lgh_obj)))
 555                                         GOTO(out, rc = 0);
 556                                 /* <2 records means no more records
 557                                  * if the last record we processed was
 558                                  * the final one, then the underlying
 559                                  * object might have been destroyed yet.
 560                                  * we better don't access that.. */
 561                                 mutex_lock(&loghandle->lgh_hdr_mutex);
 562                                 records = loghandle->lgh_hdr->llh_count;
 563                                 mutex_unlock(&loghandle->lgh_hdr_mutex);
 564                                 if (records <= 1)
 565                                         GOTO(out, rc = 0);
 566                                 CDEBUG(D_OTHER, "Re-read last llog buffer for "
 567                                        "new records, index %u, last %u\n",
 568                                        index, loghandle->lgh_last_idx);
 569                                 /* save offset inside buffer for the re-read */
 570                                 buf_offset = (char *)rec - (char *)buf;
 571                                 cur_offset = chunk_offset;
 572                                 repeated = true;
 573                                 goto repeat;
 574                         }
 575
 576                         repeated = false;
 577
 578                         if (rec->lrh_len == 0 || rec->lrh_len > chunk_size) {
 579                                 CWARN("%s: invalid length %d in llog "DFID
 580                                       "record for index %d/%d\n",
 581                                        loghandle->lgh_ctxt->loc_obd->obd_name,
 582                                        rec->lrh_len,
 583                                        PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
 584                                        rec->lrh_index, index);
 585
 586                                 GOTO(out, rc = -EINVAL);
 587                         }
 588
 589                         if (rec->lrh_index < index) {
 590                                 CDEBUG(D_OTHER, "skipping lrh_index %d\n",
 591                                        rec->lrh_index);
 592                                 continue;
 593                         }
 594
 595                         if (rec->lrh_index != index) {
 596                                 CERROR("%s: "DFID" Invalid record: index %u"
 597                                        " but expected %u\n",
 598                                        loghandle->lgh_ctxt->loc_obd->obd_name,
 599                                        PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
 600                                        rec->lrh_index, index);
 601                                 GOTO(out, rc = -ERANGE);
 602                         }
 603
 604                         CDEBUG(D_OTHER,
 605                                "lrh_index: %d lrh_len: %d (%d remains)\n",
 606                                rec->lrh_index, rec->lrh_len,
 607                                (int)(buf + chunk_size - (char *)rec));
 608
 609                         /* lgh_cur_offset is used only at llog_test_3 */
 610                         loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
 611                                                     chunk_offset;
 612
 613                         /* if set, process the callback on this record */
 614                         if (ext2_test_bit(index, LLOG_HDR_BITMAP(llh))) {
 615                                 struct llog_cookie *lgc;
 616                                 __u64   tmp_off;
 617                                 int     tmp_idx;
 618                         /* the bitmap could be changed during processing
 619                          * records from the chunk. For wrapped catalog
 620                          * it means we can read deleted record and try to
 621                          * process it. Check this case and reread the chunk.
 622                          * Checking the race with llog_add the bit is set
 623                          * after incrementation of lgh_last_idx */
 624                                 if (index == lh_last_idx &&
 625                                     lh_last_idx !=
 626                                     (loghandle->lgh_last_idx + 1)) {
 627                                         /* save offset inside buffer for
 628                                          *  the re-read */
 629                                         buf_offset = (char *)rec - (char *)buf;
 630                                         cur_offset = chunk_offset;
 631                                         repeated = true;
 632                                         goto repeat;
 633
 634                                 }
 635                                 if (lti != NULL) {
 636                                         lgc = &lti->lgi_cookie;
 637                                         /* store lu_env for recursive calls */
 638                                         tmp_off = lgc->lgc_offset;
 639                                         tmp_idx = lgc->lgc_index;
 640
 641                                         lgc->lgc_offset = (char *)rec -
 642                                                 (char *)buf + chunk_offset;
 643                                         lgc->lgc_index = rec->lrh_index;
 644                                 }
 645                                 /* using lu_env for passing record offset to
 646                                  * llog_write through various callbacks */
 647                                 rc = lpi->lpi_cb(lpi->lpi_env, loghandle, rec,
 648                                                  lpi->lpi_cbdata);
 649                                 last_called_index = index;
 650
 651                                 if (lti != NULL) {
 652                                         lgc->lgc_offset = tmp_off;
 653                                         lgc->lgc_index = tmp_idx;
 654                                 }
 655
 656                                 if (rc == LLOG_PROC_BREAK) {
 657                                         GOTO(out, rc);
 658                                 } else if (rc == LLOG_DEL_RECORD) {
 659                                         rc = llog_cancel_rec(lpi->lpi_env,
 660                                                              loghandle,
 661                                                              rec->lrh_index);
 662                                 }
 663                                 if (rc)
 664                                         GOTO(out, rc);
 665                                 /* some stupid callbacks directly cancel records
 666                                  * and delete llog. Check it and stop
 667                                  * processing. */
 668                                 if (loghandle->lgh_hdr == NULL ||
 669                                     loghandle->lgh_hdr->llh_count == 1)
 670                                         GOTO(out, rc = 0);
 671                         }
 672                         /* exit if the last index is reached */
 673                         if (index >= last_index)
 674                                 GOTO(out, rc = 0);
 675                         ++index;
 676                 }
 677         }
 678
 679 out:
 680         if (cd != NULL)
 681                 cd->lpcd_last_idx = last_called_index;
 682
 683         if (unlikely(rc == -EIO && loghandle->lgh_obj != NULL)) {
 684                 if (dt_object_remote(loghandle->lgh_obj)) {
 685                         /* If it is remote object, then -EIO might means
 686                          * disconnection or eviction, let's return -EAGAIN,
 687                          * so for update recovery log processing, it will
 688                          * retry until the umount or abort recovery, see
 689                          * lod_sub_recovery_thread() */
 690                         CERROR("%s retry remote llog process\n",
 691                                loghandle->lgh_ctxt->loc_obd->obd_name);
 692                         rc = -EAGAIN;
 693                 } else {
 694                         /* something bad happened to the processing of a local
 695                          * llog file, probably I/O error or the log got
 696                          * corrupted to be able to finally release the log we
 697                          * discard any remaining bits in the header */
 698                         CERROR("%s: Local llog found corrupted #"DOSTID":%x"
 699                                " %s index %d count %d\n",
 700                                loghandle->lgh_ctxt->loc_obd->obd_name,
 701                                POSTID(&loghandle->lgh_id.lgl_oi),
 702                                loghandle->lgh_id.lgl_ogen,
 703                                ((llh->llh_flags & LLOG_F_IS_CAT) ? "catalog" :
 704                                 "plain"), index, llh->llh_count);
 705
 706                         while (index <= last_index) {
 707                                 if (ext2_test_bit(index,
 708                                                   LLOG_HDR_BITMAP(llh)) != 0)
 709                                         llog_cancel_rec(lpi->lpi_env, loghandle,
 710                                                         index);
 711                                 index++;
 712                         }
 713                         rc = 0;
 714                 }
 715         }
 716
 717         OBD_FREE_LARGE(buf, chunk_size);
 718         lpi->lpi_rc = rc;
 719         return 0;
 720 }
 721
 722 static int llog_process_thread_daemonize(void *arg)
 723 {
 724         struct llog_process_info        *lpi = arg;
 725         struct lu_env                    env;
 726         int                              rc;
 727         struct nsproxy                  *new_ns, *curr_ns = current->nsproxy;
 728
 729         task_lock(lpi->lpi_reftask);
 730         new_ns = lpi->lpi_reftask->nsproxy;
 731         if (curr_ns != new_ns) {
 732                 get_nsproxy(new_ns);
 733
 734                 current->nsproxy = new_ns;
 735                 /* XXX: we should call put_nsproxy() instead of
 736                  * atomic_dec(&ns->count) directly. But put_nsproxy() cannot be
 737                  * used outside of the kernel itself, because it calls
 738                  * free_nsproxy() which is not exported by the kernel
 739                  * (defined in kernel/nsproxy.c) */
 740                 atomic_dec(&curr_ns->count);
 741         }
 742         task_unlock(lpi->lpi_reftask);
 743
 744         unshare_fs_struct();
 745
 746         /* client env has no keys, tags is just 0 */
 747         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
 748         if (rc)
 749                 goto out;
 750         lpi->lpi_env = &env;
 751
 752         rc = llog_process_thread(arg);
 753
 754         lu_env_fini(&env);
 755 out:
 756         complete(&lpi->lpi_completion);
 757         return rc;
 758 }
 759
 760 int llog_process_or_fork(const struct lu_env *env,
 761                          struct llog_handle *loghandle,
 762                          llog_cb_t cb, void *data, void *catdata, bool fork)
 763 {
 764         struct llog_process_info *lpi;
 765         int                      rc;
 766
 767         ENTRY;
 768
 769         OBD_ALLOC_PTR(lpi);
 770         if (lpi == NULL) {
 771                 CERROR("cannot alloc pointer\n");
 772                 RETURN(-ENOMEM);
 773         }
 774         lpi->lpi_loghandle = loghandle;
 775         lpi->lpi_cb        = cb;
 776         lpi->lpi_cbdata    = data;
 777         lpi->lpi_catdata   = catdata;
 778
 779         if (fork) {
 780                 struct task_struct *task;
 781
 782                 /* The new thread can't use parent env,
 783                  * init the new one in llog_process_thread_daemonize. */
 784                 lpi->lpi_env = NULL;
 785                 init_completion(&lpi->lpi_completion);
 786                 /* take reference to current, so that
 787                  * llog_process_thread_daemonize() can use it to switch to
 788                  * namespace associated with current  */
 789                 lpi->lpi_reftask = current;
 790                 task = kthread_run(llog_process_thread_daemonize, lpi,
 791                                    "llog_process_thread");
 792                 if (IS_ERR(task)) {
 793                         rc = PTR_ERR(task);
 794                         CERROR("%s: cannot start thread: rc = %d\n",
 795                                loghandle->lgh_ctxt->loc_obd->obd_name, rc);
 796                         GOTO(out_lpi, rc);
 797                 }
 798                 wait_for_completion(&lpi->lpi_completion);
 799         } else {
 800                 lpi->lpi_env = env;
 801                 llog_process_thread(lpi);
 802         }
 803         rc = lpi->lpi_rc;
 804
 805 out_lpi:
 806         OBD_FREE_PTR(lpi);
 807         RETURN(rc);
 808 }
 809 EXPORT_SYMBOL(llog_process_or_fork);
 810
 811 int llog_process(const struct lu_env *env, struct llog_handle *loghandle,
 812                  llog_cb_t cb, void *data, void *catdata)
 813 {
 814         int rc;
 815         rc = llog_process_or_fork(env, loghandle, cb, data, catdata, true);
 816         return rc == LLOG_DEL_PLAIN ? 0 : rc;
 817 }
 818 EXPORT_SYMBOL(llog_process);
 819
 820 int llog_reverse_process(const struct lu_env *env,
 821                          struct llog_handle *loghandle, llog_cb_t cb,
 822                          void *data, void *catdata)
 823 {
 824         struct llog_log_hdr *llh = loghandle->lgh_hdr;
 825         struct llog_process_cat_data *cd = catdata;
 826         void *buf;
 827         int rc = 0, first_index = 1, index, idx;
 828         __u32   chunk_size = llh->llh_hdr.lrh_len;
 829         ENTRY;
 830
 831         OBD_ALLOC_LARGE(buf, chunk_size);
 832         if (buf == NULL)
 833                 RETURN(-ENOMEM);
 834
 835         if (cd != NULL)
 836                 first_index = cd->lpcd_first_idx + 1;
 837         if (cd != NULL && cd->lpcd_last_idx)
 838                 index = cd->lpcd_last_idx;
 839         else
 840                 index = LLOG_HDR_BITMAP_SIZE(llh) - 1;
 841
 842         while (rc == 0) {
 843                 struct llog_rec_hdr *rec;
 844                 struct llog_rec_tail *tail;
 845
 846                 /* skip records not set in bitmap */
 847                 while (index >= first_index &&
 848                        !ext2_test_bit(index, LLOG_HDR_BITMAP(llh)))
 849                         --index;
 850
 851                 LASSERT(index >= first_index - 1);
 852                 if (index == first_index - 1)
 853                         break;
 854
 855                 /* get the buf with our target record; avoid old garbage */
 856                 memset(buf, 0, chunk_size);
 857                 rc = llog_prev_block(env, loghandle, index, buf, chunk_size);
 858                 if (rc)
 859                         GOTO(out, rc);
 860
 861                 rec = buf;
 862                 idx = rec->lrh_index;
 863                 CDEBUG(D_RPCTRACE, "index %u : idx %u\n", index, idx);
 864                 while (idx < index) {
 865                         rec = (void *)rec + rec->lrh_len;
 866                         if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
 867                                 lustre_swab_llog_rec(rec);
 868                         idx ++;
 869                 }
 870                 LASSERT(idx == index);
 871                 tail = (void *)rec + rec->lrh_len - sizeof(*tail);
 872
 873                 /* process records in buffer, starting where we found one */
 874                 while ((void *)tail > buf) {
 875                         if (tail->lrt_index == 0)
 876                                 GOTO(out, rc = 0); /* no more records */
 877
 878                         /* if set, process the callback on this record */
 879                         if (ext2_test_bit(index, LLOG_HDR_BITMAP(llh))) {
 880                                 rec = (void *)tail - tail->lrt_len +
 881                                       sizeof(*tail);
 882
 883                                 rc = cb(env, loghandle, rec, data);
 884                                 if (rc == LLOG_PROC_BREAK) {
 885                                         GOTO(out, rc);
 886                                 } else if (rc == LLOG_DEL_RECORD) {
 887                                         rc = llog_cancel_rec(env, loghandle,
 888                                                              tail->lrt_index);
 889                                 }
 890                                 if (rc)
 891                                         GOTO(out, rc);
 892                         }
 893
 894                         /* previous record, still in buffer? */
 895                         --index;
 896                         if (index < first_index)
 897                                 GOTO(out, rc = 0);
 898                         tail = (void *)tail - tail->lrt_len;
 899                 }
 900         }
 901
 902 out:
 903         if (buf != NULL)
 904                 OBD_FREE_LARGE(buf, chunk_size);
 905         RETURN(rc);
 906 }
 907 EXPORT_SYMBOL(llog_reverse_process);
 908
 909 /**
 910  * new llog API
 911  *
 912  * API functions:
 913  *      llog_open - open llog, may not exist
 914  *      llog_exist - check if llog exists
 915  *      llog_close - close opened llog, pair for open, frees llog_handle
 916  *      llog_declare_create - declare llog creation
 917  *      llog_create - create new llog on disk, need transaction handle
 918  *      llog_declare_write_rec - declaration of llog write
 919  *      llog_write_rec - write llog record on disk, need transaction handle
 920  *      llog_declare_add - declare llog catalog record addition
 921  *      llog_add - add llog record in catalog, need transaction handle
 922  */
 923 int llog_exist(struct llog_handle *loghandle)
 924 {
 925         struct llog_operations  *lop;
 926         int                      rc;
 927
 928         ENTRY;
 929
 930         rc = llog_handle2ops(loghandle, &lop);
 931         if (rc)
 932                 RETURN(rc);
 933         if (lop->lop_exist == NULL)
 934                 RETURN(-EOPNOTSUPP);
 935
 936         rc = lop->lop_exist(loghandle);
 937         RETURN(rc);
 938 }
 939 EXPORT_SYMBOL(llog_exist);
 940
 941 int llog_declare_create(const struct lu_env *env,
 942                         struct llog_handle *loghandle, struct thandle *th)
 943 {
 944         struct llog_operations  *lop;
 945         int                      raised, rc;
 946
 947         ENTRY;
 948
 949         rc = llog_handle2ops(loghandle, &lop);
 950         if (rc)
 951                 RETURN(rc);
 952         if (lop->lop_declare_create == NULL)
 953                 RETURN(-EOPNOTSUPP);
 954
 955         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
 956         if (!raised)
 957                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
 958         rc = lop->lop_declare_create(env, loghandle, th);
 959         if (!raised)
 960                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
 961         RETURN(rc);
 962 }
 963
 964 int llog_create(const struct lu_env *env, struct llog_handle *handle,
 965                 struct thandle *th)
 966 {
 967         struct llog_operations  *lop;
 968         int                      raised, rc;
 969
 970         ENTRY;
 971
 972         rc = llog_handle2ops(handle, &lop);
 973         if (rc)
 974                 RETURN(rc);
 975         if (lop->lop_create == NULL)
 976                 RETURN(-EOPNOTSUPP);
 977
 978         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
 979         if (!raised)
 980                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
 981         rc = lop->lop_create(env, handle, th);
 982         if (!raised)
 983                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
 984         RETURN(rc);
 985 }
 986
 987 int llog_declare_write_rec(const struct lu_env *env,
 988                            struct llog_handle *handle,
 989                            struct llog_rec_hdr *rec, int idx,
 990                            struct thandle *th)
 991 {
 992         struct llog_operations  *lop;
 993         int                      raised, rc;
 994
 995         ENTRY;
 996
 997         rc = llog_handle2ops(handle, &lop);
 998         if (rc)
 999                 RETURN(rc);
1000         LASSERT(lop);
1001         if (lop->lop_declare_write_rec == NULL)
1002                 RETURN(-EOPNOTSUPP);
1003
1004         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
1005         if (!raised)
1006                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
1007         rc = lop->lop_declare_write_rec(env, handle, rec, idx, th);
1008         if (!raised)
1009                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
1010         RETURN(rc);
1011 }
1012
1013 int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
1014                    struct llog_rec_hdr *rec, struct llog_cookie *logcookies,
1015                    int idx, struct thandle *th)
1016 {
1017         struct llog_operations  *lop;
1018         int                      raised, rc, buflen;
1019
1020         ENTRY;
1021
1022         /* API sanity checks */
1023         if (handle == NULL) {
1024                 CERROR("loghandle is missed\n");
1025                 RETURN(-EPROTO);
1026         } else if (handle->lgh_obj == NULL) {
1027                 CERROR("loghandle %p with NULL object\n",
1028                         handle);
1029                 RETURN(-EPROTO);
1030         } else if (th == NULL) {
1031                 CERROR("%s: missed transaction handle\n",
1032                         handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name);
1033                 RETURN(-EPROTO);
1034         } else if (handle->lgh_hdr == NULL) {
1035                 CERROR("%s: loghandle %p with no header\n",
1036                         handle->lgh_obj->do_lu.lo_dev->ld_obd->obd_name,
1037                         handle);
1038                 RETURN(-EPROTO);
1039         }
1040
1041         rc = llog_handle2ops(handle, &lop);
1042         if (rc)
1043                 RETURN(rc);
1044
1045         if (lop->lop_write_rec == NULL)
1046                 RETURN(-EOPNOTSUPP);
1047
1048         buflen = rec->lrh_len;
1049         LASSERT(cfs_size_round(buflen) == buflen);
1050
1051         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
1052         if (!raised)
1053                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
1054         rc = lop->lop_write_rec(env, handle, rec, logcookies, idx, th);
1055         if (!raised)
1056                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
1057         RETURN(rc);
1058 }
1059
1060 int llog_add(const struct lu_env *env, struct llog_handle *lgh,
1061              struct llog_rec_hdr *rec, struct llog_cookie *logcookies,
1062              struct thandle *th)
1063 {
1064         int raised, rc;
1065
1066         ENTRY;
1067
1068         if (lgh->lgh_logops->lop_add == NULL)
1069                 RETURN(-EOPNOTSUPP);
1070
1071         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
1072         if (!raised)
1073                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
1074         rc = lgh->lgh_logops->lop_add(env, lgh, rec, logcookies, th);
1075         if (!raised)
1076                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
1077         RETURN(rc);
1078 }
1079 EXPORT_SYMBOL(llog_add);
1080
1081 int llog_declare_add(const struct lu_env *env, struct llog_handle *lgh,
1082                      struct llog_rec_hdr *rec, struct thandle *th)
1083 {
1084         int raised, rc;
1085
1086         ENTRY;
1087
1088         if (lgh->lgh_logops->lop_declare_add == NULL)
1089                 RETURN(-EOPNOTSUPP);
1090
1091         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
1092         if (!raised)
1093                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
1094         rc = lgh->lgh_logops->lop_declare_add(env, lgh, rec, th);
1095         if (!raised)
1096                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
1097         RETURN(rc);
1098 }
1099 EXPORT_SYMBOL(llog_declare_add);
1100
1101 /**
1102  * Helper function to open llog or create it if doesn't exist.
1103  * It hides all transaction handling from caller.
1104  */
1105 int llog_open_create(const struct lu_env *env, struct llog_ctxt *ctxt,
1106                      struct llog_handle **res, struct llog_logid *logid,
1107                      char *name)
1108 {
1109         struct dt_device        *d;
1110         struct thandle          *th;
1111         int                      rc;
1112
1113         ENTRY;
1114
1115         rc = llog_open(env, ctxt, res, logid, name, LLOG_OPEN_NEW);
1116         if (rc)
1117                 RETURN(rc);
1118
1119         if (llog_exist(*res))
1120                 RETURN(0);
1121
1122         LASSERT((*res)->lgh_obj != NULL);
1123
1124         d = lu2dt_dev((*res)->lgh_obj->do_lu.lo_dev);
1125
1126         if (unlikely(unlikely(d->dd_rdonly)))
1127                 RETURN(-EROFS);
1128
1129         th = dt_trans_create(env, d);
1130         if (IS_ERR(th))
1131                 GOTO(out, rc = PTR_ERR(th));
1132
1133         /* Create update llog object synchronously, which
1134          * happens during inialization process see
1135          * lod_sub_prep_llog(), to make sure the update
1136          * llog object is created before corss-MDT writing
1137          * updates into the llog object */
1138         if (ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID)
1139                 th->th_sync = 1;
1140
1141         th->th_wait_submit = 1;
1142         rc = llog_declare_create(env, *res, th);
1143         if (rc == 0) {
1144                 rc = dt_trans_start_local(env, d, th);
1145                 if (rc == 0)
1146                         rc = llog_create(env, *res, th);
1147         }
1148         dt_trans_stop(env, d, th);
1149 out:
1150         if (rc)
1151                 llog_close(env, *res);
1152         RETURN(rc);
1153 }
1154 EXPORT_SYMBOL(llog_open_create);
1155
1156 /**
1157  * Helper function to delete existent llog.
1158  */
1159 int llog_erase(const struct lu_env *env, struct llog_ctxt *ctxt,
1160                struct llog_logid *logid, char *name)
1161 {
1162         struct llog_handle      *handle;
1163         int                      rc = 0, rc2;
1164
1165         ENTRY;
1166
1167         /* nothing to erase */
1168         if (name == NULL && logid == NULL)
1169                 RETURN(0);
1170
1171         rc = llog_open(env, ctxt, &handle, logid, name, LLOG_OPEN_EXISTS);
1172         if (rc < 0)
1173                 RETURN(rc);
1174
1175         rc = llog_init_handle(env, handle, LLOG_F_IS_PLAIN, NULL);
1176         if (rc == 0)
1177                 rc = llog_destroy(env, handle);
1178
1179         rc2 = llog_close(env, handle);
1180         if (rc == 0)
1181                 rc = rc2;
1182         RETURN(rc);
1183 }
1184 EXPORT_SYMBOL(llog_erase);
1185
1186 /*
1187  * Helper function for write record in llog.
1188  * It hides all transaction handling from caller.
1189  * Valid only with local llog.
1190  */
1191 int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
1192                struct llog_rec_hdr *rec, int idx)
1193 {
1194         struct dt_device        *dt;
1195         struct thandle          *th;
1196         bool                    need_cookie;
1197         int                     rc;
1198
1199         ENTRY;
1200
1201         LASSERT(loghandle);
1202         LASSERT(loghandle->lgh_ctxt);
1203         LASSERT(loghandle->lgh_obj != NULL);
1204
1205         dt = lu2dt_dev(loghandle->lgh_obj->do_lu.lo_dev);
1206
1207         if (unlikely(unlikely(dt->dd_rdonly)))
1208                 RETURN(-EROFS);
1209
1210         th = dt_trans_create(env, dt);
1211         if (IS_ERR(th))
1212                 RETURN(PTR_ERR(th));
1213
1214         rc = llog_declare_write_rec(env, loghandle, rec, idx, th);
1215         if (rc)
1216                 GOTO(out_trans, rc);
1217
1218         th->th_wait_submit = 1;
1219         rc = dt_trans_start_local(env, dt, th);
1220         if (rc)
1221                 GOTO(out_trans, rc);
1222
1223         need_cookie = !(idx == LLOG_HEADER_IDX || idx == LLOG_NEXT_IDX);
1224
1225         down_write(&loghandle->lgh_lock);
1226         if (need_cookie) {
1227                 struct llog_thread_info *lti = llog_info(env);
1228
1229                 /* cookie comes from llog_process_thread */
1230                 rc = llog_write_rec(env, loghandle, rec, &lti->lgi_cookie,
1231                                     rec->lrh_index, th);
1232                 /* upper layer didn`t pass cookie so change rc */
1233                 rc = (rc == 1 ? 0 : rc);
1234         } else {
1235                 rc = llog_write_rec(env, loghandle, rec, NULL, idx, th);
1236         }
1237
1238         up_write(&loghandle->lgh_lock);
1239 out_trans:
1240         dt_trans_stop(env, dt, th);
1241         RETURN(rc);
1242 }
1243 EXPORT_SYMBOL(llog_write);
1244
1245 int llog_open(const struct lu_env *env, struct llog_ctxt *ctxt,
1246               struct llog_handle **lgh, struct llog_logid *logid,
1247               char *name, enum llog_open_param open_param)
1248 {
1249         int      raised;
1250         int      rc;
1251
1252         ENTRY;
1253
1254         LASSERT(ctxt);
1255         LASSERT(ctxt->loc_logops);
1256
1257         if (ctxt->loc_logops->lop_open == NULL) {
1258                 *lgh = NULL;
1259                 RETURN(-EOPNOTSUPP);
1260         }
1261
1262         *lgh = llog_alloc_handle();
1263         if (*lgh == NULL)
1264                 RETURN(-ENOMEM);
1265         (*lgh)->lgh_ctxt = ctxt;
1266         (*lgh)->lgh_logops = ctxt->loc_logops;
1267
1268         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
1269         if (!raised)
1270                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
1271         rc = ctxt->loc_logops->lop_open(env, *lgh, logid, name, open_param);
1272         if (!raised)
1273                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
1274         if (rc) {
1275                 llog_free_handle(*lgh);
1276                 *lgh = NULL;
1277         }
1278         RETURN(rc);
1279 }
1280 EXPORT_SYMBOL(llog_open);
1281
1282 int llog_close(const struct lu_env *env, struct llog_handle *loghandle)
1283 {
1284         struct llog_operations  *lop;
1285         int                      rc;
1286
1287         ENTRY;
1288
1289         rc = llog_handle2ops(loghandle, &lop);
1290         if (rc)
1291                 GOTO(out, rc);
1292         if (lop->lop_close == NULL)
1293                 GOTO(out, rc = -EOPNOTSUPP);
1294         rc = lop->lop_close(env, loghandle);
1295 out:
1296         llog_handle_put(loghandle);
1297         RETURN(rc);
1298 }
1299 EXPORT_SYMBOL(llog_close);
1300
1301 /**
1302  * Helper function to get the llog size in records. It is used by MGS
1303  * mostly to check that config llog exists and contains data.
1304  *
1305  * \param[in] env       execution environment
1306  * \param[in] ctxt      llog context
1307  * \param[in] name      llog name
1308  *
1309  * \retval              true if there are records in llog besides a header
1310  * \retval              false on error or llog without records
1311  */
1312 int llog_is_empty(const struct lu_env *env, struct llog_ctxt *ctxt,
1313                   char *name)
1314 {
1315         struct llog_handle      *llh;
1316         int                      rc = 0;
1317
1318         rc = llog_open(env, ctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
1319         if (rc < 0) {
1320                 if (likely(rc == -ENOENT))
1321                         rc = 0;
1322                 GOTO(out, rc);
1323         }
1324
1325         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
1326         if (rc)
1327                 GOTO(out_close, rc);
1328         rc = llog_get_size(llh);
1329
1330 out_close:
1331         llog_close(env, llh);
1332 out:
1333         /* The header is record 1, the llog is still considered as empty
1334          * if there is only header */
1335         return (rc <= 1);
1336 }
1337 EXPORT_SYMBOL(llog_is_empty);
1338
1339 int llog_copy_handler(const struct lu_env *env, struct llog_handle *llh,
1340                       struct llog_rec_hdr *rec, void *data)
1341 {
1342         struct llog_handle      *copy_llh = data;
1343
1344         /* Append all records */
1345         return llog_write(env, copy_llh, rec, LLOG_NEXT_IDX);
1346 }
1347
1348 /* backup plain llog */
1349 int llog_backup(const struct lu_env *env, struct obd_device *obd,
1350                 struct llog_ctxt *ctxt, struct llog_ctxt *bctxt,
1351                 char *name, char *backup)
1352 {
1353         struct llog_handle      *llh, *bllh;
1354         int                      rc;
1355
1356         ENTRY;
1357
1358         /* open original log */
1359         rc = llog_open(env, ctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
1360         if (rc < 0) {
1361                 /* the -ENOENT case is also reported to the caller
1362                  * but silently so it should handle that if needed.
1363                  */
1364                 if (rc != -ENOENT)
1365                         CERROR("%s: failed to open log %s: rc = %d\n",
1366                                obd->obd_name, name, rc);
1367                 RETURN(rc);
1368         }
1369
1370         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
1371         if (rc)
1372                 GOTO(out_close, rc);
1373
1374         /* Make sure there's no old backup log */
1375         rc = llog_erase(env, bctxt, NULL, backup);
1376         if (rc < 0 && rc != -ENOENT)
1377                 GOTO(out_close, rc);
1378
1379         /* open backup log */
1380         rc = llog_open_create(env, bctxt, &bllh, NULL, backup);
1381         if (rc) {
1382                 CERROR("%s: failed to open backup logfile %s: rc = %d\n",
1383                        obd->obd_name, backup, rc);
1384                 GOTO(out_close, rc);
1385         }
1386
1387         /* check that backup llog is not the same object as original one */
1388         if (llh->lgh_obj == bllh->lgh_obj) {
1389                 CERROR("%s: backup llog %s to itself (%s), objects %p/%p\n",
1390                        obd->obd_name, name, backup, llh->lgh_obj,
1391                        bllh->lgh_obj);
1392                 GOTO(out_backup, rc = -EEXIST);
1393         }
1394
1395         rc = llog_init_handle(env, bllh, LLOG_F_IS_PLAIN, NULL);
1396         if (rc)
1397                 GOTO(out_backup, rc);
1398
1399         /* Copy log record by record */
1400         rc = llog_process_or_fork(env, llh, llog_copy_handler, (void *)bllh,
1401                                   NULL, false);
1402         if (rc)
1403                 CERROR("%s: failed to backup log %s: rc = %d\n",
1404                        obd->obd_name, name, rc);
1405 out_backup:
1406         llog_close(env, bllh);
1407 out_close:
1408         llog_close(env, llh);
1409         RETURN(rc);
1410 }
1411 EXPORT_SYMBOL(llog_backup);
1412
1413 /* Get size of llog */
1414 __u64 llog_size(const struct lu_env *env, struct llog_handle *llh)
1415 {
1416         int rc;
1417         struct lu_attr la;
1418
1419         rc = llh->lgh_obj->do_ops->do_attr_get(env, llh->lgh_obj, &la);
1420         if (rc) {
1421                 CERROR("%s: attr_get failed for "DFID": rc = %d\n",
1422                        llh->lgh_ctxt->loc_obd->obd_name,
1423                        PFID(&llh->lgh_id.lgl_oi.oi_fid), rc);
1424                 return 0;
1425         }
1426
1427         return la.la_size;
1428 }
1429 EXPORT_SYMBOL(llog_size);
1430