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 /* Allocate a new log or catalog handle */
  57 struct llog_handle *llog_alloc_handle(void)
  58 {
  59         struct llog_handle *loghandle;
  60         ENTRY;
  61
  62         OBD_ALLOC_PTR(loghandle);
  63         if (loghandle == NULL)
  64                 RETURN(ERR_PTR(-ENOMEM));
  65
  66         cfs_init_rwsem(&loghandle->lgh_lock);
  67         cfs_spin_lock_init(&loghandle->lgh_hdr_lock);
  68         CFS_INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);
  69
  70         RETURN(loghandle);
  71 }
  72 EXPORT_SYMBOL(llog_alloc_handle);
  73
  74
  75 void llog_free_handle(struct llog_handle *loghandle)
  76 {
  77         if (!loghandle)
  78                 return;
  79
  80         if (!loghandle->lgh_hdr)
  81                 goto out;
  82         if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)
  83                 cfs_list_del_init(&loghandle->u.phd.phd_entry);
  84         if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
  85                 LASSERT(cfs_list_empty(&loghandle->u.chd.chd_head));
  86         LASSERT(sizeof(*(loghandle->lgh_hdr)) == LLOG_CHUNK_SIZE);
  87         OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);
  88
  89 out:
  90         OBD_FREE_PTR(loghandle);
  91 }
  92 EXPORT_SYMBOL(llog_free_handle);
  93
  94 /* returns negative on error; 0 if success; 1 if success & log destroyed */
  95 int llog_cancel_rec(struct llog_handle *loghandle, int index)
  96 {
  97         struct llog_log_hdr *llh = loghandle->lgh_hdr;
  98         int rc = 0;
  99         ENTRY;
 100
 101         CDEBUG(D_RPCTRACE, "Canceling %d in log "LPX64"\n",
 102                index, loghandle->lgh_id.lgl_oid);
 103
 104         if (index == 0) {
 105                 CERROR("Can't cancel index 0 which is header\n");
 106                 RETURN(-EINVAL);
 107         }
 108
 109         cfs_spin_lock(&loghandle->lgh_hdr_lock);
 110         if (!ext2_clear_bit(index, llh->llh_bitmap)) {
 111                 cfs_spin_unlock(&loghandle->lgh_hdr_lock);
 112                 CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
 113                 RETURN(-ENOENT);
 114         }
 115
 116         llh->llh_count--;
 117
 118         if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
 119             (llh->llh_count == 1) &&
 120             (loghandle->lgh_last_idx == (LLOG_BITMAP_BYTES * 8) - 1)) {
 121                 cfs_spin_unlock(&loghandle->lgh_hdr_lock);
 122                 rc = llog_destroy(loghandle);
 123                 if (rc < 0) {
 124                         CERROR("%s: can't destroy empty llog #"LPX64"#"LPX64
 125                                "#%08x: rc = %d\n",
 126                                loghandle->lgh_ctxt->loc_obd->obd_name,
 127                                loghandle->lgh_id.lgl_oid,
 128                                loghandle->lgh_id.lgl_oseq,
 129                                loghandle->lgh_id.lgl_ogen, rc);
 130                         GOTO(out_err, rc);
 131                 }
 132                 RETURN(1);
 133         }
 134         cfs_spin_unlock(&loghandle->lgh_hdr_lock);
 135
 136         rc = llog_write_rec(loghandle, &llh->llh_hdr, NULL, 0, NULL, 0);
 137         if (rc < 0) {
 138                 CERROR("%s: fail to write header for llog #"LPX64"#"LPX64
 139                        "#%08x: rc = %d\n",
 140                        loghandle->lgh_ctxt->loc_obd->obd_name,
 141                        loghandle->lgh_id.lgl_oid,
 142                        loghandle->lgh_id.lgl_oseq,
 143                        loghandle->lgh_id.lgl_ogen, rc);
 144                 GOTO(out_err, rc);
 145         }
 146         RETURN(0);
 147 out_err:
 148         cfs_spin_lock(&loghandle->lgh_hdr_lock);
 149         ext2_set_bit(index, llh->llh_bitmap);
 150         llh->llh_count++;
 151         cfs_spin_unlock(&loghandle->lgh_hdr_lock);
 152         return rc;
 153 }
 154 EXPORT_SYMBOL(llog_cancel_rec);
 155
 156 int llog_init_handle(struct llog_handle *handle, int flags,
 157                      struct obd_uuid *uuid)
 158 {
 159         int rc;
 160         struct llog_log_hdr *llh;
 161         ENTRY;
 162         LASSERT(handle->lgh_hdr == NULL);
 163
 164         OBD_ALLOC_PTR(llh);
 165         if (llh == NULL)
 166                 RETURN(-ENOMEM);
 167         handle->lgh_hdr = llh;
 168         /* first assign flags to use llog_client_ops */
 169         llh->llh_flags = flags;
 170         rc = llog_read_header(handle);
 171         if (rc == 0) {
 172                 flags = llh->llh_flags;
 173                 if (uuid && !obd_uuid_equals(uuid, &llh->llh_tgtuuid)) {
 174                         CERROR("uuid mismatch: %s/%s\n", (char *)uuid->uuid,
 175                                (char *)llh->llh_tgtuuid.uuid);
 176                         rc = -EEXIST;
 177                 }
 178                 GOTO(out, rc);
 179         } else if (rc != LLOG_EEMPTY || !flags) {
 180                 /* set a pesudo flag for initialization */
 181                 flags = LLOG_F_IS_CAT;
 182                 GOTO(out, rc);
 183         }
 184         rc = 0;
 185
 186         handle->lgh_last_idx = 0; /* header is record with index 0 */
 187         llh->llh_count = 1;         /* for the header record */
 188         llh->llh_hdr.lrh_type = LLOG_HDR_MAGIC;
 189         llh->llh_hdr.lrh_len = llh->llh_tail.lrt_len = LLOG_CHUNK_SIZE;
 190         llh->llh_hdr.lrh_index = llh->llh_tail.lrt_index = 0;
 191         llh->llh_timestamp = cfs_time_current_sec();
 192         if (uuid)
 193                 memcpy(&llh->llh_tgtuuid, uuid, sizeof(llh->llh_tgtuuid));
 194         llh->llh_bitmap_offset = offsetof(typeof(*llh),llh_bitmap);
 195         ext2_set_bit(0, llh->llh_bitmap);
 196
 197 out:
 198         if (flags & LLOG_F_IS_CAT) {
 199                 LASSERT(cfs_list_empty(&handle->u.chd.chd_head));
 200                 CFS_INIT_LIST_HEAD(&handle->u.chd.chd_head);
 201                 llh->llh_size = sizeof(struct llog_logid_rec);
 202         } else if (!(flags & LLOG_F_IS_PLAIN)) {
 203                 CERROR("Unknown flags: %#x (Expected %#x or %#x\n",
 204                        flags, LLOG_F_IS_CAT, LLOG_F_IS_PLAIN);
 205                 rc = -EINVAL;
 206         }
 207
 208         if (rc) {
 209                 OBD_FREE_PTR(llh);
 210                 handle->lgh_hdr = NULL;
 211         }
 212         RETURN(rc);
 213 }
 214 EXPORT_SYMBOL(llog_init_handle);
 215
 216 int llog_close(struct llog_handle *loghandle)
 217 {
 218         struct llog_operations *lop;
 219         int rc;
 220         ENTRY;
 221
 222         rc = llog_handle2ops(loghandle, &lop);
 223         if (rc)
 224                 GOTO(out, rc);
 225         if (lop->lop_close == NULL)
 226                 GOTO(out, -EOPNOTSUPP);
 227         rc = lop->lop_close(loghandle);
 228  out:
 229         llog_free_handle(loghandle);
 230         RETURN(rc);
 231 }
 232 EXPORT_SYMBOL(llog_close);
 233
 234 static int llog_process_thread(void *arg)
 235 {
 236         struct llog_process_info     *lpi = (struct llog_process_info *)arg;
 237         struct llog_handle           *loghandle = lpi->lpi_loghandle;
 238         struct llog_log_hdr          *llh = loghandle->lgh_hdr;
 239         struct llog_process_cat_data *cd  = lpi->lpi_catdata;
 240         char                         *buf;
 241         __u64                         cur_offset = LLOG_CHUNK_SIZE;
 242         __u64                         last_offset;
 243         int                           rc = 0, index = 1, last_index;
 244         int                           saved_index = 0, last_called_index = 0;
 245
 246         LASSERT(llh);
 247
 248         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 249         if (!buf) {
 250                 lpi->lpi_rc = -ENOMEM;
 251 #ifdef __KERNEL__
 252                 cfs_complete(&lpi->lpi_completion);
 253 #endif
 254                 return 0;
 255         }
 256
 257         if (!(lpi->lpi_flags & LLOG_FLAG_NODEAMON))
 258                 cfs_daemonize_ctxt("llog_process_thread");
 259
 260         if (cd != NULL) {
 261                 last_called_index = cd->lpcd_first_idx;
 262                 index = cd->lpcd_first_idx + 1;
 263         }
 264         if (cd != NULL && cd->lpcd_last_idx)
 265                 last_index = cd->lpcd_last_idx;
 266         else
 267                 last_index = LLOG_BITMAP_BYTES * 8 - 1;
 268
 269         while (rc == 0) {
 270                 struct llog_rec_hdr *rec;
 271
 272                 /* skip records not set in bitmap */
 273                 while (index <= last_index &&
 274                        !ext2_test_bit(index, llh->llh_bitmap))
 275                         ++index;
 276
 277                 LASSERT(index <= last_index + 1);
 278                 if (index == last_index + 1)
 279                         break;
 280
 281                 CDEBUG(D_OTHER, "index: %d last_index %d\n",
 282                        index, last_index);
 283
 284                 /* get the buf with our target record; avoid old garbage */
 285                 memset(buf, 0, LLOG_CHUNK_SIZE);
 286                 last_offset = cur_offset;
 287                 rc = llog_next_block(loghandle, &saved_index, index,
 288                                      &cur_offset, buf, LLOG_CHUNK_SIZE);
 289                 if (rc)
 290                         GOTO(out, rc);
 291
 292                 /* NB: when rec->lrh_len is accessed it is already swabbed
 293                  * since it is used at the "end" of the loop and the rec
 294                  * swabbing is done at the beginning of the loop. */
 295                 for (rec = (struct llog_rec_hdr *)buf;
 296                      (char *)rec < buf + LLOG_CHUNK_SIZE;
 297                      rec = (struct llog_rec_hdr *)((char *)rec + rec->lrh_len)){
 298
 299                         CDEBUG(D_OTHER, "processing rec 0x%p type %#x\n",
 300                                rec, rec->lrh_type);
 301
 302                         if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
 303                                 lustre_swab_llog_rec(rec, NULL);
 304
 305                         CDEBUG(D_OTHER, "after swabbing, type=%#x idx=%d\n",
 306                                rec->lrh_type, rec->lrh_index);
 307
 308                         if (rec->lrh_index == 0)
 309                                 GOTO(out, 0); /* no more records */
 310
 311                         if (rec->lrh_len == 0 || rec->lrh_len >LLOG_CHUNK_SIZE){
 312                                 CWARN("invalid length %d in llog record for "
 313                                       "index %d/%d\n", rec->lrh_len,
 314                                       rec->lrh_index, index);
 315                                 GOTO(out, rc = -EINVAL);
 316                         }
 317
 318                         if (rec->lrh_index < index) {
 319                                 CDEBUG(D_OTHER, "skipping lrh_index %d\n",
 320                                        rec->lrh_index);
 321                                 continue;
 322                         }
 323
 324                         CDEBUG(D_OTHER,
 325                                "lrh_index: %d lrh_len: %d (%d remains)\n",
 326                                rec->lrh_index, rec->lrh_len,
 327                                (int)(buf + LLOG_CHUNK_SIZE - (char *)rec));
 328
 329                         loghandle->lgh_cur_idx = rec->lrh_index;
 330                         loghandle->lgh_cur_offset = (char *)rec - (char *)buf +
 331                                                     last_offset;
 332
 333                         /* if set, process the callback on this record */
 334                         if (ext2_test_bit(index, llh->llh_bitmap)) {
 335                                 rc = lpi->lpi_cb(loghandle, rec,
 336                                                  lpi->lpi_cbdata);
 337                                 last_called_index = index;
 338                                 if (rc == LLOG_PROC_BREAK) {
 339                                         GOTO(out, rc);
 340                                 } else if (rc == LLOG_DEL_RECORD) {
 341                                         llog_cancel_rec(loghandle,
 342                                                         rec->lrh_index);
 343                                         rc = 0;
 344                                 }
 345                                 if (rc)
 346                                         GOTO(out, rc);
 347                         } else {
 348                                 CDEBUG(D_OTHER, "Skipped index %d\n", index);
 349                         }
 350
 351                         /* next record, still in buffer? */
 352                         ++index;
 353                         if (index > last_index)
 354                                 GOTO(out, rc = 0);
 355                 }
 356         }
 357
 358  out:
 359         if (cd != NULL)
 360                 cd->lpcd_last_idx = last_called_index;
 361         if (buf)
 362                 OBD_FREE(buf, LLOG_CHUNK_SIZE);
 363         lpi->lpi_rc = rc;
 364 #ifdef __KERNEL__
 365         cfs_complete(&lpi->lpi_completion);
 366 #endif
 367         return 0;
 368 }
 369
 370 int llog_process_flags(struct llog_handle *loghandle, llog_cb_t cb,
 371                        void *data, void *catdata, int flags)
 372 {
 373         struct llog_process_info *lpi;
 374         int                      rc;
 375         ENTRY;
 376
 377         OBD_ALLOC_PTR(lpi);
 378         if (lpi == NULL) {
 379                 CERROR("cannot alloc pointer\n");
 380                 RETURN(-ENOMEM);
 381         }
 382         lpi->lpi_loghandle = loghandle;
 383         lpi->lpi_cb        = cb;
 384         lpi->lpi_cbdata    = data;
 385         lpi->lpi_catdata   = catdata;
 386         lpi->lpi_flags     = flags;
 387
 388 #ifdef __KERNEL__
 389         cfs_init_completion(&lpi->lpi_completion);
 390         rc = cfs_create_thread(llog_process_thread, lpi, CFS_DAEMON_FLAGS);
 391         if (rc < 0) {
 392                 CERROR("cannot start thread: %d\n", rc);
 393                 OBD_FREE_PTR(lpi);
 394                 RETURN(rc);
 395         }
 396         cfs_wait_for_completion(&lpi->lpi_completion);
 397 #else
 398         llog_process_thread(lpi);
 399 #endif
 400         rc = lpi->lpi_rc;
 401         OBD_FREE_PTR(lpi);
 402         RETURN(rc);
 403 }
 404 EXPORT_SYMBOL(llog_process_flags);
 405
 406 int llog_process(struct llog_handle *loghandle, llog_cb_t cb,
 407                  void *data, void *catdata)
 408 {
 409         return llog_process_flags(loghandle, cb, data, catdata, 0);
 410 }
 411 EXPORT_SYMBOL(llog_process);
 412
 413 inline int llog_get_size(struct llog_handle *loghandle)
 414 {
 415         if (loghandle && loghandle->lgh_hdr)
 416                 return loghandle->lgh_hdr->llh_count;
 417         return 0;
 418 }
 419 EXPORT_SYMBOL(llog_get_size);
 420
 421 int llog_reverse_process(struct llog_handle *loghandle, llog_cb_t cb,
 422                          void *data, void *catdata)
 423 {
 424         struct llog_log_hdr *llh = loghandle->lgh_hdr;
 425         struct llog_process_cat_data *cd = catdata;
 426         void *buf;
 427         int rc = 0, first_index = 1, index, idx;
 428         ENTRY;
 429
 430         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 431         if (!buf)
 432                 RETURN(-ENOMEM);
 433
 434         if (cd != NULL)
 435                 first_index = cd->lpcd_first_idx + 1;
 436         if (cd != NULL && cd->lpcd_last_idx)
 437                 index = cd->lpcd_last_idx;
 438         else
 439                 index = LLOG_BITMAP_BYTES * 8 - 1;
 440
 441         while (rc == 0) {
 442                 struct llog_rec_hdr *rec;
 443                 struct llog_rec_tail *tail;
 444
 445                 /* skip records not set in bitmap */
 446                 while (index >= first_index &&
 447                        !ext2_test_bit(index, llh->llh_bitmap))
 448                         --index;
 449
 450                 LASSERT(index >= first_index - 1);
 451                 if (index == first_index - 1)
 452                         break;
 453
 454                 /* get the buf with our target record; avoid old garbage */
 455                 memset(buf, 0, LLOG_CHUNK_SIZE);
 456                 rc = llog_prev_block(loghandle, index, buf, LLOG_CHUNK_SIZE);
 457                 if (rc)
 458                         GOTO(out, rc);
 459
 460                 rec = buf;
 461                 idx = le32_to_cpu(rec->lrh_index);
 462                 if (idx < index)
 463                         CDEBUG(D_RPCTRACE, "index %u : idx %u\n", index, idx);
 464                 while (idx < index) {
 465                         rec = ((void *)rec + le32_to_cpu(rec->lrh_len));
 466                         idx ++;
 467                 }
 468                 tail = (void *)rec + le32_to_cpu(rec->lrh_len) - sizeof(*tail);
 469
 470                 /* process records in buffer, starting where we found one */
 471                 while ((void *)tail > buf) {
 472                         rec = (void *)tail - le32_to_cpu(tail->lrt_len) +
 473                                 sizeof(*tail);
 474
 475                         if (rec->lrh_index == 0)
 476                                 GOTO(out, 0); /* no more records */
 477
 478                         /* if set, process the callback on this record */
 479                         if (ext2_test_bit(index, llh->llh_bitmap)) {
 480                                 rc = cb(loghandle, rec, data);
 481                                 if (rc == LLOG_PROC_BREAK) {
 482                                         GOTO(out, rc);
 483                                 }
 484                                 if (rc)
 485                                         GOTO(out, rc);
 486                         }
 487
 488                         /* previous record, still in buffer? */
 489                         --index;
 490                         if (index < first_index)
 491                                 GOTO(out, rc = 0);
 492                         tail = (void *)rec - sizeof(*tail);
 493                 }
 494         }
 495
 496 out:
 497         if (buf)
 498                 OBD_FREE(buf, LLOG_CHUNK_SIZE);
 499         RETURN(rc);
 500 }
 501 EXPORT_SYMBOL(llog_reverse_process);