//#define Trace
|
|
// ParallelDeflateOutputStream.cs
|
// ------------------------------------------------------------------
|
//
|
// A DeflateStream that does compression only, it uses a
|
// divide-and-conquer approach with multiple threads to exploit multiple
|
// CPUs for the DEFLATE computation.
|
//
|
// last saved: <2011-July-31 14:49:40>
|
//
|
// ------------------------------------------------------------------
|
//
|
// Copyright (c) 2009-2011 by Dino Chiesa
|
// All rights reserved!
|
//
|
// This code module is part of DotNetZip, a zipfile class library.
|
//
|
// ------------------------------------------------------------------
|
//
|
// This code is licensed under the Microsoft Public License.
|
// See the file License.txt for the license details.
|
// More info on: http://dotnetzip.codeplex.com
|
//
|
// ------------------------------------------------------------------
|
|
using System;
|
using System.Collections.Generic;
|
using System.Threading;
|
using HH.WMS.Utils.Ionic.Zlib;
|
using System.IO;
|
|
|
namespace HH.WMS.Utils.Ionic.Zlib
|
{
|
internal class WorkItem
|
{
|
public byte[] buffer;
|
public byte[] compressed;
|
public int crc;
|
public int index;
|
public int ordinal;
|
public int inputBytesAvailable;
|
public int compressedBytesAvailable;
|
public ZlibCodec compressor;
|
|
public WorkItem(int size,
|
HH.WMS.Utils.Ionic.Zlib.CompressionLevel compressLevel,
|
CompressionStrategy strategy,
|
int ix)
|
{
|
this.buffer= new byte[size];
|
// alloc 5 bytes overhead for every block (margin of safety= 2)
|
int n = size + ((size / 32768)+1) * 5 * 2;
|
this.compressed = new byte[n];
|
this.compressor = new ZlibCodec();
|
this.compressor.InitializeDeflate(compressLevel, false);
|
this.compressor.OutputBuffer = this.compressed;
|
this.compressor.InputBuffer = this.buffer;
|
this.index = ix;
|
}
|
}
|
|
/// <summary>
|
/// A class for compressing streams using the
|
/// Deflate algorithm with multiple threads.
|
/// </summary>
|
///
|
/// <remarks>
|
/// <para>
|
/// This class performs DEFLATE compression through writing. For
|
/// more information on the Deflate algorithm, see IETF RFC 1951,
|
/// "DEFLATE Compressed Data Format Specification version 1.3."
|
/// </para>
|
///
|
/// <para>
|
/// This class is similar to <see cref="HH.WMS.Utils.Ionic.Zlib.DeflateStream"/>, except
|
/// that this class is for compression only, and this implementation uses an
|
/// approach that employs multiple worker threads to perform the DEFLATE. On
|
/// a multi-cpu or multi-core computer, the performance of this class can be
|
/// significantly higher than the single-threaded DeflateStream, particularly
|
/// for larger streams. How large? Anything over 10mb is a good candidate
|
/// for parallel compression.
|
/// </para>
|
///
|
/// <para>
|
/// The tradeoff is that this class uses more memory and more CPU than the
|
/// vanilla DeflateStream, and also is less efficient as a compressor. For
|
/// large files the size of the compressed data stream can be less than 1%
|
/// larger than the size of a compressed data stream from the vanialla
|
/// DeflateStream. For smaller files the difference can be larger. The
|
/// difference will also be larger if you set the BufferSize to be lower than
|
/// the default value. Your mileage may vary. Finally, for small files, the
|
/// ParallelDeflateOutputStream can be much slower than the vanilla
|
/// DeflateStream, because of the overhead associated to using the thread
|
/// pool.
|
/// </para>
|
///
|
/// </remarks>
|
/// <seealso cref="HH.WMS.Utils.Ionic.Zlib.DeflateStream" />
|
public class ParallelDeflateOutputStream : System.IO.Stream
|
{
|
|
private static readonly int IO_BUFFER_SIZE_DEFAULT = 64 * 1024; // 128k
|
private static readonly int BufferPairsPerCore = 4;
|
|
private System.Collections.Generic.List<WorkItem> _pool;
|
private bool _leaveOpen;
|
private bool emitting;
|
private System.IO.Stream _outStream;
|
private int _maxBufferPairs;
|
private int _bufferSize = IO_BUFFER_SIZE_DEFAULT;
|
private AutoResetEvent _newlyCompressedBlob;
|
//private ManualResetEvent _writingDone;
|
//private ManualResetEvent _sessionReset;
|
private object _outputLock = new object();
|
private bool _isClosed;
|
private bool _firstWriteDone;
|
private int _currentlyFilling;
|
private int _lastFilled;
|
private int _lastWritten;
|
private int _latestCompressed;
|
private int _Crc32;
|
private HH.WMS.Utils.Ionic.Crc.CRC32 _runningCrc;
|
private object _latestLock = new object();
|
private System.Collections.Generic.Queue<int> _toWrite;
|
private System.Collections.Generic.Queue<int> _toFill;
|
private Int64 _totalBytesProcessed;
|
private HH.WMS.Utils.Ionic.Zlib.CompressionLevel _compressLevel;
|
private volatile Exception _pendingException;
|
private bool _handlingException;
|
private object _eLock = new Object(); // protects _pendingException
|
|
// This bitfield is used only when Trace is defined.
|
//private TraceBits _DesiredTrace = TraceBits.Write | TraceBits.WriteBegin |
|
//TraceBits.WriteDone | TraceBits.Lifecycle | TraceBits.Fill | TraceBits.Flush |
|
//TraceBits.Session;
|
|
//private TraceBits _DesiredTrace = TraceBits.WriteBegin | TraceBits.WriteDone | TraceBits.Synch | TraceBits.Lifecycle | TraceBits.Session ;
|
|
private TraceBits _DesiredTrace =
|
TraceBits.Session |
|
TraceBits.Compress |
|
TraceBits.WriteTake |
|
TraceBits.WriteEnter |
|
TraceBits.EmitEnter |
|
TraceBits.EmitDone |
|
TraceBits.EmitLock |
|
TraceBits.EmitSkip |
|
TraceBits.EmitBegin;
|
|
/// <summary>
|
/// Create a ParallelDeflateOutputStream.
|
/// </summary>
|
/// <remarks>
|
///
|
/// <para>
|
/// This stream compresses data written into it via the DEFLATE
|
/// algorithm (see RFC 1951), and writes out the compressed byte stream.
|
/// </para>
|
///
|
/// <para>
|
/// The instance will use the default compression level, the default
|
/// buffer sizes and the default number of threads and buffers per
|
/// thread.
|
/// </para>
|
///
|
/// <para>
|
/// This class is similar to <see cref="HH.WMS.Utils.Ionic.Zlib.DeflateStream"/>,
|
/// except that this implementation uses an approach that employs
|
/// multiple worker threads to perform the DEFLATE. On a multi-cpu or
|
/// multi-core computer, the performance of this class can be
|
/// significantly higher than the single-threaded DeflateStream,
|
/// particularly for larger streams. How large? Anything over 10mb is
|
/// a good candidate for parallel compression.
|
/// </para>
|
///
|
/// </remarks>
|
///
|
/// <example>
|
///
|
/// This example shows how to use a ParallelDeflateOutputStream to compress
|
/// data. It reads a file, compresses it, and writes the compressed data to
|
/// a second, output file.
|
///
|
/// <code>
|
/// byte[] buffer = new byte[WORKING_BUFFER_SIZE];
|
/// int n= -1;
|
/// String outputFile = fileToCompress + ".compressed";
|
/// using (System.IO.Stream input = System.IO.File.OpenRead(fileToCompress))
|
/// {
|
/// using (var raw = System.IO.File.Create(outputFile))
|
/// {
|
/// using (Stream compressor = new ParallelDeflateOutputStream(raw))
|
/// {
|
/// while ((n= input.Read(buffer, 0, buffer.Length)) != 0)
|
/// {
|
/// compressor.Write(buffer, 0, n);
|
/// }
|
/// }
|
/// }
|
/// }
|
/// </code>
|
/// <code lang="VB">
|
/// Dim buffer As Byte() = New Byte(4096) {}
|
/// Dim n As Integer = -1
|
/// Dim outputFile As String = (fileToCompress & ".compressed")
|
/// Using input As Stream = File.OpenRead(fileToCompress)
|
/// Using raw As FileStream = File.Create(outputFile)
|
/// Using compressor As Stream = New ParallelDeflateOutputStream(raw)
|
/// Do While (n <> 0)
|
/// If (n > 0) Then
|
/// compressor.Write(buffer, 0, n)
|
/// End If
|
/// n = input.Read(buffer, 0, buffer.Length)
|
/// Loop
|
/// End Using
|
/// End Using
|
/// End Using
|
/// </code>
|
/// </example>
|
/// <param name="stream">The stream to which compressed data will be written.</param>
|
public ParallelDeflateOutputStream(System.IO.Stream stream)
|
: this(stream, CompressionLevel.Default, CompressionStrategy.Default, false)
|
{
|
}
|
|
/// <summary>
|
/// Create a ParallelDeflateOutputStream using the specified CompressionLevel.
|
/// </summary>
|
/// <remarks>
|
/// See the <see cref="ParallelDeflateOutputStream(System.IO.Stream)"/>
|
/// constructor for example code.
|
/// </remarks>
|
/// <param name="stream">The stream to which compressed data will be written.</param>
|
/// <param name="level">A tuning knob to trade speed for effectiveness.</param>
|
public ParallelDeflateOutputStream(System.IO.Stream stream, CompressionLevel level)
|
: this(stream, level, CompressionStrategy.Default, false)
|
{
|
}
|
|
/// <summary>
|
/// Create a ParallelDeflateOutputStream and specify whether to leave the captive stream open
|
/// when the ParallelDeflateOutputStream is closed.
|
/// </summary>
|
/// <remarks>
|
/// See the <see cref="ParallelDeflateOutputStream(System.IO.Stream)"/>
|
/// constructor for example code.
|
/// </remarks>
|
/// <param name="stream">The stream to which compressed data will be written.</param>
|
/// <param name="leaveOpen">
|
/// true if the application would like the stream to remain open after inflation/deflation.
|
/// </param>
|
public ParallelDeflateOutputStream(System.IO.Stream stream, bool leaveOpen)
|
: this(stream, CompressionLevel.Default, CompressionStrategy.Default, leaveOpen)
|
{
|
}
|
|
/// <summary>
|
/// Create a ParallelDeflateOutputStream and specify whether to leave the captive stream open
|
/// when the ParallelDeflateOutputStream is closed.
|
/// </summary>
|
/// <remarks>
|
/// See the <see cref="ParallelDeflateOutputStream(System.IO.Stream)"/>
|
/// constructor for example code.
|
/// </remarks>
|
/// <param name="stream">The stream to which compressed data will be written.</param>
|
/// <param name="level">A tuning knob to trade speed for effectiveness.</param>
|
/// <param name="leaveOpen">
|
/// true if the application would like the stream to remain open after inflation/deflation.
|
/// </param>
|
public ParallelDeflateOutputStream(System.IO.Stream stream, CompressionLevel level, bool leaveOpen)
|
: this(stream, CompressionLevel.Default, CompressionStrategy.Default, leaveOpen)
|
{
|
}
|
|
/// <summary>
|
/// Create a ParallelDeflateOutputStream using the specified
|
/// CompressionLevel and CompressionStrategy, and specifying whether to
|
/// leave the captive stream open when the ParallelDeflateOutputStream is
|
/// closed.
|
/// </summary>
|
/// <remarks>
|
/// See the <see cref="ParallelDeflateOutputStream(System.IO.Stream)"/>
|
/// constructor for example code.
|
/// </remarks>
|
/// <param name="stream">The stream to which compressed data will be written.</param>
|
/// <param name="level">A tuning knob to trade speed for effectiveness.</param>
|
/// <param name="strategy">
|
/// By tweaking this parameter, you may be able to optimize the compression for
|
/// data with particular characteristics.
|
/// </param>
|
/// <param name="leaveOpen">
|
/// true if the application would like the stream to remain open after inflation/deflation.
|
/// </param>
|
public ParallelDeflateOutputStream(System.IO.Stream stream,
|
CompressionLevel level,
|
CompressionStrategy strategy,
|
bool leaveOpen)
|
{
|
TraceOutput(TraceBits.Lifecycle | TraceBits.Session, "-------------------------------------------------------");
|
TraceOutput(TraceBits.Lifecycle | TraceBits.Session, "Create {0:X8}", this.GetHashCode());
|
_outStream = stream;
|
_compressLevel= level;
|
Strategy = strategy;
|
_leaveOpen = leaveOpen;
|
this.MaxBufferPairs = 16; // default
|
}
|
|
|
/// <summary>
|
/// The ZLIB strategy to be used during compression.
|
/// </summary>
|
///
|
public CompressionStrategy Strategy
|
{
|
get;
|
private set;
|
}
|
|
/// <summary>
|
/// The maximum number of buffer pairs to use.
|
/// </summary>
|
///
|
/// <remarks>
|
/// <para>
|
/// This property sets an upper limit on the number of memory buffer
|
/// pairs to create. The implementation of this stream allocates
|
/// multiple buffers to facilitate parallel compression. As each buffer
|
/// fills up, this stream uses <see
|
/// cref="System.Threading.ThreadPool.QueueUserWorkItem(WaitCallback)">
|
/// ThreadPool.QueueUserWorkItem()</see>
|
/// to compress those buffers in a background threadpool thread. After a
|
/// buffer is compressed, it is re-ordered and written to the output
|
/// stream.
|
/// </para>
|
///
|
/// <para>
|
/// A higher number of buffer pairs enables a higher degree of
|
/// parallelism, which tends to increase the speed of compression on
|
/// multi-cpu computers. On the other hand, a higher number of buffer
|
/// pairs also implies a larger memory consumption, more active worker
|
/// threads, and a higher cpu utilization for any compression. This
|
/// property enables the application to limit its memory consumption and
|
/// CPU utilization behavior depending on requirements.
|
/// </para>
|
///
|
/// <para>
|
/// For each compression "task" that occurs in parallel, there are 2
|
/// buffers allocated: one for input and one for output. This property
|
/// sets a limit for the number of pairs. The total amount of storage
|
/// space allocated for buffering will then be (N*S*2), where N is the
|
/// number of buffer pairs, S is the size of each buffer (<see
|
/// cref="BufferSize"/>). By default, DotNetZip allocates 4 buffer
|
/// pairs per CPU core, so if your machine has 4 cores, and you retain
|
/// the default buffer size of 128k, then the
|
/// ParallelDeflateOutputStream will use 4 * 4 * 2 * 128kb of buffer
|
/// memory in total, or 4mb, in blocks of 128kb. If you then set this
|
/// property to 8, then the number will be 8 * 2 * 128kb of buffer
|
/// memory, or 2mb.
|
/// </para>
|
///
|
/// <para>
|
/// CPU utilization will also go up with additional buffers, because a
|
/// larger number of buffer pairs allows a larger number of background
|
/// threads to compress in parallel. If you find that parallel
|
/// compression is consuming too much memory or CPU, you can adjust this
|
/// value downward.
|
/// </para>
|
///
|
/// <para>
|
/// The default value is 16. Different values may deliver better or
|
/// worse results, depending on your priorities and the dynamic
|
/// performance characteristics of your storage and compute resources.
|
/// </para>
|
///
|
/// <para>
|
/// This property is not the number of buffer pairs to use; it is an
|
/// upper limit. An illustration: Suppose you have an application that
|
/// uses the default value of this property (which is 16), and it runs
|
/// on a machine with 2 CPU cores. In that case, DotNetZip will allocate
|
/// 4 buffer pairs per CPU core, for a total of 8 pairs. The upper
|
/// limit specified by this property has no effect.
|
/// </para>
|
///
|
/// <para>
|
/// The application can set this value at any time, but it is effective
|
/// only before the first call to Write(), which is when the buffers are
|
/// allocated.
|
/// </para>
|
/// </remarks>
|
public int MaxBufferPairs
|
{
|
get
|
{
|
return _maxBufferPairs;
|
}
|
set
|
{
|
if (value < 4)
|
throw new ArgumentException("MaxBufferPairs",
|
"Value must be 4 or greater.");
|
_maxBufferPairs = value;
|
}
|
}
|
|
/// <summary>
|
/// The size of the buffers used by the compressor threads.
|
/// </summary>
|
/// <remarks>
|
///
|
/// <para>
|
/// The default buffer size is 128k. The application can set this value
|
/// at any time, but it is effective only before the first Write().
|
/// </para>
|
///
|
/// <para>
|
/// Larger buffer sizes implies larger memory consumption but allows
|
/// more efficient compression. Using smaller buffer sizes consumes less
|
/// memory but may result in less effective compression. For example,
|
/// using the default buffer size of 128k, the compression delivered is
|
/// within 1% of the compression delivered by the single-threaded <see
|
/// cref="HH.WMS.Utils.Ionic.Zlib.DeflateStream"/>. On the other hand, using a
|
/// BufferSize of 8k can result in a compressed data stream that is 5%
|
/// larger than that delivered by the single-threaded
|
/// <c>DeflateStream</c>. Excessively small buffer sizes can also cause
|
/// the speed of the ParallelDeflateOutputStream to drop, because of
|
/// larger thread scheduling overhead dealing with many many small
|
/// buffers.
|
/// </para>
|
///
|
/// <para>
|
/// The total amount of storage space allocated for buffering will be
|
/// (N*S*2), where N is the number of buffer pairs, and S is the size of
|
/// each buffer (this property). There are 2 buffers used by the
|
/// compressor, one for input and one for output. By default, DotNetZip
|
/// allocates 4 buffer pairs per CPU core, so if your machine has 4
|
/// cores, then the number of buffer pairs used will be 16. If you
|
/// accept the default value of this property, 128k, then the
|
/// ParallelDeflateOutputStream will use 16 * 2 * 128kb of buffer memory
|
/// in total, or 4mb, in blocks of 128kb. If you set this property to
|
/// 64kb, then the number will be 16 * 2 * 64kb of buffer memory, or
|
/// 2mb.
|
/// </para>
|
///
|
/// </remarks>
|
public int BufferSize
|
{
|
get { return _bufferSize;}
|
set
|
{
|
if (value < 1024)
|
throw new ArgumentOutOfRangeException("BufferSize",
|
"BufferSize must be greater than 1024 bytes");
|
_bufferSize = value;
|
}
|
}
|
|
/// <summary>
|
/// The CRC32 for the data that was written out, prior to compression.
|
/// </summary>
|
/// <remarks>
|
/// This value is meaningful only after a call to Close().
|
/// </remarks>
|
public int Crc32 { get { return _Crc32; } }
|
|
|
/// <summary>
|
/// The total number of uncompressed bytes processed by the ParallelDeflateOutputStream.
|
/// </summary>
|
/// <remarks>
|
/// This value is meaningful only after a call to Close().
|
/// </remarks>
|
public Int64 BytesProcessed { get { return _totalBytesProcessed; } }
|
|
|
private void _InitializePoolOfWorkItems()
|
{
|
_toWrite = new Queue<int>();
|
_toFill = new Queue<int>();
|
_pool = new System.Collections.Generic.List<WorkItem>();
|
int nTasks = BufferPairsPerCore * Environment.ProcessorCount;
|
nTasks = Math.Min(nTasks, _maxBufferPairs);
|
for(int i=0; i < nTasks; i++)
|
{
|
_pool.Add(new WorkItem(_bufferSize, _compressLevel, Strategy, i));
|
_toFill.Enqueue(i);
|
}
|
|
_newlyCompressedBlob = new AutoResetEvent(false);
|
_runningCrc = new HH.WMS.Utils.Ionic.Crc.CRC32();
|
_currentlyFilling = -1;
|
_lastFilled = -1;
|
_lastWritten = -1;
|
_latestCompressed = -1;
|
}
|
|
|
|
|
/// <summary>
|
/// Write data to the stream.
|
/// </summary>
|
///
|
/// <remarks>
|
///
|
/// <para>
|
/// To use the ParallelDeflateOutputStream to compress data, create a
|
/// ParallelDeflateOutputStream with CompressionMode.Compress, passing a
|
/// writable output stream. Then call Write() on that
|
/// ParallelDeflateOutputStream, providing uncompressed data as input. The
|
/// data sent to the output stream will be the compressed form of the data
|
/// written.
|
/// </para>
|
///
|
/// <para>
|
/// To decompress data, use the <see cref="HH.WMS.Utils.Ionic.Zlib.DeflateStream"/> class.
|
/// </para>
|
///
|
/// </remarks>
|
/// <param name="buffer">The buffer holding data to write to the stream.</param>
|
/// <param name="offset">the offset within that data array to find the first byte to write.</param>
|
/// <param name="count">the number of bytes to write.</param>
|
public override void Write(byte[] buffer, int offset, int count)
|
{
|
bool mustWait = false;
|
|
// This method does this:
|
// 0. handles any pending exceptions
|
// 1. write any buffers that are ready to be written,
|
// 2. fills a work buffer; when full, flip state to 'Filled',
|
// 3. if more data to be written, goto step 1
|
|
if (_isClosed)
|
throw new InvalidOperationException();
|
|
// dispense any exceptions that occurred on the BG threads
|
if (_pendingException != null)
|
{
|
_handlingException = true;
|
var pe = _pendingException;
|
_pendingException = null;
|
throw pe;
|
}
|
|
if (count == 0) return;
|
|
if (!_firstWriteDone)
|
{
|
// Want to do this on first Write, first session, and not in the
|
// constructor. We want to allow MaxBufferPairs to
|
// change after construction, but before first Write.
|
_InitializePoolOfWorkItems();
|
_firstWriteDone = true;
|
}
|
|
|
do
|
{
|
// may need to make buffers available
|
EmitPendingBuffers(false, mustWait);
|
|
mustWait = false;
|
// use current buffer, or get a new buffer to fill
|
int ix = -1;
|
if (_currentlyFilling >= 0)
|
{
|
ix = _currentlyFilling;
|
TraceOutput(TraceBits.WriteTake,
|
"Write notake wi({0}) lf({1})",
|
ix,
|
_lastFilled);
|
}
|
else
|
{
|
TraceOutput(TraceBits.WriteTake, "Write take?");
|
if (_toFill.Count == 0)
|
{
|
// no available buffers, so... need to emit
|
// compressed buffers.
|
mustWait = true;
|
continue;
|
}
|
|
ix = _toFill.Dequeue();
|
TraceOutput(TraceBits.WriteTake,
|
"Write take wi({0}) lf({1})",
|
ix,
|
_lastFilled);
|
++_lastFilled; // TODO: consider rollover?
|
}
|
|
WorkItem workitem = _pool[ix];
|
|
int limit = ((workitem.buffer.Length - workitem.inputBytesAvailable) > count)
|
? count
|
: (workitem.buffer.Length - workitem.inputBytesAvailable);
|
|
workitem.ordinal = _lastFilled;
|
|
TraceOutput(TraceBits.Write,
|
"Write lock wi({0}) ord({1}) iba({2})",
|
workitem.index,
|
workitem.ordinal,
|
workitem.inputBytesAvailable
|
);
|
|
// copy from the provided buffer to our workitem, starting at
|
// the tail end of whatever data we might have in there currently.
|
Buffer.BlockCopy(buffer,
|
offset,
|
workitem.buffer,
|
workitem.inputBytesAvailable,
|
limit);
|
|
count -= limit;
|
offset += limit;
|
workitem.inputBytesAvailable += limit;
|
if (workitem.inputBytesAvailable == workitem.buffer.Length)
|
{
|
// No need for interlocked.increment: the Write()
|
// method is documented as not multi-thread safe, so
|
// we can assume Write() calls come in from only one
|
// thread.
|
TraceOutput(TraceBits.Write,
|
"Write QUWI wi({0}) ord({1}) iba({2}) nf({3})",
|
workitem.index,
|
workitem.ordinal,
|
workitem.inputBytesAvailable );
|
|
if (!ThreadPool.QueueUserWorkItem( _DeflateOne, workitem ))
|
throw new Exception("Cannot enqueue workitem");
|
|
_currentlyFilling = -1; // will get a new buffer next time
|
}
|
else
|
_currentlyFilling = ix;
|
|
if (count > 0)
|
TraceOutput(TraceBits.WriteEnter, "Write more");
|
}
|
while (count > 0); // until no more to write
|
|
TraceOutput(TraceBits.WriteEnter, "Write exit");
|
return;
|
}
|
|
|
|
private void _FlushFinish()
|
{
|
// After writing a series of compressed buffers, each one closed
|
// with Flush.Sync, we now write the final one as Flush.Finish,
|
// and then stop.
|
byte[] buffer = new byte[128];
|
var compressor = new ZlibCodec();
|
int rc = compressor.InitializeDeflate(_compressLevel, false);
|
compressor.InputBuffer = null;
|
compressor.NextIn = 0;
|
compressor.AvailableBytesIn = 0;
|
compressor.OutputBuffer = buffer;
|
compressor.NextOut = 0;
|
compressor.AvailableBytesOut = buffer.Length;
|
rc = compressor.Deflate(FlushType.Finish);
|
|
if (rc != ZlibConstants.Z_STREAM_END && rc != ZlibConstants.Z_OK)
|
throw new Exception("deflating: " + compressor.Message);
|
|
if (buffer.Length - compressor.AvailableBytesOut > 0)
|
{
|
TraceOutput(TraceBits.EmitBegin,
|
"Emit begin flush bytes({0})",
|
buffer.Length - compressor.AvailableBytesOut);
|
|
_outStream.Write(buffer, 0, buffer.Length - compressor.AvailableBytesOut);
|
|
TraceOutput(TraceBits.EmitDone,
|
"Emit done flush");
|
}
|
|
compressor.EndDeflate();
|
|
_Crc32 = _runningCrc.Crc32Result;
|
}
|
|
|
private void _Flush(bool lastInput)
|
{
|
if (_isClosed)
|
throw new InvalidOperationException();
|
|
if (emitting) return;
|
|
// compress any partial buffer
|
if (_currentlyFilling >= 0)
|
{
|
WorkItem workitem = _pool[_currentlyFilling];
|
_DeflateOne(workitem);
|
_currentlyFilling = -1; // get a new buffer next Write()
|
}
|
|
if (lastInput)
|
{
|
EmitPendingBuffers(true, false);
|
_FlushFinish();
|
}
|
else
|
{
|
EmitPendingBuffers(false, false);
|
}
|
}
|
|
|
|
/// <summary>
|
/// Flush the stream.
|
/// </summary>
|
public override void Flush()
|
{
|
if (_pendingException != null)
|
{
|
_handlingException = true;
|
var pe = _pendingException;
|
_pendingException = null;
|
throw pe;
|
}
|
if (_handlingException)
|
return;
|
|
_Flush(false);
|
}
|
|
|
/// <summary>
|
/// Close the stream.
|
/// </summary>
|
/// <remarks>
|
/// You must call Close on the stream to guarantee that all of the data written in has
|
/// been compressed, and the compressed data has been written out.
|
/// </remarks>
|
public override void Close()
|
{
|
TraceOutput(TraceBits.Session, "Close {0:X8}", this.GetHashCode());
|
|
if (_pendingException != null)
|
{
|
_handlingException = true;
|
var pe = _pendingException;
|
_pendingException = null;
|
throw pe;
|
}
|
|
if (_handlingException)
|
return;
|
|
if (_isClosed) return;
|
|
_Flush(true);
|
|
if (!_leaveOpen)
|
_outStream.Close();
|
|
_isClosed= true;
|
}
|
|
|
|
// workitem 10030 - implement a new Dispose method
|
|
/// <summary>Dispose the object</summary>
|
/// <remarks>
|
/// <para>
|
/// Because ParallelDeflateOutputStream is IDisposable, the
|
/// application must call this method when finished using the instance.
|
/// </para>
|
/// <para>
|
/// This method is generally called implicitly upon exit from
|
/// a <c>using</c> scope in C# (<c>Using</c> in VB).
|
/// </para>
|
/// </remarks>
|
new public void Dispose()
|
{
|
TraceOutput(TraceBits.Lifecycle, "Dispose {0:X8}", this.GetHashCode());
|
Close();
|
_pool = null;
|
Dispose(true);
|
}
|
|
|
|
/// <summary>The Dispose method</summary>
|
/// <param name="disposing">
|
/// indicates whether the Dispose method was invoked by user code.
|
/// </param>
|
protected override void Dispose(bool disposing)
|
{
|
base.Dispose(disposing);
|
}
|
|
|
/// <summary>
|
/// Resets the stream for use with another stream.
|
/// </summary>
|
/// <remarks>
|
/// Because the ParallelDeflateOutputStream is expensive to create, it
|
/// has been designed so that it can be recycled and re-used. You have
|
/// to call Close() on the stream first, then you can call Reset() on
|
/// it, to use it again on another stream.
|
/// </remarks>
|
///
|
/// <param name="stream">
|
/// The new output stream for this era.
|
/// </param>
|
///
|
/// <example>
|
/// <code>
|
/// ParallelDeflateOutputStream deflater = null;
|
/// foreach (var inputFile in listOfFiles)
|
/// {
|
/// string outputFile = inputFile + ".compressed";
|
/// using (System.IO.Stream input = System.IO.File.OpenRead(inputFile))
|
/// {
|
/// using (var outStream = System.IO.File.Create(outputFile))
|
/// {
|
/// if (deflater == null)
|
/// deflater = new ParallelDeflateOutputStream(outStream,
|
/// CompressionLevel.Best,
|
/// CompressionStrategy.Default,
|
/// true);
|
/// deflater.Reset(outStream);
|
///
|
/// while ((n= input.Read(buffer, 0, buffer.Length)) != 0)
|
/// {
|
/// deflater.Write(buffer, 0, n);
|
/// }
|
/// }
|
/// }
|
/// }
|
/// </code>
|
/// </example>
|
public void Reset(Stream stream)
|
{
|
TraceOutput(TraceBits.Session, "-------------------------------------------------------");
|
TraceOutput(TraceBits.Session, "Reset {0:X8} firstDone({1})", this.GetHashCode(), _firstWriteDone);
|
|
if (!_firstWriteDone) return;
|
|
// reset all status
|
_toWrite.Clear();
|
_toFill.Clear();
|
foreach (var workitem in _pool)
|
{
|
_toFill.Enqueue(workitem.index);
|
workitem.ordinal = -1;
|
}
|
|
_firstWriteDone = false;
|
_totalBytesProcessed = 0L;
|
_runningCrc = new HH.WMS.Utils.Ionic.Crc.CRC32();
|
_isClosed= false;
|
_currentlyFilling = -1;
|
_lastFilled = -1;
|
_lastWritten = -1;
|
_latestCompressed = -1;
|
_outStream = stream;
|
}
|
|
|
|
|
private void EmitPendingBuffers(bool doAll, bool mustWait)
|
{
|
// When combining parallel deflation with a ZipSegmentedStream, it's
|
// possible for the ZSS to throw from within this method. In that
|
// case, Close/Dispose will be called on this stream, if this stream
|
// is employed within a using or try/finally pair as required. But
|
// this stream is unaware of the pending exception, so the Close()
|
// method invokes this method AGAIN. This can lead to a deadlock.
|
// Therefore, failfast if re-entering.
|
|
if (emitting) return;
|
emitting = true;
|
if (doAll || mustWait)
|
_newlyCompressedBlob.WaitOne();
|
|
do
|
{
|
int firstSkip = -1;
|
int millisecondsToWait = doAll ? 200 : (mustWait ? -1 : 0);
|
int nextToWrite = -1;
|
|
do
|
{
|
if (Monitor.TryEnter(_toWrite, millisecondsToWait))
|
{
|
nextToWrite = -1;
|
try
|
{
|
if (_toWrite.Count > 0)
|
nextToWrite = _toWrite.Dequeue();
|
}
|
finally
|
{
|
Monitor.Exit(_toWrite);
|
}
|
|
if (nextToWrite >= 0)
|
{
|
WorkItem workitem = _pool[nextToWrite];
|
if (workitem.ordinal != _lastWritten + 1)
|
{
|
// out of order. requeue and try again.
|
TraceOutput(TraceBits.EmitSkip,
|
"Emit skip wi({0}) ord({1}) lw({2}) fs({3})",
|
workitem.index,
|
workitem.ordinal,
|
_lastWritten,
|
firstSkip);
|
|
lock(_toWrite)
|
{
|
_toWrite.Enqueue(nextToWrite);
|
}
|
|
if (firstSkip == nextToWrite)
|
{
|
// We went around the list once.
|
// None of the items in the list is the one we want.
|
// Now wait for a compressor to signal again.
|
_newlyCompressedBlob.WaitOne();
|
firstSkip = -1;
|
}
|
else if (firstSkip == -1)
|
firstSkip = nextToWrite;
|
|
continue;
|
}
|
|
firstSkip = -1;
|
|
TraceOutput(TraceBits.EmitBegin,
|
"Emit begin wi({0}) ord({1}) cba({2})",
|
workitem.index,
|
workitem.ordinal,
|
workitem.compressedBytesAvailable);
|
|
_outStream.Write(workitem.compressed, 0, workitem.compressedBytesAvailable);
|
_runningCrc.Combine(workitem.crc, workitem.inputBytesAvailable);
|
_totalBytesProcessed += workitem.inputBytesAvailable;
|
workitem.inputBytesAvailable = 0;
|
|
TraceOutput(TraceBits.EmitDone,
|
"Emit done wi({0}) ord({1}) cba({2}) mtw({3})",
|
workitem.index,
|
workitem.ordinal,
|
workitem.compressedBytesAvailable,
|
millisecondsToWait);
|
|
_lastWritten = workitem.ordinal;
|
_toFill.Enqueue(workitem.index);
|
|
// don't wait next time through
|
if (millisecondsToWait == -1) millisecondsToWait = 0;
|
}
|
}
|
else
|
nextToWrite = -1;
|
|
} while (nextToWrite >= 0);
|
|
} while (doAll && (_lastWritten != _latestCompressed));
|
|
emitting = false;
|
}
|
|
|
|
#if OLD
|
private void _PerpetualWriterMethod(object state)
|
{
|
TraceOutput(TraceBits.WriterThread, "_PerpetualWriterMethod START");
|
|
try
|
{
|
do
|
{
|
// wait for the next session
|
TraceOutput(TraceBits.Synch | TraceBits.WriterThread, "Synch _sessionReset.WaitOne(begin) PWM");
|
_sessionReset.WaitOne();
|
TraceOutput(TraceBits.Synch | TraceBits.WriterThread, "Synch _sessionReset.WaitOne(done) PWM");
|
|
if (_isDisposed) break;
|
|
TraceOutput(TraceBits.Synch | TraceBits.WriterThread, "Synch _sessionReset.Reset() PWM");
|
_sessionReset.Reset();
|
|
// repeatedly write buffers as they become ready
|
WorkItem workitem = null;
|
HH.WMS.Utils.Ionic.Zlib.CRC32 c= new HH.WMS.Utils.Ionic.Zlib.CRC32();
|
do
|
{
|
workitem = _pool[_nextToWrite % _pc];
|
lock(workitem)
|
{
|
if (_noMoreInputForThisSegment)
|
TraceOutput(TraceBits.Write,
|
"Write drain wi({0}) stat({1}) canuse({2}) cba({3})",
|
workitem.index,
|
workitem.status,
|
(workitem.status == (int)WorkItem.Status.Compressed),
|
workitem.compressedBytesAvailable);
|
|
do
|
{
|
if (workitem.status == (int)WorkItem.Status.Compressed)
|
{
|
TraceOutput(TraceBits.WriteBegin,
|
"Write begin wi({0}) stat({1}) cba({2})",
|
workitem.index,
|
workitem.status,
|
workitem.compressedBytesAvailable);
|
|
workitem.status = (int)WorkItem.Status.Writing;
|
_outStream.Write(workitem.compressed, 0, workitem.compressedBytesAvailable);
|
c.Combine(workitem.crc, workitem.inputBytesAvailable);
|
_totalBytesProcessed += workitem.inputBytesAvailable;
|
_nextToWrite++;
|
workitem.inputBytesAvailable= 0;
|
workitem.status = (int)WorkItem.Status.Done;
|
|
TraceOutput(TraceBits.WriteDone,
|
"Write done wi({0}) stat({1}) cba({2})",
|
workitem.index,
|
workitem.status,
|
workitem.compressedBytesAvailable);
|
|
|
Monitor.Pulse(workitem);
|
break;
|
}
|
else
|
{
|
int wcycles = 0;
|
// I've locked a workitem I cannot use.
|
// Therefore, wake someone else up, and then release the lock.
|
while (workitem.status != (int)WorkItem.Status.Compressed)
|
{
|
TraceOutput(TraceBits.WriteWait,
|
"Write waiting wi({0}) stat({1}) nw({2}) nf({3}) nomore({4})",
|
workitem.index,
|
workitem.status,
|
_nextToWrite, _nextToFill,
|
_noMoreInputForThisSegment );
|
|
if (_noMoreInputForThisSegment && _nextToWrite == _nextToFill)
|
break;
|
|
wcycles++;
|
|
// wake up someone else
|
Monitor.Pulse(workitem);
|
// release and wait
|
Monitor.Wait(workitem);
|
|
if (workitem.status == (int)WorkItem.Status.Compressed)
|
TraceOutput(TraceBits.WriteWait,
|
"Write A-OK wi({0}) stat({1}) iba({2}) cba({3}) cyc({4})",
|
workitem.index,
|
workitem.status,
|
workitem.inputBytesAvailable,
|
workitem.compressedBytesAvailable,
|
wcycles);
|
}
|
|
if (_noMoreInputForThisSegment && _nextToWrite == _nextToFill)
|
break;
|
|
}
|
}
|
while (true);
|
}
|
|
if (_noMoreInputForThisSegment)
|
TraceOutput(TraceBits.Write,
|
"Write nomore nw({0}) nf({1}) break({2})",
|
_nextToWrite, _nextToFill, (_nextToWrite == _nextToFill));
|
|
if (_noMoreInputForThisSegment && _nextToWrite == _nextToFill)
|
break;
|
|
} while (true);
|
|
|
// Finish:
|
// After writing a series of buffers, closing each one with
|
// Flush.Sync, we now write the final one as Flush.Finish, and
|
// then stop.
|
byte[] buffer = new byte[128];
|
ZlibCodec compressor = new ZlibCodec();
|
int rc = compressor.InitializeDeflate(_compressLevel, false);
|
compressor.InputBuffer = null;
|
compressor.NextIn = 0;
|
compressor.AvailableBytesIn = 0;
|
compressor.OutputBuffer = buffer;
|
compressor.NextOut = 0;
|
compressor.AvailableBytesOut = buffer.Length;
|
rc = compressor.Deflate(FlushType.Finish);
|
|
if (rc != ZlibConstants.Z_STREAM_END && rc != ZlibConstants.Z_OK)
|
throw new Exception("deflating: " + compressor.Message);
|
|
if (buffer.Length - compressor.AvailableBytesOut > 0)
|
{
|
TraceOutput(TraceBits.WriteBegin,
|
"Write begin flush bytes({0})",
|
buffer.Length - compressor.AvailableBytesOut);
|
|
_outStream.Write(buffer, 0, buffer.Length - compressor.AvailableBytesOut);
|
|
TraceOutput(TraceBits.WriteBegin,
|
"Write done flush");
|
}
|
|
compressor.EndDeflate();
|
|
_Crc32 = c.Crc32Result;
|
|
// signal that writing is complete:
|
TraceOutput(TraceBits.Synch, "Synch _writingDone.Set() PWM");
|
_writingDone.Set();
|
}
|
while (true);
|
}
|
catch (System.Exception exc1)
|
{
|
lock(_eLock)
|
{
|
// expose the exception to the main thread
|
if (_pendingException!=null)
|
_pendingException = exc1;
|
}
|
}
|
|
TraceOutput(TraceBits.WriterThread, "_PerpetualWriterMethod FINIS");
|
}
|
#endif
|
|
|
|
|
private void _DeflateOne(Object wi)
|
{
|
// compress one buffer
|
WorkItem workitem = (WorkItem) wi;
|
try
|
{
|
int myItem = workitem.index;
|
HH.WMS.Utils.Ionic.Crc.CRC32 crc = new HH.WMS.Utils.Ionic.Crc.CRC32();
|
|
// calc CRC on the buffer
|
crc.SlurpBlock(workitem.buffer, 0, workitem.inputBytesAvailable);
|
|
// deflate it
|
DeflateOneSegment(workitem);
|
|
// update status
|
workitem.crc = crc.Crc32Result;
|
TraceOutput(TraceBits.Compress,
|
"Compress wi({0}) ord({1}) len({2})",
|
workitem.index,
|
workitem.ordinal,
|
workitem.compressedBytesAvailable
|
);
|
|
lock(_latestLock)
|
{
|
if (workitem.ordinal > _latestCompressed)
|
_latestCompressed = workitem.ordinal;
|
}
|
lock (_toWrite)
|
{
|
_toWrite.Enqueue(workitem.index);
|
}
|
_newlyCompressedBlob.Set();
|
}
|
catch (System.Exception exc1)
|
{
|
lock(_eLock)
|
{
|
// expose the exception to the main thread
|
if (_pendingException!=null)
|
_pendingException = exc1;
|
}
|
}
|
}
|
|
|
|
|
private bool DeflateOneSegment(WorkItem workitem)
|
{
|
ZlibCodec compressor = workitem.compressor;
|
int rc= 0;
|
compressor.ResetDeflate();
|
compressor.NextIn = 0;
|
|
compressor.AvailableBytesIn = workitem.inputBytesAvailable;
|
|
// step 1: deflate the buffer
|
compressor.NextOut = 0;
|
compressor.AvailableBytesOut = workitem.compressed.Length;
|
do
|
{
|
compressor.Deflate(FlushType.None);
|
}
|
while (compressor.AvailableBytesIn > 0 || compressor.AvailableBytesOut == 0);
|
|
// step 2: flush (sync)
|
rc = compressor.Deflate(FlushType.Sync);
|
|
workitem.compressedBytesAvailable= (int) compressor.TotalBytesOut;
|
return true;
|
}
|
|
|
[System.Diagnostics.ConditionalAttribute("Trace")]
|
private void TraceOutput(TraceBits bits, string format, params object[] varParams)
|
{
|
if ((bits & _DesiredTrace) != 0)
|
{
|
lock(_outputLock)
|
{
|
int tid = Thread.CurrentThread.GetHashCode();
|
#if !SILVERLIGHT
|
Console.ForegroundColor = (ConsoleColor) (tid % 8 + 8);
|
#endif
|
Console.Write("{0:000} PDOS ", tid);
|
Console.WriteLine(format, varParams);
|
#if !SILVERLIGHT
|
Console.ResetColor();
|
#endif
|
}
|
}
|
}
|
|
|
// used only when Trace is defined
|
[Flags]
|
enum TraceBits : uint
|
{
|
None = 0,
|
NotUsed1 = 1,
|
EmitLock = 2,
|
EmitEnter = 4, // enter _EmitPending
|
EmitBegin = 8, // begin to write out
|
EmitDone = 16, // done writing out
|
EmitSkip = 32, // writer skipping a workitem
|
EmitAll = 58, // All Emit flags
|
Flush = 64,
|
Lifecycle = 128, // constructor/disposer
|
Session = 256, // Close/Reset
|
Synch = 512, // thread synchronization
|
Instance = 1024, // instance settings
|
Compress = 2048, // compress task
|
Write = 4096, // filling buffers, when caller invokes Write()
|
WriteEnter = 8192, // upon entry to Write()
|
WriteTake = 16384, // on _toFill.Take()
|
All = 0xffffffff,
|
}
|
|
|
|
/// <summary>
|
/// Indicates whether the stream supports Seek operations.
|
/// </summary>
|
/// <remarks>
|
/// Always returns false.
|
/// </remarks>
|
public override bool CanSeek
|
{
|
get { return false; }
|
}
|
|
|
/// <summary>
|
/// Indicates whether the stream supports Read operations.
|
/// </summary>
|
/// <remarks>
|
/// Always returns false.
|
/// </remarks>
|
public override bool CanRead
|
{
|
get {return false;}
|
}
|
|
/// <summary>
|
/// Indicates whether the stream supports Write operations.
|
/// </summary>
|
/// <remarks>
|
/// Returns true if the provided stream is writable.
|
/// </remarks>
|
public override bool CanWrite
|
{
|
get { return _outStream.CanWrite; }
|
}
|
|
/// <summary>
|
/// Reading this property always throws a NotSupportedException.
|
/// </summary>
|
public override long Length
|
{
|
get { throw new NotSupportedException(); }
|
}
|
|
/// <summary>
|
/// Returns the current position of the output stream.
|
/// </summary>
|
/// <remarks>
|
/// <para>
|
/// Because the output gets written by a background thread,
|
/// the value may change asynchronously. Setting this
|
/// property always throws a NotSupportedException.
|
/// </para>
|
/// </remarks>
|
public override long Position
|
{
|
get { return _outStream.Position; }
|
set { throw new NotSupportedException(); }
|
}
|
|
/// <summary>
|
/// This method always throws a NotSupportedException.
|
/// </summary>
|
/// <param name="buffer">
|
/// The buffer into which data would be read, IF THIS METHOD
|
/// ACTUALLY DID ANYTHING.
|
/// </param>
|
/// <param name="offset">
|
/// The offset within that data array at which to insert the
|
/// data that is read, IF THIS METHOD ACTUALLY DID
|
/// ANYTHING.
|
/// </param>
|
/// <param name="count">
|
/// The number of bytes to write, IF THIS METHOD ACTUALLY DID
|
/// ANYTHING.
|
/// </param>
|
/// <returns>nothing.</returns>
|
public override int Read(byte[] buffer, int offset, int count)
|
{
|
throw new NotSupportedException();
|
}
|
|
/// <summary>
|
/// This method always throws a NotSupportedException.
|
/// </summary>
|
/// <param name="offset">
|
/// The offset to seek to....
|
/// IF THIS METHOD ACTUALLY DID ANYTHING.
|
/// </param>
|
/// <param name="origin">
|
/// The reference specifying how to apply the offset.... IF
|
/// THIS METHOD ACTUALLY DID ANYTHING.
|
/// </param>
|
/// <returns>nothing. It always throws.</returns>
|
public override long Seek(long offset, System.IO.SeekOrigin origin)
|
{
|
throw new NotSupportedException();
|
}
|
|
/// <summary>
|
/// This method always throws a NotSupportedException.
|
/// </summary>
|
/// <param name="value">
|
/// The new value for the stream length.... IF
|
/// THIS METHOD ACTUALLY DID ANYTHING.
|
/// </param>
|
public override void SetLength(long value)
|
{
|
throw new NotSupportedException();
|
}
|
|
}
|
|
}
|
