/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/***********************************************************************
 *   $Id: tdriver-bru.cpp 9210 2013-01-21 14:10:42Z rdempsey $
 *
 *
 ***********************************************************************/
#include <iostream>
#include <sstream>
#include <fstream>
#include <iomanip>
#include <vector>
#include <set>

#include <pthread.h>
#include <time.h>
#include <sys/time.h>

#include <boost/filesystem/path.hpp>
#include <boost/filesystem/operations.hpp>
#include <boost/thread/mutex.hpp>

#include <cppunit/extensions/HelperMacros.h>
#include <cppunit/extensions/TestFactoryRegistry.h>
#include <cppunit/ui/text/TestRunner.h>

#include "bucketdl.h"
#include "elementtype.h"
#include "stopwatch.cpp"

#include "bucketreuse.h"

// #undef CPPUNIT_ASSERT
// #define CPPUNIT_ASSERT(x)

using namespace std;
using namespace boost;
using namespace joblist;

// Stopwatch timer;

//------------------------------------------------------------------------------
// TestDriver class derived from CppUnit
//------------------------------------------------------------------------------

class BucketReUseDriver : public CppUnit::TestFixture
{
  CPPUNIT_TEST_SUITE(BucketReUseDriver);
  CPPUNIT_TEST(parseConfig);
  CPPUNIT_TEST(createFiles);
  CPPUNIT_TEST(reuseFiles);
  CPPUNIT_TEST(newversion);
  CPPUNIT_TEST(concurrent);
  CPPUNIT_TEST(concurrent_newversion);
  CPPUNIT_TEST(concurrent_race);
  CPPUNIT_TEST_SUITE_END();

 private:
 public:
  //--------------------------------------------------------------------------
  // setup method run prior to each unit test, inherited from base
  //--------------------------------------------------------------------------
  void setUp()
  {
    clock_gettime(CLOCK_REALTIME, &ts);
  }

  //--------------------------------------------------------------------------
  // validate results from a unit test, inherited from base
  //--------------------------------------------------------------------------
  void validateResults()
  {
  }

  //--------------------------------------------------------------------------
  // test functions
  //--------------------------------------------------------------------------
  void parseConfig();
  void createFiles();
  void reuseFiles();
  void newversion();
  void concurrent();
  void concurrent_newversion();
  void concurrent_race();

 private:
  //--------------------------------------------------------------------------
  // initialize method
  // oid : reference to the column OID used for test
  //--------------------------------------------------------------------------
  void initControl(execplan::CalpontSystemCatalog::OID& oid);

  //--------------------------------------------------------------------------
  // validate results
  // files: the filenames to be verified if exist
  // exist: expect if the files are created or deleted
  //--------------------------------------------------------------------------
  void validateFileExist(set<string>& files, bool exist);

  static void* insertThread(void*);
  static void* readThread(void*);

  static void* scanThread(void*);
  static void* reuseThread(void*);
  static void* raceThread(void*);

  struct ThreadArg
  {
    uint64_t id;                              // thread id
    uint64_t version;                         // db version
    uint64_t buckets;                         // max bucket numbers
    uint64_t elements;                        // max number of elem per bucket
    uint64_t total;                           // total number of elements
    execplan::CalpontSystemCatalog::OID oid;  // column OID
    BucketDataList* dl;                       // datalist

    // for sync threads
    bool* flag;
    pthread_mutex_t* mutex;
    pthread_cond_t* cond;

    // for file status check
    set<string>* files;

    ThreadArg()
     : id(0)
     , version(0)
     , buckets(0)
     , elements(0)
     , total(0)
     , dl(NULL)
     , flag(NULL)
     , mutex(NULL)
     , cond(NULL)
     , files(NULL)
    {
    }
  };

  static const string column;

 public:
  static struct timespec ts;
};

CPPUNIT_TEST_SUITE_REGISTRATION(BucketReUseDriver);

const string BucketReUseDriver::column = "tpch.lineitem.l_orderkey";
struct timespec BucketReUseDriver::ts;

//------------------------------------------------------------------------------
// main entry point
//------------------------------------------------------------------------------
int main(int argc, char** argv)
{
  CppUnit::TextUi::TestRunner runner;
  CppUnit::TestFactoryRegistry& registry = CppUnit::TestFactoryRegistry::getRegistry();
  runner.addTest(registry.makeTest());
  bool wasSuccessful = runner.run("", false);
  return (wasSuccessful ? 0 : 1);
}

