/*
   Copyright (c) 2019 by The ThreadDB Project
   All Rights Reserved.

   ThreadDB undergoes the BSD License 2.0. You should have received a copy along with this program; if not, write to the ThreadDB Project.
   To obtain a full unlimited version contact thethreaddbproject(at)gmail.com.
   
   mainsample.cpp - General testrun of the threaddb functionality.
*/
   
#include <threaddbCPP.h>

#include <vector>
#include <list>
#include <set>
#include <map>
#include <memory>
#include <thread>
#include <stdexcept>
#include <iostream>
#include <stdio.h>
#include <string.h>
#include <sstream>
#include <chrono>

#if defined(_WIN32)
    #define PATH_SEP "\\"
#else
    #define PATH_SEP "/"
#endif

namespace tdb
{
    // [Example1.2 example]
    void threadStore(size_t idx_p, tdb::database* pthreaddb_p, const std::vector<uint64_t>* pPackageId_p) {
        const std::vector < uint64_t >& rPackageId(*pPackageId_p);
        std::string myString = "This is the new horizon approaching.";
        for (size_t iter = 0; iter < 60000; ++iter)
        {
            (*pthreaddb_p).Store(rPackageId[(iter + iter % 123 + idx_p) % 123], myString.size(), myString.c_str());
        }
    };
    // [Example1.2 example]

    // [Example2.2 example]
    void threadRecover(size_t idx_p, tdb::database* pthreaddb_p, const std::vector<uint64_t>* pPackageId_p, const char* pStringValue_p, size_t CheckLen_p) {
        const std::vector < uint64_t >& rPackageId(*pPackageId_p);
        char readBuffer[4096];

        const std::string myString = pStringValue_p;

        for (size_t packageIdx = 0; packageIdx < 123; ++packageIdx)
        {
            tdb::ReadInfo readHandle = (*pthreaddb_p).Open(rPackageId[(idx_p + packageIdx) % 123]);

            while ((*pthreaddb_p).Recover(myString.size(), readBuffer, readHandle))
            {
                if (strncmp(readBuffer, myString.c_str(), CheckLen_p))
                {
                    throw std::runtime_error("Data integrity error in tdb::database");
                }
            }
        }
    };
    // [Example2.2 example]
}

