Whamcloud - gitweb
LU-1302 llog: pass lu_env as parametr in llog functions
[fs/lustre-release.git] / lustre / mds / mds_log.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, Whamcloud, Inc.
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/mds/mds_log.c
37  *
38  * Author: Peter Braam <braam@clusterfs.com>
39  * Author: Andreas Dilger <adilger@clusterfs.com>
40  * Author: Phil Schwan <phil@clusterfs.com>
41  */
42
43 #define DEBUG_SUBSYSTEM S_MDS
44
45 #include <linux/module.h>
46 #include <linux/version.h>
47
48 #include <libcfs/list.h>
49 #include <obd_class.h>
50 #include <lustre_fsfilt.h>
51 #include <lustre_mds.h>
52 #include <lustre_log.h>
53 #include "mds_internal.h"
54
55 static int mds_llog_origin_add(const struct lu_env *env,
56                                struct llog_ctxt *ctxt,
57                                struct llog_rec_hdr *rec,
58                                struct lov_stripe_md *lsm,
59                                struct llog_cookie *logcookies, int numcookies)
60 {
61         struct obd_device *obd = ctxt->loc_obd;
62         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
63         struct llog_ctxt *lctxt;
64         int rc;
65         ENTRY;
66
67         lctxt = llog_get_context(lov_obd, ctxt->loc_idx);
68         rc = llog_add(env, lctxt, rec, lsm, logcookies, numcookies);
69         llog_ctxt_put(lctxt);
70
71         RETURN(rc);
72 }
73
74 static int mds_llog_origin_connect(struct llog_ctxt *ctxt,
75                                    struct llog_logid *logid,
76                                    struct llog_gen *gen,
77                                    struct obd_uuid *uuid)
78 {
79         struct obd_device *obd = ctxt->loc_obd;
80         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
81         struct llog_ctxt *lctxt;
82         int rc;
83         ENTRY;
84
85         lctxt = llog_get_context(lov_obd, ctxt->loc_idx);
86         rc = llog_connect(lctxt, logid, gen, uuid);
87         llog_ctxt_put(lctxt);
88         RETURN(rc);
89 }
90
91 static struct llog_operations mds_ost_orig_logops = {
92         lop_add:        mds_llog_origin_add,
93         lop_connect:    mds_llog_origin_connect,
94 };
95
96 static int mds_llog_repl_cancel(const struct lu_env *env,
97                                 struct llog_ctxt *ctxt,
98                                 struct lov_stripe_md *lsm,
99                                 int count, struct llog_cookie *cookies,
100                                 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(env, 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(const struct lu_env *env,
121                                     struct llog_handle *llh,
122                                     struct llog_rec_hdr *hdr, void *data)
123 {
124         struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
125         struct llog_cookie cookie;
126         long long endrec = *(long long *)data;
127         int rc, err;
128         struct obd_device *obd;
129         void *trans_h;
130         struct inode *inode;
131         ENTRY;
132
133         /* This is always a (sub)log, not the catalog */
134         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
135
136         if (rec->cr.cr_index > endrec)
137                 /* records are in order, so we're done */
138                 RETURN(LLOG_PROC_BREAK);
139
140         cookie.lgc_lgl = llh->lgh_id;
141         cookie.lgc_index = hdr->lrh_index;
142         obd = llh->lgh_ctxt->loc_exp->exp_obd;
143         inode = llh->lgh_file->f_dentry->d_inode;
144
145         /* XXX This is a workaround for the deadlock of changelog adding vs.
146          * changelog cancelling. Changelog adding always start transaction
147          * before acquiring the catlog lock (lgh_lock), whereas, changelog
148          * cancelling do start transaction after holding catlog lock.
149          *
150          * We start the transaction earlier here to keep the locking ordering:
151          * 'start transaction -> catlog lock'. LU-81. */
152         trans_h = fsfilt_start_log(obd, inode, FSFILT_OP_CANCEL_UNLINK,
153                                    NULL, 1);
154         if (IS_ERR(trans_h)) {
155                 CERROR("fsfilt_start_log failed: %ld\n", PTR_ERR(trans_h));
156                 RETURN(PTR_ERR(trans_h));
157         }
158
159         /* cancel them one at a time.  I suppose we could store up the cookies
160            and cancel them all at once; probably more efficient, but this is
161            done as a user call, so who cares... */
162         rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1,
163                                      &cookie);
164
165         err = fsfilt_commit(obd, inode, trans_h, 0);
166         if (err) {
167                 CERROR("fsfilt_commit failed: %d\n", err);
168                 rc = (rc >= 0) ? err : rc;
169         }
170
171         RETURN(rc < 0 ? rc : 0);
172 }
173
174 static int llog_changelog_cancel(const struct lu_env *env,
175                                  struct llog_ctxt *ctxt,
176                                  struct lov_stripe_md *lsm, int count,
177                                  struct llog_cookie *cookies, int flags)
178 {
179         struct llog_handle *cathandle = ctxt->loc_handle;
180         int rc;
181         ENTRY;
182
183         /* This should only be called with the catalog handle */
184         LASSERT(cathandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT);
185
186         rc = llog_cat_process(env, cathandle, llog_changelog_cancel_cb,
187                               (void *)cookies, 0, 0);
188         if (rc >= 0)
189                 /* 0 or 1 means we're done */
190                 rc = 0;
191         else
192                 CERROR("cancel idx %u of catalog "LPX64" rc=%d\n",
193                        cathandle->lgh_last_idx, cathandle->lgh_id.lgl_oid, rc);
194
195         RETURN(rc);
196 }
197
198 int mds_changelog_llog_init(struct obd_device *obd, struct obd_device *tgt)
199 {
200         int rc;
201
202         /* see osc_llog_init */
203         changelog_orig_logops = llog_lvfs_ops;
204         changelog_orig_logops.lop_setup = llog_obd_origin_setup;
205         changelog_orig_logops.lop_cleanup = llog_obd_origin_cleanup;
206         changelog_orig_logops.lop_add = llog_obd_origin_add;
207         changelog_orig_logops.lop_cancel = llog_changelog_cancel;
208
209         rc = llog_setup_named(obd, &obd->obd_olg, LLOG_CHANGELOG_ORIG_CTXT,
210                               tgt, 1, NULL, CHANGELOG_CATALOG,
211                               &changelog_orig_logops);
212         if (rc) {
213                 CERROR("changelog llog setup failed %d\n", rc);
214                 RETURN(rc);
215         }
216
217         rc = llog_setup_named(obd, &obd->obd_olg, LLOG_CHANGELOG_USER_ORIG_CTXT,
218                               tgt, 1, NULL, CHANGELOG_USERS,
219                               &changelog_orig_logops);
220         if (rc) {
221                 CERROR("changelog users llog setup failed %d\n", rc);
222                 RETURN(rc);
223         }
224
225         RETURN(rc);
226 }
227 EXPORT_SYMBOL(mds_changelog_llog_init);
228
229 int mds_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
230                   struct obd_device *disk_obd, int *index)
231 {
232         struct obd_device *lov_obd = obd->u.mds.mds_lov_obd;
233         struct llog_ctxt *ctxt;
234         int rc;
235         ENTRY;
236
237         LASSERT(olg == &obd->obd_olg);
238         rc = llog_setup(obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT, disk_obd,
239                         0, NULL, &mds_ost_orig_logops);
240         if (rc)
241                 RETURN(rc);
242
243         rc = llog_setup(obd, &obd->obd_olg, LLOG_SIZE_REPL_CTXT, disk_obd,
244                         0, NULL, &mds_size_repl_logops);
245         if (rc)
246                 GOTO(err_llog, rc);
247
248         rc = obd_llog_init(lov_obd, &lov_obd->obd_olg, disk_obd, index);
249         if (rc) {
250                 CERROR("lov_llog_init err %d\n", rc);
251                 GOTO(err_cleanup, rc);
252         }
253
254         RETURN(rc);
255 err_cleanup:
256         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
257         if (ctxt)
258                 llog_cleanup(ctxt);
259 err_llog:
260         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
261         if (ctxt)
262                 llog_cleanup(ctxt);
263         return rc;
264 }
265
266 int mds_llog_finish(struct obd_device *obd, int count)
267 {
268         struct llog_ctxt *ctxt;
269         int rc = 0, rc2 = 0;
270         ENTRY;
271
272         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
273         if (ctxt)
274                 rc = llog_cleanup(ctxt);
275
276         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
277         if (ctxt)
278                 rc2 = llog_cleanup(ctxt);
279         if (!rc)
280                 rc = rc2;
281
282         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
283         if (ctxt)
284                 rc2 = llog_cleanup(ctxt);
285         if (!rc)
286                 rc = rc2;
287
288         ctxt = llog_get_context(obd, LLOG_CHANGELOG_USER_ORIG_CTXT);
289         if (ctxt)
290                 rc2 = llog_cleanup(ctxt);
291         if (!rc)
292                 rc = rc2;
293
294         RETURN(rc);
295 }
296
297 static int mds_llog_add_unlink(struct obd_device *obd,
298                                struct lov_stripe_md *lsm, obd_count count,
299                                struct llog_cookie *logcookie, int cookies)
300 {
301         struct llog_unlink_rec *lur;
302         struct llog_ctxt *ctxt;
303         int rc;
304
305         if (cookies < lsm->lsm_stripe_count)
306                 RETURN(rc = -EFBIG);
307
308         /* first prepare unlink log record */
309         OBD_ALLOC_PTR(lur);
310         if (!lur)
311                 RETURN(rc = -ENOMEM);
312         lur->lur_hdr.lrh_len = lur->lur_tail.lrt_len = sizeof(*lur);
313         lur->lur_hdr.lrh_type = MDS_UNLINK_REC;
314         lur->lur_count = count;
315
316         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
317         rc = llog_add(NULL, ctxt, &lur->lur_hdr, lsm, logcookie, cookies);
318         llog_ctxt_put(ctxt);
319
320         OBD_FREE_PTR(lur);
321         RETURN(rc);
322 }
323
324 int mds_log_op_unlink(struct obd_device *obd,
325                       struct lov_mds_md *lmm, int lmm_size,
326                       struct llog_cookie *logcookies, int cookies_size)
327 {
328         struct mds_obd *mds = &obd->u.mds;
329         struct lov_stripe_md *lsm = NULL;
330         int rc;
331         ENTRY;
332
333         if (IS_ERR(mds->mds_lov_obd))
334                 RETURN(PTR_ERR(mds->mds_lov_obd));
335
336         rc = obd_unpackmd(mds->mds_lov_exp, &lsm, lmm, lmm_size);
337         if (rc < 0)
338                 RETURN(rc);
339         rc = mds_llog_add_unlink(obd, lsm, 0, logcookies,
340                                  cookies_size / sizeof(struct llog_cookie));
341         obd_free_memmd(mds->mds_lov_exp, &lsm);
342         RETURN(rc);
343 }
344 EXPORT_SYMBOL(mds_log_op_unlink);
345
346 int mds_log_op_orphan(struct obd_device *obd, struct lov_stripe_md *lsm,
347                       obd_count count)
348 {
349         struct mds_obd *mds = &obd->u.mds;
350         struct llog_cookie logcookie;
351         int rc;
352         ENTRY;
353
354         if (IS_ERR(mds->mds_lov_obd))
355                 RETURN(PTR_ERR(mds->mds_lov_obd));
356
357         rc = mds_llog_add_unlink(obd, lsm, count - 1, &logcookie, 1);
358         RETURN(rc);
359 }
360