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.

829 lines
30 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. inline bool CheckFlag(const uint64_t value, const uint64_t flag) {
  37. return (value & flag) != 0;
  38. }
  39. inline uint64_t UnsetFlag(const uint64_t value, const uint64_t flag) {
  40. return value & (~flag);
  41. }
  42. inline uint64_t SetFlag(const uint64_t value, const uint64_t flag) {
  43. return value | flag;
  44. }
  45. constexpr uint64_t FLAG_WAIT_WEAK = 0b1ULL << 63;
  46. constexpr uint64_t FLAG_HANDLE_PF = 0b1ULL << 62;
  47. constexpr uint64_t FLAG_ACCESS_WEAK = 0b1ULL << 61;
  48. constexpr uint64_t FLAG_FORCE_MAP_PAGES = 0b1ULL << 60;
  49. constexpr uint64_t FLAG_DEFAULT = 0ULL;
  50. class Cache;
  51. /*
  52. * Class Description:
  53. * Holds all required information on one cache entry and is used
  54. * both internally by the Cache and externally by the user.
  55. *
  56. * Important Usage Notes:
  57. * The pointer is only updated in WaitOnCompletion() which
  58. * therefore must be called by the user at some point in order
  59. * to use the cached data. Using this class as T for
  60. * std::shared_ptr<T> is not recommended as references are
  61. * already counted internally.
  62. *
  63. * Cache Lifetime:
  64. * As long as the instance is referenced, the pointer it stores
  65. * is guaranteed to be either nullptr or pointing to a valid copy.
  66. *
  67. * Implementation Detail:
  68. * Performs self-reference counting with a shared atomic integer.
  69. * Therefore on creating a copy the reference count is increased
  70. * and with the destructor it is deacresed. If the last copy is
  71. * destroyed the actual underlying data is freed and all shared
  72. * variables deleted.
  73. *
  74. * Notes on Thread Safety:
  75. * Class is thread safe in any possible state and performs
  76. * reference counting and deallocation itself entirely atomically.
  77. */
  78. class CacheData {
  79. public:
  80. using dml_handler = dml::handler<dml::mem_copy_operation, std::allocator<uint8_t>>;
  81. private:
  82. // set to false if we do not own the cache pointer
  83. bool delete_ = false;
  84. // data source and size of the block
  85. uint8_t* src_;
  86. size_t size_;
  87. // global reference counting object
  88. std::atomic<int32_t>* active_;
  89. // global cache-location pointer
  90. std::atomic<uint8_t*>* cache_;
  91. // object-local incomplete cache location pointer
  92. // contract: only access when being in sole posession of handlers
  93. uint8_t** incomplete_cache_;
  94. // flags inherited from parent cache
  95. uint64_t flags_ = 0;
  96. // dml handler vector pointer which is used
  97. // to wait on caching task completion
  98. std::atomic<std::vector<dml_handler>*>* handlers_;
  99. // invalid handlers pointer as we need a secondary
  100. // invalid state due to issues with waiting
  101. std::vector<dml_handler>* invalid_handlers_;
  102. // deallocates the global cache-location
  103. // and invalidates it
  104. void Deallocate();
  105. size_t GetSize() const { return size_; }
  106. uint8_t* GetSource() const { return src_; }
  107. int32_t GetRefCount() const { return active_->load(); }
  108. void SetCacheToSource() { cache_->store(src_); delete_ = false; }
  109. void SetTaskHandlersAndCache(uint8_t* cache, std::vector<dml_handler>* handlers);
  110. // initializes the class after which it is thread safe
  111. // but may only be destroyed safely after setting handlers
  112. void Init(std::vector<dml_handler>* invalid_handlers);
  113. friend Cache;
  114. public:
  115. CacheData(uint8_t* data, const size_t size);
  116. CacheData(const CacheData& other);
  117. ~CacheData();
  118. // waits on completion of caching operations
  119. // for this task and is safe to be called in
  120. // any state of the object, if the flag
  121. // FLAG_WAIT_WEAK is set for this instance
  122. // (can also be inherited from the creating
  123. // Cache-Instance or on copy from another
  124. // CacheData-Instance), WaitOnCompletion
  125. // provides no validity guarantees to the
  126. // cache pointer (GetDataLocation() may
  127. // return nullptr even after return
  128. // of the wait function). On error this
  129. // function will set the cache pointer
  130. // to the source to provide validity
  131. // guarantees after returning.
  132. void WaitOnCompletion();
  133. // returns the cache data location for this
  134. // instance which is valid as long as the
  135. // instance is alive
  136. // !!! this may also return a nullptr !!!
  137. // see WaitOnCompletion() for how to achieve
  138. // validity guarantees if required
  139. uint8_t* GetDataLocation() const { return cache_->load(); }
  140. void SetFlags(const uint64_t flags) { flags_ = flags; }
  141. uint64_t GetFlags() const { return flags_; }
  142. };
  143. /*
  144. * Class Description:
  145. * Class will handle access to data through internal copies.
  146. * These are obtained via work submission to the Intel DSA which takes
  147. * care of asynchronously duplicating the data. The user will define
  148. * where these copies lie and which system nodes will perform the copy.
  149. * This is done through policy functions set during initialization.
  150. *
  151. * Placement Policy:
  152. * The Placement Policy Function decides on which node a particular
  153. * entry is to be placed, given the current executing node and the
  154. * data source node and data size. This in turn means that for one
  155. * datum, multiple cached copies may exist at one time.
  156. *
  157. * Cache Lifetime:
  158. * When accessing the cache, a CacheData-object will be returned.
  159. * As long as this object lives, the pointer which it holds is
  160. * guaranteed to be either nullptr or a valid copy. When destroyed
  161. * the entry is marked for deletion which is only carried out
  162. * when system memory pressure drives an automated cache flush.
  163. *
  164. * Restrictions:
  165. * - Overlapping Pointers may lead to undefined behaviour during
  166. * manual cache invalidation which should not be used if you
  167. * intend to have these types of pointers
  168. * - Cache Invalidation may only be performed manually and gives
  169. * no ordering guarantees. Therefore, it is the users responsibility
  170. * to ensure that results after invalidation have been generated
  171. * using the latest state of data. The cache is best suited
  172. * to static data.
  173. *
  174. * Notes on Thread Safety:
  175. * - Cache is completely thread-safe after initialization
  176. * - CacheData-class will handle deallocation of data itself by
  177. * performing self-reference-counting atomically and only
  178. * deallocating if the last reference is destroyed
  179. * - The internal cache state has one lock which is either
  180. * acquired shared for reading the state (upon accessing an already
  181. * cached element) or unique (accessing a new element, flushing, invalidating)
  182. * - Waiting on copy completion is done over an atomic-wait in copies
  183. * of the original CacheData-instance
  184. * - Overall this class may experience performance issues due to the use
  185. * of locking (in any configuration), lock contention (worsens with higher
  186. * core count, node count and utilization) and atomics (worse in the same
  187. * situations as lock contention)
  188. *
  189. * Improving Performance:
  190. * When data is never shared between threads or memory size for the cache is
  191. * not an issue you may consider having one Cache-instance per thread and removing
  192. * the lock in Cache and modifying the reference counting and waiting mechanisms
  193. * of CacheData accordingly (although this is high effort and will yield little due
  194. * to the atomics not being shared among cores/nodes).
  195. * Otherwise, one Cache-instance per node could also be considered. This will allow
  196. * the placement policy function to be barebones and reduces the lock contention and
  197. * synchronization impact of the atomic variables.
  198. */
  199. class Cache {
  200. public:
  201. // cache policy is defined as a type here to allow flexible usage of the cacher
  202. // given a numa destination node (where the data will be needed), the numa source
  203. // node (current location of the data) and the data size, this function should
  204. // return optimal cache placement
  205. // dst node and returned value can differ if the system, for example, has HBM
  206. // attached accessible directly to node n under a different node id m
  207. typedef int (CachePolicy)(const int numa_dst_node, const int numa_src_node, const size_t data_size);
  208. // copy policy specifies the copy-executing nodes for a given task
  209. // which allows flexibility in assignment for optimizing raw throughput
  210. // or choosing a conservative usage policy
  211. typedef std::vector<int> (CopyPolicy)(const int numa_dst_node, const int numa_src_node, const size_t data_size);
  212. private:
  213. // flags to store options duh
  214. uint64_t flags_ = 0;
  215. // secondary invalid handlers vector
  216. // needed due to wake-up issues in CacheData::WaitOnCompletion
  217. std::vector<CacheData::dml_handler> invalid_handlers_;
  218. // map from [dst-numa-node,map2]
  219. // map2 from [data-ptr,cache-structure]
  220. struct LockedNodeCacheState {
  221. std::shared_mutex cache_mutex_;
  222. std::unordered_map<uint8_t*, CacheData> node_cache_state_;
  223. };
  224. std::unordered_map<uint8_t, LockedNodeCacheState*> cache_state_;
  225. CachePolicy* cache_policy_function_ = nullptr;
  226. CopyPolicy* copy_policy_function_ = nullptr;
  227. // function used to submit a copy task on a specific node to the dml
  228. // engine on that node - will change the current threads node assignment
  229. // to achieve this so take care to restore this
  230. dml::handler<dml::mem_copy_operation, std::allocator<uint8_t>> ExecuteCopy(
  231. const uint8_t* src, uint8_t* dst, const size_t size, const int node
  232. ) const;
  233. // allocates the required memory on the destination node
  234. // and then submits task to the dml library for processing
  235. // and attaches the handlers to the cache data structure
  236. void SubmitTask(CacheData* task, const int dst_node, const int src_node);
  237. // querries the policy functions for the given data and size
  238. // to obtain destination cache node, also returns the datas
  239. // source node for further usage
  240. // output may depend on the calling threads node assignment
  241. // as this is set as the "optimal placement" node
  242. void GetCacheNode(uint8_t* src, const size_t size, int* OUT_DST_NODE, int* OUT_SRC_NODE) const;
  243. // allocates memory of size "size" on the numa node "node"
  244. // and returns nullptr if this is not possible, also may
  245. // try to flush the cache of the requested node to
  246. // alleviate encountered shortage
  247. uint8_t* AllocOnNode(const size_t size, const int node);
  248. // checks whether the cache contains an entry for
  249. // the given data in the given memory node and
  250. // returns it, otherwise returns nullptr
  251. std::unique_ptr<CacheData> GetFromCache(uint8_t* src, const size_t size, const int dst_node);
  252. public:
  253. ~Cache();
  254. Cache() = default;
  255. Cache(const Cache& other) = delete;
  256. // initializes the cache with the two policy functions
  257. // only after this is it safe to use in a threaded environment
  258. void Init(CachePolicy* cache_policy_function, CopyPolicy* copy_policy_function);
  259. // function to perform data access through the cache, behaviour depends
  260. // on flags, by default will also perform prefetch, otherwise with
  261. // FLAG_ACCESS_WEAK set will not perform prefetch and instead return
  262. // a cache entry with the data source as cache location on cache miss,
  263. // this flag must be set for each invocation, the flags set for the
  264. // entire cache will not be evaluated for this
  265. std::unique_ptr<CacheData> Access(uint8_t* data, const size_t size, const uint64_t flags = FLAG_DEFAULT);
  266. // flushes the cache of inactive entries
  267. // if node is -1 then the whole cache is
  268. // checked and otherwise the specified
  269. // node - no checks on node validity
  270. void Flush(const int node = -1);
  271. // forces out all entries from the
  272. // cache and therefore will also "forget"
  273. // still-in-use entries, these will still
  274. // be properly deleted, but the cache
  275. // will be fresh - use for testing
  276. void Clear();
  277. void Invalidate(uint8_t* data);
  278. void SetFlags(const uint64_t flags) { flags_ = flags; }
  279. uint64_t GetFlags() { return flags_; }
  280. };
  281. }
  282. inline void dsacache::Cache::Clear() {
  283. for (auto& nc : cache_state_) {
  284. std::unique_lock<std::shared_mutex> lock(nc.second->cache_mutex_);
  285. nc.second->node_cache_state_.clear();
  286. }
  287. }
  288. inline void dsacache::Cache::Init(CachePolicy* cache_policy_function, CopyPolicy* copy_policy_function) {
  289. cache_policy_function_ = cache_policy_function;
  290. copy_policy_function_ = copy_policy_function;
  291. // initialize numa library
  292. numa_available();
  293. // obtain all available nodes
  294. // and those we may allocate
  295. // memory on
  296. const int nodes_max = numa_num_configured_nodes();
  297. const bitmask* valid_nodes = numa_get_mems_allowed();
  298. // prepare the cache state with entries
  299. // for all given nodes
  300. for (int node = 0; node < nodes_max; node++) {
  301. if (numa_bitmask_isbitset(valid_nodes, node)) {
  302. void* block = numa_alloc_onnode(sizeof(LockedNodeCacheState), node);
  303. auto* state = new(block)LockedNodeCacheState;
  304. cache_state_.insert({node,state});
  305. }
  306. }
  307. }
  308. inline std::unique_ptr<dsacache::CacheData> dsacache::Cache::Access(uint8_t* data, const size_t size, const uint64_t flags) {
  309. // get destination numa node for the cache
  310. int dst_node = -1;
  311. int src_node = -1;
  312. GetCacheNode(data, size, &dst_node, &src_node);
  313. // check whether the data is already cached
  314. std::unique_ptr<CacheData> task = GetFromCache(data, size, dst_node);
  315. if (task != nullptr) {
  316. return std::move(task);
  317. }
  318. // at this point the requested data is not present in cache
  319. // and we create a caching task for it, copying our current flags
  320. task = std::make_unique<CacheData>(data, size);
  321. task->SetFlags(flags_);
  322. // when the ACCESS_WEAK flag is set for the flags parameter (!)
  323. // and we have reached this point, there was no cache entry
  324. // present for the requested data and therefore we abort
  325. // but to keep validity, we return the previously created
  326. // CacheData struct, setting the cache variable to the
  327. // data source location
  328. if (CheckFlag(flags, FLAG_ACCESS_WEAK)) {
  329. std::cerr << "[x] WEAK ACCESS FAILED!" << std::endl;
  330. task->SetCacheToSource();
  331. return std::move(task);
  332. }
  333. // the following operation adds the task to the cache state
  334. // which requires unique locking of the current nodes entry
  335. {
  336. LockedNodeCacheState* local_cache_state = cache_state_[dst_node];
  337. std::unique_lock<std::shared_mutex> lock(local_cache_state->cache_mutex_);
  338. const auto state = local_cache_state->node_cache_state_.emplace(task->GetSource(), *task);
  339. // if state.second is false then no insertion took place
  340. // which means that concurrently whith this thread
  341. // some other thread must have accessed the same
  342. // resource in which case we return the other
  343. // threads data cache structure
  344. if (!state.second) {
  345. return std::move(std::make_unique<CacheData>(state.first->second));
  346. }
  347. // initialize the task now for thread safety
  348. // as we are now sure that we will submit work
  349. // to it and will not delete it beforehand
  350. // of the one in cache state - must be
  351. // performed for the local and cache-state
  352. // instance as Init will modify values that
  353. // are not shared but copied on copy-construct
  354. state.first->second.Init(&invalid_handlers_);
  355. task->Init(&invalid_handlers_);
  356. }
  357. SubmitTask(task.get(), dst_node, src_node);
  358. return std::move(task);
  359. }
  360. inline uint8_t* dsacache::Cache::AllocOnNode(const size_t size, const int node) {
  361. // allocate data on this node and flush the unused parts of the
  362. // cache if the operation fails and retry once
  363. // TODO: smarter flush strategy could keep some stuff cached
  364. // check currently free memory to see if the data fits
  365. long long int free_space = 0;
  366. numa_node_size64(node, &free_space);
  367. if (free_space < size) {
  368. // dst node lacks memory space so we flush the cache for this
  369. // node hoping to free enough currently unused entries to make
  370. // the second allocation attempt successful
  371. Flush(node);
  372. // re-test by getting the free space and checking again
  373. numa_node_size64(node, &free_space);
  374. if (free_space < size) {
  375. return nullptr;
  376. }
  377. }
  378. uint8_t* dst = reinterpret_cast<uint8_t*>(numa_alloc_onnode(size, node));
  379. if (dst == nullptr) {
  380. return nullptr;
  381. }
  382. if (CheckFlag(flags_, FLAG_FORCE_MAP_PAGES)) {
  383. static const size_t page_size_b = getpagesize();
  384. for (size_t i = 0; i < size; i += page_size_b) {
  385. dst[i] = 0;
  386. }
  387. }
  388. return dst;
  389. }
  390. inline void dsacache::Cache::SubmitTask(CacheData* task, const int dst_node, const int src_node) {
  391. uint8_t* dst = AllocOnNode(task->GetSize(), dst_node);
  392. if (dst == nullptr) {
  393. return;
  394. }
  395. // querry copy policy function for the nodes to use for the copy
  396. const std::vector<int> executing_nodes = copy_policy_function_(dst_node, src_node, task->GetSize());
  397. const size_t task_count = executing_nodes.size();
  398. // each task will copy one fair part of the total size
  399. // and in case the total size is not a factor of the
  400. // given task count the last node must copy the remainder
  401. const size_t size = task->GetSize() / task_count;
  402. const size_t last_size = size + task->GetSize() % task_count;
  403. // save the current numa node mask to restore later
  404. // as executing the copy task will place this thread
  405. // on a different node
  406. auto handlers = new std::vector<CacheData::dml_handler>();
  407. for (uint32_t i = 0; i < task_count; i++) {
  408. const size_t local_size = i + 1 == task_count ? size : last_size;
  409. const size_t local_offset = i * size;
  410. const uint8_t* local_src = task->GetSource() + local_offset;
  411. uint8_t* local_dst = dst + local_offset;
  412. handlers->emplace_back(ExecuteCopy(local_src, local_dst, local_size, executing_nodes[i]));
  413. }
  414. task->SetTaskHandlersAndCache(dst, handlers);
  415. }
  416. inline dml::handler<dml::mem_copy_operation, std::allocator<uint8_t>> dsacache::Cache::ExecuteCopy(
  417. const uint8_t* src, uint8_t* dst, const size_t size, const int node
  418. ) const {
  419. dml::const_data_view srcv = dml::make_view(src, size);
  420. dml::data_view dstv = dml::make_view(dst, size);
  421. if (CheckFlag(flags_, FLAG_HANDLE_PF)) {
  422. return dml::submit<dml::software>(
  423. dml::mem_copy.block_on_fault(), srcv, dstv,
  424. dml::execution_interface<dml::software,std::allocator<uint8_t>>(), node
  425. );
  426. }
  427. else {
  428. return dml::submit<dml::software>(
  429. dml::mem_copy, srcv, dstv,
  430. dml::execution_interface<dml::software,std::allocator<uint8_t>>(), node
  431. );
  432. }
  433. }
  434. inline void dsacache::Cache::GetCacheNode(uint8_t* src, const size_t size, int* OUT_DST_NODE, int* OUT_SRC_NODE) const {
  435. // obtain numa node of current thread to determine where the data is needed
  436. const int current_cpu = sched_getcpu();
  437. const int current_node = numa_node_of_cpu(current_cpu);
  438. // obtain node that the given data pointer is allocated on
  439. *OUT_SRC_NODE = -1;
  440. get_mempolicy(OUT_SRC_NODE, NULL, 0, (void*)src, MPOL_F_NODE | MPOL_F_ADDR);
  441. // querry cache policy function for the destination numa node
  442. *OUT_DST_NODE = cache_policy_function_(current_node, *OUT_SRC_NODE, size);
  443. }
  444. inline void dsacache::Cache::Flush(const int node) {
  445. // this lambda is used because below we have two code paths that
  446. // flush nodes, either one single or all successively
  447. const auto FlushNode = [](std::unordered_map<uint8_t*,CacheData>& map) {
  448. // begin at the front of the map
  449. auto it = map.begin();
  450. // loop until we reach the end of the map
  451. while (it != map.end()) {
  452. // if the iterator points to an inactive element
  453. // then we may erase it
  454. if (it->second.GetRefCount() <= 1) {
  455. // erase the iterator from the map
  456. map.erase(it);
  457. // as the erasure invalidated out iterator
  458. // we must start at the beginning again
  459. it = map.begin();
  460. }
  461. else {
  462. // if element is active just move over to the next one
  463. it++;
  464. }
  465. }
  466. };
  467. // we require exclusive lock as we modify the cache state
  468. // node == -1 means that cache on all nodes should be flushed
  469. if (node == -1) {
  470. for (auto& nc : cache_state_) {
  471. std::unique_lock<std::shared_mutex> lock(nc.second->cache_mutex_);
  472. FlushNode(nc.second->node_cache_state_);
  473. }
  474. }
  475. else {
  476. std::unique_lock<std::shared_mutex> lock(cache_state_[node]->cache_mutex_);
  477. FlushNode(cache_state_[node]->node_cache_state_);
  478. }
  479. }
  480. inline std::unique_ptr<dsacache::CacheData> dsacache::Cache::GetFromCache(uint8_t* src, const size_t size, const int dst_node) {
  481. // the best situation is if this data is already cached
  482. // which we check in an unnamed block in which the cache
  483. // is locked for reading to prevent another thread
  484. // from marking the element we may find as unused and
  485. // clearing it
  486. LockedNodeCacheState* local_cache_state = cache_state_[dst_node];
  487. // lock the cache state in shared-mode because we read
  488. std::shared_lock<std::shared_mutex> lock(local_cache_state->cache_mutex_);
  489. // search for the data in our cache state structure at the given node
  490. const auto search = local_cache_state->node_cache_state_.find(src);
  491. // if the data is in our structure we continue
  492. if (search != local_cache_state->node_cache_state_.end()) {
  493. // now check whether the sizes match
  494. if (search->second.GetSize() >= size) {
  495. // return a unique copy of the entry which uses the object
  496. // lifetime and destructor to safely handle deallocation
  497. return std::move(std::make_unique<CacheData>(search->second));
  498. }
  499. else {
  500. // if the sizes missmatch then we clear the current entry from cache
  501. // which will cause its deletion only after the last possible outside
  502. // reference is also destroyed
  503. local_cache_state->node_cache_state_.erase(search);
  504. }
  505. }
  506. return nullptr;
  507. }
  508. void dsacache::Cache::Invalidate(uint8_t* data) {
  509. // as the cache is modified we must obtain a unique writers lock
  510. // loop through all per-node-caches available
  511. for (auto node : cache_state_) {
  512. std::unique_lock<std::shared_mutex> lock(node.second->cache_mutex_);
  513. // search for an entry for the given data pointer
  514. auto search = node.second->node_cache_state_.find(data);
  515. if (search != node.second->node_cache_state_.end()) {
  516. // if the data is represented in-cache
  517. // then it will be erased to re-trigger
  518. // caching on next access
  519. node.second->node_cache_state_.erase(search);
  520. }
  521. }
  522. }
  523. inline dsacache::Cache::~Cache() {
  524. for (auto node : cache_state_) {
  525. node.second->~LockedNodeCacheState();
  526. numa_free(reinterpret_cast<void*>(node.second), sizeof(LockedNodeCacheState));
  527. }
  528. }
  529. inline dsacache::CacheData::CacheData(uint8_t* data, const size_t size) {
  530. src_ = data;
  531. size_ = size;
  532. delete_ = false;
  533. active_ = new std::atomic<int32_t>(1);
  534. cache_ = new std::atomic<uint8_t*>(data);
  535. handlers_ = new std::atomic<std::vector<dml_handler>*>();
  536. incomplete_cache_ = new uint8_t*(nullptr);
  537. }
  538. inline dsacache::CacheData::CacheData(const dsacache::CacheData& other) {
  539. // we copy the ptr to the global atomic reference counter
  540. // and increase the amount of active references
  541. active_ = other.active_;
  542. const int current_active = active_->fetch_add(1);
  543. src_ = other.src_;
  544. size_ = other.size_;
  545. cache_ = other.cache_;
  546. flags_ = other.flags_;
  547. incomplete_cache_ = other.incomplete_cache_;
  548. handlers_ = other.handlers_;
  549. invalid_handlers_ = other.invalid_handlers_;
  550. }
  551. inline dsacache::CacheData::~CacheData() {
  552. // due to fetch_sub returning the preivously held value
  553. // we must subtract one locally to get the current value
  554. const int32_t v = active_->fetch_sub(1) - 1;
  555. // if the returned value is zero or lower
  556. // then we must execute proper deletion
  557. // as this was the last reference
  558. if (v == 0) {
  559. // on deletion we must ensure that all offloaded
  560. // operations have completed successfully
  561. // for this we must unset the possibly active
  562. // flag for weak waiting as we wish completion
  563. // guarantees afterwards
  564. flags_ = UnsetFlag(flags_, FLAG_WAIT_WEAK);
  565. WaitOnCompletion();
  566. // only then can we deallocate the memory
  567. Deallocate();
  568. delete active_;
  569. delete cache_;
  570. delete handlers_;
  571. delete incomplete_cache_;
  572. }
  573. }
  574. inline void dsacache::CacheData::Deallocate() {
  575. // although deallocate should only be called from
  576. // a safe context to do so, it can not hurt to
  577. // defensively perform the operation atomically
  578. // and check for incomplete cache if no deallocation
  579. // takes place for the retrieved local cache
  580. uint8_t* cache_local = cache_->exchange(nullptr);
  581. if (cache_local != nullptr && delete_) numa_free(cache_local, size_);
  582. else if (*incomplete_cache_ != nullptr) numa_free(*incomplete_cache_, size_);
  583. else;
  584. }
  585. inline void dsacache::CacheData::WaitOnCompletion() {
  586. // first check if waiting is even neccessary as a valid
  587. // cache pointer signals that no waiting is to be performed
  588. if (cache_->load() != nullptr) {
  589. return;
  590. }
  591. // then check if the handlers are available
  592. handlers_->wait(nullptr);
  593. // exchange the global handlers pointer with nullptr to have a local
  594. // copy - this signals that this thread is the sole owner and therefore
  595. // responsible for waiting for them. we can not set to nullptr here but
  596. // set to secondary invalid value in order to prevent deadlocks from
  597. // the above waiting construct, where threads may miss the short period
  598. // in which the handlers are not nullptr. we could use double width cas
  599. // which however is more expensive and therefore introduce the second
  600. // invalid state to solve the aba occurring here
  601. // see https://en.wikipedia.org/wiki/ABA_problem for more info
  602. std::vector<dml_handler>* local_handlers = handlers_->exchange(invalid_handlers_);
  603. // ensure that no other thread snatched the handlers before us
  604. // and in case one did, wait again and then return
  605. if (local_handlers == invalid_handlers_) {
  606. cache_->wait(nullptr);
  607. return;
  608. }
  609. // at this point we are responsible for waiting for the handlers
  610. // and handling any error that comes through them gracefully
  611. bool error = false;
  612. for (auto& handler : *local_handlers) {
  613. if (CheckFlag(flags_, FLAG_WAIT_WEAK) && !handler.is_finished()) {
  614. std::cerr << "[x] WEAK WAIT FAIL!" << std::endl;
  615. handlers_->store(local_handlers);
  616. return;
  617. }
  618. auto result = handler.get();
  619. if (result.status != dml::status_code::ok) {
  620. // if one of the copy tasks failed we abort the whole task
  621. // after all operations are completed on it
  622. std::cerr << "[x] ERROR (" << dml::StatusCodeToString(result.status) << ") FOUND FOR TASK IN WAIT!" << std::endl;
  623. error = true;
  624. }
  625. }
  626. // at this point all handlers have been waited for
  627. // and therefore may be decomissioned
  628. delete local_handlers;
  629. // handle errors now by aborting the cache
  630. if (error) {
  631. cache_->store(src_);
  632. numa_free(*incomplete_cache_, size_);
  633. delete_ = false;
  634. *incomplete_cache_ = nullptr;
  635. }
  636. else {
  637. cache_->store(*incomplete_cache_);
  638. }
  639. // notify all waiting threads so they wake up quickly
  640. cache_->notify_all();
  641. handlers_->notify_all();
  642. }
  643. void dsacache::CacheData::SetTaskHandlersAndCache(uint8_t* cache, std::vector<dml_handler>* handlers) {
  644. *incomplete_cache_ = cache;
  645. handlers_->store(handlers);
  646. handlers_->notify_one();
  647. }
  648. void dsacache::CacheData::Init(std::vector<dml_handler>* invalid_handlers) {
  649. cache_->store(nullptr);
  650. delete_ = true;
  651. invalid_handlers_ = invalid_handlers;
  652. }