1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
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
69
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
96
97
98 @Override
99 public boolean isPersistent() {
100 return true;
101 }
102
103
104
105
106
107
108
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
131
132
133
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
145
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
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
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
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
223
224
225
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
255
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 }