| 50 | const BalancedSplitter bs(instance_num, thread_num); |
| 51 | BlockingCounter bc(thread_num); |
| 52 | FOR_RANGE(int64_t, thread_id, 0, thread_num) { |
| 53 | const Range range = bs.At(thread_id); |
| 54 | Singleton<ThreadPool>::Get()->AddWork([=, &bc]() { |
| 55 | FOR_RANGE(int64_t, i, range.begin(), range.end()) { |
| 56 | T* in_ptr = tmp_buffer->mut_dptr<T>() + i * stride; |
| 57 | T* val_ptr = values->mut_dptr<T>() + i; |
| 58 | int64_t* ind_ptr = indices->mut_dptr<int64_t>() + i; |
| 59 | std::vector<std::pair<T, int64_t>> elements(stride); |
| 60 | T mode = 0; |
| 61 | int64_t mode_idx = 0; |
| 62 | int64_t temp_freq = 0; |
| 63 | int64_t max_freq = 0; |
| 64 | FOR_RANGE(int64_t, idx, 0, stride) { |
| 65 | elements[idx] = std::make_pair(*(in_ptr + idx), idx); |
| 66 | } |
| 67 | std::sort(elements.begin(), elements.end(), |
| 68 | [=](const auto& i, const auto& j) { return i.first < j.first; }); |
| 69 | FOR_RANGE(int64_t, idx, 0, stride) { |
| 70 | temp_freq++; |
| 71 | if ((idx == stride - 1) || (elements[idx].first != elements[idx + 1].first)) { |
| 72 | if (temp_freq > max_freq) { |
| 73 | mode = elements[idx].first; |
| 74 | mode_idx = elements[idx].second; |
| 75 | max_freq = temp_freq; |
| 76 | } |
| 77 | temp_freq = 0; |
| 78 | } |
| 79 | } |
| 80 | *val_ptr = mode; |
| 81 | *ind_ptr = mode_idx; |
| 82 | } |
| 83 | bc.Decrease(); |
| 84 | }); |
| 85 | } |
| 86 | bc.WaitForeverUntilCntEqualZero(); |
| 87 | } |
| 88 | bool AlwaysComputeWhenAllOutputsEmpty() const override { return false; } |