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