Project

General

Profile

« Previous | Next » 

Revision c2405714

Added by David Sorber over 9 years ago

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.

View differences:

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;
}

Also available in: Unified diff