21871133388

Committed 10 Feb 2026 03:29PM UTC coverage: 61.096% (-0.04%) from 61.137%

Build # 21871133388

Build Type

push

github

Committed by

vigna

Commit Message

Moved to rand 0.10.0, Lender 0.5.0, mmap-rs 0.7.0, sux 0.11.1

Run Details

5071 of 8300 relevant lines covered (61.1%)

49326295.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.07

/webgraph/src/transform/simplify.rs

/*
 * SPDX-FileCopyrightText: 2023 Inria
 *
 * SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
 */

use crate::graphs::{
    arc_list_graph, no_selfloops_graph::NoSelfLoopsGraph, union_graph::UnionGraph,
};
use crate::labels::Left;
use crate::traits::{
    LenderIntoIter, RayonChannelIterExt, SequentialGraph, SortedIterator, SortedLender,
    SplitLabeling,
};
use crate::utils::sort_pairs::{KMergeIters, SortPairs};
use crate::utils::{CodecIter, DefaultBatchCodec, MemoryUsage};
use anyhow::{Context, Result};
use dsi_progress_logger::prelude::*;
use itertools::Itertools;
use lender::*;
use tempfile::Builder;

use super::transpose;

/// Returns a simplified (i.e., undirected and loopless) version of the provided
/// sorted (both on nodes and successors) graph as a [sequential
/// graph](crate::traits::SequentialGraph).
///
/// This method exploits the fact that the input graph is already sorted,
/// sorting half the number of arcs of [`simplify`].
pub fn simplify_sorted<G: SequentialGraph>(
    graph: G,
    memory_usage: MemoryUsage,
) -> Result<
    NoSelfLoopsGraph<
        UnionGraph<
            G,
            Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>,
        >,
    >,
>
where
    for<'a> G::Lender<'a>: SortedLender,
    for<'a, 'b> LenderIntoIter<'a, G::Lender<'b>>: SortedIterator,
{
    let transpose = transpose(&graph, memory_usage).context("Could not transpose the graph")?;
    Ok(NoSelfLoopsGraph(UnionGraph(graph, transpose)))
}

/// Returns a simplified (i.e., undirected and loopless) version of the provided
/// graph as a [sequential graph](crate::traits::SequentialGraph).
///
/// Note that if the graph is sorted (both on nodes and successors), it is
/// recommended to use [`simplify_sorted`].
///
/// For the meaning of the additional parameter, see [`SortPairs`].
pub fn simplify(
    graph: &impl SequentialGraph,
    memory_usage: MemoryUsage,
) -> Result<
    Left<
        arc_list_graph::ArcListGraph<
            impl Iterator<Item = ((usize, usize), ())> + Clone + Send + Sync + 'static,
        >,
    >,
> {
    let dir = Builder::new().prefix("simplify_").tempdir()?;
    let mut sorted = SortPairs::new(memory_usage, dir.path())?;

    let mut pl = ProgressLogger::default();
    pl.item_name("node")
        .expected_updates(Some(graph.num_nodes()));
    pl.start("Creating batches...");
    // create batches of sorted edges
    let mut iter = graph.iter();
    while let Some((src, succ)) = iter.next() {
        for dst in succ {
            if src != dst {
                sorted.push(src, dst)?;
                sorted.push(dst, src)?;
            }
        }
        pl.light_update();
    }
    // merge the batches
    let filter: fn(&((usize, usize), ())) -> bool = |((src, dst), ())| src != dst;
    let iter = Itertools::dedup(sorted.iter()?.filter(filter));
    let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), iter);
    pl.done();

    Ok(Left(sorted))
}

/// Returns a simplified (i.e., undirected and loopless) version of the provided
/// graph as a [sequential graph](crate::traits::SequentialGraph).
///
/// This method uses splitting to sort in parallel different parts of the graph.
///
/// For the meaning of the additional parameter, see [`SortPairs`].
pub fn simplify_split<S>(
    graph: &S,
    memory_usage: MemoryUsage,
) -> Result<
    Left<
        arc_list_graph::ArcListGraph<
            itertools::Dedup<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>,
        >,
    >,
