19278972970

Committed 11 Nov 2025 09:25PM UTC coverage: 62.316% (+14.3%) from 48.052%

Build # 19278972970

Build Type

push

github

Committed by

zommiommy

Commit Message

BatchCodec print stats and encoding time

Run Details

40 of 53 new or added lines in 4 files covered. (75.47%)

814 existing lines in 46 files now uncovered.

5252 of 8428 relevant lines covered (62.32%)

29580067.92 hits per line

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

54.24

/webgraph/src/transform/perm.rs

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

use crate::graphs::arc_list_graph;
use crate::prelude::sort_pairs::KMergeIters;
use crate::prelude::*;
use anyhow::{ensure, Context, Result};
use dsi_progress_logger::prelude::*;
use lender::*;
use rayon::ThreadPool;
use tempfile::Builder;
use value_traits::slices::SliceByValue;

/// Returns a [sequential](crate::traits::SequentialGraph) permuted graph.
///
/// Note that if the graph is [splittable](SplitLabeling),
/// [`permute_split`] will be much faster.
///
/// This assumes that the permutation is bijective. For the meaning of the
/// additional parameter, see
/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
pub fn permute(
    graph: &impl SequentialGraph,
    perm: &impl SliceByValue<Value = usize>,
    memory_usage: MemoryUsage,
) -> Result<Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>> {
    ensure!(
        perm.len() == graph.num_nodes(),
        "The given permutation has {} values and thus it's incompatible with a graph with {} nodes.",
        perm.len(),
        graph.num_nodes(),
    );
    let dir = Builder::new().prefix("permute_").tempdir()?;
    log::info!(
        "Creating a temporary directory for the sorted pairs: {}",
        dir.path().display()
    );

    // create a stream where to dump the sorted pairs
    let mut sorted = SortPairs::new(memory_usage, dir.path())?;

    // get a permuted view
    let pgraph = PermutedGraph { graph, perm };

    let mut pl = ProgressLogger::default();
    pl.item_name("node")
        .expected_updates(Some(graph.num_nodes()));
    pl.start("Creating batches...");
    // create batches of sorted edges
    for_!( (src, succ) in pgraph.iter() {
        for dst in succ {
            sorted.push(src, dst)?;
        }
        pl.light_update();
    });

    // get a graph on the sorted data
    let edges = sorted.iter().context("Could not read arcs")?;
    let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), edges);
    pl.done();

    Ok(Left(sorted))
}

