Russian Qt Forum

semaphore.acquire();
allTasks.addTask(string);
qDebug() << "nakidal" << semaphore.available() << QThread::currentThread();
qDebug() << "tasklist size from main:" << allTasks.getSize() << QThread::currentThread();

do {
    string = Tasks->getTask();
    sem->release();
    qDebug() << "zabral" << QThread::currentThread();
    qDebug() << "tasklist size from thread:" << Tasks->getSize();
    ....
while (!stoped && (2 * QThread::idealThreadCount() != sem->available()));

#ifndef TASKING_H
#define TASKING_H

#include <QStringList>
#include <QWaitCondition>
#include <QMutex>
#include <QString>

class Tasking
{
public:
    Tasking();
    QString getTask();
    void addTask(QString string);
    void clear();
    int getSize();

private:
    QStringList stringList;
    QWaitCondition cond;
    QMutex mutex;
};

#endif // TASKING_H

#include "tasking.h"

Tasking::Tasking()
{
}

QString Tasking::getTask()
{
    QMutexLocker locker(&mutex);
    while(stringList.empty())
        //cond.wait(&mutex, ULONG_MAX);
        cond.wait(&mutex);

    return stringList.takeFirst();
}

void Tasking::addTask(QString newString)
{
    QMutexLocker locker(&mutex);
    stringList.append(newString);
    cond.wakeOne();
}

int Tasking::getSize()
{
    return stringList.size();
}

void Tasking::clear()
{
    QMutexLocker locker(&mutex);
    stringList.clear();
}

#ifndef SEARCHTHREAD_H
#define SEARCHTHREAD_H

#include <QThread>
#include <QStringList>
#include <QSemaphore>

class Tasking;

class SearchThread : public QThread
{
    Q_OBJECT
public:
    SearchThread(Tasking *allTasks, QStringList numberList, int direction, QSemaphore *semaphore, QThread *parent = 0);

signals:
    void write(QString);
    void ended();

public slots:
    void run();
    void stop();

private:
    volatile bool stoped;
    int direct;
    QString string;
    QStringList list;
    Tasking *Tasks;
    QSemaphore *sem;
};

#endif // SEARCHTHREAD_H

#include "searchthread.h"
#include "tasking.h"
#include <QDebug>
#include <QTextCodec>

SearchThread::SearchThread(Tasking *allTasks, QStringList smallList, int direction, QSemaphore *semaphore, QThread *parent) :
    QThread(parent)
{
    Tasks = allTasks;
    list = smallList;
    direct = direction;
    sem = semaphore;
    stoped = false;
}

void SearchThread::stop()
{
    stoped = true;
}

void SearchThread::run()
{
    do {
        string = Tasks->getTask();
        sem->release();
        qDebug() << "zabral" << QThread::currentThread();
        qDebug() << "tasklist size from thread:" << Tasks->getSize();

        //qDebug() << string << QThread::currentThread() << stoped;
        QTextCodec::setCodecForCStrings(QTextCodec::codecForName("Windows-1251"));

        for (int i = 0; i < list.size(); ++i) {
            if (-1 != string.indexOf(list[i]))
                emit write(string);
                break;
            }
        }
        emit ended();
    }
    while (!stoped && (2 * QThread::idealThreadCount() != sem->available()));
}

#ifndef MAINSEARCHTHREAD_H
#define MAINSEARCHTHREAD_H

#include <QThread>
#include <QStringList>
#include <QMutex>
#include <QWaitCondition>

class SearchThread;

class MainSearchThread : public QThread
{
    Q_OBJECT
public:
    MainSearchThread(QString bigFile,
                     QString smallFile,
                     QString outFile,
                     int direction,
                     QObject *parent = 0);

signals:
    void warning(const QString);
    void progress(int);
    void ended(bool);
    void ended(QString);
    void stop();

public slots:
    void write(QString);
    void counter();
    void cancel();

protected:
    void run();

private:
    QList<SearchThread*> threadList;

    volatile bool canceled;
    QString bigFileName;
    QString smallFileName;
    QString outFileName;
    int direct;
    volatile qint64 count;
    volatile qint64 bigSize;
    QStringList list;
    qint64 lineCount(const QString FileName);
};

#endif // MAINSEARCHTHREAD_H

#include "mainsearchthread.h"
#include "searchthread.h"
#include "tasking.h"
#include <QFile>
#include <QTextStream>
#include <QThread>
#include <QSemaphore>
#include <QDebug>

MainSearchThread::MainSearchThread(QString bigFile,
                                   QString smallFile,
                                   QString outFile,
                                   int direction,
                                   QObject *parent) :
    QThread(parent),
    canceled(false),
    count(0),
    bigSize(0)
{
    bigFileName = bigFile;
    smallFileName = smallFile;
    outFileName = outFile;
    direct = direction;
}

void MainSearchThread::cancel()
{
    canceled = true;
}

