test_class.declarator: time = new MockTime()
test_class.modifier: private final
test_class.original_string: private final MockTime time = new MockTime();
test_class.type: MockTime
test_class.var_name: time
test_class.declarator: systemTime = new SystemTime()
test_class.modifier: private final
test_class.original_string: private final SystemTime systemTime = new SystemTime();
test_class.type: SystemTime
test_class.var_name: systemTime
test_class.declarator: metrics = new Metrics(time)
test_class.modifier: private final
test_class.original_string: private final Metrics metrics = new Metrics(time);
test_class.type: Metrics
test_class.var_name: metrics
test_class.declarator: maxBlockTimeMs = 2000
test_class.modifier: private final
test_class.original_string: private final long maxBlockTimeMs = 2000;
test_class.type: long
test_class.var_name: maxBlockTimeMs
test_class.declarator: metricGroup = "TestMetrics"
test_class.modifier: private final
test_class.original_string: private final String metricGroup = "TestMetrics";
test_class.type: String
test_class.var_name: metricGroup
test_class.file: clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java
test_class.identifier: BufferPoolTest
test_case.body: @Test
      public void testDelayedAllocation() throws Exception {
          BufferPool pool = new BufferPool(5 * 1024, 1024, metrics, time, metricGroup);
          ByteBuffer buffer = pool.allocate(1024, maxBlockTimeMs);
          CountDownLatch doDealloc = asyncDeallocate(pool, buffer);
          CountDownLatch allocation = asyncAllocate(pool, 5 * 1024);
          assertEquals("Allocation shouldn't have happened yet, waiting on memory.", 1L, allocation.getCount());
          doDealloc.countDown(); // return the memory
          assertTrue("Allocation should succeed soon after de-allocation", allocation.await(1, TimeUnit.SECONDS));
      }
