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                 RETURN(-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                         RETURN(-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                 RETURN(-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         llog_ctxt_put(ctxt);
 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                 RETURN(-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                 RETURN(-EINVAL);
 136
 137         disk_obd = ctxt->loc_exp->exp_obd;
 138         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 139
 140         rc = llog_create(ctxt, &loghandle, logid, NULL);
 141         if (rc)
 142                 GOTO(out_pop, rc);
 143
 144         rc = lustre_pack_reply(req, 2, size, NULL);
 145         if (rc)
 146                 GOTO(out_close, rc = -ENOMEM);
 147
 148         body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof (*body));
 149         body->lgd_logid = loghandle->lgh_id;
 150         flags = body->lgd_llh_flags;
 151         rc = llog_init_handle(loghandle, LLOG_F_IS_PLAIN, NULL);
 152         if (rc)
 153                 GOTO(out_close, rc);
 154         rc = llog_destroy(loghandle);
 155         if (rc)
 156                 GOTO(out_close, rc);
 157         llog_free_handle(loghandle);
 158
 159 out_close:
 160         if (rc)
 161                 llog_close(loghandle);
 162 out_pop:
 163         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 164         llog_ctxt_put(ctxt);
 165         RETURN(rc);
 166 }
 167
 168 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 169 {
 170         struct obd_export *exp = req->rq_export;
 171         struct obd_device *obd = exp->exp_obd;
 172         struct obd_device *disk_obd;
 173         struct llog_handle  *loghandle;
 174         struct llogd_body *body;
 175         struct lvfs_run_ctxt saved;
 176         struct llog_ctxt *ctxt;
 177         __u32 flags;
 178         __u8 *buf;
 179         void * ptr;
 180         int size[3] = { sizeof(struct ptlrpc_body),
 181                         sizeof(*body),
 182                         LLOG_CHUNK_SIZE };
 183         int rc, rc2;
 184         ENTRY;
 185
 186         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 187                                   lustre_swab_llogd_body);
 188         if (body == NULL) {
 189                 CERROR ("Can't unpack llogd_body\n");
 190                 RETURN(-EFAULT);
 191         }
 192
 193         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 194         if (!buf)
 195                 RETURN(-ENOMEM);
 196
 197         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 198         if (ctxt == NULL)
 199                 GOTO(out_free, rc = -EINVAL);
 200         disk_obd = ctxt->loc_exp->exp_obd;
 201         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 202
 203         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 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, 3, size, NULL);
 221         if (rc)
 222                 GOTO(out_close, rc = -ENOMEM);
 223
 224         ptr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof (body));
 225         memcpy(ptr, body, sizeof(*body));
 226
 227         ptr = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF+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, &disk_obd->obd_lvfs_ctxt, NULL);
 237         llog_ctxt_put(ctxt);
 238 out_free:
 239         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 240         RETURN(rc);
 241 }
 242
 243 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 244 {
 245         struct obd_export *exp = req->rq_export;
 246         struct obd_device *obd = exp->exp_obd;
 247         struct llog_handle  *loghandle;
 248         struct llogd_body *body;
 249         struct obd_device *disk_obd;
 250         struct lvfs_run_ctxt saved;
 251         struct llog_ctxt *ctxt;
 252         __u32 flags;
 253         __u8 *buf;
 254         void * ptr;
 255         int size[] = { sizeof(struct ptlrpc_body),
 256                        sizeof(*body),
 257                        LLOG_CHUNK_SIZE };
 258         int rc, rc2;
 259         ENTRY;
 260
 261         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 262                                   lustre_swab_llogd_body);
 263         if (body == NULL) {
 264                 CERROR ("Can't unpack llogd_body\n");
 265                 RETURN(-EFAULT);
 266         }
 267
 268         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 269         if (!buf)
 270                 RETURN(-ENOMEM);
 271
 272         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 273         LASSERT(ctxt != NULL);
 274         disk_obd = ctxt->loc_exp->exp_obd;
 275         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 276
 277         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 278         if (rc)
 279                 GOTO(out_pop, rc);
 280
 281         flags = body->lgd_llh_flags;
 282         rc = llog_init_handle(loghandle, flags, NULL);
 283         if (rc)
 284                 GOTO(out_close, rc);
 285
 286         memset(buf, 0, LLOG_CHUNK_SIZE);
 287         rc = llog_prev_block(loghandle, body->lgd_index,
 288                              buf, LLOG_CHUNK_SIZE);
 289         if (rc)
 290                 GOTO(out_close, rc);
 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         llog_ctxt_put(ctxt);
 310         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 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                 RETURN(-EFAULT);
 334         }
 335
 336         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 337         if (ctxt == NULL)
 338                 RETURN(-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 out_pop:
 364         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 365         llog_ctxt_put(ctxt);
 366         RETURN(rc);
 367 }
 368
 369 int llog_origin_handle_close(struct ptlrpc_request *req)
 370 {
 371         int rc;
 372
 373         rc = 0;
 374
 375         RETURN(rc);
 376 }
 377
 378 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 379 {
 380         struct obd_device *obd = req->rq_export->exp_obd;
 381         struct obd_device *disk_obd;
 382         struct llog_cookie *logcookies;
 383         struct llog_ctxt *ctxt = NULL;
 384         int num_cookies, rc = 0, err, i;
 385         struct lvfs_run_ctxt saved;
 386         struct llog_handle *cathandle;
 387         struct inode *inode;
 388         void *handle;
 389         ENTRY;
 390
 391         logcookies = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF,
 392                                     sizeof(*logcookies));
 393         num_cookies = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF) /
 394                       sizeof(*logcookies);
 395         if (logcookies == NULL || num_cookies == 0) {
 396                 DEBUG_REQ(D_HA, req, "no cookies sent");
 397                 RETURN(-EFAULT);
 398         }
 399
 400         ctxt = llog_get_context(obd, logcookies->lgc_subsys);
 401         if (ctxt == NULL) {
 402                 CWARN("llog subsys not setup or already cleanup\n");
 403                 RETURN(-ENOENT);
 404         }
 405
 406         disk_obd = ctxt->loc_exp->exp_obd;
 407         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 408         for (i = 0; i < num_cookies; i++, logcookies++) {
 409                 cathandle = ctxt->loc_handle;
 410                 LASSERT(cathandle != NULL);
 411                 inode = cathandle->lgh_file->f_dentry->d_inode;
 412
 413                 handle = fsfilt_start_log(disk_obd, inode,
 414                                           FSFILT_OP_CANCEL_UNLINK, NULL, 1);
 415                 if (IS_ERR(handle)) {
 416                         CERROR("fsfilt_start failed: %ld\n", PTR_ERR(handle));
 417                         GOTO(pop_ctxt, rc = PTR_ERR(handle));
 418                 }
 419
 420                 rc = llog_cat_cancel_records(cathandle, 1, logcookies);
 421
 422                 err = fsfilt_commit(disk_obd, inode, handle, 0);
 423                 if (err) {
 424                         CERROR("error committing transaction: %d\n", err);
 425                         if (!rc)
 426                                 rc = err;
 427                         GOTO(pop_ctxt, rc);
 428                 }
 429         }
 430 pop_ctxt:
 431         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 432         if (rc)
 433                 CERROR("cancel %d llog-records failed: %d\n", num_cookies, rc);
 434         else
 435                 CDEBUG(D_RPCTRACE, "cancel %d llog-records\n", num_cookies);
 436
 437         llog_ctxt_put(ctxt);
 438         RETURN(rc);
 439 }
 440 EXPORT_SYMBOL(llog_origin_handle_cancel);
 441
 442 static int llog_catinfo_config(struct obd_device *obd, char *buf, int buf_len,
 443                                char *client)
 444 {
 445         struct mds_obd *mds = &obd->u.mds;
 446         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 447         struct lvfs_run_ctxt saved;
 448         struct llog_handle *handle = NULL;
 449         char name[4][64];
 450         int rc, i, l, remains = buf_len;
 451         char *out = buf;
 452
 453         if (ctxt == NULL || mds == NULL)
 454                 GOTO(release_ctxt, rc = -EOPNOTSUPP);
 455
 456         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 457
 458         sprintf(name[0], "%s", mds->mds_profile);
 459         sprintf(name[1], "%s-clean", mds->mds_profile);
 460         sprintf(name[2], "%s", client);
 461         sprintf(name[3], "%s-clean", client);
 462
 463         for (i = 0; i < 4; i++) {
 464                 int index, uncanceled = 0;
 465                 rc = llog_create(ctxt, &handle, NULL, name[i]);
 466                 if (rc)
 467                         GOTO(out_pop, rc);
 468                 rc = llog_init_handle(handle, 0, NULL);
 469                 if (rc) {
 470                         llog_close(handle);
 471                         GOTO(out_pop, rc = -ENOENT);
 472                 }
 473
 474                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index ++) {
 475                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 476                                 uncanceled++;
 477                 }
 478
 479                 l = snprintf(out, remains, "[Log Name]: %s\nLog Size: %llu\n"
 480                              "Last Index: %d\nUncanceled Records: %d\n\n",
 481                              name[i],
 482                              i_size_read(handle->lgh_file->f_dentry->d_inode),
 483                              handle->lgh_last_idx, uncanceled);
 484                 out += l;
 485                 remains -= l;
 486
 487                 llog_close(handle);
 488                 if (remains <= 0)
 489                         break;
 490         }
 491 out_pop:
 492         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 493 release_ctxt:
 494         llog_ctxt_put(ctxt);
 495         RETURN(rc);
 496 }
 497
 498 struct cb_data {
 499         struct llog_ctxt *ctxt;
 500         char *out;
 501         int  remains;
 502         int  init;
 503 };
 504
 505 static int llog_catinfo_cb(struct llog_handle *cat,
 506                            struct llog_rec_hdr *rec, void *data)
 507 {
 508         static char *out = NULL;
 509         static int remains = 0;
 510         struct llog_ctxt *ctxt = NULL;
 511         struct llog_handle *handle;
 512         struct llog_logid *logid;
 513         struct llog_logid_rec *lir;
 514         int l, rc, index, count = 0;
 515         struct cb_data *cbd = (struct cb_data*)data;
 516
 517         if (cbd->init) {
 518                 out = cbd->out;
 519                 remains = cbd->remains;
 520                 cbd->init = 0;
 521         }
 522
 523         if (!(cat->lgh_hdr->llh_flags & LLOG_F_IS_CAT))
 524                 RETURN(-EINVAL);
 525
 526         if (!cbd->ctxt)
 527                 RETURN(-EINVAL);
 528
 529         lir = (struct llog_logid_rec *)rec;
 530         logid = &lir->lid_id;
 531         rc = llog_create(ctxt, &handle, logid, NULL);
 532         if (rc)
 533                 RETURN(-EINVAL);
 534         rc = llog_init_handle(handle, 0, NULL);
 535         if (rc)
 536                 GOTO(out_close, rc);
 537
 538         for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 539                 if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 540                         count++;
 541         }
 542
 543         l = snprintf(out, remains, "\t[Log ID]: #"LPX64"#"LPX64"#%08x\n"
 544                      "\tLog Size: %llu\n\tLast Index: %d\n"
 545                      "\tUncanceled Records: %d\n",
 546                      logid->lgl_oid, logid->lgl_ogr, logid->lgl_ogen,
 547                      i_size_read(handle->lgh_file->f_dentry->d_inode),
 548                      handle->lgh_last_idx, count);
 549         out += l;
 550         remains -= l;
 551         cbd->out = out;
 552         cbd->remains = remains;
 553         if (remains <= 0) {
 554                 CWARN("Not enough memory\n");
 555                 rc = -ENOMEM;
 556         }
 557
 558 out_close:
 559         llog_close(handle);
 560         RETURN(rc);
 561 }
 562
 563 static int llog_catinfo_deletions(struct obd_device *obd, char *buf,
 564                                   int buf_len)
 565 {
 566         struct mds_obd *mds = &obd->u.mds;
 567         struct llog_handle *handle;
 568         struct lvfs_run_ctxt saved;
 569         int size, i, count;
 570         struct llog_catid *idarray;
 571         struct llog_logid *id;
 572         char name[32] = CATLIST;
 573         int rc;
 574         struct cb_data data;
 575         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 576
 577         if (ctxt == NULL || mds == NULL)
 578                 GOTO(release_ctxt, rc = -EOPNOTSUPP);
 579
 580         count = mds->mds_lov_desc.ld_tgt_count;
 581         size = sizeof(*idarray) * count;
 582
 583         OBD_ALLOC(idarray, size);
 584         if (!idarray)
 585                 GOTO(release_ctxt, rc = -ENOMEM);
 586
 587         rc = llog_get_cat_list(obd, obd, name, count, idarray);
 588         if (rc)
 589                 GOTO(out_free, rc);
 590
 591         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 592
 593         data.ctxt = ctxt;
 594         data.out = buf;
 595         data.remains = buf_len;
 596         for (i = 0; i < count; i++) {
 597                 int l, index, uncanceled = 0;
 598
 599                 id = &idarray[i].lci_logid;
 600                 rc = llog_create(ctxt, &handle, id, NULL);
 601                 if (rc)
 602                         GOTO(out_pop, rc);
 603                 rc = llog_init_handle(handle, 0, NULL);
 604                 if (rc) {
 605                         llog_close(handle);
 606                         GOTO(out_pop, rc = -ENOENT);
 607                 }
 608                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 609                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 610                                 uncanceled++;
 611                 }
 612                 l = snprintf(data.out, data.remains,
 613                              "\n[Catlog ID]: #"LPX64"#"LPX64"#%08x  "
 614                              "[Log Count]: %d\n",
 615                              id->lgl_oid, id->lgl_ogr, id->lgl_ogen,
 616                              uncanceled);
 617
 618                 data.out += l;
 619                 data.remains -= l;
 620                 data.init = 1;
 621
 622                 llog_process(handle, llog_catinfo_cb, &data, NULL);
 623                 llog_close(handle);
 624
 625                 if (data.remains <= 0)
 626                         break;
 627         }
 628 out_pop:
 629         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 630 out_free:
 631         OBD_FREE(idarray, size);
 632 release_ctxt:
 633         llog_ctxt_put(ctxt);
 634
 635         RETURN(rc);
 636 }
 637
 638 int llog_catinfo(struct ptlrpc_request *req)
 639 {
 640         struct obd_export *exp = req->rq_export;
 641         struct obd_device *obd = exp->exp_obd;
 642         char *keyword;
 643         char *buf, *reply;
 644         int rc, buf_len = LLOG_CHUNK_SIZE;
 645         int size[2] = { sizeof(struct ptlrpc_body), buf_len };
 646
 647         OBD_ALLOC(buf, buf_len);
 648         if (buf == NULL)
 649                 return -ENOMEM;
 650         memset(buf, 0, buf_len);
 651
 652         keyword = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF, 0);
 653
 654         if (strcmp(keyword, "config") == 0) {
 655                 char *client = lustre_msg_string(req->rq_reqmsg,
 656                                                  REQ_REC_OFF + 1, 0);
 657                 rc = llog_catinfo_config(obd, buf, buf_len, client);
 658         } else if (strcmp(keyword, "deletions") == 0) {
 659                 rc = llog_catinfo_deletions(obd, buf, buf_len);
 660         } else {
 661                 rc = -EOPNOTSUPP;
 662         }
 663
 664         rc = lustre_pack_reply(req, 2, size, NULL);
 665         if (rc)
 666                 GOTO(out_free, rc = -ENOMEM);
 667
 668         reply = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, buf_len);
 669         if (strlen(buf) == 0)
 670                 sprintf(buf, "%s", "No log informations\n");
 671         memcpy(reply, buf, buf_len);
 672
 673 out_free:
 674         OBD_FREE(buf, buf_len);
 675         return rc;
 676 }
 677
 678 #else /* !__KERNEL__ */
 679 int llog_origin_handle_create(struct ptlrpc_request *req)
 680 {
 681         LBUG();
 682         return 0;
 683 }
 684
 685 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 686 {
 687         LBUG();
 688         return 0;
 689 }
 690
 691 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 692 {
 693         LBUG();
 694         return 0;
 695 }
 696 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 697 {
 698         LBUG();
 699         return 0;
 700 }
 701 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 702 {
 703         LBUG();
 704         return 0;
 705 }
 706 int llog_origin_handle_close(struct ptlrpc_request *req)
 707 {
 708         LBUG();
 709         return 0;
 710 }
 711 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 712 {
 713         LBUG();
 714         return 0;
 715 }
 716 #endif