This contains my bachelors thesis and associated tex files, code snippets and maybe more. Topic: Data Movement in Heterogeneous Memories with Intel Data Streaming Accelerator
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

741 lines
27 KiB

  1. #pragma once
  2. #include <iostream>
  3. #include <unordered_map>
  4. #include <shared_mutex>
  5. #include <mutex>
  6. #include <memory>
  7. #include <sched.h>
  8. #include <numa.h>
  9. #include <numaif.h>
  10. #include <dml/dml.hpp>
  11. namespace dml {
  12. inline const std::string StatusCodeToString(const dml::status_code code) {
  13. switch (code) {
  14. case dml::status_code::ok: return "ok";
  15. case dml::status_code::false_predicate: return "false predicate";
  16. case dml::status_code::partial_completion: return "partial completion";
  17. case dml::status_code::nullptr_error: return "nullptr error";
  18. case dml::status_code::bad_size: return "bad size";
  19. case dml::status_code::bad_length: return "bad length";
  20. case dml::status_code::inconsistent_size: return "inconsistent size";
  21. case dml::status_code::dualcast_bad_padding: return "dualcast bad padding";
  22. case dml::status_code::bad_alignment: return "bad alignment";
  23. case dml::status_code::buffers_overlapping: return "buffers overlapping";
  24. case dml::status_code::delta_delta_empty: return "delta delta empty";
  25. case dml::status_code::batch_overflow: return "batch overflow";
  26. case dml::status_code::execution_failed: return "execution failed";
  27. case dml::status_code::unsupported_operation: return "unsupported operation";
  28. case dml::status_code::queue_busy: return "queue busy";
  29. case dml::status_code::error: return "unknown error";
  30. case dml::status_code::config_error: return "config error";
  31. default: return "unhandled error";
  32. }
  33. }
  34. }
  35. namespace dsacache {
  36. class Cache;
  37. /*
  38. * Class Description:
  39. * Holds all required information on one cache entry and is used
  40. * both internally by the Cache and externally by the user.
  41. *
  42. * Important Usage Notes:
  43. * The pointer is only updated in WaitOnCompletion() which
  44. * therefore must be called by the user at some point in order
  45. * to use the cached data. Using this class as T for
  46. * std::shared_ptr<T> is not recommended as references are
  47. * already counted internally.
  48. *
  49. * Cache Lifetime:
  50. * As long as the instance is referenced, the pointer it stores
  51. * is guaranteed to be either nullptr or pointing to a valid copy.
  52. *
  53. * Implementation Detail:
  54. * Performs self-reference counting with a shared atomic integer.
  55. * Therefore on creating a copy the reference count is increased
  56. * and with the destructor it is deacresed. If the last copy is
  57. * destroyed the actual underlying data is freed and all shared
  58. * variables deleted.
  59. *
  60. * Notes on Thread Safety:
  61. * Class is thread safe in any possible state and performs
  62. * reference counting and deallocation itself entirely atomically.
  63. */
  64. class CacheData {
  65. public:
  66. using dml_handler = dml::handler<dml::mem_copy_operation, std::allocator<uint8_t>>;
  67. private:
  68. // data source and size of the block
  69. uint8_t* src_;
  70. size_t size_;
  71. // global reference counting object
  72. std::atomic<int32_t>* active_;
  73. // global cache-location pointer
  74. std::atomic<uint8_t*>* cache_;
  75. // object-local incomplete cache location pointer
  76. // which is only available in the first instance
  77. uint8_t* incomplete_cache_;
  78. // dml handler vector pointer which is only
  79. // available in the first instance
  80. std::unique_ptr<std::vector<dml_handler>> handlers_;
  81. // deallocates the global cache-location
  82. // and invalidates it
  83. void Deallocate();
  84. // checks whether there are at least two
  85. // valid references to this object which
  86. // is done as the cache always has one
  87. // internally to any living instance
  88. bool Active() const;
  89. friend Cache;
  90. public:
  91. CacheData(uint8_t* data, const size_t size);
  92. CacheData(const CacheData& other);
  93. ~CacheData();
  94. // waits on completion of caching operations
  95. // for this task and is safe to be called in
  96. // any state of the object
  97. void WaitOnCompletion();
  98. // returns the cache data location for this
  99. // instance which is valid as long as the
  100. // instance is alive - !!! this may also
  101. // yield a nullptr !!!
  102. uint8_t* GetDataLocation() const;
  103. };
  104. /*
  105. * Class Description:
  106. * Class will handle access to data through internal copies.
  107. * These are obtained via work submission to the Intel DSA which takes
  108. * care of asynchronously duplicating the data. The user will define
  109. * where these copies lie and which system nodes will perform the copy.
  110. * This is done through policy functions set during initialization.
  111. *
  112. * Placement Policy:
  113. * The Placement Policy Function decides on which node a particular
  114. * entry is to be placed, given the current executing node and the
  115. * data source node and data size. This in turn means that for one
  116. * datum, multiple cached copies may exist at one time.
  117. *
  118. * Cache Lifetime:
  119. * When accessing the cache, a CacheData-object will be returned.
  120. * As long as this object lives, the pointer which it holds is
  121. * guaranteed to be either nullptr or a valid copy. When destroyed
  122. * the entry is marked for deletion which is only carried out
  123. * when system memory pressure drives an automated cache flush.
  124. *
  125. * Restrictions:
  126. * - Overlapping Pointers may lead to undefined behaviour during
  127. * manual cache invalidation which should not be used if you
  128. * intend to have these types of pointers
  129. * - Cache Invalidation may only be performed manually and gives
  130. * no ordering guarantees. Therefore, it is the users responsibility
  131. * to ensure that results after invalidation have been generated
  132. * using the latest state of data. The cache is best suited
  133. * to static data.
  134. *
  135. * Notes on Thread Safety:
  136. * - Cache is completely thread-safe after initialization
  137. * - CacheData-class will handle deallocation of data itself by
  138. * performing self-reference-counting atomically and only
  139. * deallocating if the last reference is destroyed
  140. * - The internal cache state has one lock which is either
  141. * acquired shared for reading the state (upon accessing an already
  142. * cached element) or unique (accessing a new element, flushing, invalidating)
  143. * - Waiting on copy completion is done over an atomic-wait in copies
  144. * of the original CacheData-instance
  145. * - Overall this class may experience performance issues due to the use
  146. * of locking (in any configuration), lock contention (worsens with higher
  147. * core count, node count and utilization) and atomics (worse in the same
  148. * situations as lock contention)
  149. *
  150. * Improving Performance:
  151. * When data is never shared between threads or memory size for the cache is
  152. * not an issue you may consider having one Cache-instance per thread and removing
  153. * the lock in Cache and modifying the reference counting and waiting mechanisms
  154. * of CacheData accordingly (although this is high effort and will yield little due
  155. * to the atomics not being shared among cores/nodes).
  156. * Otherwise, one Cache-instance per node could also be considered. This will allow
  157. * the placement policy function to be barebones and reduces the lock contention and
  158. * synchronization impact of the atomic variables.
  159. */
  160. class Cache {
  161. public:
  162. // cache policy is defined as a type here to allow flexible usage of the cacher
  163. // given a numa destination node (where the data will be needed), the numa source
  164. // node (current location of the data) and the data size, this function should
  165. // return optimal cache placement
  166. // dst node and returned value can differ if the system, for example, has HBM
  167. // attached accessible directly to node n under a different node id m
  168. typedef int (CachePolicy)(const int numa_dst_node, const int numa_src_node, const size_t data_size);
  169. // copy policy specifies the copy-executing nodes for a given task
  170. // which allows flexibility in assignment for optimizing raw throughput
  171. // or choosing a conservative usage policy
  172. typedef std::vector<int> (CopyPolicy)(const int numa_dst_node, const int numa_src_node, const size_t data_size);
  173. private:
  174. // mutex for accessing the cache state map
  175. // map from [dst-numa-node,map2]
  176. // map2 from [data-ptr,cache-structure]
  177. struct LockedNodeCacheState {
  178. std::shared_mutex cache_mutex_;
  179. std::unordered_map<uint8_t*, CacheData> node_cache_state_;
  180. };
  181. std::unordered_map<uint8_t, LockedNodeCacheState*> cache_state_;
  182. CachePolicy* cache_policy_function_ = nullptr;
  183. CopyPolicy* copy_policy_function_ = nullptr;
  184. // function used to submit a copy task on a specific node to the dml
  185. // engine on that node - will change the current threads node assignment
  186. // to achieve this so take care to restore this
  187. dml::handler<dml::mem_copy_operation, std::allocator<uint8_t>> ExecuteCopy(
  188. const uint8_t* src, uint8_t* dst, const size_t size, const int node
  189. ) const;
  190. // allocates the required memory on the destination node
  191. // and then submits task to the dml library for processing
  192. // and attaches the handlers to the cache data structure
  193. void SubmitTask(CacheData* task, const int dst_node, const int src_node);
  194. // querries the policy functions for the given data and size
  195. // to obtain destination cache node, also returns the datas
  196. // source node for further usage
  197. // output may depend on the calling threads node assignment
  198. // as this is set as the "optimal placement" node
  199. void GetCacheNode(uint8_t* src, const size_t size, int* OUT_DST_NODE, int* OUT_SRC_NODE) const;
  200. // allocates memory of size "size" on the numa node "node"
  201. // and returns nullptr if this is not possible, also may
  202. // try to flush the cache of the requested node to
  203. // alleviate encountered shortage
  204. uint8_t* AllocOnNode(const size_t size, const int node);
  205. // checks whether the cache contains an entry for
  206. // the given data in the given memory node and
  207. // returns it, otherwise returns nullptr
  208. std::unique_ptr<CacheData> GetFromCache(uint8_t* src, const size_t size, const int dst_node);
  209. public:
  210. ~Cache();
  211. Cache() = default;
  212. Cache(const Cache& other) = delete;
  213. // initializes the cache with the two policy functions
  214. // only after this is it safe to use in a threaded environment
  215. void Init(CachePolicy* cache_policy_function, CopyPolicy* copy_policy_function);
  216. // function to perform data access through the cache
  217. std::unique_ptr<CacheData> Access(uint8_t* data, const size_t size);
  218. // flushes the cache of inactive entries
  219. // if node is -1 then the whole cache is
  220. // checked and otherwise the specified
  221. // node - no checks on node validity
  222. void Flush(const int node = -1);
  223. // forces out all entries from the
  224. // cache and therefore will also "forget"
  225. // still-in-use entries, these will still
  226. // be properly deleted, but the cache
  227. // will be fresh - use for testing
  228. void Clear();
  229. void Invalidate(uint8_t* data);
  230. };
  231. }
  232. inline void dsacache::Cache::Clear() {
  233. for (auto& nc : cache_state_) {
  234. std::unique_lock<std::shared_mutex> lock(nc.second->cache_mutex_);
  235. nc.second->node_cache_state_.clear();
  236. }
  237. }
  238. inline void dsacache::Cache::Init(CachePolicy* cache_policy_function, CopyPolicy* copy_policy_function) {
  239. cache_policy_function_ = cache_policy_function;
  240. copy_policy_function_ = copy_policy_function;
  241. // initialize numa library
  242. numa_available();
  243. // obtain all available nodes
  244. // and those we may allocate
  245. // memory on
  246. const int nodes_max = numa_num_configured_nodes();
  247. const bitmask* valid_nodes = numa_get_mems_allowed();
  248. // prepare the cache state with entries
  249. // for all given nodes
  250. for (int node = 0; node < nodes_max; node++) {
  251. if (numa_bitmask_isbitset(valid_nodes, node)) {
  252. void* block = numa_alloc_onnode(sizeof(LockedNodeCacheState), node);
  253. auto* state = new(block)LockedNodeCacheState;
  254. cache_state_.insert({node,state});
  255. }
  256. }
  257. }
  258. inline std::unique_ptr<dsacache::CacheData> dsacache::Cache::Access(uint8_t* data, const size_t size) {
  259. // get destination numa node for the cache
  260. int dst_node = -1;
  261. int src_node = -1;
  262. GetCacheNode(data, size, &dst_node, &src_node);
  263. // TODO: at this point it could be beneficial to check whether
  264. // TODO: the given destination node is present as an entry
  265. // TODO: in the cache state to see if it is valid
  266. // check whether the data is already cached
  267. std::unique_ptr<CacheData> task = GetFromCache(data, size, dst_node);
  268. if (task != nullptr) {
  269. return std::move(task);
  270. }
  271. // at this point the requested data is not present in cache
  272. // and we create a caching task for it
  273. task = std::make_unique<CacheData>(data, size);
  274. {
  275. LockedNodeCacheState* local_cache_state = cache_state_[dst_node];
  276. std::unique_lock<std::shared_mutex> lock(local_cache_state->cache_mutex_);
  277. const auto state = local_cache_state->node_cache_state_.emplace(task->src_, *task);
  278. // if state.second is false then no insertion took place
  279. // which means that concurrently whith this thread
  280. // some other thread must have accessed the same
  281. // resource in which case we return the other
  282. // threads data cache structure
  283. if (!state.second) {
  284. std::cout << "[!] Found another cache instance for 0x" << std::hex << (uint64_t)task->src_ << std::dec << std::endl;
  285. return std::move(std::make_unique<CacheData>(state.first->second));
  286. }
  287. }
  288. SubmitTask(task.get(), dst_node, src_node);
  289. return std::move(task);
  290. }
  291. inline uint8_t* dsacache::Cache::AllocOnNode(const size_t size, const int node) {
  292. // allocate data on this node and flush the unused parts of the
  293. // cache if the operation fails and retry once
  294. // TODO: smarter flush strategy could keep some stuff cached
  295. // check currently free memory to see if the data fits
  296. long long int free_space = 0;
  297. numa_node_size64(node, &free_space);
  298. if (free_space < size) {
  299. std::cout << "[!] Memory shortage when allocating " << size << "B on node " << node << std::endl;
  300. // dst node lacks memory space so we flush the cache for this
  301. // node hoping to free enough currently unused entries to make
  302. // the second allocation attempt successful
  303. Flush(node);
  304. // re-test by getting the free space and checking again
  305. numa_node_size64(node, &free_space);
  306. if (free_space < size) {
  307. std::cout << "[x] Memory shortage after flush when allocating " << size << "B on node " << node << std::endl;
  308. return nullptr;
  309. }
  310. }
  311. uint8_t* dst = reinterpret_cast<uint8_t*>(numa_alloc_onnode(size, node));
  312. if (dst == nullptr) {
  313. std::cout << "[x] Allocation try failed for " << size << "B on node " << node << std::endl;
  314. return nullptr;
  315. }
  316. return dst;
  317. }
  318. inline void dsacache::Cache::SubmitTask(CacheData* task, const int dst_node, const int src_node) {
  319. uint8_t* dst = AllocOnNode(task->size_, dst_node);
  320. if (dst == nullptr) {
  321. std::cout << "[x] Allocation failed so we can not cache" << std::endl;
  322. return;
  323. }
  324. task->incomplete_cache_ = dst;
  325. // querry copy policy function for the nodes to use for the copy
  326. const std::vector<int> executing_nodes = copy_policy_function_(dst_node, src_node, task->size_);
  327. const size_t task_count = executing_nodes.size();
  328. // each task will copy one fair part of the total size
  329. // and in case the total size is not a factor of the
  330. // given task count the last node must copy the remainder
  331. const size_t size = task->size_ / task_count;
  332. const size_t last_size = size + task->size_ % task_count;
  333. // save the current numa node mask to restore later
  334. // as executing the copy task will place this thread
  335. // on a different node
  336. bitmask* nodemask = numa_get_run_node_mask();
  337. for (uint32_t i = 0; i < task_count; i++) {
  338. const size_t local_size = i + 1 == task_count ? size : last_size;
  339. const size_t local_offset = i * size;
  340. const uint8_t* local_src = task->src_ + local_offset;
  341. uint8_t* local_dst = dst + local_offset;
  342. task->handlers_->emplace_back(ExecuteCopy(local_src, local_dst, local_size, executing_nodes[i]));
  343. }
  344. // restore the previous nodemask
  345. numa_run_on_node_mask(nodemask);
  346. numa_free_nodemask(nodemask);
  347. }
  348. inline dml::handler<dml::mem_copy_operation, std::allocator<uint8_t>> dsacache::Cache::ExecuteCopy(
  349. const uint8_t* src, uint8_t* dst, const size_t size, const int node
  350. ) const {
  351. dml::const_data_view srcv = dml::make_view(src, size);
  352. dml::data_view dstv = dml::make_view(dst, size);
  353. numa_run_on_node(node);
  354. return dml::submit<dml::automatic>(dml::mem_copy.block_on_fault(), srcv, dstv);
  355. }
  356. inline void dsacache::Cache::GetCacheNode(uint8_t* src, const size_t size, int* OUT_DST_NODE, int* OUT_SRC_NODE) const {
  357. // obtain numa node of current thread to determine where the data is needed
  358. const int current_cpu = sched_getcpu();
  359. const int current_node = numa_node_of_cpu(current_cpu);
  360. // obtain node that the given data pointer is allocated on
  361. *OUT_SRC_NODE = -1;
  362. get_mempolicy(OUT_SRC_NODE, NULL, 0, (void*)src, MPOL_F_NODE | MPOL_F_ADDR);
  363. // querry cache policy function for the destination numa node
  364. *OUT_DST_NODE = cache_policy_function_(current_node, *OUT_SRC_NODE, size);
  365. }
  366. inline void dsacache::Cache::Flush(const int node) {
  367. // this lambda is used because below we have two code paths that
  368. // flush nodes, either one single or all successively
  369. const auto FlushNode = [](std::unordered_map<uint8_t*,CacheData>& map) {
  370. // begin at the front of the map
  371. auto it = map.begin();
  372. // loop until we reach the end of the map
  373. while (it != map.end()) {
  374. // if the iterator points to an inactive element
  375. // then we may erase it
  376. if (it->second.Active() == false) {
  377. // erase the iterator from the map
  378. map.erase(it);
  379. // as the erasure invalidated out iterator
  380. // we must start at the beginning again
  381. it = map.begin();
  382. }
  383. else {
  384. // if element is active just move over to the next one
  385. it++;
  386. }
  387. }
  388. };
  389. // we require exclusive lock as we modify the cache state
  390. // node == -1 means that cache on all nodes should be flushed
  391. if (node == -1) {
  392. for (auto& nc : cache_state_) {
  393. std::unique_lock<std::shared_mutex> lock(nc.second->cache_mutex_);
  394. FlushNode(nc.second->node_cache_state_);
  395. }
  396. }
  397. else {
  398. std::unique_lock<std::shared_mutex> lock(cache_state_[node]->cache_mutex_);
  399. FlushNode(cache_state_[node]->node_cache_state_);
  400. }
  401. }
  402. inline std::unique_ptr<dsacache::CacheData> dsacache::Cache::GetFromCache(uint8_t* src, const size_t size, const int dst_node) {
  403. // the best situation is if this data is already cached
  404. // which we check in an unnamed block in which the cache
  405. // is locked for reading to prevent another thread
  406. // from marking the element we may find as unused and
  407. // clearing it
  408. LockedNodeCacheState* local_cache_state = cache_state_[dst_node];
  409. // lock the cache state in shared-mode because we read
  410. std::shared_lock<std::shared_mutex> lock(local_cache_state->cache_mutex_);
  411. // search for the data in our cache state structure at the given node
  412. const auto search = local_cache_state->node_cache_state_.find(src);
  413. // if the data is in our structure we continue
  414. if (search != local_cache_state->node_cache_state_.end()) {
  415. // now check whether the sizes match
  416. if (search->second.size_ >= size) {
  417. // return a unique copy of the entry which uses the object
  418. // lifetime and destructor to safely handle deallocation
  419. return std::move(std::make_unique<CacheData>(search->second));
  420. }
  421. else {
  422. // if the sizes missmatch then we clear the current entry from cache
  423. // which will cause its deletion only after the last possible outside
  424. // reference is also destroyed
  425. local_cache_state->node_cache_state_.erase(search);
  426. }
  427. }
  428. return nullptr;
  429. }
  430. void dsacache::Cache::Invalidate(uint8_t* data) {
  431. // as the cache is modified we must obtain a unique writers lock
  432. // loop through all per-node-caches available
  433. for (auto node : cache_state_) {
  434. std::unique_lock<std::shared_mutex> lock(node.second->cache_mutex_);
  435. // search for an entry for the given data pointer
  436. auto search = node.second->node_cache_state_.find(data);
  437. if (search != node.second->node_cache_state_.end()) {
  438. // if the data is represented in-cache
  439. // then it will be erased to re-trigger
  440. // caching on next access
  441. node.second->node_cache_state_.erase(search);
  442. }
  443. }
  444. }
  445. inline dsacache::Cache::~Cache() {
  446. for (auto node : cache_state_) {
  447. std::unique_lock<std::shared_mutex> lock(node.second->cache_mutex_);
  448. node.second->~LockedNodeCacheState();
  449. numa_free(reinterpret_cast<void*>(node.second), sizeof(LockedNodeCacheState));
  450. }
  451. }
  452. inline dsacache::CacheData::CacheData(uint8_t* data, const size_t size) {
  453. src_ = data;
  454. size_ = size;
  455. active_ = new std::atomic<int32_t>(1);
  456. cache_ = new std::atomic<uint8_t*>();
  457. incomplete_cache_ = nullptr;
  458. handlers_ = std::make_unique<std::vector<dml_handler>>();
  459. }
  460. inline dsacache::CacheData::CacheData(const dsacache::CacheData& other) {
  461. // we copy the ptr to the global atomic reference counter
  462. // and increase the amount of active references
  463. active_ = other.active_;
  464. const int current_active = active_->fetch_add(1);
  465. // source and size will be copied too
  466. // as well as the reference to the global
  467. // atomic cache pointer
  468. src_ = other.src_;
  469. size_ = other.size_;
  470. cache_ = other.cache_;
  471. // incomplete cache and handlers will not
  472. // be copied because only the first instance
  473. // will wait on the completion of handlers
  474. incomplete_cache_ = nullptr;
  475. handlers_ = nullptr;
  476. }
  477. inline dsacache::CacheData::~CacheData() {
  478. // if this is the first instance of this cache structure
  479. // and it has not been waited on and is now being destroyed
  480. // we must wait on completion here to ensure the cache
  481. // remains in a valid state
  482. if (handlers_ != nullptr) {
  483. WaitOnCompletion();
  484. }
  485. // due to fetch_sub returning the preivously held value
  486. // we must subtract one locally to get the current value
  487. const int32_t v = active_->fetch_sub(1) - 1;
  488. // if the returned value is zero or lower
  489. // then we must execute proper deletion
  490. // as this was the last reference
  491. if (v <= 0) {
  492. Deallocate();
  493. delete active_;
  494. delete cache_;
  495. }
  496. }
  497. inline void dsacache::CacheData::Deallocate() {
  498. // although deallocate should only be called from
  499. // a safe context to do so, it can not hurt to
  500. // defensively perform the operation atomically
  501. uint8_t* cache_local = cache_->exchange(nullptr);
  502. if (cache_local != nullptr) numa_free(cache_local, size_);
  503. // if the cache was never waited for then incomplete_cache_
  504. // may still contain a valid pointer which has to be freed
  505. if (incomplete_cache_ != nullptr) numa_free(incomplete_cache_, size_);
  506. }
  507. inline uint8_t* dsacache::CacheData::GetDataLocation() const {
  508. return cache_->load();
  509. }
  510. inline bool dsacache::CacheData::Active() const {
  511. // this entry is active if more than one
  512. // reference exists to it, as the Cache
  513. // will always keep one internally until
  514. // the entry is cleared from cache
  515. return active_->load() > 1;
  516. }
  517. inline void dsacache::CacheData::WaitOnCompletion() {
  518. // the cache data entry can be in two states
  519. // either it is the original one which has not
  520. // been waited for in which case the handlers
  521. // are non-null or it is not
  522. if (handlers_ == nullptr) {
  523. // when no handlers are attached to this cache entry we wait on a
  524. // value change for the cache structure from nullptr to non-null
  525. // which will either go through immediately if the cache is valid
  526. // already or wait until the handler-owning thread notifies us
  527. cache_->wait(nullptr);
  528. }
  529. else {
  530. // when the handlers are non-null there are some DSA task handlers
  531. // available on which we must wait here
  532. // abort is set if any operation encountered an error
  533. bool abort = false;
  534. for (auto& handler : *handlers_) {
  535. auto result = handler.get();
  536. if (result.status != dml::status_code::ok) {
  537. std::cerr << "[x] Encountered bad status code for operation: " << dml::StatusCodeToString(result.status) << std::endl;
  538. // if one of the copy tasks failed we abort the whole task
  539. // after all operations are completed on it
  540. abort = true;
  541. }
  542. }
  543. // the handlers are cleared after all have completed
  544. handlers_ = nullptr;
  545. // now we act depending on whether an abort has been
  546. // called for which signals operation incomplete
  547. if (abort) {
  548. // store nullptr in the cache location
  549. cache_->store(nullptr);
  550. // then free the now incomplete cache
  551. // TODO: it would be possible to salvage the
  552. // TODO: operation at this point but this
  553. // TODO: is quite complicated so we just abort
  554. numa_free(incomplete_cache_, size_);
  555. }
  556. else {
  557. // incomplete cache is now safe to use and therefore we
  558. // swap it with the global cache state of this entry
  559. // and notify potentially waiting threads
  560. cache_->store(incomplete_cache_);
  561. }
  562. // as a last step all waiting threads must
  563. // be notified (copies of this will wait on value
  564. // change of the cache) and the incomplete cache
  565. // is cleared to nullptr as it is not incomplete
  566. cache_->notify_all();
  567. incomplete_cache_ = nullptr;
  568. }
  569. }