diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/ZHandle.cc | 102 |
1 files changed, 89 insertions, 13 deletions
diff --git a/src/ZHandle.cc b/src/ZHandle.cc index aa7bde7..8153343 100644 --- a/src/ZHandle.cc +++ b/src/ZHandle.cc @@ -1,5 +1,7 @@ #include "ZHandle.h" #include "Task.h" +#include "Async.h" +#include "TaskMan.h" Balau::ZStream::ZStream(const IO<Handle> & h, int level, header_t header) : m_h(h) { m_zin.zalloc = m_zout.zalloc = NULL; @@ -57,6 +59,28 @@ const char * Balau::ZStream::getName() { return m_name.to_charp(); } +namespace { + +class AsyncOpInflate : public Balau::AsyncOperation { + public: + AsyncOpInflate(z_stream * zin, int * r, Balau::Events::Custom * evt) : m_zin(zin), m_r(r), m_evt(evt) { } + virtual bool needsMainQueue() { return false; } + virtual bool needsFinishWorker() { return true; } + virtual void run() { + *m_r = inflate(m_zin, Z_SYNC_FLUSH); + } + virtual void done() { + m_evt->doSignal(); + delete this; + } + private: + z_stream * m_zin; + int * m_r; + Balau::Events::Custom * m_evt; +}; + +}; + static const int BLOCK_SIZE = 1024; ssize_t Balau::ZStream::read(void * buf, size_t count) throw (GeneralException) { @@ -66,22 +90,30 @@ ssize_t Balau::ZStream::read(void * buf, size_t count) throw (GeneralException) AAssert(m_h->canRead(), "Can't call ZStream::read on a non-readable handle."); size_t readTotal = 0; + const int block_size = BLOCK_SIZE * (m_useAsyncOp ? 16 : 1); m_zin.next_out = (Bytef *) buf; m_zin.avail_out = count; if (!m_in) { - m_zin.next_in = m_in = (uint8_t *) malloc(BLOCK_SIZE); + m_zin.next_in = m_in = (uint8_t *) malloc(block_size); m_zin.avail_in = 0; } while ((count != 0) && !m_h->isClosed() && !m_h->isEOF()) { if (m_zin.avail_in == 0) { m_zin.next_in = m_in; - size_t r = m_h->read(m_in, BLOCK_SIZE); + size_t r = m_h->read(m_in, block_size); if (r <= 0) return readTotal; m_zin.avail_in = r; } - Task::operationYield(); - int r = inflate(&m_zin, Z_SYNC_FLUSH); + int r = 0; + if (m_useAsyncOp) { + Events::Custom evt; + createAsyncOp(new AsyncOpInflate(&m_zin, &r, &evt)); + Task::operationYield(&evt); + } else { + r = inflate(&m_zin, Z_SYNC_FLUSH); + Task::operationYield(); + } size_t didRead = count - m_zin.avail_out; readTotal += didRead; count -= didRead; @@ -94,6 +126,28 @@ ssize_t Balau::ZStream::read(void * buf, size_t count) throw (GeneralException) return readTotal; } +namespace { + +class AsyncOpDeflate : public Balau::AsyncOperation { + public: + AsyncOpDeflate(z_stream * zout, int * r, int flush, Balau::Events::Custom * evt) : m_zout(zout), m_r(r), m_flush(flush), m_evt(evt) { } + virtual bool needsMainQueue() { return false; } + virtual bool needsFinishWorker() { return true; } + virtual void run() { + *m_r = deflate(m_zout, m_flush); + } + virtual void done() { + m_evt->doSignal(); + delete this; + } + private: + z_stream * m_zout; + int * m_r, m_flush; + Balau::Events::Custom * m_evt; +}; + +}; + ssize_t Balau::ZStream::write(const void * buf, size_t count) throw (GeneralException) { if (m_closed || m_eof) return 0; @@ -101,18 +155,28 @@ ssize_t Balau::ZStream::write(const void * buf, size_t count) throw (GeneralExce AAssert(m_h->canWrite(), "Can't call ZStream::write on a non-writable handle."); size_t wroteTotal = 0; + const int block_size = BLOCK_SIZE * (m_useAsyncOp ? 16 : 1); m_zout.next_in = (Bytef *) const_cast<void *>(buf); m_zout.avail_in = count; - void * obuf = alloca(BLOCK_SIZE); + void * obuf = m_useAsyncOp ? malloc(block_size) : alloca(block_size); while ((count != 0) && !m_h->isClosed()) { m_zout.next_out = (Bytef *) obuf; - m_zout.avail_out = BLOCK_SIZE; - int r = deflate(&m_zout, Z_NO_FLUSH); + m_zout.avail_out = block_size; + int r = 0; + if (m_useAsyncOp) { + Events::Custom evt; + createAsyncOp(new AsyncOpDeflate(&m_zout, &r, Z_NO_FLUSH, &evt)); + Task::operationYield(&evt); + } else { + r = deflate(&m_zout, Z_NO_FLUSH); + Task::operationYield(); + } EAssert(r == Z_OK, "deflate() didn't return Z_OK but %i", r); - Task::operationYield(); - size_t compressed = BLOCK_SIZE - m_zout.avail_out; + size_t compressed = block_size - m_zout.avail_out; if (compressed) { size_t w = m_h->forceWrite(obuf, compressed); + if (m_useAsyncOp) + free(obuf); if (w <= 0) return wroteTotal; } @@ -124,18 +188,30 @@ ssize_t Balau::ZStream::write(const void * buf, size_t count) throw (GeneralExce } void Balau::ZStream::doFlush(bool finish) { - void * buf = alloca(BLOCK_SIZE); + AAssert(m_h->canWrite(), "Can't call ZStream::doFlush on a non-writable handle."); + + const int block_size = BLOCK_SIZE * (m_useAsyncOp ? 16 : 1); + void * buf = m_useAsyncOp ? malloc(block_size) : alloca(block_size); m_zout.next_in = NULL; m_zout.avail_in = 0; int r; do { m_zout.next_out = (Bytef *) buf; - m_zout.avail_out = BLOCK_SIZE; - r = deflate(&m_zout, finish ? Z_FINISH : Z_SYNC_FLUSH); + m_zout.avail_out = block_size; + if (m_useAsyncOp) { + Events::Custom evt; + createAsyncOp(new AsyncOpDeflate(&m_zout, &r, finish ? Z_FINISH : Z_SYNC_FLUSH, &evt)); + Task::operationYield(&evt); + } else { + r = deflate(&m_zout, finish ? Z_FINISH : Z_SYNC_FLUSH); + Task::operationYield(); + } EAssert((r == Z_OK) || ((r == Z_STREAM_END) && finish), "deflate() didn't return Z_OK or Z_STREAM_END, but %i (finish = %s)", r, finish ? "true" : "false"); - size_t compressed = BLOCK_SIZE - m_zout.avail_out; + size_t compressed = block_size - m_zout.avail_out; if (compressed) { size_t w = m_h->forceWrite(buf, compressed); + if (m_useAsyncOp) + free(buf); if (w <= 0) return; } |