source: inundation/parallel/parallel_shallow_water.py @ 2004

Last change on this file since 2004 was 1697, checked in by steve, 19 years ago

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