Whamcloud - gitweb
LU-4528 llog: don't write llog in 3-steps
[fs/lustre-release.git] / lustre / obdclass / llog_test.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2012, 2013, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/obdclass/llog_test.c
37  *
38  * Author: Phil Schwan <phil@clusterfs.com>
39  * Author: Mikhail Pershin <mike.pershin@intel.com>
40  */
41
42 #define DEBUG_SUBSYSTEM S_CLASS
43
44 #include <linux/module.h>
45 #include <linux/init.h>
46
47 #include <obd_class.h>
48 #include <lustre_fid.h>
49 #include <lustre_log.h>
50
51 /* This is slightly more than the number of records that can fit into a
52  * single llog file, because the llog_log_header takes up some of the
53  * space in the first block that cannot be used for the bitmap. */
54 #define LLOG_TEST_RECNUM  (LLOG_CHUNK_SIZE * 8)
55
56 static int llog_test_rand;
57 static struct obd_uuid uuid = { .uuid = "test_uuid" };
58 static struct llog_logid cat_logid;
59
60 struct llog_mini_rec {
61         struct llog_rec_hdr     lmr_hdr;
62         struct llog_rec_tail    lmr_tail;
63 } __attribute__((packed));
64
65 static int verify_handle(char *test, struct llog_handle *llh, int num_recs)
66 {
67         int i;
68         int last_idx = 0;
69         int active_recs = 0;
70
71         for (i = 0; i < LLOG_BITMAP_BYTES * 8; i++) {
72                 if (ext2_test_bit(i, llh->lgh_hdr->llh_bitmap)) {
73                         last_idx = i;
74                         active_recs++;
75                 }
76         }
77
78         if (active_recs != num_recs) {
79                 CERROR("%s: expected %d active recs after write, found %d\n",
80                        test, num_recs, active_recs);
81                 RETURN(-ERANGE);
82         }
83
84         if (llh->lgh_hdr->llh_count != num_recs) {
85                 CERROR("%s: handle->count is %d, expected %d after write\n",
86                        test, llh->lgh_hdr->llh_count, num_recs);
87                 RETURN(-ERANGE);
88         }
89
90         if (llh->lgh_last_idx < last_idx) {
91                 CERROR("%s: handle->last_idx is %d, expected %d after write\n",
92                        test, llh->lgh_last_idx, last_idx);
93                 RETURN(-ERANGE);
94         }
95
96         RETURN(0);
97 }
98
99 /* Test named-log create/open, close */
100 static int llog_test_1(const struct lu_env *env,
101                        struct obd_device *obd, char *name)
102 {
103         struct llog_handle      *llh;
104         struct llog_ctxt        *ctxt;
105         int rc;
106         int rc2;
107
108         ENTRY;
109
110         CWARN("1a: create a log with name: %s\n", name);
111         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
112         LASSERT(ctxt);
113
114         rc = llog_open_create(env, ctxt, &llh, NULL, name);
115         if (rc) {
116                 CERROR("1a: llog_create with name %s failed: %d\n", name, rc);
117                 GOTO(out, rc);
118         }
119         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, &uuid);
120         if (rc) {
121                 CERROR("1a: can't init llog handle: %d\n", rc);
122                 GOTO(out_close, rc);
123         }
124
125         rc = verify_handle("1", llh, 1);
126
127         CWARN("1b: close newly-created log\n");
128 out_close:
129         rc2 = llog_close(env, llh);
130         if (rc2) {
131                 CERROR("1b: close log %s failed: %d\n", name, rc2);
132                 if (rc == 0)
133                         rc = rc2;
134         }
135 out:
136         llog_ctxt_put(ctxt);
137         RETURN(rc);
138 }
139
140 /* Test named-log reopen; returns opened log on success */
141 static int llog_test_2(const struct lu_env *env, struct obd_device *obd,
142                        char *name, struct llog_handle **llh)
143 {
144         struct llog_ctxt        *ctxt;
145         struct llog_handle      *loghandle;
146         struct llog_logid        logid;
147         int                      rc;
148
149         ENTRY;
150
151         CWARN("2a: re-open a log with name: %s\n", name);
152         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
153         LASSERT(ctxt);
154
155         rc = llog_open(env, ctxt, llh, NULL, name, LLOG_OPEN_EXISTS);
156         if (rc) {
157                 CERROR("2a: re-open log with name %s failed: %d\n", name, rc);
158                 GOTO(out_put, rc);
159         }
160
161         rc = llog_init_handle(env, *llh, LLOG_F_IS_PLAIN, &uuid);
162         if (rc) {
163                 CERROR("2a: can't init llog handle: %d\n", rc);
164                 GOTO(out_close_llh, rc);
165         }
166
167         rc = verify_handle("2", *llh, 1);
168         if (rc)
169                 GOTO(out_close_llh, rc);
170
171         /* XXX: there is known issue with tests 2b, MGS is not able to create
172          * anonymous llog, exit now to allow following tests run.
173          * It is fixed in upcoming llog over OSD code */
174         GOTO(out_put, rc);
175
176         CWARN("2b: create a log without specified NAME & LOGID\n");
177         rc = llog_open_create(env, ctxt, &loghandle, NULL, NULL);
178         if (rc) {
179                 CERROR("2b: create log failed\n");
180                 GOTO(out_close_llh, rc);
181         }
182         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, &uuid);
183         if (rc) {
184                 CERROR("2b: can't init llog handle: %d\n", rc);
185                 GOTO(out_close, rc);
186         }
187
188         logid = loghandle->lgh_id;
189         llog_close(env, loghandle);
190
191         CWARN("2c: re-open the log by LOGID\n");
192         rc = llog_open(env, ctxt, &loghandle, &logid, NULL, LLOG_OPEN_EXISTS);
193         if (rc) {
194                 CERROR("2c: re-open log by LOGID failed\n");
195                 GOTO(out_close_llh, rc);
196         }
197
198         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, &uuid);
199         if (rc) {
200                 CERROR("2c: can't init llog handle: %d\n", rc);
201                 GOTO(out_close, rc);
202         }
203
204         CWARN("2b: destroy this log\n");
205         rc = llog_destroy(env, loghandle);
206         if (rc)
207                 CERROR("2d: destroy log failed\n");
208 out_close:
209         llog_close(env, loghandle);
210 out_close_llh:
211         if (rc)
212                 llog_close(env, *llh);
213 out_put:
214         llog_ctxt_put(ctxt);
215
216         RETURN(rc);
217 }
218
219 /* Test record writing, single and in bulk */
220 static int llog_test_3(const struct lu_env *env, struct obd_device *obd,
221                        struct llog_handle *llh)
222 {
223         struct llog_gen_rec      lgr;
224         int                      rc, i;
225         int                      num_recs = 1; /* 1 for the header */
226
227         ENTRY;
228
229         lgr.lgr_hdr.lrh_len = lgr.lgr_tail.lrt_len = sizeof(lgr);
230         lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
231
232         CWARN("3a: write one create_rec\n");
233         rc = llog_write(env, llh,  &lgr.lgr_hdr, LLOG_NEXT_IDX);
234         num_recs++;
235         if (rc < 0) {
236                 CERROR("3a: write one log record failed: %d\n", rc);
237                 RETURN(rc);
238         }
239
240         rc = verify_handle("3a", llh, num_recs);
241         if (rc)
242                 RETURN(rc);
243
244         CWARN("3c: write 1000 more log records\n");
245         for (i = 0; i < 1000; i++) {
246                 rc = llog_write(env, llh, &lgr.lgr_hdr, LLOG_NEXT_IDX);
247                 if (rc < 0) {
248                         CERROR("3c: write 1000 records failed at #%d: %d\n",
249                                i + 1, rc);
250                         RETURN(rc);
251                 }
252                 num_recs++;
253         }
254
255         rc = verify_handle("3c", llh, num_recs);
256         if (rc)
257                 RETURN(rc);
258
259         CWARN("3d: write records with variable size until BITMAP_SIZE, "
260               "return -ENOSPC\n");
261         for (i = 0; i < LLOG_BITMAP_SIZE(llh->lgh_hdr) + 1; i++) {
262                 char                     buf[64];
263                 struct llog_rec_hdr     *hdr = (void *)&buf;
264
265                 memset(buf, 0, sizeof buf);
266                 if ((i % 2) == 0)
267                         hdr->lrh_len = 40;
268                 else
269                         hdr->lrh_len = 64;
270                 hdr->lrh_type = OBD_CFG_REC;
271                 rc = llog_write(env, llh, hdr, LLOG_NEXT_IDX);
272
273                 if (rc == -ENOSPC) {
274                         break;
275                 } else if (rc < 0) {
276                         CERROR("3d: write recs failed at #%d: %d\n",
277                                i + 1, rc);
278                         RETURN(rc);
279                 }
280                 num_recs++;
281         }
282         if (rc != -ENOSPC) {
283                 CWARN("3d: write record more than BITMAP size!\n");
284                 RETURN(-EINVAL);
285         }
286         CWARN("3d: wrote %d more records before end of llog is reached\n",
287               num_recs);
288
289         rc = verify_handle("3d", llh, num_recs);
290
291         RETURN(rc);
292 }
293
294 /* Test catalogue additions */
295 static int llog_test_4(const struct lu_env *env, struct obd_device *obd)
296 {
297         struct llog_handle      *cath;
298         char                     name[10];
299         int                      rc, rc2, i, buflen;
300         struct llog_mini_rec     lmr;
301         struct llog_cookie       cookie;
302         struct llog_ctxt        *ctxt;
303         int                      num_recs = 0;
304         char                    *buf;
305         struct llog_rec_hdr     *rec;
306
307         ENTRY;
308
309         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
310         LASSERT(ctxt);
311
312         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
313         lmr.lmr_hdr.lrh_type = 0xf00f00;
314
315         sprintf(name, "%x", llog_test_rand + 1);
316         CWARN("4a: create a catalog log with name: %s\n", name);
317         rc = llog_open_create(env, ctxt, &cath, NULL, name);
318         if (rc) {
319                 CERROR("4a: llog_create with name %s failed: %d\n", name, rc);
320                 GOTO(ctxt_release, rc);
321         }
322         rc = llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);
323         if (rc) {
324                 CERROR("4a: can't init llog handle: %d\n", rc);
325                 GOTO(out, rc);
326         }
327
328         num_recs++;
329         cat_logid = cath->lgh_id;
330
331         CWARN("4b: write 1 record into the catalog\n");
332         rc = llog_cat_add(env, cath, &lmr.lmr_hdr, &cookie);
333         if (rc != 1) {
334                 CERROR("4b: write 1 catalog record failed at: %d\n", rc);
335                 GOTO(out, rc);
336         }
337         num_recs++;
338         rc = verify_handle("4b", cath, 2);
339         if (rc)
340                 GOTO(out, rc);
341
342         rc = verify_handle("4b", cath->u.chd.chd_current_log, num_recs);
343         if (rc)
344                 GOTO(out, rc);
345
346         CWARN("4c: cancel 1 log record\n");
347         rc = llog_cat_cancel_records(env, cath, 1, &cookie);
348         if (rc) {
349                 CERROR("4c: cancel 1 catalog based record failed: %d\n", rc);
350                 GOTO(out, rc);
351         }
352         num_recs--;
353
354         rc = verify_handle("4c", cath->u.chd.chd_current_log, num_recs);
355         if (rc)
356                 GOTO(out, rc);
357
358         CWARN("4d: write %d more log records\n", LLOG_TEST_RECNUM);
359         for (i = 0; i < LLOG_TEST_RECNUM; i++) {
360                 rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL);
361                 if (rc) {
362                         CERROR("4d: write %d records failed at #%d: %d\n",
363                                LLOG_TEST_RECNUM, i + 1, rc);
364                         GOTO(out, rc);
365                 }
366                 num_recs++;
367         }
368
369         /* make sure new plain llog appears */
370         rc = verify_handle("4d", cath, 3);
371         if (rc)
372                 GOTO(out, rc);
373
374         CWARN("4e: add 5 large records, one record per block\n");
375         buflen = LLOG_CHUNK_SIZE;
376         OBD_ALLOC(buf, buflen);
377         if (buf == NULL)
378                 GOTO(out, rc = -ENOMEM);
379         for (i = 0; i < 5; i++) {
380                 rec = (void *)buf;
381                 rec->lrh_len = buflen;
382                 rec->lrh_type = OBD_CFG_REC;
383                 rc = llog_cat_add(env, cath, rec, NULL);
384                 if (rc) {
385                         CERROR("4e: write 5 records failed at #%d: %d\n",
386                                i + 1, rc);
387                         GOTO(out_free, rc);
388                 }
389                 num_recs++;
390         }
391 out_free:
392         OBD_FREE(buf, buflen);
393 out:
394         CWARN("4f: put newly-created catalog\n");
395         rc2 = llog_cat_close(env, cath);
396         if (rc2) {
397                 CERROR("4: close log %s failed: %d\n", name, rc2);
398                 if (rc == 0)
399                         rc = rc2;
400         }
401 ctxt_release:
402         llog_ctxt_put(ctxt);
403         RETURN(rc);
404 }
405
406 static int cat_counter;
407
408 static int cat_print_cb(const struct lu_env *env, struct llog_handle *llh,
409                         struct llog_rec_hdr *rec, void *data)
410 {
411         struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
412         struct lu_fid            fid = {0};
413
414         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
415                 CERROR("invalid record in catalog\n");
416                 RETURN(-EINVAL);
417         }
418
419         logid_to_fid(&lir->lid_id, &fid);
420
421         CWARN("seeing record at index %d - "DFID" in log "DFID"\n",
422               rec->lrh_index, PFID(&fid),
423               PFID(lu_object_fid(&llh->lgh_obj->do_lu)));
424
425         cat_counter++;
426
427         RETURN(0);
428 }
429
430 static int plain_counter;
431
432 static int plain_print_cb(const struct lu_env *env, struct llog_handle *llh,
433                           struct llog_rec_hdr *rec, void *data)
434 {
435         struct lu_fid fid = {0};
436
437         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
438                 CERROR("log is not plain\n");
439                 RETURN(-EINVAL);
440         }
441
442         logid_to_fid(&llh->lgh_id, &fid);
443
444         CDEBUG(D_INFO, "seeing record at index %d in log "DFID"\n",
445                rec->lrh_index, PFID(&fid));
446
447         plain_counter++;
448
449         RETURN(0);
450 }
451
452 static int cancel_count;
453
454 static int llog_cancel_rec_cb(const struct lu_env *env,
455                               struct llog_handle *llh,
456                               struct llog_rec_hdr *rec, void *data)
457 {
458         struct llog_cookie cookie;
459
460         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
461                 CERROR("log is not plain\n");
462                 RETURN(-EINVAL);
463         }
464
465         cookie.lgc_lgl = llh->lgh_id;
466         cookie.lgc_index = rec->lrh_index;
467
468         llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1, &cookie);
469         cancel_count++;
470         if (cancel_count == LLOG_TEST_RECNUM)
471                 RETURN(-LLOG_EEMPTY);
472         RETURN(0);
473 }
474
475 /* Test log and catalogue processing */
476 static int llog_test_5(const struct lu_env *env, struct obd_device *obd)
477 {
478         struct llog_handle      *llh = NULL;
479         char                     name[10];
480         int                      rc, rc2;
481         struct llog_mini_rec     lmr;
482         struct llog_ctxt        *ctxt;
483
484         ENTRY;
485
486         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
487         LASSERT(ctxt);
488
489         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
490         lmr.lmr_hdr.lrh_type = 0xf00f00;
491
492         CWARN("5a: re-open catalog by id\n");
493         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
494         if (rc) {
495                 CERROR("5a: llog_create with logid failed: %d\n", rc);
496                 GOTO(out_put, rc);
497         }
498
499         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
500         if (rc) {
501                 CERROR("5a: can't init llog handle: %d\n", rc);
502                 GOTO(out, rc);
503         }
504
505         CWARN("5b: print the catalog entries.. we expect 2\n");
506         cat_counter = 0;
507         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
508         if (rc) {
509                 CERROR("5b: process with cat_print_cb failed: %d\n", rc);
510                 GOTO(out, rc);
511         }
512         if (cat_counter != 2) {
513                 CERROR("5b: %d entries in catalog\n", cat_counter);
514                 GOTO(out, rc = -EINVAL);
515         }
516
517         CWARN("5c: Cancel %d records, see one log zapped\n", LLOG_TEST_RECNUM);
518         cancel_count = 0;
519         rc = llog_cat_process(env, llh, llog_cancel_rec_cb, "foobar", 0, 0);
520         if (rc != -LLOG_EEMPTY) {
521                 CERROR("5c: process with cat_cancel_cb failed: %d\n", rc);
522                 GOTO(out, rc);
523         }
524
525         CWARN("5c: print the catalog entries.. we expect 1\n");
526         cat_counter = 0;
527         rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);
528         if (rc) {
529                 CERROR("5c: process with cat_print_cb failed: %d\n", rc);
530                 GOTO(out, rc);
531         }
532         if (cat_counter != 1) {
533                 CERROR("5c: %d entries in catalog\n", cat_counter);
534                 GOTO(out, rc = -EINVAL);
535         }
536
537         CWARN("5d: add 1 record to the log with many canceled empty pages\n");
538         rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
539         if (rc) {
540                 CERROR("5d: add record to the log with many canceled empty "
541                        "pages failed\n");
542                 GOTO(out, rc);
543         }
544
545         CWARN("5e: print plain log entries.. expect 6\n");
546         plain_counter = 0;
547         rc = llog_cat_process(env, llh, plain_print_cb, "foobar", 0, 0);
548         if (rc) {
549                 CERROR("5e: process with plain_print_cb failed: %d\n", rc);
550                 GOTO(out, rc);
551         }
552         if (plain_counter != 6) {
553                 CERROR("5e: found %d records\n", plain_counter);
554                 GOTO(out, rc = -EINVAL);
555         }
556
557         CWARN("5f: print plain log entries reversely.. expect 6\n");
558         plain_counter = 0;
559         rc = llog_cat_reverse_process(env, llh, plain_print_cb, "foobar");
560         if (rc) {
561                 CERROR("5f: reversely process with plain_print_cb failed:"
562                        "%d\n", rc);
563                 GOTO(out, rc);
564         }
565         if (plain_counter != 6) {
566                 CERROR("5f: found %d records\n", plain_counter);
567                 GOTO(out, rc = -EINVAL);
568         }
569
570 out:
571         CWARN("5g: close re-opened catalog\n");
572         rc2 = llog_cat_close(env, llh);
573         if (rc2) {
574                 CERROR("5g: close log %s failed: %d\n", name, rc2);
575                 if (rc == 0)
576                         rc = rc2;
577         }
578 out_put:
579         llog_ctxt_put(ctxt);
580
581         RETURN(rc);
582 }
583
584 /* Test client api; open log by name and process */
585 static int llog_test_6(const struct lu_env *env, struct obd_device *obd,
586                        char *name)
587 {
588         struct obd_device       *mgc_obd;
589         struct llog_ctxt        *ctxt;
590         struct obd_uuid         *mgs_uuid;
591         struct obd_export       *exp;
592         struct obd_uuid          uuid = { "LLOG_TEST6_UUID" };
593         struct llog_handle      *llh = NULL;
594         struct llog_ctxt        *nctxt;
595         int                      rc, rc2;
596
597         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
598         LASSERT(ctxt);
599         mgs_uuid = &ctxt->loc_exp->exp_obd->obd_uuid;
600
601         CWARN("6a: re-open log %s using client API\n", name);
602         mgc_obd = class_find_client_obd(mgs_uuid, LUSTRE_MGC_NAME, NULL);
603         if (mgc_obd == NULL) {
604                 CERROR("6a: no MGC devices connected to %s found.\n",
605                        mgs_uuid->uuid);
606                 GOTO(ctxt_release, rc = -ENOENT);
607         }
608
609         rc = obd_connect(NULL, &exp, mgc_obd, &uuid,
610                          NULL /* obd_connect_data */, NULL);
611         if (rc != -EALREADY) {
612                 CERROR("6a: connect on connected MGC (%s) failed to return"
613                        " -EALREADY", mgc_obd->obd_name);
614                 if (rc == 0)
615                         obd_disconnect(exp);
616                 GOTO(ctxt_release, rc = -EINVAL);
617         }
618
619         nctxt = llog_get_context(mgc_obd, LLOG_CONFIG_REPL_CTXT);
620         rc = llog_open(env, nctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);
621         if (rc) {
622                 CERROR("6a: llog_open failed %d\n", rc);
623                 GOTO(nctxt_put, rc);
624         }
625
626         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
627         if (rc) {
628                 CERROR("6a: llog_init_handle failed %d\n", rc);
629                 GOTO(parse_out, rc);
630         }
631
632         plain_counter = 1; /* llog header is first record */
633         CWARN("6b: process log %s using client API\n", name);
634         rc = llog_process(env, llh, plain_print_cb, NULL, NULL);
635         if (rc)
636                 CERROR("6b: llog_process failed %d\n", rc);
637         CWARN("6b: processed %d records\n", plain_counter);
638
639         rc = verify_handle("6b", llh, plain_counter);
640         if (rc)
641                 GOTO(parse_out, rc);
642
643         plain_counter = 1; /* llog header is first record */
644         CWARN("6c: process log %s reversely using client API\n", name);
645         rc = llog_reverse_process(env, llh, plain_print_cb, NULL, NULL);
646         if (rc)
647                 CERROR("6c: llog_reverse_process failed %d\n", rc);
648         CWARN("6c: processed %d records\n", plain_counter);
649
650         rc = verify_handle("6c", llh, plain_counter);
651         if (rc)
652                 GOTO(parse_out, rc);
653
654 parse_out:
655         rc2 = llog_close(env, llh);
656         if (rc2) {
657                 CERROR("6: llog_close failed: rc = %d\n", rc2);
658                 if (rc == 0)
659                         rc = rc2;
660         }
661 nctxt_put:
662         llog_ctxt_put(nctxt);
663 ctxt_release:
664         llog_ctxt_put(ctxt);
665         RETURN(rc);
666 }
667
668 static union {
669         struct llog_rec_hdr             lrh;   /* common header */
670         struct llog_logid_rec           llr;   /* LLOG_LOGID_MAGIC */
671         struct llog_unlink64_rec        lur;   /* MDS_UNLINK64_REC */
672         struct llog_setattr64_rec       lsr64; /* MDS_SETATTR64_REC */
673         struct llog_size_change_rec     lscr;  /* OST_SZ_REC */
674         struct llog_changelog_rec       lcr;   /* CHANGELOG_REC */
675         struct llog_changelog_user_rec  lcur;  /* CHANGELOG_USER_REC */
676         struct llog_gen_rec             lgr;   /* LLOG_GEN_REC */
677 } llog_records;
678
679 static int test_7_print_cb(const struct lu_env *env, struct llog_handle *llh,
680                            struct llog_rec_hdr *rec, void *data)
681 {
682         struct lu_fid fid = {0};
683
684         logid_to_fid(&llh->lgh_id, &fid);
685
686         CDEBUG(D_OTHER, "record type %#x at index %d in log "DFID"\n",
687                rec->lrh_type, rec->lrh_index, PFID(&fid));
688
689         plain_counter++;
690         return 0;
691 }
692
693 static int test_7_cancel_cb(const struct lu_env *env, struct llog_handle *llh,
694                             struct llog_rec_hdr *rec, void *data)
695 {
696         plain_counter++;
697         /* test LLOG_DEL_RECORD is working */
698         return LLOG_DEL_RECORD;
699 }
700
701 static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt)
702 {
703         struct llog_handle      *llh;
704         int                      rc = 0, i, process_count;
705         int                      num_recs = 0;
706
707         ENTRY;
708
709         rc = llog_open_create(env, ctxt, &llh, NULL, NULL);
710         if (rc) {
711                 CERROR("7_sub: create log failed\n");
712                 RETURN(rc);
713         }
714
715         rc = llog_init_handle(env, llh,
716                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
717                               &uuid);
718         if (rc) {
719                 CERROR("7_sub: can't init llog handle: %d\n", rc);
720                 GOTO(out_close, rc);
721         }
722         for (i = 0; i < LLOG_BITMAP_SIZE(llh->lgh_hdr); i++) {
723                 rc = llog_write(env, llh, &llog_records.lrh, LLOG_NEXT_IDX);
724                 if (rc == -ENOSPC) {
725                         break;
726                 } else if (rc < 0) {
727                         CERROR("7_sub: write recs failed at #%d: %d\n",
728                                i + 1, rc);
729                         GOTO(out_close, rc);
730                 }
731                 num_recs++;
732         }
733         if (rc != -ENOSPC) {
734                 CWARN("7_sub: write record more than BITMAP size!\n");
735                 GOTO(out_close, rc = -EINVAL);
736         }
737
738         rc = verify_handle("7_sub", llh, num_recs + 1);
739         if (rc) {
740                 CERROR("7_sub: verify handle failed: %d\n", rc);
741                 GOTO(out_close, rc);
742         }
743         if (num_recs < LLOG_BITMAP_SIZE(llh->lgh_hdr) - 1)
744                 CWARN("7_sub: records are not aligned, written %d from %u\n",
745                       num_recs, LLOG_BITMAP_SIZE(llh->lgh_hdr) - 1);
746
747         plain_counter = 0;
748         rc = llog_process(env, llh, test_7_print_cb, "test 7", NULL);
749         if (rc) {
750                 CERROR("7_sub: llog process failed: %d\n", rc);
751                 GOTO(out_close, rc);
752         }
753         process_count = plain_counter;
754         if (process_count != num_recs) {
755                 CERROR("7_sub: processed %d records from %d total\n",
756                        process_count, num_recs);
757                 GOTO(out_close, rc = -EINVAL);
758         }
759
760         plain_counter = 0;
761         rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);
762         if (rc && rc != LLOG_DEL_PLAIN) {
763                 CERROR("7_sub: reverse llog process failed: %d\n", rc);
764                 GOTO(out_close, rc);
765         }
766         if (process_count != plain_counter) {
767                 CERROR("7_sub: Reverse/direct processing found different"
768                        "number of records: %d/%d\n",
769                        plain_counter, process_count);
770                 GOTO(out_close, rc = -EINVAL);
771         }
772         if (llog_exist(llh)) {
773                 CERROR("7_sub: llog exists but should be zapped\n");
774                 GOTO(out_close, rc = -EEXIST);
775         }
776
777         rc = verify_handle("7_sub", llh, 1);
778 out_close:
779         if (rc)
780                 llog_destroy(env, llh);
781         llog_close(env, llh);
782         RETURN(rc);
783 }
784
785 /* Test all llog records writing and processing */
786 static int llog_test_7(const struct lu_env *env, struct obd_device *obd)
787 {
788         struct llog_ctxt        *ctxt;
789         int                      rc;
790
791         ENTRY;
792
793         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
794
795         CWARN("7a: test llog_logid_rec\n");
796         llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);
797         llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);
798         llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;
799
800         rc = llog_test_7_sub(env, ctxt);
801         if (rc) {
802                 CERROR("7a: llog_logid_rec test failed\n");
803                 GOTO(out, rc);
804         }
805
806         CWARN("7b: test llog_unlink64_rec\n");
807         llog_records.lur.lur_hdr.lrh_len = sizeof(llog_records.lur);
808         llog_records.lur.lur_tail.lrt_len = sizeof(llog_records.lur);
809         llog_records.lur.lur_hdr.lrh_type = MDS_UNLINK64_REC;
810
811         rc = llog_test_7_sub(env, ctxt);
812         if (rc) {
813                 CERROR("7b: llog_unlink_rec test failed\n");
814                 GOTO(out, rc);
815         }
816
817         CWARN("7c: test llog_setattr64_rec\n");
818         llog_records.lsr64.lsr_hdr.lrh_len = sizeof(llog_records.lsr64);
819         llog_records.lsr64.lsr_tail.lrt_len = sizeof(llog_records.lsr64);
820         llog_records.lsr64.lsr_hdr.lrh_type = MDS_SETATTR64_REC;
821
822         rc = llog_test_7_sub(env, ctxt);
823         if (rc) {
824                 CERROR("7c: llog_setattr64_rec test failed\n");
825                 GOTO(out, rc);
826         }
827
828         CWARN("7d: test llog_size_change_rec\n");
829         llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);
830         llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);
831         llog_records.lscr.lsc_hdr.lrh_type = OST_SZ_REC;
832
833         rc = llog_test_7_sub(env, ctxt);
834         if (rc) {
835                 CERROR("7d: llog_size_change_rec test failed\n");
836                 GOTO(out, rc);
837         }
838
839         CWARN("7e: test llog_changelog_rec\n");
840         llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);
841         llog_records.lcr.cr_tail.lrt_len = sizeof(llog_records.lcr);
842         llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;
843
844         rc = llog_test_7_sub(env, ctxt);
845         if (rc) {
846                 CERROR("7e: llog_changelog_rec test failed\n");
847                 GOTO(out, rc);
848         }
849
850         CWARN("7f: test llog_changelog_user_rec\n");
851         llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);
852         llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);
853         llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;
854
855         rc = llog_test_7_sub(env, ctxt);
856         if (rc) {
857                 CERROR("7f: llog_changelog_user_rec test failed\n");
858                 GOTO(out, rc);
859         }
860
861         CWARN("7g: test llog_gen_rec\n");
862         llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);
863         llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);
864         llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;
865
866         rc = llog_test_7_sub(env, ctxt);
867         if (rc) {
868                 CERROR("7g: llog_size_change_rec test failed\n");
869                 GOTO(out, rc);
870         }
871 out:
872         llog_ctxt_put(ctxt);
873         RETURN(rc);
874 }
875
876 static int llog_truncate(const struct lu_env *env, struct dt_object *o)
877 {
878         struct lu_attr           la;
879         struct thandle          *th;
880         struct dt_device        *d;
881         int                      rc;
882         ENTRY;
883
884         LASSERT(o);
885         d = lu2dt_dev(o->do_lu.lo_dev);
886         LASSERT(d);
887
888         rc = dt_attr_get(env, o, &la, NULL);
889         if (rc)
890                 RETURN(rc);
891
892         CDEBUG(D_OTHER, "original size "LPU64"\n", la.la_size);
893         rc = sizeof(struct llog_log_hdr) + sizeof(struct llog_mini_rec);
894         if (la.la_size < rc) {
895                 CERROR("too small llog: "LPU64"\n", la.la_size);
896                 RETURN(0);
897         }
898
899         /* drop 2 records */
900         la.la_size = la.la_size - (sizeof(struct llog_mini_rec) * 2);
901         la.la_valid = LA_SIZE;
902
903         th = dt_trans_create(env, d);
904         if (IS_ERR(th))
905                 RETURN(PTR_ERR(th));
906
907         rc = dt_declare_attr_set(env, o, &la, th);
908         if (rc)
909                 GOTO(stop, rc);
910
911         rc = dt_declare_punch(env, o, la.la_size, OBD_OBJECT_EOF, th);
912
913         rc = dt_trans_start_local(env, d, th);
914         if (rc)
915                 GOTO(stop, rc);
916
917         rc = dt_punch(env, o, la.la_size, OBD_OBJECT_EOF, th, BYPASS_CAPA);
918         if (rc)
919                 GOTO(stop, rc);
920
921         rc = dt_attr_set(env, o, &la, th, BYPASS_CAPA);
922         if (rc)
923                 GOTO(stop, rc);
924
925 stop:
926         dt_trans_stop(env, d, th);
927
928         RETURN(rc);
929 }
930
931 static int test_8_cb(const struct lu_env *env, struct llog_handle *llh,
932                           struct llog_rec_hdr *rec, void *data)
933 {
934         plain_counter++;
935         return 0;
936 }
937
938 static int llog_test_8(const struct lu_env *env, struct obd_device *obd)
939 {
940         struct llog_handle      *llh = NULL;
941         char                     name[10];
942         int                      rc, rc2, i;
943         int                      orig_counter;
944         struct llog_mini_rec     lmr;
945         struct llog_ctxt        *ctxt;
946         struct dt_object        *obj = NULL;
947
948         ENTRY;
949
950         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
951         LASSERT(ctxt);
952
953         lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;
954         lmr.lmr_hdr.lrh_type = 0xf00f00;
955
956         CWARN("8a: fill the first plain llog\n");
957         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
958         if (rc) {
959                 CERROR("8a: llog_create with logid failed: %d\n", rc);
960                 GOTO(out_put, rc);
961         }
962
963         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
964         if (rc) {
965                 CERROR("8a: can't init llog handle: %d\n", rc);
966                 GOTO(out, rc);
967         }
968
969         plain_counter = 0;
970         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
971         if (rc != 0) {
972                 CERROR("5a: process with cat_cancel_cb failed: %d\n", rc);
973                 GOTO(out, rc);
974         }
975         orig_counter = plain_counter;
976
977         for (i = 0; i < 100; i++) {
978                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
979                 if (rc) {
980                         CERROR("5a: add record failed\n");
981                         GOTO(out, rc);
982                 }
983         }
984
985         /* grab the current plain llog, we'll corrupt it later */
986         obj = llh->u.chd.chd_current_log->lgh_obj;
987         LASSERT(obj);
988         lu_object_get(&obj->do_lu);
989         CWARN("8a: pin llog "DFID"\n", PFID(lu_object_fid(&obj->do_lu)));
990
991         rc2 = llog_cat_close(env, llh);
992         if (rc2) {
993                 CERROR("8a: close log %s failed: %d\n", name, rc2);
994                 if (rc == 0)
995                         rc = rc2;
996                 GOTO(out_put, rc);
997         }
998
999         CWARN("8b: fill the second plain llog\n");
1000         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1001         if (rc) {
1002                 CERROR("8b: llog_create with logid failed: %d\n", rc);
1003                 GOTO(out_put, rc);
1004         }
1005
1006         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1007         if (rc) {
1008                 CERROR("8b: can't init llog handle: %d\n", rc);
1009                 GOTO(out, rc);
1010         }
1011
1012         for (i = 0; i < 100; i++) {
1013                 rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL);
1014                 if (rc) {
1015                         CERROR("8b: add record failed\n");
1016                         GOTO(out, rc);
1017                 }
1018         }
1019         CWARN("8b: second llog "DFID"\n",
1020                 PFID(lu_object_fid(&llh->u.chd.chd_current_log->lgh_obj->do_lu)));
1021
1022         rc2 = llog_cat_close(env, llh);
1023         if (rc2) {
1024                 CERROR("8b: close log %s failed: %d\n", name, rc2);
1025                 if (rc == 0)
1026                         rc = rc2;
1027                 GOTO(out_put, rc);
1028         }
1029
1030         CWARN("8c: drop two records from the first plain llog\n");
1031         llog_truncate(env, obj);
1032
1033         CWARN("8d: count survived records\n");
1034         rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);
1035         if (rc) {
1036                 CERROR("8d: llog_create with logid failed: %d\n", rc);
1037                 GOTO(out_put, rc);
1038         }
1039
1040         rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);
1041         if (rc) {
1042                 CERROR("8d: can't init llog handle: %d\n", rc);
1043                 GOTO(out, rc);
1044         }
1045
1046         plain_counter = 0;
1047         rc = llog_cat_process(env, llh, test_8_cb, "foobar", 0, 0);
1048         if (rc != 0) {
1049                 CERROR("8d: process with cat_cancel_cb failed: %d\n", rc);
1050                 GOTO(out, rc);
1051         }
1052
1053         if (orig_counter + 200 - 2 != plain_counter) {
1054                 CERROR("found %d records (expected %d)\n", plain_counter,
1055                        orig_counter + 200 - 2);
1056                 rc = -EIO;
1057         }
1058
1059 out:
1060         CWARN("8d: close re-opened catalog\n");
1061         rc2 = llog_cat_close(env, llh);
1062         if (rc2) {
1063                 CERROR("8d: close log %s failed: %d\n", name, rc2);
1064                 if (rc == 0)
1065                         rc = rc2;
1066         }
1067 out_put:
1068         llog_ctxt_put(ctxt);
1069
1070         if (obj != NULL)
1071                 lu_object_put(env, &obj->do_lu);
1072
1073         RETURN(rc);
1074 }
1075
1076 /* -------------------------------------------------------------------------
1077  * Tests above, boring obd functions below
1078  * ------------------------------------------------------------------------- */
1079 static int llog_run_tests(const struct lu_env *env, struct obd_device *obd)
1080 {
1081         struct llog_handle      *llh = NULL;
1082         struct llog_ctxt        *ctxt;
1083         int                      rc, err;
1084         char                     name[10];
1085
1086         ENTRY;
1087         ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);
1088         LASSERT(ctxt);
1089
1090         sprintf(name, "%x", llog_test_rand);
1091
1092         rc = llog_test_1(env, obd, name);
1093         if (rc)
1094                 GOTO(cleanup_ctxt, rc);
1095
1096         rc = llog_test_2(env, obd, name, &llh);
1097         if (rc)
1098                 GOTO(cleanup_ctxt, rc);
1099
1100         rc = llog_test_3(env, obd, llh);
1101         if (rc)
1102                 GOTO(cleanup, rc);
1103
1104         rc = llog_test_4(env, obd);
1105         if (rc)
1106                 GOTO(cleanup, rc);
1107
1108         rc = llog_test_5(env, obd);
1109         if (rc)
1110                 GOTO(cleanup, rc);
1111
1112         rc = llog_test_6(env, obd, name);
1113         if (rc)
1114                 GOTO(cleanup, rc);
1115
1116         rc = llog_test_7(env, obd);
1117         if (rc)
1118                 GOTO(cleanup, rc);
1119
1120         rc = llog_test_8(env, obd);
1121         if (rc)
1122                 GOTO(cleanup, rc);
1123
1124 cleanup:
1125         err = llog_destroy(env, llh);
1126         if (err)
1127                 CERROR("cleanup: llog_destroy failed: %d\n", err);
1128         llog_close(env, llh);
1129         if (rc == 0)
1130                 rc = err;
1131 cleanup_ctxt:
1132         llog_ctxt_put(ctxt);
1133         return rc;
1134 }
1135
1136 static int llog_test_cleanup(struct obd_device *obd)
1137 {
1138         struct obd_device       *tgt;
1139         struct lu_env            env;
1140         int                      rc;
1141
1142         ENTRY;
1143
1144         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1145         if (rc)
1146                 RETURN(rc);
1147
1148         tgt = obd->obd_lvfs_ctxt.dt->dd_lu_dev.ld_obd;
1149         rc = llog_cleanup(&env, llog_get_context(tgt, LLOG_TEST_ORIG_CTXT));
1150         if (rc)
1151                 CERROR("failed to llog_test_llog_finish: %d\n", rc);
1152         lu_env_fini(&env);
1153         RETURN(rc);
1154 }
1155
1156 static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
1157 {
1158         struct obd_device       *tgt;
1159         struct llog_ctxt        *ctxt;
1160         struct dt_object        *o;
1161         struct lu_env            env;
1162         struct lu_context        test_session;
1163         int                      rc;
1164
1165         ENTRY;
1166
1167         if (lcfg->lcfg_bufcount < 2) {
1168                 CERROR("requires a TARGET OBD name\n");
1169                 RETURN(-EINVAL);
1170         }
1171
1172         if (lcfg->lcfg_buflens[1] < 1) {
1173                 CERROR("requires a TARGET OBD name\n");
1174                 RETURN(-EINVAL);
1175         }
1176
1177         /* disk obd */
1178         tgt = class_name2obd(lustre_cfg_string(lcfg, 1));
1179         if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {
1180                 CERROR("target device not attached or not set up (%s)\n",
1181                        lustre_cfg_string(lcfg, 1));
1182                 RETURN(-EINVAL);
1183         }
1184
1185         rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
1186         if (rc)
1187                 RETURN(rc);
1188
1189         rc = lu_context_init(&test_session, LCT_SERVER_SESSION);
1190         if (rc)
1191                 GOTO(cleanup_env, rc);
1192         test_session.lc_thread = (struct ptlrpc_thread *)current;
1193         lu_context_enter(&test_session);
1194         env.le_ses = &test_session;
1195
1196         CWARN("Setup llog-test device over %s device\n",
1197               lustre_cfg_string(lcfg, 1));
1198
1199         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1200         obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);
1201
1202         rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,
1203                         &llog_osd_ops);
1204         if (rc)
1205                 GOTO(cleanup_session, rc);
1206
1207         /* use MGS llog dir for tests */
1208         ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);
1209         LASSERT(ctxt);
1210         o = ctxt->loc_dir;
1211         llog_ctxt_put(ctxt);
1212
1213         ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);
1214         LASSERT(ctxt);
1215         ctxt->loc_dir = o;
1216         llog_ctxt_put(ctxt);
1217
1218         llog_test_rand = cfs_rand();
1219
1220         rc = llog_run_tests(&env, tgt);
1221         if (rc)
1222                 llog_test_cleanup(obd);
1223 cleanup_session:
1224         lu_context_exit(&test_session);
1225         lu_context_fini(&test_session);
1226 cleanup_env:
1227         lu_env_fini(&env);
1228         RETURN(rc);
1229 }
1230
1231 static struct obd_ops llog_obd_ops = {
1232         .o_owner       = THIS_MODULE,
1233         .o_setup       = llog_test_setup,
1234         .o_cleanup     = llog_test_cleanup,
1235 };
1236
1237 static int __init llog_test_init(void)
1238 {
1239         return class_register_type(&llog_obd_ops, NULL, true, NULL,
1240 #ifndef HAVE_ONLY_PROCFS_SEQ
1241                                    NULL,
1242 #endif
1243                                    "llog_test", NULL);
1244 }
1245
1246 static void __exit llog_test_exit(void)
1247 {
1248         class_unregister_type("llog_test");
1249 }
1250
1251 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1252 MODULE_DESCRIPTION("llog test module");
1253 MODULE_LICENSE("GPL");
1254
1255 module_init(llog_test_init);
1256 module_exit(llog_test_exit);