struct timespec atTime();
ostream& operator<<(ostream& os, const struct timespec& t);

void BucketReUseDriver::parseConfig()
{
  cout << "ut: parseConfig start...\n" << endl;

  // before run this test, make sure "tpch.lineitem.l_orderkey" is in the Columnstore.xml
  BucketReuseManager* control = BucketReuseManager::instance();
  ResourceManager rm;
  BucketReuseManager::instance()->startup(rm);

  // list all the predicates if any listed in the Columnstore.xml file
  for (BucketReuseMap::iterator it = control->fControlMap.begin(); it != control->fControlMap.end(); it++)
    cout << *(it->second);

  cout << endl;

  size_t schemap = column.find_first_of(".");
  size_t columnp = column.find_last_of(".");
  CPPUNIT_ASSERT(schemap != string::npos && columnp != string::npos);

  execplan::CalpontSystemCatalog::TableColName columnName;
  columnName.schema = column.substr(0, schemap);
  columnName.table = column.substr(schemap + 1, columnp - schemap - 1);
  columnName.column = column.substr(columnp + 1);

  string filter = "allrows";

  CPPUNIT_ASSERT(control->fControlMap.find(BucketFileKey(columnName, filter)) != control->fControlMap.end());

  cout << "ut: parseConfig done!\n" << endl;
}

void BucketReUseDriver::createFiles()
{
  cout << "ut: createFiles start...\n" << endl;

  ThreadArg arg;
  arg.id = 1;
  arg.version = 1;
  arg.buckets = 2;
  arg.elements = 2;
  arg.total = 1000000;

  execplan::CalpontSystemCatalog::OID oid;
  initControl(oid);
  arg.oid = oid;

  set<string> files;
  arg.files = &files;

  pthread_t t;
  pthread_create(&t, NULL, scanThread, &arg);
  pthread_join(t, NULL);

  validateFileExist(files, true);

  cout << "ut: createFiles done!\n" << endl;
}

void BucketReUseDriver::reuseFiles()
{
  cout << "ut: reuseFiles start...\n" << endl;

  ThreadArg arg1;
  arg1.id = 1;
  arg1.version = 1;
  arg1.buckets = 2;
  arg1.elements = 2;
  arg1.total = 1000000;

  execplan::CalpontSystemCatalog::OID oid;
  initControl(oid);
  arg1.oid = oid;

  set<string> files1;
  arg1.files = &files1;

  // create the files
  pthread_t t1;
  pthread_create(&t1, NULL, scanThread, &arg1);
  pthread_join(t1, NULL);

  validateFileExist(files1, true);

  // use new datalist, reuse case
  ThreadArg arg2 = arg1;
  arg2.id = 2;

  set<string> files2;
  arg2.files = &files2;

  pthread_t t2;
  pthread_create(&t2, NULL, reuseThread, &arg2);
  pthread_join(t2, NULL);

  validateFileExist(files2, true);

  cout << "ut: reuseFiles done!\n" << endl;
}

void BucketReUseDriver::newversion()
{
  cout << "ut: newversion start...\n" << endl;

  ThreadArg arg1;
  arg1.id = 1;
  arg1.version = 1;
  arg1.buckets = 2;
  arg1.elements = 2;
  arg1.total = 1000000;

  execplan::CalpontSystemCatalog::OID oid;
  initControl(oid);
  arg1.oid = oid;

  set<string> files1;
  arg1.files = &files1;

  // create the files
  pthread_t t1;
  pthread_create(&t1, NULL, scanThread, &arg1);
  pthread_join(t1, NULL);

  validateFileExist(files1, true);

  // new version
  ThreadArg arg2 = arg1;
  arg2.id = 2;
  arg2.version = 2;

  set<string> files2;
  arg2.files = &files2;

  pthread_t t2;
  pthread_create(&t2, NULL, scanThread, &arg2);
  pthread_join(t2, NULL);

  pthread_yield();

  validateFileExist(files1, false);
  validateFileExist(files2, true);

  // read from the new files with new datalist
  ThreadArg arg3 = arg2;
  arg3.id = 3;
  arg3.dl = NULL;

  set<string> files3;
  arg3.files = &files3;

  pthread_t t3;
  pthread_create(&t3, NULL, reuseThread, &arg3);
  pthread_join(t3, NULL);

  validateFileExist(files3, true);

  cout << "ut: newversion done!\n" << endl;
}

