[120] | 1 | /** |
---|
| 2 | * Licensed to the Apache Software Foundation (ASF) under one |
---|
| 3 | * or more contributor license agreements. See the NOTICE file |
---|
| 4 | * distributed with this work for additional information |
---|
| 5 | * regarding copyright ownership. The ASF licenses this file |
---|
| 6 | * to you under the Apache License, Version 2.0 (the |
---|
| 7 | * "License"); you may not use this file except in compliance |
---|
| 8 | * with the License. You may obtain a copy of the License at |
---|
| 9 | * |
---|
| 10 | * http://www.apache.org/licenses/LICENSE-2.0 |
---|
| 11 | * |
---|
| 12 | * Unless required by applicable law or agreed to in writing, software |
---|
| 13 | * distributed under the License is distributed on an "AS IS" BASIS, |
---|
| 14 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
---|
| 15 | * See the License for the specific language governing permissions and |
---|
| 16 | * limitations under the License. |
---|
| 17 | */ |
---|
| 18 | |
---|
| 19 | package org.apache.hadoop.hdfs; |
---|
| 20 | |
---|
| 21 | import java.io.IOException; |
---|
| 22 | import java.util.Date; |
---|
| 23 | |
---|
| 24 | import org.apache.commons.logging.Log; |
---|
| 25 | import org.apache.commons.logging.LogFactory; |
---|
| 26 | import org.apache.hadoop.conf.Configuration; |
---|
| 27 | import org.apache.hadoop.fs.FSDataInputStream; |
---|
| 28 | import org.apache.hadoop.fs.FSDataOutputStream; |
---|
| 29 | import org.apache.hadoop.fs.FileSystem; |
---|
| 30 | import org.apache.hadoop.fs.Path; |
---|
| 31 | import org.apache.hadoop.mapred.JobConf; |
---|
| 32 | import org.apache.hadoop.util.StringUtils; |
---|
| 33 | |
---|
| 34 | /** |
---|
| 35 | * This program executes a specified operation that applies load to |
---|
| 36 | * the NameNode. Possible operations include create/writing files, |
---|
| 37 | * opening/reading files, renaming files, and deleting files. |
---|
| 38 | * |
---|
| 39 | * When run simultaneously on multiple nodes, this program functions |
---|
| 40 | * as a stress-test and benchmark for namenode, especially when |
---|
| 41 | * the number of bytes written to each file is small. |
---|
| 42 | * |
---|
| 43 | * This version does not use the map reduce framework |
---|
| 44 | * |
---|
| 45 | */ |
---|
| 46 | public class NNBenchWithoutMR { |
---|
| 47 | |
---|
| 48 | private static final Log LOG = LogFactory.getLog( |
---|
| 49 | "org.apache.hadoop.hdfs.NNBench"); |
---|
| 50 | |
---|
| 51 | // variable initialzed from command line arguments |
---|
| 52 | private static long startTime = 0; |
---|
| 53 | private static int numFiles = 0; |
---|
| 54 | private static long bytesPerBlock = 1; |
---|
| 55 | private static long blocksPerFile = 0; |
---|
| 56 | private static long bytesPerFile = 1; |
---|
| 57 | private static Path baseDir = null; |
---|
| 58 | |
---|
| 59 | // variables initialized in main() |
---|
| 60 | private static FileSystem fileSys = null; |
---|
| 61 | private static Path taskDir = null; |
---|
| 62 | private static String uniqueId = null; |
---|
| 63 | private static byte[] buffer; |
---|
| 64 | private static long maxExceptionsPerFile = 200; |
---|
| 65 | |
---|
| 66 | /** |
---|
| 67 | * Returns when the current number of seconds from the epoch equals |
---|
| 68 | * the command line argument given by <code>-startTime</code>. |
---|
| 69 | * This allows multiple instances of this program, running on clock |
---|
| 70 | * synchronized nodes, to start at roughly the same time. |
---|
| 71 | */ |
---|
| 72 | static void barrier() { |
---|
| 73 | long sleepTime; |
---|
| 74 | while ((sleepTime = startTime - System.currentTimeMillis()) > 0) { |
---|
| 75 | try { |
---|
| 76 | Thread.sleep(sleepTime); |
---|
| 77 | } catch (InterruptedException ex) { |
---|
| 78 | } |
---|
| 79 | } |
---|
| 80 | } |
---|
| 81 | |
---|
| 82 | static private void handleException(String operation, Throwable e, |
---|
| 83 | int singleFileExceptions) { |
---|
| 84 | LOG.warn("Exception while " + operation + ": " + |
---|
| 85 | StringUtils.stringifyException(e)); |
---|
| 86 | if (singleFileExceptions >= maxExceptionsPerFile) { |
---|
| 87 | throw new RuntimeException(singleFileExceptions + |
---|
| 88 | " exceptions for a single file exceeds threshold. Aborting"); |
---|
| 89 | } |
---|
| 90 | } |
---|
| 91 | |
---|
| 92 | /** |
---|
| 93 | * Create and write to a given number of files. Repeat each remote |
---|
| 94 | * operation until is suceeds (does not throw an exception). |
---|
| 95 | * |
---|
| 96 | * @return the number of exceptions caught |
---|
| 97 | */ |
---|
| 98 | static int createWrite() { |
---|
| 99 | int totalExceptions = 0; |
---|
| 100 | FSDataOutputStream out = null; |
---|
| 101 | boolean success = false; |
---|
| 102 | for (int index = 0; index < numFiles; index++) { |
---|
| 103 | int singleFileExceptions = 0; |
---|
| 104 | do { // create file until is succeeds or max exceptions reached |
---|
| 105 | try { |
---|
| 106 | out = fileSys.create( |
---|
| 107 | new Path(taskDir, "" + index), false, 512, (short)1, bytesPerBlock); |
---|
| 108 | success = true; |
---|
| 109 | } catch (IOException ioe) { |
---|
| 110 | success=false; |
---|
| 111 | totalExceptions++; |
---|
| 112 | handleException("creating file #" + index, ioe, ++singleFileExceptions); |
---|
| 113 | } |
---|
| 114 | } while (!success); |
---|
| 115 | long toBeWritten = bytesPerFile; |
---|
| 116 | while (toBeWritten > 0) { |
---|
| 117 | int nbytes = (int) Math.min(buffer.length, toBeWritten); |
---|
| 118 | toBeWritten -= nbytes; |
---|
| 119 | try { // only try once |
---|
| 120 | out.write(buffer, 0, nbytes); |
---|
| 121 | } catch (IOException ioe) { |
---|
| 122 | totalExceptions++; |
---|
| 123 | handleException("writing to file #" + index, ioe, ++singleFileExceptions); |
---|
| 124 | } |
---|
| 125 | } |
---|
| 126 | do { // close file until is succeeds |
---|
| 127 | try { |
---|
| 128 | out.close(); |
---|
| 129 | success = true; |
---|
| 130 | } catch (IOException ioe) { |
---|
| 131 | success=false; |
---|
| 132 | totalExceptions++; |
---|
| 133 | handleException("closing file #" + index, ioe, ++singleFileExceptions); |
---|
| 134 | } |
---|
| 135 | } while (!success); |
---|
| 136 | } |
---|
| 137 | return totalExceptions; |
---|
| 138 | } |
---|
| 139 | |
---|
| 140 | /** |
---|
| 141 | * Open and read a given number of files. |
---|
| 142 | * |
---|
| 143 | * @return the number of exceptions caught |
---|
| 144 | */ |
---|
| 145 | static int openRead() { |
---|
| 146 | int totalExceptions = 0; |
---|
| 147 | FSDataInputStream in = null; |
---|
| 148 | for (int index = 0; index < numFiles; index++) { |
---|
| 149 | int singleFileExceptions = 0; |
---|
| 150 | try { |
---|
| 151 | in = fileSys.open(new Path(taskDir, "" + index), 512); |
---|
| 152 | long toBeRead = bytesPerFile; |
---|
| 153 | while (toBeRead > 0) { |
---|
| 154 | int nbytes = (int) Math.min(buffer.length, toBeRead); |
---|
| 155 | toBeRead -= nbytes; |
---|
| 156 | try { // only try once |
---|
| 157 | in.read(buffer, 0, nbytes); |
---|
| 158 | } catch (IOException ioe) { |
---|
| 159 | totalExceptions++; |
---|
| 160 | handleException("reading from file #" + index, ioe, ++singleFileExceptions); |
---|
| 161 | } |
---|
| 162 | } |
---|
| 163 | in.close(); |
---|
| 164 | } catch (IOException ioe) { |
---|
| 165 | totalExceptions++; |
---|
| 166 | handleException("opening file #" + index, ioe, ++singleFileExceptions); |
---|
| 167 | } |
---|
| 168 | } |
---|
| 169 | return totalExceptions; |
---|
| 170 | } |
---|
| 171 | |
---|
| 172 | /** |
---|
| 173 | * Rename a given number of files. Repeat each remote |
---|
| 174 | * operation until is suceeds (does not throw an exception). |
---|
| 175 | * |
---|
| 176 | * @return the number of exceptions caught |
---|
| 177 | */ |
---|
| 178 | static int rename() { |
---|
| 179 | int totalExceptions = 0; |
---|
| 180 | boolean success = false; |
---|
| 181 | for (int index = 0; index < numFiles; index++) { |
---|
| 182 | int singleFileExceptions = 0; |
---|
| 183 | do { // rename file until is succeeds |
---|
| 184 | try { |
---|
| 185 | boolean result = fileSys.rename( |
---|
| 186 | new Path(taskDir, "" + index), new Path(taskDir, "A" + index)); |
---|
| 187 | success = true; |
---|
| 188 | } catch (IOException ioe) { |
---|
| 189 | success=false; |
---|
| 190 | totalExceptions++; |
---|
| 191 | handleException("creating file #" + index, ioe, ++singleFileExceptions); |
---|
| 192 | } |
---|
| 193 | } while (!success); |
---|
| 194 | } |
---|
| 195 | return totalExceptions; |
---|
| 196 | } |
---|
| 197 | |
---|
| 198 | /** |
---|
| 199 | * Delete a given number of files. Repeat each remote |
---|
| 200 | * operation until is suceeds (does not throw an exception). |
---|
| 201 | * |
---|
| 202 | * @return the number of exceptions caught |
---|
| 203 | */ |
---|
| 204 | static int delete() { |
---|
| 205 | int totalExceptions = 0; |
---|
| 206 | boolean success = false; |
---|
| 207 | for (int index = 0; index < numFiles; index++) { |
---|
| 208 | int singleFileExceptions = 0; |
---|
| 209 | do { // delete file until is succeeds |
---|
| 210 | try { |
---|
| 211 | boolean result = fileSys.delete(new Path(taskDir, "A" + index), true); |
---|
| 212 | success = true; |
---|
| 213 | } catch (IOException ioe) { |
---|
| 214 | success=false; |
---|
| 215 | totalExceptions++; |
---|
| 216 | handleException("creating file #" + index, ioe, ++singleFileExceptions); |
---|
| 217 | } |
---|
| 218 | } while (!success); |
---|
| 219 | } |
---|
| 220 | return totalExceptions; |
---|
| 221 | } |
---|
| 222 | |
---|
| 223 | /** |
---|
| 224 | * This launches a given namenode operation (<code>-operation</code>), |
---|
| 225 | * starting at a given time (<code>-startTime</code>). The files used |
---|
| 226 | * by the openRead, rename, and delete operations are the same files |
---|
| 227 | * created by the createWrite operation. Typically, the program |
---|
| 228 | * would be run four times, once for each operation in this order: |
---|
| 229 | * createWrite, openRead, rename, delete. |
---|
| 230 | * |
---|
| 231 | * <pre> |
---|
| 232 | * Usage: nnbench |
---|
| 233 | * -operation <one of createWrite, openRead, rename, or delete> |
---|
| 234 | * -baseDir <base output/input DFS path> |
---|
| 235 | * -startTime <time to start, given in seconds from the epoch> |
---|
| 236 | * -numFiles <number of files to create, read, rename, or delete> |
---|
| 237 | * -blocksPerFile <number of blocks to create per file> |
---|
| 238 | * [-bytesPerBlock <number of bytes to write to each block, default is 1>] |
---|
| 239 | * [-bytesPerChecksum <value for io.bytes.per.checksum>] |
---|
| 240 | * </pre> |
---|
| 241 | * |
---|
| 242 | * @throws IOException indicates a problem with test startup |
---|
| 243 | */ |
---|
| 244 | public static void main(String[] args) throws IOException { |
---|
| 245 | String version = "NameNodeBenchmark.0.3"; |
---|
| 246 | System.out.println(version); |
---|
| 247 | int bytesPerChecksum = -1; |
---|
| 248 | |
---|
| 249 | String usage = |
---|
| 250 | "Usage: nnbench " + |
---|
| 251 | " -operation <one of createWrite, openRead, rename, or delete> " + |
---|
| 252 | " -baseDir <base output/input DFS path> " + |
---|
| 253 | " -startTime <time to start, given in seconds from the epoch> " + |
---|
| 254 | " -numFiles <number of files to create> " + |
---|
| 255 | " -blocksPerFile <number of blocks to create per file> " + |
---|
| 256 | " [-bytesPerBlock <number of bytes to write to each block, default is 1>] " + |
---|
| 257 | " [-bytesPerChecksum <value for io.bytes.per.checksum>]" + |
---|
| 258 | "Note: bytesPerBlock MUST be a multiple of bytesPerChecksum"; |
---|
| 259 | |
---|
| 260 | String operation = null; |
---|
| 261 | for (int i = 0; i < args.length; i++) { // parse command line |
---|
| 262 | if (args[i].equals("-baseDir")) { |
---|
| 263 | baseDir = new Path(args[++i]); |
---|
| 264 | } else if (args[i].equals("-numFiles")) { |
---|
| 265 | numFiles = Integer.parseInt(args[++i]); |
---|
| 266 | } else if (args[i].equals("-blocksPerFile")) { |
---|
| 267 | blocksPerFile = Integer.parseInt(args[++i]); |
---|
| 268 | } else if (args[i].equals("-bytesPerBlock")) { |
---|
| 269 | bytesPerBlock = Long.parseLong(args[++i]); |
---|
| 270 | } else if (args[i].equals("-bytesPerChecksum")) { |
---|
| 271 | bytesPerChecksum = Integer.parseInt(args[++i]); |
---|
| 272 | } else if (args[i].equals("-startTime")) { |
---|
| 273 | startTime = Long.parseLong(args[++i]) * 1000; |
---|
| 274 | } else if (args[i].equals("-operation")) { |
---|
| 275 | operation = args[++i]; |
---|
| 276 | } else { |
---|
| 277 | System.out.println(usage); |
---|
| 278 | System.exit(-1); |
---|
| 279 | } |
---|
| 280 | } |
---|
| 281 | bytesPerFile = bytesPerBlock * blocksPerFile; |
---|
| 282 | |
---|
| 283 | JobConf jobConf = new JobConf(new Configuration(), NNBench.class); |
---|
| 284 | |
---|
| 285 | if ( bytesPerChecksum < 0 ) { // if it is not set in cmdline |
---|
| 286 | bytesPerChecksum = jobConf.getInt("io.bytes.per.checksum", 512); |
---|
| 287 | } |
---|
| 288 | jobConf.set("io.bytes.per.checksum", Integer.toString(bytesPerChecksum)); |
---|
| 289 | |
---|
| 290 | System.out.println("Inputs: "); |
---|
| 291 | System.out.println(" operation: " + operation); |
---|
| 292 | System.out.println(" baseDir: " + baseDir); |
---|
| 293 | System.out.println(" startTime: " + startTime); |
---|
| 294 | System.out.println(" numFiles: " + numFiles); |
---|
| 295 | System.out.println(" blocksPerFile: " + blocksPerFile); |
---|
| 296 | System.out.println(" bytesPerBlock: " + bytesPerBlock); |
---|
| 297 | System.out.println(" bytesPerChecksum: " + bytesPerChecksum); |
---|
| 298 | |
---|
| 299 | if (operation == null || // verify args |
---|
| 300 | baseDir == null || |
---|
| 301 | numFiles < 1 || |
---|
| 302 | blocksPerFile < 1 || |
---|
| 303 | bytesPerBlock < 0 || |
---|
| 304 | bytesPerBlock % bytesPerChecksum != 0) |
---|
| 305 | { |
---|
| 306 | System.err.println(usage); |
---|
| 307 | System.exit(-1); |
---|
| 308 | } |
---|
| 309 | |
---|
| 310 | fileSys = FileSystem.get(jobConf); |
---|
| 311 | uniqueId = java.net.InetAddress.getLocalHost().getHostName(); |
---|
| 312 | taskDir = new Path(baseDir, uniqueId); |
---|
| 313 | // initialize buffer used for writing/reading file |
---|
| 314 | buffer = new byte[(int) Math.min(bytesPerFile, 32768L)]; |
---|
| 315 | |
---|
| 316 | Date execTime; |
---|
| 317 | Date endTime; |
---|
| 318 | long duration; |
---|
| 319 | int exceptions = 0; |
---|
| 320 | barrier(); // wait for coordinated start time |
---|
| 321 | execTime = new Date(); |
---|
| 322 | System.out.println("Job started: " + startTime); |
---|
| 323 | if (operation.equals("createWrite")) { |
---|
| 324 | if (!fileSys.mkdirs(taskDir)) { |
---|
| 325 | throw new IOException("Mkdirs failed to create " + taskDir.toString()); |
---|
| 326 | } |
---|
| 327 | exceptions = createWrite(); |
---|
| 328 | } else if (operation.equals("openRead")) { |
---|
| 329 | exceptions = openRead(); |
---|
| 330 | } else if (operation.equals("rename")) { |
---|
| 331 | exceptions = rename(); |
---|
| 332 | } else if (operation.equals("delete")) { |
---|
| 333 | exceptions = delete(); |
---|
| 334 | } else { |
---|
| 335 | System.err.println(usage); |
---|
| 336 | System.exit(-1); |
---|
| 337 | } |
---|
| 338 | endTime = new Date(); |
---|
| 339 | System.out.println("Job ended: " + endTime); |
---|
| 340 | duration = (endTime.getTime() - execTime.getTime()) /1000; |
---|
| 341 | System.out.println("The " + operation + " job took " + duration + " seconds."); |
---|
| 342 | System.out.println("The job recorded " + exceptions + " exceptions."); |
---|
| 343 | } |
---|
| 344 | } |
---|