Whamcloud - gitweb
LU-1346 libcfs: replace libcfs wrappers with kernel API
[fs/lustre-release.git] / lustre / obdclass / llog.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 /*
31  * This file is part of Lustre, http://www.lustre.org/
32  * Lustre is a trademark of Sun Microsystems, Inc.
33  *
34  * lustre/obdclass/llog.c
35  *
36  * OST<->MDS recovery logging infrastructure.
37  * Invariants in implementation:
38  * - we do not share logs among different OST<->MDS connections, so that
39  *   if an OST or MDS fails it need only look at log(s) relevant to itself
40  *
41  * Author: Andreas Dilger <adilger@clusterfs.com>
42  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
43  * Author: Mikhail Pershin <tappro@whamcloud.com>
44  */
45
46 #define DEBUG_SUBSYSTEM S_LOG
47
48 #ifndef __KERNEL__
49 #include <liblustre.h>
50 #endif
51
52 #include <obd_class.h>
53 #include <lustre_log.h>
54 #include "llog_internal.h"
55
56 /*
57  * Allocate a new log or catalog handle
58  * Used inside llog_open().
59  */
60 struct llog_handle *llog_alloc_handle(void)
61 {
62         struct llog_handle *loghandle;
63
64         OBD_ALLOC_PTR(loghandle);
65         if (loghandle == NULL)
66                 return ERR_PTR(-ENOMEM);
67
68         init_rwsem(&loghandle->lgh_lock);
69         spin_lock_init(&loghandle->lgh_hdr_lock);
70         CFS_INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);
71
72         return loghandle;
73 }
74
75 /*
76  * Free llog handle and header data if exists. Used in llog_close() only
77  */
78 void llog_free_handle(struct llog_handle *loghandle)
79 {
80         if (!loghandle)
81                 return;
82
83         if (!loghandle->lgh_hdr)
84                 goto out;
85         if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)
86                 cfs_list_del_init(&loghandle->u.phd.phd_entry);
87         if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
88                 LASSERT(cfs_list_empty(&loghandle->u.chd.chd_head));
89         LASSERT(sizeof(*(loghandle->lgh_hdr)) == LLOG_CHUNK_SIZE);
90         OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);
91
92 out:
93         OBD_FREE_PTR(loghandle);
94 }
95
96 /* returns negative on error; 0 if success; 1 if success & log destroyed */
97 int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
98                     int index)
99 {
100         struct llog_log_hdr *llh = loghandle->lgh_hdr;
101         int rc = 0;
102         ENTRY;
103
104         CDEBUG(D_RPCTRACE, "Canceling %d in log "LPX64"\n",
105                index, loghandle->lgh_id.lgl_oid);
106
107         if (index == 0) {
108                 CERROR("Can't cancel index 0 which is header\n");
109                 RETURN(-EINVAL);
110         }
111
112         spin_lock(&loghandle->lgh_hdr_lock);
113         if (!ext2_clear_bit(index, llh->llh_bitmap)) {
114                 spin_unlock(&loghandle->lgh_hdr_lock);
115                 CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
116                 RETURN(-ENOENT);
117         }
118
119         llh->llh_count--;
120
121         if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
122             (llh->llh_count == 1) &&
123             (loghandle->lgh_last_idx == (LLOG_BITMAP_BYTES * 8) - 1)) {
124                 spin_unlock(&loghandle->lgh_hdr_lock);
125                 rc = llog_destroy(env, loghandle);
126                 if (rc < 0) {
127                         CERROR("%s: can't destroy empty llog #"LPX64"#"LPX64
128                                "#%08x: rc = %d\n",
129                                loghandle->lgh_ctxt->loc_obd->obd_name,
130                                loghandle->lgh_id.lgl_oid,
131                                loghandle->lgh_id.lgl_oseq,
132                                loghandle->lgh_id.lgl_ogen, rc);
133                         GOTO(out_err, rc);
134                 }
135                 RETURN(1);
136         }
137         spin_unlock(&loghandle->lgh_hdr_lock);
138
139         rc = llog_write(env, loghandle, &llh->llh_hdr, NULL, 0, NULL, 0);
140         if (rc < 0) {
141                 CERROR("%s: fail to write header for llog #"LPX64"#"LPX64
142                        "#%08x: rc = %d\n",
143                        loghandle->lgh_ctxt->loc_obd->obd_name,
144                        loghandle->lgh_id.lgl_oid,
145                        loghandle->lgh_id.lgl_oseq,
146                        loghandle->lgh_id.lgl_ogen, rc);
147                 GOTO(out_err, rc);
148         }
149         RETURN(0);
150 out_err:
151         spin_lock(&loghandle->lgh_hdr_lock);
152         ext2_set_bit(index, llh->llh_bitmap);
153         llh->llh_count++;
154         spin_unlock(&loghandle->lgh_hdr_lock);
155         return rc;
156 }
157 EXPORT_SYMBOL(llog_cancel_rec);
158
159 static int llog_read_header(const struct lu_env *env,
160                             struct llog_handle *handle,
161                             struct obd_uuid *uuid)
162 {
163         struct llog_operations *lop;
164         int rc;
165
166         rc = llog_handle2ops(handle, &lop);
167         if (rc)
168                 RETURN(rc);
169
170         if (lop->lop_read_header == NULL)
171                 RETURN(-EOPNOTSUPP);
172
173         rc = lop->lop_read_header(env, handle);
174         if (rc == LLOG_EEMPTY) {
175                 struct llog_log_hdr *llh = handle->lgh_hdr;
176
177                 handle->lgh_last_idx = 0; /* header is record with index 0 */
178                 llh->llh_count = 1;         /* for the header record */
179                 llh->llh_hdr.lrh_type = LLOG_HDR_MAGIC;
180                 llh->llh_hdr.lrh_len = llh->llh_tail.lrt_len = LLOG_CHUNK_SIZE;
181                 llh->llh_hdr.lrh_index = llh->llh_tail.lrt_index = 0;
182                 llh->llh_timestamp = cfs_time_current_sec();
183                 if (uuid)
184                         memcpy(&llh->llh_tgtuuid, uuid,
185                                sizeof(llh->llh_tgtuuid));
186                 llh->llh_bitmap_offset = offsetof(typeof(*llh), llh_bitmap);
187                 ext2_set_bit(0, llh->llh_bitmap);
188                 rc = 0;
189         }
190         return rc;
191 }
192
193 int llog_init_handle(const struct lu_env *env, struct llog_handle *handle,
194                      int flags, struct obd_uuid *uuid)
195 {
196         struct llog_log_hdr     *llh;
197         int                      rc;
198
199         ENTRY;
200         LASSERT(handle->lgh_hdr == NULL);
201
202         OBD_ALLOC_PTR(llh);
203         if (llh == NULL)
204                 RETURN(-ENOMEM);
205         handle->lgh_hdr = llh;
206         /* first assign flags to use llog_client_ops */
207         llh->llh_flags = flags;
208         rc = llog_read_header(env, handle, uuid);
209         if (rc == 0) {
210                 if (unlikely((llh->llh_flags & LLOG_F_IS_PLAIN &&
211                               flags & LLOG_F_IS_CAT) ||
212                              (llh->llh_flags & LLOG_F_IS_CAT &&
213                               flags & LLOG_F_IS_PLAIN))) {
214                         CERROR("%s: llog type is %s but initializing %s\n",
215                                handle->lgh_ctxt->loc_obd->obd_name,
216                                llh->llh_flags & LLOG_F_IS_CAT ?
217                                "catalog" : "plain",
218                                flags & LLOG_F_IS_CAT ? "catalog" : "plain");
219                         GOTO(out, rc = -EINVAL);
220                 } else if (llh->llh_flags &
221                            (LLOG_F_IS_PLAIN | LLOG_F_IS_CAT)) {
222                         /*
223                          * it is possible to open llog without specifying llog
224                          * type so it is taken from llh_flags
225                          */
226                         flags = llh->llh_flags;
227                 } else {
228                         /* for some reason the llh_flags has no type set */
229                         CERROR("llog type is not specified!\n");
230                         GOTO(out, rc = -EINVAL);
231                 }
232                 if (unlikely(uuid &&
233                              !obd_uuid_equals(uuid, &llh->llh_tgtuuid))) {
234                         CERROR("%s: llog uuid mismatch: %s/%s\n",
235                                handle->lgh_ctxt->loc_obd->obd_name,
236                                (char *)uuid->uuid,
237                                (char *)llh->llh_tgtuuid.uuid);
238                         GOTO(out, rc = -EEXIST);
239                 }
240         }
241         if (flags & LLOG_F_IS_CAT) {
242                 LASSERT(cfs_list_empty(&handle->u.chd.chd_head));
243                 CFS_INIT_LIST_HEAD(&handle->u.chd.chd_head);
244                 llh->llh_size = sizeof(struct llog_logid_rec);
245         } else if (!(flags & LLOG_F_IS_PLAIN)) {
246                 CERROR("%s: unknown flags: %#x (expected %#x or %#x)\n",
247                        handle->lgh_ctxt->loc_obd->obd_name,
248                        flags, LLOG_F_IS_CAT, LLOG_F_IS_PLAIN);
249                 rc = -EINVAL;
250         }
251 out:
252         if (rc) {
253                 OBD_FREE_PTR(llh);
254                 handle->lgh_hdr = NULL;
255         }
256         RETURN(rc);
257 }
258 EXPORT_SYMBOL(llog_init_handle);
259
260 static int llog_process_thread(void *arg)
261 {
262         struct llog_process_info        *lpi = arg;
263         struct llog_handle              *loghandle = lpi->lpi_loghandle;
264         struct llog_log_hdr             *llh = loghandle->lgh_hdr;
265         struct llog_process_cat_data    *cd  = lpi->lpi_catdata;
266         char                            *buf;
267         __u64                            cur_offset = LLOG_CHUNK_SIZE;
268         __u64                            last_offset;
269         int                              rc = 0, index = 1, last_index;
270         int                              saved_index = 0;
271         int                              last_called_index = 0;
272
273         ENTRY;
274
275         LASSERT(llh);
276
277         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
278         if (!buf) {
279                 lpi->lpi_rc = -ENOMEM;
280                 RETURN(0);
281         }
282
283         if (cd != NULL) {
284                 last_called_index = cd->lpcd_first_idx;
285                 index = cd->lpcd_first_idx + 1;
286         }
287         if (cd != NULL && cd->lpcd_last_idx)
288                 last_index = cd->lpcd_last_idx;
289         else
290                 last_index = LLOG_BITMAP_BYTES * 8 - 1;
291
292         while (rc == 0) {
293                 struct llog_rec_hdr *rec;
294
295                 /* skip records not set in bitmap */
296                 while (index <= last_index &&
297                        !ext2_test_bit(index, llh->llh_bitmap))
298                         ++index;
299
300                 LASSERT(index <= last_index + 1);
301                 if (index == last_index + 1)
302                         break;
303 repeat:
304                 CDEBUG(D_OTHER, "index: %d last_index %d\n",
305                        index, last_index);
306
307                 /* get the buf with our target record; avoid old garbage */
308                 memset(buf, 0, LLOG_CHUNK_SIZE);
309                 last_offset = cur_offset;
310                 rc = llog_next_block(lpi->lpi_env, loghandle, &saved_index,
311                                      index, &cur_offset, buf, LLOG_CHUNK_SIZE);
312                 if (rc)
313                         GOTO(out, rc);
314
315                 /* NB: when rec->lrh_len is accessed it is already swabbed
316                  * since it is used at the "end" of the loop and the rec
317                  * swabbing is done at the beginning of the loop. */
318                 for (rec = (struct llog_rec_hdr *)buf;
319                      (char *)rec < buf + LLOG_CHUNK_SIZE;
320                      rec = (struct llog_rec_hdr *)((char *)rec + rec->lrh_len)){
321
322                         CDEBUG(D_OTHER, "processing rec 0x%p type %#x\n",
323                                rec, rec->lrh_type);
324
325                         if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
326                                 lustre_swab_llog_rec(rec);
327
328                         CDEBUG(D_OTHER, "after swabbing, type=%#x idx=%d\n",
329                                rec->lrh_type, rec->lrh_index);
330
331                         if (rec->lrh_index == 0) {
332                                 /* probably another rec just got added? */
333                                 if (index <= loghandle->lgh_last_idx)
334                                         GOTO(repeat, rc = 0);
335                                 GOTO(out, rc = 0); /* no more records */
336                         }
337                         if (rec->lrh_len == 0 ||
338                             rec->lrh_len > LLOG_CHUNK_SIZE) {
339                                 CWARN("invalid length %d in llog record for "
340                                       "index %d/%d\n", rec->lrh_len,
341                                       rec->lrh_index, index);
342                                 GOTO(out, rc = -EINVAL);
343                         }
344
345                         if (rec->lrh_index < index) {
346                                 CDEBUG(D_OTHER, "skipping lrh_index %d\n",
347                                        rec->lrh_index);
348                                 continue;
349                         }
350
351                         CDEBUG(D_OTHER,
352                                "lrh_index: %d lrh_len: %d (%d remains)\n",
353                                rec->lrh_index, rec->lrh_len,
354                                (int)(buf + LLOG_CHUNK_SIZE - (char *)rec));
355
356                         loghandle->lgh_cur_idx = rec->lrh_index;
357                         loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
358                                                     last_offset;
359
360                         /* if set, process the callback on this record */
361                         if (ext2_test_bit(index, llh->llh_bitmap)) {
362                                 rc = lpi->lpi_cb(lpi->lpi_env, loghandle, rec,
363                                                  lpi->lpi_cbdata);
364                                 last_called_index = index;
365                                 if (rc == LLOG_PROC_BREAK) {
366                                         GOTO(out, rc);
367                                 } else if (rc == LLOG_DEL_RECORD) {
368                                         llog_cancel_rec(lpi->lpi_env,
369                                                         loghandle,
370                                                         rec->lrh_index);
371                                         rc = 0;
372                                 }
373                                 if (rc)
374                                         GOTO(out, rc);
375                         } else {
376                                 CDEBUG(D_OTHER, "Skipped index %d\n", index);
377                         }
378
379                         /* next record, still in buffer? */
380                         ++index;
381                         if (index > last_index)
382                                 GOTO(out, rc = 0);
383                 }
384         }
385
386 out:
387         if (cd != NULL)
388                 cd->lpcd_last_idx = last_called_index;
389
390         OBD_FREE(buf, LLOG_CHUNK_SIZE);
391         lpi->lpi_rc = rc;
392         return 0;
393 }
394
395 #ifdef __KERNEL__
396 static int llog_process_thread_daemonize(void *arg)
397 {
398         struct llog_process_info        *lpi = arg;
399         struct lu_env                    env;
400         int                              rc;
401
402         cfs_daemonize_ctxt("llog_process_thread");
403
404         /* client env has no keys, tags is just 0 */
405         rc = lu_env_init(&env, LCT_LOCAL);
406         if (rc)
407                 goto out;
408         lpi->lpi_env = &env;
409
410         rc = llog_process_thread(arg);
411
412         lu_env_fini(&env);
413 out:
414         complete(&lpi->lpi_completion);
415         return rc;
416 }
417 #endif
418
419 int llog_process_or_fork(const struct lu_env *env,
420                          struct llog_handle *loghandle,
421                          llog_cb_t cb, void *data, void *catdata, bool fork)
422 {
423         struct llog_process_info *lpi;
424         int                      rc;
425
426         ENTRY;
427
428         OBD_ALLOC_PTR(lpi);
429         if (lpi == NULL) {
430                 CERROR("cannot alloc pointer\n");
431                 RETURN(-ENOMEM);
432         }
433         lpi->lpi_loghandle = loghandle;
434         lpi->lpi_cb        = cb;
435         lpi->lpi_cbdata    = data;
436         lpi->lpi_catdata   = catdata;
437
438 #ifdef __KERNEL__
439         if (fork) {
440                 /* The new thread can't use parent env,
441                  * init the new one in llog_process_thread_daemonize. */
442                 lpi->lpi_env = NULL;
443                 init_completion(&lpi->lpi_completion);
444                 rc = cfs_create_thread(llog_process_thread_daemonize, lpi,
445                                        CFS_DAEMON_FLAGS);
446                 if (rc < 0) {
447                         CERROR("%s: cannot start thread: rc = %d\n",
448                                loghandle->lgh_ctxt->loc_obd->obd_name, rc);
449                         OBD_FREE_PTR(lpi);
450                         RETURN(rc);
451                 }
452                 wait_for_completion(&lpi->lpi_completion);
453         } else {
454                 lpi->lpi_env = env;
455                 llog_process_thread(lpi);
456         }
457 #else
458         lpi->lpi_env = env;
459         llog_process_thread(lpi);
460 #endif
461         rc = lpi->lpi_rc;
462         OBD_FREE_PTR(lpi);
463         RETURN(rc);
464 }
465 EXPORT_SYMBOL(llog_process_or_fork);
466
467 int llog_process(const struct lu_env *env, struct llog_handle *loghandle,
468                  llog_cb_t cb, void *data, void *catdata)
469 {
470         return llog_process_or_fork(env, loghandle, cb, data, catdata, true);
471 }
472 EXPORT_SYMBOL(llog_process);
473
474 inline int llog_get_size(struct llog_handle *loghandle)
475 {
476         if (loghandle && loghandle->lgh_hdr)
477                 return loghandle->lgh_hdr->llh_count;
478         return 0;
479 }
480 EXPORT_SYMBOL(llog_get_size);
481
482 int llog_reverse_process(const struct lu_env *env,
483                          struct llog_handle *loghandle, llog_cb_t cb,
484                          void *data, void *catdata)
485 {
486         struct llog_log_hdr *llh = loghandle->lgh_hdr;
487         struct llog_process_cat_data *cd = catdata;
488         void *buf;
489         int rc = 0, first_index = 1, index, idx;
490         ENTRY;
491
492         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
493         if (!buf)
494                 RETURN(-ENOMEM);
495
496         if (cd != NULL)
497                 first_index = cd->lpcd_first_idx + 1;
498         if (cd != NULL && cd->lpcd_last_idx)
499                 index = cd->lpcd_last_idx;
500         else
501                 index = LLOG_BITMAP_BYTES * 8 - 1;
502
503         while (rc == 0) {
504                 struct llog_rec_hdr *rec;
505                 struct llog_rec_tail *tail;
506
507                 /* skip records not set in bitmap */
508                 while (index >= first_index &&
509                        !ext2_test_bit(index, llh->llh_bitmap))
510                         --index;
511
512                 LASSERT(index >= first_index - 1);
513                 if (index == first_index - 1)
514                         break;
515
516                 /* get the buf with our target record; avoid old garbage */
517                 memset(buf, 0, LLOG_CHUNK_SIZE);
518                 rc = llog_prev_block(env, loghandle, index, buf,
519                                      LLOG_CHUNK_SIZE);
520                 if (rc)
521                         GOTO(out, rc);
522
523                 rec = buf;
524                 idx = rec->lrh_index;
525                 CDEBUG(D_RPCTRACE, "index %u : idx %u\n", index, idx);
526                 while (idx < index) {
527                         rec = (void *)rec + rec->lrh_len;
528                         if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
529                                 lustre_swab_llog_rec(rec);
530                         idx ++;
531                 }
532                 LASSERT(idx == index);
533                 tail = (void *)rec + rec->lrh_len - sizeof(*tail);
534
535                 /* process records in buffer, starting where we found one */
536                 while ((void *)tail > buf) {
537                         if (tail->lrt_index == 0)
538                                 GOTO(out, rc = 0); /* no more records */
539
540                         /* if set, process the callback on this record */
541                         if (ext2_test_bit(index, llh->llh_bitmap)) {
542                                 rec = (void *)tail - tail->lrt_len +
543                                       sizeof(*tail);
544
545                                 rc = cb(env, loghandle, rec, data);
546                                 if (rc == LLOG_PROC_BREAK) {
547                                         GOTO(out, rc);
548                                 } else if (rc == LLOG_DEL_RECORD) {
549                                         llog_cancel_rec(env, loghandle,
550                                                         tail->lrt_index);
551                                         rc = 0;
552                                 }
553                                 if (rc)
554                                         GOTO(out, rc);
555                         }
556
557                         /* previous record, still in buffer? */
558                         --index;
559                         if (index < first_index)
560                                 GOTO(out, rc = 0);
561                         tail = (void *)tail - tail->lrt_len;
562                 }
563         }
564
565 out:
566         if (buf)
567                 OBD_FREE(buf, LLOG_CHUNK_SIZE);
568         RETURN(rc);
569 }
570 EXPORT_SYMBOL(llog_reverse_process);
571
572 /**
573  * new llog API
574  *
575  * API functions:
576  *      llog_open - open llog, may not exist
577  *      llog_exist - check if llog exists
578  *      llog_close - close opened llog, pair for open, frees llog_handle
579  *      llog_declare_create - declare llog creation
580  *      llog_create - create new llog on disk, need transaction handle
581  *      llog_declare_write_rec - declaration of llog write
582  *      llog_write_rec - write llog record on disk, need transaction handle
583  *      llog_declare_add - declare llog catalog record addition
584  *      llog_add - add llog record in catalog, need transaction handle
585  */
586 int llog_exist(struct llog_handle *loghandle)
587 {
588         struct llog_operations  *lop;
589         int                      rc;
590
591         ENTRY;
592
593         rc = llog_handle2ops(loghandle, &lop);
594         if (rc)
595                 RETURN(rc);
596         if (lop->lop_exist == NULL)
597                 RETURN(-EOPNOTSUPP);
598
599         rc = lop->lop_exist(loghandle);
600         RETURN(rc);
601 }
602 EXPORT_SYMBOL(llog_exist);
603
604 int llog_declare_create(const struct lu_env *env,
605                         struct llog_handle *loghandle, struct thandle *th)
606 {
607         struct llog_operations  *lop;
608         int                      raised, rc;
609
610         ENTRY;
611
612         rc = llog_handle2ops(loghandle, &lop);
613         if (rc)
614                 RETURN(rc);
615         if (lop->lop_declare_create == NULL)
616                 RETURN(-EOPNOTSUPP);
617
618         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
619         if (!raised)
620                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
621         rc = lop->lop_declare_create(env, loghandle, th);
622         if (!raised)
623                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
624         RETURN(rc);
625 }
626 EXPORT_SYMBOL(llog_declare_create);
627
628 int llog_create(const struct lu_env *env, struct llog_handle *handle,
629                 struct thandle *th)
630 {
631         struct llog_operations  *lop;
632         int                      raised, rc;
633
634         ENTRY;
635
636         rc = llog_handle2ops(handle, &lop);
637         if (rc)
638                 RETURN(rc);
639         if (lop->lop_create == NULL)
640                 RETURN(-EOPNOTSUPP);
641
642         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
643         if (!raised)
644                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
645         rc = lop->lop_create(env, handle, th);
646         if (!raised)
647                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
648         RETURN(rc);
649 }
650 EXPORT_SYMBOL(llog_create);
651
652 int llog_declare_write_rec(const struct lu_env *env,
653                            struct llog_handle *handle,
654                            struct llog_rec_hdr *rec, int idx,
655                            struct thandle *th)
656 {
657         struct llog_operations  *lop;
658         int                      raised, rc;
659
660         ENTRY;
661
662         rc = llog_handle2ops(handle, &lop);
663         if (rc)
664                 RETURN(rc);
665         LASSERT(lop);
666         if (lop->lop_declare_write_rec == NULL)
667                 RETURN(-EOPNOTSUPP);
668
669         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
670         if (!raised)
671                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
672         rc = lop->lop_declare_write_rec(env, handle, rec, idx, th);
673         if (!raised)
674                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
675         RETURN(rc);
676 }
677 EXPORT_SYMBOL(llog_declare_write_rec);
678
679 int llog_write_rec(const struct lu_env *env, struct llog_handle *handle,
680                    struct llog_rec_hdr *rec, struct llog_cookie *logcookies,
681                    int numcookies, void *buf, int idx, struct thandle *th)
682 {
683         struct llog_operations  *lop;
684         int                      raised, rc, buflen;
685
686         ENTRY;
687
688         rc = llog_handle2ops(handle, &lop);
689         if (rc)
690                 RETURN(rc);
691
692         LASSERT(lop);
693         if (lop->lop_write_rec == NULL)
694                 RETURN(-EOPNOTSUPP);
695
696         if (buf)
697                 buflen = rec->lrh_len + sizeof(struct llog_rec_hdr) +
698                          sizeof(struct llog_rec_tail);
699         else
700                 buflen = rec->lrh_len;
701         LASSERT(cfs_size_round(buflen) == buflen);
702
703         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
704         if (!raised)
705                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
706         rc = lop->lop_write_rec(env, handle, rec, logcookies, numcookies,
707                                 buf, idx, th);
708         if (!raised)
709                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
710         RETURN(rc);
711 }
712 EXPORT_SYMBOL(llog_write_rec);
713
714 int llog_add(const struct lu_env *env, struct llog_handle *lgh,
715              struct llog_rec_hdr *rec, struct llog_cookie *logcookies,
716              void *buf, struct thandle *th)
717 {
718         int raised, rc;
719
720         ENTRY;
721
722         if (lgh->lgh_logops->lop_add == NULL)
723                 RETURN(-EOPNOTSUPP);
724
725         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
726         if (!raised)
727                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
728         rc = lgh->lgh_logops->lop_add(env, lgh, rec, logcookies, buf, th);
729         if (!raised)
730                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
731         RETURN(rc);
732 }
733 EXPORT_SYMBOL(llog_add);
734
735 int llog_declare_add(const struct lu_env *env, struct llog_handle *lgh,
736                      struct llog_rec_hdr *rec, struct thandle *th)
737 {
738         int raised, rc;
739
740         ENTRY;
741
742         if (lgh->lgh_logops->lop_declare_add == NULL)
743                 RETURN(-EOPNOTSUPP);
744
745         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
746         if (!raised)
747                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
748         rc = lgh->lgh_logops->lop_declare_add(env, lgh, rec, th);
749         if (!raised)
750                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
751         RETURN(rc);
752 }
753 EXPORT_SYMBOL(llog_declare_add);
754
755 /**
756  * Helper function to open llog or create it if doesn't exist.
757  * It hides all transaction handling from caller.
758  */
759 int llog_open_create(const struct lu_env *env, struct llog_ctxt *ctxt,
760                      struct llog_handle **res, struct llog_logid *logid,
761                      char *name)
762 {
763         struct thandle  *th;
764         int              rc;
765
766         ENTRY;
767
768         rc = llog_open(env, ctxt, res, logid, name, LLOG_OPEN_NEW);
769         if (rc)
770                 RETURN(rc);
771
772         if (llog_exist(*res))
773                 RETURN(0);
774
775         if ((*res)->lgh_obj != NULL) {
776                 struct dt_device *d;
777
778                 d = lu2dt_dev((*res)->lgh_obj->do_lu.lo_dev);
779
780                 th = dt_trans_create(env, d);
781                 if (IS_ERR(th))
782                         GOTO(out, rc = PTR_ERR(th));
783
784                 rc = llog_declare_create(env, *res, th);
785                 if (rc == 0) {
786                         rc = dt_trans_start_local(env, d, th);
787                         if (rc == 0)
788                                 rc = llog_create(env, *res, th);
789                 }
790                 dt_trans_stop(env, d, th);
791         } else {
792                 /* lvfs compat code */
793                 LASSERT((*res)->lgh_file == NULL);
794                 rc = llog_create(env, *res, NULL);
795         }
796 out:
797         if (rc)
798                 llog_close(env, *res);
799         RETURN(rc);
800 }
801 EXPORT_SYMBOL(llog_open_create);
802
803 /**
804  * Helper function to delete existent llog.
805  */
806 int llog_erase(const struct lu_env *env, struct llog_ctxt *ctxt,
807                struct llog_logid *logid, char *name)
808 {
809         struct llog_handle      *handle;
810         int                      rc = 0, rc2;
811
812         ENTRY;
813
814         /* nothing to erase */
815         if (name == NULL && logid == NULL)
816                 RETURN(0);
817
818         rc = llog_open(env, ctxt, &handle, logid, name, LLOG_OPEN_EXISTS);
819         if (rc < 0)
820                 RETURN(rc);
821
822         rc = llog_init_handle(env, handle, LLOG_F_IS_PLAIN, NULL);
823         if (rc == 0)
824                 rc = llog_destroy(env, handle);
825
826         rc2 = llog_close(env, handle);
827         if (rc == 0)
828                 rc = rc2;
829         RETURN(rc);
830 }
831 EXPORT_SYMBOL(llog_erase);
832
833 /*
834  * Helper function for write record in llog.
835  * It hides all transaction handling from caller.
836  * Valid only with local llog.
837  */
838 int llog_write(const struct lu_env *env, struct llog_handle *loghandle,
839                struct llog_rec_hdr *rec, struct llog_cookie *reccookie,
840                int cookiecount, void *buf, int idx)
841 {
842         int rc;
843
844         ENTRY;
845
846         LASSERT(loghandle);
847         LASSERT(loghandle->lgh_ctxt);
848
849         if (loghandle->lgh_obj != NULL) {
850                 struct dt_device        *dt;
851                 struct thandle          *th;
852
853                 dt = lu2dt_dev(loghandle->lgh_obj->do_lu.lo_dev);
854
855                 th = dt_trans_create(env, dt);
856                 if (IS_ERR(th))
857                         RETURN(PTR_ERR(th));
858
859                 rc = llog_declare_write_rec(env, loghandle, rec, idx, th);
860                 if (rc)
861                         GOTO(out_trans, rc);
862
863                 rc = dt_trans_start_local(env, dt, th);
864                 if (rc)
865                         GOTO(out_trans, rc);
866
867                 down_write(&loghandle->lgh_lock);
868                 rc = llog_write_rec(env, loghandle, rec, reccookie,
869                                     cookiecount, buf, idx, th);
870                 up_write(&loghandle->lgh_lock);
871 out_trans:
872                 dt_trans_stop(env, dt, th);
873         } else { /* lvfs compatibility */
874                 down_write(&loghandle->lgh_lock);
875                 rc = llog_write_rec(env, loghandle, rec, reccookie,
876                                     cookiecount, buf, idx, NULL);
877                 up_write(&loghandle->lgh_lock);
878         }
879         RETURN(rc);
880 }
881 EXPORT_SYMBOL(llog_write);
882
883 int llog_open(const struct lu_env *env, struct llog_ctxt *ctxt,
884               struct llog_handle **lgh, struct llog_logid *logid,
885               char *name, enum llog_open_param open_param)
886 {
887         int      raised;
888         int      rc;
889
890         ENTRY;
891
892         LASSERT(ctxt);
893         LASSERT(ctxt->loc_logops);
894
895         if (ctxt->loc_logops->lop_open == NULL) {
896                 *lgh = NULL;
897                 RETURN(-EOPNOTSUPP);
898         }
899
900         *lgh = llog_alloc_handle();
901         if (*lgh == NULL)
902                 RETURN(-ENOMEM);
903         (*lgh)->lgh_ctxt = ctxt;
904         (*lgh)->lgh_logops = ctxt->loc_logops;
905
906         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
907         if (!raised)
908                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
909         rc = ctxt->loc_logops->lop_open(env, *lgh, logid, name, open_param);
910         if (!raised)
911                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
912         if (rc) {
913                 llog_free_handle(*lgh);
914                 *lgh = NULL;
915         }
916         RETURN(rc);
917 }
918 EXPORT_SYMBOL(llog_open);
919
920 int llog_close(const struct lu_env *env, struct llog_handle *loghandle)
921 {
922         struct llog_operations  *lop;
923         int                      rc;
924
925         ENTRY;
926
927         rc = llog_handle2ops(loghandle, &lop);
928         if (rc)
929                 GOTO(out, rc);
930         if (lop->lop_close == NULL)
931                 GOTO(out, -EOPNOTSUPP);
932         rc = lop->lop_close(env, loghandle);
933 out:
934         llog_free_handle(loghandle);
935         RETURN(rc);
936 }
937 EXPORT_SYMBOL(llog_close);