ckcalculator.cpp

Go to the documentation of this file.

/*
 * libxcks
 * Copyright (C) 2022 Julien Couot
 *
 * This program 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 3 of the License, or (at your
 * option) any later version.
 *
 * This program 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 program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
//---------------------------------------------------------------------------
#include <thread>
#include <mutex>
#include <shared_mutex>
#include <vector>
#include <algorithm>
 
#include "libxcks/ckcalculator.hpp"
#include "libxcks/checksum.hpp"
#include "libxcks/ckfactory.hpp"
//---------------------------------------------------------------------------
 
 
using namespace std;
 
 
namespace libxcks
{
using ArrayChecksum = vector<Checksum*>;
 
 
//###########################################################################
// A thread safe checksums calculator.
//###########################################################################
 
class ThreadSafeChecksumCalculator
{
 protected:
  ArrayChecksum checksums;     
  size_t curCk;                
  mutable shared_mutex mutex;  
 
 public:
  ThreadSafeChecksumCalculator() = delete;
 
  ThreadSafeChecksumCalculator(const ThreadSafeChecksumCalculator&) = delete;
 
  ThreadSafeChecksumCalculator& operator=(const ThreadSafeChecksumCalculator&) = delete;
 
  ThreadSafeChecksumCalculator(const ArrayChecksum& ckArray) : checksums(ckArray)
  {
    resetChecksumToProcess();
  }
 
  void resetChecksumToProcess()
  {
    unique_lock lock(mutex);  // shared_lock for read only
    curCk = 0u;
  }
 
  Checksum* nextChecksumToProcess()
  {
    unique_lock lock(mutex);
    if (curCk < checksums.size())
      return checksums[curCk++];
    else
      return nullptr;
  }
 
  static void updateChecksum(ThreadSafeChecksumCalculator* c, const uint8_t* buf, size_t len)
  {
    Checksum* chk;
 
    while ((chk = c->nextChecksumToProcess()) != nullptr)
      chk->update(buf, len);
  }
};
//---------------------------------------------------------------------------
 
 
 
 
//###########################################################################
// ChecksumCalculator members
//###########################################################################
 
// Default progress checksum calculator instance.
DefaultChecksumCalculatorProgress ChecksumCalculator::defaultChecksumCalculatorProgress;
 
 
/*
 * Constructor.
 */
ChecksumCalculator::ChecksumCalculator(ChecksumCalculatorProgress& progressHandler,
                                       const unsigned int newNbThreads,
                                       const size_t newBufferSize) :
                                                       progress(progressHandler)
{
  setBufferSize(newBufferSize);
  setNbThreads(newNbThreads);
}
//---------------------------------------------------------------------------
 
 
/*
 * Gets the size of the buffer to use for reading in the input stream.
 */
size_t ChecksumCalculator::getBufferSize() const
{
  return bufferSize;
}
//---------------------------------------------------------------------------
 
 
/*
 * Sets the size of the buffer to use for reading in the input stream.
 */
size_t ChecksumCalculator::setBufferSize(const size_t bufSize)
{
  size_t old = bufferSize;
  if ((bufSize >= min_buffer_size) && (bufSize <= max_buffer_size))
    bufferSize = bufSize;
  else
    bufferSize = default_buffer_size;
  return old;
}
//---------------------------------------------------------------------------
 
 
/*
 * Returns the number of threads to use to compute checksums.
 */
unsigned int ChecksumCalculator::getNbThreads() const
{
  return nbThreads;
}
//---------------------------------------------------------------------------
 
 
/*
 * Sets the number of threads to use to compute checksums.
 */
unsigned int ChecksumCalculator::setNbThreads(const unsigned int newNbThreads)
{
  unsigned int old = nbThreads;
 
  if (newNbThreads == automatic_thread_number)
  {
    unsigned int autoNbThreads = thread::hardware_concurrency();
    if (autoNbThreads == 0u)
      nbThreads = default_thread_number;
    else
    {
      const int ckfKnownAlgos = ChecksumFactory::getAlgorithmsCount();
      const unsigned int knownAlgos = (ckfKnownAlgos <= 0) ? 1u : ckfKnownAlgos;
      nbThreads = min(knownAlgos, autoNbThreads);
    }
  }
  else
  {
    const int ckfKnownAlgos = ChecksumFactory::getAlgorithmsCount();
    const unsigned int knownAlgos = (ckfKnownAlgos <= 0) ? 1u : ckfKnownAlgos;
    const unsigned int maxNbThreads = max(max_thread_number, knownAlgos);
    nbThreads = min(newNbThreads, maxNbThreads);
  }
 
  return old;
}
//---------------------------------------------------------------------------
 
 
/*
 * Gets the progress handler used to show the progression.
 */
ChecksumCalculatorProgress& ChecksumCalculator::getChecksumProgress() const
{
  return progress;
}
//---------------------------------------------------------------------------
 
 
/*
 * Calculates the checksum from the given stream.
 */
ChecksumCalculator::State ChecksumCalculator::calculate(ChecksumValue& sumValue,
                                                        const ChecksumAlgoId id,
                                                        std::istream& is)
{
  ArrayChecksumAlgoId ids;
  ids.push_back(id);
  ArrayChecksumValue values;
  State res = calculate(values, ids, is);
  if (!values.empty())
    sumValue = values.front();
  return res;
}
//---------------------------------------------------------------------------
 
 
/*
 * Calculates the checksums from the given stream.
 */
