diff options
Diffstat (limited to 'CPP/7zip/Archive/Common/CoderMixer2MT.cpp')
| -rwxr-xr-x | CPP/7zip/Archive/Common/CoderMixer2MT.cpp | 257 |
1 files changed, 68 insertions, 189 deletions
diff --git a/CPP/7zip/Archive/Common/CoderMixer2MT.cpp b/CPP/7zip/Archive/Common/CoderMixer2MT.cpp index 8a37a10d..bff689dd 100755 --- a/CPP/7zip/Archive/Common/CoderMixer2MT.cpp +++ b/CPP/7zip/Archive/Common/CoderMixer2MT.cpp @@ -3,16 +3,11 @@ #include "StdAfx.h" #include "CoderMixer2MT.h" -#include "CrossThreadProgress.h" -using namespace NWindows; -using namespace NSynchronization; +namespace NCoderMixer { -namespace NCoderMixer2 { - -CThreadCoderInfo::CThreadCoderInfo(UInt32 numInStreams, UInt32 numOutStreams): - ExitEvent(NULL), - CCoderInfo(numInStreams, numOutStreams) +CCoder2::CCoder2(UInt32 numInStreams, UInt32 numOutStreams): + CCoderInfo2(numInStreams, numOutStreams) { InStreams.Reserve(NumInStreams); InStreamPointers.Reserve(NumInStreams); @@ -20,62 +15,38 @@ CThreadCoderInfo::CThreadCoderInfo(UInt32 numInStreams, UInt32 numOutStreams): OutStreamPointers.Reserve(NumOutStreams); } -class CCoderInfoFlusher2 +void CCoder2::Execute() { Code(NULL); } + +void CCoder2::Code(ICompressProgressInfo *progress) { - CThreadCoderInfo *m_CoderInfo; -public: - CCoderInfoFlusher2(CThreadCoderInfo *coderInfo): m_CoderInfo(coderInfo) {} - ~CCoderInfoFlusher2() + InStreamPointers.Clear(); + OutStreamPointers.Clear(); + UInt32 i; + for (i = 0; i < NumInStreams; i++) { - int i; - for (i = 0; i < m_CoderInfo->InStreams.Size(); i++) - m_CoderInfo->InStreams[i].Release(); - for (i = 0; i < m_CoderInfo->OutStreams.Size(); i++) - m_CoderInfo->OutStreams[i].Release(); - m_CoderInfo->CompressionCompletedEvent.Set(); + if (InSizePointers[i] != NULL) + InSizePointers[i] = &InSizes[i]; + InStreamPointers.Add(InStreams[i]); } -}; - -bool CThreadCoderInfo::WaitAndCode() -{ - HANDLE events[2] = { ExitEvent, CompressEvent }; - DWORD waitResult = ::WaitForMultipleObjects(2, events, FALSE, INFINITE); - if (waitResult == WAIT_OBJECT_0 + 0) - return false; - + for (i = 0; i < NumOutStreams; i++) { - InStreamPointers.Clear(); - OutStreamPointers.Clear(); - UInt32 i; - for (i = 0; i < NumInStreams; i++) - { - if (InSizePointers[i] != NULL) - InSizePointers[i] = &InSizes[i]; - InStreamPointers.Add(InStreams[i]); - } - for (i = 0; i < NumOutStreams; i++) - { - if (OutSizePointers[i] != NULL) - OutSizePointers[i] = &OutSizes[i]; - OutStreamPointers.Add(OutStreams[i]); - } - CCoderInfoFlusher2 coderInfoFlusher(this); - if (Coder) - Result = Coder->Code(InStreamPointers[0], - OutStreamPointers[0], - InSizePointers[0], - OutSizePointers[0], - Progress); - else - Result = Coder2->Code(&InStreamPointers.Front(), - &InSizePointers.Front(), - NumInStreams, - &OutStreamPointers.Front(), - &OutSizePointers.Front(), - NumOutStreams, - Progress); + if (OutSizePointers[i] != NULL) + OutSizePointers[i] = &OutSizes[i]; + OutStreamPointers.Add(OutStreams[i]); + } + if (Coder) + Result = Coder->Code(InStreamPointers[0], OutStreamPointers[0], + InSizePointers[0], OutSizePointers[0], progress); + else + Result = Coder2->Code(&InStreamPointers.Front(), &InSizePointers.Front(), NumInStreams, + &OutStreamPointers.Front(), &OutSizePointers.Front(), NumOutStreams, progress); + { + int i; + for (i = 0; i < InStreams.Size(); i++) + InStreams[i].Release(); + for (i = 0; i < OutStreams.Size(); i++) + OutStreams[i].Release(); } - return true; } static void SetSizes(const UInt64 **srcSizes, CRecordVector<UInt64> &sizes, @@ -99,54 +70,15 @@ static void SetSizes(const UInt64 **srcSizes, CRecordVector<UInt64> &sizes, } -void CThreadCoderInfo::SetCoderInfo(const UInt64 **inSizes, - const UInt64 **outSizes, ICompressProgressInfo *progress) +void CCoder2::SetCoderInfo(const UInt64 **inSizes, const UInt64 **outSizes) { - Progress = progress; SetSizes(inSizes, InSizes, InSizePointers, NumInStreams); SetSizes(outSizes, OutSizes, OutSizePointers, NumOutStreams); } -static THREAD_FUNC_DECL CoderThread(void *threadCoderInfo) -{ - for (;;) - { - if (!((CThreadCoderInfo *)threadCoderInfo)->WaitAndCode()) - return 0; - } -} - ////////////////////////////////////// // CCoderMixer2MT -static THREAD_FUNC_DECL MainCoderThread(void *threadCoderInfo) -{ - for (;;) - { - if (!((CCoderMixer2MT *)threadCoderInfo)->MyCode()) - return 0; - } -} - -CCoderMixer2MT::CCoderMixer2MT() -{ - if (CreateEvents() != S_OK) - throw 271824; - if (_mainThread.Create(MainCoderThread, this) != S_OK) - throw 271825; -} - -CCoderMixer2MT::~CCoderMixer2MT() -{ - _exitEvent.Set(); - _mainThread.Wait(); - for(int i = 0; i < _threads.Size(); i++) - { - _threads[i].Wait(); - _threads[i].Close(); - } -} - HRESULT CCoderMixer2MT::SetBindInfo(const CBindInfo &bindInfo) { _bindInfo = bindInfo; @@ -161,48 +93,23 @@ HRESULT CCoderMixer2MT::SetBindInfo(const CBindInfo &bindInfo) void CCoderMixer2MT::AddCoderCommon() { - int index = _coderInfoVector.Size(); - const CCoderStreamsInfo &CoderStreamsInfo = _bindInfo.Coders[index]; - - CThreadCoderInfo threadCoderInfo(CoderStreamsInfo.NumInStreams, - CoderStreamsInfo.NumOutStreams); - _coderInfoVector.Add(threadCoderInfo); - CThreadCoderInfo *tci = &_coderInfoVector.Back(); - tci->CreateEvents(); - tci->ExitEvent = _exitEvent; - - NWindows::CThread newThread; - _threads.Add(newThread); - if (_threads.Back().Create(CoderThread, tci) != S_OK) - throw 271824; + const CCoderStreamsInfo &c = _bindInfo.Coders[_coders.Size()]; + CCoder2 threadCoderInfo(c.NumInStreams, c.NumOutStreams); + _coders.Add(threadCoderInfo); } void CCoderMixer2MT::AddCoder(ICompressCoder *coder) { AddCoderCommon(); - _coderInfoVector.Back().Coder = coder; + _coders.Back().Coder = coder; } void CCoderMixer2MT::AddCoder2(ICompressCoder2 *coder) { AddCoderCommon(); - _coderInfoVector.Back().Coder2 = coder; + _coders.Back().Coder2 = coder; } -/* -void CCoderMixer2MT::FinishAddingCoders() -{ - for(int i = 0; i < _coderInfoVector.Size(); i++) - { - DWORD id; - HANDLE newThread = ::CreateThread(NULL, 0, CoderThread, - &_coderInfoVector[i], 0, &id); - if (newThread == 0) - throw 271824; - _threads.Add(newThread); - } -} -*/ void CCoderMixer2MT::ReInit() { @@ -211,17 +118,16 @@ void CCoderMixer2MT::ReInit() } -STDMETHODIMP CCoderMixer2MT::Init(ISequentialInStream **inStreams, - ISequentialOutStream **outStreams) +HRESULT CCoderMixer2MT::Init(ISequentialInStream **inStreams, ISequentialOutStream **outStreams) { /* - if (_coderInfoVector.Size() != _bindInfo.Coders.Size()) + if (_coders.Size() != _bindInfo.Coders.Size()) throw 0; */ int i; - for(i = 0; i < _coderInfoVector.Size(); i++) + for(i = 0; i < _coders.Size(); i++) { - CThreadCoderInfo &coderInfo = _coderInfoVector[i]; + CCoder2 &coderInfo = _coders[i]; const CCoderStreamsInfo &coderStreamsInfo = _bindInfo.Coders[i]; coderInfo.InStreams.Clear(); UInt32 j; @@ -241,45 +147,26 @@ STDMETHODIMP CCoderMixer2MT::Init(ISequentialInStream **inStreams, _bindInfo.FindOutStream(bindPair.OutIndex, outCoderIndex, outCoderStreamIndex); _streamBinders[i].CreateStreams( - &_coderInfoVector[inCoderIndex].InStreams[inCoderStreamIndex], - &_coderInfoVector[outCoderIndex].OutStreams[outCoderStreamIndex]); + &_coders[inCoderIndex].InStreams[inCoderStreamIndex], + &_coders[outCoderIndex].OutStreams[outCoderStreamIndex]); } for(i = 0; i < _bindInfo.InStreams.Size(); i++) { UInt32 inCoderIndex, inCoderStreamIndex; _bindInfo.FindInStream(_bindInfo.InStreams[i], inCoderIndex, inCoderStreamIndex); - _coderInfoVector[inCoderIndex].InStreams[inCoderStreamIndex] = inStreams[i]; + _coders[inCoderIndex].InStreams[inCoderStreamIndex] = inStreams[i]; } for(i = 0; i < _bindInfo.OutStreams.Size(); i++) { UInt32 outCoderIndex, outCoderStreamIndex; _bindInfo.FindOutStream(_bindInfo.OutStreams[i], outCoderIndex, outCoderStreamIndex); - _coderInfoVector[outCoderIndex].OutStreams[outCoderStreamIndex] = outStreams[i]; + _coders[outCoderIndex].OutStreams[outCoderStreamIndex] = outStreams[i]; } return S_OK; } - -bool CCoderMixer2MT::MyCode() -{ - HANDLE events[2] = { _exitEvent, _startCompressingEvent }; - DWORD waitResult = ::WaitForMultipleObjects(2, events, FALSE, INFINITE); - if (waitResult == WAIT_OBJECT_0 + 0) - return false; - - int i; - for(i = 0; i < _coderInfoVector.Size(); i++) - _coderInfoVector[i].CompressEvent.Set(); - for (i = 0; i < _coderInfoVector.Size(); i++) - _coderInfoVector[i].CompressionCompletedEvent.Lock(); - - _compressingFinishedEvent.Set(); - return true; -} - - STDMETHODIMP CCoderMixer2MT::Code(ISequentialInStream **inStreams, const UInt64 ** /* inSizes */, UInt32 numInStreams, @@ -294,56 +181,48 @@ STDMETHODIMP CCoderMixer2MT::Code(ISequentialInStream **inStreams, Init(inStreams, outStreams); - _compressingFinishedEvent.Reset(); // ? - - CCrossThreadProgress *progressSpec = new CCrossThreadProgress; - CMyComPtr<ICompressProgressInfo> crossProgress = progressSpec; - RINOK(progressSpec->Create()); - progressSpec->Init(); - _coderInfoVector[_progressCoderIndex].Progress = crossProgress; + int i; + for (i = 0; i < _coders.Size(); i++) + if (i != _progressCoderIndex) + { + RINOK(_coders[i].Create()); + } + + for (i = 0; i < _coders.Size(); i++) + if (i != _progressCoderIndex) + _coders[i].Start(); - _startCompressingEvent.Set(); + _coders[_progressCoderIndex].Code(progress); + for (i = 0; i < _coders.Size(); i++) + if (i != _progressCoderIndex) + _coders[i].WaitFinish(); - for (;;) + for (i = 0; i < _coders.Size(); i++) { - HANDLE events[2] = {_compressingFinishedEvent, progressSpec->ProgressEvent }; - DWORD waitResult = ::WaitForMultipleObjects(2, events, FALSE, INFINITE); - if (waitResult == WAIT_OBJECT_0 + 0) - break; - if (progress != NULL) - progressSpec->Result = progress->SetRatioInfo(progressSpec->InSize, - progressSpec->OutSize); - else - progressSpec->Result = S_OK; - progressSpec->WaitEvent.Set(); + HRESULT result = _coders[i].Result; + if (result == E_ABORT) + return result; } - - int i; - for(i = 0; i < _coderInfoVector.Size(); i++) + for (i = 0; i < _coders.Size(); i++) { - HRESULT result = _coderInfoVector[i].Result; + HRESULT result = _coders[i].Result; if (result == S_FALSE) return result; } - for(i = 0; i < _coderInfoVector.Size(); i++) + for (i = 0; i < _coders.Size(); i++) { - HRESULT result = _coderInfoVector[i].Result; + HRESULT result = _coders[i].Result; if (result != S_OK && result != E_FAIL) return result; } - for(i = 0; i < _coderInfoVector.Size(); i++) + for (i = 0; i < _coders.Size(); i++) { - HRESULT result = _coderInfoVector[i].Result; + HRESULT result = _coders[i].Result; if (result != S_OK) return result; } return S_OK; } -UInt64 CCoderMixer2MT::GetWriteProcessedSize(UInt32 binderIndex) const -{ - return _streamBinders[binderIndex].ProcessedSize; -} - } |