4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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
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
27 * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
28 * Use is subject to license terms.
30 * Copyright (c) 2011, Whamcloud, Inc.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
36 * lustre/ptlrpc/llog_server.c
38 * remote api for llog - server side
40 * Author: Andreas Dilger <adilger@clusterfs.com>
43 #define DEBUG_SUBSYSTEM S_LOG
46 #include <liblustre.h>
49 #include <obd_class.h>
50 #include <lustre_log.h>
51 #include <lustre_net.h>
52 #include <lustre_fsfilt.h>
54 #if defined(__KERNEL__) && defined(LUSTRE_LOG_SERVER)
56 int llog_origin_handle_create(struct ptlrpc_request *req)
58 struct obd_export *exp = req->rq_export;
59 struct obd_device *obd = exp->exp_obd;
60 struct obd_device *disk_obd;
61 struct llog_handle *loghandle;
62 struct llogd_body *body;
63 struct lvfs_run_ctxt saved;
64 struct llog_logid *logid = NULL;
65 struct llog_ctxt *ctxt;
70 body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
74 if (body->lgd_logid.lgl_oid > 0)
75 logid = &body->lgd_logid;
77 if (req_capsule_field_present(&req->rq_pill, &RMF_NAME, RCL_CLIENT)) {
78 name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
81 CDEBUG(D_INFO, "%s: opening log %s\n", obd->obd_name, name);
84 ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
86 CDEBUG(D_WARNING, "%s: no ctxt. group=%p idx=%d name=%s\n",
87 obd->obd_name, &obd->obd_olg, body->lgd_ctxt_idx, name);
90 disk_obd = ctxt->loc_exp->exp_obd;
91 push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
93 rc = llog_create(req->rq_svc_thread->t_env, ctxt, &loghandle, logid,
98 rc = req_capsule_server_pack(&req->rq_pill);
100 GOTO(out_close, rc = -ENOMEM);
102 body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
103 body->lgd_logid = loghandle->lgh_id;
107 rc2 = llog_close(req->rq_svc_thread->t_env, loghandle);
111 pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
115 EXPORT_SYMBOL(llog_origin_handle_create);
117 int llog_origin_handle_destroy(struct ptlrpc_request *req)
119 struct obd_export *exp = req->rq_export;
120 struct obd_device *obd = exp->exp_obd;
121 struct obd_device *disk_obd;
122 struct llog_handle *loghandle;
123 struct llogd_body *body;
124 struct lvfs_run_ctxt saved;
125 struct llog_logid *logid = NULL;
126 struct llog_ctxt *ctxt;
130 body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
134 if (body->lgd_logid.lgl_oid > 0)
135 logid = &body->lgd_logid;
137 ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
141 disk_obd = ctxt->loc_exp->exp_obd;
142 push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
144 rc = llog_create(req->rq_svc_thread->t_env, ctxt, &loghandle, logid,
149 rc = req_capsule_server_pack(&req->rq_pill);
151 GOTO(out_close, rc = -ENOMEM);
153 body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
154 body->lgd_logid = loghandle->lgh_id;
155 rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle,
156 LLOG_F_IS_PLAIN, NULL);
159 rc = llog_destroy(req->rq_svc_thread->t_env, loghandle);
162 llog_free_handle(loghandle);
166 llog_close(req->rq_svc_thread->t_env, loghandle);
168 pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
172 EXPORT_SYMBOL(llog_origin_handle_destroy);
174 int llog_origin_handle_next_block(struct ptlrpc_request *req)
176 struct obd_export *exp = req->rq_export;
177 struct obd_device *obd = exp->exp_obd;
178 struct obd_device *disk_obd;
179 struct llog_handle *loghandle;
180 struct llogd_body *body;
181 struct llogd_body *repbody;
182 struct lvfs_run_ctxt saved;
183 struct llog_ctxt *ctxt;
190 body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
194 OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
198 ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
200 GOTO(out_free, rc = -ENODEV);
201 disk_obd = ctxt->loc_exp->exp_obd;
202 push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
204 rc = llog_create(req->rq_svc_thread->t_env, ctxt, &loghandle,
205 &body->lgd_logid, NULL);
209 flags = body->lgd_llh_flags;
210 rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
215 memset(buf, 0, LLOG_CHUNK_SIZE);
216 rc = llog_next_block(req->rq_svc_thread->t_env, loghandle,
217 &body->lgd_saved_index, body->lgd_index,
218 &body->lgd_cur_offset, buf, LLOG_CHUNK_SIZE);
222 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
224 rc = req_capsule_server_pack(&req->rq_pill);
226 GOTO(out_close, rc = -ENOMEM);
228 repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
231 ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
232 memcpy(ptr, buf, LLOG_CHUNK_SIZE);
235 rc2 = llog_close(req->rq_svc_thread->t_env, loghandle);
239 pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
242 OBD_FREE(buf, LLOG_CHUNK_SIZE);
245 EXPORT_SYMBOL(llog_origin_handle_next_block);
247 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
249 struct obd_export *exp = req->rq_export;
250 struct obd_device *obd = exp->exp_obd;
251 struct llog_handle *loghandle;
252 struct llogd_body *body;
253 struct llogd_body *repbody;
254 struct obd_device *disk_obd;
255 struct lvfs_run_ctxt saved;
256 struct llog_ctxt *ctxt;
263 body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
267 OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
271 ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
273 GOTO(out_free, rc = -ENODEV);
275 disk_obd = ctxt->loc_exp->exp_obd;
276 push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
278 rc = llog_create(req->rq_svc_thread->t_env, ctxt, &loghandle,
279 &body->lgd_logid, NULL);
283 flags = body->lgd_llh_flags;
284 rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
289 memset(buf, 0, LLOG_CHUNK_SIZE);
290 rc = llog_prev_block(req->rq_svc_thread->t_env, loghandle,
291 body->lgd_index, buf, LLOG_CHUNK_SIZE);
295 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
297 rc = req_capsule_server_pack(&req->rq_pill);
299 GOTO(out_close, rc = -ENOMEM);
301 repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
304 ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
305 memcpy(ptr, buf, LLOG_CHUNK_SIZE);
308 rc2 = llog_close(req->rq_svc_thread->t_env, loghandle);
313 pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
316 OBD_FREE(buf, LLOG_CHUNK_SIZE);
319 EXPORT_SYMBOL(llog_origin_handle_prev_block);
321 int llog_origin_handle_read_header(struct ptlrpc_request *req)
323 struct obd_export *exp = req->rq_export;
324 struct obd_device *obd = exp->exp_obd;
325 struct obd_device *disk_obd;
326 struct llog_handle *loghandle;
327 struct llogd_body *body;
328 struct llog_log_hdr *hdr;
329 struct lvfs_run_ctxt saved;
330 struct llog_ctxt *ctxt;
335 body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
339 ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
343 disk_obd = ctxt->loc_exp->exp_obd;
344 push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
346 rc = llog_create(req->rq_svc_thread->t_env, ctxt, &loghandle,
347 &body->lgd_logid, NULL);
352 * llog_init_handle() reads the llog header
354 flags = body->lgd_llh_flags;
355 rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
360 rc = req_capsule_server_pack(&req->rq_pill);
362 GOTO(out_close, rc = -ENOMEM);
364 hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
365 *hdr = *loghandle->lgh_hdr;
368 rc2 = llog_close(req->rq_svc_thread->t_env, loghandle);
372 pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
376 EXPORT_SYMBOL(llog_origin_handle_read_header);
378 int llog_origin_handle_close(struct ptlrpc_request *req)
384 EXPORT_SYMBOL(llog_origin_handle_close);
386 int llog_origin_handle_cancel(struct ptlrpc_request *req)
388 struct obd_device *obd = req->rq_export->exp_obd;
389 int num_cookies, rc = 0, err, i, failed = 0;
390 struct obd_device *disk_obd;
391 struct llog_cookie *logcookies;
392 struct llog_ctxt *ctxt = NULL;
393 struct lvfs_run_ctxt saved;
394 struct llog_handle *cathandle;
399 logcookies = req_capsule_client_get(&req->rq_pill, &RMF_LOGCOOKIES);
400 num_cookies = req_capsule_get_size(&req->rq_pill, &RMF_LOGCOOKIES,
401 RCL_CLIENT) / sizeof(*logcookies);
402 if (logcookies == NULL || num_cookies == 0) {
403 DEBUG_REQ(D_HA, req, "No llog cookies sent");
407 ctxt = llog_get_context(obd, logcookies->lgc_subsys);
411 disk_obd = ctxt->loc_exp->exp_obd;
412 push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
413 for (i = 0; i < num_cookies; i++, logcookies++) {
414 cathandle = ctxt->loc_handle;
415 LASSERT(cathandle != NULL);
416 inode = cathandle->lgh_file->f_dentry->d_inode;
418 handle = fsfilt_start_log(disk_obd, inode,
419 FSFILT_OP_CANCEL_UNLINK, NULL, 1);
420 if (IS_ERR(handle)) {
421 CERROR("fsfilt_start_log() failed: %ld\n",
423 GOTO(pop_ctxt, rc = PTR_ERR(handle));
426 rc = llog_cat_cancel_records(req->rq_svc_thread->t_env,
427 cathandle, 1, logcookies);
430 * Do not raise -ENOENT errors for resent rpcs. This rec already
434 (lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)) {
436 * Do not change this message, reply-single.sh test_59b
437 * expects to find this in log.
439 CDEBUG(D_RPCTRACE, "RESENT cancel req %p - ignored\n",
442 } else if (rc == 0) {
443 CDEBUG(D_RPCTRACE, "Canceled %d llog-records\n",
447 err = fsfilt_commit(disk_obd, inode, handle, 0);
449 CERROR("Error committing transaction: %d\n", err);
459 pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
461 CERROR("Cancel %d of %d llog-records failed: %d\n",
462 failed, num_cookies, rc);
467 EXPORT_SYMBOL(llog_origin_handle_cancel);
469 #else /* !__KERNEL__ */
470 int llog_origin_handle_create(struct ptlrpc_request *req)
476 int llog_origin_handle_destroy(struct ptlrpc_request *req)
482 int llog_origin_handle_next_block(struct ptlrpc_request *req)
487 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
492 int llog_origin_handle_read_header(struct ptlrpc_request *req)
497 int llog_origin_handle_close(struct ptlrpc_request *req)
502 int llog_origin_handle_cancel(struct ptlrpc_request *req)