Доброго времени суток!

Недавно столкнулся с задачей:

(привожу задание как оно есть)

Пусть у нас на диске есть файл размером 4 гигабайта. Его можно представить, как 2^30 32-битных беззнаковых чисел. Нужно отсортировать этот файл. То есть программа должна сгенерировать еще один файл размером в 4 гигабайта, в котором содержатся те же числа, что и в исходном, но упорядочены по возрастанию. Стандартные алгоритмы сортировки (qsort, std::sort) напрямую применить невозможно, так как для их выполнения нужно как минимум 4 гигабайта оперативной памяти. Но отсортировать числа на диске возможно, если использовать дополнительное пространство на жестком диске.

Нужно написать консольную программу, которая в argv[1] получает имя файла, который нужно отсортировать (размер файла до 16Gb, размер кратен четырем),
в argv[2] — имя файла, в который нужно записать отсортированные данные.

Ограничения:
1. Программа не может рассчитывать, что возможно выделить более 256Mb памяти.
2. Программа должна эффективно использовать несколько ядер.

Решение получилось следующее:

#include <iostream>
#include <sstream>
#include <stdio.h>
#include <vector>
#include <time.h>
#include <algorithm>
#include <process.h>
#include "windows.h"


#define MEMORY_LIMIT 268435456 // 256Mb

unsigned int nMemForThread = MEMORY_LIMIT / 4;

CRITICAL_SECTION csMergePool;
CRITICAL_SECTION csFileAccess;
CRITICAL_SECTION csChunkName;
CRITICAL_SECTION csMergeError;
CRITICAL_SECTION csSortError;

using namespace std;

// for qsort
int compare(const void *a, const void *b)
{
    return ( *(int*)a - *(int*)b );
}

//this function(generateNextChunkFileName) is only to be called within critical section
string generateNextChunkFileName()
{
    static int nChunkNum = 0;
    string strChunkFileName("chunk_");

    ostringstream oss;
    oss << nChunkNum;
    strChunkFileName += oss.str();
    ++nChunkNum;

    return strChunkFileName;
}

enum errType
{
    et_NoError = 0,
    et_InputData,
    et_Access,
    et_Write,
    et_Read,
    et_Merge,
    et_Unexpected
};

/**
* class SorterException
*/
class SorterException
{
    errType m_error;
    string m_strErrorMessage;
    string m_strParam;

public:
    static const char *pchExcFileAccessError;
    static const char *pchExcInputDataError;
    static const char *pchExcReadError;
    static const char *pchExcWriteError;
    static const char *pchExcMergeError;
    static const char *pchExcUnexpectedError;

    SorterException(const string &strErrorMessage, errType error) : 
    m_error(error), m_strErrorMessage(strErrorMessage), m_strParam("") {}

    void setParam(const string & strParam)
    {
        m_strParam = strParam;
    }

    string getErrorMessage() const
    {
        return m_strErrorMessage;
    }

    errType getErrorType() const
    {
        return m_error;
    }

    static void generateException(errType error)
    {
        switch (error)
        {
        case et_InputData:
            throw SorterException(pchExcInputDataError, et_InputData);

        case et_Access:
            throw SorterException(pchExcFileAccessError, et_Access);

        case et_Write:
            throw SorterException(pchExcWriteError, et_Write);

        case et_Read:
            throw SorterException(pchExcReadError, et_Read);

        case et_Merge:
            throw SorterException(pchExcMergeError, et_Merge);

        default:
            throw SorterException(pchExcUnexpectedError, et_Unexpected);
        }
    }

    friend ostream & operator<<(ostream &stream, SorterException &exc)
    {
        stream << exc.m_strErrorMessage;

        if (!exc.m_strParam.empty())
        {
            stream << " " << exc.m_strParam;
        }

        stream << endl;

        return stream;
    }
};
//static members
const char* SorterException::pchExcFileAccessError = "Error occured while accessing file";
const char* SorterException::pchExcInputDataError  = "Input data is wrong";
const char* SorterException::pchExcReadError       = "Error occured while reading from file";
const char* SorterException::pchExcWriteError      = "Error occured while writing into file";
const char* SorterException::pchExcMergeError      = "Error occured while merging";
const char* SorterException::pchExcUnexpectedError = "Unexpected error";
//---------------------------------------------------------------------------------------------------

struct FileWrapper
{
    FILE *file;
    const char *m_pchFileName;
    long long m_nFileSize;

    void open(const char *pchInputFileName, const char *mode)
    {
        if (!file)
        {
            m_pchFileName = pchInputFileName;
            file = fopen(m_pchFileName, mode);

            if (!file)
            {
                SorterException::generateException(et_Access);
            }
        }
    }

    void close()
    {
        if (file)
        {
            if (fclose(file))
            {
                SorterException::generateException(et_Access);
            }

            file = NULL;
        }
    }

    void write(void *dest, size_t nSize, size_t nCount)
    {
        fwrite(dest, nSize, nCount, file);

        if (ferror(file))
        {
            SorterException ex(SorterException::pchExcWriteError, et_Write);
            ex.setParam(m_pchFileName);
            throw ex;
        }
    }

    size_t read(void *dest, size_t nSize, size_t nCount)
    {
        size_t nReadCount = fread(dest, nSize, nCount, file);

        if (ferror(file))
        {
            SorterException ex(SorterException::pchExcReadError, et_Read);
            ex.setParam(m_pchFileName);
            throw ex;
        }

        return nReadCount;
    }

    long long getSize()
    {
        if (file && m_nFileSize == 0)
        {
            int nSeekEnd = _fseeki64(file, 0, SEEK_END);
            m_nFileSize = static_cast<long long>(_ftelli64(file));            
            int nSeekStart = _fseeki64(file, 0, SEEK_SET);

            if (nSeekEnd || nSeekStart)
            {
                SorterException::generateException(et_Access);
            }
        }

        return m_nFileSize;
    }

    FileWrapper() : file(NULL), m_pchFileName(NULL), m_nFileSize(0) {}

    ~FileWrapper()
    {
        try
        {
            close();
        }
        catch (...)
        {
            // write to log
        }
    }
};

/**
* class Merger
* Merge files
*/

class Merger
{
    static string m_strResultOfMergeFileName;
    static bool m_bStopMerge;
    static bool m_bImmediateExit;
    static vector<string> m_vPool2Put;
    static size_t m_nMergeIndex;
    static errType m_error;

    string m_strCurrMergeFileName;
    string m_strLeftSource;
    string m_strRightSource;

    Merger(const Merger&);
    Merger& operator = (const Merger&);

    void swapPools()
    {
        m_vPool2Take.clear();
        m_vPool2Take.swap(m_vPool2Put);
    }

public:

    static vector<string> m_vPool2Take;

    Merger() {}

    void setLeftSourceFileName(const string &strLeft)
    {
        m_strLeftSource = strLeft;
    }

    void setRightSourceFileName(const string &strRight)
    {
        m_strRightSource = strRight;
    }

    void setCurrMergeFileName(const string &strMergeFileName)
    {
        m_strCurrMergeFileName = strMergeFileName;
    }

    const string& getCurrMergeFileName() const
    {
        return m_strCurrMergeFileName;
    }

    static void setError(errType error)
    {
        m_error = error;
    }

    static errType getError()
    {
        return m_error;
    }

    static void setImmediateExit(bool bExit)
    {
        m_bImmediateExit = bExit;
    }

    static unsigned __stdcall mergeInThread(void *arg)
    {
        if (!m_bImmediateExit)
        {
            Merger *merger = static_cast<Merger*>(arg);

            string strFirstChunkFileName;
            string strSecondChunkFileName;

            while (1)
            {
                if (m_bStopMerge)
                {
                    break;
                }

                bool bMerge = false;

                // critical section
                EnterCriticalSection(&csMergePool);

                size_t nShiftIndex = 0;

                if (m_vPool2Take.size() > 2)
                {
                    nShiftIndex = m_nMergeIndex + 2;
                }
                else
                {
                    // left to merge two files
                    nShiftIndex = m_nMergeIndex + 1;
                    m_bStopMerge = true;
                }

                if (nShiftIndex < m_vPool2Take.size())
                {
                    merger->setLeftSourceFileName(m_vPool2Take[m_nMergeIndex]);
                    merger->setRightSourceFileName(m_vPool2Take[nShiftIndex]);

                    ++m_nMergeIndex;

                    if (!(m_nMergeIndex % 2))
                    {
                        m_nMergeIndex += 2;
                    }

                    bMerge = true;
                }
                else if (m_vPool2Put.size() == m_vPool2Take.size() / 2)
                {
                    merger->swapPools();
                    m_nMergeIndex = 0;
                }

                LeaveCriticalSection(&csMergePool);

                try
                {
                    if (bMerge)
                    {
                        EnterCriticalSection(&csChunkName);
                        merger->setCurrMergeFileName(generateNextChunkFileName());
                        LeaveCriticalSection(&csChunkName);

                        merger->mergeTwoFilesIntoOne();

                        EnterCriticalSection(&csMergePool);
                        m_vPool2Put.push_back(merger->getCurrMergeFileName());
                        LeaveCriticalSection(&csMergePool);
                    }
                }
                catch(SorterException &ex)
                {
                    EnterCriticalSection(&csMergeError);

                    if (ex.getErrorType() != et_NoError)
                    {
                        Merger::setError(ex.getErrorType());
                        m_bStopMerge = true;
                    }

                    LeaveCriticalSection(&csMergeError);
                }
                catch (...)
                {
                    EnterCriticalSection(&csMergeError);
                    Merger::setError(et_Unexpected);
                    m_bStopMerge = true;
                    LeaveCriticalSection(&csMergeError);
                }
            }
        }

        _endthreadex(0);
        return 0;
    }

    void mergeTwoFilesIntoOne()
    {
        FileWrapper left;
        left.open(m_strLeftSource.c_str(), "rb");

        FileWrapper right;
        right.open(m_strRightSource.c_str(), "rb");

        long long nLeftSize = left.getSize();
        long long nRightSize = right.getSize();

        if (m_bStopMerge)
        {
            m_strCurrMergeFileName = m_strResultOfMergeFileName;
        }

        FileWrapper mergeFile;
        mergeFile.open(m_strCurrMergeFileName.c_str(), "wb");

        vector<int> vLeft((nMemForThread / sizeof(int)) / 2);
        vector<int> vRight((nMemForThread / sizeof(int)) / 2);
        vector<int> vResultOfMerge(nMemForThread / sizeof(int));

        long long nCurrElemNumberFromLeft = 0;
        long long nCurrElemNumberFromRight = 0;

        long long nDefaultElemsToRead = (nMemForThread / sizeof(int)) / 2;

        while (nLeftSize && nRightSize)
        {
            //---------------------read left source--------------------------
            long long nElemsToRead = nDefaultElemsToRead;

            if (nElemsToRead > nLeftSize / sizeof(int))
            {
                nElemsToRead = nLeftSize / sizeof(int);
                vLeft.resize(static_cast<size_t>(nElemsToRead));
            }

            nCurrElemNumberFromLeft = left.read(&vLeft[0], sizeof(int), static_cast<size_t>(nElemsToRead));
            nLeftSize -= nCurrElemNumberFromLeft * sizeof(int);

            //---------------------read right source--------------------------
            nElemsToRead = nDefaultElemsToRead;

            if (nElemsToRead > nRightSize / sizeof(int))
            {
                nElemsToRead = nRightSize / sizeof(int);
                vRight.resize(static_cast<size_t>(nElemsToRead));
            }

            nCurrElemNumberFromRight = right.read(&vRight[0], sizeof(int), static_cast<size_t>(nElemsToRead));
            nRightSize -= nCurrElemNumberFromRight * sizeof(int);

            //---------------------merge in memory and then flush---------------
            merge(vLeft.begin(), vLeft.end(), vRight.begin(), vRight.end(), vResultOfMerge.begin());
            // flush just merged data
            mergeFile.write(&vResultOfMerge[0], sizeof(int), vResultOfMerge.size());
        }

        left.close();
        right.close();
        mergeFile.close();

        remove(m_strLeftSource.c_str());
        remove(m_strRightSource.c_str());

        if (nLeftSize || nRightSize)
        {
            // somthing wrong happend
            SorterException::generateException(et_Merge);
        }
    }

