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