|
|
@ -22,7 +22,6 @@ using aggregation = Aggregation<uint64_t, Sum, load_mode::Stream>; |
|
|
|
|
|
|
|
dsacache::Cache CACHE_; |
|
|
|
|
|
|
|
std::array<std::atomic<int32_t>, GROUP_COUNT> PREFETCHED_CHUNKS_; |
|
|
|
std::vector<std::barrier<NopStruct>*> BARRIERS_; |
|
|
|
std::shared_future<void> LAUNCH_; |
|
|
|
|
|
|
@ -31,13 +30,16 @@ uint64_t* DATA_B_; |
|
|
|
uint16_t* MASK_A_; |
|
|
|
uint64_t* DATA_DST_; |
|
|
|
|
|
|
|
// if more b than j -> perform b normal, subsplit j
|
|
|
|
// if more j than b -> subsplit b like it is now
|
|
|
|
void scan_b(size_t gid, size_t tid) { |
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)].clear(); |
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)].resize(1); |
|
|
|
|
|
|
|
LAUNCH_.wait(); |
|
|
|
|
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_BEGIN] = std::chrono::steady_clock::now(); |
|
|
|
|
|
|
|
template<size_t TC_CACHING> |
|
|
|
void caching(size_t gid, size_t tid) { |
|
|
|
constexpr size_t VIRT_TID_INCREMENT = TC_CACHING / TC_AGGRJ; |
|
|
|
constexpr size_t SUBCHUNK_THREAD_RATIO = TC_AGGRJ / (TC_CACHING == 0 ? 1 : TC_CACHING); |
|
|
|
constexpr size_t VIRT_TID_INCREMENT = TC_SCANB / TC_AGGRJ; |
|
|
|
constexpr size_t SUBCHUNK_THREAD_RATIO = TC_AGGRJ / (TC_SCANB == 0 ? 1 : TC_SCANB); |
|
|
|
constexpr bool CACHE_SUBCHUNKING = SUBCHUNK_THREAD_RATIO > 1; |
|
|
|
constexpr bool CACHE_OVERCHUNKING = VIRT_TID_INCREMENT > 1; |
|
|
|
|
|
|
@ -55,9 +57,6 @@ void caching(size_t gid, size_t tid) { |
|
|
|
uint64_t* sub_chunk_ptr = &chunk_ptr[j * SUBCHUNK_SIZE_ELEMENTS]; |
|
|
|
|
|
|
|
CACHE_.Access(reinterpret_cast<uint8_t*>(sub_chunk_ptr), SUBCHUNK_SIZE_B); |
|
|
|
|
|
|
|
PREFETCHED_CHUNKS_[gid]++; |
|
|
|
PREFETCHED_CHUNKS_[gid].notify_one(); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
@ -68,9 +67,6 @@ void caching(size_t gid, size_t tid) { |
|
|
|
uint64_t* chunk_ptr = get_chunk<TC_AGGRJ>(DATA_B_, chunk_index, tid_virt); |
|
|
|
|
|
|
|
CACHE_.Access(reinterpret_cast<uint8_t*>(chunk_ptr), CHUNK_SIZE_B); |
|
|
|
|
|
|
|
PREFETCHED_CHUNKS_[gid]++; |
|
|
|
PREFETCHED_CHUNKS_[gid].notify_one(); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
@ -80,24 +76,10 @@ void caching(size_t gid, size_t tid) { |
|
|
|
uint64_t* chunk_ptr = get_chunk<TC_SCANB>(DATA_B_, chunk_index, tid); |
|
|
|
|
|
|
|
CACHE_.Access(reinterpret_cast<uint8_t*>(chunk_ptr), CHUNK_SIZE_B); |
|
|
|
|
|
|
|
PREFETCHED_CHUNKS_[gid]++; |
|
|
|
PREFETCHED_CHUNKS_[gid].notify_one(); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
void scan_b(size_t gid, size_t tid) { |
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)].clear(); |
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)].resize(1); |
|
|
|
|
|
|
|
LAUNCH_.wait(); |
|
|
|
|
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_BEGIN] = std::chrono::steady_clock::now(); |
|
|
|
|
|
|
|
caching<TC_SCANB>(gid, tid); |
|
|
|
|
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_WAIT] = std::chrono::steady_clock::now(); |
|
|
|
THREAD_TIMING_[SCANB_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_END] = std::chrono::steady_clock::now(); |
|
|
|
} |
|
|
@ -117,13 +99,13 @@ void scan_a(size_t gid, size_t tid) { |
|
|
|
|
|
|
|
filter::apply_same(mask_ptr, nullptr, chunk_ptr, CMP_A, CHUNK_SIZE_B / TC_SCANA); |
|
|
|
|
|
|
|
BARRIERS_[gid]->arrive_and_wait(); |
|
|
|
} |
|
|
|
|
|
|
|
THREAD_TIMING_[SCANA_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_WAIT] = std::chrono::steady_clock::now(); |
|
|
|
THREAD_TIMING_[SCANA_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_END] = std::chrono::steady_clock::now(); |
|
|
|
|
|
|
|
BARRIERS_[gid]->arrive_and_drop(); |
|
|
|
|
|
|
|
THREAD_TIMING_[SCANA_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_END] = std::chrono::steady_clock::now(); |
|
|
|
} |
|
|
|
|
|
|
|
void aggr_j(size_t gid, size_t tid) { |
|
|
@ -132,19 +114,19 @@ void aggr_j(size_t gid, size_t tid) { |
|
|
|
CACHE_HITS_[UniqueIndex(gid,tid)] = 0; |
|
|
|
|
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)].clear(); |
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)].resize(RUN_COUNT); |
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)].resize(1); |
|
|
|
|
|
|
|
__m512i aggregator = aggregation::OP::zero(); |
|
|
|
|
|
|
|
LAUNCH_.wait(); |
|
|
|
|
|
|
|
BARRIERS_[gid]->arrive_and_wait(); |
|
|
|
for (size_t i = 0; i < RUN_COUNT; i++) { |
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)][i][TIME_STAMP_BEGIN] = std::chrono::steady_clock::now(); |
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_BEGIN] = std::chrono::steady_clock::now(); |
|
|
|
|
|
|
|
BARRIERS_[gid]->arrive_and_drop(); |
|
|
|
|
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)][i][TIME_STAMP_WAIT] = std::chrono::steady_clock::now(); |
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_WAIT] = std::chrono::steady_clock::now(); |
|
|
|
|
|
|
|
for (size_t i = 0; i < RUN_COUNT; i++) { |
|
|
|
const size_t chunk_index = get_chunk_index(gid, i); |
|
|
|
uint64_t* chunk_ptr = get_chunk<TC_AGGRJ>(DATA_B_, chunk_index, tid); |
|
|
|
uint16_t* mask_ptr_a = get_mask<TC_AGGRJ>(MASK_A_, chunk_index, tid); |
|
|
@ -174,10 +156,9 @@ BARRIERS_[gid]->arrive_and_wait(); |
|
|
|
uint64_t tmp = _mm512_reduce_add_epi64(aggregator); |
|
|
|
aggregator = aggregation::apply_masked(aggregator, data_ptr, mask_ptr_a, SUBCHUNK_SIZE_B); |
|
|
|
|
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)][i][TIME_STAMP_END] = std::chrono::steady_clock::now(); |
|
|
|
} |
|
|
|
|
|
|
|
BARRIERS_[gid]->arrive_and_drop(); |
|
|
|
THREAD_TIMING_[AGGRJ_TIMING_INDEX][UniqueIndex(gid,tid)][0][TIME_STAMP_END] = std::chrono::steady_clock::now(); |
|
|
|
|
|
|
|
aggregation::happly(&DATA_DST_[UniqueIndex(gid,tid)], aggregator); |
|
|
|
} |
|
|
|