void BucketReUseDriver::concurrent()
{
  cout << "ut: concurrent start...\n" << endl;

  ThreadArg arg1;
  arg1.id = 1;
  arg1.version = 1;
  arg1.buckets = 2;
  arg1.elements = 2;
  arg1.total = 1000000;

  execplan::CalpontSystemCatalog::OID oid;
  initControl(oid);
  arg1.oid = oid;

  set<string> files1;
  arg1.files = &files1;

  // create the files
  pthread_t t1;
  pthread_create(&t1, NULL, scanThread, &arg1);

  sleep(1);

  // reuse case, current thread
  ThreadArg arg2 = arg1;
  arg2.id = 2;
  set<string> files2;
  arg2.files = &files2;

  pthread_t t2;
  pthread_create(&t2, NULL, reuseThread, &arg2);

  pthread_join(t1, NULL);
  validateFileExist(files1, true);

  pthread_join(t2, NULL);
  validateFileExist(files2, true);

  cout << "ut: concurrent done!\n" << endl;
}

void BucketReUseDriver::concurrent_newversion()
{
  cout << "ut: concurrent_newversion start...\n" << endl;

  bool flag = false;
  pthread_mutex_t mutex;
  pthread_mutex_init(&mutex, 0);
  pthread_cond_t cond;
  pthread_cond_init(&cond, 0);

  ThreadArg arg1;
  arg1.id = 1;
  arg1.version = 1;
  arg1.buckets = 2;
  arg1.elements = 2;
  arg1.total = 1000000;

  execplan::CalpontSystemCatalog::OID oid;
  initControl(oid);
  arg1.oid = oid;

  set<string> files1;
  arg1.files = &files1;

  // create the files
  pthread_t t1;
  pthread_create(&t1, NULL, scanThread, &arg1);

  // reuse case, current thread
  ThreadArg arg2 = arg1;
  arg2.id = 2;
  set<string> files2;
  arg2.files = &files2;

  sleep(1);

  pthread_t t2;
  pthread_create(&t2, NULL, reuseThread, &arg2);

  // new version
  ThreadArg arg3 = arg1;
  arg3.id = 3;
  arg3.version = 3;
  arg3.flag = &flag;
  arg3.mutex = &mutex;
  arg3.cond = &cond;
  set<string> files3;
  arg3.files = &files3;

  pthread_t t3;
  pthread_create(&t3, NULL, scanThread, &arg3);

  sleep(3);

  pthread_mutex_lock(&mutex);
  flag = true;
  pthread_cond_broadcast(&cond);
  pthread_mutex_unlock(&mutex);

  pthread_join(t1, NULL);
  pthread_join(t2, NULL);
  pthread_join(t3, NULL);

  // let cleanup thread do its job
  pthread_yield();

  validateFileExist(files1, false);
  validateFileExist(files2, false);
  validateFileExist(files3, true);

  pthread_mutex_destroy(&mutex);
  pthread_cond_destroy(&cond);

  cout << "ut: concurrent_newversion done!\n" << endl;
}

void BucketReUseDriver::concurrent_race()
{
  cout << "ut: concurrent_race start...\n" << endl;

  ThreadArg arg1;
  arg1.id = 1;
  arg1.version = 1;
  arg1.buckets = 2;
  arg1.elements = 2;
  arg1.total = 2000000;

  execplan::CalpontSystemCatalog::OID oid;
  initControl(oid);
  arg1.oid = oid;

  set<string> files1;
  arg1.files = &files1;

  ThreadArg arg2 = arg1;
  arg2.id = 2;
  set<string> files2;
  arg2.files = &files2;

  // start the version 1 threads
  pthread_t t1;
  pthread_t t2;
  pthread_create(&t1, NULL, raceThread, &arg1);
  pthread_create(&t2, NULL, raceThread, &arg2);

  // let the version 1 threads register
  sleep(1);

  ThreadArg arg3 = arg1;
  arg3.id = 3;
  arg3.version = 4;
  arg3.total = 1000000;
  set<string> files3;
  arg3.files = &files3;

  ThreadArg arg4 = arg3;
  arg4.id = 4;
  set<string> files4;
  arg4.files = &files4;

  // start the version 4 threads
  pthread_t t3;
  pthread_t t4;
  pthread_create(&t3, NULL, raceThread, &arg3);
  pthread_create(&t4, NULL, raceThread, &arg4);

  pthread_join(t3, NULL);
  pthread_join(t4, NULL);
  validateFileExist(files1, true);
  validateFileExist(files4, true);

  pthread_join(t1, NULL);
  pthread_join(t2, NULL);
  validateFileExist(files1, false);
  validateFileExist(files4, true);

  cout << "ut: concurrent_race done!\n" << endl;
}

