| /** |
| * 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.fs.slive; |
| |
| import java.io.IOException; |
| import java.util.List; |
| import java.util.Random; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.slive.DataWriter.GenerateOutput; |
| import org.apache.hadoop.fs.slive.OperationOutput.OutputType; |
| |
| /** |
| * Operation which selects a random file and a random number of bytes to create |
| * that file with (from the write size option) and selects a random block size |
| * (from the block size option) and a random replication amount (from the |
| * replication option) and attempts to create a file with those options. |
| * |
| * This operation will capture statistics on success for bytes written, time |
| * taken (milliseconds), and success count and on failure it will capture the |
| * number of failures and the time taken (milliseconds) to fail. |
| */ |
| class CreateOp extends Operation { |
| |
| private static final Log LOG = LogFactory.getLog(CreateOp.class); |
| |
| private static int DEF_IO_BUFFER_SIZE = 4096; |
| |
| private static final String IO_BUF_CONFIG = ("io.file.buffer.size"); |
| |
| CreateOp(ConfigExtractor cfg, Random rnd) { |
| super(CreateOp.class.getSimpleName(), cfg, rnd); |
| } |
| |
| /** |
| * Returns the block size to use (aligned to nearest BYTES_PER_CHECKSUM if |
| * configuration says a value exists) - this will avoid the warnings caused by |
| * this not occurring and the file will not be created if it is not correct... |
| * |
| * @return long |
| */ |
| private long determineBlockSize() { |
| Range<Long> blockSizeRange = getConfig().getBlockSize(); |
| long blockSize = Range.betweenPositive(getRandom(), blockSizeRange); |
| Long byteChecksum = getConfig().getByteCheckSum(); |
| if (byteChecksum == null) { |
| return blockSize; |
| } |
| // adjust to nearest multiple |
| long full = (blockSize / byteChecksum) * byteChecksum; |
| long toFull = blockSize - full; |
| if (toFull >= (byteChecksum / 2)) { |
| full += byteChecksum; |
| } |
| // adjust if over extended |
| if (full > blockSizeRange.getUpper()) { |
| full = blockSizeRange.getUpper(); |
| } |
| if (full < blockSizeRange.getLower()) { |
| full = blockSizeRange.getLower(); |
| } |
| return full; |
| } |
| |
| /** |
| * Gets the replication amount |
| * |
| * @return short |
| */ |
| private short determineReplication() { |
| Range<Short> replicationAmountRange = getConfig().getReplication(); |
| Range<Long> repRange = new Range<Long>(replicationAmountRange.getLower() |
| .longValue(), replicationAmountRange.getUpper().longValue()); |
| short replicationAmount = (short) Range.betweenPositive(getRandom(), |
| repRange); |
| return replicationAmount; |
| } |
| |
| /** |
| * Gets the output buffering size to use |
| * |
| * @return int |
| */ |
| private int getBufferSize() { |
| return getConfig().getConfig().getInt(IO_BUF_CONFIG, DEF_IO_BUFFER_SIZE); |
| } |
| |
| /** |
| * Gets the file to create |
| * |
| * @return Path |
| */ |
| protected Path getCreateFile() { |
| Path fn = getFinder().getFile(); |
| return fn; |
| } |
| |
| @Override // Operation |
| List<OperationOutput> run(FileSystem fs) { |
| List<OperationOutput> out = super.run(fs); |
| FSDataOutputStream os = null; |
| try { |
| Path fn = getCreateFile(); |
| Range<Long> writeSizeRange = getConfig().getWriteSize(); |
| long writeSize = 0; |
| long blockSize = determineBlockSize(); |
| short replicationAmount = determineReplication(); |
| if (getConfig().shouldWriteUseBlockSize()) { |
| writeSizeRange = getConfig().getBlockSize(); |
| } |
| writeSize = Range.betweenPositive(getRandom(), writeSizeRange); |
| long bytesWritten = 0; |
| long timeTaken = 0; |
| int bufSize = getBufferSize(); |
| boolean overWrite = false; |
| DataWriter writer = new DataWriter(getRandom()); |
| LOG.info("Attempting to create file at " + fn + " of size " |
| + Helper.toByteInfo(writeSize) + " using blocksize " |
| + Helper.toByteInfo(blockSize) + " and replication amount " |
| + replicationAmount); |
| { |
| // open & create |
| long startTime = Timer.now(); |
| os = fs.create(fn, overWrite, bufSize, replicationAmount, blockSize); |
| timeTaken += Timer.elapsed(startTime); |
| // write the given length |
| GenerateOutput stats = writer.writeSegment(writeSize, os); |
| bytesWritten += stats.getBytesWritten(); |
| timeTaken += stats.getTimeTaken(); |
| // capture close time |
| startTime = Timer.now(); |
| os.close(); |
| os = null; |
| timeTaken += Timer.elapsed(startTime); |
| } |
| LOG.info("Created file at " + fn + " of size " |
| + Helper.toByteInfo(bytesWritten) + " bytes using blocksize " |
| + Helper.toByteInfo(blockSize) + " and replication amount " |
| + replicationAmount + " in " + timeTaken + " milliseconds"); |
| // collect all the stats |
| out.add(new OperationOutput(OutputType.LONG, getType(), |
| ReportWriter.OK_TIME_TAKEN, timeTaken)); |
| out.add(new OperationOutput(OutputType.LONG, getType(), |
| ReportWriter.BYTES_WRITTEN, bytesWritten)); |
| out.add(new OperationOutput(OutputType.LONG, getType(), |
| ReportWriter.SUCCESSES, 1L)); |
| } catch (IOException e) { |
| out.add(new OperationOutput(OutputType.LONG, getType(), |
| ReportWriter.FAILURES, 1L)); |
| LOG.warn("Error with creating", e); |
| } finally { |
| if (os != null) { |
| try { |
| os.close(); |
| } catch (IOException e) { |
| LOG.warn("Error closing create stream", e); |
| } |
| } |
| } |
| return out; |
| } |
| } |