@Test public void testLABThreading() throws Exception { Configuration conf = new Configuration(); MultithreadedTestUtil.TestContext ctx = new MultithreadedTestUtil.TestContext(conf); final AtomicInteger totalAllocated = new AtomicInteger(); final MemStoreLAB mslab = new MemStoreLAB(); List<List<AllocRecord>> allocations = Lists.newArrayList(); for (int i = 0; i < 10; i++) { final List<AllocRecord> allocsByThisThread = Lists.newLinkedList(); allocations.add(allocsByThisThread); TestThread t = new MultithreadedTestUtil.RepeatingTestThread(ctx) { private Random r = new Random(); @Override public void doAnAction() throws Exception { int size = r.nextInt(1000); Allocation alloc = mslab.allocateBytes(size); totalAllocated.addAndGet(size); allocsByThisThread.add(new AllocRecord(alloc, size)); } }; ctx.addThread(t); } ctx.startThreads(); while (totalAllocated.get() < 50*1024*1024 && ctx.shouldRun()) { Thread.sleep(10); } ctx.stop(); Map<byte[], Map<Integer, AllocRecord>> mapsByChunk = Maps.newHashMap(); int sizeCounted = 0; for (AllocRecord rec : Iterables.concat(allocations)) { sizeCounted += rec.size; if (rec.size == 0) continue; Map<Integer, AllocRecord> mapForThisByteArray = mapsByChunk.get(rec.alloc.getData()); if (mapForThisByteArray == null) { mapForThisByteArray = Maps.newTreeMap(); mapsByChunk.put(rec.alloc.getData(), mapForThisByteArray); } AllocRecord oldVal = mapForThisByteArray.put(rec.alloc.getOffset(), rec); assertNull("Already had an entry " + oldVal + " for allocation " + rec, oldVal); } assertEquals("Sanity check test", sizeCounted, totalAllocated.get()); for (Map<Integer, AllocRecord> allocsInChunk : mapsByChunk.values()) { int expectedOff = 0; for (AllocRecord alloc : allocsInChunk.values()) { assertEquals(expectedOff, alloc.alloc.getOffset()); assertTrue("Allocation " + alloc + " overruns buffer", alloc.alloc.getOffset() + alloc.size <= alloc.alloc.getData().length); expectedOff += alloc.size; } } }
public Allocation allocateBytes(int size) { Preconditions.checkArgument(size >= 0, "negative size"); if (size > maxAlloc) { return null; } while (true) { Chunk c = getOrMakeChunk(); int allocOffset = c.alloc(size); if (allocOffset != -1) { return new Allocation(c.data, allocOffset); } tryRetireChunk(c); } }
MemStoreLAB { public Allocation allocateBytes(int size) { Preconditions.checkArgument(size >= 0, "negative size"); if (size > maxAlloc) { return null; } while (true) { Chunk c = getOrMakeChunk(); int allocOffset = c.alloc(size); if (allocOffset != -1) { return new Allocation(c.data, allocOffset); } tryRetireChunk(c); } } }
MemStoreLAB { public Allocation allocateBytes(int size) { Preconditions.checkArgument(size >= 0, "negative size"); if (size > maxAlloc) { return null; } while (true) { Chunk c = getOrMakeChunk(); int allocOffset = c.alloc(size); if (allocOffset != -1) { return new Allocation(c.data, allocOffset); } tryRetireChunk(c); } } MemStoreLAB();  MemStoreLAB(Configuration conf); }
MemStoreLAB { public Allocation allocateBytes(int size) { Preconditions.checkArgument(size >= 0, "negative size"); if (size > maxAlloc) { return null; } while (true) { Chunk c = getOrMakeChunk(); int allocOffset = c.alloc(size); if (allocOffset != -1) { return new Allocation(c.data, allocOffset); } tryRetireChunk(c); } } MemStoreLAB();  MemStoreLAB(Configuration conf); Allocation allocateBytes(int size); }
MemStoreLAB { public Allocation allocateBytes(int size) { Preconditions.checkArgument(size >= 0, "negative size"); if (size > maxAlloc) { return null; } while (true) { Chunk c = getOrMakeChunk(); int allocOffset = c.alloc(size); if (allocOffset != -1) { return new Allocation(c.data, allocOffset); } tryRetireChunk(c); } } MemStoreLAB();  MemStoreLAB(Configuration conf); Allocation allocateBytes(int size);  }