void BucketReUseDriver::initControl(execplan::CalpontSystemCatalog::OID& oid)
{
  // make sure the column for testing is in the map, "column" is a class variable
  // to test with another column, only one place to change
  // -- to validate the config parsing, run parseConfig --
  config::Config* cf = config::Config::makeConfig();
  vector<string> columns;
  cf->getConfig("HashBucketReuse", "Predicate", columns);

  bool found = true;

  for (vector<string>::iterator it = columns.begin(); it != columns.end(); ++it)
  {
    if (it->compare(0, column.size(), column) == 0)
    {
      found = true;
      break;
    }
  }

  string filter = "allrows";
  BucketReuseManager* control = BucketReuseManager::instance();

  if (found == false)
  {
    cout << "tpch.lineitem.l_orderkey is not in the Columnstore.xml!)" << endl;
    cout << "insert it to countinue unit test" << endl;

    size_t schemap = column.find_first_of(".");
    size_t columnp = column.find_last_of(".");
    CPPUNIT_ASSERT(schemap != string::npos && columnp != string::npos);

    execplan::CalpontSystemCatalog::TableColName tcn;
    tcn.schema = column.substr(0, schemap);
    tcn.table = column.substr(schemap + 1, columnp - schemap - 1);
    tcn.column = column.substr(columnp + 1);

    control->fConfigMap.insert(pair<string, BucketFileKey>(column, BucketFileKey(tcn, filter)));
  }

  ResourceManager rm;
  // now  start the BucketReuseManager
  BucketReuseManager::instance()->startup(rm);

  // get the oid for registration
  size_t schemap = column.find_first_of(".");
  size_t columnp = column.find_last_of(".");
  CPPUNIT_ASSERT(schemap != string::npos && columnp != string::npos);

  execplan::CalpontSystemCatalog::TableColName columnName;
  columnName.schema = column.substr(0, schemap);
  columnName.table = column.substr(schemap + 1, columnp - schemap - 1);
  columnName.column = column.substr(columnp + 1);

  CPPUNIT_ASSERT(control->fControlMap.find(BucketFileKey(columnName, filter)) != control->fControlMap.end());
}

void BucketReUseDriver::validateFileExist(set<string>& files, bool exist)
{
  cout << "\ncheck if files exist or not:" << endl;

  for (set<string>::iterator i = files.begin(); i != files.end(); i++)
  {
    filesystem::path p(i->c_str());
    cout << (*i) << "-- ";

    if (exist)
    {
      CPPUNIT_ASSERT(filesystem::exists(p));
      cout << "OK" << endl;
    }
    else
    {
      CPPUNIT_ASSERT(!filesystem::exists(p));
      cout << "GONE" << endl;
    }
  }

  cout << endl;
}

void* BucketReUseDriver::insertThread(void* arg)
{
  ThreadArg* a = reinterpret_cast<ThreadArg*>(arg);
  BucketDataList* dl = a->dl;
  CPPUNIT_ASSERT(dl != NULL);

  cout << "thread " << a->id << " start at " << atTime() << endl;

  BucketReuseControlEntry* entry = dl->reuseControl();

  for (uint64_t i = 0; i < a->buckets; i++)
  {
    stringstream ss;
    ss << entry->baseName() << "." << i;
    filesystem::path p(ss.str().c_str());
    a->files->insert(ss.str());
  }

  ElementType e;

  for (uint64_t i = 0; i < a->total; i++)
  {
    e.first = i;
    e.second = i * 10 + a->version;  // include the version in values
    dl->insert(e);
  }

  cout << "thread[" << a->id << "] last element inserted at " << atTime() << endl;
  dl->endOfInput();

  cout << "thread " << a->id << " finished at " << atTime() << endl;

  return NULL;
}

