Whamcloud - gitweb
a1d5de22b530814f15b56504757aaff19ea75dd8
[fs/lustre-release.git] / lustre / obdclass / llog_obd.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
39 #define DEBUG_SUBSYSTEM S_LOG
40
41 #ifndef EXPORT_SYMTAB
42 #define EXPORT_SYMTAB
43 #endif
44
45 #ifndef __KERNEL__
46 #include <liblustre.h>
47 #endif
48
49 #include <obd_class.h>
50 #include <lustre_log.h>
51 #include <libcfs/list.h>
52 #include "llog_internal.h"
53
54 /* helper functions for calling the llog obd methods */
55 static struct llog_ctxt* llog_new_ctxt(struct obd_device *obd)
56 {
57         struct llog_ctxt *ctxt;
58
59         OBD_ALLOC_PTR(ctxt);
60         if (!ctxt)
61                 return NULL;
62
63         ctxt->loc_obd = obd;
64         cfs_atomic_set(&ctxt->loc_refcount, 1);
65
66         return ctxt;
67 }
68
69 static void llog_ctxt_destroy(struct llog_ctxt *ctxt)
70 {
71         if (ctxt->loc_exp) {
72                 class_export_put(ctxt->loc_exp);
73                 ctxt->loc_exp = NULL;
74         }
75         if (ctxt->loc_imp) {
76                 class_import_put(ctxt->loc_imp);
77                 ctxt->loc_imp = NULL;
78         }
79         LASSERT(ctxt->loc_llcd == NULL);
80         OBD_FREE_PTR(ctxt);
81 }
82
83 int __llog_ctxt_put(struct llog_ctxt *ctxt)
84 {
85         struct obd_llog_group *olg = ctxt->loc_olg;
86         struct obd_device *obd;
87         int rc = 0;
88
89         cfs_spin_lock(&olg->olg_lock);
90         if (!cfs_atomic_dec_and_test(&ctxt->loc_refcount)) {
91                 cfs_spin_unlock(&olg->olg_lock);
92                 return rc;
93         }
94         olg->olg_ctxts[ctxt->loc_idx] = NULL;
95         cfs_spin_unlock(&olg->olg_lock);
96
97         if (ctxt->loc_lcm)
98                 lcm_put(ctxt->loc_lcm);
99
100         obd = ctxt->loc_obd;
101         cfs_spin_lock(&obd->obd_dev_lock);
102         /* sync with llog ctxt user thread */
103         cfs_spin_unlock(&obd->obd_dev_lock);
104
105         /* obd->obd_starting is needed for the case of cleanup
106          * in error case while obd is starting up. */
107         LASSERTF(obd->obd_starting == 1 ||
108                  obd->obd_stopping == 1 || obd->obd_set_up == 0,
109                  "wrong obd state: %d/%d/%d\n", !!obd->obd_starting,
110                  !!obd->obd_stopping, !!obd->obd_set_up);
111
112         /* cleanup the llog ctxt here */
113         if (CTXTP(ctxt, cleanup))
114                 rc = CTXTP(ctxt, cleanup)(ctxt);
115
116         llog_ctxt_destroy(ctxt);
117         cfs_waitq_signal(&olg->olg_waitq);
118         return rc;
119 }
120 EXPORT_SYMBOL(__llog_ctxt_put);
121
122 int llog_cleanup(struct llog_ctxt *ctxt)
123 {
124         struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
125         struct obd_llog_group *olg;
126         int rc, idx;
127         ENTRY;
128
129         LASSERT(ctxt != NULL);
130         LASSERT(ctxt != LP_POISON);
131
132         olg = ctxt->loc_olg;
133         LASSERT(olg != NULL);
134         LASSERT(olg != LP_POISON);
135
136         idx = ctxt->loc_idx;
137
138         /* 
139          * Banlance the ctxt get when calling llog_cleanup()
140          */
141         LASSERT(cfs_atomic_read(&ctxt->loc_refcount) < LI_POISON);
142         LASSERT(cfs_atomic_read(&ctxt->loc_refcount) > 1);
143         llog_ctxt_put(ctxt);
144
145         /* 
146          * Try to free the ctxt. 
147          */
148         rc = __llog_ctxt_put(ctxt);
149         if (rc)
150                 CERROR("Error %d while cleaning up ctxt %p\n",
151                        rc, ctxt);
152
153         l_wait_event(olg->olg_waitq,
154                      llog_group_ctxt_null(olg, idx), &lwi);
155
156         RETURN(rc);
157 }
158 EXPORT_SYMBOL(llog_cleanup);
159
160 int llog_setup_named(struct obd_device *obd,  struct obd_llog_group *olg,
161                      int index, struct obd_device *disk_obd, int count,
162                      struct llog_logid *logid, const char *logname,
163                      struct llog_operations *op)
164 {
165         struct llog_ctxt *ctxt;
166         int rc = 0;
167         ENTRY;
168
169         if (index < 0 || index >= LLOG_MAX_CTXTS)
170                 RETURN(-EINVAL);
171
172         LASSERT(olg != NULL);
173
174         ctxt = llog_new_ctxt(obd);
175         if (!ctxt)
176                 RETURN(-ENOMEM);
177
178         ctxt->loc_obd = obd;
179         ctxt->loc_olg = olg;
180         ctxt->loc_idx = index;
181         ctxt->loc_logops = op;
182         cfs_mutex_init(&ctxt->loc_mutex);
183         ctxt->loc_exp = class_export_get(disk_obd->obd_self_export);
184         ctxt->loc_flags = LLOG_CTXT_FLAG_UNINITIALIZED;
185
186         rc = llog_group_set_ctxt(olg, ctxt, index);
187         if (rc) {
188                 llog_ctxt_destroy(ctxt);
189                 if (rc == -EEXIST) {
190                         ctxt = llog_group_get_ctxt(olg, index);
191                         if (ctxt) {
192                                 /*
193                                  * mds_lov_update_desc() might call here multiple
194                                  * times. So if the llog is already set up then
195                                  * don't to do it again. 
196                                  */
197                                 CDEBUG(D_CONFIG, "obd %s ctxt %d already set up\n",
198                                        obd->obd_name, index);
199                                 LASSERT(ctxt->loc_olg == olg);
200                                 LASSERT(ctxt->loc_obd == obd);
201                                 LASSERT(ctxt->loc_exp == disk_obd->obd_self_export);
202                                 LASSERT(ctxt->loc_logops == op);
203                                 llog_ctxt_put(ctxt);
204                         }
205                         rc = 0;
206                 }
207                 RETURN(rc);
208         }
209
210         if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LLOG_SETUP)) {
211                 rc = -ENOTSUPP;
212         } else {
213                 if (op->lop_setup)
214                         rc = op->lop_setup(obd, olg, index, disk_obd, count,
215                                            logid, logname);
216         }
217
218         if (rc) {
219                 CERROR("obd %s ctxt %d lop_setup=%p failed %d\n",
220                        obd->obd_name, index, op->lop_setup, rc);
221                 llog_ctxt_put(ctxt);
222         } else {
223                 CDEBUG(D_CONFIG, "obd %s ctxt %d is initialized\n",
224                        obd->obd_name, index);
225                 ctxt->loc_flags &= ~LLOG_CTXT_FLAG_UNINITIALIZED;
226         }
227
228         RETURN(rc);
229 }
230 EXPORT_SYMBOL(llog_setup_named);
231
232 int llog_setup(struct obd_device *obd,  struct obd_llog_group *olg,
233                int index, struct obd_device *disk_obd, int count,
234                struct llog_logid *logid, struct llog_operations *op)
235 {
236         return llog_setup_named(obd,olg,index,disk_obd,count,logid,NULL,op);
237 }
238 EXPORT_SYMBOL(llog_setup);
239
240 int llog_sync(struct llog_ctxt *ctxt, struct obd_export *exp)
241 {
242         int rc = 0;
243         ENTRY;
244
245         if (!ctxt)
246                 RETURN(0);
247
248         if (CTXTP(ctxt, sync))
249                 rc = CTXTP(ctxt, sync)(ctxt, exp);
250
251         RETURN(rc);
252 }
253 EXPORT_SYMBOL(llog_sync);
254
255 int llog_add(struct llog_ctxt *ctxt, struct llog_rec_hdr *rec,
256              struct lov_stripe_md *lsm, struct llog_cookie *logcookies,
257              int numcookies)
258 {
259         int raised, rc;
260         ENTRY;
261
262         if (!ctxt) {
263                 CERROR("No ctxt\n");
264                 RETURN(-ENODEV);
265         }
266
267         if (ctxt->loc_flags & LLOG_CTXT_FLAG_UNINITIALIZED)
268                 RETURN(-ENXIO);
269
270
271         CTXT_CHECK_OP(ctxt, add, -EOPNOTSUPP);
272         raised = cfs_cap_raised(CFS_CAP_SYS_RESOURCE);
273         if (!raised)
274                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
275         rc = CTXTP(ctxt, add)(ctxt, rec, lsm, logcookies, numcookies);
276         if (!raised)
277                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
278         RETURN(rc);
279 }
280 EXPORT_SYMBOL(llog_add);
281
282 int llog_cancel(struct llog_ctxt *ctxt, struct lov_stripe_md *lsm,
283                 int count, struct llog_cookie *cookies, int flags)
284 {
285         int rc;
286         ENTRY;
287
288         if (!ctxt) {
289                 CERROR("No ctxt\n");
290                 RETURN(-ENODEV);
291         }
292
293         CTXT_CHECK_OP(ctxt, cancel, -EOPNOTSUPP);
294         rc = CTXTP(ctxt, cancel)(ctxt, lsm, count, cookies, flags);
295         RETURN(rc);
296 }
297 EXPORT_SYMBOL(llog_cancel);
298
299 /* callback func for llog_process in llog_obd_origin_setup */
300 static int cat_cancel_cb(struct llog_handle *cathandle,
301                           struct llog_rec_hdr *rec, void *data)
302 {
303         struct llog_logid_rec *lir = (struct llog_logid_rec *)rec;
304         struct llog_handle *loghandle;
305         struct llog_log_hdr *llh;
306         int rc, index;
307         ENTRY;
308
309         if (rec->lrh_type != LLOG_LOGID_MAGIC) {
310                 CERROR("invalid record in catalog\n");
311                 RETURN(-EINVAL);
312         }
313         CDEBUG(D_HA, "processing log "LPX64":%x at index %u of catalog "
314                LPX64"\n", lir->lid_id.lgl_oid, lir->lid_id.lgl_ogen,
315                rec->lrh_index, cathandle->lgh_id.lgl_oid);
316
317         rc = llog_cat_id2handle(cathandle, &loghandle, &lir->lid_id);
318         if (rc) {
319                 CERROR("Cannot find handle for log "LPX64"\n",
320                        lir->lid_id.lgl_oid);
321                 if (rc == -ENOENT) {
322                         index = rec->lrh_index;
323                         goto cat_cleanup;
324                 }
325                 RETURN(rc);
326         }
327
328         llh = loghandle->lgh_hdr;
329         if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
330             (llh->llh_count == 1)) {
331                 rc = llog_destroy(loghandle);
332                 if (rc)
333                         CERROR("failure destroying log in postsetup: %d\n", rc);
334
335                 index = loghandle->u.phd.phd_cookie.lgc_index;
336                 llog_free_handle(loghandle);
337
338 cat_cleanup:
339                 LASSERT(index);
340                 llog_cat_set_first_idx(cathandle, index);
341                 rc = llog_cancel_rec(cathandle, index);
342                 if (rc == 0)
343                         CDEBUG(D_HA, "cancel log "LPX64":%x at index %u of catalog "
344                               LPX64"\n", lir->lid_id.lgl_oid,
345                               lir->lid_id.lgl_ogen, rec->lrh_index,
346                               cathandle->lgh_id.lgl_oid);
347         }
348
349         RETURN(rc);
350 }
351
352 /* lop_setup method for filter/osc */
353 // XXX how to set exports
354 int llog_obd_origin_setup(struct obd_device *obd, struct obd_llog_group *olg,
355                           int index, struct obd_device *disk_obd, int count,
356                           struct llog_logid *logid, const char *name)
357 {
358         struct llog_ctxt *ctxt;
359         struct llog_handle *handle;
360         struct lvfs_run_ctxt saved;
361         int rc;
362         ENTRY;
363
364         if (count == 0)
365                 RETURN(0);
366
367         LASSERT(count == 1);
368
369         LASSERT(olg != NULL);
370         ctxt = llog_group_get_ctxt(olg, index);
371         if (!ctxt)
372                 RETURN(-ENODEV);
373
374         if (logid && logid->lgl_oid) {
375                 rc = llog_create(ctxt, &handle, logid, NULL);
376         } else {
377                 rc = llog_create(ctxt, &handle, NULL, (char *)name);
378                 if (!rc && logid)
379                         *logid = handle->lgh_id;
380         }
381         if (rc)
382                 GOTO(out, rc);
383
384         ctxt->loc_handle = handle;
385         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
386         rc = llog_init_handle(handle, LLOG_F_IS_CAT, NULL);
387         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
388         if (rc)
389                 GOTO(out, rc);
390
391         rc = llog_process(handle, (llog_cb_t)cat_cancel_cb, NULL, NULL);
392         if (rc)
393                 CERROR("llog_process() with cat_cancel_cb failed: %d\n", rc);
394         GOTO(out, rc);
395 out:
396         llog_ctxt_put(ctxt);
397         return rc;
398 }
399 EXPORT_SYMBOL(llog_obd_origin_setup);
400
401 int llog_obd_origin_cleanup(struct llog_ctxt *ctxt)
402 {
403         struct llog_handle *cathandle, *n, *loghandle;
404         struct llog_log_hdr *llh;
405         int rc, index;
406         ENTRY;
407
408         if (!ctxt)
409                 RETURN(0);
410
411         cathandle = ctxt->loc_handle;
412         if (cathandle) {
413                 cfs_list_for_each_entry_safe(loghandle, n,
414                                              &cathandle->u.chd.chd_head,
415                                              u.phd.phd_entry) {
416                         llh = loghandle->lgh_hdr;
417                         if ((llh->llh_flags &
418                                 LLOG_F_ZAP_WHEN_EMPTY) &&
419                             (llh->llh_count == 1)) {
420                                 rc = llog_destroy(loghandle);
421                                 if (rc)
422                                         CERROR("failure destroying log during "
423                                                "cleanup: %d\n", rc);
424
425                                 index = loghandle->u.phd.phd_cookie.lgc_index;
426                                 llog_free_handle(loghandle);
427
428                                 LASSERT(index);
429                                 llog_cat_set_first_idx(cathandle, index);
430                                 rc = llog_cancel_rec(cathandle, index);
431                                 if (rc == 0)
432                                         CDEBUG(D_RPCTRACE, "cancel plain log at"
433                                                "index %u of catalog "LPX64"\n",
434                                                index,cathandle->lgh_id.lgl_oid);
435                         }
436                 }
437                 llog_cat_put(ctxt->loc_handle);
438         }
439         RETURN(0);
440 }
441 EXPORT_SYMBOL(llog_obd_origin_cleanup);
442
443 /* add for obdfilter/sz and mds/unlink */
444 int llog_obd_origin_add(struct llog_ctxt *ctxt,
445                         struct llog_rec_hdr *rec, struct lov_stripe_md *lsm,
446                         struct llog_cookie *logcookies, int numcookies)
447 {
448         struct llog_handle *cathandle;
449         int rc;
450         ENTRY;
451
452         cathandle = ctxt->loc_handle;
453         LASSERT(cathandle != NULL);
454         rc = llog_cat_add_rec(cathandle, rec, logcookies, NULL);
455         if (rc != 0 && rc != 1)
456                 CERROR("write one catalog record failed: %d\n", rc);
457         RETURN(rc);
458 }
459 EXPORT_SYMBOL(llog_obd_origin_add);
460
461 int obd_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
462                   struct obd_device *disk_obd, int *index)
463 {
464         int rc;
465         ENTRY;
466         OBD_CHECK_DT_OP(obd, llog_init, 0);
467         OBD_COUNTER_INCREMENT(obd, llog_init);
468
469         rc = OBP(obd, llog_init)(obd, olg, disk_obd, index);
470         RETURN(rc);
471 }
472 EXPORT_SYMBOL(obd_llog_init);
473
474 int obd_llog_finish(struct obd_device *obd, int count)
475 {
476         int rc;
477         ENTRY;
478         OBD_CHECK_DT_OP(obd, llog_finish, 0);
479         OBD_COUNTER_INCREMENT(obd, llog_finish);
480
481         rc = OBP(obd, llog_finish)(obd, count);
482         RETURN(rc);
483 }
484 EXPORT_SYMBOL(obd_llog_finish);