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, 2014, 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         if (llh->lgh_last_idx < last_idx) {
  91                 CERROR("%s: handle->last_idx is %d, expected %d after write\n",
  92                        test, llh->lgh_last_idx, last_idx);
  93                 RETURN(-ERANGE);
  94         }
  95
  96         RETURN(0);
  97 }
  98
  99 /* Test named-log create/open, close */
 100 static int llog_test_1(const struct lu_env *env,
 101                        struct obd_device *obd, char *name)
 102 {
 103         struct llog_handle      *llh;
 104         struct llog_ctxt        *ctxt;
 105         int rc;
 106         int rc2;
 107
 108         ENTRY;
 109
 110         CWARN("1a: create a log with name: %s\n", name);
 111         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 112         LASSERT(ctxt);
 113
 114         rc = llog_open_create(env, ctxt, &llh, NULL, name);
 115         if (rc) {
 116                 CERROR("1a: llog_create with name %s failed: %d\n", name, rc);
 117                 GOTO(out, rc);
 118         }
 119         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, &uuid);
 120         if (rc) {
 121                 CERROR("1a: can't init llog handle: %d\n", rc);
 122                 GOTO(out_close, rc);
 123         }
 124
 125         rc = verify_handle("1", llh, 1);
 126
 127         CWARN("1b: close newly-created log\n");
 128 out_close:
 129         rc2 = llog_close(env, llh);
 130         if (rc2) {
 131                 CERROR("1b: close log %s failed: %d\n", name, rc2);
 132                 if (rc == 0)
 133                         rc = rc2;
 134         }
 135 out:
 136         llog_ctxt_put(ctxt);
 137         RETURN(rc);
 138 }
 139
 140 /* Test named-log reopen; returns opened log on success */
 141 static int llog_test_2(const struct lu_env *env, struct obd_device *obd,
 142                        char *name, struct llog_handle **llh)
 143 {
 144         struct llog_ctxt        *ctxt;
 145         struct llog_handle      *loghandle;
 146         struct llog_logid        logid;
 147         int                      rc;
 148
 149         ENTRY;
 150
 151         CWARN("2a: re-open a log with name: %s\n", name);
 152         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 153         LASSERT(ctxt);
 154
 155         rc = llog_open(env, ctxt, llh, NULL, name, LLOG_OPEN_EXISTS);
 156         if (rc) {
 157                 CERROR("2a: re-open log with name %s failed: %d\n", name, rc);
 158                 GOTO(out_put, rc);
 159         }
 160
 161         rc = llog_init_handle(env, *llh, LLOG_F_IS_PLAIN, &uuid);
 162         if (rc) {
 163                 CERROR("2a: can't init llog handle: %d\n", rc);
 164                 GOTO(out_close_llh, rc);
 165         }
 166
 167         rc = verify_handle("2", *llh, 1);
 168         if (rc)
 169                 GOTO(out_close_llh, rc);
 170
 171         /* XXX: there is known issue with tests 2b, MGS is not able to create
 172          * anonymous llog, exit now to allow following tests run.
 173          * It is fixed in upcoming llog over OSD code */
 174         GOTO(out_put, rc);
 175
 176         CWARN("2b: create a log without specified NAME & LOGID\n");
 177         rc = llog_open_create(env, ctxt, &loghandle, NULL, NULL);
 178         if (rc) {
 179                 CERROR("2b: create log failed\n");
 180                 GOTO(out_close_llh, rc);
 181         }
 182         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, &uuid);
 183         if (rc) {
 184                 CERROR("2b: can't init llog handle: %d\n", rc);
 185                 GOTO(out_close, rc);
 186         }
 187
 188         logid = loghandle->lgh_id;
 189         llog_close(env, loghandle);
 190
 191         CWARN("2c: re-open the log by LOGID\n");
 192         rc = llog_open(env, ctxt, &loghandle, &logid, NULL, LLOG_OPEN_EXISTS);
 193         if (rc) {
 194                 CERROR("2c: re-open log by LOGID failed\n");
 195                 GOTO(out_close_llh, rc);
 196         }
 197
 198         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, &uuid);
 199         if (rc) {
 200                 CERROR("2c: can't init llog handle: %d\n", rc);
 201                 GOTO(out_close, rc);
 202         }
 203
 204         CWARN("2b: destroy this log\n");
 205         rc = llog_destroy(env, loghandle);
 206         if (rc)
 207                 CERROR("2d: destroy log failed\n");
 208 out_close:
 209         llog_close(env, loghandle);
 210 out_close_llh:
 211         if (rc)
 212                 llog_close(env, *llh);
 213 out_put:
 214         llog_ctxt_put(ctxt);
 215
 216         RETURN(rc);
 217 }
 218
 219 static int records;
 220 static off_t rec_offset;
 221 static int paddings;
 222 static int start_idx;
 223
 224 /*
 225  * Test 3 callback.
 226  * - check lgh_cur_offset correctness
 227  * - check record index consistency
 228  * - modify each record in-place
 229  * - add new record during *last_idx processing
 230  */
 231 static int test3_check_n_add_cb(const struct lu_env *env,
 232                                 struct llog_handle *lgh,
 233                                 struct llog_rec_hdr *rec, void *data)
 234 {
 235         struct llog_gen_rec *lgr = (struct llog_gen_rec *)rec;
 236         int *last_rec = data;
 237         int rc;
 238
 239         if (lgh->lgh_hdr->llh_flags & LLOG_F_IS_FIXSIZE) {
 240                 LASSERT(lgh->lgh_hdr->llh_size > 0);
 241                 if (lgh->lgh_cur_offset != lgh->lgh_hdr->llh_hdr.lrh_len +
 242                                 (start_idx + records - 1) *
 243                                 lgh->lgh_hdr->llh_size)
 244                         CERROR("Wrong record offset in cur_off: "LPU64", should"
 245                                " be %u\n", lgh->lgh_cur_offset,
 246                                lgh->lgh_hdr->llh_hdr.lrh_len +
 247                                (start_idx + records - 1) *
 248                                lgh->lgh_hdr->llh_size);
 249         } else {
 250                 size_t chunk_size = lgh->lgh_hdr->llh_hdr.lrh_len;
 251
 252                 /* For variable size records the start offset is unknown, trust
 253                  * the first value and check others are consistent with it. */
 254                 if (rec_offset == 0)
 255                         rec_offset = lgh->lgh_cur_offset;
 256
 257                 if (lgh->lgh_cur_offset != rec_offset) {
 258                         /* there can be padding record */
 259                         if ((lgh->lgh_cur_offset % chunk_size == 0) &&
 260                             (lgh->lgh_cur_offset - rec_offset <
 261                              rec->lrh_len + LLOG_MIN_REC_SIZE)) {
 262                                 rec_offset = lgh->lgh_cur_offset;
 263                                 paddings++;
 264                         } else {
 265                                 CERROR("Wrong record offset in cur_off: "LPU64
 266                                        ", should be %lld (rec len %u)\n",
 267                                        lgh->lgh_cur_offset,
 268                                        (long long)rec_offset, rec->lrh_len);
 269                         }
 270                 }
 271                 rec_offset += rec->lrh_len;
 272         }
 273
 274         if ((start_idx + records + paddings) != rec->lrh_index)
 275                 CERROR("Record with wrong index was read: %u, expected %u\n",
 276                        rec->lrh_index, start_idx + records + paddings);
 277
 278         /* modify all records in place */
 279         lgr->lgr_gen.conn_cnt = rec->lrh_index;
 280         rc = llog_write(env, lgh, rec, rec->lrh_index);
 281         if (rc < 0)
 282                 CERROR("cb_test_3: cannot modify record while processing\n");
 283
 284         /* Add new record to the llog at *last_rec position one by one to
 285          * check that last block is re-read during processing */
 286         if ((start_idx + records + paddings) == *last_rec ||
 287             (start_idx + records + paddings) == (*last_rec + 1)) {
 288                 rc = llog_write(env, lgh, rec, LLOG_NEXT_IDX);
 289                 if (rc < 0)
 290                         CERROR("cb_test_3: cannot add new record while "
 291                                "processing\n");
 292         }
 293         records++;
 294
 295         return rc;
 296 }
 297
 298 /* Check in-place modifications were done for all records*/
 299 static int test3_check_cb(const struct lu_env *env, struct llog_handle *lgh,
 300                           struct llog_rec_hdr *rec, void *data)
 301 {
 302         struct llog_gen_rec *lgr = (struct llog_gen_rec *)rec;
 303
 304         if (lgr->lgr_gen.conn_cnt != rec->lrh_index) {
 305                 CERROR("cb_test_3: record %u is not modified\n",
 306                        rec->lrh_index);
 307                 return -EINVAL;
 308         }
 309         records++;
 310         return 0;
 311 }
 312
 313 static int llog_test3_process(const struct lu_env *env,
 314                               struct llog_handle *lgh,
 315                               llog_cb_t cb, int start)
 316 {
 317         struct llog_process_cat_data cd;
 318         int last_idx; /* new record will be injected here */
 319         int rc = 0;
 320
 321         CWARN("test3: processing records from index %d to the end\n",
 322               start);
 323         cd.lpcd_first_idx = start - 1;
 324         cd.lpcd_last_idx = 0;
 325         records = paddings = 0;
 326         last_idx = lgh->lgh_last_idx;
 327         rc = llog_process(env, lgh, cb, &last_idx, &cd);
 328         if (rc < 0)
 329                 return rc;
 330         CWARN("test3: total %u records processed with %u paddings\n",
 331               records, paddings);
 332         return records;
 333 }
 334
 335 /* Test plain llog functionality */
 336 static int llog_test_3(const struct lu_env *env, struct obd_device *obd,
 337                        struct llog_handle *llh)
 338 {
 339         char buf[128];
 340         struct llog_rec_hdr *hdr = (void *)buf;
 341         int rc, i;
 342         int num_recs = 1; /* 1 for the header */
 343         int expected;
 344
 345         ENTRY;
 346
 347         hdr->lrh_len = sizeof(struct llog_gen_rec);
 348         hdr->lrh_type = LLOG_GEN_REC;
 349         llh->lgh_hdr->llh_size = sizeof(struct llog_gen_rec);
 350         llh->lgh_hdr->llh_flags |= LLOG_F_IS_FIXSIZE;
 351
 352         /* Fill the llog with 64-bytes records, use 1023 records,
 353          * so last chunk will be partially full. Don't change this
 354          * value until record size is changed.
 355          */
 356         CWARN("3a: write 1023 fixed-size llog records\n");
 357         for (i = 0; i < 1023; i++) {
 358                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 359                 if (rc < 0) {
 360                         CERROR("3a: write 1023 records failed at #%d: %d\n",
 361                                i + 1, rc);
 362                         RETURN(rc);
 363                 }
 364                 num_recs++;
 365         }
 366
 367         rc = verify_handle("3a", llh, num_recs);
 368         if (rc)
 369                 RETURN(rc);
 370
 371         /*
 372          * Test fixed-size records processing:
 373          * - search the needed index
 374          * - go through all records from that index
 375          * - check all indices are growing monotonically and exist
 376          * - modify each record
 377          *
 378          * NB: test3_check_n_add adds two new records while processing
 379          * after last record. There were 1023 records created so the last chunk
 380          * misses exactly one record. Therefore one of new records will be
 381          * the last in the current chunk and second causes the new chunk to be
 382          * created.
 383          */
 384         rec_offset = 0;
 385         start_idx = 501;
 386         expected = 525;
 387         rc = llog_test3_process(env, llh, test3_check_n_add_cb, start_idx);
 388         if (rc < 0)
 389                 RETURN(rc);
 390
 391         /* extra record is created during llog_process() */
 392         if (rc != expected) {
 393                 CERROR("3a: process total %d records but expect %d\n",
 394                        rc, expected);
 395                 RETURN(-ERANGE);
 396         }
 397
 398         num_recs += 2;
 399
 400         /* test modification in place */
 401         rc = llog_test3_process(env, llh, test3_check_cb, start_idx);
 402         if (rc < 0)
 403                 RETURN(rc);
 404
 405         if (rc != expected) {
 406                 CERROR("3a: process total %d records but expect %d\n",
 407                        rc, expected);
 408                 RETURN(-ERANGE);
 409         }
 410
 411         CWARN("3b: write 566 variable size llog records\n");
 412
 413         /* Drop llh_size to 0 to mark llog as variable-size and write
 414          * header to make this change permanent. */
 415         llh->lgh_hdr->llh_flags &= ~LLOG_F_IS_FIXSIZE;
 416         llog_write(env, llh, &llh->lgh_hdr->llh_hdr, LLOG_HEADER_IDX);
 417
 418         hdr->lrh_type = OBD_CFG_REC;
 419
 420         /* there are 1025 64-bytes records in llog already,
 421          * the last chunk contains single record, i.e. 64 bytes.
 422          * Each pair of variable size records is 200 bytes, so
 423          * we will have the following distribution per chunks:
 424          * block 1: 64 + 80(80/120) + 80 + 48(pad) = 81 iterations
 425          * block 2: 80(120/80) + 120 + 72(pad) = 81 itereations
 426          * block 3: 80(80/120) + 80 + 112(pad) = 81 iterations
 427          * -- the same as block 2 again and so on.
 428          * block 7: 80(80/120) = 80 iterations and 192 bytes remain
 429          * Total 6 * 81 + 80 = 566 itereations.
 430          * Callback will add another 120 bytes in the end of the last chunk
 431          * and another 120 bytes will cause padding (72 bytes) plus 120
 432          * bytes in the new block.
 433          */
 434         for (i = 0; i < 566; i++) {
 435                 if ((i % 2) == 0)
 436                         hdr->lrh_len = 80;
 437                 else
 438                         hdr->lrh_len = 120;
 439
 440                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 441                 if (rc < 0) {
 442                         CERROR("3a: write 566 records failed at #%d: %d\n",
 443                                i + 1, rc);
 444                         RETURN(rc);
 445                 }
 446                 num_recs++;
 447         }
 448
 449         rc = verify_handle("3b", llh, num_recs);
 450         if (rc)
 451                 RETURN(rc);
 452
 453         start_idx = 1026;
 454         expected = 568;
 455         rc = llog_test3_process(env, llh, test3_check_n_add_cb, start_idx);
 456         if (rc < 0)
 457                 RETURN(rc);
 458
 459         if (rc != expected) {
 460                 CERROR("3b: process total %d records but expect %d\n",
 461                        rc, expected);
 462                 RETURN(-ERANGE);
 463         }
 464
 465         num_recs += 2;
 466
 467         /* test modification in place */
 468         rc = llog_test3_process(env, llh, test3_check_cb, start_idx);
 469         if (rc < 0)
 470                 RETURN(rc);
 471
 472         if (rc != expected) {
 473                 CERROR("3b: process total %d records but expect %d\n",
 474                        rc, expected);
 475                 RETURN(-ERANGE);
 476         }
 477
 478         CWARN("3c: write records with variable size until BITMAP_SIZE, "
 479               "return -ENOSPC\n");
 480         while (num_recs < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr)) {
 481                 if ((num_recs % 2) == 0)
 482                         hdr->lrh_len = 80;
 483                 else
 484                         hdr->lrh_len = 128;
 485
 486                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 487                 if (rc == -ENOSPC) {
 488                         break;
 489                 } else if (rc < 0) {
 490                         CERROR("3c: write recs failed at #%d: %d\n",
 491                                num_recs, rc);
 492                         RETURN(rc);
 493                 }
 494                 num_recs++;
 495         }
 496
 497         if (rc != -ENOSPC) {
 498                 CWARN("3c: write record more than BITMAP size!\n");
 499                 RETURN(-EINVAL);
 500         }
 501         CWARN("3c: wrote %d more records before end of llog is reached\n",
 502               num_recs);
 503
 504         rc = verify_handle("3d", llh, num_recs);
 505
 506         RETURN(rc);
 507 }
 508
 509 /* Test catalogue additions */
 510 static int llog_test_4(const struct lu_env *env, struct obd_device *obd)
 511 {
 512         struct llog_handle      *cath;
 513         char                     name[10];
 514         int                      rc, rc2, i, buflen;
 515         struct llog_mini_rec     lmr;
 516         struct llog_cookie       cookie;
 517         struct llog_ctxt        *ctxt;
 518         int                      num_recs = 0;
 519         char                    *buf;
 520         struct llog_rec_hdr     *rec;
 521
 522         ENTRY;
 523
 524         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 525         LASSERT(ctxt);
 526
 527         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 528         lmr.lmr_hdr.lrh_type = 0xf00f00;
 529
 530         sprintf(name, "%x", llog_test_rand + 1);
 531         CWARN("4a: create a catalog log with name: %s\n", name);
 532         rc = llog_open_create(env, ctxt, &cath, NULL, name);
 533         if (rc) {
 534                 CERROR("4a: llog_create with name %s failed: %d\n", name, rc);
 535                 GOTO(ctxt_release, rc);
 536         }
 537         rc = llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);
 538         if (rc) {
 539                 CERROR("4a: can't init llog handle: %d\n", rc);
 540                 GOTO(out, rc);
 541         }
 542
 543         num_recs++;
 544         cat_logid = cath->lgh_id;
 545
 546         CWARN("4b: write 1 record into the catalog\n");
 547         rc = llog_cat_add(env, cath, &lmr.lmr_hdr, &cookie);
 548         if (rc != 1) {
 549                 CERROR("4b: write 1 catalog record failed at: %d\n", rc);
 550                 GOTO(out, rc);
 551         }
 552         num_recs++;
 553         rc = verify_handle("4b", cath, 2);
 554         if (rc)
 555                 GOTO(out, rc);
 556
 557         rc = verify_handle("4b", cath->u.chd.chd_current_log, num_recs);
 558         if (rc)
 559                 GOTO(out, rc);
 560
 561         CWARN("4c: cancel 1 log record\n");
 562         rc = llog_cat_cancel_records(env, cath, 1, &cookie);
 563         if (rc) {
 564                 CERROR("4c: cancel 1 catalog based record failed: %d\n", rc);
 565                 GOTO(out, rc);
 566         }
 567         num_recs--;
 568
 569         rc = verify_handle("4c", cath->u.chd.chd_current_log, num_recs);
 570         if (rc)
 571                 GOTO(out, rc);
 572
 573         CWARN("4d: write %d more log records\n", LLOG_TEST_RECNUM);
 574         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
 575                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
 576                 if (rc) {
 577                         CERROR("4d: write %d records failed at #%d: %d\n",
 578                                LLOG_TEST_RECNUM, i + 1, rc);
 579                         GOTO(out, rc);
 580                 }
 581                 num_recs++;
 582         }
 583
 584         /* make sure new plain llog appears */
 585         rc = verify_handle("4d", cath, 3);
 586         if (rc)
 587                 GOTO(out, rc);
 588
 589         CWARN("4e: add 5 large records, one record per block\n");
 590         buflen = LLOG_MIN_CHUNK_SIZE;
 591         OBD_ALLOC(buf, buflen);
 592         if (buf == NULL)
 593                 GOTO(out, rc = -ENOMEM);
 594         for (i = 0; i < 5; i++) {
 595                 rec = (void *)buf;
 596                 rec->lrh_len = buflen;
 597                 rec->lrh_type = OBD_CFG_REC;
 598                 rc = llog_cat_add(env, cath, rec, NULL);
 599                 if (rc) {
 600                         CERROR("4e: write 5 records failed at #%d: %d\n",
 601                                i + 1, rc);
 602                         GOTO(out_free, rc);
 603                 }
 604                 num_recs++;
 605         }
 606 out_free:
 607         OBD_FREE(buf, buflen);
 608 out:
 609         CWARN("4f: put newly-created catalog\n");
 610         rc2 = llog_cat_close(env, cath);
 611         if (rc2) {
 612                 CERROR("4: close log %s failed: %d\n", name, rc2);
 613                 if (rc == 0)
 614                         rc = rc2;
 615         }
 616 ctxt_release:
 617         llog_ctxt_put(ctxt);
 618         RETURN(rc);
 619 }
 620
 621 static int cat_counter;
 622
 623 static int cat_print_cb(const struct lu_env *env, struct llog_handle *llh,
 624                         struct llog_rec_hdr *rec, void *data)
 625 {
 626         struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
 627         struct lu_fid            fid = {0};
 628
 629         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
 630                 CERROR("invalid record in catalog\n");
 631                 RETURN(-EINVAL);
 632         }
 633
 634         logid_to_fid(&lir->lid_id, &fid);
 635
 636         CWARN("seeing record at index %d - "DFID" in log "DFID"\n",
 637               rec->lrh_index, PFID(&fid),
 638               PFID(lu_object_fid(&llh->lgh_obj->do_lu)));
 639
 640         cat_counter++;
 641
 642         RETURN(0);
 643 }
 644
 645 static int plain_counter;
 646
 647 static int plain_print_cb(const struct lu_env *env, struct llog_handle *llh,
 648                           struct llog_rec_hdr *rec, void *data)
 649 {
 650         struct lu_fid fid = {0};
 651
 652         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 653                 CERROR("log is not plain\n");
 654                 RETURN(-EINVAL);
 655         }
 656
 657         logid_to_fid(&llh->lgh_id, &fid);
 658
 659         CDEBUG(D_INFO, "seeing record at index %d in log "DFID"\n",
 660                rec->lrh_index, PFID(&fid));
 661
 662         plain_counter++;
 663
 664         RETURN(0);
 665 }
 666
 667 static int cancel_count;
 668
 669 static int llog_cancel_rec_cb(const struct lu_env *env,
 670                               struct llog_handle *llh,
 671                               struct llog_rec_hdr *rec, void *data)
 672 {
 673         struct llog_cookie cookie;
 674
 675         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 676                 CERROR("log is not plain\n");
 677                 RETURN(-EINVAL);
 678         }
 679
 680         cookie.lgc_lgl = llh->lgh_id;
 681         cookie.lgc_index = rec->lrh_index;
 682
 683         llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1, &cookie);
 684         cancel_count++;
 685         if (cancel_count == LLOG_TEST_RECNUM)
 686                 RETURN(-LLOG_EEMPTY);
 687         RETURN(0);
 688 }
 689
 690 /* Test log and catalogue processing */
 691 static int llog_test_5(const struct lu_env *env, struct obd_device *obd)
 692 {
 693         struct llog_handle      *llh = NULL;
 694         char                     name[10];
 695         int                      rc, rc2;
 696         struct llog_mini_rec     lmr;
 697         struct llog_ctxt        *ctxt;
 698
 699         ENTRY;
 700
 701         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 702         LASSERT(ctxt);
 703
 704         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 705         lmr.lmr_hdr.lrh_type = 0xf00f00;
 706
 707         CWARN("5a: re-open catalog by id\n");
 708         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
 709         if (rc) {
 710                 CERROR("5a: llog_create with logid failed: %d\n", rc);
 711                 GOTO(out_put, rc);
 712         }
 713
 714         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
 715         if (rc) {
 716                 CERROR("5a: can't init llog handle: %d\n", rc);
 717                 GOTO(out, rc);
 718         }
 719
 720         CWARN("5b: print the catalog entries.. we expect 2\n");
 721         cat_counter = 0;
 722         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 723         if (rc) {
 724                 CERROR("5b: process with cat_print_cb failed: %d\n", rc);
 725                 GOTO(out, rc);
 726         }
 727         if (cat_counter != 2) {
 728                 CERROR("5b: %d entries in catalog\n", cat_counter);
 729                 GOTO(out, rc = -EINVAL);
 730         }
 731
 732         CWARN("5c: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
 733         cancel_count = 0;
 734         rc = llog_cat_process(env, llh, llog_cancel_rec_cb, "foobar", 0, 0);
 735         if (rc != -LLOG_EEMPTY) {
 736                 CERROR("5c: process with llog_cancel_rec_cb failed: %d\n", rc);
 737                 GOTO(out, rc);
 738         }
 739
 740         CWARN("5c: print the catalog entries.. we expect 1\n");
 741         cat_counter = 0;
 742         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 743         if (rc) {
 744                 CERROR("5c: process with cat_print_cb failed: %d\n", rc);
 745                 GOTO(out, rc);
 746         }
 747         if (cat_counter != 1) {
 748                 CERROR("5c: %d entries in catalog\n", cat_counter);
 749                 GOTO(out, rc = -EINVAL);
 750         }
 751
 752         CWARN("5d: add 1 record to the log with many canceled empty pages\n");
 753         rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
 754         if (rc) {
 755                 CERROR("5d: add record to the log with many canceled empty "
 756                        "pages failed\n");
 757                 GOTO(out, rc);
 758         }
 759
 760         CWARN("5e: print plain log entries.. expect 6\n");
 761         plain_counter = 0;
 762         rc = llog_cat_process(env, llh, plain_print_cb, "foobar", 0, 0);
 763         if (rc) {
 764                 CERROR("5e: process with plain_print_cb failed: %d\n", rc);
 765                 GOTO(out, rc);
 766         }
 767         if (plain_counter != 6) {
 768                 CERROR("5e: found %d records\n", plain_counter);
 769                 GOTO(out, rc = -EINVAL);
 770         }
 771
 772         CWARN("5f: print plain log entries reversely.. expect 6\n");
 773         plain_counter = 0;
 774         rc = llog_cat_reverse_process(env, llh, plain_print_cb, "foobar");
 775         if (rc) {
 776                 CERROR("5f: reversely process with plain_print_cb failed: "
 777                        "%d\n", rc);
 778                 GOTO(out, rc);
 779         }
 780         if (plain_counter != 6) {
 781                 CERROR("5f: found %d records\n", plain_counter);
 782                 GOTO(out, rc = -EINVAL);
 783         }
 784
 785 out:
 786         CWARN("5g: close re-opened catalog\n");
 787         rc2 = llog_cat_close(env, llh);
 788         if (rc2) {
 789                 CERROR("5g: close log %s failed: %d\n", name, rc2);
 790                 if (rc == 0)
 791                         rc = rc2;
 792         }
 793 out_put:
 794         llog_ctxt_put(ctxt);
 795
 796         RETURN(rc);
 797 }
 798
 799 /* Test client api; open log by name and process */
 800 static int llog_test_6(const struct lu_env *env, struct obd_device *obd,
 801                        char *name)
 802 {
 803         struct obd_device       *mgc_obd;
 804         struct llog_ctxt        *ctxt;
 805         struct obd_uuid         *mgs_uuid;
 806         struct obd_export       *exp;
 807         struct obd_uuid          uuid = { "LLOG_TEST6_UUID" };
 808         struct llog_handle      *llh = NULL;
 809         struct llog_ctxt        *nctxt;
 810         int                      rc, rc2;
 811
 812         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 813         LASSERT(ctxt);
 814         mgs_uuid = &ctxt->loc_exp->exp_obd->obd_uuid;
 815
 816         CWARN("6a: re-open log %s using client API\n", name);
 817         mgc_obd = class_find_client_obd(mgs_uuid, LUSTRE_MGC_NAME, NULL);
 818         if (mgc_obd == NULL) {
 819                 CERROR("6a: no MGC devices connected to %s found.\n",
 820                        mgs_uuid->uuid);
 821                 GOTO(ctxt_release, rc = -ENOENT);
 822         }
 823
 824         rc = obd_connect(NULL, &exp, mgc_obd, &uuid,
 825                          NULL /* obd_connect_data */, NULL);
 826         if (rc != -EALREADY) {
 827                 CERROR("6a: connect on connected MGC (%s) failed to return"
 828                        " -EALREADY\n", mgc_obd->obd_name);
 829                 if (rc == 0)
 830                         obd_disconnect(exp);
 831                 GOTO(ctxt_release, rc = -EINVAL);
 832         }
 833
 834         nctxt = llog_get_context(mgc_obd, LLOG_CONFIG_REPL_CTXT);
 835         rc = llog_open(env, nctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
 836         if (rc) {
 837                 CERROR("6a: llog_open failed %d\n", rc);
 838                 GOTO(nctxt_put, rc);
 839         }
 840
 841         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
 842         if (rc) {
 843                 CERROR("6a: llog_init_handle failed %d\n", rc);
 844                 GOTO(parse_out, rc);
 845         }
 846
 847         plain_counter = 1; /* llog header is first record */
 848         CWARN("6b: process log %s using client API\n", name);
 849         rc = llog_process(env, llh, plain_print_cb, NULL, NULL);
 850         if (rc)
 851                 CERROR("6b: llog_process failed %d\n", rc);
 852         CWARN("6b: processed %d records\n", plain_counter);
 853
 854         rc = verify_handle("6b", llh, plain_counter);
 855         if (rc)
 856                 GOTO(parse_out, rc);
 857
 858         plain_counter = 1; /* llog header is first record */
 859         CWARN("6c: process log %s reversely using client API\n", name);
 860         rc = llog_reverse_process(env, llh, plain_print_cb, NULL, NULL);
 861         if (rc)
 862                 CERROR("6c: llog_reverse_process failed %d\n", rc);
 863         CWARN("6c: processed %d records\n", plain_counter);
 864
 865         rc = verify_handle("6c", llh, plain_counter);
 866         if (rc)
 867                 GOTO(parse_out, rc);
 868
 869 parse_out:
 870         rc2 = llog_close(env, llh);
 871         if (rc2) {
 872                 CERROR("6: llog_close failed: rc = %d\n", rc2);
 873                 if (rc == 0)
 874                         rc = rc2;
 875         }
 876 nctxt_put:
 877         llog_ctxt_put(nctxt);
 878 ctxt_release:
 879         llog_ctxt_put(ctxt);
 880         RETURN(rc);
 881 }
 882
 883 static union {
 884         struct llog_rec_hdr             lrh;   /* common header */
 885         struct llog_logid_rec           llr;   /* LLOG_LOGID_MAGIC */
 886         struct llog_unlink64_rec        lur;   /* MDS_UNLINK64_REC */
 887         struct llog_setattr64_rec       lsr64; /* MDS_SETATTR64_REC */
 888         struct llog_size_change_rec     lscr;  /* OST_SZ_REC */
 889         struct llog_changelog_rec       lcr;   /* CHANGELOG_REC */
 890         struct llog_changelog_user_rec  lcur;  /* CHANGELOG_USER_REC */
 891         struct llog_gen_rec             lgr;   /* LLOG_GEN_REC */
 892 } llog_records;
 893
 894 static int test_7_print_cb(const struct lu_env *env, struct llog_handle *llh,
 895                            struct llog_rec_hdr *rec, void *data)
 896 {
 897         struct lu_fid fid = {0};
 898
 899         logid_to_fid(&llh->lgh_id, &fid);
 900
 901         CDEBUG(D_OTHER, "record type %#x at index %d in log "DFID"\n",
 902                rec->lrh_type, rec->lrh_index, PFID(&fid));
 903
 904         plain_counter++;
 905         return 0;
 906 }
 907
 908 static int test_7_cancel_cb(const struct lu_env *env, struct llog_handle *llh,
 909                             struct llog_rec_hdr *rec, void *data)
 910 {
 911         plain_counter++;
 912         /* test LLOG_DEL_RECORD is working */
 913         return LLOG_DEL_RECORD;
 914 }
 915
 916 static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
 917 {
 918         struct llog_handle      *llh;
 919         int                      rc = 0, i, process_count;
 920         int                      num_recs = 0;
 921
 922         ENTRY;
 923
 924         rc = llog_open_create(env, ctxt, &llh, NULL, NULL);
 925         if (rc) {
 926                 CERROR("7_sub: create log failed\n");
 927                 RETURN(rc);
 928         }
 929
 930         rc = llog_init_handle(env, llh,
 931                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
 932                               &uuid);
 933         if (rc) {
 934                 CERROR("7_sub: can't init llog handle: %d\n", rc);
 935                 GOTO(out_close, rc);
 936         }
 937         for (i = 0; i < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr); i++) {
 938                 rc = llog_write(env, llh, &llog_records.lrh, LLOG_NEXT_IDX);
 939                 if (rc == -ENOSPC) {
 940                         break;
 941                 } else if (rc < 0) {
 942                         CERROR("7_sub: write recs failed at #%d: %d\n",
 943                                i + 1, rc);
 944                         GOTO(out_close, rc);
 945                 }
 946                 num_recs++;
 947         }
 948         if (rc != -ENOSPC) {
 949                 CWARN("7_sub: write record more than BITMAP size!\n");
 950                 GOTO(out_close, rc = -EINVAL);
 951         }
 952
 953         rc = verify_handle("7_sub", llh, num_recs + 1);
 954         if (rc) {
 955                 CERROR("7_sub: verify handle failed: %d\n", rc);
 956                 GOTO(out_close, rc);
 957         }
 958         if (num_recs < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1)
 959                 CWARN("7_sub: records are not aligned, written %d from %u\n",
 960                       num_recs, LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1);
 961
 962         plain_counter = 0;
 963         rc = llog_process(env, llh, test_7_print_cb, "test 7", NULL);
 964         if (rc) {
 965                 CERROR("7_sub: llog process failed: %d\n", rc);
 966                 GOTO(out_close, rc);
 967         }
 968         process_count = plain_counter;
 969         if (process_count != num_recs) {
 970                 CERROR("7_sub: processed %d records from %d total\n",
 971                        process_count, num_recs);
 972                 GOTO(out_close, rc = -EINVAL);
 973         }
 974
 975         plain_counter = 0;
 976         rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);
 977         if (rc && rc != LLOG_DEL_PLAIN) {
 978                 CERROR("7_sub: reverse llog process failed: %d\n", rc);
 979                 GOTO(out_close, rc);
 980         }
 981         if (process_count != plain_counter) {
 982                 CERROR("7_sub: Reverse/direct processing found different"
 983                        "number of records: %d/%d\n",
 984                        plain_counter, process_count);
 985                 GOTO(out_close, rc = -EINVAL);
 986         }
 987         if (llog_exist(llh)) {
 988                 CERROR("7_sub: llog exists but should be zapped\n");
 989                 GOTO(out_close, rc = -EEXIST);
 990         }
 991
 992         rc = verify_handle("7_sub", llh, 1);
 993 out_close:
 994         if (rc)
 995                 llog_destroy(env, llh);
 996         llog_close(env, llh);
 997         RETURN(rc);
 998 }
 999
