Whamcloud - gitweb
2a30db50a30c11a2115351b71a025f894f68093b
[fs/lustre-release.git] / lustre / ptlrpc / llog_client.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, Intel Corporation.
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/ptlrpc/llog_client.c
37  *
38  * remote api for llog - client side
39  *
40  * Author: Andreas Dilger <adilger@clusterfs.com>
41  */
42
43 #define DEBUG_SUBSYSTEM S_LOG
44
45 #include <libcfs/libcfs.h>
46
47 #include <obd_class.h>
48 #include <lustre_log.h>
49 #include <lustre_net.h>
50 #include <libcfs/list.h>
51
52 #define LLOG_CLIENT_ENTRY(ctxt, imp) do {                             \
53         mutex_lock(&ctxt->loc_mutex);                             \
54         if (ctxt->loc_imp) {                                          \
55                 imp = class_import_get(ctxt->loc_imp);                \
56         } else {                                                      \
57                 CERROR("ctxt->loc_imp == NULL for context idx %d."    \
58                        "Unable to complete MDS/OSS recovery,"         \
59                        "but I'll try again next time.  Not fatal.\n", \
60                        ctxt->loc_idx);                                \
61                 imp = NULL;                                           \
62                 mutex_unlock(&ctxt->loc_mutex);                   \
63                 return (-EINVAL);                                     \
64         }                                                             \
65         mutex_unlock(&ctxt->loc_mutex);                           \
66 } while(0)
67
68 #define LLOG_CLIENT_EXIT(ctxt, imp) do {                              \
69         mutex_lock(&ctxt->loc_mutex);                             \
70         if (ctxt->loc_imp != imp)                                     \
71                 CWARN("loc_imp has changed from %p to %p\n",          \
72                        ctxt->loc_imp, imp);                           \
73         class_import_put(imp);                                        \
74         mutex_unlock(&ctxt->loc_mutex);                           \
75 } while(0)
76
77 /* This is a callback from the llog_* functions.
78  * Assumes caller has already pushed us into the kernel context. */
79 static int llog_client_open(const struct lu_env *env,
80                             struct llog_handle *lgh, struct llog_logid *logid,
81                             char *name, enum llog_open_param open_param)
82 {
83         struct obd_import     *imp;
84         struct llogd_body     *body;
85         struct llog_ctxt      *ctxt = lgh->lgh_ctxt;
86         struct ptlrpc_request *req = NULL;
87         int                    rc;
88         ENTRY;
89
90         LLOG_CLIENT_ENTRY(ctxt, imp);
91
92         /* client cannot create llog */
93         LASSERTF(open_param != LLOG_OPEN_NEW, "%#x\n", open_param);
94         LASSERT(lgh);
95
96         req = ptlrpc_request_alloc(imp, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
97         if (req == NULL)
98                 GOTO(out, rc = -ENOMEM);
99
100         if (name)
101                 req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
102                                      strlen(name) + 1);
103
104         rc = ptlrpc_request_pack(req, LUSTRE_LOG_VERSION,
105                                  LLOG_ORIGIN_HANDLE_CREATE);
106         if (rc) {
107                 ptlrpc_request_free(req);
108                 req = NULL;
109                 GOTO(out, rc);
110         }
111         ptlrpc_request_set_replen(req);
112
113         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
114         if (logid)
115                 body->lgd_logid = *logid;
116         body->lgd_ctxt_idx = ctxt->loc_idx - 1;
117
118         if (name) {
119                 char *tmp;
120                 tmp = req_capsule_client_sized_get(&req->rq_pill, &RMF_NAME,
121                                                    strlen(name) + 1);
122                 LASSERT(tmp);
123                 strcpy(tmp, name);
124         }
125
126         rc = ptlrpc_queue_wait(req);
127         if (rc)
128                 GOTO(out, rc);
129
130         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
131         if (body == NULL)
132                 GOTO(out, rc = -EFAULT);
133
134         lgh->lgh_id = body->lgd_logid;
135         lgh->lgh_ctxt = ctxt;
136         EXIT;
137 out:
138         LLOG_CLIENT_EXIT(ctxt, imp);
139         ptlrpc_req_finished(req);
140         return rc;
141 }
142
143 static int llog_client_destroy(const struct lu_env *env,
144                                struct llog_handle *loghandle)
145 {
146         struct obd_import     *imp;
147         struct ptlrpc_request *req = NULL;
148         struct llogd_body     *body;
149         int                    rc;
150         ENTRY;
151
152         LLOG_CLIENT_ENTRY(loghandle->lgh_ctxt, imp);
153         req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_DESTROY,
154                                         LUSTRE_LOG_VERSION,
155                                         LLOG_ORIGIN_HANDLE_DESTROY);
156         if (req == NULL)
157                 GOTO(err_exit, rc =-ENOMEM);
158
159         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
160         body->lgd_logid = loghandle->lgh_id;
161         body->lgd_llh_flags = loghandle->lgh_hdr->llh_flags;
162
163         if (!(body->lgd_llh_flags & LLOG_F_IS_PLAIN))
164                 CERROR("%s: wrong llog flags %x\n", imp->imp_obd->obd_name,
165                        body->lgd_llh_flags);
166
167         ptlrpc_request_set_replen(req);
168         rc = ptlrpc_queue_wait(req);
169
170         ptlrpc_req_finished(req);
171 err_exit:
172         LLOG_CLIENT_EXIT(loghandle->lgh_ctxt, imp);
173         RETURN(rc);
174 }
175
176
177 static int llog_client_next_block(const struct lu_env *env,
178                                   struct llog_handle *loghandle,
179                                   int *cur_idx, int next_idx,
180                                   __u64 *cur_offset, void *buf, int len)
181 {
182         struct obd_import     *imp;
183         struct ptlrpc_request *req = NULL;
184         struct llogd_body     *body;
185         void                  *ptr;
186         int                    rc;
187         ENTRY;
188
189         LLOG_CLIENT_ENTRY(loghandle->lgh_ctxt, imp);
190         req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK,
191                                         LUSTRE_LOG_VERSION,
192                                         LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
193         if (req == NULL)
194                 GOTO(err_exit, rc =-ENOMEM);
195
196         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
197         body->lgd_logid = loghandle->lgh_id;
198         body->lgd_ctxt_idx = loghandle->lgh_ctxt->loc_idx - 1;
199         body->lgd_llh_flags = loghandle->lgh_hdr->llh_flags;
200         body->lgd_index = next_idx;
201         body->lgd_saved_index = *cur_idx;
202         body->lgd_len = len;
203         body->lgd_cur_offset = *cur_offset;
204
205         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER, len);
206         ptlrpc_request_set_replen(req);
207         rc = ptlrpc_queue_wait(req);
208         if (rc)
209                 GOTO(out, rc);
210
211         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
212         if (body == NULL)
213                 GOTO(out, rc =-EFAULT);
214
215         /* The log records are swabbed as they are processed */
216         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
217         if (ptr == NULL)
218                 GOTO(out, rc =-EFAULT);
219
220         *cur_idx = body->lgd_saved_index;
221         *cur_offset = body->lgd_cur_offset;
222
223         memcpy(buf, ptr, len);
224         EXIT;
225 out:
226         ptlrpc_req_finished(req);
227 err_exit:
228         LLOG_CLIENT_EXIT(loghandle->lgh_ctxt, imp);
229         return rc;
230 }
231
232 static int llog_client_prev_block(const struct lu_env *env,
233                                   struct llog_handle *loghandle,
234                                   int prev_idx, void *buf, int len)
235 {
236         struct obd_import     *imp;
237         struct ptlrpc_request *req = NULL;
238         struct llogd_body     *body;
239         void                  *ptr;
240         int                    rc;
241         ENTRY;
242
243         LLOG_CLIENT_ENTRY(loghandle->lgh_ctxt, imp);
244         req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK,
245                                         LUSTRE_LOG_VERSION,
246                                         LLOG_ORIGIN_HANDLE_PREV_BLOCK);
247         if (req == NULL)
248                 GOTO(err_exit, rc = -ENOMEM);
249
250         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
251         body->lgd_logid = loghandle->lgh_id;
252         body->lgd_ctxt_idx = loghandle->lgh_ctxt->loc_idx - 1;
253         body->lgd_llh_flags = loghandle->lgh_hdr->llh_flags;
254         body->lgd_index = prev_idx;
255         body->lgd_len = len;
256
257         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER, len);
258         ptlrpc_request_set_replen(req);
259
260         rc = ptlrpc_queue_wait(req);
261         if (rc)
262                 GOTO(out, rc);
263
264         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
265         if (body == NULL)
266                 GOTO(out, rc =-EFAULT);
267
268         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
269         if (ptr == NULL)
270                 GOTO(out, rc =-EFAULT);
271
272         memcpy(buf, ptr, len);
273         EXIT;
274 out:
275         ptlrpc_req_finished(req);
276 err_exit:
277         LLOG_CLIENT_EXIT(loghandle->lgh_ctxt, imp);
278         return rc;
279 }
280
281 static int llog_client_read_header(const struct lu_env *env,
282                                    struct llog_handle *handle)
283 {
284         struct obd_import     *imp;
285         struct ptlrpc_request *req = NULL;
286         struct llogd_body     *body;
287         struct llog_log_hdr   *hdr;
288         struct llog_rec_hdr   *llh_hdr;
289         int                    rc;
290         ENTRY;
291
292         LLOG_CLIENT_ENTRY(handle->lgh_ctxt, imp);
293         req = ptlrpc_request_alloc_pack(imp,&RQF_LLOG_ORIGIN_HANDLE_READ_HEADER,
294                                         LUSTRE_LOG_VERSION,
295                                         LLOG_ORIGIN_HANDLE_READ_HEADER);
296         if (req == NULL)
297                 GOTO(err_exit, rc = -ENOMEM);
298
299         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
300         body->lgd_logid = handle->lgh_id;
301         body->lgd_ctxt_idx = handle->lgh_ctxt->loc_idx - 1;
302         body->lgd_llh_flags = handle->lgh_hdr->llh_flags;
303
304         ptlrpc_request_set_replen(req);
305         rc = ptlrpc_queue_wait(req);
306         if (rc)
307                 GOTO(out, rc);
308
309         hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
310         if (hdr == NULL)
311                 GOTO(out, rc =-EFAULT);
312
313         memcpy(handle->lgh_hdr, hdr, sizeof (*hdr));
314         handle->lgh_last_idx = handle->lgh_hdr->llh_tail.lrt_index;
315
316         /* sanity checks */
317         llh_hdr = &handle->lgh_hdr->llh_hdr;
318         if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
319                 CERROR("bad log header magic: %#x (expecting %#x)\n",
320                        llh_hdr->lrh_type, LLOG_HDR_MAGIC);
321                 rc = -EIO;
322         } else if (llh_hdr->lrh_len != LLOG_CHUNK_SIZE) {
323                 CERROR("incorrectly sized log header: %#x "
324                        "(expecting %#x)\n",
325                        llh_hdr->lrh_len, LLOG_CHUNK_SIZE);
326                 CERROR("you may need to re-run lconf --write_conf.\n");
327                 rc = -EIO;
328         }
329         EXIT;
330 out:
331         ptlrpc_req_finished(req);
332 err_exit:
333         LLOG_CLIENT_EXIT(handle->lgh_ctxt, imp);
334         return rc;
335 }
336
337 static int llog_client_close(const struct lu_env *env,
338                              struct llog_handle *handle)
339 {
340         /* this doesn't call LLOG_ORIGIN_HANDLE_CLOSE because
341            the servers all close the file at the end of every
342            other LLOG_ RPC. */
343         return(0);
344 }
345
346 struct llog_operations llog_client_ops = {
347         .lop_next_block         = llog_client_next_block,
348         .lop_prev_block         = llog_client_prev_block,
349         .lop_read_header        = llog_client_read_header,
350         .lop_open               = llog_client_open,
351         .lop_destroy            = llog_client_destroy,
352         .lop_close              = llog_client_close,
353 };
354 EXPORT_SYMBOL(llog_client_ops);