Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: compressor didn't chunkify big payload #3

Merged
merged 8 commits into from
Apr 26, 2022
Merged

fix: compressor didn't chunkify big payload #3

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
0757db6
Update the compressor to chunk the bytes as per the standard
g11tech Apr 14, 2022
6e22a09
bump package version
g11tech Apr 14, 2022
ab41c6d
test for multi chunk compress
g11tech Apr 14, 2022
ce0a908
remove the dupilicated test
g11tech Apr 14, 2022
5c0a678
revering version bump as releases are auto created
g11tech Apr 15, 2022
d6c2786
benchmarks
g11tech Apr 25, 2022
0e7e5ef
log unhandled exception
g11tech Apr 25, 2022
21c7b4c
remove self
g11tech Apr 26, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 27 additions & 12 deletions lib/compress-stream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,11 @@
/**
* As per the snappy framing format for streams, the size of any uncompressed chunk can be
* no longer than 65536 bytes.
*
* From: https://github.com/google/snappy/blob/main/framing_format.txt#L90:L92
*/
const UNCOMPRESSED_CHUNK_SIZE = 65536;

var Transform = require('stream').Transform
, util = require('util')

Expand Down Expand Up @@ -52,18 +60,25 @@ CompressStream.prototype._uncompressed = function (chunk) {

CompressStream.prototype._transform = function (chunk, enc, callback) {
var self = this
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess we don't need self anymore now

async function compressChunks() {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why async function?

Copy link
Author

@g11tech g11tech Apr 15, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the API of transform is to call callback on the completion of transformation of the chunk, so don't want to holdup transform

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But there is no await in this body, and compressChunks() is called without any handling. Errors will become unhandled rejections. Can you just make it

  function compressChunks() {

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess the callback can error and cause unhandled rejection.

Maybe the implementation can look like:

new Promise(() => {
  ...
  callback(...);
  ...
}).catch((e) => logUnexpectedError(e);

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done, just console logged as nothing fancy available in lib

try {
for (let startFrom = 0; startFrom < chunk.length; startFrom += UNCOMPRESSED_CHUNK_SIZE) {
const endAt = startFrom + Math.min(chunk.length - startFrom, UNCOMPRESSED_CHUNK_SIZE);
const bytesChunk = chunk.slice(startFrom, endAt);
const compressed = snappy.compressSync(bytesChunk)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why switch to the sync version? If we want to do that we should do it in another PR as study in depth the implications

Copy link
Author

@g11tech g11tech Apr 15, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dapplion the async version where I chunk bytes from outside (like call compressStream.write(chunk)) which uses the original flow, compared to this where chunking happen in the transform using this way is 2X better slower

(i.e. compressSync based solution is coming out ahead):
image
(orig-snappy is the original with data chunked at compressStream.write level)

If I use async version inside with await Promise((resolve)=>... wrap to make sure the async returned data is written in a proper serial order, its 10-5% slower because of the overhead, if I don't care about the serial order in which chunks are written (which leads to incorrect stream), then async chunking version inside transform is 5% better.

I think for now this is our best bet 🙂

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay sounds good to me. Can you commit a benchmark tho? To inform us of the cost of compressing an uncompressing objects of 10Kb, 100Kb, 1000Kb

if (compressed.length < bytesChunk.length)
self._compressed(bytesChunk, compressed)
else
self._uncompressed(bytesChunk)

}
callback();
} catch (err) {
return callback(err);
}
}

snappy.compress(chunk, function (err, compressed) {
if (err)
return callback(err)

if (compressed.length < chunk.length)
self._compressed(chunk, compressed)
else
self._uncompressed(chunk)

callback()
})
compressChunks();
}

module.exports = CompressStream
module.exports = CompressStream
31 changes: 30 additions & 1 deletion test/compress-test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,16 @@
var spawn = require('child_process').spawn

, createCompressStream = require('../').createCompressStream
, test = require('tap').test
, largerInput = require('fs').readFileSync(__filename)
, largerInputString = largerInput.toString()

const UNCOMPRESSED_CHUNK_SIZE = 65536
let superLargeInput = largerInput;
for (let i = largerInput.length; i <= UNCOMPRESSED_CHUNK_SIZE; i += largerInput.length) {
superLargeInput = Buffer.concat([superLargeInput, largerInput]);
}
const superLargeInputString = superLargeInput.toString();

test('compress small string', function (t) {
var child = spawn('python', [ '-m', 'snappy', '-d' ])
, compressStream = createCompressStream()
Expand Down Expand Up @@ -48,3 +54,26 @@ test('compress large string', function (t) {
compressStream.write(largerInputString)
compressStream.end()
})


test('compress very very large string', function (t) {
var child = spawn('python', [ '-m', 'snappy', '-d' ])
, compressStream = createCompressStream()
, data = ''

child.stdout.on('data', function (chunk) {
data = data + chunk.toString()
})

child.stdout.on('end', function () {
t.equal(data, superLargeInputString)
t.end()
})

child.stderr.pipe(process.stderr)

compressStream.pipe(child.stdin)

compressStream.write(superLargeInputString)
compressStream.end()
})