Compare commits
35
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fc73d1b060 | ||
|
|
3c76a402e5 | ||
|
|
f50d5b274d | ||
|
|
c371153bfc | ||
|
|
63da7bb974 | ||
|
|
285f1500cb | ||
|
|
03e0e99768 | ||
|
|
e75752005b | ||
|
|
354ecff2f1 | ||
|
|
94603260f6 | ||
|
|
d65c9b3a5f | ||
|
|
a2aab3a2b4 | ||
|
|
e6162e8838 | ||
|
|
43169c243f | ||
|
|
5d7abed881 | ||
|
|
f580ec8edb | ||
|
|
c4a120705f | ||
|
|
3b1bc31f45 | ||
|
|
e8037363ac | ||
|
|
0a57bda871 | ||
|
|
a52c65b06a | ||
|
|
4624e4ff59 | ||
|
|
7567de93ae | ||
|
|
97ac8f4324 | ||
|
|
40a9a10600 | ||
|
|
5d8ddb3fb6 | ||
|
|
4d3f9752d0 | ||
|
|
0a2b398702 | ||
|
|
497a2b1695 | ||
|
|
1eb893a18e | ||
|
|
da864d5ecb | ||
|
|
86fe3f04c1 | ||
|
|
dd8f8a7826 | ||
|
|
f81abcb640 | ||
|
|
6aab19f72a |
@@ -0,0 +1,38 @@
|
||||
### Description
|
||||
_Add comprehensive description of proposed changes_
|
||||
|
||||
|
||||
Fixes # - _issue number(s) if exists_
|
||||
|
||||
### Type of change
|
||||
|
||||
_Choose one or multiple, leave empty if none of the other choices apply_
|
||||
|
||||
_Add respective label(s) to PR if you have permissions_
|
||||
|
||||
- [ ] bug fix - _change which fixes an issue_
|
||||
- [ ] new feature - _change which adds functionality_
|
||||
- [ ] tests - _change in tests_
|
||||
- [ ] infrastructure - _change in infrastructure and CI_
|
||||
- [ ] documentation - _documentation update_
|
||||
|
||||
### Tests
|
||||
|
||||
- [ ] added - _required for new features and for some bug fixes_
|
||||
- [ ] not needed
|
||||
|
||||
### Documentation
|
||||
|
||||
- [ ] updated in # - _add PR number_
|
||||
- [ ] needs to be updated
|
||||
- [ ] not needed
|
||||
|
||||
### Breaks backward compatibility
|
||||
- [ ] Yes
|
||||
- [ ] No
|
||||
- [ ] Unknown
|
||||
|
||||
### Notify the following users
|
||||
_List users with `@` to send notifications_
|
||||
|
||||
### Other information
|
||||
+1
-1
@@ -38,7 +38,7 @@ You can use issues to report a problem, make a feature request, or add comments
|
||||
|
||||
You can find all [open oneTBB pull requests](https://github.com/oneapi-src/oneTBB/pulls) on GitHub.
|
||||
|
||||
No anonymous contributions are accepted. The name in the commit message Signed-of-by line and your email must match the change authorship information. Make sure your .gitconfig is set up correctly so you can use `git commit -s` for signing your patches:
|
||||
No anonymous contributions are accepted. The name in the commit message Signed-off-by line and your email must match the change authorship information. Make sure your .gitconfig is set up correctly so you can use `git commit -s` for signing your patches:
|
||||
|
||||
`git config --global user.name "Taylor Developer"`
|
||||
|
||||
|
||||
-34
@@ -1,34 +0,0 @@
|
||||
# Installation from Sources
|
||||
|
||||
Required Software:
|
||||
* DPC++/C++ Compiler
|
||||
* Microsoft* Visual Studio* (Windows* OS only)
|
||||
* GNU Compilers (gcc) 4.8.5–10
|
||||
* GNU C Library (glibc) version 2.17–2.32
|
||||
* Clang* 4.2–11.0
|
||||
|
||||
The project uses CMake build configuration. Make sure that you have CMake version 3.1 or higher.
|
||||
|
||||
Before installing the library, learn how to [configure, build, and test](cmake\README.md) the project.
|
||||
|
||||
NOTE: Be careful about installation. Avoid commands like `make install` unless you fully understand the consequences.
|
||||
|
||||
The following installation components are supported:
|
||||
|
||||
`runtime` - oneTBB runtime package (core shared libraries and .dll files on Windows).
|
||||
`devel` - oneTBB development package (header files, CMake integration files, library symbolic links, and .lib files on Windows).
|
||||
`tbb4py` - oneTBB Module for Python.
|
||||
|
||||
To install the specific components after configuration and build, run:
|
||||
|
||||
```bash
|
||||
cmake -DCOMPONENT=<component> [-DBUILD_TYPE=<build-type>] -P cmake_install.cmake
|
||||
```
|
||||
|
||||
Simple packaging using CPack is supported.
|
||||
The following commands allow to create a simple portable package that includes header files, libraries, and integration files for CMake:
|
||||
|
||||
```bash
|
||||
cmake <options> ..
|
||||
cpack
|
||||
```
|
||||
@@ -5,8 +5,7 @@ oneAPI Threading Building Blocks (oneTBB) lets you easily write parallel C++ pro
|
||||
full advantage of multicore performance that are portable, composable, and have future-proof scalability.
|
||||
|
||||
## Installation
|
||||
You can install oneTBB from [GitHub*](https://github.com/oneapi-src/oneTBB/releases).
|
||||
See [Installation from Sources](INSTALL.md) for details.
|
||||
You can install oneTBB from GitHub*, see [Installation from Sources](INSTALL.md) for details.
|
||||
|
||||
## Documentation
|
||||
* [oneTBB Specification](https://spec.oneapi.com/versions/latest/elements/oneTBB/source/nested-index.html)
|
||||
@@ -17,7 +16,7 @@ See [Installation from Sources](INSTALL.md) for details.
|
||||
|
||||
## Support
|
||||
Please report issues and suggestions via
|
||||
[GitHub issues](https://github.com/oneapi-src/oneTBB/issues).
|
||||
[GitHub issues](https://github.com/oneapi-src/oneTBB/issues). See our [documentation](./CONTRIBUTING.md##Issues) to learn how to work with issues.
|
||||
|
||||
## How to Contribute
|
||||
We welcome community contributions, so check our [contributing guidelines](CONTRIBUTING.md)
|
||||
|
||||
@@ -33,7 +33,7 @@ if (MSVC_VERSION LESS_EQUAL 1900)
|
||||
set(TBB_TEST_COMPILE_FLAGS ${TBB_TEST_COMPILE_FLAGS} /wd4503)
|
||||
endif()
|
||||
set(TBB_LIB_COMPILE_FLAGS -D_CRT_SECURE_NO_WARNINGS /GS)
|
||||
set(TBB_COMMON_COMPILE_FLAGS /volatile:iso /FS)
|
||||
set(TBB_COMMON_COMPILE_FLAGS /volatile:iso /FS /EHsc)
|
||||
|
||||
# Ignore /WX set through add_compile_options() or added to CMAKE_CXX_FLAGS if TBB_STRICT is disabled.
|
||||
if (NOT TBB_STRICT AND COMMAND tbb_remove_compile_flag)
|
||||
|
||||
+27
-3
@@ -16,9 +16,18 @@ of multi-core processors. Use |short_name| to write scalable applications that:
|
||||
- Emphasize data parallel programming
|
||||
- Take advantage of concurrent collections and parallel algorithms
|
||||
|
||||
System Requirements
|
||||
*******************
|
||||
|
||||
Refer to the `oneTBB System Requirements <https://software.intel.com/content/www/us/en/develop/articles/intel-oneapi-threading-building-blocks-system-requirements.html>`_.
|
||||
|
||||
|
||||
Before You Begin
|
||||
****************
|
||||
|
||||
Download |short_name| as a `stand-alone product <https://software.intel.com/content/www/us/en/develop/articles/oneapi-standalone-components.html#onetbb>`_
|
||||
or as a part of the `Intel(R) oneAPI Base Toolkit <https://software.intel.com/content/www/us/en/develop/tools/oneapi/base-toolkit/download.html>`_.
|
||||
|
||||
After installing |short_name|, you need to set the environment variables:
|
||||
|
||||
#. Go to the oneTBB installation directory (``<install_dir>``). By default, ``<install_dir>`` is the following:
|
||||
@@ -43,10 +52,25 @@ After installing |short_name|, you need to set the environment variables:
|
||||
``vars.bat in <install_dir>/tbb/latest/env``
|
||||
|
||||
|
||||
System Requirements
|
||||
*******************
|
||||
Example
|
||||
*******
|
||||
|
||||
Refer to the `oneTBB System Requirements <https://software.intel.com/content/www/us/en/develop/articles/intel-oneapi-threading-building-blocks-system-requirements.html>`_.
|
||||
Below you can find a typical example for a |short_name| algorithm.
|
||||
The sample calculates a sum of all integer numbers from 1 to 100.
|
||||
|
||||
.. code:: cpp
|
||||
|
||||
int sum = oneapi::tbb::parallel_reduce(oneapi::tbb::blocked_range<int>(1,101), 0,
|
||||
[](oneapi::tbb::blocked_range<int> const& r, int init) -> int {
|
||||
for (int v = r.begin(); v != r.end(); v++ ) {
|
||||
init += v;
|
||||
}
|
||||
return init;
|
||||
},
|
||||
[](int lhs, int rhs) -> int {
|
||||
return lhs + rhs;
|
||||
}
|
||||
);
|
||||
|
||||
Find more
|
||||
*********
|
||||
|
||||
@@ -1,19 +1,3 @@
|
||||
//configure tms
|
||||
var wapLocalCode = 'us-en';
|
||||
//dynamically set per localized site, see mapping table for values
|
||||
var wapSection = "oneapi-docs";
|
||||
//load tms
|
||||
(function() {
|
||||
var host = (window.document.location.protocol == 'http:') ? "http://www.intel.com" : "https://www.intel.com";
|
||||
var url = host+"/content/dam/www/global/wap/tms-loader.js"; //wap file url
|
||||
var po = document.createElement('script');
|
||||
po.type = 'text/javascript';
|
||||
po.async = true;
|
||||
po.src = url;
|
||||
var s = document.getElementsByTagName('script')[0];
|
||||
s.parentNode.insertBefore(po, s);
|
||||
})();
|
||||
|
||||
window.MathJax = {
|
||||
TeX: {
|
||||
Macros: {
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
{% extends "!layout.html" %}
|
||||
{% block extrahead %}
|
||||
<script defer type="text/javascript" src="https://www.intel.com/content/dam/www/global/wap/performance-config.js" ></script>
|
||||
<script type="text/javascript">
|
||||
// Configure TMS settings
|
||||
var wapLocalCode = 'us-en'; // Dynamically set per localized site, see mapping table for values
|
||||
var wapSection = "oneapi-tbb"; // WAP team will give you a unique section for your site
|
||||
// Load TMS
|
||||
(function () {
|
||||
var url = 'https://www.intel.com/content/dam/www/global/wap/tms-loader.js'; // WAP file URL
|
||||
var po = document.createElement('script'); po.type = 'text/javascript'; po.async = true; po.src = url;
|
||||
var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(po, s);
|
||||
})();
|
||||
</script>
|
||||
{% endblock %}
|
||||
+37
-1
@@ -1,6 +1,42 @@
|
||||
|full_name|
|
||||
===========
|
||||
|
||||
|
||||
This document contains information about |full_name|.
|
||||
|short_name| is a flexible performance library that can be found in the |base_tk|
|
||||
or as a stand-alone product. More information and specifications can be found on the
|
||||
|full_name| main page.
|
||||
|
||||
Documentation for older versions of |short_name| is available for download only.
|
||||
For a list of available documentation downloads by product version, see these pages:
|
||||
|
||||
* `Download Documentation for Intel Parallel Studio XE <https://software.intel.com/content/www/us/en/develop/articles/download-documentation-intel-parallel-studio-xe-current-previous.html>`_
|
||||
* `Download Documentation for Intel System Studio <https://software.intel.com/content/www/us/en/develop/articles/download-documentation-intel-system-studio-current-previous.html>`_
|
||||
|
||||
The following are some important topics for the ``novice user``:
|
||||
|
||||
**Get Started with oneAPI Threading Building Blocks (oneTBB)**
|
||||
|
||||
`Get Started with oneTBB <https://software.intel.com/content/www/us/en/develop/documentation/get-started-with-onetbb/top.html>`_ gives you a brief explanation of what oneTBB is.
|
||||
|
||||
**oneTBB Benefits**
|
||||
|
||||
:ref:`Benefits` describes how |short_name| differs from typical threading packages.
|
||||
|
||||
**Package Contents**
|
||||
|
||||
|short_name| includes dynamic library files and header files for Windows*, Linux*, and macOS*
|
||||
operating systems as described in :ref:`Package_Contents`.
|
||||
|
||||
|
||||
The following is important topic for the ``experienced user``:
|
||||
|
||||
**Migrating from Threading Building Blocks (TBB)**
|
||||
|
||||
Learn how to migrate from TBB to oneTBB with :ref:`Migration_Guide`.
|
||||
|
||||
|
||||
|
||||
.. toctree::
|
||||
:maxdepth: 2
|
||||
|
||||
@@ -10,4 +46,4 @@
|
||||
intro/Benefits
|
||||
tbb_userguide/title
|
||||
reference/reference
|
||||
intro/notices_and_disclaimers
|
||||
intro/notices_and_disclaimers
|
||||
|
||||
@@ -325,3 +325,131 @@ task:
|
||||
}
|
||||
tg.wait();
|
||||
}
|
||||
|
||||
|
||||
Scheduler Bypass
|
||||
----------------
|
||||
|
||||
TBB ``task::execute()`` method can return a pointer to a task that can be executed next by the current thread.
|
||||
This might reduce scheduling overheads compared to direct ``spawn``. Similar to ``spawn``, the returned task
|
||||
is not guaranteed to be executed next by the current thread.
|
||||
|
||||
.. code:: cpp
|
||||
|
||||
#include <tbb/task.h>
|
||||
|
||||
// Assuming OtherTask is defined.
|
||||
|
||||
struct Task : tbb::task {
|
||||
task* execute(){
|
||||
// some work to do ...
|
||||
|
||||
auto* other_p = new(this->parent().allocate_child()) OtherTask{};
|
||||
this->parent().add_ref_count();
|
||||
|
||||
return other_p;
|
||||
}
|
||||
};
|
||||
|
||||
int main(){
|
||||
// Assuming RootTask is defined.
|
||||
RootTask& root = *new(tbb::task::allocate_root()) RootTask{};
|
||||
|
||||
Task& child = *new(root.allocate_child()) Task{/*params*/};
|
||||
|
||||
root.add_ref_count();
|
||||
|
||||
tbb::task_spawn(child);
|
||||
|
||||
root.wait_for_all();;
|
||||
}
|
||||
|
||||
In oneTBB this can be done using the preview feature of ``oneapi::tbb::task_group``.
|
||||
|
||||
.. code:: cpp
|
||||
|
||||
#define TBB_PREVIEW_TASK_GROUP_EXTENSIONS 1
|
||||
#include <oneapi/tbb/task_group.h>
|
||||
|
||||
// Assuming OtherTask is defined.
|
||||
|
||||
int main(){
|
||||
oneapi::tbb::task_group tg;
|
||||
|
||||
tg.run([&tg](){
|
||||
//some work to do ...
|
||||
|
||||
return tg.defer(OtherTask{});
|
||||
});
|
||||
|
||||
tg.wait();
|
||||
}
|
||||
|
||||
Here ``oneapi::tbb::task_group::defer`` adds a new task into the ``tg``. However, the task is not put into a
|
||||
queue of tasks ready for execution via ``oneapi::tbb::task_group::run``, but bypassed to the executing thread directly
|
||||
via function return value.
|
||||
|
||||
Deferred task creation
|
||||
----------------------
|
||||
The TBB low-level task API separates the task creation from the actual spawning. This separation allows to
|
||||
postpone the task spawning, while the parent task and final result production are blocked from premature leave.
|
||||
For example, ``RootTask``, ``ChildTask``, and ``CallBackTask`` are the user-side functors that
|
||||
inherit ``tbb::task`` and implement its interface. Then, blocking the ``RootTask`` from leaving prematurely
|
||||
and waiting on it is implemented as follows:
|
||||
|
||||
.. code:: cpp
|
||||
|
||||
#include <tbb/task.h>
|
||||
|
||||
int main() {
|
||||
// Assuming RootTask, ChildTask, and CallBackTask are defined.
|
||||
RootTask& root = *new(tbb::task::allocate_root()) RootTask{};
|
||||
|
||||
ChildTask& child = *new(root.allocate_child()) ChildTask{/*params*/};
|
||||
CallBackTask& cb_task = *new(root.allocate_child()) CallBackTask{/*params*/};
|
||||
|
||||
root.set_ref_count(3);
|
||||
|
||||
tbb::task::spawn(child);
|
||||
|
||||
register_callback([cb_task&](){
|
||||
tbb::task::enqueue(cb_task);
|
||||
});
|
||||
|
||||
root.wait_for_all();
|
||||
// Control flow will reach here only after both ChildTask and CallBackTask are executed,
|
||||
// i.e. after the callback is called
|
||||
}
|
||||
|
||||
In oneTBB this can be done using the preview feature of ``oneapi::tbb::task_group``.
|
||||
|
||||
.. code:: cpp
|
||||
|
||||
#define TBB_PREVIEW_TASK_GROUP_EXTENSIONS 1
|
||||
#include <oneapi/tbb/task_group.h>
|
||||
|
||||
int main(){
|
||||
oneapi::tbb::task_group tg;
|
||||
oneapi::tbb::task_arena arena;
|
||||
// Assuming ChildTask and CallBackTask are defined.
|
||||
|
||||
auto cb = tg.defer(CallBackTask{/*params*/});
|
||||
|
||||
register_callback([&tg, c = std::move(cb), &arena]{
|
||||
arena.enqueue(c);
|
||||
});
|
||||
|
||||
tg.run(ChildTask{/*params*/});
|
||||
|
||||
|
||||
tg.wait();
|
||||
// Control flow gets here once both ChildTask and CallBackTask are executed
|
||||
// i.e. after the callback is called
|
||||
}
|
||||
|
||||
Here ``oneapi::tbb::task_group::defer`` adds a new task into the ``tg``. However, the task is not spawned until
|
||||
``oneapi::tbb::task_arena::enqueue`` is called.
|
||||
|
||||
.. note::
|
||||
The call to ``oneapi::tbb::task_group::wait`` will not return control until both ``ChildTask`` and
|
||||
``CallBackTask`` are executed.
|
||||
|
||||
@@ -19,9 +19,9 @@
|
||||
|
||||
#if __TBB_PREVIEW_COLLABORATIVE_CALL_ONCE
|
||||
|
||||
#include "tbb/task_arena.h"
|
||||
#include "tbb/task_group.h"
|
||||
#include "tbb/task.h"
|
||||
#include "task_arena.h"
|
||||
#include "task_group.h"
|
||||
#include "task.h"
|
||||
|
||||
#include <atomic>
|
||||
|
||||
|
||||
@@ -524,6 +524,28 @@ protected:
|
||||
return cancellation_status ? canceled : complete;
|
||||
}
|
||||
|
||||
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
|
||||
task_group_status internal_run_and_wait(d2::task_handle&& h) {
|
||||
if (h == nullptr) {
|
||||
throw_exception(exception_id::bad_task_handle);
|
||||
}
|
||||
|
||||
using acs = d2::task_handle_accessor;
|
||||
if (&acs::ctx_of(h) != &context()) {
|
||||
throw_exception(exception_id::bad_task_handle_wrong_task_group);
|
||||
}
|
||||
|
||||
bool cancellation_status = false;
|
||||
try_call([&] {
|
||||
execute_and_wait(*acs::release(h), context(), m_wait_ctx, context());
|
||||
}).on_completion([&] {
|
||||
// TODO: the reset method is not thread-safe. Ensure the correct behavior.
|
||||
cancellation_status = context().is_group_execution_cancelled();
|
||||
context().reset();
|
||||
});
|
||||
return cancellation_status ? canceled : complete;
|
||||
}
|
||||
#endif
|
||||
template<typename F>
|
||||
task* prepare_task(F&& f) {
|
||||
m_wait_ctx.reserve();
|
||||
@@ -632,6 +654,12 @@ public:
|
||||
task_group_status run_and_wait(const F& f) {
|
||||
return internal_run_and_wait(f);
|
||||
}
|
||||
|
||||
#if __TBB_PREVIEW_TASK_GROUP_EXTENSIONS
|
||||
task_group_status run_and_wait(d2::task_handle&& h) {
|
||||
return internal_run_and_wait(std::move(h));
|
||||
}
|
||||
#endif
|
||||
}; // class task_group
|
||||
|
||||
#if TBB_PREVIEW_ISOLATED_TASK_GROUP
|
||||
|
||||
@@ -23,7 +23,7 @@
|
||||
// Product version
|
||||
#define TBB_VERSION_MAJOR 2021
|
||||
// Update version
|
||||
#define TBB_VERSION_MINOR 4
|
||||
#define TBB_VERSION_MINOR 5
|
||||
// "Patch" version for custom releases
|
||||
#define TBB_VERSION_PATCH 0
|
||||
// Suffix string
|
||||
@@ -34,7 +34,7 @@
|
||||
// OneAPI oneTBB specification version
|
||||
#define ONETBB_SPEC_VERSION "1.0"
|
||||
// Full interface version
|
||||
#define TBB_INTERFACE_VERSION 12040
|
||||
#define TBB_INTERFACE_VERSION 12050
|
||||
// Major interface version
|
||||
#define TBB_INTERFACE_VERSION_MAJOR (TBB_INTERFACE_VERSION/1000)
|
||||
// Minor interface version
|
||||
|
||||
+15
-12
@@ -217,9 +217,6 @@ struct arena_base : padded<intrusive_list_node> {
|
||||
task_stream<back_nonnull_accessor> my_critical_task_stream;
|
||||
#endif
|
||||
|
||||
//! The number of workers requested by the external thread owning the arena.
|
||||
unsigned my_max_num_workers;
|
||||
|
||||
//! The total number of workers that are requested from the resource manager.
|
||||
int my_total_num_workers_requested;
|
||||
|
||||
@@ -254,18 +251,9 @@ struct arena_base : padded<intrusive_list_node> {
|
||||
//! The market that owns this arena.
|
||||
market* my_market;
|
||||
|
||||
//! ABA prevention marker.
|
||||
std::uintptr_t my_aba_epoch;
|
||||
|
||||
//! Default task group context.
|
||||
d1::task_group_context* my_default_ctx;
|
||||
|
||||
//! The number of slots in the arena.
|
||||
unsigned my_num_slots;
|
||||
|
||||
//! The number of reserved slots (can be occupied only by external threads).
|
||||
unsigned my_num_reserved_slots;
|
||||
|
||||
#if __TBB_ENQUEUE_ENFORCED_CONCURRENCY
|
||||
// arena needs an extra worker despite a global limit
|
||||
std::atomic<bool> my_global_concurrency_mode;
|
||||
@@ -284,6 +272,21 @@ struct arena_base : padded<intrusive_list_node> {
|
||||
int my_local_concurrency_requests;
|
||||
#endif /* __TBB_ENQUEUE_ENFORCED_CONCURRENCY*/
|
||||
|
||||
//! ABA prevention marker.
|
||||
std::uintptr_t my_aba_epoch;
|
||||
//! The number of slots in the arena.
|
||||
unsigned my_num_slots;
|
||||
//! The number of reserved slots (can be occupied only by external threads).
|
||||
unsigned my_num_reserved_slots;
|
||||
//! The number of workers requested by the external thread owning the arena.
|
||||
unsigned my_max_num_workers;
|
||||
|
||||
//! The target serialization epoch for callers of adjust_job_count_estimate
|
||||
int my_adjust_demand_target_epoch;
|
||||
|
||||
//! The current serialization epoch for callers of adjust_job_count_estimate
|
||||
d1::waitable_atomic<int> my_adjust_demand_current_epoch;
|
||||
|
||||
#if TBB_USE_ASSERT
|
||||
//! Used to trap accesses to the object after its destruction.
|
||||
std::uintptr_t my_guard;
|
||||
|
||||
+4
-4
@@ -578,14 +578,14 @@ void market::adjust_demand ( arena& a, int delta, bool mandatory ) {
|
||||
my_num_workers_requested += delta;
|
||||
__TBB_ASSERT(my_num_workers_requested <= (int)effective_soft_limit, NULL);
|
||||
|
||||
target_epoch = my_adjust_demand_target_epoch++;
|
||||
target_epoch = a.my_adjust_demand_target_epoch++;
|
||||
}
|
||||
|
||||
my_adjust_demand_current_epoch.wait_until(target_epoch, /* context = */ target_epoch, std::memory_order_acquire);
|
||||
a.my_adjust_demand_current_epoch.wait_until(target_epoch, /* context = */ target_epoch, std::memory_order_relaxed);
|
||||
// Must be called outside of any locks
|
||||
my_server->adjust_job_count_estimate( delta );
|
||||
my_adjust_demand_current_epoch.exchange(target_epoch + 1);
|
||||
my_adjust_demand_current_epoch.notify_relaxed(target_epoch + 1);
|
||||
a.my_adjust_demand_current_epoch.exchange(target_epoch + 1);
|
||||
a.my_adjust_demand_current_epoch.notify_relaxed(target_epoch + 1);
|
||||
}
|
||||
|
||||
void market::process( job& j ) {
|
||||
|
||||
@@ -105,12 +105,6 @@ private:
|
||||
//! Number of workers currently requested from RML
|
||||
int my_num_workers_requested;
|
||||
|
||||
//! The target serialization epoch for callers of adjust_job_count_estimate
|
||||
int my_adjust_demand_target_epoch;
|
||||
|
||||
//! The current serialization epoch for callers of adjust_job_count_estimate
|
||||
d1::waitable_atomic<int> my_adjust_demand_current_epoch;
|
||||
|
||||
//! First unused index of worker
|
||||
/** Used to assign indices to the new workers coming from RML, and busy part
|
||||
of my_workers array. **/
|
||||
|
||||
+109
-55
@@ -168,7 +168,11 @@ struct queuing_rw_mutex_impl {
|
||||
s.my_state.store(d1::queuing_rw_mutex::scoped_lock::state_t(write ? STATE_WRITER : STATE_READER), std::memory_order_relaxed);
|
||||
s.my_internal_lock.store(RELEASED, std::memory_order_relaxed);
|
||||
|
||||
queuing_rw_mutex::scoped_lock* predecessor = m.q_tail.exchange(&s, std::memory_order_release);
|
||||
|
||||
// The CAS must have release semantics, because we are
|
||||
// "sending" the fields initialized above to other actors.
|
||||
// We need acquire semantics, because we are acquiring the predecessor (or mutex if no predecessor)
|
||||
queuing_rw_mutex::scoped_lock* predecessor = m.q_tail.exchange(&s, std::memory_order_acq_rel);
|
||||
|
||||
if( write ) { // Acquiring for write
|
||||
|
||||
@@ -177,7 +181,8 @@ struct queuing_rw_mutex_impl {
|
||||
predecessor = tricky_pointer(predecessor) & ~FLAG;
|
||||
__TBB_ASSERT( !predecessor->my_next, "the predecessor has another successor!");
|
||||
tricky_pointer::store(predecessor->my_next, &s, std::memory_order_release);
|
||||
spin_wait_until_eq(s.my_going, 1U);
|
||||
// We are acquiring the mutex
|
||||
spin_wait_until_eq(s.my_going, 1U, std::memory_order_acquire);
|
||||
}
|
||||
|
||||
} else { // Acquiring for read
|
||||
@@ -194,10 +199,18 @@ struct queuing_rw_mutex_impl {
|
||||
} else {
|
||||
// Load predecessor->my_state now, because once predecessor->my_next becomes
|
||||
// non-NULL, we must assume that *predecessor might be destroyed.
|
||||
pred_state = STATE_READER;
|
||||
predecessor->my_state.compare_exchange_strong(pred_state, STATE_READER_UNBLOCKNEXT, std::memory_order_acq_rel);
|
||||
pred_state = predecessor->my_state.load(std::memory_order_relaxed);
|
||||
if (pred_state == STATE_READER) {
|
||||
// Notify the previous reader to unblock us.
|
||||
predecessor->my_state.compare_exchange_strong(pred_state, STATE_READER_UNBLOCKNEXT, std::memory_order_relaxed);
|
||||
}
|
||||
if (pred_state == STATE_ACTIVEREADER) { // either we initially read it or CAS failed
|
||||
// Active reader means that the predecessor already acquired the mutex and cannot notify us.
|
||||
// Therefore, we need to acquire the mutex ourselves by re-reading predecessor state.
|
||||
(void)predecessor->my_state.load(std::memory_order_acquire);
|
||||
}
|
||||
}
|
||||
tricky_pointer::store(s.my_prev, predecessor, std::memory_order_relaxed); // TODO: order?
|
||||
tricky_pointer::store(s.my_prev, predecessor, std::memory_order_relaxed);
|
||||
__TBB_ASSERT( !( tricky_pointer(predecessor) & FLAG ), "use of corrupted pointer!" );
|
||||
__TBB_ASSERT( !predecessor->my_next.load(std::memory_order_relaxed), "the predecessor has another successor!");
|
||||
tricky_pointer::store(predecessor->my_next, &s, std::memory_order_release);
|
||||
@@ -206,13 +219,18 @@ struct queuing_rw_mutex_impl {
|
||||
sync_prepare_done = true;
|
||||
ITT_NOTIFY(sync_prepare, s.my_mutex);
|
||||
#endif
|
||||
// We are acquiring the mutex
|
||||
spin_wait_until_eq(s.my_going, 1U, std::memory_order_acquire);
|
||||
}
|
||||
}
|
||||
|
||||
// The protected state must have been acquired here before it can be further released to any other reader(s):
|
||||
unsigned char old_state = STATE_READER;
|
||||
s.my_state.compare_exchange_strong(old_state, STATE_ACTIVEREADER, std::memory_order_acq_rel);
|
||||
// When this reader is signaled by previous actor it acquires the mutex.
|
||||
// We need to build happens-before relation with all other coming readers that will read our ACTIVEREADER
|
||||
// without blocking on my_going. Therefore, we need to publish ACTIVEREADER with release semantics.
|
||||
// On fail it is relaxed, because we will build happens-before on my_going.
|
||||
s.my_state.compare_exchange_strong(old_state, STATE_ACTIVEREADER, std::memory_order_release, std::memory_order_relaxed);
|
||||
if( old_state!=STATE_READER ) {
|
||||
#if __TBB_USE_ITT_NOTIFY
|
||||
if( !sync_prepare_done )
|
||||
@@ -220,20 +238,16 @@ struct queuing_rw_mutex_impl {
|
||||
#endif
|
||||
// Failed to become active reader -> need to unblock the next waiting reader first
|
||||
__TBB_ASSERT( s.my_state.load(std::memory_order_relaxed)==STATE_READER_UNBLOCKNEXT, "unexpected state" );
|
||||
spin_wait_while_eq(s.my_next, 0U);
|
||||
spin_wait_while_eq(s.my_next, 0U, std::memory_order_acquire);
|
||||
/* my_state should be changed before unblocking the next otherwise it might finish
|
||||
and another thread can get our old state and left blocked */
|
||||
s.my_state.store(STATE_ACTIVEREADER, std::memory_order_relaxed);
|
||||
tricky_pointer::load(s.my_next, std::memory_order_relaxed)->my_going.store(1U, std::memory_order_release);
|
||||
}
|
||||
__TBB_ASSERT( s.my_state.load(std::memory_order_relaxed) ==STATE_ACTIVEREADER, "unlocked reader is active reader" );
|
||||
__TBB_ASSERT(s.my_state.load(std::memory_order_relaxed) == STATE_ACTIVEREADER, "unlocked reader is active reader");
|
||||
}
|
||||
|
||||
ITT_NOTIFY(sync_acquired, s.my_mutex);
|
||||
|
||||
// Force acquire so that user's critical section receives correct values
|
||||
// from processor that was previously in the user's critical section.
|
||||
atomic_fence(std::memory_order_acquire); // TODO: consider exchange on line 171
|
||||
}
|
||||
|
||||
//! A method to acquire queuing_rw_mutex if it is free
|
||||
@@ -253,13 +267,11 @@ struct queuing_rw_mutex_impl {
|
||||
s.my_internal_lock.store(RELEASED, std::memory_order_relaxed);
|
||||
|
||||
// The CAS must have release semantics, because we are
|
||||
// "sending" the fields initialized above to other processors.
|
||||
// "sending" the fields initialized above to other actors.
|
||||
// We need acquire semantics, because we are acquiring the mutex
|
||||
d1::queuing_rw_mutex::scoped_lock* expected = nullptr;
|
||||
if (!m.q_tail.compare_exchange_strong(expected, &s, std::memory_order_release))
|
||||
if (!m.q_tail.compare_exchange_strong(expected, &s, std::memory_order_acq_rel))
|
||||
return false; // Someone already took the lock
|
||||
// Force acquire so that user's critical section receives correct values
|
||||
// from processor that was previously in the user's critical section.
|
||||
atomic_fence(std::memory_order_acquire);
|
||||
s.my_mutex = &m;
|
||||
ITT_NOTIFY(sync_acquired, s.my_mutex);
|
||||
return true;
|
||||
@@ -280,32 +292,50 @@ struct queuing_rw_mutex_impl {
|
||||
d1::queuing_rw_mutex::scoped_lock* next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
if( !next ) {
|
||||
d1::queuing_rw_mutex::scoped_lock* expected = &s;
|
||||
if( s.my_mutex->q_tail.compare_exchange_strong(expected, nullptr, std::memory_order_release) ) {
|
||||
// Release mutex on success otherwise wait for successor publication
|
||||
if( s.my_mutex->q_tail.compare_exchange_strong(expected, nullptr,
|
||||
std::memory_order_release, std::memory_order_relaxed) )
|
||||
{
|
||||
// this was the only item in the queue, and the queue is now empty.
|
||||
goto done;
|
||||
}
|
||||
spin_wait_while_eq(s.my_next, 0U);
|
||||
spin_wait_while_eq(s.my_next, 0U, std::memory_order_relaxed);
|
||||
next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
}
|
||||
next->my_going.store(2U, std::memory_order_relaxed); // protect next queue node from being destroyed too early
|
||||
if( next->my_state.load()==STATE_UPGRADE_WAITING ) {
|
||||
// If the next is STATE_UPGRADE_WAITING, it is expected to acquire all other released readers via release
|
||||
// sequence in next->my_state. In that case, we need to preserve release sequence in next->my_state
|
||||
// contributed by other reader. So, there are two approaches not to break the release sequence:
|
||||
// 1. Use read-modify-write (exchange) operation to store with release the UPGRADE_LOSER state;
|
||||
// 2. Acquire the release sequence and store the sequence and UPGRADE_LOSER state.
|
||||
// The second approach seems better on x86 because it does not involve interlocked operations.
|
||||
// Therefore, we read next->my_state with acquire while it is not required for else branch to get the
|
||||
// release sequence.
|
||||
if( next->my_state.load(std::memory_order_acquire)==STATE_UPGRADE_WAITING ) {
|
||||
// the next waiting for upgrade means this writer was upgraded before.
|
||||
acquire_internal_lock(s);
|
||||
// Responsibility transition, the one who reads uncorrupted my_prev will do release.
|
||||
// Guarantee that above store of 2 into next->my_going happens-before resetting of next->my_prev
|
||||
d1::queuing_rw_mutex::scoped_lock* tmp = tricky_pointer::exchange(next->my_prev, nullptr, std::memory_order_release);
|
||||
next->my_state.store(STATE_UPGRADE_LOSER, std::memory_order_relaxed);
|
||||
// Pass the release sequence that we acquired with the above load of next->my_state.
|
||||
next->my_state.store(STATE_UPGRADE_LOSER, std::memory_order_release);
|
||||
// We are releasing the mutex
|
||||
next->my_going.store(1U, std::memory_order_release);
|
||||
unblock_or_wait_on_internal_lock(s, get_flag(tmp));
|
||||
} else {
|
||||
// next->state cannot be STATE_UPGRADE_REQUESTED
|
||||
__TBB_ASSERT( next->my_state.load(std::memory_order_relaxed) & (STATE_COMBINED_WAITINGREADER | STATE_WRITER), "unexpected state" );
|
||||
__TBB_ASSERT( !( next->my_prev.load(std::memory_order_relaxed) & FLAG ), "use of corrupted pointer!" );
|
||||
tricky_pointer::store(next->my_prev, nullptr, std::memory_order_relaxed);
|
||||
// Guarantee that above store of 2 into next->my_going happens-before resetting of next->my_prev
|
||||
tricky_pointer::store(next->my_prev, nullptr, std::memory_order_release);
|
||||
// We are releasing the mutex
|
||||
next->my_going.store(1U, std::memory_order_release);
|
||||
}
|
||||
|
||||
} else { // Acquired for read
|
||||
|
||||
// The basic idea it to build happens-before relation with left and right readers via prev and next. In addition,
|
||||
// the first reader should acquire the left (prev) signal and propagate to right (next). To simplify, we always
|
||||
// build happens-before relation between left and right (left is happened before right).
|
||||
queuing_rw_mutex::scoped_lock *tmp = nullptr;
|
||||
retry:
|
||||
// Addition to the original paper: Mark my_prev as in use
|
||||
@@ -317,12 +347,9 @@ struct queuing_rw_mutex_impl {
|
||||
// Failed to acquire the lock on predecessor. The predecessor either unlinks or upgrades.
|
||||
// In the second case, it could or could not know my "in use" flag - need to check
|
||||
// Responsibility transition, the one who reads uncorrupted my_prev will do release.
|
||||
tmp = tricky_pointer::compare_exchange_strong(s.my_prev, tricky_pointer(predecessor) | FLAG, predecessor, std::memory_order_release);
|
||||
tmp = tricky_pointer::compare_exchange_strong(s.my_prev, tricky_pointer(predecessor) | FLAG, predecessor, std::memory_order_acquire);
|
||||
if( !(tricky_pointer(tmp) & FLAG) ) {
|
||||
// Wait for the predecessor to change my_prev (e.g. during unlink)
|
||||
// TODO: spin_wait condition seems never reachable
|
||||
__TBB_ASSERT(tricky_pointer::load(s.my_prev, std::memory_order_relaxed) != (tricky_pointer(predecessor) | FLAG), nullptr);
|
||||
tricky_pointer::spin_wait_while_eq( s.my_prev, tricky_pointer(predecessor)|FLAG );
|
||||
// Now owner of predecessor is waiting for _us_ to release its lock
|
||||
release_internal_lock(*predecessor);
|
||||
}
|
||||
@@ -338,13 +365,13 @@ struct queuing_rw_mutex_impl {
|
||||
tricky_pointer::store(predecessor->my_next, nullptr, std::memory_order_release);
|
||||
|
||||
d1::queuing_rw_mutex::scoped_lock* expected = &s;
|
||||
if( !tricky_pointer::load(s.my_next, std::memory_order_relaxed) && !s.my_mutex->q_tail.compare_exchange_strong(expected, predecessor, std::memory_order_release) ) {
|
||||
if( !tricky_pointer::load(s.my_next, std::memory_order_acquire) && !s.my_mutex->q_tail.compare_exchange_strong(expected, predecessor, std::memory_order_release) ) {
|
||||
spin_wait_while_eq( s.my_next, 0U, std::memory_order_acquire );
|
||||
}
|
||||
__TBB_ASSERT( !(s.my_next.load(std::memory_order_relaxed) & FLAG), "use of corrupted pointer" );
|
||||
|
||||
// ensure acquire semantics of reading 'my_next'
|
||||
if(d1::queuing_rw_mutex::scoped_lock *const l_next = tricky_pointer::load(s.my_next, std::memory_order_acquire) ) { // I->next != nil, TODO: rename to next after clearing up and adapting the n in the comment two lines below
|
||||
// my_next is acquired either with load or spin_wait.
|
||||
if(d1::queuing_rw_mutex::scoped_lock *const l_next = tricky_pointer::load(s.my_next, std::memory_order_relaxed) ) { // I->next != nil, TODO: rename to next after clearing up and adapting the n in the comment two lines below
|
||||
// Equivalent to I->next->prev = I->prev but protected against (prev[n]&FLAG)!=0
|
||||
tmp = tricky_pointer::exchange(l_next->my_prev, predecessor, std::memory_order_release);
|
||||
// I->prev->next = I->next;
|
||||
@@ -359,9 +386,12 @@ struct queuing_rw_mutex_impl {
|
||||
d1::queuing_rw_mutex::scoped_lock* next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
if( !next ) {
|
||||
d1::queuing_rw_mutex::scoped_lock* expected = &s;
|
||||
if( !s.my_mutex->q_tail.compare_exchange_strong(expected, nullptr, std::memory_order_release) ) {
|
||||
spin_wait_while_eq( s.my_next, 0U );
|
||||
next = tricky_pointer::load(s.my_next, std::memory_order_relaxed);
|
||||
// Release mutex on success otherwise wait for successor publication
|
||||
if( !s.my_mutex->q_tail.compare_exchange_strong(expected, nullptr,
|
||||
std::memory_order_release, std::memory_order_relaxed) )
|
||||
{
|
||||
spin_wait_while_eq( s.my_next, 0U, std::memory_order_relaxed );
|
||||
next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
} else {
|
||||
goto unlock_self;
|
||||
}
|
||||
@@ -375,7 +405,8 @@ struct queuing_rw_mutex_impl {
|
||||
unblock_or_wait_on_internal_lock(s, get_flag(tmp));
|
||||
}
|
||||
done:
|
||||
spin_wait_while_eq( s.my_going, 2U );
|
||||
// Lifetime synchronization, no need to build happens-before relation
|
||||
spin_wait_while_eq( s.my_going, 2U, std::memory_order_relaxed );
|
||||
|
||||
s.initialize();
|
||||
}
|
||||
@@ -384,26 +415,41 @@ struct queuing_rw_mutex_impl {
|
||||
if ( s.my_state.load(std::memory_order_relaxed) == STATE_ACTIVEREADER ) return true; // Already a reader
|
||||
|
||||
ITT_NOTIFY(sync_releasing, s.my_mutex);
|
||||
s.my_state.store(STATE_READER, std::memory_order_relaxed);
|
||||
if( ! tricky_pointer::load(s.my_next, std::memory_order_relaxed)) {
|
||||
d1::queuing_rw_mutex::scoped_lock* next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
if( !next ) {
|
||||
s.my_state.store(STATE_READER, std::memory_order_seq_cst);
|
||||
// the following load of q_tail must not be reordered with setting STATE_READER above
|
||||
if( &s==s.my_mutex->q_tail.load() ) {
|
||||
if( &s == s.my_mutex->q_tail.load(std::memory_order_seq_cst) ) {
|
||||
unsigned char old_state = STATE_READER;
|
||||
s.my_state.compare_exchange_strong(old_state, STATE_ACTIVEREADER, std::memory_order_release);
|
||||
// When this reader is signaled by previous actor it acquires the mutex.
|
||||
// We need to build happens-before relation with all other coming readers that will read our ACTIVEREADER
|
||||
// without blocking on my_going. Therefore, we need to publish ACTIVEREADER with release semantics.
|
||||
// On fail it is relaxed, because we will build happens-before on my_going.
|
||||
s.my_state.compare_exchange_strong(old_state, STATE_ACTIVEREADER, std::memory_order_release, std::memory_order_relaxed);
|
||||
if( old_state==STATE_READER )
|
||||
return true; // Downgrade completed
|
||||
}
|
||||
/* wait for the next to register */
|
||||
spin_wait_while_eq( s.my_next, 0U );
|
||||
spin_wait_while_eq(s.my_next, 0U, std::memory_order_relaxed);
|
||||
next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
}
|
||||
d1::queuing_rw_mutex::scoped_lock *const next = tricky_pointer::load(s.my_next, std::memory_order_acquire);
|
||||
|
||||
__TBB_ASSERT( next, "still no successor at this point!" );
|
||||
if( next->my_state & STATE_COMBINED_WAITINGREADER )
|
||||
if( next->my_state.load(std::memory_order_relaxed) & STATE_COMBINED_WAITINGREADER )
|
||||
next->my_going.store(1U, std::memory_order_release);
|
||||
else if( next->my_state==STATE_UPGRADE_WAITING )
|
||||
// If the next is STATE_UPGRADE_WAITING, it is expected to acquire all other released readers via release
|
||||
// sequence in next->my_state. In that case, we need to preserve release sequence in next->my_state
|
||||
// contributed by other reader. So, there are two approaches not to break the release sequence:
|
||||
// 1. Use read-modify-write (exchange) operation to store with release the UPGRADE_LOSER state;
|
||||
// 2. Acquire the release sequence and store the sequence and UPGRADE_LOSER state.
|
||||
// The second approach seems better on x86 because it does not involve interlocked operations.
|
||||
// Therefore, we read next->my_state with acquire while it is not required for else branch to get the
|
||||
// release sequence.
|
||||
else if( next->my_state.load(std::memory_order_acquire)==STATE_UPGRADE_WAITING )
|
||||
// the next waiting for upgrade means this writer was upgraded before.
|
||||
next->my_state.store(STATE_UPGRADE_LOSER, std::memory_order_relaxed);
|
||||
s.my_state.store(STATE_ACTIVEREADER, std::memory_order_relaxed);
|
||||
// To safe release sequence on next->my_state read it with acquire
|
||||
next->my_state.store(STATE_UPGRADE_LOSER, std::memory_order_release);
|
||||
s.my_state.store(STATE_ACTIVEREADER, std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -419,16 +465,20 @@ struct queuing_rw_mutex_impl {
|
||||
queuing_rw_mutex::scoped_lock* me = &s;
|
||||
|
||||
ITT_NOTIFY(sync_releasing, s.my_mutex);
|
||||
s.my_state.store(STATE_UPGRADE_REQUESTED, std::memory_order_relaxed);
|
||||
// Publish ourselves into my_state that other UPGRADE_WAITING actors can acquire our state.
|
||||
s.my_state.store(STATE_UPGRADE_REQUESTED, std::memory_order_release);
|
||||
requested:
|
||||
__TBB_ASSERT( !(s.my_next.load(std::memory_order_relaxed) & FLAG), "use of corrupted pointer!" );
|
||||
acquire_internal_lock(s);
|
||||
d1::queuing_rw_mutex::scoped_lock* expected = &s;
|
||||
if( !s.my_mutex->q_tail.compare_exchange_strong(expected, tricky_pointer(me)|FLAG, std::memory_order_release) ) {
|
||||
spin_wait_while_eq( s.my_next, 0U );
|
||||
if( !s.my_mutex->q_tail.compare_exchange_strong(expected, tricky_pointer(me)|FLAG, std::memory_order_acq_rel) ) {
|
||||
spin_wait_while_eq( s.my_next, 0U, std::memory_order_relaxed );
|
||||
queuing_rw_mutex::scoped_lock * next;
|
||||
next = tricky_pointer::fetch_add(s.my_next, FLAG, std::memory_order_acquire);
|
||||
unsigned short n_state = next->my_state;
|
||||
// While we were READER the next READER might reach STATE_UPGRADE_WAITING state.
|
||||
// Therefore, it did not build happens before relation with us and we need to acquire the
|
||||
// next->my_state to build the happens before relation ourselves
|
||||
unsigned short n_state = next->my_state.load(std::memory_order_acquire);
|
||||
/* the next reader can be blocked by our state. the best thing to do is to unblock it */
|
||||
if( n_state & STATE_COMBINED_WAITINGREADER )
|
||||
next->my_going.store(1U, std::memory_order_release);
|
||||
@@ -439,7 +489,7 @@ struct queuing_rw_mutex_impl {
|
||||
// save next|FLAG for simplicity of following comparisons
|
||||
tmp = tricky_pointer(next)|FLAG;
|
||||
for( atomic_backoff b; tricky_pointer::load(s.my_next, std::memory_order_relaxed)==tmp; b.pause() ) {
|
||||
if( s.my_state & STATE_COMBINED_UPGRADING ) {
|
||||
if( s.my_state.load(std::memory_order_acquire) & STATE_COMBINED_UPGRADING ) {
|
||||
if( tricky_pointer::load(s.my_next, std::memory_order_acquire)==tmp )
|
||||
tricky_pointer::store(s.my_next, next, std::memory_order_relaxed);
|
||||
goto waiting;
|
||||
@@ -458,7 +508,11 @@ struct queuing_rw_mutex_impl {
|
||||
} // if( this != my_mutex->q_tail... )
|
||||
{
|
||||
unsigned char old_state = STATE_UPGRADE_REQUESTED;
|
||||
s.my_state.compare_exchange_strong(old_state, STATE_UPGRADE_WAITING, std::memory_order_acquire);
|
||||
// If we reach STATE_UPGRADE_WAITING state we do not build happens-before relation with READER on
|
||||
// left. We delegate this responsibility to READER on left when it try upgrading. Therefore, we are releasing
|
||||
// on success.
|
||||
// Otherwise, on fail, we already acquired the next->my_state.
|
||||
s.my_state.compare_exchange_strong(old_state, STATE_UPGRADE_WAITING, std::memory_order_release, std::memory_order_relaxed);
|
||||
}
|
||||
waiting:
|
||||
__TBB_ASSERT( !( s.my_next.load(std::memory_order_relaxed) & FLAG ), "use of corrupted pointer!" );
|
||||
@@ -477,11 +531,14 @@ struct queuing_rw_mutex_impl {
|
||||
// While the predecessor pointer (my_prev) is in use (FLAG is set), we can safely update the node`s state.
|
||||
// Corrupted pointer transitions responsibility to release the predecessor`s node on us.
|
||||
unsigned char old_state = STATE_UPGRADE_REQUESTED;
|
||||
predecessor->my_state.compare_exchange_strong(old_state, STATE_UPGRADE_WAITING, std::memory_order_release);
|
||||
// Try to build happens before with the upgrading READER on left. If fail, the predecessor state is not
|
||||
// important for us because it will acquire our state.
|
||||
predecessor->my_state.compare_exchange_strong(old_state, STATE_UPGRADE_WAITING, std::memory_order_release,
|
||||
std::memory_order_relaxed);
|
||||
}
|
||||
if( !success ) {
|
||||
// Responsibility transition, the one who reads uncorrupted my_prev will do release.
|
||||
tmp = tricky_pointer::compare_exchange_strong(s.my_prev, tricky_pointer(predecessor)|FLAG, predecessor, std::memory_order_release);
|
||||
tmp = tricky_pointer::compare_exchange_strong(s.my_prev, tricky_pointer(predecessor)|FLAG, predecessor, std::memory_order_acquire);
|
||||
if( tricky_pointer(tmp) & FLAG ) {
|
||||
tricky_pointer::spin_wait_while_eq(s.my_prev, predecessor);
|
||||
predecessor = tricky_pointer::load(s.my_prev, std::memory_order_relaxed);
|
||||
@@ -509,9 +566,6 @@ struct queuing_rw_mutex_impl {
|
||||
// now wait for the predecessor to finish working with my fields
|
||||
spin_wait_while_eq( s.my_going, 2U );
|
||||
|
||||
// Acquire critical section indirectly from previous owner or directly from predecessor (TODO: not clear).
|
||||
atomic_fence(std::memory_order_acquire); // on either "my_mutex->q_tail" or "my_going" (TODO: not clear)
|
||||
|
||||
bool result = ( s.my_state != STATE_UPGRADE_LOSER );
|
||||
s.my_state.store(STATE_WRITER, std::memory_order_relaxed);
|
||||
s.my_going.store(1U, std::memory_order_relaxed);
|
||||
|
||||
@@ -40,8 +40,12 @@ TEST_CASE("Basic Locable requirement test") {
|
||||
GeneralTest<oneapi::tbb::spin_rw_mutex>("Spin RW Mutex");
|
||||
GeneralTest<oneapi::tbb::queuing_mutex>("Queuing Mutex");
|
||||
GeneralTest<oneapi::tbb::queuing_rw_mutex>("Queuing RW Mutex");
|
||||
// TODO: Consider adding Thread Sanitizer (note that accesses inside the transaction
|
||||
// considered as races by Thread Sanitizer)
|
||||
#if !__TBB_USE_THREAD_SANITIZER
|
||||
GeneralTest<oneapi::tbb::speculative_spin_mutex>("Speculative Spin Mutex");
|
||||
GeneralTest<oneapi::tbb::speculative_spin_rw_mutex>("Speculative Spin RW Mutex");
|
||||
#endif
|
||||
// NullMutexes
|
||||
GeneralTest<oneapi::tbb::null_mutex, utils::AtomicCounter<oneapi::tbb::null_mutex>>("Null Mutex", false);
|
||||
GeneralTest<oneapi::tbb::null_rw_mutex, utils::AtomicCounter<oneapi::tbb::null_rw_mutex>>("Null RW Mutex", false);
|
||||
@@ -56,8 +60,10 @@ TEST_CASE("Lockable requirement test") {
|
||||
TestTryAcquire<oneapi::tbb::spin_rw_mutex>("Spin RW Mutex");
|
||||
TestTryAcquire<oneapi::tbb::queuing_mutex>("Queuing Mutex");
|
||||
TestTryAcquire<oneapi::tbb::queuing_rw_mutex>("Queuing RW Mutex");
|
||||
#if !__TBB_USE_THREAD_SANITIZER
|
||||
TestTryAcquire<oneapi::tbb::speculative_spin_mutex>("Speculative Spin Mutex");
|
||||
TestTryAcquire<oneapi::tbb::speculative_spin_rw_mutex>("Speculative Spin RW Mutex");
|
||||
#endif
|
||||
TestTryAcquire<oneapi::tbb::null_mutex>("Null Mutex");
|
||||
}
|
||||
|
||||
@@ -67,16 +73,18 @@ TEST_CASE("Shared mutexes (reader/writer) test") {
|
||||
// General reader writer capabilities + upgrade/downgrade
|
||||
TestReaderWriterLock<oneapi::tbb::spin_rw_mutex>("Spin RW Mutex");
|
||||
TestReaderWriterLock<oneapi::tbb::queuing_rw_mutex>("Queuing RW Mutex");
|
||||
TestReaderWriterLock<oneapi::tbb::speculative_spin_rw_mutex>("Speculative Spin RW Mutex");
|
||||
TestNullRWMutex<oneapi::tbb::null_rw_mutex>("Null RW Mutex");
|
||||
// Single threaded read/write try_acquire operations
|
||||
TestTryAcquireReader<oneapi::tbb::spin_rw_mutex>("Spin RW Mutex");
|
||||
TestTryAcquireReader<oneapi::tbb::queuing_rw_mutex>("Queuing RW Mutex");
|
||||
TestRWStateMultipleChange<oneapi::tbb::spin_rw_mutex>("Spin RW Mutex");
|
||||
TestRWStateMultipleChange<oneapi::tbb::queuing_rw_mutex>("Queuing RW Mutex");
|
||||
TestTryAcquireReader<oneapi::tbb::null_rw_mutex>("Null RW Mutex");
|
||||
#if !__TBB_USE_THREAD_SANITIZER
|
||||
TestReaderWriterLock<oneapi::tbb::speculative_spin_rw_mutex>("Speculative Spin RW Mutex");
|
||||
TestTryAcquireReader<oneapi::tbb::speculative_spin_rw_mutex>("Speculative Spin RW Mutex");
|
||||
TestRWStateMultipleChange<oneapi::tbb::speculative_spin_rw_mutex>("Speculative Spin RW Mutex");
|
||||
TestTryAcquireReader<oneapi::tbb::null_rw_mutex>("Null RW Mutex");
|
||||
#endif
|
||||
}
|
||||
|
||||
//! Testing ISO C++ Mutex and Shared Mutex requirements.
|
||||
|
||||
@@ -247,7 +247,7 @@ void test(ConnectJoinNodeFunc&& connect_join_node) {
|
||||
|
||||
remove_edge(bn3, input_port<2>(*jn));
|
||||
remove_edge(bn2, input_port<1>(*jn));
|
||||
remove_edge(bn1, input_port<0>(*jn));
|
||||
remove_edge(bn1, *jn); //Removes an edge between a sender and port 0 of a multi-input successor.
|
||||
remove_edge(*jn, qn);
|
||||
|
||||
run_and_check(g, bn1, bn2, bn3, qn, /*expected=*/false);
|
||||
|
||||
@@ -125,6 +125,18 @@ struct put_dec_body : utils::NoAssign {
|
||||
|
||||
};
|
||||
|
||||
template< typename Sender, typename Receiver >
|
||||
void make_edge_impl(Sender& sender, Receiver& receiver){
|
||||
#if __GNUC__ < 12 && !TBB_USE_DEBUG
|
||||
// Seemingly, GNU compiler generates incorrect code for the call of limiter.register_successor in release (-03)
|
||||
// The function pointer to make_edge workarounds the issue for unknown reason
|
||||
auto make_edge_ptr = tbb::flow::make_edge<int>;
|
||||
make_edge_ptr(sender, receiver);
|
||||
#else
|
||||
tbb::flow::make_edge(sender, receiver);
|
||||
#endif
|
||||
}
|
||||
|
||||
template< typename T >
|
||||
void test_puts_with_decrements( int num_threads, tbb::flow::limiter_node< T >& lim , tbb::flow::graph& g) {
|
||||
parallel_receiver<T> r(g);
|
||||
@@ -355,14 +367,7 @@ void test_reserve_release_messages() {
|
||||
broad.try_put(1); //failed message retrieved.
|
||||
g.wait_for_all();
|
||||
|
||||
#if __GNUC__ && __GNUC__ < 12 && !TBB_USE_DEBUG
|
||||
// Seemingly, GNU compiler generates incorrect code for the call of limiter.register_successor in release (-03)
|
||||
// The function pointer to make_edge workarounds the issue for unknown reason
|
||||
auto make_edge_ptr = make_edge<int>;
|
||||
make_edge_ptr(limit, output_queue); //putting the successor back
|
||||
#else
|
||||
make_edge(limit, output_queue); //putting the successor back
|
||||
#endif
|
||||
make_edge_impl(limit, output_queue); //putting the successor back
|
||||
|
||||
broad.try_put(1); //drop the count
|
||||
|
||||
@@ -445,7 +450,9 @@ void test_try_put_without_successors() {
|
||||
int try_put_num{3};
|
||||
tbb::flow::buffer_node<int> bn(g);
|
||||
tbb::flow::limiter_node<int> ln(g, try_put_num);
|
||||
|
||||
tbb::flow::make_edge(bn, ln);
|
||||
|
||||
int i = 1;
|
||||
for (; i <= try_put_num; i++)
|
||||
bn.try_put(i);
|
||||
@@ -457,7 +464,9 @@ void test_try_put_without_successors() {
|
||||
return int{};
|
||||
}
|
||||
);
|
||||
tbb::flow::make_edge(ln, fn);
|
||||
|
||||
make_edge_impl(ln, fn);
|
||||
|
||||
g.wait_for_all();
|
||||
CHECK((counter == i * try_put_num / 2));
|
||||
|
||||
|
||||
@@ -33,7 +33,9 @@
|
||||
//! \brief Test for [mutex.spin_mutex mutex.spin_rw_mutex mutex.queuing_mutex mutex.queuing_rw_mutexmutex.speculative_spin_mutex mutex.speculative_spin_rw_mutex] specifications
|
||||
|
||||
// TODO: Investigate why RTM doesn't work on some macOS.
|
||||
#if __TBB_TSX_INTRINSICS_PRESENT && !__APPLE__
|
||||
// TODO: Consider adding Thread Sanitizer (note that accesses inside the transaction
|
||||
// considered as races by Thread Sanitizer)
|
||||
#if __TBB_TSX_INTRINSICS_PRESENT && !__APPLE__ && !__TBB_USE_THREAD_SANITIZER
|
||||
|
||||
inline static bool IsInsideTx() {
|
||||
return _xtest() != 0;
|
||||
|
||||
@@ -255,6 +255,12 @@ private:
|
||||
|
||||
// Simple test for basic resumable tasks functionality
|
||||
void TestSuspendResume() {
|
||||
#if __TBB_USE_SANITIZERS
|
||||
constexpr int iter_size = 100;
|
||||
#else
|
||||
constexpr int iter_size = 50000;
|
||||
#endif
|
||||
|
||||
std::atomic<int> global_epoch; global_epoch = 0;
|
||||
EpochAsyncActivity async(4, global_epoch);
|
||||
|
||||
@@ -263,7 +269,7 @@ void TestSuspendResume() {
|
||||
inner_par_iters = outer_par_iters = 0;
|
||||
|
||||
tbb::parallel_for(0, N, [&](int) {
|
||||
for (int i = 0; i < 100; ++i) {
|
||||
for (int i = 0; i < iter_size; ++i) {
|
||||
ets_fiber.local() = i;
|
||||
|
||||
int local_epoch;
|
||||
@@ -286,7 +292,7 @@ void TestSuspendResume() {
|
||||
++outer_par_iters;
|
||||
});
|
||||
CHECK(outer_par_iters == N);
|
||||
CHECK(inner_par_iters == N*N*100);
|
||||
CHECK(inner_par_iters == N*N*iter_size);
|
||||
}
|
||||
|
||||
// During cleanup external thread's local task pool may
|
||||
|
||||
@@ -1173,6 +1173,23 @@ TEST_CASE("Task handle run"){
|
||||
CHECK_MESSAGE(h == nullptr, "Delayed task can be executed only once");
|
||||
}
|
||||
|
||||
//! Basic test for task handle
|
||||
//! \brief \ref interface \ref requirement
|
||||
TEST_CASE("Task handle run_and_wait"){
|
||||
tbb::task_handle h;
|
||||
|
||||
tbb::task_group tg;
|
||||
bool run {false};
|
||||
|
||||
h = tg.defer([&]{
|
||||
run = true;
|
||||
});
|
||||
CHECK_MESSAGE(run == false, "delayed task should not be run until run(task_handle) is called");
|
||||
tg.run_and_wait(std::move(h));
|
||||
CHECK_MESSAGE(run == true, "Delayed task should be completed when task_group::wait exits");
|
||||
|
||||
CHECK_MESSAGE(h == nullptr, "Delayed task can be executed only once");
|
||||
}
|
||||
//! Test for empty check
|
||||
//! \brief \ref interface
|
||||
TEST_CASE("Task handle empty check"){
|
||||
@@ -1308,7 +1325,7 @@ TEST_CASE("task_handle cannot be scheduled into different task_group"){
|
||||
//! The test for error in task_handle being scheduled into task_group different from one it was created from
|
||||
//! \brief \ref requirement
|
||||
TEST_CASE("task_handle cannot be scheduled into other task_group of the same context"
|
||||
* doctest::should_fail() //Implementation is no there yet, as it is not clear that is the expected behaviour
|
||||
* doctest::should_fail() //Implementation is no there yet, as it is not clear that is the expected behavior
|
||||
* doctest::skip() //skip the test for now, to not pollute the test log
|
||||
)
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user