test_case.class_method_signature: BufferPoolTest.testDelayedAllocation()
test_case.full_signature: @Test public void testDelayedAllocation()
test_case.identifier: testDelayedAllocation
test_case.invocations: allocate asyncDeallocate asyncAllocate assertEquals getCount countDown assertTrue await
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testDelayedAllocation()
focal_class.declarator: totalMemory
focal_class.modifier: private final
focal_class.original_string: private final long totalMemory;
focal_class.type: long
focal_class.var_name: totalMemory
focal_class.declarator: poolableSize
focal_class.modifier: private final
focal_class.original_string: private final int poolableSize;
focal_class.type: int
focal_class.var_name: poolableSize
focal_class.declarator: lock
focal_class.modifier: private final
focal_class.original_string: private final ReentrantLock lock;
focal_class.type: ReentrantLock
focal_class.var_name: lock
focal_class.declarator: free
focal_class.modifier: private final
focal_class.original_string: private final Deque<ByteBuffer> free;
focal_class.type: Deque<ByteBuffer>
focal_class.var_name: free
focal_class.declarator: waiters
focal_class.modifier: private final
focal_class.original_string: private final Deque<Condition> waiters;
focal_class.type: Deque<Condition>
focal_class.var_name: waiters
focal_class.declarator: availableMemory
focal_class.modifier: private
focal_class.original_string: private long availableMemory;
focal_class.type: long
focal_class.var_name: availableMemory
focal_class.declarator: metrics
focal_class.modifier: private final
focal_class.original_string: private final Metrics metrics;
focal_class.type: Metrics
focal_class.var_name: metrics
focal_class.declarator: time
focal_class.modifier: private final
focal_class.original_string: private final Time time;
focal_class.type: Time
focal_class.var_name: time
focal_class.declarator: waitTime
focal_class.modifier: private final
focal_class.original_string: private final Sensor waitTime;
focal_class.type: Sensor
focal_class.var_name: waitTime
focal_class.file: clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java
focal_class.identifier: BufferPool
focal_class.class_method_signature: BufferPool.BufferPool(long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName)
focal_class.full_signature: public  BufferPool(long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName)
focal_class.identifier: BufferPool
focal_class.modifiers: public
focal_class.parameters: (long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName)
focal_class.signature:  BufferPool(long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName)
focal_class.class_method_signature: BufferPool.allocate(int size, long maxTimeToBlockMs)
focal_class.full_signature: public ByteBuffer allocate(int size, long maxTimeToBlockMs)
focal_class.identifier: allocate
focal_class.modifiers: public
focal_class.parameters: (int size, long maxTimeToBlockMs)
focal_class.return: ByteBuffer
focal_class.signature: ByteBuffer allocate(int size, long maxTimeToBlockMs)
focal_class.class_method_signature: BufferPool.freeUp(int size)
focal_class.full_signature: private void freeUp(int size)
focal_class.identifier: freeUp
focal_class.modifiers: private
focal_class.parameters: (int size)
focal_class.return: void
focal_class.signature: void freeUp(int size)
focal_class.class_method_signature: BufferPool.deallocate(ByteBuffer buffer, int size)
focal_class.full_signature: public void deallocate(ByteBuffer buffer, int size)
focal_class.identifier: deallocate
focal_class.modifiers: public
focal_class.parameters: (ByteBuffer buffer, int size)
focal_class.return: void
focal_class.signature: void deallocate(ByteBuffer buffer, int size)
focal_class.class_method_signature: BufferPool.deallocate(ByteBuffer buffer)
focal_class.full_signature: public void deallocate(ByteBuffer buffer)
focal_class.identifier: deallocate
focal_class.modifiers: public
focal_class.parameters: (ByteBuffer buffer)
focal_class.return: void
focal_class.signature: void deallocate(ByteBuffer buffer)
focal_class.class_method_signature: BufferPool.availableMemory()
focal_class.full_signature: public long availableMemory()
focal_class.identifier: availableMemory
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: long
focal_class.signature: long availableMemory()
focal_class.class_method_signature: BufferPool.unallocatedMemory()
focal_class.full_signature: public long unallocatedMemory()
focal_class.identifier: unallocatedMemory
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: long
focal_class.signature: long unallocatedMemory()
focal_class.class_method_signature: BufferPool.queued()
focal_class.full_signature: public int queued()
focal_class.identifier: queued
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int queued()
focal_class.class_method_signature: BufferPool.poolableSize()
focal_class.full_signature: public int poolableSize()
focal_class.identifier: poolableSize
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int poolableSize()
focal_class.class_method_signature: BufferPool.totalMemory()
focal_class.full_signature: public long totalMemory()
focal_class.identifier: totalMemory
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: long
focal_class.signature: long totalMemory()
focal_class.class_method_signature: BufferPool.waiters()
focal_class.full_signature:  Deque<Condition> waiters()
focal_class.identifier: waiters
focal_class.parameters: ()
focal_class.return: Deque<Condition>
focal_class.signature: Deque<Condition> waiters()
focal_method.body: public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {
          if (size > this.totalMemory) {
              throw new IllegalArgumentException("Attempt to allocate " + size
                  + " bytes, but there is a hard limit of "
                  + this.totalMemory
                  + " on memory allocations.");
          }
  
          this.lock.lock();
          try {
              // check if we have a free buffer of the right size pooled
              // 校验是否有合适的小的空闲的buffer
              if (size == poolableSize && !this.free.isEmpty()) {
                  return this.free.pollFirst();
              }
  
              // now check if the request is immediately satisfiable with the
              // memory on hand or if we need to block
              // 校验现在的内存是否可以立即满足请求，或者是否需要阻塞
              int freeListSize = this.free.size() * this.poolableSize;
  
              // 如果可用内存+freeList大小大于申请大小
              if (this.availableMemory + freeListSize >= size) {
                  // we have enough unallocated or pooled memory to immediately
                  // satisfy the request
                  // 有足够的被释放或存放在池中的内存来立即满足请求
                  freeUp(size);
                  this.availableMemory -= size;
                  lock.unlock();
                  return ByteBuffer.allocate(size);
              } else {// 现在可用的内存大小无法满足
                  // we are out of memory and will have to block
                  // 需要的内存可能会导致溢出，所以需要阻塞
                  int accumulated = 0;
                  ByteBuffer buffer = null;
  
                  Condition moreMemory = this.lock.newCondition();
                  long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);
                  this.waiters.addLast(moreMemory);
  
                  // loop over and over until we have a buffer or have reserved
                  // enough memory to allocate one
                  // 循环等待，直到获取到申请到足够的内存
                  while (accumulated < size) {
                      long startWaitNs = time.nanoseconds();
                      long timeNs;
                      boolean waitingTimeElapsed;
                      try {
                          // 当前condition进行等待 remainingTimeToBlockNs 毫秒
                          waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
                      } catch (InterruptedException e) {
                          this.waiters.remove(moreMemory);
                          throw e;
                      } finally {
                          long endWaitNs = time.nanoseconds();
                          timeNs = Math.max(0L, endWaitNs - startWaitNs);
                          // 统计阻塞时间
                          this.waitTime.record(timeNs, time.milliseconds());
                      }
  
                      // 代表超时了
                      if (waitingTimeElapsed) {
                          this.waiters.remove(moreMemory);
                          throw new TimeoutException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms.");
                      }
  
                      // 没超时，剩余时间等于自己减去耗时
                      remainingTimeToBlockNs -= timeNs;
                      // check if we can satisfy this request from the free list,
                      // otherwise allocate memory
                      // 检查free list是否可以满足请求，不满足则申请内存
                      if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {
                          // just grab a buffer from the free list
                          // 和前面是一样的，从free list中取一个出来
                          buffer = this.free.pollFirst();
                          accumulated = size;
                      } else {
  
                          //todo: 走到这里说明申请的大小要大于poolableSize，或者free为空
                          // we'll need to allocate memory, but we may only get
                          // part of what we need on this iteration
                          // 需要申请内存，但是在这个循环可能只能从中获取需要内存的一部分，也就是说太大了，会再获取一次
                          // size：要申请的大小。
                          freeUp(size - accumulated);
  
                          int got = (int) Math.min(size - accumulated, this.availableMemory);
                          this.availableMemory -= got;
                          accumulated += got;
                      }
                  }
  
                  // remove the condition for this thread to let the next thread
                  // in line start getting memory
                  Condition removed = this.waiters.removeFirst();
                  if (removed != moreMemory) {
                      throw new IllegalStateException("Wrong condition: this shouldn't happen.");
                  }
  
                  // signal any additional waiters if there is more memory left
                  // over for them
                  // 通知其他waiters去拿内存
                  if (this.availableMemory > 0 || !this.free.isEmpty()) {
                      if (!this.waiters.isEmpty()) {
                          this.waiters.peekFirst()
                                      .signal();
                      }
                  }
  
                  // unlock and return the buffer
                  lock.unlock();
                  if (buffer == null) {// buffer = null 代表内存时直接从free中轮询释放的
                      return ByteBuffer.allocate(size);
                  } else {// buffer不为空，是直接复用free中的内存
                      return buffer;
                  }
              }
          } finally {
              if (lock.isHeldByCurrentThread()) {
                  lock.unlock();
              }
          }
      }
focal_method.class_method_signature: BufferPool.allocate(int size, long maxTimeToBlockMs)
focal_method.full_signature: public ByteBuffer allocate(int size, long maxTimeToBlockMs)
focal_method.identifier: allocate
focal_method.invocations: lock isEmpty pollFirst size freeUp unlock allocate newCondition toNanos addLast nanoseconds await remove nanoseconds max record milliseconds remove isEmpty pollFirst freeUp min removeFirst isEmpty isEmpty signal peekFirst unlock allocate isHeldByCurrentThread unlock
focal_method.modifiers: public
focal_method.parameters: (int size, long maxTimeToBlockMs)
focal_method.return: ByteBuffer
focal_method.signature: ByteBuffer allocate(int size, long maxTimeToBlockMs)