3

我目前有一个算法，可以散列一个键并使用 map::count 检查它的唯一性。如何优化？我也忘了提到这是线程的。

int coll = 0;
map<long, bool> mymap;
#pragma omp parallel for
for (int i = 0; i < 256; i++)
  for (int j = 0; j < 256; j++)
    for (int k = 0; k < 256; k++)
    {
      string temp;
      temp = i;
      temp += j;
      temp += k;
      temp += temp;
      long myhash = hash(temp.c_str());

     if (mymap.count(myhash))
      {
        #pragma omp atomic
        coll++;
        cout << "Collision at " << i << " " << j << " " << k << endl;
      }
      else
      {
        #pragma omp critical
        mymap[myhash] = true;
      }
    }

cout << "Number of collisions: " << coll << endl;
cout << "Map size: " << mymap.size() << endl;

经过多次试验和错误，这是我可以制作的最佳版本，使用 1GB RAM 在 82.5 秒内生成 4294967296 个密钥。

#include <iostream>
#include <string>
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <sys/time.h>
#include <iomanip>
#include <omp.h>
#include <vector>
#include <fstream>
#include <ios>
#include <unistd.h>
using namespace std;

class Timer 
{
private:
  timeval startTime;

public:

  void start()
  {
    gettimeofday(&startTime, NULL);
  }

  double stop()
  {
    timeval endTime;
    long seconds, useconds;
    double duration;

    gettimeofday(&endTime, NULL);

    seconds  = endTime.tv_sec  - startTime.tv_sec;
    useconds = endTime.tv_usec - startTime.tv_usec;

    duration = seconds + useconds/1000000.0;

    return duration;
  }

  static void printTime(double duration)
  {
    cout << setprecision(10) << fixed << duration << " seconds" << endl;
  }
};

static inline long hash(const char* str)
{
  return (*(long*)str)>> 0;
}  
int coll;
vector<bool> test;

void process_mem_usage(double& vm_usage, double& resident_set)
{
  using std::ios_base;
  using std::ifstream;
  using std::string;

  vm_usage     = 0.0;
  resident_set = 0.0;

  // 'file' stat seems to give the most reliable results
  //
  ifstream stat_stream("/proc/self/stat",ios_base::in);

  // dummy vars for leading entries in stat that we don't care about
  //
  string pid, comm, state, ppid, pgrp, session, tty_nr;
  string tpgid, flags, minflt, cminflt, majflt, cmajflt;
  string utime, stime, cutime, cstime, priority, nice;
  string O, itrealvalue, starttime;

  // the two fields we want
  //
  unsigned long vsize;
  long rss;

  stat_stream >> pid >> comm >> state >> ppid >> pgrp >> session >> tty_nr
          >> tpgid >> flags >> minflt >> cminflt >> majflt >> cmajflt
          >> utime >> stime >> cutime >> cstime >> priority >> nice
          >> O >> itrealvalue >> starttime >> vsize >> rss; // don't care about the rest

  stat_stream.close();

  long page_size_kb = sysconf(_SC_PAGE_SIZE) / 1024; // in case x86-64 is configured to use 2MB pages
  vm_usage     = vsize / 1024.0;
  resident_set = rss * page_size_kb;
}
Timer timer;
void signal_handlerkill(int sig)
{
  cout << "Number of collisions: " << coll << endl;
  //cout << test.size() << endl;
  double vm, rss;
  process_mem_usage(vm, rss);
  vm /= 1024.0;
  rss /= 1024.0;
  cout << "VM: " << vm << "MB" << endl;
  timer.printTime(timer.stop());
  exit(1);
}

int main()
{
  signal(SIGINT, signal_handlerkill);
  timer = Timer();
  timer.start();
  coll = 0;

  for (long i = 0; i < 4294967296+1; i++)
  {
    test.push_back(0); //Set up the vector
  }

  #pragma omp parallel for 
  for (int i = 0; i < 256; i++)
    for (int j = 0; j < 256; j++)
      for (int k = 0; k < 256; k++)
        for (int l = 0; l < 256; l++)
        {
          const char temp[4] = {i, j, k, l};
          long myhash = (*(long*)temp);

          if(test.at(myhash))
          {
            #pragma omp atomic
            coll++;
          }
          else
          {
            test[myhash].flip();
          }
        }

  cout << "Number of collisions: " << coll << endl;
  double vm, rss;
  process_mem_usage(vm, rss);
  vm /= 1024.0;
  rss /= 1024.0;
  cout << "VM: " << vm << "MB" << endl;
  timer.printTime(timer.stop());

  return 0;
}
4

5 回答 5

4

在空间方面，您可以使用 aset而不是 a map，因为该bool值是无用的。

此外，如果您使用 C++11，anunordered_set可能会提供更好的性能。

还，

temp = i;
temp += j;
temp += k;
temp += temp;

