1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/other-licenses/7zstub/src/7zip/Archive/Common/CoderMixer2MT.cpp Wed Dec 31 06:09:35 2014 +0100
1.3 @@ -0,0 +1,359 @@
1.4 +// CoderMixer2MT.cpp
1.5 +
1.6 +#include "StdAfx.h"
1.7 +
1.8 +#include "CoderMixer2MT.h"
1.9 +#include "CrossThreadProgress.h"
1.10 +
1.11 +using namespace NWindows;
1.12 +using namespace NSynchronization;
1.13 +
1.14 +namespace NCoderMixer2 {
1.15 +
1.16 +CThreadCoderInfo::CThreadCoderInfo(UInt32 numInStreams, UInt32 numOutStreams):
1.17 + ExitEvent(NULL),
1.18 + CompressEvent(NULL),
1.19 + CompressionCompletedEvent(NULL),
1.20 + CCoderInfo(numInStreams, numOutStreams)
1.21 +{
1.22 + InStreams.Reserve(NumInStreams);
1.23 + InStreamPointers.Reserve(NumInStreams);
1.24 + OutStreams.Reserve(NumOutStreams);
1.25 + OutStreamPointers.Reserve(NumOutStreams);
1.26 +}
1.27 +
1.28 +void CThreadCoderInfo::CreateEvents()
1.29 +{
1.30 + CompressEvent = new CAutoResetEvent(false);
1.31 + CompressionCompletedEvent = new CAutoResetEvent(false);
1.32 +}
1.33 +
1.34 +CThreadCoderInfo::~CThreadCoderInfo()
1.35 +{
1.36 + if (CompressEvent != NULL)
1.37 + delete CompressEvent;
1.38 + if (CompressionCompletedEvent != NULL)
1.39 + delete CompressionCompletedEvent;
1.40 +}
1.41 +
1.42 +class CCoderInfoFlusher2
1.43 +{
1.44 + CThreadCoderInfo *m_CoderInfo;
1.45 +public:
1.46 + CCoderInfoFlusher2(CThreadCoderInfo *coderInfo): m_CoderInfo(coderInfo) {}
1.47 + ~CCoderInfoFlusher2()
1.48 + {
1.49 + int i;
1.50 + for (i = 0; i < m_CoderInfo->InStreams.Size(); i++)
1.51 + m_CoderInfo->InStreams[i].Release();
1.52 + for (i = 0; i < m_CoderInfo->OutStreams.Size(); i++)
1.53 + m_CoderInfo->OutStreams[i].Release();
1.54 + m_CoderInfo->CompressionCompletedEvent->Set();
1.55 + }
1.56 +};
1.57 +
1.58 +bool CThreadCoderInfo::WaitAndCode()
1.59 +{
1.60 + HANDLE events[2] = { ExitEvent, *CompressEvent };
1.61 + DWORD waitResult = ::WaitForMultipleObjects(2, events, FALSE, INFINITE);
1.62 + if (waitResult == WAIT_OBJECT_0 + 0)
1.63 + return false;
1.64 +
1.65 + {
1.66 + InStreamPointers.Clear();
1.67 + OutStreamPointers.Clear();
1.68 + UInt32 i;
1.69 + for (i = 0; i < NumInStreams; i++)
1.70 + {
1.71 + if (InSizePointers[i] != NULL)
1.72 + InSizePointers[i] = &InSizes[i];
1.73 + InStreamPointers.Add(InStreams[i]);
1.74 + }
1.75 + for (i = 0; i < NumOutStreams; i++)
1.76 + {
1.77 + if (OutSizePointers[i] != NULL)
1.78 + OutSizePointers[i] = &OutSizes[i];
1.79 + OutStreamPointers.Add(OutStreams[i]);
1.80 + }
1.81 + CCoderInfoFlusher2 coderInfoFlusher(this);
1.82 + if (Coder)
1.83 + Result = Coder->Code(InStreamPointers[0],
1.84 + OutStreamPointers[0],
1.85 + InSizePointers[0],
1.86 + OutSizePointers[0],
1.87 + Progress);
1.88 + else
1.89 + Result = Coder2->Code(&InStreamPointers.Front(),
1.90 + &InSizePointers.Front(),
1.91 + NumInStreams,
1.92 + &OutStreamPointers.Front(),
1.93 + &OutSizePointers.Front(),
1.94 + NumOutStreams,
1.95 + Progress);
1.96 + }
1.97 + return true;
1.98 +}
1.99 +
1.100 +static void SetSizes(const UInt64 **srcSizes, CRecordVector<UInt64> &sizes,
1.101 + CRecordVector<const UInt64 *> &sizePointers, UInt32 numItems)
1.102 +{
1.103 + sizes.Clear();
1.104 + sizePointers.Clear();
1.105 + for(UInt32 i = 0; i < numItems; i++)
1.106 + {
1.107 + if (srcSizes == 0 || srcSizes[i] == NULL)
1.108 + {
1.109 + sizes.Add(0);
1.110 + sizePointers.Add(NULL);
1.111 + }
1.112 + else
1.113 + {
1.114 + sizes.Add(*srcSizes[i]);
1.115 + sizePointers.Add(&sizes.Back());
1.116 + }
1.117 + }
1.118 +}
1.119 +
1.120 +
1.121 +void CThreadCoderInfo::SetCoderInfo(const UInt64 **inSizes,
1.122 + const UInt64 **outSizes, ICompressProgressInfo *progress)
1.123 +{
1.124 + Progress = progress;
1.125 + SetSizes(inSizes, InSizes, InSizePointers, NumInStreams);
1.126 + SetSizes(outSizes, OutSizes, OutSizePointers, NumOutStreams);
1.127 +}
1.128 +
1.129 +static DWORD WINAPI CoderThread(void *threadCoderInfo)
1.130 +{
1.131 + while(true)
1.132 + {
1.133 + if (!((CThreadCoderInfo *)threadCoderInfo)->WaitAndCode())
1.134 + return 0;
1.135 + }
1.136 +}
1.137 +
1.138 +//////////////////////////////////////
1.139 +// CCoderMixer2MT
1.140 +
1.141 +static DWORD WINAPI MainCoderThread(void *threadCoderInfo)
1.142 +{
1.143 + while(true)
1.144 + {
1.145 + if (!((CCoderMixer2MT *)threadCoderInfo)->MyCode())
1.146 + return 0;
1.147 + }
1.148 +}
1.149 +
1.150 +CCoderMixer2MT::CCoderMixer2MT()
1.151 +{
1.152 + if (!_mainThread.Create(MainCoderThread, this))
1.153 + throw 271825;
1.154 +}
1.155 +
1.156 +CCoderMixer2MT::~CCoderMixer2MT()
1.157 +{
1.158 + _exitEvent.Set();
1.159 + _mainThread.Wait();
1.160 + for(int i = 0; i < _threads.Size(); i++)
1.161 + {
1.162 + _threads[i].Wait();
1.163 + _threads[i].Close();
1.164 + }
1.165 +}
1.166 +
1.167 +void CCoderMixer2MT::SetBindInfo(const CBindInfo &bindInfo)
1.168 +{
1.169 + _bindInfo = bindInfo;
1.170 + _streamBinders.Clear();
1.171 + for(int i = 0; i < _bindInfo.BindPairs.Size(); i++)
1.172 + {
1.173 + _streamBinders.Add(CStreamBinder());
1.174 + _streamBinders.Back().CreateEvents();
1.175 + }
1.176 +}
1.177 +
1.178 +void CCoderMixer2MT::AddCoderCommon()
1.179 +{
1.180 + int index = _coderInfoVector.Size();
1.181 + const CCoderStreamsInfo &CoderStreamsInfo = _bindInfo.Coders[index];
1.182 +
1.183 + CThreadCoderInfo threadCoderInfo(CoderStreamsInfo.NumInStreams,
1.184 + CoderStreamsInfo.NumOutStreams);
1.185 + _coderInfoVector.Add(threadCoderInfo);
1.186 + _coderInfoVector.Back().CreateEvents();
1.187 + _coderInfoVector.Back().ExitEvent = _exitEvent;
1.188 + _compressingCompletedEvents.Add(*_coderInfoVector.Back().CompressionCompletedEvent);
1.189 +
1.190 + NWindows::CThread newThread;
1.191 + _threads.Add(newThread);
1.192 + if (!_threads.Back().Create(CoderThread, &_coderInfoVector.Back()))
1.193 + throw 271824;
1.194 +}
1.195 +
1.196 +void CCoderMixer2MT::AddCoder(ICompressCoder *coder)
1.197 +{
1.198 + AddCoderCommon();
1.199 + _coderInfoVector.Back().Coder = coder;
1.200 +}
1.201 +
1.202 +void CCoderMixer2MT::AddCoder2(ICompressCoder2 *coder)
1.203 +{
1.204 + AddCoderCommon();
1.205 + _coderInfoVector.Back().Coder2 = coder;
1.206 +}
1.207 +
1.208 +/*
1.209 +void CCoderMixer2MT::FinishAddingCoders()
1.210 +{
1.211 + for(int i = 0; i < _coderInfoVector.Size(); i++)
1.212 + {
1.213 + DWORD id;
1.214 + HANDLE newThread = ::CreateThread(NULL, 0, CoderThread,
1.215 + &_coderInfoVector[i], 0, &id);
1.216 + if (newThread == 0)
1.217 + throw 271824;
1.218 + _threads.Add(newThread);
1.219 + }
1.220 +}
1.221 +*/
1.222 +
1.223 +void CCoderMixer2MT::ReInit()
1.224 +{
1.225 + for(int i = 0; i < _streamBinders.Size(); i++)
1.226 + _streamBinders[i].ReInit();
1.227 +}
1.228 +
1.229 +
1.230 +STDMETHODIMP CCoderMixer2MT::Init(ISequentialInStream **inStreams,
1.231 + ISequentialOutStream **outStreams)
1.232 +{
1.233 + if (_coderInfoVector.Size() != _bindInfo.Coders.Size())
1.234 + throw 0;
1.235 + int i;
1.236 + for(i = 0; i < _coderInfoVector.Size(); i++)
1.237 + {
1.238 + CThreadCoderInfo &coderInfo = _coderInfoVector[i];
1.239 + const CCoderStreamsInfo &coderStreamsInfo = _bindInfo.Coders[i];
1.240 + coderInfo.InStreams.Clear();
1.241 + UInt32 j;
1.242 + for(j = 0; j < coderStreamsInfo.NumInStreams; j++)
1.243 + coderInfo.InStreams.Add(NULL);
1.244 + coderInfo.OutStreams.Clear();
1.245 + for(j = 0; j < coderStreamsInfo.NumOutStreams; j++)
1.246 + coderInfo.OutStreams.Add(NULL);
1.247 + }
1.248 +
1.249 + for(i = 0; i < _bindInfo.BindPairs.Size(); i++)
1.250 + {
1.251 + const CBindPair &bindPair = _bindInfo.BindPairs[i];
1.252 + UInt32 inCoderIndex, inCoderStreamIndex;
1.253 + UInt32 outCoderIndex, outCoderStreamIndex;
1.254 + _bindInfo.FindInStream(bindPair.InIndex, inCoderIndex, inCoderStreamIndex);
1.255 + _bindInfo.FindOutStream(bindPair.OutIndex, outCoderIndex, outCoderStreamIndex);
1.256 +
1.257 + _streamBinders[i].CreateStreams(
1.258 + &_coderInfoVector[inCoderIndex].InStreams[inCoderStreamIndex],
1.259 + &_coderInfoVector[outCoderIndex].OutStreams[outCoderStreamIndex]);
1.260 + }
1.261 +
1.262 + for(i = 0; i < _bindInfo.InStreams.Size(); i++)
1.263 + {
1.264 + UInt32 inCoderIndex, inCoderStreamIndex;
1.265 + _bindInfo.FindInStream(_bindInfo.InStreams[i], inCoderIndex, inCoderStreamIndex);
1.266 + _coderInfoVector[inCoderIndex].InStreams[inCoderStreamIndex] = inStreams[i];
1.267 + }
1.268 +
1.269 + for(i = 0; i < _bindInfo.OutStreams.Size(); i++)
1.270 + {
1.271 + UInt32 outCoderIndex, outCoderStreamIndex;
1.272 + _bindInfo.FindOutStream(_bindInfo.OutStreams[i], outCoderIndex, outCoderStreamIndex);
1.273 + _coderInfoVector[outCoderIndex].OutStreams[outCoderStreamIndex] = outStreams[i];
1.274 + }
1.275 + return S_OK;
1.276 +}
1.277 +
1.278 +
1.279 +bool CCoderMixer2MT::MyCode()
1.280 +{
1.281 + HANDLE events[2] = { _exitEvent, _startCompressingEvent };
1.282 + DWORD waitResult = ::WaitForMultipleObjects(2, events, FALSE, INFINITE);
1.283 + if (waitResult == WAIT_OBJECT_0 + 0)
1.284 + return false;
1.285 +
1.286 + for(int i = 0; i < _coderInfoVector.Size(); i++)
1.287 + _coderInfoVector[i].CompressEvent->Set();
1.288 + DWORD result = ::WaitForMultipleObjects(_compressingCompletedEvents.Size(),
1.289 + &_compressingCompletedEvents.Front(), TRUE, INFINITE);
1.290 +
1.291 + _compressingFinishedEvent.Set();
1.292 +
1.293 + return true;
1.294 +}
1.295 +
1.296 +
1.297 +STDMETHODIMP CCoderMixer2MT::Code(ISequentialInStream **inStreams,
1.298 + const UInt64 **inSizes,
1.299 + UInt32 numInStreams,
1.300 + ISequentialOutStream **outStreams,
1.301 + const UInt64 **outSizes,
1.302 + UInt32 numOutStreams,
1.303 + ICompressProgressInfo *progress)
1.304 +{
1.305 + if (numInStreams != (UInt32)_bindInfo.InStreams.Size() ||
1.306 + numOutStreams != (UInt32)_bindInfo.OutStreams.Size())
1.307 + return E_INVALIDARG;
1.308 +
1.309 + Init(inStreams, outStreams);
1.310 +
1.311 + _compressingFinishedEvent.Reset(); // ?
1.312 +
1.313 + CCrossThreadProgress *progressSpec = new CCrossThreadProgress;
1.314 + CMyComPtr<ICompressProgressInfo> crossProgress = progressSpec;
1.315 + progressSpec->Init();
1.316 + _coderInfoVector[_progressCoderIndex].Progress = crossProgress;
1.317 +
1.318 + _startCompressingEvent.Set();
1.319 +
1.320 +
1.321 + while (true)
1.322 + {
1.323 + HANDLE events[2] = {_compressingFinishedEvent, progressSpec->ProgressEvent };
1.324 + DWORD waitResult = ::WaitForMultipleObjects(2, events, FALSE, INFINITE);
1.325 + if (waitResult == WAIT_OBJECT_0 + 0)
1.326 + break;
1.327 + if (progress != NULL)
1.328 + progressSpec->Result = progress->SetRatioInfo(progressSpec->InSize,
1.329 + progressSpec->OutSize);
1.330 + else
1.331 + progressSpec->Result = S_OK;
1.332 + progressSpec->WaitEvent.Set();
1.333 + }
1.334 +
1.335 + int i;
1.336 + for(i = 0; i < _coderInfoVector.Size(); i++)
1.337 + {
1.338 + HRESULT result = _coderInfoVector[i].Result;
1.339 + if (result == S_FALSE)
1.340 + return result;
1.341 + }
1.342 + for(i = 0; i < _coderInfoVector.Size(); i++)
1.343 + {
1.344 + HRESULT result = _coderInfoVector[i].Result;
1.345 + if (result != S_OK && result != E_FAIL)
1.346 + return result;
1.347 + }
1.348 + for(i = 0; i < _coderInfoVector.Size(); i++)
1.349 + {
1.350 + HRESULT result = _coderInfoVector[i].Result;
1.351 + if (result != S_OK)
1.352 + return result;
1.353 + }
1.354 + return S_OK;
1.355 +}
1.356 +
1.357 +UInt64 CCoderMixer2MT::GetWriteProcessedSize(UInt32 binderIndex) const
1.358 +{
1.359 + return _streamBinders[binderIndex].ProcessedSize;
1.360 +}
1.361 +
1.362 +}
