Skip to content

Multipass performance optimizations (~4.3x speedup) #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
schulzchristian merged 1 commit into from
Apr 2, 2026
Merged

Multipass performance optimizations (~4.3x speedup) #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion code_for_graphs/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,23 @@ if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)
set(CMAKE_BUILD_TYPE "Release")
endif()

# Profile-Guided Optimization (PGO) support
option(PGO_GENERATE "Build with PGO instrumentation (phase 1)" OFF)
option(PGO_USE "Build with PGO optimization from collected profiles (phase 2)" OFF)
if(PGO_GENERATE AND PGO_USE)
message(FATAL_ERROR "PGO_GENERATE and PGO_USE are mutually exclusive. Use PGO_GENERATE first, run a training workload, then rebuild with PGO_USE.")
endif()
if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
if(PGO_GENERATE)
add_definitions(-fprofile-generate=${CMAKE_BINARY_DIR}/pgo_data)
set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fprofile-generate=${CMAKE_BINARY_DIR}/pgo_data")
endif()
if(PGO_USE)
add_definitions(-fprofile-use=${CMAKE_BINARY_DIR}/pgo_data -fprofile-correction)
set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fprofile-use=${CMAKE_BINARY_DIR}/pgo_data")
endif()
Comment on lines +28 to +41

Copilot AI Apr 2, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PGO_GENERATE and PGO_USE can both be enabled simultaneously, which will add conflicting -fprofile-generate / -fprofile-use flags and can cause build failures. Consider enforcing mutual exclusivity (or explicit precedence) between these options.

Copilot uses AI. Check for mistakes.
endif()

# tweak compiler flags
CHECK_CXX_COMPILER_FLAG(-funroll-loops COMPILER_SUPPORTS_FUNROLL_LOOPS)
if(COMPILER_SUPPORTS_FUNROLL_LOOPS)
Expand Down Expand Up @@ -90,7 +107,7 @@ include_directories(${CMAKE_CURRENT_SOURCE_DIR}/lib/tools)



set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-builtin-malloc -fno-builtin-calloc -fno-builtin-realloc")
# Removed -fno-builtin-malloc flags to let compiler optimize allocation patterns