1000 /* Test all llog records writing and processing */
1001 static int llog_test_7(const struct lu_env *env, struct obd_device *obd)
1002 {
1003         struct llog_ctxt        *ctxt;
1004         int                      rc;
1005
1006         ENTRY;
1007
1008         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1009
1010         CWARN("7a: test llog_logid_rec\n");
1011         llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);
1012         llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);
1013         llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
1014
1015         rc = llog_test_7_sub(env, ctxt);
1016         if (rc) {
1017                 CERROR("7a: llog_logid_rec test failed\n");
1018                 GOTO(out, rc);
1019         }
1020
1021         CWARN("7b: test llog_unlink64_rec\n");
1022         llog_records.lur.lur_hdr.lrh_len = sizeof(llog_records.lur);
1023         llog_records.lur.lur_tail.lrt_len = sizeof(llog_records.lur);
1024         llog_records.lur.lur_hdr.lrh_type = MDS_UNLINK64_REC;
1025
1026         rc = llog_test_7_sub(env, ctxt);
1027         if (rc) {
1028                 CERROR("7b: llog_unlink_rec test failed\n");
1029                 GOTO(out, rc);
1030         }
1031
1032         CWARN("7c: test llog_setattr64_rec\n");
1033         llog_records.lsr64.lsr_hdr.lrh_len = sizeof(llog_records.lsr64);
1034         llog_records.lsr64.lsr_tail.lrt_len = sizeof(llog_records.lsr64);
1035         llog_records.lsr64.lsr_hdr.lrh_type = MDS_SETATTR64_REC;
1036
1037         rc = llog_test_7_sub(env, ctxt);
1038         if (rc) {
1039                 CERROR("7c: llog_setattr64_rec test failed\n");
1040                 GOTO(out, rc);
1041         }
1042
1043         CWARN("7d: test llog_size_change_rec\n");
1044         llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);
1045         llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);
1046         llog_records.lscr.lsc_hdr.lrh_type = OST_SZ_REC;
1047
1048         rc = llog_test_7_sub(env, ctxt);
1049         if (rc) {
1050                 CERROR("7d: llog_size_change_rec test failed\n");
1051                 GOTO(out, rc);
1052         }
1053
1054         CWARN("7e: test llog_changelog_rec\n");
1055         /* Direct access to cr_do_not_use: peculiar case for this test */
1056         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
1057         llog_records.lcr.cr_do_not_use.lrt_len = sizeof(llog_records.lcr);
1058         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
1059
1060         rc = llog_test_7_sub(env, ctxt);
1061         if (rc) {
1062                 CERROR("7e: llog_changelog_rec test failed\n");
1063                 GOTO(out, rc);
1064         }
1065
1066         CWARN("7f: test llog_changelog_user_rec\n");
1067         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
1068         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
1069         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
1070
1071         rc = llog_test_7_sub(env, ctxt);
1072         if (rc) {
1073                 CERROR("7f: llog_changelog_user_rec test failed\n");
1074                 GOTO(out, rc);
1075         }
1076
1077         CWARN("7g: test llog_gen_rec\n");
1078         llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);
1079         llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);
1080         llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
1081
1082         rc = llog_test_7_sub(env, ctxt);
1083         if (rc) {
1084                 CERROR("7g: llog_size_change_rec test failed\n");
1085                 GOTO(out, rc);
1086         }
1087 out:
1088         llog_ctxt_put(ctxt);
1089         RETURN(rc);
1090 }
1091
1092 static int llog_truncate(const struct lu_env *env, struct dt_object *o)
1093 {
1094         struct lu_attr           la;
1095         struct thandle          *th;
1096         struct dt_device        *d;
1097         int                      rc;
1098         ENTRY;
1099
1100         LASSERT(o);
1101         d = lu2dt_dev(o->do_lu.lo_dev);
1102         LASSERT(d);
1103
1104         rc = dt_attr_get(env, o, &la);
1105         if (rc)
1106                 RETURN(rc);
1107
1108         CDEBUG(D_OTHER, "original size "LPU64"\n", la.la_size);
1109         rc = sizeof(struct llog_log_hdr) + sizeof(struct llog_mini_rec);
1110         if (la.la_size < rc) {
1111                 CERROR("too small llog: "LPU64"\n", la.la_size);
1112                 RETURN(0);
1113         }
1114
1115         /* drop 2 records */
1116         la.la_size = la.la_size - (sizeof(struct llog_mini_rec) * 2);
1117         la.la_valid = LA_SIZE;
1118
1119         th = dt_trans_create(env, d);
1120         if (IS_ERR(th))
1121                 RETURN(PTR_ERR(th));
1122
1123         rc = dt_declare_attr_set(env, o, &la, th);
1124         if (rc)
1125                 GOTO(stop, rc);
1126
1127         rc = dt_declare_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
1128
1129         rc = dt_trans_start_local(env, d, th);
1130         if (rc)
1131                 GOTO(stop, rc);
1132
1133         rc = dt_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
1134         if (rc)
1135                 GOTO(stop, rc);
1136
1137         rc = dt_attr_set(env, o, &la, th);
1138         if (rc)
1139                 GOTO(stop, rc);
1140
1141 stop:
1142         dt_trans_stop(env, d, th);
1143
1144         RETURN(rc);
1145 }
1146
1147 static int test_8_cb(const struct lu_env *env, struct llog_handle *llh,
1148                           struct llog_rec_hdr *rec, void *data)
1149 {
1150         plain_counter++;
1151         return 0;
1152 }
1153
1154 static int llog_test_8(const struct lu_env *env, struct obd_device *obd)
1155 {
1156         struct llog_handle      *llh = NULL;
1157         char                     name[10];
1158         int                      rc, rc2, i;
1159         int                      orig_counter;
1160         struct llog_mini_rec     lmr;
1161         struct llog_ctxt        *ctxt;
1162         struct dt_object        *obj = NULL;
1163
1164         ENTRY;
1165
1166         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1167         LASSERT(ctxt);
1168
1169         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
1170         lmr.lmr_hdr.lrh_type = 0xf00f00;
1171
1172         CWARN("8a: fill the first plain llog\n");
1173         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1174         if (rc) {
1175                 CERROR("8a: llog_create with logid failed: %d\n", rc);
1176                 GOTO(out_put, rc);
1177         }
1178
1179         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1180         if (rc) {
1181                 CERROR("8a: can't init llog handle: %d\n", rc);
1182                 GOTO(out, rc);
1183         }
1184
1185         plain_counter = 0;
1186         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1187         if (rc != 0) {
1188                 CERROR("5a: process with test_8_cb failed: %d\n", rc);
1189                 GOTO(out, rc);
1190         }
1191         orig_counter = plain_counter;
1192
1193         for (i = 0; i < 100; i++) {
1194                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1195                 if (rc) {
1196                         CERROR("5a: add record failed\n");
1197                         GOTO(out, rc);
1198                 }
1199         }
1200
1201         /* grab the current plain llog, we'll corrupt it later */
1202         obj = llh->u.chd.chd_current_log->lgh_obj;
1203         LASSERT(obj);
1204         lu_object_get(&obj->do_lu);
1205         CWARN("8a: pin llog "DFID"\n", PFID(lu_object_fid(&obj->do_lu)));
1206
1207         rc2 = llog_cat_close(env, llh);
1208         if (rc2) {
1209                 CERROR("8a: close log %s failed: %d\n", name, rc2);
1210                 if (rc == 0)
1211                         rc = rc2;
1212                 GOTO(out_put, rc);
1213         }
1214
1215         CWARN("8b: fill the second plain llog\n");
1216         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1217         if (rc) {
1218                 CERROR("8b: llog_create with logid failed: %d\n", rc);
1219                 GOTO(out_put, rc);
1220         }
1221
1222         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1223         if (rc) {
1224                 CERROR("8b: can't init llog handle: %d\n", rc);
1225                 GOTO(out, rc);
1226         }
1227
1228         for (i = 0; i < 100; i++) {
1229                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1230                 if (rc) {
1231                         CERROR("8b: add record failed\n");
1232                         GOTO(out, rc);
1233                 }
1234         }
1235         CWARN("8b: second llog "DFID"\n",
1236                 PFID(lu_object_fid(&llh->u.chd.chd_current_log->lgh_obj->do_lu)));
1237
1238         rc2 = llog_cat_close(env, llh);
1239         if (rc2) {
1240                 CERROR("8b: close log %s failed: %d\n", name, rc2);
1241                 if (rc == 0)
1242                         rc = rc2;
1243                 GOTO(out_put, rc);
1244         }
1245
1246         CWARN("8c: drop two records from the first plain llog\n");
1247         llog_truncate(env, obj);
1248
1249         CWARN("8d: count survived records\n");
1250         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1251         if (rc) {
1252                 CERROR("8d: llog_create with logid failed: %d\n", rc);
1253                 GOTO(out_put, rc);
1254         }
1255
1256         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1257         if (rc) {
1258                 CERROR("8d: can't init llog handle: %d\n", rc);
1259                 GOTO(out, rc);
1260         }
1261
1262         plain_counter = 0;
1263         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1264         if (rc != 0) {
1265                 CERROR("8d: process with test_8_cb failed: %d\n", rc);
1266                 GOTO(out, rc);
1267         }
1268
1269         if (orig_counter + 200 - 2 != plain_counter) {
1270                 CERROR("found %d records (expected %d)\n", plain_counter,
1271                        orig_counter + 200 - 2);
1272                 rc = -EIO;
1273         }
1274
1275 out:
1276         CWARN("8d: close re-opened catalog\n");
1277         rc2 = llog_cat_close(env, llh);
1278         if (rc2) {
1279                 CERROR("8d: close log %s failed: %d\n", name, rc2);
1280                 if (rc == 0)
1281                         rc = rc2;
1282         }
1283 out_put:
1284         llog_ctxt_put(ctxt);
1285
1286         if (obj != NULL)
1287                 lu_object_put(env, &obj->do_lu);
1288
1289         RETURN(rc);
1290 }
1291
1292 /* -------------------------------------------------------------------------
1293  * Tests above, boring obd functions below
1294  * ------------------------------------------------------------------------- */
1295 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
1296 {
1297         struct llog_handle      *llh = NULL;
1298         struct llog_ctxt        *ctxt;
1299         int                      rc, err;
1300         char                     name[10];
1301
1302         ENTRY;
1303         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1304         LASSERT(ctxt);
1305
1306         sprintf(name, "%x", llog_test_rand);
1307
1308         rc = llog_test_1(env, obd, name);
1309         if (rc)
1310                 GOTO(cleanup_ctxt, rc);
1311
1312         rc = llog_test_2(env, obd, name, &llh);
1313         if (rc)
1314                 GOTO(cleanup_ctxt, rc);
1315
1316         rc = llog_test_3(env, obd, llh);
1317         if (rc)
1318                 GOTO(cleanup, rc);
1319
1320         rc = llog_test_4(env, obd);
1321         if (rc)
1322                 GOTO(cleanup, rc);
1323
1324         rc = llog_test_5(env, obd);
1325         if (rc)
1326                 GOTO(cleanup, rc);
1327
1328         rc = llog_test_6(env, obd, name);
1329         if (rc)
1330                 GOTO(cleanup, rc);
1331
1332         rc = llog_test_7(env, obd);
1333         if (rc)
1334                 GOTO(cleanup, rc);
1335
1336         rc = llog_test_8(env, obd);
1337         if (rc)
1338                 GOTO(cleanup, rc);
1339
1340 cleanup:
1341         err = llog_destroy(env, llh);
1342         if (err)
1343                 CERROR("cleanup: llog_destroy failed: %d\n", err);
1344         llog_close(env, llh);
1345         if (rc == 0)
1346                 rc = err;
1347 cleanup_ctxt:
1348         llog_ctxt_put(ctxt);
1349         return rc;
1350 }
1351
1352 static int llog_test_cleanup(struct obd_device *obd)
1353 {
1354         struct obd_device       *tgt;
1355         struct lu_env            env;
1356         int                      rc;
1357
1358         ENTRY;
1359
1360         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1361         if (rc)
1362                 RETURN(rc);
1363
1364         tgt = obd->obd_lvfs_ctxt.dt->dd_lu_dev.ld_obd;
1365         rc = llog_cleanup(&env, llog_get_context(tgt, LLOG_TEST_ORIG_CTXT));
1366         if (rc)
1367                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
1368         lu_env_fini(&env);
1369         RETURN(rc);
1370 }
1371
1372 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
1373 {
1374         struct obd_device       *tgt;
1375         struct llog_ctxt        *ctxt;
1376         struct dt_object        *o;
1377         struct lu_env            env;
1378         struct lu_context        test_session;
1379         int                      rc;
1380
1381         ENTRY;
1382
1383         if (lcfg->lcfg_bufcount < 2) {
1384                 CERROR("requires a TARGET OBD name\n");
1385                 RETURN(-EINVAL);
1386         }
1387
1388         if (lcfg->lcfg_buflens[1] < 1) {
1389                 CERROR("requires a TARGET OBD name\n");
1390                 RETURN(-EINVAL);
1391         }
1392
1393         /* disk obd */
1394         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
1395         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
1396                 CERROR("target device not attached or not set up (%s)\n",
1397                        lustre_cfg_string(lcfg, 1));
1398                 RETURN(-EINVAL);
1399         }
1400
1401         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1402         if (rc)
1403                 RETURN(rc);
1404
1405         rc = lu_context_init(&test_session, LCT_SERVER_SESSION);
1406         if (rc)
1407                 GOTO(cleanup_env, rc);
1408         test_session.lc_thread = (struct ptlrpc_thread *)current;
1409         lu_context_enter(&test_session);
1410         env.le_ses = &test_session;
1411
1412         CWARN("Setup llog-test device over %s device\n",
1413               lustre_cfg_string(lcfg, 1));
1414
1415         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1416         obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);
1417
1418         rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,
1419                         &llog_osd_ops);
1420         if (rc)
1421                 GOTO(cleanup_session, rc);
1422
1423         /* use MGS llog dir for tests */
1424         ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);
1425         LASSERT(ctxt);
1426         o = ctxt->loc_dir;
1427         llog_ctxt_put(ctxt);
1428
1429         ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);
1430         LASSERT(ctxt);
1431         ctxt->loc_dir = o;
1432         llog_ctxt_put(ctxt);
1433
1434         llog_test_rand = cfs_rand();
1435
1436         rc = llog_run_tests(&env, tgt);
1437         if (rc)
1438                 llog_test_cleanup(obd);
1439 cleanup_session:
1440         lu_context_exit(&test_session);
1441         lu_context_fini(&test_session);
1442 cleanup_env:
1443         lu_env_fini(&env);
1444         RETURN(rc);
1445 }
1446
1447 static struct obd_ops llog_obd_ops = {
1448         .o_owner       = THIS_MODULE,
1449         .o_setup       = llog_test_setup,
1450         .o_cleanup     = llog_test_cleanup,
1451 };
1452
1453 static int __init llog_test_init(void)
1454 {
1455         return class_register_type(&llog_obd_ops, NULL, true, NULL,
1456                                    "llog_test", NULL);
1457 }
1458
1459 static void __exit llog_test_exit(void)
1460 {
1461         class_unregister_type("llog_test");
1462 }
1463
1464 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1465 MODULE_DESCRIPTION("llog test module");
1466 MODULE_LICENSE("GPL");
1467
1468 module_init(llog_test_init);
1469 module_exit(llog_test_exit);