text: /*
      RawSpeed - RAW file Decompressor.
  
      Copyright (C) 2017 Roman Lebedev
  
      This library is free software; you can redistribute it and/or
      modify it under the terms of the GNU Lesser General Public
      License as published by the Free Software Foundation; either
      version 2 of the License, or (at your option) any later version.
  
      This library is distributed in the hope that it will be useful,
      but WITHOUT ANY WARRANTY; without even the implied warranty of
      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
      Lesser General Public License for more details.
  
      You should have received a copy of the GNU Lesser General Public
      License along with this library; if not, write to the Free Software
      Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
  
  #include "rawspeedconfig.h" // for HAVE_PTHREAD
  #include "decompressors/AbstractParallelizedDecompressor.h"
  #include "common/Point.h"                 // for iPoint2D
  #include "common/Threading.h"             // for sliceUp
  #include "decoders/RawDecoderException.h" // for ThrowRDE
  #include <cassert>                        // for assert
  #include <memory>                         // for allocator_traits<>::value_...
  #include <string>                         // for string
  #include <vector>                         // for vector
  
  namespace rawspeed {
  
  void AbstractParallelizedDecompressor::decompressOne(uint32 pieces) const {
    RawDecompressorThread t(this, 1);
    t.taskNo = 0;
    t.start = 0;
    t.end = pieces;
  
    RawDecompressorThread::start_routine(&t);
  
    std::string firstErr;
    if (mRaw->isTooManyErrors(1, &firstErr)) {
      ThrowRDE("Too many errors encountered. Giving up. First Error:\n%s",
               firstErr.c_str());
    }
  };
  
  #ifdef HAVE_PTHREAD
  void AbstractParallelizedDecompressor::startThreading(uint32 pieces) const {
    assert(pieces > 0);
    assert(getThreadCount() > 0);
    const auto buckets = sliceUp(getThreadCount(), pieces);
  
    if (buckets.size() == 1)
      return decompressOne(pieces);
  
    std::vector<RawDecompressorThread> threads(
        buckets.size(), RawDecompressorThread(this, buckets.size()));
  
    /* Initialize and set thread detached attribute */
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
  
    uint32 i = 0;
    bool fail = false;
    int offset = 0;
    for (auto& t : threads) {
      t.taskNo = i;
  
      t.start = offset;
      t.end = t.start + buckets[i];
  
      assert(t.start < pieces);
      assert(t.end > 0);
      assert(t.end > t.start);
      assert(t.end <= pieces);
  
  #ifndef NDEBUG
      if (i > 0) {
        assert(t.start > threads[i - 1].start);
        assert(t.end > threads[i - 1].end);
        assert(t.start == threads[i - 1].end);
      }
  #endif
  
      if (pthread_create(&t.threadid, &attr, RawDecompressorThread::start_routine,
                         &t) != 0) {
        // If a failure occurs, we need to wait for the already created threads to
        // finish
        fail = true;
        while (threads.size() > i)
          threads.pop_back();
        break;
      }
  
      offset = t.end;
      i++;
    }
  
    for (auto& t : threads)
      pthread_join(t.threadid, nullptr);
  
    pthread_attr_destroy(&attr);
  
    if (fail)
      ThrowRDE("Unable to start threads");
  
    std::string firstErr;
    if (mRaw->isTooManyErrors(1, &firstErr)) {
      ThrowRDE("Too many errors encountered. Giving up. First Error:\n%s",
               firstErr.c_str());
    }
  }
  #else
  void AbstractParallelizedDecompressor::startThreading(uint32 pieces) const {
    decompressOne(pieces);
  }
  #endif
  
  void AbstractParallelizedDecompressor::decode() const {
    startThreading(mRaw->dim.y);
  }
  
  } // namespace rawspeed
  