@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()); } }
@Override public void setNumBuffers(int numBuffers) throws IOException { int numExcessBuffers; synchronized (availableMemorySegments) { checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least %s buffers, but tried to set to %s", numberOfRequiredMemorySegments, numBuffers); if (numBuffers > maxNumberOfMemorySegments) { currentPoolSize = maxNumberOfMemorySegments; } else { currentPoolSize = numBuffers; } returnExcessMemorySegments(); numExcessBuffers = numberOfRequestedMemorySegments - currentPoolSize; } if (owner.isPresent() && numExcessBuffers > 0) { owner.get().releaseMemory(numExcessBuffers); } }
LocalBufferPool implements BufferPool { @Override public void setNumBuffers(int numBuffers) throws IOException { int numExcessBuffers; synchronized (availableMemorySegments) { checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least %s buffers, but tried to set to %s", numberOfRequiredMemorySegments, numBuffers); if (numBuffers > maxNumberOfMemorySegments) { currentPoolSize = maxNumberOfMemorySegments; } else { currentPoolSize = numBuffers; } returnExcessMemorySegments(); numExcessBuffers = numberOfRequestedMemorySegments - currentPoolSize; } if (owner.isPresent() && numExcessBuffers > 0) { owner.get().releaseMemory(numExcessBuffers); } } }
LocalBufferPool implements BufferPool { @Override public void setNumBuffers(int numBuffers) throws IOException { int numExcessBuffers; synchronized (availableMemorySegments) { checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least %s buffers, but tried to set to %s", numberOfRequiredMemorySegments, numBuffers); if (numBuffers > maxNumberOfMemorySegments) { currentPoolSize = maxNumberOfMemorySegments; } else { currentPoolSize = numBuffers; } returnExcessMemorySegments(); numExcessBuffers = numberOfRequestedMemorySegments - currentPoolSize; } if (owner.isPresent() && numExcessBuffers > 0) { owner.get().releaseMemory(numExcessBuffers); } } LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments);   LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments,
			int maxNumberOfMemorySegments);   LocalBufferPool(
		NetworkBufferPool networkBufferPool,
		int numberOfRequiredMemorySegments,
		int maxNumberOfMemorySegments,
		Optional<BufferPoolOwner> owner); }
LocalBufferPool implements BufferPool { @Override public void setNumBuffers(int numBuffers) throws IOException { int numExcessBuffers; synchronized (availableMemorySegments) { checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least %s buffers, but tried to set to %s", numberOfRequiredMemorySegments, numBuffers); if (numBuffers > maxNumberOfMemorySegments) { currentPoolSize = maxNumberOfMemorySegments; } else { currentPoolSize = numBuffers; } returnExcessMemorySegments(); numExcessBuffers = numberOfRequestedMemorySegments - currentPoolSize; } if (owner.isPresent() && numExcessBuffers > 0) { owner.get().releaseMemory(numExcessBuffers); } } LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments);   LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments,
			int maxNumberOfMemorySegments);   LocalBufferPool(
		NetworkBufferPool networkBufferPool,
		int numberOfRequiredMemorySegments,
		int maxNumberOfMemorySegments,
		Optional<BufferPoolOwner> owner); @Override boolean isDestroyed(); @Override int getNumberOfRequiredMemorySegments(); @Override int getMaxNumberOfMemorySegments(); @Override int getNumberOfAvailableMemorySegments(); @Override int getNumBuffers(); @Override int bestEffortGetNumOfUsedBuffers(); @Override Buffer requestBuffer(); @Override Buffer requestBufferBlocking(); @Override BufferBuilder requestBufferBuilderBlocking(); @Override void recycle(MemorySegment segment); @Override void lazyDestroy(); @Override boolean addBufferListener(BufferListener listener); @Override void setNumBuffers(int numBuffers); @Override String toString(); }
LocalBufferPool implements BufferPool { @Override public void setNumBuffers(int numBuffers) throws IOException { int numExcessBuffers; synchronized (availableMemorySegments) { checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least %s buffers, but tried to set to %s", numberOfRequiredMemorySegments, numBuffers); if (numBuffers > maxNumberOfMemorySegments) { currentPoolSize = maxNumberOfMemorySegments; } else { currentPoolSize = numBuffers; } returnExcessMemorySegments(); numExcessBuffers = numberOfRequestedMemorySegments - currentPoolSize; } if (owner.isPresent() && numExcessBuffers > 0) { owner.get().releaseMemory(numExcessBuffers); } } LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments);   LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments,
			int maxNumberOfMemorySegments);   LocalBufferPool(
		NetworkBufferPool networkBufferPool,
		int numberOfRequiredMemorySegments,
		int maxNumberOfMemorySegments,
		Optional<BufferPoolOwner> owner); @Override boolean isDestroyed(); @Override int getNumberOfRequiredMemorySegments(); @Override int getMaxNumberOfMemorySegments(); @Override int getNumberOfAvailableMemorySegments(); @Override int getNumBuffers(); @Override int bestEffortGetNumOfUsedBuffers(); @Override Buffer requestBuffer(); @Override Buffer requestBufferBlocking(); @Override BufferBuilder requestBufferBuilderBlocking(); @Override void recycle(MemorySegment segment); @Override void lazyDestroy(); @Override boolean addBufferListener(BufferListener listener); @Override void setNumBuffers(int numBuffers); @Override String toString();  }