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.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
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
63
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
89
90
91 @Override
92 public boolean isPersistent() {
93 return true;
94 }
95
96
97
98
99
100
101
102
103 @Override
104 public int read(ByteBuffer dstBuffer, long offset) throws IOException {
105 return accessFile(readAccessor, dstBuffer, offset);
106 }
107
108
109
110
111
112
113
114 @Override
115 public void write(ByteBuffer srcBuffer, long offset) throws IOException {
116 accessFile(writeAccessor, srcBuffer, offset);
117 }
118
119
120
121
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
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
168 buffer.limit((int) (buffer.limit() - remainingAccessDataLen + sizePerFile - accessOffset));
169 }
170 int accessLen = accessor.access(fileChannel, buffer, accessOffset);
171
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
191
192
193
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 }