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