How to exchange variables between processes in Python in real time_ish. Is the only way to use queues and pipes - Stack Overflow

admin管理员组

文章数量:1279122

from multiprocessing import Process, Manager, Lock, Value
import time
    
    class Niao:
        def __init__(self, axle_angle, position_x, position_y, energy, speed):
            self.axle_angle = axle_angle
            self.position_x = position_x
            self.position_y = position_y
            self.energy = energy
            self.speed = speed 
            
        def move(self, new_x, new_y):
            self.position_x.value = new_x
            self.position_y.value = new_y
    
    class SharedWorld:
        def __init__(self, manager):
            self.lock = Lock()
            self.niao_dict = manager.dict()  # Shared dictionary for Niao objects
    
        def add_niao(self, key, niao):
            with self.lock:
                self.niao_dict[key] = {
                    'axle_angle': niao.axle_angle.value,  # Store the value directly
                    'position_x': niao.position_x.value,  # Store the value directly
                    'position_y': niao.position_y.value,  # Store the value directly
                    'energy': niao.energy.value,          # Store the value directly
                    'speed': niao.speed.value              # Store the value directly
                }
    
        def get_niao(self, key):
            with self.lock:
                niao_data = self.niao_dict[key]
                return Niao(
                    Value('d',niao_data['axle_angle']),  # Create new Value object
                    Value('d', niao_data['position_x']),  # Create new Value object
                    Value('d', niao_data['position_y']),  # Create new Value object
                    Value('d', niao_data['energy']),      # Create new Value object
                    Value('d', niao_data['speed'])        # Create new Value object
                )
           
        def move_niao(self, key, new_x, new_y):
            with self.lock:
                self.niao_dict[key]['position_x'] = new_x  # Update the value directly
                self.niao_dict[key]['position_y'] = new_y  # Update the value directly
    
    def niao_worker(shared_world, key):
        while True:
            with shared_world.lock:  # Lock access to shared data
                niao_data = shared_world.niao_dict[key]
                print(f"Niao Worker accessing: Niao Position ({niao_data['position_x']}, 
                      {niao_data['position_y']})")
                pos_x = niao_data['position_x']
                pos_y = niao_data['position_y']
            
            # Move Niao object
            shared_world.move_niao(key, pos_x + 5.0, pos_y + 6.0)
    
            with shared_world.lock:  # Lock access to shared data
                niao_data = shared_world.niao_dict[key]
                print(f"Niao Worker accessing post update: Niao Position ({niao_data['position_x']}, {niao_data['position_y']})")
            
            time.sleep(1)  # Delay for 1 second
    
    def worker(shared_world, key):
        while True:        
            niao = shared_world.get_niao('niao_0')
            print(f"Worker accessing: Position ({niao.position_x.value}, {niao.position_y.value})")
            # Delay to reduce the loop's speed
            time.sleep(1)  # Delay for 1 second (adjust as needed)
    
    if __name__ == "__main__":
        manager = Manager()
        shared_world = SharedWorld(manager)
    
        # Add Niao objects to the shared world
        shared_world.add_niao('niao_0', Niao(
            Value('d', 0.0),  # niao_axle_angle
            Value('d', 0.0),  # niao_position_x
            Value('d', 0.0),  # niao_position_y
            Value('d', 0.0),  # niao_energy
            Value('d', 0.0)   # niao_speed
        ))
    
        # Create and start Niao processes
        niao_processes = []
        for key in ['niao_0']:
            p = Process(target=niao_worker, args=(shared_world, key))
            niao_processes.append(p)
            p.start()
    
        # Create and start Food processes
        food_processes = []
        for key in ['food_1']:
            p = Process(target=worker, args=(shared_world, key))
            food_processes.append(p)
            p.start()
    
        # Wait for all processes to finish
        for p in niao_processes + food_processes:
            p.join()

This code is the culmination of multiple attempts to share variables between processes in Python. I think I have understood that the dictionaries return shadow copies of the data and that is why the data is not being updated. I am attempting to run multiple niao objects in real time at 25 fps but all the objects need to know the position of each other and much more. I have limited programming experience but being able to exchange real data between processes is such an obvious thing to do that I struggle to understand why it is so difficult. I do not want to use queues and pipes. I have looked at using Cython to improve access speed but concluded this only added to the complexity and it should be possible to do this with Python. Can anyone tell me that exchanging parameters this way is possible.

