View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one or more
5    * contributor license agreements. See the NOTICE file distributed with this
6    * work for additional information regarding copyright ownership. The ASF
7    * licenses this file to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance with the License.
9    * You may obtain a copy of the License at
10   *
11   * http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16   * License for the specific language governing permissions and limitations
17   * under the License.
18   */
19  package org.apache.hadoop.hbase.io.hfile.bucket;
20  
21  import java.io.File;
22  import java.io.IOException;
23  import java.io.RandomAccessFile;
24  import java.nio.ByteBuffer;
25  import java.nio.channels.FileChannel;
26  import java.util.Arrays;
27  
28  import org.apache.commons.logging.Log;
29  import org.apache.commons.logging.LogFactory;
30  import org.apache.hadoop.hbase.classification.InterfaceAudience;
31  import org.apache.hadoop.util.StringUtils;
32  
33  /**
34   * IO engine that stores data to a file on the local file system.
35   */
36  @InterfaceAudience.Private
37  public class FileIOEngine implements IOEngine {
38    private static final Log LOG = LogFactory.getLog(FileIOEngine.class);
39    public static final String FILE_DELIMITER = ",";
40    private final String[] filePaths;
41    private final FileChannel[] fileChannels;
42    private final RandomAccessFile[] rafs;
43  
44    private final long sizePerFile;
45    private final long capacity;
46  
47    private FileReadAccessor readAccessor = new FileReadAccessor();
48    private FileWriteAccessor writeAccessor = new FileWriteAccessor();
49  
50    public FileIOEngine(long capacity, String... filePaths) throws IOException {
51      this.sizePerFile = capacity / filePaths.length;
52      this.capacity = this.sizePerFile * filePaths.length;
53      this.filePaths = filePaths;
54      this.fileChannels = new FileChannel[filePaths.length];
55      this.rafs = new RandomAccessFile[filePaths.length];
56      for (int i = 0; i < filePaths.length; i++) {
57        String filePath = filePaths[i];
58        try {
59          rafs[i] = new RandomAccessFile(filePath, "rw");
60          long totalSpace = new File(filePath).getTotalSpace();
61          if (totalSpace < sizePerFile) {
62            // The next setting length will throw exception,logging this message
63            // is just used for the detail reason of exception,
64            String msg = "Only " + StringUtils.byteDesc(totalSpace)
65                + " total space under " + filePath + ", not enough for requested "
66                + StringUtils.byteDesc(sizePerFile);
67            LOG.warn(msg);
68          }
69          rafs[i].setLength(sizePerFile);
70          fileChannels[i] = rafs[i].getChannel();
71          LOG.info("Allocating cache " + StringUtils.byteDesc(sizePerFile)
72              + ", on the path:" + filePath);
73        } catch (IOException fex) {
74          LOG.error("Failed allocating cache on " + filePath, fex);
75          shutdown();
76          throw fex;
77        }
78      }
79    }
80  
81    @Override
82    public String toString() {
83      return "ioengine=" + this.getClass().getSimpleName() + ", paths="
84          + Arrays.asList(filePaths) + ", capacity=" + String.format("%,d", this.capacity);
85    }
86  
87    /**
88     * File IO engine is always able to support persistent storage for the cache
89     * @return true
90     */
91    @Override
92    public boolean isPersistent() {
93      return true;
94    }
95  
96    /**
97     * Transfers data from file to the given byte buffer
98     * @param dstBuffer the given byte buffer into which bytes are to be written
99     * @param offset The offset in the file where the first byte to be read
100    * @return number of bytes read
101    * @throws IOException
102    */
103   @Override
104   public int read(ByteBuffer dstBuffer, long offset) throws IOException {
105     return accessFile(readAccessor, dstBuffer, offset);
106   }
107 
108   /**
109    * Transfers data from the given byte buffer to file
110    * @param srcBuffer the given byte buffer from which bytes are to be read
111    * @param offset The offset in the file where the first byte to be written
112    * @throws IOException
113    */
114   @Override
115   public void write(ByteBuffer srcBuffer, long offset) throws IOException {
116     accessFile(writeAccessor, srcBuffer, offset);
117   }
118 
119   /**
120    * Sync the data to file after writing
121    * @throws IOException
122    */
123   @Override
124   public void sync() throws IOException {
125     for (int i = 0; i < fileChannels.length; i++) {
126       try {
127         if (fileChannels[i] != null) {
128           fileChannels[i].force(true);
129         }
130       } catch (IOException ie) {
131         LOG.warn("Failed syncing data to " + this.filePaths[i]);
132         throw ie;
133       }
134     }
135   }
136 
137   /**
138    * Close the file
139    */
140   @Override
141   public void shutdown() {
142     for (int i = 0; i < filePaths.length; i++) {
143       try {
144         if (fileChannels[i] != null) {
145           fileChannels[i].close();
146         }
147         if (rafs[i] != null) {
148           rafs[i].close();
149         }
150       } catch (IOException ex) {
151         LOG.error("Failed closing " + filePaths[i] + " when shudown the IOEngine", ex);
152       }
153     }
154   }
155 
156   private int accessFile(FileAccessor accessor, ByteBuffer buffer, long globalOffset)
157       throws IOException {
158     int startFileNum = getFileNum(globalOffset);
159     int remainingAccessDataLen = buffer.remaining();
160     int endFileNum = getFileNum(globalOffset + remainingAccessDataLen - 1);
161     int accessFileNum = startFileNum;
162     long accessOffset = getAbsoluteOffsetInFile(accessFileNum, globalOffset);
163     int bufLimit = buffer.limit();
164     while (true) {
165       FileChannel fileChannel = fileChannels[accessFileNum];
166       if (endFileNum > accessFileNum) {
167         // short the limit;
168         buffer.limit((int) (buffer.limit() - remainingAccessDataLen + sizePerFile - accessOffset));
169       }
170       int accessLen = accessor.access(fileChannel, buffer, accessOffset);
171       // recover the limit
172       buffer.limit(bufLimit);
173       if (accessLen < remainingAccessDataLen) {
174         remainingAccessDataLen -= accessLen;
175         accessFileNum++;
176         accessOffset = 0;
177       } else {
178         break;
179       }
180       if (accessFileNum >= fileChannels.length) {
181         throw new IOException("Required data len " + StringUtils.byteDesc(buffer.remaining())
182             + " exceed the engine's capacity " + StringUtils.byteDesc(capacity) + " where offset="
183             + globalOffset);
184       }
185     }
186     return bufLimit;
187   }
188 
189   /**
190    * Get the absolute offset in given file with the relative global offset.
191    * @param fileNum
192    * @param globalOffset
193    * @return the absolute offset
194    */
195   private long getAbsoluteOffsetInFile(int fileNum, long globalOffset) {
196     return globalOffset - fileNum * sizePerFile;
197   }
198 
199   private int getFileNum(long offset) {
200     if (offset < 0) {
201       throw new IllegalArgumentException("Unexpected offset " + offset);
202     }
203     int fileNum = (int) (offset / sizePerFile);
204     if (fileNum >= fileChannels.length) {
205       throw new RuntimeException("Not expected offset " + offset + " where capacity=" + capacity);
206     }
207     return fileNum;
208   }
209 
210   private static interface FileAccessor {
211     int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
212         throws IOException;
213   }
214 
215   private static class FileReadAccessor implements FileAccessor {
216     @Override
217     public int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
218         throws IOException {
219       return fileChannel.read(byteBuffer, accessOffset);
220     }
221   }
222 
223   private static class FileWriteAccessor implements FileAccessor {
224     @Override
225     public int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
226         throws IOException {
227       return fileChannel.write(byteBuffer, accessOffset);
228     }
229   }
230 
231 }