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.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
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
65
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
91
92
93 @Override
94 public boolean isPersistent() {
95 return true;
96 }
97
98
99
100
101
102
103
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
126
127
128
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
140
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
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
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
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
217
218
219
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 }