Changeset 8513 for trunk/anuga_core/source/anuga_parallel/parallel_generic_communications.py

Timestamp:

Aug 14, 2012, 8:46:07 PM (13 years ago)

Author:

steve

Message:

Adding working asynchronous update_ghosts

File:

• trunk/anuga_core/source/anuga_parallel/parallel_generic_communications.py (modified) (4 diffs)

trunk/anuga_core/source/anuga_parallel/parallel_generic_communications.py

@@ -33 +33 @@
     """
 
-
-
     import time
 
@@ -41 +39 @@
     t0 = time.time()
 
-    allreduce = True
-    
-    if allreduce:
-        import anuga_parallel.pypar_extras as par_extras
-        par_extras.allreduce(domain.local_timestep, pypar.MIN,
+
+    import anuga_parallel.pypar_extras as par_exts
+
+    par_exts.allreduce(domain.local_timestep, pypar.MIN,
                       buffer=domain.global_timestep,
                       bypass=True)
-        domain.communication_reduce_time += time.time()-t0
-    else:
-        pypar.reduce(domain.local_timestep, pypar.MIN, 0,
-                      buffer=domain.global_timestep,
-                      bypass=True)
-
-        domain.communication_reduce_time += time.time()-t0
-
-
-        #Broadcast minimal timestep to all processors
-        t0 = time.time()
-        pypar.broadcast(domain.global_timestep, 0)#, bypass=True)
-
-        domain.communication_broadcast_time += time.time()-t0
+
+    domain.communication_reduce_time += time.time()-t0
+
+
+
+
+#    pypar.reduce(domain.local_timestep, pypar.MIN, 0,
+#                      buffer=domain.global_timestep,
+#                      bypass=True)
+#
+#
+#    domain.communication_reduce_time += time.time()-t0
+
+
+    #Broadcast minimal timestep to all processors
+    t0 = time.time()
+    #pypar.broadcast(domain.global_timestep, 0)#, bypass=True)
+
+    domain.communication_broadcast_time += time.time()-t0
 
     #old_flux_timestep = domain.flux_timestep
@@ -68 +70 @@
     
 
-def communicate_ghosts(domain):
+def communicate_ghosts_blocking(domain):
 
     # We must send the information from the full cells and
@@ -130 +132 @@
     domain.communication_time += time.time()-t0
 
+
+
+def communicate_ghosts_asynchronous(domain):
+
+    # We must send the information from the full cells and
+    # receive the information for the ghost cells
+    # We have a dictionary of lists with ghosts expecting updates from
+    # the separate processors
+    # Using isend and irecv
+
+    import numpy as num
+    import time
+    t0 = time.time()
+
+    # update of non-local ghost cells by copying full cell data into the
+    # Xout buffer arrays
+
+    #iproc == domain.processor
+
+    #Setup send buffer arrays for sending full data to other processors
+    for send_proc in domain.full_send_dict:
+        Idf  = domain.full_send_dict[send_proc][0]
+        Xout = domain.full_send_dict[send_proc][2]
+
+        for i, q in enumerate(domain.conserved_quantities):
+            #print 'Store send data',i,q
+            Q_cv =  domain.quantities[q].centroid_values
+            Xout[:,i] = num.take(Q_cv, Idf)
+
+
+
+#    from pprint import pprint
+#
+#    if pypar.rank() == 0:
+#        print 'Before commun 0'
+#        pprint(domain.full_send_dict)
+#
+#    if pypar.rank() == 1:
+#        print 'Before commun 1'
+#        pprint(domain.full_send_dict)
+
+
+    # Do all the comuunication using isend/irecv via the buffers in the
+    # full_send_dict and ghost_recv_dict
+    from anuga_parallel.pypar_extras import mpiextras
+
+    mpiextras.send_recv_via_dicts(domain.full_send_dict,domain.ghost_recv_dict)
+
+#
+#    if pypar.rank() == 0:
+#        print 'After commun 0'
+#        pprint(domain.ghost_recv_dict)
+#
+#    if pypar.rank() == 1:
+#        print 'After commun 1'
+#        pprint(domain.ghost_recv_dict)
+
+    # Now copy data from receive buffers to the domain
+    for recv_proc in domain.ghost_recv_dict:
+        Idg  = domain.ghost_recv_dict[recv_proc][0]
+        X    = domain.ghost_recv_dict[recv_proc][2]
+
+        #print recv_proc
+        #print X
+
+        for i, q in enumerate(domain.conserved_quantities):
+            #print 'Read receive data',i,q
+            Q_cv =  domain.quantities[q].centroid_values
+            num.put(Q_cv, Idg, X[:,i])
+
+
+    domain.communication_time += time.time()-t0
+
+