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         /* force catalog wrap for 5th plain LLOG */
1484         cfs_fail_loc = CFS_FAIL_SKIP|OBD_FAIL_CAT_RECORDS;
1485         cfs_fail_val = 4;
1486
1487         CWARN("10b: write %d log records\n", LLOG_TEST_RECNUM);
1488         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1489                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1490                 if (rc) {
1491                         CERROR("10b: write %d records failed at #%d: %d\n",
1492                                LLOG_TEST_RECNUM, i + 1, rc);
1493                         GOTO(out, rc);
1494                 }
1495         }
1496
1497         /* make sure 2 new plain llog appears in catalog (+1 with hdr) */
1498         rc = verify_handle("10b", cath, 3);
1499         if (rc)
1500                 GOTO(out, rc);
1501
1502         /* sync device to commit all recent LLOG changes to disk and avoid
1503          * to consume a huge space with delayed journal commit callbacks
1504          * particularly on low memory nodes or VMs */
1505         rc = dt_sync(env, dt);
1506         if (rc) {
1507                 CERROR("10b: sync failed: %d\n", rc);
1508                 GOTO(out, rc);
1509         }
1510
1511         CWARN("10c: write %d more log records\n", 2 * LLOG_TEST_RECNUM);
1512         for (i = 0; i < 2 * LLOG_TEST_RECNUM; i++) {
1513                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1514                 if (rc) {
1515                         CERROR("10c: write %d records failed at #%d: %d\n",
1516                                2*LLOG_TEST_RECNUM, i + 1, rc);
1517                         GOTO(out, rc);
1518                 }
1519         }
1520
1521         /* make sure 2 new plain llog appears in catalog (+1 with hdr) */
1522         rc = verify_handle("10c", cath, 5);
1523         if (rc)
1524                 GOTO(out, rc);
1525
1526         /* sync device to commit all recent LLOG changes to disk and avoid
1527          * to consume a huge space with delayed journal commit callbacks
1528          * particularly on low memory nodes or VMs */
1529         rc = dt_sync(env, dt);
1530         if (rc) {
1531                 CERROR("10c: sync failed: %d\n", rc);
1532                 GOTO(out, rc);
1533         }
1534
1535         /* fill last allocated plain LLOG and reach -ENOSPC condition
1536          * because no slot available in Catalog */
1537         enospc = 0;
1538         eok = 0;
1539         CWARN("10c: write %d more log records\n", LLOG_TEST_RECNUM);
1540         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1541                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1542                 if (rc && rc != -ENOSPC) {
1543                         CERROR("10c: write %d records failed at #%d: %d\n",
1544                                LLOG_TEST_RECNUM, i + 1, rc);
1545                         GOTO(out, rc);
1546                 }
1547                 /* after last added plain LLOG has filled up, all new
1548                  * records add should fail with -ENOSPC */
1549                 if (rc == -ENOSPC) {
1550                         enospc++;
1551                 } else {
1552                         enospc = 0;
1553                         eok++;
1554                 }
1555         }
1556
1557         if ((enospc == 0) && (enospc+eok != LLOG_TEST_RECNUM)) {
1558                 CERROR("10c: all last records adds should have failed with"
1559                        " -ENOSPC\n");
1560                 GOTO(out, rc = -EINVAL);
1561         }
1562
1563         CWARN("10c: wrote %d records then %d failed with ENOSPC\n", eok,
1564               enospc);
1565
1566         /* make sure no new record in Catalog */
1567         rc = verify_handle("10c", cath, 5);
1568         if (rc)
1569                 GOTO(out, rc);
1570
1571         /* Catalog should have reached its max size for test */
1572         rc = dt_attr_get(env, cath->lgh_obj, &la);
1573         if (rc) {
1574                 CERROR("10c: failed to get catalog attrs: %d\n", rc);
1575                 GOTO(out, rc);
1576         }
1577         cat_max_size = la.la_size;
1578
1579         /* cancel all 1st plain llog records to empty it, this will also cause
1580          * its catalog entry to be freed for next forced wrap in 10e */
1581         CWARN("10d: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1582         cancel_count = 0;
1583         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1584         if (rc != -LLOG_EEMPTY) {
1585                 CERROR("10d: process with llog_cancel_rec_cb failed: %d\n", rc);
1586                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1587                  * not reached */
1588                 if (rc == 0)
1589                         rc = -ERANGE;
1590                 GOTO(out, rc);
1591         }
1592
1593         CWARN("10d: print the catalog entries.. we expect 3\n");
1594         cat_counter = 0;
1595         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1596         if (rc) {
1597                 CERROR("10d: process with cat_print_cb failed: %d\n", rc);
1598                 GOTO(out, rc);
1599         }
1600         if (cat_counter != 3) {
1601                 CERROR("10d: %d entries in catalog\n", cat_counter);
1602                 GOTO(out, rc = -EINVAL);
1603         }
1604
1605         /* verify one down in catalog (+1 with hdr) */
1606         rc = verify_handle("10d", cath, 4);
1607         if (rc)
1608                 GOTO(out, rc);
1609
1610         /* sync device to commit all recent LLOG changes to disk and avoid
1611          * to consume a huge space with delayed journal commit callbacks
1612          * particularly on low memory nodes or VMs */
1613         rc = dt_sync(env, dt);
1614         if (rc) {
1615                 CERROR("10d: sync failed: %d\n", rc);
1616                 GOTO(out, rc);
1617         }
1618
1619         enospc = 0;
1620         eok = 0;
1621         CWARN("10e: write %d more log records\n", LLOG_TEST_RECNUM);
1622         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1623                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1624                 if (rc && rc != -ENOSPC) {
1625                         CERROR("10e: write %d records failed at #%d: %d\n",
1626                                LLOG_TEST_RECNUM, i + 1, rc);
1627                         GOTO(out, rc);
1628                 }
1629                 /* after last added plain LLOG has filled up, all new
1630                  * records add should fail with -ENOSPC */
1631                 if (rc == -ENOSPC) {
1632                         enospc++;
1633                 } else {
1634                         enospc = 0;
1635                         eok++;
1636                 }
1637         }
1638
1639         if ((enospc == 0) && (enospc+eok != LLOG_TEST_RECNUM)) {
1640                 CERROR("10e: all last records adds should have failed with"
1641                        " -ENOSPC\n");
1642                 GOTO(out, rc = -EINVAL);
1643         }
1644
1645         CWARN("10e: wrote %d records then %d failed with ENOSPC\n", eok,
1646               enospc);
1647
1648         CWARN("10e: print the catalog entries.. we expect 4\n");
1649         cat_counter = 0;
1650         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1651         if (rc) {
1652                 CERROR("10d: process with cat_print_cb failed: %d\n", rc);
1653                 GOTO(out, rc);
1654         }
1655         if (cat_counter != 4) {
1656                 CERROR("10d: %d entries in catalog\n", cat_counter);
1657                 GOTO(out, rc = -EINVAL);
1658         }
1659
1660         /* make sure 1 new plain llog appears in catalog (+1 with hdr) */
1661         rc = verify_handle("10e", cath, 5);
1662         if (rc)
1663                 GOTO(out, rc);
1664
1665         /* verify catalog has wrap around */
1666         if (cath->lgh_last_idx > cath->lgh_hdr->llh_cat_idx) {
1667                 CERROR("10e: catalog failed to wrap around\n");
1668                 GOTO(out, rc = -EINVAL);
1669         }
1670
1671         rc = dt_attr_get(env, cath->lgh_obj, &la);
1672         if (rc) {
1673                 CERROR("10e: failed to get catalog attrs: %d\n", rc);
1674                 GOTO(out, rc);
1675         }
1676
1677         if (la.la_size != cat_max_size) {
1678                 CERROR("10e: catalog size has changed after it has wrap around,"
1679                        " current size = "LPU64", expected size = "LPU64"\n",
1680                        la.la_size, cat_max_size);
1681                 GOTO(out, rc = -EINVAL);
1682         }
1683         CWARN("10e: catalog successfully wrap around, last_idx %d, first %d\n",
1684               cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1685
1686         /* sync device to commit all recent LLOG changes to disk and avoid
1687          * to consume a huge space with delayed journal commit callbacks
1688          * particularly on low memory nodes or VMs */
1689         rc = dt_sync(env, dt);
1690         if (rc) {
1691                 CERROR("10e: sync failed: %d\n", rc);
1692                 GOTO(out, rc);
1693         }
1694
1695         /* cancel more records to free one more slot in Catalog
1696          * see if it is re-allocated when adding more records */
1697         CWARN("10f: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1698         cancel_count = 0;
1699         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1700         if (rc != -LLOG_EEMPTY) {
1701                 CERROR("10f: process with llog_cancel_rec_cb failed: %d\n", rc);
1702                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1703                  * not reached */
1704                 if (rc == 0)
1705                         rc = -ERANGE;
1706                 GOTO(out, rc);
1707         }
1708
1709         CWARN("10f: print the catalog entries.. we expect 3\n");
1710         cat_counter = 0;
1711         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1712         if (rc) {
1713                 CERROR("10f: process with cat_print_cb failed: %d\n", rc);
1714                 GOTO(out, rc);
1715         }
1716         if (cat_counter != 3) {
1717                 CERROR("10f: %d entries in catalog\n", cat_counter);
1718                 GOTO(out, rc = -EINVAL);
1719         }
1720
1721         /* verify one down in catalog (+1 with hdr) */
1722         rc = verify_handle("10f", cath, 4);
1723         if (rc)
1724                 GOTO(out, rc);
1725
1726         /* sync device to commit all recent LLOG changes to disk and avoid
1727          * to consume a huge space with delayed journal commit callbacks
1728          * particularly on low memory nodes or VMs */
1729         rc = dt_sync(env, dt);
1730         if (rc) {
1731                 CERROR("10f: sync failed: %d\n", rc);
1732                 GOTO(out, rc);
1733         }
1734
1735         enospc = 0;
1736         eok = 0;
1737         CWARN("10f: write %d more log records\n", LLOG_TEST_RECNUM);
1738         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
1739                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
1740                 if (rc && rc != -ENOSPC) {
1741                         CERROR("10f: write %d records failed at #%d: %d\n",
1742                                LLOG_TEST_RECNUM, i + 1, rc);
1743                         GOTO(out, rc);
1744                 }
1745                 /* after last added plain LLOG has filled up, all new
1746                  * records add should fail with -ENOSPC */
1747                 if (rc == -ENOSPC) {
1748                         enospc++;
1749                 } else {
1750                         enospc = 0;
1751                         eok++;
1752                 }
1753         }
1754
1755         if ((enospc == 0) && (enospc+eok != LLOG_TEST_RECNUM)) {
1756                 CERROR("10f: all last records adds should have failed with"
1757                        " -ENOSPC\n");
1758                 GOTO(out, rc = -EINVAL);
1759         }
1760
1761         CWARN("10f: wrote %d records then %d failed with ENOSPC\n", eok,
1762               enospc);
1763
1764         /* make sure 1 new plain llog appears in catalog (+1 with hdr) */
1765         rc = verify_handle("10f", cath, 5);
1766         if (rc)
1767                 GOTO(out, rc);
1768
1769         /* verify lgh_last_idx = llh_cat_idx = 2 now */
1770         if (cath->lgh_last_idx != cath->lgh_hdr->llh_cat_idx ||
1771             cath->lgh_last_idx != 2) {
1772                 CERROR("10f: lgh_last_idx = %d vs 2, llh_cat_idx = %d vs 2\n",
1773                        cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1774                 GOTO(out, rc = -EINVAL);
1775         }
1776
1777         rc = dt_attr_get(env, cath->lgh_obj, &la);
1778         if (rc) {
1779                 CERROR("10f: failed to get catalog attrs: %d\n", rc);
1780                 GOTO(out, rc);
1781         }
1782
1783         if (la.la_size != cat_max_size) {
1784                 CERROR("10f: catalog size has changed after it has wrap around,"
1785                        " current size = "LPU64", expected size = "LPU64"\n",
1786                        la.la_size, cat_max_size);
1787                 GOTO(out, rc = -EINVAL);
1788         }
1789
1790         /* sync device to commit all recent LLOG changes to disk and avoid
1791          * to consume a huge space with delayed journal commit callbacks
1792          * particularly on low memory nodes or VMs */
1793         rc = dt_sync(env, dt);
1794         if (rc) {
1795                 CERROR("10f: sync failed: %d\n", rc);
1796                 GOTO(out, rc);
1797         }
1798
1799         /* will llh_cat_idx also successfully wrap ? */
1800
1801         /* cancel all records in the plain LLOGs referenced by 2 last indexes in
1802          * Catalog */
1803
1804         /* cancel more records to free one more slot in Catalog */
1805         CWARN("10g: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1806         cancel_count = 0;
1807         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1808         if (rc != -LLOG_EEMPTY) {
1809                 CERROR("10g: process with llog_cancel_rec_cb failed: %d\n", rc);
1810                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1811                  * not reached */
1812                 if (rc == 0)
1813                         rc = -ERANGE;
1814                 GOTO(out, rc);
1815         }
1816
1817         CWARN("10g: print the catalog entries.. we expect 3\n");
1818         cat_counter = 0;
1819         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1820         if (rc) {
1821                 CERROR("10g: process with cat_print_cb failed: %d\n", rc);
1822                 GOTO(out, rc);
1823         }
1824         if (cat_counter != 3) {
1825                 CERROR("10g: %d entries in catalog\n", cat_counter);
1826                 GOTO(out, rc = -EINVAL);
1827         }
1828
1829         /* verify one down in catalog (+1 with hdr) */
1830         rc = verify_handle("10g", cath, 4);
1831         if (rc)
1832                 GOTO(out, rc);
1833
1834         /* sync device to commit all recent LLOG changes to disk and avoid
1835          * to consume a huge space with delayed journal commit callbacks
1836          * particularly on low memory nodes or VMs */
1837         rc = dt_sync(env, dt);
1838         if (rc) {
1839                 CERROR("10g: sync failed: %d\n", rc);
1840                 GOTO(out, rc);
1841         }
1842
1843         /* cancel more records to free one more slot in Catalog */
1844         CWARN("10g: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1845         cancel_count = 0;
1846         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1847         if (rc != -LLOG_EEMPTY) {
1848                 CERROR("10g: process with llog_cancel_rec_cb failed: %d\n", rc);
1849                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1850                  * not reached */
1851                 if (rc == 0)
1852                         rc = -ERANGE;
1853                 GOTO(out, rc);
1854         }
1855
1856         CWARN("10g: print the catalog entries.. we expect 2\n");
1857         cat_counter = 0;
1858         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1859         if (rc) {
1860                 CERROR("10g: process with cat_print_cb failed: %d\n", rc);
1861                 GOTO(out, rc);
1862         }
1863         if (cat_counter != 2) {
1864                 CERROR("10g: %d entries in catalog\n", cat_counter);
1865                 GOTO(out, rc = -EINVAL);
1866         }
1867
1868         /* verify one down in catalog (+1 with hdr) */
1869         rc = verify_handle("10g", cath, 3);
1870         if (rc)
1871                 GOTO(out, rc);
1872
1873         /* verify lgh_last_idx = 2 and llh_cat_idx = 0 now */
1874         if (cath->lgh_hdr->llh_cat_idx != 0 ||
1875             cath->lgh_last_idx != 2) {
1876                 CERROR("10g: lgh_last_idx = %d vs 2, llh_cat_idx = %d vs 0\n",
1877                        cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1878                 GOTO(out, rc = -EINVAL);
1879         }
1880
1881         /* sync device to commit all recent LLOG changes to disk and avoid
1882          * to consume a huge space with delayed journal commit callbacks
1883          * particularly on low memory nodes or VMs */
1884         rc = dt_sync(env, dt);
1885         if (rc) {
1886                 CERROR("10g: sync failed: %d\n", rc);
1887                 GOTO(out, rc);
1888         }
1889
1890         /* cancel more records to free one more slot in Catalog */
1891         CWARN("10g: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
1892         cancel_count = 0;
1893         rc = llog_cat_process(env, cath, llog_cancel_rec_cb, "foobar", 0, 0);
1894         if (rc != -LLOG_EEMPTY) {
1895                 CERROR("10g: process with llog_cancel_rec_cb failed: %d\n", rc);
1896                 /* need to indicate error if for any reason LLOG_TEST_RECNUM is
1897                  * not reached */
1898                 if (rc == 0)
1899                         rc = -ERANGE;
1900                 GOTO(out, rc);
1901         }
1902
1903         CWARN("10g: print the catalog entries.. we expect 1\n");
1904         cat_counter = 0;
1905         rc = llog_process(env, cath, cat_print_cb, "test 10", NULL);
1906         if (rc) {
1907                 CERROR("10g: process with cat_print_cb failed: %d\n", rc);
1908                 GOTO(out, rc);
1909         }
1910         if (cat_counter != 1) {
1911                 CERROR("10g: %d entries in catalog\n", cat_counter);
1912                 GOTO(out, rc = -EINVAL);
1913         }
1914
1915         /* verify one down in catalog (+1 with hdr) */
1916         rc = verify_handle("10g", cath, 2);
1917         if (rc)
1918                 GOTO(out, rc);
1919
1920         /* verify lgh_last_idx = 2 and llh_cat_idx = 1 now */
1921         if (cath->lgh_hdr->llh_cat_idx != 1 ||
1922             cath->lgh_last_idx != 2) {
1923                 CERROR("10g: lgh_last_idx = %d vs 2, llh_cat_idx = %d vs 1\n",
1924                        cath->lgh_last_idx, cath->lgh_hdr->llh_cat_idx);
1925                 GOTO(out, rc = -EINVAL);
1926         }
1927
1928         CWARN("10g: llh_cat_idx has also successfully wrapped!\n");
1929
1930 out:
1931         cfs_fail_loc = 0;
1932         cfs_fail_val = 0;
1933
1934         CWARN("10: put newly-created catalog\n");
1935         rc2 = llog_cat_close(env, cath);
1936         if (rc2) {
1937                 CERROR("10: close log %s failed: %d\n", name, rc2);
1938                 if (rc == 0)
1939                         rc = rc2;
1940         }
1941 ctxt_release:
1942         llog_ctxt_put(ctxt);
1943         RETURN(rc);
1944 }
1945
1946 /* -------------------------------------------------------------------------
1947  * Tests above, boring obd functions below
1948  * ------------------------------------------------------------------------- */
1949 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
1950 {
1951         struct llog_handle      *llh = NULL;
1952         struct llog_ctxt        *ctxt;
1953         int                      rc, err;
1954         char                     name[10];
1955
1956         ENTRY;
1957         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1958         LASSERT(ctxt);
1959
1960         sprintf(name, "%x", llog_test_rand);
1961
1962         rc = llog_test_1(env, obd, name);
1963         if (rc)
1964                 GOTO(cleanup_ctxt, rc);
1965
1966         rc = llog_test_2(env, obd, name, &llh);
1967         if (rc)
1968                 GOTO(cleanup_ctxt, rc);
1969
1970         rc = llog_test_3(env, obd, llh);
1971         if (rc)
1972                 GOTO(cleanup, rc);
1973
1974         rc = llog_test_4(env, obd);
1975         if (rc)
1976                 GOTO(cleanup, rc);
1977
1978         rc = llog_test_5(env, obd);
1979         if (rc)
1980                 GOTO(cleanup, rc);
1981
1982         rc = llog_test_6(env, obd, name);
1983         if (rc)
1984                 GOTO(cleanup, rc);
1985
1986         rc = llog_test_7(env, obd);
1987         if (rc)
1988                 GOTO(cleanup, rc);
1989
1990         rc = llog_test_8(env, obd);
1991         if (rc)
1992                 GOTO(cleanup, rc);
1993
1994         rc = llog_test_9(env, obd);
1995         if (rc != 0)
1996                 GOTO(cleanup, rc);
1997
1998         rc = llog_test_10(env, obd);
1999         if (rc)
2000                 GOTO(cleanup, rc);
2001
2002 cleanup:
2003         err = llog_destroy(env, llh);
2004         if (err)
2005                 CERROR("cleanup: llog_destroy failed: %d\n", err);
2006         llog_close(env, llh);
2007         if (rc == 0)
2008                 rc = err;
2009 cleanup_ctxt:
2010         llog_ctxt_put(ctxt);
2011         return rc;
2012 }
2013
2014 static int llog_test_cleanup(struct obd_device *obd)
2015 {
2016         struct obd_device       *tgt;
2017         struct lu_env            env;
2018         int                      rc;
2019
2020         ENTRY;
2021
2022         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
2023         if (rc)
2024                 RETURN(rc);
2025
2026         tgt = obd->obd_lvfs_ctxt.dt->dd_lu_dev.ld_obd;
2027         rc = llog_cleanup(&env, llog_get_context(tgt, LLOG_TEST_ORIG_CTXT));
2028         if (rc)
2029                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
2030         lu_env_fini(&env);
2031         RETURN(rc);
2032 }
2033
2034 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
2035 {
2036         struct obd_device       *tgt;
2037         struct llog_ctxt        *ctxt;
2038         struct dt_object        *o;
2039         struct lu_env            env;
2040         struct lu_context        test_session;
2041         int                      rc;
2042
2043         ENTRY;
2044
2045         if (lcfg->lcfg_bufcount < 2) {
2046                 CERROR("requires a TARGET OBD name\n");
2047                 RETURN(-EINVAL);
2048         }
2049
2050         if (lcfg->lcfg_buflens[1] < 1) {
2051                 CERROR("requires a TARGET OBD name\n");
2052                 RETURN(-EINVAL);
2053         }
2054
2055         /* disk obd */
2056         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
2057         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
2058                 CERROR("target device not attached or not set up (%s)\n",
2059                        lustre_cfg_string(lcfg, 1));
2060                 RETURN(-EINVAL);
2061         }
2062
2063         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
2064         if (rc)
2065                 RETURN(rc);
2066
2067         rc = lu_context_init(&test_session, LCT_SERVER_SESSION);
2068         if (rc)
2069                 GOTO(cleanup_env, rc);
2070         test_session.lc_thread = (struct ptlrpc_thread *)current;
2071         lu_context_enter(&test_session);
2072         env.le_ses = &test_session;
2073
2074         CWARN("Setup llog-test device over %s device\n",
2075               lustre_cfg_string(lcfg, 1));
2076
2077         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
2078         obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);
2079
2080         rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,
2081                         &llog_osd_ops);
2082         if (rc)
2083                 GOTO(cleanup_session, rc);
2084
2085         /* use MGS llog dir for tests */
2086         ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);
2087         LASSERT(ctxt);
2088         o = ctxt->loc_dir;
2089         llog_ctxt_put(ctxt);
2090
2091         ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);
2092         LASSERT(ctxt);
2093         ctxt->loc_dir = o;
2094         llog_ctxt_put(ctxt);
2095
2096         llog_test_rand = cfs_rand();
2097
2098         rc = llog_run_tests(&env, tgt);
2099         if (rc)
2100                 llog_test_cleanup(obd);
2101 cleanup_session:
2102         lu_context_exit(&test_session);
2103         lu_context_fini(&test_session);
2104 cleanup_env:
2105         lu_env_fini(&env);
2106         RETURN(rc);
2107 }
2108
2109 static struct obd_ops llog_obd_ops = {
2110         .o_owner       = THIS_MODULE,
2111         .o_setup       = llog_test_setup,
2112         .o_cleanup     = llog_test_cleanup,
2113 };
2114
2115 static int __init llog_test_init(void)
2116 {
2117         return class_register_type(&llog_obd_ops, NULL, true, NULL,
2118                                    "llog_test", NULL);
2119 }
2120
2121 static void __exit llog_test_exit(void)
2122 {
2123         class_unregister_type("llog_test");
2124 }
2125
2126 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2127 MODULE_DESCRIPTION("Lustre Log test module");
2128 MODULE_VERSION(LUSTRE_VERSION_STRING);
2129 MODULE_LICENSE("GPL");
2130
2131 module_init(llog_test_init);
2132 module_exit(llog_test_exit);