//#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 System.IO;  namespace 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,                        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="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="Ionic.Zlib.DeflateStream" />    internal 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 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 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="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 &amp; ".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 &lt;&gt; 0)        ///                 If (n &gt; 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="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 Ionic.Crc.CRC32();            _currentlyFilling = -1;            _lastFilled = -1;            _lastWritten = -1;            _latestCompressed = -1;        }            /// <summary>

Lines 1–500 of 1401. Load more