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 /* Test record writing, single and in bulk */
 220 static int llog_test_3(const struct lu_env *env, struct obd_device *obd,
 221                        struct llog_handle *llh)
 222 {
 223         struct llog_gen_rec      lgr;
 224         int                      rc, i;
 225         int                      num_recs = 1; /* 1 for the header */
 226
 227         ENTRY;
 228
 229         lgr.lgr_hdr.lrh_len = lgr.lgr_tail.lrt_len = sizeof(lgr);
 230         lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
 231
 232         CWARN("3a: write one create_rec\n");
 233         rc = llog_write(env, llh,  &lgr.lgr_hdr, LLOG_NEXT_IDX);
 234         num_recs++;
 235         if (rc < 0) {
 236                 CERROR("3a: write one log record failed: %d\n", rc);
 237                 RETURN(rc);
 238         }
 239
 240         rc = verify_handle("3a", llh, num_recs);
 241         if (rc)
 242                 RETURN(rc);
 243
 244         CWARN("3c: write 1000 more log records\n");
 245         for (i = 0; i < 1000; i++) {
 246                 rc = llog_write(env, llh, &lgr.lgr_hdr, LLOG_NEXT_IDX);
 247                 if (rc < 0) {
 248                         CERROR("3c: write 1000 records failed at #%d: %d\n",
 249                                i + 1, rc);
 250                         RETURN(rc);
 251                 }
 252                 num_recs++;
 253         }
 254
 255         rc = verify_handle("3c", llh, num_recs);
 256         if (rc)
 257                 RETURN(rc);
 258
 259         CWARN("3d: write records with variable size until BITMAP_SIZE, "
 260               "return -ENOSPC\n");
 261         for (i = 0; i < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) + 1; i++) {
 262                 char                     buf[64];
 263                 struct llog_rec_hdr     *hdr = (void *)&buf;
 264
 265                 memset(buf, 0, sizeof buf);
 266                 if ((i % 2) == 0)
 267                         hdr->lrh_len = 40;
 268                 else
 269                         hdr->lrh_len = 64;
 270                 hdr->lrh_type = OBD_CFG_REC;
 271                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
 272
 273                 if (rc == -ENOSPC) {
 274                         break;
 275                 } else if (rc < 0) {
 276                         CERROR("3d: write recs failed at #%d: %d\n",
 277                                i + 1, rc);
 278                         RETURN(rc);
 279                 }
 280                 num_recs++;
 281         }
 282         if (rc != -ENOSPC) {
 283                 CWARN("3d: write record more than BITMAP size!\n");
 284                 RETURN(-EINVAL);
 285         }
 286         CWARN("3d: wrote %d more records before end of llog is reached\n",
 287               num_recs);
 288
 289         rc = verify_handle("3d", llh, num_recs);
 290
 291         RETURN(rc);
 292 }
 293
 294 /* Test catalogue additions */
 295 static int llog_test_4(const struct lu_env *env, struct obd_device *obd)
 296 {
 297         struct llog_handle      *cath;
 298         char                     name[10];
 299         int                      rc, rc2, i, buflen;
 300         struct llog_mini_rec     lmr;
 301         struct llog_cookie       cookie;
 302         struct llog_ctxt        *ctxt;
 303         int                      num_recs = 0;
 304         char                    *buf;
 305         struct llog_rec_hdr     *rec;
 306
 307         ENTRY;
 308
 309         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 310         LASSERT(ctxt);
 311
 312         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 313         lmr.lmr_hdr.lrh_type = 0xf00f00;
 314
 315         sprintf(name, "%x", llog_test_rand + 1);
 316         CWARN("4a: create a catalog log with name: %s\n", name);
 317         rc = llog_open_create(env, ctxt, &cath, NULL, name);
 318         if (rc) {
 319                 CERROR("4a: llog_create with name %s failed: %d\n", name, rc);
 320                 GOTO(ctxt_release, rc);
 321         }
 322         rc = llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);
 323         if (rc) {
 324                 CERROR("4a: can't init llog handle: %d\n", rc);
 325                 GOTO(out, rc);
 326         }
 327
 328         num_recs++;
 329         cat_logid = cath->lgh_id;
 330
 331         CWARN("4b: write 1 record into the catalog\n");
 332         rc = llog_cat_add(env, cath, &lmr.lmr_hdr, &cookie);
 333         if (rc != 1) {
 334                 CERROR("4b: write 1 catalog record failed at: %d\n", rc);
 335                 GOTO(out, rc);
 336         }
 337         num_recs++;
 338         rc = verify_handle("4b", cath, 2);
 339         if (rc)
 340                 GOTO(out, rc);
 341
 342         rc = verify_handle("4b", cath->u.chd.chd_current_log, num_recs);
 343         if (rc)
 344                 GOTO(out, rc);
 345
 346         CWARN("4c: cancel 1 log record\n");
 347         rc = llog_cat_cancel_records(env, cath, 1, &cookie);
 348         if (rc) {
 349                 CERROR("4c: cancel 1 catalog based record failed: %d\n", rc);
 350                 GOTO(out, rc);
 351         }
 352         num_recs--;
 353
 354         rc = verify_handle("4c", cath->u.chd.chd_current_log, num_recs);
 355         if (rc)
 356                 GOTO(out, rc);
 357
 358         CWARN("4d: write %d more log records\n", LLOG_TEST_RECNUM);
 359         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
 360                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
 361                 if (rc) {
 362                         CERROR("4d: write %d records failed at #%d: %d\n",
 363                                LLOG_TEST_RECNUM, i + 1, rc);
 364                         GOTO(out, rc);
 365                 }
 366                 num_recs++;
 367         }
 368
 369         /* make sure new plain llog appears */
 370         rc = verify_handle("4d", cath, 3);
 371         if (rc)
 372                 GOTO(out, rc);
 373
 374         CWARN("4e: add 5 large records, one record per block\n");
 375         buflen = LLOG_MIN_CHUNK_SIZE;
 376         OBD_ALLOC(buf, buflen);
 377         if (buf == NULL)
 378                 GOTO(out, rc = -ENOMEM);
 379         for (i = 0; i < 5; i++) {
 380                 rec = (void *)buf;
 381                 rec->lrh_len = buflen;
 382                 rec->lrh_type = OBD_CFG_REC;
 383                 rc = llog_cat_add(env, cath, rec, NULL);
 384                 if (rc) {
 385                         CERROR("4e: write 5 records failed at #%d: %d\n",
 386                                i + 1, rc);
 387                         GOTO(out_free, rc);
 388                 }
 389                 num_recs++;
 390         }
 391 out_free:
 392         OBD_FREE(buf, buflen);
 393 out:
 394         CWARN("4f: put newly-created catalog\n");
 395         rc2 = llog_cat_close(env, cath);
 396         if (rc2) {
 397                 CERROR("4: close log %s failed: %d\n", name, rc2);
 398                 if (rc == 0)
 399                         rc = rc2;
 400         }
 401 ctxt_release:
 402         llog_ctxt_put(ctxt);
 403         RETURN(rc);
 404 }
 405
 406 static int cat_counter;
 407
 408 static int cat_print_cb(const struct lu_env *env, struct llog_handle *llh,
 409                         struct llog_rec_hdr *rec, void *data)
 410 {
 411         struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
 412         struct lu_fid            fid = {0};
 413
 414         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
 415                 CERROR("invalid record in catalog\n");
 416                 RETURN(-EINVAL);
 417         }
 418
 419         logid_to_fid(&lir->lid_id, &fid);
 420
 421         CWARN("seeing record at index %d - "DFID" in log "DFID"\n",
 422               rec->lrh_index, PFID(&fid),
 423               PFID(lu_object_fid(&llh->lgh_obj->do_lu)));
 424
 425         cat_counter++;
 426
 427         RETURN(0);
 428 }
 429
 430 static int plain_counter;
 431
 432 static int plain_print_cb(const struct lu_env *env, struct llog_handle *llh,
 433                           struct llog_rec_hdr *rec, void *data)
 434 {
 435         struct lu_fid fid = {0};
 436
 437         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 438                 CERROR("log is not plain\n");
 439                 RETURN(-EINVAL);
 440         }
 441
 442         logid_to_fid(&llh->lgh_id, &fid);
 443
 444         CDEBUG(D_INFO, "seeing record at index %d in log "DFID"\n",
 445                rec->lrh_index, PFID(&fid));
 446
 447         plain_counter++;
 448
 449         RETURN(0);
 450 }
 451
 452 static int cancel_count;
 453
 454 static int llog_cancel_rec_cb(const struct lu_env *env,
 455                               struct llog_handle *llh,
 456                               struct llog_rec_hdr *rec, void *data)
 457 {
 458         struct llog_cookie cookie;
 459
 460         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 461                 CERROR("log is not plain\n");
 462                 RETURN(-EINVAL);
 463         }
 464
 465         cookie.lgc_lgl = llh->lgh_id;
 466         cookie.lgc_index = rec->lrh_index;
 467
 468         llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1, &cookie);
 469         cancel_count++;
 470         if (cancel_count == LLOG_TEST_RECNUM)
 471                 RETURN(-LLOG_EEMPTY);
 472         RETURN(0);
 473 }
 474
 475 /* Test log and catalogue processing */
 476 static int llog_test_5(const struct lu_env *env, struct obd_device *obd)
 477 {
 478         struct llog_handle      *llh = NULL;
 479         char                     name[10];
 480         int                      rc, rc2;
 481         struct llog_mini_rec     lmr;
 482         struct llog_ctxt        *ctxt;
 483
 484         ENTRY;
 485
 486         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 487         LASSERT(ctxt);
 488
 489         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 490         lmr.lmr_hdr.lrh_type = 0xf00f00;
 491
 492         CWARN("5a: re-open catalog by id\n");
 493         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
 494         if (rc) {
 495                 CERROR("5a: llog_create with logid failed: %d\n", rc);
 496                 GOTO(out_put, rc);
 497         }
 498
 499         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
 500         if (rc) {
 501                 CERROR("5a: can't init llog handle: %d\n", rc);
 502                 GOTO(out, rc);
 503         }
 504
 505         CWARN("5b: print the catalog entries.. we expect 2\n");
 506         cat_counter = 0;
 507         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 508         if (rc) {
 509                 CERROR("5b: process with cat_print_cb failed: %d\n", rc);
 510                 GOTO(out, rc);
 511         }
 512         if (cat_counter != 2) {
 513                 CERROR("5b: %d entries in catalog\n", cat_counter);
 514                 GOTO(out, rc = -EINVAL);
 515         }
 516
 517         CWARN("5c: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
 518         cancel_count = 0;
 519         rc = llog_cat_process(env, llh, llog_cancel_rec_cb, "foobar", 0, 0);
 520         if (rc != -LLOG_EEMPTY) {
 521                 CERROR("5c: process with llog_cancel_rec_cb failed: %d\n", rc);
 522                 GOTO(out, rc);
 523         }
 524
 525         CWARN("5c: print the catalog entries.. we expect 1\n");
 526         cat_counter = 0;
 527         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 528         if (rc) {
 529                 CERROR("5c: process with cat_print_cb failed: %d\n", rc);
 530                 GOTO(out, rc);
 531         }
 532         if (cat_counter != 1) {
 533                 CERROR("5c: %d entries in catalog\n", cat_counter);
 534                 GOTO(out, rc = -EINVAL);
 535         }
 536
 537         CWARN("5d: add 1 record to the log with many canceled empty pages\n");
 538         rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
 539         if (rc) {
 540                 CERROR("5d: add record to the log with many canceled empty "
 541                        "pages failed\n");
 542                 GOTO(out, rc);
 543         }
 544
 545         CWARN("5e: print plain log entries.. expect 6\n");
 546         plain_counter = 0;
 547         rc = llog_cat_process(env, llh, plain_print_cb, "foobar", 0, 0);
 548         if (rc) {
 549                 CERROR("5e: process with plain_print_cb failed: %d\n", rc);
 550                 GOTO(out, rc);
 551         }
 552         if (plain_counter != 6) {
 553                 CERROR("5e: found %d records\n", plain_counter);
 554                 GOTO(out, rc = -EINVAL);
 555         }
 556
 557         CWARN("5f: print plain log entries reversely.. expect 6\n");
 558         plain_counter = 0;
 559         rc = llog_cat_reverse_process(env, llh, plain_print_cb, "foobar");
 560         if (rc) {
 561                 CERROR("5f: reversely process with plain_print_cb failed: "
 562                        "%d\n", rc);
 563                 GOTO(out, rc);
 564         }
 565         if (plain_counter != 6) {
 566                 CERROR("5f: found %d records\n", plain_counter);
 567                 GOTO(out, rc = -EINVAL);
 568         }
 569
 570 out:
 571         CWARN("5g: close re-opened catalog\n");
 572         rc2 = llog_cat_close(env, llh);
 573         if (rc2) {
 574                 CERROR("5g: close log %s failed: %d\n", name, rc2);
 575                 if (rc == 0)
 576                         rc = rc2;
 577         }
 578 out_put:
 579         llog_ctxt_put(ctxt);
 580
 581         RETURN(rc);
 582 }
 583
 584 /* Test client api; open log by name and process */
 585 static int llog_test_6(const struct lu_env *env, struct obd_device *obd,
 586                        char *name)
 587 {
 588         struct obd_device       *mgc_obd;
 589         struct llog_ctxt        *ctxt;
 590         struct obd_uuid         *mgs_uuid;
 591         struct obd_export       *exp;
 592         struct obd_uuid          uuid = { "LLOG_TEST6_UUID" };
 593         struct llog_handle      *llh = NULL;
 594         struct llog_ctxt        *nctxt;
 595         int                      rc, rc2;
 596
 597         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 598         LASSERT(ctxt);
 599         mgs_uuid = &ctxt->loc_exp->exp_obd->obd_uuid;
 600
 601         CWARN("6a: re-open log %s using client API\n", name);
 602         mgc_obd = class_find_client_obd(mgs_uuid, LUSTRE_MGC_NAME, NULL);
 603         if (mgc_obd == NULL) {
 604                 CERROR("6a: no MGC devices connected to %s found.\n",
 605                        mgs_uuid->uuid);
 606                 GOTO(ctxt_release, rc = -ENOENT);
 607         }
 608
 609         rc = obd_connect(NULL, &exp, mgc_obd, &uuid,
 610                          NULL /* obd_connect_data */, NULL);
 611         if (rc != -EALREADY) {
 612                 CERROR("6a: connect on connected MGC (%s) failed to return"
 613                        " -EALREADY\n", mgc_obd->obd_name);
 614                 if (rc == 0)
 615                         obd_disconnect(exp);
 616                 GOTO(ctxt_release, rc = -EINVAL);
 617         }
 618
 619         nctxt = llog_get_context(mgc_obd, LLOG_CONFIG_REPL_CTXT);
 620         rc = llog_open(env, nctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
 621         if (rc) {
 622                 CERROR("6a: llog_open failed %d\n", rc);
 623                 GOTO(nctxt_put, rc);
 624         }
 625
 626         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
 627         if (rc) {
 628                 CERROR("6a: llog_init_handle failed %d\n", rc);
 629                 GOTO(parse_out, rc);
 630         }
 631
 632         plain_counter = 1; /* llog header is first record */
 633         CWARN("6b: process log %s using client API\n", name);
 634         rc = llog_process(env, llh, plain_print_cb, NULL, NULL);
 635         if (rc)
 636                 CERROR("6b: llog_process failed %d\n", rc);
 637         CWARN("6b: processed %d records\n", plain_counter);
 638
 639         rc = verify_handle("6b", llh, plain_counter);
 640         if (rc)
 641                 GOTO(parse_out, rc);
 642
 643         plain_counter = 1; /* llog header is first record */
 644         CWARN("6c: process log %s reversely using client API\n", name);
 645         rc = llog_reverse_process(env, llh, plain_print_cb, NULL, NULL);
 646         if (rc)
 647                 CERROR("6c: llog_reverse_process failed %d\n", rc);
 648         CWARN("6c: processed %d records\n", plain_counter);
 649
 650         rc = verify_handle("6c", llh, plain_counter);
 651         if (rc)
 652                 GOTO(parse_out, rc);
 653
 654 parse_out:
 655         rc2 = llog_close(env, llh);
 656         if (rc2) {
 657                 CERROR("6: llog_close failed: rc = %d\n", rc2);
 658                 if (rc == 0)
 659                         rc = rc2;
 660         }
 661 nctxt_put:
 662         llog_ctxt_put(nctxt);
 663 ctxt_release:
 664         llog_ctxt_put(ctxt);
 665         RETURN(rc);
 666 }
 667
 668 static union {
 669         struct llog_rec_hdr             lrh;   /* common header */
 670         struct llog_logid_rec           llr;   /* LLOG_LOGID_MAGIC */
 671         struct llog_unlink64_rec        lur;   /* MDS_UNLINK64_REC */
 672         struct llog_setattr64_rec       lsr64; /* MDS_SETATTR64_REC */
 673         struct llog_size_change_rec     lscr;  /* OST_SZ_REC */
 674         struct llog_changelog_rec       lcr;   /* CHANGELOG_REC */
 675         struct llog_changelog_user_rec  lcur;  /* CHANGELOG_USER_REC */
 676         struct llog_gen_rec             lgr;   /* LLOG_GEN_REC */
 677 } llog_records;
 678
 679 static int test_7_print_cb(const struct lu_env *env, struct llog_handle *llh,
 680                            struct llog_rec_hdr *rec, void *data)
 681 {
 682         struct lu_fid fid = {0};
 683
 684         logid_to_fid(&llh->lgh_id, &fid);
 685
 686         CDEBUG(D_OTHER, "record type %#x at index %d in log "DFID"\n",
 687                rec->lrh_type, rec->lrh_index, PFID(&fid));
 688
 689         plain_counter++;
 690         return 0;
 691 }
 692
 693 static int test_7_cancel_cb(const struct lu_env *env, struct llog_handle *llh,
 694                             struct llog_rec_hdr *rec, void *data)
 695 {
 696         plain_counter++;
 697         /* test LLOG_DEL_RECORD is working */
 698         return LLOG_DEL_RECORD;
 699 }
 700
 701 static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
 702 {
 703         struct llog_handle      *llh;
 704         int                      rc = 0, i, process_count;
 705         int                      num_recs = 0;
 706
 707         ENTRY;
 708
 709         rc = llog_open_create(env, ctxt, &llh, NULL, NULL);
 710         if (rc) {
 711                 CERROR("7_sub: create log failed\n");
 712                 RETURN(rc);
 713         }
 714
 715         rc = llog_init_handle(env, llh,
 716                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
 717                               &uuid);
 718         if (rc) {
 719                 CERROR("7_sub: can't init llog handle: %d\n", rc);
 720                 GOTO(out_close, rc);
 721         }
 722         for (i = 0; i < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr); i++) {
 723                 rc = llog_write(env, llh, &llog_records.lrh, LLOG_NEXT_IDX);
 724                 if (rc == -ENOSPC) {
 725                         break;
 726                 } else if (rc < 0) {
 727                         CERROR("7_sub: write recs failed at #%d: %d\n",
 728                                i + 1, rc);
 729                         GOTO(out_close, rc);
 730                 }
 731                 num_recs++;
 732         }
 733         if (rc != -ENOSPC) {
 734                 CWARN("7_sub: write record more than BITMAP size!\n");
 735                 GOTO(out_close, rc = -EINVAL);
 736         }
 737
 738         rc = verify_handle("7_sub", llh, num_recs + 1);
 739         if (rc) {
 740                 CERROR("7_sub: verify handle failed: %d\n", rc);
 741                 GOTO(out_close, rc);
 742         }
 743         if (num_recs < LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1)
 744                 CWARN("7_sub: records are not aligned, written %d from %u\n",
 745                       num_recs, LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1);
 746
 747         plain_counter = 0;
 748         rc = llog_process(env, llh, test_7_print_cb, "test 7", NULL);
 749         if (rc) {
 750                 CERROR("7_sub: llog process failed: %d\n", rc);
 751                 GOTO(out_close, rc);
 752         }
 753         process_count = plain_counter;
 754         if (process_count != num_recs) {
 755                 CERROR("7_sub: processed %d records from %d total\n",
 756                        process_count, num_recs);
 757                 GOTO(out_close, rc = -EINVAL);
 758         }
 759
 760         plain_counter = 0;
 761         rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);
 762         if (rc && rc != LLOG_DEL_PLAIN) {
 763                 CERROR("7_sub: reverse llog process failed: %d\n", rc);
 764                 GOTO(out_close, rc);
 765         }
 766         if (process_count != plain_counter) {
 767                 CERROR("7_sub: Reverse/direct processing found different"
 768                        "number of records: %d/%d\n",
 769                        plain_counter, process_count);
 770                 GOTO(out_close, rc = -EINVAL);
 771         }
 772         if (llog_exist(llh)) {
 773                 CERROR("7_sub: llog exists but should be zapped\n");
 774                 GOTO(out_close, rc = -EEXIST);
 775         }
 776
 777         rc = verify_handle("7_sub", llh, 1);
 778 out_close:
 779         if (rc)
 780                 llog_destroy(env, llh);
 781         llog_close(env, llh);
 782         RETURN(rc);
 783 }
 784
 785 /* Test all llog records writing and processing */
 786 static int llog_test_7(const struct lu_env *env, struct obd_device *obd)
 787 {
 788         struct llog_ctxt        *ctxt;
 789         int                      rc;
 790
 791         ENTRY;
 792
 793         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 794
 795         CWARN("7a: test llog_logid_rec\n");
 796         llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);
 797         llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);
 798         llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
 799
 800         rc = llog_test_7_sub(env, ctxt);
 801         if (rc) {
 802                 CERROR("7a: llog_logid_rec test failed\n");
 803                 GOTO(out, rc);
 804         }
 805
 806         CWARN("7b: test llog_unlink64_rec\n");
 807         llog_records.lur.lur_hdr.lrh_len = sizeof(llog_records.lur);
 808         llog_records.lur.lur_tail.lrt_len = sizeof(llog_records.lur);
 809         llog_records.lur.lur_hdr.lrh_type = MDS_UNLINK64_REC;
 810
 811         rc = llog_test_7_sub(env, ctxt);
 812         if (rc) {
 813                 CERROR("7b: llog_unlink_rec test failed\n");
 814                 GOTO(out, rc);
 815         }
 816
 817         CWARN("7c: test llog_setattr64_rec\n");
 818         llog_records.lsr64.lsr_hdr.lrh_len = sizeof(llog_records.lsr64);
 819         llog_records.lsr64.lsr_tail.lrt_len = sizeof(llog_records.lsr64);
 820         llog_records.lsr64.lsr_hdr.lrh_type = MDS_SETATTR64_REC;
 821
 822         rc = llog_test_7_sub(env, ctxt);
 823         if (rc) {
 824                 CERROR("7c: llog_setattr64_rec test failed\n");
 825                 GOTO(out, rc);
 826         }
 827
 828         CWARN("7d: test llog_size_change_rec\n");
 829         llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);
 830         llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);
 831         llog_records.lscr.lsc_hdr.lrh_type = OST_SZ_REC;
 832
 833         rc = llog_test_7_sub(env, ctxt);
 834         if (rc) {
 835                 CERROR("7d: llog_size_change_rec test failed\n");
 836                 GOTO(out, rc);
 837         }
 838
 839         CWARN("7e: test llog_changelog_rec\n");
 840         /* Direct access to cr_do_not_use: peculiar case for this test */
 841         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
 842         llog_records.lcr.cr_do_not_use.lrt_len = sizeof(llog_records.lcr);
 843         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
 844
 845         rc = llog_test_7_sub(env, ctxt);
 846         if (rc) {
 847                 CERROR("7e: llog_changelog_rec test failed\n");
 848                 GOTO(out, rc);
 849         }
 850
 851         CWARN("7f: test llog_changelog_user_rec\n");
 852         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
 853         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
 854         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
 855
 856         rc = llog_test_7_sub(env, ctxt);
 857         if (rc) {
 858                 CERROR("7f: llog_changelog_user_rec test failed\n");
 859                 GOTO(out, rc);
 860         }
 861
 862         CWARN("7g: test llog_gen_rec\n");
 863         llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);
 864         llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);
 865         llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
 866
 867         rc = llog_test_7_sub(env, ctxt);
 868         if (rc) {
 869                 CERROR("7g: llog_size_change_rec test failed\n");
 870                 GOTO(out, rc);
 871         }
 872 out:
 873         llog_ctxt_put(ctxt);
 874         RETURN(rc);
 875 }
 876
 877 static int llog_truncate(const struct lu_env *env, struct dt_object *o)
 878 {
 879         struct lu_attr           la;
 880         struct thandle          *th;
 881         struct dt_device        *d;
 882         int                      rc;
 883         ENTRY;
 884
 885         LASSERT(o);
 886         d = lu2dt_dev(o->do_lu.lo_dev);
 887         LASSERT(d);
 888
 889         rc = dt_attr_get(env, o, &la);
 890         if (rc)
 891                 RETURN(rc);
 892
 893         CDEBUG(D_OTHER, "original size "LPU64"\n", la.la_size);
 894         rc = sizeof(struct llog_log_hdr) + sizeof(struct llog_mini_rec);
 895         if (la.la_size < rc) {
 896                 CERROR("too small llog: "LPU64"\n", la.la_size);
 897                 RETURN(0);
 898         }
 899
 900         /* drop 2 records */
 901         la.la_size = la.la_size - (sizeof(struct llog_mini_rec) * 2);
 902         la.la_valid = LA_SIZE;
 903
 904         th = dt_trans_create(env, d);
 905         if (IS_ERR(th))
 906                 RETURN(PTR_ERR(th));
 907
 908         rc = dt_declare_attr_set(env, o, &la, th);
 909         if (rc)
 910                 GOTO(stop, rc);
 911
 912         rc = dt_declare_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
 913
 914         rc = dt_trans_start_local(env, d, th);
 915         if (rc)
 916                 GOTO(stop, rc);
 917
 918         rc = dt_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
 919         if (rc)
 920                 GOTO(stop, rc);
 921
 922         rc = dt_attr_set(env, o, &la, th);
 923         if (rc)
 924                 GOTO(stop, rc);
 925
 926 stop:
 927         dt_trans_stop(env, d, th);
 928
 929         RETURN(rc);
 930 }
 931
 932 static int test_8_cb(const struct lu_env *env, struct llog_handle *llh,
 933                           struct llog_rec_hdr *rec, void *data)
 934 {
 935         plain_counter++;
 936         return 0;
 937 }
 938
 939 static int llog_test_8(const struct lu_env *env, struct obd_device *obd)
 940 {
 941         struct llog_handle      *llh = NULL;
 942         char                     name[10];
 943         int                      rc, rc2, i;
 944         int                      orig_counter;
 945         struct llog_mini_rec     lmr;
 946         struct llog_ctxt        *ctxt;
 947         struct dt_object        *obj = NULL;
 948
 949         ENTRY;
 950
 951         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 952         LASSERT(ctxt);
 953
 954         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 955         lmr.lmr_hdr.lrh_type = 0xf00f00;
 956
 957         CWARN("8a: fill the first plain llog\n");
 958         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
 959         if (rc) {
 960                 CERROR("8a: llog_create with logid failed: %d\n", rc);
 961                 GOTO(out_put, rc);
 962         }
 963
 964         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
 965         if (rc) {
 966                 CERROR("8a: can't init llog handle: %d\n", rc);
 967                 GOTO(out, rc);
 968         }
 969
 970         plain_counter = 0;
 971         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
 972         if (rc != 0) {
 973                 CERROR("5a: process with test_8_cb failed: %d\n", rc);
 974                 GOTO(out, rc);
 975         }
 976         orig_counter = plain_counter;
 977
 978         for (i = 0; i < 100; i++) {
 979                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
 980                 if (rc) {
 981                         CERROR("5a: add record failed\n");
 982                         GOTO(out, rc);
 983                 }
 984         }
 985
 986         /* grab the current plain llog, we'll corrupt it later */
 987         obj = llh->u.chd.chd_current_log->lgh_obj;
 988         LASSERT(obj);
 989         lu_object_get(&obj->do_lu);
 990         CWARN("8a: pin llog "DFID"\n", PFID(lu_object_fid(&obj->do_lu)));
 991
 992         rc2 = llog_cat_close(env, llh);
 993         if (rc2) {
 994                 CERROR("8a: close log %s failed: %d\n", name, rc2);
 995                 if (rc == 0)
 996                         rc = rc2;
 997                 GOTO(out_put, rc);
 998         }
 999
1000         CWARN("8b: fill the second plain llog\n");
1001         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1002         if (rc) {
1003                 CERROR("8b: llog_create with logid failed: %d\n", rc);
1004                 GOTO(out_put, rc);
1005         }
1006
1007         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1008         if (rc) {
1009                 CERROR("8b: can't init llog handle: %d\n", rc);
1010                 GOTO(out, rc);
1011         }
1012
1013         for (i = 0; i < 100; i++) {
1014                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1015                 if (rc) {
1016                         CERROR("8b: add record failed\n");
1017                         GOTO(out, rc);
1018                 }
1019         }
1020         CWARN("8b: second llog "DFID"\n",
1021                 PFID(lu_object_fid(&llh->u.chd.chd_current_log->lgh_obj->do_lu)));
1022
1023         rc2 = llog_cat_close(env, llh);
1024         if (rc2) {
1025                 CERROR("8b: close log %s failed: %d\n", name, rc2);
1026                 if (rc == 0)
1027                         rc = rc2;
1028                 GOTO(out_put, rc);
1029         }
1030
1031         CWARN("8c: drop two records from the first plain llog\n");
1032         llog_truncate(env, obj);
1033
1034         CWARN("8d: count survived records\n");
1035         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1036         if (rc) {
1037                 CERROR("8d: llog_create with logid failed: %d\n", rc);
1038                 GOTO(out_put, rc);
1039         }
1040
1041         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1042         if (rc) {
1043                 CERROR("8d: can't init llog handle: %d\n", rc);
1044                 GOTO(out, rc);
1045         }
1046
1047         plain_counter = 0;
1048         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1049         if (rc != 0) {
1050                 CERROR("8d: process with test_8_cb failed: %d\n", rc);
1051                 GOTO(out, rc);
1052         }
1053
1054         if (orig_counter + 200 - 2 != plain_counter) {
1055                 CERROR("found %d records (expected %d)\n", plain_counter,
1056                        orig_counter + 200 - 2);
1057                 rc = -EIO;
1058         }
1059
1060 out:
1061         CWARN("8d: close re-opened catalog\n");
1062         rc2 = llog_cat_close(env, llh);
1063         if (rc2) {
1064                 CERROR("8d: close log %s failed: %d\n", name, rc2);
1065                 if (rc == 0)
1066                         rc = rc2;
1067         }
1068 out_put:
1069         llog_ctxt_put(ctxt);
1070
1071         if (obj != NULL)
1072                 lu_object_put(env, &obj->do_lu);
1073
1074         RETURN(rc);
1075 }
1076
1077 /* -------------------------------------------------------------------------
1078  * Tests above, boring obd functions below
1079  * ------------------------------------------------------------------------- */
1080 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
1081 {
1082         struct llog_handle      *llh = NULL;
1083         struct llog_ctxt        *ctxt;
1084         int                      rc, err;
1085         char                     name[10];
1086
1087         ENTRY;
1088         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1089         LASSERT(ctxt);
1090
1091         sprintf(name, "%x", llog_test_rand);
1092
1093         rc = llog_test_1(env, obd, name);
1094         if (rc)
1095                 GOTO(cleanup_ctxt, rc);
1096
1097         rc = llog_test_2(env, obd, name, &llh);
1098         if (rc)
1099                 GOTO(cleanup_ctxt, rc);
1100
1101         rc = llog_test_3(env, obd, llh);
1102         if (rc)
1103                 GOTO(cleanup, rc);
1104
1105         rc = llog_test_4(env, obd);
1106         if (rc)
1107                 GOTO(cleanup, rc);
1108
1109         rc = llog_test_5(env, obd);
1110         if (rc)
1111                 GOTO(cleanup, rc);
1112
1113         rc = llog_test_6(env, obd, name);
1114         if (rc)
1115                 GOTO(cleanup, rc);
1116
1117         rc = llog_test_7(env, obd);
1118         if (rc)
1119                 GOTO(cleanup, rc);
1120
1121         rc = llog_test_8(env, obd);
1122         if (rc)
1123                 GOTO(cleanup, rc);
1124
1125 cleanup:
1126         err = llog_destroy(env, llh);
1127         if (err)
1128                 CERROR("cleanup: llog_destroy failed: %d\n", err);
1129         llog_close(env, llh);
1130         if (rc == 0)
1131                 rc = err;
1132 cleanup_ctxt:
1133         llog_ctxt_put(ctxt);
1134         return rc;
1135 }
1136
1137 static int llog_test_cleanup(struct obd_device *obd)
1138 {
1139         struct obd_device       *tgt;
1140         struct lu_env            env;
1141         int                      rc;
1142
1143         ENTRY;
1144
1145         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1146         if (rc)
1147                 RETURN(rc);
1148
1149         tgt = obd->obd_lvfs_ctxt.dt->dd_lu_dev.ld_obd;
1150         rc = llog_cleanup(&env, llog_get_context(tgt, LLOG_TEST_ORIG_CTXT));
1151         if (rc)
1152                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
1153         lu_env_fini(&env);
1154         RETURN(rc);
1155 }
1156
1157 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
1158 {
1159         struct obd_device       *tgt;
1160         struct llog_ctxt        *ctxt;
1161         struct dt_object        *o;
1162         struct lu_env            env;
1163         struct lu_context        test_session;
1164         int                      rc;
1165
1166         ENTRY;
1167
1168         if (lcfg->lcfg_bufcount < 2) {
1169                 CERROR("requires a TARGET OBD name\n");
1170                 RETURN(-EINVAL);
1171         }
1172
1173         if (lcfg->lcfg_buflens[1] < 1) {
1174                 CERROR("requires a TARGET OBD name\n");
1175                 RETURN(-EINVAL);
1176         }
1177
1178         /* disk obd */
1179         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
1180         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
1181                 CERROR("target device not attached or not set up (%s)\n",
1182                        lustre_cfg_string(lcfg, 1));
1183                 RETURN(-EINVAL);
1184         }
1185
1186         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1187         if (rc)
1188                 RETURN(rc);
1189
1190         rc = lu_context_init(&test_session, LCT_SERVER_SESSION);
1191         if (rc)
1192                 GOTO(cleanup_env, rc);
1193         test_session.lc_thread = (struct ptlrpc_thread *)current;
1194         lu_context_enter(&test_session);
1195         env.le_ses = &test_session;
1196
1197         CWARN("Setup llog-test device over %s device\n",
1198               lustre_cfg_string(lcfg, 1));
1199
1200         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1201         obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);
1202
1203         rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,
1204                         &llog_osd_ops);
1205         if (rc)
1206                 GOTO(cleanup_session, rc);
1207
1208         /* use MGS llog dir for tests */
1209         ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);
1210         LASSERT(ctxt);
1211         o = ctxt->loc_dir;
1212         llog_ctxt_put(ctxt);
1213
1214         ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);
1215         LASSERT(ctxt);
1216         ctxt->loc_dir = o;
1217         llog_ctxt_put(ctxt);
1218
1219         llog_test_rand = cfs_rand();
1220
1221         rc = llog_run_tests(&env, tgt);
1222         if (rc)
1223                 llog_test_cleanup(obd);
1224 cleanup_session:
1225         lu_context_exit(&test_session);
1226         lu_context_fini(&test_session);
1227 cleanup_env:
1228         lu_env_fini(&env);
1229         RETURN(rc);
1230 }
1231
1232 static struct obd_ops llog_obd_ops = {
1233         .o_owner       = THIS_MODULE,
1234         .o_setup       = llog_test_setup,
1235         .o_cleanup     = llog_test_cleanup,
1236 };
1237
1238 static int __init llog_test_init(void)
1239 {
1240         return class_register_type(&llog_obd_ops, NULL, true, NULL,
1241                                    "llog_test", NULL);
1242 }
1243
1244 static void __exit llog_test_exit(void)
1245 {
1246         class_unregister_type("llog_test");
1247 }
1248
1249 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1250 MODULE_DESCRIPTION("llog test module");
1251 MODULE_LICENSE("GPL");
1252
1253 module_init(llog_test_init);
1254 module_exit(llog_test_exit);