File: host_function_worker_thread_pool.cpp

package info (click to toggle)
intel-compute-runtime 26.05.37020.3-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 83,596 kB
  • sloc: cpp: 976,037; lisp: 2,096; sh: 704; makefile: 162
file content (89 lines) | stat: -rw-r--r-- 2,102 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
/*
 * Copyright (C) 2025 Intel Corporation
 *
 * SPDX-License-Identifier: MIT
 *
 */

#include "shared/source/command_stream/host_function_worker_thread_pool.h"

#include "shared/source/command_stream/host_function.h"

namespace NEO {

HostFunctionThreadPool::HostFunctionThreadPool(int32_t threadsInThreadPoolLimit) {

    if (threadsInThreadPoolLimit == HostFunctionThreadPoolHelper::unlimitedThreads) {
        unlimitedThreads = true;
    } else {
        threadsLimit = static_cast<uint32_t>(threadsInThreadPoolLimit);
    }
}

HostFunctionThreadPool::~HostFunctionThreadPool() = default;

void HostFunctionThreadPool::registerThread() noexcept {

    if ((threads.size() < threadsLimit) || unlimitedThreads) {
        threads.emplace_back(([this](std::stop_token st) {
            this->workerLoop(st);
        }));
    }
}

void HostFunctionThreadPool::shutdown() noexcept {

    for (auto &thread : threads) {
        thread.request_stop();
    }

    semaphore.release(static_cast<ptrdiff_t>(threads.size()));

    for (auto &thread : threads) {
        thread.join();
    }

    {
        std::lock_guard lock{this->hostFunctionsMutex};
        hostFunctions.clear();
    }

    threads.clear();
}

void HostFunctionThreadPool::registerHostFunctionToExecute(HostFunctionStreamer *streamer, HostFunction &&hostFunction) {

    {
        std::unique_lock lock{this->hostFunctionsMutex};
        hostFunctions.emplace_back(streamer, std::move(hostFunction));
    }
    semaphore.release();
}

void NEO::HostFunctionThreadPool::workerLoop(std::stop_token st) noexcept {

    while (st.stop_requested() == false) {

        semaphore.acquire();

        if (st.stop_requested()) {
            return;
        }

        executeHostFunction();
    }
}

void HostFunctionThreadPool::executeHostFunction() noexcept {

    std::unique_lock lock{this->hostFunctionsMutex};
    auto [streamer, hostFunction] = std::move(hostFunctions.front());
    hostFunctions.pop_front();
    lock.unlock();

    hostFunction.invoke();

    streamer->signalHostFunctionCompletion(hostFunction);
}

} // namespace NEO