Skip to content

feature: switch threader_for to int64 indices #3471

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

Draft
Alexandr-Solovev wants to merge 13 commits into
base: main
Choose a base branch
from
Draft

feature: switch threader_for to int64 indices #3471

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
a4c06de
init commit for replace threader_for
Alexandr-Solovev Jan 7, 2026
4ba268e
fix abi
Alexandr-Solovev Jan 29, 2026
3a569e7
Merge branch 'main' into dev/asolovev_threader_for_upd
Alexandr-Solovev Jan 29, 2026
c0f983e
minor fix
Alexandr-Solovev Jan 30, 2026
2a313dc
Merge branch 'dev/asolovev_threader_for_upd' of https://github.com/Al…
Alexandr-Solovev Jan 30, 2026
6ace7c3
Merge branch 'main' into dev/asolovev_threader_for_upd
Alexandr-Solovev Feb 5, 2026
aaf4c6a
fixes for int64
Alexandr-Solovev Feb 6, 2026
ea5cbfe
init refactoring wip
Alexandr-Solovev Feb 9, 2026
e9048ac
Merge branch 'main' into dev/asolovev_threader_for_upd
Alexandr-Solovev Feb 16, 2026
31b9cf0
fixes
Alexandr-Solovev Feb 16, 2026
7676940
Merge branch 'main' into dev/asolovev_threader_for_upd
Alexandr-Solovev Feb 16, 2026
0ec337a
minor fixes
Alexandr-Solovev Feb 16, 2026
bffdf4e
fixes
Alexandr-Solovev Feb 20, 2026
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
9 changes: 9 additions & 0 deletions .github/.abignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,15 @@
; limitations under the License.
;

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;;; DAAL THREADING LAYER DESELECTIONS ;;;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
; deselect threading layer functions

[suppress_function]
symbol_name = _daal_threader_for_int64
drop = yes

;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;;; DAAL DESELECTIONS ;;;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;
Expand Down
12 changes: 2 additions & 10 deletions cpp/daal/src/externals/core_threading_win_dll.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,7 @@ FARPROC load_daal_thr_func(const char * ordinal)
typedef void * (*_threaded_malloc_t)(const size_t, const size_t);
typedef void (*_threaded_free_t)(void *);

typedef void (*_daal_threader_for_t)(int, int, const void *, daal::functype);
typedef void (*_daal_threader_for_int64_t)(int64_t, const void *, daal::functype_int64);
typedef void (*_daal_threader_for_t)(int64_t, int64_t, const void *, daal::functype);
typedef void (*_daal_threader_for_int32ptr_t)(const int *, const int *, const void *, daal::functype_int32ptr);
typedef void (*_daal_threader_for_simple_t)(int, int, const void *, daal::functype);
typedef void (*_daal_static_threader_for_t)(size_t, const void *, daal::functype_static);
Expand Down Expand Up @@ -179,7 +178,7 @@ DAAL_EXPORT void _threaded_scalable_free(void * ptr)
_threaded_free_ptr(ptr);
}

