apaam/mlpack
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Simulated kdB-trees in THOR experiment

Browse Source
This commit is contained in:
rriegel
2008-07-17 19:14:19 +00:00
parent a346dae0ae
commit 594c69cd54
13 changed files with 951 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastlib2/fastlib/thor/allnn.cc
+6 -2
View File
@@ -6,7 +6,7 @@
* TODO: Currently doesn't output anything.
*/
#include "thor/thor.h"
#include "fastlib/thor/thor.h"
#include "fastlib/fastlib.h"
/**
@@ -83,6 +83,7 @@ class Allnn {
const QPostponed& postponed,
const QPoint& q_point,
index_t q_index) {}
void Seed(const Param& param, const QPoint& q_point) {}
};
struct QSummaryResult {
@@ -120,6 +121,8 @@ class Allnn {
}
void FinishReaccumulate(const Param& param, const QNode& q_node) {}
void Seed(const Param& param, const QNode& q_node) {}
};
/**
@@ -138,6 +141,7 @@ class Allnn {
const QPoint& q_point,
index_t q_index,
const RNode& r_node,
const Delta& delta,
const QSummaryResult& unapplied_summary_results,
QResult* q_result,
GlobalResult* global_result) {
@@ -196,7 +200,7 @@ class Allnn {
*/
static bool ConsiderPairIntrinsic(const Param& param,
const QNode& q_node, const RNode& r_node,
Delta* delta,
const Delta& parent_delta, Delta* delta,
GlobalResult* global_result, QPostponed* q_postponed) {
return true;
}
fastlib2/fastlib/thor/build.py
+2
View File
@@ -12,6 +12,8 @@ distribcache.h
gnp.h
kdtree.h
kdtree_impl.h
kdbtree.h
kdbtree_impl.h
rpc_base.h
rpc.h
rpc_sock.h
fastlib2/fastlib/thor/dfs.h
+4 -3
View File
@@ -17,7 +17,7 @@ template<typename GNP>
class DualTreeDepthFirst {
FORBID_ACCIDENTAL_COPIES(DualTreeDepthFirst);
private:
public:
struct QMutables {
typename GNP::QSummaryResult summary_result;
typename GNP::QPostponed postponed;
@@ -35,7 +35,7 @@ class DualTreeDepthFirst {
CacheArray<typename GNP::QPoint> q_points_;
CacheArray<typename GNP::QNode> q_nodes_;
CacheArray<typename GNP::QResult> q_results_;
SubsetArray<QMutables> q_mutables_;
CacheArray<QMutables> q_mutables_; // SubsetArray
CacheArray<typename GNP::RPoint> r_points_;
CacheArray<typename GNP::RNode> r_nodes_;
@@ -64,7 +64,8 @@ class DualTreeDepthFirst {
DistributedCache *q_nodes,
DistributedCache *r_points,
DistributedCache *r_nodes,
DistributedCache *q_results);
DistributedCache *q_results,
DistributedCache *q_mutables); //
/**
* Gets the global result after computation.
fastlib2/fastlib/thor/dfs_impl.h
+32 -21
View File
@@ -18,14 +18,36 @@ void DualTreeDepthFirst<GNP>::Doit(
DistributedCache *q_nodes,
DistributedCache *r_points,
DistributedCache *r_nodes,
DistributedCache *q_results) {
DistributedCache *q_results,
DistributedCache *q_mutables) { //
param_.Copy(param_in);
q_nodes_.Init(q_nodes, BlockDevice::M_READ);
r_points_.Init(r_points, BlockDevice::M_READ);
r_nodes_.Init(r_nodes, BlockDevice::M_READ);
r_points_.Init(r_points, BlockDevice::M_READ);
q_mutables_.Init(q_mutables, BlockDevice::M_OVERWRITE,
q_root_index, q_end_index); //
q_nodes_.Init(q_nodes, BlockDevice::M_READ);
//QMutables default_mutable;
//default_mutable.summary_result.Init(param_);
//default_mutable.postponed.Init(param_);
//q_mutables_.Init(default_mutable, q_root_index, q_end_index);
// Seed summary_results
{
CacheWriteIter<QMutables> q_mutables_iter(&q_mutables_, //
q_root_index);
CacheReadIter<typename GNP::QNode> q_nodes_iter(&q_nodes_,
q_root_index);
for (int i = q_root_index; i < q_end_index; ++i,
q_mutables_iter.Next(), q_nodes_iter.Next()) {
q_mutables_iter->summary_result.Seed(param_, *q_nodes_iter);
}
}
const typename GNP::QNode *q_root = q_nodes_.StartRead(q_root_index);
q_results_.Init(q_results, BlockDevice::M_OVERWRITE,
q_root->begin(), q_root->end());
q_points_.Init(q_points, BlockDevice::M_READ,
@@ -39,26 +61,12 @@ void DualTreeDepthFirst<GNP>::Doit(
q_root->begin());
for (int i = q_root->begin(); i < q_root->end(); ++i,
q_results_iter.Next(), q_points_iter.Next()) {
(*q_results_iter).Seed(param_, *q_points_iter);
q_results_iter->Seed(param_, *q_points_iter);
}
}
q_nodes_.StopRead(q_root_index);
QMutables default_mutable;
default_mutable.summary_result.Init(param_);
default_mutable.postponed.Init(param_);
q_mutables_.Init(default_mutable, q_root_index, q_end_index);
// Seed summary_results
{
CacheReadIter<typename GNP::QNode> q_nodes_iter(&q_nodes_,
q_root_index);
for (int i = q_root_index; i < q_end_index; ++i, q_nodes_iter.Next()) {
q_mutables_[i].summary_result.Seed(param_, *q_nodes_iter);
}
}
global_result_.Init(param_);
r_root_ = r_nodes_.StartRead(0);
@@ -73,7 +81,8 @@ void DualTreeDepthFirst<GNP>::Begin_(index_t q_root_index) {
typename GNP::Delta delta;
typename GNP::Delta empty_delta;
CacheRead<typename GNP::QNode> q_root(&q_nodes_, q_root_index);
QMutables *q_root_mut = &q_mutables_[q_root_index];
//QMutables *q_root_mut = &q_mutables_[q_root_index];
CacheWrite<QMutables> q_root_mut(&q_mutables_, q_root_index);
empty_delta.Init(param_);
delta.Init(param_);
@@ -118,7 +127,8 @@ void DualTreeDepthFirst<GNP>::PushDownPostprocess_(
} else {
for (index_t k = 0; k < 2; k++) {
index_t q_child_i = q_node->child(k);
QMutables *q_child_mut = &q_mutables_[q_child_i];
//QMutables *q_child_mut = &q_mutables_[q_child_i];
CacheWrite<QMutables> q_child_mut(&q_mutables_, q_child_i);
q_child_mut->postponed.ApplyPostponed(param_, q_node_mut->postponed);
@@ -169,7 +179,8 @@ void DualTreeDepthFirst<GNP>::Pair_(
typename GNP::Delta child_delta;
index_t q_child_i = q_node->child(k);
CacheRead<typename GNP::QNode> q_child(&q_nodes_, q_child_i);
QMutables *q_child_mut = &q_mutables_[q_child_i];
//QMutables *q_child_mut = &q_mutables_[q_child_i];
CacheWrite<QMutables> q_child_mut(&q_mutables_, q_child_i);
child_delta.Init(param_);
q_child_mut->postponed.ApplyPostponed(
fastlib2/fastlib/thor/kdbtree.h
+161
View File
@@ -0,0 +1,161 @@
// Copyright 2008 Georgia Institute of Technology. All rights reserved.
/**
* @file thor/kdbtree.h
*
* Tools for kdB-trees.
*
* @experimental
*/
#ifndef THOR_KDBTREE_H
#define THOR_KDBTREE_H
#include "fastlib/base/base.h"
#include "thortree.h"
#include "cachearray.h"
#include "kdtree.h"
#include "fastlib/file/textfile.h"
#include "fastlib/data/dataset.h"
#include "fastlib/tree/bounds.h"
#include "fastlib/col/arraylist.h"
#include "fastlib/fx/fx.h"
/**
* Single-threaded kdB-tree builder.
*
* Mimics adding points one at a time in accordance with the splitting
* rules for kdB_FD trees. Specifically, splits are always median
* splits, and the first split made for as many points can fit in a
* block is reused for all points that ultimately inhabit that region.
* Nodes are maintained in a height-balanced data layout (broken only
* to avoid creating pages containing just one subregion).
*
* To assist with dual-tree computation, this structure stores more
* information for binary nodes than a traditional kdB-tree. It keeps
* bounding boxes and cached statistics. It also forms nodes beneath
* the minimum level needed to fit a region's points within one block
* to optimize for base-case performance.
*/
template<typename TPoint, typename TNode, typename TParam>
class KdBTreeBuilder {
public:
typedef TNode Node;
typedef TPoint Point;
typedef typename TNode::Bound Bound;
typedef TParam Param;
typedef ThorTreeDecomposition<Node> TreeDecomposition;
typedef typename TreeDecomposition::DecompNode DecompNode;
private:
struct HrectPartitionCondition {
int dimension;
double value;
HrectPartitionCondition(int dimension_in, double value_in)
: dimension(dimension_in)
, value(value_in) {}
bool is_left(const Vector& vector) const {
return vector.get(dimension) < value;
}
};
private:
const Param* param_;
CacheArray<Point> inputs_;
CacheArray<Point> points_;
CacheArray<Node> nodes_;
index_t point_block_size_;
index_t node_block_size_;
index_t leaf_size_;
index_t n_points_;
index_t spare_block_i_;
public:
/**
* Builds a kdB-tree.
*
* See class comments.
*/
void Doit(
fx_module* module, const Param* param_in,
index_t begin_index, index_t end_index,
DistributedCache* input_in,
DistributedCache* points_create,
DistributedCache* nodes_create,
TreeDecomposition* decomposition);
private:
void FindBoundingBox_(index_t begin_index, index_t end_index, Bound* bound);
index_t Build_(index_t begin_col, index_t end_col);
void Insert_(const Point &input, index_t node_i);
index_t SplitLeaf_(index_t node_i, bool make_new_block);
index_t CreateChildren_(index_t node_i);
index_t PackNodes_(
index_t dest_i, index_t parent_i, Node *parent,
index_t offset_i, index_t *node_ip);
void FixChildrenParents_(index_t parent_i, const Node *parent);
void MedianSplit_(const Node *node, Bound *left_bound, Bound *right_bound);
index_t SplitLeafPages_(index_t node_i);
void Postprocess_(Node *node);
};
namespace thor {
/**
* Creates a THOR kd-tree.
*
* @param param the parameter object for initializations
* @param points_channel the channel for the points cache
* @param nodes_channel the channel for the nodes cache
* @param block_size_kb the upper-limit block size in kilobytes
* @param megs the number of megabytes dedicated to the nodes cache
* @param module where to get tree-building parameters such as leaf_size
* @param n_points the number of data points
* @param input_cache the cache of points to read from
* @param points_cache the cache of points to initialize and create
* @param nodes_cache the cache of nodes to initialize and create
* @param decomposition the resulting tree decomposition
*/
template<typename Point, typename Node, typename Param>
void CreateKdBTreeMaster(const Param& param,
int points_channel, int nodes_channel,
int block_size_kb, double megs, datanode *module,
index_t n_points, DistributedCache *input_cache,
DistributedCache *points_cache, DistributedCache *nodes_cache,
ThorTreeDecomposition<Node> *decomposition);
/**
* Creates a THOR kd-tree from an existing cache.
*
* Parameters taken from the @c module data node:
* @li @c leaf_size number of points per leaf
* @li @c block_size_kb maximum block size, in kilobytes
* @li @c megs cache size, in megabytes (floating-point allowed)
*
* @param param parameter object for initializing new nodes
* @param points_channel the channel for the points distributed cache
* @param nodes_channel the channel for the nodes distributed cache
* @param extra_channel a free channel used for internal purposes
* @param module the module to load config parameters from
* @param n_points the number of points
* @param input_cache the data points to read from
* @param tree_out the tree encapsulation to create
*/
template<typename Point, typename Node, typename Param>
void CreateKdBTree(const Param& param,
int points_channel, int nodes_channel, int extra_channel,
datanode *module, index_t n_points, DistributedCache *input_cache,
ThorTree<Param, Point, Node> *tree_out);
};
#include "kdbtree_impl.h"
#endif
fastlib2/fastlib/thor/kdbtree_impl.h
+601
View File
@@ -0,0 +1,601 @@
/* Template implementations for kdtree.h. */
template<typename TPoint, typename TNode, typename TParam>
void KdBTreeBuilder<TPoint, TNode, TParam>::Doit(
fx_module* module, const Param* param_in,
index_t begin_index, index_t end_index,
DistributedCache* input_in,
DistributedCache* points_create,
DistributedCache* nodes_create,
TreeDecomposition* decomposition) {
DEBUG_ASSERT_MSG(rpc::n_peers() == 1, "kdB-trees only coded for serial!");
param_ = param_in;
n_points_ = end_index - begin_index;
inputs_.Init(input_in, BlockDevice::M_READ);
points_.Init(points_create, BlockDevice::M_CREATE);
nodes_.Init(nodes_create, BlockDevice::M_CREATE);
index_t dimension;
{
CacheRead<Point> first_point(&inputs_, inputs_.begin_index());
dimension = first_point->vec().length();
}
point_block_size_ = points_.n_block_elems();
node_block_size_ = nodes_.n_block_elems();
leaf_size_ = fx_param_int(module, "leaf_size", 32);
if (leaf_size_ > point_block_size_) {
NONFATAL("Decreasing leaf size from %d to %d due to block size!\n",
int(leaf_size_), int(point_block_size_));
leaf_size_ = point_block_size_;
}
spare_block_i_ = -index_t(1);
fx_timer_start(module, "tree_build");
index_t node_i = Build_(begin_index, end_index);
decomposition->Init(new DecompNode(
typename TreeDecomposition::Info(0, 1),
&nodes_, node_i, nodes_.end_index()));
fx_timer_stop(module, "tree_build");
}
template<typename TPoint, typename TNode, typename TParam>
index_t KdBTreeBuilder<TPoint, TNode, TParam>::Build_(
index_t begin_col, index_t end_col) {
index_t point_i = points_.AllocD(0, point_block_size_);
index_t root_i = nodes_.AllocD(0, node_block_size_);
{
CacheWrite<Node> root(&nodes_, root_i);
root->set_range(point_i, 0);
root->bound().Reset();
root->set_leaf();
root->set_parent(-index_t(1));
root->set_subnodes_in_page(1);
}
NOTIFY("Phase 1 of tree building...");
/* Add points one by one */
CacheReadIter<Point> input(&inputs_, begin_col);
for (int i = end_col - begin_col; i--; input.Next()) {
Insert_(*input, root_i);
fl_print_progress("points in tree",
(n_points_ - i) * 100 / n_points_);
}
NOTIFY("Phase 2 of tree building...");
/* Build leaf-level partitions */
SplitLeafPages_(root_i);
NOTIFY("Phase 3 of tree building...");
/* Fill statistics and node ranges */
CacheWrite<Node> root(&nodes_, root_i);
Postprocess_(&*root);
NOTIFY("Done!");
return root_i;
}
template<typename TPoint, typename TNode, typename TParam>
void KdBTreeBuilder<TPoint, TNode, TParam>::Postprocess_(
Node *node) {
if (!node->is_leaf()) {
/* Recurse on children */
CacheWrite<Node> left(&nodes_, node->child(0));
CacheWrite<Node> right(&nodes_, node->child(1));
Postprocess_(&*left);
Postprocess_(&*right);
/* Update ranges; done here due to shuffling */
node->set_range(min(left->begin(), right->begin()),
max(left->end(), right->end()),
left->count() + right->count());
/* Update stats; done here to reduce cost */
node->stat().Accumulate(*param_, left->stat(),
left->bound(), left->count());
node->stat().Accumulate(*param_, right->stat(),
right->bound(), right->count());
} else {
/* Finally compute stats for the leaf */
CacheReadIter<Point> point_iter(&points_, node->begin());
for (index_t remaining = node->count(); remaining--;
point_iter.Next()) {
node->stat().Accumulate(*param_, *point_iter);
}
}
/* Both leafs and inner node stats need postprocessing */
node->stat().Postprocess(*param_, node->bound(), node->count());
}
template<typename TPoint, typename TNode, typename TParam>
index_t KdBTreeBuilder<TPoint, TNode, TParam>::SplitLeafPages_(
index_t node_i) {
const Node *node = nodes_.StartRead(node_i);
if (!node->is_leaf()) {
/* Recurse on children */
for (int k = 0; k < Node::CARDINALITY; k++) {
index_t child_i = node->child(k);
nodes_.StopRead(node_i);
node_i = SplitLeafPages_(child_i);
node = nodes_.StartRead(node_i);
}
} else if (node->count() > leaf_size_) {
/* Fits in block, but split to cut cost of base-case */
nodes_.StopRead(node_i);
node_i = SplitLeaf_(node_i, false);
return SplitLeafPages_(node_i);
}
/* Parent may have moved due to splitting! */
index_t parent_i = node->parent();
nodes_.StopRead(node_i);
return parent_i;
}
template<typename TPoint, typename TNode, typename TParam>
void KdBTreeBuilder<TPoint, TNode, TParam>::Insert_(
const Point &input, index_t node_i) {
/* Find the leaf to which the new point belongs */
Node *node = nodes_.StartWrite(node_i);
for (;;) {
/* Incorporate the point into nodes' bounding boxes */
node->bound() |= input.vec();
if (node->is_leaf()) {
break; /* The point will go here! */
}
index_t left_i = node->child(0);
index_t right_i = node->child(1);
nodes_.StopWrite(node_i);
/* Decide if we move into the left or right child */
Node *left = nodes_.StartWrite(left_i);
Node *right = nodes_.StartWrite(right_i);
if (left->bound().MinDistanceSq(input.vec())
<= right->bound().MinDistanceSq(input.vec())) {
node_i = left_i;
node = left;
nodes_.StopWrite(right_i);
} else {
node_i = right_i;
node = right;
nodes_.StopWrite(left_i);
}
}
/* Do we need to split? */
if (node->count() >= point_block_size_) {
nodes_.StopWrite(node_i);
node_i = SplitLeaf_(node_i, true);
Insert_(input, node_i);
} else {
CacheWrite<Point> point(&points_, node->begin() + node->count());
mem::CopyBytes(&*point, &input, points_.n_elem_bytes());
ot::SemiCopy<Point>(reinterpret_cast<char *>(&*point), &input);
node->set_range(node->begin(), node->count() + 1);
nodes_.StopWrite(node_i);
}
}
template<typename TPoint, typename TNode, typename TParam>
index_t KdBTreeBuilder<TPoint, TNode, TParam>::SplitLeaf_(
index_t node_i, bool make_new_block) {
/* Add children to the newly split leaf; may restructure tree */
node_i = CreateChildren_(node_i);
CacheRead<Node> node(&nodes_, node_i);
CacheWrite<Node> left(&nodes_, node->child(0));
CacheWrite<Node> right(&nodes_, node->child(1));
left->bound().Reset();
left->set_leaf();
left->set_parent(node_i);
left->set_subnodes_in_page(0);
right->bound().Reset();
right->set_leaf();
right->set_parent(node_i);
right->set_subnodes_in_page(0);
MedianSplit_(&*node, &left->bound(), &right->bound());
/* Copy right side's points into new block, if appropriate */
index_t point_i = node->begin() + node->count() / 2;
if (make_new_block) {
CacheReadIter<Point> old_iter(&points_, point_i);
point_i = points_.AllocD(0, point_block_size_);
CacheWriteIter<Point> new_iter(&points_, point_i);
for (int remaining = node->count() / 2; remaining--;
old_iter.Next(), new_iter.Next()) {
mem::CopyBytes(&*new_iter, &*old_iter, points_.n_elem_bytes());
ot::SemiCopy(reinterpret_cast<char *>(&*new_iter), &*old_iter);
}
}
/* Tell children where their points are */
left->set_range(node->begin(), node->count() / 2);
right->set_range(point_i, node->count() / 2);
return node_i;
}
template<typename TPoint, typename TNode, typename TParam>
index_t KdBTreeBuilder<TPoint, TNode, TParam>::CreateChildren_(
index_t node_i) {
/* Find the parent of the current tree segment */
index_t parent_i = node_i;
Node *parent;
for (;;) {
/* We're adding two subnodes along a leaf-to-root path */
parent = nodes_.StartWrite(parent_i);
parent->set_subnodes_in_page(parent->subnodes_in_page() + 2);
if (parent->parent() == -index_t(1)
|| nodes_.Blockid(parent_i) != nodes_.Blockid(node_i)) {
/* The parent is the first node outside the current block */
break;
}
/* Move up a node; note: added "parent()" to thor_struct.h */
index_t next_i = parent->parent();
nodes_.StopWrite(parent_i);
parent_i = next_i;
}
/* If there's room, add children; note: left child begins page */
if (parent->subnodes_in_page() <= node_block_size_) {
index_t block_i = parent->parent() == -index_t(1)
? parent_i : parent->child(0);
CacheWrite<Node> node(&nodes_, node_i);
node->set_child(0, block_i + parent->subnodes_in_page() - 2);
node->set_child(1, block_i + parent->subnodes_in_page() - 1);
nodes_.StopWrite(parent_i);
return node_i;
}
/* Special handling for root page */
if (parent->parent() == -index_t(1)) {
index_t block_i = spare_block_i_ != -index_t(1)
? spare_block_i_
: nodes_.AllocD(0, node_block_size_);
index_t left_i = parent->child(0);
index_t right_i = parent->child(1);
CacheWrite<Node> left(&nodes_, left_i);
CacheWrite<Node> right(&nodes_, right_i);
if (left->is_leaf() ||
nodes_.Blockid(left->child(0)) != nodes_.Blockid(left_i)) {
if (node_i == left_i) {
left->set_child(0, block_i);
left->set_child(1, block_i + 1);
block_i = nodes_.AllocD(0, node_block_size_);
}
PackNodes_(block_i, right_i, &*right, 0, &node_i);
} else if (right->is_leaf() ||
nodes_.Blockid(right->child(0)) != nodes_.Blockid(right_i)) {
if (node_i == right_i) {
right->set_child(0, block_i);
right->set_child(1, block_i + 1);
block_i = nodes_.AllocD(0, node_block_size_);
}
PackNodes_(block_i, left_i, &*left, 0, &node_i);
} else {
PackNodes_(block_i, left_i, &*left, 0, &node_i);
block_i = nodes_.AllocD(0, node_block_size_);
PackNodes_(block_i, right_i, &*right, 0, &node_i);
}
parent->set_subnodes_in_page(3);
nodes_.StopWrite(parent_i);
spare_block_i_ = -index_t(1);
return node_i;
}
/* Remember old children; note: old_left_i is first in block */
index_t old_left_i = parent->child(0);
index_t old_right_i = parent->child(1);
/* Children should be in the same, different block from parent */
DEBUG_ASSERT((old_left_i & nodes_.n_block_elems_mask()) == 0);
DEBUG_ASSERT(nodes_.Blockid(old_left_i) == nodes_.Blockid(old_right_i));
DEBUG_ASSERT(nodes_.Blockid(old_left_i) != nodes_.Blockid(parent_i));
/* Old page parent isn't one any more, so reset subnode count */
parent->set_subnodes_in_page(0);
nodes_.StopWrite(parent_i);
parent_i = CreateChildren_(parent_i);
/* Identify new children */
parent = nodes_.StartWrite(parent_i);
index_t new_left_i = parent->child(0);
index_t new_right_i = parent->child(1);
nodes_.StopWrite(parent_i);
/* New page for split, or at least for work */
index_t block_i = spare_block_i_ != -index_t(1)
? spare_block_i_
: nodes_.AllocD(0, node_block_size_);
index_t for_copy_i = block_i;
/* Copy left child up, fixing parent ref */
CacheWrite<Node> new_left(&nodes_, new_left_i);
CacheRead<Node> old_left(&nodes_, old_left_i);
mem::CopyBytes(&*new_left, &*old_left, nodes_.n_elem_bytes());
ot::SemiCopy(reinterpret_cast<char *>(&*new_left), &*old_left);
new_left->set_parent(parent_i);
/* Copy right child up, fixing parent ref */
CacheWrite<Node> new_right(&nodes_, new_right_i);
CacheRead<Node> old_right(&nodes_, old_right_i);
mem::CopyBytes(&*new_right, &*old_right, nodes_.n_elem_bytes());
ot::SemiCopy(reinterpret_cast<char *>(&*new_right), &*old_right);
new_right->set_parent(parent_i);
/* Special handling if copied node is a stub */
if (new_left->is_leaf() ||
nodes_.Blockid(new_left->child(0)) != nodes_.Blockid(old_left_i)) {
if (node_i == old_left_i) {
/* Worst-case, essentially; we still need the block */
node_i = new_left_i;
new_left->set_child(0, block_i);
new_left->set_child(1, block_i + 1);
block_i += 2;
for_copy_i = block_i;
} else {
spare_block_i_ = block_i;
if (!new_left->is_leaf()) {
/* Still fix refs for nodes pointing to another block */
FixChildrenParents_(new_left_i, &*new_left);
}
}
/* Pack into new block, to be copied back (hence offsets) */
block_i = PackNodes_(block_i, new_right_i, &*new_right,
old_left_i - block_i, &node_i);
DEBUG_ASSERT(block_i - for_copy_i == new_right->subnodes_in_page());
} else if (new_right->is_leaf() ||
nodes_.Blockid(new_right->child(0)) != nodes_.Blockid(old_left_i)) {
if (node_i == old_right_i) {
/* Worst-case, essentially; we still need the block */
node_i = new_right_i;
new_right->set_child(0, block_i);
new_right->set_child(1, block_i + 1);
block_i += 2;
for_copy_i = block_i;
} else {
spare_block_i_ = block_i;
if (!new_right->is_leaf()) {
/* Still fix refs for nodes pointing to another block */
FixChildrenParents_(new_right_i, &*new_right);
}
}
/* Pack into new block, to be copied back (hence offsets) */
block_i = PackNodes_(block_i, new_left_i, &*new_left,
old_left_i - block_i, &node_i);
DEBUG_ASSERT(block_i - for_copy_i == new_left->subnodes_in_page());
} else {
spare_block_i_ = -index_t(1);
/* Pack right side into the new block */
block_i = PackNodes_(block_i, new_right_i, &*new_right, 0, &node_i);
DEBUG_ASSERT(block_i - for_copy_i == new_right->subnodes_in_page());
for_copy_i = block_i;
/* Pack left side in prep to copy it back */
block_i = PackNodes_(block_i, new_left_i, &*new_left,
old_left_i - block_i, &node_i);
DEBUG_ASSERT(block_i - for_copy_i == new_left->subnodes_in_page());
}
/* Copy back to orig (refs already fixed) */
CacheWriteIter<Node> old_iter(&nodes_, old_left_i);
CacheReadIter<Node> new_iter(&nodes_, for_copy_i);
for (int remaining = block_i - for_copy_i; remaining--;
old_iter.Next(), new_iter.Next()) {
mem::CopyBytes(&*old_iter, &*new_iter, nodes_.n_elem_bytes());
ot::SemiCopy(reinterpret_cast<char *>(&*old_iter), &*new_iter);
}
if (node_i >= for_copy_i) {
node_i += old_left_i - for_copy_i;
}
return node_i;
}
template<typename TPoint, typename TNode, typename TParam>
index_t KdBTreeBuilder<TPoint, TNode, TParam>::PackNodes_(
index_t dest_i, index_t parent_i, Node *parent,
index_t offset_i, index_t *node_ip) {
for (int k = 0; k < Node::CARDINALITY; ++k) {
index_t child_i = parent->child(k);
index_t next_i = dest_i + offset_i;
parent->set_child(k, next_i);
CacheWrite<Node> dest(&nodes_, dest_i);
CacheRead<Node> child(&nodes_, child_i);
mem::CopyBytes(&*dest, &*child, nodes_.n_elem_bytes());
ot::SemiCopy(reinterpret_cast<char *>(&*dest), &*child);
dest->set_parent(parent_i);
++dest_i;
if (node_ip && child_i == *node_ip) {
/* Found splitting node; leave room for children. */
*node_ip = next_i - offset_i;
dest->set_child(0, next_i + 1);
dest->set_child(1, next_i + 2);
dest_i += 2;
} else if (!dest->is_leaf()) {
if (nodes_.Blockid(dest->child(0)) == nodes_.Blockid(child_i)) {
/* Children in same block; recurse */
dest_i = PackNodes_(dest_i, next_i, &*dest, offset_i, node_ip);
} else {
/* Still fix parent refs for out-of-block */
FixChildrenParents_(next_i, &*dest);
}
}
}
return dest_i;
}
template<typename TPoint, typename TNode, typename TParam>
void KdBTreeBuilder<TPoint, TNode, TParam>::FixChildrenParents_(
index_t parent_i, const Node *parent) {
DEBUG_ASSERT(!parent->is_leaf());
CacheWrite<Node> left(&nodes_, parent->child(0));
CacheWrite<Node> right(&nodes_, parent->child(1));
left->set_parent(parent_i);
right->set_parent(parent_i);
}
template<typename TPoint, typename TNode, typename TParam>
void KdBTreeBuilder<TPoint, TNode, TParam>::MedianSplit_(
const Node *node, Bound *left_bound, Bound *right_bound) {
/* Short loop to find widest dimension */
index_t split_dim = BIG_BAD_NUMBER;
double max_width = -1;
for (index_t d = 0; d < node->bound().dim(); d++) {
double w = node->bound().get(d).width();
if (w > max_width) {
max_width = w;
split_dim = d;
}
}
index_t goal_pos = node->begin() + node->count() / 2;
index_t begin_pos = node->begin();
index_t end_pos = node->end();
DRange current_range = node->bound().get(split_dim);
Bound temp_left_bound;
Bound temp_right_bound;
temp_left_bound.Init(node->bound().dim());
temp_right_bound.Init(node->bound().dim());
index_t split_pos;
for (;;) {
/* Choose a split val likely close to median */
double split_val = current_range.interpolate(
(goal_pos - begin_pos) / double(end_pos - begin_pos));
/* Enact the split, rearranging points */
temp_left_bound.Reset();
temp_right_bound.Reset();
split_pos = thor::Partition(
HrectPartitionCondition(split_dim, split_val),
begin_pos, end_pos - begin_pos, &points_,
&temp_left_bound, &temp_right_bound);
/* Pull in range to more tightly bound median */
if (split_pos == goal_pos) {
*left_bound |= temp_left_bound;
*right_bound |= temp_right_bound;
break;
} else if (split_pos < goal_pos) {
*left_bound |= temp_left_bound;
current_range = temp_right_bound.get(split_dim);
if (current_range.width() == 0) {
break; /* Identical elements */
}
begin_pos = split_pos;
} else if (split_pos > goal_pos) {
*right_bound |= temp_right_bound;
current_range = temp_left_bound.get(split_dim);
if (current_range.width() == 0) {
break; /* Identical elements */
}
end_pos = split_pos;
}
}
if (split_pos != goal_pos) {
/* Identical elements; compute actual bound */
FindBoundingBox_(begin_pos, goal_pos, left_bound);
FindBoundingBox_(goal_pos, end_pos, right_bound);
}
}
template<typename TPoint, typename TNode, typename TParam>
void KdBTreeBuilder<TPoint, TNode, TParam>::FindBoundingBox_(
index_t begin_index, index_t end_index, Bound* bound) {
CacheReadIter<Point> point(&points_, begin_index);
for (index_t i = end_index - begin_index; i--; point.Next()) {
*bound |= point->vec();
}
}
template<typename Point, typename Node, typename Param>
void thor::CreateKdBTreeMaster(const Param& param,
int points_channel, int nodes_channel,
int block_size_kb, double megs, datanode *module,
index_t n_points, DistributedCache *input_cache,
DistributedCache *points_cache, DistributedCache *nodes_cache,
ThorTreeDecomposition<Node> *decomposition) {
Point example_point;
CacheArray<Point>::GetDefaultElement(input_cache, &example_point);
CacheArray<Point>::CreateCacheMaster(points_channel,
CacheArray<Point>::ConvertBlockSize(example_point, block_size_kb),
example_point, megs, points_cache);
Node example_node;
example_node.stat().Init(param);
example_node.bound().Init(example_point.vec().length());
CacheArray<Node>::CreateCacheMaster(nodes_channel,
CacheArray<Node>::ConvertBlockSize(example_node, block_size_kb),
example_node, megs, nodes_cache);
KdBTreeBuilder<Point, Node, Param> builder;
builder.Doit(module, &param, 0, n_points, input_cache,
points_cache, nodes_cache, decomposition);
}
template<typename Point, typename Node, typename Param>
void thor::CreateKdBTree(const Param& param,
int points_channel, int nodes_channel, int extra_channel,
datanode *module, index_t n_points, DistributedCache *input_cache,
ThorTree<Param, Point, Node> *tree_out) {
DistributedCache *points_cache = new DistributedCache();
DistributedCache *nodes_cache = new DistributedCache();
double megs = fx_param_double(module, "megs", 1000);
int block_size_kb = fx_param_int(module, "block_size_kb", 64);
/* Not sure what all this does; might be important */
Broadcaster<ThorTreeDecomposition<Node> > broadcaster;
ThorTreeDecomposition<Node> decomposition;
CreateKdBTreeMaster<Point, Node>(param,
points_channel, nodes_channel, block_size_kb, megs, module,
n_points, input_cache, points_cache, nodes_cache, &decomposition);
/* Probably engineered for parallel; shouldn't hurt */
broadcaster.SetData(decomposition);
points_cache->Sync();
nodes_cache->Sync();
broadcaster.Doit(extra_channel);
tree_out->Init(param, broadcaster.get(), points_cache, nodes_cache);
}
fastlib2/fastlib/thor/kdtree_impl.h
+14
View File
@@ -20,6 +20,9 @@ index_t thor::Partition(
index_t begin, index_t count,
PointCache* points,
Bound* left_bound, Bound* right_bound) {
static index_t processed_tot = 0;
static index_t processed_cur = 0;
index_t left_i = begin;
index_t right_i = begin + count - 1;
@@ -37,6 +40,7 @@ index_t thor::Partition(
}
*left_bound |= left_v->vec();
left_i++;
processed_cur++; //
}
for (;;) {
@@ -48,12 +52,22 @@ index_t thor::Partition(
}
*right_bound |= right_v->vec();
right_i--;
processed_cur++; //
}
points->Swap(left_i, right_i);
DEBUG_ASSERT(left_i <= right_i);
right_i--;
processed_cur++; //
//
if (processed_cur > 10000000) {
processed_tot += processed_cur;
processed_cur = 0;
NOTIFY("Partitioned %"LI"d points total...", processed_tot);
}
//
}
}
fastlib2/fastlib/thor/thor.h
+1
View File
@@ -12,6 +12,7 @@
#include "rbfs.h"
#include "thor_utils.h"
#include "kdtree.h"
#include "kdbtree.h"
#include "fastlib/tree/bounds.h"
fastlib2/fastlib/thor/thor_struct.h
+48 -2
View File
@@ -7,6 +7,8 @@
#ifndef THOR_THORSTRUCT_H
#define THOR_THORSTRUCT_H
#define LIT_DISK_TREES
#include "cachearray.h"
#include "fastlib/la/matrix.h"
@@ -46,20 +48,42 @@ class ThorNode {
index_t children_[t_cardinality];
#ifdef LIT_DISK_TREES
index_t end_;
index_t parent_;
index_t subnodes_in_page_;
#endif LIT_DISK_TREES
OT_DEF_BASIC(ThorNode) {
OT_MY_OBJECT(begin_);
OT_MY_OBJECT(count_);
OT_MY_OBJECT(bound_);
OT_MY_OBJECT(stat_);
OT_MY_ARRAY(children_);
#ifdef LIT_DISK_TREES
OT_MY_OBJECT(end_);
OT_MY_OBJECT(parent_);
OT_MY_OBJECT(subnodes_in_page_);
#endif LIT_DISK_TREES
}
public:
void set_range(index_t begin_in, index_t count_in) {
begin_ = begin_in;
count_ = count_in;
#ifdef LIT_DISK_TREES
end_ = begin_ + count_;
#endif
}
#ifdef LIT_DISK_TREES
void set_range(index_t begin_in, index_t end_in, index_t count_in) {
begin_ = begin_in;
count_ = count_in;
end_ = end_in;
}
#endif
const Bound& bound() const {
return bound_;
}
@@ -89,6 +113,24 @@ class ThorNode {
children_[0] = -index_t(1);
}
#ifdef LIT_DISK_TREES
index_t parent() const {
return parent_;
}
void set_parent(index_t parent_index) {
parent_ = parent_index;
}
index_t subnodes_in_page() const {
return subnodes_in_page_;
}
void set_subnodes_in_page(index_t subnodes) {
subnodes_in_page_ = subnodes;
}
#endif
bool is_leaf() const {
return children_[0] == -index_t(1);
}
@@ -104,7 +146,11 @@ class ThorNode {
* Gets the index one beyond the last index in the series.
*/
index_t end() const {
#ifdef LIT_DISK_TREES
return end_;
#else
return begin_ + count_;
#endif
}
/**
@@ -123,7 +169,7 @@ class ThorNode {
void PrintSelf() const {
printf("node: %d to %d: %d points total\n",
begin_, begin_ + count_ - 1, count_);
begin(), end() - 1, count());
}
};
@@ -243,7 +289,7 @@ class ThorSkeletonNode {
*
* @param array the array to read information from if the node is not
* already in the skeleton tree
* @param k child number (k.e. 0 for left and 1 for right)
* @param k child number (i.e. 0 for left and 1 for right)
*/
ThorSkeletonNode *GetChild(CacheArray<Node> *array, int k) {
if (children_[k] == NULL && !node_.is_leaf()) {
fastlib2/fastlib/thor/thor_utils.h
+5 -2
View File
@@ -11,6 +11,7 @@
#define THOR_UTILS_H
#include "kdtree.h"
#include "kdbtree.h"
#include "sched.h"
#include "rpc.h"
@@ -49,6 +50,7 @@ class ThreadedDualTreeSolver {
DistributedCache *r_points_cache_;
DistributedCache *r_nodes_cache_;
DistributedCache *q_results_cache_;
DistributedCache *q_mutables_cache_;
typename GNP::GlobalResult global_result_;
DualTreeRecursionStats stats_;
Mutex mutex_;
@@ -72,7 +74,8 @@ class ThreadedDualTreeSolver {
const typename GNP::Param& param,
DistributedCache *q_points_cache_in, DistributedCache *q_nodes_cache_in,
DistributedCache *r_points_cache_in, DistributedCache *r_nodes_cache_in,
DistributedCache *q_results_cache_in);
DistributedCache *q_results_cache_in,
DistributedCache *q_mutables_cache_in);
/**
* Gets the GNP's global-result of the entire computation.
@@ -184,7 +187,7 @@ index_t ReadPoints(
template<typename GNP, typename SerialSolver, typename QTree, typename RTree>
void RpcDualTree(datanode *module, int base_channel,
const typename GNP::Param& param, QTree *q, RTree *r,
DistributedCache *q_results,
DistributedCache *q_results, DistributedCache *q_mutables,
typename GNP::GlobalResult *global_result_out);
/**
fastlib2/fastlib/thor/thor_utils_impl.h
+19 -6
View File
@@ -6,7 +6,8 @@ void thor::ThreadedDualTreeSolver<GNP, Solver>::Doit(
const typename GNP::Param& param,
DistributedCache *q_points_cache_in, DistributedCache *q_nodes_cache_in,
DistributedCache *r_points_cache_in, DistributedCache *r_nodes_cache_in,
DistributedCache *q_results_cache_in) {
DistributedCache *q_results_cache_in,
DistributedCache *q_mutables_cache_in) { //
param_ = &param;
work_queue_ = work_queue_in;
rank_ = rank;
@@ -16,6 +17,7 @@ void thor::ThreadedDualTreeSolver<GNP, Solver>::Doit(
r_points_cache_ = r_points_cache_in;
r_nodes_cache_ = r_nodes_cache_in;
q_results_cache_ = q_results_cache_in;
q_mutables_cache_ = q_mutables_cache_in; //
global_result_.Init(*param_);
stats_.Init();
@@ -59,7 +61,7 @@ void thor::ThreadedDualTreeSolver<GNP, Solver>::ThreadBody_() {
work[i].node_index, work[i].node_end_index,
q_points_cache_, q_nodes_cache_,
r_points_cache_, r_nodes_cache_,
q_results_cache_);
q_results_cache_, q_mutables_cache_); //
mutex_.Lock();
stats_.Add(solver.stats());
@@ -159,7 +161,7 @@ index_t thor::ReadPoints(const Param& param, int points_channel,
template<typename GNP, typename SerialSolver, typename QTree, typename RTree>
void thor::RpcDualTree(datanode *module, int base_channel,
const typename GNP::Param& param, QTree *q, RTree *r,
DistributedCache *q_results,
DistributedCache *q_results, DistributedCache *q_mutables, //
typename GNP::GlobalResult *global_result_out) {
int n_threads = fx_param_int(module, "n_threads", 2);
RemoteSchedulerBackend *work_backend = NULL;
@@ -191,7 +193,8 @@ void thor::RpcDualTree(datanode *module, int base_channel,
fx_timer_start(module, "gnp");
ThreadedDualTreeSolver<GNP, SerialSolver> solver;
solver.Doit(n_threads, rpc::rank(), work_queue, param,
&q->points(), &q->nodes(), &r->points(), &r->nodes(), q_results);
&q->points(), &q->nodes(), &r->points(), &r->nodes(),
q_results, q_mutables); //
rpc::Barrier(base_channel + 2);
fx_timer_stop(module, "gnp");
@@ -203,6 +206,7 @@ void thor::RpcDualTree(datanode *module, int base_channel,
r->nodes().StartSync();
}
q_results->StartSync();
q_mutables->StartSync(); //
q->points().WaitSync(fx_submodule(io_module, "q_points"));
q->nodes().WaitSync(fx_submodule(io_module, "q_nodes"));
if (!mem::PtrsEqual(q, r)) {
@@ -210,6 +214,7 @@ void thor::RpcDualTree(datanode *module, int base_channel,
r->nodes().WaitSync(fx_submodule(io_module, "r_nodes"));
}
q_results->WaitSync(fx_submodule(io_module, "q_results"));
q_mutables->WaitSync(fx_submodule(io_module, "q_mutables")); //
fx_timer_stop(module, "write_results");
#ifdef DEBUG
@@ -249,6 +254,7 @@ void thor::MonochromaticDualTreeMain(datanode *module, const char *gnp_name) {
index_t n_points;
ThorTree<typename GNP::Param, typename GNP::QPoint, typename GNP::QNode> tree;
DistributedCache q_results;
DistributedCache q_mutables; //
typename GNP::Param param;
rpc::Init();
@@ -271,7 +277,7 @@ void thor::MonochromaticDualTreeMain(datanode *module, const char *gnp_name) {
fx_timer_start(module, "tree");
CreateKdTree<typename GNP::QPoint, typename GNP::QNode>(
param, DATA_CHANNEL + 2, DATA_CHANNEL + 3,
param, DATA_CHANNEL + 2, DATA_CHANNEL + 3, // DATA_CHANNEL + 4,
fx_submodule(module, "tree"), n_points, points_cache, &tree);
fx_timer_stop(module, "tree");
@@ -280,10 +286,17 @@ void thor::MonochromaticDualTreeMain(datanode *module, const char *gnp_name) {
tree.CreateResultCache(Q_RESULTS_CHANNEL, default_result,
results_megs, &q_results);
// Alert! Temporary hack; does not support rbfs
typename Solver::QMutables default_mutable; //
default_mutable.summary_result.Init(param); //
default_mutable.postponed.Init(param); //
tree.CreateMutableCache(Q_RESULTS_CHANNEL + 1, default_mutable, //
results_megs, &q_mutables);
typename GNP::GlobalResult global_result;
RpcDualTree<GNP, Solver>(
fx_submodule(module, "gnp"), GNP_CHANNEL, param,
&tree, &tree, &q_results, &global_result);
&tree, &tree, &q_results, &q_mutables, &global_result);
rpc::Done();
}
fastlib2/fastlib/thor/thortree.h
+10
View File
@@ -244,6 +244,16 @@ class ThorTree {
template<typename Result>
void CreateResultCacheWorker(int channel,
double megs, DistributedCache *results);
template<typename Mutable>
void CreateMutableCache(int channel, const Mutable& default_mutable,
double megs, DistributedCache *mutables);
template<typename Mutable>
void CreateMutableCacheMaster(int channel, const Mutable& default_mutable,
double megs, DistributedCache *mutables);
template<typename Mutable>
void CreateMutableCacheWorker(int channel,
double megs, DistributedCache *mutables);
};
/**
fastlib2/fastlib/thor/thortree_impl.h
+48
View File
@@ -54,6 +54,54 @@ void ThorTree<TParam, TPoint, TNode>::CreateResultCache(
}
}
template<typename TParam, typename TPoint, typename TNode>
template<typename Mutable>
void ThorTree<TParam, TPoint, TNode>::CreateMutableCacheMaster(
int channel, const Mutable& default_mutable,
double megs, DistributedCache *mutables) {
DEBUG_ASSERT_MSG(rpc::rank() == 0, "Only master calls this");
index_t block_size = CacheArray<Point>::GetNumBlockElements(nodes_);
CacheArray<Mutable>::CreateCacheMaster(channel,
block_size, default_mutable, megs, mutables);
for (int i = 0; i < rpc::n_peers(); i++) {
const TreeGrain *grain = &decomp_.grain_by_owner(i);
if (grain->is_valid()) {
BlockDevice::blockid_t begin_block =
(grain->node_index + block_size - 1) / block_size;
BlockDevice::blockid_t end_block =
(grain->node_end_index + block_size - 1) / block_size;
DEBUG_ASSERT(mutables->n_blocks() == begin_block);
mutables->AllocBlocks(end_block - begin_block, i);
}
}
mutables->StartSync();
mutables->WaitSync();
}
template<typename TParam, typename TPoint, typename TNode>
template<typename Mutable>
void ThorTree<TParam, TPoint, TNode>::CreateMutableCacheWorker(
int channel, double megs, DistributedCache *mutables) {
DEBUG_ASSERT_MSG(rpc::rank() != 0, "Only workers call this");
CacheArray<Mutable>::CreateCacheWorker(channel, megs, mutables);
mutables->StartSync();
mutables->WaitSync();
}
template<typename TParam, typename TPoint, typename TNode>
template<typename Mutable>
void ThorTree<TParam, TPoint, TNode>::CreateMutableCache(
int channel, const Mutable& default_mutable,
double megs, DistributedCache *mutables) {
if (rpc::rank() == 0) {
CreateResultCacheMaster(channel, default_mutable, megs, mutables);
} else {
CreateResultCacheWorker<Mutable>(channel, megs, mutables);
}
}
template<typename TParam, typename TPoint, typename TNode>
template<typename Result, typename Visitor>
void ThorTree<TParam, TPoint, TNode>::Update(