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