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