Review: Repair multi-threading code

@@ -577,6 +577,7 @@
 #define	LIBXFS_LIO_TYPE_RAW		0x3
 
 #define LIBXFS_BBTOOFF64(bbs)	(((xfs_off_t)(bbs)) << BBSHIFT)
+extern int libxfs_nproc(void);
 
 #include <xfs/xfs_ialloc.h>
 #include <xfs/xfs_rtalloc.h>

@@ -118,3 +118,9 @@
 {
 	return (sizeof(void *));
 }
+
+int
+platform_nproc(void)
+{
+	return 1;
+}

@@ -178,3 +178,9 @@
 {
 	return (sizeof(void *));
 }
+
+int
+platform_nproc(void)
+{
+	return 1;
+}

@@ -703,3 +703,9 @@
 	c = asctime(localtime(&t));
 	fprintf(fp, "%s", c);
 }
+
+int
+libxfs_nproc(void)
+{
+	return platform_nproc();
+}

@@ -33,5 +33,6 @@
 extern int platform_direct_blockdev (void);
 extern int platform_align_blockdev (void);
 extern int platform_aio_init (int aio_count);
+extern int platform_nproc(void);
 
 #endif	/* LIBXFS_INIT_H */

@@ -19,6 +19,7 @@
 #include <xfs/libxfs.h>
 #include <aio.h>
 #include <diskinfo.h>
+#include <sys/sysmp.h>
 
 extern char *progname;
 extern __int64_t findsize(char *);

@@ -102,3 +103,10 @@
 {
 	return (sizeof(void *));
 }
+
+int
+platform_nproc(void)
+{
+	return sysmp(MP_NPROCS);
+}
+

@@ -207,3 +207,24 @@
 		return getpagesize();
 	return max_block_alignment;
 }
+
+int
+platform_nproc(void)
+{
+	FILE *fp;
+	char buffer[256];
+	int nproc;
+
+	fp = fopen("/proc/stat", "r");
+	if (fp) {
+		nproc = 0;
+		while (fgets(buffer, sizeof(buffer), fp)) {
+			if (strncmp(buffer, "cpu", 3) == 0)
+				nproc++;
+		}
+		fclose(fp);
+		if (nproc > 1)
+			return (nproc - 1);	/* discard the initial aggregate cpu entry */
+	}
+	return 1;
+}

@@ -26,6 +26,7 @@
 #include "dir.h"
 #include "dinode.h"
 #include "prefetch.h"
+#include "threads.h"
 #include "versions.h"
 
 /*

@@ -148,16 +149,19 @@
 		if (check_inode_block(mp, ino) == 0)
 			return(0);
 
+		PREPAIR_RW_WRITE_LOCK(&per_ag_lock[agno]);
 		switch (state = get_agbno_state(mp, agno, agbno))  {
 		case XR_E_INO:
 			do_warn(
 		_("uncertain inode block %d/%d already known\n"),
 				agno, agbno);
+			PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 			break;
 		case XR_E_UNKNOWN:
 		case XR_E_FREE1:
 		case XR_E_FREE:
 			set_agbno_state(mp, agno, agbno, XR_E_INO);
+			PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 			break;
 		case XR_E_MULT:
 		case XR_E_INUSE:

@@ -170,6 +174,7 @@
 		_("inode block %d/%d multiply claimed, (state %d)\n"),
 				agno, agbno, state);
 			set_agbno_state(mp, agno, agbno, XR_E_MULT);
+			PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 			return(0);
 		default:
 			do_warn(

@@ -176,6 +181,7 @@
 		_("inode block %d/%d bad state, (state %d)\n"),
 				agno, agbno, state);
 			set_agbno_state(mp, agno, agbno, XR_E_INO);
+			PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 			break;
 		}
 

@@ -425,6 +431,7 @@
 	 * user data -- we're probably here as a result of a directory
 	 * entry or an iunlinked pointer
 	 */
