PartMerge.cpp

/*
 * This file is part of the BSGS distribution (https://github.com/JeanLucPons/Kangaroo).
 * Copyright (c) 2020 Jean Luc PONS.
 *
 * 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 3.
 *
 * 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, see <http://www.gnu.org/licenses/>.
 */

#include "Kangaroo.h"
#include <fstream>
#include "SECPK1/IntGroup.h"
#include "Timer.h"
#include <string.h>
#define _USE_MATH_DEFINES
#include <math.h>
#include <algorithm>
#ifndef WIN64
#include <dirent.h>
#include <pthread.h>
#include <sys/stat.h>
#define _strdup strdup
#endif

using namespace std;

string Kangaroo::GetPartName(std::string &partName, int i, bool tmpPart)
{

  char tmp[256];
  if (tmpPart)
    sprintf(tmp, "part%03d.tmp", i);
  else
    sprintf(tmp, "part%03d", i);
  string pName = partName + "/" + string(tmp);

  return pName;
}

FILE *Kangaroo::OpenPart(std::string &partName, char *mode, int i, bool tmpPart)
{

  string fName = GetPartName(partName, i, tmpPart);
  FILE *f = fopen(fName.c_str(), mode);
  if (f == NULL)
  {
    ::printf("OpenPart: Cannot open %s for mode %s\n", fName.c_str(), mode);
    ::printf("%s\n", ::strerror(errno));
  }

  return f;
}

void Kangaroo::CreateEmptyPartWork(std::string &partName)
{

#ifdef WIN64

  WIN32_FIND_DATA ffd;
  HANDLE hFind;

  hFind = FindFirstFile(partName.c_str(), &ffd);
  if (hFind != INVALID_HANDLE_VALUE)
  {
    ::printf("CreateEmptyPartWork: %s exists\n", partName.c_str());
    return;
  }

  if (CreateDirectory(partName.c_str(), NULL) == 0)
  {
    ::printf("CreateDirectory Error: %d\n", GetLastError());
    return;
  }

#else

  struct stat buffer;
  if (stat(partName.c_str(), &buffer) == 0)
  {
    ::printf("CreateEmptyPartWork: %s exists\n", partName.c_str());
    return;
  }

  if (mkdir(partName.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH) != 0)
  {
    ::printf("mkdir(%s) Error:\n", partName.c_str());
    perror("");
    return;
  }

#endif

  // Header
  string hName = partName + "/header";
  FILE *f = fopen(hName.c_str(), "wb");
  if (f == NULL)
  {
    ::printf("CreateEmptyPartWork: Cannot open %s for writing\n", partName.c_str());
    ::printf("%s\n", ::strerror(errno));
    return;
  }

  fclose(f);

  // Part
  for (int i = 0; i < MERGE_PART; i++)
  {

    FILE *f = OpenPart(partName, "wb", i);
    if (f == NULL)
      return;

    for (int j = 0; j < H_PER_PART; j++)
    {
      uint32_t z = 0;
      fwrite(&z, sizeof(uint32_t), 1, f);
      fwrite(&z, sizeof(uint32_t), 1, f);
    }

    fclose(f);
  }

  ::printf("CreateEmptyPartWork %s done\n", partName.c_str());
}