    static void setMergeFileName(const string &strFName)
    {
        m_strResultOfMergeFileName = strFName;
    }
};
// static members
string Merger::m_strResultOfMergeFileName;
bool Merger::m_bStopMerge = false;
bool Merger::m_bImmediateExit = false;
vector<string> Merger::m_vPool2Put(0);
vector<string> Merger::m_vPool2Take(0);
size_t Merger::m_nMergeIndex = 0;
errType Merger::m_error = et_NoError;
//---------------------------------------------------------------------------------------------------

/**
* class ChunkSorter
* Read and sort input data from file by small chunks
* After chunk is sorted it is flushed on disk for merge
*/

class ChunkSorter
{
    static bool m_bNoMoreData;
    static bool m_bAbortExit;
    static vector<string> *m_pChunkNamePool; //!< is to be set before using ChunkSorter
    static errType m_error;
    vector<int> m_vChunkToSort;

    ChunkSorter(const ChunkSorter&);
    ChunkSorter& operator = (const ChunkSorter&);

    void sortAndFlush()
    {
        while (1)
        {
            EnterCriticalSection(&csFileAccess);

            size_t nReadElems = m_fileToSort.read(&m_vChunkToSort[0], sizeof(int), nMemForThread / sizeof(int));

            if (feof(m_fileToSort.file))
            {
                m_bNoMoreData = true;
            }

            LeaveCriticalSection(&csFileAccess);

            if (nReadElems)
            {
                // sort data
                qsort(&m_vChunkToSort[0], m_vChunkToSort.size(), sizeof(int), compare);

                EnterCriticalSection(&csChunkName);
                string strChunkFileName = generateNextChunkFileName();
                LeaveCriticalSection(&csChunkName);

                // flush sorted data
                FileWrapper f;
                f.open(strChunkFileName.c_str(), "wb");
                f.write(&m_vChunkToSort[0], sizeof(int), m_vChunkToSort.size());
                f.close();

                // critical section
                EnterCriticalSection(&csMergePool);
                m_pChunkNamePool->push_back(strChunkFileName);
                LeaveCriticalSection(&csMergePool);
            }

            if (m_bNoMoreData || m_bAbortExit)
            {
                // end up with sorting
                m_vChunkToSort.clear();
                break;
            }
        }
    }

public:

    static FileWrapper m_fileToSort;
    static long long m_nFileSize;

    ChunkSorter() :
    m_vChunkToSort(nMemForThread / sizeof(int))
    {
    }

    static void setError(errType error)
    {
        m_error = error;
    }

    static errType getError()
    {
        return m_error;
    }

    static unsigned __stdcall thread_func(void* arg)
    {
        ChunkSorter *sorter = static_cast<ChunkSorter*>(arg);

        try
        {
            if (!m_pChunkNamePool)
            {
                SorterException::generateException(et_InputData);
            }

            sorter->sortAndFlush();
        }
        catch(SorterException &ex)
        {
            EnterCriticalSection(&csSortError);

            if (ex.getErrorType() != et_NoError)
            {
                ChunkSorter::setError(ex.getErrorType());
            }

            m_bAbortExit = true;
            LeaveCriticalSection(&csSortError);
        }
        catch (...)
        {
            EnterCriticalSection(&csSortError);
            ChunkSorter::setError(et_Unexpected);
            m_bAbortExit = true;
            LeaveCriticalSection(&csSortError);
        }

        _endthreadex(0);
        return 0;
    }

    static void setFileToSort(const char *pchFileName)
    {
        m_fileToSort.open(pchFileName, "rb");
        m_nFileSize = m_fileToSort.getSize();
    }

    // method is to be called to check input data size after file is set
    static void checkInputDataSize()
    {
        if (m_nFileSize == 0 || m_nFileSize % sizeof(int))
        {
            SorterException::generateException(et_InputData);
        }
    }

    static void setNamePool(vector<string> *vNamePool)
    {
        m_pChunkNamePool = vNamePool;
    }
};

// static members
FileWrapper ChunkSorter::m_fileToSort;
long long ChunkSorter::m_nFileSize = 0;
bool ChunkSorter::m_bNoMoreData = false;
bool ChunkSorter::m_bAbortExit = false;
errType ChunkSorter::m_error = et_NoError;
vector<string>* ChunkSorter::m_pChunkNamePool = NULL;
//---------------------------------------------------------------------------------------------------


int main (int argc, char* argv[])
{
    try
    {
        if (argc < 2)
        {
            SorterException::generateException(et_InputData);
        }

        string strInputFileName = argv[1];
        string strOutputFileName = argv[2];

        ChunkSorter::setFileToSort(strInputFileName.c_str());
        ChunkSorter::checkInputDataSize();

        if (ChunkSorter::m_nFileSize <= MEMORY_LIMIT)
        {
            // data size is small enough to be allocated and sorted in memory
            vector<int> vDataToSort(static_cast<size_t>(ChunkSorter::m_nFileSize / sizeof(int)));

            ChunkSorter::m_fileToSort.read(&vDataToSort[0], sizeof(int), vDataToSort.capacity());

            qsort(&vDataToSort[0], vDataToSort.size(), sizeof(int), compare);

            FileWrapper merge;
            merge.open(strOutputFileName.c_str(), "wb");
            merge.write(&vDataToSort[0], sizeof(int), vDataToSort.size());
        }
        else
        {
            ChunkSorter::setNamePool(&Merger::m_vPool2Take);
            vector<char> vTime(10);
            _strtime_s(&vTime[0], vTime.size());
            cout << "start time: " << &vTime[0] << endl;

            InitializeCriticalSection(&csMergePool);
            InitializeCriticalSection(&csFileAccess);
            InitializeCriticalSection(&csChunkName);
            InitializeCriticalSection(&csMergeError);
            InitializeCriticalSection(&csSortError);

            //--------------------------------chunk_sorter initialization------------------------------------
            ChunkSorter SortAgent1;
            ChunkSorter SortAgent2;
            ChunkSorter SortAgent3;
            ChunkSorter SortAgent4;

            unsigned Agent1ThreadID;
            unsigned Agent2ThreadID;
            unsigned Agent3ThreadID;
            unsigned Agent4ThreadID;

            HANDLE hSortThreads[4];
            hSortThreads[0] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent1, 0, &Agent1ThreadID);
            hSortThreads[1] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent2, 0, &Agent2ThreadID);
            hSortThreads[2] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent3, 0, &Agent3ThreadID);
            hSortThreads[3] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent4, 0, &Agent4ThreadID);

            //------------------------------------merger initialization---------------------------------------
            Merger::setMergeFileName(strOutputFileName);
            Merger mergerAgent1;
            Merger mergerAgent2;

            unsigned mergerThredID1;
            unsigned mergerThredID2;

            HANDLE hMergeThreads[2];
            hMergeThreads[0] = (HANDLE)_beginthreadex(NULL, 0, &Merger::mergeInThread, &mergerAgent1, CREATE_SUSPENDED, &mergerThredID1);
            hMergeThreads[1] = (HANDLE)_beginthreadex(NULL, 0, &Merger::mergeInThread, &mergerAgent2, CREATE_SUSPENDED, &mergerThredID2);
            //-----------------------------------------------------------------------------------------------

            // wait until sorters finish their job and then resume threads to merge
            WaitForMultipleObjectsEx(4, hSortThreads, true, INFINITE, false);

            CloseHandle(hSortThreads[3]);
            CloseHandle(hSortThreads[2]);
            CloseHandle(hSortThreads[1]);
            CloseHandle(hSortThreads[0]);

            // check sort error here
            bool bSortError = false;

            if (ChunkSorter::getError() != et_NoError)
            {
                Merger::setImmediateExit(true);
                bSortError = true;
            }

            _strtime_s(&vTime[0], vTime.size());
            cout << "resume threads time: " << &vTime[0] << endl;

            ResumeThread(hMergeThreads[0]);
            ResumeThread(hMergeThreads[1]);

            // wait untill all threads finish
            WaitForMultipleObjectsEx(2, hMergeThreads, true, INFINITE, false);

            CloseHandle(hMergeThreads[1]);
            CloseHandle(hMergeThreads[0]);

            DeleteCriticalSection(&csChunkName);
            DeleteCriticalSection(&csFileAccess);
            DeleteCriticalSection(&csMergePool);

            DeleteCriticalSection(&csMergeError);
            DeleteCriticalSection(&csSortError);

            if (bSortError)
            {
                SorterException::generateException(ChunkSorter::getError());
            }

            // check merge error here
            if (Merger::getError() != et_NoError)
            {
                SorterException::generateException(Merger::getError());
            }

            _strtime_s(&vTime[0], vTime.size());
            cout << "end time: " << &vTime[0] << endl;
        }
    }

    catch (SorterException &ex)
    {
        cout << ex;
    }

    catch (...)
    {
        cerr << "Unknown error" << endl;
    }

    return 0;
}

Сразу скажу, что сделал изначально с ошибкой — сортирует файлы размером <= 256Mb либо размером кратным этому числу, т.е. файл размером 260Mb не будет отсортирован корректно, потеряется последний int(если sizeof(int) == 4). Но вопрос не в этом, программа компилируется и даже сортирует файл: 4G она обрабатывает за ~4 часа. Просьба покритиковать код, указать на явные ошибки(если есть) работы с потоками(до этой задачи никогда не работал с потоками вообще) ну и проблемы кода в целом.

Здравствуйте, letsurf, Вы писали:

L>Но вопрос не в этом, программа компилируется и даже сортирует файл: 4G она обрабатывает за ~4 часа. Просьба покритиковать код

Критикую. 4 часа на сортировку файла в 4 Гб это писец, мягко говоря. В Яндекс тебя не возьмут. Выкини и перепиши по новой

Здравствуйте, kaa.python, Вы писали:

KP>Здравствуйте, letsurf, Вы писали:

L>>Но вопрос не в этом, программа компилируется и даже сортирует файл: 4G она обрабатывает за ~4 часа. Просьба покритиковать код

KP>Критикую. 4 часа на сортировку файла в 4 Гб это писец, мягко говоря. В Яндекс тебя не возьмут. Выкини и перепиши по новой

Ну вот и не взяли . Спасибо

#include <iostream>
#include <sstream>
#include <stdio.h>
#include <vector>
#include <time.h>
#include <algorithm>
#include <process.h>
#include "windows.h"


#define MEMORY_LIMIT 268435456 // 256Mb

unsigned int nMemForThread = MEMORY_LIMIT / 4;

CRITICAL_SECTION csMergePool;
CRITICAL_SECTION csFileAccess;
CRITICAL_SECTION csChunkName;
CRITICAL_SECTION csMergeError;
CRITICAL_SECTION csSortError;

