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