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  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  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_server.c
  37  *
  38  * remote api for llog - server side
  39  *
  40  * Author: Andreas Dilger <adilger@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_LOG
  44
  45 #ifndef EXPORT_SYMTAB
  46 #define EXPORT_SYMTAB
  47 #endif
  48
  49 #ifndef __KERNEL__
  50 #include <liblustre.h>
  51 #endif
  52
  53 #include <obd_class.h>
  54 #include <lustre_log.h>
  55 #include <lustre_net.h>
  56 #include <libcfs/list.h>
  57 #include <lustre_fsfilt.h>
  58
  59 #if defined(__KERNEL__) && defined(LUSTRE_LOG_SERVER)
  60
  61 int llog_origin_handle_create(struct ptlrpc_request *req)
  62 {
  63         struct obd_export    *exp = req->rq_export;
  64         struct obd_device    *obd = exp->exp_obd;
  65         struct obd_device    *disk_obd;
  66         struct llog_handle   *loghandle;
  67         struct llogd_body    *body;
  68         struct lvfs_run_ctxt  saved;
  69         struct llog_logid    *logid = NULL;
  70         struct llog_ctxt     *ctxt;
  71         char                 *name = NULL;
  72         int                   rc, rc2;
  73         ENTRY;
  74
  75         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
  76         if (body == NULL)
  77                 RETURN(-EFAULT);
  78
  79         if (body->lgd_logid.lgl_oid > 0)
  80                 logid = &body->lgd_logid;
  81
  82         if (req_capsule_field_present(&req->rq_pill, &RMF_NAME, RCL_CLIENT)) {
  83                 name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
  84                 if (name == NULL)
  85                         RETURN(-EFAULT);
  86                 CDEBUG(D_INFO, "opening log %s\n", name);
  87         }
  88
  89         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
  90         if (ctxt == NULL)
  91                 RETURN(-ENODEV);
  92         disk_obd = ctxt->loc_exp->exp_obd;
  93         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
  94
  95         rc = llog_create(ctxt, &loghandle, logid, name);
  96         if (rc)
  97                 GOTO(out_pop, rc);
  98
  99         rc = req_capsule_server_pack(&req->rq_pill);
 100         if (rc)
 101                 GOTO(out_close, rc = -ENOMEM);
 102
 103         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 104         body->lgd_logid = loghandle->lgh_id;
 105
 106         GOTO(out_close, rc);
 107 out_close:
 108         rc2 = llog_close(loghandle);
 109         if (!rc)
 110                 rc = rc2;
 111 out_pop:
 112         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 113         llog_ctxt_put(ctxt);
 114         return rc;
 115 }
 116
 117 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 118 {
 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;
 127         __u32                 flags;
 128         int                   rc;
 129         ENTRY;
 130
 131         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 132         if (body == NULL)
 133                 RETURN(-EFAULT);
 134
 135         if (body->lgd_logid.lgl_oid > 0)
 136                 logid = &body->lgd_logid;
 137
 138         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 139         if (ctxt == NULL)
 140                 RETURN(-ENODEV);
 141
 142         disk_obd = ctxt->loc_exp->exp_obd;
 143         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 144
 145         rc = llog_create(ctxt, &loghandle, logid, NULL);
 146         if (rc)
 147                 GOTO(out_pop, rc);
 148
 149         rc = req_capsule_server_pack(&req->rq_pill);
 150         if (rc)
 151                 GOTO(out_close, rc = -ENOMEM);
 152
 153         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 154         body->lgd_logid = loghandle->lgh_id;
 155         flags = body->lgd_llh_flags;
 156         rc = llog_init_handle(loghandle, LLOG_F_IS_PLAIN, NULL);
 157         if (rc)
 158                 GOTO(out_close, rc);
 159         rc = llog_destroy(loghandle);
 160         if (rc)
 161                 GOTO(out_close, rc);
 162         llog_free_handle(loghandle);
 163         GOTO(out_close, rc);
 164 out_close:
 165         if (rc)
 166                 llog_close(loghandle);
 167 out_pop:
 168         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 169         llog_ctxt_put(ctxt);
 170         return rc;
 171 }
 172
 173 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 174 {
 175         struct obd_export   *exp = req->rq_export;
 176         struct obd_device   *obd = exp->exp_obd;
 177         struct obd_device   *disk_obd;
 178         struct llog_handle  *loghandle;
 179         struct llogd_body   *body;
 180         struct llogd_body   *repbody;
 181         struct lvfs_run_ctxt saved;
 182         struct llog_ctxt    *ctxt;
 183         __u32                flags;
 184         __u8                *buf;
 185         void                *ptr;
 186         int                  rc, rc2;
 187         ENTRY;
 188
 189         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 190         if (body == NULL)
 191                 RETURN(-EFAULT);
 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 = -ENODEV);
 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         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 220                              LLOG_CHUNK_SIZE);
 221         rc = req_capsule_server_pack(&req->rq_pill);
 222         if (rc)
 223                 GOTO(out_close, rc = -ENOMEM);
 224
 225         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 226         *repbody = *body;
 227
 228         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 229         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
 230         GOTO(out_close, rc);
 231 out_close:
 232         rc2 = llog_close(loghandle);
 233         if (!rc)
 234                 rc = rc2;
 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 llogd_body    *repbody;
 250         struct obd_device    *disk_obd;
 251         struct lvfs_run_ctxt  saved;
 252         struct llog_ctxt     *ctxt;
 253         __u32                 flags;
 254         __u8                 *buf;
 255         void                 *ptr;
 256         int                   rc, rc2;
 257         ENTRY;
 258
 259         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 260         if (body == NULL)
 261                 RETURN(-EFAULT);
 262
 263         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 264         if (!buf)
 265                 RETURN(-ENOMEM);
 266
 267         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 268         if (ctxt == NULL)
 269                 GOTO(out_free, rc = -ENODEV);
 270
 271         disk_obd = ctxt->loc_exp->exp_obd;
 272         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 273
 274         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 275         if (rc)
 276                 GOTO(out_pop, rc);
 277
 278         flags = body->lgd_llh_flags;
 279         rc = llog_init_handle(loghandle, flags, NULL);
 280         if (rc)
 281                 GOTO(out_close, rc);
 282
 283         memset(buf, 0, LLOG_CHUNK_SIZE);
 284         rc = llog_prev_block(loghandle, body->lgd_index,
 285                              buf, LLOG_CHUNK_SIZE);
 286         if (rc)
 287                 GOTO(out_close, rc);
 288
 289         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 290                              LLOG_CHUNK_SIZE);
 291         rc = req_capsule_server_pack(&req->rq_pill);
 292         if (rc)
 293                 GOTO(out_close, rc = -ENOMEM);
 294
 295         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 296         *repbody = *body;
 297
 298         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 299         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
 300         GOTO(out_close, rc);
 301 out_close:
 302         rc2 = llog_close(loghandle);
 303         if (!rc)
 304                 rc = rc2;
 305
 306 out_pop:
 307         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 308         llog_ctxt_put(ctxt);
 309 out_free:
 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                   rc, rc2;
 326         ENTRY;
 327
 328         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 329         if (body == NULL)
 330                 RETURN(-EFAULT);
 331
 332         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 333         if (ctxt == NULL)
 334                 RETURN(-ENODEV);
 335
 336         disk_obd = ctxt->loc_exp->exp_obd;
 337         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 338
 339         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 340         if (rc)
 341                 GOTO(out_pop, rc);
 342
 343         /*
 344          * llog_init_handle() reads the llog header
 345          */
 346         flags = body->lgd_llh_flags;
 347         rc = llog_init_handle(loghandle, flags, NULL);
 348         if (rc)
 349                 GOTO(out_close, rc);
 350
 351         rc = req_capsule_server_pack(&req->rq_pill);
 352         if (rc)
 353                 GOTO(out_close, rc = -ENOMEM);
 354
 355         hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
 356         *hdr = *loghandle->lgh_hdr;
 357         GOTO(out_close, rc);
 358 out_close:
 359         rc2 = llog_close(loghandle);
 360         if (!rc)
 361                 rc = rc2;
 362 out_pop:
 363         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 364         llog_ctxt_put(ctxt);
 365         return rc;
 366 }
 367
 368 int llog_origin_handle_close(struct ptlrpc_request *req)
 369 {
 370         ENTRY;
 371         /* Nothing to do */
 372         RETURN(0);
 373 }
 374
 375 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 376 {
 377         struct obd_device *obd = req->rq_export->exp_obd;
 378         int num_cookies, rc = 0, err, i, failed = 0;
 379         struct obd_device *disk_obd;
 380         struct llog_cookie *logcookies;
 381         struct llog_ctxt *ctxt = NULL;
 382         struct lvfs_run_ctxt saved;
 383         struct llog_handle *cathandle;
 384         struct inode *inode;
 385         void *handle;
 386         ENTRY;
 387
 388         logcookies = req_capsule_client_get(&req->rq_pill, &RMF_LOGCOOKIES);
 389         num_cookies = req_capsule_get_size(&req->rq_pill, &RMF_LOGCOOKIES,
 390                                            RCL_CLIENT) / sizeof(*logcookies);
 391         if (logcookies == NULL || num_cookies == 0) {
 392                 DEBUG_REQ(D_HA, req, "No llog cookies sent");
 393                 RETURN(-EFAULT);
 394         }
 395
 396         ctxt = llog_get_context(obd, logcookies->lgc_subsys);
 397         if (ctxt == NULL)
 398                 RETURN(-ENODEV);
 399
 400         disk_obd = ctxt->loc_exp->exp_obd;
 401         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 402         for (i = 0; i < num_cookies; i++, logcookies++) {
 403                 cathandle = ctxt->loc_handle;
 404                 LASSERT(cathandle != NULL);
 405                 inode = cathandle->lgh_file->f_dentry->d_inode;
 406
 407                 handle = fsfilt_start_log(disk_obd, inode,
 408                                           FSFILT_OP_CANCEL_UNLINK, NULL, 1);
 409                 if (IS_ERR(handle)) {
 410                         CERROR("fsfilt_start_log() failed: %ld\n",
 411                                PTR_ERR(handle));
 412                         GOTO(pop_ctxt, rc = PTR_ERR(handle));
 413                 }
 414
 415                 rc = llog_cat_cancel_records(cathandle, 1, logcookies);
 416
 417                 /*
 418                  * Do not raise -ENOENT errors for resent rpcs. This rec already
 419                  * might be killed.
 420                  */
 421                 if (rc == -ENOENT &&
 422                     (lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)) {
 423                         /*
 424                          * Do not change this message, reply-single.sh test_59b
 425                          * expects to find this in log.
 426                          */
 427                         CDEBUG(D_RPCTRACE, "RESENT cancel req %p - ignored\n",
 428                                req);
 429                         rc = 0;
 430                 } else if (rc == 0) {
 431                         CDEBUG(D_RPCTRACE, "Canceled %d llog-records\n",
 432                                num_cookies);
 433                 }
 434
 435                 err = fsfilt_commit(disk_obd, inode, handle, 0);
 436                 if (err) {
 437                         CERROR("Error committing transaction: %d\n", err);
 438                         if (!rc)
 439                                 rc = err;
 440                         failed++;
 441                         GOTO(pop_ctxt, rc);
 442                 } else if (rc)
 443                         failed++;
 444         }
 445         GOTO(pop_ctxt, rc);
 446 pop_ctxt:
 447         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 448         if (rc)
 449                 CERROR("Cancel %d of %d llog-records failed: %d\n",
 450                        failed, num_cookies, rc);
 451
 452         llog_ctxt_put(ctxt);
 453         return rc;
 454 }
 455 EXPORT_SYMBOL(llog_origin_handle_cancel);
 456
 457 static int llog_catinfo_config(struct obd_device *obd, char *buf, int buf_len,
 458                                char *client)
 459 {
 460         struct mds_obd       *mds = &obd->u.mds;
 461         struct llog_ctxt     *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 462         struct lvfs_run_ctxt  saved;
 463         struct llog_handle   *handle = NULL;
 464         char                  name[4][64];
 465         int                   rc, i, l, remains = buf_len;
 466         char                 *out = buf;
 467         ENTRY;
 468
 469         if (ctxt == NULL || mds == NULL)
 470                 GOTO(release_ctxt, rc = -ENODEV);
 471
 472         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 473
 474         sprintf(name[0], "%s", mds->mds_profile);
 475         sprintf(name[1], "%s-clean", mds->mds_profile);
 476         sprintf(name[2], "%s", client);
 477         sprintf(name[3], "%s-clean", client);
 478
 479         for (i = 0; i < 4; i++) {
 480                 int index, uncanceled = 0;
 481                 rc = llog_create(ctxt, &handle, NULL, name[i]);
 482                 if (rc)
 483                         GOTO(out_pop, rc);
 484                 rc = llog_init_handle(handle, 0, NULL);
 485                 if (rc) {
 486                         llog_close(handle);
 487                         GOTO(out_pop, rc = -ENOENT);
 488                 }
 489
 490                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index ++) {
 491                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 492                                 uncanceled++;
 493                 }
 494
 495                 l = snprintf(out, remains, "[Log Name]: %s\nLog Size: %llu\n"
 496                              "Last Index: %d\nUncanceled Records: %d\n\n",
 497                              name[i],
 498                              i_size_read(handle->lgh_file->f_dentry->d_inode),
 499                              handle->lgh_last_idx, uncanceled);
 500                 out += l;
 501                 remains -= l;
 502
 503                 llog_close(handle);
 504                 if (remains <= 0)
 505                         break;
 506         }
 507         GOTO(out_pop, rc);
 508 out_pop:
 509         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 510 release_ctxt:
 511         llog_ctxt_put(ctxt);
 512         return rc;
 513 }
 514
 515 struct cb_data {
 516         struct llog_ctxt *ctxt;
 517         char *out;
 518         int  remains;
 519         int  init;
 520 };
 521
 522 static int llog_catinfo_cb(struct llog_handle *cat,
 523                            struct llog_rec_hdr *rec, void *data)
 524 {
 525         static char *out = NULL;
 526         static int remains = 0;
 527         struct llog_ctxt *ctxt = NULL;
 528         struct llog_handle *handle;
 529         struct llog_logid *logid;
 530         struct llog_logid_rec *lir;
 531         int l, rc, index, count = 0;
 532         struct cb_data *cbd = (struct cb_data*)data;
 533         ENTRY;
 534
 535         if (cbd->init) {
 536                 out = cbd->out;
 537                 remains = cbd->remains;
 538                 cbd->init = 0;
 539         }
 540
 541         if (!(cat->lgh_hdr->llh_flags & LLOG_F_IS_CAT))
 542                 RETURN(-EINVAL);
 543
 544         if (!cbd->ctxt)
 545                 RETURN(-ENODEV);
 546
 547         lir = (struct llog_logid_rec *)rec;
 548         logid = &lir->lid_id;
 549         rc = llog_create(ctxt, &handle, logid, NULL);
 550         if (rc)
 551                 RETURN(-EINVAL);
 552         rc = llog_init_handle(handle, 0, NULL);
 553         if (rc)
 554                 GOTO(out_close, rc);
 555
 556         for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 557                 if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 558                         count++;
 559         }
 560
 561         l = snprintf(out, remains, "\t[Log ID]: #"LPX64"#"LPX64"#%08x\n"
 562                      "\tLog Size: %llu\n\tLast Index: %d\n"
 563                      "\tUncanceled Records: %d\n",
 564                      logid->lgl_oid, logid->lgl_ogr, logid->lgl_ogen,
 565                      i_size_read(handle->lgh_file->f_dentry->d_inode),
 566                      handle->lgh_last_idx, count);
 567         out += l;
 568         remains -= l;
 569         cbd->out = out;
 570         cbd->remains = remains;
 571         if (remains <= 0) {
 572                 CWARN("Not enough memory\n");
 573                 rc = -ENOMEM;
 574         }
 575         EXIT;
 576 out_close:
 577         llog_close(handle);
 578         return rc;
 579 }
 580
 581 static int llog_catinfo_deletions(struct obd_device *obd, char *buf,
 582                                   int buf_len)
 583 {
 584         struct mds_obd *mds = &obd->u.mds;
 585         struct llog_handle *handle;
 586         struct lvfs_run_ctxt saved;
 587         int size, i, count;
 588         struct llog_catid *idarray;
 589         char name[32] = CATLIST;
 590         struct cb_data data;
 591         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 592         int rc;
 593         ENTRY;
 594
 595         if (ctxt == NULL || mds == NULL)
 596                 GOTO(release_ctxt, rc = -ENODEV);
 597
 598         count = mds->mds_lov_desc.ld_tgt_count;
 599         size = sizeof(*idarray) * count;
 600
 601         OBD_VMALLOC(idarray, size);
 602         if (!idarray)
 603                 GOTO(release_ctxt, rc = -ENOMEM);
 604
 605         mutex_down(&obd->obd_olg.olg_cat_processing);
 606         rc = llog_get_cat_list(obd, obd, name, 0, count, idarray);
 607         if (rc)
 608                 GOTO(out_free, rc);
 609
 610         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 611
 612         data.ctxt = ctxt;
 613         data.out = buf;
 614         data.remains = buf_len;
 615         for (i = 0; i < count; i++) {
 616                 int l, index, uncanceled = 0;
 617
 618                 rc = llog_create(ctxt, &handle, &idarray[i].lci_logid, NULL);
 619                 if (rc)
 620                         GOTO(out_pop, rc);
 621                 rc = llog_init_handle(handle, 0, NULL);
 622                 if (rc) {
 623                         llog_close(handle);
 624                         GOTO(out_pop, rc = -ENOENT);
 625                 }
 626                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 627                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 628                                 uncanceled++;
 629                 }
 630                 l = snprintf(data.out, data.remains,
 631                              "\n[Catlog ID]: #"LPX64"#"LPX64"#%08x  "
 632                              "[Log Count]: %d\n",
 633                              idarray[i].lci_logid.lgl_oid,
 634                              idarray[i].lci_logid.lgl_ogr,
 635                              idarray[i].lci_logid.lgl_ogen, uncanceled);
 636
 637                 data.out += l;
 638                 data.remains -= l;
 639                 data.init = 1;
 640
 641                 llog_process(handle, llog_catinfo_cb, &data, NULL);
 642                 llog_close(handle);
 643
 644                 if (data.remains <= 0)
 645                         break;
 646         }
 647         EXIT;
 648 out_pop:
 649         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 650 out_free:
 651         mutex_up(&obd->obd_olg.olg_cat_processing);
 652         OBD_VFREE(idarray, size);
 653 release_ctxt:
 654         llog_ctxt_put(ctxt);
 655         return rc;
 656 }
 657
 658 int llog_catinfo(struct ptlrpc_request *req)
 659 {
 660         struct obd_export *exp = req->rq_export;
 661         struct obd_device *obd = exp->exp_obd;
 662         char              *keyword;
 663         char              *buf, *reply;
 664         int                rc;
 665         ENTRY;
 666
 667         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 668         if (buf == NULL)
 669                 RETURN(-ENOMEM);
 670
 671         memset(buf, 0, LLOG_CHUNK_SIZE);
 672
 673         keyword = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 674         LASSERT(keyword);
 675
 676         if (strcmp(keyword, "config") == 0) {
 677                 char *client = req_capsule_client_get(&req->rq_pill,
 678                                                       &RMF_STRING);
 679
 680                 LASSERT(client);
 681                 rc = llog_catinfo_config(obd, buf, LLOG_CHUNK_SIZE, client);
 682         } else if (strcmp(keyword, "deletions") == 0) {
 683                 rc = llog_catinfo_deletions(obd, buf, LLOG_CHUNK_SIZE);
 684         } else {
 685                 rc = -EOPNOTSUPP;
 686         }
 687
 688         req_capsule_set_size(&req->rq_pill, &RMF_STRING, RCL_SERVER,
 689                              LLOG_CHUNK_SIZE);
 690         rc = req_capsule_server_pack(&req->rq_pill);
 691         if (rc)
 692                 GOTO(out_free, rc = -ENOMEM);
 693
 694         reply = req_capsule_server_get(&req->rq_pill, &RMF_STRING);
 695         if (strlen(buf) == 0)
 696                 sprintf(buf, "%s", "No log informations\n");
 697         memcpy(reply, buf, LLOG_CHUNK_SIZE);
 698         EXIT;
 699 out_free:
 700         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 701         return rc;
 702 }
 703
 704 #else /* !__KERNEL__ */
 705 int llog_origin_handle_create(struct ptlrpc_request *req)
 706 {
 707         LBUG();
 708         return 0;
 709 }
 710
 711 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 712 {
 713         LBUG();
 714         return 0;
 715 }
 716
 717 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 718 {
 719         LBUG();
 720         return 0;
 721 }
 722 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 723 {
 724         LBUG();
 725         return 0;
 726 }
 727 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 728 {
 729         LBUG();
 730         return 0;
 731 }
 732 int llog_origin_handle_close(struct ptlrpc_request *req)
 733 {
 734         LBUG();
 735         return 0;
 736 }
 737 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 738 {
 739         LBUG();
 740         return 0;
 741 }
 742 #endif