bool Kangaroo::MergePartition(TH_PARAM *p)
{

  uint32_t part = p->hStart;
  string p1Name = string(p->part1Name);
  string p2Name = string(p->part2Name);

  FILE *f1 = OpenPart(p1Name, "rb", part, false);
  if (f1 == NULL)
    return false;
  FILE *f2 = OpenPart(p2Name, "rb", part, false);
  if (f2 == NULL)
    return false;
  FILE *f = OpenPart(p1Name, "wb", part, true);
  if (f == NULL)
    return false;

  uint32_t hStart = part * (HASH_SIZE / MERGE_PART);
  uint32_t hStop = (part + 1) * (HASH_SIZE / MERGE_PART);

  uint32_t hDP;
  uint32_t hDuplicate;
  Int d1;
  uint32_t type1;
  Int d2;
  uint32_t type2;

  for (uint32_t h = hStart; h < hStop && !endOfSearch; h++)
  {

    int mStatus = HashTable::MergeH(h, f1, f2, f, &hDP, &hDuplicate, &d1, &type1, &d2, &type2);
    switch (mStatus)
    {
    case ADD_OK:
      break;
    case ADD_COLLISION:
      CollisionCheck(&d1, type1, &d2, type2);
      break;
    }

    // Counting overflow after (2^32)*MARGE_PART DP
    p->hStop += hDP;
    collisionInSameHerd += hDuplicate;
  }

  ::fclose(f1);
  ::fclose(f2);
  ::fclose(f);

  // Rename
  string oldName = GetPartName(p1Name, part, true);
  string newName = GetPartName(p1Name, part, false);
  if (!endOfSearch)
  {
    remove(newName.c_str());
    rename(oldName.c_str(), newName.c_str());
  }
  else
  {
    remove(oldName.c_str());
  }

  return true;
}
// Threaded proc
#ifdef WIN64
extern DWORD WINAPI _mergeThread(LPVOID lpParam);
#else
extern void *_mergeThread(void *lpParam);
#endif