void* BucketReUseDriver::readThread(void* arg)
{
  ThreadArg* a = reinterpret_cast<ThreadArg*>(arg);
  BucketDataList* dl = a->dl;
  CPPUNIT_ASSERT(dl != NULL);

  cout << "thread " << a->id << " start at " << atTime() << endl;

  BucketReuseControlEntry* entry = dl->reuseControl();

  for (uint64_t i = 0; i < a->buckets; i++)
  {
    stringstream ss;
    ss << entry->baseName() << "." << i;
    filesystem::path p(ss.str().c_str());
    a->files->insert(ss.str());
  }

  ElementType e;
  uint64_t min = 0xffffffff, max = 0, count = 0;
  uint64_t k[a->buckets];  // count of each bucket
  bool firstRead = true;

  for (uint64_t i = 0; i < a->buckets; i++)
  {
    k[i] = 0;
    uint64_t it = dl->getIterator(i);

    while (dl->next(i, it, &e))
    {
      if (firstRead)
      {
        cout << "thread[" << a->id << "] first read at " << atTime() << endl;
        firstRead = false;
      }

      if (e.second < min)
        min = e.second;

      if (e.second > max)
        max = e.second;

      // output the first 10 of each bucket or last 10 of the datalist
      // if (count < 10 || (a->total - count) < 10 || k[i] < 10)
      if (count < 2 || (a->total - count) < 2 || k[i] < 2)
        cout << "thread[" << a->id << "] bucket:" << i << " e(" << e.first << ", " << e.second << ")" << endl;

      count++;
      k[i]++;
    }
  }

  cout << "\nthread[" << a->id << "] element: count = " << count << ", min/max = " << min << "/" << max
       << ", elements in each bucket: ";

  for (uint64_t i = 0; i < a->buckets; i++)
    cout << k[i] << " ";

  cout << endl;

  cout << "thread " << a->id << " finished at " << atTime() << endl;

  return NULL;
}

void* BucketReUseDriver::scanThread(void* arg)
{
  ThreadArg* a = reinterpret_cast<ThreadArg*>(arg);
  BucketDataList* dl = a->dl;
  CPPUNIT_ASSERT(dl == NULL);

  if (a->cond != NULL)
  {
    pthread_mutex_lock(a->mutex);

    while (*(a->flag) != true)
      pthread_cond_wait(a->cond, a->mutex);

    pthread_mutex_unlock(a->mutex);
  }

  cout << "thread " << a->id << " start at " << atTime() << endl;

  string dummy;
  bool scan = false;
  boost::shared_ptr<execplan::CalpontSystemCatalog> c =
      execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(0x80000000);
  execplan::CalpontSystemCatalog::TableColName tcn = c->colName(a->oid);
  BucketReuseControlEntry* entry =
      BucketReuseManager::instance()->userRegister(tcn, dummy, a->version, a->buckets, scan);
  CPPUNIT_ASSERT(scan == true);

  ResourceManager rm;
  dl = new BucketDataList(a->buckets, 1, a->elements, rm);
  dl->setElementMode(1);
  dl->reuseControl(entry, !scan);

  ThreadArg arg1 = *a;
  arg1.id *= 10;
  arg1.dl = dl;

  pthread_t t1;
  pthread_create(&t1, NULL, insertThread, &arg1);

  ThreadArg arg2 = arg1;
  arg2.id += 1;
  pthread_t t2;
  pthread_create(&t2, NULL, readThread, &arg2);

  pthread_join(t1, NULL);
  pthread_join(t2, NULL);

  delete dl;

  cout << "thread " << a->id << " finished at " << atTime() << endl;

  return NULL;
}

void* BucketReUseDriver::reuseThread(void* arg)
{
  ThreadArg* a = reinterpret_cast<ThreadArg*>(arg);
  BucketDataList* dl = a->dl;
  CPPUNIT_ASSERT(dl == NULL);

  if (a->cond != NULL)
  {
    pthread_mutex_lock(a->mutex);

    while (*(a->flag) != true)
      pthread_cond_wait(a->cond, a->mutex);

    pthread_mutex_unlock(a->mutex);
  }

  cout << "thread " << a->id << " start at " << atTime() << endl;

  string dummy;
  bool scan = true;
  boost::shared_ptr<execplan::CalpontSystemCatalog> c =
      execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(0x80000000);
  execplan::CalpontSystemCatalog::TableColName tcn = c->colName(a->oid);
  BucketReuseControlEntry* entry =
      BucketReuseManager::instance()->userRegister(tcn, dummy, a->version, a->buckets, scan);
  CPPUNIT_ASSERT(scan == false);

  ResourceManager rm;
  dl = new BucketDataList(a->buckets, 1, a->elements, rm);
  dl->setElementMode(1);
  dl->reuseControl(entry, !scan);

  ThreadArg arg1 = *a;
  arg1.id *= 10;
  arg1.dl = dl;

  pthread_t t1;
  pthread_create(&t1, NULL, readThread, &arg1);

  if (entry->fileStatus() == BucketReuseControlEntry::progress_c)
  {
    boost::mutex::scoped_lock lock(BucketReuseManager::instance()->getMutex());
    dl->reuseControl()->stateChange().wait(lock);
  }
  else
  {
    CPPUNIT_ASSERT((entry->fileStatus() == BucketReuseControlEntry::using_c) ||
                   (entry->fileStatus() == BucketReuseControlEntry::ready_c));
  }

  // the bucket files are ready
  dl->restoreBucketInformation();
  dl->endOfInput();

  pthread_join(t1, NULL);

  delete dl;

  cout << "thread " << a->id << " finished at " << atTime() << endl;

  return NULL;
}