void TestDB(size_t PackageSize_p, size_t PackageCacheSize_p, const char* pTmpFolder_p, const char* pTmpPackageFolder_p)
{
    const std::string myString = "This is the new horizon approaching.";
    const std::string myString_ = "This is the new horizon approaching!";
    const std::string myString__ = "This is the new horizon approaching";

    std::vector<uint64_t> packageID;
    std::list<std::shared_ptr<tdb::ItemInfo> > itemList;

    char readBuffer[4096];

    std::list< std::string > createdFiles;

    std::cout << "Testrun started - Packagesize: " << PackageSize_p << " Cached Packages: " << PackageCacheSize_p <<
				  " Package header flushing: " << ( pTmpPackageFolder_p!=std::string("")? "true":"false" ) << std::endl;
    try
    {
		// [Example1.1 example]
        std::cout << "Creating database" << std::endl;
        tdb::database database(PackageSize_p, PackageCacheSize_p, pTmpPackageFolder_p);

        std::cout << "Creating threads" << std::endl;
        createdFiles.push_back(database.NewThread(pTmpFolder_p));
        createdFiles.push_back(database.NewThread(pTmpFolder_p));
        createdFiles.push_back(database.NewThread(pTmpFolder_p));
        createdFiles.push_back(database.NewThread(pTmpFolder_p));

        std::cout << "Creating packages" << std::endl;
        for (int iter = 0; iter < 123; ++iter)
        {
            packageID.push_back(database.NewPackage());
        }

        std::vector< std::shared_ptr<tdb::ItemInfo> > itemStart;
        itemStart.resize(123);

        std::cout << "Multithreaded filling started" << std::endl;
        {
            std::list< std::shared_ptr<std::thread> > threadGroup;

            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadStore, 1, &database, &packageID)));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadStore, 99, &database, &packageID)));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadStore, 24, &database, &packageID)));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadStore, 120, &database, &packageID)));

            for (size_t iter = 0; iter < 60000; ++iter)
            {
                if (iter == 10000)
                {
                    std::cout << "Moving database file" << std::endl;
                    std::ostringstream os;
                    os << pTmpFolder_p << PATH_SEP << "mymovedfile.tdb";
                    database.RelocateFileTo(0, os.str().c_str(), eMoveFileTo);
                    std::cout << "Moved database file to: " << os.str() << std::endl;
                    createdFiles.push_back(os.str());
                }

                itemList.push_back(std::shared_ptr<tdb::ItemInfo>(new tdb::ItemInfo()));

                const size_t packageNr = (iter + iter % 123) % 123;

                if (itemStart[packageNr] == 0)
                {
                    itemStart[packageNr] = itemList.back();
                }

                database.Store(packageID[packageNr], myString.size(), myString.c_str(), itemList.back().get());
            }

            for (auto iter(threadGroup.begin()); iter != threadGroup.end(); ++iter)
                (*iter)->join();
        }
        std::cout << "Multithreaded filling finished" << std::endl;

        std::cout << "Synchronize database buffers" << std::endl;
        database.Synchronize();
        
        // [Example1.1 example]

        // [Example2.1 example]

        std::cout << "Multithreaded stream reading started" << std::endl;
        {
            std::list< std::shared_ptr<std::thread> > threadGroup;

            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 1, &database, &packageID, myString.c_str(), myString.size())));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 99, &database, &packageID, myString.c_str(), myString.size())));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 24, &database, &packageID, myString.c_str(), myString.size())));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 120, &database, &packageID, myString.c_str(), myString.size())));

            for (size_t packageIdx = 0; packageIdx < 123; ++packageIdx)
            {
                tdb::ReadInfo readHandle = database.Open(packageID[packageIdx]);

                while (database.Recover(myString.size() * 10, readBuffer, readHandle))
                {
                }
            }

            for (auto iter(threadGroup.begin()); iter != threadGroup.end(); ++iter)
                (*iter)->join();
        }
        std::cout << "Multithreaded stream reading finished" << std::endl;

        std::cout << "Singlethreaded item access/replace started" << std::endl;
        for (auto iter = itemList.begin(); iter != itemList.end(); ++iter)
        {
            tdb::ReadInfo readHandle = database.Open(**iter);
            size_t bytesRead = database.Recover(myString.size(), readBuffer, readHandle);

            readBuffer[myString_.size()] = 0;
            if (std::string(readBuffer) != myString)
            {
                throw std::runtime_error("Data integrity error in tdb::database");
            }

            database.Replace(myString.size(), myString_.c_str(), **iter);
        }

        std::cout << "Singlethreaded item access/replace finished" << std::endl;

        // [Example2.1 example]

        std::cout << "Single threaded item access started" << std::endl;

        for (auto iter = itemStart.begin(); iter != itemStart.end(); ++iter)
        {
            if ((*iter) != 0)
            {
                tdb::ReadInfo readHandle = database.Open(**iter);
                for (size_t bytesRead = database.Recover(myString.size(), readBuffer, readHandle);
                     bytesRead>0;
                     bytesRead = database.Recover(myString.size(), readBuffer, readHandle))
                {
                    readBuffer[bytesRead-1] = 0;
                    if (std::string(readBuffer) != myString__)
                    {
                        throw std::runtime_error("Data integrity error in tdb::database");
                    }

                };
            }
        }
        std::cout << "Singlethreaded item access finished" << std::endl;

        std::cout << "Saving database index file" << std::endl;

        std::ostringstream os;
        os << pTmpFolder_p << PATH_SEP << "testrun.tdb";
        database.Save(os.str().c_str());

        createdFiles.push_back(os.str().c_str());
    }
    catch (const std::runtime_error& exception)
    {
        std::cout << exception.what();
        exit(1);
    }

    try
    {
        std::ostringstream os;
        os << pTmpFolder_p << PATH_SEP << "testrun.tdb";

        std::cout << "Creating file based database" << std::endl;
        tdb::database database(os.str().c_str(), PackageCacheSize_p);

        std::cout << "Singlethreaded item access started" << std::endl;
        for (auto iter = itemList.begin(); iter != itemList.end(); ++iter)
        {
            tdb::ReadInfo readHandle = database.Open(**iter);
            size_t bytesRead = database.Recover(myString.size(), readBuffer, readHandle);

            readBuffer[myString_.size()] = 0;
            if (std::string(readBuffer) != myString_)
            {
                throw std::runtime_error("Data integrity error in tdb::database");
            }
        }
        std::cout << "Singlethreaded item access finished" << std::endl;

        std::cout << "Multithreaded stream reading started" << std::endl;
        {
            std::list< std::shared_ptr<std::thread> > threadGroup;

            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 1, &database, &packageID, myString_.c_str(), myString_.size() - 1)));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 99, &database, &packageID, myString_.c_str(), myString_.size() - 1)));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 24, &database, &packageID, myString_.c_str(), myString_.size() - 1)));
            threadGroup.push_back(std::shared_ptr<std::thread>(new std::thread(&tdb::threadRecover, 120, &database, &packageID, myString_.c_str(), myString_.size() - 1)));

            for (size_t packageIdx = 0; packageIdx < 123; ++packageIdx)
            {
                tdb::ReadInfo readHandle = database.Open(packageID[packageIdx]);

                while (database.Recover(myString.size() * 10, readBuffer, readHandle))
                {
                }
            }

            for (auto iter(threadGroup.begin()); iter != threadGroup.end(); ++iter)
                (*iter)->join();
        }

        std::cout << "Multithreaded stream reading finished" << std::endl;

        createdFiles.push_back(database.NewThread(pTmpFolder_p));
        createdFiles.push_back(database.NewThread(pTmpFolder_p));
        createdFiles.push_back(database.NewThread(pTmpFolder_p));
        createdFiles.push_back(database.NewThread(pTmpFolder_p));

        std::cout << "Singlethreaded filling started" << std::endl;

        for (size_t iter = 0; iter < 60000; ++iter)
        {
            database.Store(packageID[(iter + iter % 123) % 123], myString.size(), myString.c_str());
        }

        std::cout << "Singlethreaded filling finished" << std::endl;

        std::cout << "Singlethreaded stream reading started" << std::endl;

        for (size_t packageIdx = 0; packageIdx < 123; ++packageIdx)
        {
            tdb::ReadInfo readHandle = database.Open(packageID[packageIdx]);

            while (database.Recover(myString.size() * 10, readBuffer, readHandle))
            {
            }
        }

        std::cout << "Singlethreaded stream reading finised" << std::endl;
    }
    catch (const std::runtime_error& exception)
    {
        std::cout << exception.what();
        exit(1);
    }

    std::cout << "Remove temporary files" << std::endl;

    for (auto iter(createdFiles.begin()); iter != createdFiles.end(); ++iter)
    {
        try
        {
            std::remove((*iter).c_str());
        }
        catch (...)
        {
            ;
        }
    }
}

