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