void* BucketReUseDriver::raceThread(void* arg)
{
  ThreadArg* a = reinterpret_cast<ThreadArg*>(arg);
  BucketDataList* dl = a->dl;
  CPPUNIT_ASSERT(dl == NULL);

  if (a->cond != NULL)
  {
    pthread_mutex_lock(a->mutex);

    while (*(a->flag) != true)
      pthread_cond_wait(a->cond, a->mutex);

    pthread_mutex_unlock(a->mutex);
  }

  cout << "thread " << a->id << " start at " << atTime() << endl;

  string dummy;
  bool scan = true;
  ResourceManager rm;
  BucketReuseControlEntry* entry = NULL;
  {
    boost::mutex::scoped_lock lock(BucketReuseManager::instance()->getMutex());
    boost::shared_ptr<execplan::CalpontSystemCatalog> c =
        execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(0x80000000);
    execplan::CalpontSystemCatalog::TableColName tcn = c->colName(a->oid);
    entry = BucketReuseManager::instance()->userRegister(tcn, dummy, a->version, a->buckets, scan);

    dl = new BucketDataList(a->buckets, 1, a->elements, rm);
    dl->setElementMode(1);
    dl->reuseControl(entry, !scan);
  }

  if (scan == true)
  {
    CPPUNIT_ASSERT(entry->fileStatus() == BucketReuseControlEntry::progress_c);

    ThreadArg arg1 = *a;
    arg1.id *= 10;
    arg1.dl = dl;

    pthread_t t1;
    pthread_create(&t1, NULL, insertThread, &arg1);

    ThreadArg arg2 = arg1;
    arg2.id += 1;
    pthread_t t2;
    pthread_create(&t2, NULL, readThread, &arg2);

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
  }
  else
  {
    ThreadArg arg1 = *a;
    arg1.id *= 10;
    arg1.dl = dl;

    pthread_t t1;
    pthread_create(&t1, NULL, readThread, &arg1);

    if (entry->fileStatus() == BucketReuseControlEntry::progress_c)
    {
      boost::mutex::scoped_lock lock(BucketReuseManager::instance()->getMutex());
      dl->reuseControl()->stateChange().wait(lock);
    }
    else
    {
      CPPUNIT_ASSERT((entry->fileStatus() == BucketReuseControlEntry::using_c) ||
                     (entry->fileStatus() == BucketReuseControlEntry::ready_c));
    }

    // the bucket files are ready
    dl->restoreBucketInformation();
    dl->endOfInput();

    pthread_join(t1, NULL);
  }

  delete dl;

  cout << "thread " << a->id << " finished at " << atTime() << endl;

  return NULL;
}

timespec atTime()
{
  timespec ts, ts1, ts2;
  ts1 = BucketReUseDriver::ts;
  clock_gettime(CLOCK_REALTIME, &ts2);

  if (ts2.tv_nsec < ts1.tv_nsec)
  {
    ts.tv_sec = ts2.tv_sec - ts1.tv_sec - 1;
    ts.tv_nsec = ts2.tv_nsec + 1000000000 - ts1.tv_nsec;
  }
  else
  {
    ts.tv_sec = ts2.tv_sec - ts1.tv_sec;
    ts.tv_nsec = ts2.tv_nsec - ts1.tv_nsec;
  }

  return ts;
}

ostream& operator<<(ostream& os, const struct timespec& t)
{
  os << t.tv_sec << "." << setw(9) << setfill('0') << t.tv_nsec << "s";
  return os;
}