2 +----------------------------------------------------------------------+
4 +----------------------------------------------------------------------+
5 | Copyright (c) 2010-present Facebook, Inc. (http://www.facebook.com) |
6 +----------------------------------------------------------------------+
7 | This source file is subject to version 3.01 of the PHP license, |
8 | that is bundled with this package in the file LICENSE, and is |
9 | available through the world-wide-web at the following url: |
10 | http://www.php.net/license/3_01.txt |
11 | If you did not receive a copy of the PHP license and are unable to |
12 | obtain it through the world-wide-web, please send a note to |
13 | license@php.net so we can mail you a copy immediately. |
14 +----------------------------------------------------------------------+
23 #include <type_traits>
29 #include <folly/ScopeGuard.h>
31 #include "hphp/runtime/vm/treadmill.h"
32 #include "hphp/runtime/base/program-functions.h"
34 namespace HPHP::HHBBC
{
38 //////////////////////////////////////////////////////////////////////
41 * Before using the parallel module, you can configure these to change
42 * how much parallelism is used.
44 extern size_t num_threads
;
45 extern size_t final_threads
;
47 //////////////////////////////////////////////////////////////////////
52 auto size_info(Items
&& items
) {
53 auto const size
= items
.size();
54 if (!size
) return std::make_tuple(size
, size
);
55 auto const threads
= std::min(num_threads
, size
);
56 return std::make_tuple(size
, threads
);
59 template<class Func
, class Item
>
60 auto caller(const Func
& func
, Item
&& item
, size_t worker
) ->
61 decltype(func(std::forward
<Item
>(item
), worker
)) {
62 return func(std::forward
<Item
>(item
), worker
);
65 template<class Func
, class Item
>
66 auto caller(const Func
& func
, Item
&& item
, size_t worker
) ->
67 decltype(func(std::forward
<Item
>(item
))) {
68 return func(std::forward
<Item
>(item
));
74 * Call a function on each element of `inputs', in parallel.
76 * If `func' throws an exception, some of the work will not be
79 template<class Func
, class Items
>
80 void for_each(Items
&& inputs
, Func func
) {
81 std::atomic
<bool> failed
{false};
82 std::atomic
<size_t> index
{0};
83 auto const info
= detail::size_info(inputs
);
84 auto const size
= std::get
<0>(info
);
86 auto const threads
= std::get
<1>(info
);
88 std::vector
<std::thread
> workers
;
89 for (auto worker
= size_t{0}; worker
< threads
; ++worker
) {
90 workers
.push_back(std::thread([&, worker
] {
92 HphpSessionAndThread _
{Treadmill::SessionKind::HHBBC
};
94 auto const i
= index
++;
98 std::forward
<Items
>(inputs
)[i
],
102 } catch (const std::exception
& e
) {
104 "worker thread exited with exception: %s\n", e
.what());
110 for (auto& t
: workers
) t
.join();
112 if (failed
) throw std::runtime_error("parallel::for_each failed");
115 //////////////////////////////////////////////////////////////////////
118 * Call a function that produces a return value for each element of
119 * `inputs' in parallel, and collect the results.
121 * Requires: the type returned from the function call must be
122 * DefaultConstructible, and either MoveAssignable or Assignable.
124 * If `func' throws an exception, the results of the output vector
125 * will contain some default-constructed values.
127 template<class Func
, class Items
>
128 auto map(Items
&& inputs
, Func func
) -> std::vector
<decltype(func(inputs
[0]))> {
129 auto const info
= detail::size_info(inputs
);
130 auto const size
= std::get
<0>(info
);
131 std::vector
<decltype(func(inputs
[0]))> retVec(size
);
132 if (!size
) return retVec
;
133 auto const threads
= std::get
<1>(info
);
135 auto const retMem
= &retVec
[0];
137 std::atomic
<bool> failed
{false};
138 std::atomic
<size_t> index
{0};
140 std::vector
<std::thread
> workers
;
141 for (auto worker
= size_t{0}; worker
< threads
; ++worker
) {
142 workers
.push_back(std::thread([&] {
144 HphpSessionAndThread _
{Treadmill::SessionKind::HHBBC
};
146 auto const i
= index
++;
147 if (i
>= size
) break;
148 retMem
[i
] = func(std::forward
<Items
>(inputs
)[i
]);
150 } catch (const std::runtime_error
& e
) {
152 "worker thread exited with exception: %s\n", e
.what());
158 for (auto& t
: workers
) t
.join();
159 if (failed
) throw std::runtime_error("parallel::map failed");
164 //////////////////////////////////////////////////////////////////////
167 * Call a function that produces a return value for each integer from
168 * 0 to count-1 in parallel, and collect the results.
170 * Requires: the type returned from the function call must be
171 * DefaultConstructible, and either MoveAssignable or Assignable.
173 * If `func' throws an exception, the results of the output vector
174 * will contain some default-constructed values.
176 template<typename Func
>
177 auto gen(size_t count
, Func func
) -> std::vector
<decltype(func(0))> {
178 std::vector
<decltype(func(0))> retVec(count
);
179 if (!count
) return retVec
;
181 auto const threads
= std::min(num_threads
, count
);
183 std::atomic
<bool> failed
{false};
184 std::atomic
<size_t> index
{0};
186 std::vector
<std::thread
> workers
;
187 for (auto worker
= size_t{0}; worker
< threads
; ++worker
) {
188 workers
.emplace_back(
191 HphpSessionAndThread _
{Treadmill::SessionKind::HHBBC
};
193 auto const i
= index
++;
194 if (i
>= count
) break;
197 } catch (const std::runtime_error
& e
) {
199 stderr
, "worker thread exited with exception: %s\n", e
.what()
207 for (auto& t
: workers
) t
.join();
208 if (failed
) throw std::runtime_error("parallel::gen failed");
213 //////////////////////////////////////////////////////////////////////
217 template<size_t Total
>
218 void populateThreads(std::array
<std::thread
, Total
>&,
222 template<size_t Total
, typename Func
, typename
... Funcs
>
223 void populateThreads(std::array
<std::thread
, Total
>& threads
,
224 std::atomic
<bool>& failed
,
228 threads
[idx
] = std::thread(
231 HphpSessionAndThread _
{Treadmill::SessionKind::HHBBC
};
233 } catch (const std::runtime_error
& e
) {
235 stderr
, "worker thread exited with exception: %s\n", e
.what()
241 populateThreads(threads
, failed
, idx
+1, std::forward
<Funcs
>(funcs
)...);
246 // Run N callables in parallel
247 template<typename
... Funcs
>
248 void parallel(Funcs
&&... funcs
) {
249 std::array
<std::thread
, sizeof...(Funcs
)> threads
;
250 std::atomic
<bool> failed
{false};
251 detail::populateThreads(threads
, failed
, 0, std::forward
<Funcs
>(funcs
)...);
252 for (auto& t
: threads
) t
.join();
253 if (failed
) throw std::runtime_error("parallel::parallel failed");
256 //////////////////////////////////////////////////////////////////////
260 //////////////////////////////////////////////////////////////////////