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 (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2011, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/ptlrpc/llog_server.c
  39  *
  40  * remote api for llog - server side
  41  *
  42  * Author: Andreas Dilger <adilger@clusterfs.com>
  43  */
  44
  45 #define DEBUG_SUBSYSTEM S_LOG
  46
  47 #ifndef EXPORT_SYMTAB
  48 #define EXPORT_SYMTAB
  49 #endif
  50
  51 #ifndef __KERNEL__
  52 #include <liblustre.h>
  53 #endif
  54
  55 #include <obd_class.h>
  56 #include <lustre_log.h>
  57 #include <lustre_net.h>
  58 #include <libcfs/list.h>
  59 #include <lustre_fsfilt.h>
  60
  61 #if defined(__KERNEL__) && defined(LUSTRE_LOG_SERVER)
  62
  63 int llog_origin_handle_create(struct ptlrpc_request *req)
  64 {
  65         struct obd_export    *exp = req->rq_export;
  66         struct obd_device    *obd = exp->exp_obd;
  67         struct obd_device    *disk_obd;
  68         struct llog_handle   *loghandle;
  69         struct llogd_body    *body;
  70         struct lvfs_run_ctxt  saved;
  71         struct llog_logid    *logid = NULL;
  72         struct llog_ctxt     *ctxt;
  73         char                 *name = NULL;
  74         int                   rc, rc2;
  75         ENTRY;
  76
  77         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
  78         if (body == NULL)
  79                 RETURN(-EFAULT);
  80
  81         if (body->lgd_logid.lgl_oid > 0)
  82                 logid = &body->lgd_logid;
  83
  84         if (req_capsule_field_present(&req->rq_pill, &RMF_NAME, RCL_CLIENT)) {
  85                 name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
  86                 if (name == NULL)
  87                         RETURN(-EFAULT);
  88                 CDEBUG(D_INFO, "%s: opening log %s\n", obd->obd_name, name);
  89         }
  90
  91         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
  92         if (ctxt == NULL) {
  93                 CDEBUG(D_WARNING, "%s: no ctxt. group=%p idx=%d name=%s\n",
  94                        obd->obd_name, &obd->obd_olg, body->lgd_ctxt_idx, name);
  95                 RETURN(-ENODEV);
  96         }
  97         disk_obd = ctxt->loc_exp->exp_obd;
  98         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
  99
 100         rc = llog_create(ctxt, &loghandle, logid, name);
 101         if (rc)
 102                 GOTO(out_pop, rc);
 103
 104         rc = req_capsule_server_pack(&req->rq_pill);
 105         if (rc)
 106                 GOTO(out_close, rc = -ENOMEM);
 107
 108         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 109         body->lgd_logid = loghandle->lgh_id;
 110
 111         GOTO(out_close, rc);
 112 out_close:
 113         rc2 = llog_close(loghandle);
 114         if (!rc)
 115                 rc = rc2;
 116 out_pop:
 117         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 118         llog_ctxt_put(ctxt);
 119         return rc;
 120 }
 121
 122 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 123 {
 124         struct obd_export    *exp = req->rq_export;
 125         struct obd_device    *obd = exp->exp_obd;
 126         struct obd_device    *disk_obd;
 127         struct llog_handle   *loghandle;
 128         struct llogd_body    *body;
 129         struct lvfs_run_ctxt  saved;
 130         struct llog_logid    *logid = NULL;
 131         struct llog_ctxt     *ctxt;
 132         int                   rc;
 133         ENTRY;
 134
 135         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 136         if (body == NULL)
 137                 RETURN(-EFAULT);
 138
 139         if (body->lgd_logid.lgl_oid > 0)
 140                 logid = &body->lgd_logid;
 141
 142         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 143         if (ctxt == NULL)
 144                 RETURN(-ENODEV);
 145
 146         disk_obd = ctxt->loc_exp->exp_obd;
 147         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 148
 149         rc = llog_create(ctxt, &loghandle, logid, NULL);
 150         if (rc)
 151                 GOTO(out_pop, rc);
 152
 153         rc = req_capsule_server_pack(&req->rq_pill);
 154         if (rc)
 155                 GOTO(out_close, rc = -ENOMEM);
 156
 157         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 158         body->lgd_logid = loghandle->lgh_id;
 159         rc = llog_init_handle(loghandle, LLOG_F_IS_PLAIN, NULL);
 160         if (rc)
 161                 GOTO(out_close, rc);
 162         rc = llog_destroy(loghandle);
 163         if (rc)
 164                 GOTO(out_close, rc);
 165         llog_free_handle(loghandle);
 166         GOTO(out_close, rc);
 167 out_close:
 168         if (rc)
 169                 llog_close(loghandle);
 170 out_pop:
 171         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 172         llog_ctxt_put(ctxt);
 173         return rc;
 174 }
 175
 176 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 177 {
 178         struct obd_export   *exp = req->rq_export;
 179         struct obd_device   *obd = exp->exp_obd;
 180         struct obd_device   *disk_obd;
 181         struct llog_handle  *loghandle;
 182         struct llogd_body   *body;
 183         struct llogd_body   *repbody;
 184         struct lvfs_run_ctxt saved;
 185         struct llog_ctxt    *ctxt;
 186         __u32                flags;
 187         __u8                *buf;
 188         void                *ptr;
 189         int                  rc, rc2;
 190         ENTRY;
 191
 192         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 193         if (body == NULL)
 194                 RETURN(-EFAULT);
 195
 196         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 197         if (!buf)
 198                 RETURN(-ENOMEM);
 199
 200         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 201         if (ctxt == NULL)
 202                 GOTO(out_free, rc = -ENODEV);
 203         disk_obd = ctxt->loc_exp->exp_obd;
 204         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 205
 206         rc = llog_create(ctxt, &loghandle, &body->lgd_logid, NULL);
 207         if (rc)
 208                 GOTO(out_pop, rc);
 209
 210         flags = body->lgd_llh_flags;
 211         rc = llog_init_handle(loghandle, flags, NULL);
 212         if (rc)
 213                 GOTO(out_close, rc);
 214
 215         memset(buf, 0, LLOG_CHUNK_SIZE);
 216         rc = llog_next_block(loghandle, &body->lgd_saved_index,
 217                              body->lgd_index,
 218                              &body->lgd_cur_offset, buf, LLOG_CHUNK_SIZE);
 219         if (rc)
 220                 GOTO(out_close, rc);
 221
 222         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 223                              LLOG_CHUNK_SIZE);
 224         rc = req_capsule_server_pack(&req->rq_pill);
 225         if (rc)
 226                 GOTO(out_close, rc = -ENOMEM);
 227
 228         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 229         *repbody = *body;
 230
 231         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 232         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
 233         GOTO(out_close, rc);
 234 out_close:
 235         rc2 = llog_close(loghandle);
 236         if (!rc)
 237                 rc = rc2;
 238 out_pop:
 239         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 240         llog_ctxt_put(ctxt);
 241 out_free:
 242         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 243         return rc;
 244 }
 245
 246 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 247 {
 248         struct obd_export    *exp = req->rq_export;
 249         struct obd_device    *obd = exp->exp_obd;
 250         struct llog_handle   *loghandle;
 251         struct llogd_body    *body;
 252         struct llogd_body    *repbody;
 253         struct obd_device    *disk_obd;
 254         struct lvfs_run_ctxt  saved;
 255         struct llog_ctxt     *ctxt;
 256         __u32                 flags;
 257         __u8                 *buf;
 258         void                 *ptr;
 259         int                   rc, rc2;
 260         ENTRY;
 261
 262         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 263         if (body == NULL)
 264                 RETURN(-EFAULT);
 265
 266         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 267         if (!buf)
 268                 RETURN(-ENOMEM);
 269
 270         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 271         if (ctxt == NULL)
 272                 GOTO(out_free, rc = -ENODEV);
 273
 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         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 293                              LLOG_CHUNK_SIZE);
 294         rc = req_capsule_server_pack(&req->rq_pill);
 295         if (rc)
 296                 GOTO(out_close, rc = -ENOMEM);
 297
 298         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 299         *repbody = *body;
 300
 301         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 302         memcpy(ptr, buf, LLOG_CHUNK_SIZE);
 303         GOTO(out_close, rc);
 304 out_close:
 305         rc2 = llog_close(loghandle);
 306         if (!rc)
 307                 rc = rc2;
 308
 309 out_pop:
 310         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 311         llog_ctxt_put(ctxt);
 312 out_free:
 313         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 314         return rc;
 315 }
 316
 317 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 318 {
 319         struct obd_export    *exp = req->rq_export;
 320         struct obd_device    *obd = exp->exp_obd;
 321         struct obd_device    *disk_obd;
 322         struct llog_handle   *loghandle;
 323         struct llogd_body    *body;
 324         struct llog_log_hdr  *hdr;
 325         struct lvfs_run_ctxt  saved;
 326         struct llog_ctxt     *ctxt;
 327         __u32                 flags;
 328         int                   rc, rc2;
 329         ENTRY;
 330
 331         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 332         if (body == NULL)
 333                 RETURN(-EFAULT);
 334
 335         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
 336         if (ctxt == NULL)
 337                 RETURN(-ENODEV);
 338
 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         /*
 347          * llog_init_handle() reads the llog header
 348          */
 349         flags = body->lgd_llh_flags;
 350         rc = llog_init_handle(loghandle, flags, NULL);
 351         if (rc)
 352                 GOTO(out_close, rc);
 353
 354         rc = req_capsule_server_pack(&req->rq_pill);
 355         if (rc)
 356                 GOTO(out_close, rc = -ENOMEM);
 357
 358         hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
 359         *hdr = *loghandle->lgh_hdr;
 360         GOTO(out_close, rc);
 361 out_close:
 362         rc2 = llog_close(loghandle);
 363         if (!rc)
 364                 rc = rc2;
 365 out_pop:
 366         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 367         llog_ctxt_put(ctxt);
 368         return rc;
 369 }
 370
 371 int llog_origin_handle_close(struct ptlrpc_request *req)
 372 {
 373         ENTRY;
 374         /* Nothing to do */
 375         RETURN(0);
 376 }
 377
 378 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 379 {
 380         struct obd_device *obd = req->rq_export->exp_obd;
 381         int num_cookies, rc = 0, err, i, failed = 0;
 382         struct obd_device *disk_obd;
 383         struct llog_cookie *logcookies;
 384         struct llog_ctxt *ctxt = NULL;
 385         struct lvfs_run_ctxt saved;
 386         struct llog_handle *cathandle;
 387         struct inode *inode;
 388         void *handle;
 389         ENTRY;
 390
 391         logcookies = req_capsule_client_get(&req->rq_pill, &RMF_LOGCOOKIES);
 392         num_cookies = req_capsule_get_size(&req->rq_pill, &RMF_LOGCOOKIES,
 393                                            RCL_CLIENT) / sizeof(*logcookies);
 394         if (logcookies == NULL || num_cookies == 0) {
 395                 DEBUG_REQ(D_HA, req, "No llog cookies sent");
 396                 RETURN(-EFAULT);
 397         }
 398
 399         ctxt = llog_get_context(obd, logcookies->lgc_subsys);
 400         if (ctxt == NULL)
 401                 RETURN(-ENODEV);
 402
 403         disk_obd = ctxt->loc_exp->exp_obd;
 404         push_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 405         for (i = 0; i < num_cookies; i++, logcookies++) {
 406                 cathandle = ctxt->loc_handle;
 407                 LASSERT(cathandle != NULL);
 408                 inode = cathandle->lgh_file->f_dentry->d_inode;
 409
 410                 handle = fsfilt_start_log(disk_obd, inode,
 411                                           FSFILT_OP_CANCEL_UNLINK, NULL, 1);
 412                 if (IS_ERR(handle)) {
 413                         CERROR("fsfilt_start_log() failed: %ld\n",
 414                                PTR_ERR(handle));
 415                         GOTO(pop_ctxt, rc = PTR_ERR(handle));
 416                 }
 417
 418                 rc = llog_cat_cancel_records(cathandle, 1, logcookies);
 419
 420                 /*
 421                  * Do not raise -ENOENT errors for resent rpcs. This rec already
 422                  * might be killed.
 423                  */
 424                 if (rc == -ENOENT &&
 425                     (lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)) {
 426                         /*
 427                          * Do not change this message, reply-single.sh test_59b
 428                          * expects to find this in log.
 429                          */
 430                         CDEBUG(D_RPCTRACE, "RESENT cancel req %p - ignored\n",
 431                                req);
 432                         rc = 0;
 433                 } else if (rc == 0) {
 434                         CDEBUG(D_RPCTRACE, "Canceled %d llog-records\n",
 435                                num_cookies);
 436                 }
 437
 438                 err = fsfilt_commit(disk_obd, inode, handle, 0);
 439                 if (err) {
 440                         CERROR("Error committing transaction: %d\n", err);
 441                         if (!rc)
 442                                 rc = err;
 443                         failed++;
 444                         GOTO(pop_ctxt, rc);
 445                 } else if (rc)
 446                         failed++;
 447         }
 448         GOTO(pop_ctxt, rc);
 449 pop_ctxt:
 450         pop_ctxt(&saved, &disk_obd->obd_lvfs_ctxt, NULL);
 451         if (rc)
 452                 CERROR("Cancel %d of %d llog-records failed: %d\n",
 453                        failed, num_cookies, rc);
 454
 455         llog_ctxt_put(ctxt);
 456         return rc;
 457 }
 458 EXPORT_SYMBOL(llog_origin_handle_cancel);
 459
 460 static int llog_catinfo_config(struct obd_device *obd, char *buf, int buf_len,
 461                                char *client)
 462 {
 463         struct mds_obd       *mds = &obd->u.mds;
 464         struct llog_ctxt     *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 465         struct lvfs_run_ctxt  saved;
 466         struct llog_handle   *handle = NULL;
 467         char                  name[4][64];
 468         int                   rc, i, l, remains = buf_len;
 469         char                 *out = buf;
 470         ENTRY;
 471
 472         if (ctxt == NULL || mds == NULL)
 473                 GOTO(release_ctxt, rc = -ENODEV);
 474
 475         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 476
 477         sprintf(name[0], "%s", mds->mds_profile);
 478         sprintf(name[1], "%s-clean", mds->mds_profile);
 479         sprintf(name[2], "%s", client);
 480         sprintf(name[3], "%s-clean", client);
 481
 482         for (i = 0; i < 4; i++) {
 483                 int index, uncanceled = 0;
 484                 rc = llog_create(ctxt, &handle, NULL, name[i]);
 485                 if (rc)
 486                         GOTO(out_pop, rc);
 487                 rc = llog_init_handle(handle, 0, NULL);
 488                 if (rc) {
 489                         llog_close(handle);
 490                         GOTO(out_pop, rc = -ENOENT);
 491                 }
 492
 493                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index ++) {
 494                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 495                                 uncanceled++;
 496                 }
 497
 498                 l = snprintf(out, remains, "[Log Name]: %s\nLog Size: %llu\n"
 499                              "Last Index: %d\nUncanceled Records: %d\n\n",
 500                              name[i],
 501                              i_size_read(handle->lgh_file->f_dentry->d_inode),
 502                              handle->lgh_last_idx, uncanceled);
 503                 out += l;
 504                 remains -= l;
 505
 506                 llog_close(handle);
 507                 if (remains <= 0)
 508                         break;
 509         }
 510         GOTO(out_pop, rc);
 511 out_pop:
 512         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 513 release_ctxt:
 514         llog_ctxt_put(ctxt);
 515         return rc;
 516 }
 517
 518 struct cb_data {
 519         struct llog_ctxt *ctxt;
 520         char *out;
 521         int  remains;
 522         int  init;
 523 };
 524
 525 static int llog_catinfo_cb(struct llog_handle *cat,
 526                            struct llog_rec_hdr *rec, void *data)
 527 {
 528         static char *out = NULL;
 529         static int remains = 0;
 530         struct llog_ctxt *ctxt = NULL;
 531         struct llog_handle *handle = NULL;
 532         struct llog_logid *logid;
 533         struct llog_logid_rec *lir;
 534         int l, rc, index, count = 0;
 535         struct cb_data *cbd = (struct cb_data*)data;
 536         ENTRY;
 537
 538         if (cbd->init) {
 539                 out = cbd->out;
 540                 remains = cbd->remains;
 541                 cbd->init = 0;
 542         }
 543
 544         if (!(cat->lgh_hdr->llh_flags & LLOG_F_IS_CAT))
 545                 RETURN(-EINVAL);
 546
 547         if (!cbd->ctxt)
 548                 RETURN(-ENODEV);
 549
 550         lir = (struct llog_logid_rec *)rec;
 551         logid = &lir->lid_id;
 552         rc = llog_create(ctxt, &handle, logid, NULL);
 553         if (rc)
 554                 RETURN(-EINVAL);
 555         rc = llog_init_handle(handle, 0, NULL);
 556         if (rc)
 557                 GOTO(out_close, rc);
 558
 559         for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 560                 if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 561                         count++;
 562         }
 563
 564         l = snprintf(out, remains, "\t[Log ID]: #"LPX64"#"LPX64"#%08x\n"
 565                      "\tLog Size: %llu\n\tLast Index: %d\n"
 566                      "\tUncanceled Records: %d\n",
 567                      logid->lgl_oid, logid->lgl_oseq, logid->lgl_ogen,
 568                      i_size_read(handle->lgh_file->f_dentry->d_inode),
 569                      handle->lgh_last_idx, count);
 570         out += l;
 571         remains -= l;
 572         cbd->out = out;
 573         cbd->remains = remains;
 574         if (remains <= 0) {
 575                 CWARN("Not enough memory\n");
 576                 rc = -ENOMEM;
 577         }
 578         EXIT;
 579 out_close:
 580         llog_close(handle);
 581         return rc;
 582 }
 583
 584 static int llog_catinfo_deletions(struct obd_device *obd, char *buf,
 585                                   int buf_len)
 586 {
 587         struct mds_obd *mds = &obd->u.mds;
 588         struct llog_handle *handle;
 589         struct lvfs_run_ctxt saved;
 590         int size, i, count;
 591         struct llog_catid *idarray;
 592         char name[32] = CATLIST;
 593         struct cb_data data;
 594         struct llog_ctxt *ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 595         int rc;
 596         ENTRY;
 597
 598         if (ctxt == NULL || mds == NULL)
 599                 GOTO(release_ctxt, rc = -ENODEV);
 600
 601         count = mds->mds_lov_desc.ld_tgt_count;
 602         size = sizeof(*idarray) * count;
 603
 604         OBD_ALLOC_LARGE(idarray, size);
 605         if (!idarray)
 606                 GOTO(release_ctxt, rc = -ENOMEM);
 607
 608         cfs_mutex_down(&obd->obd_olg.olg_cat_processing);
 609         rc = llog_get_cat_list(obd, name, 0, count, idarray);
 610         if (rc)
 611                 GOTO(out_free, rc);
 612
 613         push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 614
 615         data.ctxt = ctxt;
 616         data.out = buf;
 617         data.remains = buf_len;
 618         for (i = 0; i < count; i++) {
 619                 int l, index, uncanceled = 0;
 620
 621                 rc = llog_create(ctxt, &handle, &idarray[i].lci_logid, NULL);
 622                 if (rc)
 623                         GOTO(out_pop, rc);
 624                 rc = llog_init_handle(handle, 0, NULL);
 625                 if (rc) {
 626                         llog_close(handle);
 627                         GOTO(out_pop, rc = -ENOENT);
 628                 }
 629                 for (index = 1; index < (LLOG_BITMAP_BYTES * 8); index++) {
 630                         if (ext2_test_bit(index, handle->lgh_hdr->llh_bitmap))
 631                                 uncanceled++;
 632                 }
 633                 l = snprintf(data.out, data.remains,
 634                              "\n[Catlog ID]: #"LPX64"#"LPX64"#%08x  "
 635                              "[Log Count]: %d\n",
 636                              idarray[i].lci_logid.lgl_oid,
 637                              idarray[i].lci_logid.lgl_oseq,
 638                              idarray[i].lci_logid.lgl_ogen, uncanceled);
 639
 640                 data.out += l;
 641                 data.remains -= l;
 642                 data.init = 1;
 643
 644                 llog_process(handle, llog_catinfo_cb, &data, NULL);
 645                 llog_close(handle);
 646
 647                 if (data.remains <= 0)
 648                         break;
 649         }
 650         EXIT;
 651 out_pop:
 652         pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 653 out_free:
 654         cfs_mutex_up(&obd->obd_olg.olg_cat_processing);
 655         OBD_FREE_LARGE(idarray, size);
 656 release_ctxt:
 657         llog_ctxt_put(ctxt);
 658         return rc;
 659 }
 660
 661 int llog_catinfo(struct ptlrpc_request *req)
 662 {
 663         struct obd_export *exp = req->rq_export;
 664         struct obd_device *obd = exp->exp_obd;
 665         char              *keyword;
 666         char              *buf, *reply;
 667         int                rc;
 668         ENTRY;
 669
 670         OBD_ALLOC(buf, LLOG_CHUNK_SIZE);
 671         if (buf == NULL)
 672                 RETURN(-ENOMEM);
 673
 674         memset(buf, 0, LLOG_CHUNK_SIZE);
 675
 676         keyword = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 677         LASSERT(keyword);
 678
 679         if (strcmp(keyword, "config") == 0) {
 680                 char *client = req_capsule_client_get(&req->rq_pill,
 681                                                       &RMF_STRING);
 682
 683                 LASSERT(client);
 684                 rc = llog_catinfo_config(obd, buf, LLOG_CHUNK_SIZE, client);
 685         } else if (strcmp(keyword, "deletions") == 0) {
 686                 rc = llog_catinfo_deletions(obd, buf, LLOG_CHUNK_SIZE);
 687         } else {
 688                 rc = -EOPNOTSUPP;
 689         }
 690
 691         req_capsule_set_size(&req->rq_pill, &RMF_STRING, RCL_SERVER,
 692                              LLOG_CHUNK_SIZE);
 693         rc = req_capsule_server_pack(&req->rq_pill);
 694         if (rc)
 695                 GOTO(out_free, rc = -ENOMEM);
 696
 697         reply = req_capsule_server_get(&req->rq_pill, &RMF_STRING);
 698         if (strlen(buf) == 0)
 699                 sprintf(buf, "%s", "No log informations\n");
 700         memcpy(reply, buf, LLOG_CHUNK_SIZE);
 701         EXIT;
 702 out_free:
 703         OBD_FREE(buf, LLOG_CHUNK_SIZE);
 704         return rc;
 705 }
 706
 707 #else /* !__KERNEL__ */
 708 int llog_origin_handle_create(struct ptlrpc_request *req)
 709 {
 710         LBUG();
 711         return 0;
 712 }
 713
 714 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 715 {
 716         LBUG();
 717         return 0;
 718 }
 719
 720 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 721 {
 722         LBUG();
 723         return 0;
 724 }
 725 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 726 {
 727         LBUG();
 728         return 0;
 729 }
 730 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 731 {
 732         LBUG();
 733         return 0;
 734 }
 735 int llog_origin_handle_close(struct ptlrpc_request *req)
 736 {
 737         LBUG();
 738         return 0;
 739 }
 740 int llog_origin_handle_cancel(struct ptlrpc_request *req)
 741 {
 742         LBUG();
 743         return 0;
 744 }
 745 #endif