lustre/ptlrpc/recov_thread.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/recov_thread.c
  37  *
  38  * OST<->MDS recovery logging thread.
  39  * Invariants in implementation:
  40  * - we do not share logs among different OST<->MDS connections, so that
  41  *   if an OST or MDS fails it need only look at log(s) relevant to itself
  42  *
  43  * Author: Andreas Dilger   <adilger@clusterfs.com>
  44  *         Yury Umanets     <yury.umanets@sun.com>
  45  *         Alexey Lyashkov  <alexey.lyashkov@sun.com>
  46  */
  47
  48 #define DEBUG_SUBSYSTEM S_LOG
  49
  50 #ifndef EXPORT_SYMTAB
  51 # define EXPORT_SYMTAB
  52 #endif
  53
  54 #ifdef __KERNEL__
  55 # include <libcfs/libcfs.h>
  56 #else
  57 # include <libcfs/list.h>
  58 # include <liblustre.h>
  59 #endif
  60
  61 #include <obd_class.h>
  62 #include <obd_support.h>
  63 #include <obd_class.h>
  64 #include <lustre_net.h>
  65 #include <lnet/types.h>
  66 #include <libcfs/list.h>
  67 #include <lustre_log.h>
  68 #include "ptlrpc_internal.h"
  69
  70 static atomic_t                   llcd_count = ATOMIC_INIT(0);
  71 static cfs_mem_cache_t           *llcd_cache = NULL;
  72
  73 #ifdef __KERNEL__
  74 enum {
  75         LLOG_LCM_FL_START       = 1 << 0,
  76         LLOG_LCM_FL_EXIT        = 1 << 1
  77 };
  78
  79 /**
  80  * Allocate new llcd from cache, init it and return to caller.
  81  * Bumps number of objects allocated.
  82  */
  83 static struct llog_canceld_ctxt *llcd_alloc(void)
  84 {
  85         struct llog_canceld_ctxt *llcd;
  86         int llcd_size;
  87
  88         /*
  89          * Payload of lustre_msg V2 is bigger.
  90          */
  91         llcd_size = CFS_PAGE_SIZE -
  92                 lustre_msg_size(LUSTRE_MSG_MAGIC_V2, 1, NULL);
  93         llcd_size += offsetof(struct llog_canceld_ctxt, llcd_cookies);
  94         OBD_SLAB_ALLOC(llcd, llcd_cache, CFS_ALLOC_STD, llcd_size);
  95         if (!llcd)
  96                 return NULL;
  97
  98         CFS_INIT_LIST_HEAD(&llcd->llcd_list);
  99         llcd->llcd_size = llcd_size;
 100         llcd->llcd_cookiebytes = 0;
 101         atomic_inc(&llcd_count);
 102         return llcd;
 103 }
 104
 105 /**
 106  * Returns passed llcd to cache.
 107  */
 108 static void llcd_free(struct llog_canceld_ctxt *llcd)
 109 {
 110         LASSERT(atomic_read(&llcd_count) > 0);
 111         OBD_SLAB_FREE(llcd, llcd_cache, llcd->llcd_size);
 112         atomic_dec(&llcd_count);
 113 }
 114
 115 /**
 116  * Copy passed @cookies to @llcd.
 117  */
 118 static void llcd_copy(struct llog_canceld_ctxt *llcd,
 119                       struct llog_cookie *cookies)
 120 {
 121         memcpy((char *)llcd->llcd_cookies + llcd->llcd_cookiebytes,
 122               cookies, sizeof(*cookies));
 123         llcd->llcd_cookiebytes += sizeof(*cookies);
 124 }
 125
 126 /**
 127  * Checks if passed cookie fits into llcd free space buffer. Returns
 128  * 1 if yes and 0 otherwise.
 129  */
 130 static int llcd_fit(struct llog_canceld_ctxt *llcd,
 131                  struct llog_cookie *cookies)
 132 {
 133         return (llcd->llcd_size -
 134                 llcd->llcd_cookiebytes) >= sizeof(*cookies);
 135 }
 136
 137 static void llcd_print(struct llog_canceld_ctxt *llcd,
 138                        const char *func, int line)
 139 {
 140         CDEBUG(D_RPCTRACE, "Llcd (%p) at %s:%d:\n", llcd, func, line);
 141         CDEBUG(D_RPCTRACE, "  size: %d\n", llcd->llcd_size);
 142         CDEBUG(D_RPCTRACE, "  ctxt: %p\n", llcd->llcd_ctxt);
 143         CDEBUG(D_RPCTRACE, "  lcm : %p\n", llcd->llcd_lcm);
 144         CDEBUG(D_RPCTRACE, "  cookiebytes : %d\n", llcd->llcd_cookiebytes);
 145 }
 146
 147 /**
 148  * Llcd completion function. Called uppon llcd send finish regardless
 149  * sending result. Error is passed in @rc. Note, that this will be called
 150  * in cleanup time when all inflight rpcs aborted.
 151  */
 152 static int
 153 llcd_interpret(const struct lu_env *env,
 154                struct ptlrpc_request *req, void *noused, int rc)
 155 {
 156         struct llog_canceld_ctxt *llcd = req->rq_async_args.pointer_arg[0];
 157         CDEBUG(D_RPCTRACE, "Sent llcd %p (%d)\n", llcd, rc);
 158         llcd_free(llcd);
 159         return 0;
 160 }
 161
 162 /**
 163  * Send @llcd to remote node. Free llcd uppon completion or error. Sending
 164  * is performed in async style so this function will return asap without
 165  * blocking.
 166  */
 167 static int llcd_send(struct llog_canceld_ctxt *llcd)
 168 {
 169         char *bufs[2] = { NULL, (char *)llcd->llcd_cookies };
 170         struct obd_import *import = NULL;
 171         struct llog_commit_master *lcm;
 172         struct ptlrpc_request *req;
 173         struct llog_ctxt *ctxt;
 174         int rc;
 175         ENTRY;
 176
 177         ctxt = llcd->llcd_ctxt;
 178         if (!ctxt) {
 179                 CERROR("Invalid llcd with NULL ctxt found (%p)\n",
 180                        llcd);
 181                 llcd_print(llcd, __FUNCTION__, __LINE__);
 182                 LBUG();
 183         }
 184         LASSERT_SEM_LOCKED(&ctxt->loc_sem);
 185
 186         if (llcd->llcd_cookiebytes == 0)
 187                 GOTO(exit, rc = 0);
 188
 189         lcm = llcd->llcd_lcm;
 190
 191         /*
 192          * Check if we're in exit stage. Do not send llcd in
 193          * this case.
 194          */
 195         if (test_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags))
 196                 GOTO(exit, rc = -ENODEV);
 197
 198         CDEBUG(D_RPCTRACE, "Sending llcd %p\n", llcd);
 199
 200         import = llcd->llcd_ctxt->loc_imp;
 201         if (!import || (import == LP_POISON) ||
 202             (import->imp_client == LP_POISON)) {
 203                 CERROR("Invalid import %p for llcd %p\n",
 204                        import, llcd);
 205                 GOTO(exit, rc = -ENODEV);
 206         }
 207
 208         OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_DELAY_RECOV, 10);
 209
 210         /*
 211          * No need to get import here as it is already done in
 212          * llog_receptor_accept().
 213          */
 214         req = ptlrpc_request_alloc(import, &RQF_LOG_CANCEL);
 215         if (req == NULL) {
 216                 CERROR("Can't allocate request for sending llcd %p\n",
 217                        llcd);
 218                 GOTO(exit, rc = -ENOMEM);
 219         }
 220         req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES,
 221                              RCL_CLIENT, llcd->llcd_cookiebytes);
 222
 223         rc = ptlrpc_request_bufs_pack(req, LUSTRE_LOG_VERSION,
 224                                       OBD_LOG_CANCEL, bufs, NULL);
 225         if (rc) {
 226                 ptlrpc_request_free(req);
 227                 GOTO(exit, rc);
 228         }
 229
 230         ptlrpc_at_set_req_timeout(req);
 231         ptlrpc_request_set_replen(req);
 232
 233         /* bug 5515 */
 234         req->rq_request_portal = LDLM_CANCEL_REQUEST_PORTAL;
 235         req->rq_reply_portal = LDLM_CANCEL_REPLY_PORTAL;
 236         req->rq_interpret_reply = (ptlrpc_interpterer_t)llcd_interpret;
 237         req->rq_async_args.pointer_arg[0] = llcd;
 238         rc = ptlrpc_set_add_new_req(&lcm->lcm_pc, req);
 239         if (rc) {
 240                 ptlrpc_request_free(req);
 241                 GOTO(exit, rc);
 242         }
 243         RETURN(0);
 244 exit:
 245         CDEBUG(D_RPCTRACE, "Refused llcd %p\n", llcd);
 246         llcd_free(llcd);
 247         return rc;
 248 }
 249
 250 /**
 251  * Attach @llcd to @ctxt. Establish llcd vs. ctxt reserve connection
 252  * so hat they can refer each other.
 253  */
 254 static int
 255 llcd_attach(struct llog_ctxt *ctxt, struct llog_canceld_ctxt *llcd)
 256 {
 257         struct llog_commit_master *lcm;
 258
 259         LASSERT(ctxt != NULL && llcd != NULL);
 260         LASSERT_SEM_LOCKED(&ctxt->loc_sem);
 261         LASSERT(ctxt->loc_llcd == NULL);
 262         lcm = ctxt->loc_lcm;
 263         spin_lock(&lcm->lcm_lock);
 264         atomic_inc(&lcm->lcm_count);
 265         list_add_tail(&llcd->llcd_list, &lcm->lcm_llcds);
 266         spin_unlock(&lcm->lcm_lock);
 267         CDEBUG(D_RPCTRACE, "Attach llcd %p to ctxt %p (%d)\n",
 268                llcd, ctxt, atomic_read(&lcm->lcm_count));
 269         llcd->llcd_ctxt = llog_ctxt_get(ctxt);
 270         llcd->llcd_lcm = ctxt->loc_lcm;
 271         ctxt->loc_llcd = llcd;
 272         return 0;
 273 }
 274
 275 /**
 276  * Opposite to llcd_attach(). Detaches llcd from its @ctxt. This makes
 277  * sure that this llcd will not be found another time we try to cancel.
 278  */
 279 static struct llog_canceld_ctxt *llcd_detach(struct llog_ctxt *ctxt)
 280 {
 281         struct llog_commit_master *lcm;
 282         struct llog_canceld_ctxt *llcd;
 283
 284         LASSERT(ctxt != NULL);
 285         LASSERT_SEM_LOCKED(&ctxt->loc_sem);
 286
 287         llcd = ctxt->loc_llcd;
 288         if (!llcd)
 289                 return NULL;
 290
 291         lcm = ctxt->loc_lcm;
 292         if (atomic_read(&lcm->lcm_count) == 0) {
 293                 CERROR("Invalid detach occured %p:%p\n", ctxt, llcd);
 294                 llcd_print(llcd, __FUNCTION__, __LINE__);
 295                 LBUG();
 296         }
 297         spin_lock(&lcm->lcm_lock);
 298         LASSERT(!list_empty(&llcd->llcd_list));
 299         list_del_init(&llcd->llcd_list);
 300         atomic_dec(&lcm->lcm_count);
 301         spin_unlock(&lcm->lcm_lock);
 302         ctxt->loc_llcd = NULL;
 303
 304         CDEBUG(D_RPCTRACE, "Detach llcd %p from ctxt %p (%d)\n",
 305                llcd, ctxt, atomic_read(&lcm->lcm_count));
 306
 307         llog_ctxt_put(ctxt);
 308         return llcd;
 309 }
 310
 311 /**
 312  * Return @llcd cached in @ctxt. Allocate new one if required. Attach it
 313  * to ctxt so that it may be used for gathering cookies and sending.
 314  */
 315 static struct llog_canceld_ctxt *llcd_get(struct llog_ctxt *ctxt)
 316 {
 317         struct llog_canceld_ctxt *llcd;
 318
 319         llcd = llcd_alloc();
 320         if (!llcd) {
 321                 CERROR("Couldn't alloc an llcd for ctxt %p\n", ctxt);
 322                 return NULL;
 323         }
 324         llcd_attach(ctxt, llcd);
 325         return llcd;
 326 }
 327
 328 /**
 329  * Deatch llcd from its @ctxt. Free llcd.
 330  */
 331 static void llcd_put(struct llog_ctxt *ctxt)
 332 {
 333         struct llog_commit_master *lcm;
 334         struct llog_canceld_ctxt *llcd;
 335
 336         lcm = ctxt->loc_lcm;
 337         llcd = llcd_detach(ctxt);
 338         if (llcd)
 339                 llcd_free(llcd);
 340 }
 341
 342 /**
 343  * Detach llcd from its @ctxt so that nobody will find it with try to
 344  * re-use. Send llcd to remote node.
 345  */
 346 static int llcd_push(struct llog_ctxt *ctxt)
 347 {
 348         struct llog_canceld_ctxt *llcd;
 349         int rc;
 350
 351         /*
 352          * Make sure that this llcd will not be sent again as we detach
 353          * it from ctxt.
 354          */
 355         llcd = llcd_detach(ctxt);
 356         if (!llcd) {
 357                 CERROR("Invalid detached llcd found %p\n", llcd);
 358                 llcd_print(llcd, __FUNCTION__, __LINE__);
 359                 LBUG();
 360         }
 361
 362         rc = llcd_send(llcd);
 363         if (rc)
 364                 CERROR("Couldn't send llcd %p (%d)\n", llcd, rc);
 365         return rc;
 366 }
 367
 368 /**
 369  * Start recovery thread which actually deals llcd sending. This
 370  * is all ptlrpc standard thread based so there is not much of work
 371  * to do.
 372  */
 373 int llog_recov_thread_start(struct llog_commit_master *lcm)
 374 {
 375         int rc;
 376         ENTRY;
 377
 378         rc = ptlrpcd_start(lcm->lcm_name, &lcm->lcm_pc);
 379         if (rc) {
 380                 CERROR("Error %d while starting recovery thread %s\n",
 381                        rc, lcm->lcm_name);
 382                 RETURN(rc);
 383         }
 384         RETURN(rc);
 385 }
 386 EXPORT_SYMBOL(llog_recov_thread_start);
 387
 388 /**
 389  * Stop recovery thread. Complement to llog_recov_thread_start().
 390  */
 391 void llog_recov_thread_stop(struct llog_commit_master *lcm, int force)
 392 {
 393         ENTRY;
 394
 395         /*
 396          * Let all know that we're stopping. This will also make
 397          * llcd_send() refuse any new llcds.
 398          */
 399         set_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags);
 400
 401         /*
 402          * Stop processing thread. No new rpcs will be accepted for
 403          * for processing now.
 404          */
 405         ptlrpcd_stop(&lcm->lcm_pc, force);
 406
 407         /*
 408          * By this point no alive inflight llcds should be left. Only
 409          * those forgotten in sync may still be attached to ctxt. Let's
 410          * print them.
 411          */
 412         if (atomic_read(&lcm->lcm_count) != 0) {
 413                 struct llog_canceld_ctxt *llcd;
 414                 struct list_head         *tmp;
 415
 416                 CERROR("Busy llcds found (%d) on lcm %p\n",
 417                        atomic_read(&lcm->lcm_count) == 0, lcm);
 418
 419                 spin_lock(&lcm->lcm_lock);
 420                 list_for_each(tmp, &lcm->lcm_llcds) {
 421                         llcd = list_entry(tmp, struct llog_canceld_ctxt,
 422                                           llcd_list);
 423                         llcd_print(llcd, __FUNCTION__, __LINE__);
 424                 }
 425                 spin_unlock(&lcm->lcm_lock);
 426         }
 427         EXIT;
 428 }
 429 EXPORT_SYMBOL(llog_recov_thread_stop);
 430
 431 /**
 432  * Initialize commit master structure and start recovery thread on it.
 433  */
 434 struct llog_commit_master *llog_recov_thread_init(char *name)
 435 {
 436         struct llog_commit_master *lcm;
 437         int rc;
 438         ENTRY;
 439
 440         OBD_ALLOC_PTR(lcm);
 441         if (!lcm)
 442                 RETURN(NULL);
 443
 444         /*
 445          * Try to create threads with unique names.
 446          */
 447         snprintf(lcm->lcm_name, sizeof(lcm->lcm_name),
 448                  "ll_log_commit_%s", name);
 449
 450         strncpy(lcm->lcm_name, name, sizeof(lcm->lcm_name));
 451         atomic_set(&lcm->lcm_count, 0);
 452         spin_lock_init(&lcm->lcm_lock);
 453         CFS_INIT_LIST_HEAD(&lcm->lcm_llcds);
 454         rc = llog_recov_thread_start(lcm);
 455         if (rc) {
 456                 CERROR("Can't start commit thread, rc %d\n", rc);
 457                 GOTO(out, rc);
 458         }
 459         RETURN(lcm);
 460 out:
 461         OBD_FREE_PTR(lcm);
 462         return NULL;
 463 }
 464 EXPORT_SYMBOL(llog_recov_thread_init);
 465
 466 /**
 467  * Finalize commit master and its recovery thread.
 468  */
 469 void llog_recov_thread_fini(struct llog_commit_master *lcm, int force)
 470 {
 471         ENTRY;
 472         llog_recov_thread_stop(lcm, force);
 473         OBD_FREE_PTR(lcm);
 474         EXIT;
 475 }
 476 EXPORT_SYMBOL(llog_recov_thread_fini);
 477
 478 static int llog_recov_thread_replay(struct llog_ctxt *ctxt,
 479                                     void *cb, void *arg)
 480 {
 481         struct obd_device *obd = ctxt->loc_obd;
 482         struct llog_process_cat_args *lpca;
 483         int rc;
 484         ENTRY;
 485
 486         if (obd->obd_stopping)
 487                 RETURN(-ENODEV);
 488
 489         /*
 490          * This will be balanced in llog_cat_process_thread()
 491          */
 492         OBD_ALLOC_PTR(lpca);
 493         if (!lpca)
 494                 RETURN(-ENOMEM);
 495
 496         lpca->lpca_cb = cb;
 497         lpca->lpca_arg = arg;
 498
 499         /*
 500          * This will be balanced in llog_cat_process_thread()
 501          */
 502         lpca->lpca_ctxt = llog_ctxt_get(ctxt);
 503         if (!lpca->lpca_ctxt) {
 504                 OBD_FREE_PTR(lpca);
 505                 RETURN(-ENODEV);
 506         }
 507         rc = cfs_kernel_thread(llog_cat_process_thread, lpca,
 508                                CLONE_VM | CLONE_FILES);
 509         if (rc < 0) {
 510                 CERROR("Error starting llog_cat_process_thread(): %d\n", rc);
 511                 OBD_FREE_PTR(lpca);
 512                 llog_ctxt_put(ctxt);
 513         } else {
 514                 CDEBUG(D_HA, "Started llog_cat_process_thread(): %d\n", rc);
 515                 rc = 0;
 516         }
 517
 518         RETURN(rc);
 519 }
 520
 521 int llog_obd_repl_connect(struct llog_ctxt *ctxt,
 522                           struct llog_logid *logid, struct llog_gen *gen,
 523                           struct obd_uuid *uuid)
 524 {
 525         int rc;
 526         ENTRY;
 527
 528         /*
 529          * Send back cached llcd from llog before recovery if we have any.
 530          * This is void is nothing cached is found there.
 531          */
 532         llog_sync(ctxt, NULL);
 533
 534         /*
 535          * Start recovery in separate thread.
 536          */
 537         mutex_down(&ctxt->loc_sem);
 538         ctxt->loc_gen = *gen;
 539         rc = llog_recov_thread_replay(ctxt, ctxt->llog_proc_cb, logid);
 540         mutex_up(&ctxt->loc_sem);
 541
 542         RETURN(rc);
 543 }
 544 EXPORT_SYMBOL(llog_obd_repl_connect);
 545
 546 /**
 547  * Deleted objects have a commit callback that cancels the MDS
 548  * log record for the deletion. The commit callback calls this
 549  * function.
 550  */
 551 int llog_obd_repl_cancel(struct llog_ctxt *ctxt,
 552                          struct lov_stripe_md *lsm, int count,
 553                          struct llog_cookie *cookies, int flags)
 554 {
 555         struct llog_commit_master *lcm;
 556         struct llog_canceld_ctxt *llcd;
 557         int rc = 0;
 558         ENTRY;
 559
 560         LASSERT(ctxt != NULL);
 561
 562         mutex_down(&ctxt->loc_sem);
 563         lcm = ctxt->loc_lcm;
 564
 565         /*
 566          * Let's check if we have all structures alive. We also check for
 567          * possible shutdown. Do nothing if we're stopping.
 568          */
 569         if (ctxt->loc_imp == NULL) {
 570                 CDEBUG(D_RPCTRACE, "No import for ctxt %p\n", ctxt);
 571                 GOTO(out, rc = -ENODEV);
 572         }
 573
 574         if (test_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags)) {
 575                 CDEBUG(D_RPCTRACE, "Commit thread is stopping for ctxt %p\n",
 576                        ctxt);
 577                 GOTO(out, rc = -ENODEV);
 578         }
 579
 580         llcd = ctxt->loc_llcd;
 581
 582         if (count > 0 && cookies != NULL) {
 583                 /*
 584                  * Get new llcd from ctxt if required.
 585                  */
 586                 if (!llcd) {
 587                         llcd = llcd_get(ctxt);
 588                         if (!llcd)
 589                                 GOTO(out, rc = -ENOMEM);
 590                         /*
 591                          * Allocation is successful, let's check for stop
 592                          * flag again to fall back as soon as possible.
 593                          */
 594                         if (test_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags))
 595                                 GOTO(out, rc = -ENODEV);
 596                 }
 597
 598                 /*
 599                  * Llcd does not have enough room for @cookies. Let's push
 600                  * it out and allocate new one.
 601                  */
 602                 if (!llcd_fit(llcd, cookies)) {
 603                         rc = llcd_push(ctxt);
 604                         if (rc)
 605                                 GOTO(out, rc);
 606                         llcd = llcd_get(ctxt);
 607                         if (!llcd)
 608                                 GOTO(out, rc = -ENOMEM);
 609                         /*
 610                          * Allocation is successful, let's check for stop
 611                          * flag again to fall back as soon as possible.
 612                          */
 613                         if (test_bit(LLOG_LCM_FL_EXIT, &lcm->lcm_flags))
 614                                 GOTO(out, rc = -ENODEV);
 615                 }
 616
 617                 /*
 618                  * Copy cookies to @llcd, no matter old or new allocated
 619                  * one.
 620                  */
 621                 llcd_copy(llcd, cookies);
 622         }
 623
 624         /*
 625          * Let's check if we need to send copied @cookies asap. If yes
 626          * then do it.
 627          */
 628         if (llcd && (flags & OBD_LLOG_FL_SENDNOW)) {
 629                 rc = llcd_push(ctxt);
 630                 if (rc)
 631                         GOTO(out, rc);
 632         }
 633         EXIT;
 634 out:
 635         if (rc)
 636                 llcd_put(ctxt);
 637         mutex_up(&ctxt->loc_sem);
 638         return rc;
 639 }
 640 EXPORT_SYMBOL(llog_obd_repl_cancel);
 641
 642 int llog_obd_repl_sync(struct llog_ctxt *ctxt, struct obd_export *exp)
 643 {
 644         int rc = 0;
 645         ENTRY;
 646
 647         /*
 648          * Flush any remaining llcd.
 649          */
 650         mutex_down(&ctxt->loc_sem);
 651         if (exp && (ctxt->loc_imp == exp->exp_imp_reverse)) {
 652                 /*
 653                  * This is ost->mds connection, we can't be sure that mds
 654                  * can still receive cookies, let's killed the cached llcd.
 655                  */
 656                 CDEBUG(D_RPCTRACE, "Kill cached llcd\n");
 657                 llcd_put(ctxt);
 658                 mutex_up(&ctxt->loc_sem);
 659         } else {
 660                 /*
 661                  * This is either llog_sync() from generic llog code or sync
 662                  * on client disconnect. In either way let's do it and send
 663                  * llcds to the target with waiting for completion.
 664                  */
 665                 CDEBUG(D_RPCTRACE, "Sync cached llcd\n");
 666                 mutex_up(&ctxt->loc_sem);
 667                 rc = llog_cancel(ctxt, NULL, 0, NULL, OBD_LLOG_FL_SENDNOW);
 668         }
 669         RETURN(rc);
 670 }
 671 EXPORT_SYMBOL(llog_obd_repl_sync);
 672
 673 #else /* !__KERNEL__ */
 674
 675 int llog_obd_repl_cancel(struct llog_ctxt *ctxt,
 676                          struct lov_stripe_md *lsm, int count,
 677                          struct llog_cookie *cookies, int flags)
 678 {
 679         return 0;
 680 }
 681 #endif
 682
 683 /**
 684  * Module init time fucntion. Initializes slab for llcd objects.
 685  */
 686 int llog_recov_init(void)
 687 {
 688         int llcd_size;
 689
 690         llcd_size = CFS_PAGE_SIZE -
 691                 lustre_msg_size(LUSTRE_MSG_MAGIC_V2, 1, NULL);
 692         llcd_size += offsetof(struct llog_canceld_ctxt, llcd_cookies);
 693         llcd_cache = cfs_mem_cache_create("llcd_cache", llcd_size, 0, 0);
 694         if (!llcd_cache) {
 695                 CERROR("Error allocating llcd cache\n");
 696                 return -ENOMEM;
 697         }
 698         return 0;
 699 }
 700
 701 /**
 702  * Module fini time fucntion. Releases slab for llcd objects.
 703  */
 704 void llog_recov_fini(void)
 705 {
 706         /*
 707          * Kill llcd cache when thread is stopped and we're sure no
 708          * llcd in use left.
 709          */
 710         if (llcd_cache) {
 711                 /*
 712                  * In 2.6.22 cfs_mem_cache_destroy() will not return error
 713                  * for busy resources. Let's check it another way.
 714                  */
 715                 LASSERTF(atomic_read(&llcd_count) == 0,
 716                          "Can't destroy llcd cache! Number of "
 717                          "busy llcds: %d\n", atomic_read(&llcd_count));
 718                 cfs_mem_cache_destroy(llcd_cache);
 719                 llcd_cache = NULL;
 720         }
 721 }