1 /***************************************************************************
2 * Copyright (C) 2005-2018 by the Quassel Project *
3 * devel@quassel-irc.org *
5 * This program is free software; you can redistribute it and/or modify *
6 * it under the terms of the GNU General Public License as published by *
7 * the Free Software Foundation; either version 2 of the License, or *
8 * (at your option) version 3. *
10 * This program is distributed in the hope that it will be useful, *
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of *
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
13 * GNU General Public License for more details. *
15 * You should have received a copy of the GNU General Public License *
16 * along with this program; if not, write to the *
17 * Free Software Foundation, Inc., *
18 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. *
19 ***************************************************************************/
21 #include "compressor.h"
26 const int maxBufferSize = 64 * 1024 * 1024; // protect us from zip bombs
27 const int ioBufferSize = 64 * 1024; // chunk size for inflate/deflate; should not be too large as we preallocate that space!
29 Compressor::Compressor(QTcpSocket *socket, Compressor::CompressionLevel level, QObject *parent)
36 connect(socket, &QIODevice::readyRead, this, &Compressor::readData);
39 if (level != NoCompression)
43 // something went wrong during initialization... but we can only emit an error after RemotePeer has connected its signal
44 QTimer::singleShot(0, this, [this]() { emit error(); });
48 // It's possible that more data has already arrived during the handshake, so readyRead() wouldn't be triggered.
49 // However, we want to give RemotePeer a chance to connect to our signals, so trigger this asynchronously.
50 if (socket->bytesAvailable())
51 QTimer::singleShot(0, this, &Compressor::readData);
55 Compressor::~Compressor()
57 // release resources allocated by zlib
59 inflateEnd(_inflater);
63 deflateEnd(_deflater);
69 bool Compressor::initStreams()
72 switch(compressionLevel()) {
80 zlevel = Z_DEFAULT_COMPRESSION;
83 _inflater = new z_stream;
84 memset(_inflater, 0, sizeof(z_stream));
85 if (Z_OK != inflateInit(_inflater)) {
86 qWarning() << "Could not initialize the inflate stream!";
90 _deflater = new z_stream;
91 memset(_deflater, 0, sizeof(z_stream));
92 if (Z_OK != deflateInit(_deflater, zlevel)) {
93 qWarning() << "Could not initialize the deflate stream!";
97 _inputBuffer.reserve(ioBufferSize); // pre-allocate space
98 _outputBuffer.resize(ioBufferSize); // not a typo; we never change the size of this buffer anyway (we *do* for _inputBuffer!)
100 qDebug() << "Enabling compression...";
107 qint64 Compressor::bytesAvailable() const
109 return _readBuffer.size();
113 qint64 Compressor::read(char *data, qint64 maxSize)
116 maxSize = _readBuffer.size();
118 qint64 n = qMin(maxSize, (qint64)_readBuffer.size());
119 memcpy(data, _readBuffer.constData(), n);
121 // TODO: don't copy for every read
122 if (n == _readBuffer.size())
125 _readBuffer = _readBuffer.mid(n);
127 // If there's still data left in the socket buffer, make sure to schedule a read
128 if (_socket->bytesAvailable())
129 QTimer::singleShot(0, this, &Compressor::readData);
135 // The usual usage pattern is to write a blocksize first, followed by the actual data.
136 // By setting NoFlush, one can indicate that the write buffer should not immediately be
137 // written, which should make things a bit more efficient.
138 qint64 Compressor::write(const char *data, qint64 count, WriteBufferHint flush)
140 int pos = _writeBuffer.size();
141 _writeBuffer.resize(pos + count);
142 memcpy(_writeBuffer.data() + pos, data, count);
144 if (flush != NoFlush)
151 void Compressor::readData()
153 // don't try to read more data if we're already closing
154 if (_socket->state() != QAbstractSocket::ConnectedState)
157 if (!_socket->bytesAvailable() || _readBuffer.size() >= maxBufferSize)
160 if (compressionLevel() == NoCompression) {
161 _readBuffer.append(_socket->read(maxBufferSize - _readBuffer.size()));
166 // We let zlib directly append to the readBuffer, which means we pre-allocate extra space for ioBufferSize.
167 // Afterwards, we'll shrink the buffer appropriately. Since shrinking should not reallocate, the readBuffer's
168 // capacity should over time adapt to the largest message sizes we encounter. However, this is not a bad thing
169 // considering that otherwise (using an intermediate buffer) we'd copy around data for every single message.
170 // TODO: Benchmark if it would still make sense to squeeze the buffer from time to time (e.g. after initial sync)!
172 while (_socket->bytesAvailable() && _readBuffer.size() + ioBufferSize < maxBufferSize && _inputBuffer.size() < ioBufferSize) {
173 _readBuffer.resize(_readBuffer.size() + ioBufferSize);
174 _inputBuffer.append(_socket->read(ioBufferSize - _inputBuffer.size()));
176 _inflater->next_in = reinterpret_cast<unsigned char *>(_inputBuffer.data());
177 _inflater->avail_in = _inputBuffer.size();
178 _inflater->next_out = reinterpret_cast<unsigned char *>(_readBuffer.data() + _readBuffer.size() - ioBufferSize);
179 _inflater->avail_out = ioBufferSize;
181 const unsigned char *orig_out = _inflater->next_out; // so we see if we have actually produced any output
183 int status = inflate(_inflater, Z_SYNC_FLUSH); // get as much data as possible
185 // adjust input and output buffers
186 _readBuffer.resize(_inflater->next_out - reinterpret_cast<unsigned char *>(_readBuffer.data()));
187 if (_inflater->avail_in > 0)
188 memmove(_inputBuffer.data(), _inflater->next_in, _inflater->avail_in);
189 _inputBuffer.resize(_inflater->avail_in);
191 if (_inflater->next_out != orig_out)
199 qWarning() << "Error while decompressing stream:" << status;
200 emit error(StreamError);
203 // means that we need more input to continue, so this is not an actual error
206 qWarning() << "Reached end of zlib stream!"; // this should really never happen
209 // just try to get more out of the stream
213 //qDebug() << "inflate in:" << _inflater->total_in << "out:" << _inflater->total_out << "ratio:" << (double)_inflater->total_in/_inflater->total_out;
217 void Compressor::writeData()
219 if (compressionLevel() == NoCompression) {
220 _socket->write(_writeBuffer);
221 _writeBuffer.clear();
225 _deflater->next_in = reinterpret_cast<unsigned char *>(_writeBuffer.data());
226 _deflater->avail_in = _writeBuffer.size();
230 _deflater->next_out = reinterpret_cast<unsigned char *>(_outputBuffer.data());
231 _deflater->avail_out = ioBufferSize;
232 status = deflate(_deflater, Z_PARTIAL_FLUSH);
233 if (status != Z_OK && status != Z_BUF_ERROR) {
234 qWarning() << "Error while compressing stream:" << status;
235 emit error(StreamError);
239 if (_deflater->avail_out == static_cast<unsigned int>(ioBufferSize))
240 continue; // nothing to write here
242 if (!_socket->write(_outputBuffer.constData(), ioBufferSize - _deflater->avail_out)) {
243 qWarning() << "Error while writing to socket:" << _socket->errorString();
244 emit error(DeviceError);
247 } while (_deflater->avail_out == 0); // the output buffer being full is the only reason we should have to loop here!
249 if (_deflater->avail_in > 0) {
250 qWarning() << "Oops, something weird happened: data still remaining in write buffer!";
251 emit error(StreamError);
254 _writeBuffer.resize(0);
256 //qDebug() << "deflate in:" << _deflater->total_in << "out:" << _deflater->total_out << "ratio:" << (double)_deflater->total_out/_deflater->total_in;
260 void Compressor::flush()
262 if (compressionLevel() == NoCompression && _socket->state() == QAbstractSocket::ConnectedState)
265 // FIXME: missing impl for enabled compression; but then we're not using this method yet