/*
(non-Javadoc)
@see net.timewalker.ffmq4.storage.data.impl.AbstractBlockBasedDataStore#writeFirstBlock()
340472
@Override
	protected void writeFirstBlock() throws DataStoreException
    {
    	try
    	{
	    	allocationTableRandomAccessFile.seek(AT_HEADER_FIRSTBLOCK_OFFSET);
	        allocationTableRandomAccessFile.writeInt(firstBlock);
    	}
    	catch (IOException e)
    	{
    		throw new DataStoreException("Cannot write to allocation table file : "+allocationTableFile.getAbsolutePath(),e);
    	}
    }
6256
@Override
	public boolean write(Writer writer) throws IOException {

		writeBlock(writer, EmblPadding.PA_PADDING, accession);

		return true;
	}
994863
public void write(DataOutput out) throws IOException {
    out.writeLong(blockId);
    out.writeLong(blockGenStamp);
    super.write(out);
  }
1193140
public synchronized void reset() throws BlockStoreException {
        try {
            WriteBatch batch = db.createWriteBatch();
            try {
                DBIterator it = db.iterator();
                try {
                    it.seekToFirst();
                    while (it.hasNext())
                        batch.delete(it.next().getKey());
                    db.write(batch);
                } finally {
                    it.close();
                }
            } finally {
                batch.close();
            }
            initStoreIfNeeded();
        } catch (IOException e) {
            throw new BlockStoreException(e);
        }
    }
663666
private void sendNextSketchBlock() {
        byte[] rawBlock = sketchBlocksToSend.get(currSketchBlockNum);
        Buffer block = new Buffer();
        block.write(rawBlock);
        sendMessage(BeanMessageID.BL_FW_BLOCK, block);

        resetSketchBlockSendTimeout();

        int blocksSent = currSketchBlockNum + 1;
        int totalBlocks = sketchBlocksToSend.size();
        onSketchUploadProgress.onResult(UploadProgress.create(blocksSent, totalBlocks));

        currSketchBlockNum++;
        if ( currSketchBlockNum >= sketchBlocksToSend.size() ) {
            resetSketchUploadState();
        }
    }
105033
void storeBlock(BlockId blockId, ByteBuffer block) throws IOException {
        synchronized (m_accessLock) {
            if (m_blockPathMap.containsKey(blockId)) {
                throw new IllegalArgumentException("Request to store block that is already stored: "
                                                    + blockId.toString());
            }

            int origPosition = block.position();
            block.position(0);
            Path blockPath = makeBlockPath(blockId);
            try (SeekableByteChannel channel = Files.newByteChannel(blockPath, OPEN_OPTIONS, PERMISSIONS)) {
                channel.write(block);
            }
            finally {
                block.position(origPosition);
            }

            m_blockPathMap.put(blockId, blockPath);
        }
    }
37252
@Override
  public WriteLockedOffHeapClockCache<K, V> newInstance() {
    StorageEngine<? super K, ? super V> storageEngine = storageEngineFactory.newInstance();
    try {
      if (evictionListener == null) {
        return new WriteLockedOffHeapClockCache<>(tableSource, storageEngine, tableSize);
      } else {
        return new EvictionListeningWriteLockedOffHeapClockCache<>(evictionListener, tableSource, storageEngine, tableSize);
      }
    } catch (RuntimeException e) {
      storageEngine.destroy();
      throw e;
    }
  }
389682
private void beginMainBlock() {
        int blockNo = registerBlock(null); // =0
        BlockTabRec btr = blockTab[blockNo];
        btr.tPosBegin = 0;
        btr.tPosContentsBegin = 0;
        openBlocksTab[currentNestingLevel] = blockNo;
        currentNestingLevel++;
    }
355405
protected void firstWrite()
        throws IOException
    {
        if (_response.isCommitted())
            return;
        
        // Nobble the OutputStream for HEAD requests
        if (HttpRequest.__HEAD.equals(_request.getMethod()))
            _outputStream.nullOutput();    
            
        int length=_response.getIntField(HttpFields.__ContentLength);
        if (length>=0)
            _outputStream.setContentLength(length);
    }
322624
@Override
	protected void writeDataBlock(byte[] data, int offset, int len, int blockHandle) throws DataStoreException
    {
    	byte[] blockData = new byte[blockSize];
    	System.arraycopy(data, offset, blockData, 0, len);
    	
    	dirtyBlockTable.markDirty(blockHandle, blockData);    	
    	journal.writeDataBlock(blockHandle, (long)blockHandle*blockSize, blockData);
    }
1012773
private void writeBytesSlow(final byte[] bytes, int off, int len)
    {
        while (len > 0)
        {
            final Block block = current;
            final int amount = Math.min(len, block.remaining());
            System.arraycopy(bytes, off, block.data, block.limit, amount);
            block.limit += amount;
            off += amount;
            len -= amount;
            if (block.remaining() == 0)
            {
                if (index == blocks.size() - 1)
                {
                    allocateNewBlock();
                }
                index++;
                current = blocks.get(index);
            }
        }

    }
116905
@Override
  public void addBlock(BlockInfo newblock) {
    if (this.blocks == null) {
      this.blocks = new BlockInfo[1];
      this.blocks[0] = newblock;
    } else {
      int size = this.blocks.length;
      BlockInfo[] newlist = new BlockInfo[size + 1];
      System.arraycopy(this.blocks, 0, newlist, 0, size);
      newlist[size] = newblock;
      this.blocks = newlist;
    }
  }
255153
private void copyBlock(DFSClient dfs, LocatedBlock lblock,
                         OutputStream fos) throws Exception {
    int failures = 0;
    InetSocketAddress targetAddr = null;
    TreeSet<DatanodeInfo> deadNodes = new TreeSet<DatanodeInfo>();
    Socket s = null;
    BlockReader blockReader = null; 
    Block block = lblock.getBlock(); 

    while (s == null) {
      DatanodeInfo chosenNode;
      
      try {
        chosenNode = bestNode(dfs, lblock.getLocations(), deadNodes);
        targetAddr = NetUtils.createSocketAddr(chosenNode.getName());
      }  catch (IOException ie) {
        if (failures >= DFSClient.MAX_BLOCK_ACQUIRE_FAILURES) {
          throw new IOException("Could not obtain block " + lblock);
        }
        LOG.info("Could not obtain block from any node:  " + ie);
        try {
          Thread.sleep(10000);
        }  catch (InterruptedException iex) {
        }
        deadNodes.clear();
        failures++;
        continue;
      }
      try {
        s = new Socket();
        s.connect(targetAddr, HdfsConstants.READ_TIMEOUT);
        s.setSoTimeout(HdfsConstants.READ_TIMEOUT);
        
        blockReader = 
            BlockReader.newBlockReader(DataTransferProtocol.DATA_TRANSFER_VERSION,
                                               nn.getNamesystem().getFSImage().storage.namespaceID,
                                               s, targetAddr.toString() + ":" + 
                                               block.getBlockId(), 
                                               block.getBlockId(), 
                                               block.getGenerationStamp(), 
                                               0, -1,
                                               conf.getInt("io.file.buffer.size", 4096));
        
      }  catch (IOException ex) {
        // Put chosen node into dead list, continue
        LOG.info("Failed to connect to " + targetAddr + ":" + ex);
        deadNodes.add(chosenNode);
        if (s != null) {
          try {
            s.close();
          } catch (IOException iex) {
          }
        }
        s = null;
      }
    }
    if (blockReader == null) {
      throw new Exception("Could not open data stream for " + lblock.getBlock());
    }
    byte[] buf = new byte[1024];
    int cnt = 0;
    boolean success = true;
    long bytesRead = 0;
    try {
      while ((cnt = blockReader.read(buf, 0, buf.length)) > 0) {
        fos.write(buf, 0, cnt);
        bytesRead += cnt;
      }
      if ( bytesRead != block.getNumBytes() ) {
        throw new IOException("Recorded block size is " + block.getNumBytes() + 
                              ", but datanode returned " +bytesRead+" bytes");
      }
    } catch (Exception e) {
      e.printStackTrace();
      success = false;
    } finally {
      try {s.close(); } catch (Exception e1) {}
    }
    if (!success)
      throw new Exception("Could not copy block data for " + lblock.getBlock());
  }
907503
final boolean addStoredBlockInternal(Block block,
                               DatanodeDescriptor node,
                               DatanodeDescriptor delNodeHint,
                               boolean initialBlockReport,
                               InitialReportWorker worker) throws IOException {
    InjectionHandler.processEvent(InjectionEvent.FSNAMESYSTEM_ADDSTORED_BLOCK,
        node);
    // either this is a direct call protected by writeLock, or
    // this is parallel initial block report
    assert (hasWriteLock() || (worker != null));
    
    // the call is invoked by the initial block report worker thread
    final boolean parallelInitialBlockReport = (worker != null);
    
    BlockInfo storedBlock = blocksMap.getStoredBlock(block);
    
    // handle the case for standby when we receive a block
    // that we don't know about yet - this is for block reports,
    // where we do not explicitly check this beforehand
    if (storedBlock == null && nameNode.shouldRetryAbsentBlocks() // standby
        && nameNode.shouldRetryAbsentBlock(block, storedBlock)) {
      // block should be retried
      return false;
    }
    
    if (storedBlock == null) {
      // then we need to do some special processing.
      storedBlock = blocksMap.getStoredBlockWithoutMatchingGS(block);

      if (storedBlock == null) {
        rejectAddStoredBlock(
          new Block(block), node,
          "Block not in blockMap with any generation stamp",
          initialBlockReport, parallelInitialBlockReport);
        return false;
      }

      INodeFile inode = storedBlock.getINode();
      if (inode == null) {
        rejectAddStoredBlock(
          new Block(block), node,
          "Block does not correspond to any file",
          initialBlockReport, parallelInitialBlockReport);
        return false;
      }

      boolean reportedOldGS = block.getGenerationStamp() < storedBlock.getGenerationStamp();
      boolean reportedNewGS = block.getGenerationStamp() > storedBlock.getGenerationStamp();
      boolean underConstruction = inode.isUnderConstruction();
      boolean isLastBlock = inode.getLastBlock() != null &&
        inode.getLastBlock().getBlockId() == block.getBlockId();

      // Don't add blocks to the DN when they're part of the in-progress last block
      // and have an inconsistent generation stamp. Instead just add them to targets
      // for recovery purposes. They will get added to the node when
      // commitBlockSynchronization runs
      if (reportedOldGS || reportedNewGS) {  // mismatched generation stamp
        if (underConstruction && isLastBlock) {
          NameNode.stateChangeLog.info(
              "BLOCK* NameSystem.addStoredBlock: "
              + "Targets updated: block " + block + " on " + node.getName() +
              " is added as a target for block " + storedBlock + " with size " +
              block.getNumBytes());
          
          lockParallelBRLock(parallelInitialBlockReport);
          try {
            ((INodeFileUnderConstruction) inode).addTarget(node,
                block.getGenerationStamp());
          } finally {
            unlockParallelBRLock(parallelInitialBlockReport);
          }
        } else {
          rejectAddStoredBlock(
              new Block(block), node,
              "Reported block has mismatched generation stamp " +
              "but is not the last block of " +
              "an under-construction file. (current generation is " +
              storedBlock.getGenerationStamp() + ")",
              initialBlockReport, parallelInitialBlockReport);
        }
        return false;
      }
    }

    INodeFile fileINode = storedBlock.getINode();
    if (fileINode == null) {
      rejectAddStoredBlock(
          new Block(block), node,
          "Block does not correspond to any file",
          initialBlockReport, parallelInitialBlockReport);
      return false;
    }

    assert storedBlock != null : "Block must be stored by now";

    // add block to the data-node
    boolean added;
    if (!parallelInitialBlockReport) {
      // full insert
      added = node.addBlock(storedBlock);
    } else {
      // insert DN descriptor into stored block
      int index = node.addBlockWithoutInsertion(storedBlock);
      added = index >= 0;
      // inform the worker so it can insert it into its local list
      worker.setCurrentStoredBlock(storedBlock, index);
    }
    
    // Is the block being reported the last block of an underconstruction file?
    boolean blockUnderConstruction = false;
    
    if (fileINode.isUnderConstruction()) {
      Block last = fileINode.getLastBlock();
      if (last == null) {
        // This should never happen, but better to handle it properly than to throw
        // an NPE below.
        LOG.error("Null blocks for reported block=" + block + " stored=" + storedBlock +
          " inode=" + fileINode);
        return false;
      }
      blockUnderConstruction = last.equals(storedBlock);
    }

    // block == storedBlock when this addStoredBlock is the result of a block report
    if (block.getNumBytes() != storedBlock.getNumBytes()) {
      if (!checkBlockSize(block, storedBlock.getINode())) {
        try {
          // New replica has an invalid block size. Mark it as corrupted.
          LOG.warn("Mark new replica " + block + " from " + node.getName() +
            "as corrupt because its length " + block.getNumBytes() +
            " is not valid");
          markBlockAsCorrupt(block, node, parallelInitialBlockReport);
        } catch (IOException e) {
          LOG.warn("Error in deleting bad block " + block + e);
        }
      } else {
        long cursize = storedBlock.getNumBytes();
        if (cursize == 0) {
          storedBlock.setNumBytes(block.getNumBytes());
        } else if (cursize != block.getNumBytes()) {
          String logMsg = "Inconsistent size for block " + block + 
            " reported from " + node.getName() +
            " current size is " + cursize +
            " reported size is " + block.getNumBytes();
          // If the block is still under construction this isn't likely
          // to be a problem, so just log at INFO level.
          if (blockUnderConstruction) {
            if (cursize != 1) { // cursize == 1 implies block was fsynced
              LOG.info(logMsg);
            }
          } else {
            LOG.warn(logMsg);
          }

          try {
            if (cursize > block.getNumBytes() && !blockUnderConstruction) {
              // new replica is smaller in size than existing block.
              // Mark the new replica as corrupt.
              LOG.warn("Mark new replica " + block + " from " + node.getName() +
                "as corrupt because its length is shorter than existing ones");
              markBlockAsCorrupt(block, node, parallelInitialBlockReport);
            } else {
              // new replica is larger in size than existing block.
              if (!blockUnderConstruction) {
                // Mark pre-existing replicas as corrupt.
                int numNodes = blocksMap.numNodes(block);
                int count = 0;
                DatanodeDescriptor nodes[] = new DatanodeDescriptor[numNodes];
                Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(block);
                for (; it != null && it.hasNext(); ) {
                  DatanodeDescriptor dd = it.next();
                  if (!dd.equals(node)) {
                    nodes[count++] = dd;
                  }
                }
                for (int j = 0; j < count; j++) {
                  LOG.warn("Mark existing replica " + block + " from " + node.getName() +
                  " as corrupt because its length is shorter than the new one");
                  markBlockAsCorrupt(block, nodes[j], parallelInitialBlockReport);
                }
              }
              //
              // change the size of block in blocksMap
              //
              storedBlock.setNumBytes(block.getNumBytes());
            }
          } catch (IOException e) {
            LOG.warn("Error in deleting bad block " + block + e);
          }
        }
      }
      block = storedBlock;
    } else {
      block = storedBlock;
    }
    assert storedBlock == block : "Block must be stored by now";

    int curReplicaDelta = 0;

    if (added) {
      curReplicaDelta = 1;
    } else {
      NameNode.stateChangeLog.info("BLOCK* NameSystem.addStoredBlock: "
        + "Redundant addStoredBlock request received for "
        + block + " on " + node.getName()
        + " size " + block.getNumBytes());
    }

    // filter out containingNodes that are marked for decommission.
    NumberReplicas num = countNodes(storedBlock);
    int numCurrentReplica = 0;
    int numLiveReplicas = num.liveReplicas();

    boolean popReplQueuesBefore = isPopulatingReplQueuesInternal();

    if (!popReplQueuesBefore) {
      // if we haven't populated the replication queues
      // then use a cheaper method to count
      // we only need live and decommissioned replicas.
      numCurrentReplica = numLiveReplicas + num.decommissionedReplicas();
    } else {
      // count live & decommissioned replicas
      numCurrentReplica = numLiveReplicas + pendingReplications.getNumReplicas(block);
    }

    if (blockUnderConstruction) {
      lockParallelBRLock(parallelInitialBlockReport);
      try {
        added = ((INodeFileUnderConstruction) fileINode).addTarget(node,
            block.getGenerationStamp());
      } finally {
        unlockParallelBRLock(parallelInitialBlockReport);
      }
    }

    // check whether safe replication is reached for the block. Do not increment
    // safe block count if this block has already been reported by the same
    // datanode before. We used the check for 'added' to achieve this.
    if (added && isInSafeModeInternal()) {
      int numSafeReplicas = numLiveReplicas + num.decommissionedReplicas();
      if (blockUnderConstruction) {
        // If this is the last block under construction then all the replicas so
        // far have been added to the "targets" field of
        // INodeFileUnderConstruction and hence lookup the replication factor
        // from there.
        // In this code path, a "valid" replica is added to the block, and we
        // only count "valid" replica to be safe replica. It is possible that
        // earlier a replica with higher generation stamp has been added and a
        // false was returned. In that case, if we only count number of targets
        // we will miss to count in the case that we first accepted more than
        // minReplication's number of higher generation stamp blocks but later
        // blocks with "valid" generation stamps themselves reached minimum
        // replication again.
        DatanodeDescriptor[] validDNs = ((INodeFileUnderConstruction) fileINode)
            .getValidTargets();
        numSafeReplicas = (validDNs != null) ? validDNs.length : 0;
      }
      
      if (!parallelInitialBlockReport && added) {
        // regular add stored block
        incrementSafeBlockCount(numSafeReplicas, initialBlockReport);
      } else if (parallelInitialBlockReport && added) {
        // for parallel initial block report increment local worker variable
        worker.incrementSafeBlockCount(numSafeReplicas);
      }
    }

    if (!popReplQueuesBefore && isPopulatingReplQueuesInternal()) {
      // we have just initialized the repl queues
      // must recompute num
      num = countNodes(storedBlock);
      numLiveReplicas = num.liveReplicas();
      numCurrentReplica = numLiveReplicas +
        pendingReplications.getNumReplicas(block);
    }

    //
    // if file is being actively written to and it is the last block, 
    // then do not check replication-factor here.
    //
    if (blockUnderConstruction) {
      return true;
    }

    // do not handle mis-replicated blocks during start up
    if (!isPopulatingReplQueuesInternal()) {
      return true;
    }

    if (curReplicaDelta == 0) {
      return true;
    }
    
    // handle underReplication
    short blockReplication = fileINode.getBlockReplication(storedBlock);
    updateNeededReplicationQueue(storedBlock, curReplicaDelta, numCurrentReplica,
        num.decommissionedReplicas, node, blockReplication);
    // handle over-replication
    if (numCurrentReplica > blockReplication) {
      // Put block into a queue and handle excess block asyncly
      if (delNodeHint == null || node == delNodeHint) {
        overReplicatedBlocks.add(block);
      } else {
        overReplicatedBlocks.add(new OverReplicatedBlock(block, node, delNodeHint));
      }
    }
    // If the file replication has reached desired value
    // we can remove any corrupt replicas the block may have
    int corruptReplicasCount = corruptReplicas.numCorruptReplicas(block);
    int numCorruptNodes = num.corruptReplicas();
    if (numCorruptNodes != corruptReplicasCount) {
      LOG.warn("Inconsistent number of corrupt replicas for " +
        block + "blockMap has " + numCorruptNodes +
        " but corrupt replicas map has " + corruptReplicasCount);
    }
    if ((corruptReplicasCount > 0) && (numLiveReplicas >= blockReplication)) {
      invalidateCorruptReplicas(block);
    }
    return true;
  }
1261998
protected void writeBlock(Writer writer, String header, 
			String block) throws IOException {
		writeBlock(writer, null, header, block);
	}
706316
public void writePadBlocks(int blockCount) throws IOException {

        for (int i = 0; i < blockCount; i++) {
            write(ZERO_BLOCK, ZERO_BLOCK.length);
        }
    }
272203
private void write(byte []buffer, int offset, int length, boolean isFinal)
  {
    Objects.requireNonNull(buffer);

    _frameOut.write(buffer, offset, length, isFinal);
  }