distributed/adjlist/initialize.hpp

0001 // Copyright (C) 2007 Douglas Gregor
0002
0003 // Use, modification and distribution is subject to the Boost Software
0004 // License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
0005 // http://www.boost.org/LICENSE_1_0.txt)
0006
0007 // This file contains code for the distributed adjacency list's
0008 // initializations. It should not be included directly by users.
0009
0010 #ifndef BOOST_GRAPH_DISTRIBUTED_ADJLIST_INITIALIZE_HPP
0011 #define BOOST_GRAPH_DISTRIBUTED_ADJLIST_INITIALIZE_HPP
0012
0013 #ifndef BOOST_GRAPH_USE_MPI
0014 #error "Parallel BGL files should not be included unless <boost/graph/use_mpi.hpp> has been included"
0015 #endif
0016
0017 namespace boost {
0018
0019 template<PBGL_DISTRIB_ADJLIST_TEMPLATE_PARMS>
0020 template<typename EdgeIterator>
0021 void
0022 PBGL_DISTRIB_ADJLIST_TYPE::
0023 initialize(EdgeIterator first, EdgeIterator last,
0024            vertices_size_type, const base_distribution_type& distribution,
0025            vecS)
0026 {
0027   process_id_type id = process_id(process_group_);
0028   while (first != last) {
0029     if ((process_id_type)distribution(first->first) == id) {
0030       vertex_descriptor source(id, distribution.local(first->first));
0031       vertex_descriptor target(distribution(first->second),
0032                                distribution.local(first->second));
0033       add_edge(source, target, *this);
0034     }
0035     ++first;
0036   }
0037
0038   synchronize(process_group_);
0039 }
0040
0041 template<PBGL_DISTRIB_ADJLIST_TEMPLATE_PARMS>
0042 template<typename EdgeIterator, typename EdgePropertyIterator>
0043 void
0044 PBGL_DISTRIB_ADJLIST_TYPE::
0045 initialize(EdgeIterator first, EdgeIterator last,
0046            EdgePropertyIterator ep_iter,
0047            vertices_size_type, const base_distribution_type& distribution,
0048            vecS)
0049 {
0050   process_id_type id = process_id(process_group_);
0051   while (first != last) {
0052     if (static_cast<process_id_type>(distribution(first->first)) == id) {
0053       vertex_descriptor source(id, distribution.local(first->first));
0054       vertex_descriptor target(distribution(first->second),
0055                                distribution.local(first->second));
0056       add_edge(source, target, *ep_iter, *this);
0057     }
0058     ++first;
0059     ++ep_iter;
0060   }
0061
0062   synchronize(process_group_);
0063 }
0064
0065 template<PBGL_DISTRIB_ADJLIST_TEMPLATE_PARMS>
0066 template<typename EdgeIterator, typename EdgePropertyIterator,
0067          typename VertexListS>
0068 void
0069 PBGL_DISTRIB_ADJLIST_TYPE::
0070 initialize(EdgeIterator first, EdgeIterator last,
0071            EdgePropertyIterator ep_iter,
0072            vertices_size_type n, const base_distribution_type& distribution,
0073            VertexListS)
0074 {
0075   using boost::parallel::inplace_all_to_all;
0076
0077   typedef vertices_size_type vertex_number_t;
0078   typedef typename std::iterator_traits<EdgePropertyIterator>::value_type
0079     edge_property_init_t;
0080
0081   typedef std::pair<vertex_descriptor, vertex_number_t>
0082     st_pair;
0083   typedef std::pair<st_pair, edge_property_init_t> delayed_edge_t;
0084
0085   process_group_type pg = process_group();
0086   process_id_type id = process_id(pg);
0087
0088   // Vertex indices
0089   std::vector<local_vertex_descriptor> index_to_vertex;
0090   index_to_vertex.reserve(num_vertices(*this));
0091   BGL_FORALL_VERTICES_T(v, base(), inherited)
0092     index_to_vertex.push_back(v);
0093
0094   // The list of edges we can't add immediately.
0095   std::vector<delayed_edge_t> delayed_edges;
0096
0097   std::vector<std::vector<vertex_number_t> > descriptor_requests;
0098   descriptor_requests.resize(num_processes(pg));
0099
0100   // Add all of the edges we can, up to the point where we run
0101   // into a descriptor we don't know.
0102   while (first != last) {
0103     if (distribution(first->first) == id) {
0104       if (distribution(first->second) != id) break;
0105       vertex_descriptor source
0106         (id, index_to_vertex[distribution.local(first->first)]);
0107       vertex_descriptor target
0108         (distribution(first->second),
0109          index_to_vertex[distribution.local(first->second)]);
0110       add_edge(source, target, *ep_iter, *this);
0111     }
0112     ++first;
0113     ++ep_iter;
0114   }
0115
0116   // Queue all of the remaining edges and determine the set of
0117   // descriptors we need to know about.
0118   while (first != last) {
0119     if (distribution(first->first) == id) {
0120       vertex_descriptor source
0121         (id, index_to_vertex[distribution.local(first->first)]);
0122       process_id_type dest = distribution(first->second);
0123       if (dest != id) {
0124         descriptor_requests[dest]
0125           .push_back(distribution.local(first->second));
0126         // Compact request list if we need to
0127         if (descriptor_requests[dest].size() >
0128               distribution.block_size(dest, n)) {
0129           std::sort(descriptor_requests[dest].begin(),
0130                     descriptor_requests[dest].end());
0131           descriptor_requests[dest].erase(
0132             std::unique(descriptor_requests[dest].begin(),
0133                         descriptor_requests[dest].end()),
0134             descriptor_requests[dest].end());
0135         }
0136       }
0137
0138       // Save the edge for later
0139       delayed_edges.push_back
0140         (delayed_edge_t(st_pair(source, first->second), *ep_iter));
0141     }
0142     ++first;
0143     ++ep_iter;
0144   }
0145
0146   // Compact descriptor requests
0147   for (process_id_type dest = 0; dest < num_processes(pg); ++dest) {
0148     std::sort(descriptor_requests[dest].begin(),
0149               descriptor_requests[dest].end());
0150     descriptor_requests[dest].erase(
0151       std::unique(descriptor_requests[dest].begin(),
0152                   descriptor_requests[dest].end()),
0153       descriptor_requests[dest].end());
0154   }
0155
0156   // Send out all of the descriptor requests
0157   std::vector<std::vector<vertex_number_t> > in_descriptor_requests;
0158   in_descriptor_requests.resize(num_processes(pg));
0159   inplace_all_to_all(pg, descriptor_requests, in_descriptor_requests);
0160
0161   // Reply to all of the descriptor requests
0162   std::vector<std::vector<local_vertex_descriptor> >
0163     descriptor_responses;
0164   descriptor_responses.resize(num_processes(pg));
0165   for (process_id_type dest = 0; dest < num_processes(pg); ++dest) {
0166     for (std::size_t i = 0; i < in_descriptor_requests[dest].size(); ++i) {
0167       local_vertex_descriptor v =
0168         index_to_vertex[in_descriptor_requests[dest][i]];
0169       descriptor_responses[dest].push_back(v);
0170     }
0171     in_descriptor_requests[dest].clear();
0172   }
0173   in_descriptor_requests.clear();
0174   inplace_all_to_all(pg, descriptor_responses);
0175
0176   // Add the queued edges
0177   for(typename std::vector<delayed_edge_t>::iterator i
0178         = delayed_edges.begin(); i != delayed_edges.end(); ++i) {
0179     process_id_type dest = distribution(i->first.second);
0180     local_vertex_descriptor tgt_local;
0181     if (dest == id) {
0182       tgt_local = index_to_vertex[distribution.local(i->first.second)];
0183     } else {
0184       std::vector<vertex_number_t>& requests = descriptor_requests[dest];
0185       typename std::vector<vertex_number_t>::iterator pos =
0186         std::lower_bound(requests.begin(), requests.end(),
0187                          distribution.local(i->first.second));
0188       tgt_local = descriptor_responses[dest][pos - requests.begin()];
0189     }
0190     add_edge(i->first.first, vertex_descriptor(dest, tgt_local),
0191              i->second, *this);
0192   }
0193   synchronize(process_group_);
0194 }
0195
0196 template<PBGL_DISTRIB_ADJLIST_TEMPLATE_PARMS>
0197 template<typename EdgeIterator, typename VertexListS>
0198 void
0199 PBGL_DISTRIB_ADJLIST_TYPE::
0200 initialize(EdgeIterator first, EdgeIterator last,
0201            vertices_size_type n, const base_distribution_type& distribution,
0202            VertexListS)
0203 {
0204   using boost::parallel::inplace_all_to_all;
0205
0206   typedef vertices_size_type vertex_number_t;
0207
0208   typedef std::pair<vertex_descriptor, vertex_number_t> delayed_edge_t;
0209
0210   process_group_type pg = process_group();
0211   process_id_type id = process_id(pg);
0212
0213   // Vertex indices
0214   std::vector<local_vertex_descriptor> index_to_vertex;
0215   index_to_vertex.reserve(num_vertices(*this));
0216   BGL_FORALL_VERTICES_T(v, base(), inherited)
0217     index_to_vertex.push_back(v);
0218
0219   // The list of edges we can't add immediately.
0220   std::vector<delayed_edge_t> delayed_edges;
0221
0222   std::vector<std::vector<vertex_number_t> > descriptor_requests;
0223   descriptor_requests.resize(num_processes(pg));
0224
0225   // Add all of the edges we can, up to the point where we run
0226   // into a descriptor we don't know.
0227   while (first != last) {
0228     if (distribution(first->first) == id) {
0229       if (distribution(first->second) != id) break;
0230       vertex_descriptor source
0231         (id, index_to_vertex[distribution.local(first->first)]);
0232       vertex_descriptor target
0233         (distribution(first->second),
0234          index_to_vertex[distribution.local(first->second)]);
0235       add_edge(source, target, *this);
0236     }
0237     ++first;
0238   }
0239
0240   // Queue all of the remaining edges and determine the set of
0241   // descriptors we need to know about.
0242   while (first != last) {
0243     if (distribution(first->first) == id) {
0244       vertex_descriptor source
0245         (id, index_to_vertex[distribution.local(first->first)]);
0246       process_id_type dest = distribution(first->second);
0247       if (dest != id) {
0248         descriptor_requests[dest]
0249           .push_back(distribution.local(first->second));
0250         // Compact request list if we need to
0251         if (descriptor_requests[dest].size() >
0252               distribution.block_size(dest, n)) {
0253           std::sort(descriptor_requests[dest].begin(),
0254                     descriptor_requests[dest].end());
0255           descriptor_requests[dest].erase(
0256             std::unique(descriptor_requests[dest].begin(),
0257                         descriptor_requests[dest].end()),
0258             descriptor_requests[dest].end());
0259         }
0260       }
0261
0262       // Save the edge for later
0263       delayed_edges.push_back(delayed_edge_t(source, first->second));
0264     }
0265     ++first;
0266   }
0267
0268   // Compact descriptor requests
0269   for (process_id_type dest = 0; dest < num_processes(pg); ++dest) {
0270     std::sort(descriptor_requests[dest].begin(),
0271               descriptor_requests[dest].end());
0272     descriptor_requests[dest].erase(
0273       std::unique(descriptor_requests[dest].begin(),
0274                   descriptor_requests[dest].end()),
0275       descriptor_requests[dest].end());
0276   }
0277
0278   // Send out all of the descriptor requests
0279   std::vector<std::vector<vertex_number_t> > in_descriptor_requests;
0280   in_descriptor_requests.resize(num_processes(pg));
0281   inplace_all_to_all(pg, descriptor_requests, in_descriptor_requests);
0282
0283   // Reply to all of the descriptor requests
0284   std::vector<std::vector<local_vertex_descriptor> >
0285     descriptor_responses;
0286   descriptor_responses.resize(num_processes(pg));
0287   for (process_id_type dest = 0; dest < num_processes(pg); ++dest) {
0288     for (std::size_t i = 0; i < in_descriptor_requests[dest].size(); ++i) {
0289       local_vertex_descriptor v =
0290         index_to_vertex[in_descriptor_requests[dest][i]];
0291       descriptor_responses[dest].push_back(v);
0292     }
0293     in_descriptor_requests[dest].clear();
0294   }
0295   in_descriptor_requests.clear();
0296   inplace_all_to_all(pg, descriptor_responses);
0297
0298   // Add the queued edges
0299   for(typename std::vector<delayed_edge_t>::iterator i
0300         = delayed_edges.begin(); i != delayed_edges.end(); ++i) {
0301     process_id_type dest = distribution(i->second);
0302     local_vertex_descriptor tgt_local;
0303     if (dest == id) {
0304       tgt_local = index_to_vertex[distribution.local(i->second)];
0305     } else {
0306       std::vector<vertex_number_t>& requests = descriptor_requests[dest];
0307       typename std::vector<vertex_number_t>::iterator pos =
0308         std::lower_bound(requests.begin(), requests.end(),
0309                          distribution.local(i->second));
0310       tgt_local = descriptor_responses[dest][pos - requests.begin()];
0311     }
0312     add_edge(i->first, vertex_descriptor(dest, tgt_local), *this);
0313   }
0314   synchronize(process_group_);
0315 }
0316
0317 }   // end namespace boost
0318
0319 #endif // BOOST_GRAPH_DISTRIBUTED_ADJLIST_INITIALIZE_HPP