+	PREPAIR_RW_WRITE_LOCK(&per_ag_lock[agno]);
 	for (j = 0, cur_agbno = chunk_start_agbno;
 			cur_agbno < chunk_stop_agbno; cur_agbno++)  {
 		switch (state = get_agbno_state(mp, agno, cur_agbno))  {

@@ -447,9 +454,12 @@
 			break;
 		}
 
-		if (j)
+		if (j) {
+			PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 			return(0);
+		}
 	}
+	PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 
 	/*
 	 * ok, chunk is good.  put the record into the tree if required,

@@ -472,6 +482,7 @@
 
 	set_inode_used(irec_p, agino - start_agino);
 
+	PREPAIR_RW_WRITE_LOCK(&per_ag_lock[agno]);
 	for (cur_agbno = chunk_start_agbno;
 			cur_agbno < chunk_stop_agbno; cur_agbno++)  {
 		switch (state = get_agbno_state(mp, agno, cur_agbno))  {

@@ -501,6 +512,7 @@
 			break;
 		}
 	}
+	PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 
 	return(ino_cnt);
 }

@@ -700,6 +712,7 @@
 	/*
 	 * mark block as an inode block in the incore bitmap
 	 */
+	PREPAIR_RW_WRITE_LOCK(&per_ag_lock[agno]);
 	switch (state = get_agbno_state(mp, agno, agbno))  {
 	case XR_E_INO:	/* already marked */
 		break;

@@ -717,6 +730,7 @@
 			XFS_AGB_TO_FSB(mp, agno, agbno), state);
 		break;
 	}
+	PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 
 	while (!done)  {
 		/*

@@ -869,6 +883,7 @@
 			ibuf_offset = 0;
 			agbno++;
 
+			PREPAIR_RW_WRITE_LOCK(&per_ag_lock[agno]);
 			switch (state = get_agbno_state(mp, agno, agbno))  {
 			case XR_E_INO:	/* already marked */
 				break;

@@ -888,6 +903,7 @@
 					XFS_AGB_TO_FSB(mp, agno, agbno), state);
 				break;
 			}