>
where
    S: SequentialGraph + SplitLabeling,
{
    let num_threads = rayon::current_num_threads();
    let (tx, rx) = crossbeam_channel::unbounded();

    let mut dirs = vec![];

    rayon::in_place_scope(|scope| {
        let mut thread_id = 0;
        #[allow(clippy::explicit_counter_loop)] // enumerate requires some extra bounds here
        for iter in graph.split_iter(num_threads) {
            let tx = tx.clone();
            let dir = Builder::new()
                .prefix(&format!("simplify_split_{thread_id}_"))
                .tempdir()
                .expect("Could not create a temporary directory");
            let dir_path = dir.path().to_path_buf();
            dirs.push(dir);
            scope.spawn(move |_| {
                log::debug!("Spawned thread {thread_id}");
                let mut sorted = SortPairs::new(memory_usage, dir_path).unwrap();
                for_!( (src, succ) in iter {
                    for dst in succ {
                        if src != dst {
                            sorted.push(src, dst).unwrap();
                            sorted.push(dst, src).unwrap();
                        }
                    }
                });
                let result = sorted.iter().context("Could not read arcs").unwrap();
                tx.send(result).expect("Could not send the sorted pairs");
                log::debug!("Thread {thread_id} finished");
            });
            thread_id += 1;
        }
    });
    drop(tx);

    // get a graph on the sorted data
    log::debug!("Waiting for threads to finish");
    let edges: KMergeIters<CodecIter<DefaultBatchCodec>> = rx.into_rayon_iter().sum();
    let edges = edges.dedup();
    log::debug!("All threads finished");
    let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), edges);

    drop(dirs);
    Ok(Left(sorted))
}

