Pins a buffer to the specified block, potentially waiting until a buffer
becomes available. If no buffer becomes available within a fixed time
period, then repins all currently holding blocks.

@param blk
a block ID
@return the buffer pinned to that block
971135
public Buffer pin(BlockId blk) {
		// Try to find out if this block has been pinned by this transaction
		PinnedBuffer pinnedBuff = pinnedBuffers.get(blk);
		if (pinnedBuff != null) {
			pinnedBuff.pinnedCount++;
			return pinnedBuff.buffer;
		}
		
		// This transaction has pinned too many buffers
		if (pinnedBuffers.size() == BUFFER_POOL_SIZE)
			throw new BufferAbortException();
		
		// Pinning process
		try {
			Buffer buff;
			long timestamp = System.currentTimeMillis();
			boolean waitedBeforeGotBuffer = false;

			// Try to pin a buffer or the pinned buffer for the given BlockId
			buff = bufferPool.pin(blk);

			// If there is no such buffer or no available buffer,
			// wait for it
			if (buff == null) {
				waitedBeforeGotBuffer = true;
				synchronized (bufferPool) {
					waitingThreads.add(Thread.currentThread());

					while (buff == null && !waitingTooLong(timestamp)) {
						bufferPool.wait(MAX_TIME);
						if (waitingThreads.get(0).equals(Thread.currentThread()))
							buff = bufferPool.pin(blk);
					}

					waitingThreads.remove(Thread.currentThread());
				}
			}

			// If it still has no buffer after a long wait,
			// release and re-pin all buffers it has
			if (buff == null) {
				repin();
				buff = pin(blk);
			} else {
				pinnedBuffers.put(buff.block(), new PinnedBuffer(buff));
			}

			// TODO: Add some comment here
			if (waitedBeforeGotBuffer) {
				synchronized (bufferPool) {
					bufferPool.notifyAll();
				}
			}

			return buff;
		} catch (InterruptedException e) {
			throw new BufferAbortException();
		}
	}
1335499
private void ensureBufferAvailable(TableBlock block) throws IOException {
        if (!block.hasBuffer()) {
            synchronized (this) {
                if (!block.hasBuffer()) {
                    ByteBuffer buffer = getBuffer();
                    block.setBuffer(buffer);
                }
            }
        }
    }
1231501
public static LargeBlockTask getLoadTask(BlockId blockId, ByteBuffer block) {
        return new LargeBlockTask() {
            @Override
            public LargeBlockResponse call() throws Exception {
                Exception theException = null;
                try {
                    LargeBlockManager.getInstance().loadBlock(blockId, block);
                }
                catch (Exception exc) {
                    theException = exc;
                }

                return new LargeBlockResponse(theException);
            }
        };
    }
1215315
@Override
	public BufferOrEvent getNextNonBlocked() throws Exception {
		while (true) {
			// process buffered BufferOrEvents before grabbing new ones
			Optional<BufferOrEvent> next;
			if (currentBuffered == null) {
				next = inputGate.getNextBufferOrEvent();
			}
			else {
				next = Optional.ofNullable(currentBuffered.getNext());
				if (!next.isPresent()) {
					completeBufferedSequence();
					return getNextNonBlocked();
				}
			}

			if (!next.isPresent()) {
				if (!endOfStream) {
					// end of input stream. stream continues with the buffered data
					endOfStream = true;
					releaseBlocksAndResetBarriers();
					return getNextNonBlocked();
				}
				else {
					// final end of both input and buffered data
					return null;
				}
			}

			BufferOrEvent bufferOrEvent = next.get();
			if (isBlocked(bufferOrEvent.getChannelIndex())) {
				// if the channel is blocked, we just store the BufferOrEvent
				bufferBlocker.add(bufferOrEvent);
				checkSizeLimit();
			}
			else if (bufferOrEvent.isBuffer()) {
				return bufferOrEvent;
			}
			else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
				if (!endOfStream) {
					// process barriers only if there is a chance of the checkpoint completing
					processBarrier((CheckpointBarrier) bufferOrEvent.getEvent(), bufferOrEvent.getChannelIndex());
				}
			}
			else if (bufferOrEvent.getEvent().getClass() == CancelCheckpointMarker.class) {
				processCancellationBarrier((CancelCheckpointMarker) bufferOrEvent.getEvent());
			}
			else {
				if (bufferOrEvent.getEvent().getClass() == EndOfPartitionEvent.class) {
					processEndOfPartition();
				}
				return bufferOrEvent;
			}
		}
	}