+			PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 
 		} else if (irec_offset == XFS_INODES_PER_CHUNK)  {
 			/*

184 lines not on this page · raw message

@@ -19,6 +19,7 @@
 #include <libxfs.h>
 #include "dir_stack.h"
 #include "err_protos.h"
+#include "threads.h"
 
 /*
  * a directory stack for holding directories while

@@ -29,7 +30,10 @@
 
 static dir_stack_t	dirstack_freelist;
 static int		dirstack_init = 0;
+static pthread_mutex_t	dirstack_mutex;
+static pthread_mutexattr_t dirstack_mutexattr;
 
+
 void
 dir_stack_init(dir_stack_t *stack)
 {

@@ -38,6 +42,11 @@
 
 	if (dirstack_init == 0)  {
 		dirstack_init = 1;
+		PREPAIR_MTX_ATTR_INIT(&dirstack_mutexattr);
+#ifdef PTHREAD_MUTEX_SPINBLOCK_NP
+		PREPAIR_MTX_ATTR_SET(&dirstack_mutexattr, PTHREAD_MUTEX_SPINBLOCK_NP);
+#endif
+		PREPAIR_MTX_LOCK_INIT(&dirstack_mutex, &dirstack_mutexattr);
 		dir_stack_init(&dirstack_freelist);
 	}
 

@@ -85,8 +94,10 @@
 {
 	dir_stack_elem_t *elem;
 
+	PREPAIR_MTX_LOCK(&dirstack_mutex);
 	if (dirstack_freelist.cnt == 0)  {
 		if ((elem = malloc(sizeof(dir_stack_elem_t))) == NULL)  {
+			PREPAIR_MTX_UNLOCK(&dirstack_mutex);
 			do_error(
 		_("couldn't malloc dir stack element, try more swap\n"));
 			exit(1);

@@ -94,6 +105,7 @@
 	} else  {
 		elem = dir_stack_pop(&dirstack_freelist);
 	}
+	PREPAIR_MTX_UNLOCK(&dirstack_mutex);
 
 	elem->ino = ino;
 

@@ -116,7 +128,9 @@
 	ino = elem->ino;
 	elem->ino = NULLFSINO;
 
+	PREPAIR_MTX_LOCK(&dirstack_mutex);
 	dir_stack_push(&dirstack_freelist, elem);
+	PREPAIR_MTX_UNLOCK(&dirstack_mutex);
 
 	return(ino);
 }

@@ -191,8 +191,10 @@
 EXTERN __uint32_t	sb_unit;
 EXTERN __uint32_t	sb_width;
 
-extern size_t ts_dirbuf_size;
-extern size_t ts_dir_freemap_size;
-extern size_t ts_attr_freemap_size;
+extern size_t 		ts_dirbuf_size;
+extern size_t 		ts_dir_freemap_size;
+extern size_t 		ts_attr_freemap_size;
+
+EXTERN pthread_rwlock_t	*per_ag_lock;
 
 #endif /* _XFS_REPAIR_GLOBAL_H */

@@ -23,6 +23,7 @@
 #include "agheader.h"
 #include "protos.h"
 #include "err_protos.h"
+#include "threads.h"
 
 /*
  * push a block allocation record onto list.  assumes list

@@ -64,6 +65,7 @@
 		do_error(_("couldn't allocate block map pointers\n"));
 		return;
 	}
+	PREPAIR_RW_LOCK_ALLOC(per_ag_lock, agno);
 	for (i = 0; i < agno; i++)  {
 		size = roundup((numblocks+(NBBY/XR_BB)-1) / (NBBY/XR_BB),
 		       		sizeof(__uint64_t));

@@ -75,6 +77,7 @@
 			return;
 		}
 		bzero(ba_bmap[i], size);
+		PREPAIR_RW_LOCK_INIT(&per_ag_lock[i], NULL);
 	}
 
 	if (rtblocks == 0)  {

@@ -24,6 +24,7 @@
 #include "protos.h"
 #include "err_protos.h"
 #include "avl64.h"
+#include "threads.h"
 #define ALLOC_NUM_EXTS		100
 
 /*

@@ -92,6 +93,13 @@
 static ba_rec_t		*rt_ba_list;
 
 /*
+ * locks.
+ */
+static pthread_rwlock_t ext_flist_lock;
+static pthread_rwlock_t rt_ext_tree_lock;
+static pthread_rwlock_t rt_ext_flist_lock;
+
+/*
  * extent tree stuff is avl trees of duplicate extents,
  * sorted in order by block number.  there is one tree per ag.
  */

@@ -104,6 +112,7 @@
 	extent_tree_node_t *new;
 	extent_alloc_rec_t *rec;
 
+	PREPAIR_RW_WRITE_LOCK(&ext_flist_lock);
 	if (ext_flist.cnt == 0)  {
 		ASSERT(ext_flist.list == NULL);
 

@@ -130,6 +139,7 @@
 	ext_flist.list = (extent_tree_node_t *) new->avl_node.avl_nextino;
 	ext_flist.cnt--;
 	new->avl_node.avl_nextino = NULL;
+	PREPAIR_RW_UNLOCK(&ext_flist_lock);
 
 	/* initialize node */
 

@@ -145,9 +155,11 @@
 void
 release_extent_tree_node(extent_tree_node_t *node)
 {
+	PREPAIR_RW_WRITE_LOCK(&ext_flist_lock);
 	node->avl_node.avl_nextino = (avlnode_t *) ext_flist.list;
 	ext_flist.list = node;
 	ext_flist.cnt++;
+	PREPAIR_RW_UNLOCK(&ext_flist_lock);
 
 	return;
 }

@@ -658,6 +670,7 @@
 	rt_extent_tree_node_t *new;
 	rt_extent_alloc_rec_t *rec;
 
+	PREPAIR_RW_WRITE_LOCK(&rt_ext_flist_lock);
 	if (rt_ext_flist.cnt == 0)  {
 		ASSERT(rt_ext_flist.list == NULL);
 

@@ -684,6 +697,7 @@
 	rt_ext_flist.list = (rt_extent_tree_node_t *) new->avl_node.avl_nextino;
 	rt_ext_flist.cnt--;
 	new->avl_node.avl_nextino = NULL;
+	PREPAIR_RW_UNLOCK(&rt_ext_flist_lock);
 
 	/* initialize node */
 

@@ -762,6 +776,7 @@
 	xfs_drtbno_t new_startblock;
 	xfs_extlen_t new_blockcount;
 
+	PREPAIR_RW_WRITE_LOCK(&rt_ext_tree_lock);
 	avl64_findranges(rt_ext_tree_ptr, startblock - 1,
 		startblock + blockcount + 1,
 		(avl64node_t **) &first, (avl64node_t **) &last);

@@ -779,6 +794,7 @@
 			do_error(_("duplicate extent range\n"));
 		}
 
+		PREPAIR_RW_UNLOCK(&rt_ext_tree_lock);
 		return;
 	}
 

@@ -802,8 +818,10 @@
 		 * just bail if the new extent is contained within an old one
 		 */
 		if (ext->rt_startblock <= startblock &&
-				ext->rt_blockcount >= blockcount)
+				ext->rt_blockcount >= blockcount) {
+			PREPAIR_RW_UNLOCK(&rt_ext_tree_lock);
 			return;
+		}
 		/*
 		 * now check for overlaps and adjacent extents
 		 */

