def finalize(self):
     """
     End being parallel
     """
     if self.is_parallel is True:
         import pypar
         pypar.finalize()

def run():
    """
    Run the process, handling any parallelisation.
    """
    
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config",
                        help="Configuration file",
                        type=str)
    parser.add_argument("-i", "--inputfile",
                        help="Input DEM file (ascii format)",
                        type=str)
    parser.add_argument("-o", "--output", 
                        help="Output path",
                        type=str)
    parser.add_argument("-v", "--verbose", 
                        help=("Verbose output (not available when invoking"
                                "parallel run)") )
                                
    args = parser.parse_args() 
                          
    logfile = 'topomult.log'
    loglevel = 'INFO'
    
    if args.verbose:
        verbose = args.verbose
    else:
        verbose = False

    if args.config:
        cfg = ConfigParser.ConfigParser()
        cfg.read(args.config)

        input_file = cfg.get('Input', 'Filename')
        output_path = cfg.get('Output', 'Path')
        logfile = cfg.get('Logging', 'LogFile')
        loglevel = cfg.get('Logging', 'LogLevel')
        verbose = cfg.get('Logging', 'Verbose')
        
    if args.inputfile:
        input_file = args.inputfile

    if args.output:
        output_path = args.output
    
    attemptParallel()
    if pp.size() > 1 and pp.rank() > 0:
        logfile += '-' + str(pp.rank())
        verbose = False  # to stop output to console

    flStartLog(logfile, loglevel, verbose)
    
    pp.barrier()
    work(input_file, output_path,
             ['n','s','e','w','ne','nw','se','sw'])
    pp.barrier()
    
    pp.finalize()

    def runMapReduce(self):
        if self.MPI_myid == 0:
            self.result = self.master()
        else:
            self.slave()

        pypar.finalize()
        logging.debug('[PROCESS %d]: MPI environment finalized.'%(self.MPI_myid, ))
        return

def abnormalexit(reason):
    """this tells each worker node to exit, then kills the server process.
       this should only be called by the server node"""
    print 'abnormal exit'
    print reason
    sendtoall(('Die', 0))
    pypar.barrier()
    pypar.finalize()
    sys.exit(2)

def run_client():
    '''
    Runs
    '''
    # Identification
    myid = pypar.rank() # id of this process
    nproc = pypar.size() # number of processors

    print "I am client", myid
    pypar.finalize()

 def run(self):
     if self.myid == 0:
         self.work.masterBeforeWork()
         self.master()
         self.work.masterAfterWork()
     else:
         self.work.slaveBeforeWork()
         self.slave()
         self.work.slaveAfterWork()
     
     pypar.finalize()
     if self.myid != 0:
         sys.exit()

def two_example():
    txt = ["yes", "no", "when", "what the", "a", "5ive!"]

    rank = pypar.rank()
    size = pypar.size()

    print
    print "I am processor %d of %d. " % (rank, size)
    for i, ele in enumerate(txt):
        if i % size == rank:
            print "i" + str(i) + " P" + str(rank) + " len " + str(len(ele)) + " for " + ele

    pypar.finalize()

def main():
    
#    #=========================================================================
#    #==============    Electronic TISE   =====================================
#    #=========================================================================
#    #Get parameters.
#    m_max, nu_max, mu_max, R_grid, beta, theta = el_tise_problem_parameters()
#   
#    #Do calculations.
#    electronic_BO.save_electronic_eigenstates(m_max, nu_max, mu_max, R_grid, beta, theta)
#
#    #=========================================================================   
#    #==============    Electronic TDSE   =====================================
#    #=========================================================================
#    #Get parameters.
#    filename, m, q, E_lim = el_tdse_problem_parameters()
#    
#    #Do calculations.
#    tdse = tdse_electron.TDSE_length_z(filename = filename)
#    tdse.BO_dipole_couplings(m, q, E_lim)
#
#    #=========================================================================   
#    #==============    Vibrational TISE   ====================================
    #=========================================================================
    #Get problem parameters.
    filename_el, nr_kept, xmin, xmax, xsize, order = vib_problem_parameters()
    
    #Do calculations.
    vibrational_BO.save_all_eigenstates(filename_el, nr_kept, 
	xmin, xmax, xsize, order)
    
    #=========================================================================   
    #==============    Vibrational TDSE   ====================================
    #=========================================================================
    #Get problem parameters.
    filename_el, nr_kept, xmin, xmax, xsize, order = vib_problem_parameters()
    
    #Do calculations.
    vibrational_BO.save_all_couplings(filename_el, nr_kept, 
    	xmin, xmax, xsize, order)
    
    #Calculate couplings for the eigenfunction basis.
    #vibrational_BO.save_eigenfunction_couplings(filename_el, nr_kept, 
    #	xmin, xmax, xsize, order)
    
    #=========================================================================
   
    pypar.finalize()

