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 the Lustre file system, http://www.lustre.org
   8  *   Lustre is a trademark of Cluster File Systems, Inc.
   9  *
  10  *   You may have signed or agreed to another license before downloading
  11  *   this software.  If so, you are bound by the terms and conditions
  12  *   of that agreement, and the following does not apply to you.  See the
  13  *   LICENSE file included with this distribution for more information.
  14  *
  15  *   If you did not agree to a different license, then this copy of Lustre
  16  *   is open source software; you can redistribute it and/or modify it
  17  *   under the terms of version 2 of the GNU General Public License as
  18  *   published by the Free Software Foundation.
  19  *
  20  *   In either case, Lustre is distributed in the hope that it will be
  21  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  22  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  23  *   license text for more details.
  24  *
  25  *  remote api for llog - server side
  26  *
  27  */
  28
  29 #define DEBUG_SUBSYSTEM S_LOG
  30
  31 #ifndef EXPORT_SYMTAB
  32 #define EXPORT_SYMTAB
  33 #endif
  34
  35 #ifndef __KERNEL__
  36 #include <liblustre.h>
  37 #endif
  38
  39 #include <obd_class.h>
  40 #include <lustre_log.h>
  41 #include <lustre_net.h>
  42 #include <libcfs/list.h>
  43 #include <lustre_fsfilt.h>
  44
  45 #if defined(__KERNEL__) && defined(LUSTRE_LOG_SERVER)
  46
  47 int llog_origin_handle_create(struct ptlrpc_request *req)
  48 {
  49         struct obd_export *exp = req->rq_export;
  50         struct obd_device *obd = exp->exp_obd;
  51         struct obd_device *disk_obd;
  52         struct llog_handle  *loghandle;
  53         struct llogd_body *body;
  54         struct lvfs_run_ctxt saved;
  55         struct llog_logid *logid = NULL;
  56         struct llog_ctxt *ctxt;
  57         char * name = NULL;
  58         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
  59         int rc, rc2;
  60         ENTRY;
  61
  62         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
  63                                  lustre_swab_llogd_body);
  64         if (body == NULL) {
  65                 CERROR ("Can't unpack llogd_body\n");
  66                 GOTO(out, rc =-EFAULT);
  67         }
  68
  69         if (body->lgd_logid.lgl_oid > 0)
  70                 logid = &body->lgd_logid;
  71
  72         if (lustre_msg_bufcount(req->rq_reqmsg) > 2) {
  73                 name = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF + 1, 0);
  74                 if (name == NULL) {
  75                         CERROR("Can't unpack name\n");
  76                         GOTO(out, rc = -EFAULT);
  77                 }
  78                 CDEBUG(D_INFO, "opening log %s\n", name);
  79         }
  80
  81         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
  82         if (ctxt == NULL)
  83                 GOTO(out, rc = -EINVAL);
  84         disk_obd = ctxt->loc_exp->exp_obd;
  85         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
  86
  87         rc = llog_create(ctxt, &loghandle, logid, name);
  88         if (rc)
  89                 GOTO(out_pop, rc);
  90
  91         rc = lustre_pack_reply(req, 2, size, NULL);
  92         if (rc)
  93                 GOTO(out_close, rc = -ENOMEM);
  94
  95         body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*body));
  96         body->lgd_logid = loghandle->lgh_id;
  97
  98 out_close:
  99         rc2 = llog_close(loghandle);
 100         if (!rc)
 101                 rc = rc2;
 102 out_pop:
 103         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 104 out:
 105         RETURN(rc);
 106 }
 107
 108 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 109 {
 110         struct obd_export *exp = req->rq_export;
 111         struct obd_device *obd = exp->exp_obd;
 112         struct obd_device *disk_obd;
 113         struct llog_handle  *loghandle;
 114         struct llogd_body *body;
 115         struct lvfs_run_ctxt saved;
 116         struct llog_logid *logid = NULL;
 117         struct llog_ctxt *ctxt;
 118         int size[] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 119         int rc;
 120         __u32 flags;
 121         ENTRY;
 122
 123         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 124                                  lustre_swab_llogd_body);
 125         if (body == NULL) {
 126                 CERROR ("Can't unpack llogd_body\n");
 127                 GOTO(out, rc =-EFAULT);
 128         }
 129
 130         if (body->lgd_logid.lgl_oid > 0)
 131                 logid = &body->lgd_logid;
 132
 133         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 134         if (ctxt == NULL)
 135                 GOTO(out, rc = -EINVAL);
 136         disk_obd = ctxt->loc_exp->exp_obd;
 137         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 138
 139         rc = llog_create(ctxt, &loghandle, logid, NULL);
 140         if (rc)
 141                 GOTO(out_pop, rc);
 142
 143         rc = lustre_pack_reply(req, 2, size, NULL);
 144         if (rc)
 145                 GOTO(out_close, rc = -ENOMEM);
 146
 147         body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof (*body));
 148         body->lgd_logid = loghandle->lgh_id;
 149         flags = body->lgd_llh_flags;
 150         rc = llog_init_handle(loghandle, LLOG_F_IS_PLAIN, NULL);
 151         if (rc)
 152                 GOTO(out_close, rc);
 153         rc = llog_destroy(loghandle);
 154         if (rc)
 155                 GOTO(out_close, rc);
 156         llog_free_handle(loghandle);
 157
 158 out_close:
 159         if (rc)
 160                 llog_close(loghandle);
 161 out_pop:
 162         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 163 out:
 164         RETURN(rc);
 165 }
 166
 167 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 168 {
 169         struct obd_export *exp = req->rq_export;
 170         struct obd_device *obd = exp->exp_obd;
 171         struct obd_device *disk_obd;
 172         struct llog_handle  *loghandle;
 173         struct llogd_body *body;
 174         struct lvfs_run_ctxt saved;
 175         struct llog_ctxt *ctxt;
 176         __u32 flags;
 177         __u8 *buf;
 178         void * ptr;
 179         int size[3] = { sizeof(struct ptlrpc_body),
 180                         sizeof(*body),
 181                         LLOG_CHUNK_SIZE };
 182         int rc, rc2;
 183         ENTRY;
 184
 185         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 186                                   lustre_swab_llogd_body);
 187         if (body == NULL) {
 188                 CERROR ("Can't unpack llogd_body\n");
 189                 GOTO(out, rc =-EFAULT);
 190         }
 191
 192         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 193         if (!buf)
 194                 GOTO(out, rc = -ENOMEM);
 195
 196         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 197         if (ctxt == NULL)
 198                 GOTO(out_free, rc = -EINVAL);
 199         disk_obd = ctxt->loc_exp->exp_obd;
 200         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 201
 202         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 203         if (rc)
 204                 GOTO(out_pop, rc);
 205
 206         flags = body->lgd_llh_flags;
 207         rc = llog_init_handle(loghandle, flags, NULL);
 208         if (rc)
 209                 GOTO(out_close, rc);
 210
 211         memset(buf, 0, LLOG_CHUNK_SIZE);
 212         rc = llog_next_block(loghandle, &body->lgd_saved_index,
 213                              body->lgd_index,
 214                              &body->lgd_cur_offset, buf, LLOG_CHUNK_SIZE);
 215         if (rc)
 216                 GOTO(out_close, rc);
 217
 218
 219         rc = lustre_pack_reply(req, 3, size, NULL);
 220         if (rc)
 221                 GOTO(out_close, rc = -ENOMEM);
 222
 223         ptr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof (body));
 224         memcpy(ptr, body, sizeof(*body));
 225
 226         ptr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF + 1, LLOG_CHUNK_SIZE);
 227         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
 228
 229 out_close:
 230         rc2 = llog_close(loghandle);
 231         if (!rc)
 232                 rc = rc2;
 233
 234 out_pop:
 235         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 236 out_free:
 237         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 238 out:
 239         RETURN(rc);
 240 }
 241
 242 int llog_origin_handle_prev_block(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 obd_device *disk_obd;
 249         struct lvfs_run_ctxt saved;
 250         struct llog_ctxt *ctxt;
 251         __u32 flags;
 252         __u8 *buf;
 253         void * ptr;
 254         int size[] = { sizeof(struct ptlrpc_body),
 255                        sizeof(*body),
 256                        LLOG_CHUNK_SIZE };
 257         int rc, rc2;
 258         ENTRY;
 259
 260         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 261                                   lustre_swab_llogd_body);
 262         if (body == NULL) {
 263                 CERROR ("Can't unpack llogd_body\n");
 264                 GOTO(out, rc =-EFAULT);
 265         }
 266
 267         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 268         if (!buf)
 269                 GOTO(out, rc = -ENOMEM);
 270
 271         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 272         LASSERT(ctxt != NULL);
 273         disk_obd = ctxt->loc_exp->exp_obd;
 274         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 275
 276         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 277         if (rc)
 278                 GOTO(out_pop, rc);
 279
 280         flags = body->lgd_llh_flags;
 281         rc = llog_init_handle(loghandle, flags, NULL);
 282         if (rc)
 283                 GOTO(out_close, rc);
 284
 285         memset(buf, 0, LLOG_CHUNK_SIZE);
 286         rc = llog_prev_block(loghandle, body->lgd_index,
 287                              buf, LLOG_CHUNK_SIZE);
 288         if (rc)
 289                 GOTO(out_close, rc);
 290
 291
 292         rc = lustre_pack_reply(req, 3, size, NULL);
 293         if (rc)
 294                 GOTO(out_close, rc = -ENOMEM);
 295
 296         ptr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(body));
 297         memcpy(ptr, body, sizeof(*body));
 298
 299         ptr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF+1, LLOG_CHUNK_SIZE);
 300         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
 301
 302 out_close:
 303         rc2 = llog_close(loghandle);
 304         if (!rc)
 305                 rc = rc2;
 306
 307 out_pop:
 308         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 309         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 310 out:
 311         RETURN(rc);
 312 }
 313
 314 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 315 {
 316         struct obd_export *exp = req->rq_export;
 317         struct obd_device *obd = exp->exp_obd;
 318         struct obd_device *disk_obd;
 319         struct llog_handle  *loghandle;
 320         struct llogd_body *body;
 321         struct llog_log_hdr *hdr;
 322         struct lvfs_run_ctxt saved;
 323         struct llog_ctxt *ctxt;
 324         __u32 flags;
 325         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*hdr) };
 326         int rc, rc2;
 327         ENTRY;
 328
 329         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 330                                   lustre_swab_llogd_body);
 331         if (body == NULL) {
 332                 CERROR ("Can't unpack llogd_body\n");
 333                 GOTO(out, rc =-EFAULT);
 334         }
 335
 336         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 337         if (ctxt == NULL)
 338                 GOTO(out, rc = -EINVAL);
 339         disk_obd = ctxt->loc_exp->exp_obd;
 340         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 341
 342         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 343         if (rc)
 344                 GOTO(out_pop, rc);
 345
 346         /* init_handle reads the header */
 347         flags = body->lgd_llh_flags;
 348         rc = llog_init_handle(loghandle, flags, NULL);
 349         if (rc)
 350                 GOTO(out_close, rc);
 351
 352         rc = lustre_pack_reply(req, 2, size, NULL);
 353         if (rc)
 354                 GOTO(out_close, rc = -ENOMEM);
 355
 356         hdr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*hdr));
 357         memcpy(hdr, loghandle->lgh_hdr, sizeof(*hdr));
 358
 359 out_close:
 360         rc2 = llog_close(loghandle);
 361         if (!rc)
 362                 rc = rc2;
 363
 364 out_pop:
 365         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 366
 367 out:
 368         RETURN(rc);
 369 }
 370
 371 int llog_origin_handle_close(struct ptlrpc_request *req)
 372 {
 373         int rc;
 374
 375         rc = 0;
 376
 377         RETURN(rc);
 378 }
 379
 380 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 381 {
 382         struct obd_device *obd = req->rq_export->exp_obd;
 383         struct obd_device *disk_obd;
 384         struct llog_cookie *logcookies;
 385         struct llog_ctxt *ctxt;
 386         int num_cookies, rc = 0, err, i;
 387         struct lvfs_run_ctxt saved;
 388         struct llog_handle *cathandle;
 389         struct inode *inode;
 390         void *handle;
 391         ENTRY;
 392
 393         logcookies = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF,
 394                                     sizeof(*logcookies));
 395         num_cookies = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF) /
 396                       sizeof(*logcookies);
 397         if (logcookies == NULL || num_cookies == 0) {
 398                 DEBUG_REQ(D_HA, req, "no cookies sent");
 399                 RETURN(-EFAULT);
 400         }
 401
 402         ctxt = llog_get_context(obd, logcookies->lgc_subsys);
 403         if (ctxt == NULL) {
 404                 CWARN("llog subsys not setup or already cleanup\n");
 405                 RETURN(-ENOENT);
 406         }
 407
 408         disk_obd = ctxt->loc_exp->exp_obd;
 409         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 410         for (i = 0; i < num_cookies; i++, logcookies++) {
 411                 cathandle = ctxt->loc_handle;
 412                 LASSERT(cathandle != NULL);
 413                 inode = cathandle->lgh_file->f_dentry->d_inode;
 414
 415                 handle = fsfilt_start_log(disk_obd, inode,
 416                                           FSFILT_OP_CANCEL_UNLINK, NULL, 1);
 417                 if (IS_ERR(handle)) {
 418                         CERROR("fsfilt_start failed: %ld\n", PTR_ERR(handle));
 419                         GOTO(pop_ctxt, rc = PTR_ERR(handle));
 420                 }
 421
 422                 rc = llog_cat_cancel_records(cathandle, 1, logcookies);
 423
 424                 err = fsfilt_commit(disk_obd, inode, handle, 0);
 425                 if (err) {
 426                         CERROR("error committing transaction: %d\n", err);
 427                         if (!rc)
 428                                 rc = err;
 429                         GOTO(pop_ctxt, rc);
 430                 }
 431         }
 432 pop_ctxt:
 433         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 434         if (rc)
 435                 CERROR("cancel %d llog-records failed: %d\n", num_cookies, rc);
 436         else
 437                 CDEBUG(D_HA, "cancel %d llog-records\n", num_cookies);
 438
 439         RETURN(rc);
 440 }
 441 EXPORT_SYMBOL(llog_origin_handle_cancel);
 442
 443 static int llog_catinfo_config(struct obd_device *obd, char *buf, int buf_len,
 444                                char *client)
 445 {
 446         struct mds_obd *mds = &obd->u.mds;
 447         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 448         struct lvfs_run_ctxt saved;
 449         struct llog_handle *handle = NULL;
 450         char name[4][64];
 451         int rc, i, l, remains = buf_len;
 452         char *out = buf;
 453
 454         if (ctxt == NULL || mds == NULL)
 455                 RETURN(-EOPNOTSUPP);
 456
 457         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 458
 459         sprintf(name[0], "%s", mds->mds_profile);
 460         sprintf(name[1], "%s-clean", mds->mds_profile);
 461         sprintf(name[2], "%s", client);
 462         sprintf(name[3], "%s-clean", client);
 463
 464         for (i = 0; i < 4; i++) {
 465                 int index, uncanceled = 0;
 466                 rc = llog_create(ctxt, &handle, NULL, name[i]);
 467                 if (rc)
 468                         GOTO(out_pop, rc);
 469                 rc = llog_init_handle(handle, 0, NULL);
 470                 if (rc) {
 471                         llog_close(handle);
 472                         GOTO(out_pop, rc = -ENOENT);
 473                 }
 474
 475                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index ++) {
 476                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 477                                 uncanceled++;
 478                 }
 479
 480                 l = snprintf(out, remains, "[Log Name]: %s\nLog Size: %llu\n"
 481                              "Last Index: %d\nUncanceled Records: %d\n\n",
 482                              name[i],
 483                              handle->lgh_file->f_dentry->d_inode->i_size,
 484                              handle->lgh_last_idx, uncanceled);
 485                 out += l;
 486                 remains -= l;
 487
 488                 llog_close(handle);
 489                 if (remains <= 0)
 490                         break;
 491         }
 492 out_pop:
 493         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 494         RETURN(rc);
 495 }
 496
 497 struct cb_data {
 498         struct llog_ctxt *ctxt;
 499         char *out;
 500         int  remains;
 501         int  init;
 502 };
 503
 504 static int llog_catinfo_cb(struct llog_handle *cat,
 505                            struct llog_rec_hdr *rec, void *data)
 506 {
 507         static char *out = NULL;
 508         static int remains = 0;
 509         struct llog_ctxt *ctxt;
 510         struct llog_handle *handle;
 511         struct llog_logid *logid;
 512         struct llog_logid_rec *lir;
 513         int l, rc, index, count = 0;
 514         struct cb_data *cbd = (struct cb_data*)data;
 515
 516         if (cbd->init) {
 517                 out = cbd->out;
 518                 remains = cbd->remains;
 519                 cbd->init = 0;
 520         }
 521         ctxt = cbd->ctxt;
 522
 523         if (!(cat->lgh_hdr->llh_flags & LLOG_F_IS_CAT))
 524                 RETURN(-EINVAL);
 525
 526         lir = (struct llog_logid_rec *)rec;
 527         logid = &lir->lid_id;
 528         rc = llog_create(ctxt, &handle, logid, NULL);
 529         if (rc)
 530                 RETURN(-EINVAL);
 531         rc = llog_init_handle(handle, 0, NULL);
 532         if (rc)
 533                 GOTO(out_close, rc);
 534
 535         for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 536                 if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 537                         count++;
 538         }
 539
 540         l = snprintf(out, remains, "\t[Log ID]: #"LPX64"#"LPX64"#%08x\n"
 541                      "\tLog Size: %llu\n\tLast Index: %d\n"
 542                      "\tUncanceled Records: %d\n",
 543                      logid->lgl_oid, logid->lgl_ogr, logid->lgl_ogen,
 544                      handle->lgh_file->f_dentry->d_inode->i_size,
 545                      handle->lgh_last_idx, count);
 546         out += l;
 547         remains -= l;
 548         cbd->out = out;
 549         cbd->remains = remains;
 550         if (remains <= 0) {
 551                 CWARN("Not enough memory\n");
 552                 rc = -ENOMEM;
 553         }
 554
 555 out_close:
 556         llog_close(handle);
 557         RETURN(rc);
 558 }
 559
 560 static int llog_catinfo_deletions(struct obd_device *obd, char *buf,
 561                                   int buf_len)
 562 {
 563         struct mds_obd *mds = &obd->u.mds;
 564         struct llog_handle *handle;
 565         struct lvfs_run_ctxt saved;
 566         int size, i, count;
 567         struct llog_catid *idarray;
 568         struct llog_logid *id;
 569         char name[32] = CATLIST;
 570         int rc;
 571         struct cb_data data;
 572         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 573
 574         if (ctxt == NULL || mds == NULL)
 575                 RETURN(-EOPNOTSUPP);
 576
 577         count = mds->mds_lov_desc.ld_tgt_count;
 578         size = sizeof(*idarray) * count;
 579
 580         OBD_ALLOC(idarray, size);
 581         if (!idarray)
 582                 RETURN(-ENOMEM);
 583
 584         rc = llog_get_cat_list(obd, obd, name, count, idarray);
 585         if (rc)
 586                 GOTO(out_free, rc);
 587
 588         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 589
 590         data.ctxt = ctxt;
 591         data.out = buf;
 592         data.remains = buf_len;
 593         for (i = 0; i < count; i++) {
 594                 int l, index, uncanceled = 0;
 595
 596                 id = &idarray[i].lci_logid;
 597                 rc = llog_create(ctxt, &handle, id, NULL);
 598                 if (rc)
 599                         GOTO(out_pop, rc);
 600                 rc = llog_init_handle(handle, 0, NULL);
 601                 if (rc) {
 602                         llog_close(handle);
 603                         GOTO(out_pop, rc = -ENOENT);
 604                 }
 605                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 606                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 607                                 uncanceled++;
 608                 }
 609                 l = snprintf(data.out, data.remains,
 610                              "\n[Catlog ID]: #"LPX64"#"LPX64"#%08x  "
 611                              "[Log Count]: %d\n",
 612                              id->lgl_oid, id->lgl_ogr, id->lgl_ogen,
 613                              uncanceled);
 614
 615                 data.out += l;
 616                 data.remains -= l;
 617                 data.init = 1;
 618
 619                 llog_process(handle, llog_catinfo_cb, &data, NULL);
 620                 llog_close(handle);
 621
 622                 if (data.remains <= 0)
 623                         break;
 624         }
 625 out_pop:
 626         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 627 out_free:
 628         OBD_FREE(idarray, size);
 629         RETURN(rc);
 630 }
 631
 632 int llog_catinfo(struct ptlrpc_request *req)
 633 {
 634         struct obd_export *exp = req->rq_export;
 635         struct obd_device *obd = exp->exp_obd;
 636         char *keyword;
 637         char *buf, *reply;
 638         int rc, buf_len = LLOG_CHUNK_SIZE;
 639         int size[2] = { sizeof(struct ptlrpc_body), buf_len };
 640
 641         OBD_ALLOC(buf, buf_len);
 642         if (buf == NULL)
 643                 return -ENOMEM;
 644         memset(buf, 0, buf_len);
 645
 646         keyword = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF, 0);
 647
 648         if (strcmp(keyword, "config") == 0) {
 649                 char *client = lustre_msg_string(req->rq_reqmsg,
 650                                                  REQ_REC_OFF + 1, 0);
 651                 rc = llog_catinfo_config(obd, buf, buf_len, client);
 652         } else if (strcmp(keyword, "deletions") == 0) {
 653                 rc = llog_catinfo_deletions(obd, buf, buf_len);
 654         } else {
 655                 rc = -EOPNOTSUPP;
 656         }
 657
 658         rc = lustre_pack_reply(req, 2, size, NULL);
 659         if (rc)
 660                 GOTO(out_free, rc = -ENOMEM);
 661
 662         reply = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, buf_len);
 663         if (strlen(buf) == 0)
 664                 sprintf(buf, "%s", "No log informations\n");
 665         memcpy(reply, buf, buf_len);
 666
 667 out_free:
 668         OBD_FREE(buf, buf_len);
 669         return rc;
 670 }
 671
 672 #else /* !__KERNEL__ */
 673 int llog_origin_handle_create(struct ptlrpc_request *req)
 674 {
 675         LBUG();
 676         return 0;
 677 }
 678
 679 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 680 {
 681         LBUG();
 682         return 0;
 683 }
 684
 685 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 686 {
 687         LBUG();
 688         return 0;
 689 }
 690 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 691 {
 692         LBUG();
 693         return 0;
 694 }
 695 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 696 {
 697         LBUG();
 698         return 0;
 699 }
 700 int llog_origin_handle_close(struct ptlrpc_request *req)
 701 {
 702         LBUG();
 703         return 0;
 704 }
 705 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 706 {
 707         LBUG();
 708         return 0;
 709 }
 710 #endif