from multiprocessing import Process, Manager, Lock, Value
import time
    
    class Niao:
        def __init__(self, axle_angle, position_x, position_y, energy, speed):
            self.axle_angle = axle_angle
            self.position_x = position_x
            self.position_y = position_y
            self.energy = energy
            self.speed = speed 
            
        def move(self, new_x, new_y):
            self.position_x.value = new_x
            self.position_y.value = new_y
    
    class SharedWorld:
        def __init__(self, manager):
            self.lock = Lock()
            self.niao_dict = manager.dict()  # Shared dictionary for Niao objects
    
        def add_niao(self, key, niao):
            with self.lock:
                self.niao_dict[key] = {
                    'axle_angle': niao.axle_angle.value,  # Store the value directly
                    'position_x': niao.position_x.value,  # Store the value directly
                    'position_y': niao.position_y.value,  # Store the value directly
                    'energy': niao.energy.value,          # Store the value directly
                    'speed': niao.speed.value              # Store the value directly
                }
    
        def get_niao(self, key):
            with self.lock:
                niao_data = self.niao_dict[key]
                return Niao(
                    Value('d',niao_data['axle_angle']),  # Create new Value object
                    Value('d', niao_data['position_x']),  # Create new Value object
                    Value('d', niao_data['position_y']),  # Create new Value object
                    Value('d', niao_data['energy']),      # Create new Value object
                    Value('d', niao_data['speed'])        # Create new Value object
                )
           
        def move_niao(self, key, new_x, new_y):
            with self.lock:
                self.niao_dict[key]['position_x'] = new_x  # Update the value directly
                self.niao_dict[key]['position_y'] = new_y  # Update the value directly
    
    def niao_worker(shared_world, key):
        while True:
            with shared_world.lock:  # Lock access to shared data
                niao_data = shared_world.niao_dict[key]
                print(f"Niao Worker accessing: Niao Position ({niao_data['position_x']}, 
                      {niao_data['position_y']})")
                pos_x = niao_data['position_x']
                pos_y = niao_data['position_y']
            
            # Move Niao object
            shared_world.move_niao(key, pos_x + 5.0, pos_y + 6.0)
    
            with shared_world.lock:  # Lock access to shared data
                niao_data = shared_world.niao_dict[key]
                print(f"Niao Worker accessing post update: Niao Position ({niao_data['position_x']}, {niao_data['position_y']})")
            
            time.sleep(1)  # Delay for 1 second
    
    def worker(shared_world, key):
        while True:        
            niao = shared_world.get_niao('niao_0')
            print(f"Worker accessing: Position ({niao.position_x.value}, {niao.position_y.value})")
            # Delay to reduce the loop's speed
            time.sleep(1)  # Delay for 1 second (adjust as needed)
    
    if __name__ == "__main__":
        manager = Manager()
        shared_world = SharedWorld(manager)
    
        # Add Niao objects to the shared world
        shared_world.add_niao('niao_0', Niao(
            Value('d', 0.0),  # niao_axle_angle
            Value('d', 0.0),  # niao_position_x
            Value('d', 0.0),  # niao_position_y
            Value('d', 0.0),  # niao_energy
            Value('d', 0.0)   # niao_speed
        ))
    
        # Create and start Niao processes
        niao_processes = []
        for key in ['niao_0']:
            p = Process(target=niao_worker, args=(shared_world, key))
            niao_processes.append(p)
            p.start()
    
        # Create and start Food processes
        food_processes = []
        for key in ['food_1']:
            p = Process(target=worker, args=(shared_world, key))
            food_processes.append(p)
            p.start()
    
        # Wait for all processes to finish
        for p in niao_processes + food_processes:
            p.join()

This code is the culmination of multiple attempts to share variables between processes in Python. I think I have understood that the dictionaries return shadow copies of the data and that is why the data is not being updated. I am attempting to run multiple niao objects in real time at 25 fps but all the objects need to know the position of each other and much more. I have limited programming experience but being able to exchange real data between processes is such an obvious thing to do that I struggle to understand why it is so difficult. I do not want to use queues and pipes. I have looked at using Cython to improve access speed but concluded this only added to the complexity and it should be possible to do this with Python. Can anyone tell me that exchanging parameters this way is possible.

• python
• real-time
• shared-data

Share Improve this question Follow asked Feb 24 at 10:47 Peter MasonPeter Mason 133 bronze badges 5

• 1 This sounds like a job for just one process. You don't need separate processes for every actor in a simulation. – user2357112 Commented Feb 24 at 10:52
• The problem as you've described it is a hard problem in any language, and frequently involves some sort of messaging layer (like RabbitMQ) or persistence layer (a database, like PostgreSQL) (both come with maintenance and performance overheads as well). A "process" intrinsically isolates one part's memory from others, and it might not be the abstraction you're looking for. – David Maze Commented Feb 24 at 11:52
• Thank you. That would take me in me in an unexpected direction. However using one process would limit the number of participants. 100 ms of simulation takes about 20 ms and with the additional work to accommodate the interaction this would severely limit the number of participants besides what do I do with the other cpu cores? – Peter Mason Commented Feb 24 at 12:00
• To David Maze, Thank you. As I said my programming experience is limited and it is nice to know it is difficult and not only because of my knowledge base. P.S. How do the gaming people solve this? – Peter Mason Commented Feb 24 at 12:04
• What is "niao" please? – Mark Setchell Commented Feb 24 at 18:29

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

