/**
* @throws Exception if failed.
*/
public void testMultiThreaded() throws Exception {
GridUnsafeMemory mem = new GridUnsafeMemory(0);
X.println("___ Started");
Random rnd = new GridRandom();
for (int i = 0; i < 20; i++) {
GridHadoopJobInfo job = new JobInfo();
final GridHadoopTaskContext taskCtx = new TaskContext();
final GridHadoopConcurrentHashMultimap m = new GridHadoopConcurrentHashMultimap(job, mem, 16);
final ConcurrentMap<Integer, Collection<Integer>> mm = new ConcurrentHashMap<>();
X.println("___ MT");
multithreaded(new Callable<Object>() {
@Override public Object call() throws Exception {
X.println("___ TH in");
Random rnd = new GridRandom();
IntWritable key = new IntWritable();
IntWritable val = new IntWritable();
GridHadoopMultimap.Adder a = m.startAdding(taskCtx);
for (int i = 0; i < 50000; i++) {
int k = rnd.nextInt(32000);
int v = rnd.nextInt();
key.set(k);
val.set(v);
a.write(key, val);
Collection<Integer> list = mm.get(k);
if (list == null) {
list = new ConcurrentLinkedQueue<>();
Collection<Integer> old = mm.putIfAbsent(k, list);
if (old != null)
list = old;
}
list.add(v);
}
a.close();
X.println("___ TH out");
return null;
}
}, 3 + rnd.nextInt(27));
X.println("___ Check: " + m.capacity());
assertEquals(mm.size(), m.keys());
assertTrue(m.capacity() > 32000);
GridHadoopTaskInput in = m.input(taskCtx);
while (in.next()) {
IntWritable key = (IntWritable) in.key();
Iterator<?> valsIter = in.values();
Collection<Integer> vals = mm.remove(key.get());
assertNotNull(vals);
while (valsIter.hasNext()) {
IntWritable val = (IntWritable) valsIter.next();
assertTrue(vals.remove(val.get()));
}
assertTrue(vals.isEmpty());
}
in.close();
m.close();
assertEquals(0, mem.allocatedSize());
}
}