using namespace std;

// for qsort
int compare(const void *a, const void *b)
{
    return ( *(int*)a - *(int*)b );
}

//this function(generateNextChunkFileName) is only to be called within critical section
string generateNextChunkFileName()
{
    static int nChunkNum = 0;
    string strChunkFileName("chunk_");

    ostringstream oss;
    oss << nChunkNum;
    strChunkFileName += oss.str();
    ++nChunkNum;

    return strChunkFileName;
}

enum errType
{
    et_NoError = 0,
    et_InputData,
    et_Access,
    et_Write,
    et_Read,
    et_Merge,
    et_Unexpected
};

/**
* class SorterException
*/
class SorterException
{
    errType m_error;
    string m_strErrorMessage;
    string m_strParam;

public:
    static const char *pchExcFileAccessError;
    static const char *pchExcInputDataError;
    static const char *pchExcReadError;
    static const char *pchExcWriteError;
    static const char *pchExcMergeError;
    static const char *pchExcUnexpectedError;

    SorterException(const string &strErrorMessage, errType error) : 
    m_error(error), m_strErrorMessage(strErrorMessage), m_strParam("") {}

    void setParam(const string & strParam)
    {
        m_strParam = strParam;
    }

    string getErrorMessage() const
    {
        return m_strErrorMessage;
    }

    errType getErrorType() const
    {
        return m_error;
    }

    static void generateException(errType error)
    {
        switch (error)
        {
        case et_InputData:
            throw SorterException(pchExcInputDataError, et_InputData);

        case et_Access:
            throw SorterException(pchExcFileAccessError, et_Access);

        case et_Write:
            throw SorterException(pchExcWriteError, et_Write);

        case et_Read:
            throw SorterException(pchExcReadError, et_Read);

        case et_Merge:
            throw SorterException(pchExcMergeError, et_Merge);

        default:
            throw SorterException(pchExcUnexpectedError, et_Unexpected);
        }
    }

    friend ostream & operator<<(ostream &stream, SorterException &exc)
    {
        stream << exc.m_strErrorMessage;

        if (!exc.m_strParam.empty())
        {
            stream << " " << exc.m_strParam;
        }

        stream << endl;

        return stream;
    }
};
//static members
const char* SorterException::pchExcFileAccessError = "Error occured while accessing file";
const char* SorterException::pchExcInputDataError  = "Input data is wrong";
const char* SorterException::pchExcReadError       = "Error occured while reading from file";
const char* SorterException::pchExcWriteError      = "Error occured while writing into file";
const char* SorterException::pchExcMergeError      = "Error occured while merging";
const char* SorterException::pchExcUnexpectedError = "Unexpected error";
//---------------------------------------------------------------------------------------------------

struct FileWrapper
{
    FILE *file;
    const char *m_pchFileName;
    long long m_nFileSize;

    void open(const char *pchInputFileName, const char *mode)
    {
        if (!file)
        {
            m_pchFileName = pchInputFileName;
            file = fopen(m_pchFileName, mode);

            if (!file)
            {
                SorterException::generateException(et_Access);
            }
        }
    }

    void close()
    {
        if (file)
        {
            if (fclose(file))
            {
                SorterException::generateException(et_Access);
            }

            file = NULL;
        }
    }

    void write(void *dest, size_t nSize, size_t nCount)
    {
        fwrite(dest, nSize, nCount, file);

        if (ferror(file))
        {
            SorterException ex(SorterException::pchExcWriteError, et_Write);
            ex.setParam(m_pchFileName);
            throw ex;
        }
    }

    size_t read(void *dest, size_t nSize, size_t nCount)
    {
        size_t nReadCount = fread(dest, nSize, nCount, file);

        if (ferror(file))
        {
            SorterException ex(SorterException::pchExcReadError, et_Read);
            ex.setParam(m_pchFileName);
            throw ex;
        }

        return nReadCount;
    }

    long long getSize()
    {
        if (file && m_nFileSize == 0)
        {
            int nSeekEnd = _fseeki64(file, 0, SEEK_END);
            m_nFileSize = static_cast<long long>(_ftelli64(file));            
            int nSeekStart = _fseeki64(file, 0, SEEK_SET);

            if (nSeekEnd || nSeekStart)
            {
                SorterException::generateException(et_Access);
            }
        }

        return m_nFileSize;
    }

    FileWrapper() : file(NULL), m_pchFileName(NULL), m_nFileSize(0) {}

    ~FileWrapper()
    {
        try
        {
            close();
        }
        catch (...)
        {
            // write to log
        }
    }
};

/**
* class Merger
* Merge files
*/

class Merger
{
    static string m_strResultOfMergeFileName;
    static bool m_bStopMerge;
    static bool m_bImmediateExit;
    static vector<string> m_vPool2Put;
    static size_t m_nMergeIndex;
    static errType m_error;

    string m_strCurrMergeFileName;
    string m_strLeftSource;
    string m_strRightSource;

    Merger(const Merger&);
    Merger& operator = (const Merger&);

    void swapPools()
    {
        m_vPool2Take.clear();
        m_vPool2Take.swap(m_vPool2Put);
    }

public:

    static vector<string> m_vPool2Take;

    Merger() {}

    void setLeftSourceFileName(const string &strLeft)
    {
        m_strLeftSource = strLeft;
    }

    void setRightSourceFileName(const string &strRight)
    {
        m_strRightSource = strRight;
    }

    void setCurrMergeFileName(const string &strMergeFileName)
    {
        m_strCurrMergeFileName = strMergeFileName;
    }

    const string& getCurrMergeFileName() const
    {
        return m_strCurrMergeFileName;
    }

    static void setError(errType error)
    {
        m_error = error;
    }

    static errType getError()
    {
        return m_error;
    }

    static void setImmediateExit(bool bExit)
    {
        m_bImmediateExit = bExit;
    }

    static unsigned __stdcall mergeInThread(void *arg)
    {
        if (!m_bImmediateExit)
        {
            Merger *merger = static_cast<Merger*>(arg);

            string strFirstChunkFileName;
            string strSecondChunkFileName;

            while (1)
            {
                if (m_bStopMerge)
                {
                    break;
                }

                bool bMerge = false;

                // critical section
                EnterCriticalSection(&csMergePool);

                size_t nShiftIndex = 0;

                if (m_vPool2Take.size() > 2)
                {
                    nShiftIndex = m_nMergeIndex + 2;
                }
                else
                {
                    // left to merge two files
                    nShiftIndex = m_nMergeIndex + 1;
                    m_bStopMerge = true;
                }

                if (nShiftIndex < m_vPool2Take.size())
                {
                    merger->setLeftSourceFileName(m_vPool2Take[m_nMergeIndex]);
                    merger->setRightSourceFileName(m_vPool2Take[nShiftIndex]);

                    ++m_nMergeIndex;

                    if (!(m_nMergeIndex % 2))
                    {
                        m_nMergeIndex += 2;
                    }

                    bMerge = true;
                }
                else if (m_vPool2Put.size() == m_vPool2Take.size() / 2)
                {
                    merger->swapPools();
                    m_nMergeIndex = 0;
                }

                LeaveCriticalSection(&csMergePool);

                try
                {
                    if (bMerge)
                    {
                        EnterCriticalSection(&csChunkName);
                        merger->setCurrMergeFileName(generateNextChunkFileName());
                        LeaveCriticalSection(&csChunkName);

                        merger->mergeTwoFilesIntoOne();

                        EnterCriticalSection(&csMergePool);
                        m_vPool2Put.push_back(merger->getCurrMergeFileName());
                        LeaveCriticalSection(&csMergePool);
                    }
                }
                catch(SorterException &ex)
                {
                    EnterCriticalSection(&csMergeError);

                    if (ex.getErrorType() != et_NoError)
                    {
                        Merger::setError(ex.getErrorType());
                        m_bStopMerge = true;
                    }

                    LeaveCriticalSection(&csMergeError);
                }
                catch (...)
                {
                    EnterCriticalSection(&csMergeError);
                    Merger::setError(et_Unexpected);
                    m_bStopMerge = true;
                    LeaveCriticalSection(&csMergeError);
                }
            }
        }

        _endthreadex(0);
        return 0;
    }

    void mergeTwoFilesIntoOne()
    {
        FileWrapper left;
        left.open(m_strLeftSource.c_str(), "rb");

        FileWrapper right;
        right.open(m_strRightSource.c_str(), "rb");

        long long nLeftSize = left.getSize();
        long long nRightSize = right.getSize();

        if (m_bStopMerge)
        {
            m_strCurrMergeFileName = m_strResultOfMergeFileName;
        }

        FileWrapper mergeFile;
        mergeFile.open(m_strCurrMergeFileName.c_str(), "wb");

        vector<int> vLeft((nMemForThread / sizeof(int)) / 2);
        vector<int> vRight((nMemForThread / sizeof(int)) / 2);
        vector<int> vResultOfMerge(nMemForThread / sizeof(int));

        long long nCurrElemNumberFromLeft = 0;
        long long nCurrElemNumberFromRight = 0;

        long long nDefaultElemsToRead = (nMemForThread / sizeof(int)) / 2;

        while (nLeftSize && nRightSize)
        {
            //---------------------read left source--------------------------
            long long nElemsToRead = nDefaultElemsToRead;

            if (nElemsToRead > nLeftSize / sizeof(int))
            {
                nElemsToRead = nLeftSize / sizeof(int);
                vLeft.resize(static_cast<size_t>(nElemsToRead));
            }

            nCurrElemNumberFromLeft = left.read(&vLeft[0], sizeof(int), static_cast<size_t>(nElemsToRead));
            nLeftSize -= nCurrElemNumberFromLeft * sizeof(int);

            //---------------------read right source--------------------------
            nElemsToRead = nDefaultElemsToRead;

            if (nElemsToRead > nRightSize / sizeof(int))
            {
                nElemsToRead = nRightSize / sizeof(int);
                vRight.resize(static_cast<size_t>(nElemsToRead));
            }

            nCurrElemNumberFromRight = right.read(&vRight[0], sizeof(int), static_cast<size_t>(nElemsToRead));
            nRightSize -= nCurrElemNumberFromRight * sizeof(int);

            //---------------------merge in memory and then flush---------------
            merge(vLeft.begin(), vLeft.end(), vRight.begin(), vRight.end(), vResultOfMerge.begin());
            // flush just merged data
            mergeFile.write(&vResultOfMerge[0], sizeof(int), vResultOfMerge.size());
        }

        left.close();
        right.close();
        mergeFile.close();

        remove(m_strLeftSource.c_str());
        remove(m_strRightSource.c_str());

        if (nLeftSize || nRightSize)
        {
            // somthing wrong happend
            SorterException::generateException(et_Merge);
        }
    }

    static void setMergeFileName(const string &strFName)
    {
        m_strResultOfMergeFileName = strFName;
    }
};
// static members
string Merger::m_strResultOfMergeFileName;
bool Merger::m_bStopMerge = false;
bool Merger::m_bImmediateExit = false;
vector<string> Merger::m_vPool2Put(0);
vector<string> Merger::m_vPool2Take(0);
size_t Merger::m_nMergeIndex = 0;
errType Merger::m_error = et_NoError;
//---------------------------------------------------------------------------------------------------

/**
* class ChunkSorter
* Read and sort input data from file by small chunks
* After chunk is sorted it is flushed on disk for merge
*/