stringstream可能比使用 a甚至 char 数组具有更大的开销。

于 2012-05-15T18:14:37.047 回答
3

使用insert而不是operator[]. 插入函数返回一对。第二个值表示，如果该值是实际插入的，即你可以重写你的代码如下：

if (!mymap.insert(std::make_pair(myhash, true)).second) {
    coll++;
    cout << "Collision at " << i << " " << j << " " << k << endl;
}
于 2012-05-15T18:14:15.057 回答
2

好吧，我在这里回答了这个问题：https ://stackoverflow.com/a/10606381/389833 ，它是这样的：

    int coll = 0;
typedef map<long, bool> MY_MAP_TYPE;
MY_MAP_TYPE mymap;
string temp;
long myhash;
for (int i = 0; i < 256; i++)
    for (int j = 0; j < 256; j++)
        for (int k = 0; k < 256; k++)
        {
            temp = i;
            temp += j;
            temp += k;
            temp += temp;
            myhash = hash(temp.c_str());
            if( mymap.insert( MY_MAP_TYPE::value_type( myhash, true ) ).second == false)
            {
                coll++;
                cout << "Collision at " << i << " " << j << " " << k << endl;
            }
        }
于 2012-05-15T18:30:57.107 回答
1

根据散列的大小，您可以用空间换取 CPU 时间，并且只使用布尔向量而不是映射来进行常数时间查找。如果范围是 0 - 256 3（此处唯一值的数量），它应该只需要大约 2 MB，因为许多实现中的 STL 向量将在内部将 bool 向量压缩为位。当然，如果您的哈希函数可以返回非常大的值，例如 2 32甚至 2 64 ，这将不会有效（或者可能根本不起作用） 。

于 2012-05-15T18:26:49.127 回答
1

如果您只关心 6 个字符串，那么您可以通过以下方式轻松优化您生成的循环：

for (int i = 0; i < 256; i++)
  for (int j = 0; j < 256; j++)
    for (int k = 0; k < 256; k++)
    {
/*
      string temp;
      temp = i;
      temp += j;
      temp += k;
      temp += temp;
      myhash = hash(temp.c_str());
*/  
      // effectively, the same as above
      const char temp[7] = {i, j, k, i, j, k, '\0'};
      myhash = hash(temp);
    }

以上与insert所建议的相结合，应该会提供很好的性能提升。

编辑：

所以，你在下面评论这个版本“更慢”让我很怀疑：

  1. 你是如何分析的
  2. 哈希函数的实现

这些都是有问题的，因为在我的机器上运行这段代码（暂时忽略 3.3GHz 幻数，因为那是我 CPU 的速度）：

#include <iostream>
#include <vector>
#include <boost/functional/hash.hpp>
#include <x86intrin.h>

using namespace std;

uint64_t f(std::vector<uint64_t>& values)
{
    boost::hash<std::string> hasher;

    uint64_t start = __rdtsc();
    int z = 0;

    for (int i = 0; i < 256; i++)
    {
      for (int j = 0; j < 256; j++)
      {
        for (int k = 0; k < 256; k++)
        {
          string temp;
          temp = i;
          temp += j;
          temp += k;
          temp += temp;

          values[z++] = hasher(temp);
        }
      }
    }

    return (__rdtsc()) - start;
}

uint64_t g(std::vector<uint64_t>& values)
{
    boost::hash<std::string> hasher;

    uint64_t start = __rdtsc();
    int z = 0;

    for (int i = 0; i < 256; i++)
    {
      for (int j = 0; j < 256; j++)
      {
        for (int k = 0; k < 256; k++)
        {
          const char temp[7] = {i, j, k, i, j, k, '\0'};
          values[z++] = hasher(std::string(temp, 6));
        }
      }
    }

    return (__rdtsc()) - start;
}

static const double freq = 3300000000.0;
static const int elements = 1024 * 1024 * 16;

int main()
{
    std::vector<uint64_t> values_f(elements);
    std::vector<uint64_t> values_g(elements);

    uint64_t delta_f = f(values_f);
    uint64_t delta_g = g(values_g);

    cout << "F: " << (delta_f * 1000.0) / freq << "ms \n";
    cout << "G: " << (delta_g * 1000.0) / freq << "ms \n";

    for(int x = 0; x < elements; ++x)
    {
        if(values_f[x] != values_g[x])
        {
            cout << "Error: Expected "
                 << values_f[x] << " received "
                 << values_g[x] << "!\n";
        }
    }

    return 0;
}

给出这个输出：

F: 3297.17ms 
G: 736.444ms

表明构建的版本std::string（甚至在技术上不是必需的）性能比进行连接的版本好得多。我的情况的不同之处在于使用boost::hash(并且显然使用 astd::vector而不是 a std::mapor std::set，但这并不会使测试偏向于任何一个结果。

于 2012-05-15T19:42:52.450 回答