1 /* -*- mode: c; c-basic-offset: 8; -*-
2 * vim: noexpandtab sw=8 ts=8 sts=0:
6 * Defines functions of journalling api
8 * Copyright (C) 2003, 2004 Oracle. All rights reserved.
10 * This program is free software; you can redistribute it and/or
11 * modify it under the terms of the GNU General Public
12 * License as published by the Free Software Foundation; either
13 * version 2 of the License, or (at your option) any later version.
15 * This program is distributed in the hope that it will be useful,
16 * but WITHOUT ANY WARRANTY; without even the implied warranty of
17 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
18 * General Public License for more details.
20 * You should have received a copy of the GNU General Public
21 * License along with this program; if not, write to the
22 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
23 * Boston, MA 021110-1307, USA.
27 #include <linux/types.h>
28 #include <linux/slab.h>
29 #include <linux/highmem.h>
30 #include <linux/kthread.h>
32 #define MLOG_MASK_PREFIX ML_JOURNAL
33 #include <cluster/masklog.h>
39 #include "extent_map.h"
40 #include "heartbeat.h"
43 #include "localalloc.h"
50 #include "buffer_head_io.h"
52 DEFINE_SPINLOCK(trans_inc_lock);
54 static int ocfs2_force_read_journal(struct inode *inode);
55 static int ocfs2_recover_node(struct ocfs2_super *osb,
57 static int __ocfs2_recovery_thread(void *arg);
58 static int ocfs2_commit_cache(struct ocfs2_super *osb);
59 static int ocfs2_wait_on_mount(struct ocfs2_super *osb);
60 static void ocfs2_handle_cleanup_locks(struct ocfs2_journal *journal,
61 struct ocfs2_journal_handle *handle);
62 static void ocfs2_commit_unstarted_handle(struct ocfs2_journal_handle *handle);
63 static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
65 static int ocfs2_trylock_journal(struct ocfs2_super *osb,
67 static int ocfs2_recover_orphans(struct ocfs2_super *osb,
69 static int ocfs2_commit_thread(void *arg);
71 static int ocfs2_commit_cache(struct ocfs2_super *osb)
76 struct ocfs2_journal *journal = NULL;
80 journal = osb->journal;
82 /* Flush all pending commits and checkpoint the journal. */
83 down_write(&journal->j_trans_barrier);
85 if (atomic_read(&journal->j_num_trans) == 0) {
86 up_write(&journal->j_trans_barrier);
87 mlog(0, "No transactions for me to flush!\n");
91 journal_lock_updates(journal->j_journal);
92 status = journal_flush(journal->j_journal);
93 journal_unlock_updates(journal->j_journal);
95 up_write(&journal->j_trans_barrier);
100 old_id = ocfs2_inc_trans_id(journal);
102 flushed = atomic_read(&journal->j_num_trans);
103 atomic_set(&journal->j_num_trans, 0);
104 up_write(&journal->j_trans_barrier);
106 mlog(0, "commit_thread: flushed transaction %lu (%u handles)\n",
107 journal->j_trans_id, flushed);
109 ocfs2_kick_vote_thread(osb);
110 wake_up(&journal->j_checkpointed);
116 struct ocfs2_journal_handle *ocfs2_alloc_handle(struct ocfs2_super *osb)
118 struct ocfs2_journal_handle *retval = NULL;
120 retval = kcalloc(1, sizeof(*retval), GFP_NOFS);
122 mlog(ML_ERROR, "Failed to allocate memory for journal "
127 retval->num_locks = 0;
128 retval->k_handle = NULL;
130 INIT_LIST_HEAD(&retval->locks);
131 INIT_LIST_HEAD(&retval->inode_list);
132 retval->journal = osb->journal;
137 /* pass it NULL and it will allocate a new handle object for you. If
138 * you pass it a handle however, it may still return error, in which
139 * case it has free'd the passed handle for you. */
140 struct ocfs2_journal_handle *ocfs2_start_trans(struct ocfs2_super *osb,
141 struct ocfs2_journal_handle *handle,
145 journal_t *journal = osb->journal->j_journal;
147 mlog_entry("(max_buffs = %d)\n", max_buffs);
149 BUG_ON(!osb || !osb->journal->j_journal);
151 if (ocfs2_is_hard_readonly(osb)) {
156 BUG_ON(osb->journal->j_state == OCFS2_JOURNAL_FREE);
157 BUG_ON(max_buffs <= 0);
159 /* JBD might support this, but our journalling code doesn't yet. */
160 if (journal_current_handle()) {
161 mlog(ML_ERROR, "Recursive transaction attempted!\n");
166 handle = ocfs2_alloc_handle(osb);
169 mlog(ML_ERROR, "Failed to allocate memory for journal "
174 down_read(&osb->journal->j_trans_barrier);
176 /* actually start the transaction now */
177 handle->k_handle = journal_start(journal, max_buffs);
178 if (IS_ERR(handle->k_handle)) {
179 up_read(&osb->journal->j_trans_barrier);
181 ret = PTR_ERR(handle->k_handle);
182 handle->k_handle = NULL;
185 if (is_journal_aborted(journal)) {
186 ocfs2_abort(osb->sb, "Detected aborted journal");
192 atomic_inc(&(osb->journal->j_num_trans));
194 mlog_exit_ptr(handle);
199 ocfs2_commit_unstarted_handle(handle); /* will kfree handle */
205 void ocfs2_handle_add_inode(struct ocfs2_journal_handle *handle,
211 atomic_inc(&inode->i_count);
213 /* we're obviously changing it... */
214 mutex_lock(&inode->i_mutex);
217 BUG_ON(OCFS2_I(inode)->ip_handle);
218 BUG_ON(!list_empty(&OCFS2_I(inode)->ip_handle_list));
220 OCFS2_I(inode)->ip_handle = handle;
221 list_move_tail(&(OCFS2_I(inode)->ip_handle_list), &(handle->inode_list));
224 static void ocfs2_handle_unlock_inodes(struct ocfs2_journal_handle *handle)
226 struct list_head *p, *n;
228 struct ocfs2_inode_info *oi;
230 list_for_each_safe(p, n, &handle->inode_list) {
231 oi = list_entry(p, struct ocfs2_inode_info,
233 inode = &oi->vfs_inode;
235 OCFS2_I(inode)->ip_handle = NULL;
236 list_del_init(&OCFS2_I(inode)->ip_handle_list);
238 mutex_unlock(&inode->i_mutex);
243 /* This is trivial so we do it out of the main commit
244 * paths. Beware, it can be called from start_trans too! */
245 static void ocfs2_commit_unstarted_handle(struct ocfs2_journal_handle *handle)
249 ocfs2_handle_unlock_inodes(handle);
250 /* You are allowed to add journal locks before the transaction
252 ocfs2_handle_cleanup_locks(handle->journal, handle);
259 void ocfs2_commit_trans(struct ocfs2_journal_handle *handle)
261 handle_t *jbd_handle;
263 struct ocfs2_journal *journal = handle->journal;
269 if (!handle->k_handle) {
270 ocfs2_commit_unstarted_handle(handle);
275 /* release inode semaphores we took during this transaction */
276 ocfs2_handle_unlock_inodes(handle);
278 /* ocfs2_extend_trans may have had to call journal_restart
279 * which will always commit the transaction, but may return
280 * error for any number of reasons. If this is the case, we
281 * clear k_handle as it's not valid any more. */
282 if (handle->k_handle) {
283 jbd_handle = handle->k_handle;
285 /* actually stop the transaction. if we've set h_sync,
286 * it'll have been committed when we return */
287 retval = journal_stop(jbd_handle);
290 mlog(ML_ERROR, "Could not commit transaction\n");
294 handle->k_handle = NULL; /* it's been free'd in journal_stop */
297 ocfs2_handle_cleanup_locks(journal, handle);
299 up_read(&journal->j_trans_barrier);
306 * 'nblocks' is what you want to add to the current
307 * transaction. extend_trans will either extend the current handle by
308 * nblocks, or commit it and start a new one with nblocks credits.
310 * WARNING: This will not release any semaphores or disk locks taken
311 * during the transaction, so make sure they were taken *before*
312 * start_trans or we'll have ordering deadlocks.
314 * WARNING2: Note that we do *not* drop j_trans_barrier here. This is
315 * good because transaction ids haven't yet been recorded on the
316 * cluster locks associated with this handle.
318 int ocfs2_extend_trans(handle_t *handle, int nblocks)
327 mlog(0, "Trying to extend transaction by %d blocks\n", nblocks);
329 status = journal_extend(handle, nblocks);
336 mlog(0, "journal_extend failed, trying journal_restart\n");
337 status = journal_restart(handle, nblocks);
351 int ocfs2_journal_access(struct ocfs2_journal_handle *handle,
353 struct buffer_head *bh,
362 mlog_entry("bh->b_blocknr=%llu, type=%d (\"%s\"), bh->b_size = %zu\n",
363 (unsigned long long)bh->b_blocknr, type,
364 (type == OCFS2_JOURNAL_ACCESS_CREATE) ?
365 "OCFS2_JOURNAL_ACCESS_CREATE" :
366 "OCFS2_JOURNAL_ACCESS_WRITE",
369 /* we can safely remove this assertion after testing. */
370 if (!buffer_uptodate(bh)) {
371 mlog(ML_ERROR, "giving me a buffer that's not uptodate!\n");
372 mlog(ML_ERROR, "b_blocknr=%llu\n",
373 (unsigned long long)bh->b_blocknr);
377 /* Set the current transaction information on the inode so
378 * that the locking code knows whether it can drop it's locks
379 * on this inode or not. We're protected from the commit
380 * thread updating the current transaction id until
381 * ocfs2_commit_trans() because ocfs2_start_trans() took
382 * j_trans_barrier for us. */
383 ocfs2_set_inode_lock_trans(OCFS2_SB(inode->i_sb)->journal, inode);
385 mutex_lock(&OCFS2_I(inode)->ip_io_mutex);
387 case OCFS2_JOURNAL_ACCESS_CREATE:
388 case OCFS2_JOURNAL_ACCESS_WRITE:
389 status = journal_get_write_access(handle->k_handle, bh);
392 case OCFS2_JOURNAL_ACCESS_UNDO:
393 status = journal_get_undo_access(handle->k_handle, bh);
398 mlog(ML_ERROR, "Uknown access type!\n");
400 mutex_unlock(&OCFS2_I(inode)->ip_io_mutex);
403 mlog(ML_ERROR, "Error %d getting %d access to buffer!\n",
410 int ocfs2_journal_dirty(struct ocfs2_journal_handle *handle,
411 struct buffer_head *bh)
415 mlog_entry("(bh->b_blocknr=%llu)\n",
416 (unsigned long long)bh->b_blocknr);
418 status = journal_dirty_metadata(handle->k_handle, bh);
420 mlog(ML_ERROR, "Could not dirty metadata buffer. "
421 "(bh->b_blocknr=%llu)\n",
422 (unsigned long long)bh->b_blocknr);
428 int ocfs2_journal_dirty_data(handle_t *handle,
429 struct buffer_head *bh)
431 int err = journal_dirty_data(handle, bh);
434 /* TODO: When we can handle it, abort the handle and go RO on
440 /* We always assume you're adding a metadata lock at level 'ex' */
441 int ocfs2_handle_add_lock(struct ocfs2_journal_handle *handle,
445 struct ocfs2_journal_lock *lock;
449 lock = kmem_cache_alloc(ocfs2_lock_cache, GFP_NOFS);
458 lock->jl_inode = inode;
460 list_add_tail(&(lock->jl_lock_list), &(handle->locks));
469 static void ocfs2_handle_cleanup_locks(struct ocfs2_journal *journal,
470 struct ocfs2_journal_handle *handle)
472 struct list_head *p, *n;
473 struct ocfs2_journal_lock *lock;
476 list_for_each_safe(p, n, &(handle->locks)) {
477 lock = list_entry(p, struct ocfs2_journal_lock,
479 list_del(&lock->jl_lock_list);
482 inode = lock->jl_inode;
483 ocfs2_meta_unlock(inode, 1);
484 if (atomic_read(&inode->i_count) == 1)
486 "Inode %llu, I'm doing a last iput for!",
487 (unsigned long long)OCFS2_I(inode)->ip_blkno);
489 kmem_cache_free(ocfs2_lock_cache, lock);
493 #define OCFS2_DEFAULT_COMMIT_INTERVAL (HZ * 5)
495 void ocfs2_set_journal_params(struct ocfs2_super *osb)
497 journal_t *journal = osb->journal->j_journal;
499 spin_lock(&journal->j_state_lock);
500 journal->j_commit_interval = OCFS2_DEFAULT_COMMIT_INTERVAL;
501 if (osb->s_mount_opt & OCFS2_MOUNT_BARRIER)
502 journal->j_flags |= JFS_BARRIER;
504 journal->j_flags &= ~JFS_BARRIER;
505 spin_unlock(&journal->j_state_lock);
508 int ocfs2_journal_init(struct ocfs2_journal *journal, int *dirty)
511 struct inode *inode = NULL; /* the journal inode */
512 journal_t *j_journal = NULL;
513 struct ocfs2_dinode *di = NULL;
514 struct buffer_head *bh = NULL;
515 struct ocfs2_super *osb;
522 osb = journal->j_osb;
524 /* already have the inode for our journal */
525 inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
532 if (is_bad_inode(inode)) {
533 mlog(ML_ERROR, "access error (bad inode)\n");
540 SET_INODE_JOURNAL(inode);
541 OCFS2_I(inode)->ip_open_count++;
543 /* Skip recovery waits here - journal inode metadata never
544 * changes in a live cluster so it can be considered an
545 * exception to the rule. */
546 status = ocfs2_meta_lock_full(inode, NULL, &bh, 1,
547 OCFS2_META_LOCK_RECOVERY);
549 if (status != -ERESTARTSYS)
550 mlog(ML_ERROR, "Could not get lock on journal!\n");
555 di = (struct ocfs2_dinode *)bh->b_data;
557 if (inode->i_size < OCFS2_MIN_JOURNAL_SIZE) {
558 mlog(ML_ERROR, "Journal file size (%lld) is too small!\n",
564 mlog(0, "inode->i_size = %lld\n", inode->i_size);
565 mlog(0, "inode->i_blocks = %llu\n",
566 (unsigned long long)inode->i_blocks);
567 mlog(0, "inode->ip_clusters = %u\n", OCFS2_I(inode)->ip_clusters);
569 /* call the kernels journal init function now */
570 j_journal = journal_init_inode(inode);
571 if (j_journal == NULL) {
572 mlog(ML_ERROR, "Linux journal layer error\n");
577 mlog(0, "Returned from journal_init_inode\n");
578 mlog(0, "j_journal->j_maxlen = %u\n", j_journal->j_maxlen);
580 *dirty = (le32_to_cpu(di->id1.journal1.ij_flags) &
581 OCFS2_JOURNAL_DIRTY_FL);
583 journal->j_journal = j_journal;
584 journal->j_inode = inode;
587 ocfs2_set_journal_params(osb);
589 journal->j_state = OCFS2_JOURNAL_LOADED;
595 ocfs2_meta_unlock(inode, 1);
599 OCFS2_I(inode)->ip_open_count--;
608 static int ocfs2_journal_toggle_dirty(struct ocfs2_super *osb,
613 struct ocfs2_journal *journal = osb->journal;
614 struct buffer_head *bh = journal->j_bh;
615 struct ocfs2_dinode *fe;
619 fe = (struct ocfs2_dinode *)bh->b_data;
620 if (!OCFS2_IS_VALID_DINODE(fe)) {
621 /* This is called from startup/shutdown which will
622 * handle the errors in a specific manner, so no need
623 * to call ocfs2_error() here. */
624 mlog(ML_ERROR, "Journal dinode %llu has invalid "
625 "signature: %.*s", (unsigned long long)fe->i_blkno, 7,
631 flags = le32_to_cpu(fe->id1.journal1.ij_flags);
633 flags |= OCFS2_JOURNAL_DIRTY_FL;
635 flags &= ~OCFS2_JOURNAL_DIRTY_FL;
636 fe->id1.journal1.ij_flags = cpu_to_le32(flags);
638 status = ocfs2_write_block(osb, bh, journal->j_inode);
648 * If the journal has been kmalloc'd it needs to be freed after this
651 void ocfs2_journal_shutdown(struct ocfs2_super *osb)
653 struct ocfs2_journal *journal = NULL;
655 struct inode *inode = NULL;
656 int num_running_trans = 0;
662 journal = osb->journal;
666 inode = journal->j_inode;
668 if (journal->j_state != OCFS2_JOURNAL_LOADED)
671 /* need to inc inode use count as journal_destroy will iput. */
675 num_running_trans = atomic_read(&(osb->journal->j_num_trans));
676 if (num_running_trans > 0)
677 mlog(0, "Shutting down journal: must wait on %d "
678 "running transactions!\n",
681 /* Do a commit_cache here. It will flush our journal, *and*
682 * release any locks that are still held.
683 * set the SHUTDOWN flag and release the trans lock.
684 * the commit thread will take the trans lock for us below. */
685 journal->j_state = OCFS2_JOURNAL_IN_SHUTDOWN;
687 /* The OCFS2_JOURNAL_IN_SHUTDOWN will signal to commit_cache to not
688 * drop the trans_lock (which we want to hold until we
689 * completely destroy the journal. */
690 if (osb->commit_task) {
691 /* Wait for the commit thread */
692 mlog(0, "Waiting for ocfs2commit to exit....\n");
693 kthread_stop(osb->commit_task);
694 osb->commit_task = NULL;
697 BUG_ON(atomic_read(&(osb->journal->j_num_trans)) != 0);
699 status = ocfs2_journal_toggle_dirty(osb, 0);
703 /* Shutdown the kernel journal system */
704 journal_destroy(journal->j_journal);
706 OCFS2_I(inode)->ip_open_count--;
708 /* unlock our journal */
709 ocfs2_meta_unlock(inode, 1);
711 brelse(journal->j_bh);
712 journal->j_bh = NULL;
714 journal->j_state = OCFS2_JOURNAL_FREE;
716 // up_write(&journal->j_trans_barrier);
723 static void ocfs2_clear_journal_error(struct super_block *sb,
729 olderr = journal_errno(journal);
731 mlog(ML_ERROR, "File system error %d recorded in "
732 "journal %u.\n", olderr, slot);
733 mlog(ML_ERROR, "File system on device %s needs checking.\n",
736 journal_ack_err(journal);
737 journal_clear_err(journal);
741 int ocfs2_journal_load(struct ocfs2_journal *journal)
744 struct ocfs2_super *osb;
751 osb = journal->j_osb;
753 status = journal_load(journal->j_journal);
755 mlog(ML_ERROR, "Failed to load journal!\n");
759 ocfs2_clear_journal_error(osb->sb, journal->j_journal, osb->slot_num);
761 status = ocfs2_journal_toggle_dirty(osb, 1);
767 /* Launch the commit thread */
768 osb->commit_task = kthread_run(ocfs2_commit_thread, osb, "ocfs2cmt");
769 if (IS_ERR(osb->commit_task)) {
770 status = PTR_ERR(osb->commit_task);
771 osb->commit_task = NULL;
772 mlog(ML_ERROR, "unable to launch ocfs2commit thread, error=%d",
783 /* 'full' flag tells us whether we clear out all blocks or if we just
784 * mark the journal clean */
785 int ocfs2_journal_wipe(struct ocfs2_journal *journal, int full)
793 status = journal_wipe(journal->j_journal, full);
799 status = ocfs2_journal_toggle_dirty(journal->j_osb, 0);
809 * JBD Might read a cached version of another nodes journal file. We
810 * don't want this as this file changes often and we get no
811 * notification on those changes. The only way to be sure that we've
812 * got the most up to date version of those blocks then is to force
813 * read them off disk. Just searching through the buffer cache won't
814 * work as there may be pages backing this file which are still marked
815 * up to date. We know things can't change on this file underneath us
816 * as we have the lock by now :)
818 static int ocfs2_force_read_journal(struct inode *inode)
822 u64 v_blkno, p_blkno;
823 #define CONCURRENT_JOURNAL_FILL 32
824 struct buffer_head *bhs[CONCURRENT_JOURNAL_FILL];
828 BUG_ON(inode->i_blocks !=
829 ocfs2_align_bytes_to_sectors(i_size_read(inode)));
831 memset(bhs, 0, sizeof(struct buffer_head *) * CONCURRENT_JOURNAL_FILL);
833 mlog(0, "Force reading %llu blocks\n",
834 (unsigned long long)(inode->i_blocks >>
835 (inode->i_sb->s_blocksize_bits - 9)));
839 (inode->i_blocks >> (inode->i_sb->s_blocksize_bits - 9))) {
841 status = ocfs2_extent_map_get_blocks(inode, v_blkno,
849 if (p_blocks > CONCURRENT_JOURNAL_FILL)
850 p_blocks = CONCURRENT_JOURNAL_FILL;
852 /* We are reading journal data which should not
853 * be put in the uptodate cache */
854 status = ocfs2_read_blocks(OCFS2_SB(inode->i_sb),
855 p_blkno, p_blocks, bhs, 0,
862 for(i = 0; i < p_blocks; i++) {
871 for(i = 0; i < CONCURRENT_JOURNAL_FILL; i++)
878 struct ocfs2_la_recovery_item {
879 struct list_head lri_list;
881 struct ocfs2_dinode *lri_la_dinode;
882 struct ocfs2_dinode *lri_tl_dinode;
885 /* Does the second half of the recovery process. By this point, the
886 * node is marked clean and can actually be considered recovered,
887 * hence it's no longer in the recovery map, but there's still some
888 * cleanup we can do which shouldn't happen within the recovery thread
889 * as locking in that context becomes very difficult if we are to take
890 * recovering nodes into account.
892 * NOTE: This function can and will sleep on recovery of other nodes
893 * during cluster locking, just like any other ocfs2 process.
895 void ocfs2_complete_recovery(void *data)
898 struct ocfs2_super *osb = data;
899 struct ocfs2_journal *journal = osb->journal;
900 struct ocfs2_dinode *la_dinode, *tl_dinode;
901 struct ocfs2_la_recovery_item *item;
902 struct list_head *p, *n;
903 LIST_HEAD(tmp_la_list);
907 mlog(0, "completing recovery from keventd\n");
909 spin_lock(&journal->j_lock);
910 list_splice_init(&journal->j_la_cleanups, &tmp_la_list);
911 spin_unlock(&journal->j_lock);
913 list_for_each_safe(p, n, &tmp_la_list) {
914 item = list_entry(p, struct ocfs2_la_recovery_item, lri_list);
915 list_del_init(&item->lri_list);
917 mlog(0, "Complete recovery for slot %d\n", item->lri_slot);
919 la_dinode = item->lri_la_dinode;
921 mlog(0, "Clean up local alloc %llu\n",
922 (unsigned long long)la_dinode->i_blkno);
924 ret = ocfs2_complete_local_alloc_recovery(osb,
932 tl_dinode = item->lri_tl_dinode;
934 mlog(0, "Clean up truncate log %llu\n",
935 (unsigned long long)tl_dinode->i_blkno);
937 ret = ocfs2_complete_truncate_log_recovery(osb,
945 ret = ocfs2_recover_orphans(osb, item->lri_slot);
952 mlog(0, "Recovery completion\n");
956 /* NOTE: This function always eats your references to la_dinode and
957 * tl_dinode, either manually on error, or by passing them to
958 * ocfs2_complete_recovery */
959 static void ocfs2_queue_recovery_completion(struct ocfs2_journal *journal,
961 struct ocfs2_dinode *la_dinode,
962 struct ocfs2_dinode *tl_dinode)
964 struct ocfs2_la_recovery_item *item;
966 item = kmalloc(sizeof(struct ocfs2_la_recovery_item), GFP_NOFS);
968 /* Though we wish to avoid it, we are in fact safe in
969 * skipping local alloc cleanup as fsck.ocfs2 is more
970 * than capable of reclaiming unused space. */
981 INIT_LIST_HEAD(&item->lri_list);
982 item->lri_la_dinode = la_dinode;
983 item->lri_slot = slot_num;
984 item->lri_tl_dinode = tl_dinode;
986 spin_lock(&journal->j_lock);
987 list_add_tail(&item->lri_list, &journal->j_la_cleanups);
988 queue_work(ocfs2_wq, &journal->j_recovery_work);
989 spin_unlock(&journal->j_lock);
992 /* Called by the mount code to queue recovery the last part of
993 * recovery for it's own slot. */
994 void ocfs2_complete_mount_recovery(struct ocfs2_super *osb)
996 struct ocfs2_journal *journal = osb->journal;
999 /* No need to queue up our truncate_log as regular
1000 * cleanup will catch that. */
1001 ocfs2_queue_recovery_completion(journal,
1003 osb->local_alloc_copy,
1005 ocfs2_schedule_truncate_log_flush(osb, 0);
1007 osb->local_alloc_copy = NULL;
1012 static int __ocfs2_recovery_thread(void *arg)
1014 int status, node_num;
1015 struct ocfs2_super *osb = arg;
1019 status = ocfs2_wait_on_mount(osb);
1025 status = ocfs2_super_lock(osb, 1);
1031 while(!ocfs2_node_map_is_empty(osb, &osb->recovery_map)) {
1032 node_num = ocfs2_node_map_first_set_bit(osb,
1033 &osb->recovery_map);
1034 if (node_num == O2NM_INVALID_NODE_NUM) {
1035 mlog(0, "Out of nodes to recover.\n");
1039 status = ocfs2_recover_node(osb, node_num);
1042 "Error %d recovering node %d on device (%u,%u)!\n",
1044 MAJOR(osb->sb->s_dev), MINOR(osb->sb->s_dev));
1045 mlog(ML_ERROR, "Volume requires unmount.\n");
1049 ocfs2_recovery_map_clear(osb, node_num);
1051 ocfs2_super_unlock(osb, 1);
1053 /* We always run recovery on our own orphan dir - the dead
1054 * node(s) may have voted "no" on an inode delete earlier. A
1055 * revote is therefore required. */
1056 ocfs2_queue_recovery_completion(osb->journal, osb->slot_num, NULL,
1060 mutex_lock(&osb->recovery_lock);
1062 !ocfs2_node_map_is_empty(osb, &osb->recovery_map)) {
1063 mutex_unlock(&osb->recovery_lock);
1067 osb->recovery_thread_task = NULL;
1068 mb(); /* sync with ocfs2_recovery_thread_running */
1069 wake_up(&osb->recovery_event);
1071 mutex_unlock(&osb->recovery_lock);
1074 /* no one is callint kthread_stop() for us so the kthread() api
1075 * requires that we call do_exit(). And it isn't exported, but
1076 * complete_and_exit() seems to be a minimal wrapper around it. */
1077 complete_and_exit(NULL, status);
1081 void ocfs2_recovery_thread(struct ocfs2_super *osb, int node_num)
1083 mlog_entry("(node_num=%d, osb->node_num = %d)\n",
1084 node_num, osb->node_num);
1086 mutex_lock(&osb->recovery_lock);
1087 if (osb->disable_recovery)
1090 /* People waiting on recovery will wait on
1091 * the recovery map to empty. */
1092 if (!ocfs2_recovery_map_set(osb, node_num))
1093 mlog(0, "node %d already be in recovery.\n", node_num);
1095 mlog(0, "starting recovery thread...\n");
1097 if (osb->recovery_thread_task)
1100 osb->recovery_thread_task = kthread_run(__ocfs2_recovery_thread, osb,
1102 if (IS_ERR(osb->recovery_thread_task)) {
1103 mlog_errno((int)PTR_ERR(osb->recovery_thread_task));
1104 osb->recovery_thread_task = NULL;
1108 mutex_unlock(&osb->recovery_lock);
1109 wake_up(&osb->recovery_event);
1114 /* Does the actual journal replay and marks the journal inode as
1115 * clean. Will only replay if the journal inode is marked dirty. */
1116 static int ocfs2_replay_journal(struct ocfs2_super *osb,
1123 struct inode *inode = NULL;
1124 struct ocfs2_dinode *fe;
1125 journal_t *journal = NULL;
1126 struct buffer_head *bh = NULL;
1128 inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
1130 if (inode == NULL) {
1135 if (is_bad_inode(inode)) {
1142 SET_INODE_JOURNAL(inode);
1144 status = ocfs2_meta_lock_full(inode, NULL, &bh, 1,
1145 OCFS2_META_LOCK_RECOVERY);
1147 mlog(0, "status returned from ocfs2_meta_lock=%d\n", status);
1148 if (status != -ERESTARTSYS)
1149 mlog(ML_ERROR, "Could not lock journal!\n");
1154 fe = (struct ocfs2_dinode *) bh->b_data;
1156 flags = le32_to_cpu(fe->id1.journal1.ij_flags);
1158 if (!(flags & OCFS2_JOURNAL_DIRTY_FL)) {
1159 mlog(0, "No recovery required for node %d\n", node_num);
1163 mlog(ML_NOTICE, "Recovering node %d from slot %d on device (%u,%u)\n",
1165 MAJOR(osb->sb->s_dev), MINOR(osb->sb->s_dev));
1167 OCFS2_I(inode)->ip_clusters = le32_to_cpu(fe->i_clusters);
1169 status = ocfs2_force_read_journal(inode);
1175 mlog(0, "calling journal_init_inode\n");
1176 journal = journal_init_inode(inode);
1177 if (journal == NULL) {
1178 mlog(ML_ERROR, "Linux journal layer error\n");
1183 status = journal_load(journal);
1188 journal_destroy(journal);
1192 ocfs2_clear_journal_error(osb->sb, journal, slot_num);
1194 /* wipe the journal */
1195 mlog(0, "flushing the journal.\n");
1196 journal_lock_updates(journal);
1197 status = journal_flush(journal);
1198 journal_unlock_updates(journal);
1202 /* This will mark the node clean */
1203 flags = le32_to_cpu(fe->id1.journal1.ij_flags);
1204 flags &= ~OCFS2_JOURNAL_DIRTY_FL;
1205 fe->id1.journal1.ij_flags = cpu_to_le32(flags);
1207 status = ocfs2_write_block(osb, bh, inode);
1214 journal_destroy(journal);
1217 /* drop the lock on this nodes journal */
1219 ocfs2_meta_unlock(inode, 1);
1232 * Do the most important parts of node recovery:
1233 * - Replay it's journal
1234 * - Stamp a clean local allocator file
1235 * - Stamp a clean truncate log
1236 * - Mark the node clean
1238 * If this function completes without error, a node in OCFS2 can be
1239 * said to have been safely recovered. As a result, failure during the
1240 * second part of a nodes recovery process (local alloc recovery) is
1241 * far less concerning.
1243 static int ocfs2_recover_node(struct ocfs2_super *osb,
1248 struct ocfs2_slot_info *si = osb->slot_info;
1249 struct ocfs2_dinode *la_copy = NULL;
1250 struct ocfs2_dinode *tl_copy = NULL;
1252 mlog_entry("(node_num=%d, osb->node_num = %d)\n",
1253 node_num, osb->node_num);
1255 mlog(0, "checking node %d\n", node_num);
1257 /* Should not ever be called to recover ourselves -- in that
1258 * case we should've called ocfs2_journal_load instead. */
1259 BUG_ON(osb->node_num == node_num);
1261 slot_num = ocfs2_node_num_to_slot(si, node_num);
1262 if (slot_num == OCFS2_INVALID_SLOT) {
1264 mlog(0, "no slot for this node, so no recovery required.\n");
1268 mlog(0, "node %d was using slot %d\n", node_num, slot_num);
1270 status = ocfs2_replay_journal(osb, node_num, slot_num);
1276 /* Stamp a clean local alloc file AFTER recovering the journal... */
1277 status = ocfs2_begin_local_alloc_recovery(osb, slot_num, &la_copy);
1283 /* An error from begin_truncate_log_recovery is not
1284 * serious enough to warrant halting the rest of
1286 status = ocfs2_begin_truncate_log_recovery(osb, slot_num, &tl_copy);
1290 /* Likewise, this would be a strange but ultimately not so
1291 * harmful place to get an error... */
1292 ocfs2_clear_slot(si, slot_num);
1293 status = ocfs2_update_disk_slots(osb, si);
1297 /* This will kfree the memory pointed to by la_copy and tl_copy */
1298 ocfs2_queue_recovery_completion(osb->journal, slot_num, la_copy,
1308 /* Test node liveness by trylocking his journal. If we get the lock,
1309 * we drop it here. Return 0 if we got the lock, -EAGAIN if node is
1310 * still alive (we couldn't get the lock) and < 0 on error. */
1311 static int ocfs2_trylock_journal(struct ocfs2_super *osb,
1315 struct inode *inode = NULL;
1317 inode = ocfs2_get_system_file_inode(osb, JOURNAL_SYSTEM_INODE,
1319 if (inode == NULL) {
1320 mlog(ML_ERROR, "access error\n");
1324 if (is_bad_inode(inode)) {
1325 mlog(ML_ERROR, "access error (bad inode)\n");
1331 SET_INODE_JOURNAL(inode);
1333 flags = OCFS2_META_LOCK_RECOVERY | OCFS2_META_LOCK_NOQUEUE;
1334 status = ocfs2_meta_lock_full(inode, NULL, NULL, 1, flags);
1336 if (status != -EAGAIN)
1341 ocfs2_meta_unlock(inode, 1);
1349 /* Call this underneath ocfs2_super_lock. It also assumes that the
1350 * slot info struct has been updated from disk. */
1351 int ocfs2_mark_dead_nodes(struct ocfs2_super *osb)
1353 int status, i, node_num;
1354 struct ocfs2_slot_info *si = osb->slot_info;
1356 /* This is called with the super block cluster lock, so we
1357 * know that the slot map can't change underneath us. */
1359 spin_lock(&si->si_lock);
1360 for(i = 0; i < si->si_num_slots; i++) {
1361 if (i == osb->slot_num)
1363 if (ocfs2_is_empty_slot(si, i))
1366 node_num = si->si_global_node_nums[i];
1367 if (ocfs2_node_map_test_bit(osb, &osb->recovery_map, node_num))
1369 spin_unlock(&si->si_lock);
1371 /* Ok, we have a slot occupied by another node which
1372 * is not in the recovery map. We trylock his journal
1373 * file here to test if he's alive. */
1374 status = ocfs2_trylock_journal(osb, i);
1376 /* Since we're called from mount, we know that
1377 * the recovery thread can't race us on
1378 * setting / checking the recovery bits. */
1379 ocfs2_recovery_thread(osb, node_num);
1380 } else if ((status < 0) && (status != -EAGAIN)) {
1385 spin_lock(&si->si_lock);
1387 spin_unlock(&si->si_lock);
1395 static int ocfs2_queue_orphans(struct ocfs2_super *osb,
1397 struct inode **head)
1400 struct inode *orphan_dir_inode = NULL;
1402 unsigned long offset, blk, local;
1403 struct buffer_head *bh = NULL;
1404 struct ocfs2_dir_entry *de;
1405 struct super_block *sb = osb->sb;
1407 orphan_dir_inode = ocfs2_get_system_file_inode(osb,
1408 ORPHAN_DIR_SYSTEM_INODE,
1410 if (!orphan_dir_inode) {
1416 mutex_lock(&orphan_dir_inode->i_mutex);
1417 status = ocfs2_meta_lock(orphan_dir_inode, NULL, NULL, 0);
1425 while(offset < i_size_read(orphan_dir_inode)) {
1426 blk = offset >> sb->s_blocksize_bits;
1428 bh = ocfs2_bread(orphan_dir_inode, blk, &status, 0);
1439 while(offset < i_size_read(orphan_dir_inode)
1440 && local < sb->s_blocksize) {
1441 de = (struct ocfs2_dir_entry *) (bh->b_data + local);
1443 if (!ocfs2_check_dir_entry(orphan_dir_inode,
1451 local += le16_to_cpu(de->rec_len);
1452 offset += le16_to_cpu(de->rec_len);
1454 /* I guess we silently fail on no inode? */
1455 if (!le64_to_cpu(de->inode))
1457 if (de->file_type > OCFS2_FT_MAX) {
1459 "block %llu contains invalid de: "
1460 "inode = %llu, rec_len = %u, "
1461 "name_len = %u, file_type = %u, "
1463 (unsigned long long)bh->b_blocknr,
1464 (unsigned long long)le64_to_cpu(de->inode),
1465 le16_to_cpu(de->rec_len),
1472 if (de->name_len == 1 && !strncmp(".", de->name, 1))
1474 if (de->name_len == 2 && !strncmp("..", de->name, 2))
1477 iter = ocfs2_iget(osb, le64_to_cpu(de->inode),
1478 OCFS2_FI_FLAG_NOLOCK);
1482 mlog(0, "queue orphan %llu\n",
1483 (unsigned long long)OCFS2_I(iter)->ip_blkno);
1484 /* No locking is required for the next_orphan
1485 * queue as there is only ever a single
1486 * process doing orphan recovery. */
1487 OCFS2_I(iter)->ip_next_orphan = *head;
1494 ocfs2_meta_unlock(orphan_dir_inode, 0);
1496 mutex_unlock(&orphan_dir_inode->i_mutex);
1497 iput(orphan_dir_inode);
1501 static int ocfs2_orphan_recovery_can_continue(struct ocfs2_super *osb,
1506 spin_lock(&osb->osb_lock);
1507 ret = !osb->osb_orphan_wipes[slot];
1508 spin_unlock(&osb->osb_lock);
1512 static void ocfs2_mark_recovering_orphan_dir(struct ocfs2_super *osb,
1515 spin_lock(&osb->osb_lock);
1516 /* Mark ourselves such that new processes in delete_inode()
1517 * know to quit early. */
1518 ocfs2_node_map_set_bit(osb, &osb->osb_recovering_orphan_dirs, slot);
1519 while (osb->osb_orphan_wipes[slot]) {
1520 /* If any processes are already in the middle of an
1521 * orphan wipe on this dir, then we need to wait for
1523 spin_unlock(&osb->osb_lock);
1524 wait_event_interruptible(osb->osb_wipe_event,
1525 ocfs2_orphan_recovery_can_continue(osb, slot));
1526 spin_lock(&osb->osb_lock);
1528 spin_unlock(&osb->osb_lock);
1531 static void ocfs2_clear_recovering_orphan_dir(struct ocfs2_super *osb,
1534 ocfs2_node_map_clear_bit(osb, &osb->osb_recovering_orphan_dirs, slot);
1538 * Orphan recovery. Each mounted node has it's own orphan dir which we
1539 * must run during recovery. Our strategy here is to build a list of
1540 * the inodes in the orphan dir and iget/iput them. The VFS does
1541 * (most) of the rest of the work.
1543 * Orphan recovery can happen at any time, not just mount so we have a
1544 * couple of extra considerations.
1546 * - We grab as many inodes as we can under the orphan dir lock -
1547 * doing iget() outside the orphan dir risks getting a reference on
1549 * - We must be sure not to deadlock with other processes on the
1550 * system wanting to run delete_inode(). This can happen when they go
1551 * to lock the orphan dir and the orphan recovery process attempts to
1552 * iget() inside the orphan dir lock. This can be avoided by
1553 * advertising our state to ocfs2_delete_inode().
1555 static int ocfs2_recover_orphans(struct ocfs2_super *osb,
1559 struct inode *inode = NULL;
1561 struct ocfs2_inode_info *oi;
1563 mlog(0, "Recover inodes from orphan dir in slot %d\n", slot);
1565 ocfs2_mark_recovering_orphan_dir(osb, slot);
1566 ret = ocfs2_queue_orphans(osb, slot, &inode);
1567 ocfs2_clear_recovering_orphan_dir(osb, slot);
1569 /* Error here should be noted, but we want to continue with as
1570 * many queued inodes as we've got. */
1575 oi = OCFS2_I(inode);
1576 mlog(0, "iput orphan %llu\n", (unsigned long long)oi->ip_blkno);
1578 iter = oi->ip_next_orphan;
1580 spin_lock(&oi->ip_lock);
1581 /* Delete voting may have set these on the assumption
1582 * that the other node would wipe them successfully.
1583 * If they are still in the node's orphan dir, we need
1584 * to reset that state. */
1585 oi->ip_flags &= ~(OCFS2_INODE_DELETED|OCFS2_INODE_SKIP_DELETE);
1587 /* Set the proper information to get us going into
1588 * ocfs2_delete_inode. */
1589 oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
1590 oi->ip_orphaned_slot = slot;
1591 spin_unlock(&oi->ip_lock);
1601 static int ocfs2_wait_on_mount(struct ocfs2_super *osb)
1603 /* This check is good because ocfs2 will wait on our recovery
1604 * thread before changing it to something other than MOUNTED
1606 wait_event(osb->osb_mount_event,
1607 atomic_read(&osb->vol_state) == VOLUME_MOUNTED ||
1608 atomic_read(&osb->vol_state) == VOLUME_DISABLED);
1610 /* If there's an error on mount, then we may never get to the
1611 * MOUNTED flag, but this is set right before
1612 * dismount_volume() so we can trust it. */
1613 if (atomic_read(&osb->vol_state) == VOLUME_DISABLED) {
1614 mlog(0, "mount error, exiting!\n");
1621 static int ocfs2_commit_thread(void *arg)
1624 struct ocfs2_super *osb = arg;
1625 struct ocfs2_journal *journal = osb->journal;
1627 /* we can trust j_num_trans here because _should_stop() is only set in
1628 * shutdown and nobody other than ourselves should be able to start
1629 * transactions. committing on shutdown might take a few iterations
1630 * as final transactions put deleted inodes on the list */
1631 while (!(kthread_should_stop() &&
1632 atomic_read(&journal->j_num_trans) == 0)) {
1634 wait_event_interruptible(osb->checkpoint_event,
1635 atomic_read(&journal->j_num_trans)
1636 || kthread_should_stop());
1638 status = ocfs2_commit_cache(osb);
1642 if (kthread_should_stop() && atomic_read(&journal->j_num_trans)){
1644 "commit_thread: %u transactions pending on "
1646 atomic_read(&journal->j_num_trans));
1653 /* Look for a dirty journal without taking any cluster locks. Used for
1654 * hard readonly access to determine whether the file system journals
1655 * require recovery. */
1656 int ocfs2_check_journals_nolocks(struct ocfs2_super *osb)
1660 struct buffer_head *di_bh;
1661 struct ocfs2_dinode *di;
1662 struct inode *journal = NULL;
1664 for(slot = 0; slot < osb->max_slots; slot++) {
1665 journal = ocfs2_get_system_file_inode(osb,
1666 JOURNAL_SYSTEM_INODE,
1668 if (!journal || is_bad_inode(journal)) {
1675 ret = ocfs2_read_block(osb, OCFS2_I(journal)->ip_blkno, &di_bh,
1682 di = (struct ocfs2_dinode *) di_bh->b_data;
1684 if (le32_to_cpu(di->id1.journal1.ij_flags) &
1685 OCFS2_JOURNAL_DIRTY_FL)