ChecksumCalculator::State ChecksumCalculator::calculate(ArrayChecksumValue& sumValues,
                                                        const ArrayChecksumAlgoId& ids,
                                                        std::istream& is)
{
  // Check if the input stream is valid.
  if (!is.good())
    return State::ReadError;
 
  // Retrieve algorithms for computing the sums.
  ArrayChecksum checksums;
  checksums.reserve(ids.size());
  for (const ChecksumAlgoId id : ids)
  {
    bool notFound = true;
    ArrayChecksum::const_iterator it = checksums.cbegin();
    while (notFound && it != checksums.cend())
    {
      if ((*it)->getID() == id)
        notFound = false;
      else
        it++;
    }
    if (notFound)
    {
      Checksum* c = ChecksumFactory::getNewInstance(id);
      if (c != nullptr)
        checksums.push_back(c);
    }
  }
 
  // Check if at least a checksum instance is OK.
  if (checksums.empty())
  // Nothing has been added in the checksums array, so don't clear values with
  // delete.
    return State::NoValidAlgoId;
 
  // Initializes the buffer.
  const size_t bufSize = getBufferSize();
  uint8_t* buff = new uint8_t[bufSize];
  if (buff == nullptr)
    return State::OutOfMemory;
 
  // Calculating the checksum.
  bool cancelled = false;
  size_t read;
  ios::iostate errorState = ios::goodbit;
  size_t i;
  const size_t s = checksums.size();
 
  // Initialising the thread with the number of available threads.
  unsigned int nThreads = getNbThreads();
  if (nThreads > static_cast<unsigned int>(s))
    nThreads = static_cast<unsigned int>(s);
  if (nThreads == 0u)
    nThreads = 1u;
 
  if (nThreads > 1u)
  {
    vector<thread> threads(nThreads);
    ThreadSafeChecksumCalculator tscc(checksums);
 
    while (!cancelled && !is.eof() && (errorState & ios::goodbit) == ios::goodbit)
    {
      read = is.read(reinterpret_cast<char*>(buff), bufSize).gcount();
      if (read > 0u && read <= bufSize)
      {
        for (i = 0u; i < nThreads; i++)
          threads[i] = thread(ThreadSafeChecksumCalculator::updateChecksum, &tscc, buff, read);
        for (i = 0u; i < nThreads; i++)
          threads[i].join();
        tscc.resetChecksumToProcess();
        getChecksumProgress().update(read, cancelled);
      }
      errorState = is.rdstate();
    }
  }
  else  // one algorithm, faster with no threads
  {
    while (!cancelled && !is.eof() && (errorState & ios::goodbit) == ios::goodbit)
    {
      read = is.read(reinterpret_cast<char*>(buff), bufSize).gcount();
      if (read > 0u && read <= bufSize)
      {
        for (i = 0u; i < s; i++)
          checksums[i]->update(buff, read);
        getChecksumProgress().update(read, cancelled);
      }
      errorState = is.rdstate();
    }
  }
 
  // Cleans-up the memory
  delete[] buff;
 
  if (cancelled)
  {
    for_each(checksums.begin(), checksums.end(), [](Checksum* c){ delete c; });
    return State::CancelledByUser;
  }
 
  if (((errorState & ios::goodbit) != ios::goodbit && (errorState & ios::eofbit) != ios::eofbit))
  {
    for_each(checksums.begin(), checksums.end(), [](Checksum* c){ delete c; });
    return State::ReadError;
  }
 
  // Clearing the array of calculated values of the checksums
  sumValues.clear();
 
  // Copying to values in the calculated values of the checksums array
  for (const Checksum* c : checksums)
  {
    sumValues.emplace_back(ChecksumValue(*c));
    delete c;
  }
  //for_each(checksums.begin(), checksums.end(), [](Checksum* c){ delete c; });  // done upper
 
  return State::Ok;
}
//---------------------------------------------------------------------------
 
 
 
 
//###########################################################################
// ChecksumFileCalculator members
//###########################################################################
 
 
/*
 * Constructor.
 */
ChecksumFileCalculator::ChecksumFileCalculator(ChecksumCalculatorProgress& progressHandler,
                                               const unsigned int newNbThreads,
                                               const size_t newBufferSize) :
                                                 ChecksumCalculator(progressHandler,
                                                                    newNbThreads,
                                                                    newBufferSize)
{
}
//---------------------------------------------------------------------------
 
 
/*
 * Calculates the checksum from the given file.
 */
ChecksumFileCalculator::State ChecksumFileCalculator::calculate(ChecksumValue& sumValue,
                                                                const ChecksumAlgoId id,
                                                                const std::filesystem::path& filepath)
{
  error_code ec;
  if (!filesystem::exists(filepath, ec))
    return State::FileNotFound;
 
  ifstream is(filepath, ios_base::in | ios_base::binary);
  if (!is)
    return State::CantOpenFile;
 
  return ChecksumCalculator::calculate(sumValue, id, is);
}
//---------------------------------------------------------------------------
 
 
/*
 * Calculates the checksums from the given file.
 */
ChecksumFileCalculator::State ChecksumFileCalculator::calculate(ArrayChecksumValue& sumValues,
                                                                const ArrayChecksumAlgoId& ids,
                                                                const std::filesystem::path& filepath)
{
  error_code ec;
  if (!filesystem::exists(filepath, ec))
    return State::FileNotFound;
 
  ifstream is(filepath, ios_base::in | ios_base::binary);
  if (!is)
    return State::CantOpenFile;
 
  return ChecksumCalculator::calculate(sumValues, ids, is);
}
//---------------------------------------------------------------------------
 
}  // namespace libxcks
//---------------------------------------------------------------------------