VPF::О грамотной остановке потоков

CaHko

Дата 23.12.2008, 15:38 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 3
Регистрация: 23.12.2008

Репутация: нет
Всего: нет

Была задача реализовать параллельное выполнение маленьких подзадачач,на примере qsort

моя версия такая

Код


#include "stdafx.h"

class CriticalSection
{
public:
    CriticalSection()
    {
        InitializeCriticalSection(&_cs);
    }

    ~CriticalSection()
    {
        DeleteCriticalSection(&_cs);
    }

    void Lock()
    {
        EnterCriticalSection(&_cs);
    }

    void Unlock()
    {
        LeaveCriticalSection(&_cs);
    }

private:
    CRITICAL_SECTION _cs;
};

class Event
{
public:
    Event()
    {
        _event = CreateEvent(NULL, TRUE, FALSE, NULL);
    }

    ~Event()
    {
        CloseHandle(_event);
    }

    void Signal()
    {
        SetEvent(_event);
    }

    void Reset()
    {
        ResetEvent(_event);
    }

    void Wait(int timeout = INFINITE)
    {
        WaitForSingleObject(_event, timeout);
    }

 /*   static int WaitForTwo(const Event& ev1, const Event& ev2, int timeout = INFINITE)
    {
        HANDLE h[2];
        h[0] = ev1._event;
        h[1] = ev2._event;
        int res = WaitForMultipleObjects(2, h, FALSE, timeout);
        switch (res)
        {
            case WAIT_TIMEOUT:
                return 0;
            case WAIT_FAILED:
                return -1;
            default:
                return res - WAIT_OBJECT_0 + 1;
        }
    }
*/
private:
    HANDLE _event;
};

class ThreadPool;

class Callback
{
public:
    virtual void Invoke(ThreadPool* pool) = 0;
};

class MessageQueue
{
public:
    void Push(Callback* callback)
    {
        _cs.Lock();
        _queue.push(callback);
        _notEmpty.Signal();
        _cs.Unlock();
    }

    Callback* Pop()
    {
        while (true)
        {
            _notEmpty.Wait();
            _cs.Lock();
            if (!_queue.empty())
            {
                Callback* obj = _queue.front();
                _queue.pop();
                if (_queue.empty()) 
                    _notEmpty.Reset();
                _cs.Unlock();
                return obj;
            }
            _cs.Unlock();
        }
    }

private:
    std::queue<Callback*> _queue;

    Event _notEmpty;
    CriticalSection _cs;
};


class ExecutorThread
{
public:
    ExecutorThread() : _messageQueue(NULL), _pool(NULL), _thread(NULL)
    {
    }

    void Start(int number, ThreadPool* pool, MessageQueue* mq)
    {
        _number = number;
        _pool = pool;
        _messageQueue = mq;
        _thread = CreateThread(NULL, 0, ThreadProc, this, 0, NULL);
    }

protected:
    void Run()
    {
        while (true)
        {
            Callback* cb = _messageQueue->Pop();
            if (cb == NULL) return;
            std::cout << "Running in " << _number << " thread...\n";
            cb->Invoke(_pool);
        }
    }

private:
    static DWORD WINAPI ThreadProc(LPVOID lpParameter)
    {
        ExecutorThread* zis = (ExecutorThread*)lpParameter;
        zis->Run();
        return 0;
    }

    MessageQueue* _messageQueue;
    ThreadPool* _pool;
    HANDLE _thread;
    int _number;
};

class ThreadPool
{
public:
    ThreadPool(int threads)
    {
        _threads.resize(threads);
        for (unsigned int i = 0; i < _threads.size(); i++)
            _threads[i].Start(i+1, this, &_messageQueue);
    }

    void Invoke(Callback* callback)
    {
        _messageQueue.Push(callback);
    }

    void Stop()
    {
    }

public:
    MessageQueue _messageQueue;
    std::vector<ExecutorThread> _threads;
};


class PrintCallback : public Callback
{
public:
    PrintCallback(int i) : _i(i)
    { }

    virtual void Invoke(ThreadPool* pool)
    {
        Sleep(200);
        std::cout << _i << "\n";
        pool->Invoke(new PrintCallback(_i+1));
        pool->Invoke(new PrintCallback(_i+5));
    }

private:
    int _i;
};

int _tmain(int argc, _TCHAR* argv[])
{
    ThreadPool pool(20);
    pool.Invoke(new PrintCallback(0));
    Sleep(1000);
    pool.Stop();

    return 0;
}

Но как грамотно убить много потоков,чтобы не писать sleep'ы? и я не совсем понял как работают функции treadfunc и т.д,как мне привязать к invoke qsort? Надеюсь на помощь smile

Lazin

Дата 23.12.2008, 15:51 (ссылка)

(нет голосов)

Загрузка ...

Эксперт

Профиль
Группа: Завсегдатай
Сообщений: 3820
Регистрация: 11.12.2006
Где: paranoid oil empi re

Репутация: 8
Всего: 154

WaitForMultipleObjects, в качестве параметра передаешь массив хэндлов потоков и WaitAll = TRUE, функция заблокирует поток до тех пор, пока все потоки не завершатся

Добавлено через 2 минуты и 45 секунд
WaitForMultipleObjects, в качестве параметра передаешь массив хэндлов потоков и WaitAll = TRUE, функция заблокирует поток до тех пор, пока все потоки не завершатся

--------------------

C++ is the steampunk of programming languages: modern ideas realized with victorian technology.

CaHko

Дата 23.12.2008, 17:09 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 3
Регистрация: 23.12.2008

Репутация: нет
Всего: нет

Это я понял. А кто может объяснить как qsort привязать?

Lazin

Дата 23.12.2008, 17:34 (ссылка)

(нет голосов)

Загрузка ...

Эксперт

Профиль
Группа: Завсегдатай
Сообщений: 3820
Регистрация: 11.12.2006
Где: paranoid oil empi re

Репутация: 8
Всего: 154

что значит привязать?
если имеется ввиду распараллелить, то, поскольку алгоритм quicsort построен по принципу разделяй и властвуй, то есть сначала задача делится на маленькие части, а потом их результаты соединяются, то логично использовать алгоритм reduce, а его легко распараллелить. Можно посмотреть как это делается в библиотеке threading building blocks, там есть шаблон parallel_reduce(pdf)

--------------------

C++ is the steampunk of programming languages: modern ideas realized with victorian technology.

CaHko

Дата 23.12.2008, 17:54 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 3
Регистрация: 23.12.2008

Репутация: нет
Всего: нет

Привязать: Та прога которую я привел в начале,и должна была эффективно решать задачу(т.е с использованием много поточности) на примере qsort,но как это привязать к ней qsort я не понял,поэтому просто дописал прогу,чтобы она работала.

xvr

Дата 24.12.2008, 20:32 (ссылка)

(нет голосов)

Загрузка ...

Эксперт

Профиль
Группа: Комодератор
Сообщений: 7046
Регистрация: 28.8.2007
Где: Дублин, Ирландия

Репутация: 40
Всего: 223

Цитата(CaHko @ 23.12.2008, 17:54)

Для начала, ты должен переделать qsort, что бы она не вызывала себя рекурсивно, а рождала потоки

0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| C/C++: Системное программирование и WinAPI \| Следующая тема »