int main(int argc, char* argv[])
{
    std::string tmpFolder;

    if (argc < 2)
    {
        std::cout << "The testrun of threadDB requires a folder name to store the temporary database files" << std::endl;
        exit(1);
    }

    tmpFolder = argv[1];
    tmpFolder.erase(tmpFolder.find_last_not_of("\\/") + 1);

	typedef std::chrono::system_clock::time_point timepoint;

	// First run is with a package buffer limit of 30 packages, this will cause package flushing only.
	{
		timepoint t1 = std::chrono::system_clock::now();

		TestDB(4096, 30, tmpFolder.c_str(), "");

		timepoint t2 = std::chrono::system_clock::now();
        std::chrono::duration<double, std::milli> fp_ms = t2 - t1;

		std::cout << "Executing time perturbed package access with package flushing: " << fp_ms.count()/1000.0 << std::endl;
	}

	// Second run is with a package buffer limit of 30 packages, this will cause package and package header flushing.
	{
		timepoint t1 = std::chrono::system_clock::now();

		TestDB(4096, 80, tmpFolder.c_str(), tmpFolder.c_str());

		timepoint t2 = std::chrono::system_clock::now();
        std::chrono::duration<double, std::milli> fp_ms = t2 - t1;

		std::cout << "Executing time perturbed package access with package and package header flushing: " << fp_ms.count()/1000.0 << std::endl;
	}

    // Third run is with a package buffer limit of 200 packages, no package collisions. The enlarged Record size will also minimize disk synchronization
	{
		timepoint t1 = std::chrono::system_clock::now();

		TestDB(4194304, 200, tmpFolder.c_str(), tmpFolder.c_str());
		
		timepoint t2 = std::chrono::system_clock::now();
		std::chrono::duration<double, std::milli> fp_ms = t2 - t1;

		std::cout << "Executing time perturbed package access without package flushing: " << fp_ms.count()/1000.0 << std::endl;
	}
}