1	/*
2	* SPDX-FileCopyrightText: 2023 Inria
3	*
4	* SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
5	*/
6
7	use crate::graphs::{
8	arc_list_graph, no_selfloops_graph::NoSelfLoopsGraph, union_graph::UnionGraph,
9	};
10	use crate::labels::Left;
11	use crate::traits::{
12	LenderIntoIter, RayonChannelIterExt, SequentialGraph, SortedIterator, SortedLender,
13	SplitLabeling,
14	};
15	use crate::utils::sort_pairs::{KMergeIters, SortPairs};
16	use crate::utils::{CodecIter, DefaultBatchCodec, MemoryUsage};
17	use anyhow::{Context, Result};
18	use dsi_progress_logger::prelude::*;
19	use itertools::Itertools;
20	use lender::*;
21	use tempfile::Builder;
22
23	use super::transpose;
24
25	/// Returns a simplified (i.e., undirected and loopless) version of the provided
26	/// sorted (both on nodes and successors) graph as a [sequential
27	/// graph](crate::traits::SequentialGraph).
28	///
29	/// This method exploits the fact that the input graph is already sorted,
30	/// sorting half the number of arcs of [`simplify`].
31	pub fn simplify_sorted<G: SequentialGraph>(	1✔
32	graph: G,
33	memory_usage: MemoryUsage,
34	) -> Result<
35	NoSelfLoopsGraph<
36	UnionGraph<
37	G,
38	Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>,
39	>,
40	>,
41	>
42	where
43	for<'a> G::Lender<'a>: SortedLender,
44	for<'a, 'b> LenderIntoIter<'a, G::Lender<'b>>: SortedIterator,
45	{
46	let transpose = transpose(&graph, memory_usage).context("Could not transpose the graph")?;	5✔
47	Ok(NoSelfLoopsGraph(UnionGraph(graph, transpose)))	1✔
48	}
49
50	/// Returns a simplified (i.e., undirected and loopless) version of the provided
51	/// graph as a [sequential graph](crate::traits::SequentialGraph).
52	///
53	/// Note that if the graph is sorted (both on nodes and successors), it is
54	/// recommended to use [`simplify_sorted`].
55	///
56	/// For the meaning of the additional parameter, see [`SortPairs`].
57	pub fn simplify(	×
58	graph: &impl SequentialGraph,
59	memory_usage: MemoryUsage,
60	) -> Result<
61	Left<
62	arc_list_graph::ArcListGraph<
63	impl Iterator<Item = ((usize, usize), ())> + Clone + Send + Sync + 'static,
64	>,
65	>,
66	> {
67	let dir = Builder::new().prefix("simplify_").tempdir()?;	×
68	let mut sorted = SortPairs::new(memory_usage, dir.path())?;	×
69
70	let mut pl = ProgressLogger::default();	×
71	pl.item_name("node")	×
72	.expected_updates(Some(graph.num_nodes()));	×
73	pl.start("Creating batches...");	×
74	// create batches of sorted edges
75	let mut iter = graph.iter();	×
76	while let Some((src, succ)) = iter.next() {	×
77	for dst in succ {	×
78	if src != dst {	×
79	sorted.push(src, dst)?;	×
80	sorted.push(dst, src)?;	×
81	}
82	}
83	pl.light_update();	×
84	}
85	// merge the batches
86	let filter: fn(&((usize, usize), ())) -> bool = \|((src, dst), ())\| src != dst;	×
87	let iter = Itertools::dedup(sorted.iter()?.filter(filter));	×
88	let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), iter);	×
89	pl.done();	×
90
91	Ok(Left(sorted))	×
92	}
93
94	/// Returns a simplified (i.e., undirected and loopless) version of the provided
95	/// graph as a [sequential graph](crate::traits::SequentialGraph).
96	///
97	/// This method uses splitting to sort in parallel different parts of the graph.
98	///
99	/// For the meaning of the additional parameter, see [`SortPairs`].
100	pub fn simplify_split<S>(	4✔
101	graph: &S,
102	memory_usage: MemoryUsage,
103	) -> Result<
104	Left<
105	arc_list_graph::ArcListGraph<
106	itertools::Dedup<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>,
107	>,
108	>,
109	>
110	where
111	S: SequentialGraph + SplitLabeling,
112	{
113	let num_threads = rayon::current_num_threads();	8✔
114	let (tx, rx) = crossbeam_channel::unbounded();	12✔
115
116	let mut dirs = vec![];	8✔
117
118	rayon::in_place_scope(\|scope\| {	8✔
119	let mut thread_id = 0;	8✔
120	#[allow(clippy::explicit_counter_loop)] // enumerate requires some extra bounds here
121	for iter in graph.split_iter(num_threads) {	28✔
122	let tx = tx.clone();	48✔
123	let dir = Builder::new()	48✔
124	.prefix(&format!("simplify_split_{thread_id}_"))	16✔
125	.tempdir()	16✔
126	.expect("Could not create a temporary directory");	32✔
127	let dir_path = dir.path().to_path_buf();	48✔
128	dirs.push(dir);	48✔
129	scope.spawn(move \|_\| {	48✔
130	log::debug!("Spawned thread {thread_id}");	16✔
131	let mut sorted = SortPairs::new(memory_usage, dir_path).unwrap();	80✔
132	for_!( (src, succ) in iter {	2,604,472✔
133	for dst in succ {	14,166,836✔
134	if src != dst {	25,379,448✔
135	sorted.push(src, dst).unwrap();	75,089,040✔
136	sorted.push(dst, src).unwrap();	50,059,360✔
137	}
138	}
139	});
140	let result = sorted.iter().context("Could not read arcs").unwrap();	80✔
141	tx.send(result).expect("Could not send the sorted pairs");	80✔
142	log::debug!("Thread {thread_id} finished");	16✔
143	});
144	thread_id += 1;	16✔
145	}
146	});
147	drop(tx);	8✔
148
149	// get a graph on the sorted data
150	log::debug!("Waiting for threads to finish");	4✔
151	let edges: KMergeIters<CodecIter<DefaultBatchCodec>> = rx.into_rayon_iter().sum();	20✔
152	let edges = edges.dedup();	12✔
153	log::debug!("All threads finished");	4✔
154	let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), edges);	20✔
155
156	drop(dirs);	8✔
157	Ok(Left(sorted))	4✔
158	}

vigna / webgraph-rs / 21871133388

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous