lustre/obdclass/llog_cat.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2015, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/obdclass/llog_cat.c
  33  *
  34  * OST<->MDS recovery logging infrastructure.
  35  *
  36  * Invariants in implementation:
  37  * - we do not share logs among different OST<->MDS connections, so that
  38  *   if an OST or MDS fails it need only look at log(s) relevant to itself
  39  *
  40  * Author: Andreas Dilger <adilger@clusterfs.com>
  41  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
  42  * Author: Mikhail Pershin <mike.pershin@intel.com>
  43  */
  44
  45 #define DEBUG_SUBSYSTEM S_LOG
  46
  47
  48 #include <obd_class.h>
  49
  50 #include "llog_internal.h"
  51
  52 /* Create a new log handle and add it to the open list.
  53  * This log handle will be closed when all of the records in it are removed.
  54  *
  55  * Assumes caller has already pushed us into the kernel context and is locking.
  56  */
  57 static int llog_cat_new_log(const struct lu_env *env,
  58                             struct llog_handle *cathandle,
  59                             struct llog_handle *loghandle,
  60                             struct thandle *th)
  61 {
  62         struct llog_thread_info *lgi = llog_info(env);
  63         struct llog_logid_rec   *rec = &lgi->lgi_logid;
  64         struct thandle *handle = NULL;
  65         struct dt_device *dt = NULL;
  66         struct llog_log_hdr     *llh = cathandle->lgh_hdr;
  67         int                      rc, index;
  68
  69         ENTRY;
  70
  71         index = (cathandle->lgh_last_idx + 1) %
  72                 (OBD_FAIL_PRECHECK(OBD_FAIL_CAT_RECORDS) ? (cfs_fail_val + 1) :
  73                                                 LLOG_HDR_BITMAP_SIZE(llh));
  74
  75         /* check that new llog index will not overlap with the first one.
  76          * - llh_cat_idx is the index just before the first/oldest still in-use
  77          *      index in catalog
  78          * - lgh_last_idx is the last/newest used index in catalog
  79          *
  80          * When catalog is not wrapped yet then lgh_last_idx is always larger
  81          * than llh_cat_idx. After the wrap around lgh_last_idx re-starts
  82          * from 0 and llh_cat_idx becomes the upper limit for it
  83          *
  84          * Check if catalog has already wrapped around or not by comparing
  85          * last_idx and cat_idx */
  86         if ((index == llh->llh_cat_idx + 1 && llh->llh_count > 1) ||
  87             (index == 0 && llh->llh_cat_idx == 0)) {
  88                 CWARN("%s: there are no more free slots in catalog\n",
  89                       loghandle->lgh_ctxt->loc_obd->obd_name);
  90                 RETURN(-ENOSPC);
  91         }
  92
  93         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED))
  94                 RETURN(-ENOSPC);
  95
  96         if (loghandle->lgh_hdr != NULL) {
  97                 /* If llog object is remote and creation is failed, lgh_hdr
  98                  * might be left over here, free it first */
  99                 LASSERT(!llog_exist(loghandle));
 100                 OBD_FREE_LARGE(loghandle->lgh_hdr, loghandle->lgh_hdr_size);
 101                 loghandle->lgh_hdr = NULL;
 102         }
 103
 104         if (th == NULL) {
 105                 dt = lu2dt_dev(cathandle->lgh_obj->do_lu.lo_dev);
 106
 107                 handle = dt_trans_create(env, dt);
 108                 if (IS_ERR(handle))
 109                         RETURN(PTR_ERR(handle));
 110
 111                 /* Create update llog object synchronously, which
 112                  * happens during inialization process see
 113                  * lod_sub_prep_llog(), to make sure the update
 114                  * llog object is created before corss-MDT writing
 115                  * updates into the llog object */
 116                 if (cathandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID)
 117                         handle->th_sync = 1;
 118
 119                 handle->th_wait_submit = 1;
 120
 121                 rc = llog_declare_create(env, loghandle, handle);
 122                 if (rc != 0)
 123                         GOTO(out, rc);
 124
 125                 rec->lid_hdr.lrh_len = sizeof(*rec);
 126                 rec->lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
 127                 rec->lid_id = loghandle->lgh_id;
 128                 rc = llog_declare_write_rec(env, cathandle, &rec->lid_hdr, -1,
 129                                             handle);
 130                 if (rc != 0)
 131                         GOTO(out, rc);
 132
 133                 rc = dt_trans_start_local(env, dt, handle);
 134                 if (rc != 0)
 135                         GOTO(out, rc);
 136
 137                 th = handle;
 138         }
 139
 140         rc = llog_create(env, loghandle, th);
 141         /* if llog is already created, no need to initialize it */
 142         if (rc == -EEXIST) {
 143                 GOTO(out, rc = 0);
 144         } else if (rc != 0) {
 145                 CERROR("%s: can't create new plain llog in catalog: rc = %d\n",
 146                        loghandle->lgh_ctxt->loc_obd->obd_name, rc);
 147                 GOTO(out, rc);
 148         }
 149
 150         rc = llog_init_handle(env, loghandle,
 151                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
 152                               &cathandle->lgh_hdr->llh_tgtuuid);
 153         if (rc < 0)
 154                 GOTO(out, rc);
 155
 156         /* build the record for this log in the catalog */
 157         rec->lid_hdr.lrh_len = sizeof(*rec);
 158         rec->lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
 159         rec->lid_id = loghandle->lgh_id;
 160
 161         /* append the new record into catalog. The new index will be
 162          * assigned to the record and updated in rec header */
 163         rc = llog_write_rec(env, cathandle, &rec->lid_hdr,
 164                             &loghandle->u.phd.phd_cookie, LLOG_NEXT_IDX, th);
 165         if (rc < 0)
 166                 GOTO(out_destroy, rc);
 167
 168         CDEBUG(D_OTHER, "new plain log "DOSTID":%x for index %u of catalog"
 169                DOSTID"\n", POSTID(&loghandle->lgh_id.lgl_oi),
 170                loghandle->lgh_id.lgl_ogen, rec->lid_hdr.lrh_index,
 171                POSTID(&cathandle->lgh_id.lgl_oi));
 172
 173         loghandle->lgh_hdr->llh_cat_idx = rec->lid_hdr.lrh_index;
 174 out:
 175         if (handle != NULL) {
 176                 handle->th_result = rc >= 0 ? 0 : rc;
 177                 dt_trans_stop(env, dt, handle);
 178         }
 179         RETURN(rc);
 180
 181 out_destroy:
 182         /* to signal llog_cat_close() it shouldn't try to destroy the llog,
 183          * we want to destroy it in this transaction, otherwise the object
 184          * becomes an orphan */
 185         loghandle->lgh_hdr->llh_flags &= ~LLOG_F_ZAP_WHEN_EMPTY;
 186         /* this is to mimic full log, so another llog_cat_current_log()
 187          * can skip it and ask for another onet */
 188         loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) + 1;
 189         llog_trans_destroy(env, loghandle, th);
 190         RETURN(rc);
 191 }
 192
 193 /* Open an existent log handle and add it to the open list.
 194  * This log handle will be closed when all of the records in it are removed.
 195  *
 196  * Assumes caller has already pushed us into the kernel context and is locking.
 197  * We return a lock on the handle to ensure nobody yanks it from us.
 198  *
 199  * This takes extra reference on llog_handle via llog_handle_get() and require
 200  * this reference to be put by caller using llog_handle_put()
 201  */
 202 int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
 203                        struct llog_handle **res, struct llog_logid *logid)
 204 {
 205         struct llog_handle      *loghandle;
 206         enum llog_flag           fmt;
 207         int                      rc = 0;
 208
 209         ENTRY;
 210
 211         if (cathandle == NULL)
 212                 RETURN(-EBADF);
 213
 214         fmt = cathandle->lgh_hdr->llh_flags & LLOG_F_EXT_MASK;
 215         down_write(&cathandle->lgh_lock);
 216         list_for_each_entry(loghandle, &cathandle->u.chd.chd_head,
 217                             u.phd.phd_entry) {
 218                 struct llog_logid *cgl = &loghandle->lgh_id;
 219
 220                 if (ostid_id(&cgl->lgl_oi) == ostid_id(&logid->lgl_oi) &&
 221                     ostid_seq(&cgl->lgl_oi) == ostid_seq(&logid->lgl_oi)) {
 222                         if (cgl->lgl_ogen != logid->lgl_ogen) {
 223                                 CERROR("%s: log "DOSTID" generation %x != %x\n",
 224                                        loghandle->lgh_ctxt->loc_obd->obd_name,
 225                                        POSTID(&logid->lgl_oi), cgl->lgl_ogen,
 226                                        logid->lgl_ogen);
 227                                 continue;
 228                         }
 229                         loghandle->u.phd.phd_cat_handle = cathandle;
 230                         up_write(&cathandle->lgh_lock);
 231                         GOTO(out, rc = 0);
 232                 }
 233         }
 234         up_write(&cathandle->lgh_lock);
 235
 236         rc = llog_open(env, cathandle->lgh_ctxt, &loghandle, logid, NULL,
 237                        LLOG_OPEN_EXISTS);
 238         if (rc < 0) {
 239                 CERROR("%s: error opening log id "DOSTID":%x: rc = %d\n",
 240                        cathandle->lgh_ctxt->loc_obd->obd_name,
 241                        POSTID(&logid->lgl_oi), logid->lgl_ogen, rc);
 242                 RETURN(rc);
 243         }
 244
 245         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN | fmt, NULL);
 246         if (rc < 0) {
 247                 llog_close(env, loghandle);
 248                 loghandle = NULL;
 249                 RETURN(rc);
 250         }
 251
 252         down_write(&cathandle->lgh_lock);
 253         list_add_tail(&loghandle->u.phd.phd_entry, &cathandle->u.chd.chd_head);
 254         up_write(&cathandle->lgh_lock);
 255
 256         loghandle->u.phd.phd_cat_handle = cathandle;
 257         loghandle->u.phd.phd_cookie.lgc_lgl = cathandle->lgh_id;
 258         loghandle->u.phd.phd_cookie.lgc_index =
 259                                 loghandle->lgh_hdr->llh_cat_idx;
 260         EXIT;
 261 out:
 262         llog_handle_get(loghandle);
 263         *res = loghandle;
 264         return 0;
 265 }
 266
 267 int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle)
 268 {
 269         struct llog_handle      *loghandle, *n;
 270         int                      rc;
 271
 272         ENTRY;
 273
 274         list_for_each_entry_safe(loghandle, n, &cathandle->u.chd.chd_head,
 275                                  u.phd.phd_entry) {
 276                 struct llog_log_hdr     *llh = loghandle->lgh_hdr;
 277                 int                      index;
 278
 279                 /* unlink open-not-created llogs */
 280                 list_del_init(&loghandle->u.phd.phd_entry);
 281                 llh = loghandle->lgh_hdr;
 282                 if (loghandle->lgh_obj != NULL && llh != NULL &&
 283                     (llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
 284                     (llh->llh_count == 1)) {
 285                         rc = llog_destroy(env, loghandle);
 286                         if (rc)
 287                                 CERROR("%s: failure destroying log during "
 288                                        "cleanup: rc = %d\n",
 289                                        loghandle->lgh_ctxt->loc_obd->obd_name,
 290                                        rc);
 291
 292                         index = loghandle->u.phd.phd_cookie.lgc_index;
 293                         llog_cat_cleanup(env, cathandle, NULL, index);
 294                 }
 295                 llog_close(env, loghandle);
 296         }
 297         /* if handle was stored in ctxt, remove it too */
 298         if (cathandle->lgh_ctxt->loc_handle == cathandle)
 299                 cathandle->lgh_ctxt->loc_handle = NULL;
 300         rc = llog_close(env, cathandle);
 301         RETURN(rc);
 302 }
 303 EXPORT_SYMBOL(llog_cat_close);
 304
 305 /**
 306  * lockdep markers for nested struct llog_handle::lgh_lock locking.
 307  */
 308 enum {
 309         LLOGH_CAT,
 310         LLOGH_LOG
 311 };
 312
 313 /** Return the currently active log handle.  If the current log handle doesn't
 314  * have enough space left for the current record, start a new one.
 315  *
 316  * If reclen is 0, we only want to know what the currently active log is,
 317  * otherwise we get a lock on this log so nobody can steal our space.
 318  *
 319  * Assumes caller has already pushed us into the kernel context and is locking.
 320  *
 321  * NOTE: loghandle is write-locked upon successful return
 322  */
 323 static struct llog_handle *llog_cat_current_log(struct llog_handle *cathandle,
 324                                                 struct thandle *th)
 325 {
 326         struct llog_handle *loghandle = NULL;
 327         ENTRY;
 328
 329
 330         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED2)) {
 331                 down_write_nested(&cathandle->lgh_lock, LLOGH_CAT);
 332                 GOTO(next, loghandle);
 333         }
 334
 335         down_read_nested(&cathandle->lgh_lock, LLOGH_CAT);
 336         loghandle = cathandle->u.chd.chd_current_log;
 337         if (loghandle) {
 338                 struct llog_log_hdr *llh;
 339
 340                 down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
 341                 llh = loghandle->lgh_hdr;
 342                 if (llh == NULL || !llog_is_full(loghandle)) {
 343                         up_read(&cathandle->lgh_lock);
 344                         RETURN(loghandle);
 345                 } else {
 346                         up_write(&loghandle->lgh_lock);
 347                 }
 348         }
 349         up_read(&cathandle->lgh_lock);
 350
 351         /* time to use next log */
 352
 353         /* first, we have to make sure the state hasn't changed */
 354         down_write_nested(&cathandle->lgh_lock, LLOGH_CAT);
 355         loghandle = cathandle->u.chd.chd_current_log;
 356         if (loghandle) {
 357                 struct llog_log_hdr *llh;
 358
 359                 down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
 360                 llh = loghandle->lgh_hdr;
 361                 LASSERT(llh);
 362                 if (!llog_is_full(loghandle))
 363                         GOTO(out_unlock, loghandle);
 364                 else
 365                         up_write(&loghandle->lgh_lock);
 366         }
 367
 368 next:
 369         /* Sigh, the chd_next_log and chd_current_log is initialized
 370          * in declare phase, and we do not serialize the catlog
 371          * accessing, so it might be possible the llog creation
 372          * thread (see llog_cat_declare_add_rec()) did not create
 373          * llog successfully, then the following thread might
 374          * meet this situation. */
 375         if (IS_ERR_OR_NULL(cathandle->u.chd.chd_next_log)) {
 376                 CERROR("%s: next log does not exist!\n",
 377                        cathandle->lgh_ctxt->loc_obd->obd_name);
 378                 loghandle = ERR_PTR(-EIO);
 379                 if (cathandle->u.chd.chd_next_log == NULL) {
 380                         /* Store the error in chd_next_log, so
 381                          * the following process can get correct
 382                          * failure value */
 383                         cathandle->u.chd.chd_next_log = loghandle;
 384                 }
 385                 GOTO(out_unlock, loghandle);
 386         }
 387
 388         CDEBUG(D_INODE, "use next log\n");
 389
 390         loghandle = cathandle->u.chd.chd_next_log;
 391         cathandle->u.chd.chd_current_log = loghandle;
 392         cathandle->u.chd.chd_next_log = NULL;
 393         down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
 394
 395 out_unlock:
 396         up_write(&cathandle->lgh_lock);
 397         LASSERT(loghandle);
 398         RETURN(loghandle);
 399 }
 400
 401 static int llog_cat_update_header(const struct lu_env *env,
 402                            struct llog_handle *cathandle)
 403 {
 404         struct llog_handle *loghandle;
 405         int rc;
 406         ENTRY;
 407
 408         /* refresh llog */
 409         down_write(&cathandle->lgh_lock);
 410         if (!cathandle->lgh_stale) {
 411                 up_write(&cathandle->lgh_lock);
 412                 RETURN(0);
 413         }
 414         list_for_each_entry(loghandle, &cathandle->u.chd.chd_head,
 415                             u.phd.phd_entry) {
 416                 if (!llog_exist(loghandle))
 417                         continue;
 418
 419                 rc = llog_read_header(env, loghandle, NULL);
 420                 if (rc != 0) {
 421                         up_write(&cathandle->lgh_lock);
 422                         GOTO(out, rc);
 423                 }
 424         }
 425         rc = llog_read_header(env, cathandle, NULL);
 426         if (rc == 0)
 427                 cathandle->lgh_stale = 0;
 428         up_write(&cathandle->lgh_lock);
 429         if (rc != 0)
 430                 GOTO(out, rc);
 431 out:
 432         RETURN(rc);
 433 }
 434
 435 /* Add a single record to the recovery log(s) using a catalog
 436  * Returns as llog_write_record
 437  *
 438  * Assumes caller has already pushed us into the kernel context.
 439  */
 440 int llog_cat_add_rec(const struct lu_env *env, struct llog_handle *cathandle,
 441                      struct llog_rec_hdr *rec, struct llog_cookie *reccookie,
 442                      struct thandle *th)
 443 {
 444         struct llog_handle *loghandle;
 445         int rc, retried = 0;
 446         ENTRY;
 447
 448         LASSERT(rec->lrh_len <= cathandle->lgh_ctxt->loc_chunk_size);
 449
 450 retry:
 451         loghandle = llog_cat_current_log(cathandle, th);
 452         if (IS_ERR(loghandle))
 453                 RETURN(PTR_ERR(loghandle));
 454
 455         /* loghandle is already locked by llog_cat_current_log() for us */
 456         if (!llog_exist(loghandle)) {
 457                 rc = llog_cat_new_log(env, cathandle, loghandle, th);
 458                 if (rc < 0) {
 459                         up_write(&loghandle->lgh_lock);
 460                         /* nobody should be trying to use this llog */
 461                         down_write(&cathandle->lgh_lock);
 462                         if (cathandle->u.chd.chd_current_log == loghandle)
 463                                 cathandle->u.chd.chd_current_log = NULL;
 464                         up_write(&cathandle->lgh_lock);
 465                         RETURN(rc);
 466                 }
 467         }
 468         /* now let's try to add the record */
 469         rc = llog_write_rec(env, loghandle, rec, reccookie, LLOG_NEXT_IDX, th);
 470         if (rc < 0) {
 471                 CDEBUG_LIMIT(rc == -ENOSPC ? D_HA : D_ERROR,
 472                              "llog_write_rec %d: lh=%p\n", rc, loghandle);
 473                 /* -ENOSPC is returned if no empty records left
 474                  * and when it's lack of space on the stogage.
 475                  * there is no point to try again if it's the second
 476                  * case. many callers (like llog test) expect ENOSPC,
 477                  * so we preserve this error code, but look for the
 478                  * actual cause here */
 479                 if (rc == -ENOSPC && llog_is_full(loghandle))
 480                         rc = -ENOBUFS;
 481         }
 482         up_write(&loghandle->lgh_lock);
 483
 484         if (rc == -ENOBUFS) {
 485                 if (retried++ == 0)
 486                         GOTO(retry, rc);
 487                 CERROR("%s: error on 2nd llog: rc = %d\n",
 488                        cathandle->lgh_ctxt->loc_obd->obd_name, rc);
 489         }
 490
 491         RETURN(rc);
 492 }
 493 EXPORT_SYMBOL(llog_cat_add_rec);
 494
 495 int llog_cat_declare_add_rec(const struct lu_env *env,
 496                              struct llog_handle *cathandle,
 497                              struct llog_rec_hdr *rec, struct thandle *th)
 498 {
 499         struct llog_thread_info *lgi = llog_info(env);
 500         struct llog_logid_rec   *lirec = &lgi->lgi_logid;
 501         struct llog_handle      *loghandle, *next;
 502         int                      rc = 0;
 503
 504         ENTRY;
 505
 506         if (cathandle->u.chd.chd_current_log == NULL) {
 507                 /* declare new plain llog */
 508                 down_write(&cathandle->lgh_lock);
 509                 if (cathandle->u.chd.chd_current_log == NULL) {
 510                         rc = llog_open(env, cathandle->lgh_ctxt, &loghandle,
 511                                        NULL, NULL, LLOG_OPEN_NEW);
 512                         if (rc == 0) {
 513                                 cathandle->u.chd.chd_current_log = loghandle;
 514                                 list_add_tail(&loghandle->u.phd.phd_entry,
 515                                               &cathandle->u.chd.chd_head);
 516                         }
 517                 }
 518                 up_write(&cathandle->lgh_lock);
 519         } else if (cathandle->u.chd.chd_next_log == NULL ||
 520                    IS_ERR(cathandle->u.chd.chd_next_log)) {
 521                 /* declare next plain llog */
 522                 down_write(&cathandle->lgh_lock);
 523                 if (cathandle->u.chd.chd_next_log == NULL ||
 524                     IS_ERR(cathandle->u.chd.chd_next_log)) {
 525                         rc = llog_open(env, cathandle->lgh_ctxt, &loghandle,
 526                                        NULL, NULL, LLOG_OPEN_NEW);
 527                         if (rc == 0) {
 528                                 cathandle->u.chd.chd_next_log = loghandle;
 529                                 list_add_tail(&loghandle->u.phd.phd_entry,
 530                                               &cathandle->u.chd.chd_head);
 531                         }
 532                 }
 533                 up_write(&cathandle->lgh_lock);
 534         }
 535         if (rc)
 536                 GOTO(out, rc);
 537
 538         lirec->lid_hdr.lrh_len = sizeof(*lirec);
 539
 540         if (!llog_exist(cathandle->u.chd.chd_current_log)) {
 541                 if (dt_object_remote(cathandle->lgh_obj)) {
 542                         /* For remote operation, if we put the llog object
 543                          * creation in the current transaction, then the
 544                          * llog object will not be created on the remote
 545                          * target until the transaction stop, if other
 546                          * operations start before the transaction stop,
 547                          * and use the same llog object, will be dependent
 548                          * on the success of this transaction. So let's
 549                          * create the llog object synchronously here to
 550                          * remove the dependency. */
 551 create_again:
 552                         down_read_nested(&cathandle->lgh_lock, LLOGH_CAT);
 553                         loghandle = cathandle->u.chd.chd_current_log;
 554                         down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
 555                         if (cathandle->lgh_stale) {
 556                                 up_write(&loghandle->lgh_lock);
 557                                 up_read(&cathandle->lgh_lock);
 558                                 GOTO(out, rc = -EIO);
 559                         }
 560                         if (!llog_exist(loghandle)) {
 561                                 rc = llog_cat_new_log(env, cathandle, loghandle,
 562                                                       NULL);
 563                                 if (rc == -ESTALE)
 564                                         cathandle->lgh_stale = 1;
 565                         }
 566                         up_write(&loghandle->lgh_lock);
 567                         up_read(&cathandle->lgh_lock);
 568                         if (rc == -ESTALE) {
 569                                 rc = llog_cat_update_header(env, cathandle);
 570                                 if (rc != 0)
 571                                         GOTO(out, rc);
 572                                 goto create_again;
 573                         } else if (rc < 0) {
 574                                 GOTO(out, rc);
 575                         }
 576                 } else {
 577                         rc = llog_declare_create(env,
 578                                         cathandle->u.chd.chd_current_log, th);
 579                         if (rc)
 580                                 GOTO(out, rc);
 581                         llog_declare_write_rec(env, cathandle,
 582                                                &lirec->lid_hdr, -1, th);
 583                 }
 584         }
 585
 586 write_again:
 587         /* declare records in the llogs */
 588         rc = llog_declare_write_rec(env, cathandle->u.chd.chd_current_log,
 589                                     rec, -1, th);
 590         if (rc == -ESTALE) {
 591                 down_write(&cathandle->lgh_lock);
 592                 if (cathandle->lgh_stale) {
 593                         up_write(&cathandle->lgh_lock);
 594                         GOTO(out, rc = -EIO);
 595                 }
 596
 597                 cathandle->lgh_stale = 1;
 598                 up_write(&cathandle->lgh_lock);
 599                 rc = llog_cat_update_header(env, cathandle);
 600                 if (rc != 0)
 601                         GOTO(out, rc);
 602                 goto write_again;
 603         } else if (rc < 0) {
 604                 GOTO(out, rc);
 605         }
 606
 607         next = cathandle->u.chd.chd_next_log;
 608         if (!IS_ERR_OR_NULL(next)) {
 609                 if (!llog_exist(next)) {
 610                         if (dt_object_remote(cathandle->lgh_obj)) {
 611                                 /* For remote operation, if we put the llog
 612                                  * object creation in the current transaction,
 613                                  * then the llog object will not be created on
 614                                  * the remote target until the transaction stop,
 615                                  * if other operations start before the
 616                                  * transaction stop, and use the same llog
 617                                  * object, will be dependent on the success of
 618                                  * this transaction. So let's create the llog
 619                                  * object synchronously here to remove the
 620                                  * dependency. */
 621                                 down_write_nested(&cathandle->lgh_lock,
 622                                                  LLOGH_CAT);
 623                                 next = cathandle->u.chd.chd_next_log;
 624                                 if (IS_ERR_OR_NULL(next)) {
 625                                         /* Sigh, another thread just tried,
 626                                          * let's fail as well */
 627                                         up_write(&cathandle->lgh_lock);
 628                                         if (next == NULL)
 629                                                 rc = -EIO;
 630                                         else
 631                                                 rc = PTR_ERR(next);
 632                                         GOTO(out, rc);
 633                                 }
 634
 635                                 down_write_nested(&next->lgh_lock, LLOGH_LOG);
 636                                 if (!llog_exist(next)) {
 637                                         rc = llog_cat_new_log(env, cathandle,
 638                                                               next, NULL);
 639                                         if (rc < 0)
 640                                                 cathandle->u.chd.chd_next_log =
 641                                                                 ERR_PTR(rc);
 642                                 }
 643                                 up_write(&next->lgh_lock);
 644                                 up_write(&cathandle->lgh_lock);
 645                                 if (rc < 0)
 646                                         GOTO(out, rc);
 647                         } else {
 648                                 rc = llog_declare_create(env, next, th);
 649                                 llog_declare_write_rec(env, cathandle,
 650                                                 &lirec->lid_hdr, -1, th);
 651                         }
 652                 }
 653                 /* XXX: we hope for declarations made for existing llog
 654                  *      this might be not correct with some backends
 655                  *      where declarations are expected against specific
 656                  *      object like ZFS with full debugging enabled */
 657                 /*llog_declare_write_rec(env, next, rec, -1, th);*/
 658         }
 659 out:
 660         RETURN(rc);
 661 }
 662 EXPORT_SYMBOL(llog_cat_declare_add_rec);
 663
 664 int llog_cat_add(const struct lu_env *env, struct llog_handle *cathandle,
 665                  struct llog_rec_hdr *rec, struct llog_cookie *reccookie)
 666 {
 667         struct llog_ctxt        *ctxt;
 668         struct dt_device        *dt;
 669         struct thandle          *th = NULL;
 670         int                      rc;
 671
 672         ctxt = cathandle->lgh_ctxt;
 673         LASSERT(ctxt);
 674         LASSERT(ctxt->loc_exp);
 675
 676         LASSERT(cathandle->lgh_obj != NULL);
 677         dt = lu2dt_dev(cathandle->lgh_obj->do_lu.lo_dev);
 678
 679         th = dt_trans_create(env, dt);
 680         if (IS_ERR(th))
 681                 RETURN(PTR_ERR(th));
 682
 683         rc = llog_cat_declare_add_rec(env, cathandle, rec, th);
 684         if (rc)
 685                 GOTO(out_trans, rc);
 686
 687         rc = dt_trans_start_local(env, dt, th);
 688         if (rc)
 689                 GOTO(out_trans, rc);
 690         rc = llog_cat_add_rec(env, cathandle, rec, reccookie, th);
 691 out_trans:
 692         dt_trans_stop(env, dt, th);
 693         RETURN(rc);
 694 }
 695 EXPORT_SYMBOL(llog_cat_add);
 696
 697 /* For each cookie in the cookie array, we clear the log in-use bit and either:
 698  * - the log is empty, so mark it free in the catalog header and delete it
 699  * - the log is not empty, just write out the log header
 700  *
 701  * The cookies may be in different log files, so we need to get new logs
 702  * each time.
 703  *
 704  * Assumes caller has already pushed us into the kernel context.
 705  */
 706 int llog_cat_cancel_records(const struct lu_env *env,
 707                             struct llog_handle *cathandle, int count,
 708                             struct llog_cookie *cookies)
 709 {
 710         int i, index, rc = 0, failed = 0;
 711
 712         ENTRY;
 713
 714         for (i = 0; i < count; i++, cookies++) {
 715                 struct llog_handle      *loghandle;
 716                 struct llog_logid       *lgl = &cookies->lgc_lgl;
 717                 int                      lrc;
 718
 719                 rc = llog_cat_id2handle(env, cathandle, &loghandle, lgl);
 720                 if (rc) {
 721                         CERROR("%s: cannot find handle for llog "DOSTID": %d\n",
 722                                cathandle->lgh_ctxt->loc_obd->obd_name,
 723                                POSTID(&lgl->lgl_oi), rc);
 724                         failed++;
 725                         continue;
 726                 }
 727
 728                 lrc = llog_cancel_rec(env, loghandle, cookies->lgc_index);
 729                 if (lrc == LLOG_DEL_PLAIN) { /* log has been destroyed */
 730                         index = loghandle->u.phd.phd_cookie.lgc_index;
 731                         rc = llog_cat_cleanup(env, cathandle, loghandle,
 732                                               index);
 733                 } else if (lrc == -ENOENT) {
 734                         if (rc == 0) /* ENOENT shouldn't rewrite any error */
 735                                 rc = lrc;
 736                 } else if (lrc < 0) {
 737                         failed++;
 738                         rc = lrc;
 739                 }
 740                 llog_handle_put(loghandle);
 741         }
 742         if (rc)
 743                 CERROR("%s: fail to cancel %d of %d llog-records: rc = %d\n",
 744                        cathandle->lgh_ctxt->loc_obd->obd_name, failed, count,
 745                        rc);
 746
 747         RETURN(rc);
 748 }
 749 EXPORT_SYMBOL(llog_cat_cancel_records);
 750
 751 static int llog_cat_process_cb(const struct lu_env *env,
 752                                struct llog_handle *cat_llh,
 753                                struct llog_rec_hdr *rec, void *data)
 754 {
 755         struct llog_process_data *d = data;
 756         struct llog_logid_rec *lir = (struct llog_logid_rec *)rec;
 757         struct llog_handle *llh;
 758         struct llog_log_hdr *hdr;
 759         int rc;
 760
 761         ENTRY;
 762         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
 763                 CERROR("invalid record in catalog\n");
 764                 RETURN(-EINVAL);
 765         }
 766         CDEBUG(D_HA, "processing log "DOSTID":%x at index %u of catalog "
 767                DOSTID"\n", POSTID(&lir->lid_id.lgl_oi), lir->lid_id.lgl_ogen,
 768                rec->lrh_index, POSTID(&cat_llh->lgh_id.lgl_oi));
 769
 770         rc = llog_cat_id2handle(env, cat_llh, &llh, &lir->lid_id);
 771         if (rc) {
 772                 CERROR("%s: cannot find handle for llog "DOSTID": %d\n",
 773                        cat_llh->lgh_ctxt->loc_obd->obd_name,
 774                        POSTID(&lir->lid_id.lgl_oi), rc);
 775                 if (rc == -ENOENT || rc == -ESTALE) {
 776                         /* After a server crash, a stub of index
 777                          * record in catlog could be kept, because
 778                          * plain log destroy + catlog index record
 779                          * deletion are not atomic. So we end up with
 780                          * an index but no actual record. Destroy the
 781                          * index and move on. */
 782                         rc = llog_cat_cleanup(env, cat_llh, NULL,
 783                                               rec->lrh_index);
 784                 }
 785
 786                 RETURN(rc);
 787         }
 788
 789         /* clean old empty llogs, do not consider current llog in use */
 790         /* ignore remote (lgh_obj=NULL) llogs */
 791         hdr = llh->lgh_hdr;
 792         if ((hdr->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
 793             hdr->llh_count == 1 && cat_llh->lgh_obj != NULL &&
 794             llh != cat_llh->u.chd.chd_current_log) {
 795                 rc = llog_destroy(env, llh);
 796                 if (rc)
 797                         CERROR("%s: fail to destroy empty log: rc = %d\n",
 798                                llh->lgh_ctxt->loc_obd->obd_name, rc);
 799                 GOTO(out, rc = LLOG_DEL_PLAIN);
 800         }
 801
 802         if (rec->lrh_index < d->lpd_startcat) {
 803                 /* Skip processing of the logs until startcat */
 804                 rc = 0;
 805         } else if (d->lpd_startidx > 0) {
 806                 struct llog_process_cat_data cd;
 807
 808                 cd.lpcd_first_idx = d->lpd_startidx;
 809                 cd.lpcd_last_idx = 0;
 810                 rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
 811                                           &cd, false);
 812                 /* Continue processing the next log from idx 0 */
 813                 d->lpd_startidx = 0;
 814         } else {
 815                 rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
 816                                           NULL, false);
 817         }
 818
 819 out:
 820         /* The empty plain log was destroyed while processing */
 821         if (rc == LLOG_DEL_PLAIN)
 822                 rc = llog_cat_cleanup(env, cat_llh, llh,
 823                                       llh->u.phd.phd_cookie.lgc_index);
 824         llog_handle_put(llh);
 825
 826         RETURN(rc);
 827 }
 828
 829 int llog_cat_process_or_fork(const struct lu_env *env,
 830                              struct llog_handle *cat_llh, llog_cb_t cat_cb,
 831                              llog_cb_t cb, void *data, int startcat,
 832                              int startidx, bool fork)
 833 {
 834         struct llog_process_data d;
 835         struct llog_log_hdr *llh = cat_llh->lgh_hdr;
 836         int rc;
 837         ENTRY;
 838
 839         LASSERT(llh->llh_flags & LLOG_F_IS_CAT);
 840         d.lpd_data = data;
 841         d.lpd_cb = cb;
 842         d.lpd_startcat = startcat;
 843         d.lpd_startidx = startidx;
 844
 845         if (llh->llh_cat_idx >= cat_llh->lgh_last_idx &&
 846             llh->llh_count > 1) {
 847                 struct llog_process_cat_data cd;
 848
 849                 CWARN("catlog "DOSTID" crosses index zero\n",
 850                       POSTID(&cat_llh->lgh_id.lgl_oi));
 851
 852                 cd.lpcd_first_idx = llh->llh_cat_idx;
 853                 cd.lpcd_last_idx = 0;
 854                 rc = llog_process_or_fork(env, cat_llh, cat_cb,
 855                                           &d, &cd, fork);
 856                 if (rc != 0)
 857                         RETURN(rc);
 858
 859                 cd.lpcd_first_idx = 0;
 860                 cd.lpcd_last_idx = cat_llh->lgh_last_idx;
 861                 rc = llog_process_or_fork(env, cat_llh, cat_cb,
 862                                           &d, &cd, fork);
 863         } else {
 864                 rc = llog_process_or_fork(env, cat_llh, cat_cb,
 865                                           &d, NULL, fork);
 866         }
 867
 868         RETURN(rc);
 869 }
 870
 871 int llog_cat_process(const struct lu_env *env, struct llog_handle *cat_llh,
 872                      llog_cb_t cb, void *data, int startcat, int startidx)
 873 {
 874         return llog_cat_process_or_fork(env, cat_llh, llog_cat_process_cb,
 875                                         cb, data, startcat, startidx, false);
 876 }
 877 EXPORT_SYMBOL(llog_cat_process);
 878
 879 static int llog_cat_size_cb(const struct lu_env *env,
 880                              struct llog_handle *cat_llh,
 881                              struct llog_rec_hdr *rec, void *data)
 882 {
 883         struct llog_process_data *d = data;
 884         struct llog_logid_rec *lir = (struct llog_logid_rec *)rec;
 885         struct llog_handle *llh;
 886         int rc;
 887         __u64 *cum_size = d->lpd_data;
 888         __u64 size;
 889
 890         ENTRY;
 891         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
 892                 CERROR("%s: invalid record in catalog, rc = %d\n",
 893                        cat_llh->lgh_ctxt->loc_obd->obd_name, -EINVAL);
 894                 RETURN(-EINVAL);
 895         }
 896         CDEBUG(D_HA, "processing log "DOSTID":%x at index %u of catalog "
 897                DOSTID"\n", POSTID(&lir->lid_id.lgl_oi), lir->lid_id.lgl_ogen,
 898                rec->lrh_index, POSTID(&cat_llh->lgh_id.lgl_oi));
 899
 900         rc = llog_cat_id2handle(env, cat_llh, &llh, &lir->lid_id);
 901         if (rc) {
 902                 CWARN("%s: cannot find handle for llog "DOSTID": rc = %d\n",
 903                       cat_llh->lgh_ctxt->loc_obd->obd_name,
 904                       POSTID(&lir->lid_id.lgl_oi), rc);
 905                 RETURN(0);
 906         }
 907         size = llog_size(env, llh);
 908         *cum_size += size;
 909
 910         CDEBUG(D_INFO, "Add llog entry "DOSTID" size %llu\n",
 911                POSTID(&llh->lgh_id.lgl_oi), size);
 912
 913         llog_handle_put(llh);
 914
 915         RETURN(0);
 916
 917 }
 918
 919 __u64 llog_cat_size(const struct lu_env *env, struct llog_handle *cat_llh)
 920 {
 921         __u64 size = llog_size(env, cat_llh);
 922
 923         llog_cat_process_or_fork(env, cat_llh, llog_cat_size_cb,
 924                                  NULL, &size, 0, 0, false);
 925
 926         return size;
 927 }
 928 EXPORT_SYMBOL(llog_cat_size);
 929
 930 static int llog_cat_reverse_process_cb(const struct lu_env *env,
 931                                        struct llog_handle *cat_llh,
 932                                        struct llog_rec_hdr *rec, void *data)
 933 {
 934         struct llog_process_data *d = data;
 935         struct llog_logid_rec *lir = (struct llog_logid_rec *)rec;
 936         struct llog_handle *llh;
 937         struct llog_log_hdr *hdr;
 938         int rc;
 939
 940         if (le32_to_cpu(rec->lrh_type) != LLOG_LOGID_MAGIC) {
 941                 CERROR("invalid record in catalog\n");
 942                 RETURN(-EINVAL);
 943         }
 944         CDEBUG(D_HA, "processing log "DOSTID":%x at index %u of catalog "
 945                DOSTID"\n", POSTID(&lir->lid_id.lgl_oi), lir->lid_id.lgl_ogen,
 946                le32_to_cpu(rec->lrh_index), POSTID(&cat_llh->lgh_id.lgl_oi));
 947
 948         rc = llog_cat_id2handle(env, cat_llh, &llh, &lir->lid_id);
 949         if (rc) {
 950                 CERROR("%s: cannot find handle for llog "DOSTID": %d\n",
 951                        cat_llh->lgh_ctxt->loc_obd->obd_name,
 952                        POSTID(&lir->lid_id.lgl_oi), rc);
 953                 if (rc == -ENOENT || rc == -ESTALE) {
 954                         /* After a server crash, a stub of index
 955                          * record in catlog could be kept, because
 956                          * plain log destroy + catlog index record
 957                          * deletion are not atomic. So we end up with
 958                          * an index but no actual record. Destroy the
 959                          * index and move on. */
 960                         rc = llog_cat_cleanup(env, cat_llh, NULL,
 961                                               rec->lrh_index);
 962                 }
 963
 964                 RETURN(rc);
 965         }
 966
 967         /* clean old empty llogs, do not consider current llog in use */
 968         hdr = llh->lgh_hdr;
 969         if ((hdr->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
 970             hdr->llh_count == 1 &&
 971             llh != cat_llh->u.chd.chd_current_log) {
 972                 rc = llog_destroy(env, llh);
 973                 if (rc)
 974                         CERROR("%s: fail to destroy empty log: rc = %d\n",
 975                                llh->lgh_ctxt->loc_obd->obd_name, rc);
 976                 GOTO(out, rc = LLOG_DEL_PLAIN);
 977         }
 978
 979         rc = llog_reverse_process(env, llh, d->lpd_cb, d->lpd_data, NULL);
 980
 981 out:
 982         /* The empty plain was destroyed while processing */
 983         if (rc == LLOG_DEL_PLAIN)
 984                 rc = llog_cat_cleanup(env, cat_llh, llh,
 985                                       llh->u.phd.phd_cookie.lgc_index);
 986
 987         llog_handle_put(llh);
 988         RETURN(rc);
 989 }
 990
 991 int llog_cat_reverse_process(const struct lu_env *env,
 992                              struct llog_handle *cat_llh,
 993                              llog_cb_t cb, void *data)
 994 {
 995         struct llog_process_data d;
 996         struct llog_process_cat_data cd;
 997         struct llog_log_hdr *llh = cat_llh->lgh_hdr;
 998         int rc;
 999         ENTRY;
1000
1001         LASSERT(llh->llh_flags & LLOG_F_IS_CAT);
1002         d.lpd_data = data;
1003         d.lpd_cb = cb;
1004
1005         if (llh->llh_cat_idx >= cat_llh->lgh_last_idx &&
1006             llh->llh_count > 1) {
1007                 CWARN("catalog "DOSTID" crosses index zero\n",
1008                       POSTID(&cat_llh->lgh_id.lgl_oi));
1009
1010                 cd.lpcd_first_idx = 0;
1011                 cd.lpcd_last_idx = cat_llh->lgh_last_idx;
1012                 rc = llog_reverse_process(env, cat_llh,
1013                                           llog_cat_reverse_process_cb,
1014                                           &d, &cd);
1015                 if (rc != 0)
1016                         RETURN(rc);
1017
1018                 cd.lpcd_first_idx = le32_to_cpu(llh->llh_cat_idx);
1019                 cd.lpcd_last_idx = 0;
1020                 rc = llog_reverse_process(env, cat_llh,
1021                                           llog_cat_reverse_process_cb,
1022                                           &d, &cd);
1023         } else {
1024                 rc = llog_reverse_process(env, cat_llh,
1025                                           llog_cat_reverse_process_cb,
1026                                           &d, NULL);
1027         }
1028
1029         RETURN(rc);
1030 }
1031 EXPORT_SYMBOL(llog_cat_reverse_process);
1032
1033 static int llog_cat_set_first_idx(struct llog_handle *cathandle, int idx)
1034 {
1035         struct llog_log_hdr *llh = cathandle->lgh_hdr;
1036         int bitmap_size;
1037
1038         ENTRY;
1039
1040         bitmap_size = LLOG_HDR_BITMAP_SIZE(llh);
1041         /*
1042          * The llh_cat_idx equals to the first used index minus 1
1043          * so if we canceled the first index then llh_cat_idx
1044          * must be renewed.
1045          */
1046         if (llh->llh_cat_idx == (idx - 1)) {
1047                 llh->llh_cat_idx = idx;
1048
1049                 while (idx != cathandle->lgh_last_idx) {
1050                         idx = (idx + 1) % bitmap_size;
1051                         if (!ext2_test_bit(idx, LLOG_HDR_BITMAP(llh))) {
1052                                 /* update llh_cat_idx for each unset bit,
1053                                  * expecting the next one is set */
1054                                 llh->llh_cat_idx = idx;
1055                         } else if (idx == 0) {
1056                                 /* skip header bit */
1057                                 llh->llh_cat_idx = 0;
1058                                 continue;
1059                         } else {
1060                                 /* the first index is found */
1061                                 break;
1062                         }
1063                 }
1064
1065                 CDEBUG(D_RPCTRACE, "Set catlog "DOSTID" first idx %u,"
1066                        " (last_idx %u)\n", POSTID(&cathandle->lgh_id.lgl_oi),
1067                        llh->llh_cat_idx, cathandle->lgh_last_idx);
1068         }
1069
1070         RETURN(0);
1071 }
1072
1073 /* Cleanup deleted plain llog traces from catalog */
1074 int llog_cat_cleanup(const struct lu_env *env, struct llog_handle *cathandle,
1075                      struct llog_handle *loghandle, int index)
1076 {
1077         int rc;
1078
1079         LASSERT(index);
1080         if (loghandle != NULL) {
1081                 /* remove destroyed llog from catalog list and
1082                  * chd_current_log variable */
1083                 down_write(&cathandle->lgh_lock);
1084                 if (cathandle->u.chd.chd_current_log == loghandle)
1085                         cathandle->u.chd.chd_current_log = NULL;
1086                 list_del_init(&loghandle->u.phd.phd_entry);
1087                 up_write(&cathandle->lgh_lock);
1088                 LASSERT(index == loghandle->u.phd.phd_cookie.lgc_index);
1089                 /* llog was opened and keep in a list, close it now */
1090                 llog_close(env, loghandle);
1091         }
1092
1093         /* do not attempt to cleanup on-disk llog if on client side */
1094         if (cathandle->lgh_obj == NULL)
1095                 return 0;
1096
1097         /* remove plain llog entry from catalog by index */
1098         llog_cat_set_first_idx(cathandle, index);
1099         rc = llog_cancel_rec(env, cathandle, index);
1100         if (rc == 0)
1101                 CDEBUG(D_HA, "cancel plain log at index"
1102                        " %u of catalog "DOSTID"\n",
1103                        index, POSTID(&cathandle->lgh_id.lgl_oi));
1104         return rc;
1105 }