DAAL_EXPORT void _daal_threader_for(int n, int threads_request, const void * a, daal::functype func)
DAAL_EXPORT void _daal_threader_for(int64_t n, int64_t threads_request, const void * a, daal::functype func)
{
load_daal_thr_dll();
static _daal_threader_for_t _daal_threader_for_ptr = (_daal_threader_for_t)load_daal_thr_func("_daal_threader_for");
Expand All @@ -202,13 +201,6 @@ DAAL_EXPORT void _daal_threader_for_int32ptr(const int * begin, const int * end,
_daal_threader_for_int32ptr_ptr(begin, end, a, func);
}

DAAL_EXPORT void _daal_threader_for_int64(int64_t n, const void * a, daal::functype_int64 func)
{
load_daal_thr_dll();
static _daal_threader_for_int64_t _daal_threader_for_int64_ptr = (_daal_threader_for_int64_t)load_daal_thr_func("_daal_threader_for_int64");
_daal_threader_for_int64_ptr(n, a, func);
}

DAAL_EXPORT void _daal_static_threader_for(size_t n, const void * a, daal::functype_static func)
{
load_daal_thr_dll();
Expand Down
24 changes: 1 addition & 23 deletions cpp/daal/src/threading/threading.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,29 +117,7 @@ DAAL_EXPORT size_t _setNumberOfThreads(const size_t numThreads, void ** globalCo
return 1;
}

DAAL_EXPORT void _daal_threader_for(int n, int reserved, const void * a, daal::functype func)
{
if (daal::threader_env()->getNumberOfThreads() > 1)
{
tbb::parallel_for(tbb::blocked_range<int>(0, n, 1), [&](tbb::blocked_range<int> r) {
int i;
for (i = r.begin(); i < r.end(); i++)
{
func(i, a);
}
});
}
else
{
int i;
for (i = 0; i < n; i++)
{
func(i, a);
}
}
}

DAAL_EXPORT void _daal_threader_for_int64(int64_t n, const void * a, daal::functype_int64 func)
DAAL_EXPORT void _daal_threader_for(int64_t n, int64_t reserved, const void * a, daal::functype func)
{
if (daal::threader_env()->getNumberOfThreads() > 1)
{
Expand Down
26 changes: 2 additions & 24 deletions cpp/daal/src/threading/threading.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,10 +87,9 @@ extern "C"
{
DAAL_EXPORT int _daal_threader_get_max_threads();
DAAL_EXPORT int _daal_threader_get_current_thread_index();
DAAL_EXPORT void _daal_threader_for(int n, int threads_request, const void * a, daal::functype func);
DAAL_EXPORT void _daal_threader_for(int64_t n, int64_t threads_request, const void * a, daal::functype func);
DAAL_EXPORT void _daal_threader_reduce(const size_t n, const size_t grainSize, daal::Reducer & reducer);
DAAL_EXPORT void _daal_static_threader_reduce(const size_t n, const size_t grainSize, daal::Reducer & reducer);
DAAL_EXPORT void _daal_threader_for_int64(int64_t n, const void * a, daal::functype_int64 func);
DAAL_EXPORT void _daal_threader_for_simple(int n, int threads_request, const void * a, daal::functype func);
DAAL_EXPORT void _daal_threader_for_int32ptr(const int * begin, const int * end, const void * a, daal::functype_int32ptr func);
DAAL_EXPORT void _daal_static_threader_for(size_t n, const void * a, daal::functype_static func);
Expand Down Expand Up @@ -266,34 +265,13 @@ inline void threader_func_break(int i, bool & needBreak, const void * a)
/// @param[in] reserved Parameter reserved for the future. Currently unused.
Copy link

Copilot AI Feb 5, 2026

Choose a reason for hiding this comment

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

The parameter name 'reserved' in the documentation no longer accurately describes the second parameter, which is now 'threads_request' with a specific purpose. The documentation should be updated to reflect that this parameter controls the number of threads requested for parallel execution.

Copilot generated this review using guidance from repository custom instructions.
Copy link
Copy Markdown
Contributor

@david-cortes-intel david-cortes-intel Feb 5, 2026

Choose a reason for hiding this comment

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

There is a mismatch between the headers and the C++ files regarding the names of the parameters.

Copy link
Copy Markdown
Contributor Author

@Alexandr-Solovev Alexandr-Solovev Feb 6, 2026

Choose a reason for hiding this comment

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

Actually, it's a common issue for all threader functions. WIll be update all of the functions

/// @param[in] func Callable object that defines the loop body.
template <typename F>
inline void threader_for(int n, int reserved, const F & func)
inline void threader_for(int64_t n, int64_t reserved, const F & func)
{
const void * a = static_cast<const void *>(&func);

_daal_threader_for(n, reserved, a, threader_func<F>);
}

/// Pass a function to be executed in a for loop to the threading layer.
/// The maximal number of iterations in the loop is `2^63 - 1 (INT64_MAX)`.
/// The default scheduling of the threading layer is used to assign
/// the iterations of the loop to threads.
/// The iterations of the loop should be logically independent.
/// Data dependencies between the iterations are allowed, but may requre the use
/// of synchronization primitives.
///
/// @tparam F Callable object of type `[/* captures */](int64_t i) -> void`,
/// where `i` is the loop's iteration index, `0 <= i < n`.
///
/// @param[in] n Number of iterations in the for loop.
/// @param[in] func Callable object that defines the loop body.
template <typename F>
inline void threader_for_int64(int64_t n, const F & func)
{
const void * a = static_cast<const void *>(&func);

_daal_threader_for_int64(n, a, threader_func<F>);
}

/// Pass a function to be executed in a for loop to the threading layer.
/// The maximal number of iterations in the loop is 2^31 - 1.
Copy link

Copilot AI Feb 5, 2026

Choose a reason for hiding this comment

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

The documentation comment is now incorrect - it states the maximum is 2^31 - 1, but the function now accepts int64_t which allows up to 2^63 - 1 iterations. This comment should be updated to reflect the new 64-bit limit.

Copilot generated this review using guidance from repository custom instructions.
Copy link
Copy Markdown
Contributor

@david-cortes-intel david-cortes-intel Feb 5, 2026

Choose a reason for hiding this comment

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

Bot is right here.

///
Expand Down
2 changes: 1 addition & 1 deletion cpp/oneapi/dal/algo/basic_statistics/backend/cpu/apply_weights_cpu.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ void apply_weights(const pr::ndview<Float, 1>& weights, pr::ndview<Float, 2>& sa
const bk::uniform_blocking blocking(r_count, threading_block);
const auto block_count = blocking.get_block_count();

de::threader_for_int64(block_count, [&](std::int64_t b) -> void {
de::threader_for(block_count, block_count, [&](std::int64_t b) -> void {
const auto f_row = blocking.get_block_start_index(b);
const auto l_row = blocking.get_block_end_index(b);

Expand Down
10 changes: 2 additions & 8 deletions cpp/oneapi/dal/backend/interop/threading.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,19 +25,13 @@ ONEDAL_EXPORT int _onedal_threader_get_current_thread_index() {
return _daal_threader_get_current_thread_index();
}

ONEDAL_EXPORT void _onedal_threader_for(std::int32_t n,
std::int32_t threads_request,
ONEDAL_EXPORT void _onedal_threader_for(std::int64_t n,
std::int64_t threads_request,
const void *a,
oneapi::dal::preview::functype func) {
_daal_threader_for(n, threads_request, a, static_cast<daal::functype>(func));
}

ONEDAL_EXPORT void _onedal_threader_for_int64(std::int64_t n,
const void *a,
oneapi::dal::preview::functype_int64 func) {
_daal_threader_for_int64(n, a, static_cast<daal::functype_int64>(func));
}

ONEDAL_EXPORT void _onedal_threader_for_simple(std::int32_t n,
std::int32_t threads_request,
const void *a,
Expand Down
19 changes: 4 additions & 15 deletions cpp/oneapi/dal/detail/threading.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,15 +55,11 @@ ONEDAL_EXPORT int _onedal_threader_get_max_threads();

ONEDAL_EXPORT int _onedal_threader_get_current_thread_index();

ONEDAL_EXPORT void _onedal_threader_for(std::int32_t n,
std::int32_t threads_request,
ONEDAL_EXPORT void _onedal_threader_for(std::int64_t n,
std::int64_t threads_request,
const void *a,
oneapi::dal::preview::functype func);

ONEDAL_EXPORT void _onedal_threader_for_int64(std::int64_t n,
const void *a,
oneapi::dal::preview::functype_int64 func);

ONEDAL_EXPORT void _onedal_threader_for_simple(std::int32_t n,
std::int32_t threads_request,
const void *a,
Expand Down Expand Up @@ -155,21 +151,14 @@ inline void threader_func_blocked_size(std::size_t f, std::size_t l, const void
}

template <typename F>
inline ONEDAL_EXPORT void threader_for(std::int32_t n,
std::int32_t threads_request,
inline ONEDAL_EXPORT void threader_for(std::int64_t n,
std::int64_t threads_request,
const F &lambda) {
const void *a = static_cast<const void *>(&lambda);

_onedal_threader_for(n, threads_request, a, threader_func<F>);
}

template <typename F>
inline ONEDAL_EXPORT void threader_for_int64(std::int64_t n, const F &lambda) {
const void *a = static_cast<const void *>(&lambda);

_onedal_threader_for_int64(n, a, threader_func_int64<F>);
}

template <typename F>
inline ONEDAL_EXPORT void threader_for_simple(std::int32_t n,
std::int32_t threads_request,
Expand Down
26 changes: 13 additions & 13 deletions cpp/oneapi/dal/io/csv/detail/read_graph_kernel_impl.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,7 @@ template <typename Graph>
struct collect_degrees_from_edge_list<Graph, /* IsDirected = */ false> {
template <typename EdgeList, typename AtomicType>
auto operator()(const EdgeList &edges, AtomicType *degrees_cv) {
dal::detail::threader_for_int64(edges.size(), [&](std::int64_t u) {
dal::detail::threader_for(edges.size(), edges.size(), [&](std::int64_t u) {
++degrees_cv[std::get<0>(edges[u])];
++degrees_cv[std::get<1>(edges[u])];
});
Expand All @@ -168,7 +168,7 @@ template <typename Graph>
struct collect_degrees_from_edge_list<Graph, /* IsDirected = */ true> {
template <typename EdgeList, typename AtomicType>
auto operator()(const EdgeList &edges, AtomicType *degrees_cv) {
dal::detail::threader_for_int64(edges.size(), [&](std::int64_t u) {
dal::detail::threader_for(edges.size(), edges.size(), [&](std::int64_t u) {
++degrees_cv[std::get<0>(edges[u])];
});
}
Expand Down Expand Up @@ -227,7 +227,7 @@ ONEDAL_EXPORT std::int64_t compute_prefix_sum<std::int64_t, std::int32_t>(

template <typename Index, typename AtomicIndex>
void fill_from_atomics(Index *arr, AtomicIndex *atomic_arr, std::int64_t elements_count) {
dal::detail::threader_for_int64(elements_count, [&](std::int64_t n) {
dal::detail::threader_for(elements_count, elements_count, [&](std::int64_t n) {
arr[n] = atomic_arr[n].load();
});
}
Expand All @@ -241,7 +241,7 @@ struct fill_unfiltered_neighs<Graph, /* IsDirected = */ false> {
auto operator()(const edge_list<Vertex> &edges,
AtomicEdge *rows_vec_atomic,
Vertex *unfiltered_neighs) {
dal::detail::threader_for_int64(edges.size(), [&](std::int64_t u) {
dal::detail::threader_for(edges.size(), edges.size(), [&](std::int64_t u) {
unfiltered_neighs[++rows_vec_atomic[edges[u].first] - 1] = edges[u].second;
unfiltered_neighs[++rows_vec_atomic[edges[u].second] - 1] = edges[u].first;
});
Expand All @@ -251,7 +251,7 @@ struct fill_unfiltered_neighs<Graph, /* IsDirected = */ false> {
auto operator()(const weighted_edge_list<Vertex, Weight> &edges,
AtomicEdge *rows_vec_atomic,
std::pair<Vertex, Weight> *unfiltered_neighs_vals) {
dal::detail::threader_for_int64(edges.size(), [&](std::int64_t u) {
dal::detail::threader_for(edges.size(), edges.size(), [&](std::int64_t u) {
unfiltered_neighs_vals[++rows_vec_atomic[std::get<0>(edges[u])] - 1] =
std::make_pair(std::get<1>(edges[u]), std::get<2>(edges[u]));
unfiltered_neighs_vals[++rows_vec_atomic[std::get<1>(edges[u])] - 1] =
Expand All @@ -266,7 +266,7 @@ struct fill_unfiltered_neighs<Graph, /* IsDirected = */ true> {
auto operator()(const edge_list<Vertex> &edges,
AtomicEdge *rows_vec_atomic,
Vertex *unfiltered_neighs) {
dal::detail::threader_for_int64(edges.size(), [&](std::int64_t u) {
dal::detail::threader_for(edges.size(), edges.size(), [&](std::int64_t u) {
unfiltered_neighs[++rows_vec_atomic[edges[u].first] - 1] = edges[u].second;
});
}
Expand All @@ -275,7 +275,7 @@ struct fill_unfiltered_neighs<Graph, /* IsDirected = */ true> {
auto operator()(const weighted_edge_list<Vertex, Weight> &edges,
AtomicEdge *rows_vec_atomic,
std::pair<Vertex, Weight> *unfiltered_neighs_vals) {
dal::detail::threader_for_int64(edges.size(), [&](std::int64_t u) {
dal::detail::threader_for(edges.size(), edges.size(), [&](std::int64_t u) {
unfiltered_neighs_vals[++rows_vec_atomic[std::get<0>(edges[u])] - 1] =
std::make_pair(std::get<1>(edges[u]), std::get<2>(edges[u]));
});
Expand All @@ -289,7 +289,7 @@ void fill_filtered_neighs(const EdgeIndex *unfiltered_offsets,
const EdgeIndex *filtered_offsets,
VertexIndex *filtered_neighs,
std::int64_t vertex_count) {
dal::detail::threader_for_int64(vertex_count, [&](std::int64_t u) {
dal::detail::threader_for(vertex_count, vertex_count, [&](std::int64_t u) {
auto u_neighs = filtered_neighs + filtered_offsets[u];
auto u_neighs_unf = unfiltered_neighs + unfiltered_offsets[u];
for (VertexIndex i = 0; i < filtered_degrees[u]; i++) {
Expand All @@ -306,7 +306,7 @@ void fill_filtered_neighs(const Edge *unfiltered_offsets,
Vertex *filtered_neighs,
Weight *filtered_vals,
std::int64_t vertex_count) {
dal::detail::threader_for_int64(vertex_count, [&](std::int64_t u) {
dal::detail::threader_for(vertex_count, vertex_count, [&](std::int64_t u) {
auto u_neighs = filtered_neighs + filtered_offsets[u];
auto u_neighs_vals = filtered_vals + filtered_offsets[u];
auto u_neighs_unf = unfiltered_neighs_vals + unfiltered_offsets[u];
Expand All @@ -332,7 +332,7 @@ void filter_neighbors_and_fill_new_degrees(VertexIndex *unfiltered_neighs,
VertexIndex *new_degrees,
std::int64_t vertex_count) {
//removing self-loops, multiple edges from graph, and make neighbors in CSR sorted
dal::detail::threader_for_int64(vertex_count, [&](std::int64_t u) {
dal::detail::threader_for(vertex_count, vertex_count, [&](std::int64_t u) {
auto start_p = unfiltered_neighs + unfiltered_offsets[u];
auto end_p = unfiltered_neighs + unfiltered_offsets[u + 1];

Expand All @@ -350,7 +350,7 @@ void filter_neighbors_and_fill_new_degrees(std::pair<Vertex, Weight> *unfiltered
Vertex *new_degrees,
std::int64_t vertex_count) {
//removing self-loops, multiple edges from graph, and make neighbors in CSR sorted
dal::detail::threader_for_int64(vertex_count, [&](std::int64_t u) {
dal::detail::threader_for(vertex_count, vertex_count, [&](std::int64_t u) {
auto start_p = unfiltered_neighs_vals + unfiltered_offsets[u];
auto end_p = unfiltered_neighs_vals + unfiltered_offsets[u + 1];

Expand Down Expand Up @@ -478,7 +478,7 @@ void convert_to_csr_impl(const edge_list<typename graph_traits<Graph>::vertex_ty
auto edge_offsets_tup =
ra_vertex_edge.template allocate_array<vertex_edge_set_t>(vertex_count + 1);
auto rows_vertex = std::get<1>(edge_offsets_tup);
dal::detail::threader_for_int64(vertex_count + 1, [&](std::int64_t u) {
dal::detail::threader_for(vertex_count + 1, vertex_count + 1, [&](std::int64_t u) {
rows_vertex[u] = static_cast<vertex_edge_t>(edge_offsets_data[u]);
});

Expand Down Expand Up @@ -601,7 +601,7 @@ void convert_to_csr_impl(
auto edge_offsets_tup =
ra_vertex_edge.template allocate_array<vertex_edge_set_t>(vertex_count + 1);
auto rows_vertex = std::get<1>(edge_offsets_tup);
dal::detail::threader_for_int64(vertex_count + 1, [&](std::int64_t u) {
dal::detail::threader_for(vertex_count + 1, vertex_count + 1, [&](std::int64_t u) {
rows_vertex[u] = static_cast<vertex_edge_t>(edge_offsets_data[u]);
});

Expand Down
2 changes: 1 addition & 1 deletion cpp/oneapi/dal/table/backend/convert/copy_convert_impl_cpu.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,7 @@ void copy_convert(const detail::host_policy& policy,
const std::int64_t row_count = shape.first;
const std::int64_t col_count = shape.second;

detail::threader_for_int64(row_count, [&](std::int64_t i) -> void {
detail::threader_for(row_count, row_count, [&](std::int64_t i) -> void {
Copy link

Copilot AI Feb 5, 2026

Choose a reason for hiding this comment

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

The second parameter appears redundant - passing row_count twice suggests the API might benefit from clearer parameter naming or documentation. Consider whether threads_request should match the iteration count or if this represents a threading hint that should be calculated differently.

Suggested change
detail::threader_for(row_count, row_count, [&](std::int64_t i) -> void {
const std::int64_t threads_request = detail::threader_get_max_threads();
detail::threader_for(row_count, threads_request, [&](std::int64_t i) -> void {

Copilot uses AI. Check for mistakes.
auto* out_raw_ptr = out_ptrs[i];
const auto* inp_raw_ptr = inp_ptrs[i];

Expand Down
Loading