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