// Threaded proc
#ifdef WIN64
DWORD WINAPI _mergePartThread(LPVOID lpParam)
{
#else
void *_mergePartThread(void *lpParam)
{
#endif
  TH_PARAM *p = (TH_PARAM *)lpParam;
  p->obj->MergePartition(p);
  p->isRunning = false;
  return 0;
}

bool Kangaroo::MergeWorkPartPart(std::string &part1Name, std::string &part2Name)
{

  double t0;
  double t1;
  uint32_t v1;
  uint32_t v2;

  t0 = Timer::get_tick();

  // ---------------------------------------------------
  string file1 = part1Name + "/header";
  bool partIsEmpty = IsEmpty(file1);
  string file2 = part2Name + "/header";
  if (IsEmpty(file2))
  {
    ::printf("MergeWorkPartPart: partition #2 is empty\n");
    return true;
  }

  uint32_t dp1;
  Point k1;
  uint64_t count1;
  double time1;
  Int RS1;
  Int RE1;

  if (!partIsEmpty)
  {

    FILE *f1 = ReadHeader(file1, &v1, HEADW);
    if (f1 == NULL)
      return false;

    // Read global param
    ::fread(&dp1, sizeof(uint32_t), 1, f1);
    ::fread(&RS1.bits64, 32, 1, f1);
    RS1.bits64[4] = 0;
    ::fread(&RE1.bits64, 32, 1, f1);
    RE1.bits64[4] = 0;
    ::fread(&k1.x.bits64, 32, 1, f1);
    k1.x.bits64[4] = 0;
    ::fread(&k1.y.bits64, 32, 1, f1);
    k1.y.bits64[4] = 0;
    ::fread(&count1, sizeof(uint64_t), 1, f1);
    ::fread(&time1, sizeof(double), 1, f1);

    k1.z.SetInt32(1);
    if (!secp->EC(k1))
    {
      ::printf("MergeWorkPartPart: key1 does not lie on elliptic curve\n");
      ::fclose(f1);
      return true;
    }
    ::fclose(f1);
  }

  // ---------------------------------------------------

  FILE *f2 = ReadHeader(file2, &v2, HEADW);
  if (f2 == NULL)
  {
    return true;
  }

  uint32_t dp2;
  Point k2;
  uint64_t count2;
  double time2;
  Int RS2;
  Int RE2;

  // Read global param
  ::fread(&dp2, sizeof(uint32_t), 1, f2);
  ::fread(&RS2.bits64, 32, 1, f2);
  RS2.bits64[4] = 0;
  ::fread(&RE2.bits64, 32, 1, f2);
  RE2.bits64[4] = 0;
  ::fread(&k2.x.bits64, 32, 1, f2);
  k2.x.bits64[4] = 0;
  ::fread(&k2.y.bits64, 32, 1, f2);
  k2.y.bits64[4] = 0;
  ::fread(&count2, sizeof(uint64_t), 1, f2);
  ::fread(&time2, sizeof(double), 1, f2);

  k2.z.SetInt32(1);
  if (!secp->EC(k2))
  {
    ::printf("MergeWorkPartPart: key2 does not lie on elliptic curve\n");
    ::fclose(f2);
    return true;
  }

  if (!partIsEmpty)
  {

    if (v1 != v2)
    {
      ::printf("MergeWorkPartPart: cannot merge workfile of different version\n");
      ::fclose(f2);
      return true;
    }

    if (!RS1.IsEqual(&RS2) || !RE1.IsEqual(&RE2))
    {

      ::printf("MergeWorkPartPart: File range differs\n");
      ::printf("RS1: %s\n", RS1.GetBase16().c_str());
      ::printf("RE1: %s\n", RE1.GetBase16().c_str());
      ::printf("RS2: %s\n", RS2.GetBase16().c_str());
      ::printf("RE2: %s\n", RE2.GetBase16().c_str());
      ::fclose(f2);
      return true;
    }

    if (!k1.equals(k2))
    {

      ::printf("MergeWorkPartPart: key differs, multiple keys not yet supported\n");
      ::fclose(f2);
      return true;
    }
  }
  else
  {

    dp1 = dp2;
    k1 = k2;
    count1 = 0;
    time1 = 0;
    RS1.Set(&RS2);
    RE1.Set(&RE2);
  }
  fclose(f2);

  ::printf("%s: [DP%d]\n", part1Name.c_str(), dp1);
  ::printf("%s: [DP%d]\n", part2Name.c_str(), dp2);

  // Set starting parameters
  endOfSearch = false;
  keysToSearch.clear();
  keysToSearch.push_back(k1);
  keyIdx = 0;
  collisionInSameHerd = 0;
  rangeStart.Set(&RS1);
  rangeEnd.Set(&RE1);
  InitRange();
  InitSearchKey();

  // Write new header
  FILE *f = fopen(file1.c_str(), "wb");
  if (f == NULL)
  {
    ::printf("MergeWorkPart: Cannot open %s for writing\n", file1.c_str());
    ::printf("%s\n", ::strerror(errno));
    fclose(f2);
    return true;
  }
  dpSize = (dp1 < dp2) ? dp1 : dp2;
  if (!SaveHeader(file1, f, HEADW, count1 + count2, time1 + time2))
  {
    fclose(f2);
    return true;
  }
  fclose(f);

  int nbCore = Timer::getCoreNumber();
  int l2 = (int)log2(nbCore);
  int nbThread = (int)pow(2.0, l2);
  if (nbThread > 16)
    nbThread = 16;

#ifndef WIN64
  setvbuf(stdout, NULL, _IONBF, 0);
#endif

  ::printf("Thread: %d\n", nbThread);
  ::printf("Merging");

  TH_PARAM *params = (TH_PARAM *)malloc(nbThread * sizeof(TH_PARAM));
  THREAD_HANDLE *thHandles = (THREAD_HANDLE *)malloc(nbThread * sizeof(THREAD_HANDLE));
  memset(params, 0, nbThread * sizeof(TH_PARAM));
  uint64_t nbDP = 0;

  for (int p = 0; p < MERGE_PART && !endOfSearch; p += nbThread)
  {

    printf(".");

    for (int i = 0; i < nbThread; i++)
    {
      params[i].threadId = i;
      params[i].isRunning = true;
      params[i].hStart = p + i;
      params[i].hStop = 0;
      params[i].part1Name = _strdup(part1Name.c_str());
      params[i].part2Name = _strdup(part2Name.c_str());
      thHandles[i] = LaunchThread(_mergePartThread, params + i);
    }

    JoinThreads(thHandles, nbThread);
    FreeHandles(thHandles, nbThread);

    for (int i = 0; i < nbThread; i++)
    {
      free(params[i].part1Name);
      free(params[i].part2Name);
      nbDP += params[i].hStop;
    }
  }

  free(params);
  free(thHandles);

  t1 = Timer::get_tick();

  if (!endOfSearch)
  {

    ::printf("Done [2^%.3f DP][%s]\n", log2((double)nbDP), GetTimeStr(t1 - t0).c_str());
  }
  else
  {

#ifdef WIN64
    ::printf("Dead kangaroo: %I64d\n", collisionInSameHerd);
#else
    ::printf("Dead kangaroo: %" PRId64 "\n", collisionInSameHerd);
#endif
    ::printf("Total f1+f2: DP count 2^%.2f\n", log2((double)nbDP));
    return true;
  }

#ifdef WIN64
  ::printf("Dead kangaroo: %I64d\n", collisionInSameHerd);
#else
  ::printf("Dead kangaroo: %" PRId64 "\n", collisionInSameHerd);
#endif
  ::printf("Total f1+f2: DP count 2^%.2f\n", log2((double)nbDP));

  return false;
}

bool Kangaroo::FillEmptyPartFromFile(std::string &partName, std::string &fileName, bool printStat)
{

  double t0;
  double t1;
  uint32_t v1;

  uint32_t dp1;
  Point k1;
  uint64_t count1;
  double time1;
  Int RS1;
  Int RE1;

  t0 = Timer::get_tick();

  FILE *f1 = ReadHeader(fileName, &v1, HEADW);
  if (f1 == NULL)
    return true;

  // Read global param
  ::fread(&dp1, sizeof(uint32_t), 1, f1);
  ::fread(&RS1.bits64, 32, 1, f1);
  RS1.bits64[4] = 0;
  ::fread(&RE1.bits64, 32, 1, f1);
  RE1.bits64[4] = 0;
  ::fread(&k1.x.bits64, 32, 1, f1);
  k1.x.bits64[4] = 0;
  ::fread(&k1.y.bits64, 32, 1, f1);
  k1.y.bits64[4] = 0;
  ::fread(&count1, sizeof(uint64_t), 1, f1);
  ::fread(&time1, sizeof(double), 1, f1);

  k1.z.SetInt32(1);
  if (!secp->EC(k1))
  {
    ::printf("FillEmptyPartFromFile: key1 does not lie on elliptic curve\n");
    ::fclose(f1);
    return true;
  }

  // Save header
  dpSize = dp1;
  keysToSearch.clear();
  keysToSearch.push_back(k1);
  keyIdx = 0;
  collisionInSameHerd = 0;
  rangeStart.Set(&RS1);
  rangeEnd.Set(&RE1);
  InitRange();
  InitSearchKey();

  string file1 = partName + "/header";
  FILE *f = fopen(file1.c_str(), "wb");
  if (f == NULL)
  {
    ::printf("FillEmptyPartFromFile: Cannot open %s for writing\n", file1.c_str());
    ::printf("%s\n", ::strerror(errno));
    return true;
  }
  if (!SaveHeader(file1, f, HEADW, count1, time1))
  {
    return true;
  }
  ::fclose(f);

  ::printf("Part %s: [DP%d]\n", partName.c_str(), dp1);
  ::printf("File %s: [DP%d]\n", fileName.c_str(), dp1);

  uint64_t nbDP = 0;
  ::printf("Filling");

  // Save parts
  for (int p = 0; p < MERGE_PART; p++)
  {

    if (p % (MERGE_PART / 64) == 0)
      ::printf(".");

    FILE *f = OpenPart(partName, "wb", p, false);
    uint32_t hStart = p * (HASH_SIZE / MERGE_PART);
    uint32_t hStop = (p + 1) * (HASH_SIZE / MERGE_PART);

    uint32_t nbItem;
    uint32_t maxItem;
    unsigned char buff[32];

    for (uint32_t h = hStart; h < hStop; h++)
    {
      ::fread(&nbItem, sizeof(uint32_t), 1, f1);
      ::fread(&maxItem, sizeof(uint32_t), 1, f1);
      ::fwrite(&nbItem, sizeof(uint32_t), 1, f);
      ::fwrite(&maxItem, sizeof(uint32_t), 1, f);
      for (uint32_t i = 0; i < nbItem; i++)
      {
        ::fread(&buff, 32, 1, f1);
        ::fwrite(&buff, 32, 1, f);
      }
      nbDP += nbItem;
    }

    ::fclose(f);
  }

  ::fclose(f1);
  t1 = Timer::get_tick();

  ::printf("Done [2^%.3f DP][%s]\n", log2((double)nbDP), GetTimeStr(t1 - t0).c_str());

  return false;
}

bool Kangaroo::MergeWorkPart(std::string &partName, std::string &file2, bool printStat)
{

  double t0;
  double t1;
  uint32_t v1;
  uint32_t v2;

#ifndef WIN64
  setvbuf(stdout, NULL, _IONBF, 0);
#endif

  t0 = Timer::get_tick();

  // ---------------------------------------------------
  string file1 = partName + "/header";
  if (IsEmpty(file1))
    return FillEmptyPartFromFile(partName, file2, printStat);

  uint32_t dp1;
  Point k1;
  uint64_t count1;
  double time1;
  Int RS1;
  Int RE1;

  FILE *f1 = ReadHeader(file1, &v1, HEADW);
  if (f1 == NULL)
    return true;

  // Read global param
  ::fread(&dp1, sizeof(uint32_t), 1, f1);
  ::fread(&RS1.bits64, 32, 1, f1);
  RS1.bits64[4] = 0;
  ::fread(&RE1.bits64, 32, 1, f1);
  RE1.bits64[4] = 0;
  ::fread(&k1.x.bits64, 32, 1, f1);
  k1.x.bits64[4] = 0;
  ::fread(&k1.y.bits64, 32, 1, f1);
  k1.y.bits64[4] = 0;
  ::fread(&count1, sizeof(uint64_t), 1, f1);
  ::fread(&time1, sizeof(double), 1, f1);

  k1.z.SetInt32(1);
  if (!secp->EC(k1))
  {
    ::printf("MergeWorkPart: key1 does not lie on elliptic curve\n");
    ::fclose(f1);
    return true;
  }
  ::fclose(f1);

  // ---------------------------------------------------

  FILE *f2 = ReadHeader(file2, &v2, HEADW);
  if (f2 == NULL)
  {
    return true;
  }

  uint32_t dp2;
  Point k2;
  uint64_t count2;
  double time2;
  Int RS2;
  Int RE2;

  // Read global param
  ::fread(&dp2, sizeof(uint32_t), 1, f2);
  ::fread(&RS2.bits64, 32, 1, f2);
  RS2.bits64[4] = 0;
  ::fread(&RE2.bits64, 32, 1, f2);
  RE2.bits64[4] = 0;
  ::fread(&k2.x.bits64, 32, 1, f2);
  k2.x.bits64[4] = 0;
  ::fread(&k2.y.bits64, 32, 1, f2);
  k2.y.bits64[4] = 0;
  ::fread(&count2, sizeof(uint64_t), 1, f2);
  ::fread(&time2, sizeof(double), 1, f2);

  k2.z.SetInt32(1);
  if (!secp->EC(k2))
  {
    ::printf("MergeWorkPart: key2 does not lie on elliptic curve\n");
    ::fclose(f2);
    return true;
  }

  if (v1 != v2)
  {
    ::printf("MergeWorkPart: cannot merge workfile of different version\n");
    ::fclose(f2);
    return true;
  }

  if (!RS1.IsEqual(&RS2) || !RE1.IsEqual(&RE2))
  {

    ::printf("MergeWorkPart: File range differs\n");
    ::printf("RS1: %s\n", RS1.GetBase16().c_str());
    ::printf("RE1: %s\n", RE1.GetBase16().c_str());
    ::printf("RS2: %s\n", RS2.GetBase16().c_str());
    ::printf("RE2: %s\n", RE2.GetBase16().c_str());
    ::fclose(f2);
    return true;
  }

  if (!k1.equals(k2))
  {

    ::printf("MergeWorkPart: key differs, multiple keys not yet supported\n");
    ::fclose(f2);
    return true;
  }

  ::printf("Part %s: [DP%d]\n", partName.c_str(), dp1);
  ::printf("File %s: [DP%d]\n", file2.c_str(), dp2);

  endOfSearch = false;

  // Set starting parameters
  keysToSearch.clear();
  keysToSearch.push_back(k1);
  keyIdx = 0;
  collisionInSameHerd = 0;
  rangeStart.Set(&RS1);
  rangeEnd.Set(&RE1);
  InitRange();
  InitSearchKey();

  t0 = Timer::get_tick();

  ::printf("Merging");

  // Save header
  FILE *f = fopen(file1.c_str(), "wb");
  if (f == NULL)
  {
    ::printf("MergeWorkPart: Cannot open %s for writing\n", file1.c_str());
    ::printf("%s\n", ::strerror(errno));
    fclose(f2);
    return true;
  }
  dpSize = (dp1 < dp2) ? dp1 : dp2;
  if (!SaveHeader(file1, f, HEADW, count1 + count2, time1 + time2))
  {
    fclose(f2);
    return true;
  }
  fclose(f);

  uint64_t nbDP = 0;
  uint32_t hDP;
  uint32_t hDuplicate;
  Int d1;
  uint32_t type1;
  Int d2;
  uint32_t type2;

  for (int part = 0; part < MERGE_PART && !endOfSearch; part++)
  {

    if (part % (MERGE_PART / 64) == 0)
      ::printf(".");

    uint32_t hStart = part * (HASH_SIZE / MERGE_PART);
    uint32_t hStop = (part + 1) * (HASH_SIZE / MERGE_PART);

    // Load hashtables
    FILE *f1 = OpenPart(partName, "rb", part);
    FILE *f = OpenPart(partName, "wb", part, true);

    for (uint32_t h = hStart; h < hStop && !endOfSearch; h++)
    {

      int mStatus = HashTable::MergeH(h, f1, f2, f, &hDP, &hDuplicate, &d1, &type1, &d2, &type2);
      switch (mStatus)
      {
      case ADD_OK:
        break;
      case ADD_COLLISION:
        CollisionCheck(&d1, type1, &d2, type2);
        break;
      }

      nbDP += hDP;
      collisionInSameHerd += hDuplicate;
    }

    fclose(f1);
    fclose(f);

    // Rename
    string oldName = GetPartName(partName, part, true);
    string newName = GetPartName(partName, part, false);
    if (!endOfSearch)
    {
      remove(newName.c_str());
      rename(oldName.c_str(), newName.c_str());
    }
    else
    {
      remove(oldName.c_str());
    }
  }

  fclose(f2);

  t1 = Timer::get_tick();

  if (!endOfSearch)
  {

    ::printf("Done [2^%.3f DP][%s]\n", log2((double)nbDP), GetTimeStr(t1 - t0).c_str());
  }
  else
  {

#ifdef WIN64
    ::printf("Dead kangaroo: %I64d\n", collisionInSameHerd);
#else
    ::printf("Dead kangaroo: %" PRId64 "\n", collisionInSameHerd);
#endif
    ::printf("Total f1+f2: DP count 2^%.2f\n", log2((double)nbDP));
    return true;
  }

  if (printStat)
  {
#ifdef WIN64
    ::printf("Dead kangaroo: %I64d\n", collisionInSameHerd);
#else
    ::printf("Dead kangaroo: %" PRId64 "\n", collisionInSameHerd);
#endif
    ::printf("Total f1+f2: DP count 2^%.2f\n", log2((double)nbDP));
  }
  else
  {
    offsetTime = time1 + time2;
    offsetCount = count1 + count2;
  }

  return false;
}