Whamcloud - gitweb
LU-1303 lod: transfer default striping from parent/fs
[fs/lustre-release.git] / lustre / fid / fid_store.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) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 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/fid/fid_store.c
37  *
38  * Lustre Sequence Manager
39  *
40  * Author: Yury Umanets <umka@clusterfs.com>
41  */
42
43 #define DEBUG_SUBSYSTEM S_FID
44
45 #ifdef __KERNEL__
46 # include <libcfs/libcfs.h>
47 # include <linux/module.h>
48 #else /* __KERNEL__ */
49 # include <liblustre.h>
50 #endif
51
52 #include <obd.h>
53 #include <obd_class.h>
54 #include <dt_object.h>
55 #include <md_object.h>
56 #include <obd_support.h>
57 #include <lustre_req_layout.h>
58 #include <lustre_fid.h>
59 #include "fid_internal.h"
60
61 #ifdef __KERNEL__
62
63 static struct lu_buf *seq_store_buf(struct seq_thread_info *info)
64 {
65         struct lu_buf *buf;
66
67         buf = &info->sti_buf;
68         buf->lb_buf = &info->sti_space;
69         buf->lb_len = sizeof(info->sti_space);
70         return buf;
71 }
72
73 struct seq_update_callback {
74         struct dt_txn_commit_cb suc_cb;
75         struct lu_server_seq   *suc_seq;
76 };
77
78 void seq_update_cb(struct lu_env *env, struct thandle *th,
79                    struct dt_txn_commit_cb *cb, int err)
80 {
81         struct seq_update_callback *ccb;
82
83         ccb = container_of0(cb, struct seq_update_callback, suc_cb);
84
85         LASSERT(ccb->suc_seq != NULL);
86
87         ccb->suc_seq->lss_need_sync = 0;
88         OBD_FREE_PTR(ccb);
89 }
90
91 struct thandle *seq_store_trans_create(struct lu_server_seq *seq,
92                                        const struct lu_env *env)
93 {
94         struct dt_device *dt_dev;
95
96         dt_dev = lu2dt_dev(seq->lss_obj->do_lu.lo_dev);
97         return dt_trans_create(env, dt_dev);
98 }
99
100 int seq_store_trans_start(struct lu_server_seq *seq, const struct lu_env *env,
101                           struct thandle *th)
102 {
103         struct dt_device *dt_dev;
104         ENTRY;
105
106         dt_dev = lu2dt_dev(seq->lss_obj->do_lu.lo_dev);
107
108         return dt_trans_start(env, dt_dev, th);
109 }
110
111 int seq_update_cb_add(struct thandle *th, struct lu_server_seq *seq)
112 {
113         struct seq_update_callback *ccb;
114         struct dt_txn_commit_cb    *dcb;
115         int                        rc;
116
117         OBD_ALLOC_PTR(ccb);
118         if (ccb == NULL)
119                 return -ENOMEM;
120
121         ccb->suc_seq       = seq;
122         seq->lss_need_sync = 1;
123
124         dcb            = &ccb->suc_cb;
125         dcb->dcb_func  = seq_update_cb;
126         CFS_INIT_LIST_HEAD(&dcb->dcb_linkage);
127         strncpy(dcb->dcb_name, "seq_update_cb", MAX_COMMIT_CB_STR_LEN);
128         dcb->dcb_name[MAX_COMMIT_CB_STR_LEN - 1] = '\0';
129
130         rc = dt_trans_cb_add(th, dcb);
131         if (rc)
132                 OBD_FREE_PTR(ccb);
133         return rc;
134 }
135
136 int seq_declare_store_write(struct lu_server_seq *seq,
137                             const struct lu_env *env,
138                             struct thandle *th)
139 {
140         struct dt_object *dt_obj = seq->lss_obj;
141         int rc;
142         ENTRY;
143
144         rc = dt_obj->do_body_ops->dbo_declare_write(env, dt_obj,
145                                                     sizeof(struct lu_seq_range),
146                                                     0, th);
147         return rc;
148 }
149
150 /* This function implies that caller takes care about locking. */
151 int seq_store_write(struct lu_server_seq *seq,
152                     const struct lu_env *env,
153                     struct thandle *th)
154 {
155         struct dt_object *dt_obj = seq->lss_obj;
156         struct seq_thread_info *info;
157         loff_t pos = 0;
158         int rc;
159         ENTRY;
160
161         info = lu_context_key_get(&env->le_ctx, &seq_thread_key);
162         LASSERT(info != NULL);
163
164         /* Store ranges in le format. */
165         range_cpu_to_le(&info->sti_space, &seq->lss_space);
166
167         rc = dt_obj->do_body_ops->dbo_write(env, dt_obj,
168                                             seq_store_buf(info),
169                                             &pos, th, BYPASS_CAPA, 1);
170         if (rc == sizeof(info->sti_space)) {
171                 CDEBUG(D_INFO, "%s: Space - "DRANGE"\n",
172                        seq->lss_name, PRANGE(&seq->lss_space));
173                 rc = 0;
174         } else if (rc >= 0) {
175                 rc = -EIO;
176         }
177
178
179         RETURN(rc);
180 }
181
182 int seq_store_update(const struct lu_env *env, struct lu_server_seq *seq,
183                      struct lu_seq_range *out, int sync)
184 {
185         struct dt_device *dt_dev;
186         struct thandle *th;
187         int rc;
188         ENTRY;
189
190         dt_dev = lu2dt_dev(seq->lss_obj->do_lu.lo_dev);
191
192         th = seq_store_trans_create(seq, env);
193         if (IS_ERR(th))
194                 RETURN(PTR_ERR(th));
195
196         rc = seq_declare_store_write(seq, env, th);
197         if (rc)
198                 GOTO(exit, rc);
199
200         if (out != NULL) {
201                 rc = fld_declare_server_create(seq->lss_site->ms_server_fld,
202                                                env, th);
203                 if (rc)
204                         GOTO(exit, rc);
205         }
206
207         rc = seq_store_trans_start(seq, env, th);
208         if (rc)
209                 GOTO(exit, rc);
210
211         rc = seq_store_write(seq, env, th);
212         if (rc) {
213                 CERROR("%s: Can't write space data, rc %d\n",
214                        seq->lss_name, rc);
215                 GOTO(exit,rc);
216         } else if (out != NULL) {
217                 rc = fld_server_create(seq->lss_site->ms_server_fld,
218                                        env, out, th);
219                 if (rc) {
220                         CERROR("%s: Can't Update fld database, rc %d\n",
221                                seq->lss_name, rc);
222                         GOTO(exit,rc);
223                 }
224         }
225
226         /* next sequence update will need sync until this update is committed
227          * in case of sync operation this is not needed obviously */
228         if (!sync)
229                 /* if callback can't be added then sync always */
230                 sync = !!seq_update_cb_add(th, seq);
231
232         th->th_sync |= sync;
233 exit:
234         dt_trans_stop(env, dt_dev, th);
235         return rc;
236 }
237
238 /*
239  * This function implies that caller takes care about locking or locking is not
240  * needed (init time).
241  */
242 int seq_store_read(struct lu_server_seq *seq,
243                    const struct lu_env *env)
244 {
245         struct dt_object *dt_obj = seq->lss_obj;
246         struct seq_thread_info *info;
247         loff_t pos = 0;
248         int rc;
249         ENTRY;
250
251         info = lu_context_key_get(&env->le_ctx, &seq_thread_key);
252         LASSERT(info != NULL);
253
254         rc = dt_obj->do_body_ops->dbo_read(env, dt_obj, seq_store_buf(info),
255                                            &pos, BYPASS_CAPA);
256
257         if (rc == sizeof(info->sti_space)) {
258                 range_le_to_cpu(&seq->lss_space, &info->sti_space);
259                 CDEBUG(D_INFO, "%s: Space - "DRANGE"\n",
260                        seq->lss_name, PRANGE(&seq->lss_space));
261                 rc = 0;
262         } else if (rc == 0) {
263                 rc = -ENODATA;
264         } else if (rc >= 0) {
265                 CERROR("%s: Read only %d bytes of %d\n", seq->lss_name,
266                        rc, (int)sizeof(info->sti_space));
267                 rc = -EIO;
268         }
269
270         RETURN(rc);
271 }
272
273 int seq_store_init(struct lu_server_seq *seq,
274                    const struct lu_env *env,
275                    struct dt_device *dt)
276 {
277         struct dt_object *dt_obj;
278         struct lu_fid fid;
279         const char *name;
280         int rc;
281         ENTRY;
282
283         name = seq->lss_type == LUSTRE_SEQ_SERVER ?
284                 LUSTRE_SEQ_SRV_NAME : LUSTRE_SEQ_CTL_NAME;
285
286         dt_obj = dt_store_open(env, dt, "", name, &fid);
287         if (!IS_ERR(dt_obj)) {
288                 seq->lss_obj = dt_obj;
289                 rc = 0;
290         } else {
291                 CERROR("%s: Can't find \"%s\" obj %d\n",
292                        seq->lss_name, name, (int)PTR_ERR(dt_obj));
293                 rc = PTR_ERR(dt_obj);
294         }
295
296         RETURN(rc);
297 }
298
299 void seq_store_fini(struct lu_server_seq *seq,
300                     const struct lu_env *env)
301 {
302         ENTRY;
303
304         if (seq->lss_obj != NULL) {
305                 if (!IS_ERR(seq->lss_obj))
306                         lu_object_put(env, &seq->lss_obj->do_lu);
307                 seq->lss_obj = NULL;
308         }
309
310         EXIT;
311 }
312 #endif