[3185] | 1 | import sys |
---|
| 2 | from os import sep |
---|
| 3 | |
---|
| 4 | # FIXME: This should be removed. Set evironment variable PYTHONPATH to |
---|
| 5 | # directory ...../anuga/inundation instead |
---|
| 6 | #sys.path.append('..'+sep+'pyvolution') |
---|
| 7 | |
---|
| 8 | """Class Parallel_Shallow_Water_Domain - |
---|
| 9 | 2D triangular domains for finite-volume computations of |
---|
| 10 | the shallow water equation, with extra structures to allow |
---|
| 11 | communication between other Parallel_Domains and itself |
---|
| 12 | |
---|
| 13 | This module contains a specialisation of class Domain |
---|
| 14 | from module shallow_water.py |
---|
| 15 | |
---|
| 16 | Ole Nielsen, Stephen Roberts, Duncan Gray, Christopher Zoppou |
---|
| 17 | Geoscience Australia, 2004-2005 |
---|
| 18 | """ |
---|
| 19 | |
---|
| 20 | import logging, logging.config |
---|
| 21 | logger = logging.getLogger('parallel') |
---|
| 22 | logger.setLevel(logging.WARNING) |
---|
| 23 | |
---|
| 24 | try: |
---|
| 25 | logging.config.fileConfig('log.ini') |
---|
| 26 | except: |
---|
| 27 | pass |
---|
| 28 | |
---|
| 29 | from pyvolution.shallow_water import * |
---|
| 30 | from Numeric import zeros, Float, Int, ones, allclose, array |
---|
| 31 | #from pypar_dist import pypar |
---|
| 32 | import pypar |
---|
| 33 | |
---|
| 34 | class Parallel_Domain(Domain): |
---|
| 35 | |
---|
| 36 | def __init__(self, coordinates, vertices, boundary = None, |
---|
| 37 | full_send_dict = None, ghost_recv_dict = None): |
---|
| 38 | |
---|
| 39 | Domain.__init__(self, |
---|
| 40 | coordinates, |
---|
| 41 | vertices, |
---|
| 42 | boundary, |
---|
| 43 | full_send_dict=full_send_dict, |
---|
| 44 | ghost_recv_dict=ghost_recv_dict, |
---|
| 45 | processor=pypar.rank(), |
---|
| 46 | numproc=pypar.size()) |
---|
| 47 | |
---|
| 48 | N = self.number_of_elements |
---|
| 49 | |
---|
| 50 | # self.processor = pypar.rank() |
---|
| 51 | # self.numproc = pypar.size() |
---|
| 52 | # |
---|
| 53 | # # Setup Communication Buffers |
---|
| 54 | # self.nsys = 3 |
---|
| 55 | # for key in full_send_dict: |
---|
| 56 | # buffer_shape = full_send_dict[key][0].shape[0] |
---|
| 57 | # full_send_dict[key].append(zeros( (buffer_shape,self.nsys) ,Float)) |
---|
| 58 | # |
---|
| 59 | # |
---|
| 60 | # for key in ghost_recv_dict: |
---|
| 61 | # buffer_shape = ghost_recv_dict[key][0].shape[0] |
---|
| 62 | # ghost_recv_dict[key].append(zeros( (buffer_shape,self.nsys) ,Float)) |
---|
| 63 | # |
---|
| 64 | # self.full_send_dict = full_send_dict |
---|
| 65 | self.ghost_recv_dict = ghost_recv_dict |
---|
| 66 | |
---|
| 67 | # Buffers for synchronisation of timesteps |
---|
| 68 | self.local_timestep = zeros(1, Float) |
---|
| 69 | self.global_timestep = zeros(1, Float) |
---|
| 70 | |
---|
| 71 | self.local_timesteps = zeros(self.numproc, Float) |
---|
| 72 | |
---|
| 73 | |
---|
| 74 | self.communication_time = 0.0 |
---|
| 75 | self.communication_reduce_time = 0.0 |
---|
| 76 | self.communication_broadcast_time = 0.0 |
---|
| 77 | |
---|
| 78 | |
---|
| 79 | def set_name(self, name): |
---|
| 80 | """Assign name based on processor number |
---|
| 81 | """ |
---|
| 82 | |
---|
| 83 | # Call parents method with processor number attached. |
---|
| 84 | Domain.set_name(self, name + '_P%d_%d' %(self.processor, self.numproc)) |
---|
| 85 | |
---|
| 86 | |
---|
| 87 | def check_integrity(self): |
---|
| 88 | Domain.check_integrity(self) |
---|
| 89 | |
---|
| 90 | msg = 'Will need to check global and local numbering' |
---|
| 91 | assert self.conserved_quantities[0] == 'stage', msg |
---|
| 92 | assert self.conserved_quantities[1] == 'xmomentum', msg |
---|
| 93 | assert self.conserved_quantities[2] == 'ymomentum', msg |
---|
| 94 | |
---|
| 95 | |
---|
| 96 | def update_timestep_1(self, yieldstep, finaltime): |
---|
| 97 | """Calculate local timestep using broadcasts |
---|
| 98 | """ |
---|
| 99 | |
---|
| 100 | #LINDA: |
---|
| 101 | # Moved below so timestep is found before doing update |
---|
| 102 | |
---|
| 103 | #Domain.update_timestep(self, yieldstep, finaltime) |
---|
| 104 | |
---|
| 105 | import time |
---|
| 106 | |
---|
| 107 | |
---|
| 108 | t0 = time.time() |
---|
| 109 | |
---|
| 110 | #Broadcast local timestep from every processor to every other |
---|
| 111 | for pid in range(self.numproc): |
---|
| 112 | #print 'P%d calling broadcast from %d' %(self.processor, pid) |
---|
| 113 | self.local_timestep[0] = self.timestep |
---|
| 114 | pypar.broadcast(self.local_timestep, pid, bypass=True) |
---|
| 115 | self.local_timesteps[pid] = self.local_timestep[0] |
---|
| 116 | |
---|
| 117 | self.timestep = min(self.local_timesteps) |
---|
| 118 | |
---|
| 119 | pypar.barrier() |
---|
| 120 | self.communication_broadcast_time += time.time()-t0 |
---|
| 121 | |
---|
| 122 | # LINDA: |
---|
| 123 | # Moved timestep to here |
---|
| 124 | |
---|
| 125 | Domain.update_timestep(self, yieldstep, finaltime) |
---|
| 126 | |
---|
| 127 | |
---|
| 128 | def update_timestep(self, yieldstep, finaltime): |
---|
| 129 | """Calculate local timestep |
---|
| 130 | """ |
---|
| 131 | |
---|
| 132 | # LINDA: Moved below so timestep is updated before |
---|
| 133 | # calculating statistic |
---|
| 134 | |
---|
| 135 | #Compute minimal timestep on local process |
---|
| 136 | #Domain.update_timestep(self, yieldstep, finaltime) |
---|
| 137 | |
---|
| 138 | pypar.barrier() |
---|
| 139 | |
---|
| 140 | import time |
---|
| 141 | #Compute minimal timestep across all processes |
---|
| 142 | self.local_timestep[0] = self.timestep |
---|
| 143 | use_reduce_broadcast = True |
---|
| 144 | if use_reduce_broadcast: |
---|
| 145 | t0 = time.time() |
---|
| 146 | pypar.reduce(self.local_timestep, pypar.MIN, 0, |
---|
| 147 | buffer=self.global_timestep, |
---|
| 148 | bypass=True) |
---|
| 149 | |
---|
| 150 | else: |
---|
| 151 | #Alternative: Try using straight send and receives |
---|
| 152 | t0 = time.time() |
---|
| 153 | self.global_timestep[0] = self.timestep |
---|
| 154 | |
---|
| 155 | if self.processor == 0: |
---|
| 156 | for i in range(1, self.numproc): |
---|
| 157 | pypar.receive(i, |
---|
| 158 | buffer=self.local_timestep, |
---|
| 159 | bypass=True) |
---|
| 160 | |
---|
| 161 | if self.local_timestep[0] < self.global_timestep[0]: |
---|
| 162 | self.global_timestep[0] = self.local_timestep[0] |
---|
| 163 | else: |
---|
| 164 | pypar.send(self.local_timestep, 0, |
---|
| 165 | use_buffer=True, bypass=True) |
---|
| 166 | |
---|
| 167 | |
---|
| 168 | self.communication_reduce_time += time.time()-t0 |
---|
| 169 | |
---|
| 170 | |
---|
| 171 | #Broadcast minimal timestep to all |
---|
| 172 | t0 = time.time() |
---|
| 173 | pypar.broadcast(self.global_timestep, 0, |
---|
| 174 | bypass=True) |
---|
| 175 | |
---|
| 176 | self.communication_broadcast_time += time.time()-t0 |
---|
| 177 | |
---|
| 178 | |
---|
| 179 | self.timestep = self.global_timestep[0] |
---|
| 180 | |
---|
| 181 | # LINDA: |
---|
| 182 | # update local stats now |
---|
| 183 | |
---|
| 184 | #Compute minimal timestep on local process |
---|
| 185 | Domain.update_timestep(self, yieldstep, finaltime) |
---|
| 186 | |
---|
| 187 | #update_timestep = update_timestep_1 |
---|
| 188 | |
---|
| 189 | def update_ghosts(self): |
---|
| 190 | |
---|
| 191 | # We must send the information from the full cells and |
---|
| 192 | # receive the information for the ghost cells |
---|
| 193 | # We have a dictionary of lists with ghosts expecting updates from |
---|
| 194 | # the separate processors |
---|
| 195 | |
---|
| 196 | |
---|
| 197 | from Numeric import take,put |
---|
| 198 | import time |
---|
| 199 | t0 = time.time() |
---|
| 200 | |
---|
| 201 | # update of non-local ghost cells |
---|
| 202 | for iproc in range(self.numproc): |
---|
| 203 | if iproc == self.processor: |
---|
| 204 | #Send data from iproc processor to other processors |
---|
| 205 | for send_proc in self.full_send_dict: |
---|
| 206 | if send_proc != iproc: |
---|
| 207 | |
---|
| 208 | Idf = self.full_send_dict[send_proc][0] |
---|
| 209 | Xout = self.full_send_dict[send_proc][2] |
---|
| 210 | |
---|
| 211 | for i, q in enumerate(self.conserved_quantities): |
---|
| 212 | #print 'Send',i,q |
---|
| 213 | Q_cv = self.quantities[q].centroid_values |
---|
| 214 | Xout[:,i] = take(Q_cv, Idf) |
---|
| 215 | |
---|
| 216 | pypar.send(Xout, send_proc, |
---|
| 217 | use_buffer=True, bypass = True) |
---|
| 218 | |
---|
| 219 | |
---|
| 220 | else: |
---|
| 221 | #Receive data from the iproc processor |
---|
| 222 | if self.ghost_recv_dict.has_key(iproc): |
---|
| 223 | |
---|
| 224 | Idg = self.ghost_recv_dict[iproc][0] |
---|
| 225 | X = self.ghost_recv_dict[iproc][2] |
---|
| 226 | |
---|
| 227 | X = pypar.receive(iproc, buffer=X, bypass = True) |
---|
| 228 | |
---|
| 229 | for i, q in enumerate(self.conserved_quantities): |
---|
| 230 | #print 'Receive',i,q |
---|
| 231 | Q_cv = self.quantities[q].centroid_values |
---|
| 232 | put(Q_cv, Idg, X[:,i]) |
---|
| 233 | |
---|
| 234 | #local update of ghost cells |
---|
| 235 | iproc = self.processor |
---|
| 236 | if self.full_send_dict.has_key(iproc): |
---|
| 237 | |
---|
| 238 | # LINDA: |
---|
| 239 | # now store full as local id, global id, value |
---|
| 240 | Idf = self.full_send_dict[iproc][0] |
---|
| 241 | |
---|
| 242 | # LINDA: |
---|
| 243 | # now store ghost as local id, global id, value |
---|
| 244 | Idg = self.ghost_recv_dict[iproc][0] |
---|
| 245 | |
---|
| 246 | for i, q in enumerate(self.conserved_quantities): |
---|
| 247 | #print 'LOCAL SEND RECEIVE',i,q |
---|
| 248 | Q_cv = self.quantities[q].centroid_values |
---|
| 249 | put(Q_cv, Idg, take(Q_cv, Idf)) |
---|
| 250 | |
---|
| 251 | self.communication_time += time.time()-t0 |
---|
| 252 | |
---|
| 253 | |
---|
| 254 | def write_time(self): |
---|
| 255 | if self.min_timestep == self.max_timestep: |
---|
| 256 | print 'Processor %d, Time = %.4f, delta t = %.8f, steps=%d (%d)'\ |
---|
| 257 | %(self.processor, self.time, self.min_timestep, self.number_of_steps, |
---|
| 258 | self.number_of_first_order_steps) |
---|
| 259 | elif self.min_timestep > self.max_timestep: |
---|
| 260 | print 'Processor %d, Time = %.4f, steps=%d (%d)'\ |
---|
| 261 | %(self.processor, self.time, self.number_of_steps, |
---|
| 262 | self.number_of_first_order_steps) |
---|
| 263 | else: |
---|
| 264 | print 'Processor %d, Time = %.4f, delta t in [%.8f, %.8f], steps=%d (%d)'\ |
---|
| 265 | %(self.processor, self.time, self.min_timestep, |
---|
| 266 | self.max_timestep, self.number_of_steps, |
---|
| 267 | self.number_of_first_order_steps) |
---|
| 268 | |
---|
| 269 | |
---|
| 270 | def evolve(self, yieldstep = None, finaltime = None): |
---|
| 271 | """Specialisation of basic evolve method from parent class |
---|
| 272 | """ |
---|
| 273 | |
---|
| 274 | #Initialise real time viz if requested |
---|
| 275 | if self.time == 0.0: |
---|
| 276 | pass |
---|
| 277 | |
---|
| 278 | #Call basic machinery from parent class |
---|
| 279 | for t in Domain.evolve(self, yieldstep, finaltime): |
---|
| 280 | |
---|
| 281 | #Pass control on to outer loop for more specific actions |
---|
| 282 | yield(t) |
---|