/// Returns a [sequential](crate::traits::SequentialGraph) permuted graph
/// starting from a [splittable](SplitLabeling) graph.
///
/// Note that if the graph is not [splittable](SplitLabeling) you must use
/// [`permute`], albeit it will be slower.
///
/// This assumes that the permutation is bijective. For the meaning of the
/// additional parameter, see
/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
pub fn permute_split<S, P>(
    graph: &S,
    perm: &P,
    memory_usage: MemoryUsage,
    threads: &ThreadPool,
) -> Result<Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>>
where
    S: SequentialGraph + SplitLabeling,
    P: SliceByValue<Value = usize> + Send + Sync + Clone,
    for<'a> <S as SequentialLabeling>::Lender<'a>: Send + Sync + Clone + ExactSizeLender,
{
    ensure!(
        perm.len() == graph.num_nodes(),
        "The given permutation has {} values and thus it's incompatible with a graph with {} nodes.",
        perm.len(),
        graph.num_nodes(),
    );

    // get a permuted view
    let pgraph = PermutedGraph { graph, perm };

    let num_threads = threads.current_num_threads();
    let mut dirs = vec![];

    let edges = threads.in_place_scope(|scope| {
        let (tx, rx) = std::sync::mpsc::channel();

        for (thread_id, iter) in pgraph.split_iter(num_threads).enumerate() {
            let tx = tx.clone();
            let dir = Builder::new()
                .prefix(&format!("permute_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 / num_threads, dir_path).unwrap();
                for_!( (src, succ) in iter {
                    for dst in succ {
                        sorted.push(src, dst).unwrap();
                    }
                });
                tx.send(sorted.iter().context("Could not read arcs").unwrap())
                    .expect("Could not send the sorted pairs");
                log::debug!("Thread {thread_id} finished");
            });
        }
        drop(tx);

        // get a graph on the sorted data
        log::debug!("Waiting for threads to finish");
        rx.iter().sum()
    });

    log::debug!("All threads finished");
    Ok(Left(arc_list_graph::ArcListGraph::new_labeled(
        graph.num_nodes(),
        edges,
    )))
}

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::arc_list_graph;
8	use crate::prelude::sort_pairs::KMergeIters;
9	use crate::prelude::*;
10	use anyhow::{ensure, Context, Result};
11	use dsi_progress_logger::prelude::*;
12	use lender::*;
13	use rayon::ThreadPool;
14	use tempfile::Builder;
15	use value_traits::slices::SliceByValue;
16
17	/// Returns a [sequential](crate::traits::SequentialGraph) permuted graph.
18	///
19	/// Note that if the graph is [splittable](SplitLabeling),
20	/// [`permute_split`] will be much faster.
21	///
22	/// This assumes that the permutation is bijective. For the meaning of the
23	/// additional parameter, see
24	/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
UNCOV 25	pub fn permute(	×
26	graph: &impl SequentialGraph,
27	perm: &impl SliceByValue<Value = usize>,
28	memory_usage: MemoryUsage,
29	) -> Result<Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>> {
UNCOV 30	ensure!(	×
31	perm.len() == graph.num_nodes(),	×
32	"The given permutation has {} values and thus it's incompatible with a graph with {} nodes.",	×
33	perm.len(),	×
34	graph.num_nodes(),	×
35	);
UNCOV 36	let dir = Builder::new().prefix("permute_").tempdir()?;	×
37	log::info!(	×
38	"Creating a temporary directory for the sorted pairs: {}",	×
39	dir.path().display()	×
40	);
41
42	// create a stream where to dump the sorted pairs
UNCOV 43	let mut sorted = SortPairs::new(memory_usage, dir.path())?;	×
44
45	// get a permuted view
UNCOV 46	let pgraph = PermutedGraph { graph, perm };	×
47
UNCOV 48	let mut pl = ProgressLogger::default();	×
49	pl.item_name("node")	×
50	.expected_updates(Some(graph.num_nodes()));	×
51	pl.start("Creating batches...");	×
52	// create batches of sorted edges
UNCOV 53	for_!( (src, succ) in pgraph.iter() {	×
54	for dst in succ {	×
55	sorted.push(src, dst)?;	×
56	}
UNCOV 57	pl.light_update();	×
58	});
59
60	// get a graph on the sorted data
UNCOV 61	let edges = sorted.iter().context("Could not read arcs")?;	×
62	let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), edges);	×
63	pl.done();	×
64
UNCOV 65	Ok(Left(sorted))	×
66	}
67
68	/// Returns a [sequential](crate::traits::SequentialGraph) permuted graph
69	/// starting from a [splittable](SplitLabeling) graph.
70	///
71	/// Note that if the graph is not [splittable](SplitLabeling) you must use
72	/// [`permute`], albeit it will be slower.
73	///
74	/// This assumes that the permutation is bijective. For the meaning of the
75	/// additional parameter, see
76	/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
77	pub fn permute_split<S, P>(	4✔
78	graph: &S,
79	perm: &P,
80	memory_usage: MemoryUsage,
81	threads: &ThreadPool,
82	) -> Result<Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>>
83	where
84	S: SequentialGraph + SplitLabeling,
85	P: SliceByValue<Value = usize> + Send + Sync + Clone,
86	for<'a> <S as SequentialLabeling>::Lender<'a>: Send + Sync + Clone + ExactSizeLender,
87	{
88	ensure!(	4✔
89	perm.len() == graph.num_nodes(),	12✔
UNCOV 90	"The given permutation has {} values and thus it's incompatible with a graph with {} nodes.",	×
UNCOV 91	perm.len(),	×
92	graph.num_nodes(),	×
93	);
94
95	// get a permuted view
96	let pgraph = PermutedGraph { graph, perm };	8✔
97
98	let num_threads = threads.current_num_threads();	12✔
99	let mut dirs = vec![];	8✔
100
101	let edges = threads.in_place_scope(\|scope\| {	16✔
102	let (tx, rx) = std::sync::mpsc::channel();	12✔
103
104	for (thread_id, iter) in pgraph.split_iter(num_threads).enumerate() {	48✔
105	let tx = tx.clone();	48✔
106	let dir = Builder::new()	48✔
107	.prefix(&format!("permute_split_{thread_id}_"))	32✔
108	.tempdir()	16✔
109	.expect("Could not create a temporary directory");	32✔
110	let dir_path = dir.path().to_path_buf();	48✔
111	dirs.push(dir);	48✔
112	scope.spawn(move \|_\| {	48✔
113	log::debug!("Spawned thread {thread_id}");	16✔
114	let mut sorted = SortPairs::new(memory_usage / num_threads, dir_path).unwrap();	80✔
115	for_!( (src, succ) in iter {	2,604,472✔
116	for dst in succ {	39,896,052✔
117	sorted.push(src, dst).unwrap();	51,458,432✔
118	}
119	});
120	tx.send(sorted.iter().context("Could not read arcs").unwrap())	96✔
121	.expect("Could not send the sorted pairs");	32✔
122	log::debug!("Thread {thread_id} finished");	16✔
123	});
124	}
125	drop(tx);	8✔
126
127	// get a graph on the sorted data
128	log::debug!("Waiting for threads to finish");	4✔
129	rx.iter().sum()	12✔
130	});
131
132	log::debug!("All threads finished");	4✔
133	Ok(Left(arc_list_graph::ArcListGraph::new_labeled(	8✔
134	graph.num_nodes(),	8✔
135	edges,	4✔
136	)))
137	}

vigna / webgraph-rs / 19278972970

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