class ChunkSorter
{
    static bool m_bNoMoreData;
    static bool m_bAbortExit;
    static vector<string> *m_pChunkNamePool; //!< is to be set before using ChunkSorter
    static errType m_error;
    vector<int> m_vChunkToSort;

    ChunkSorter(const ChunkSorter&);
    ChunkSorter& operator = (const ChunkSorter&);

    void sortAndFlush()
    {
        while (1)
        {
            EnterCriticalSection(&csFileAccess);

            size_t nReadElems = m_fileToSort.read(&m_vChunkToSort[0], sizeof(int), nMemForThread / sizeof(int));

            if (feof(m_fileToSort.file))
            {
                m_bNoMoreData = true;
            }

            LeaveCriticalSection(&csFileAccess);

            if (nReadElems)
            {
                // sort data
                qsort(&m_vChunkToSort[0], m_vChunkToSort.size(), sizeof(int), compare);

                EnterCriticalSection(&csChunkName);
                string strChunkFileName = generateNextChunkFileName();
                LeaveCriticalSection(&csChunkName);

                // flush sorted data
                FileWrapper f;
                f.open(strChunkFileName.c_str(), "wb");
                f.write(&m_vChunkToSort[0], sizeof(int), m_vChunkToSort.size());
                f.close();

                // critical section
                EnterCriticalSection(&csMergePool);
                m_pChunkNamePool->push_back(strChunkFileName);
                LeaveCriticalSection(&csMergePool);
            }

            if (m_bNoMoreData || m_bAbortExit)
            {
                // end up with sorting
                m_vChunkToSort.clear();
                break;
            }
        }
    }

public:

    static FileWrapper m_fileToSort;
    static long long m_nFileSize;

    ChunkSorter() :
    m_vChunkToSort(nMemForThread / sizeof(int))
    {
    }

    static void setError(errType error)
    {
        m_error = error;
    }

    static errType getError()
    {
        return m_error;
    }

    static unsigned __stdcall thread_func(void* arg)
    {
        ChunkSorter *sorter = static_cast<ChunkSorter*>(arg);

        try
        {
            if (!m_pChunkNamePool)
            {
                SorterException::generateException(et_InputData);
            }

            sorter->sortAndFlush();
        }
        catch(SorterException &ex)
        {
            EnterCriticalSection(&csSortError);

            if (ex.getErrorType() != et_NoError)
            {
                ChunkSorter::setError(ex.getErrorType());
            }

            m_bAbortExit = true;
            LeaveCriticalSection(&csSortError);
        }
        catch (...)
        {
            EnterCriticalSection(&csSortError);
            ChunkSorter::setError(et_Unexpected);
            m_bAbortExit = true;
            LeaveCriticalSection(&csSortError);
        }

        _endthreadex(0);
        return 0;
    }

    static void setFileToSort(const char *pchFileName)
    {
        m_fileToSort.open(pchFileName, "rb");
        m_nFileSize = m_fileToSort.getSize();
    }

    // method is to be called to check input data size after file is set
    static void checkInputDataSize()
    {
        if (m_nFileSize == 0 || m_nFileSize % sizeof(int))
        {
            SorterException::generateException(et_InputData);
        }
    }

    static void setNamePool(vector<string> *vNamePool)
    {
        m_pChunkNamePool = vNamePool;
    }
};

// static members
FileWrapper ChunkSorter::m_fileToSort;
long long ChunkSorter::m_nFileSize = 0;
bool ChunkSorter::m_bNoMoreData = false;
bool ChunkSorter::m_bAbortExit = false;
errType ChunkSorter::m_error = et_NoError;
vector<string>* ChunkSorter::m_pChunkNamePool = NULL;
//---------------------------------------------------------------------------------------------------


int main (int argc, char* argv[])
{
    try
    {
        if (argc < 2)
        {
            SorterException::generateException(et_InputData);
        }

        string strInputFileName = argv[1];
        string strOutputFileName = argv[2];

        ChunkSorter::setFileToSort(strInputFileName.c_str());
        ChunkSorter::checkInputDataSize();

        if (ChunkSorter::m_nFileSize <= MEMORY_LIMIT)
        {
            // data size is small enough to be allocated and sorted in memory
            vector<int> vDataToSort(static_cast<size_t>(ChunkSorter::m_nFileSize / sizeof(int)));

            ChunkSorter::m_fileToSort.read(&vDataToSort[0], sizeof(int), vDataToSort.capacity());

            qsort(&vDataToSort[0], vDataToSort.size(), sizeof(int), compare);

            FileWrapper merge;
            merge.open(strOutputFileName.c_str(), "wb");
            merge.write(&vDataToSort[0], sizeof(int), vDataToSort.size());
        }
        else
        {
            ChunkSorter::setNamePool(&Merger::m_vPool2Take);
            vector<char> vTime(10);
            _strtime_s(&vTime[0], vTime.size());
            cout << "start time: " << &vTime[0] << endl;

            InitializeCriticalSection(&csMergePool);
            InitializeCriticalSection(&csFileAccess);
            InitializeCriticalSection(&csChunkName);
            InitializeCriticalSection(&csMergeError);
            InitializeCriticalSection(&csSortError);

            //--------------------------------chunk_sorter initialization------------------------------------
            ChunkSorter SortAgent1;
            ChunkSorter SortAgent2;
            ChunkSorter SortAgent3;
            ChunkSorter SortAgent4;

            unsigned Agent1ThreadID;
            unsigned Agent2ThreadID;
            unsigned Agent3ThreadID;
            unsigned Agent4ThreadID;

            HANDLE hSortThreads[4];
            hSortThreads[0] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent1, 0, &Agent1ThreadID);
            hSortThreads[1] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent2, 0, &Agent2ThreadID);
            hSortThreads[2] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent3, 0, &Agent3ThreadID);
            hSortThreads[3] = (HANDLE)_beginthreadex(NULL, 0, &ChunkSorter::thread_func, &SortAgent4, 0, &Agent4ThreadID);

            //------------------------------------merger initialization---------------------------------------
            Merger::setMergeFileName(strOutputFileName);
            Merger mergerAgent1;
            Merger mergerAgent2;

            unsigned mergerThredID1;
            unsigned mergerThredID2;

            HANDLE hMergeThreads[2];
            hMergeThreads[0] = (HANDLE)_beginthreadex(NULL, 0, &Merger::mergeInThread, &mergerAgent1, CREATE_SUSPENDED, &mergerThredID1);
            hMergeThreads[1] = (HANDLE)_beginthreadex(NULL, 0, &Merger::mergeInThread, &mergerAgent2, CREATE_SUSPENDED, &mergerThredID2);
            //-----------------------------------------------------------------------------------------------

            // wait until sorters finish their job and then resume threads to merge
            WaitForMultipleObjectsEx(4, hSortThreads, true, INFINITE, false);

            CloseHandle(hSortThreads[3]);
            CloseHandle(hSortThreads[2]);
            CloseHandle(hSortThreads[1]);
            CloseHandle(hSortThreads[0]);

            // check sort error here
            bool bSortError = false;

            if (ChunkSorter::getError() != et_NoError)
            {
                Merger::setImmediateExit(true);
                bSortError = true;
            }

            _strtime_s(&vTime[0], vTime.size());
            cout << "resume threads time: " << &vTime[0] << endl;

            ResumeThread(hMergeThreads[0]);
            ResumeThread(hMergeThreads[1]);

            // wait untill all threads finish
            WaitForMultipleObjectsEx(2, hMergeThreads, true, INFINITE, false);

            CloseHandle(hMergeThreads[1]);
            CloseHandle(hMergeThreads[0]);

            DeleteCriticalSection(&csChunkName);
            DeleteCriticalSection(&csFileAccess);
            DeleteCriticalSection(&csMergePool);

            DeleteCriticalSection(&csMergeError);
            DeleteCriticalSection(&csSortError);

            if (bSortError)
            {
                SorterException::generateException(ChunkSorter::getError());
            }

            // check merge error here
            if (Merger::getError() != et_NoError)
            {
                SorterException::generateException(Merger::getError());
            }

            _strtime_s(&vTime[0], vTime.size());
            cout << "end time: " << &vTime[0] << endl;
        }
    }

    catch (SorterException &ex)
    {
        cout << ex;
    }

    catch (...)
    {
        cerr << "Unknown error" << endl;
    }

    return 0;
}

Здравствуйте, letsurf, Вы писали:

L>Пусть у нас на диске есть файл размером 4 гигабайта. Его можно представить, как 2^30 32-битных беззнаковых чисел. Нужно отсортировать этот файл. То есть программа должна сгенерировать еще один файл размером в 4 гигабайта, в котором содержатся те же числа, что и в исходном, но упорядочены по возрастанию. Стандартные алгоритмы сортировки (qsort, std::sort) напрямую применить невозможно, так как для их выполнения нужно как минимум 4 гигабайта оперативной памяти. Но отсортировать числа на диске возможно, если использовать дополнительное пространство на жестком диске.

L>Нужно написать консольную программу, которая в argv[1] получает имя файла, который нужно отсортировать (размер файла до 16Gb, размер кратен четырем),

У тебя есть файл с 32-битными числами.
Таких чисел может быть up to 4 млрд.
Нужно вычислить индексы вхождения каждого числа в исходный файл.
Индексы вхождения можно записать в промежуточный файл.
Промежуточный файл размером 16Gb, где в каждые 4b записывается индекс вхождения соответствующего числа в исходный файл.
Потом на основе индексов вхождения собираем файл результата.

Здравствуйте, letsurf, Вы писали:

L>Доброго времени суток!

L>Недавно столкнулся с задачей:

L>(привожу задание как оно есть)

L>Пусть у нас на диске есть файл размером 4 гигабайта. Его можно представить, как 2^30 32-битных беззнаковых чисел. Нужно отсортировать этот файл. То есть программа должна сгенерировать еще один файл размером в 4 гигабайта, в котором содержатся те же числа, что и в исходном, но упорядочены по возрастанию. Стандартные алгоритмы сортировки (qsort, std::sort) напрямую применить невозможно, так как для их выполнения нужно как минимум 4 гигабайта оперативной памяти. Но отсортировать числа на диске возможно, если использовать дополнительное пространство на жестком диске.

L>Нужно написать консольную программу, которая в argv[1] получает имя файла, который нужно отсортировать (размер файла до 16Gb, размер кратен четырем),
L>в argv[2] — имя файла, в который нужно записать отсортированные данные.

L>Ограничения:
L>1. Программа не может рассчитывать, что возможно выделить более 256Mb памяти.
L>2. Программа должна эффективно использовать несколько ядер.
Запустить несколько процессов, которые будут читать из исходного файла отдельные непересекающиеся фрагменты. Размер фрагмента — 256Mb, так что получается 16 процессов. Прочитав фрагмент целиком в память напускаем на него qsort. Получаем отсортированный фрагмент исходного файла, сохраняем его на диск. Получаем 16 отсортированных фрагментов исходного файла.
Теперь остается "слить" промежуточные файлы в один — сортировка слиянием, тут проблем нет. Можно запустить один процесс, который будет сливать все 16 фрагментов, а можно в несколько этапов слияние делать. Например, запустить 8 процессов слияния, которые из 16 фрагментов сдалют 8 по 512Mb, потом из 8 — 4 и т.д. На этом этапе все упрется в производительность файловой системы — из фрагментов надо будет читать по 32бита.

Здравствуйте, alsemm, Вы писали:

A>Здравствуйте, letsurf, Вы писали:

