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