source: inundation/ga/storm_surge/parallel/pytiming.py @ 1363

Last change on this file since 1363 was 1363, checked in by steve, 20 years ago

Changes to visualisation. Can use visualise_color_stage to color stage, but it uses python update (slow).

File size: 6.5 KB
Line 
1#!/usr/bin/env python
2
3# Timing of MPI module for Python and estimation of latency and bandwidth
4#
5# Send numerical array in a ring from processor 0 to 1 etc back to 0
6# Perform timings and compare different sending strategies
7#
8# OMN, OCT 2001
9
10import time, sys, pypar, Numeric
11
12# The send/recv routines
13#
14import pypar
15
16#from mpiext import send_array, receive_array  #most direct methods
17#Not available in final install - use bypass form for now
18
19#--------------------------------------------------------------
20
21method = 2     # Use
22                  # 0: automatically allocated buffers
23                  # 1: user-supplied buffers
24                  # 2: Use bypass - let pypar use direct mpi_ext call
25                  # 3: direct call to mpiext (with buffers),
26                  #    only fractionally better than bypass
27                 
28vanilla = False  # Force use of vanilla format (1)
29consistency_check = False # Check correctness
30
31
32#--------------------------------------------------------------
33# linfit
34#
35def linfit(x, y):
36  """Fit a and b to the model y = ax + b. Return a,b,variance
37  """
38 
39  Sx = Sy = SSoN = SxoN = norm = varest = 0.0
40  N = len(x)
41  assert len(y) == N, "x and y must have same length"
42 
43  for i in range(N):
44    #print("x,y = %f, %f\n",x[i],y[i])
45    Sx  = Sx + x[i]
46    Sy  = Sy + y[i]
47 
48  SxoN = Sx/N
49 
50  a = 0.0 
51  for i in range(N):
52    t    = x[i] - SxoN
53    SSoN = SSoN + t*t
54    a    = a + t*y[i]
55
56
57  a = a/SSoN            # a = (N Sxy - SxSy)/(NSxx - Sx^2) */
58  b = (Sy - Sx*a)/N
59 
60  # Quality - variance estimate \sum_i r_i^2 /(m-n)
61  for i in range(N): 
62    norm = norm + float(x[i])*x[i]
63    res = y[i] - a*x[i] - b
64    varest = varest + res*res
65
66  varest = varest/norm/(N-2)
67  return a, b, varest
68
69
70
71
72#--------------------------------------------------------------
73# Main program
74#
75MAXI  = 10         # Number of blocks
76MAXM  = 500000     # Largest block
77BLOCK = MAXM/MAXI  # Block size
78
79repeats = 10
80msgid = 0
81vanilla = 0 #Select vanilla mode (slower but general)
82
83numprocs = pypar.size()
84myid = pypar.rank()
85processor_name = pypar.Get_processor_name()
86
87if myid == 0:
88  # Main process - Create message, pass on, verify correctness and log timing
89  #
90  print "MAXM = %d, number of processors = %d" %(MAXM, numprocs)
91  print "Measurements are repeated %d times for reliability" %repeats
92
93if numprocs < 2:
94  print "Program needs at least two processors - aborting\n"
95  pypar.Abort()
96   
97pypar.Barrier() #Synchronize all before timing   
98print "I am process %d on %s" %(myid,processor_name)
99
100
101#Initialise data and timings
102#
103from RandomArray import uniform, seed
104seed(17, 53)
105A = uniform(0.0,100.0,MAXM)
106elsize = A.itemsize()
107#print elsize
108
109noelem  = [0]*MAXI
110bytes   = [0]*MAXI         
111avgtime = [0.0]*MAXI         
112mintime = [ 1000000.0]*MAXI     
113maxtime = [-1000000.0]*MAXI           
114
115
116
117
118if myid == 0:   
119  # Determine timer overhead
120  cpuOH = 1.0;
121  for k in range(repeats):   # Repeat to get reliable timings
122    t1 = pypar.Wtime()
123    t2 = pypar.Wtime()
124    if t2-t1 < cpuOH: cpuOH = t2-t1
125   
126  print "Timing overhead is %f seconds.\n" %cpuOH         
127
128     
129# Pass msg circularly   
130for k in range(repeats):
131  if myid == 0:
132    print "Run %d of %d" %(k+1,repeats)
133   
134  for i in range(MAXI):
135    m=BLOCK*i+1       
136   
137    noelem[i] = m
138   
139    pypar.Barrier() # Synchronize
140   
141    if myid == 0:
142      #
143      # Main process
144      #
145      t1 = pypar.Wtime()
146
147      if method==0:
148        pypar.send(A[:m], destination=1, tag=msgid, vanilla=vanilla)
149        C = pypar.receive(numprocs-1, tag=msgid, vanilla=vanilla)
150      elif method == 1: 
151        pypar.send(A[:m], use_buffer=True, destination=1,
152                   tag=msgid, vanilla=vanilla)
153        C = pypar.receive(numprocs-1, buffer=A[:m], tag=msgid, vanilla=vanilla)
154      elif method==2:
155        pypar.send(A[:m], use_buffer=True, destination=1,
156                   tag=msgid, vanilla=vanilla, bypass=True)
157        C = pypar.receive(numprocs-1, buffer=A[:m], tag=msgid,
158                          vanilla=vanilla, bypass=True)
159      else:
160        raise 'Unknown method'
161        #send_array(A[:m], 1, msgid)   
162        #stat = receive_array(A[:m], numprocs-1, msgid)
163        #C = A[:m]
164       
165      t2 = pypar.Wtime() - t1 - cpuOH
166      t2 = t2/numprocs
167      avgtime[i] = avgtime[i] + t2
168      if t2 < mintime[i]: mintime[i] = t2
169      if t2 > maxtime[i]: maxtime[i] = t2
170
171      # Uncomment to verify integrity of data
172      # However, this may affect accuracy of timings for some reason.
173      #
174      if consistency_check:
175        assert Numeric.alltrue(C == A[:m])
176    else:
177      #
178      # Parallel process - get msg and pass it on
179      #
180
181      if method==0:
182        C = pypar.receive(myid-1, tag=msgid, vanilla=vanilla)
183        pypar.send(A[:m], destination=(myid+1)%numprocs,
184                   tag=msgid, vanilla=vanilla)
185      elif method==1: 
186        C = pypar.receive(myid-1, buffer=A[:m], tag=msgid, vanilla=vanilla)
187        pypar.send(A[:m], use_buffer=True, destination=(myid+1)%numprocs,
188                   tag=msgid, vanilla=vanilla)                       
189      elif method==2:
190        # Use pypar bypass
191        C = pypar.receive(myid-1, buffer=A[:m], tag=msgid,
192                          vanilla=vanilla, bypass=True)
193        pypar.send(A[:m], use_buffer=True, destination=(myid+1)%numprocs,
194                   tag=msgid, vanilla=vanilla, bypass=True)
195      else:
196        raise 'Unknown method'
197        # Use direct mpiext call
198        #stat = receive_array(A[:m], myid-1, msgid)               
199        #send_array(A[:m], (myid+1)%numprocs, msgid)
200
201
202# Output stats
203#
204if myid == 0:
205  print "Bytes transferred   time (micro seconds)"
206  print "                    min        avg        max "     
207  print "----------------------------------------------"
208     
209  for i in range(MAXI):
210    avgtime[i] = avgtime[i]/repeats*1.0e6 #Average micro seconds
211    mintime[i] = mintime[i]*1.0e6         #Min micro seconds       
212    maxtime[i] = maxtime[i]*1.0e6         #Min micro seconds             
213         
214    m = noelem[i]
215    bytes[i] = elsize*noelem[i]       
216     
217    print "%10d    %10d %10d %10d" %(bytes[i], mintime[i], avgtime[i], maxtime[i]) 
218
219
220  Tbw, Tlat, varest = linfit(bytes, mintime)
221  print "\nLinear regression on best timings (t = t_l + t_b * bytes):\n",
222  print "  t_b = %f\n  t_l = %f" %(Tbw,Tlat)
223  print "  Estimated relative variance = %.9f\n" %varest
224       
225  print "Estimated bandwith (1/t_b):  %.3f Mb/s" %(1.0/Tbw)   
226  print "Estimated latency:           %d micro s" %int(mintime[0]-bytes[0]*Tbw) 
227
228   
229pypar.Finalize()
230
231
232
233
234 
235 
Note: See TracBrowser for help on using the repository browser.