@@ -831,6 +849,7 @@
 		do_error(_("duplicate extent range\n"));
 	}
 
+	PREPAIR_RW_UNLOCK(&rt_ext_tree_lock);
 	return;
 }
 

@@ -841,10 +860,15 @@
 int
 search_rt_dup_extent(xfs_mount_t *mp, xfs_drtbno_t bno)
 {
-	if (avl64_findrange(rt_ext_tree_ptr, bno) != NULL)
-		return(1);
+	int ret;
 
-	return(0);
+	PREPAIR_RW_READ_LOCK(&rt_ext_tree_lock);
+	if (avl64_findrange(rt_ext_tree_ptr, bno) != NULL)
+		ret = 1;
+	else
+		ret = 0;
+	PREPAIR_RW_UNLOCK(&rt_ext_tree_lock);
+	return(ret);
 }
 
 static __uint64_t

@@ -873,6 +897,9 @@
 
 	ba_list = NULL;
 	rt_ba_list = NULL;
+	PREPAIR_RW_LOCK_INIT(&ext_flist_lock, NULL);
+	PREPAIR_RW_LOCK_INIT(&rt_ext_tree_lock, NULL);
+	PREPAIR_RW_LOCK_INIT(&rt_ext_flist_lock, NULL);
 
 	if ((extent_tree_ptrs = malloc(agcount *
 					sizeof(avltree_desc_t *))) == NULL)

@@ -22,8 +22,10 @@
 #include "incore.h"
 #include "agheader.h"
 #include "protos.h"
+#include "threads.h"
 #include "err_protos.h"
 
+static pthread_rwlock_t ino_flist_lock;
 extern avlnode_t	*avl_firstino(avlnode_t *root);
 
 /*

@@ -69,6 +71,7 @@
 	ino_tree_node_t *new;
 	avlnode_t *node;
 
+	PREPAIR_RW_WRITE_LOCK(&ino_flist_lock);
 	if (ino_flist.cnt == 0)  {
 		ASSERT(ino_flist.list == NULL);
 

@@ -92,6 +95,7 @@
 	ino_flist.cnt--;
 	node = &new->avl_node;
 	node->avl_nextino = node->avl_forw = node->avl_back = NULL;
+	PREPAIR_RW_UNLOCK(&ino_flist_lock);
 
 	/* initialize node */
 

@@ -115,6 +119,7 @@
 	ino_rec->avl_node.avl_forw = NULL;
 	ino_rec->avl_node.avl_back = NULL;
 
+	PREPAIR_RW_WRITE_LOCK(&ino_flist_lock);
 	if (ino_flist.list != NULL)  {
 		ASSERT(ino_flist.cnt > 0);
 		ino_rec->avl_node.avl_nextino = (avlnode_t *) ino_flist.list;

@@ -132,6 +137,7 @@
 		if (ino_rec->ino_un.plist != NULL)
 			free(ino_rec->ino_un.plist);
 	}
+	PREPAIR_RW_UNLOCK(&ino_flist_lock);
 
 	return;
 }

