| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hadoop.raid; |
| |
| import java.io.BufferedOutputStream; |
| import java.io.ByteArrayInputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.io.DataInputStream; |
| import java.io.DataOutputStream; |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileOutputStream; |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.OutputStream; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.regex.Pattern; |
| import java.util.Random; |
| import java.net.InetSocketAddress; |
| import java.net.Socket; |
| import java.nio.channels.SocketChannel; |
| |
| import org.apache.hadoop.util.DataChecksum; |
| import org.apache.hadoop.hdfs.DistributedFileSystem; |
| import org.apache.hadoop.hdfs.protocol.Block; |
| import org.apache.hadoop.hdfs.protocol.DataTransferProtocol; |
| import org.apache.hadoop.hdfs.protocol.DatanodeInfo; |
| import org.apache.hadoop.hdfs.protocol.FSConstants; |
| import org.apache.hadoop.hdfs.protocol.FSConstants.DatanodeReportType; |
| import org.apache.hadoop.hdfs.protocol.LocatedBlock; |
| import org.apache.hadoop.hdfs.protocol.LocatedBlocks; |
| import org.apache.hadoop.hdfs.server.common.HdfsConstants; |
| import org.apache.hadoop.hdfs.server.datanode.FSDataset; |
| import org.apache.hadoop.hdfs.server.datanode.RaidBlockSender; |
| import org.apache.hadoop.io.Text; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.conf.Configured; |
| import org.apache.hadoop.fs.ChecksumException; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.hdfs.BlockMissingException; |
| import org.apache.hadoop.hdfs.RaidDFSUtil; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.hadoop.net.NetUtils; |
| |
| import org.apache.hadoop.raid.RaidNode; |
| import org.apache.hadoop.raid.RaidUtils; |
| import org.apache.hadoop.raid.protocol.PolicyInfo.ErasureCodeType; |
| |
| |
| /** |
| * contains the core functionality of the block fixer |
| * |
| * raid.blockfix.interval - interval between checks for corrupt files |
| * |
| * raid.blockfix.history.interval - interval before fixing same file again |
| * |
| * raid.blockfix.read.timeout - read time out |
| * |
| * raid.blockfix.write.timeout - write time out |
| */ |
| public class BlockFixer extends Configured implements Runnable { |
| public static final Log LOG = LogFactory.getLog( |
| "org.apache.hadoop.raid.BlockFixer"); |
| |
| public static final String BLOCKFIX_INTERVAL = "raid.blockfix.interval"; |
| public static final String BLOCKFIX_HISTORY_INTERVAL = |
| "raid.blockfix.history.interval"; |
| public static final String BLOCKFIX_READ_TIMEOUT = |
| "raid.blockfix.read.timeout"; |
| public static final String BLOCKFIX_WRITE_TIMEOUT = |
| "raid.blockfix.write.timeout"; |
| |
| public static final long DEFAULT_BLOCKFIX_INTERVAL = 60 * 1000; // 1 min |
| public static final long DEFAULT_BLOCKFIX_HISTORY_INTERVAL = |
| 60 * 60 * 1000; // 60 mins |
| |
| private java.util.HashMap<String, java.util.Date> history; |
| private long numFilesFixed = 0; |
| private String xorPrefix; |
| private String rsPrefix; |
| private Encoder xorEncoder; |
| private Decoder xorDecoder; |
| private Encoder rsEncoder; |
| private Decoder rsDecoder; |
| |
| // interval between checks for corrupt files |
| protected long blockFixInterval = DEFAULT_BLOCKFIX_INTERVAL; |
| |
| // interval before fixing same file again |
| protected long historyInterval = DEFAULT_BLOCKFIX_HISTORY_INTERVAL; |
| |
| public volatile boolean running = true; |
| |
| |
| public BlockFixer(Configuration conf) throws IOException { |
| super(conf); |
| history = new java.util.HashMap<String, java.util.Date>(); |
| blockFixInterval = getConf().getInt(BLOCKFIX_INTERVAL, |
| (int) blockFixInterval); |
| xorPrefix = RaidNode.xorDestinationPath(getConf()).toUri().getPath(); |
| if (!xorPrefix.endsWith(Path.SEPARATOR)) { |
| xorPrefix += Path.SEPARATOR; |
| } |
| int stripeLength = RaidNode.getStripeLength(getConf()); |
| xorEncoder = new XOREncoder(getConf(), stripeLength); |
| xorDecoder = new XORDecoder(getConf(), stripeLength); |
| rsPrefix = RaidNode.rsDestinationPath(getConf()).toUri().getPath(); |
| if (!rsPrefix.endsWith(Path.SEPARATOR)) { |
| rsPrefix += Path.SEPARATOR; |
| } |
| int parityLength = RaidNode.rsParityLength(getConf()); |
| rsEncoder = new ReedSolomonEncoder(getConf(), stripeLength, parityLength); |
| rsDecoder = new ReedSolomonDecoder(getConf(), stripeLength, parityLength); |
| } |
| |
| public void run() { |
| while (running) { |
| try { |
| LOG.info("BlockFixer continuing to run..."); |
| doFix(); |
| } catch (Exception e) { |
| LOG.error(StringUtils.stringifyException(e)); |
| } catch (Error err) { |
| LOG.error("Exiting after encountering " + |
| StringUtils.stringifyException(err)); |
| throw err; |
| } |
| } |
| } |
| |
| public long filesFixed() { |
| return numFilesFixed; |
| } |
| |
| void doFix() throws InterruptedException, IOException { |
| while (running) { |
| // Sleep before proceeding to fix files. |
| Thread.sleep(blockFixInterval); |
| |
| // Purge history older than the history interval. |
| purgeHistory(); |
| |
| List<Path> corruptFiles = getCorruptFiles(); |
| if (corruptFiles.isEmpty()) { |
| // If there are no corrupt files, retry after some time. |
| continue; |
| } |
| LOG.info("Found " + corruptFiles.size() + " corrupt files."); |
| |
| sortCorruptFiles(corruptFiles); |
| |
| for (Path srcPath: corruptFiles) { |
| if (!running) break; |
| try { |
| fixFile(srcPath); |
| } catch (IOException ie) { |
| LOG.error("Hit error while processing " + srcPath + |
| ": " + StringUtils.stringifyException(ie)); |
| // Do nothing, move on to the next file. |
| } |
| } |
| } |
| } |
| |
| |
| void fixFile(Path srcPath) throws IOException { |
| |
| if (RaidNode.isParityHarPartFile(srcPath)) { |
| processCorruptParityHarPartFile(srcPath); |
| return; |
| } |
| |
| // The corrupted file is a XOR parity file |
| if (isXorParityFile(srcPath)) { |
| processCorruptParityFile(srcPath, xorEncoder); |
| return; |
| } |
| |
| // The corrupted file is a ReedSolomon parity file |
| if (isRsParityFile(srcPath)) { |
| processCorruptParityFile(srcPath, rsEncoder); |
| return; |
| } |
| |
| // The corrupted file is a source file |
| RaidNode.ParityFilePair ppair = |
| RaidNode.xorParityForSource(srcPath, getConf()); |
| Decoder decoder = null; |
| if (ppair != null) { |
| decoder = xorDecoder; |
| } else { |
| ppair = RaidNode.rsParityForSource(srcPath, getConf()); |
| if (ppair != null) { |
| decoder = rsDecoder; |
| } |
| } |
| |
| // If we have a parity file, process the file and fix it. |
| if (ppair != null) { |
| processCorruptFile(srcPath, ppair, decoder); |
| } |
| |
| } |
| |
| /** |
| * We maintain history of fixed files because a fixed file may appear in |
| * the list of corrupt files if we loop around too quickly. |
| * This function removes the old items in the history so that we can |
| * recognize files that have actually become corrupt since being fixed. |
| */ |
| void purgeHistory() { |
| // Default history interval is 1 hour. |
| long historyInterval = getConf().getLong( |
| BLOCKFIX_HISTORY_INTERVAL, 3600*1000); |
| java.util.Date cutOff = new java.util.Date( |
| System.currentTimeMillis()-historyInterval); |
| List<String> toRemove = new java.util.ArrayList<String>(); |
| |
| for (String key: history.keySet()) { |
| java.util.Date item = history.get(key); |
| if (item.before(cutOff)) { |
| toRemove.add(key); |
| } |
| } |
| for (String key: toRemove) { |
| LOG.info("Removing " + key + " from history"); |
| history.remove(key); |
| } |
| } |
| |
| /** |
| * @return A list of corrupt files as obtained from the namenode |
| */ |
| List<Path> getCorruptFiles() throws IOException { |
| DistributedFileSystem dfs = getDFS(new Path("/")); |
| |
| String[] nnCorruptFiles = RaidDFSUtil.getCorruptFiles(getConf()); |
| List<Path> corruptFiles = new LinkedList<Path>(); |
| for (String file: nnCorruptFiles) { |
| if (!history.containsKey(file)) { |
| corruptFiles.add(new Path(file)); |
| } |
| } |
| RaidUtils.filterTrash(getConf(), corruptFiles); |
| return corruptFiles; |
| } |
| |
| /** |
| * Sorts source files ahead of parity files. |
| */ |
| void sortCorruptFiles(List<Path> files) { |
| // TODO: We should first fix the files that lose more blocks |
| Comparator<Path> comp = new Comparator<Path>() { |
| public int compare(Path p1, Path p2) { |
| if (isXorParityFile(p2) || isRsParityFile(p2)) { |
| // If p2 is a parity file, p1 is smaller. |
| return -1; |
| } |
| if (isXorParityFile(p1) || isRsParityFile(p1)) { |
| // If p1 is a parity file, p2 is smaller. |
| return 1; |
| } |
| // If both are source files, they are equal. |
| return 0; |
| } |
| }; |
| Collections.sort(files, comp); |
| } |
| |
| /** |
| * Reads through a corrupt source file fixing corrupt blocks on the way. |
| * @param srcPath Path identifying the corrupt file. |
| * @throws IOException |
| */ |
| void processCorruptFile(Path srcPath, RaidNode.ParityFilePair parityPair, |
| Decoder decoder) throws IOException { |
| LOG.info("Processing corrupt file " + srcPath); |
| |
| DistributedFileSystem srcFs = getDFS(srcPath); |
| FileStatus srcStat = srcFs.getFileStatus(srcPath); |
| long blockSize = srcStat.getBlockSize(); |
| long srcFileSize = srcStat.getLen(); |
| String uriPath = srcPath.toUri().getPath(); |
| |
| int numBlocksFixed = 0; |
| List<LocatedBlock> corrupt = |
| RaidDFSUtil.corruptBlocksInFile(srcFs, uriPath, 0, srcFileSize); |
| for (LocatedBlock lb: corrupt) { |
| Block corruptBlock = lb.getBlock(); |
| long corruptOffset = lb.getStartOffset(); |
| |
| LOG.info("Found corrupt block " + corruptBlock + |
| ", offset " + corruptOffset); |
| |
| final long blockContentsSize = |
| Math.min(blockSize, srcFileSize - corruptOffset); |
| File localBlockFile = |
| File.createTempFile(corruptBlock.getBlockName(), ".tmp"); |
| localBlockFile.deleteOnExit(); |
| |
| try { |
| decoder.recoverBlockToFile(srcFs, srcPath, parityPair.getFileSystem(), |
| parityPair.getPath(), blockSize, corruptOffset, localBlockFile, |
| blockContentsSize); |
| |
| // We have a the contents of the block, send them. |
| DatanodeInfo datanode = chooseDatanode(lb.getLocations()); |
| computeMetdataAndSendFixedBlock( |
| datanode, localBlockFile, lb, blockContentsSize); |
| numBlocksFixed++; |
| |
| LOG.info("Adding " + srcPath + " to history"); |
| history.put(srcPath.toString(), new java.util.Date()); |
| } finally { |
| localBlockFile.delete(); |
| } |
| } |
| LOG.info("Fixed " + numBlocksFixed + " blocks in " + srcPath); |
| numFilesFixed++; |
| } |
| |
| /** |
| * checks whether file is xor parity file |
| */ |
| boolean isXorParityFile(Path p) { |
| String pathStr = p.toUri().getPath(); |
| if (pathStr.contains(RaidNode.HAR_SUFFIX)) { |
| return false; |
| } |
| return pathStr.startsWith(xorPrefix); |
| } |
| |
| /** |
| * checks whether file is rs parity file |
| */ |
| boolean isRsParityFile(Path p) { |
| String pathStr = p.toUri().getPath(); |
| if (pathStr.contains(RaidNode.HAR_SUFFIX)) { |
| return false; |
| } |
| return pathStr.startsWith(rsPrefix); |
| } |
| |
| /** |
| * Returns a DistributedFileSystem hosting the path supplied. |
| */ |
| protected DistributedFileSystem getDFS(Path p) throws IOException { |
| return (DistributedFileSystem) p.getFileSystem(getConf()); |
| } |
| |
| /** |
| * Fixes corrupt blocks in a parity file. |
| * This function uses the corresponding source file to regenerate parity |
| * file blocks. |
| */ |
| void processCorruptParityFile(Path parityPath, Encoder encoder) |
| throws IOException { |
| LOG.info("Processing corrupt file " + parityPath); |
| Path srcPath = sourcePathFromParityPath(parityPath); |
| if (srcPath == null) { |
| LOG.warn("Unusable parity file " + parityPath); |
| return; |
| } |
| |
| DistributedFileSystem parityFs = getDFS(parityPath); |
| FileStatus parityStat = parityFs.getFileStatus(parityPath); |
| long blockSize = parityStat.getBlockSize(); |
| long parityFileSize = parityStat.getLen(); |
| FileStatus srcStat = getDFS(srcPath).getFileStatus(srcPath); |
| long srcFileSize = srcStat.getLen(); |
| |
| // Check timestamp. |
| if (srcStat.getModificationTime() != parityStat.getModificationTime()) { |
| LOG.info("Mismatching timestamp for " + srcPath + " and " + parityPath + |
| ", moving on..."); |
| return; |
| } |
| |
| String uriPath = parityPath.toUri().getPath(); |
| int numBlocksFixed = 0; |
| List<LocatedBlock> corrupt = RaidDFSUtil.corruptBlocksInFile( |
| parityFs, uriPath, 0, parityFileSize); |
| for (LocatedBlock lb: corrupt) { |
| Block corruptBlock = lb.getBlock(); |
| long corruptOffset = lb.getStartOffset(); |
| |
| LOG.info("Found corrupt block " + corruptBlock + |
| ", offset " + corruptOffset); |
| |
| File localBlockFile = |
| File.createTempFile(corruptBlock.getBlockName(), ".tmp"); |
| localBlockFile.deleteOnExit(); |
| |
| try { |
| encoder.recoverParityBlockToFile(parityFs, srcPath, srcFileSize, |
| blockSize, parityPath, corruptOffset, localBlockFile); |
| // We have a the contents of the block, send them. |
| DatanodeInfo datanode = chooseDatanode(lb.getLocations()); |
| computeMetdataAndSendFixedBlock( |
| datanode, localBlockFile, lb, blockSize); |
| |
| numBlocksFixed++; |
| LOG.info("Adding " + parityPath + " to history"); |
| history.put(parityPath.toString(), new java.util.Date()); |
| } finally { |
| localBlockFile.delete(); |
| } |
| } |
| LOG.info("Fixed " + numBlocksFixed + " blocks in " + parityPath); |
| numFilesFixed++; |
| } |
| |
| /** |
| * Reads through a parity HAR part file, fixing corrupt blocks on the way. |
| * A HAR block can contain many file blocks, as long as the HAR part file |
| * block size is a multiple of the file block size. |
| */ |
| void processCorruptParityHarPartFile(Path partFile) throws IOException { |
| LOG.info("Processing corrupt file " + partFile); |
| // Get some basic information. |
| DistributedFileSystem dfs = getDFS(partFile); |
| FileStatus partFileStat = dfs.getFileStatus(partFile); |
| long partFileSize = partFileStat.getLen(); |
| long partFileBlockSize = partFileStat.getBlockSize(); |
| LOG.info(partFile + " has block size " + partFileBlockSize); |
| |
| // Find the path to the index file. |
| // Parity file HARs are only one level deep, so the index files is at the |
| // same level as the part file. |
| String harDirectory = partFile.toUri().getPath(); // Temporarily. |
| harDirectory = |
| harDirectory.substring(0, harDirectory.lastIndexOf(Path.SEPARATOR)); |
| Path indexFile = new Path(harDirectory + "/" + HarIndex.indexFileName); |
| FileStatus indexStat = dfs.getFileStatus(indexFile); |
| // Parses through the HAR index file. |
| HarIndex harIndex = new HarIndex(dfs.open(indexFile), indexStat.getLen()); |
| |
| String uriPath = partFile.toUri().getPath(); |
| int numBlocksFixed = 0; |
| List<LocatedBlock> corrupt = RaidDFSUtil.corruptBlocksInFile( |
| dfs, uriPath, 0, partFileSize); |
| for (LocatedBlock lb: corrupt) { |
| Block corruptBlock = lb.getBlock(); |
| long corruptOffset = lb.getStartOffset(); |
| |
| File localBlockFile = |
| File.createTempFile(corruptBlock.getBlockName(), ".tmp"); |
| localBlockFile.deleteOnExit(); |
| processCorruptParityHarPartBlock( |
| dfs, partFile, corruptBlock, corruptOffset, partFileStat, harIndex, |
| localBlockFile); |
| // Now we have recovered the part file block locally, send it. |
| try { |
| DatanodeInfo datanode = chooseDatanode(lb.getLocations()); |
| computeMetdataAndSendFixedBlock(datanode, localBlockFile, |
| lb, localBlockFile.length()); |
| numBlocksFixed++; |
| |
| LOG.info("Adding " + partFile + " to history"); |
| history.put(partFile.toString(), new java.util.Date()); |
| } finally { |
| localBlockFile.delete(); |
| } |
| } |
| LOG.info("Fixed " + numBlocksFixed + " blocks in " + partFile); |
| numFilesFixed++; |
| } |
| |
| /** |
| * This fixes a single part file block by recovering in sequence each |
| * parity block in the part file block. |
| */ |
| private void processCorruptParityHarPartBlock( |
| FileSystem dfs, Path partFile, Block corruptBlock, long corruptOffset, |
| FileStatus partFileStat, HarIndex harIndex, File localBlockFile) |
| throws IOException { |
| String partName = partFile.toUri().getPath(); // Temporarily. |
| partName = partName.substring(1 + partName.lastIndexOf(Path.SEPARATOR)); |
| |
| OutputStream out = new FileOutputStream(localBlockFile); |
| |
| try { |
| // A HAR part file block could map to several parity files. We need to |
| // use all of them to recover this block. |
| final long corruptEnd = Math.min(corruptOffset + partFileStat.getBlockSize(), |
| partFileStat.getLen()); |
| for (long offset = corruptOffset; offset < corruptEnd; ) { |
| HarIndex.IndexEntry entry = harIndex.findEntry(partName, offset); |
| if (entry == null) { |
| String msg = "Corrupt index file has no matching index entry for " + |
| partName + ":" + offset; |
| LOG.warn(msg); |
| throw new IOException(msg); |
| } |
| Path parityFile = new Path(entry.fileName); |
| Encoder encoder; |
| if (isXorParityFile(parityFile)) { |
| encoder = xorEncoder; |
| } else if (isRsParityFile(parityFile)) { |
| encoder = rsEncoder; |
| } else { |
| String msg = "Could not figure out parity file correctly"; |
| LOG.warn(msg); |
| throw new IOException(msg); |
| } |
| Path srcFile = sourcePathFromParityPath(parityFile); |
| FileStatus srcStat = dfs.getFileStatus(srcFile); |
| if (srcStat.getModificationTime() != entry.mtime) { |
| String msg = "Modification times of " + parityFile + " and " + srcFile + |
| " do not match."; |
| LOG.warn(msg); |
| throw new IOException(msg); |
| } |
| long corruptOffsetInParity = offset - entry.startOffset; |
| LOG.info(partFile + ":" + offset + " maps to " + |
| parityFile + ":" + corruptOffsetInParity + |
| " and will be recovered from " + srcFile); |
| encoder.recoverParityBlockToStream(dfs, srcFile, srcStat.getLen(), |
| srcStat.getBlockSize(), parityFile, corruptOffsetInParity, out); |
| // Finished recovery of one parity block. Since a parity block has the |
| // same size as a source block, we can move offset by source block size. |
| offset += srcStat.getBlockSize(); |
| LOG.info("Recovered " + srcStat.getBlockSize() + " part file bytes "); |
| if (offset > corruptEnd) { |
| String msg = |
| "Recovered block spills across part file blocks. Cannot continue..."; |
| throw new IOException(msg); |
| } |
| } |
| } finally { |
| out.close(); |
| } |
| } |
| |
| /** |
| * Choose a datanode (hostname:portnumber). The datanode is chosen at |
| * random from the live datanodes. |
| * @param locationsToAvoid locations to avoid. |
| * @return A string in the format name:port. |
| * @throws IOException |
| */ |
| private DatanodeInfo chooseDatanode(DatanodeInfo[] locationsToAvoid) |
| throws IOException { |
| DistributedFileSystem dfs = getDFS(new Path("/")); |
| DatanodeInfo[] live = dfs.getClient().datanodeReport( |
| DatanodeReportType.LIVE); |
| LOG.info("Choosing a datanode from " + live.length + |
| " live nodes while avoiding " + locationsToAvoid.length); |
| Random rand = new Random(); |
| DatanodeInfo chosen = null; |
| int maxAttempts = 1000; |
| for (int i = 0; i < maxAttempts && chosen == null; i++) { |
| int idx = rand.nextInt(live.length); |
| chosen = live[idx]; |
| for (DatanodeInfo avoid: locationsToAvoid) { |
| if (chosen.name.equals(avoid.name)) { |
| LOG.info("Avoiding " + avoid.name); |
| chosen = null; |
| break; |
| } |
| } |
| } |
| if (chosen == null) { |
| throw new IOException("Could not choose datanode"); |
| } |
| LOG.info("Choosing datanode " + chosen.name); |
| return chosen; |
| } |
| |
| /** |
| * Reads data from the data stream provided and computes metadata. |
| */ |
| static DataInputStream computeMetadata( |
| Configuration conf, InputStream dataStream) throws IOException { |
| ByteArrayOutputStream mdOutBase = new ByteArrayOutputStream(1024*1024); |
| DataOutputStream mdOut = new DataOutputStream(mdOutBase); |
| |
| // First, write out the version. |
| mdOut.writeShort(FSDataset.METADATA_VERSION); |
| |
| // Create a summer and write out its header. |
| int bytesPerChecksum = conf.getInt("io.bytes.per.checksum", 512); |
| DataChecksum sum = DataChecksum.newDataChecksum( |
| DataChecksum.CHECKSUM_CRC32, |
| bytesPerChecksum); |
| sum.writeHeader(mdOut); |
| |
| // Buffer to read in a chunk of data. |
| byte[] buf = new byte[bytesPerChecksum]; |
| // Buffer to store the checksum bytes. |
| byte[] chk = new byte[sum.getChecksumSize()]; |
| |
| // Read data till we reach the end of the input stream. |
| int bytesSinceFlush = 0; |
| while (true) { |
| // Read some bytes. |
| int bytesRead = dataStream.read( |
| buf, bytesSinceFlush, bytesPerChecksum-bytesSinceFlush); |
| if (bytesRead == -1) { |
| if (bytesSinceFlush > 0) { |
| boolean reset = true; |
| sum.writeValue(chk, 0, reset); // This also resets the sum. |
| // Write the checksum to the stream. |
| mdOut.write(chk, 0, chk.length); |
| bytesSinceFlush = 0; |
| } |
| break; |
| } |
| // Update the checksum. |
| sum.update(buf, bytesSinceFlush, bytesRead); |
| bytesSinceFlush += bytesRead; |
| |
| // Flush the checksum if necessary. |
| if (bytesSinceFlush == bytesPerChecksum) { |
| boolean reset = true; |
| sum.writeValue(chk, 0, reset); // This also resets the sum. |
| // Write the checksum to the stream. |
| mdOut.write(chk, 0, chk.length); |
| bytesSinceFlush = 0; |
| } |
| } |
| |
| byte[] mdBytes = mdOutBase.toByteArray(); |
| return new DataInputStream(new ByteArrayInputStream(mdBytes)); |
| } |
| |
| private void computeMetdataAndSendFixedBlock( |
| DatanodeInfo datanode, |
| File localBlockFile, LocatedBlock block, long blockSize |
| ) throws IOException { |
| |
| LOG.info("Computing metdata"); |
| InputStream blockContents = null; |
| DataInputStream blockMetadata = null; |
| try { |
| blockContents = new FileInputStream(localBlockFile); |
| blockMetadata = computeMetadata(getConf(), blockContents); |
| blockContents.close(); |
| // Reopen |
| blockContents = new FileInputStream(localBlockFile); |
| sendFixedBlock(datanode, blockContents, blockMetadata, block, blockSize); |
| } finally { |
| if (blockContents != null) { |
| blockContents.close(); |
| blockContents = null; |
| } |
| if (blockMetadata != null) { |
| blockMetadata.close(); |
| blockMetadata = null; |
| } |
| } |
| } |
| |
| /** |
| * Send a generated block to a datanode. |
| * @param datanode Chosen datanode name in host:port form. |
| * @param blockContents Stream with the block contents. |
| * @param corruptBlock Block identifying the block to be sent. |
| * @param blockSize size of the block. |
| * @throws IOException |
| */ |
| private void sendFixedBlock( |
| DatanodeInfo datanode, |
| final InputStream blockContents, DataInputStream metadataIn, |
| LocatedBlock block, long blockSize |
| ) throws IOException { |
| InetSocketAddress target = NetUtils.createSocketAddr(datanode.name); |
| Socket sock = SocketChannel.open().socket(); |
| |
| int readTimeout = getConf().getInt(BLOCKFIX_READ_TIMEOUT, |
| HdfsConstants.READ_TIMEOUT); |
| NetUtils.connect(sock, target, readTimeout); |
| sock.setSoTimeout(readTimeout); |
| |
| int writeTimeout = getConf().getInt(BLOCKFIX_WRITE_TIMEOUT, |
| HdfsConstants.WRITE_TIMEOUT); |
| |
| OutputStream baseStream = NetUtils.getOutputStream(sock, writeTimeout); |
| DataOutputStream out = new DataOutputStream( |
| new BufferedOutputStream(baseStream, FSConstants.SMALL_BUFFER_SIZE)); |
| |
| boolean corruptChecksumOk = false; |
| boolean chunkOffsetOK = false; |
| boolean verifyChecksum = true; |
| boolean transferToAllowed = false; |
| |
| try { |
| LOG.info("Sending block " + block.getBlock() + |
| " from " + sock.getLocalSocketAddress().toString() + |
| " to " + sock.getRemoteSocketAddress().toString() + |
| " " + blockSize + " bytes"); |
| RaidBlockSender blockSender = new RaidBlockSender( |
| block.getBlock(), blockSize, 0, blockSize, |
| corruptChecksumOk, chunkOffsetOK, verifyChecksum, transferToAllowed, |
| metadataIn, new RaidBlockSender.InputStreamFactory() { |
| @Override |
| public InputStream createStream(long offset) throws IOException { |
| // we are passing 0 as the offset above, so we can safely ignore |
| // the offset passed |
| return blockContents; |
| } |
| }); |
| |
| DatanodeInfo[] nodes = new DatanodeInfo[]{datanode}; |
| DataTransferProtocol.Sender.opWriteBlock( |
| out, block.getBlock(), 1, |
| DataTransferProtocol.BlockConstructionStage.PIPELINE_SETUP_CREATE, |
| 0, blockSize, 0, "", null, nodes, block.getBlockToken()); |
| blockSender.sendBlock(out, baseStream); |
| |
| LOG.info("Sent block " + block.getBlock() + " to " + datanode.name); |
| } finally { |
| out.close(); |
| } |
| } |
| |
| /** |
| * returns the source file corresponding to a parity file |
| */ |
| Path sourcePathFromParityPath(Path parityPath) { |
| String parityPathStr = parityPath.toUri().getPath(); |
| if (parityPathStr.startsWith(xorPrefix)) { |
| // Remove the prefix to get the source file. |
| String src = parityPathStr.replaceFirst(xorPrefix, "/"); |
| return new Path(src); |
| } else if (parityPathStr.startsWith(rsPrefix)) { |
| // Remove the prefix to get the source file. |
| String src = parityPathStr.replaceFirst(rsPrefix, "/"); |
| return new Path(src); |
| } |
| return null; |
| } |
| } |
| |