A>Запустить несколько процессов, которые будут читать из исходного файла отдельные непересекающиеся фрагменты. Размер фрагмента — 256Mb, так что получается 16 процессов. Прочитав фрагмент целиком в память напускаем на него qsort. Получаем отсортированный фрагмент исходного файла, сохраняем его на диск. Получаем 16 отсортированных фрагментов исходного файла.
A>Теперь остается "слить" промежуточные файлы в один — сортировка слиянием, тут проблем нет. Можно запустить один процесс, который будет сливать все 16 фрагментов, а можно в несколько этапов слияние делать. Например, запустить 8 процессов слияния, которые из 16 фрагментов сдалют 8 по 512Mb, потом из 8 — 4 и т.д. На этом этапе все упрется в производительность файловой системы — из фрагментов надо будет читать по 32бита.

Примерно так я и сделал, но, видимо, это не самый быстрый способ.

Скорее всего сортировку подсчетом было бы быстрее, но не факт.

Здравствуйте, letsurf, Вы писали:

L>Здравствуйте, alsemm, Вы писали:

A>>Здравствуйте, letsurf, Вы писали:

A>>Запустить несколько процессов, которые будут читать из исходного файла отдельные непересекающиеся фрагменты. Размер фрагмента — 256Mb, так что получается 16 процессов. Прочитав фрагмент целиком в память напускаем на него qsort. Получаем отсортированный фрагмент исходного файла, сохраняем его на диск. Получаем 16 отсортированных фрагментов исходного файла.
A>>Теперь остается "слить" промежуточные файлы в один — сортировка слиянием, тут проблем нет. Можно запустить один процесс, который будет сливать все 16 фрагментов, а можно в несколько этапов слияние делать. Например, запустить 8 процессов слияния, которые из 16 фрагментов сдалют 8 по 512Mb, потом из 8 — 4 и т.д. На этом этапе все упрется в производительность файловой системы — из фрагментов надо будет читать по 32бита.

L>Примерно так я и сделал, но, видимо, это не самый быстрый способ.
Это не способ небыстрый, а реализация кривая. Потому что если открыть исходный файл только один раз (static FileWrapper m_fileToSort), т.е. создать единственный дескриптор FILE*, то читать через него асинхронно из нескольких потоков очень тяжело.
Мне вообще непонятно, что у тебя в коде происходит. Т.е. может оно конечно и выдает корректный результат, но асинхронностью тут и не пахнет.

Здравствуйте, alsemm, Вы писали:

L>>Примерно так я и сделал, но, видимо, это не самый быстрый способ.
A>Это не способ небыстрый, а реализация кривая. Потому что если открыть исходный файл только один раз (static FileWrapper m_fileToSort), т.е. создать единственный дескриптор FILE*, то читать через него асинхронно из нескольких потоков очень тяжело.
A>Мне вообще непонятно, что у тебя в коде происходит. Т.е. может оно конечно и выдает корректный результат, но асинхронностью тут и не пахнет.

Исходный файл читается цепочками по 64Mb, которые сортируются qsort'ом и складываются на диск. Этот процесс не занимает много времени, основное же время у меня уходит на слияние этих цепочек.

Здравствуйте, rus blood, Вы писали:

RB>Нужно вычислить индексы вхождения каждого числа в исходный файл.

Индексы-то зачем? Нужны счётчики, раз это сортировка подсчётом.

Здравствуйте, Кодт, Вы писали:

К>Индексы-то зачем? Нужны счётчики, раз это сортировка подсчётом.
Гы. Произвольный доступ убъет всю производительность.

Здравствуйте, letsurf, Вы писали:

L>Ограничения:
L>1. Программа не может рассчитывать, что возможно выделить более 256Mb памяти.
L>2. Программа должна эффективно использовать несколько ядер.

Просто как идея, не знаю на сколько будет (не)эффективно: cчитывать числа из исходного файла и записывать в n файлов, каждый их которых может уместить m 32-битных беззнаковых чисел. Таким образом в первый файл попадут числа из диапозона (0;m — 1), во второй (m;2*m — 1) и т.д. Теперь считываем в цикле содержимое мелких файлов, сортируем и пишем в выходной файл.

из очевидных недостатков это конечно крайне низний технологический уровень
-есть такая штука как stxxl — это как раз и есть stl для объемов данных которые в память не влезают, но вы ей не воспользовались
-RAII презрет, зачем эти голые CRITICAL_SECTION Enter\Leave, что нельзя было обертками воспользоваться из ATL или буста?
-файлы зачем-то читаются в память — нафиг это нужно? с файл мэппингом было бы гораздо проще
-зачем-то вручную создаются потоки — опять дикость. Гораздо лучше было бы воспользоваться ОМP

ошибок достаточно много, даже если вы и решили задачу правильно слив засчитан

Здравствуйте, rm822, Вы писали:

R>из очевидных недостатков это конечно крайне низний технологический уровень
R>-есть такая штука как stxxl — это как раз и есть stl для объемов данных которые в память не влезают, но вы ей не воспользовались
R>-RAII презрет, зачем эти голые CRITICAL_SECTION Enter\Leave, что нельзя было обертками воспользоваться из ATL или буста?
R>-файлы зачем-то читаются в память — нафиг это нужно? с файл мэппингом было бы гораздо проще
R>-зачем-то вручную создаются потоки — опять дикость. Гораздо лучше было бы воспользоваться ОМP

R>ошибок достаточно много, даже если вы и решили задачу правильно слив засчитан

все по существу, спасибо.

Привет!
Преждевременная оптимизация и здесь тоже зло. Первым делом избавься от идеи многопоточности для первого варианта решения задачи. Очень легко наступить на грабли, получив два параллельных потока, работающих одновременно с диском.
К чему четыре потока чанксортеров, если они читают из файла под критической секцией? Только чтобы параллельно запустить сортировку в другом, который уже свой чанк прочитал? Это можно было бы сделать более прямым способом, да и не нужно, по крайне мере в первой версии — сортировка чанка вряд ли занимает существенное время, чтобы её сразу делать параллельно.
Чанксортер читает под критической секцией, потом сортирует и пишет в файл без оной — писать может параллельно с чтением другим чанксортером — io деградирует?
Не до конца понял, как происходит merge, но по-моему так: 2 куска читаются в память, затем сливаются std::merge-ом в память, затем сбрасываются на диск. Во-первых, читать в память такими кусками ни к чему, можно сливать по мере чтения. Тут можно сделать что-то вроде istream_iterator или переписать руками без использования merge. Можно, конечно, и читать в память кусками и сливать std::merge-ом по-старому, но не выделять память под массивы левой, правой и результирующей части в mergeTwoFilesIntoOne при каждом слиянии двух файлов! Ну и во-вторых, не стоит сливать файлы всего по два, тем более в двух merger-ах, которые параллельно насилуют диск и перегонять бедные int-ы по несколько раз из файлов в файл. Нужно слить сразу все чанки — смотри в сторону std::priority_queue.

должно быть написано не "как 2^30 32-битных беззнаковых чисел" а "как 2^32 32-битных беззнаковых чисел"
или где неточность?

Здравствуйте, Andruh, Вы писали:

A>К чему четыре потока чанксортеров, если они читают из файла под критической секцией? Только чтобы параллельно запустить сортировку в другом, который уже свой чанк прочитал?

Да, для этого.

A>Не до конца понял, как происходит merge, но по-моему так: 2 куска читаются в память, затем сливаются std::merge-ом в память, затем сбрасываются на диск.

все верно, именно это он и делает.

A>Можно, конечно, и читать в память кусками и сливать std::merge-ом по-старому, но не выделять память под массивы левой, правой и результирующей части в mergeTwoFilesIntoOne при каждом слиянии двух файлов!

ага, на это уже сам обратил внимание.

A>Ну и во-вторых, не стоит сливать файлы всего по два, тем более в двух merger-ах, которые параллельно насилуют диск и перегонять бедные int-ы по несколько раз из файлов в файл. Нужно слить сразу все чанки — смотри в сторону std::priority_queue.

спасибо.

еще неточность "Пусть у нас на диске есть файл размером 4 гигабайта" должно быть "Пусть у нас на диске есть файл размером от 4 гигабайт"?

Здравствуйте, vitali_y, Вы писали:

_>должно быть написано не "как 2^30 32-битных беззнаковых чисел" а "как 2^32 32-битных беззнаковых чисел"
_>или где неточность?

нет, все верное написано, т.е. 2^30 32-битных числа занимают как раз 4Г

>>нет, все верное написано, т.е. 2^30 32-битных числа занимают как раз 4Г

да верно

Здравствуйте, vitali_y, Вы писали:

_>еще неточность "Пусть у нас на диске есть файл размером 4 гигабайта" должно быть "Пусть у нас на диске есть файл размером от 4 гигабайт"?

по-русски, я думаю, правильно будет "4 гигабайта"

ну размер от 4 до 16GB — тогда переполнения при сортировке подсчетом не будет — точнее DWORD хватит — идея скорее всего тут такая.
сколько ваш алгоритм работает — как понимаю 16 GB файл читается с диска примерно 10 минут,
т.е. чисто эмпирически — программа которая тратит максимум минут 20 — рабочая программа — если значительно больше — выбран неправильный способ.
могу ошибаться...

_>>еще неточность "Пусть у нас на диске есть файл размером 4 гигабайта" должно быть "Пусть у нас на диске есть файл размером от 4 гигабайт"?
L>по-русски, я думаю, правильно будет "4 гигабайта"

имел ввиду "от 4" — вообщем это не важно — просто мне нравятся однозначные формулировки, что бы лишних вопросов не возникало.

Здравствуйте, WolfHound, Вы писали:

К>>Индексы-то зачем? Нужны счётчики, раз это сортировка подсчётом.
WH>Гы. Произвольный доступ убъет всю производительность.

Конечно, убьёт.
Всё равно, индексы-то зачем?
Устраивать 4 миллиарда забегов по исходному файлу, что ли?

ну и идея практической реализации:
мы ограничены 256MB — т.е. для подсчета используем некий хеш (хеш-таблицу),
где наименее востребованные во времени данные сбрасываем на диск либо упаковываем — тут нужно подумать — чтобы разные крайние случаи обрабатывались эфФективно —
т.е. случай 16GB и все числа встречаются по 1 разу — ну и близкие...

"2. Программа должна эффективно использовать несколько ядер."
— ну а этим тебя запутали.

К>>Индексы-то зачем? Нужны счётчики, раз это сортировка подсчётом.
WH>Гы. Произвольный доступ убъет всю производительность.

вы о чем?

Здравствуйте, Кодт, Вы писали:

К>Здравствуйте, WolfHound, Вы писали:

К>>>Индексы-то зачем? Нужны счётчики, раз это сортировка подсчётом.
WH>>Гы. Произвольный доступ убъет всю производительность.

К>Конечно, убьёт.

Не убъет если подсчитывать за проход 256к индексов (в памяти).

Здравствуйте, letsurf, Вы писали:

L>Доброго времени суток!

L>Недавно столкнулся с задачей:

L>(привожу задание как оно есть)

L>Пусть у нас на диске есть файл размером 4 гигабайта. Его можно представить, как 2^30 32-битных беззнаковых чисел. Нужно отсортировать этот файл. То есть программа должна сгенерировать еще один файл размером в 4 гигабайта, в котором содержатся те же числа, что и в исходном, но упорядочены по возрастанию. Стандартные алгоритмы сортировки (qsort, std::sort) напрямую применить невозможно, так как для их выполнения нужно как минимум 4 гигабайта оперативной памяти. Но отсортировать числа на диске возможно, если использовать дополнительное пространство на жестком диске.

