diff --git a/fastlib2/fastlib/thor/allnn.cc b/fastlib2/fastlib/thor/allnn.cc index 7cec066312..8c3c2f4199 100644 --- a/fastlib2/fastlib/thor/allnn.cc +++ b/fastlib2/fastlib/thor/allnn.cc @@ -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; } diff --git a/fastlib2/fastlib/thor/build.py b/fastlib2/fastlib/thor/build.py index 0e65a6fdad..4be1ab7f2d 100644 --- a/fastlib2/fastlib/thor/build.py +++ b/fastlib2/fastlib/thor/build.py @@ -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 diff --git a/fastlib2/fastlib/thor/dfs.h b/fastlib2/fastlib/thor/dfs.h index 25733e1a36..b18dab7eb0 100644 --- a/fastlib2/fastlib/thor/dfs.h +++ b/fastlib2/fastlib/thor/dfs.h @@ -17,7 +17,7 @@ template 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 q_points_; CacheArray q_nodes_; CacheArray q_results_; - SubsetArray q_mutables_; + CacheArray q_mutables_; // SubsetArray CacheArray r_points_; CacheArray 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. diff --git a/fastlib2/fastlib/thor/dfs_impl.h b/fastlib2/fastlib/thor/dfs_impl.h index 6a47a3d992..10a574f657 100644 --- a/fastlib2/fastlib/thor/dfs_impl.h +++ b/fastlib2/fastlib/thor/dfs_impl.h @@ -18,14 +18,36 @@ void DualTreeDepthFirst::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 q_mutables_iter(&q_mutables_, // + q_root_index); + CacheReadIter 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::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 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::Begin_(index_t q_root_index) { typename GNP::Delta delta; typename GNP::Delta empty_delta; CacheRead 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 q_root_mut(&q_mutables_, q_root_index); empty_delta.Init(param_); delta.Init(param_); @@ -118,7 +127,8 @@ void DualTreeDepthFirst::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 q_child_mut(&q_mutables_, q_child_i); q_child_mut->postponed.ApplyPostponed(param_, q_node_mut->postponed); @@ -169,7 +179,8 @@ void DualTreeDepthFirst::Pair_( typename GNP::Delta child_delta; index_t q_child_i = q_node->child(k); CacheRead 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 q_child_mut(&q_mutables_, q_child_i); child_delta.Init(param_); q_child_mut->postponed.ApplyPostponed( diff --git a/fastlib2/fastlib/thor/kdbtree.h b/fastlib2/fastlib/thor/kdbtree.h new file mode 100644 index 0000000000..a1181080eb --- /dev/null +++ b/fastlib2/fastlib/thor/kdbtree.h @@ -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 +class KdBTreeBuilder { + public: + typedef TNode Node; + typedef TPoint Point; + typedef typename TNode::Bound Bound; + typedef TParam Param; + typedef ThorTreeDecomposition 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 inputs_; + CacheArray points_; + CacheArray 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 + 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 *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 + void CreateKdBTree(const Param& param, + int points_channel, int nodes_channel, int extra_channel, + datanode *module, index_t n_points, DistributedCache *input_cache, + ThorTree *tree_out); + +}; + +#include "kdbtree_impl.h" + +#endif diff --git a/fastlib2/fastlib/thor/kdbtree_impl.h b/fastlib2/fastlib/thor/kdbtree_impl.h new file mode 100644 index 0000000000..10f8c39529 --- /dev/null +++ b/fastlib2/fastlib/thor/kdbtree_impl.h @@ -0,0 +1,601 @@ +/* Template implementations for kdtree.h. */ + +template +void KdBTreeBuilder::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 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 +index_t KdBTreeBuilder::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 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 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 root(&nodes_, root_i); + Postprocess_(&*root); + + NOTIFY("Done!"); + + return root_i; +} + +template +void KdBTreeBuilder::Postprocess_( + Node *node) { + if (!node->is_leaf()) { + /* Recurse on children */ + CacheWrite left(&nodes_, node->child(0)); + CacheWrite 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_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 +index_t KdBTreeBuilder::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 +void KdBTreeBuilder::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(&points_, node->begin() + node->count()); + mem::CopyBytes(&*point, &input, points_.n_elem_bytes()); + ot::SemiCopy(reinterpret_cast(&*point), &input); + + node->set_range(node->begin(), node->count() + 1); + nodes_.StopWrite(node_i); + } +} + +template +index_t KdBTreeBuilder::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(&nodes_, node_i); + CacheWrite left(&nodes_, node->child(0)); + CacheWrite 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 old_iter(&points_, point_i); + point_i = points_.AllocD(0, point_block_size_); + CacheWriteIter 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(&*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 +index_t KdBTreeBuilder::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(&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 left(&nodes_, left_i); + CacheWrite 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 new_left(&nodes_, new_left_i); + CacheRead old_left(&nodes_, old_left_i); + mem::CopyBytes(&*new_left, &*old_left, nodes_.n_elem_bytes()); + ot::SemiCopy(reinterpret_cast(&*new_left), &*old_left); + new_left->set_parent(parent_i); + + /* Copy right child up, fixing parent ref */ + CacheWrite new_right(&nodes_, new_right_i); + CacheRead old_right(&nodes_, old_right_i); + mem::CopyBytes(&*new_right, &*old_right, nodes_.n_elem_bytes()); + ot::SemiCopy(reinterpret_cast(&*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 old_iter(&nodes_, old_left_i); + CacheReadIter 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(&*old_iter), &*new_iter); + } + + if (node_i >= for_copy_i) { + node_i += old_left_i - for_copy_i; + } + + return node_i; +} + +template +index_t KdBTreeBuilder::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 dest(&nodes_, dest_i); + CacheRead child(&nodes_, child_i); + mem::CopyBytes(&*dest, &*child, nodes_.n_elem_bytes()); + ot::SemiCopy(reinterpret_cast(&*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 +void KdBTreeBuilder::FixChildrenParents_( + index_t parent_i, const Node *parent) { + DEBUG_ASSERT(!parent->is_leaf()); + CacheWrite left(&nodes_, parent->child(0)); + CacheWrite right(&nodes_, parent->child(1)); + left->set_parent(parent_i); + right->set_parent(parent_i); +} + +template +void KdBTreeBuilder::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 +void KdBTreeBuilder::FindBoundingBox_( + index_t begin_index, index_t end_index, Bound* bound) { + CacheReadIter point(&points_, begin_index); + for (index_t i = end_index - begin_index; i--; point.Next()) { + *bound |= point->vec(); + } +} + +template +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 *decomposition) { + Point example_point; + CacheArray::GetDefaultElement(input_cache, &example_point); + + CacheArray::CreateCacheMaster(points_channel, + CacheArray::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::CreateCacheMaster(nodes_channel, + CacheArray::ConvertBlockSize(example_node, block_size_kb), + example_node, megs, nodes_cache); + + KdBTreeBuilder builder; + builder.Doit(module, ¶m, 0, n_points, input_cache, + points_cache, nodes_cache, decomposition); +} + +template +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 *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 > broadcaster; + ThorTreeDecomposition decomposition; + + CreateKdBTreeMaster(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); +} diff --git a/fastlib2/fastlib/thor/kdtree_impl.h b/fastlib2/fastlib/thor/kdtree_impl.h index 92f65d25ec..32e934ab07 100644 --- a/fastlib2/fastlib/thor/kdtree_impl.h +++ b/fastlib2/fastlib/thor/kdtree_impl.h @@ -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); + } + // } } diff --git a/fastlib2/fastlib/thor/thor.h b/fastlib2/fastlib/thor/thor.h index 47466a2f0d..c8d7d05b10 100644 --- a/fastlib2/fastlib/thor/thor.h +++ b/fastlib2/fastlib/thor/thor.h @@ -12,6 +12,7 @@ #include "rbfs.h" #include "thor_utils.h" #include "kdtree.h" +#include "kdbtree.h" #include "fastlib/tree/bounds.h" diff --git a/fastlib2/fastlib/thor/thor_struct.h b/fastlib2/fastlib/thor/thor_struct.h index 508e163437..928b179175 100644 --- a/fastlib2/fastlib/thor/thor_struct.h +++ b/fastlib2/fastlib/thor/thor_struct.h @@ -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 *array, int k) { if (children_[k] == NULL && !node_.is_leaf()) { diff --git a/fastlib2/fastlib/thor/thor_utils.h b/fastlib2/fastlib/thor/thor_utils.h index 7f290ae3f3..a9b210b6ae 100644 --- a/fastlib2/fastlib/thor/thor_utils.h +++ b/fastlib2/fastlib/thor/thor_utils.h @@ -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 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); /** diff --git a/fastlib2/fastlib/thor/thor_utils_impl.h b/fastlib2/fastlib/thor/thor_utils_impl.h index 5853114c38..7b1bad0ca1 100644 --- a/fastlib2/fastlib/thor/thor_utils_impl.h +++ b/fastlib2/fastlib/thor/thor_utils_impl.h @@ -6,7 +6,8 @@ void thor::ThreadedDualTreeSolver::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_ = ¶m; work_queue_ = work_queue_in; rank_ = rank; @@ -16,6 +17,7 @@ void thor::ThreadedDualTreeSolver::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::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 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 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 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( - 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( fx_submodule(module, "gnp"), GNP_CHANNEL, param, - &tree, &tree, &q_results, &global_result); + &tree, &tree, &q_results, &q_mutables, &global_result); rpc::Done(); } diff --git a/fastlib2/fastlib/thor/thortree.h b/fastlib2/fastlib/thor/thortree.h index a7b0152476..d4f7107d19 100644 --- a/fastlib2/fastlib/thor/thortree.h +++ b/fastlib2/fastlib/thor/thortree.h @@ -244,6 +244,16 @@ class ThorTree { template void CreateResultCacheWorker(int channel, double megs, DistributedCache *results); + + template + void CreateMutableCache(int channel, const Mutable& default_mutable, + double megs, DistributedCache *mutables); + template + void CreateMutableCacheMaster(int channel, const Mutable& default_mutable, + double megs, DistributedCache *mutables); + template + void CreateMutableCacheWorker(int channel, + double megs, DistributedCache *mutables); }; /** diff --git a/fastlib2/fastlib/thor/thortree_impl.h b/fastlib2/fastlib/thor/thortree_impl.h index 1a5473ed2f..597d64a875 100644 --- a/fastlib2/fastlib/thor/thortree_impl.h +++ b/fastlib2/fastlib/thor/thortree_impl.h @@ -54,6 +54,54 @@ void ThorTree::CreateResultCache( } } +template +template +void ThorTree::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::GetNumBlockElements(nodes_); + CacheArray::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 +template +void ThorTree::CreateMutableCacheWorker( + int channel, double megs, DistributedCache *mutables) { + DEBUG_ASSERT_MSG(rpc::rank() != 0, "Only workers call this"); + CacheArray::CreateCacheWorker(channel, megs, mutables); + mutables->StartSync(); + mutables->WaitSync(); +} + +template +template +void ThorTree::CreateMutableCache( + int channel, const Mutable& default_mutable, + double megs, DistributedCache *mutables) { + if (rpc::rank() == 0) { + CreateResultCacheMaster(channel, default_mutable, megs, mutables); + } else { + CreateResultCacheWorker(channel, megs, mutables); + } +} + template template void ThorTree::Update(