lustre/mds/mds_log.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/mds/mds_log.c
  39  *
  40  * Author: Peter Braam <braam@clusterfs.com>
  41  * Author: Andreas Dilger <adilger@clusterfs.com>
  42  * Author: Phil Schwan <phil@clusterfs.com>
  43  */
  44
  45 #define DEBUG_SUBSYSTEM S_MDS
  46
  47 #ifndef AUTOCONF_INCLUDED
  48 #include <linux/config.h>
  49 #endif
  50 #include <linux/module.h>
  51 #include <linux/version.h>
  52
  53 #include <libcfs/list.h>
  54 #include <obd_class.h>
  55 #include <lustre_fsfilt.h>
  56 #include <lustre_mds.h>
  57 #include <lustre_log.h>
  58 #include "mds_internal.h"
  59
  60 static int mds_llog_origin_add(struct llog_ctxt *ctxt, struct llog_rec_hdr *rec,
  61                                struct lov_stripe_md *lsm,
  62                                struct llog_cookie *logcookies, int numcookies)
  63 {
  64         struct obd_device *obd = ctxt->loc_obd;
  65         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
  66         struct llog_ctxt *lctxt;
  67         int rc;
  68         ENTRY;
  69
  70         lctxt = llog_get_context(lov_obd, ctxt->loc_idx);
  71         rc = llog_add(lctxt, rec, lsm, logcookies, numcookies);
  72         llog_ctxt_put(lctxt);
  73
  74         RETURN(rc);
  75 }
  76
  77 static int mds_llog_origin_connect(struct llog_ctxt *ctxt,
  78                                    struct llog_logid *logid,
  79                                    struct llog_gen *gen,
  80                                    struct obd_uuid *uuid)
  81 {
  82         struct obd_device *obd = ctxt->loc_obd;
  83         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
  84         struct llog_ctxt *lctxt;
  85         int rc;
  86         ENTRY;
  87
  88         lctxt = llog_get_context(lov_obd, ctxt->loc_idx);
  89         rc = llog_connect(lctxt, logid, gen, uuid);
  90         llog_ctxt_put(lctxt);
  91         RETURN(rc);
  92 }
  93
  94 static struct llog_operations mds_ost_orig_logops = {
  95         lop_add:        mds_llog_origin_add,
  96         lop_connect:    mds_llog_origin_connect,
  97 };
  98
  99 static int mds_llog_repl_cancel(struct llog_ctxt *ctxt, struct lov_stripe_md *lsm,
 100                           int count, struct llog_cookie *cookies, int flags)
 101 {
 102         struct obd_device *obd = ctxt->loc_obd;
 103         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
 104         struct llog_ctxt *lctxt;
 105         int rc;
 106         ENTRY;
 107
 108         lctxt = llog_get_context(lov_obd, ctxt->loc_idx);
 109         rc = llog_cancel(lctxt, lsm, count, cookies, flags);
 110         llog_ctxt_put(lctxt);
 111         RETURN(rc);
 112 }
 113
 114 static struct llog_operations mds_size_repl_logops = {
 115         lop_cancel:     mds_llog_repl_cancel,
 116 };
 117
 118 static struct llog_operations changelog_orig_logops;
 119
 120 static int llog_changelog_cancel_cb(struct llog_handle *llh,
 121                                     struct llog_rec_hdr *hdr, void *data)
 122 {
 123         struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
 124         struct llog_cookie cookie;
 125         long long endrec = *(long long *)data;
 126         int rc, err;
 127         struct obd_device *obd;
 128         void *trans_h;
 129         struct inode *inode;
 130         ENTRY;
 131
 132         /* This is always a (sub)log, not the catalog */
 133         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
 134
 135         if (rec->cr.cr_index > endrec)
 136                 /* records are in order, so we're done */
 137                 RETURN(LLOG_PROC_BREAK);
 138
 139         cookie.lgc_lgl = llh->lgh_id;
 140         cookie.lgc_index = hdr->lrh_index;
 141         obd = llh->lgh_ctxt->loc_exp->exp_obd;
 142         inode = llh->lgh_file->f_dentry->d_inode;
 143
 144         /* XXX This is a workaround for the deadlock of changelog adding vs.
 145          * changelog cancelling. Changelog adding always start transaction
 146          * before acquiring the catlog lock (lgh_lock), whereas, changelog
 147          * cancelling do start transaction after holding catlog lock.
 148          *
 149          * We start the transaction earlier here to keep the locking ordering:
 150          * 'start transaction -> catlog lock'. LU-81. */
 151         trans_h = fsfilt_start_log(obd, inode, FSFILT_OP_CANCEL_UNLINK,
 152                                    NULL, 1);
 153         if (IS_ERR(trans_h)) {
 154                 CERROR("fsfilt_start_log failed: %ld\n", PTR_ERR(trans_h));
 155                 RETURN(PTR_ERR(trans_h));
 156         }
 157
 158         /* cancel them one at a time.  I suppose we could store up the cookies
 159            and cancel them all at once; probably more efficient, but this is
 160            done as a user call, so who cares... */
 161         rc = llog_cat_cancel_records(llh->u.phd.phd_cat_handle, 1, &cookie);
 162
 163         err = fsfilt_commit(obd, inode, trans_h, 0);
 164         if (err) {
 165                 CERROR("fsfilt_commit failed: %d\n", err);
 166                 rc = (rc >= 0) ? err : rc;
 167         }
 168
 169         RETURN(rc < 0 ? rc : 0);
 170 }
 171
 172 static int llog_changelog_cancel(struct llog_ctxt *ctxt,
 173                                  struct lov_stripe_md *lsm, int count,
 174                                  struct llog_cookie *cookies, int flags)
 175 {
 176         struct llog_handle *cathandle = ctxt->loc_handle;
 177         int rc;
 178         ENTRY;
 179
 180         /* This should only be called with the catalog handle */
 181         LASSERT(cathandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT);
 182
 183         rc = llog_cat_process(cathandle, llog_changelog_cancel_cb,
 184                               (void *)cookies, 0, 0);
 185         if (rc >= 0)
 186                 /* 0 or 1 means we're done */
 187                 rc = 0;
 188         else
 189                 CERROR("cancel idx %u of catalog "LPX64" rc=%d\n",
 190                        cathandle->lgh_last_idx, cathandle->lgh_id.lgl_oid, rc);
 191
 192         RETURN(rc);
 193 }
 194
 195 int mds_changelog_llog_init(struct obd_device *obd, struct obd_device *tgt)
 196 {
 197         int rc;
 198
 199         /* see osc_llog_init */
 200         changelog_orig_logops = llog_lvfs_ops;
 201         changelog_orig_logops.lop_setup = llog_obd_origin_setup;
 202         changelog_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
 203         changelog_orig_logops.lop_add = llog_obd_origin_add;
 204         changelog_orig_logops.lop_cancel = llog_changelog_cancel;
 205
 206         rc = llog_setup_named(obd, &obd->obd_olg, LLOG_CHANGELOG_ORIG_CTXT,
 207                               tgt, 1, NULL, CHANGELOG_CATALOG,
 208                               &changelog_orig_logops);
 209         if (rc) {
 210                 CERROR("changelog llog setup failed %d\n", rc);
 211                 RETURN(rc);
 212         }
 213
 214         rc = llog_setup_named(obd, &obd->obd_olg, LLOG_CHANGELOG_USER_ORIG_CTXT,
 215                               tgt, 1, NULL, CHANGELOG_USERS,
 216                               &changelog_orig_logops);
 217         if (rc) {
 218                 CERROR("changelog users llog setup failed %d\n", rc);
 219                 RETURN(rc);
 220         }
 221
 222         RETURN(rc);
 223 }
 224 EXPORT_SYMBOL(mds_changelog_llog_init);
 225
 226 int mds_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
 227                   struct obd_device *disk_obd, int *index)
 228 {
 229         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
 230         struct llog_ctxt *ctxt;
 231         int rc;
 232         ENTRY;
 233
 234         LASSERT(olg == &obd->obd_olg);
 235         rc = llog_setup(obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT, disk_obd,
 236                         0, NULL, &mds_ost_orig_logops);
 237         if (rc)
 238                 RETURN(rc);
 239
 240         rc = llog_setup(obd, &obd->obd_olg, LLOG_SIZE_REPL_CTXT, disk_obd,
 241                         0, NULL, &mds_size_repl_logops);
 242         if (rc)
 243                 GOTO(err_llog, rc);
 244
 245         rc = obd_llog_init(lov_obd, &lov_obd->obd_olg, disk_obd, index);
 246         if (rc) {
 247                 CERROR("lov_llog_init err %d\n", rc);
 248                 GOTO(err_cleanup, rc);
 249         }
 250
 251         RETURN(rc);
 252 err_cleanup:
 253         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
 254         if (ctxt)
 255                 llog_cleanup(ctxt);
 256 err_llog:
 257         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
 258         if (ctxt)
 259                 llog_cleanup(ctxt);
 260         return rc;
 261 }
 262
 263 int mds_llog_finish(struct obd_device *obd, int count)
 264 {
 265         struct llog_ctxt *ctxt;
 266         int rc = 0, rc2 = 0;
 267         ENTRY;
 268
 269         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
 270         if (ctxt)
 271                 rc = llog_cleanup(ctxt);
 272
 273         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
 274         if (ctxt)
 275                 rc2 = llog_cleanup(ctxt);
 276         if (!rc)
 277                 rc = rc2;
 278
 279         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
 280         if (ctxt)
 281                 rc2 = llog_cleanup(ctxt);
 282         if (!rc)
 283                 rc = rc2;
 284
 285         ctxt = llog_get_context(obd, LLOG_CHANGELOG_USER_ORIG_CTXT);
 286         if (ctxt)
 287                 rc2 = llog_cleanup(ctxt);
 288         if (!rc)
 289                 rc = rc2;
 290
 291         RETURN(rc);
 292 }
 293
 294 static int mds_llog_add_unlink(struct obd_device *obd,
 295                                struct lov_stripe_md *lsm, obd_count count,
 296                                struct llog_cookie *logcookie, int cookies)
 297 {
 298         struct llog_unlink_rec *lur;
 299         struct llog_ctxt *ctxt;
 300         int rc;
 301
 302         if (cookies < lsm->lsm_stripe_count)
 303                 RETURN(rc = -EFBIG);
 304
 305         /* first prepare unlink log record */
 306         OBD_ALLOC_PTR(lur);
 307         if (!lur)
 308                 RETURN(rc = -ENOMEM);
 309         lur->lur_hdr.lrh_len = lur->lur_tail.lrt_len = sizeof(*lur);
 310         lur->lur_hdr.lrh_type = MDS_UNLINK_REC;
 311         lur->lur_count = count;
 312
 313         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
 314         rc = llog_add(ctxt, &lur->lur_hdr, lsm, logcookie, cookies);
 315         llog_ctxt_put(ctxt);
 316
 317         OBD_FREE_PTR(lur);
 318         RETURN(rc);
 319 }
 320
 321 int mds_log_op_unlink(struct obd_device *obd,
 322                       struct lov_mds_md *lmm, int lmm_size,
 323                       struct llog_cookie *logcookies, int cookies_size)
 324 {
 325         struct mds_obd *mds = &obd->u.mds;
 326         struct lov_stripe_md *lsm = NULL;
 327         int rc;
 328         ENTRY;
 329
 330         if (IS_ERR(mds->mds_lov_obd))
 331                 RETURN(PTR_ERR(mds->mds_lov_obd));
 332
 333         rc = obd_unpackmd(mds->mds_lov_exp, &lsm, lmm, lmm_size);
 334         if (rc < 0)
 335                 RETURN(rc);
 336         rc = mds_llog_add_unlink(obd, lsm, 0, logcookies,
 337                                  cookies_size / sizeof(struct llog_cookie));
 338         obd_free_memmd(mds->mds_lov_exp, &lsm);
 339         RETURN(rc);
 340 }
 341 EXPORT_SYMBOL(mds_log_op_unlink);
 342
 343 int mds_log_op_orphan(struct obd_device *obd, struct lov_stripe_md *lsm,
 344                       obd_count count)
 345 {
 346         struct mds_obd *mds = &obd->u.mds;
 347         struct llog_cookie logcookie;
 348         int rc;
 349         ENTRY;
 350
 351         if (IS_ERR(mds->mds_lov_obd))
 352                 RETURN(PTR_ERR(mds->mds_lov_obd));
 353
 354         rc = mds_llog_add_unlink(obd, lsm, count - 1, &logcookie, 1);
 355         RETURN(rc);
 356 }
 357