Whamcloud - gitweb
b=19427 correct lmm_object_id and reserve fids for fid-on-OST.
[fs/lustre-release.git] / lustre / lov / lov_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  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
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/lov/lov_log.c
37  *
38  * Author: Phil Schwan <phil@clusterfs.com>
39  * Author: Peter Braam <braam@clusterfs.com>
40  * Author: Mike Shaver <shaver@clusterfs.com>
41  */
42
43 #ifndef EXPORT_SYMTAB
44 # define EXPORT_SYMTAB
45 #endif
46 #define DEBUG_SUBSYSTEM S_LOV
47 #ifdef __KERNEL__
48 #include <libcfs/libcfs.h>
49 #else
50 #include <liblustre.h>
51 #endif
52
53 #include <obd_support.h>
54 #include <lustre_lib.h>
55 #include <lustre_net.h>
56 #include <lustre/lustre_idl.h>
57 #include <lustre_dlm.h>
58 #include <lustre_mds.h>
59 #include <obd_class.h>
60 #include <obd_lov.h>
61 #include <obd_ost.h>
62 #include <lprocfs_status.h>
63
64 #include "lov_internal.h"
65
66 /* Add log records for each OSC that this object is striped over, and return
67  * cookies for each one.  We _would_ have nice abstraction here, except that
68  * we need to keep cookies in stripe order, even if some are NULL, so that
69  * the right cookies are passed back to the right OSTs at the client side.
70  * Unset cookies should be all-zero (which will never occur naturally). */
71 static int lov_llog_origin_add(struct llog_ctxt *ctxt, struct llog_rec_hdr *rec,
72                                struct lov_stripe_md *lsm,
73                                struct llog_cookie *logcookies, int numcookies)
74 {
75         struct obd_device *obd = ctxt->loc_obd;
76         struct lov_obd *lov = &obd->u.lov;
77         int i, rc = 0, cookies = 0;
78         ENTRY;
79
80         LASSERTF(logcookies && numcookies >= lsm->lsm_stripe_count,
81                  "logcookies %p, numcookies %d lsm->lsm_stripe_count %d \n",
82                  logcookies, numcookies, lsm->lsm_stripe_count);
83
84         for (i = 0; i < lsm->lsm_stripe_count; i++) {
85                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
86                 struct obd_device *child =
87                         lov->lov_tgts[loi->loi_ost_idx]->ltd_exp->exp_obd;
88                 struct llog_ctxt *cctxt = llog_get_context(child, ctxt->loc_idx);
89
90                 /* fill mds unlink/setattr log record */
91                 switch (rec->lrh_type) {
92                 case MDS_UNLINK_REC: {
93                         struct llog_unlink_rec *lur = (struct llog_unlink_rec *)rec;
94                         lur->lur_oid = loi->loi_id;
95                         lur->lur_oseq = loi->loi_seq;
96                         break;
97                 }
98                 case MDS_SETATTR_REC: {
99                         struct llog_setattr_rec *lsr = (struct llog_setattr_rec *)rec;
100                         lsr->lsr_oid = loi->loi_id;
101                         lsr->lsr_oseq = loi->loi_seq;
102                         break;
103                 }
104                 case MDS_SETATTR64_REC: {
105                         struct llog_setattr64_rec *lsr = (struct llog_setattr64_rec *)rec;
106                         lsr->lsr_oid = loi->loi_id;
107                         lsr->lsr_oseq = loi->loi_seq;
108                         break;
109                 }
110                 default:
111                         break;
112                 }
113                 /* inject error in llog_add() below */
114                 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_FAIL_LOV_LOG_ADD)) {
115                         llog_ctxt_put(cctxt);
116                         cctxt = NULL;
117                 }
118                 rc = llog_add(cctxt, rec, NULL, logcookies + cookies,
119                                numcookies - cookies);
120                 llog_ctxt_put(cctxt);
121                 if (rc < 0) {
122                         CERROR("Can't add llog (rc = %d) for stripe %i\n",
123                                rc, cookies);
124                         memset(logcookies + cookies, 0,
125                                sizeof(struct llog_cookie));
126                         rc = 1; /* skip this cookie */
127                 }
128                 /* Note that rc is always 1 if llog_add was successful */
129                 cookies += rc;
130         }
131         RETURN(cookies);
132 }
133
134 static int lov_llog_origin_connect(struct llog_ctxt *ctxt,
135                                    struct llog_logid *logid,
136                                    struct llog_gen *gen,
137                                    struct obd_uuid *uuid)
138 {
139         struct obd_device *obd = ctxt->loc_obd;
140         struct lov_obd *lov = &obd->u.lov;
141         int i, rc = 0, err = 0;
142         ENTRY;
143
144         obd_getref(obd);
145         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
146                 struct obd_device *child;
147                 struct llog_ctxt *cctxt;
148
149                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
150                         continue;
151                 if (uuid && !obd_uuid_equals(uuid, &lov->lov_tgts[i]->ltd_uuid))
152                         continue;
153                 CDEBUG(D_CONFIG, "connect %d/%d\n", i, lov->desc.ld_tgt_count);
154                 child = lov->lov_tgts[i]->ltd_exp->exp_obd;
155                 cctxt = llog_get_context(child, ctxt->loc_idx);
156                 rc = llog_connect(cctxt, logid, gen, uuid);
157                 llog_ctxt_put(cctxt);
158
159                 if (rc) {
160                         CERROR("error osc_llog_connect tgt %d (%d)\n", i, rc);
161                         if (!err)
162                                 err = rc;
163                 }
164         }
165         obd_putref(obd);
166
167         RETURN(err);
168 }
169
170 /* the replicators commit callback */
171 static int lov_llog_repl_cancel(struct llog_ctxt *ctxt, struct lov_stripe_md *lsm,
172                           int count, struct llog_cookie *cookies, int flags)
173 {
174         struct lov_obd *lov;
175         struct obd_device *obd = ctxt->loc_obd;
176         int rc = 0, i;
177         ENTRY;
178
179         LASSERT(lsm != NULL);
180         LASSERT(count == lsm->lsm_stripe_count);
181
182         lov = &obd->u.lov;
183         obd_getref(obd);
184         for (i = 0; i < count; i++, cookies++) {
185                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
186                 struct obd_device *child =
187                         lov->lov_tgts[loi->loi_ost_idx]->ltd_exp->exp_obd;
188                 struct llog_ctxt *cctxt =
189                         llog_get_context(child, ctxt->loc_idx);
190                 int err;
191
192                 err = llog_cancel(cctxt, NULL, 1, cookies, flags);
193                 llog_ctxt_put(cctxt);
194                 if (err && lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
195                         CERROR("error: objid "LPX64" subobj "LPX64
196                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
197                                loi->loi_id, loi->loi_ost_idx, err);
198                         if (!rc)
199                                 rc = err;
200                 }
201         }
202         obd_putref(obd);
203         RETURN(rc);
204 }
205
206 static struct llog_operations lov_mds_ost_orig_logops = {
207         lop_add: lov_llog_origin_add,
208         lop_connect: lov_llog_origin_connect
209 };
210
211 static struct llog_operations lov_size_repl_logops = {
212         lop_cancel: lov_llog_repl_cancel
213 };
214
215 int lov_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
216                   struct obd_device *disk_obd, int *index)
217 {
218         struct lov_obd *lov = &obd->u.lov;
219         struct obd_device *child;
220         int i, rc = 0;
221         ENTRY;
222
223         LASSERT(olg == &obd->obd_olg);
224         rc = llog_setup(obd, olg, LLOG_MDS_OST_ORIG_CTXT, disk_obd, 0, NULL,
225                         &lov_mds_ost_orig_logops);
226         if (rc)
227                 RETURN(rc);
228
229         rc = llog_setup(obd, olg, LLOG_SIZE_REPL_CTXT, disk_obd, 0, NULL,
230                         &lov_size_repl_logops);
231         if (rc)
232                 GOTO(err_cleanup, rc);
233
234         obd_getref(obd);
235         /* count may not match lov->desc.ld_tgt_count during dynamic ost add */
236         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
237                 if (!lov->lov_tgts[i])
238                         continue;
239
240                 if (index && i != *index)
241                         continue;
242
243                 child = lov->lov_tgts[i]->ltd_obd;
244                 rc = obd_llog_init(child, &child->obd_olg, disk_obd, &i);
245                 if (rc)
246                         CERROR("error osc_llog_init idx %d osc '%s' tgt '%s' "
247                                "(rc=%d)\n", i, child->obd_name,
248                                disk_obd->obd_name, rc);
249                 rc = 0;
250         }
251         obd_putref(obd);
252         GOTO(err_cleanup, rc);
253 err_cleanup:
254         if (rc) {
255                 struct llog_ctxt *ctxt =
256                         llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
257                 if (ctxt)
258                         llog_cleanup(ctxt);
259                 ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
260                 if (ctxt)
261                         llog_cleanup(ctxt);
262         }
263         return rc;
264 }
265
266 int lov_llog_finish(struct obd_device *obd, int count)
267 {
268         struct llog_ctxt *ctxt;
269         int rc = 0, rc2 = 0;
270         ENTRY;
271
272         /* cleanup our llogs only if the ctxts have been setup
273          * (client lov doesn't setup, mds lov does). */
274         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
275         if (ctxt)
276                 rc = llog_cleanup(ctxt);
277
278         ctxt = llog_get_context(obd, LLOG_SIZE_REPL_CTXT);
279         if (ctxt)
280                 rc2 = llog_cleanup(ctxt);
281         if (!rc)
282                 rc = rc2;
283
284         /* lov->tgt llogs are cleaned during osc_cleanup. */
285         RETURN(rc);
286 }