Whamcloud - gitweb
e4146dc965c215c00390fd935a4fd610578630fe
[fs/lustre-release.git] / lustre / obdclass / llog.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  Copyright (C) 2001-2003 Cluster File Systems, Inc.
5  *   Author: Andreas Dilger <adilger@clusterfs.com>
6  *
7  *   This file is part of Lustre, http://www.lustre.org.
8  *
9  *   Lustre is free software; you can redistribute it and/or
10  *   modify it under the terms of version 2 of the GNU General Public
11  *   License as published by the Free Software Foundation.
12  *
13  *   Lustre is distributed in the hope that it will be useful,
14  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
15  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16  *   GNU General Public License for more details.
17  *
18  *   You should have received a copy of the GNU General Public License
19  *   along with Lustre; if not, write to the Free Software
20  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
21  *
22  * OST<->MDS recovery logging infrastructure.
23  *
24  * Invariants in implementation:
25  * - we do not share logs among different OST<->MDS connections, so that
26  *   if an OST or MDS fails it need only look at log(s) relevant to itself
27  */
28
29 #define DEBUG_SUBSYSTEM S_LOG
30
31 #ifndef EXPORT_SYMTAB
32 #define EXPORT_SYMTAB
33 #endif
34
35 #ifdef __KERNEL__
36 #include <linux/fs.h>
37 #else
38 #include <liblustre.h>
39 #endif
40
41 #include <linux/obd_class.h>
42 #include <linux/lustre_log.h>
43 #include <portals/list.h>
44
45 /* Allocate a new log or catalog handle */
46 struct llog_handle *llog_alloc_handle(void)
47 {
48         struct llog_handle *loghandle;
49         ENTRY;
50
51         OBD_ALLOC(loghandle, sizeof(*loghandle));
52         if (loghandle == NULL)
53                 RETURN(ERR_PTR(-ENOMEM));
54
55         init_rwsem(&loghandle->lgh_lock);
56
57         RETURN(loghandle);
58 }
59 EXPORT_SYMBOL(llog_alloc_handle);
60
61
62 void llog_free_handle(struct llog_handle *loghandle)
63 {
64         if (!loghandle)
65                 return;
66
67         if (!loghandle->lgh_hdr)
68                 goto out;
69         if (le32_to_cpu(loghandle->lgh_hdr->llh_flags) & LLOG_F_IS_PLAIN)
70                 list_del_init(&loghandle->u.phd.phd_entry);
71         if (le32_to_cpu(loghandle->lgh_hdr->llh_flags) & LLOG_F_IS_CAT)
72                 LASSERT(list_empty(&loghandle->u.chd.chd_head));
73         OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);
74
75  out:
76         OBD_FREE(loghandle, sizeof(*loghandle));
77 }
78 EXPORT_SYMBOL(llog_free_handle);
79
80 /* returns negative on error; 0 if success; 1 if success & log destroyed */
81 int llog_cancel_rec(struct llog_handle *loghandle, int index)
82 {
83         struct llog_log_hdr *llh = loghandle->lgh_hdr;
84         int rc = 0;
85         ENTRY;
86
87         CDEBUG(D_HA, "canceling %d in log "LPX64"\n",
88                index, loghandle->lgh_id.lgl_oid);
89
90         if (index == 0) {
91                 CERROR("cannot cancel index 0 (which is header)\n");
92                 RETURN(-EINVAL);
93         }
94
95         if (!ext2_clear_bit(index, llh->llh_bitmap)) {
96                 CERROR("catalog index %u already clear?\n", index);
97                 RETURN(-EINVAL);
98         }
99
100         llh->llh_count = cpu_to_le32(le32_to_cpu(llh->llh_count) - 1);
101
102         if ((le32_to_cpu(llh->llh_flags) & LLOG_F_ZAP_WHEN_EMPTY) &&
103             (le32_to_cpu(llh->llh_count) == 1) &&
104             (loghandle->lgh_last_idx == (LLOG_BITMAP_BYTES * 8) - 1)) {
105                 rc = llog_destroy(loghandle);
106                 if (rc)
107                         CERROR("failure destroying log after last cancel: %d\n",
108                                rc);
109                 LASSERT(rc == 0);
110                 RETURN(1);
111         }
112
113         rc = llog_write_rec(loghandle, &llh->llh_hdr, NULL, 0, NULL, 0);
114         if (rc)
115                 CERROR("failure re-writing header %d\n", rc);
116         LASSERT(rc == 0);
117         RETURN(rc);
118 }
119 EXPORT_SYMBOL(llog_cancel_rec);
120
121 int llog_init_handle(struct llog_handle *handle, int flags,
122                      struct obd_uuid *uuid)
123 {
124         int rc;
125         struct llog_log_hdr *llh;
126         ENTRY;
127         LASSERT(handle->lgh_hdr == NULL);
128
129         OBD_ALLOC(llh, sizeof(*llh));
130         if (llh == NULL)
131                 RETURN(-ENOMEM);
132         handle->lgh_hdr = llh;
133         /* first assign flags to use llog_client_ops */
134         llh->llh_flags = cpu_to_le32(flags);
135         rc = llog_read_header(handle);
136         if (rc == 0) {
137                 flags = le32_to_cpu(llh->llh_flags);
138                 if (uuid)
139                         LASSERT(obd_uuid_equals(uuid, &llh->llh_tgtuuid));
140                 GOTO(out, rc);
141         } else if (rc != LLOG_EEMPTY || !flags) {
142                 /* set a pesudo flag for initialization */
143                 flags = LLOG_F_IS_CAT;
144                 GOTO(out, rc);
145         }
146         rc = 0;
147
148         handle->lgh_last_idx = 0; /* header is record with index 0 */
149         llh->llh_count = cpu_to_le32(1);         /* for the header record */
150         llh->llh_hdr.lrh_type = cpu_to_le32(LLOG_HDR_MAGIC);
151         llh->llh_hdr.lrh_len = llh->llh_tail.lrt_len =
152                 cpu_to_le32(LLOG_CHUNK_SIZE);
153         llh->llh_hdr.lrh_index = llh->llh_tail.lrt_index = 0;
154         llh->llh_timestamp = cpu_to_le64(LTIME_S(CURRENT_TIME));
155         if (uuid)
156                 memcpy(&llh->llh_tgtuuid, uuid, sizeof(llh->llh_tgtuuid));
157         llh->llh_bitmap_offset = cpu_to_le32(offsetof(typeof(*llh),llh_bitmap));
158         ext2_set_bit(0, llh->llh_bitmap);
159
160 out:
161         if (flags & LLOG_F_IS_CAT) {
162                 INIT_LIST_HEAD(&handle->u.chd.chd_head);
163                 llh->llh_size = cpu_to_le32(sizeof(struct llog_logid_rec));
164         }
165         else if (flags & LLOG_F_IS_PLAIN)
166                 INIT_LIST_HEAD(&handle->u.phd.phd_entry);
167         else
168                 LBUG();
169
170         if (rc) {
171                 OBD_FREE(llh, sizeof(*llh));
172                 handle->lgh_hdr = NULL;
173         }
174         RETURN(rc);
175 }
176 EXPORT_SYMBOL(llog_init_handle);
177
178 int llog_close(struct llog_handle *loghandle)
179 {
180         struct llog_operations *lop;
181         int rc;
182         ENTRY;
183
184         rc = llog_handle2ops(loghandle, &lop);
185         if (rc)
186                 GOTO(out, rc);
187         if (lop->lop_close == NULL)
188                 GOTO(out, -EOPNOTSUPP);
189         rc = lop->lop_close(loghandle);
190  out:
191         llog_free_handle(loghandle);
192         RETURN(rc);
193 }
194 EXPORT_SYMBOL(llog_close);
195
196 int llog_process(struct llog_handle *loghandle, llog_cb_t cb,
197                  void *data, void *catdata)
198 {
199         struct llog_log_hdr *llh = loghandle->lgh_hdr;
200         struct llog_process_cat_data *cd = catdata;
201         void *buf;
202         __u64 cur_offset = LLOG_CHUNK_SIZE;
203         int rc = 0, index = 1, last_index, idx;
204         int saved_index = 0;
205         ENTRY;
206
207         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
208         if (!buf)
209                 RETURN(-ENOMEM);
210
211         if (cd != NULL)
212                 index = cd->first_idx + 1;
213         if (cd != NULL && cd->last_idx)
214                 last_index = cd->last_idx;
215         else
216                 last_index = LLOG_BITMAP_BYTES * 8 - 1;
217
218         while (rc == 0) {
219                 struct llog_rec_hdr *rec;
220
221                 /* skip records not set in bitmap */
222                 while (index <= last_index &&
223                        !ext2_test_bit(index, llh->llh_bitmap))
224                         ++index;
225
226                 LASSERT(index <= last_index + 1);
227                 if (index == last_index + 1)
228                         break;
229
230                 /* get the buf with our target record; avoid old garbage */
231                 memset(buf, 0, LLOG_CHUNK_SIZE);
232                 rc = llog_next_block(loghandle, &saved_index, index,
233                                      &cur_offset, buf, LLOG_CHUNK_SIZE);
234                 if (rc)
235                         GOTO(out, rc);
236
237                 rec = buf;
238                 idx = le32_to_cpu(rec->lrh_index);
239                 if (idx < index)
240                         CDEBUG(D_HA, "index %u : idx %u\n", index, idx);
241                 while (idx < index) {
242                         rec = ((void *)rec + le32_to_cpu(rec->lrh_len));
243                         idx ++;
244                 }
245
246                 /* process records in buffer, starting where we found one */
247                 while ((void *)rec < buf + LLOG_CHUNK_SIZE) {
248                         if (rec->lrh_index == 0)
249                                 GOTO(out, 0); /* no more records */
250
251                         /* if set, process the callback on this record */
252                         if (ext2_test_bit(index, llh->llh_bitmap)) {
253                                 rc = cb(loghandle, rec, data);
254                                 if (rc == LLOG_PROC_BREAK) {
255                                         CWARN("recovery from log: "LPX64":%x"
256                                               " stopped\n",
257                                               loghandle->lgh_id.lgl_oid,
258                                               loghandle->lgh_id.lgl_ogen);
259                                         GOTO(out, rc);
260                                 }
261                                 if (rc)
262                                         GOTO(out, rc);
263                         }
264
265                         /* next record, still in buffer? */
266                         ++index;
267                         if (index > last_index)
268                                 GOTO(out, rc = 0);
269                         rec = ((void *)rec + le32_to_cpu(rec->lrh_len));
270                 }
271         }
272
273  out:
274         if (buf)
275                 OBD_FREE(buf, LLOG_CHUNK_SIZE);
276         RETURN(rc);
277 }
278 EXPORT_SYMBOL(llog_process);