Whamcloud - gitweb
- landing of b_hd_cleanup_merge to HEAD.
[fs/lustre-release.git] / lustre / ptlrpc / llog_server.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  *  remote api for llog - server side
23  *
24  */
25
26 #define DEBUG_SUBSYSTEM S_LOG
27
28 #ifndef EXPORT_SYMTAB
29 #define EXPORT_SYMTAB
30 #endif
31
32 #ifndef __KERNEL__
33 #include <liblustre.h>
34 #else
35 #include <linux/fs.h>
36 #endif
37
38 #include <linux/obd_class.h>
39 #include <linux/lustre_log.h>
40 #include <linux/lustre_net.h>
41 #include <portals/list.h>
42 #include <linux/lustre_fsfilt.h>
43
44 #ifdef __KERNEL__
45
46 int llog_origin_handle_open(struct ptlrpc_request *req)
47 {
48         struct obd_export *exp = req->rq_export;
49         struct obd_device *obd = exp->exp_obd;
50         struct llog_handle  *loghandle;
51         struct llogd_body *body;
52         struct lvfs_run_ctxt saved;
53         struct llog_logid *logid = NULL;
54         struct llog_ctxt *ctxt;
55         char *name = NULL;
56         int rc, rc2, size = sizeof (*body);
57         ENTRY;
58
59         body = lustre_swab_reqbuf(req, 0, sizeof(*body),
60                                   lustre_swab_llogd_body);
61         if (body == NULL) {
62                 CERROR ("Can't unpack llogd_body\n");
63                 GOTO(out, rc =-EFAULT);
64         }
65
66         if (body->lgd_logid.lgl_oid > 0)
67                 logid = &body->lgd_logid;
68
69         if (req->rq_reqmsg->bufcount > 1) {
70                 name = lustre_msg_string(req->rq_reqmsg, 1, 0);
71                 if (name == NULL) {
72                         CERROR("Can't unpack name\n");
73                         GOTO(out, rc = -EFAULT);
74                 }
75         }
76
77         ctxt = llog_get_context(&obd->obd_llogs, body->lgd_ctxt_idx);
78         if (ctxt == NULL)
79                 GOTO(out, rc = -EINVAL);
80
81         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
82
83         rc = llog_open(ctxt, &loghandle, logid, name, body->lgd_llh_flags);
84         if (rc)
85                 GOTO(out_pop, rc);
86
87         rc = lustre_pack_reply(req, 1, &size, NULL);
88         if (rc)
89                 GOTO(out_close, rc = -ENOMEM);
90
91         body = lustre_msg_buf(req->rq_repmsg, 0, sizeof(*body));
92         body->lgd_logid = loghandle->lgh_id;
93
94 out_close:
95         rc2 = llog_close(loghandle);
96         if (!rc)
97                 rc = rc2;
98 out_pop:
99         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
100 out:
101         RETURN(rc);
102 }
103
104 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
105 {
106         struct obd_export *exp = req->rq_export;
107         struct obd_device *obd = exp->exp_obd;
108         struct llog_handle  *loghandle;
109         struct llogd_body *body;
110         struct lvfs_run_ctxt saved;
111         struct llog_ctxt *ctxt;
112         __u32 flags;
113         __u8 *buf;
114         void * ptr;
115         int rc, rc2, size[] = {sizeof (*body), LLOG_CHUNK_SIZE};
116         ENTRY;
117
118         body = lustre_swab_reqbuf(req, 0, sizeof(*body),
119                                   lustre_swab_llogd_body);
120         if (body == NULL) {
121                 CERROR ("Can't unpack llogd_body\n");
122                 GOTO(out, rc =-EFAULT);
123         }
124
125         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
126         if (!buf)
127                 GOTO(out, rc = -ENOMEM);
128
129         ctxt = llog_get_context(&obd->obd_llogs, body->lgd_ctxt_idx);
130         if (ctxt == NULL)
131                 GOTO(out, rc = -EINVAL);
132         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
133
134         rc = llog_open(ctxt, &loghandle, &body->lgd_logid, NULL, 0);
135         if (rc)
136                 GOTO(out_pop, rc);
137
138         flags = body->lgd_llh_flags;
139         rc = llog_init_handle(loghandle, flags, NULL);
140         if (rc)
141                 GOTO(out_close, rc);
142
143         memset(buf, 0, LLOG_CHUNK_SIZE);
144         rc = llog_prev_block(loghandle, body->lgd_index, buf, LLOG_CHUNK_SIZE);
145         if (rc)
146                 GOTO(out_close, rc);
147
148
149         rc = lustre_pack_reply(req, 2, size, NULL);
150         if (rc)
151                 GOTO(out_close, rc = -ENOMEM);
152
153         ptr = lustre_msg_buf(req->rq_repmsg, 0, sizeof (body));
154         memcpy(ptr, body, sizeof(*body));
155
156         ptr = lustre_msg_buf(req->rq_repmsg, 1, LLOG_CHUNK_SIZE);
157         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
158
159 out_close:
160         rc2 = llog_close(loghandle);
161         if (!rc)
162                 rc = rc2;
163
164 out_pop:
165         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
166         OBD_FREE(buf, LLOG_CHUNK_SIZE);
167 out:
168         RETURN(rc);
169 }
170
171 int llog_origin_handle_next_block(struct ptlrpc_request *req)
172 {
173         struct obd_export *exp = req->rq_export;
174         struct obd_device *obd = exp->exp_obd;
175         struct llog_handle  *loghandle;
176         struct llogd_body *body;
177         struct lvfs_run_ctxt saved;
178         struct llog_ctxt *ctxt;
179         __u32 flags;
180         __u8 *buf;
181         void * ptr;
182         int size[] = {sizeof (*body),
183                       LLOG_CHUNK_SIZE};
184         int rc, rc2;
185         ENTRY;
186
187         body = lustre_swab_reqbuf(req, 0, sizeof(*body),
188                                   lustre_swab_llogd_body);
189         if (body == NULL) {
190                 CERROR ("Can't unpack llogd_body\n");
191                 GOTO(out, rc =-EFAULT);
192         }
193
194         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
195         if (!buf)
196                 GOTO(out, rc = -ENOMEM);
197
198         ctxt = llog_get_context(&obd->obd_llogs, body->lgd_ctxt_idx);
199         if (ctxt == NULL)
200                 GOTO(out, rc = -EINVAL);
201         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
202
203         rc = llog_open(ctxt, &loghandle, &body->lgd_logid, NULL, 0);
204         if (rc)
205                 GOTO(out_pop, rc);
206
207         flags = body->lgd_llh_flags;
208         rc = llog_init_handle(loghandle, flags, NULL);
209         if (rc)
210                 GOTO(out_close, rc);
211
212         memset(buf, 0, LLOG_CHUNK_SIZE);
213         rc = llog_next_block(loghandle, &body->lgd_saved_index,
214                              body->lgd_index,
215                              &body->lgd_cur_offset, buf, LLOG_CHUNK_SIZE);
216         if (rc)
217                 GOTO(out_close, rc);
218
219
220         rc = lustre_pack_reply(req, 2, size, NULL);
221         if (rc)
222                 GOTO(out_close, rc = -ENOMEM);
223
224         ptr = lustre_msg_buf(req->rq_repmsg, 0, sizeof (body));
225         memcpy(ptr, body, sizeof(*body));
226
227         ptr = lustre_msg_buf(req->rq_repmsg, 1, LLOG_CHUNK_SIZE);
228         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
229
230 out_close:
231         rc2 = llog_close(loghandle);
232         if (!rc)
233                 rc = rc2;
234
235 out_pop:
236         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
237         OBD_FREE(buf, LLOG_CHUNK_SIZE);
238 out:
239         RETURN(rc);
240 }
241
242 int llog_origin_handle_read_header(struct ptlrpc_request *req)
243 {
244         struct obd_export *exp = req->rq_export;
245         struct obd_device *obd = exp->exp_obd;
246         struct llog_handle  *loghandle;
247         struct llogd_body *body;
248         struct llog_log_hdr *hdr;
249         struct lvfs_run_ctxt saved;
250         struct llog_ctxt *ctxt;
251         __u32 flags;
252         int size[] = {sizeof (*hdr)};
253         int rc, rc2;
254         ENTRY;
255
256         body = lustre_swab_reqbuf(req, 0, sizeof(*body),
257                                   lustre_swab_llogd_body);
258         if (body == NULL) {
259                 CERROR ("Can't unpack llogd_body\n");
260                 GOTO(out, rc =-EFAULT);
261         }
262
263         ctxt = llog_get_context(&obd->obd_llogs, body->lgd_ctxt_idx);
264         if (ctxt == NULL)
265                 GOTO(out, rc = -EINVAL);
266         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
267
268         rc = llog_open(ctxt, &loghandle, &body->lgd_logid, NULL, 0);
269         if (rc)
270                 GOTO(out_pop, rc);
271
272         /* init_handle reads the header */
273         flags = body->lgd_llh_flags;
274         rc = llog_init_handle(loghandle, flags, NULL);
275         if (rc)
276                 GOTO(out_close, rc);
277
278         rc = lustre_pack_reply(req, 1, size, NULL);
279         if (rc)
280                 GOTO(out_close, rc = -ENOMEM);
281
282         hdr = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*hdr));
283         memcpy(hdr, loghandle->lgh_hdr, sizeof(*hdr));
284
285 out_close:
286         rc2 = llog_close(loghandle);
287         if (!rc)
288                 rc = rc2;
289
290 out_pop:
291         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
292
293 out:
294         RETURN(rc);
295 }
296
297 int llog_origin_handle_close(struct ptlrpc_request *req)
298 {
299         int rc;
300
301         rc = 0;
302
303         RETURN(rc);
304 }
305
306 int llog_origin_handle_cancel(struct ptlrpc_request *req)
307 {
308         struct obd_device *obd = req->rq_export->exp_obd;
309         struct llog_cookie *logcookies;
310         struct llog_ctxt *ctxt;
311         int num_cookies, rc = 0, err, i;
312         struct lvfs_run_ctxt saved;
313         struct llog_handle *cathandle;
314         struct inode *inode;
315         void *handle;
316         ENTRY;
317
318         logcookies = lustre_msg_buf(req->rq_reqmsg, 0, sizeof(*logcookies));
319         num_cookies = req->rq_reqmsg->buflens[0]/sizeof(*logcookies);
320         if (logcookies == NULL || num_cookies == 0) {
321                 DEBUG_REQ(D_HA, req, "no cookies sent");
322                 RETURN(-EFAULT);
323         }
324
325         ctxt = llog_get_context(&obd->obd_llogs, logcookies->lgc_subsys);
326         if (ctxt == NULL) {
327                 CWARN("llog subsys not setup or already cleanup\n");
328                 RETURN(-ENOENT);
329         }
330
331         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
332         for (i = 0; i < num_cookies; i++, logcookies++) {
333                 cathandle = ctxt->loc_handle;
334                 LASSERT(cathandle != NULL);
335                 inode = cathandle->lgh_file->f_dentry->d_inode;
336
337                 handle = llog_fsfilt_start(ctxt, inode,
338                                            FSFILT_OP_CANCEL_UNLINK, NULL);
339                 if (IS_ERR(handle)) {
340                         CERROR("fsfilt_start failed: %ld\n", PTR_ERR(handle));
341                         GOTO(pop_ctxt, rc = PTR_ERR(handle));
342                 }
343
344                 rc = llog_cat_cancel_records(cathandle, 1, logcookies);
345
346                 err = llog_fsfilt_commit(ctxt, inode, handle, 0);
347                 if (err) {
348                         CERROR("error committing transaction: %d\n", err);
349                         if (!rc)
350                                 rc = err;
351                         GOTO(pop_ctxt, rc);
352                 }
353         }
354 pop_ctxt:
355         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
356         if (rc)
357                 CERROR("cancel %d llog-records failed: %d\n", num_cookies, rc);
358         else
359                 CDEBUG(D_HA, "cancel %d llog-records\n", num_cookies);
360
361         RETURN(rc);
362 }
363 EXPORT_SYMBOL(llog_origin_handle_cancel);
364
365 static int llog_catinfo_config(struct obd_device *obd, char *buf, int buf_len,
366                                char *client)
367 {
368         struct mds_obd *mds = &obd->u.mds;
369         struct llog_ctxt *ctxt;
370         struct lvfs_run_ctxt saved;
371         struct llog_handle *handle = NULL;
372         char name[4][64];
373         int rc, i, l, remains = buf_len;
374         char *out = buf;
375
376         ctxt = llog_get_context(&obd->obd_llogs, LLOG_CONFIG_ORIG_CTXT);
377         if (ctxt == NULL || mds == NULL)
378                 RETURN(-EOPNOTSUPP);
379
380         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
381
382         sprintf(name[0], "%s", mds->mds_profile);
383         sprintf(name[1], "%s-clean", mds->mds_profile);
384         sprintf(name[2], "%s", client);
385         sprintf(name[3], "%s-clean", client);
386
387         for (i = 0; i < 4; i++) {
388                 int index, uncanceled = 0;
389                 rc = llog_open(ctxt, &handle, NULL, name[i], 0);
390                 if (rc)
391                         GOTO(out_pop, rc);
392                 rc = llog_init_handle(handle, 0, NULL);
393                 if (rc) {
394                         llog_close(handle);
395                         GOTO(out_pop, rc = -ENOENT);
396                 }
397
398                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index ++) {
399                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
400                                 uncanceled++;
401                 }
402
403                 l = snprintf(out, remains, "[Log Name]: %s\nLog Size: %llu\n"
404                              "Last Index: %d\nUncanceled Records: %d\n\n",
405                              name[i],
406                              handle->lgh_file->f_dentry->d_inode->i_size,
407                              handle->lgh_last_idx, uncanceled);
408                 out += l;
409                 remains -= l;
410
411                 llog_close(handle);
412                 if (remains <= 0)
413                         break;
414         }
415 out_pop:
416         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
417         RETURN(rc);
418 }
419
420 struct cb_data {
421         struct llog_ctxt *ctxt;
422         char *out;
423         int  remains;
424         int  init;
425 };
426
427 static int llog_catinfo_cb(struct llog_handle *cat,
428                            struct llog_rec_hdr *rec, void *data)
429 {
430         static char *out = NULL;
431         static int remains = 0;
432         struct llog_ctxt *ctxt;
433         struct llog_handle *handle;
434         struct llog_logid *logid;
435         struct llog_logid_rec *lir;
436         int l, rc, index, count = 0;
437         struct cb_data *cbd = (struct cb_data*)data;
438
439         if (cbd->init) {
440                 out = cbd->out;
441                 remains = cbd->remains;
442                 cbd->init = 0;
443         }
444         ctxt = cbd->ctxt;
445
446         if (!(cat->lgh_hdr->llh_flags & LLOG_F_IS_CAT))
447                 RETURN(-EINVAL);
448
449         lir = (struct llog_logid_rec *)rec;
450         logid = &lir->lid_id;
451         rc = llog_open(ctxt, &handle, logid, NULL, 0);
452         if (rc)
453                 RETURN(-EINVAL);
454         rc = llog_init_handle(handle, 0, NULL);
455         if (rc)
456                 GOTO(out_close, rc);
457
458         for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
459                 if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
460                         count++;
461         }
462
463         l = snprintf(out, remains, "\t[Log ID]: #"LPX64"#"LPX64"#%08x\n"
464                      "\tLog Size: %llu\n\tLast Index: %d\n"
465                      "\tUncanceled Records: %d\n",
466                      logid->lgl_oid, logid->lgl_ogr, logid->lgl_ogen,
467                      handle->lgh_file->f_dentry->d_inode->i_size,
468                      handle->lgh_last_idx, count);
469         out += l;
470         remains -= l;
471         cbd->out = out;
472         cbd->remains = remains;
473         if (remains <= 0) {
474                 CWARN("Not enough memory\n");
475                 rc = -ENOMEM;
476         }
477
478 out_close:
479         llog_close(handle);
480         RETURN(rc);
481 }
482
483 static int llog_catinfo_deletions(struct obd_device *obd, char *buf,
484                                   int buf_len)
485 {
486         struct mds_obd *mds = &obd->u.mds;
487         struct llog_handle *handle;
488         struct lvfs_run_ctxt saved;
489         int size, i, count;
490         struct llog_catid *idarray;
491         struct llog_logid *id;
492         char name[32] = CATLIST;
493         int rc;
494         struct cb_data data;
495         struct llog_ctxt *ctxt = llog_get_context(&obd->obd_llogs, 
496                                                   LLOG_CONFIG_ORIG_CTXT);
497
498         if (ctxt == NULL || mds == NULL)
499                 RETURN(-EOPNOTSUPP);
500
501         count = mds->mds_lov_desc.ld_tgt_count;
502         size = sizeof(*idarray) * count;
503
504         OBD_ALLOC(idarray, size);
505         if (!idarray)
506                 RETURN(-ENOMEM);
507
508         rc = llog_get_cat_list(&obd->obd_lvfs_ctxt, obd->obd_fsops, 
509                                name, count, idarray);
510         if (rc)
511                 GOTO(out_free, rc);
512
513         push_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
514
515         data.ctxt = ctxt;
516         data.out = buf;
517         data.remains = buf_len;
518         for (i = 0; i < count; i++) {
519                 int l, index, uncanceled = 0;
520
521                 id = &idarray[i].lci_logid;
522                 rc = llog_open(ctxt, &handle, id, NULL, 0);
523                 if (rc)
524                         GOTO(out_pop, rc);
525                 rc = llog_init_handle(handle, 0, NULL);
526                 if (rc) {
527                         llog_close(handle);
528                         GOTO(out_pop, rc = -ENOENT);
529                 }
530                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
531                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
532                                 uncanceled++;
533                 }
534                 l = snprintf(data.out, data.remains,
535                              "\n[Catlog ID]: #"LPX64"#"LPX64"#%08x  "
536                              "[Log Count]: %d\n",
537                              id->lgl_oid, id->lgl_ogr, id->lgl_ogen,
538                              uncanceled);
539
540                 data.out += l;
541                 data.remains -= l;
542                 data.init = 1;
543
544                 llog_process(handle, llog_catinfo_cb, &data, NULL);
545                 llog_close(handle);
546
547                 if (data.remains <= 0)
548                         break;
549         }
550 out_pop:
551         pop_ctxt(&saved, ctxt->loc_lvfs_ctxt, NULL);
552 out_free:
553         OBD_FREE(idarray, size);
554         RETURN(rc);
555 }
556
557 int llog_catinfo(struct ptlrpc_request *req)
558 {
559         struct obd_export *exp = req->rq_export;
560         struct obd_device *obd = exp->exp_obd;
561         char *keyword;
562         char *buf, *reply;
563         int rc, buf_len = LLOG_CHUNK_SIZE;
564
565         OBD_ALLOC(buf, buf_len);
566         if (buf == NULL)
567                 return -ENOMEM;
568         memset(buf, 0, buf_len);
569
570         keyword = lustre_msg_string(req->rq_reqmsg, 0, 0);
571
572         if (strcmp(keyword, "config") == 0) {
573                 char *client = lustre_msg_string(req->rq_reqmsg, 1, 0);
574                 rc = llog_catinfo_config(obd, buf, buf_len, client);
575         } else if (strcmp(keyword, "deletions") == 0) {
576                 rc = llog_catinfo_deletions(obd, buf, buf_len);
577         } else {
578                 rc = -EOPNOTSUPP;
579         }
580
581         rc = lustre_pack_reply(req, 1, &buf_len, NULL);
582         if (rc)
583                 GOTO(out_free, rc = -ENOMEM);
584
585         reply = lustre_msg_buf(req->rq_repmsg, 0, buf_len);
586         if (strlen(buf) == 0)
587                 sprintf(buf, "%s", "No log informations\n");
588         memcpy(reply, buf, buf_len);
589
590 out_free:
591         OBD_FREE(buf, buf_len);
592         return rc;
593 }
594
595 #else /* !__KERNEL__ */
596 int llog_origin_handle_open(struct ptlrpc_request *req)
597 {
598         LBUG();
599         return 0;
600 }
601 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
602 {
603         LBUG();
604         return 0;
605 }
606 int llog_origin_handle_next_block(struct ptlrpc_request *req)
607 {
608         LBUG();
609         return 0;
610 }
611 int llog_origin_handle_read_header(struct ptlrpc_request *req)
612 {
613         LBUG();
614         return 0;
615 }
616 int llog_origin_handle_close(struct ptlrpc_request *req)
617 {
618         LBUG();
619         return 0;
620 }
621 int llog_origin_handle_cancel(struct ptlrpc_request *req)
622 {
623         LBUG();
624         return 0;
625 }
626 #endif