1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.ipc;
19
20 import java.io.ByteArrayInputStream;
21 import java.io.DataInput;
22 import java.io.IOException;
23 import java.io.InputStream;
24 import java.io.OutputStream;
25 import java.nio.BufferOverflowException;
26 import java.nio.ByteBuffer;
27
28 import org.apache.commons.io.IOUtils;
29 import org.apache.commons.logging.Log;
30 import org.apache.commons.logging.LogFactory;
31 import org.apache.hadoop.hbase.classification.InterfaceAudience;
32 import org.apache.hadoop.conf.Configurable;
33 import org.apache.hadoop.conf.Configuration;
34 import org.apache.hadoop.hbase.CellScanner;
35 import org.apache.hadoop.hbase.DoNotRetryIOException;
36 import org.apache.hadoop.hbase.HBaseIOException;
37 import org.apache.hadoop.hbase.codec.Codec;
38 import org.apache.hadoop.hbase.io.BoundedByteBufferPool;
39 import org.apache.hadoop.hbase.io.ByteBufferOutputStream;
40 import org.apache.hadoop.hbase.io.HeapSize;
41 import org.apache.hadoop.hbase.util.Bytes;
42 import org.apache.hadoop.hbase.util.ClassSize;
43 import org.apache.hadoop.io.compress.CodecPool;
44 import org.apache.hadoop.io.compress.CompressionCodec;
45 import org.apache.hadoop.io.compress.CompressionInputStream;
46 import org.apache.hadoop.io.compress.Compressor;
47 import org.apache.hadoop.io.compress.Decompressor;
48
49 import com.google.common.base.Preconditions;
50 import com.google.protobuf.CodedOutputStream;
51 import com.google.protobuf.Message;
52
53
54
55
56 @InterfaceAudience.Private
57 public class IPCUtil {
58
59 public static final Log LOG = LogFactory.getLog(IPCUtil.class);
60
61
62
63 private final int cellBlockDecompressionMultiplier;
64 private final int cellBlockBuildingInitialBufferSize;
65 private final Configuration conf;
66
67 public IPCUtil(final Configuration conf) {
68 super();
69 this.conf = conf;
70 this.cellBlockDecompressionMultiplier =
71 conf.getInt("hbase.ipc.cellblock.decompression.buffersize.multiplier", 3);
72
73
74
75 this.cellBlockBuildingInitialBufferSize =
76 ClassSize.align(conf.getInt("hbase.ipc.cellblock.building.initial.buffersize", 16 * 1024));
77 }
78
79
80
81
82 public static class CellScannerButNoCodecException extends HBaseIOException {};
83
84
85
86
87
88
89
90
91
92
93
94
95 @SuppressWarnings("resource")
96 public ByteBuffer buildCellBlock(final Codec codec, final CompressionCodec compressor,
97 final CellScanner cellScanner)
98 throws IOException {
99 return buildCellBlock(codec, compressor, cellScanner, null);
100 }
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117 @SuppressWarnings("resource")
118 public ByteBuffer buildCellBlock(final Codec codec, final CompressionCodec compressor,
119 final CellScanner cellScanner, final BoundedByteBufferPool pool)
120 throws IOException {
121 if (cellScanner == null) return null;
122 if (codec == null) throw new CellScannerButNoCodecException();
123 int bufferSize = this.cellBlockBuildingInitialBufferSize;
124 ByteBufferOutputStream baos = null;
125 if (pool != null) {
126 ByteBuffer bb = pool.getBuffer();
127 bufferSize = bb.capacity();
128 baos = new ByteBufferOutputStream(bb);
129 } else {
130
131 if (cellScanner instanceof HeapSize) {
132 long longSize = ((HeapSize)cellScanner).heapSize();
133
134 if (longSize > Integer.MAX_VALUE) {
135 throw new IOException("Size " + longSize + " > " + Integer.MAX_VALUE);
136 }
137 bufferSize = ClassSize.align((int)longSize);
138 }
139 baos = new ByteBufferOutputStream(bufferSize);
140 }
141 OutputStream os = baos;
142 Compressor poolCompressor = null;
143 try {
144 if (compressor != null) {
145 if (compressor instanceof Configurable) ((Configurable)compressor).setConf(this.conf);
146 poolCompressor = CodecPool.getCompressor(compressor);
147 os = compressor.createOutputStream(os, poolCompressor);
148 }
149 Codec.Encoder encoder = codec.getEncoder(os);
150 int count = 0;
151 while (cellScanner.advance()) {
152 encoder.write(cellScanner.current());
153 count++;
154 }
155 encoder.flush();
156
157
158 if (count == 0) return null;
159 } catch (BufferOverflowException e) {
160 throw new DoNotRetryIOException(e);
161 } finally {
162 os.close();
163 if (poolCompressor != null) CodecPool.returnCompressor(poolCompressor);
164 }
165 if (LOG.isTraceEnabled()) {
166 if (bufferSize < baos.size()) {
167 LOG.trace("Buffer grew from initial bufferSize=" + bufferSize + " to " + baos.size() +
168 "; up hbase.ipc.cellblock.building.initial.buffersize?");
169 }
170 }
171 return baos.getByteBuffer();
172 }
173
174
175
176
177
178
179
180 public CellScanner createCellScanner(final Codec codec, final CompressionCodec compressor,
181 final byte [] cellBlock)
182 throws IOException {
183 return createCellScanner(codec, compressor, cellBlock, 0, cellBlock.length);
184 }
185
186
187
188
189
190
191
192
193
194 public CellScanner createCellScanner(final Codec codec, final CompressionCodec compressor,
195 final byte [] cellBlock, final int offset, final int length)
196 throws IOException {
197
198
199 InputStream is = null;
200 if (compressor != null) {
201
202 if (compressor instanceof Configurable) ((Configurable)compressor).setConf(this.conf);
203 Decompressor poolDecompressor = CodecPool.getDecompressor(compressor);
204 CompressionInputStream cis =
205 compressor.createInputStream(new ByteArrayInputStream(cellBlock, offset, length),
206 poolDecompressor);
207 ByteBufferOutputStream bbos = null;
208 try {
209
210
211 bbos = new ByteBufferOutputStream((length - offset) *
212 this.cellBlockDecompressionMultiplier);
213 IOUtils.copy(cis, bbos);
214 bbos.close();
215 ByteBuffer bb = bbos.getByteBuffer();
216 is = new ByteArrayInputStream(bb.array(), 0, bb.limit());
217 } finally {
218 if (is != null) is.close();
219 if (bbos != null) bbos.close();
220
221 CodecPool.returnDecompressor(poolDecompressor);
222 }
223 } else {
224 is = new ByteArrayInputStream(cellBlock, offset, length);
225 }
226 return codec.getDecoder(is);
227 }
228
229
230
231
232
233
234
235 public static ByteBuffer getDelimitedMessageAsByteBuffer(final Message m) throws IOException {
236 if (m == null) return null;
237 int serializedSize = m.getSerializedSize();
238 int vintSize = CodedOutputStream.computeRawVarint32Size(serializedSize);
239 byte [] buffer = new byte[serializedSize + vintSize];
240
241 CodedOutputStream cos = CodedOutputStream.newInstance(buffer);
242
243 cos.writeMessageNoTag(m);
244 cos.flush();
245 cos.checkNoSpaceLeft();
246 return ByteBuffer.wrap(buffer);
247 }
248
249
250
251
252
253
254
255
256
257
258 public static int write(final OutputStream dos, final Message header, final Message param,
259 final ByteBuffer cellBlock)
260 throws IOException {
261
262
263
264 int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(header, param);
265 if (cellBlock != null) totalSize += cellBlock.remaining();
266 return write(dos, header, param, cellBlock, totalSize);
267 }
268
269 private static int write(final OutputStream dos, final Message header, final Message param,
270 final ByteBuffer cellBlock, final int totalSize)
271 throws IOException {
272
273 dos.write(Bytes.toBytes(totalSize));
274
275 header.writeDelimitedTo(dos);
276 if (param != null) param.writeDelimitedTo(dos);
277 if (cellBlock != null) dos.write(cellBlock.array(), 0, cellBlock.remaining());
278 dos.flush();
279 return totalSize;
280 }
281
282
283
284
285
286
287
288
289
290 public static void readChunked(final DataInput in, byte[] dest, int offset, int len)
291 throws IOException {
292 int maxRead = 8192;
293
294 for (; offset < len; offset += maxRead) {
295 in.readFully(dest, offset, Math.min(len - offset, maxRead));
296 }
297 }
298
299
300
301
302 public static int getTotalSizeWhenWrittenDelimited(Message ... messages) {
303 int totalSize = 0;
304 for (Message m: messages) {
305 if (m == null) continue;
306 totalSize += m.getSerializedSize();
307 totalSize += CodedOutputStream.computeRawVarint32Size(m.getSerializedSize());
308 }
309 Preconditions.checkArgument(totalSize < Integer.MAX_VALUE);
310 return totalSize;
311 }
312 }