это задачка на radix sort

I>это задачка на radix sort
вряд ли.

Здравствуйте, letsurf, Вы писали:

Предлагаю следующее решение. Но оно будет платформозависимым.

Открываем файлмэппинг.
Фаза 1
Файл делим на некие куски, размер куска кратен странице памяти, надо подобрать наилучший размер. Эти куски сортируем независимо друг от друга, открывая на кусок view и сортируя его, прямо-таки в самом файле, конечно, испортив его. Получаем набор отсортированных кусков. Сортировку можно провести многопоточно.
Фаза 2
Закрываем все view. Открываем новые view, каждый — на начало своего куска, размер ViewSize = 4 Кб (может, можно и больше, надо подсчитать). Имеем, т.о. массив указателей на начала кусков.

Проходим по этому массиву, ищем минимальный элемент по указателю. Найдя — заносим его в результат, продвигаем указатель во view, а если при этом выходим за размер view, то переоткрываем view на следующий ViewSize. В общем, слияние упорядоченых последовательностей.

Очень быстро не будет, но и не 4 часа, я думаю. Доступ везде последовательный в Фазе 2. По памяти — можно уложить в 256 Мб (и даже меньше) , если правильно подобрать размеры.

PD>Проходим по этому массиву, ищем минимальный элемент по указателю. Найдя — заносим его в результат, продвигаем указатель во view, а если при этом выходим за размер view, то переоткрываем view на следующий ViewSize. В общем, слияние упорядоченых последовательностей.

Тут можно и похитрее. Проходим по этому массиву, находим все минимальные элементы (заносим в некий дополнительный список/массив номера view, где они встречались), после чего заносим в выходной файл это число столько раз, сколько оно встретилось, после чего продвигаем указатели во всех view, где оно встретилось). Так можно сократить число проходов.

Здравствуйте, rm822, Вы писали:


R>-зачем-то вручную создаются потоки — опять дикость. Гораздо лучше было бы воспользоваться ОМP

И плодить страхокод с прагмами? Уж лучше православный С++-ный TBB.

SC>И плодить страхокод с прагмами?

сомнительное замечание. Прагмы тем и хороши, что можно их убрать и должен остаться корректный код.

Здравствуйте, dilmah, Вы писали:


SC>>И плодить страхокод с прагмами?

D>сомнительное замечание. Прагмы тем и хороши, что можно их убрать и должен остаться корректный код.

Хм, вообще, далеко не любую прагму можно убрать, я бы не поставил на работоспособность кода, из которого убрали #pragma pack к примеру.
Ну да ладно, а зачем вообще прагмы посреди кода? Да еще такие загадочные, типа

#pragma omp parallel for private(w) reduction(+:sum) schedule(static,1)

Task-based модель Tbb гораздо дружественее к С++, не требует от компилятора поддержки всякого нестандартного и дает более изящный код. Не говоря уж о о том, что пул потоков Tbb работает пошустрее, чем Omp в реализации MSCV(по крайней мере, на тот момент когда я в последний раз пользовался Omp). Наличие исходников Tbb тоже помогает, c openMP иногда просто неожиданно начинается непонятно что и ты никак не можешь узнать что именно. Я один раз целый день убил, пытаясь понять почему у меня все потоки имеют идентификатор ноль. Оказалось, MKL развлекается.

Короче, не-не-не, я однозначно отказался от OpenMP в пользу Tbb и ничуть об этом не жалею.

SC>И плодить страхокод с прагмами? Уж лучше православный С++-ный TBB.
Мне на ваши религиозные убеждения пофиг, у меня и у коллег никаких проблем с прагмами не было

I>>это задачка на radix sort
_>вряд ли.

да это классическая задача на radix sort, ну и сортировка подсчетом тут используется

Здравствуйте, vitali_y, Вы писали:

I>>>это задачка на radix sort
_>>вряд ли.

_>да это классическая задача на radix sort, ну и сортировка подсчетом тут используется

спасибо, в эту сторону я и не смотрел.

Здравствуйте, letsurf, Вы писали:

L>Доброго времени суток!

L>Недавно столкнулся с задачей:

L>(привожу задание как оно есть)

L>Пусть у нас на диске есть файл размером 4 гигабайта. Его можно представить, как 2^30 32-битных беззнаковых чисел. Нужно отсортировать этот файл. То есть программа должна сгенерировать еще один файл размером в 4 гигабайта, в котором содержатся те же числа, что и в исходном, но упорядочены по возрастанию. Стандартные алгоритмы сортировки (qsort, std::sort) напрямую применить невозможно, так как для их выполнения нужно как минимум 4 гигабайта оперативной памяти. Но отсортировать числа на диске возможно, если использовать дополнительное пространство на жестком диске.

Пробовал делать это задание
-Radix sort рулит, но требует столько же дополнительной памяти, сколько сортируемый массив => 1 chunk = 256мб/2
-merge из chunk'ов можно делать c с помощью кучи aka priority_queue
-тормозит общение с диском. fread/fwrite из MSVC с мьютексом на каждый вызов необходимо заменит на велосипед.
На реализации ассинхронности и многопоточности без использования левых библиотек я сломался

D14>Пробовал делать это задание
D14>-Radix sort рулит, но требует столько же дополнительной памяти, сколько сортируемый массив => 1 chunk = 256мб/2
D14>-merge из chunk'ов можно делать c с помощью кучи aka priority_queue
D14>-тормозит общение с диском. fread/fwrite из MSVC с мьютексом на каждый вызов необходимо заменит на велосипед.
D14>На реализации ассинхронности и многопоточности без использования левых библиотек я сломался

похоже что ты неправильно понял как пользовать числовую сортировку,
ну и как всегда не внял правилу, что преждевременная оптимизация зло.
такие задачи решаются на листике бумаги — потом проверяется реализация.
по-рабоче-крестьянски — смысл такой:
1) понадобится дополнительное место на диске объемом в исходный файл;
2) выбирается основание системы счисления — пусть N — чтобы хватило отведенных 256MB оперативки;
3) сортировка делается в несколько (точнее в N) проходов
проход 1:
a) пробежался по файлу — вычитал все младшие разряды в ОП (в отведенные 256MB)
b) отсортировал получившийся массив с помощью сортировки подсчетом;
c) выписал отсортированные по младшему разряду числа во вспомогательный файл;
проход 2:
a) пробежался по вспомогательному файлу — вычитал второй разряд в ОП (в отведенные 256MB)
b) отсортировал получившийся массив с помощью сортировки подсчетом;
c) выписал отсортированные по второму разряду числа во 1 файл;
...
в этом алгоритме в данной реализации — узкое место это диск,
какой бы многоядерный не был у тебя процессор — тут ты существенного ускорения не получишь —
тут в задаче "заковырка" — для любителей многопроцессорности.
хотя этап b) можно распараллелить — если бороться за каждую секунду.

держись товарищ не ломайся
большая просьба, раз ты озаботился реализацией, если все же есть желание убедиться, что это работает —
а это работает именно так — задача то классическая —
поделись исходником со всеми

Здравствуйте, letsurf, Вы писали:

L>Здравствуйте, alsemm, Вы писали:

L>>>Примерно так я и сделал, но, видимо, это не самый быстрый способ.
A>>Это не способ небыстрый, а реализация кривая. Потому что если открыть исходный файл только один раз (static FileWrapper m_fileToSort), т.е. создать единственный дескриптор FILE*, то читать через него асинхронно из нескольких потоков очень тяжело.
A>>Мне вообще непонятно, что у тебя в коде происходит. Т.е. может оно конечно и выдает корректный результат, но асинхронностью тут и не пахнет.

L>Исходный файл читается цепочками по 64Mb, которые сортируются qsort'ом и складываются на диск. Этот процесс не занимает много времени, основное же время у меня уходит на слияние этих цепочек.
Стало мне любопытно и я написал свое решение задачки.
Исходный файл 1.9Gb сортируется за 10 минут.
Решение тупое до безобразия. Закодировать пришлось 2-а приложения:
— первое читает из заданного файла заданный кусок, сортирует его и сохраняет отсортированный кусок в отдельный файл;
— второе получает список остортированных файлов и сливает их в один.
Еще есть четыре bash скрипта, которые запускают эти приложения.
Все барахло лежит тут — http://files.rsdn.ru/30512/sorter.zip. В архиве скрипты, msvc60 dsp/dsw чтобы собрать приложения и собранные приложения. Для запуска нужен cygwin, ну или можно под Linux собрать и гонять там.
Использовать:
1. распаковать архив
2. запустить sort.sh <путь_к_исходному_файлу> <путь_к_отсортированному_файлу>

Как работает (в общих чертах).
Запускаем асинхронно N процессов. Каждому процессу задается фрагмент исходного файла. Каждый процесс читает выделенный ему фрагмент по частям и сортирует их. Каждая отсортированная часть записывается в отдельный временный файл. Далее запускаем один процесс, которые сортирует слиянием все остортированные фрагменты.
Выяснилось, что оптимальное значение N=<число процессоров>. А дальше нужно ловить баланс между количеством фрагментов для слияния и размером этих фрагментов. Эти параметры заданы в файле config.sh. Можно с ним поиграться.

Здравствуйте, vitali_y, Вы писали:

_>1) понадобится дополнительное место на диске объемом в исходный файл;
_>2) выбирается основание системы счисления — пусть N — чтобы хватило отведенных 256MB оперативки;
_>3) сортировка делается в несколько (точнее в N) проходов
_>проход 1:
_>a) пробежался по файлу — вычитал все младшие разряды в ОП (в отведенные 256MB)
_>b) отсортировал получившийся массив с помощью сортировки подсчетом;
_>c) выписал отсортированные по младшему разряду числа во вспомогательный файл;
_>проход 2:
_>a) пробежался по вспомогательному файлу — вычитал второй разряд в ОП (в отведенные 256MB)
_>b) отсортировал получившийся массив с помощью сортировки подсчетом;
_>c) выписал отсортированные по второму разряду числа во 1 файл;

Ну, идея-то понятна. Вместо двух чанков в памяти использовать вдвое больший чанк в памяти и такой же на диске. Концептуально ничего нового по сравнению с лобовым методом, к тому же, не понятно, как к такому решению можно прийти логически — выгрузить один чанк на диск, а не эмпирически.


_>в этом алгоритме в данной реализации — узкое место это диск,
_>какой бы многоядерный не был у тебя процессор — тут ты существенного ускорения не получишь —
_>тут в задаче "заковырка" — для любителей многопроцессорности.
_>хотя этап b) можно распараллелить — если бороться за каждую секунду.

Перманентная загрузка проца >0 — было условием задачи. К тому, же я не уверен, ускорения не получишь. Думаю, при наличии нескольких физических дисков все ровно наоборот. Академические решения шуршат по паре-тройке десятков гб/мин (сначала хотел написать сек., но видимо ошибаюсь) — по-моему на такие цифры я натолкнулся после 15 минут гугления.

A>Запускаем асинхронно N процессов. Каждому процессу задается фрагмент исходного файла. Каждый процесс читает выделенный ему фрагмент по частям и сортирует их.

Параллельное чтение из файла, к сожалению, плохая идея — проверял, тормозит-с. М.б. железо дерьмовое было

Здравствуйте, D14, Вы писали:


A>>Запускаем асинхронно N процессов. Каждому процессу задается фрагмент исходного файла. Каждый процесс читает выделенный ему фрагмент по частям и сортирует их.

