Rename LeapFrog to Leapfrog
[junction.git] / samples / MapScalabilityTests / MapScalabilityTests.cpp
1 /*------------------------------------------------------------------------
2   Junction: Concurrent data structures in C++
3   Copyright (c) 2016 Jeff Preshing
4
5   Distributed under the Simplified BSD License.
6   Original location: https://github.com/preshing/junction
7
8   This software is distributed WITHOUT ANY WARRANTY; without even the
9   implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
10   See the LICENSE file for more information.
11 ------------------------------------------------------------------------*/
12
13 #include <junction/Core.h>
14 #include <turf/CPUTimer.h>
15 #include <turf/Util.h>
16 #include <turf/extra/UniqueSequence.h>
17 #include <turf/extra/JobDispatcher.h>
18 #include <turf/extra/Options.h>
19 #include <junction/extra/MapAdapter.h>
20 #include <algorithm>
21 #include <vector>
22
23 using namespace turf::intTypes;
24 typedef junction::extra::MapAdapter MapAdapter;
25
26 static const ureg NumKeysPerThread = 2000;
27 static const ureg DefaultReadsPerWrite = 4;
28 static const ureg DefaultItersPerChunk = 10000;
29 static const ureg DefaultChunks = 200;
30 static const u32 Prime = 0x4190ab09;
31
32 struct SharedState {
33     MapAdapter& adapter;
34     MapAdapter::Map* map;
35     ureg numKeysPerThread;
36     ureg numThreads;
37     ureg readsPerWrite;
38     ureg itersPerChunk;
39     turf::extra::SpinKicker spinKicker;
40     turf::Atomic<u32> doneFlag;
41
42     SharedState(MapAdapter& adapter, ureg numKeysPerThread, ureg readsPerWrite, ureg itersPerChunk)
43         : adapter(adapter), map(NULL), numKeysPerThread(numKeysPerThread), readsPerWrite(readsPerWrite),
44           itersPerChunk(itersPerChunk) {
45         doneFlag.storeNonatomic(0);
46         numThreads = 0;
47     }
48 };
49
50 class ThreadState {
51 public:
52     SharedState& m_shared;
53     MapAdapter::ThreadContext m_threadCtx;
54     ureg m_threadIndex;
55     u32 m_rangeLo;
56     u32 m_rangeHi;
57
58     u32 m_addIndex;
59     u32 m_removeIndex;
60
61     struct Stats {
62         ureg mapOpsDone;
63         double duration;
64
65         Stats() {
66             mapOpsDone = 0;
67             duration = 0;
68         }
69
70         Stats& operator+=(const Stats& other) {
71             mapOpsDone += other.mapOpsDone;
72             duration += other.duration;
73             return *this;
74         }
75
76         bool operator<(const Stats& other) const {
77             return duration < other.duration;
78         }
79     };
80
81     Stats m_stats;
82
83     ThreadState(SharedState& shared, ureg threadIndex, u32 rangeLo, u32 rangeHi)
84         : m_shared(shared), m_threadCtx(shared.adapter, threadIndex) {
85         m_threadIndex = threadIndex;
86         m_rangeLo = rangeLo;
87         m_rangeHi = rangeHi;
88         m_addIndex = rangeLo;
89         m_removeIndex = rangeLo;
90     }
91
92     void registerThread() {
93         m_threadCtx.registerThread();
94     }
95
96     void unregisterThread() {
97         m_threadCtx.unregisterThread();
98     }
99
100     void initialPopulate() {
101         TURF_ASSERT(m_addIndex == m_removeIndex);
102         MapAdapter::Map* map = m_shared.map;
103         for (ureg i = 0; i < m_shared.numKeysPerThread; i++) {
104             u32 key = m_addIndex * Prime;
105             if (key >= 2)
106                 map->set(key, (void*) uptr(key));
107             if (++m_addIndex == m_rangeHi)
108                 m_addIndex = m_rangeLo;
109         }
110     }
111
112     void run() {
113         MapAdapter::Map* map = m_shared.map;
114         turf::CPUTimer::Converter converter;
115         Stats stats;
116         ureg lookupIndex = m_rangeLo;
117         ureg remaining = m_shared.itersPerChunk;
118         if (m_threadIndex == 0)
119             m_shared.spinKicker.kick(m_shared.numThreads - 1);
120         else {
121             remaining = ~u32(0);
122             m_shared.spinKicker.waitForKick();
123         }
124
125         // ---------
126         turf::CPUTimer::Point start = turf::CPUTimer::get();
127         for (; remaining > 0; remaining--) {
128             // Add
129             if (m_shared.doneFlag.load(turf::Relaxed))
130                 break;
131             u32 key = m_addIndex * Prime;
132             if (key >= 2) {
133                 map->set(key, (void*) uptr(key));
134                 stats.mapOpsDone++;
135             }
136             if (++m_addIndex == m_rangeHi)
137                 m_addIndex = m_rangeLo;
138
139             // Lookup
140             if (s32(lookupIndex - m_removeIndex) < 0)
141                 lookupIndex = m_removeIndex;
142             for (ureg l = 0; l < m_shared.readsPerWrite; l++) {
143                 if (m_shared.doneFlag.load(turf::Relaxed))
144                     break;
145                 key = lookupIndex * Prime;
146                 if (key >= 2) {
147                     volatile void* value = map->get(key);
148                     TURF_UNUSED(value);
149                     stats.mapOpsDone++;
150                 }
151                 if (++lookupIndex == m_rangeHi)
152                     lookupIndex = m_rangeLo;
153                 if (lookupIndex == m_addIndex)
154                     lookupIndex = m_removeIndex;
155             }
156
157             // Remove
158             if (m_shared.doneFlag.load(turf::Relaxed))
159                 break;
160             key = m_removeIndex * Prime;
161             if (key >= 2) {
162                 map->erase(key);
163                 stats.mapOpsDone++;
164             }
165             if (++m_removeIndex == m_rangeHi)
166                 m_removeIndex = m_rangeLo;
167
168             // Lookup
169             if (s32(lookupIndex - m_removeIndex) < 0)
170                 lookupIndex = m_removeIndex;
171             for (ureg l = 0; l < m_shared.readsPerWrite; l++) {
172                 if (m_shared.doneFlag.load(turf::Relaxed))
173                     break;
174                 key = lookupIndex * Prime;
175                 if (key >= 2) {
176                     volatile void* value = map->get(key);
177                     TURF_UNUSED(value);
178                     stats.mapOpsDone++;
179                 }
180                 if (++lookupIndex == m_rangeHi)
181                     lookupIndex = m_rangeLo;
182                 if (lookupIndex == m_addIndex)
183                     lookupIndex = m_removeIndex;
184             }
185         }
186         if (m_threadIndex == 0)
187             m_shared.doneFlag.store(1, turf::Relaxed);
188         m_threadCtx.update();
189         turf::CPUTimer::Point end = turf::CPUTimer::get();
190         // ---------
191
192         stats.duration = converter.toSeconds(end - start);
193         m_stats = stats;
194     }
195 };
196
197 static const turf::extra::Option Options[] = {
198     {"readsPerWrite", 'r', true, "number of reads per write"},
199     {"itersPerChunk", 'i', true, "number of iterations per chunk"},
200     {"chunks", 'c', true, "number of chunks to execute"},
201     {"keepChunkFraction", 'k', true, "threshold fraction of chunk timings to keep"},
202 };
203
204 int main(int argc, const char** argv) {
205     turf::extra::Options options(Options, TURF_STATIC_ARRAY_SIZE(Options));
206     options.parse(argc, argv);
207     ureg readsPerWrite = options.getInteger("readsPerWrite", DefaultReadsPerWrite);
208     ureg itersPerChunk = options.getInteger("itersPerChunk", DefaultItersPerChunk);
209     ureg chunks = options.getInteger("chunks", DefaultChunks);
210     double keepChunkFraction = options.getDouble("keepChunkFraction", 1.0);
211
212     turf::extra::JobDispatcher dispatcher;
213     ureg numCores = dispatcher.getNumPhysicalCores();
214     TURF_ASSERT(numCores > 0);
215     MapAdapter adapter(numCores);
216
217     // Create shared state and register first thread
218     SharedState shared(adapter, NumKeysPerThread, readsPerWrite, itersPerChunk);
219     std::vector<ThreadState> threads;
220     threads.reserve(numCores);
221     for (ureg t = 0; t < numCores; t++) {
222         u32 rangeLo = 0xffffffffu / numCores * t + 1;
223         u32 rangeHi = 0xffffffffu / numCores * (t + 1) + 1;
224         threads.emplace_back(shared, t, rangeLo, rangeHi);
225     }
226     dispatcher.kickOne(0, &ThreadState::registerThread, threads[0]);
227
228     {
229         // Create the map and populate it entirely from main thread
230         MapAdapter::Map map(MapAdapter::getInitialCapacity(numCores * NumKeysPerThread));
231         shared.map = &map;
232         for (ureg t = 0; t < numCores; t++) {
233             threads[t].initialPopulate();
234         }
235
236         printf("{\n");
237         printf("'mapType': '%s',\n", MapAdapter::MapName);
238         printf("'population': %d,\n", (int) (numCores * NumKeysPerThread));
239         printf("'readsPerWrite': %d,\n", (int) readsPerWrite);
240         printf("'itersPerChunk': %d,\n", (int) itersPerChunk);
241         printf("'chunks': %d,\n", (int) chunks);
242         printf("'keepChunkFraction': %f,\n", keepChunkFraction);
243         printf("'labels': ('numThreads', 'mapOpsDone', 'totalTime'),\n"), printf("'points': [\n");
244         for (shared.numThreads = 1; shared.numThreads <= numCores; shared.numThreads++) {
245             if (shared.numThreads > 1) {
246                 // Spawn and register a new thread
247                 dispatcher.kickOne(shared.numThreads - 1, &ThreadState::registerThread, threads[shared.numThreads - 1]);
248             }
249
250             std::vector<ThreadState::Stats> kickTotals;
251             for (ureg c = 0; c < chunks; c++) {
252                 shared.doneFlag.storeNonatomic(false);
253                 dispatcher.kickMulti(&ThreadState::run, &threads[0], shared.numThreads);
254
255                 ThreadState::Stats kickTotal;
256                 for (ureg t = 0; t < shared.numThreads; t++)
257                     kickTotal += threads[t].m_stats;
258                 kickTotals.push_back(kickTotal);
259             }
260
261             std::sort(kickTotals.begin(), kickTotals.end());
262             ThreadState::Stats totals;
263             for (ureg t = 0; t < ureg(kickTotals.size() * keepChunkFraction); t++) {
264                 totals += kickTotals[t];
265             }
266
267             printf("    (%d, %d, %f),\n", int(shared.numThreads), int(totals.mapOpsDone), totals.duration);
268         }
269         printf("],\n");
270         printf("}\n");
271
272         shared.map = NULL;
273     }
274
275     dispatcher.kickMulti(&ThreadState::unregisterThread, &threads[0], threads.size());
276     return 0;
277 }