@@ -643,6 +649,7 @@
 	int i;
 	int agcount = mp->m_sb.sb_agcount;
 
+	PREPAIR_RW_LOCK_INIT(&ino_flist_lock, NULL);
 	if ((inode_tree_ptrs = malloc(agcount *
 					sizeof(avltree_desc_t *))) == NULL)
 		do_error(_("couldn't malloc inode tree descriptor table\n"));

@@ -43,12 +43,17 @@
 }
 
 static void
-ts_init(void)
+ts_create(void)
 {
 	/* create thread specific keys */
 	pthread_key_create(&dirbuf_key, NULL);
 	pthread_key_create(&dir_freemap_key, NULL);
 	pthread_key_create(&attr_freemap_key, NULL);
+}
+
+void
+ts_init(void)
+{
 
 	/* allocate thread specific storage */
 	ts_alloc(dirbuf_key, 1, ts_dirbuf_size);

@@ -136,6 +141,7 @@
 	if (!libxfs_init(args))
 		do_error(_("couldn't initialize XFS library\n"));
 
+	ts_create();
 	ts_init();
 	increase_rlimit();
 	if (do_prefetch) {

@@ -143,4 +149,5 @@
 		if (do_prefetch)
 			libxfs_lio_allocate();
 	}
+	thread_init();
 }

@@ -24,6 +24,7 @@
 #include "protos.h"
 #include "err_protos.h"
 #include "dinode.h"
+#include "threads.h"
 
 /*
  * walks an unlinked list, returns 1 on an error (bogus pointer) or

@@ -57,6 +58,7 @@
 				add_aginode_uncertain(agno, current_ino, 1);
 				agbno = XFS_AGINO_TO_AGBNO(mp, current_ino);
 
+				PREPAIR_RW_WRITE_LOCK(&per_ag_lock[agno]);
 				switch (state = get_agbno_state(mp,
 							agno, agbno))  {
 				case XR_E_UNKNOWN:

@@ -64,8 +66,10 @@
 				case XR_E_FREE1:
 					set_agbno_state(mp, agno, agbno,
 						XR_E_INO);
+					PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 					break;
 				case XR_E_BAD_STATE:
+					PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 					do_error(_(
 						"bad state in block map %d\n"),
 						state);

@@ -84,6 +88,7 @@
 					 */
 					set_agbno_state(mp, agno, agbno,
 						XR_E_INO);
+					PREPAIR_RW_UNLOCK(&per_ag_lock[agno]);
 					break;
 				}
 			}

@@ -144,6 +149,17 @@
 }
 
 void
+parallel_p3_process_aginodes(xfs_mount_t *mp, xfs_agnumber_t agno)
+{
+	/*
+	 * turn on directory processing (inode discovery) and
+	 * attribute processing (extra_attr_check)
+	 */
+	do_log(_("        - agno = %d\n"), agno);
+	process_aginodes(mp, agno, 1, 0, 1);
+}
+
+void
 phase3(xfs_mount_t *mp)
 {
 	int i, j;

@@ -171,13 +187,9 @@
 	    "        - process known inodes and perform inode discovery...\n"));
 
 	for (i = 0; i < mp->m_sb.sb_agcount; i++)  {
-		do_log(_("        - agno = %d\n"), i);
-		/*
-		 * turn on directory processing (inode discovery) and
-		 * attribute processing (extra_attr_check)
-		 */
-		process_aginodes(mp, i, 1, 0, 1);
+		queue_work(parallel_p3_process_aginodes, mp, i);
 	}
+	wait_for_workers();
 
 	/*
 	 * process newly discovered inode chunks

@@ -28,6 +28,7 @@
 #include "bmap.h"
 #include "versions.h"
 #include "dir2.h"
+#include "threads.h"
 
 
 /* ARGSUSED */

@@ -1119,6 +1120,18 @@
 
 
 void