1038894
public Long putIfAbsentChecksum(Block blk, Long newChecksum)
      throws IOException {
    Long oldChecksum = putIfAbsent(blk, newChecksum);
    if (oldChecksum!= null && !oldChecksum.equals(newChecksum)) {
      throw new IOException("Block " + blk.toString()
        + " has different checksums " + oldChecksum + "(old) and " +
          newChecksum+ "(new)");
    }
    return oldChecksum;
  }
1030534
@Override
    public void commitBlobBlocks(String blob, BlockList blockList)
            throws ServiceException {
        blobService.commitBlobBlocks(containerName, blob, blockList);
    }
42715
private CallbackResult<Boolean> renderBlock(BufferBuilder buffer, IBlockAccess world, BlockPos pos, IBlockState state)
	{
		IBlockRenderer renderer = getBlockRendererOverride(world, pos, state);
		if (renderer == null)
			renderer = blockRenderers.get(state.getBlock());
		if (renderer == null)
			return CallbackResult.noResult();

		//convert pos to immutable one
		return CallbackResult	.<Boolean> builder()
								.value(renderer.renderBlock(buffer, world, new BlockPos(pos), state))
								.withReturn(true)
								.result();
	}
564811
bbBlock findBlockForWrite(Object caller, int version, bbBlock curr, int pos)
    {
        assert mutation_in_progress(caller, version);
        assert (pos >= 0 && "invalid position, positions must be >= 0".length() > 0);
        if (pos > this._buf_limit + 1) {
            throw new BlockedBufferException("writes must be contiguous");
        }
        assert _validate();
        if (curr != null && curr.hasRoomToWrite(pos, 1) == true) {
            if (curr._offset + curr._limit == pos && curr._idx < this._next_block_position) {
                bbBlock b = this._blocks.get(curr._idx + 1);
                if (b.containsForWrite(pos)) {
                    curr = b;
                }
            }
            return curr;
        }
        // we're not going to write into curr, so find out the right block
        bbBlock block;
        if (pos == this._buf_limit) {
            // if we're at the limit the only possible (existing) block
            // will be the very last block - shortcut to optimize append
            assert this._next_block_position > 0;
            block = this._blocks.get(this._next_block_position - 1);
        }
        else if (curr != null && pos == curr._offset + curr._limit) {
            // if our current position is exactly at the end (and we already know
            // we can't write into this block if we can write at all we'll have
            // to write into the next block (inner blocks can't be 0 bytes long)
            block = this._blocks.get(curr._idx + 1);
        }
        else {
            // since we're not at the limit and we don't have a current block
            // we'll go find the block in the list (this is an abnormal case)
            // since append if usual for writing
            block = findBlockHelper(pos, 0, this._next_block_position);
        }
        assert block != null;
        assert block.containsForWrite(pos);
        // chech our candidate block to see if it's the one we'd write into
        if (block.hasRoomToWrite(pos, 1)) {
            return block;
        }
        // at this point, we can't use _curr in any event so we can just
        // move on to the next block since findHelper will have returned
        // either the right block (which it didn't) or the one just in
        // front of the right block - so let's see if there is an allocated
        // block just following this
        if (block._idx < this._next_block_position - 1) {
            block = this._blocks.get(block._idx + 1);
            return block;
        }
        // there wasn't a following block (actually a common case when
        // you're appending) so we have to go ahead an actually add a new block
        int newIdx = block._idx + 1;
        assert newIdx == this._next_block_position;
        bbBlock ret =  this.addBlock(caller
                            ,version
                            ,newIdx
                            ,pos
                            ,this.nextBlockSize(caller, version)
               );
        return ret;
    }