def main():

    # Parse command line

    options, args = cookbook.doc_optparse.parse( __doc__ )

    #try:
    pos_fname, neg_fname, out_dir = args
    align_count, mapping = rp.mapping.alignment_mapping_from_file( file( options.mapping ) )
    #except:
    #    cookbook.doc_optparse.exit()

    try:
        run( open( pos_fname ), open( neg_fname ), out_dir, options.format, align_count, mapping )
    finally:
        pypar.finalize()

    def __init__(self, aWorkList):  
       
        self.WORKTAG = 1
        self.DIETAG =  2
        
        self.MPI_myid =    pypar.rank()
        self.MPI_numproc = pypar.size()
        self.MPI_node =    pypar.get_processor_name()

        self.works = aWorkList
        self.numWorks = len(self.works)

        self.reduceFunction = None
        self.mapFunction = None
        self.result = None
   
        if self.MPI_numproc < 2:
            pypar.finalize()
            if self.MPI_myid == 0:
                raise Exception, 'ERROR: Number of processors must be greater than 2.'

def start(initializer=None, initargs=(), maxtasks=None):
    global Pool
    try:
        # make pypar available
        global pp
        import pypar as pp

        if pp.size() > 1:
            Pool = PyparPool
            if pp.rank() > 0:
                worker(initializer, initargs, maxtasks)
        else:
            # fallback to multiprocessing
            print "Using multiprocessing"
            pp.finalize()
            import multiprocessing as mp

            Pool = mp.Pool

    except ImportError:  # no pypar
        return

def do_run(pdb, i, cur, db, mutationList):
    
    if mutationList != "ALA":
        mfile = Core.Data.MutationListFile(filename=mutationList,create=True)
        mfile.removeDuplicates(autoUpdate=False)
        mutList = mfile.mutantList()
        if isRoot(myid):
            print mfile.numberOfMutants()
    else:
        mutList = Core.Data.CreateScanList(pdbFile=i, mutation='ALA', skipResidueTypes=['ALA', 'GLY'])
    
    results = DeltaStability(inputFile = i, mutationList = mutList, configurationFile='/home/satnam/proteinDesignTool.conf', workingDirectory = os.getcwd(), outputDirectory = os.getcwd())
    
    # Results are submitted to results_pdb+chain and only by one processor
    if isRoot(myid):
        cur.execute("create table if not exists results_%s_%s(mutation VARCHAR(20), score FLOAT);" % (pdb,os.path.split(mutationList)[1]))
        for mutant in range(results.stabilityResults.numberOfRows()):
            cur.execute("insert into results_%s_%s (mutation, score) VALUES (%s%s%s, %s%s%s);" % (pdb,os.path.split(mutationList)[1], '"', results.stabilityResults[mutant][0], '"', '"', results.stabilityResults[mutant][-1],'"'))
        print "Calculated %s stability and results added to database" % (pdb)
            
    pypar.finalize()

def main():
    
    # Ensure all Processors are ready
    pypar.barrier()
    print "Processor %d is ready" % (myid)
    
    # Connect to MySQL db
    db = MySQLdb.connect(host="localhost", 
                         user = "root", 
                         passwd = "samsung", 
                         db = "sat")
    cur = db.cursor()


    # Option parser from wrapper script
    parser = optparse.OptionParser()
    # PDB
    parser.add_option("-p", "--pdb", 
                      help="Choose all or a pdb id", 
                      dest="pdb", default ="all")
    # PDB directory
    parser.add_option("-d", "--dir", 
                      help="i", 
                      dest="i", default ="all")

    parser.add_option("-m", "--mutationList", 
                      help="Location of mutation list file", 
                      dest="m", default="ALA")
    
    (opts, args) = parser.parse_args()
    
    # Run calculations
    do_run(opts.pdb, opts.i, cur, db, opts.m)

    # Finalize and exit
    pypar.finalize()

  mr2 = mr.copy()
  mr2.reduce(output)
  fp.close()
  mr2.destroy()

# stats to screen
# include stats on number of nonzeroes per row

if me == 0:
  print order,"rows in matrix"
  print ntotal,"nonzeroes in matrix"

mr.reduce(nonzero)
mr.collate()
mr.reduce(degree)
mr.collate()
mr.reduce(histo)
mr.gather(1)
mr.sort_keys(ncompare)
total = 0
mr.map_kv(mr,stats)
if me == 0: print order-total,"rows with 0 nonzeroes"