+parallel_p4_process_aginodes(xfs_mount_t *mp, xfs_agnumber_t agno)
+{
+	do_log(_("        - agno = %d\n"), agno);
+	process_aginodes(mp, agno, 0, 1, 0);
+
+	/*
+	 * now recycle the per-AG duplicate extent records
+	 */
+	release_dup_extent_tree(agno);
+}
+
+void
 phase4(xfs_mount_t *mp)
 {
 	ino_tree_node_t		*irec;

@@ -1325,14 +1338,9 @@
 		 * and attribute processing is turned OFF since we did that
 		 * already in phase 3.
 		 */
-		do_log(_("        - agno = %d\n"), i);
-		process_aginodes(mp, i, 0, 1, 0);
-
-		/*
-		 * now recycle the per-AG duplicate extent records
-		 */
-		release_dup_extent_tree(i);
+		queue_work(parallel_p4_process_aginodes, mp, i);
 	}
+	wait_for_workers();
 
 	/*
 	 * free up memory used to track trealtime duplicate extents

167 lines not on this page · raw message

@@ -26,6 +26,7 @@
 #include "dinode.h"
 #include "versions.h"
 #include "prefetch.h"
+#include "threads.h"
 
 /* dinoc is a pointer to the IN-CORE dinode core */
 void

@@ -180,8 +181,9 @@
 	libxfs_bcache_purge();
 
 	for (i = 0; i < glob_agcount; i++)  {
-		phase7_alt_function(mp, i);
+		queue_work(phase7_alt_function, mp, i);
 	}
+	wait_for_workers();
 }
 
 void

@@ -44,4 +44,6 @@
 extern void *ts_attr_freemap(void);
 extern void *ts_dir_freemap(void);
 extern void *ts_dirbuf(void);
+extern void ts_init(void);
+extern void thread_init(void);
 

181 lines not on this page · raw message

308 lines not on this page · raw message

@@ -0,0 +1,37 @@
+#ifndef	_XFS_REPAIR_THREADS_H_
+#define	_XFS_REPAIR_THREADS_H_
+
+extern int		do_parallel;
+extern int		thread_count;
+/*
+**  locking variants - rwlock/mutex
+*/
+#define PREPAIR_RW_LOCK_ATTR		PTHREAD_PROCESS_PRIVATE
+
+#define	PREPAIR_RW_LOCK_ALLOC(lkp, n)				\
+	if (do_parallel) {					\
+		lkp = malloc(n*sizeof(pthread_rwlock_t));	\
+		if (lkp == NULL)				\
+			do_error("cannot alloc %d locks\n", n);	\
+			/* NO RETURN */				\
+	}
+#define PREPAIR_RW_LOCK_INIT(l,a)	if (do_parallel) pthread_rwlock_init((l),(a))
+#define PREPAIR_RW_READ_LOCK(l) 	if (do_parallel) pthread_rwlock_rdlock((l))
+#define PREPAIR_RW_WRITE_LOCK(l)	if (do_parallel) pthread_rwlock_wrlock((l))
+#define PREPAIR_RW_UNLOCK(l)		if (do_parallel) pthread_rwlock_unlock((l))
+#define PREPAIR_RW_WRITE_LOCK_NOTEST(l)	pthread_rwlock_wrlock((l))
+#define PREPAIR_RW_UNLOCK_NOTEST(l)	pthread_rwlock_unlock((l))
+#define PREPAIR_RW_LOCK_DELETE(l)	if (do_parallel) pthread_rwlock_destroy((l))
+
+#define PREPAIR_MTX_LOCK_INIT(m, a)	if (do_parallel) pthread_mutex_init((m), (a))
+#define PREPAIR_MTX_ATTR_INIT(a)	if (do_parallel) pthread_mutexattr_init((a))
+#define PREPAIR_MTX_ATTR_SET(a, l)	if (do_parallel) pthread_mutexattr_settype((a), l)
+#define PREPAIR_MTX_LOCK(m)		if (do_parallel) pthread_mutex_lock(m)
+#define PREPAIR_MTX_UNLOCK(m)		if (do_parallel) pthread_mutex_unlock(m)
+
+
+typedef void	disp_func_t(xfs_mount_t *mp, xfs_agnumber_t agno);
+extern	int	queue_work(disp_func_t func, xfs_mount_t *mp, xfs_agnumber_t agno);
+extern	void	wait_for_workers(void);
+
+#endif	/* _XFS_REPAIR_THREADS_H_ */