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) 2011, 2012, Intel, Inc.
  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         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_rec(env, llh,  &lgr.lgr_hdr, NULL, 0, NULL, -1);
 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("3b: write 10 cfg log records with 8 bytes bufs\n");
 245         for (i = 0; i < 10; i++) {
 246                 struct llog_rec_hdr     hdr;
 247                 char                    buf[8];
 248
 249                 hdr.lrh_len = 8;
 250                 hdr.lrh_type = OBD_CFG_REC;
 251                 memset(buf, 0, sizeof buf);
 252                 rc = llog_write_rec(env, llh, &hdr, NULL, 0, buf, -1);
 253                 if (rc < 0) {
 254                         CERROR("3b: write 10 records failed at #%d: %d\n",
 255                                i + 1, rc);
 256                         RETURN(rc);
 257                 }
 258                 num_recs++;
 259         }
 260
 261         rc = verify_handle("3b", llh, num_recs);
 262         if (rc)
 263                 RETURN(rc);
 264
 265         CWARN("3c: write 1000 more log records\n");
 266         for (i = 0; i < 1000; i++) {
 267                 rc = llog_write_rec(env, llh, &lgr.lgr_hdr, NULL, 0, NULL,
 268                                     -1);
 269                 if (rc < 0) {
 270                         CERROR("3c: write 1000 records failed at #%d: %d\n",
 271                                i + 1, rc);
 272                         RETURN(rc);
 273                 }
 274                 num_recs++;
 275         }
 276
 277         rc = verify_handle("3c", llh, num_recs);
 278         if (rc)
 279                 RETURN(rc);
 280
 281         CWARN("3d: write log more than BITMAP_SIZE, return -ENOSPC\n");
 282         for (i = 0; i < LLOG_BITMAP_SIZE(llh->lgh_hdr) + 1; i++) {
 283                 struct llog_rec_hdr     hdr;
 284                 char                    buf_even[24];
 285                 char                    buf_odd[32];
 286
 287                 memset(buf_odd, 0, sizeof buf_odd);
 288                 memset(buf_even, 0, sizeof buf_even);
 289                 if ((i % 2) == 0) {
 290                         hdr.lrh_len = 24;
 291                         hdr.lrh_type = OBD_CFG_REC;
 292                         rc = llog_write_rec(env, llh, &hdr, NULL, 0, buf_even,
 293                                             -1);
 294                 } else {
 295                         hdr.lrh_len = 32;
 296                         hdr.lrh_type = OBD_CFG_REC;
 297                         rc = llog_write_rec(env, llh, &hdr, NULL, 0, buf_odd,
 298                                             -1);
 299                 }
 300                 if (rc == -ENOSPC) {
 301                         break;
 302                 } else if (rc < 0) {
 303                         CERROR("3d: write recs failed at #%d: %d\n",
 304                                i + 1, rc);
 305                         RETURN(rc);
 306                 }
 307                 num_recs++;
 308         }
 309         if (rc != -ENOSPC) {
 310                 CWARN("3d: write record more than BITMAP size!\n");
 311                 RETURN(-EINVAL);
 312         }
 313         CWARN("3d: wrote %d more records before end of llog is reached\n",
 314               num_recs);
 315
 316         rc = verify_handle("3d", llh, num_recs);
 317
 318         RETURN(rc);
 319 }
 320
 321 /* Test catalogue additions */
 322 static int llog_test_4(const struct lu_env *env, struct obd_device *obd)
 323 {
 324         struct llog_handle      *cath;
 325         char                     name[10];
 326         int                      rc, rc2, i, buflen;
 327         struct llog_mini_rec     lmr;
 328         struct llog_cookie       cookie;
 329         struct llog_ctxt        *ctxt;
 330         int                      num_recs = 0;
 331         char                    *buf;
 332         struct llog_rec_hdr      rec;
 333
 334         ENTRY;
 335
 336         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 337         LASSERT(ctxt);
 338
 339         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 340         lmr.lmr_hdr.lrh_type = 0xf00f00;
 341
 342         sprintf(name, "%x", llog_test_rand + 1);
 343         CWARN("4a: create a catalog log with name: %s\n", name);
 344         rc = llog_open_create(env, ctxt, &cath, NULL, name);
 345         if (rc) {
 346                 CERROR("4a: llog_create with name %s failed: %d\n", name, rc);
 347                 GOTO(ctxt_release, rc);
 348         }
 349         llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);
 350         if (rc) {
 351                 CERROR("4a: can't init llog handle: %d\n", rc);
 352                 GOTO(out, rc);
 353         }
 354
 355         num_recs++;
 356         cat_logid = cath->lgh_id;
 357
 358         /* XXX: there is known issue with tests 4b, MGS is not able to add
 359          * anonymous plain llog, si we exit now to allow following tests run.
 360          * It is fixed in upcoming llog over OSD code */
 361         GOTO(out, rc);
 362
 363         CWARN("4b: write 1 record into the catalog\n");
 364         rc = llog_cat_add_rec(env, cath, &lmr.lmr_hdr, &cookie, NULL);
 365         if (rc != 1) {
 366                 CERROR("4b: write 1 catalog record failed at: %d\n", rc);
 367                 GOTO(out, rc);
 368         }
 369         num_recs++;
 370         rc = verify_handle("4b", cath, 2);
 371         if (rc)
 372                 GOTO(out, rc);
 373
 374         rc = verify_handle("4b", cath->u.chd.chd_current_log, num_recs);
 375         if (rc)
 376                 GOTO(out, rc);
 377
 378         CWARN("4c: cancel 1 log record\n");
 379         rc = llog_cat_cancel_records(env, cath, 1, &cookie);
 380         if (rc) {
 381                 CERROR("4c: cancel 1 catalog based record failed: %d\n", rc);
 382                 GOTO(out, rc);
 383         }
 384         num_recs--;
 385
 386         rc = verify_handle("4c", cath->u.chd.chd_current_log, num_recs);
 387         if (rc)
 388                 GOTO(out, rc);
 389
 390         CWARN("4d: write %d more log records\n", LLOG_TEST_RECNUM);
 391         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
 392                 rc = llog_cat_add_rec(env, cath, &lmr.lmr_hdr, NULL, NULL);
 393                 if (rc) {
 394                         CERROR("4d: write %d records failed at #%d: %d\n",
 395                                LLOG_TEST_RECNUM, i + 1, rc);
 396                         GOTO(out, rc);
 397                 }
 398                 num_recs++;
 399         }
 400
 401         /* make sure new plain llog appears */
 402         rc = verify_handle("4d", cath, 3);
 403         if (rc)
 404                 GOTO(out, rc);
 405
 406         CWARN("4e: add 5 large records, one record per block\n");
 407         buflen = LLOG_CHUNK_SIZE - sizeof(struct llog_rec_hdr) -
 408                  sizeof(struct llog_rec_tail);
 409         OBD_ALLOC(buf, buflen);
 410         if (buf == NULL)
 411                 GOTO(out, rc = -ENOMEM);
 412         for (i = 0; i < 5; i++) {
 413                 rec.lrh_len = buflen;
 414                 rec.lrh_type = OBD_CFG_REC;
 415                 rc = llog_cat_add_rec(env, cath, &rec, NULL, buf);
 416                 if (rc) {
 417                         CERROR("4e: write 5 records failed at #%d: %d\n",
 418                                i + 1, rc);
 419                         GOTO(out_free, rc);
 420                 }
 421                 num_recs++;
 422         }
 423 out_free:
 424         OBD_FREE(buf, buflen);
 425 out:
 426         CWARN("4f: put newly-created catalog\n");
 427         rc2 = llog_cat_close(env, cath);
 428         if (rc2) {
 429                 CERROR("4: close log %s failed: %d\n", name, rc2);
 430                 if (rc == 0)
 431                         rc = rc2;
 432         }
 433 ctxt_release:
 434         llog_ctxt_put(ctxt);
 435         RETURN(rc);
 436 }
 437
 438 static int cat_counter;
 439
 440 static int cat_print_cb(const struct lu_env *env, struct llog_handle *llh,
 441                         struct llog_rec_hdr *rec, void *data)
 442 {
 443         struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
 444
 445         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
 446                 CERROR("invalid record in catalog\n");
 447                 RETURN(-EINVAL);
 448         }
 449
 450         CWARN("seeing record at index %d - "LPX64":%x in log "LPX64"\n",
 451               rec->lrh_index, lir->lid_id.lgl_oid,
 452               lir->lid_id.lgl_ogen, llh->lgh_id.lgl_oid);
 453
 454         cat_counter++;
 455
 456         RETURN(0);
 457 }
 458
 459 static int plain_counter;
 460
 461 static int plain_print_cb(const struct lu_env *env, struct llog_handle *llh,
 462                           struct llog_rec_hdr *rec, void *data)
 463 {
 464         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 465                 CERROR("log is not plain\n");
 466                 RETURN(-EINVAL);
 467         }
 468
 469         CDEBUG(D_INFO, "seeing record at index %d in log "LPX64"\n",
 470               rec->lrh_index, llh->lgh_id.lgl_oid);
 471
 472         plain_counter++;
 473
 474         RETURN(0);
 475 }
 476
 477 static int cancel_count;
 478
 479 static int llog_cancel_rec_cb(const struct lu_env *env,
 480                               struct llog_handle *llh,
 481                               struct llog_rec_hdr *rec, void *data)
 482 {
 483         struct llog_cookie cookie;
 484
 485         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
 486                 CERROR("log is not plain\n");
 487                 RETURN(-EINVAL);
 488         }
 489
 490         cookie.lgc_lgl = llh->lgh_id;
 491         cookie.lgc_index = rec->lrh_index;
 492
 493         llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1, &cookie);
 494         cancel_count++;
 495         if (cancel_count == LLOG_TEST_RECNUM)
 496                 RETURN(-LLOG_EEMPTY);
 497         RETURN(0);
 498 }
 499
 500 /* Test log and catalogue processing */
 501 static int llog_test_5(const struct lu_env *env, struct obd_device *obd)
 502 {
 503         struct llog_handle      *llh = NULL;
 504         char                     name[10];
 505         int                      rc, rc2;
 506         struct llog_mini_rec     lmr;
 507         struct llog_ctxt        *ctxt;
 508
 509         ENTRY;
 510
 511         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 512         LASSERT(ctxt);
 513
 514         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
 515         lmr.lmr_hdr.lrh_type = 0xf00f00;
 516
 517         CWARN("5a: re-open catalog by id\n");
 518         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
 519         if (rc) {
 520                 CERROR("5a: llog_create with logid failed: %d\n", rc);
 521                 GOTO(out_put, rc);
 522         }
 523
 524         llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
 525         if (rc) {
 526                 CERROR("5a: can't init llog handle: %d\n", rc);
 527                 GOTO(out, rc);
 528         }
 529
 530         /* XXX: depends on tests 4 which is not working yet */
 531         GOTO(out, rc);
 532
 533         CWARN("5b: print the catalog entries.. we expect 2\n");
 534         cat_counter = 0;
 535         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 536         if (rc) {
 537                 CERROR("5b: process with cat_print_cb failed: %d\n", rc);
 538                 GOTO(out, rc);
 539         }
 540         if (cat_counter != 2) {
 541                 CERROR("5b: %d entries in catalog\n", cat_counter);
 542                 GOTO(out, rc = -EINVAL);
 543         }
 544
 545         CWARN("5c: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
 546         cancel_count = 0;
 547         rc = llog_cat_process(env, llh, llog_cancel_rec_cb, "foobar", 0, 0);
 548         if (rc != -LLOG_EEMPTY) {
 549                 CERROR("5c: process with cat_cancel_cb failed: %d\n", rc);
 550                 GOTO(out, rc);
 551         }
 552
 553         CWARN("5c: print the catalog entries.. we expect 1\n");
 554         cat_counter = 0;
 555         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
 556         if (rc) {
 557                 CERROR("5c: process with cat_print_cb failed: %d\n", rc);
 558                 GOTO(out, rc);
 559         }
 560         if (cat_counter != 1) {
 561                 CERROR("5c: %d entries in catalog\n", cat_counter);
 562                 GOTO(out, rc = -EINVAL);
 563         }
 564
 565         CWARN("5d: add 1 record to the log with many canceled empty pages\n");
 566         rc = llog_cat_add_rec(env, llh, &lmr.lmr_hdr, NULL, NULL);
 567         if (rc) {
 568                 CERROR("5d: add record to the log with many canceled empty "
 569                        "pages failed\n");
 570                 GOTO(out, rc);
 571         }
 572
 573         CWARN("5e: print plain log entries.. expect 6\n");
 574         plain_counter = 0;
 575         rc = llog_cat_process(env, llh, plain_print_cb, "foobar", 0, 0);
 576         if (rc) {
 577                 CERROR("5e: process with plain_print_cb failed: %d\n", rc);
 578                 GOTO(out, rc);
 579         }
 580         if (plain_counter != 6) {
 581                 CERROR("5e: found %d records\n", plain_counter);
 582                 GOTO(out, rc = -EINVAL);
 583         }
 584
 585         CWARN("5f: print plain log entries reversely.. expect 6\n");
 586         plain_counter = 0;
 587         rc = llog_cat_reverse_process(env, llh, plain_print_cb, "foobar");
 588         if (rc) {
 589                 CERROR("5f: reversely process with plain_print_cb failed:"
 590                        "%d\n", rc);
 591                 GOTO(out, rc);
 592         }
 593         if (plain_counter != 6) {
 594                 CERROR("5f: found %d records\n", plain_counter);
 595                 GOTO(out, rc = -EINVAL);
 596         }
 597
 598 out:
 599         CWARN("5g: close re-opened catalog\n");
 600         rc2 = llog_cat_close(env, llh);
 601         if (rc2) {
 602                 CERROR("5g: close log %s failed: %d\n", name, rc2);
 603                 if (rc == 0)
 604                         rc = rc2;
 605         }
 606 out_put:
 607         llog_ctxt_put(ctxt);
 608
 609         RETURN(rc);
 610 }
 611
 612 /* Test client api; open log by name and process */
 613 static int llog_test_6(const struct lu_env *env, struct obd_device *obd,
 614                        char *name)
 615 {
 616         struct obd_device       *mgc_obd;
 617         struct llog_ctxt        *ctxt;
 618         struct obd_uuid         *mgs_uuid;
 619         struct obd_export       *exp;
 620         struct obd_uuid          uuid = { "LLOG_TEST6_UUID" };
 621         struct llog_handle      *llh = NULL;
 622         struct llog_ctxt        *nctxt;
 623         int                      rc, rc2;
 624
 625         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 626         LASSERT(ctxt);
 627         mgs_uuid = &ctxt->loc_exp->exp_obd->obd_uuid;
 628
 629         CWARN("6a: re-open log %s using client API\n", name);
 630         mgc_obd = class_find_client_obd(mgs_uuid, LUSTRE_MGC_NAME, NULL);
 631         if (mgc_obd == NULL) {
 632                 CERROR("6a: no MGC devices connected to %s found.\n",
 633                        mgs_uuid->uuid);
 634                 GOTO(ctxt_release, rc = -ENOENT);
 635         }
 636
 637         rc = obd_connect(NULL, &exp, mgc_obd, &uuid,
 638                          NULL /* obd_connect_data */, NULL);
 639         if (rc != -EALREADY) {
 640                 CERROR("6a: connect on connected MDC (%s) failed to return"
 641                        " -EALREADY", mgc_obd->obd_name);
 642                 if (rc == 0)
 643                         obd_disconnect(exp);
 644                 GOTO(ctxt_release, rc = -EINVAL);
 645         }
 646
 647         nctxt = llog_get_context(mgc_obd, LLOG_CONFIG_REPL_CTXT);
 648         rc = llog_open(env, nctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
 649         if (rc) {
 650                 CERROR("6a: llog_open failed %d\n", rc);
 651                 GOTO(nctxt_put, rc);
 652         }
 653
 654         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
 655         if (rc) {
 656                 CERROR("6a: llog_init_handle failed %d\n", rc);
 657                 GOTO(parse_out, rc);
 658         }
 659
 660         plain_counter = 1; /* llog header is first record */
 661         CWARN("6b: process log %s using client API\n", name);
 662         rc = llog_process(env, llh, plain_print_cb, NULL, NULL);
 663         if (rc)
 664                 CERROR("6b: llog_process failed %d\n", rc);
 665         CWARN("6b: processed %d records\n", plain_counter);
 666
 667         rc = verify_handle("6b", llh, plain_counter);
 668         if (rc)
 669                 GOTO(parse_out, rc);
 670
 671         plain_counter = 1; /* llog header is first record */
 672         CWARN("6c: process log %s reversely using client API\n", name);
 673         rc = llog_reverse_process(env, llh, plain_print_cb, NULL, NULL);
 674         if (rc)
 675                 CERROR("6c: llog_reverse_process failed %d\n", rc);
 676         CWARN("6c: processed %d records\n", plain_counter);
 677
 678         rc = verify_handle("6c", llh, plain_counter);
 679         if (rc)
 680                 GOTO(parse_out, rc);
 681
 682 parse_out:
 683         rc2 = llog_close(env, llh);
 684         if (rc2) {
 685                 CERROR("6: llog_close failed: rc = %d\n", rc2);
 686                 if (rc == 0)
 687                         rc = rc2;
 688         }
 689 nctxt_put:
 690         llog_ctxt_put(nctxt);
 691 ctxt_release:
 692         llog_ctxt_put(ctxt);
 693         RETURN(rc);
 694 }
 695
 696 static union {
 697         struct llog_rec_hdr             lrh;   /* common header */
 698         struct llog_logid_rec           llr;   /* LLOG_LOGID_MAGIC */
 699         struct llog_unlink64_rec        lur;   /* MDS_UNLINK64_REC */
 700         struct llog_setattr64_rec       lsr64; /* MDS_SETATTR64_REC */
 701         struct llog_size_change_rec     lscr;  /* OST_SZ_REC */
 702         struct llog_changelog_rec       lcr;   /* CHANGELOG_REC */
 703         struct llog_changelog_user_rec  lcur;  /* CHANGELOG_USER_REC */
 704         struct llog_gen_rec             lgr;   /* LLOG_GEN_REC */
 705 } llog_records;
 706
 707 static int test_7_print_cb(const struct lu_env *env, struct llog_handle *llh,
 708                            struct llog_rec_hdr *rec, void *data)
 709 {
 710         struct lu_fid fid;
 711
 712         logid_to_fid(&llh->lgh_id, &fid);
 713
 714         CDEBUG(D_OTHER, "record type %#x at index %d in log "DFID"\n",
 715                rec->lrh_type, rec->lrh_index, PFID(&fid));
 716
 717         plain_counter++;
 718         return 0;
 719 }
 720
 721 static int test_7_cancel_cb(const struct lu_env *env, struct llog_handle *llh,
 722                             struct llog_rec_hdr *rec, void *data)
 723 {
 724         plain_counter++;
 725         /* test LLOG_DEL_RECORD is working */
 726         return LLOG_DEL_RECORD;
 727 }
 728
 729 static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
 730 {
 731         struct llog_handle      *llh;
 732         int                      rc = 0, i, process_count;
 733         int                      num_recs = 0;
 734
 735         ENTRY;
 736
 737         rc = llog_open_create(env, ctxt, &llh, NULL, "llt_test7");
 738         if (rc) {
 739                 CERROR("7_sub: create log failed\n");
 740                 RETURN(rc);
 741         }
 742
 743         rc = llog_init_handle(env, llh,
 744                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
 745                               &uuid);
 746         if (rc) {
 747                 CERROR("7_sub: can't init llog handle: %d\n", rc);
 748                 GOTO(out_close, rc);
 749         }
 750         for (i = 0; i < LLOG_BITMAP_SIZE(llh->lgh_hdr); i++) {
 751                 rc = llog_write_rec(env, llh, &llog_records.lrh, NULL, 0,
 752                                     NULL, -1);
 753                 if (rc == -ENOSPC) {
 754                         break;
 755                 } else if (rc < 0) {
 756                         CERROR("7_sub: write recs failed at #%d: %d\n",
 757                                i + 1, rc);
 758                         GOTO(out_close, rc);
 759                 }
 760                 num_recs++;
 761         }
 762         if (rc != -ENOSPC) {
 763                 CWARN("7_sub: write record more than BITMAP size!\n");
 764                 GOTO(out_close, rc = -EINVAL);
 765         }
 766
 767         rc = verify_handle("7_sub", llh, num_recs + 1);
 768         if (rc) {
 769                 CERROR("7_sub: verify handle failed: %d\n", rc);
 770                 GOTO(out_close, rc);
 771         }
 772         if (num_recs < LLOG_BITMAP_SIZE(llh->lgh_hdr) - 1)
 773                 CWARN("7_sub: records are not aligned, written %d from %u\n",
 774                       num_recs, LLOG_BITMAP_SIZE(llh->lgh_hdr) - 1);
 775
 776         plain_counter = 0;
 777         rc = llog_process(env, llh, test_7_print_cb, "test 7", NULL);
 778         if (rc) {
 779                 CERROR("7_sub: llog process failed: %d\n", rc);
 780                 GOTO(out_close, rc);
 781         }
 782         process_count = plain_counter;
 783         if (process_count != num_recs) {
 784                 CERROR("7_sub: processed %d records from %d total\n",
 785                        process_count, num_recs);
 786                 GOTO(out_close, rc = -EINVAL);
 787         }
 788
 789         plain_counter = 0;
 790         rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);
 791         if (rc) {
 792                 CERROR("7_sub: reverse llog process failed: %d\n", rc);
 793                 GOTO(out_close, rc);
 794         }
 795         if (process_count != plain_counter) {
 796                 CERROR("7_sub: Reverse/direct processing found different"
 797                        "number of records: %d/%d\n",
 798                        plain_counter, process_count);
 799                 GOTO(out_close, rc = -EINVAL);
 800         }
 801         if (llog_exist(llh)) {
 802                 CERROR("7_sub: llog exists but should be zapped\n");
 803                 GOTO(out_close, rc = -EEXIST);
 804         }
 805
 806         rc = verify_handle("7_sub", llh, 1);
 807 out_close:
 808         if (rc)
 809                 llog_destroy(env, llh);
 810         llog_close(env, llh);
 811         RETURN(rc);
 812 }
 813
 814 /* Test all llog records writing and processing */
 815 static int llog_test_7(const struct lu_env *env, struct obd_device *obd)
 816 {
 817         struct llog_ctxt        *ctxt;
 818         int                      rc;
 819
 820         ENTRY;
 821
 822         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 823
 824         CWARN("7a: test llog_logid_rec\n");
 825         llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);
 826         llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);
 827         llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
 828
 829         rc = llog_test_7_sub(env, ctxt);
 830         if (rc) {
 831                 CERROR("7a: llog_logid_rec test failed\n");
 832                 GOTO(out, rc);
 833         }
 834
 835         CWARN("7b: test llog_unlink64_rec\n");
 836         llog_records.lur.lur_hdr.lrh_len = sizeof(llog_records.lur);
 837         llog_records.lur.lur_tail.lrt_len = sizeof(llog_records.lur);
 838         llog_records.lur.lur_hdr.lrh_type = MDS_UNLINK64_REC;
 839
 840         rc = llog_test_7_sub(env, ctxt);
 841         if (rc) {
 842                 CERROR("7b: llog_unlink_rec test failed\n");
 843                 GOTO(out, rc);
 844         }
 845
 846         CWARN("7c: test llog_setattr64_rec\n");
 847         llog_records.lsr64.lsr_hdr.lrh_len = sizeof(llog_records.lsr64);
 848         llog_records.lsr64.lsr_tail.lrt_len = sizeof(llog_records.lsr64);
 849         llog_records.lsr64.lsr_hdr.lrh_type = MDS_SETATTR64_REC;
 850
 851         rc = llog_test_7_sub(env, ctxt);
 852         if (rc) {
 853                 CERROR("7c: llog_setattr64_rec test failed\n");
 854                 GOTO(out, rc);
 855         }
 856
 857         CWARN("7d: test llog_size_change_rec\n");
 858         llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);
 859         llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);
 860         llog_records.lscr.lsc_hdr.lrh_type = OST_SZ_REC;
 861
 862         rc = llog_test_7_sub(env, ctxt);
 863         if (rc) {
 864                 CERROR("7d: llog_size_change_rec test failed\n");
 865                 GOTO(out, rc);
 866         }
 867
 868         CWARN("7e: test llog_changelog_rec\n");
 869         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
 870         llog_records.lcr.cr_tail.lrt_len = sizeof(llog_records.lcr);
 871         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
 872
 873         rc = llog_test_7_sub(env, ctxt);
 874         if (rc) {
 875                 CERROR("7e: llog_changelog_rec test failed\n");
 876                 GOTO(out, rc);
 877         }
 878
 879         CWARN("7f: test llog_changelog_user_rec\n");
 880         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
 881         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
 882         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
 883
 884         rc = llog_test_7_sub(env, ctxt);
 885         if (rc) {
 886                 CERROR("7f: llog_changelog_user_rec test failed\n");
 887                 GOTO(out, rc);
 888         }
 889
 890         CWARN("7g: test llog_gen_rec\n");
 891         llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);
 892         llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);
 893         llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
 894
 895         rc = llog_test_7_sub(env, ctxt);
 896         if (rc) {
 897                 CERROR("7g: llog_size_change_rec test failed\n");
 898                 GOTO(out, rc);
 899         }
 900 out:
 901         llog_ctxt_put(ctxt);
 902         RETURN(rc);
 903 }
 904
 905 /* -------------------------------------------------------------------------
 906  * Tests above, boring obd functions below
 907  * ------------------------------------------------------------------------- */
 908 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
 909 {
 910         struct llog_handle      *llh = NULL;
 911         struct lvfs_run_ctxt     saved;
 912         struct llog_ctxt        *ctxt;
 913         int                      rc, err;
 914         char                     name[10];
 915
 916         ENTRY;
 917         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
 918         LASSERT(ctxt);
 919
 920         sprintf(name, "%x", llog_test_rand);
 921         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 922
 923         rc = llog_test_1(env, obd, name);
 924         if (rc)
 925                 GOTO(cleanup_ctxt, rc);
 926
 927         rc = llog_test_2(env, obd, name, &llh);
 928         if (rc)
 929                 GOTO(cleanup_ctxt, rc);
 930
 931         rc = llog_test_3(env, obd, llh);
 932         if (rc)
 933                 GOTO(cleanup, rc);
 934
 935         rc = llog_test_4(env, obd);
 936         if (rc)
 937                 GOTO(cleanup, rc);
 938
 939         rc = llog_test_5(env, obd);
 940         if (rc)
 941                 GOTO(cleanup, rc);
 942
 943         rc = llog_test_6(env, obd, name);
 944         if (rc)
 945                 GOTO(cleanup, rc);
 946
 947         rc = llog_test_7(env, obd);
 948         if (rc)
 949                 GOTO(cleanup, rc);
 950
 951 cleanup:
 952         err = llog_destroy(env, llh);
 953         if (err)
 954                 CERROR("cleanup: llog_destroy failed: %d\n", err);
 955         llog_close(env, llh);
 956         if (rc == 0)
 957                 rc = err;
 958 cleanup_ctxt:
 959         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 960         llog_ctxt_put(ctxt);
 961         return rc;
 962 }
 963
 964 #ifdef LPROCFS
 965 static struct lprocfs_vars lprocfs_llog_test_obd_vars[] = { {0} };
 966 static struct lprocfs_vars lprocfs_llog_test_module_vars[] = { {0} };
 967 static void lprocfs_llog_test_init_vars(struct lprocfs_static_vars *lvars)
 968 {
 969     lvars->module_vars  = lprocfs_llog_test_module_vars;
 970     lvars->obd_vars     = lprocfs_llog_test_obd_vars;
 971 }
 972 #endif
 973
 974 static int llog_test_cleanup(struct obd_device *obd)
 975 {
 976         int rc;
 977
 978         rc = llog_cleanup(llog_get_context(obd, LLOG_TEST_ORIG_CTXT));
 979         if (rc)
 980                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
 981         return rc;
 982 }
 983
 984 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
 985 {
 986         struct obd_device *tgt;
 987         struct lu_env env;
 988         int rc;
 989
 990         ENTRY;
 991
 992         if (lcfg->lcfg_bufcount < 2) {
 993                 CERROR("requires a TARGET OBD name\n");
 994                 RETURN(-EINVAL);
 995         }
 996
 997         if (lcfg->lcfg_buflens[1] < 1) {
 998                 CERROR("requires a TARGET OBD name\n");
 999                 RETURN(-EINVAL);
1000         }
1001
1002         /* disk obd */
1003         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
1004         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
1005                 CERROR("target device not attached or not set up (%s)\n",
1006                        lustre_cfg_string(lcfg, 1));
1007                 RETURN(-EINVAL);
1008         }
1009
1010         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1011         if (rc)
1012                 RETURN(rc);
1013
1014         CWARN("Setup llog-test device over %s device\n",
1015               lustre_cfg_string(lcfg, 1));
1016
1017         rc = llog_setup(obd, &obd->obd_olg, LLOG_TEST_ORIG_CTXT, tgt, 0, NULL,
1018                         &llog_lvfs_ops);
1019         if (rc)
1020                 GOTO(cleanup_env, rc);
1021
1022         llog_test_rand = cfs_rand();
1023
1024         rc = llog_run_tests(&env, obd);
1025         if (rc)
1026                 llog_test_cleanup(obd);
1027 cleanup_env:
1028         lu_env_fini(&env);
1029         RETURN(rc);
1030 }
1031
1032 static struct obd_ops llog_obd_ops = {
1033         .o_owner       = THIS_MODULE,
1034         .o_setup       = llog_test_setup,
1035         .o_cleanup     = llog_test_cleanup,
1036 };
1037
1038 static int __init llog_test_init(void)
1039 {
1040         struct lprocfs_static_vars lvars;
1041
1042         lprocfs_llog_test_init_vars(&lvars);
1043         return class_register_type(&llog_obd_ops, NULL,
1044                                    lvars.module_vars, "llog_test", NULL);
1045 }
1046
1047 static void __exit llog_test_exit(void)
1048 {
1049         class_unregister_type("llog_test");
1050 }
1051
1052 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1053 MODULE_DESCRIPTION("llog test module");
1054 MODULE_LICENSE("GPL");
1055
1056 module_init(llog_test_init);
1057 module_exit(llog_test_exit);