D14>Параллельное чтение из файла, к сожалению, плохая идея — проверял, тормозит-с. М.б. железо дерьмовое было
Мои замеры говорят об обратном. Железо — Samsung R60, ОС — Vista.
Исходный файл ~ 140Mb.
Размер файла на который бъется исходный для сортировки — 40Mb. 40Mb выбрано с тем расчетом, чтобы исходный файл бился на равное количество фрагментов независимо о того, сколько процессов его читают одновременно.
Запускам 2 асинхронных процесса по чтению исходного файла и сортировки его частей. Получается такой config.sh:

max_async_sorters=2
...
chunk_bytes=$((40 * 1024 * 1024))

Измеряем время всего процесса от начала сортировки фрагментов до их слияния и выдачи отсортированного файла. Повторяем сортировку 4 раза, чтобы исключить погрешности от работы свопа и т.п.:

stol@computer /cygdrive/c/Users/stol/development/sorter
$ ls -l data
-rwx------ 1 stol None 140083200 Jun 17 12:20 data

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_580_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_580_chunk_111984640-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_580_chunk_41943040-70041600 /cygdrive/c/Users/stol/AppData/Local/Temp/data_580_chunk_70041600-111984640...
merge OK

real    0m14.051s
user    0m0.530s
sys     0m0.930s

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_14620_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_14620_chunk_111984640-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_14620_chunk_41943040-70041600 /cygdrive/c/Users/stol/AppData/Local/Temp/data_14620_chunk_70041600-111984640...
merge OK

real    0m14.362s
user    0m0.302s
sys     0m0.897s

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_17688_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_17688_chunk_111984640-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_17688_chunk_41943040-70041600 /cygdrive/c/Users/stol/AppData/Local/Temp/data_17688_chunk_70041600-111984640...
merge OK

real    0m15.479s
user    0m0.286s
sys     0m1.006s

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_18380_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_18380_chunk_111984640-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_18380_chunk_41943040-70041600 /cygdrive/c/Users/stol/AppData/Local/Temp/data_18380_chunk_70041600-111984640...
merge OK

real    0m14.285s
user    0m0.286s
sys     0m1.006s

Теперь убираем асинхронность в чтении исходного файла config.sh:

max_async_sorters=1

Запускаем 4-е сортировки, как и в первом случае:

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_14800_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_14800_chunk_125829120-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_14800_chunk_41943040-83886080 /cygdrive/c/Users/stol/AppData/Local/Temp/data_14800_chunk_83886080-125829120...
merge OK

real    0m18.709s
user    0m0.135s
sys     0m0.995s

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_16972_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_16972_chunk_125829120-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_16972_chunk_41943040-83886080 /cygdrive/c/Users/stol/AppData/Local/Temp/data_16972_chunk_83886080-125829120...
merge OK

real    0m17.487s
user    0m0.332s
sys     0m0.994s

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_17512_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_17512_chunk_125829120-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_17512_chunk_41943040-83886080 /cygdrive/c/Users/stol/AppData/Local/Temp/data_17512_chunk_83886080-125829120...
merge OK

real    0m18.175s
user    0m0.287s
sys     0m0.822s

stol@computer /cygdrive/c/Users/stol/development/sorter
$ time ./sort.sh data data.sorted
merging /cygdrive/c/Users/stol/AppData/Local/Temp/data_20052_chunk_0-41943040 /cygdrive/c/Users/stol/AppData/Local/Temp/data_20052_chunk_125829120-140083200 /cygdrive/c/Users/stol/AppData/Local/Temp/data_20052_chunk_41943040-83886080 /cygdrive/c/Users/stol/AppData/Local/Temp/data_20052_chunk_83886080-125829120...
merge OK

real    0m18.769s
user    0m0.380s
sys     0m0.884s

Имееющий глаза, да увидит

Такой алгоритм отлично подходит, например, для запуска его на кластере, позволяя загрузить все доступное железо.

Слабым местом в моем решении считаю финальную стадию слияния фрагментов в один файл. Но и тут можно ускорить за счет асинхронности.
Для этого запускаем M процессов, каждый из скоторых "собирает" числа в определенном диапазоне. Первый — [0,1000000), второй — [1000000, 2000000) и т.д. до [UINT_MAX-..., UINT_MAX)
Каждый процесс имеет доступ ко всем отсортированным фрагментам исходного файла. Ну и читает из них только те числа, которые попадают в его диапазон.
Получаем на вызоде M файлов, которые сливаем в один простым cat-ом.

_>>1) понадобится дополнительное место на диске объемом в исходный файл;
_>>2) выбирается основание системы счисления — пусть N — чтобы хватило отведенных 256MB оперативки;
_>>3) сортировка делается в несколько (точнее в N) проходов
_>>проход 1:
_>>a) пробежался по файлу — вычитал все младшие разряды в ОП (в отведенные 256MB)
_>>b) отсортировал получившийся массив с помощью сортировки подсчетом;
_>>c) выписал отсортированные по младшему разряду числа во вспомогательный файл;
_>>проход 2:
_>>a) пробежался по вспомогательному файлу — вычитал второй разряд в ОП (в отведенные 256MB)
_>>b) отсортировал получившийся массив с помощью сортировки подсчетом;
_>>c) выписал отсортированные по второму разряду числа во 1 файл;

D14>Ну, идея-то понятна. Вместо двух чанков в памяти использовать вдвое больший чанк в памяти и такой же на диске. Концептуально ничего нового по сравнению с лобовым методом, к тому же, не понятно, как к такому решению можно прийти логически — выгрузить один чанк на диск, а не эмпирически.


_>>в этом алгоритме в данной реализации — узкое место это диск,
_>>какой бы многоядерный не был у тебя процессор — тут ты существенного ускорения не получишь —
_>>тут в задаче "заковырка" — для любителей многопроцессорности.
_>>хотя этап b) можно распараллелить — если бороться за каждую секунду.

D14>Перманентная загрузка проца >0 — было условием задачи. К тому, же я не уверен, ускорения не получишь. Думаю, при наличии нескольких физических дисков все ровно наоборот. Академические решения шуршат по паре-тройке десятков гб/мин (сначала хотел написать сек., но видимо ошибаюсь) — по-моему на такие цифры я натолкнулся после 15 минут гугления.

после твоих комментариев, я не уверен, что идея и алгоритм были поняты

Здравствуйте, vitali_y, Вы писали:

_>после твоих комментариев, я не уверен, что идея и алгоритм были поняты
т.е. таблицу распределений хранить в памяти (256 мб), а сортировать на диске? И огрести random access доступ по здоровому временному файлу, либо я опять не так понял?

_>>после твоих комментариев, я не уверен, что идея и алгоритм были поняты
D14>т.е. таблицу распределений хранить в памяти (256 мб), а сортировать на диске? И огрести random access доступ по здоровому временному файлу, либо я опять не так понял?

да мое предыдущее описание не верно — когда доходит до реализации — приходится еще подумать — правильно ли понята теория .
2^30 даже байт не поместится в отведенных 256MB...

Правильный алгоритм:
имеется файл данных — это f0;
алгоритм состоит в следующем:
требуется N дополнительных файлов — пронумерованы от 0 до N-1;
проход 0:
a) идем по рассматриваемому файлу (f0), вычитываем числа, обрабатываем 0 младший разряд в выбранной системе счисления;
b) в соответствии с этим разрядом (это число x от 0 до N-1) — записываю рассматриваемое число в соответствующий доп. файл с номером x;
c) исходный файл (f0) на диске стираем;
проход 1:
0) требуется еще N дополнительных файлов — пронумерованы от 0 до N-1;
a) рассматриваем доп. файлы полученные на предыдущем шаге в порядке от 0 до N-1;
b) идем по рассматриваемому файлу, вычитываем числа, обрабатываем 1 младший разряд в выбранной системе счисления;
с) в соответствии с этим разрядом (это число x от 0 до N-1) — записываю рассматриваемое число в соответствующий доп. файл с номером x;
...
после последнего шага выписываю результат в файл — числа отсортированы.

дополнительные 256MB не нужны, много процесорность не нужна.

если в качестве основания системы счисления выбрать N := 256 — то понадобится 4 прохода для сортировки.
по памяти алгоритм требует доп. места на диске равного размеру исходного файла, размер оперативной памяти незначительный,
используемый процессор — безразлично. время работы алгоритма оценивается в 5 раз вычитать объем исходных данных с диска.
т.е. быстрый диск — это здесь хорошо.
можно использовать эту задачу в качестве тестирования оборудования на производительность с доказательством того что человек,
поддавшийся рекламе и купивший самый последний процессор был обманут на деньги

Здравствуйте, vitali_y, Вы писали:

_>по памяти алгоритм требует доп. места на диске равного размеру исходного файла, размер оперативной памяти незначительный,
_>используемый процессор — безразлично. время работы алгоритма оценивается в 5 раз вычитать объем исходных данных с диска.
_>т.е. быстрый диск — это здесь хорошо.
_>можно использовать эту задачу в качестве тестирования оборудования на производительность с доказательством того что человек,
_>поддавшийся рекламе и купивший самый последний процессор был обманут на деньги
Не поленитесь, напишите реализацию. Можно будет сравнить с моим решением — http://rsdn.ru/forum/cpp.applied/3846359.1.aspx

Автор: alsemm
Дата: 17.06.10

Идея такова.

1. Создаем на диске N отсортированных подмножеств размером 256M. Банально нарезаем файл на куски по 256M , сортируем их в памяти и сбрасываем на диск.

2. Используем сортировку подсчетом от 0 до 256M значений (и дальше в цикле двигаясь вверх по 256м), делаем это проходя каждый из N отсортированных подмножеств.

3. Сортировать подсчетом будем только значения лежащие в нужном диапазоне , тем самым избегая лишние проходы по файлам.

4. подсчитанные значения скидываем на диск последовательно записывая значения.


Преимущества , везде используется последовательный доступ к данным. каждое из данных проходятся с диска только 2 раза.

Вроде все операции можно распаралелить, но боюсь проблемы будут с одновременным чтением из файла .

_>>по памяти алгоритм требует доп. места на диске равного размеру исходного файла, размер оперативной памяти незначительный,
_>>используемый процессор — безразлично. время работы алгоритма оценивается в 5 раз вычитать объем исходных данных с диска.
_>>т.е. быстрый диск — это здесь хорошо.
_>>можно использовать эту задачу в качестве тестирования оборудования на производительность с доказательством того что человек,
_>>поддавшийся рекламе и купивший самый последний процессор был обманут на деньги
A>Не поленитесь, напишите реализацию. Можно будет сравнить с моим решением — http://rsdn.ru/forum/cpp.applied/3846359.1.aspx

Автор: alsemm
Дата: 17.06.10

нет желания. сомневаюсь в работоспособности вами написанного — даже тестировать не захотелось и разбираться
в вашем алгоритме — 2 exe для меня все слишком сложно.

_>если в качестве основания системы счисления выбрать N := 256 — то понадобится 4 прохода для сортировки.
_>по памяти алгоритм требует доп. места на диске равного размеру исходного файла, размер оперативной памяти незначительный,
_>используемый процессор — безразлично. время работы алгоритма оценивается в 5 раз вычитать объем исходных данных с диска.
_>т.е. быстрый диск — это здесь хорошо.
_>можно использовать эту задачу в качестве тестирования оборудования на производительность с доказательством того что человек,
_>поддавшийся рекламе и купивший самый последний процессор был обманут на деньги

