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.ClosedChannelException;
26  import java.nio.channels.FileChannel;
27  import java.util.Arrays;
28  
29  import com.google.common.annotations.VisibleForTesting;
30  import org.apache.commons.logging.Log;
31  import org.apache.commons.logging.LogFactory;
32  import org.apache.hadoop.hbase.classification.InterfaceAudience;
33  import org.apache.hadoop.util.StringUtils;
34  
35  /**
36   * IO engine that stores data to a file on the local file system.
37   */
38  @InterfaceAudience.Private
39  public class FileIOEngine implements IOEngine {
40    private static final Log LOG = LogFactory.getLog(FileIOEngine.class);
41    public static final String FILE_DELIMITER = ",";
42    private final String[] filePaths;
43    private final FileChannel[] fileChannels;
44    private final RandomAccessFile[] rafs;
45  
46    private final long sizePerFile;
47    private final long capacity;
48  
49    private FileReadAccessor readAccessor = new FileReadAccessor();
50    private FileWriteAccessor writeAccessor = new FileWriteAccessor();
51  
52    public FileIOEngine(long capacity, String... filePaths) throws IOException {
53      this.sizePerFile = capacity / filePaths.length;
54      this.capacity = this.sizePerFile * filePaths.length;
55      this.filePaths = filePaths;
56      this.fileChannels = new FileChannel[filePaths.length];
57      this.rafs = new RandomAccessFile[filePaths.length];
58      for (int i = 0; i < filePaths.length; i++) {
59        String filePath = filePaths[i];
60        try {
61          rafs[i] = new RandomAccessFile(filePath, "rw");
62          long totalSpace = new File(filePath).getTotalSpace();
63          if (totalSpace < sizePerFile) {
64            // The next setting length will throw exception,logging this message
65            // is just used for the detail reason of exception,
66            String msg = "Only " + StringUtils.byteDesc(totalSpace)
67                + " total space under " + filePath + ", not enough for requested "
68                + StringUtils.byteDesc(sizePerFile);
69            LOG.warn(msg);
70          }
71          rafs[i].setLength(sizePerFile);
72          fileChannels[i] = rafs[i].getChannel();
73          LOG.info("Allocating cache " + StringUtils.byteDesc(sizePerFile)
74              + ", on the path:" + filePath);
75        } catch (IOException fex) {
76          LOG.error("Failed allocating cache on " + filePath, fex);
77          shutdown();
78          throw fex;
79        }
80      }
81    }
82  
83    @Override
84    public String toString() {
85      return "ioengine=" + this.getClass().getSimpleName() + ", paths="
86          + Arrays.asList(filePaths) + ", capacity=" + String.format("%,d", this.capacity);
87    }
88  
89    /**
90     * File IO engine is always able to support persistent storage for the cache
91     * @return true
92     */
93    @Override
94    public boolean isPersistent() {
95      return true;
96    }
97  
98    /**
99     * Transfers data from file to the given byte buffer
100    * @param dstBuffer the given byte buffer into which bytes are to be written
101    * @param offset The offset in the file where the first byte to be read
102    * @return number of bytes read
103    * @throws IOException
104    */
105   @Override
106   public int read(ByteBuffer dstBuffer, long offset) throws IOException {
107     if (dstBuffer.remaining() != 0) {
108       return accessFile(readAccessor, dstBuffer, offset);
109     }
110     return 0;
111   }
112 
113   @VisibleForTesting
114   void closeFileChannels() {
115     for (FileChannel fileChannel: fileChannels) {
116       try {
117         fileChannel.close();
118       } catch (IOException e) {
119         LOG.warn("Failed to close FileChannel", e);
120       }
121     }
122   }
123 
124   /**
125    * Transfers data from the given byte buffer to file
126    * @param srcBuffer the given byte buffer from which bytes are to be read
127    * @param offset The offset in the file where the first byte to be written
128    * @throws IOException
129    */
130   @Override
131   public void write(ByteBuffer srcBuffer, long offset) throws IOException {
132     if (!srcBuffer.hasRemaining()) {
133       return;
134     }
135     accessFile(writeAccessor, srcBuffer, offset);
136   }
137 
138   /**
139    * Sync the data to file after writing
140    * @throws IOException
141    */
142   @Override
143   public void sync() throws IOException {
144     for (int i = 0; i < fileChannels.length; i++) {
145       try {
146         if (fileChannels[i] != null) {
147           fileChannels[i].force(true);
148         }
149       } catch (IOException ie) {
150         LOG.warn("Failed syncing data to " + this.filePaths[i]);
151         throw ie;
152       }
153     }
154   }
155 
156   /**
157    * Close the file
158    */
159   @Override
160   public void shutdown() {
161     for (int i = 0; i < filePaths.length; i++) {
162       try {
163         if (fileChannels[i] != null) {
164           fileChannels[i].close();
165         }
166         if (rafs[i] != null) {
167           rafs[i].close();
168         }
169       } catch (IOException ex) {
170         LOG.error("Failed closing " + filePaths[i] + " when shudown the IOEngine", ex);
171       }
172     }
173   }
174 
175   private int accessFile(FileAccessor accessor, ByteBuffer buffer, long globalOffset)
176       throws IOException {
177     int startFileNum = getFileNum(globalOffset);
178     int remainingAccessDataLen = buffer.remaining();
179     int endFileNum = getFileNum(globalOffset + remainingAccessDataLen - 1);
180     int accessFileNum = startFileNum;
181     long accessOffset = getAbsoluteOffsetInFile(accessFileNum, globalOffset);
182     int bufLimit = buffer.limit();
183     while (true) {
184       FileChannel fileChannel = fileChannels[accessFileNum];
185       int accessLen = 0;
186       if (endFileNum > accessFileNum) {
187         // short the limit;
188         buffer.limit((int) (buffer.limit() - remainingAccessDataLen + sizePerFile - accessOffset));
189       }
190       try {
191         accessLen = accessor.access(fileChannel, buffer, accessOffset);
192       } catch (ClosedChannelException e) {
193         LOG.warn("Caught ClosedChannelException accessing BucketCache, reopening file. ", e);
194         refreshFileConnection(accessFileNum);
195         continue;
196       }
197       // recover the limit
198       buffer.limit(bufLimit);
199       if (accessLen < remainingAccessDataLen) {
200         remainingAccessDataLen -= accessLen;
201         accessFileNum++;
202         accessOffset = 0;
203       } else {
204         break;
205       }
206       if (accessFileNum >= fileChannels.length) {
207         throw new IOException("Required data len " + StringUtils.byteDesc(buffer.remaining())
208             + " exceed the engine's capacity " + StringUtils.byteDesc(capacity) + " where offset="
209             + globalOffset);
210       }
211     }
212     return bufLimit;
213   }
214 
215   /**
216    * Get the absolute offset in given file with the relative global offset.
217    * @param fileNum
218    * @param globalOffset
219    * @return the absolute offset
220    */
221   private long getAbsoluteOffsetInFile(int fileNum, long globalOffset) {
222     return globalOffset - fileNum * sizePerFile;
223   }
224 
225   private int getFileNum(long offset) {
226     if (offset < 0) {
227       throw new IllegalArgumentException("Unexpected offset " + offset);
228     }
229     int fileNum = (int) (offset / sizePerFile);
230     if (fileNum >= fileChannels.length) {
231       throw new RuntimeException("Not expected offset " + offset + " where capacity=" + capacity);
232     }
233     return fileNum;
234   }
235 
236   @VisibleForTesting
237   FileChannel[] getFileChannels() {
238     return fileChannels;
239   }
240 
241   @VisibleForTesting
242   void refreshFileConnection(int accessFileNum) throws IOException {
243     FileChannel fileChannel = fileChannels[accessFileNum];
244     if (fileChannel != null) {
245       fileChannel.close();
246     }
247     rafs[accessFileNum] = new RandomAccessFile(filePaths[accessFileNum], "rw");
248     fileChannels[accessFileNum] = rafs[accessFileNum].getChannel();
249   }
250 
251   private static interface FileAccessor {
252     int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
253         throws IOException;
254   }
255 
256   private static class FileReadAccessor implements FileAccessor {
257     @Override
258     public int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
259         throws IOException {
260       return fileChannel.read(byteBuffer, accessOffset);
261     }
262   }
263 
264   private static class FileWriteAccessor implements FileAccessor {
265     @Override
266     public int access(FileChannel fileChannel, ByteBuffer byteBuffer, long accessOffset)
267         throws IOException {
268       return fileChannel.write(byteBuffer, accessOffset);
269     }
270   }
271 
272 }