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, 2013, 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_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_BITMAP_BYTES * 8; i++) {
  72                 if (ext2_test_bit(i, llh->lgh_hdr->llh_bitmap)) {
  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_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_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 cat_cancel_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_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_BITMAP_SIZE(llh->lgh_hdr) - 1)
 744                 CWARN("7_sub: records are not aligned, written %d from %u\n",
 745                       num_recs, LLOG_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         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
 841         llog_records.lcr.cr_tail.lrt_len = sizeof(llog_records.lcr);
 842         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
 843
 844         rc = llog_test_7_sub(env, ctxt);
 845         if (rc) {
 846                 CERROR("7e: llog_changelog_rec test failed\n");
 847                 GOTO(out, rc);
 848         }
 849
 850         CWARN("7f: test llog_changelog_user_rec\n");
 851         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
 852         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
 853         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
 854
 855         rc = llog_test_7_sub(env, ctxt);
 856         if (rc) {
 857                 CERROR("7f: llog_changelog_user_rec test failed\n");
 858                 GOTO(out, rc);
 859         }
 860
 861         CWARN("7g: test llog_gen_rec\n");
 862         llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);
 863         llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);
 864         llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
 865
 866         rc = llog_test_7_sub(env, ctxt);
 867         if (rc) {
 868                 CERROR("7g: llog_size_change_rec test failed\n");
 869                 GOTO(out, rc);
 870         }
 871 out:
 872         llog_ctxt_put(ctxt);
 873         RETURN(rc);
 874 }
 875
 876 static int llog_truncate(const struct lu_env *env, struct dt_object *o)
 877 {
 878         struct lu_attr           la;
 879         struct thandle          *th;
 880         struct dt_device        *d;
 881         int                      rc;
 882         ENTRY;
 883
 884         LASSERT(o);
 885         d = lu2dt_dev(o->do_lu.lo_dev);
 886         LASSERT(d);
 887
 888         rc = dt_attr_get(env, o, &la, NULL);
 889         if (rc)
 890                 RETURN(rc);
 891
 892         CDEBUG(D_OTHER, "original size "LPU64"\n", la.la_size);
 893         rc = sizeof(struct llog_log_hdr) + sizeof(struct llog_mini_rec);
 894         if (la.la_size < rc) {
 895                 CERROR("too small llog: "LPU64"\n", la.la_size);
 896                 RETURN(0);
 897         }
 898
 899         /* drop 2 records */
 900         la.la_size = la.la_size - (sizeof(struct llog_mini_rec) * 2);
 901         la.la_valid = LA_SIZE;
 902
 903         th = dt_trans_create(env, d);
 904         if (IS_ERR(th))
 905                 RETURN(PTR_ERR(th));
 906
 907         rc = dt_declare_attr_set(env, o, &la, th);
 908         if (rc)
 909                 GOTO(stop, rc);
 910
 911         rc = dt_declare_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
 912
 913         rc = dt_trans_start_local(env, d, th);
 914         if (rc)
 915                 GOTO(stop, rc);
 916
 917         rc = dt_punch(env, o, la.la_size, OBD_OBJECT_EOF, th, BYPASS_CAPA);
 918         if (rc)
 919                 GOTO(stop, rc);
 920
 921         rc = dt_attr_set(env, o, &la, th, BYPASS_CAPA);
 922         if (rc)
 923                 GOTO(stop, rc);
 924
 925 stop:
 926         dt_trans_stop(env, d, th);
 927
 928         RETURN(rc);
 929 }
 930
 931 static int test_8_cb(const struct lu_env *env, struct llog_handle *llh,
 932                           struct llog_rec_hdr *rec, void *data)
 933 {
 934         plain_counter++;
 935         return 0;
 936 }
 937
 938 static int llog_test_8(const struct lu_env *env, struct obd_device *obd)
 939 {
 940         struct llog_handle      *llh = NULL;
 941         char                     name[10];
 942         int                      rc, rc2, i;
 943         int                      orig_counter;
 944         struct llog_mini_rec     lmr;
 945         struct llog_ctxt        *ctxt;
 946         struct dt_object        *obj = NULL;
 947
 948         ENTRY;
 949
 950         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 951         LASSERT(ctxt);
 952
 953         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 954         lmr.lmr_hdr.lrh_type = 0xf00f00;
 955
 956         CWARN("8a: fill the first plain llog\n");
 957         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
 958         if (rc) {
 959                 CERROR("8a: llog_create with logid failed: %d\n", rc);
 960                 GOTO(out_put, rc);
 961         }
 962
 963         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
 964         if (rc) {
 965                 CERROR("8a: can't init llog handle: %d\n", rc);
 966                 GOTO(out, rc);
 967         }
 968
 969         plain_counter = 0;
 970         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
 971         if (rc != 0) {
 972                 CERROR("5a: process with cat_cancel_cb failed: %d\n", rc);
 973                 GOTO(out, rc);
 974         }
 975         orig_counter = plain_counter;
 976
 977         for (i = 0; i < 100; i++) {
 978                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
 979                 if (rc) {
 980                         CERROR("5a: add record failed\n");
 981                         GOTO(out, rc);
 982                 }
 983         }
 984
 985         /* grab the current plain llog, we'll corrupt it later */
 986         obj = llh->u.chd.chd_current_log->lgh_obj;
 987         LASSERT(obj);
 988         lu_object_get(&obj->do_lu);
 989         CWARN("8a: pin llog "DFID"\n", PFID(lu_object_fid(&obj->do_lu)));
 990
 991         rc2 = llog_cat_close(env, llh);
 992         if (rc2) {
 993                 CERROR("8a: close log %s failed: %d\n", name, rc2);
 994                 if (rc == 0)
 995                         rc = rc2;
 996                 GOTO(out_put, rc);
 997         }
 998
 999         CWARN("8b: fill the second plain llog\n");
1000         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1001         if (rc) {
1002                 CERROR("8b: llog_create with logid failed: %d\n", rc);
1003                 GOTO(out_put, rc);
1004         }
1005
1006         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1007         if (rc) {
1008                 CERROR("8b: can't init llog handle: %d\n", rc);
1009                 GOTO(out, rc);
1010         }
1011
1012         for (i = 0; i < 100; i++) {
1013                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1014                 if (rc) {
1015                         CERROR("8b: add record failed\n");
1016                         GOTO(out, rc);
1017                 }
1018         }
1019         CWARN("8b: second llog "DFID"\n",
1020                 PFID(lu_object_fid(&llh->u.chd.chd_current_log->lgh_obj->do_lu)));
1021
1022         rc2 = llog_cat_close(env, llh);
1023         if (rc2) {
1024                 CERROR("8b: close log %s failed: %d\n", name, rc2);
1025                 if (rc == 0)
1026                         rc = rc2;
1027                 GOTO(out_put, rc);
1028         }
1029
1030         CWARN("8c: drop two records from the first plain llog\n");
1031         llog_truncate(env, obj);
1032
1033         CWARN("8d: count survived records\n");
1034         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1035         if (rc) {
1036                 CERROR("8d: llog_create with logid failed: %d\n", rc);
1037                 GOTO(out_put, rc);
1038         }
1039
1040         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1041         if (rc) {
1042                 CERROR("8d: can't init llog handle: %d\n", rc);
1043                 GOTO(out, rc);
1044         }
1045
1046         plain_counter = 0;
1047         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1048         if (rc != 0) {
1049                 CERROR("8d: process with cat_cancel_cb failed: %d\n", rc);
1050                 GOTO(out, rc);
1051         }
1052
1053         if (orig_counter + 200 - 2 != plain_counter) {
1054                 CERROR("found %d records (expected %d)\n", plain_counter,
1055                        orig_counter + 200 - 2);
1056                 rc = -EIO;
1057         }
1058
1059 out:
1060         CWARN("8d: close re-opened catalog\n");
1061         rc2 = llog_cat_close(env, llh);
1062         if (rc2) {
1063                 CERROR("8d: close log %s failed: %d\n", name, rc2);
1064                 if (rc == 0)
1065                         rc = rc2;
1066         }
1067 out_put:
1068         llog_ctxt_put(ctxt);
1069
1070         if (obj != NULL)
1071                 lu_object_put(env, &obj->do_lu);
1072
1073         RETURN(rc);
1074 }
1075
1076 /* -------------------------------------------------------------------------
1077  * Tests above, boring obd functions below
1078  * ------------------------------------------------------------------------- */
1079 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
1080 {
1081         struct llog_handle      *llh = NULL;
1082         struct llog_ctxt        *ctxt;
1083         int                      rc, err;
1084         char                     name[10];
1085
1086         ENTRY;
1087         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1088         LASSERT(ctxt);
1089
1090         sprintf(name, "%x", llog_test_rand);
1091
1092         rc = llog_test_1(env, obd, name);
1093         if (rc)
1094                 GOTO(cleanup_ctxt, rc);
1095
1096         rc = llog_test_2(env, obd, name, &llh);
1097         if (rc)
1098                 GOTO(cleanup_ctxt, rc);
1099
1100         rc = llog_test_3(env, obd, llh);
1101         if (rc)
1102                 GOTO(cleanup, rc);
1103
1104         rc = llog_test_4(env, obd);
1105         if (rc)
1106                 GOTO(cleanup, rc);
1107
1108         rc = llog_test_5(env, obd);
1109         if (rc)
1110                 GOTO(cleanup, rc);
1111
1112         rc = llog_test_6(env, obd, name);
1113         if (rc)
1114                 GOTO(cleanup, rc);
1115
1116         rc = llog_test_7(env, obd);
1117         if (rc)
1118                 GOTO(cleanup, rc);
1119
1120         rc = llog_test_8(env, obd);
1121         if (rc)
1122                 GOTO(cleanup, rc);
1123
1124 cleanup:
1125         err = llog_destroy(env, llh);
1126         if (err)
1127                 CERROR("cleanup: llog_destroy failed: %d\n", err);
1128         llog_close(env, llh);
1129         if (rc == 0)
1130                 rc = err;
1131 cleanup_ctxt:
1132         llog_ctxt_put(ctxt);
1133         return rc;
1134 }
1135
1136 static int llog_test_cleanup(struct obd_device *obd)
1137 {
1138         struct obd_device       *tgt;
1139         struct lu_env            env;
1140         int                      rc;
1141
1142         ENTRY;
1143
1144         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1145         if (rc)
1146                 RETURN(rc);
1147
1148         tgt = obd->obd_lvfs_ctxt.dt->dd_lu_dev.ld_obd;
1149         rc = llog_cleanup(&env, llog_get_context(tgt, LLOG_TEST_ORIG_CTXT));
1150         if (rc)
1151                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
1152         lu_env_fini(&env);
1153         RETURN(rc);
1154 }
1155
1156 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
1157 {
1158         struct obd_device       *tgt;
1159         struct llog_ctxt        *ctxt;
1160         struct dt_object        *o;
1161         struct lu_env            env;
1162         struct lu_context        test_session;
1163         int                      rc;
1164
1165         ENTRY;
1166
1167         if (lcfg->lcfg_bufcount < 2) {
1168                 CERROR("requires a TARGET OBD name\n");
1169                 RETURN(-EINVAL);
1170         }
1171
1172         if (lcfg->lcfg_buflens[1] < 1) {
1173                 CERROR("requires a TARGET OBD name\n");
1174                 RETURN(-EINVAL);
1175         }
1176
1177         /* disk obd */
1178         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
1179         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
1180                 CERROR("target device not attached or not set up (%s)\n",
1181                        lustre_cfg_string(lcfg, 1));
1182                 RETURN(-EINVAL);
1183         }
1184
1185         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1186         if (rc)
1187                 RETURN(rc);
1188
1189         rc = lu_context_init(&test_session, LCT_SERVER_SESSION);
1190         if (rc)
1191                 GOTO(cleanup_env, rc);
1192         test_session.lc_thread = (struct ptlrpc_thread *)current;
1193         lu_context_enter(&test_session);
1194         env.le_ses = &test_session;
1195
1196         CWARN("Setup llog-test device over %s device\n",
1197               lustre_cfg_string(lcfg, 1));
1198
1199         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1200         obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);
1201
1202         rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,
1203                         &llog_osd_ops);
1204         if (rc)
1205                 GOTO(cleanup_session, rc);
1206
1207         /* use MGS llog dir for tests */
1208         ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);
1209         LASSERT(ctxt);
1210         o = ctxt->loc_dir;
1211         llog_ctxt_put(ctxt);
1212
1213         ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);
1214         LASSERT(ctxt);
1215         ctxt->loc_dir = o;
1216         llog_ctxt_put(ctxt);
1217
1218         llog_test_rand = cfs_rand();
1219
1220         rc = llog_run_tests(&env, tgt);
1221         if (rc)
1222                 llog_test_cleanup(obd);
1223 cleanup_session:
1224         lu_context_exit(&test_session);
1225         lu_context_fini(&test_session);
1226 cleanup_env:
1227         lu_env_fini(&env);
1228         RETURN(rc);
1229 }
1230
1231 static struct obd_ops llog_obd_ops = {
1232         .o_owner       = THIS_MODULE,
1233         .o_setup       = llog_test_setup,
1234         .o_cleanup     = llog_test_cleanup,
1235 };
1236
1237 static int __init llog_test_init(void)
1238 {
1239         return class_register_type(&llog_obd_ops, NULL, true, NULL,
1240 #ifndef HAVE_ONLY_PROCFS_SEQ
1241                                    NULL,
1242 #endif
1243                                    "llog_test", NULL);
1244 }
1245
1246 static void __exit llog_test_exit(void)
1247 {
1248         class_unregister_type("llog_test");
1249 }
1250
1251 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1252 MODULE_DESCRIPTION("llog test module");
1253 MODULE_LICENSE("GPL");
1254
1255 module_init(llog_test_init);
1256 module_exit(llog_test_exit);