если в качестве основания системы счисления выбрать N := 65536 — то понадобится 2 прохода для сортировки.
если в качестве основания системы счисления выбрать N := 4294967296 — то это будет чистая сортировка подсчетом.
интуитивно, похоже что N := 256 — оптимальное значение.
реализуемый алгоритм должен воспринимать N в качестве параметра, чтобы узнать, что лучше работает на практике.

Здравствуйте, vitali_y, Вы писали:

_>нет желания. сомневаюсь в работоспособности вами написанного — даже тестировать не захотелось и разбираться
В работоспособности моего решения убедиться проще простого:
1. возьмем некий файл 'foo' и отстортируем его моим решением:

./sort.sh foo foo.sorted

2. преобразуем foo.sorted в текстовый файл так, чтобы в каждой строке было было одно 32-битное целое и сохраним в файл foo.sorted.txt:

hexdump -ve '1/4 "%08x " "\n"' foo.sorted > foo.sorted.txt

3. преобразуем исходный файл foo аналогично тому, как это сделали для foo.sorted:

hexdump -ve '1/4 "%08x " "\n"' foo > foo.txt

4. сортируем foo.txt:

sort foo.txt > foo.sorted2.txt

5. сравниваем foo.sorted.txt и foo.sorted2.txt (файлы должны быть одинаковыми):

cmp foo.sorted.txt foo.sorted2.txt

PS Свои сомнения держите при себе, или приводите веские аргументы, которые их могут обосновать.

_>>нет желания. сомневаюсь в работоспособности вами написанного — даже тестировать не захотелось и разбираться
A>В работоспособности моего решения убедиться проще простого:
A>1. возьмем некий файл 'foo' и отстортируем его моим решением:
A>

A>./sort.sh foo foo.sorted
A>

A>2. преобразуем foo.sorted в текстовый файл так, чтобы в каждой строке было было одно 32-битное целое и сохраним в файл foo.sorted.txt:
A>

A>hexdump -ve '1/4 "%08x " "\n"' foo.sorted > foo.sorted.txt
A>

A>3. преобразуем исходный файл foo аналогично тому, как это сделали для foo.sorted:
A>

A>hexdump -ve '1/4 "%08x " "\n"' foo > foo.txt
A>

A>4. сортируем foo.txt:
A>

A>sort foo.txt > foo.sorted2.txt
A>

A>5. сравниваем foo.sorted.txt и foo.sorted2.txt (файлы должны быть одинаковыми):
A>

A>cmp foo.sorted.txt foo.sorted2.txt
A>

A>PS Свои сомнения держите при себе, или приводите веские аргументы, которые их могут обосновать.

тогда вам техническое задание. программа должно работать следующим образом:
программа обрабатывает следующие параметры входной строки:
-g N -> генерирует файл заполненный рандомно сгенерированными числами размером N GB;
-generate N -> то же что и выше;
-c filename -> проверяет идут ли числа заполняющие файл в порядке возрастания;
-check filename -> то же что и выше;
-s filename -> сортирует входной файл, на выходе filename.s;
-sort filename -> то же что и выше;
-help -> показывает данную справку;
-h -> то же что и выше;
-? -> то же что и выше;

ну почему я должен держать свои сомнения при себе — я скачал ваш исходник посмотрел его, попытался понять ваше описание,
а уж потом засомневался.
вы вероятно заметили, что у нас немного разное понятие простоты...
у меня нету bash и прочих подобных прелестей и знаете это меня даже не расстраивает.
мне не ясно что вы сравниваете в п.5, что вы преобразуете в п.3 и в п.2 — вы что в ручную проверяете 4GB файл ?!

Здравствуйте, vitali_y, Вы писали:

_>ну почему я должен держать свои сомнения при себе — я скачал ваш исходник посмотрел его, попытался понять ваше описание,
_>а уж потом засомневался.
Исходников там два. Еще несколько баш-скриптов. Что именно вызывает сомнения? Конкретно, в формате "имя файла:номер строки"

_>мне не ясно что вы сравниваете в п.5, что вы преобразуете в п.3 и в п.2 — вы что в ручную проверяете 4GB файл ?!
1. п.2, п.3 — преобразование бинарного файла в текстовый с заданным форматом, см. hexdump
2. п5. — сравнение файлов, см. cmp &mdash; compare two files byte by byte

>>ну почему я должен держать свои сомнения при себе — я скачал ваш исходник посмотрел его, попытался понять ваше описание,
_>>а уж потом засомневался.
A>Исходников там два. Еще несколько баш-скриптов. Что именно вызывает сомнения? Конкретно, в формате "имя файла:номер строки"

два исходника — мне 1 достаточно — зачем 2 exe — я написал вам как нормальная программа должна выглядеть.
у вас — ненормальная — конечно с оговоркой — для моих средних способностей.
мне вас — гениев — не понять — потомки поймут — не расстраивайтесь.

_>>мне не ясно что вы сравниваете в п.5, что вы преобразуете в п.3 и в п.2 — вы что в ручную проверяете 4GB файл ?!
A>1. п.2, п.3 — преобразование бинарного файла в текстовый с заданным форматом, см. hexdump
A>2. п5. — сравнение файлов, см. cmp &mdash; compare two files byte by byte

зачем мне текстовый файл?
зачем мне cygwin? зачем мне cmp?

где оценка времени работы вашего алгоритма? она линейная, квадратичная? вы с оценкой работы алгоритма знакомы
или образование гения кулинарный техникум?

Здравствуйте, vitali_y, Вы писали:

>>>ну почему я должен держать свои сомнения при себе — я скачал ваш исходник посмотрел его, попытался понять ваше описание,
_>>>а уж потом засомневался.
A>>Исходников там два. Еще несколько баш-скриптов. Что именно вызывает сомнения? Конкретно, в формате "имя файла:номер строки"

_>два исходника — мне 1 достаточно — зачем 2 exe — я написал вам как нормальная программа должна выглядеть.
Каждая программа должна делать что-то одно. От этого она получается простая. А вся система в целом более гибкая и настраиваемая. В такой системе шелл (bash) — это "клей", который собирает мелкие программы в одну. unix way и все такое.

_>>>мне не ясно что вы сравниваете в п.5, что вы преобразуете в п.3 и в п.2 — вы что в ручную проверяете 4GB файл ?!
A>>1. п.2, п.3 — преобразование бинарного файла в текстовый с заданным форматом, см. hexdump
A>>2. п5. — сравнение файлов, см. cmp &mdash; compare two files byte by byte

_>зачем мне текстовый файл?
Задача: убедиться, что программа сортировки работает корректно. Можно конечно и на C++ слепить программу для этого или добавить в основную программу сортировки еще один параметр командной строки. А можно использовать стандартные средства, например, unix-овые (впрочем, другими я и не владею
Убедиться, что самопальная сортировка работает корректно можно если отсортировать исходный файл какими-то стандартными средствами, которые дают 100% корректный результат и сравнить с выводом нашей сортировки. Для сортировки текстовых файлов в unix есть команда sort. Но она работает с текстом и сортирует строки. А у нас исходные данные — бинарные. Значит перед сортировкой данные надо перевести в текст. Для этого есть команда hexdump. Настроим hexdump так, что в текстовом файле каждое 32-битное целое будет записано на отдельной строке. Теперь запускаем sort и получаем отсортированный текстовый файл. Остается напустить hexdump на отсортированный нашей сортировкой файл и сравнить результат с результатом, который выдал sort. Для сравнения файлов есть команды diff и cmp. Нам хватит cmp.

_>зачем мне cygwin? зачем мне cmp?
Затем чтобы не писать идиотскте тех. задания и не строить велосипеды.

_>где оценка времени работы вашего алгоритма? она линейная, квадратичная? вы с оценкой работы алгоритма знакомы
Прошу прощения, вы видимо имели ввиду оценку сложности алгоритма. Понятие "оценка времени" в теории алгоритмов я чего-то не припонимаю.
А оценка сложности получается как для сортировки слиянием, но несколько лучше, т.к. сортировку фрагментов мы распараллеливаем.

_>или образование гения кулинарный техникум?
Держите себя в руках

1) unix way у словии задачи не оговаривался.
2) вы переводите 4-16 GB бинарный в текстовый формат — какого размера у вас текстовый файл в результате?
3) оценку сложности вы не можете сделать. оценка времени — это примерно сколько времени в часах минутах секундах будет работать ваша программа. я оценил работу моего алгоритма по времени — зависит от выбора N — столько раз придется вычитать заданный объем с диска.

Здравствуйте, vitali_y, Вы писали:

_>1) unix way у словии задачи не оговаривался.
Вас что не устраивает в моем решении, что там два екзешника получилось? Ну солью я их в один, получу С++ простыню на тыщу строк с запуском потоков, их ожиданием и т.п. За лесом деревья потеряются. На фик эти сложности, если часть работы проще в шеле сделать.

_>2) вы переводите 4-16 GB бинарный в текстовый формат — какого размера у вас текстовый файл в результате?
А зачем на таких объемах проверять корректность мой сортировки? Взять любой zip на десяток мегов и проверить — этого будет достаточно.

просто шутка в картинках:

вы:
я:
я побщался с вами:
виртуально, для общего спокойствия:

ни времени оценки, ни сложности алгоритма вашего я не получил и запускать его я не буду в вашем виде.
все — давайте прекратим бесполезный разговор.

На кой черт здесь вообще потоки нужны? Операции с диском съедят выигрыш в производительности, если только функция сравнения двух элементов (а тут всего лишь сравнение чисел) не требует длительных вычислений.

По коду. Очень много букв. Можно было написать короче. Зачем-то классы притянуты за уши, хотя задача довольно процедурная.

По поводу алгоритма. Куски по 256 Мб можно отсортировать в памяти любым понравившися алгоритмом. Далее сортировкой слиянием в каждом проходе по файлу объединять 2 соседних куска данных.
Примерно получится log2(16 Гб / 256 Мб) + 1 = 7 проходов по файлу на диске.

Здравствуйте, <Аноним>, Вы писали:

А>На кой черт здесь вообще потоки нужны? Операции с диском съедят выигрыш в производительности, если только функция сравнения двух элементов (а тут всего лишь сравнение чисел) не требует длительных вычислений.

А>По коду. Очень много букв. Можно было написать короче. Зачем-то классы притянуты за уши, хотя задача довольно процедурная.

А>По поводу алгоритма. Куски по 256 Мб можно отсортировать в памяти любым понравившися алгоритмом. Далее сортировкой слиянием в каждом проходе по файлу объединять 2 соседних куска данных.
А>Примерно получится log2(16 Гб / 256 Мб) + 1 = 7 проходов по файлу на диске.

Приводили решение с 2мя доступами к данным.

Здравствуйте, Аноним, Вы писали:

А>По поводу алгоритма. Куски по 256 Мб можно отсортировать в памяти любым понравившися алгоритмом. Далее сортировкой слиянием в каждом проходе по файлу объединять 2 соседних куска данных.

сливать можно все куски разом.

Здравствуйте, VEAPUK, Вы писали:

VEA>Здравствуйте, Аноним, Вы писали:

А>>По поводу алгоритма. Куски по 256 Мб можно отсортировать в памяти любым понравившися алгоритмом. Далее сортировкой слиянием в каждом проходе по файлу объединять 2 соседних куска данных.

VEA>сливать можно все куски разом.

Ааа... Об этом я не подумал)
Можно было бы heap держать, с минимальным элементом в каждом куске. Хотя бы.

У Кормена было где-то такое упражнение на сливание множества отсортированных массивов.