| 1 | """Trying to lump parallel stuff into simpler interface |
|---|
| 2 | |
|---|
| 3 | |
|---|
| 4 | """ |
|---|
| 5 | |
|---|
| 6 | import numpy as num |
|---|
| 7 | |
|---|
| 8 | # The abstract Python-MPI interface |
|---|
| 9 | from anuga_parallel.parallel_abstraction import size, rank, get_processor_name |
|---|
| 10 | from anuga_parallel.parallel_abstraction import finalize, send, receive |
|---|
| 11 | from anuga_parallel.parallel_abstraction import pypar_available, barrier |
|---|
| 12 | |
|---|
| 13 | |
|---|
| 14 | # ANUGA parallel engine (only load if pypar can) |
|---|
| 15 | if pypar_available: |
|---|
| 16 | from anuga_parallel.distribute_mesh import send_submesh |
|---|
| 17 | from anuga_parallel.distribute_mesh import rec_submesh |
|---|
| 18 | from anuga_parallel.distribute_mesh import extract_submesh |
|---|
| 19 | |
|---|
| 20 | # Mesh partitioning using Metis |
|---|
| 21 | from anuga_parallel.distribute_mesh import build_submesh |
|---|
| 22 | from anuga_parallel.distribute_mesh import pmesh_divide_metis_with_map |
|---|
| 23 | |
|---|
| 24 | from anuga_parallel.parallel_shallow_water import Parallel_domain |
|---|
| 25 | |
|---|
| 26 | from anuga.abstract_2d_finite_volumes.neighbour_mesh import Mesh |
|---|
| 27 | |
|---|
| 28 | #------------------------------------------------------------------------------ |
|---|
| 29 | # Read in processor information |
|---|
| 30 | #------------------------------------------------------------------------------ |
|---|
| 31 | |
|---|
| 32 | numprocs = size() |
|---|
| 33 | myid = rank() |
|---|
| 34 | processor_name = get_processor_name() |
|---|
| 35 | #print 'I am processor %d of %d on node %s' %(myid, numprocs, processor_name) |
|---|
| 36 | |
|---|
| 37 | |
|---|
| 38 | |
|---|
| 39 | |
|---|
| 40 | def distribute(domain, verbose=False, debug=False, parameters = None): |
|---|
| 41 | """ Distribute the domain to all processes |
|---|
| 42 | |
|---|
| 43 | parameters values |
|---|
| 44 | """ |
|---|
| 45 | |
|---|
| 46 | |
|---|
| 47 | if debug: |
|---|
| 48 | verbose = True |
|---|
| 49 | |
|---|
| 50 | barrier() |
|---|
| 51 | |
|---|
| 52 | # FIXME: Dummy assignment (until boundaries are refactored to |
|---|
| 53 | # be independent of domains until they are applied) |
|---|
| 54 | if myid == 0: |
|---|
| 55 | bdmap = {} |
|---|
| 56 | for tag in domain.get_boundary_tags(): |
|---|
| 57 | bdmap[tag] = None |
|---|
| 58 | |
|---|
| 59 | |
|---|
| 60 | domain.set_boundary(bdmap) |
|---|
| 61 | |
|---|
| 62 | |
|---|
| 63 | if not pypar_available or numprocs == 1 : return domain # Bypass |
|---|
| 64 | |
|---|
| 65 | # For some obscure reason this communication must happen prior to |
|---|
| 66 | # the more complex mesh distribution - Oh Well! |
|---|
| 67 | if myid == 0: |
|---|
| 68 | domain_name = domain.get_name() |
|---|
| 69 | domain_dir = domain.get_datadir() |
|---|
| 70 | domain_store = domain.get_store() |
|---|
| 71 | domain_smooth = domain.smooth |
|---|
| 72 | domain_reduction = domain.reduction |
|---|
| 73 | domain_minimum_storable_height = domain.minimum_storable_height |
|---|
| 74 | domain_flow_algorithm = domain.get_flow_algorithm() |
|---|
| 75 | domain_minimum_allowed_height = domain.get_minimum_allowed_height() |
|---|
| 76 | georef = domain.geo_reference |
|---|
| 77 | number_of_global_triangles = domain.number_of_triangles |
|---|
| 78 | number_of_global_nodes = domain.number_of_nodes |
|---|
| 79 | |
|---|
| 80 | # FIXME - what other attributes need to be transferred? |
|---|
| 81 | |
|---|
| 82 | for p in xrange(1, numprocs): |
|---|
| 83 | # FIXME SR: Creates cPickle dump |
|---|
| 84 | send((domain_name, domain_dir, domain_store, \ |
|---|
| 85 | domain_smooth, domain_reduction, \ |
|---|
| 86 | domain_minimum_storable_height, domain_flow_algorithm, \ |
|---|
| 87 | domain_minimum_allowed_height, georef, \ |
|---|
| 88 | number_of_global_triangles, number_of_global_nodes), p) |
|---|
| 89 | else: |
|---|
| 90 | if verbose: print 'P%d: Receiving domain attributes' %(myid) |
|---|
| 91 | |
|---|
| 92 | domain_name, domain_dir, domain_store, \ |
|---|
| 93 | domain_smooth, domain_reduction, \ |
|---|
| 94 | domain_minimum_storable_height, domain_flow_algorithm, \ |
|---|
| 95 | domain_minimum_allowed_height, georef, number_of_global_triangles, \ |
|---|
| 96 | number_of_global_nodes = receive(0) |
|---|
| 97 | |
|---|
| 98 | |
|---|
| 99 | |
|---|
| 100 | # Distribute boundary conditions |
|---|
| 101 | # FIXME: This cannot handle e.g. Time_boundaries due to |
|---|
| 102 | # difficulties pickling functions |
|---|
| 103 | if myid == 0: |
|---|
| 104 | boundary_map = domain.boundary_map |
|---|
| 105 | for p in xrange(1, numprocs): |
|---|
| 106 | # FIXME SR: Creates cPickle dump |
|---|
| 107 | send(boundary_map, p) |
|---|
| 108 | else: |
|---|
| 109 | if verbose: print 'P%d: Receiving boundary map' %(myid) |
|---|
| 110 | |
|---|
| 111 | boundary_map = receive(0) |
|---|
| 112 | |
|---|
| 113 | |
|---|
| 114 | send_s2p = False |
|---|
| 115 | |
|---|
| 116 | if myid == 0: |
|---|
| 117 | # Partition and distribute mesh. |
|---|
| 118 | # Structures returned is in the |
|---|
| 119 | # correct form for the ANUGA data structure |
|---|
| 120 | |
|---|
| 121 | |
|---|
| 122 | points, vertices, boundary, quantities,\ |
|---|
| 123 | ghost_recv_dict, full_send_dict,\ |
|---|
| 124 | number_of_full_nodes, number_of_full_triangles,\ |
|---|
| 125 | s2p_map, p2s_map, tri_map, node_map, ghost_layer_width =\ |
|---|
| 126 | distribute_mesh(domain, verbose=verbose, debug=debug, parameters=parameters) |
|---|
| 127 | |
|---|
| 128 | # Extract l2g maps |
|---|
| 129 | tri_l2g = extract_l2g_map(tri_map) |
|---|
| 130 | node_l2g = extract_l2g_map(node_map) |
|---|
| 131 | |
|---|
| 132 | if debug: |
|---|
| 133 | print 'P%d' %myid |
|---|
| 134 | print 'tri_map ',tri_map |
|---|
| 135 | print 'node_map',node_map |
|---|
| 136 | print 'tri_l2g', tri_l2g |
|---|
| 137 | print 'node_l2g', node_l2g |
|---|
| 138 | print 's2p_map', s2p_map |
|---|
| 139 | print 'p2s_map', p2s_map |
|---|
| 140 | |
|---|
| 141 | |
|---|
| 142 | def protocol(x): |
|---|
| 143 | vanilla=False |
|---|
| 144 | import pypar |
|---|
| 145 | control_info, x = pypar.create_control_info(x, vanilla, return_object=True) |
|---|
| 146 | print 'protocol', control_info[0] |
|---|
| 147 | |
|---|
| 148 | # Send serial to parallel (s2p) and parallel to serial (p2s) triangle mapping to proc 1 .. numprocs |
|---|
| 149 | |
|---|
| 150 | |
|---|
| 151 | if send_s2p : |
|---|
| 152 | n = len(s2p_map) |
|---|
| 153 | s2p_map_keys_flat = num.reshape(num.array(s2p_map.keys(),num.int), (n,1) ) |
|---|
| 154 | s2p_map_values_flat = num.array(s2p_map.values(),num.int) |
|---|
| 155 | s2p_map_flat = num.concatenate( (s2p_map_keys_flat, s2p_map_values_flat), axis=1 ) |
|---|
| 156 | |
|---|
| 157 | n = len(p2s_map) |
|---|
| 158 | p2s_map_keys_flat = num.reshape(num.array(p2s_map.keys(),num.int), (n,2) ) |
|---|
| 159 | p2s_map_values_flat = num.reshape(num.array(p2s_map.values(),num.int) , (n,1)) |
|---|
| 160 | p2s_map_flat = num.concatenate( (p2s_map_keys_flat, p2s_map_values_flat), axis=1 ) |
|---|
| 161 | |
|---|
| 162 | for p in range(1, numprocs): |
|---|
| 163 | |
|---|
| 164 | # FIXME SR: Creates cPickle dump |
|---|
| 165 | send(s2p_map_flat, p) |
|---|
| 166 | # FIXME SR: Creates cPickle dump |
|---|
| 167 | #print p2s_map |
|---|
| 168 | send(p2s_map_flat, p) |
|---|
| 169 | else: |
|---|
| 170 | if verbose: print 'Not sending s2p_map and p2s_map' |
|---|
| 171 | s2p_map = None |
|---|
| 172 | p2s_map = None |
|---|
| 173 | |
|---|
| 174 | if verbose: print 'Communication done' |
|---|
| 175 | |
|---|
| 176 | else: |
|---|
| 177 | # Read in the mesh partition that belongs to this |
|---|
| 178 | # processor |
|---|
| 179 | if verbose: print 'P%d: Receiving submeshes' %(myid) |
|---|
| 180 | points, vertices, boundary, quantities,\ |
|---|
| 181 | ghost_recv_dict, full_send_dict,\ |
|---|
| 182 | number_of_full_nodes, number_of_full_triangles, \ |
|---|
| 183 | tri_map, node_map, ghost_layer_width =\ |
|---|
| 184 | rec_submesh(0, verbose) |
|---|
| 185 | |
|---|
| 186 | |
|---|
| 187 | |
|---|
| 188 | # Extract l2g maps |
|---|
| 189 | tri_l2g = extract_l2g_map(tri_map) |
|---|
| 190 | node_l2g = extract_l2g_map(node_map) |
|---|
| 191 | |
|---|
| 192 | # Recieve serial to parallel (s2p) and parallel to serial (p2s) triangle mapping |
|---|
| 193 | if send_s2p : |
|---|
| 194 | s2p_map_flat = receive(0) |
|---|
| 195 | s2p_map = dict.fromkeys(s2p_map_flat[:,0], s2p_map_flat[:,1:2]) |
|---|
| 196 | |
|---|
| 197 | p2s_map_flat = receive(0) |
|---|
| 198 | p2s_map_keys = [tuple(x) for x in p2s_map_flat[:,0:1]] |
|---|
| 199 | |
|---|
| 200 | p2s_map = dict.fromkeys(p2s_map_keys, p2s_map_flat[:,2]) |
|---|
| 201 | else: |
|---|
| 202 | s2p_map = None |
|---|
| 203 | p2s_map = None |
|---|
| 204 | |
|---|
| 205 | #------------------------------------------------------------------------ |
|---|
| 206 | # Build the domain for this processor using partion structures |
|---|
| 207 | #------------------------------------------------------------------------ |
|---|
| 208 | |
|---|
| 209 | if verbose: print 'myid = %g, no_full_nodes = %g, no_full_triangles = %g' % (myid, number_of_full_nodes, number_of_full_triangles) |
|---|
| 210 | |
|---|
| 211 | |
|---|
| 212 | domain = Parallel_domain(points, vertices, boundary, |
|---|
| 213 | full_send_dict=full_send_dict, |
|---|
| 214 | ghost_recv_dict=ghost_recv_dict, |
|---|
| 215 | number_of_full_nodes=number_of_full_nodes, |
|---|
| 216 | number_of_full_triangles=number_of_full_triangles, |
|---|
| 217 | geo_reference=georef, |
|---|
| 218 | number_of_global_triangles = number_of_global_triangles, |
|---|
| 219 | number_of_global_nodes = number_of_global_nodes, |
|---|
| 220 | s2p_map = s2p_map, |
|---|
| 221 | p2s_map = p2s_map, ## jj added this |
|---|
| 222 | tri_l2g = tri_l2g, ## SR added this |
|---|
| 223 | node_l2g = node_l2g, |
|---|
| 224 | ghost_layer_width = ghost_layer_width) |
|---|
| 225 | |
|---|
| 226 | #------------------------------------------------------------------------ |
|---|
| 227 | # Transfer initial conditions to each subdomain |
|---|
| 228 | #------------------------------------------------------------------------ |
|---|
| 229 | for q in quantities: |
|---|
| 230 | domain.set_quantity(q, quantities[q]) |
|---|
| 231 | |
|---|
| 232 | |
|---|
| 233 | #------------------------------------------------------------------------ |
|---|
| 234 | # Transfer boundary conditions to each subdomain |
|---|
| 235 | #------------------------------------------------------------------------ |
|---|
| 236 | boundary_map['ghost'] = None # Add binding to ghost boundary |
|---|
| 237 | domain.set_boundary(boundary_map) |
|---|
| 238 | |
|---|
| 239 | |
|---|
| 240 | #------------------------------------------------------------------------ |
|---|
| 241 | # Transfer other attributes to each subdomain |
|---|
| 242 | #------------------------------------------------------------------------ |
|---|
| 243 | domain.set_name(domain_name) |
|---|
| 244 | domain.set_datadir(domain_dir) |
|---|
| 245 | domain.set_store(domain_store) |
|---|
| 246 | domain.set_store_vertices_smoothly(domain_smooth,domain_reduction) |
|---|
| 247 | domain.set_minimum_storable_height(domain_minimum_storable_height) |
|---|
| 248 | domain.set_minimum_allowed_height(domain_minimum_allowed_height) |
|---|
| 249 | domain.set_flow_algorithm(domain_flow_algorithm) |
|---|
| 250 | domain.geo_reference = georef |
|---|
| 251 | |
|---|
| 252 | #------------------------------------------------------------------------ |
|---|
| 253 | # Return parallel domain to all nodes |
|---|
| 254 | #------------------------------------------------------------------------ |
|---|
| 255 | return domain |
|---|
| 256 | |
|---|
| 257 | |
|---|
| 258 | |
|---|
| 259 | |
|---|
| 260 | |
|---|
| 261 | |
|---|
| 262 | def distribute_mesh(domain, verbose=False, debug=False, parameters=None): |
|---|
| 263 | |
|---|
| 264 | |
|---|
| 265 | if debug: |
|---|
| 266 | verbose = True |
|---|
| 267 | |
|---|
| 268 | numprocs = size() |
|---|
| 269 | |
|---|
| 270 | |
|---|
| 271 | # Subdivide the mesh |
|---|
| 272 | if verbose: print 'Subdivide mesh' |
|---|
| 273 | new_nodes, new_triangles, new_boundary, triangles_per_proc, quantities, \ |
|---|
| 274 | s2p_map, p2s_map = \ |
|---|
| 275 | pmesh_divide_metis_with_map(domain, numprocs) |
|---|
| 276 | |
|---|
| 277 | #PETE: s2p_map (maps serial domain triangles to parallel domain triangles) |
|---|
| 278 | # sp2_map (maps parallel domain triangles to domain triangles) |
|---|
| 279 | |
|---|
| 280 | |
|---|
| 281 | |
|---|
| 282 | # Build the mesh that should be assigned to each processor, |
|---|
| 283 | # this includes ghost nodes and the communication pattern |
|---|
| 284 | if verbose: print 'Build submeshes' |
|---|
| 285 | submesh = build_submesh(new_nodes, new_triangles, new_boundary, quantities, triangles_per_proc, parameters) |
|---|
| 286 | |
|---|
| 287 | if verbose: |
|---|
| 288 | for p in range(numprocs): |
|---|
| 289 | N = len(submesh['ghost_nodes'][p]) |
|---|
| 290 | M = len(submesh['ghost_triangles'][p]) |
|---|
| 291 | print 'There are %d ghost nodes and %d ghost triangles on proc %d'\ |
|---|
| 292 | %(N, M, p) |
|---|
| 293 | |
|---|
| 294 | #if debug: |
|---|
| 295 | # from pprint import pprint |
|---|
| 296 | # pprint(submesh) |
|---|
| 297 | |
|---|
| 298 | |
|---|
| 299 | # Send the mesh partition to the appropriate processor |
|---|
| 300 | if verbose: print 'Distribute submeshes' |
|---|
| 301 | for p in range(1, numprocs): |
|---|
| 302 | send_submesh(submesh, triangles_per_proc, p, verbose) |
|---|
| 303 | |
|---|
| 304 | # Build the local mesh for processor 0 |
|---|
| 305 | points, vertices, boundary, quantities, \ |
|---|
| 306 | ghost_recv_dict, full_send_dict, tri_map, node_map, ghost_layer_width =\ |
|---|
| 307 | extract_submesh(submesh, triangles_per_proc,0) |
|---|
| 308 | |
|---|
| 309 | # Keep track of the number full nodes and triangles. |
|---|
| 310 | # This is useful later if one needs access to a ghost-free domain |
|---|
| 311 | # Here, we do it for process 0. The others are done in rec_submesh. |
|---|
| 312 | number_of_full_nodes = len(submesh['full_nodes'][0]) |
|---|
| 313 | number_of_full_triangles = len(submesh['full_triangles'][0]) |
|---|
| 314 | |
|---|
| 315 | #print |
|---|
| 316 | #for p in range(numprocs): |
|---|
| 317 | # print 'Process %d:' %(p) |
|---|
| 318 | # |
|---|
| 319 | # print 'full_triangles:' |
|---|
| 320 | # print submesh['full_triangles'][p] |
|---|
| 321 | # |
|---|
| 322 | # print 'full_nodes:' |
|---|
| 323 | # print submesh['full_nodes'][p] |
|---|
| 324 | # |
|---|
| 325 | # print 'ghost_triangles:' |
|---|
| 326 | # print submesh['ghost_triangles'][p]# |
|---|
| 327 | # |
|---|
| 328 | # print 'ghost_nodes:' |
|---|
| 329 | # print submesh['ghost_nodes'][p] |
|---|
| 330 | # print |
|---|
| 331 | # |
|---|
| 332 | #print 'Receive dict' |
|---|
| 333 | #print ghost_recv_dict |
|---|
| 334 | # |
|---|
| 335 | #print 'Send dict' |
|---|
| 336 | #print full_send_dict |
|---|
| 337 | |
|---|
| 338 | |
|---|
| 339 | # Return structures necessary for building the parallel domain |
|---|
| 340 | return points, vertices, boundary, quantities,\ |
|---|
| 341 | ghost_recv_dict, full_send_dict,\ |
|---|
| 342 | number_of_full_nodes, number_of_full_triangles, \ |
|---|
| 343 | s2p_map, p2s_map, tri_map, node_map, ghost_layer_width |
|---|
| 344 | |
|---|
| 345 | |
|---|
| 346 | |
|---|
| 347 | def extract_l2g_map(map): |
|---|
| 348 | # Extract l2g data from corresponding map |
|---|
| 349 | # Maps |
|---|
| 350 | |
|---|
| 351 | import numpy as num |
|---|
| 352 | |
|---|
| 353 | b = num.arange(len(map)) |
|---|
| 354 | |
|---|
| 355 | l_ids = num.extract(map>-1,map) |
|---|
| 356 | g_ids = num.extract(map>-1,b) |
|---|
| 357 | |
|---|
| 358 | # print len(g_ids) |
|---|
| 359 | # print len(l_ids) |
|---|
| 360 | # print l_ids |
|---|
| 361 | |
|---|
| 362 | l2g = num.zeros_like(g_ids) |
|---|
| 363 | l2g[l_ids] = g_ids |
|---|
| 364 | |
|---|
| 365 | return l2g |
|---|
| 366 | |
|---|
