Revision c2405714
Added by David Sorber over 9 years ago
| software/clustering_proto/cluster.h | ||
|---|---|---|
|
#include <atomic>
|
||
|
#include <cstdint>
|
||
|
|
||
|
const uint32_t NUM_PEERS = 16;
|
||
|
const uint32_t DATA_SIZE = 4;
|
||
|
const uint32_t VECTOR_LEN = 100;
|
||
|
|
||
|
struct cluster_node_t
|
||
|
{
|
||
|
uint32_t id;
|
||
|
std::atomic<uint32_t> lock;
|
||
|
uint32_t peers[NUM_PEERS];
|
||
|
double data[DATA_SIZE];
|
||
|
double tf_vector[VECTOR_LEN];
|
||
|
};
|
||
|
|
||
|
typedef struct cluster_node_t ClusterNode;
|
||
| software/clustering_proto/make_random_nodes.cc | ||
|---|---|---|
|
#include <fstream>
|
||
|
#include <iomanip>
|
||
|
#include <iostream>
|
||
|
#include <mutex>
|
||
|
#include <limits>
|
||
|
#include <random>
|
||
|
#include <set>
|
||
|
#include <sstream>
|
||
|
#include <thread>
|
||
|
#include <vector>
|
||
|
|
||
|
#include <fcntl.h>
|
||
|
#include <unistd.h>
|
||
|
#include <sys/mman.h>
|
||
|
#include <sys/stat.h>
|
||
|
|
||
|
#include "cluster.h"
|
||
|
|
||
|
std::mutex OUTPUT_MUTEX;
|
||
|
|
||
|
#define OUT(x) {\
|
||
|
std::lock_guard<std::mutex> lock(OUTPUT_MUTEX);\
|
||
|
x\
|
||
|
}
|
||
|
|
||
|
/**
|
||
|
* Generate a good random seed using the x86_64 rdtsc register. See:
|
||
|
* http://stackoverflow.com/questions/7617587/is-there-an-alternative-to-using-time-to-seed-a-random-number-generation
|
||
| ... | ... | |
|
return ((unsigned long long)hi << 32) | lo;
|
||
|
}
|
||
|
|
||
|
void usage(char** argv)
|
||
|
{
|
||
|
char* appname = argv[0];
|
||
|
if (argv[0][0] == '.' && argv[0][1] == '/')
|
||
|
{
|
||
|
appname = &argv[0][2];
|
||
|
}
|
||
|
|
||
|
std::cout << "\n" << appname << " <num nodes> <location>\n" << std::endl;
|
||
|
}
|
||
|
|
||
|
void usage()
|
||
|
void create_nodes_thread(
|
||
|
uint32_t tid,
|
||
|
uint32_t start_id,
|
||
|
uint32_t num_nodes,
|
||
|
const std::string& base_path)
|
||
|
{
|
||
|
std::cout << "\nfoobar <num nodes> <location>\n" << std::endl;
|
||
|
uint64_t seed = rdtsc();
|
||
|
OUT(std::cout << "T[" << std::setw(2) << tid << "] seed: 0x" << std::hex
|
||
|
<< seed << std::dec << std::endl;)
|
||
|
|
||
|
uint32_t nodes_created = 0;
|
||
|
for (uint32_t node_id = start_id; node_id < (start_id + num_nodes); ++node_id)
|
||
|
{
|
||
|
// Create a new clutser node
|
||
|
ClusterNode new_node;
|
||
|
new_node.id = node_id;
|
||
|
|
||
|
// Prime out random number generation
|
||
|
std::mt19937_64 generator(rdtsc());
|
||
|
std::uniform_real_distribution<double> tf_dist(0, 1);
|
||
|
std::uniform_int_distribution<uint32_t> term_dist(0, VECTOR_LEN - 1);
|
||
|
|
||
|
// Randomly choose how many terms will have values
|
||
|
uint32_t num_terms = term_dist(generator);
|
||
|
//~ std::cout << "Num terms: " << num_terms << std::endl;
|
||
|
|
||
|
// Create indicies, use a set to guarantee that we end up with unique
|
||
|
// indicies
|
||
|
std::set<uint32_t> indicies;
|
||
|
while (indicies.size() < num_terms)
|
||
|
{
|
||
|
indicies.insert(term_dist(generator));
|
||
|
}
|
||
|
|
||
|
// Create the TF vector be assigning frequencies
|
||
|
std::vector<double> tf_freqs(VECTOR_LEN, 0.0);
|
||
|
double total = 0.0;
|
||
|
for (auto idx : indicies)
|
||
|
{
|
||
|
tf_freqs[idx] = -std::log(tf_dist(generator));
|
||
|
total += tf_freqs[idx];
|
||
|
}
|
||
|
|
||
|
// Now normalize the frequencies
|
||
|
double scaled_total = 0.0;
|
||
|
for (auto idx : indicies)
|
||
|
{
|
||
|
tf_freqs[idx] /= total;
|
||
|
scaled_total += tf_freqs[idx];
|
||
|
}
|
||
|
|
||
|
#if 0
|
||
|
// Print out the frequencies
|
||
|
for (uint32_t idx = 0; idx < VECTOR_LEN; ++idx)
|
||
|
{
|
||
|
std::cout << std::setw(3) << idx << " -- "
|
||
|
<< std::setprecision(std::numeric_limits<double>::digits10)
|
||
|
<< tf_freqs[idx] << std::endl;
|
||
|
}
|
||
|
#endif
|
||
|
|
||
|
#if 0
|
||
|
// Print out normalized total, which should always be 1.0
|
||
|
std::cout << "Scaled total: "
|
||
|
<< std::setprecision(std::numeric_limits<double>::digits10)
|
||
|
<< scaled_total << std::endl;
|
||
|
#endif
|
||
|
|
||
|
// Build filename
|
||
|
std::ostringstream fname;
|
||
|
fname << base_path << "/node_0" << std::setw(10) << std::setfill('0')
|
||
|
<< node_id << ".bin";
|
||
|
|
||
|
// Write out the cluster node file
|
||
|
std::ofstream outfile;
|
||
|
outfile.open(fname.str().c_str(), std::ios::out | std::ios::binary);
|
||
|
if (outfile.is_open())
|
||
|
{
|
||
|
outfile.write(reinterpret_cast<const char*>(&new_node),
|
||
|
sizeof(ClusterNode));
|
||
|
}
|
||
|
outfile.close();
|
||
|
|
||
|
#if 0 // Enable to get per thread status information (this will slow things down)
|
||
|
++nodes_created;
|
||
|
double complete_percent = (double)nodes_created / num_nodes;
|
||
|
complete_percent *= 100;
|
||
|
|
||
|
OUT(std::cout << "T[" << std::setw(2) << tid << "] " << std::setw(6)
|
||
|
<< std::fixed << std::setprecision(2)
|
||
|
<< complete_percent
|
||
|
<< "% -- (" << nodes_created << "/"
|
||
|
<< num_nodes << " -- " << fname.str() << std::endl;)
|
||
|
#endif
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// g++ -std=c++11 -o make_random_nodes make_random_nodes.cc -lpthread
|
||
|
int main(int argc, char** argv)
|
||
|
{
|
||
|
if (argc < 3)
|
||
|
{
|
||
|
std::cerr << "ERROR: not enough arguments!" << std::endl;
|
||
|
usage();
|
||
|
usage(argv);
|
||
|
return -1;
|
||
|
}
|
||
|
|
||
|
// Parse arguments
|
||
|
uint32_t num_nodes = std::strtoul(argv[1], nullptr, -1);
|
||
|
uint32_t total_nodes = std::strtoul(argv[1], nullptr, 0);
|
||
|
std::string base_path(argv[2]);
|
||
|
|
||
|
uint64_t seed = rdtsc();
|
||
|
std::cout << "Seed: 0x" << std::hex << seed << std::dec << std::endl;
|
||
|
std::cout << "Total nodes to create: " << total_nodes << std::endl;
|
||
|
|
||
|
NodeData new_node;
|
||
|
uint32_t num_cores = std::thread::hardware_concurrency();
|
||
|
std::cout << "This machine appears to have " << num_cores << " cores.\n"
|
||
|
<< std::endl;
|
||
|
|
||
|
// Prime out random number generation
|
||
|
std::mt19937_64 generator(rdtsc());
|
||
|
std::uniform_real_distribution<double> tf_dist(0, 1);
|
||
|
std::uniform_int_distribution<uint32_t> term_dist(0, VECTOR_LEN - 1);
|
||
|
|
||
|
// Randomly choose how many terms will have values
|
||
|
uint32_t num_terms = term_dist(generator);
|
||
|
std::cout << "Num terms: " << num_terms << std::endl;
|
||
|
|
||
|
// Create indicies, use a set to guarantee that we end up with unique
|
||
|
// indicies
|
||
|
std::set<uint32_t> indicies;
|
||
|
std::uniform_int_distribution<uint32_t> slot_dist(0, VECTOR_LEN - 1);
|
||
|
while (indicies.size() < num_terms)
|
||
|
if (total_nodes < num_cores)
|
||
|
{
|
||
|
indicies.insert(slot_dist(generator));
|
||
|
// Too few nodes to bother firing up threads... just run directly
|
||
|
create_nodes_thread(1, 1, total_nodes, base_path);
|
||
|
}
|
||
|
|
||
|
// Create the TF vector be assigning frequencies
|
||
|
std::vector<double> tf_freqs(VECTOR_LEN, 0.0);
|
||
|
double total = 0.0;
|
||
|
for (auto idx : indicies)
|
||
|
else
|
||
|
{
|
||
|
tf_freqs[idx] = -std::log(tf_dist(generator));
|
||
|
total += tf_freqs[idx];
|
||
|
}
|
||
|
|
||
|
// Now normalize the frequencies
|
||
|
double scaled_total = 0.0;
|
||
|
for (auto idx : indicies)
|
||
|
{
|
||
|
tf_freqs[idx] /= total;
|
||
|
scaled_total += tf_freqs[idx];
|
||
|
}
|
||
|
|
||
|
#if 1
|
||
|
// Print out the frequencies
|
||
|
for (uint32_t idx = 0; idx < VECTOR_LEN; ++idx)
|
||
|
{
|
||
|
std::cout << std::setw(3) << idx << " -- "
|
||
|
<< std::setprecision(std::numeric_limits<double>::digits10)
|
||
|
<< tf_freqs[idx] << std::endl;
|
||
|
std::vector<std::thread*> threads;
|
||
|
|
||
|
uint32_t nodes_per_thread = total_nodes / num_cores;
|
||
|
|
||
|
// Spawn threads
|
||
|
for (uint32_t tid = 0; tid < num_cores; ++tid)
|
||
|
{
|
||
|
uint32_t start_id = (tid * nodes_per_thread) + 1;
|
||
|
|
||
|
// Hand "tail" to the last thread
|
||
|
if (tid == (num_cores - 1))
|
||
|
{
|
||
|
nodes_per_thread += (total_nodes % num_cores);
|
||
|
}
|
||
|
|
||
|
std::thread* thread = new std::thread(&create_nodes_thread, tid,
|
||
|
start_id, nodes_per_thread,
|
||
|
base_path);
|
||
|
threads.push_back(thread);
|
||
|
}
|
||
|
|
||
|
// Join threads
|
||
|
for (auto thread_ptr : threads)
|
||
|
{
|
||
|
if (thread_ptr->joinable())
|
||
|
{
|
||
|
thread_ptr->join();
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
#endif
|
||
|
//~ create_nodes_thread(1, 1, total_nodes, base_path);
|
||
|
|
||
|
|
||
|
// Print out normalized total, which should always be 1.0
|
||
|
std::cout << "Scaled total: "
|
||
|
<< std::setprecision(std::numeric_limits<double>::digits10)
|
||
|
<< scaled_total << std::endl;
|
||
|
|
||
|
return 0;
|
||
|
}
|
||
Adding header file that I forgot to add last time around. I also finished up a multithreaded version of the make random nodes script that can create 1 million nodes in ~20 seconds.