set(LIBKAFFPA_SOURCE_FILES
lib/tools/quality_metrics.cpp
Expand Down
61 changes: 43 additions & 18 deletions code_for_graphs/app/streammultisection.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ int main(int argn, char **argv) {

config.LogDump(stdout);
config.stream_input = true;
config.graph_filename = graph_filename;

[[maybe_unused]] bool already_fully_partitioned;

Expand All @@ -98,27 +99,45 @@ int main(int argn, char **argv) {
for (config.restream_number=0; config.restream_number<passes; config.restream_number++) {

io_t.restart();
graph_io_stream::readFirstLineStream(config, graph_filename, total_edge_cut, qap);
graph_io_stream::loadRemainingLinesToBinary(config, input);
if (config.restream_number == 0 || !config.ram_stream) {
// First pass or non-ram_stream: read file and load data
graph_io_stream::readFirstLineStream(config, graph_filename, total_edge_cut, qap);
graph_io_stream::loadRemainingLinesToBinary(config, input);
} else {
// Subsequent ram_stream passes: reset config without file I/O
config.remaining_stream_nodes = config.total_nodes;
config.remaining_stream_edges = config.total_edges;
config.total_stream_nodeweight = 0;
config.total_stream_nodecounter = 0;
config.stream_n_nodes = config.remaining_stream_nodes;
if (config.num_streams_passes > 1 + config.restream_number) {
config.stream_total_upperbound = ceil(((100+1.5*config.imbalance)/100.)*(config.remaining_stream_nodes/(double)config.k));
} else {
config.stream_total_upperbound = ceil(((100+config.imbalance)/100.)*(config.remaining_stream_nodes/(double)config.k));
}
config.quotient_nodes = config.k;
total_edge_cut = 0;
qap = 0;
config.nmbNodes = MIN(config.stream_buffer_len, config.remaining_stream_nodes);
config.n_batches = ceil(config.remaining_stream_nodes / (double)config.nmbNodes);
config.curr_batch = 0;
}
buffer_io_time += io_t.elapsed();
onepass_partitioner->instantiate_blocks(config.remaining_stream_nodes, config.remaining_stream_edges, config.k, config.imbalance);
if (config.restream_number > 0 && config.use_self_sorting_array) {
onepass_partitioner->reset_sorted_blocks();
}
if (config.stream_rec_bisection) {
onepass_partitioner->create_problem_tree(config.remaining_stream_nodes, config.remaining_stream_edges, config.k,
onepass_partitioner->create_problem_tree(config.remaining_stream_nodes, config.remaining_stream_edges, config.k,
config.enable_mapping, config.stream_rec_biss_orig_alpha, config.non_hashified_layers);
}

omp_set_schedule(config.omp_schedule, config.omp_chunk);
omp_set_num_threads(config.parallel_nodes);

#pragma omp parallel for schedule(static)
for (int i=0; i < config.parallel_nodes; i++) {
config.all_blocks_to_keys[i].resize(config.k);
for (auto & b : config.all_blocks_to_keys[i]) {
b = INVALID_PARTITION;
}
memset(config.all_blocks_to_keys[i].data(), 0xFF, config.k * sizeof(PartitionID));
config.neighbor_blocks[i].resize(config.k);
config.next_key[i] = 0;
if (config.sample_edges) {
Expand All @@ -131,16 +150,18 @@ int main(int argn, char **argv) {
}

processing_t.restart();
#pragma omp parallel for schedule(runtime)
#pragma omp parallel for schedule(runtime)
for (LongNodeID curr_node = 0; curr_node < config.n_batches; curr_node++) {
int my_thread = omp_get_thread_num();
io_t.restart();
if((config.one_pass_algorithm != ONEPASS_HASHING) && (config.one_pass_algorithm != ONEPASS_HASHING_CRC32)) {
graph_io_stream::loadBufferLinesToBinary(config, input, 1);
if (!config.ram_stream) {
io_t.restart();
if((config.one_pass_algorithm != ONEPASS_HASHING) && (config.one_pass_algorithm != ONEPASS_HASHING_CRC32)) {
graph_io_stream::loadBufferLinesToBinary(config, input, 1);
}
buffer_io_time += io_t.elapsed();
}
Comment on lines +156 to 162

Copilot AI Apr 2, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

buffer_io_time is updated from multiple threads inside the OpenMP parallel loop without synchronization, which is a data race (undefined behavior) even if it’s “just timing”. Use a reduction, an atomic add, or a per-thread accumulator merged after the parallel region.

Copilot uses AI. Check for mistakes.
buffer_io_time += io_t.elapsed();
// ***************************** perform partitioning ***************************************
t.restart();
// ***************************** perform partitioning ***************************************
if (config.dynamic_threashold) t.restart();
graph_io_stream::readNodeOnePass(config, curr_node, my_thread, input, onepass_partitioner);
PartitionID block = onepass_partitioner->solve_node(curr_node, 1, my_thread);
(*config.stream_nodes_assign)[curr_node] = block;
Expand All @@ -158,16 +179,20 @@ int main(int argn, char **argv) {
config.sampling_threashold;
config.sampling_threashold = MIN(config.sampling_threashold, 4);
}
global_mapping_time += t.elapsed();
}
total_time += processing_t.elapsed();
global_mapping_time += processing_t.elapsed();

if (config.ram_stream) {
delete input;
/* delete lines; */
if (!config.ram_stream) {
/* Non-ram_stream: input already freed per-node */
}
}

// Free cached input for ram_stream after all passes
if (config.ram_stream && input != NULL) {
delete input;
}

// output some information about the partition that we have computed
std::cout << "graph has " << config.stream_nodes_assign->size() << " nodes and " << config.total_edges << " edges" << std::endl;
std::cout << "Total processing time: " << total_time << std::endl;
Expand Down
104 changes: 92 additions & 12 deletions code_for_graphs/lib/io/graph_io_stream.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,10 @@
#include <unordered_map>
#include <list>
#include <algorithm>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>

#include "definitions.h"
#include "data_structure/graph_access.h"
Expand Down Expand Up @@ -191,20 +195,96 @@ inline void graph_io_stream::loadBufferLinesToBinary(PartitionConfig & partition
inline std::vector<std::vector<LongNodeID>>* graph_io_stream::loadLinesFromStreamToBinary(PartitionConfig & partition_config, LongNodeID num_lines) {
std::vector<std::vector<LongNodeID>>* input;
input = new std::vector<std::vector<LongNodeID>>(num_lines);
std::vector<std::string>* lines;
lines = new std::vector<std::string>(1);
LongNodeID node_counter = 0;
buffered_input *ss2 = NULL;
while( node_counter < num_lines) {
std::getline(*(partition_config.stream_in),(*lines)[0]);
if ((*lines)[0][0] == '%') { // a comment in the file
continue;

if (partition_config.ram_stream) {
// mmap the file for zero-copy access, then parse from the mapped region
std::ifstream& in = *(partition_config.stream_in);
std::streampos cur = in.tellg();

int fd = open(partition_config.graph_filename.c_str(), O_RDONLY);
void* mmap_addr = MAP_FAILED;
size_t file_size = 0;
size_t offset = 0;
if (fd != -1) {
struct stat st;
if (fstat(fd, &st) == 0 && st.st_size > 0 && cur >= 0) {
file_size = (size_t)st.st_size;
offset = (size_t)cur;
if (offset <= file_size) {
#ifdef MAP_POPULATE
mmap_addr = mmap(nullptr, file_size, PROT_READ, MAP_PRIVATE | MAP_POPULATE, fd, 0);
#else
mmap_addr = mmap(nullptr, file_size, PROT_READ, MAP_PRIVATE, fd, 0);
#endif
}
}
close(fd);
}

const char* p;
const char* pend;
size_t actual;
std::vector<char> buf; // fallback only
if (mmap_addr != MAP_FAILED) {
p = (const char*)mmap_addr + offset;
pend = (const char*)mmap_addr + file_size;
actual = file_size - offset;
} else {
in.seekg(0, std::ios::end);
size_t remaining = (size_t)in.tellg() - offset;
in.seekg(cur);
buf.resize(remaining);
in.read(buf.data(), remaining);
actual = (size_t)in.gcount();
p = buf.data();
pend = p + actual;
Comment on lines +233 to +240

Copilot AI Apr 2, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fallback path casts tellg() results to size_t without validating they’re non-negative/valid. If tellg() returns -1 (or offset > end), this underflows and can attempt to allocate a huge buffer. Add stream state checks and guard the remaining-size computation before resizing the buffer.

Suggested change
in.seekg(0, std::ios::end);
size_t remaining = (size_t)in.tellg() - offset;
in.seekg(cur);
buf.resize(remaining);
in.read(buf.data(), remaining);
actual = (size_t)in.gcount();
p = buf.data();
pend = p + actual;
// Fallback to reading from stream: guard tellg() and size computations
in.clear();
in.seekg(0, std::ios::end);
std::streampos end_pos = in.tellg();
if (!in || end_pos < 0 || static_cast<size_t>(end_pos) < offset) {
// Invalid stream position or offset beyond end: treat as no remaining data
in.clear();
in.seekg(cur);
actual = 0;
buf.clear();
p = buf.data();
pend = p;
} else {
size_t remaining = static_cast<size_t>(end_pos) - offset;
in.clear();
in.seekg(cur);
buf.resize(remaining);
in.read(buf.data(), remaining);
actual = static_cast<size_t>(in.gcount());
p = buf.data();
pend = p + actual;
}

Copilot uses AI. Check for mistakes.
}

size_t avg_entries = (num_lines > 0) ? std::max((size_t)4, actual / (num_lines * 4)) : 4;

LongNodeID node_counter = 0;
while (node_counter < num_lines && p < pend) {
if (*p == '%') {
while (p < pend && *p != '\n') p++;
if (p < pend) p++;
continue;
}
auto& vec = (*input)[node_counter];
vec.reserve(avg_entries);
while (p < pend && *p != '\n' && *p != '\r') {
while (p < pend && *p != '\n' && *p != '\r' && (*p < '0' || *p > '9')) p++;
if (p >= pend || *p == '\n' || *p == '\r') break;
LongNodeID val = 0;
while (p < pend && *p >= '0' && *p <= '9') {
val = val * 10 + (LongNodeID)(*p - '0');
p++;
}
vec.push_back(val);
}
if (p < pend && *p == '\r') p++;
if (p < pend && *p == '\n') p++;
node_counter++;
}
if (mmap_addr != MAP_FAILED) {
munmap(mmap_addr, file_size);
}
} else {
// Original streaming approach
std::vector<std::string>* lines;
lines = new std::vector<std::string>(1);
LongNodeID node_counter = 0;
buffered_input *ss2 = NULL;
while( node_counter < num_lines) {
std::getline(*(partition_config.stream_in),(*lines)[0]);
if ((*lines)[0][0] == '%') {
continue;
}
ss2 = new buffered_input(lines);
ss2->simple_scan_line((*input)[node_counter++]);
(*lines)[0].clear(); delete ss2;
}
ss2 = new buffered_input(lines);
ss2->simple_scan_line((*input)[node_counter++]);
(*lines)[0].clear(); delete ss2;
delete lines;
}
delete lines;
return input;
}

Expand Down
19 changes: 16 additions & 3 deletions code_for_graphs/lib/partition/onepass_partitioning/vertex_partitioning.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,13 +174,22 @@ inline PartitionID vertex_partitioning::solve_node(LongNodeID curr_node_id, Node
switch(sampling) {
case SAMPLING_INACTIVE_LINEAR_COMPLEXITY:
decision = solve_linear_complexity(curr_node_id, curr_node_weight, my_thread);
// Duplicate call is intentional: #pragma omp critical cannot be conditionally skipped
if (n_threads > 1) {
#pragma omp critical(update_self_sorting_vector)
this->sorted_blocks.increment(decision);
this->sorted_blocks.increment(decision);
} else {
this->sorted_blocks.increment(decision);
Comment on lines 175 to +182

Copilot AI Apr 2, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The added comment is inaccurate/misleading: the call to sorted_blocks.increment(decision) is not duplicated, and an OpenMP critical section can be conditionally executed as written. Please update/remove this comment to reflect the actual reason for the branching (skipping critical overhead when n_threads == 1).

Copilot uses AI. Check for mistakes.
}
break;
case SAMPLING_NEIGHBORS_LINEAR_COMPLEXITY:
decision = solve_sampl_neighb_linear_complex(curr_node_id, curr_node_weight, my_thread);
if (n_threads > 1) {
#pragma omp critical(update_self_sorting_vector)
this->sorted_blocks.increment(decision);
this->sorted_blocks.increment(decision);
} else {
this->sorted_blocks.increment(decision);
}
break;
case SAMPLING_INACTIVE:
decision = solve(curr_node_id, curr_node_weight, my_thread);
Expand Down Expand Up @@ -429,8 +438,12 @@ inline void vertex_partitioning::reset_sorted_blocks() {
}

inline void vertex_partitioning::decrement_sorted_block(PartitionID block) {
if (n_threads > 1) {
#pragma omp critical(update_self_sorting_vector)
this->sorted_blocks.decrement(block);
this->sorted_blocks.decrement(block);
} else {
this->sorted_blocks.decrement(block);
}
}

inline void vertex_partitioning::set_sampling_threashold(float sampling_threashold) {
Expand Down
20 changes: 19 additions & 1 deletion code_for_hypergraphs/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,24 @@ if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)
set(CMAKE_BUILD_TYPE "Release")
endif()

# Profile-Guided Optimization (PGO) support
# Usage: cmake -DPGO_GENERATE=ON .. && make && <run training workload> && cmake -DPGO_USE=ON .. && make
option(PGO_GENERATE "Build with PGO instrumentation (phase 1)" OFF)
option(PGO_USE "Build with PGO optimization from collected profiles (phase 2)" OFF)
if(PGO_GENERATE AND PGO_USE)
message(FATAL_ERROR "PGO_GENERATE and PGO_USE are mutually exclusive. Use PGO_GENERATE first, run a training workload, then rebuild with PGO_USE.")
endif()
if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
if(PGO_GENERATE)
add_definitions(-fprofile-generate=${CMAKE_BINARY_DIR}/pgo_data)
set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fprofile-generate=${CMAKE_BINARY_DIR}/pgo_data")
endif()
if(PGO_USE)
add_definitions(-fprofile-use=${CMAKE_BINARY_DIR}/pgo_data -fprofile-correction)
set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -fprofile-use=${CMAKE_BINARY_DIR}/pgo_data")
endif()
Comment on lines +29 to +42

Copilot AI Apr 2, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PGO_GENERATE and PGO_USE can currently both be enabled at the same time, which will add conflicting -fprofile-generate / -fprofile-use flags and can lead to compiler/linker errors or unpredictable behavior. Consider making these options mutually exclusive (e.g., error out when both are ON, or have PGO_USE override PGO_GENERATE).

Copilot uses AI. Check for mistakes.
endif()

# tweak compiler flags
CHECK_CXX_COMPILER_FLAG(-funroll-loops COMPILER_SUPPORTS_FUNROLL_LOOPS)
if(COMPILER_SUPPORTS_FUNROLL_LOOPS)
Expand Down Expand Up @@ -86,7 +104,7 @@ include_directories(${CMAKE_CURRENT_SOURCE_DIR}/lib/tools)



set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-builtin-malloc -fno-builtin-calloc -fno-builtin-realloc")
# Removed -fno-builtin-malloc flags to let compiler optimize allocation patterns

set(LIBKAFFPA_SOURCE_FILES
lib/tools/quality_metrics.cpp
Expand Down
2 changes: 2 additions & 0 deletions code_for_hypergraphs/app/configuration.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -616,6 +616,8 @@ inline void configuration::standard( PartitionConfig & partition_config ) {
partition_config.problem = PROBLEM_DEGREE;

partition_config.enable_mapping = false;
partition_config.evaluate_cut = true;
partition_config.evaluate_connectivity = true;
partition_config.ls_neighborhood = COMMUNICATIONGRAPH;
partition_config.communication_neighborhood_dist = 10;
partition_config.construction_algorithm = MAP_CONST_FASTHIERARCHY_TOPDOWN;
Expand Down
Loading
Loading