919371
public static <E> BlockingQueue<E> getLimitedBlockingQueue(int maxQueue) {
        return new LinkedBlockingQueue<E>(maxQueue) {
            @Override
            public boolean offer(E e) {
                return putInsteadOfOffer(this, e);
            }
        };
    }
1076370
public QuadBatch pushBatch (QuadBatch newBatch) {
    if (newBatch == null) return null;
    QuadBatch oldBatch = batch;
    batch.end();
    batch = beginBatch(newBatch);
    return oldBatch;
  }
1338684
public static <E> BlockingQueue<E> getWaitingLimitedBlockingQueue(
            long waitingMillis, int maxQueue) {
        return new LinkedBlockingQueue<E>(maxQueue) {
            @Override
            public boolean offer(E e) {
                if (size() >= maxQueue) {
                    sleep(waitingMillis);
                    log.warn("Wait For {} ms And Blocking !!! - maxQueue = {}",
                            waitingMillis, maxQueue);
                }
                return putInsteadOfOffer(this, e);
            }
        };
    }
733033
public synchronized T getNonBlocking ()
    {
        if (_count == 0) {
            return null;
        }

        // pull the object off, and clear our reference to it
        T retval = _items[_start];
        _items[_start] = null;

        _start = (_start + 1) % _size;
        _count--;

        return retval;
    }
898364
void getBlock(int offset) {

        try {
            RowSetNavigatorClient source = session.getRows(id, offset,
                baseBlockSize);

            table         = source.table;
            currentOffset = source.currentOffset;
        } catch (HsqlException e) {}
    }
408690
public StoredBlock getCheckpointBefore(long time) {
        try {
            checkArgument(time > params.getGenesisBlock().getTimeSeconds());
            // This is thread safe because the map never changes after creation.
            Map.Entry<Long, StoredBlock> entry = checkpoints.floorEntry(time);
            if (entry != null) return entry.getValue();
            Block genesis = params.getGenesisBlock().cloneAsHeader();
            return new StoredBlock(genesis, genesis.getWork(), 0);
        } catch (VerificationException e) {
            throw new RuntimeException(e);  // Cannot happen.
        }
    }
867021
private void writeBlock() throws IOException {
        if (this.debug) {
            System.err.println("WriteBlock: blkIdx = " + this.currBlkIdx);
        }

        if (this.outStream == null) {
            throw new IOException("writing to an input buffer");
        }

        this.outStream.write(this.blockBuffer, 0, this.blockSize);
        this.outStream.flush();

        this.currRecIdx = 0;
        this.currBlkIdx++;
    }
492770
public ByteBuffer get(final int index) {
		if (!validState) {
			throw new InvalidStateException();
		}
		if (log.isDebugEnabled()) {
			log.debug("get(" + index + ")");
		}
		try {
			if (useMmap) {
				final MappedByteBuffer mbb = getMmapForIndex(index);
				if (mbb != null) {
					return mbb;
				}
				// Fallback to RAF
			}
			final ByteBuffer buf = bufstack.pop();
			fileChannel.position(index * blockSize).read(buf);
			buf.rewind();
			return buf;
		} catch (Exception e) {
			log.error("Exception in get(" + index + ")", e);
		}
		return null;
	}
628571
public void finalizeAndNotifyNamenode(int namespaceId, Block blk)
      throws IOException {
    this.getFSDataset().finalizeBlock(namespaceId, blk);
    this.notifyNamenodeReceivedBlock(namespaceId, blk, null);
  }