1 /*
2 * Copyright 2012 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a copy of the License at:
7 *
8 * https://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package io.netty.handler.codec.compression;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.util.concurrent.FastThreadLocal;
20 import io.netty.util.internal.MathUtil;
21 import io.netty.util.internal.SystemPropertyUtil;
22
23 import java.util.Arrays;
24
25 /**
26 * Uncompresses an input {@link ByteBuf} encoded with Snappy compression into an
27 * output {@link ByteBuf}.
28 *
29 * See <a href="https://github.com/google/snappy/blob/master/format_description.txt">snappy format</a>.
30 */
31 public final class Snappy {
32
33 private static final int MAX_HT_SIZE = 1 << 14;
34 private static final int MIN_COMPRESSIBLE_BYTES = 15;
35
36 // used as a return value to indicate that we haven't yet read our full preamble
37 private static final int PREAMBLE_NOT_FULL = -1;
38 private static final int NOT_ENOUGH_INPUT = -1;
39
40 // constants for the tag types
41 private static final int LITERAL = 0;
42 private static final int COPY_1_BYTE_OFFSET = 1;
43 private static final int COPY_2_BYTE_OFFSET = 2;
44 private static final int COPY_4_BYTE_OFFSET = 3;
45
46 // Hash table used to compress, shared between subsequent call to .encode()
47 private static final FastThreadLocal<short[]> HASH_TABLE = new FastThreadLocal<short[]>();
48
49 private static final boolean DEFAULT_REUSE_HASHTABLE =
50 SystemPropertyUtil.getBoolean("io.netty.handler.codec.compression.snappy.reuseHashTable", false);
51
52 public Snappy() {
53 this(DEFAULT_REUSE_HASHTABLE);
54 }
55
56 Snappy(boolean reuseHashtable) {
57 this.reuseHashtable = reuseHashtable;
58 }
59
60 public static Snappy withHashTableReuse() {
61 return new Snappy(true);
62 }
63
64 private final boolean reuseHashtable;
65 private State state = State.READING_PREAMBLE;
66 private byte tag;
67 private int written;
68
69 private enum State {
70 READING_PREAMBLE,
71 READING_TAG,
72 READING_LITERAL,
73 READING_COPY
74 }
75
76 public void reset() {
77 state = State.READING_PREAMBLE;
78 tag = 0;
79 written = 0;
80 }
81
82 public void encode(final ByteBuf in, final ByteBuf out, final int length) {
83 // Write the preamble length to the output buffer
84 for (int i = 0;; i ++) {
85 int b = length >>> i * 7;
86 if ((b & 0xFFFFFF80) != 0) {
87 out.writeByte(b & 0x7f | 0x80);
88 } else {
89 out.writeByte(b);
90 break;
91 }
92 }
93
94 int inIndex = in.readerIndex();
95 final int baseIndex = inIndex;
96
97 int hashTableSize = MathUtil.findNextPositivePowerOfTwo(length);
98 hashTableSize = Math.min(hashTableSize, MAX_HT_SIZE);
99 final short[] table = getHashTable(hashTableSize);
100 final int shift = Integer.numberOfLeadingZeros(hashTableSize) + 1;
101
102 int nextEmit = inIndex;
103
104 if (length - inIndex >= MIN_COMPRESSIBLE_BYTES) {
105 int nextHash = hash(in, ++inIndex, shift);
106 outer: while (true) {
107 int skip = 32;
108
109 int candidate;
110 int nextIndex = inIndex;
111 do {
112 inIndex = nextIndex;
113 int hash = nextHash;
114 int bytesBetweenHashLookups = skip++ >> 5;
115 nextIndex = inIndex + bytesBetweenHashLookups;
116
117 // We need at least 4 remaining bytes to read the hash
118 if (nextIndex > length - 4) {
119 break outer;
120 }
121
122 nextHash = hash(in, nextIndex, shift);
123
124 // equivalent to Short.toUnsignedInt
125 // use unsigned short cast to avoid loss precision when 32767 <= length <= 65355
126 candidate = baseIndex + ((int) table[hash]) & 0xffff;
127
128 table[hash] = (short) (inIndex - baseIndex);
129 }
130 while (in.getInt(inIndex) != in.getInt(candidate));
131
132 encodeLiteral(in, out, inIndex - nextEmit);
133
134 int insertTail;
135 do {
136 int base = inIndex;
137 int matched = 4 + findMatchingLength(in, candidate + 4, inIndex + 4, length);
138 inIndex += matched;
139 int offset = base - candidate;
140 encodeCopy(out, offset, matched);
141 in.readerIndex(in.readerIndex() + matched);
142 insertTail = inIndex - 1;
143 nextEmit = inIndex;
144 if (inIndex >= length - 4) {
145 break outer;
146 }
147
148 int prevHash = hash(in, insertTail, shift);
149 table[prevHash] = (short) (inIndex - baseIndex - 1);
150 int currentHash = hash(in, insertTail + 1, shift);
151 candidate = baseIndex + table[currentHash];
152 table[currentHash] = (short) (inIndex - baseIndex);
153 }
154 while (in.getInt(insertTail + 1) == in.getInt(candidate));
155
156 nextHash = hash(in, insertTail + 2, shift);
157 ++inIndex;
158 }
159 }
160
161 // If there are any remaining characters, write them out as a literal
162 if (nextEmit < length) {
163 encodeLiteral(in, out, length - nextEmit);
164 }
165 }
166
167 /**
168 * Hashes the 4 bytes located at index, shifting the resulting hash into
169 * the appropriate range for our hash table.
170 *
171 * @param in The input buffer to read 4 bytes from
172 * @param index The index to read at
173 * @param shift The shift value, for ensuring that the resulting value is
174 * within the range of our hash table size
175 * @return A 32-bit hash of 4 bytes located at index
176 */
177 private static int hash(ByteBuf in, int index, int shift) {
178 return in.getInt(index) * 0x1e35a7bd >>> shift;
179 }
180
181 /**
182 * Returns a short[] to be used as a hashtable
183 *
184 * @param hashTableSize the size for the hashtable
185 * @return An appropriately sized empty hashtable
186 */
187 private short[] getHashTable(int hashTableSize) {
188 if (reuseHashtable) {
189 return getHashTableFastThreadLocalArrayFill(hashTableSize);
190 }
191 return new short[hashTableSize];
192 }
193
194 /**
195 * Returns a short[] from a FastThreadLocal, zeroing for correctness
196 * creating a new one and resizing it if necessary
197 *
198 * @return An appropriately sized empty hashtable
199 * @param hashTableSize
200 */
201 public static short[] getHashTableFastThreadLocalArrayFill(int hashTableSize) {
202 short[] hashTable = HASH_TABLE.get();
203 if (hashTable == null || hashTable.length < hashTableSize) {
204 hashTable = new short[hashTableSize];
205 HASH_TABLE.set(hashTable);
206 return hashTable;
207 }
208
209 Arrays.fill(hashTable, 0, hashTableSize, (short) 0);
210 return hashTable;
211 }
212
213 /**
214 * Iterates over the supplied input buffer between the supplied minIndex and
215 * maxIndex to find how long our matched copy overlaps with an already-written
216 * literal value.
217 *
218 * @param in The input buffer to scan over
219 * @param minIndex The index in the input buffer to start scanning from
220 * @param inIndex The index of the start of our copy
221 * @param maxIndex The length of our input buffer
222 * @return The number of bytes for which our candidate copy is a repeat of
223 */
224 private static int findMatchingLength(ByteBuf in, int minIndex, int inIndex, int maxIndex) {
225 int matched = 0;
226
227 while (inIndex <= maxIndex - 4 &&
228 in.getInt(inIndex) == in.getInt(minIndex + matched)) {
229 inIndex += 4;
230 matched += 4;
231 }
232
233 while (inIndex < maxIndex && in.getByte(minIndex + matched) == in.getByte(inIndex)) {
234 ++inIndex;
235 ++matched;
236 }
237
238 return matched;
239 }
240
241 /**
242 * Calculates the minimum number of bits required to encode a value. This can
243 * then in turn be used to calculate the number of septets or octets (as
244 * appropriate) to use to encode a length parameter.
245 *
246 * @param value The value to calculate the minimum number of bits required to encode
247 * @return The minimum number of bits required to encode the supplied value
248 */
249 private static int bitsToEncode(int value) {
250 int highestOneBit = Integer.highestOneBit(value);
251 int bitLength = 0;
252 while ((highestOneBit >>= 1) != 0) {
253 bitLength++;
254 }
255
256 return bitLength;
257 }
258
259 /**
260 * Writes a literal to the supplied output buffer by directly copying from
261 * the input buffer. The literal is taken from the current readerIndex
262 * up to the supplied length.
263 *
264 * @param in The input buffer to copy from
265 * @param out The output buffer to copy to
266 * @param length The length of the literal to copy
267 */
268 static void encodeLiteral(ByteBuf in, ByteBuf out, int length) {
269 if (length < 61) {
270 out.writeByte(length - 1 << 2);
271 } else {
272 int bitLength = bitsToEncode(length - 1);
273 int bytesToEncode = 1 + bitLength / 8;
274 out.writeByte(59 + bytesToEncode << 2);
275 for (int i = 0; i < bytesToEncode; i++) {
276 out.writeByte(length - 1 >> i * 8 & 0x0ff);
277 }
278 }
279
280 out.writeBytes(in, length);
281 }
282
283 private static void encodeCopyWithOffset(ByteBuf out, int offset, int length) {
284 if (length < 12 && offset < 2048) {
285 out.writeByte(COPY_1_BYTE_OFFSET | length - 4 << 2 | offset >> 8 << 5);
286 out.writeByte(offset & 0x0ff);
287 } else {
288 out.writeByte(COPY_2_BYTE_OFFSET | length - 1 << 2);
289 out.writeByte(offset & 0x0ff);
290 out.writeByte(offset >> 8 & 0x0ff);
291 }
292 }
293
294 /**
295 * Encodes a series of copies, each at most 64 bytes in length.
296 *
297 * @param out The output buffer to write the copy pointer to
298 * @param offset The offset at which the original instance lies
299 * @param length The length of the original instance
300 */
301 private static void encodeCopy(ByteBuf out, int offset, int length) {
302 while (length >= 68) {
303 encodeCopyWithOffset(out, offset, 64);
304 length -= 64;
305 }
306
307 if (length > 64) {
308 encodeCopyWithOffset(out, offset, 60);
309 length -= 60;
310 }
311
312 encodeCopyWithOffset(out, offset, length);
313 }
314
315 public void decode(ByteBuf in, ByteBuf out) {
316 while (in.isReadable()) {
317 switch (state) {
318 case READING_PREAMBLE:
319 int uncompressedLength = readPreamble(in);
320 if (uncompressedLength == PREAMBLE_NOT_FULL) {
321 // We've not yet read all of the preamble, so wait until we can
322 return;
323 }
324 if (uncompressedLength == 0) {
325 // Should never happen, but it does mean we have nothing further to do
326 return;
327 }
328 out.ensureWritable(uncompressedLength);
329 state = State.READING_TAG;
330 // fall through
331 case READING_TAG:
332 if (!in.isReadable()) {
333 return;
334 }
335 tag = in.readByte();
336 switch (tag & 0x03) {
337 case LITERAL:
338 state = State.READING_LITERAL;
339 break;
340 case COPY_1_BYTE_OFFSET:
341 case COPY_2_BYTE_OFFSET:
342 case COPY_4_BYTE_OFFSET:
343 state = State.READING_COPY;
344 break;
345 }
346 break;
347 case READING_LITERAL:
348 int literalWritten = decodeLiteral(tag, in, out);
349 if (literalWritten != NOT_ENOUGH_INPUT) {
350 state = State.READING_TAG;
351 written += literalWritten;
352 } else {
353 // Need to wait for more data
354 return;
355 }
356 break;
357 case READING_COPY:
358 int decodeWritten;
359 switch (tag & 0x03) {
360 case COPY_1_BYTE_OFFSET:
361 decodeWritten = decodeCopyWith1ByteOffset(tag, in, out, written);
362 if (decodeWritten != NOT_ENOUGH_INPUT) {
363 state = State.READING_TAG;
364 written += decodeWritten;
365 } else {
366 // Need to wait for more data
367 return;
368 }
369 break;
370 case COPY_2_BYTE_OFFSET:
371 decodeWritten = decodeCopyWith2ByteOffset(tag, in, out, written);
372 if (decodeWritten != NOT_ENOUGH_INPUT) {
373 state = State.READING_TAG;
374 written += decodeWritten;
375 } else {
376 // Need to wait for more data
377 return;
378 }
379 break;
380 case COPY_4_BYTE_OFFSET:
381 decodeWritten = decodeCopyWith4ByteOffset(tag, in, out, written);
382 if (decodeWritten != NOT_ENOUGH_INPUT) {
383 state = State.READING_TAG;
384 written += decodeWritten;
385 } else {
386 // Need to wait for more data
387 return;
388 }
389 break;
390 }
391 }
392 }
393 }
394
395 /**
396 * Reads the length varint (a series of bytes, where the lower 7 bits
397 * are data and the upper bit is a flag to indicate more bytes to be
398 * read).
399 *
400 * @param in The input buffer to read the preamble from
401 * @return The calculated length based on the input buffer, or 0 if
402 * no preamble is able to be calculated
403 */
404 private static int readPreamble(ByteBuf in) {
405 int length = 0;
406 int byteIndex = 0;
407 while (in.isReadable()) {
408 int current = in.readUnsignedByte();
409 length |= (current & 0x7f) << byteIndex++ * 7;
410 if ((current & 0x80) == 0) {
411 return length;
412 }
413
414 if (byteIndex >= 4) {
415 throw new DecompressionException("Preamble is greater than 4 bytes");
416 }
417 }
418
419 return 0;
420 }
421
422 /**
423 * Get the length varint (a series of bytes, where the lower 7 bits
424 * are data and the upper bit is a flag to indicate more bytes to be
425 * read).
426 *
427 * @param in The input buffer to get the preamble from
428 * @return The calculated length based on the input buffer, or 0 if
429 * no preamble is able to be calculated
430 */
431 int getPreamble(ByteBuf in) {
432 if (state == State.READING_PREAMBLE) {
433 int readerIndex = in.readerIndex();
434 try {
435 return readPreamble(in);
436 } finally {
437 in.readerIndex(readerIndex);
438 }
439 }
440 return 0;
441 }
442
443 /**
444 * Reads a literal from the input buffer directly to the output buffer.
445 * A "literal" is an uncompressed segment of data stored directly in the
446 * byte stream.
447 *
448 * @param tag The tag that identified this segment as a literal is also
449 * used to encode part of the length of the data
450 * @param in The input buffer to read the literal from
451 * @param out The output buffer to write the literal to
452 * @return The number of bytes appended to the output buffer, or -1 to indicate "try again later"
453 */
454 static int decodeLiteral(byte tag, ByteBuf in, ByteBuf out) {
455 in.markReaderIndex();
456 int length;
457 switch(tag >> 2 & 0x3F) {
458 case 60:
459 if (!in.isReadable()) {
460 return NOT_ENOUGH_INPUT;
461 }
462 length = in.readUnsignedByte();
463 break;
464 case 61:
465 if (in.readableBytes() < 2) {
466 return NOT_ENOUGH_INPUT;
467 }
468 length = in.readUnsignedShortLE();
469 break;
470 case 62:
471 if (in.readableBytes() < 3) {
472 return NOT_ENOUGH_INPUT;
473 }
474 length = in.readUnsignedMediumLE();
475 break;
476 case 63:
477 if (in.readableBytes() < 4) {
478 return NOT_ENOUGH_INPUT;
479 }
480 length = in.readIntLE();
481 break;
482 default:
483 length = tag >> 2 & 0x3F;
484 }
485 length += 1;
486
487 if (in.readableBytes() < length) {
488 in.resetReaderIndex();
489 return NOT_ENOUGH_INPUT;
490 }
491
492 out.writeBytes(in, length);
493 return length;
494 }
495
496 /**
497 * Reads a compressed reference offset and length from the supplied input
498 * buffer, seeks back to the appropriate place in the input buffer and
499 * writes the found data to the supplied output stream.
500 *
501 * @param tag The tag used to identify this as a copy is also used to encode
502 * the length and part of the offset
503 * @param in The input buffer to read from
504 * @param out The output buffer to write to
505 * @return The number of bytes appended to the output buffer, or -1 to indicate
506 * "try again later"
507 * @throws DecompressionException If the read offset is invalid
508 */
509 private static int decodeCopyWith1ByteOffset(byte tag, ByteBuf in, ByteBuf out, int writtenSoFar) {
510 if (!in.isReadable()) {
511 return NOT_ENOUGH_INPUT;
512 }
513
514 int initialIndex = out.writerIndex();
515 int length = 4 + ((tag & 0x01c) >> 2);
516 int offset = (tag & 0x0e0) << 8 >> 5 | in.readUnsignedByte();
517
518 validateOffset(offset, writtenSoFar);
519
520 out.markReaderIndex();
521 if (offset < length) {
522 int copies = length / offset;
523 for (; copies > 0; copies--) {
524 out.readerIndex(initialIndex - offset);
525 out.readBytes(out, offset);
526 }
527 if (length % offset != 0) {
528 out.readerIndex(initialIndex - offset);
529 out.readBytes(out, length % offset);
530 }
531 } else {
532 out.readerIndex(initialIndex - offset);
533 out.readBytes(out, length);
534 }
535 out.resetReaderIndex();
536
537 return length;
538 }
539
540 /**
541 * Reads a compressed reference offset and length from the supplied input
542 * buffer, seeks back to the appropriate place in the input buffer and
543 * writes the found data to the supplied output stream.
544 *
545 * @param tag The tag used to identify this as a copy is also used to encode
546 * the length and part of the offset
547 * @param in The input buffer to read from
548 * @param out The output buffer to write to
549 * @throws DecompressionException If the read offset is invalid
550 * @return The number of bytes appended to the output buffer, or -1 to indicate
551 * "try again later"
552 */
553 private static int decodeCopyWith2ByteOffset(byte tag, ByteBuf in, ByteBuf out, int writtenSoFar) {
554 if (in.readableBytes() < 2) {
555 return NOT_ENOUGH_INPUT;
556 }
557
558 int initialIndex = out.writerIndex();
559 int length = 1 + (tag >> 2 & 0x03f);
560 int offset = in.readUnsignedShortLE();
561
562 validateOffset(offset, writtenSoFar);
563
564 out.markReaderIndex();
565 if (offset < length) {
566 int copies = length / offset;
567 for (; copies > 0; copies--) {
568 out.readerIndex(initialIndex - offset);
569 out.readBytes(out, offset);
570 }
571 if (length % offset != 0) {
572 out.readerIndex(initialIndex - offset);
573 out.readBytes(out, length % offset);
574 }
575 } else {
576 out.readerIndex(initialIndex - offset);
577 out.readBytes(out, length);
578 }
579 out.resetReaderIndex();
580
581 return length;
582 }
583
584 /**
585 * Reads a compressed reference offset and length from the supplied input
586 * buffer, seeks back to the appropriate place in the input buffer and
587 * writes the found data to the supplied output stream.
588 *
589 * @param tag The tag used to identify this as a copy is also used to encode
590 * the length and part of the offset
591 * @param in The input buffer to read from
592 * @param out The output buffer to write to
593 * @return The number of bytes appended to the output buffer, or -1 to indicate
594 * "try again later"
595 * @throws DecompressionException If the read offset is invalid
596 */
597 private static int decodeCopyWith4ByteOffset(byte tag, ByteBuf in, ByteBuf out, int writtenSoFar) {
598 if (in.readableBytes() < 4) {
599 return NOT_ENOUGH_INPUT;
600 }
601
602 int initialIndex = out.writerIndex();
603 int length = 1 + (tag >> 2 & 0x03F);
604 int offset = in.readIntLE();
605
606 validateOffset(offset, writtenSoFar);
607
608 out.markReaderIndex();
609 if (offset < length) {
610 int copies = length / offset;
611 for (; copies > 0; copies--) {
612 out.readerIndex(initialIndex - offset);
613 out.readBytes(out, offset);
614 }
615 if (length % offset != 0) {
616 out.readerIndex(initialIndex - offset);
617 out.readBytes(out, length % offset);
618 }
619 } else {
620 out.readerIndex(initialIndex - offset);
621 out.readBytes(out, length);
622 }
623 out.resetReaderIndex();
624
625 return length;
626 }
627
628 /**
629 * Validates that the offset extracted from a compressed reference is within
630 * the permissible bounds of an offset (0 < offset < Integer.MAX_VALUE), and does not
631 * exceed the length of the chunk currently read so far.
632 *
633 * @param offset The offset extracted from the compressed reference
634 * @param chunkSizeSoFar The number of bytes read so far from this chunk
635 * @throws DecompressionException if the offset is invalid
636 */
637 private static void validateOffset(int offset, int chunkSizeSoFar) {
638 if (offset == 0) {
639 throw new DecompressionException("Offset is less than minimum permissible value");
640 }
641
642 if (offset < 0) {
643 // Due to arithmetic overflow
644 throw new DecompressionException("Offset is greater than maximum value supported by this implementation");
645 }
646
647 if (offset > chunkSizeSoFar) {
648 throw new DecompressionException("Offset exceeds size of chunk");
649 }
650 }
651
652 /**
653 * Computes the CRC32C checksum of the supplied data and performs the "mask" operation
654 * on the computed checksum
655 *
656 * @param data The input data to calculate the CRC32C checksum of
657 */
658 static int calculateChecksum(ByteBuf data) {
659 return calculateChecksum(data, data.readerIndex(), data.readableBytes());
660 }
661
662 /**
663 * Computes the CRC32C checksum of the supplied data and performs the "mask" operation
664 * on the computed checksum
665 *
666 * @param data The input data to calculate the CRC32C checksum of
667 */
668 static int calculateChecksum(ByteBuf data, int offset, int length) {
669 Crc32c crc32 = new Crc32c();
670 try {
671 crc32.update(data, offset, length);
672 return maskChecksum(crc32.getValue());
673 } finally {
674 crc32.reset();
675 }
676 }
677
678 /**
679 * Computes the CRC32C checksum of the supplied data, performs the "mask" operation
680 * on the computed checksum, and then compares the resulting masked checksum to the
681 * supplied checksum.
682 *
683 * @param expectedChecksum The checksum decoded from the stream to compare against
684 * @param data The input data to calculate the CRC32C checksum of
685 * @throws DecompressionException If the calculated and supplied checksums do not match
686 */
687 static void validateChecksum(int expectedChecksum, ByteBuf data) {
688 validateChecksum(expectedChecksum, data, data.readerIndex(), data.readableBytes());
689 }
690
691 /**
692 * Computes the CRC32C checksum of the supplied data, performs the "mask" operation
693 * on the computed checksum, and then compares the resulting masked checksum to the
694 * supplied checksum.
695 *
696 * @param expectedChecksum The checksum decoded from the stream to compare against
697 * @param data The input data to calculate the CRC32C checksum of
698 * @throws DecompressionException If the calculated and supplied checksums do not match
699 */
700 static void validateChecksum(int expectedChecksum, ByteBuf data, int offset, int length) {
701 final int actualChecksum = calculateChecksum(data, offset, length);
702 if (actualChecksum != expectedChecksum) {
703 throw new DecompressionException(
704 "mismatching checksum: " + Integer.toHexString(actualChecksum) +
705 " (expected: " + Integer.toHexString(expectedChecksum) + ')');
706 }
707 }
708
709 /**
710 * From the spec:
711 *
712 * "Checksums are not stored directly, but masked, as checksumming data and
713 * then its own checksum can be problematic. The masking is the same as used
714 * in Apache Hadoop: Rotate the checksum by 15 bits, then add the constant
715 * 0xa282ead8 (using wraparound as normal for unsigned integers)."
716 *
717 * @param checksum The actual checksum of the data
718 * @return The masked checksum
719 */
720 static int maskChecksum(long checksum) {
721 return (int) ((checksum >> 15 | checksum << 17) + 0xa282ead8);
722 }
723 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！