qint64 MainSearchThread::lineCount(const QString FileName)
{
  QFile File(FileName);
  if (!File.open(QIODevice::ReadOnly)) {
      emit warning(tr("Cannot read file %1:\n%2.").arg(File.fileName()).arg(File.errorString()));
      emit ended(tr("Error."));
      return -1;
  }

  char   buff[1024*128];
  qint64 lcount = 0;
  int    len;

  char prev = 0;
  char cur  = 0;
  for(;;) {
      if (canceled)
      {
          emit ended(tr("Canceled."));
          return -1;
      }

      len = File.read(buff, sizeof(buff));
      if (File.error()) return -1;
      if(!len) {break;}

      for (int i=0; i<len; ++i) {
          cur = buff[i];
          if      (cur  == 10) {++lcount;}
          else if (prev == 13) {++lcount;}
          prev = cur;
      }
  }
  if (cur == 13) {++lcount;}
  File.close();
  return lcount + 1;
}

void MainSearchThread::counter()
{
    count++;
    emit progress(count*100/bigSize);
}

void MainSearchThread::write(QString string)
{
    qDebug() << string;
    list.append(string);
}

void MainSearchThread::run()
{
    //определяем размер файла
    billingSize = lineCount(bigFileName);
    if ( -1 == bigSize)
        return;
    //проверка на открытие файла
    QFile bigFile(bigFileName);
    if (!bigFile.open(QIODevice::ReadOnly)) {
        emit warning(tr("Cannot read file %1:\n%2.").arg(bigFile.fileName()).arg(bigFile.errorString()));
        emit ended(tr("Error."));
        return;
    }
    //проверка открытия файла
    QFile smallFile(smallFileName);
    if (!smallFile.open(QIODevice::ReadOnly)) {
        emit warning(tr("Cannot read file %1:\n%2.").arg(smallFile.fileName()).arg(smallFile.errorString()));
        emit ended(tr("Error."));
        bigFile.close();
        return;
    }
    //берём строки из файла и кладём в список строк
    QTextStream numberStream(&smallFile);
    QStringList smallList;
    QString smallString;
    while (!smallStream.atEnd()) {
        //проверка на отмену
        if (canceled) {
            bigFile.close();
            smallFile.close();
            emit ended(true);
            emit ended(tr("Canceled."));
        }
        smallString = smallStream.readLine();
        //если строка пустая, то пропускаем её
        if ("" == smallString)
            continue;
        smallList.append(smallString);
    }

    Tasking allTasks; //создаём список заданий
    QSemaphore semaphore(2 * QThread::idealThreadCount()); //семафор для контроля раздачи заданий
    //QSemaphore semaphore(2); //семафор для контроля раздачи заданий
    //подготавливаем список потоков, раздаем списки и указатель на список заданий
    for (int i = 1; i <= QThread::idealThreadCount(); i++) {
        SearchThread *thread = new SearchThread(&allTasks, smallList, direct, &semaphore);
        connect(thread, SIGNAL(write(QString)), this, SLOT(write(QString)));
        connect(thread, SIGNAL(ended()), this, SLOT(counter()));
        connect(this, SIGNAL(stop()), thread, SLOT(stop()));
        threadList.append(thread);
        thread->start();
    }
    //устанавливаем счётчик строк в 0
    count = 0;
    //читаем построчно файл и раздаем строки списку потоков
    QTextStream bigStream(&bigFile);
    QString bigString;
    while (!bigStream.atEnd()) {
        //проверка на отмену
        if (canceled) {
            //очищаем список заданий
            allTasks.clear();
            //сообщаем об остановке
            emit stop();
            //ждем пока все потоки завершат выполнение
            for (int i = 1; i <= QThread::idealThreadCount(); i++) {
                threadList[i - 1]->wait();
            }
            bigFile.close();
            emit ended(true);
            emit ended(tr("Canceled."));
            return;
        }
        //читаем строку из файла
        bigString = bigStream.readLine();
        //если строка пустая, то пропускаем её
        if ("" == bigString)
            continue;

        semaphore.acquire();
        allTasks.addTask(bigString);
        qDebug() << "nakidal" << semaphore.available() << QThread::currentThread();
        qDebug() << "tasklist size from main:" << allTasks.getSize() << QThread::currentThread();
    }
    //необходимо сообщить потокам что всё
    emit stop();
    //ожидаем завершения всех потоков
    for (int i = 1; i <= QThread::idealThreadCount(); i++) {
        threadList[i - 1]->wait();
    }
    qDebug() << list;
    //закрываем файл
    bigFile.close();
    //открываем файл для записи найденых строк
    QFile outFile(outFileName);
    if (!outFile.open(QIODevice::WriteOnly)) {
        emit warning(tr("Cannot write file %1:\n%2.").arg(outFile.fileName()).arg(outFile.errorString()));
        return;
    }
    //записываем найденые строки
    QTextStream outStream(&outFile);
    for (int i = 0; i < list.size(); ++i)
        outStream << list[i] << "\r\n";

    outFile.close(); //закрываем файл с найдеными строками
    emit ended(true);
    emit progress(100);
    emit ended(tr("Sucsessfuly complited."));
}

QSemaphore semaphore(2 * QThread::idealThreadCount()); //симофор для контроля раздачи заданий