lustre/obdclass/llog_test.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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, 2015, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/obdclass/llog_test.c
  37  *
  38  * Author: Phil Schwan <phil@clusterfs.com>
  39  * Author: Mikhail Pershin <mike.pershin@intel.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_CLASS
  43
  44 #include <linux/module.h>
  45 #include <linux/init.h>
  46
  47 #include <obd_class.h>
  48 #include <lustre_fid.h>
  49 #include <lustre_log.h>
  50
  51 /* This is slightly more than the number of records that can fit into a
  52  * single llog file, because the llog_log_header takes up some of the
  53  * space in the first block that cannot be used for the bitmap. */
  54 #define LLOG_TEST_RECNUM  (LLOG_MIN_CHUNK_SIZE * 8)
  55
  56 static int llog_test_rand;
  57 static struct obd_uuid uuid = { .uuid = "test_uuid" };
  58 static struct llog_logid cat_logid;
  59
  60 struct llog_mini_rec {
  61         struct llog_rec_hdr     lmr_hdr;
  62         struct llog_rec_tail    lmr_tail;
  63 } __attribute__((packed));
  64
  65 static int verify_handle(char *test, struct llog_handle *llh, int num_recs)
  66 {
  67         int i;
  68         int last_idx = 0;
  69         int active_recs = 0;
  70
  71         for (i = 0; i < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr); i++) {
  72                 if (ext2_test_bit(i, LLOG_HDR_BITMAP(llh->lgh_hdr))) {
  73                         last_idx = i;
  74                         active_recs++;
  75                 }
  76         }
  77
  78         /* check the llog is sane at first, llh_count and lgh_last_idx*/
  79         if (llh->lgh_hdr->llh_count != active_recs) {
  80                 CERROR("%s: handle->count is %d, but there are %d recs found\n",
  81                        test, llh->lgh_hdr->llh_count, active_recs);
  82                 RETURN(-ERANGE);
  83         }
  84
  85         if (llh->lgh_last_idx != LLOG_HDR_TAIL(llh->lgh_hdr)->lrt_index ||
  86             (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_CAT) &&
  87              llh->lgh_last_idx < last_idx)) {
  88                 CERROR("%s: lgh_last_idx is %d (%d in the header), last found %d\n",
  89                        test, llh->lgh_last_idx,
  90                        LLOG_HDR_TAIL(llh->lgh_hdr)->lrt_index, last_idx);
  91                 RETURN(-ERANGE);
  92         }
  93
  94         /* finally checks against expected value from the caller */
  95         if (active_recs != num_recs) {
  96                 CERROR("%s: expected %d active recs after write, found %d\n",
  97                        test, num_recs, active_recs);
  98                 RETURN(-ERANGE);
  99         }
 100
 101         RETURN(0);
 102 }
 103
 104 /* Test named-log create/open, close */
 105 static int llog_test_1(const struct lu_env *env,
 106                        struct obd_device *obd, char *name)
 107 {
 108         struct llog_handle      *llh;
 109         struct llog_ctxt        *ctxt;
 110         int rc;
 111         int rc2;
 112
 113         ENTRY;
 114
 115         CWARN("1a: create a log with name: %s\n", name);
 116         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 117         LASSERT(ctxt);
 118
 119         rc = llog_open_create(env, ctxt, &llh, NULL, name);
 120         if (rc) {
 121                 CERROR("1a: llog_create with name %s failed: %d\n", name, rc);
 122                 GOTO(out, rc);
 123         }
 124         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, &uuid);
 125         if (rc) {
 126                 CERROR("1a: can't init llog handle: %d\n", rc);
 127                 GOTO(out_close, rc);
 128         }
 129
 130         rc = verify_handle("1", llh, 1);
 131
 132         CWARN("1b: close newly-created log\n");
 133 out_close:
 134         rc2 = llog_close(env, llh);
 135         if (rc2) {
 136                 CERROR("1b: close log %s failed: %d\n", name, rc2);
 137                 if (rc == 0)
 138                         rc = rc2;
 139         }
 140 out:
 141         llog_ctxt_put(ctxt);
 142         RETURN(rc);
 143 }
 144
 145 static int test_2_cancel_cb(const struct lu_env *env, struct llog_handle *llh,
 146                             struct llog_rec_hdr *rec, void *data)
 147 {
 148         return LLOG_DEL_RECORD;
 149 }
 150
 151 /* Test named-log reopen; returns opened log on success */
 152 static int llog_test_2(const struct lu_env *env, struct obd_device *obd,
 153                        char *name, struct llog_handle **llh)
 154 {
 155         struct llog_ctxt        *ctxt;
 156         struct llog_handle      *lgh;
 157         struct llog_logid        logid;
 158         int                      rc;
 159         struct llog_mini_rec     lmr;
 160
 161         ENTRY;
 162
 163         CWARN("2a: re-open a log with name: %s\n", name);
 164         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 165         LASSERT(ctxt);
 166
 167         rc = llog_open(env, ctxt, llh, NULL, name, LLOG_OPEN_EXISTS);
 168         if (rc) {
 169                 CERROR("2a: re-open log with name %s failed: %d\n", name, rc);
 170                 GOTO(out_put, rc);
 171         }
 172
 173         rc = llog_init_handle(env, *llh, LLOG_F_IS_PLAIN, &uuid);
 174         if (rc) {
 175                 CERROR("2a: can't init llog handle: %d\n", rc);
 176                 GOTO(out_close_llh, rc);
 177         }
 178
 179         rc = verify_handle("2", *llh, 1);
 180         if (rc)
 181                 GOTO(out_close_llh, rc);
 182
 183         CWARN("2b: create a log without specified NAME & LOGID\n");
 184         rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
 185         if (rc) {
 186                 CERROR("2b: create log failed\n");
 187                 GOTO(out_close_llh, rc);
 188         }
 189         rc = llog_init_handle(env, lgh, LLOG_F_IS_PLAIN, &uuid);
 190         if (rc) {
 191                 CERROR("2b: can't init llog handle: %d\n", rc);
 192                 GOTO(out_close, rc);
 193         }
 194
 195         logid = lgh->lgh_id;
 196
 197         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 198         lmr.lmr_hdr.lrh_type = 0xf02f02;
 199
 200         /* Check llog header values are correct after record add/cancel */
 201         CWARN("2b: write 1 llog records, check llh_count\n");
 202         rc = llog_write(env, lgh, &lmr.lmr_hdr, LLOG_NEXT_IDX);
 203         if (rc < 0)
 204                 GOTO(out_close, rc);
 205
 206         /* in-memory values after record addition */
 207         rc = verify_handle("2b", lgh, 2);
 208         if (rc < 0)
 209                 GOTO(out_close, rc);
 210
 211         /* re-open llog to read on-disk values */
 212         llog_close(env, lgh);
 213
 214         CWARN("2c: re-open the log by LOGID and verify llh_count\n");
 215         rc = llog_open(env, ctxt, &lgh, &logid, NULL, LLOG_OPEN_EXISTS);
 216         if (rc < 0) {
 217                 CERROR("2c: re-open log by LOGID failed\n");
 218                 GOTO(out_close_llh, rc);
 219         }
 220
 221         rc = llog_init_handle(env, lgh, LLOG_F_IS_PLAIN, &uuid);
 222         if (rc < 0) {
 223                 CERROR("2c: can't init llog handle: %d\n", rc);
 224                 GOTO(out_close, rc);
 225         }
 226
 227         /* check values just read from disk */
 228         rc = verify_handle("2c", lgh, 2);
 229         if (rc < 0)
 230                 GOTO(out_close, rc);
 231
 232         rc = llog_process(env, lgh, test_2_cancel_cb, NULL, NULL);
 233         if (rc < 0)
 234                 GOTO(out_close, rc);
 235
 236         /* in-memory values */
 237         rc = verify_handle("2c", lgh, 1);
 238         if (rc < 0)
 239                 GOTO(out_close, rc);
 240
 241         /* re-open llog to get on-disk values */
 242         llog_close(env, lgh);
 243
 244         rc = llog_open(env, ctxt, &lgh, &logid, NULL, LLOG_OPEN_EXISTS);
 245         if (rc) {
 246                 CERROR("2c: re-open log by LOGID failed\n");
 247                 GOTO(out_close_llh, rc);
 248         }
 249
 250         rc = llog_init_handle(env, lgh, LLOG_F_IS_PLAIN, &uuid);
 251         if (rc) {
 252                 CERROR("2c: can't init llog handle: %d\n", rc);
 253                 GOTO(out_close, rc);
 254         }
 255
 256         /* on-disk values after llog re-open */
 257         rc = verify_handle("2c", lgh, 1);
 258         if (rc < 0)
 259                 GOTO(out_close, rc);
 260
 261         CWARN("2d: destroy this log\n");
 262         rc = llog_destroy(env, lgh);
 263         if (rc)
 264                 CERROR("2d: destroy log failed\n");
 265 out_close:
 266         llog_close(env, lgh);
 267 out_close_llh:
 268         if (rc)
 269                 llog_close(env, *llh);
 270 out_put:
 271         llog_ctxt_put(ctxt);
 272
 273         RETURN(rc);
 274 }
 275
 276 static int test_3_rec_num;
 277 static off_t test_3_rec_off;
 278 static int test_3_paddings;
 279 static int test_3_start_idx;
 280
 281 /*
 282  * Test 3 callback.
 283  * - check lgh_cur_offset correctness
 284  * - check record index consistency
 285  * - modify each record in-place
 286  * - add new record during *last_idx processing
 287  */
 288 static int test3_check_n_add_cb(const struct lu_env *env,
 289                                 struct llog_handle *lgh,
 290                                 struct llog_rec_hdr *rec, void *data)
 291 {
 292         struct llog_gen_rec *lgr = (struct llog_gen_rec *)rec;
 293         int *last_rec = data;
 294         unsigned cur_idx = test_3_start_idx + test_3_rec_num;
 295         int rc;
 296
 297         if (lgh->lgh_hdr->llh_flags & LLOG_F_IS_FIXSIZE) {
 298                 LASSERT(lgh->lgh_hdr->llh_size > 0);
 299                 if (lgh->lgh_cur_offset != lgh->lgh_hdr->llh_hdr.lrh_len +
 300                                         (cur_idx - 1) * lgh->lgh_hdr->llh_size)
 301                         CERROR("Wrong record offset in cur_off: "LPU64", should be %u\n",
 302                                lgh->lgh_cur_offset,
 303                                lgh->lgh_hdr->llh_hdr.lrh_len +
 304                                (cur_idx - 1) * lgh->lgh_hdr->llh_size);
 305         } else {
 306                 size_t chunk_size = lgh->lgh_hdr->llh_hdr.lrh_len;
 307
 308                 /* For variable size records the start offset is unknown, trust
 309                  * the first value and check others are consistent with it. */
 310                 if (test_3_rec_off == 0)
 311                         test_3_rec_off = lgh->lgh_cur_offset;
 312
 313                 if (lgh->lgh_cur_offset != test_3_rec_off) {
 314                         /* there can be padding record */
 315                         if ((lgh->lgh_cur_offset % chunk_size == 0) &&
 316                             (lgh->lgh_cur_offset - test_3_rec_off <
 317                              rec->lrh_len + LLOG_MIN_REC_SIZE)) {
 318                                 test_3_rec_off = lgh->lgh_cur_offset;
 319                                 test_3_paddings++;
 320                         } else {
 321                                 CERROR("Wrong record offset in cur_off: "LPU64
 322                                        ", should be %lld (rec len %u)\n",
 323                                        lgh->lgh_cur_offset,
 324                                        (long long)test_3_rec_off,
 325                                        rec->lrh_len);
 326                         }
 327                 }
 328                 test_3_rec_off += rec->lrh_len;
 329         }
 330
 331         cur_idx += test_3_paddings;
 332         if (cur_idx != rec->lrh_index)
 333                 CERROR("Record with wrong index was read: %u, expected %u\n",
 334                        rec->lrh_index, cur_idx);
 335
 336         /* modify all records in place */
 337         lgr->lgr_gen.conn_cnt = rec->lrh_index;
 338         rc = llog_write(env, lgh, rec, rec->lrh_index);
 339         if (rc < 0)
 340                 CERROR("cb_test_3: cannot modify record while processing\n");
 341
 342         /* Add new record to the llog at *last_rec position one by one to
 343          * check that last block is re-read during processing */
 344         if (cur_idx == *last_rec || cur_idx == (*last_rec + 1)) {
 345                 rc = llog_write(env, lgh, rec, LLOG_NEXT_IDX);
 346                 if (rc < 0)
 347                         CERROR("cb_test_3: cannot add new record while "
 348                                "processing\n");
 349         }
 350         test_3_rec_num++;
 351
 352         return rc;
 353 }
 354
 355 /* Check in-place modifications were done for all records*/
 356 static int test3_check_cb(const struct lu_env *env, struct llog_handle *lgh,
 357                           struct llog_rec_hdr *rec, void *data)
 358 {
 359         struct llog_gen_rec *lgr = (struct llog_gen_rec *)rec;
 360
 361         if (lgr->lgr_gen.conn_cnt != rec->lrh_index) {
 362                 CERROR("cb_test_3: record %u is not modified\n",
 363                        rec->lrh_index);
 364                 return -EINVAL;
 365         }
 366         test_3_rec_num++;
 367         return 0;
 368 }
 369
 370 static int llog_test3_process(const struct lu_env *env,
 371                               struct llog_handle *lgh,
 372                               llog_cb_t cb, int start)
 373 {
 374         struct llog_process_cat_data cd;
 375         int last_idx; /* new record will be injected here */
 376         int rc = 0;
 377
 378         CWARN("test3: processing records from index %d to the end\n",
 379               start);
 380         cd.lpcd_first_idx = start - 1;
 381         cd.lpcd_last_idx = 0;
 382         test_3_rec_num = test_3_paddings = 0;
 383         last_idx = lgh->lgh_last_idx;
 384         rc = llog_process(env, lgh, cb, &last_idx, &cd);
 385         if (rc < 0)
 386                 return rc;
 387         CWARN("test3: total %u records processed with %u paddings\n",
 388               test_3_rec_num, test_3_paddings);
 389         return test_3_rec_num;
 390 }
 391
 392 /* Test plain llog functionality */
 393 static int llog_test_3(const struct lu_env *env, struct obd_device *obd,
 394                        struct llog_handle *llh)
 395 {
 396         char buf[128];
 397         struct llog_rec_hdr *hdr = (void *)buf;
 398         int rc, i;
 399         int num_recs = 1; /* 1 for the header */
 400         int expected;
 401
 402         ENTRY;
 403
 404         hdr->lrh_len = sizeof(struct llog_gen_rec);
 405         hdr->lrh_type = LLOG_GEN_REC;
 406         llh->lgh_hdr->llh_size = sizeof(struct llog_gen_rec);
 407         llh->lgh_hdr->llh_flags |= LLOG_F_IS_FIXSIZE;
 408
 409         /* Fill the llog with 64-bytes records, use 1023 records,
 410          * so last chunk will be partially full. Don't change this
 411          * value until record size is changed.
 412          */
 413         CWARN("3a: write 1023 fixed-size llog records\n");
 414         for (i = 0; i < 1023; i++) {
 415                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 416                 if (rc < 0) {
 417                         CERROR("3a: write 1023 records failed at #%d: %d\n",
 418                                i + 1, rc);
 419                         RETURN(rc);
 420                 }
 421                 num_recs++;
 422         }
 423
 424         rc = verify_handle("3a", llh, num_recs);
 425         if (rc)
 426                 RETURN(rc);
 427
 428         /*
 429          * Test fixed-size records processing:
 430          * - search the needed index
 431          * - go through all records from that index
 432          * - check all indices are growing monotonically and exist
 433          * - modify each record
 434          *
 435          * NB: test3_check_n_add adds two new records while processing
 436          * after last record. There were 1023 records created so the last chunk
 437          * misses exactly one record. Therefore one of new records will be
 438          * the last in the current chunk and second causes the new chunk to be
 439          * created.
 440          */
 441         test_3_rec_off = 0;
 442         test_3_start_idx = 501;
 443         expected = 525;
 444         rc = llog_test3_process(env, llh, test3_check_n_add_cb,
 445                                 test_3_start_idx);
 446         if (rc < 0)
 447                 RETURN(rc);
 448
 449         /* extra record is created during llog_process() */
 450         if (rc != expected) {
 451                 CERROR("3a: process total %d records but expect %d\n",
 452                        rc, expected);
 453                 RETURN(-ERANGE);
 454         }
 455
 456         num_recs += 2;
 457
 458         /* test modification in place */
 459         rc = llog_test3_process(env, llh, test3_check_cb, test_3_start_idx);
 460         if (rc < 0)
 461                 RETURN(rc);
 462
 463         if (rc != expected) {
 464                 CERROR("3a: process total %d records but expect %d\n",
 465                        rc, expected);
 466                 RETURN(-ERANGE);
 467         }
 468
 469         CWARN("3b: write 566 variable size llog records\n");
 470
 471         /* Drop llh_size to 0 to mark llog as variable-size and write
 472          * header to make this change permanent. */
 473         llh->lgh_hdr->llh_flags &= ~LLOG_F_IS_FIXSIZE;
 474         llog_write(env, llh, &llh->lgh_hdr->llh_hdr, LLOG_HEADER_IDX);
 475
 476         hdr->lrh_type = OBD_CFG_REC;
 477
 478         /* there are 1025 64-bytes records in llog already,
 479          * the last chunk contains single record, i.e. 64 bytes.
 480          * Each pair of variable size records is 200 bytes, so
 481          * we will have the following distribution per chunks:
 482          * block 1: 64 + 80(80/120) + 80 + 48(pad) = 81 iterations
 483          * block 2: 80(120/80) + 120 + 72(pad) = 81 itereations
 484          * block 3: 80(80/120) + 80 + 112(pad) = 81 iterations
 485          * -- the same as block 2 again and so on.
 486          * block 7: 80(80/120) = 80 iterations and 192 bytes remain
 487          * Total 6 * 81 + 80 = 566 itereations.
 488          * Callback will add another 120 bytes in the end of the last chunk
 489          * and another 120 bytes will cause padding (72 bytes) plus 120
 490          * bytes in the new block.
 491          */
 492         for (i = 0; i < 566; i++) {
 493                 if ((i % 2) == 0)
 494                         hdr->lrh_len = 80;
 495                 else
 496                         hdr->lrh_len = 120;
 497
 498                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 499                 if (rc < 0) {
 500                         CERROR("3b: write 566 records failed at #%d: %d\n",
 501                                i + 1, rc);
 502                         RETURN(rc);
 503                 }
 504                 num_recs++;
 505         }
 506
 507         rc = verify_handle("3b", llh, num_recs);
 508         if (rc)
 509                 RETURN(rc);
 510
 511         test_3_start_idx = 1026;
 512         expected = 568;
 513         rc = llog_test3_process(env, llh, test3_check_n_add_cb,
 514                                 test_3_start_idx);
 515         if (rc < 0)
 516                 RETURN(rc);
 517
 518         if (rc != expected) {
 519                 CERROR("3b: process total %d records but expect %d\n",
 520                        rc, expected);
 521                 RETURN(-ERANGE);
 522         }
 523
 524         num_recs += 2;
 525
 526         /* test modification in place */
 527         rc = llog_test3_process(env, llh, test3_check_cb, test_3_start_idx);
 528         if (rc < 0)
 529                 RETURN(rc);
 530
 531         if (rc != expected) {
 532                 CERROR("3b: process total %d records but expect %d\n",
 533                        rc, expected);
 534                 RETURN(-ERANGE);
 535         }
 536
 537         CWARN("3c: write records with variable size until BITMAP_SIZE, "
 538               "return -ENOSPC\n");
 539         while (num_recs < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr)) {
 540                 if ((num_recs % 2) == 0)
 541                         hdr->lrh_len = 80;
 542                 else
 543                         hdr->lrh_len = 128;
 544
 545                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 546                 if (rc == -ENOSPC) {
 547                         break;
 548                 } else if (rc < 0) {
 549                         CERROR("3c: write recs failed at #%d: %d\n",
 550                                num_recs, rc);
 551                         RETURN(rc);
 552                 }
 553                 num_recs++;
 554         }
 555
 556         if (rc != -ENOSPC) {
 557                 CWARN("3c: write record more than BITMAP size!\n");
 558                 RETURN(-EINVAL);
 559         }
 560         CWARN("3c: wrote %d more records before end of llog is reached\n",
 561               num_recs);
 562
 563         rc = verify_handle("3c", llh, num_recs);
 564
 565         RETURN(rc);
 566 }
 567
 568 /* Test catalogue additions */
 569 static int llog_test_4(const struct lu_env *env, struct obd_device *obd)
 570 {
 571         struct llog_handle      *cath;
 572         char                     name[10];
 573         int                      rc, rc2, i, buflen;
 574         struct llog_mini_rec     lmr;
 575         struct llog_cookie       cookie;
 576         struct llog_ctxt        *ctxt;
 577         int                      num_recs = 0;
 578         char                    *buf;
 579         struct llog_rec_hdr     *rec;
 580
 581         ENTRY;
 582
 583         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 584         LASSERT(ctxt);
 585
 586         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 587         lmr.lmr_hdr.lrh_type = 0xf00f00;
 588
 589         sprintf(name, "%x", llog_test_rand + 1);
 590         CWARN("4a: create a catalog log with name: %s\n", name);
 591         rc = llog_open_create(env, ctxt, &cath, NULL, name);
 592         if (rc) {
 593                 CERROR("4a: llog_create with name %s failed: %d\n", name, rc);
 594                 GOTO(ctxt_release, rc);
 595         }
 596         rc = llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);
 597         if (rc) {
 598                 CERROR("4a: can't init llog handle: %d\n", rc);
 599                 GOTO(out, rc);
 600         }
 601
 602         num_recs++;
 603         cat_logid = cath->lgh_id;
 604
 605         CWARN("4b: write 1 record into the catalog\n");
 606         rc = llog_cat_add(env, cath, &lmr.lmr_hdr, &cookie);
 607         if (rc != 1) {
 608                 CERROR("4b: write 1 catalog record failed at: %d\n", rc);
 609                 GOTO(out, rc);
 610         }
 611         num_recs++;
 612         rc = verify_handle("4b", cath, 2);
 613         if (rc)
 614                 GOTO(out, rc);
 615
 616         rc = verify_handle("4b", cath->u.chd.chd_current_log, num_recs);
 617         if (rc)
 618                 GOTO(out, rc);
 619
 620         CWARN("4c: cancel 1 log record\n");
 621         rc = llog_cat_cancel_records(env, cath, 1, &cookie);
 622         if (rc) {
 623                 CERROR("4c: cancel 1 catalog based record failed: %d\n", rc);
 624                 GOTO(out, rc);
 625         }
 626         num_recs--;
 627
 628         rc = verify_handle("4c", cath->u.chd.chd_current_log, num_recs);
 629         if (rc)
 630                 GOTO(out, rc);
 631
 632         CWARN("4d: write %d more log records\n", LLOG_TEST_RECNUM);
 633         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
 634                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
 635                 if (rc) {
 636                         CERROR("4d: write %d records failed at #%d: %d\n",
 637                                LLOG_TEST_RECNUM, i + 1, rc);
 638                         GOTO(out, rc);
 639                 }
 640                 num_recs++;
 641         }
 642
 643         /* make sure new plain llog appears */
 644         rc = verify_handle("4d", cath, 3);
 645         if (rc)
 646                 GOTO(out, rc);
 647
 648         CWARN("4e: add 5 large records, one record per block\n");
 649         buflen = LLOG_MIN_CHUNK_SIZE;
 650         OBD_ALLOC(buf, buflen);
 651         if (buf == NULL)
 652                 GOTO(out, rc = -ENOMEM);
 653         for (i = 0; i < 5; i++) {
 654                 rec = (void *)buf;
 655                 rec->lrh_len = buflen;
 656                 rec->lrh_type = OBD_CFG_REC;
 657                 rc = llog_cat_add(env, cath, rec, NULL);
 658                 if (rc) {
 659                         CERROR("4e: write 5 records failed at #%d: %d\n",
 660                                i + 1, rc);
 661                         GOTO(out_free, rc);
 662                 }
 663                 num_recs++;
 664         }
 665 out_free:
 666         OBD_FREE(buf, buflen);
 667 out:
 668         CWARN("4f: put newly-created catalog\n");
 669         rc2 = llog_cat_close(env, cath);
 670         if (rc2) {
 671                 CERROR("4: close log %s failed: %d\n", name, rc2);
 672                 if (rc == 0)
 673                         rc = rc2;
 674         }
 675 ctxt_release:
 676         llog_ctxt_put(ctxt);
 677         RETURN(rc);
 678 }
 679
 680 static int cat_counter;
 681
 682 static int cat_print_cb(const struct lu_env *env, struct llog_handle *llh,
 683                         struct llog_rec_hdr *rec, void *data)
 684 {
 685         struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
 686         struct lu_fid            fid = {0};
 687
 688         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
 689                 CERROR("invalid record in catalog\n");
 690                 RETURN(-EINVAL);
 691         }
 692
 693         logid_to_fid(&lir->lid_id, &fid);
 694
 695         CWARN("seeing record at index %d - "DFID" in log "DFID"\n",
 696               rec->lrh_index, PFID(&fid),
 697               PFID(lu_object_fid(&llh->lgh_obj->do_lu)));
 698
 699         cat_counter++;
 700
 701         RETURN(0);
 702 }
 703
 704 static int plain_counter;
 705
 706 static int plain_print_cb(const struct lu_env *env, struct llog_handle *llh,
 707                           struct llog_rec_hdr *rec, void *data)
 708 {
 709         struct lu_fid fid = {0};
 710
 711         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 712                 CERROR("log is not plain\n");
 713                 RETURN(-EINVAL);
 714         }
 715
 716         logid_to_fid(&llh->lgh_id, &fid);
 717
 718         CDEBUG(D_INFO, "seeing record at index %d in log "DFID"\n",
 719                rec->lrh_index, PFID(&fid));
 720
 721         plain_counter++;
 722
 723         RETURN(0);
 724 }
 725
 726 static int cancel_count;
 727
 728 static int llog_cancel_rec_cb(const struct lu_env *env,
 729                               struct llog_handle *llh,
 730                               struct llog_rec_hdr *rec, void *data)
 731 {
 732         struct llog_cookie cookie;
 733
 734         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 735                 CERROR("log is not plain\n");
 736                 RETURN(-EINVAL);
 737         }
 738
 739         cookie.lgc_lgl = llh->lgh_id;
 740         cookie.lgc_index = rec->lrh_index;
 741
 742         llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1, &cookie);
 743         cancel_count++;
 744         if (cancel_count == LLOG_TEST_RECNUM)
 745                 RETURN(-LLOG_EEMPTY);
 746         RETURN(0);
 747 }
 748
 749 /* Test log and catalogue processing */
 750 static int llog_test_5(const struct lu_env *env, struct obd_device *obd)
 751 {
 752         struct llog_handle      *llh = NULL;
 753         char                     name[10];
 754         int                      rc, rc2;
 755         struct llog_mini_rec     lmr;
 756         struct llog_ctxt        *ctxt;
 757
 758         ENTRY;
 759
 760         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 761         LASSERT(ctxt);
 762
 763         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 764         lmr.lmr_hdr.lrh_type = 0xf00f00;
 765
 766         CWARN("5a: re-open catalog by id\n");
 767         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
 768         if (rc) {
 769                 CERROR("5a: llog_create with logid failed: %d\n", rc);
 770                 GOTO(out_put, rc);
 771         }
 772
 773         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
 774         if (rc) {
 775                 CERROR("5a: can't init llog handle: %d\n", rc);
 776                 GOTO(out, rc);
 777         }
 778
 779         CWARN("5b: print the catalog entries.. we expect 2\n");
 780         cat_counter = 0;
 781         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 782         if (rc) {
 783                 CERROR("5b: process with cat_print_cb failed: %d\n", rc);
 784                 GOTO(out, rc);
 785         }
 786         if (cat_counter != 2) {
 787                 CERROR("5b: %d entries in catalog\n", cat_counter);
 788                 GOTO(out, rc = -EINVAL);
 789         }
 790
 791         CWARN("5c: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
 792         cancel_count = 0;
 793         rc = llog_cat_process(env, llh, llog_cancel_rec_cb, "foobar", 0, 0);
 794         if (rc != -LLOG_EEMPTY) {
 795                 CERROR("5c: process with llog_cancel_rec_cb failed: %d\n", rc);
 796                 GOTO(out, rc);
 797         }
 798
 799         CWARN("5c: print the catalog entries.. we expect 1\n");
 800         cat_counter = 0;
 801         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 802         if (rc) {
 803                 CERROR("5c: process with cat_print_cb failed: %d\n", rc);
 804                 GOTO(out, rc);
 805         }
 806         if (cat_counter != 1) {
 807                 CERROR("5c: %d entries in catalog\n", cat_counter);
 808                 GOTO(out, rc = -EINVAL);
 809         }
 810
 811         CWARN("5d: add 1 record to the log with many canceled empty pages\n");
 812         rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
 813         if (rc) {
 814                 CERROR("5d: add record to the log with many canceled empty "
 815                        "pages failed\n");
 816                 GOTO(out, rc);
 817         }
 818
 819         CWARN("5e: print plain log entries.. expect 6\n");
 820         plain_counter = 0;
 821         rc = llog_cat_process(env, llh, plain_print_cb, "foobar", 0, 0);
 822         if (rc) {
 823                 CERROR("5e: process with plain_print_cb failed: %d\n", rc);
 824                 GOTO(out, rc);
 825         }
 826         if (plain_counter != 6) {
 827                 CERROR("5e: found %d records\n", plain_counter);
 828                 GOTO(out, rc = -EINVAL);
 829         }
 830
 831         CWARN("5f: print plain log entries reversely.. expect 6\n");
 832         plain_counter = 0;
 833         rc = llog_cat_reverse_process(env, llh, plain_print_cb, "foobar");
 834         if (rc) {
 835                 CERROR("5f: reversely process with plain_print_cb failed: "
 836                        "%d\n", rc);
 837                 GOTO(out, rc);
 838         }
 839         if (plain_counter != 6) {
 840                 CERROR("5f: found %d records\n", plain_counter);
 841                 GOTO(out, rc = -EINVAL);
 842         }
 843
 844 out:
 845         CWARN("5g: close re-opened catalog\n");
 846         rc2 = llog_cat_close(env, llh);
 847         if (rc2) {
 848                 CERROR("5g: close log %s failed: %d\n", name, rc2);
 849                 if (rc == 0)
 850                         rc = rc2;
 851         }
 852 out_put:
 853         llog_ctxt_put(ctxt);
 854
 855         RETURN(rc);
 856 }
 857
 858 /* Test client api; open log by name and process */
 859 static int llog_test_6(const struct lu_env *env, struct obd_device *obd,
 860                        char *name)
 861 {
 862         struct obd_device       *mgc_obd;
 863         struct llog_ctxt        *ctxt;
 864         struct obd_uuid         *mgs_uuid;
 865         struct obd_export       *exp;
 866         struct obd_uuid          uuid = { "LLOG_TEST6_UUID" };
 867         struct llog_handle      *llh = NULL;
 868         struct llog_ctxt        *nctxt;
 869         int                      rc, rc2;
 870
 871         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 872         LASSERT(ctxt);
 873         mgs_uuid = &ctxt->loc_exp->exp_obd->obd_uuid;
 874
 875         CWARN("6a: re-open log %s using client API\n", name);
 876         mgc_obd = class_find_client_obd(mgs_uuid, LUSTRE_MGC_NAME, NULL);
 877         if (mgc_obd == NULL) {
 878                 CERROR("6a: no MGC devices connected to %s found.\n",
 879                        mgs_uuid->uuid);
 880                 GOTO(ctxt_release, rc = -ENOENT);
 881         }
 882
 883         rc = obd_connect(NULL, &exp, mgc_obd, &uuid,
 884                          NULL /* obd_connect_data */, NULL);
 885         if (rc != -EALREADY) {
 886                 CERROR("6a: connect on connected MGC (%s) failed to return"
 887                        " -EALREADY\n", mgc_obd->obd_name);
 888                 if (rc == 0)
 889                         obd_disconnect(exp);
 890                 GOTO(ctxt_release, rc = -EINVAL);
 891         }
 892
 893         nctxt = llog_get_context(mgc_obd, LLOG_CONFIG_REPL_CTXT);
 894         rc = llog_open(env, nctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
 895         if (rc) {
 896                 CERROR("6a: llog_open failed %d\n", rc);
 897                 GOTO(nctxt_put, rc);
 898         }
 899
 900         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
 901         if (rc) {
 902                 CERROR("6a: llog_init_handle failed %d\n", rc);
 903                 GOTO(parse_out, rc);
 904         }
 905
 906         plain_counter = 1; /* llog header is first record */
 907         CWARN("6b: process log %s using client API\n", name);
 908         rc = llog_process(env, llh, plain_print_cb, NULL, NULL);
 909         if (rc)
 910                 CERROR("6b: llog_process failed %d\n", rc);
 911         CWARN("6b: processed %d records\n", plain_counter);
 912
 913         rc = verify_handle("6b", llh, plain_counter);
 914         if (rc)
 915                 GOTO(parse_out, rc);
 916
 917         plain_counter = 1; /* llog header is first record */
 918         CWARN("6c: process log %s reversely using client API\n", name);
 919         rc = llog_reverse_process(env, llh, plain_print_cb, NULL, NULL);
 920         if (rc)
 921                 CERROR("6c: llog_reverse_process failed %d\n", rc);
 922         CWARN("6c: processed %d records\n", plain_counter);
 923
 924         rc = verify_handle("6c", llh, plain_counter);
 925         if (rc)
 926                 GOTO(parse_out, rc);
 927
 928 parse_out:
 929         rc2 = llog_close(env, llh);
 930         if (rc2) {
 931                 CERROR("6: llog_close failed: rc = %d\n", rc2);
 932                 if (rc == 0)
 933                         rc = rc2;
 934         }
 935 nctxt_put:
 936         llog_ctxt_put(nctxt);
 937 ctxt_release:
 938         llog_ctxt_put(ctxt);
 939         RETURN(rc);
 940 }
 941
 942 static union {
 943         struct llog_rec_hdr             lrh;   /* common header */
 944         struct llog_logid_rec           llr;   /* LLOG_LOGID_MAGIC */
 945         struct llog_unlink64_rec        lur;   /* MDS_UNLINK64_REC */
 946         struct llog_setattr64_rec       lsr64; /* MDS_SETATTR64_REC */
 947         struct llog_size_change_rec     lscr;  /* OST_SZ_REC */
 948         struct llog_changelog_rec       lcr;   /* CHANGELOG_REC */
 949         struct llog_changelog_user_rec  lcur;  /* CHANGELOG_USER_REC */
 950         struct llog_gen_rec             lgr;   /* LLOG_GEN_REC */
 951 } llog_records;
 952
 953 static int test_7_print_cb(const struct lu_env *env, struct llog_handle *llh,
 954                            struct llog_rec_hdr *rec, void *data)
 955 {
 956         struct lu_fid fid = {0};
 957
 958         logid_to_fid(&llh->lgh_id, &fid);
 959
 960         CDEBUG(D_OTHER, "record type %#x at index %d in log "DFID"\n",
 961                rec->lrh_type, rec->lrh_index, PFID(&fid));
 962
 963         plain_counter++;
 964         return 0;
 965 }
 966
 967 static int test_7_cancel_cb(const struct lu_env *env, struct llog_handle *llh,
 968                             struct llog_rec_hdr *rec, void *data)
 969 {
 970         plain_counter++;
 971         /* test LLOG_DEL_RECORD is working */
 972         return LLOG_DEL_RECORD;
 973 }
 974
 975 static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
 976 {
 977         struct llog_handle      *llh;
 978         int                      rc = 0, i, process_count;
 979         int                      num_recs = 0;
 980
 981         ENTRY;
 982
 983         rc = llog_open_create(env, ctxt, &llh, NULL, NULL);
 984         if (rc) {
 985                 CERROR("7_sub: create log failed\n");
 986                 RETURN(rc);
 987         }
 988
 989         rc = llog_init_handle(env, llh,
 990                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
 991                               &uuid);
 992         if (rc) {
 993                 CERROR("7_sub: can't init llog handle: %d\n", rc);
 994                 GOTO(out_close, rc);
 995         }
 996         for (i = 0; i < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr); i++) {
 997                 rc = llog_write(env, llh, &llog_records.lrh, LLOG_NEXT_IDX);
 998                 if (rc == -ENOSPC) {
 999                         break;
1000                 } else if (rc < 0) {
1001                         CERROR("7_sub: write recs failed at #%d: %d\n",
1002                                i + 1, rc);
1003                         GOTO(out_close, rc);
1004                 }
1005                 num_recs++;
1006         }
1007         if (rc != -ENOSPC) {
1008                 CWARN("7_sub: write record more than BITMAP size!\n");
1009                 GOTO(out_close, rc = -EINVAL);
1010         }
1011
1012         rc = verify_handle("7_sub", llh, num_recs + 1);
1013         if (rc) {
1014                 CERROR("7_sub: verify handle failed: %d\n", rc);
1015                 GOTO(out_close, rc);
1016         }
1017         if (num_recs < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1)
1018                 CWARN("7_sub: records are not aligned, written %d from %u\n",
1019                       num_recs, LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1);
1020
1021         plain_counter = 0;
1022         rc = llog_process(env, llh, test_7_print_cb, "test 7", NULL);
1023         if (rc) {
1024                 CERROR("7_sub: llog process failed: %d\n", rc);
1025                 GOTO(out_close, rc);
1026         }
1027         process_count = plain_counter;
1028         if (process_count != num_recs) {
1029                 CERROR("7_sub: processed %d records from %d total\n",
1030                        process_count, num_recs);
1031                 GOTO(out_close, rc = -EINVAL);
1032         }
1033
1034         plain_counter = 0;
1035         rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);
1036         if (rc && rc != LLOG_DEL_PLAIN) {
1037                 CERROR("7_sub: reverse llog process failed: %d\n", rc);
1038                 GOTO(out_close, rc);
1039         }
1040         if (process_count != plain_counter) {
1041                 CERROR("7_sub: Reverse/direct processing found different"
1042                        "number of records: %d/%d\n",
1043                        plain_counter, process_count);
1044                 GOTO(out_close, rc = -EINVAL);
1045         }
1046         if (llog_exist(llh)) {
1047                 CERROR("7_sub: llog exists but should be zapped\n");
1048                 GOTO(out_close, rc = -EEXIST);
1049         }
1050
1051         rc = verify_handle("7_sub", llh, 1);
1052 out_close:
1053         if (rc)
1054                 llog_destroy(env, llh);
1055         llog_close(env, llh);
1056         RETURN(rc);
1057 }
1058
1059 /* Test all llog records writing and processing */
1060 static int llog_test_7(const struct lu_env *env, struct obd_device *obd)
1061 {
1062         struct llog_ctxt        *ctxt;
1063         int                      rc;
1064
1065         ENTRY;
1066
1067         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1068
1069         CWARN("7a: test llog_logid_rec\n");
1070         llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);
1071         llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);
1072         llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
1073
1074         rc = llog_test_7_sub(env, ctxt);
1075         if (rc) {
1076                 CERROR("7a: llog_logid_rec test failed\n");
1077                 GOTO(out, rc);
1078         }
1079
1080         CWARN("7b: test llog_unlink64_rec\n");
1081         llog_records.lur.lur_hdr.lrh_len = sizeof(llog_records.lur);
1082         llog_records.lur.lur_tail.lrt_len = sizeof(llog_records.lur);
1083         llog_records.lur.lur_hdr.lrh_type = MDS_UNLINK64_REC;
1084
1085         rc = llog_test_7_sub(env, ctxt);
1086         if (rc) {
1087                 CERROR("7b: llog_unlink_rec test failed\n");
1088                 GOTO(out, rc);
1089         }
1090
1091         CWARN("7c: test llog_setattr64_rec\n");
1092         llog_records.lsr64.lsr_hdr.lrh_len = sizeof(llog_records.lsr64);
1093         llog_records.lsr64.lsr_tail.lrt_len = sizeof(llog_records.lsr64);
1094         llog_records.lsr64.lsr_hdr.lrh_type = MDS_SETATTR64_REC;
1095
1096         rc = llog_test_7_sub(env, ctxt);
1097         if (rc) {
1098                 CERROR("7c: llog_setattr64_rec test failed\n");
1099                 GOTO(out, rc);
1100         }
1101
1102         CWARN("7d: test llog_size_change_rec\n");
1103         llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);
1104         llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);
1105         llog_records.lscr.lsc_hdr.lrh_type = OST_SZ_REC;
1106
1107         rc = llog_test_7_sub(env, ctxt);
1108         if (rc) {
1109                 CERROR("7d: llog_size_change_rec test failed\n");
1110                 GOTO(out, rc);
1111         }
1112
1113         CWARN("7e: test llog_changelog_rec\n");
1114         /* Direct access to cr_do_not_use: peculiar case for this test */
1115         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
1116         llog_records.lcr.cr_do_not_use.lrt_len = sizeof(llog_records.lcr);
1117         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
1118
1119         rc = llog_test_7_sub(env, ctxt);
1120         if (rc) {
1121                 CERROR("7e: llog_changelog_rec test failed\n");
1122                 GOTO(out, rc);
1123         }
1124
1125         CWARN("7f: test llog_changelog_user_rec\n");
1126         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
1127         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
1128         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
1129
1130         rc = llog_test_7_sub(env, ctxt);
1131         if (rc) {
1132                 CERROR("7f: llog_changelog_user_rec test failed\n");
1133                 GOTO(out, rc);
1134         }
1135
1136         CWARN("7g: test llog_gen_rec\n");
1137         llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);
1138         llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);
1139         llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
1140
1141         rc = llog_test_7_sub(env, ctxt);
1142         if (rc) {
1143                 CERROR("7g: llog_size_change_rec test failed\n");
1144                 GOTO(out, rc);
1145         }
1146 out:
1147         llog_ctxt_put(ctxt);
1148         RETURN(rc);
1149 }
1150
1151 static int llog_truncate(const struct lu_env *env, struct dt_object *o)
1152 {
1153         struct lu_attr           la;
1154         struct thandle          *th;
1155         struct dt_device        *d;
1156         int                      rc;
1157         ENTRY;
1158
1159         LASSERT(o);
1160         d = lu2dt_dev(o->do_lu.lo_dev);
1161         LASSERT(d);
1162
1163         rc = dt_attr_get(env, o, &la);
1164         if (rc)
1165                 RETURN(rc);
1166
1167         CDEBUG(D_OTHER, "original size "LPU64"\n", la.la_size);
1168         rc = sizeof(struct llog_log_hdr) + sizeof(struct llog_mini_rec);
1169         if (la.la_size < rc) {
1170                 CERROR("too small llog: "LPU64"\n", la.la_size);
1171                 RETURN(0);
1172         }
1173
1174         /* drop 2 records */
1175         la.la_size = la.la_size - (sizeof(struct llog_mini_rec) * 2);
1176         la.la_valid = LA_SIZE;
1177
1178         th = dt_trans_create(env, d);
1179         if (IS_ERR(th))
1180                 RETURN(PTR_ERR(th));
1181
1182         rc = dt_declare_attr_set(env, o, &la, th);
1183         if (rc)
1184                 GOTO(stop, rc);
1185
1186         rc = dt_declare_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
1187
1188         rc = dt_trans_start_local(env, d, th);
1189         if (rc)
1190                 GOTO(stop, rc);
1191
1192         rc = dt_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
1193         if (rc)
1194                 GOTO(stop, rc);
1195
1196         rc = dt_attr_set(env, o, &la, th);
1197         if (rc)
1198                 GOTO(stop, rc);
1199
1200 stop:
1201         dt_trans_stop(env, d, th);
1202
1203         RETURN(rc);
1204 }
1205
1206 static int test_8_cb(const struct lu_env *env, struct llog_handle *llh,
1207                           struct llog_rec_hdr *rec, void *data)
1208 {
1209         plain_counter++;
1210         return 0;
1211 }
1212
1213 static int llog_test_8(const struct lu_env *env, struct obd_device *obd)
1214 {
1215         struct llog_handle      *llh = NULL;
1216         char                     name[10];
1217         int                      rc, rc2, i;
1218         int                      orig_counter;
1219         struct llog_mini_rec     lmr;
1220         struct llog_ctxt        *ctxt;
1221         struct dt_object        *obj = NULL;
1222
1223         ENTRY;
1224
1225         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1226         LASSERT(ctxt);
1227
1228         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
1229         lmr.lmr_hdr.lrh_type = 0xf00f00;
1230
1231         CWARN("8a: fill the first plain llog\n");
1232         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1233         if (rc) {
1234                 CERROR("8a: llog_create with logid failed: %d\n", rc);
1235                 GOTO(out_put, rc);
1236         }
1237
1238         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1239         if (rc) {
1240                 CERROR("8a: can't init llog handle: %d\n", rc);
1241                 GOTO(out, rc);
1242         }
1243
1244         plain_counter = 0;
1245         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1246         if (rc != 0) {
1247                 CERROR("5a: process with test_8_cb failed: %d\n", rc);
1248                 GOTO(out, rc);
1249         }
1250         orig_counter = plain_counter;
1251
1252         for (i = 0; i < 100; i++) {
1253                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1254                 if (rc) {
1255                         CERROR("5a: add record failed\n");
1256                         GOTO(out, rc);
1257                 }
1258         }
1259
1260         /* grab the current plain llog, we'll corrupt it later */
1261         obj = llh->u.chd.chd_current_log->lgh_obj;
1262         LASSERT(obj);
1263         lu_object_get(&obj->do_lu);
1264         CWARN("8a: pin llog "DFID"\n", PFID(lu_object_fid(&obj->do_lu)));
1265
1266         rc2 = llog_cat_close(env, llh);
1267         if (rc2) {
1268                 CERROR("8a: close log %s failed: %d\n", name, rc2);
1269                 if (rc == 0)
1270                         rc = rc2;
1271                 GOTO(out_put, rc);
1272         }
1273
1274         CWARN("8b: fill the second plain llog\n");
1275         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1276         if (rc) {
1277                 CERROR("8b: llog_create with logid failed: %d\n", rc);
1278                 GOTO(out_put, rc);
1279         }
1280
1281         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1282         if (rc) {
1283                 CERROR("8b: can't init llog handle: %d\n", rc);
1284                 GOTO(out, rc);
1285         }
1286
1287         for (i = 0; i < 100; i++) {
1288                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1289                 if (rc) {
1290                         CERROR("8b: add record failed\n");
1291                         GOTO(out, rc);
1292                 }
1293         }
1294         CWARN("8b: second llog "DFID"\n",
1295                 PFID(lu_object_fid(&llh->u.chd.chd_current_log->lgh_obj->do_lu)));
1296
1297         rc2 = llog_cat_close(env, llh);
1298         if (rc2) {
1299                 CERROR("8b: close log %s failed: %d\n", name, rc2);
1300                 if (rc == 0)
1301                         rc = rc2;
1302                 GOTO(out_put, rc);
1303         }
1304
1305         CWARN("8c: drop two records from the first plain llog\n");
1306         llog_truncate(env, obj);
1307
1308         CWARN("8d: count survived records\n");
1309         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1310         if (rc) {
1311                 CERROR("8d: llog_create with logid failed: %d\n", rc);
1312                 GOTO(out_put, rc);
1313         }
1314
1315         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1316         if (rc) {
1317                 CERROR("8d: can't init llog handle: %d\n", rc);
1318                 GOTO(out, rc);
1319         }
1320
1321         plain_counter = 0;
1322         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1323         if (rc != 0) {
1324                 CERROR("8d: process with test_8_cb failed: %d\n", rc);
1325                 GOTO(out, rc);
1326         }
1327
1328         if (orig_counter + 200 - 2 != plain_counter) {
1329                 CERROR("found %d records (expected %d)\n", plain_counter,
1330                        orig_counter + 200 - 2);
1331                 rc = -EIO;
1332         }
1333
1334 out:
1335         CWARN("8d: close re-opened catalog\n");
1336         rc2 = llog_cat_close(env, llh);
1337         if (rc2) {
1338                 CERROR("8d: close log %s failed: %d\n", name, rc2);
1339                 if (rc == 0)
1340                         rc = rc2;
1341         }
1342 out_put:
1343         llog_ctxt_put(ctxt);
1344
1345         if (obj != NULL)
1346                 lu_object_put(env, &obj->do_lu);
1347
1348         RETURN(rc);
1349 }
1350
1351 static int llog_test_9_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
1352 {
1353         struct llog_handle      *llh;
1354         struct lu_fid            fid;
1355         int                      rc = 0;
1356
1357         ENTRY;
1358
1359         rc = llog_open_create(env, ctxt, &llh, NULL, NULL);
1360         if (rc != 0) {
1361                 CERROR("9_sub: create log failed\n");
1362                 RETURN(rc);
1363         }
1364
1365         rc = llog_init_handle(env, llh,
1366                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
1367                               &uuid);
1368         if (rc != 0) {
1369                 CERROR("9_sub: can't init llog handle: %d\n", rc);
1370                 GOTO(out_close, rc);
1371         }
1372
1373         logid_to_fid(&llh->lgh_id, &fid);
1374         fid_to_logid(&fid, &llog_records.llr.lid_id);
1375         rc = llog_write(env, llh, &llog_records.lrh, LLOG_NEXT_IDX);
1376         if (rc < 0) {
1377                 CERROR("9_sub: write recs failed at #1: %d\n", rc);
1378                 GOTO(out_close, rc);
1379         }
1380         CWARN("9_sub: record type %x in log "DFID_NOBRACE"\n",
1381               llog_records.lrh.lrh_type, PFID(&fid));
1382 out_close:
1383         llog_close(env, llh);
1384         RETURN(rc);
1385 }
1386
1387 /* Prepare different types of llog records for llog_reader test*/
1388 static int llog_test_9(const struct lu_env *env, struct obd_device *obd)
1389 {
1390         struct llog_ctxt        *ctxt;
1391         int                      rc;
1392
1393         ENTRY;
1394
1395         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1396
1397         CWARN("9a: test llog_logid_rec\n");
1398         llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);
1399         llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);
1400         llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
1401
1402         rc = llog_test_9_sub(env, ctxt);
1403         if (rc != 0) {
1404                 CERROR("9a: llog_logid_rec test failed\n");
1405                 GOTO(out, rc);
1406         }
1407
1408         CWARN("9b: test llog_obd_cfg_rec\n");
1409         llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);
1410         llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);
1411         llog_records.lscr.lsc_hdr.lrh_type = OBD_CFG_REC;
1412
1413         rc = llog_test_9_sub(env, ctxt);
1414         if (rc != 0) {
1415                 CERROR("9b: llog_obd_cfg_rec test failed\n");
1416                 GOTO(out, rc);
1417         }
1418
1419         CWARN("9c: test llog_changelog_rec\n");
1420         /* Direct access to cr_do_not_use: peculiar case for this test */
1421         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
1422         llog_records.lcr.cr_do_not_use.lrt_len = sizeof(llog_records.lcr);
1423         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
1424
1425         rc = llog_test_9_sub(env, ctxt);
1426         if (rc != 0) {
1427                 CERROR("9c: llog_changelog_rec test failed\n");
1428                 GOTO(out, rc);
1429         }
1430
1431         CWARN("9d: test llog_changelog_user_rec\n");
1432         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
1433         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
1434         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
1435
1436         rc = llog_test_9_sub(env, ctxt);
1437         if (rc != 0) {
1438                 CERROR("9d: llog_changelog_user_rec test failed\n");
1439                 GOTO(out, rc);
1440         }
1441
1442 out:
1443         llog_ctxt_put(ctxt);
1444         RETURN(rc);
1445 }
1446
1447 /* test catalog wrap around */
1448 static int llog_test_10(const struct lu_env *env, struct obd_device *obd)
1449 {
1450         struct llog_handle      *cath;
1451         char                     name[10];
1452         int                      rc, rc2, i, enospc, eok;
1453         struct llog_mini_rec     lmr;
1454         struct llog_ctxt        *ctxt;
1455         struct lu_attr           la;
1456         __u64                    cat_max_size;
1457         struct dt_device        *dt;
1458
1459         ENTRY;
1460
1461         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1462         LASSERT(ctxt);
1463
1464         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
1465         lmr.lmr_hdr.lrh_type = 0xf00f00;
1466
1467         snprintf(name, sizeof(name), "%x", llog_test_rand + 2);
1468         CWARN("10a: create a catalog log with name: %s\n", name);
1469         rc = llog_open_create(env, ctxt, &cath, NULL, name);
1470         if (rc) {
1471                 CERROR("10a: llog_create with name %s failed: %d\n", name, rc);
1472                 GOTO(ctxt_release, rc);
1473         }
1474         rc = llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);
1475         if (rc) {
1476                 CERROR("10a: can't init llog handle: %d\n", rc);
1477                 GOTO(out, rc);
1478         }
1479
1480         cat_logid = cath->lgh_id;
1481         dt = lu2dt_dev(cath->lgh_obj->do_lu.lo_dev);
1482
1483         /* sync device to commit all recent LLOG changes to disk and avoid
1484          * to consume a huge space with delayed journal commit callbacks
1485          * particularly on low memory nodes or VMs */
1486         rc = dt_sync(env, dt);
1487         if (rc) {
1488                 CERROR("10c: sync failed: %d\n", rc);
1489                 GOTO(out, rc);
1490         }
1491
1492         /* force catalog wrap for 5th plain LLOG */
1493         cfs_fail_loc = CFS_FAIL_SKIP|OBD_FAIL_CAT_RECORDS;
1494         cfs_fail_val = 4;
1495
1496         CWARN("10b: write %d log records\n", LLOG_TEST_RECNUM);
1497         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1498                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1499                 if (rc) {
1500                         CERROR("10b: write %d records failed at #%d: %d\n",
1501                                LLOG_TEST_RECNUM, i + 1, rc);
1502                         GOTO(out, rc);
1503                 }
1504         }
1505
1506         /* make sure 2 new plain llog appears in catalog (+1 with hdr) */
1507         rc = verify_handle("10b", cath, 3);
1508         if (rc)
1509                 GOTO(out, rc);
1510
1511         /* sync device to commit all recent LLOG changes to disk and avoid
1512          * to consume a huge space with delayed journal commit callbacks
1513          * particularly on low memory nodes or VMs */
1514         rc = dt_sync(env, dt);
1515         if (rc) {
1516                 CERROR("10b: sync failed: %d\n", rc);
1517                 GOTO(out, rc);
1518         }
1519
1520         CWARN("10c: write %d more log records\n", 2 * LLOG_TEST_RECNUM);
1521         for (i = 0; i < 2 * LLOG_TEST_RECNUM; i++) {
1522                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1523                 if (rc) {
1524                         CERROR("10c: write %d records failed at #%d: %d\n",
1525                                2*LLOG_TEST_RECNUM, i + 1, rc);
1526                         GOTO(out, rc);
1527                 }
1528         }
1529
1530         /* make sure 2 new plain llog appears in catalog (+1 with hdr) */
1531         rc = verify_handle("10c", cath, 5);
1532         if (rc)
1533                 GOTO(out, rc);
1534
1535         /* sync device to commit all recent LLOG changes to disk and avoid
1536          * to consume a huge space with delayed journal commit callbacks
1537          * particularly on low memory nodes or VMs */
1538         rc = dt_sync(env, dt);
1539         if (rc) {
1540                 CERROR("10c: sync failed: %d\n", rc);
1541                 GOTO(out, rc);
1542         }
1543
1544         /* fill last allocated plain LLOG and reach -ENOSPC condition
1545          * because no slot available in Catalog */
1546         enospc = 0;
1547         eok = 0;
1548         CWARN("10c: write %d more log records\n", LLOG_TEST_RECNUM);
1549         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1550                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1551                 if (rc && rc != -ENOSPC) {
1552                         CERROR("10c: write %d records failed at #%d: %d\n",
1553                                LLOG_TEST_RECNUM, i + 1, rc);
1554                         GOTO(out, rc);
1555                 }
1556                 /* after last added plain LLOG has filled up, all new
1557                  * records add should fail with -ENOSPC */
1558                 if (rc == -ENOSPC) {
1559                         enospc++;
1560                 } else {
1561                         enospc = 0;
1562                         eok++;
1563                 }
1564         }
1565
1566         if ((enospc == 0) && (enospc+eok != LLOG_TEST_RECNUM)) {
1567                 CERROR("10c: all last records adds should have failed with"
1568                        " -ENOSPC\n");
1569                 GOTO(out, rc = -EINVAL);
1570         }
1571
1572         CWARN("10c: wrote %d records then %d failed with ENOSPC\n", eok,
1573               enospc);
1574
1575         /* make sure no new record in Catalog */
1576         rc = verify_handle("10c", cath, 5);
1577         if (rc)
1578                 GOTO(out, rc);
1579
1580         /* Catalog should have reached its max size for test */
1581         rc = dt_attr_get(env, cath->lgh_obj, &la);
1582         if (rc) {
1583                 CERROR("10c: failed to get catalog attrs: %d\n", rc);
1584                 GOTO(out, rc);
1585         }
1586         cat_max_size = la.la_size;
1587
1588         /* cancel all 1st plain llog records to empty it, this will also cause
1589          * its catalog entry to be freed for next forced wrap in 10e */
1590         CWARN("10d: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1591         cancel_count = 0;
1592         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1593         if (rc != -LLOG_EEMPTY) {
1594                 CERROR("10d: process with llog_cancel_rec_cb failed: %d\n", rc);
1595                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1596                  * not reached */
1597                 if (rc == 0)
1598                         rc = -ERANGE;
1599                 GOTO(out, rc);
1600         }
1601
1602         CWARN("10d: print the catalog entries.. we expect 3\n");
1603         cat_counter = 0;
1604         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1605         if (rc) {
1606                 CERROR("10d: process with cat_print_cb failed: %d\n", rc);
1607                 GOTO(out, rc);
1608         }
1609         if (cat_counter != 3) {
1610                 CERROR("10d: %d entries in catalog\n", cat_counter);
1611                 GOTO(out, rc = -EINVAL);
1612         }
1613
1614         /* verify one down in catalog (+1 with hdr) */
1615         rc = verify_handle("10d", cath, 4);
1616         if (rc)
1617                 GOTO(out, rc);
1618
1619         /* sync device to commit all recent LLOG changes to disk and avoid
1620          * to consume a huge space with delayed journal commit callbacks
1621          * particularly on low memory nodes or VMs */
1622         rc = dt_sync(env, dt);
1623         if (rc) {
1624                 CERROR("10d: sync failed: %d\n", rc);
1625                 GOTO(out, rc);
1626         }
1627
1628         enospc = 0;
1629         eok = 0;
1630         CWARN("10e: write %d more log records\n", LLOG_TEST_RECNUM);
1631         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1632                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1633                 if (rc && rc != -ENOSPC) {
1634                         CERROR("10e: write %d records failed at #%d: %d\n",
1635                                LLOG_TEST_RECNUM, i + 1, rc);
1636                         GOTO(out, rc);
1637                 }
1638                 /* after last added plain LLOG has filled up, all new
1639                  * records add should fail with -ENOSPC */
1640                 if (rc == -ENOSPC) {
1641                         enospc++;
1642                 } else {
1643                         enospc = 0;
1644                         eok++;
1645                 }
1646         }
1647
1648         if ((enospc == 0) && (enospc+eok != LLOG_TEST_RECNUM)) {
1649                 CERROR("10e: all last records adds should have failed with"
1650                        " -ENOSPC\n");
1651                 GOTO(out, rc = -EINVAL);
1652         }
1653
1654         CWARN("10e: wrote %d records then %d failed with ENOSPC\n", eok,
1655               enospc);
1656
1657         CWARN("10e: print the catalog entries.. we expect 4\n");
1658         cat_counter = 0;
1659         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1660         if (rc) {
1661                 CERROR("10d: process with cat_print_cb failed: %d\n", rc);
1662                 GOTO(out, rc);
1663         }
1664         if (cat_counter != 4) {
1665                 CERROR("10d: %d entries in catalog\n", cat_counter);
1666                 GOTO(out, rc = -EINVAL);
1667         }
1668
1669         /* make sure 1 new plain llog appears in catalog (+1 with hdr) */
1670         rc = verify_handle("10e", cath, 5);
1671         if (rc)
1672                 GOTO(out, rc);
1673
1674         /* verify catalog has wrap around */
1675         if (cath->lgh_last_idx > cath->lgh_hdr->llh_cat_idx) {
1676                 CERROR("10e: catalog failed to wrap around\n");
1677                 GOTO(out, rc = -EINVAL);
1678         }
1679
1680         rc = dt_attr_get(env, cath->lgh_obj, &la);
1681         if (rc) {
1682                 CERROR("10e: failed to get catalog attrs: %d\n", rc);
1683                 GOTO(out, rc);
1684         }
1685
1686         if (la.la_size != cat_max_size) {
1687                 CERROR("10e: catalog size has changed after it has wrap around,"
1688                        " current size = "LPU64", expected size = "LPU64"\n",
1689                        la.la_size, cat_max_size);
1690                 GOTO(out, rc = -EINVAL);
1691         }
1692         CWARN("10e: catalog successfully wrap around, last_idx %d, first %d\n",
1693               cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1694
1695         /* sync device to commit all recent LLOG changes to disk and avoid
1696          * to consume a huge space with delayed journal commit callbacks
1697          * particularly on low memory nodes or VMs */
1698         rc = dt_sync(env, dt);
1699         if (rc) {
1700                 CERROR("10e: sync failed: %d\n", rc);
1701                 GOTO(out, rc);
1702         }
1703
1704         /* cancel more records to free one more slot in Catalog
1705          * see if it is re-allocated when adding more records */
1706         CWARN("10f: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1707         cancel_count = 0;
1708         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1709         if (rc != -LLOG_EEMPTY) {
1710                 CERROR("10f: process with llog_cancel_rec_cb failed: %d\n", rc);
1711                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1712                  * not reached */
1713                 if (rc == 0)
1714                         rc = -ERANGE;
1715                 GOTO(out, rc);
1716         }
1717
1718         CWARN("10f: print the catalog entries.. we expect 3\n");
1719         cat_counter = 0;
1720         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1721         if (rc) {
1722                 CERROR("10f: process with cat_print_cb failed: %d\n", rc);
1723                 GOTO(out, rc);
1724         }
1725         if (cat_counter != 3) {
1726                 CERROR("10f: %d entries in catalog\n", cat_counter);
1727                 GOTO(out, rc = -EINVAL);
1728         }
1729
1730         /* verify one down in catalog (+1 with hdr) */
1731         rc = verify_handle("10f", cath, 4);
1732         if (rc)
1733                 GOTO(out, rc);
1734
1735         /* sync device to commit all recent LLOG changes to disk and avoid
1736          * to consume a huge space with delayed journal commit callbacks
1737          * particularly on low memory nodes or VMs */
1738         rc = dt_sync(env, dt);
1739         if (rc) {
1740                 CERROR("10f: sync failed: %d\n", rc);
1741                 GOTO(out, rc);
1742         }
1743
1744         enospc = 0;
1745         eok = 0;
1746         CWARN("10f: write %d more log records\n", LLOG_TEST_RECNUM);
1747         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1748                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1749                 if (rc && rc != -ENOSPC) {
1750                         CERROR("10f: write %d records failed at #%d: %d\n",
1751                                LLOG_TEST_RECNUM, i + 1, rc);
1752                         GOTO(out, rc);
1753                 }
1754                 /* after last added plain LLOG has filled up, all new
1755                  * records add should fail with -ENOSPC */
1756                 if (rc == -ENOSPC) {
1757                         enospc++;
1758                 } else {
1759                         enospc = 0;
1760                         eok++;
1761                 }
1762         }
1763
1764         if ((enospc == 0) && (enospc+eok != LLOG_TEST_RECNUM)) {
1765                 CERROR("10f: all last records adds should have failed with"
1766                        " -ENOSPC\n");
1767                 GOTO(out, rc = -EINVAL);
1768         }
1769
1770         CWARN("10f: wrote %d records then %d failed with ENOSPC\n", eok,
1771               enospc);
1772
1773         /* make sure 1 new plain llog appears in catalog (+1 with hdr) */
1774         rc = verify_handle("10f", cath, 5);
1775         if (rc)
1776                 GOTO(out, rc);
1777
1778         /* verify lgh_last_idx = llh_cat_idx = 2 now */
1779         if (cath->lgh_last_idx != cath->lgh_hdr->llh_cat_idx ||
1780             cath->lgh_last_idx != 2) {
1781                 CERROR("10f: lgh_last_idx = %d vs 2, llh_cat_idx = %d vs 2\n",
1782                        cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1783                 GOTO(out, rc = -EINVAL);
1784         }
1785
1786         rc = dt_attr_get(env, cath->lgh_obj, &la);
1787         if (rc) {
1788                 CERROR("10f: failed to get catalog attrs: %d\n", rc);
1789                 GOTO(out, rc);
1790         }
1791
1792         if (la.la_size != cat_max_size) {
1793                 CERROR("10f: catalog size has changed after it has wrap around,"
1794                        " current size = "LPU64", expected size = "LPU64"\n",
1795                        la.la_size, cat_max_size);
1796                 GOTO(out, rc = -EINVAL);
1797         }
1798
1799         /* sync device to commit all recent LLOG changes to disk and avoid
1800          * to consume a huge space with delayed journal commit callbacks
1801          * particularly on low memory nodes or VMs */
1802         rc = dt_sync(env, dt);
1803         if (rc) {
1804                 CERROR("10f: sync failed: %d\n", rc);
1805                 GOTO(out, rc);
1806         }
1807
1808         /* will llh_cat_idx also successfully wrap ? */
1809
1810         /* cancel all records in the plain LLOGs referenced by 2 last indexes in
1811          * Catalog */
1812
1813         /* cancel more records to free one more slot in Catalog */
1814         CWARN("10g: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1815         cancel_count = 0;
1816         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1817         if (rc != -LLOG_EEMPTY) {
1818                 CERROR("10g: process with llog_cancel_rec_cb failed: %d\n", rc);
1819                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1820                  * not reached */
1821                 if (rc == 0)
1822                         rc = -ERANGE;
1823                 GOTO(out, rc);
1824         }
1825
1826         CWARN("10g: print the catalog entries.. we expect 3\n");
1827         cat_counter = 0;
1828         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1829         if (rc) {
1830                 CERROR("10g: process with cat_print_cb failed: %d\n", rc);
1831                 GOTO(out, rc);
1832         }
1833         if (cat_counter != 3) {
1834                 CERROR("10g: %d entries in catalog\n", cat_counter);
1835                 GOTO(out, rc = -EINVAL);
1836         }
1837
1838         /* verify one down in catalog (+1 with hdr) */
1839         rc = verify_handle("10g", cath, 4);
1840         if (rc)
1841                 GOTO(out, rc);
1842
1843         /* sync device to commit all recent LLOG changes to disk and avoid
1844          * to consume a huge space with delayed journal commit callbacks
1845          * particularly on low memory nodes or VMs */
1846         rc = dt_sync(env, dt);
1847         if (rc) {
1848                 CERROR("10g: sync failed: %d\n", rc);
1849                 GOTO(out, rc);
1850         }
1851
1852         /* cancel more records to free one more slot in Catalog */
1853         CWARN("10g: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1854         cancel_count = 0;
1855         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1856         if (rc != -LLOG_EEMPTY) {
1857                 CERROR("10g: process with llog_cancel_rec_cb failed: %d\n", rc);
1858                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1859                  * not reached */
1860                 if (rc == 0)
1861                         rc = -ERANGE;
1862                 GOTO(out, rc);
1863         }
1864
1865         CWARN("10g: print the catalog entries.. we expect 2\n");
1866         cat_counter = 0;
1867         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1868         if (rc) {
1869                 CERROR("10g: process with cat_print_cb failed: %d\n", rc);
1870                 GOTO(out, rc);
1871         }
1872         if (cat_counter != 2) {
1873                 CERROR("10g: %d entries in catalog\n", cat_counter);
1874                 GOTO(out, rc = -EINVAL);
1875         }
1876
1877         /* verify one down in catalog (+1 with hdr) */
1878         rc = verify_handle("10g", cath, 3);
1879         if (rc)
1880                 GOTO(out, rc);
1881
1882         /* verify lgh_last_idx = 2 and llh_cat_idx = 0 now */
1883         if (cath->lgh_hdr->llh_cat_idx != 0 ||
1884             cath->lgh_last_idx != 2) {
1885                 CERROR("10g: lgh_last_idx = %d vs 2, llh_cat_idx = %d vs 0\n",
1886                        cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1887                 GOTO(out, rc = -EINVAL);
1888         }
1889
1890         /* sync device to commit all recent LLOG changes to disk and avoid
1891          * to consume a huge space with delayed journal commit callbacks
1892          * particularly on low memory nodes or VMs */
1893         rc = dt_sync(env, dt);
1894         if (rc) {
1895                 CERROR("10g: sync failed: %d\n", rc);
1896                 GOTO(out, rc);
1897         }
1898
1899         /* cancel more records to free one more slot in Catalog */
1900         CWARN("10g: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1901         cancel_count = 0;
1902         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1903         if (rc != -LLOG_EEMPTY) {
1904                 CERROR("10g: process with llog_cancel_rec_cb failed: %d\n", rc);
1905                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1906                  * not reached */
1907                 if (rc == 0)
1908                         rc = -ERANGE;
1909                 GOTO(out, rc);
1910         }
1911
1912         CWARN("10g: print the catalog entries.. we expect 1\n");
1913         cat_counter = 0;
1914         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1915         if (rc) {
1916                 CERROR("10g: process with cat_print_cb failed: %d\n", rc);
1917                 GOTO(out, rc);
1918         }
1919         if (cat_counter != 1) {
1920                 CERROR("10g: %d entries in catalog\n", cat_counter);
1921                 GOTO(out, rc = -EINVAL);
1922         }
1923
1924         /* verify one down in catalog (+1 with hdr) */
1925         rc = verify_handle("10g", cath, 2);
1926         if (rc)
1927                 GOTO(out, rc);
1928
1929         /* verify lgh_last_idx = 2 and llh_cat_idx = 1 now */
1930         if (cath->lgh_hdr->llh_cat_idx != 1 ||
1931             cath->lgh_last_idx != 2) {
1932                 CERROR("10g: lgh_last_idx = %d vs 2, llh_cat_idx = %d vs 1\n",
1933                        cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1934                 GOTO(out, rc = -EINVAL);
1935         }
1936
1937         CWARN("10g: llh_cat_idx has also successfully wrapped!\n");
1938
1939 out:
1940         cfs_fail_loc = 0;
1941         cfs_fail_val = 0;
1942
1943         CWARN("10: put newly-created catalog\n");
1944         rc2 = llog_cat_close(env, cath);
1945         if (rc2) {
1946                 CERROR("10: close log %s failed: %d\n", name, rc2);
1947                 if (rc == 0)
1948                         rc = rc2;
1949         }
1950 ctxt_release:
1951         llog_ctxt_put(ctxt);
1952         RETURN(rc);
1953 }
1954
1955 /* -------------------------------------------------------------------------
1956  * Tests above, boring obd functions below
1957  * ------------------------------------------------------------------------- */
1958 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
1959 {
1960         struct llog_handle      *llh = NULL;
1961         struct llog_ctxt        *ctxt;
1962         int                      rc, err;
1963         char                     name[10];
1964
1965         ENTRY;
1966         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1967         LASSERT(ctxt);
1968
1969         sprintf(name, "%x", llog_test_rand);
1970
1971         rc = llog_test_1(env, obd, name);
1972         if (rc)
1973                 GOTO(cleanup_ctxt, rc);
1974
1975         rc = llog_test_2(env, obd, name, &llh);
1976         if (rc)
1977                 GOTO(cleanup_ctxt, rc);
1978
1979         rc = llog_test_3(env, obd, llh);
1980         if (rc)
1981                 GOTO(cleanup, rc);
1982
1983         rc = llog_test_4(env, obd);
1984         if (rc)
1985                 GOTO(cleanup, rc);
1986
1987         rc = llog_test_5(env, obd);
1988         if (rc)
1989                 GOTO(cleanup, rc);
1990
1991         rc = llog_test_6(env, obd, name);
1992         if (rc)
1993                 GOTO(cleanup, rc);
1994
1995         rc = llog_test_7(env, obd);
1996         if (rc)
1997                 GOTO(cleanup, rc);
1998
1999         rc = llog_test_8(env, obd);
2000         if (rc)
2001                 GOTO(cleanup, rc);
2002
2003         rc = llog_test_9(env, obd);
2004         if (rc != 0)
2005                 GOTO(cleanup, rc);
2006
2007         rc = llog_test_10(env, obd);
2008         if (rc)
2009                 GOTO(cleanup, rc);
2010
2011 cleanup:
2012         err = llog_destroy(env, llh);
2013         if (err)
2014                 CERROR("cleanup: llog_destroy failed: %d\n", err);
2015         llog_close(env, llh);
2016         if (rc == 0)
2017                 rc = err;
2018 cleanup_ctxt:
2019         llog_ctxt_put(ctxt);
2020         return rc;
2021 }
2022
2023 static int llog_test_cleanup(struct obd_device *obd)
2024 {
2025         struct obd_device       *tgt;
2026         struct lu_env            env;
2027         int                      rc;
2028
2029         ENTRY;
2030
2031         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
2032         if (rc)
2033                 RETURN(rc);
2034
2035         tgt = obd->obd_lvfs_ctxt.dt->dd_lu_dev.ld_obd;
2036         rc = llog_cleanup(&env, llog_get_context(tgt, LLOG_TEST_ORIG_CTXT));
2037         if (rc)
2038                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
2039         lu_env_fini(&env);
2040         RETURN(rc);
2041 }
2042
2043 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
2044 {
2045         struct obd_device       *tgt;
2046         struct llog_ctxt        *ctxt;
2047         struct dt_object        *o;
2048         struct lu_env            env;
2049         struct lu_context        test_session;
2050         int                      rc;
2051
2052         ENTRY;
2053
2054         if (lcfg->lcfg_bufcount < 2) {
2055                 CERROR("requires a TARGET OBD name\n");
2056                 RETURN(-EINVAL);
2057         }
2058
2059         if (lcfg->lcfg_buflens[1] < 1) {
2060                 CERROR("requires a TARGET OBD name\n");
2061                 RETURN(-EINVAL);
2062         }
2063
2064         /* disk obd */
2065         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
2066         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
2067                 CERROR("target device not attached or not set up (%s)\n",
2068                        lustre_cfg_string(lcfg, 1));
2069                 RETURN(-EINVAL);
2070         }
2071
2072         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
2073         if (rc)
2074                 RETURN(rc);
2075
2076         rc = lu_context_init(&test_session, LCT_SERVER_SESSION);
2077         if (rc)
2078                 GOTO(cleanup_env, rc);
2079         test_session.lc_thread = (struct ptlrpc_thread *)current;
2080         lu_context_enter(&test_session);
2081         env.le_ses = &test_session;
2082
2083         CWARN("Setup llog-test device over %s device\n",
2084               lustre_cfg_string(lcfg, 1));
2085
2086         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
2087         obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);
2088
2089         rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,
2090                         &llog_osd_ops);
2091         if (rc)
2092                 GOTO(cleanup_session, rc);
2093
2094         /* use MGS llog dir for tests */
2095         ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);
2096         LASSERT(ctxt);
2097         o = ctxt->loc_dir;
2098         llog_ctxt_put(ctxt);
2099
2100         ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);
2101         LASSERT(ctxt);
2102         ctxt->loc_dir = o;
2103         llog_ctxt_put(ctxt);
2104
2105         llog_test_rand = cfs_rand();
2106
2107         rc = llog_run_tests(&env, tgt);
2108         if (rc)
2109                 llog_test_cleanup(obd);
2110 cleanup_session:
2111         lu_context_exit(&test_session);
2112         lu_context_fini(&test_session);
2113 cleanup_env:
2114         lu_env_fini(&env);
2115         RETURN(rc);
2116 }
2117
2118 static struct obd_ops llog_obd_ops = {
2119         .o_owner       = THIS_MODULE,
2120         .o_setup       = llog_test_setup,
2121         .o_cleanup     = llog_test_cleanup,
2122 };
2123
2124 static int __init llog_test_init(void)
2125 {
2126         return class_register_type(&llog_obd_ops, NULL, true, NULL,
2127                                    "llog_test", NULL);
2128 }
2129
2130 static void __exit llog_test_exit(void)
2131 {
2132         class_unregister_type("llog_test");
2133 }
2134
2135 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2136 MODULE_DESCRIPTION("Lustre Log test module");
2137 MODULE_VERSION(LUSTRE_VERSION_STRING);
2138 MODULE_LICENSE("GPL");
2139
2140 module_init(llog_test_init);
2141 module_exit(llog_test_exit);