source: inundation/parallel/parallel_advection.py @ 3120

Last change on this file since 3120 was 2813, checked in by steve, 19 years ago

Moving ghosts into domain.py

File size: 6.0 KB
Line 
1import sys
2from os import sep
3sys.path.append('..'+sep+'pyvolution')
4
5"""Class Parallel_Domain -
62D triangular domains for finite-volume computations of
7the advection equation, with extra structures to allow
8communication between other Parallel_Domains and itself
9
10This module contains a specialisation of class Domain from module advection.py
11
12Ole Nielsen, Stephen Roberts, Duncan Gray, Christopher Zoppou
13Geoscience Australia, 2004-2005
14"""
15
16import logging, logging.config
17logger = logging.getLogger('parallel')
18logger.setLevel(logging.WARNING)
19
20try:
21    logging.config.fileConfig('log.ini')
22except:
23    pass
24
25from pyvolution.advection import *
26from Numeric import zeros, Float, Int, ones, allclose, array
27import pypar
28
29
30class Parallel_Domain(Domain):
31
32    def __init__(self,
33                 coordinates,
34                 vertices,
35                 boundary = None,
36                 full_send_dict = None,
37                 ghost_recv_dict = None,
38                 velocity = None):
39
40        Domain.__init__(self,
41                        coordinates,
42                        vertices,
43                        boundary,
44                        velocity = velocity,
45                        full_send_dict=full_send_dict,
46                        ghost_recv_dict=ghost_recv_dict,
47                        processor=pypar.rank(),
48                        numproc=pypar.size()
49                        )
50
51        N = self.number_of_elements
52
53
54        self.communication_time = 0.0
55        self.communication_reduce_time = 0.0
56
57
58        print 'processor',self.processor
59        print 'numproc',self.numproc
60
61    def check_integrity(self):
62        Domain.check_integrity(self)
63
64        msg = 'Will need to check global and local numbering'
65        assert self.conserved_quantities[0] == 'stage', msg
66
67    def update_timestep(self, yieldstep, finaltime):
68
69        # Calculate local timestep
70        Domain.update_timestep(self, yieldstep, finaltime)
71
72        import time
73        t0 = time.time()
74
75        # For some reason it looks like pypar only reduces numeric arrays
76        # hence we need to create some dummy arrays for communication
77        ltimestep = ones( 1, Float )
78        ltimestep[0] = self.timestep
79        gtimestep = zeros( 1, Float) # Buffer for results
80
81        pypar.raw_reduce(ltimestep, gtimestep, pypar.MIN, 0)
82        pypar.broadcast(gtimestep,0)
83
84        self.timestep = gtimestep[0]
85
86        self.communication_reduce_time += time.time()-t0
87
88
89    def update_ghosts(self):
90
91        # We must send the information from the full cells and
92        # receive the information for the ghost cells
93        # We have a dictionary of lists with ghosts expecting updates from
94        # the separate processors
95
96        from Numeric import take,put
97        import time
98        t0 = time.time()
99
100        stage_cv = self.quantities['stage'].centroid_values
101
102        # update of non-local ghost cells
103        for iproc in range(self.numproc):
104            if iproc == self.processor:
105                #Send data from iproc processor to other processors
106                for send_proc in self.full_send_dict:
107                    if send_proc != iproc:
108
109                        Idf  = self.full_send_dict[send_proc][0]
110                        Xout = self.full_send_dict[send_proc][2]
111
112                        N = len(Idf)
113
114                        #for i in range(N):
115                        #    Xout[i,0] = stage_cv[Idf[i]]
116                        Xout[:,0] = take(stage_cv, Idf)
117
118                        pypar.send(Xout,send_proc)
119
120
121            else:
122                #Receive data from the iproc processor
123                if  self.ghost_recv_dict.has_key(iproc):
124
125                    # LINDA:
126                    # now store ghost as local id, global id, value
127                    Idg = self.ghost_recv_dict[iproc][0]
128                    X = self.ghost_recv_dict[iproc][2]
129
130                    X = pypar.receive(iproc,X)
131                    N = len(Idg)
132
133                    put(stage_cv, Idg, X[:,0])
134                    #for i in range(N):
135                    #    stage_cv[Idg[i]] = X[i,0]
136
137
138        #local update of ghost cells
139        iproc = self.processor
140        if self.full_send_dict.has_key(iproc):
141
142            # LINDA:
143            # now store full as local id, global id, value
144            Idf  = self.full_send_dict[iproc][0]
145
146            # LINDA:
147            # now store ghost as local id, global id, value
148            Idg = self.ghost_recv_dict[iproc][0]
149
150            N = len(Idg)
151
152            #for i in range(N):
153            #    #print i,Idg[i],Idf[i]
154            #    stage_cv[Idg[i]] = stage_cv[Idf[i]]
155
156            put(stage_cv, Idg, take(stage_cv, Idf))
157
158
159        self.communication_time += time.time()-t0
160
161
162    def write_time(self):
163        if self.min_timestep == self.max_timestep:
164            print 'Processor %d, Time = %.4f, delta t = %.8f, steps=%d (%d)'\
165                  %(self.processor, self.time, self.min_timestep, self.number_of_steps,
166                    self.number_of_first_order_steps)
167        elif self.min_timestep > self.max_timestep:
168            print 'Processor %d, Time = %.4f, steps=%d (%d)'\
169                  %(self.processor, self.time, self.number_of_steps,
170                    self.number_of_first_order_steps)
171        else:
172            print 'Processor %d, Time = %.4f, delta t in [%.8f, %.8f], steps=%d (%d)'\
173                  %(self.processor, self.time, self.min_timestep,
174                    self.max_timestep, self.number_of_steps,
175                    self.number_of_first_order_steps)
176
177
178
179    def evolve(self, yieldstep = None, finaltime = None):
180        """Specialisation of basic evolve method from parent class
181        """
182
183        #Initialise real time viz if requested
184        if self.time == 0.0:
185            pass
186
187        #Call basic machinery from parent class
188        for t in Domain.evolve(self, yieldstep, finaltime):
189
190            #Pass control on to outer loop for more specific actions
191            yield(t)
Note: See TracBrowser for help on using the repository browser.