To those that replied Thank you and to any that follow, I found this:- It works in this test code and it remains to be seen if it works in my application. It is very fast too, it has gone from 100's milliseconds to a few microseconds.

from multiprocessing import Process, Lock, Value, shared_memory
import time
import numpy as np
import signal
import os

class Niao:
    def __init__(self, axle_angle, position_x, position_y, energy, speed):
        self.axle_angle = axle_angle
        self.position_x = position_x
        self.position_y = position_y
        self.energy = energy
        self.speed = speed 
        
    def move(self, new_x, new_y):
        self.position_x.value = new_x
        self.position_y.value = new_y

class SharedWorld:
    def __init__(self):
        self.lock = Lock()
        self.niao_shm = None  # Shared memory for Niao attributes

    def create_niao_shared_memory(self):
        # Create shared memory for Niao attributes
        self.niao_shm = shared_memory.SharedMemory(create=True, size=5 * 8)  # 5 double values (8 bytes each)
        return self.niao_shm

    def add_niao(self, niao):
        with self.lock:
            # Store values in shared memory
            shm_array = np.ndarray((5,), dtype='d', buffer=self.niao_shm.buf)
            shm_array[0] = niao.axle_angle.value
            shm_array[1] = niao.position_x.value
            shm_array[2] = niao.position_y.value
            shm_array[3] = niao.energy.value
            shm_array[4] = niao.speed.value

    def get_niao(self):
        with self.lock:
            shm_array = np.ndarray((5,), dtype='d', buffer=self.niao_shm.buf)
            return Niao(
                Value('d', shm_array[0]),  # Create new Value object
                Value('d', shm_array[1]),  # Create new Value object
                Value('d', shm_array[2]),  # Create new Value object
                Value('d', shm_array[3]),  # Create new Value object
                Value('d', shm_array[4])   # Create new Value object
            )
       
    def move_niao(self, new_x, new_y):
        with self.lock:
            shm_array = np.ndarray((5,), dtype='d', buffer=self.niao_shm.buf)
            shm_array[1] = new_x  # Update position_x
            shm_array[2] = new_y  # Update position_y

def niao_worker(shared_world):
    while True:
        with shared_world.lock:  # Lock access to shared data
            shm_array = np.ndarray((5,), dtype='d', buffer=shared_world.niao_shm.buf)
            print(f"Niao Worker accessing: Niao Position ({shm_array[1]}, {shm_array[2]})")
            pos_x = shm_array[1]
            pos_y = shm_array[2]
        
        # Move Niao object
        shared_world.move_niao(pos_x + 5.0, pos_y + 6.0)

        start_time = time.time()  # Record the start time
        with shared_world.lock:  # Lock access to shared data
            shm_array = np.ndarray((5,), dtype='d', buffer=shared_world.niao_shm.buf)

        end_time = time.time()  # Record the end time
        duration_microseconds = (end_time - start_time) * 1_000_000  # Convert to microseconds
        print(f"Niao_worker access shm_array {duration_microseconds:.2f} microseconds.")            
        print(f"Niao Worker accessing post update: Niao Position ({shm_array[1]}, {shm_array[2]})")
        
        time.sleep(1)  # Delay for 1 second

def worker(shared_world):
    while True:        
        niao = shared_world.get_niao()
        print(f"Worker accessing: Position ({niao.position_x.value}, {niao.position_y.value})")
        # Delay to reduce the loop's speed
        time.sleep(1)  # Delay for 1 second (adjust as needed)

def signal_handler(sig, frame):
    print("Terminating processes...")
    os._exit(0)

if __name__ == "__main__":
    signal.signal(signal.SIGINT, signal_handler)  # Handle Ctrl+C gracefully

    shared_world = SharedWorld()
    shared_world.create_niao_shared_memory()

    # Add Niao object to the shared world
    shared_world.add_niao(Niao(
        Value('d', 0.0),  # niao_axle_angle
        Value('d', 0.0),  # niao_position_x
        Value('d', 0.0),  # niao_position_y
        Value('d', 0.0),  # niao_energy
        Value('d', 0.0)   # niao_speed
    ))

    # Create and start Niao process
    niao_process = Process(target=niao_worker, args=(shared_world,))
    niao_process.start()

    # Create and start Worker process
    worker_process = Process(target=worker, args=(shared_world,))
    worker_process.start()

    # Wait for processes to finish (they run indefinitely)
    niao_process.join()
    worker_process.join()

    # Cleanup shared memory
    shared_world.niao_shm.close()
    shared_world.niao_shm.unlink()

本文标签：