Merge branch 'develop'

Author: giadarol

ring_of_CPUs_multiturn.py

@@ -1,5 +1,7 @@
 import numpy as np
 import time
+import sys, os
+import socket
 
 from ring_of_CPUs import SingleCoreComminicator
 
@@ -13,11 +15,20 @@ def print2logandstdo(message, mode='a+'):
         fid.writelines([message+'\n'])
 
 
+def verbose_mpi_out(message, myid, mpi_verbose, mode='a+'):
+    if mpi_verbose:
+        t_now = time.mktime(time.localtime())
+        time_string = time.strftime("%d/%m/%Y %H:%M:%S", time.localtime(t_now))
+
+        with open('mpi_logfile_cpu%03d.txt'%myid, mode) as fid:
+            fid.writelines([time_string + ' - ' + message + '\n'])
+        
+
 class RingOfCPUs_multiturn(object):
     def __init__(self, sim_content, N_pieces_per_transfer=1, force_serial = False, comm=None,
                     N_parellel_rings = 1,
                     N_buffer_float_size = 1000000, N_buffer_int_size = 100,
-                    verbose = False,
+                    verbose = False, mpi_verbose = False, enable_barriers = False,
                     enable_orders_from_master = True):
 
 
@@ -30,6 +41,8 @@ def __init__(self, sim_content, N_pieces_per_transfer=1, force_serial = False, c
         self.N_parellel_rings = N_parellel_rings
 
         self.verbose = verbose
+        self.mpi_verbose = mpi_verbose
+        self.enable_barriers = enable_barriers
 
         self.enable_orders_from_master = enable_orders_from_master
 
@@ -45,28 +58,56 @@ def __init__(self, sim_content, N_pieces_per_transfer=1, force_serial = False, c
         if hasattr(sim_content, 'N_parellel_rings'):
             self.N_parellel_rings = sim_content.N_parellel_rings
 
+        if hasattr(sim_content, 'verbose'):
+            self.verbose = sim_content.verbose
+        
+        if hasattr(sim_content, 'mpi_verbose'):
+            self.mpi_verbose = sim_content.mpi_verbose
+        
+        if hasattr(sim_content, 'enable_barriers'):
+            self.enable_barriers = sim_content.enable_barriers
+        
 
         self.sim_content.ring_of_CPUs = self
 
         # choice of the communicator
         if force_serial:
             comm_info = 'Single CPU forced by user.'
             self.comm = SingleCoreComminicator()
+            self.comm_type = 'Serial'
         elif comm is not None:
             comm_info = 'Multiprocessing using communicator provided as argument.'
             self.comm = comm
+            self.comm_type = 'Multip'
         else:
             comm_info = 'Multiprocessing via MPI.'
             from mpi4py import MPI
             self.comm = MPI.COMM_WORLD
+            self.comm_type = 'MPI'
 
+        self.verbose_mpi_out = lambda message: verbose_mpi_out(message, self.comm.Get_rank(), 
+                                                                   self.mpi_verbose)
+
+        verbose_mpi_out('Debug file (cpu %d)'%(self.comm.Get_rank()), self.comm.Get_rank(), 
+                        self.mpi_verbose, mode = 'w')
+
+        self.verbose_mpi_out('Interpreter at %s (cpu %d)'%(sys.executable, self.comm.Get_rank()))
+        if self.mpi_verbose and self.comm_type=='MPI':
+           import mpi4py
+           self.verbose_mpi_out('mpi4py version: %s (cpu %d)'%(mpi4py.__version__, self.comm.Get_rank()))
+
+        self.verbose_mpi_out('Running on %s (cpu %d)'%(socket.gethostname(), self.comm.Get_rank()))         
+        
+        if self.enable_barriers:
+            self.verbose_mpi_out('At barrier 1 (cpu %d)'%self.comm.Get_rank())
+            self.comm.Barrier()
+            self.verbose_mpi_out('After barrier 1 (cpu %d)'%self.comm.Get_rank())
+
         #check if there is only one node
         if self.comm.Get_size()==1:
             #in case it is forced by user it will be rebound but there is no harm in that
             self.comm = SingleCoreComminicator()
 
-        #~ if self.N_pieces_per_transfer>1:
-            #~ raise ValueError("Not implemented!")
 
         # get info on the grid
         self.N_nodes = self.comm.Get_size()
@@ -96,8 +137,18 @@ def __init__(self, sim_content, N_pieces_per_transfer=1, force_serial = False, c
         self.buf_float = np.zeros(self.N_buffer_float_size, dtype=np.float64)
         self.buf_int = np.array(self.N_buffer_int_size*[0])
 
+        if self.enable_barriers:
+            self.verbose_mpi_out('At barrier 2 (cpu %d)'%self.comm.Get_rank())
+            self.comm.Barrier()
+            self.verbose_mpi_out('After barrier 2 (cpu %d)'%self.comm.Get_rank())
+
         self.sim_content.init_all()
 
+        if self.enable_barriers:
+            self.verbose_mpi_out('At barrier 3 (cpu %d)'%self.comm.Get_rank())
+            self.comm.Barrier()
+            self.verbose_mpi_out('After barrier 3 (cpu %d)'%self.comm.Get_rank())
+
         if self.I_am_the_master:
             print2logandstdo('PyPARIS simulation -- multiturn parallelization')#, mode='w+')
             print2logandstdo(comm_info)
@@ -108,14 +159,17 @@ def __init__(self, sim_content, N_pieces_per_transfer=1, force_serial = False, c
             print2logandstdo('Multi-ring info:')
             print2logandstdo('N_parellel_rings = %d'%self.N_parellel_rings)
             print2logandstdo('N_nodes_per_ring = %d'%self.N_nodes_per_ring)
-            import socket
-            import sys
             print2logandstdo('Running on %s'%socket.gethostname())
-            print2logandstdo('Interpreter at %s'%sys.executable)			
+            print2logandstdo('Interpreter at %s'%sys.executable)            
 
         self.left = int(np.mod(self.myid-1, self.N_nodes))
         self.right = int(np.mod(self.myid+1, self.N_nodes))
 
+        if self.enable_barriers:
+            self.verbose_mpi_out('At barrier 4 (cpu %d)'%self.comm.Get_rank())
+            self.comm.Barrier()
+            self.verbose_mpi_out('After barrier 4 (cpu %d)'%self.comm.Get_rank())
+
         if self.I_am_at_start_ring:
             self.bunches_to_be_treated = deque([])
             self.slices_to_be_treated = []
@@ -128,7 +182,13 @@ def __init__(self, sim_content, N_pieces_per_transfer=1, force_serial = False, c
         if self.I_am_the_master:
             list_bunches = sim_content.init_master()
             self.bunches_to_be_treated.extend(list_bunches)
-            
+
+        if self.enable_barriers:
+            self.verbose_mpi_out('At barrier 5 (cpu %d)'%self.comm.Get_rank())
+            self.comm.Barrier()
+            self.verbose_mpi_out('After barrier 5 (cpu %d)'%self.comm.Get_rank())
+
+
     def run(self):
 
 
@@ -228,11 +288,34 @@ def run(self):
             sendbuf = ch.combine_float_buffers(list_of_buffers_to_send)
             if len(sendbuf) > self.N_buffer_float_size:
                 raise ValueError('Float buffer (%d) is too small!\n %d required.'%(self.N_buffer_float_size, len(sendbuf)))
+
+            if self.enable_barriers:
+                self.verbose_mpi_out('At barrier L1 (cpu %d)'%self.comm.Get_rank())
+                self.comm.Barrier()
+                self.verbose_mpi_out('After barrier L1 (cpu %d)'%self.comm.Get_rank())
+
+            self.verbose_mpi_out('At Sendrecv, cpu %d/%d, iter %d'%(self.myid, self.N_nodes, iteration))
             self.comm.Sendrecv(sendbuf, dest=self.right, sendtag=self.right, 
                         recvbuf=self.buf_float, source=self.left, recvtag=self.myid)
+            self.verbose_mpi_out('After Sendrecv, cpu %d/%d, iter %d'%(self.myid, self.N_nodes, iteration))
+                
+            if self.enable_barriers:
+                self.verbose_mpi_out('At barrier L2 (cpu %d)'%self.comm.Get_rank())
+                self.comm.Barrier()
+                self.verbose_mpi_out('After barrier L2 (cpu %d)'%self.comm.Get_rank())
+
             list_received_buffers = ch.split_float_buffers(self.buf_float)
-    
-            # Handle orders (for now only to stopping the simulation)
+            
+            
+            #######################################################
+            # Handle orders (for now only to stop the simulation) #
+            #######################################################
+            if self.enable_barriers:
+                self.verbose_mpi_out('At barrier L3 (cpu %d)'%self.comm.Get_rank())
+                self.comm.Barrier()
+                self.verbose_mpi_out('After barrier L3 (cpu %d)'%self.comm.Get_rank())
+
+
             if self.enable_orders_from_master:
                 if self.I_am_the_master:
                     # send orders
@@ -241,16 +324,28 @@ def run(self):
                         raise ValueError('Int buffer is too small!')
                     self.buf_int = 0*self.buf_int
                     self.buf_int[:len(buforders)]=buforders
+
+                    self.verbose_mpi_out('At Bcast, cpu %d/%d, iter %d'%(self.myid, self.N_nodes, iteration))
                     self.comm.Bcast(self.buf_int, self.master_id)
+                    self.verbose_mpi_out('After Bcast, cpu %d/%d, iter %d'%(self.myid, self.N_nodes, iteration))
+
                 else:    
                     # receive orders from the master
+                    self.verbose_mpi_out('At Bcast, cpu %d/%d, iter %d'%(self.myid, self.N_nodes, iteration))
                     self.comm.Bcast(self.buf_int, self.master_id)
+                    self.verbose_mpi_out('After Bcast, cpu %d/%d, iter %d'%(self.myid, self.N_nodes, iteration))
+
                     orders_from_master = ch.buffer_2_list_of_strings(self.buf_int)
 
                 # check if simulation has to be ended
                 if 'stop' in orders_from_master:
                     break
 
+            if self.enable_barriers:
+                self.verbose_mpi_out('At barrier L4 (cpu %d)'%self.comm.Get_rank())
+                self.comm.Barrier()
+                self.verbose_mpi_out('After barrier L4 (cpu %d)'%self.comm.Get_rank())
+
             iteration+=1
 
             # (TEMPORARY!) To stop

test_multibunch/001_multibunch_without_ecloud/002_some_checks.py

@@ -39,7 +39,7 @@
     turn_num = np.arange(0, len(ob.mean_x[:, i_bunch]), dtype=np.float)
     axx.plot(ob.mean_x[0, i_bunch]*np.exp(-turn_num/tau_damp_x), 
             linewidth=2, color='red', linestyle='--',
-            label=r'Dumping time = %.0f turns'%tau_damp_x)
+            label=r'Damping time = %.0f turns'%tau_damp_x)
     axy.plot(ob.mean_y[0, i_bunch]*np.exp(-turn_num/tau_damp_y), 
             linewidth=2, color='red', linestyle='--',
             label=r'Damping time = %.0f turns'%tau_damp_y)

test_multibunch/002_multibunch_with_ecloud_small/pyecloud_config/simulation_parameters.input

@@ -69,3 +69,5 @@ flag_movie = 0             #1/0
 flag_sc_movie = 0          #1/0
 
 save_mp_state_time_file =  -1
+
+extract_sey = False