LocalBufferPool implements BufferPool { @Override public void setNumBuffers(int numBuffers) throws IOException { synchronized (availableMemorySegments) { checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least " + numberOfRequiredMemorySegments + " buffers, but tried to set to " + numBuffers + "."); currentPoolSize = numBuffers; returnExcessMemorySegments(); if (owner != null && numberOfRequestedMemorySegments > currentPoolSize) { owner.releaseMemory(numberOfRequestedMemorySegments - numBuffers); } } } LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments); @Override boolean isDestroyed(); @Override int getMemorySegmentSize(); @Override int getNumberOfRequiredMemorySegments(); @Override int getNumberOfAvailableMemorySegments(); @Override int getNumBuffers(); @Override void setBufferPoolOwner(BufferPoolOwner owner); @Override Buffer requestBuffer(); @Override Buffer requestBufferBlocking(); @Override void recycle(MemorySegment segment); @Override void lazyDestroy(); @Override boolean addListener(EventListener<Buffer> listener); @Override void setNumBuffers(int numBuffers); @Override String toString();  }
@Test(expected = IllegalArgumentException.class) public void testSetLessThanRequiredNumBuffers() throws IOException { localBufferPool.setNumBuffers(1); localBufferPool.setNumBuffers(0); }
@Test @SuppressWarnings("unchecked") public void testConcurrentRequestRecycle() throws ExecutionException, InterruptedException, IOException { int numConcurrentTasks = 128; int numBuffersToRequestPerTask = 1024; localBufferPool.setNumBuffers(numConcurrentTasks); Future<Boolean>[] taskResults = new Future[numConcurrentTasks]; for (int i = 0; i < numConcurrentTasks; i++) { taskResults[i] = executor.submit(new BufferRequesterTask(localBufferPool, numBuffersToRequestPerTask)); } for (int i = 0; i < numConcurrentTasks; i++) { assertTrue(taskResults[i].get()); } }