if me == 0:
  print "%g secs to generate matrix on %d procs in %d iterations" % \
        (tstop-tstart,nprocs,niterate)

mr.destroy()
  
pypar.finalize()

    from pypar_balancer import PyparWork, PyparBalancer

    NUM_NODES=pp.size()
    
    if NUM_NODES > 1:
        HAVE_MPI=1  # we have pypar, and we're running with more than one node
        
    if DEBUG:
        if MY_RANK==0:
            if HAVE_PYPAR and HAVE_MPI:
                print "Running full MPI"
            elif HAVE_PYPAR:
                print "MPI available, but not enough nodes for master/slave"

    if HAVE_PYPAR and not HAVE_MPI:
        pp.finalize() # not enough nodes to actually run master/slave... shut down MPI now.
        
except:
    if DEBUG:
        import traceback
        traceback.print_exc()
        if HAVE_PYPAR and HAVE_MPI:
            print "Running full MPI"
        elif HAVE_PYPAR:
            print "MPI available, but not enough nodes for master/slave"
        else:
            print "No MPI."
        
if HAVE_MPI:
    class GenericMPI (PyparWork):
    

import pypar                                       # Import module and initialise MPI 

proc = pypar.size()                                # Number of processes as specified by mpirun
myid = pypar.rank()                                # Id of of this process (myid in [0, proc-1]) 
node = pypar.get_processor_name()                  # Host name on which current process is running

print 'I am proc %d of %d on node %s' %(myid, proc, node)

if myid == 0:                                      # Actions for process 0:
  msg = 'P0'  
  pypar.send(msg, destination=1)                   # Send message to proces 1 (right hand neighbour)
  msg = pypar.receive(source=proc-1)               # Receive message from last process
      
  print 'Processor 0 received message "%s" from processor %d' %(msg, proc-1)

else:                                              # Actions for all other processes:

  source = myid-1                                  # Source is the process to the left
  destination = (myid+1)%proc                      # Destination is process to the right
                                                   # wrapped so that last processor will 
						   # send back to proces 0  
  
  msg = pypar.receive(source)                      # Receive message from source 
  msg = msg + 'P' + str(myid)                      # Update message     
  pypar.send(msg, destination)                     # Send message to destination   

pypar.finalize()                                   # Stop MPI 

#.........这里部分代码省略.........
        # Communicate hazard output directory name to all nodes to ensure they have exactly the same time stamp.
        for i in range(P):
            pypar.send((hazard_output_folder), i)
    else:
        # Receive correctly timestamped output directory names
        hazard_output_folder = pypar.receive(0)
        logdir = os.path.join(hazard_output_folder, 'logs')


    try:
        name = os.path.splitext(scenario)[0]
    except:
        name = 'run'


    # Wait until log dir has been created
    pypar.barrier()

    params = get_scenario_parameters(scenario)

    # Start processes staggered to avoid race conditions for disk access (otherwise it is slow to get started)
    time.sleep(2*p)

    # Logging
    s = 'Proc %i' % p
    print '     %s -' % string.ljust(s, 8),
    AIM_logfile = os.path.join(logdir, 'P%i.log' % p)
    start_logging(filename=AIM_logfile, echo=False)

    # Get cracking
    basename, _ = os.path.splitext(scenario)
    count_local = 0
    count_all = 0
    for i, file in enumerate(os.listdir(windfield_directory)):

        count_all += 1

        # Distribute jobs cyclically to processors
        if i%P == p:

            if not file.endswith('.profile'):
                continue

            count_local += 1

            windfield = '%s/%s' % (windfield_directory, file)
            windname, _ = os.path.splitext(file)
            header('Computing event %i on processor %i using wind field: %s' % (i, p, windfield))



            if dircomment is None:
                dircomment = params['eruption_comment']

            # Override or create parameters derived from native Fall3d wind field
            params['wind_profile'] = windfield
            params['wind_altitudes'] = get_layers_from_windfield(windfield) # FIXME: Try to comment this out.
            params['Meteorological_model'] = 'profile'

            if hazard_output_folder is None:
                hazard_output_folder = basename + '_hazard_outputs'

            if p == 0:
                print 'Storing multiple outputs in directory: %s' % hazard_output_folder

            # Run scenario
            aim = _run_scenario(params,
                                timestamp_output=True,
                                dircomment=dircomment + '_run%i_proc%i' % (i, p))

            # Make sure folder is present and can be shared by group
            makedir(hazard_output_folder)
            s = 'chmod -R g+w %s' % hazard_output_folder
            run(s)

            # Copy result file to output folder
            result_file = aim.scenario_name + '.res.nc'
            newname = aim.scenario_name + '.%s.res.nc' % windname # Name after wind file
            s = 'cp %s/%s %s/%s' % (aim.output_dir, result_file, hazard_output_folder, newname)
            run(s)

            # Create projectionfile in hazard output
            if i == 0:
                s = 'cp %s %s/%s' % (aim.projection_file, hazard_output_folder, 'HazardMaps.res.prj')
                run(s)

            # Clean up outputs from this scenario
            print 'P%i: Cleaning up %s' % (p, aim.output_dir)
            s = '/bin/rm -rf %s' % aim.output_dir
            run(s)

    print 'Processor %i done %i windfields' % (p, count_local)
    print 'Outputs available in directory: %s' % hazard_output_folder

    pypar.barrier()
    if p == 0:
        print 'Parallel simulation finished %i windfields in %i seconds' % (count_all, time.time() - t_start)


    pypar.finalize()

            if status.tag == DIE_TAG:
                continue_working = False
            
            # not being put to sleep, load in videos of interest and compute
            else:
                frame_pair, status = pp.receive(source=MASTER_PROCESS, tag=pp.any_tag, 
                    return_status=True)
                work_index = status.tag
                

                frame_a = np.array(Image.open(os.path.join(vidpath, tif_files[frame_pair[0]])));
                frame_b = np.array(Image.open(os.path.join(vidpath, tif_files[frame_pair[1]])));
                
                # Code below simulates a task running
                u, v = PIVCompute(frame_a, frame_b, window_size = window_size, overlap = overlap)
                print  "Received work frame pair " + str(frame_pair) + " u origin value is " + str(u[0,0])                 
                
                # package up into work array
                work_array = np.zeros((2,u.shape[0], u.shape[1]))
                work_array[0,:,:] = u
                work_array[1,:,:] = v

                result_array = work_array.copy()

                pp.send(result_array, destination=MASTER_PROCESS, tag=work_index)
        #### while
    #### if worker

    pp.finalize()

		print p.rank(), res

	if False:
		s = 0
		for i in xrange(100):
			r = p.rank()
			r = broadcast(r)
			s += (r + 1)
			p.barrier()
		print "%d %d" % ( p.rank(), s )

	if False:
		m = None
		v = None
		if root():
			m = eye_matrix(3000)
			v = range(3000)
		r = p_mv(m,v)
		if root():
			print r

	if root():
		end = p.time()
		total = end - start
		print "total time: %.14f" % total
			
	p.finalize()
		
		
		

def main():
    #--------------------#
    # server code
    #--------------------#
    if rank == 0:
        print 'server running on ', procname

        opts = task(sys.argv)

        opts.printruninfo()

        sendtoall(('Start', sys.argv))
        server = serverdata(opts)

        #set up the collector and generator
        start = time.time()

        collector = resultcollector(server)
        end = time.time()
        print end-start
        
        jobs = jobgenerator(server)

        numjobsreceived = 0
        #begin distributing work
        for proc in xrange(1, min(numnodes, jobs.numjobs+1)):
            job = jobs.next(proc)
            pypar.send(('job',job), proc, tag=OUT)
        while numjobsreceived < jobs.jobindex:#while any job is still running
            #wait for any node to send a result
            msg, status = pypar.receive(pypar.any_source, return_status=True, tag=RETURN)
            numjobsreceived += 1
            proc, response = msg

            if jobs.hasnext(proc):#see if there is more work to be done
                job = jobs.next(proc)
                pypar.send(('job',job), proc, tag=OUT)#send it to the node that just completed

            #combine the results *after* sending the new job
            #(this way the worker can proceed while the results are being combined)
            collector.collect(response)


        #all jobs collected, kill the workers
        sendtoall(('Done', 0))

        #finish up the computation
        collector.finish()
        
    #--------------------#    
    # worker code
    #--------------------#
    else:
        while True:
            start = time.time()
            (code, msg), status = pypar.receive(0, return_status=True, tag=OUT)
            end = time.time()
            print 'waiting', end-start
            if code == 'Done':#all work is done
                opts.printruninfo()
                break
            elif code == 'Die':#abnormal exit
                break
            elif code == 'Start':
                opts = task(msg)
                sys.stdout = open(opts.logprefix+'%02d.log'%rank, 'w') #logfile
                print 'client', rank, 'running on', procname                
            else:
                start = time.time()
                jobnum, job = msg
                print jobnum
                result = opts.dojob(job)#do the job
                end = time.time()
                print 'working',msg[0], end-start

                start = time.time()
                pypar.send((rank, (jobnum, result)), 0, tag=RETURN)#return the result to the server
                end = time.time()
                print 'sending', end-start

    #------------------#
    #end of parallel code
    pypar.barrier()
    pypar.finalize()