Python многопроцессорная разделяемая память; одна запись, многократное чтение - PullRequest
Новая
       154

Python многопроцессорная разделяемая память; одна запись, многократное чтение

0 голосов
/ 07 августа 2020

СИСТЕМА

  • Linux (Manjaro KDE)
  • Python 3.8.3

ПРОГРАММА: У меня есть входящие строковые данные на UDP-порт. Основной l oop буферизует процессы перед использованием селекторов для мониторинга порта UDP. Я хочу, чтобы данные UDP, которые постоянно обновляются, были доступны для каждого процесса.

TRIED:

  • Многопроцессорные очереди с maxsize = 1, и это стало головной болью и быстро сломался.
  • Многопроцессорные массивы (вот где я сейчас)

Я проверил, и массив в каждом месте, на которое я смотрю, имеет то же самое адрес памяти (думаю). По какой-то причине, когда я пытаюсь получить доступ к содержимому массива в дочернем процессе, процесс зависает.

НЕ ПРОВЕРЕНО

  • Pipes. У меня такое чувство, что это может быть путь к go. Но я уже глубоко на неизведанной территории; Я никогда раньше ими не пользовался.

ЧТО Я ХОЧУ Я хотел бы получить доступ к данным UDP из дочерних процессов - это метод camera_view.

Фиктивная строка UDP 

import socket
import random
import datetime
import time

conn = ('127.0.0.1', 6666)

def rand_value(f_val, t_val):
    result = round(random.uniform(f_val, t_val), 2)  
    result = random.uniform(f_val, t_val)
    return result

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

while True:

    time.sleep(6)
    timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    overlay = timestamp

    for i in range(9):
        val = rand_value(i*10, i*10+10)
        if i == 8: val = 'TASK: Im the real Batman'
        overlay = overlay + "," + str(val)
    
    print(overlay)
    sock.sendto(overlay.encode(), conn)

Моя программа 

import datetime
import selectors
import socket
import time
from multiprocessing import Lock, Process, Queue
from multiprocessing.sharedctypes import Array
from ctypes import c_char_p


REQUIRED_CAMERAS = 1
CAMERA_CONN = {'name':['Colour Camera'], 'ip':['127.0.0.1'], 'port':[9000]}
OVERLAY_CONN = ('0.0.0.0', 6666)
CONTROL_CONN = ('0.0.0.0', 6667)
NUMBER_OF_ITEMS_IN_OVERLAY = 10

class Camera():
    def __init__(self, cam_name, cam_ip, cam_port):
        self.ip = cam_ip
        self.port = cam_port
        self.capture = cv2.VideoCapture(0)
        self.frame_width = int(self.capture.get(3))
        self.frame_height = int(self.capture.get(4))
        self.name = cam_name


def get_overlay(data_packet):
        data = data_packet.decode()
        data = data.split(',')
        field0 = data[0]
        field1 = 'KP: ' + str(round(float(data[1]), 3))
        field2 = 'DCC: ' + str(round(float(data[2]), 2)) + 'm'
        field3 = 'E: ' + str(round(float(data[3]), 2)) + 'm'
        field4 = 'N: ' + str(round(float(data[4]), 2)) + 'm'
        field5 = 'D: ' + str(round(float(data[5]), 2)) + 'm'
        field6 = 'H: ' + str(round(float(data[6]), 2)) # + '°'
        field7 = 'R: ' + str(round(float(data[7]), 2)) # + '°'
        field8 = 'P: ' + str(round(float(data[8]), 2)) # + '°' 
        field9 = data[9]

        x = []
        for i in range(NUMBER_OF_ITEMS_IN_OVERLAY):
            x.append(eval('field' + str(i)).encode())
            # if i == 0:
            #     print(x[i])
                
        return x

def socket_reader(sock, mask, q, REQUIRED_CAMERAS, overlay):
    data_packet, sensor_ip = sock.recvfrom(1024)
    sensor_port = sock.getsockname()[1]
    print(f'SENSOR PORT {sensor_port} and SENSOR_IP {sensor_ip}')

    if sensor_port == OVERLAY_CONN[1]:
        x = get_overlay(data_packet)
        for i in range(len(x)):
            overlay[i] = x[i]
            print(f'Socket Reader {overlay}')

def camera_view(CAMERA_CONN, cam_name, camera, overlay_q, control_q, overlay):
    while True:
        print(f'PROCESS {camera} RUNNING FOR: {cam_name}')
        try:
            print(f'Camera View {overlay}')
            for i in range(len(overlay)):
                print(overlay[i])
        except:
            pass
        time.sleep(1)
        

def controller(REQUIRED_CAMERAS, CAMERA_CONN, OVERLAY_CONN, CONTROL_CONN):
    
    if REQUIRED_CAMERAS > len(CAMERA_CONN['name']):
        print(f'REQURIED_CAMERAS: {REQUIRED_CAMERAS} - more than connections in CAMERA_CONN ')
    else:
        # Set up a UDP connection for the overlay string and the control commands
        sock_overlay = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock_control = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock_overlay.bind(OVERLAY_CONN)
        sock_control.bind(CONTROL_CONN)
        
        # Set up the selector to watch over the socket
        # and trigger when data is ready for reading
        sel = selectors.DefaultSelector()
        sel.register(fileobj=sock_overlay, events=selectors.EVENT_READ, data=socket_reader)
        sel.register(fileobj=sock_control, events=selectors.EVENT_READ, data=socket_reader)
        
        # create shared memory
        overlay_q = Queue(maxsize=1)
        control_q = Queue(maxsize=1)     
        overlay = Array(c_char_p, range(NUMBER_OF_ITEMS_IN_OVERLAY))
        print(f'Init Overlay {overlay}')
        
        # Generate the processes; one per camera
        processes = []
        
        for camera in range(REQUIRED_CAMERAS):
            processes.append(Process(target=camera_view, args=(CAMERA_CONN, CAMERA_CONN['name'][camera], camera, overlay_q, control_q, overlay)))

        for process in processes:
            process.daemon = True
            process.start()
            
        # Spin over the selector
        while True:

            # Only have one connnection registered, so to stop
            # the loop spinning up the CPU, I have made it blocking 
            # with the timeout = 1 (sec) instead of =0.
            events = sel.select(timeout=None)

            for key, mask in events:
                # the selector callback is the data= from the register above
                callback = key.data
                # the callback gets the sock, mask and the sensor queues
                if key.fileobj == sock_overlay:
                    callback(key.fileobj, mask, overlay_q, REQUIRED_CAMERAS, overlay)
                else:
                    callback(key.fileobj, mask, control_q, REQUIRED_CAMERAS, overlay)


if __name__ == "__main__":
    
    controller(REQUIRED_CAMERAS, CAMERA_CONN, OVERLAY_CONN, CONTROL_CONN)

EDIT1: 

from multiprocessing import Process, Array
from ctypes import c_char_p
import time

def worker(arr):
    count = 0
    while True:
        count += 1
        val = 'val' + str(count)
        arr[0] = val
        print(arr[:])
        time.sleep(2)

def main():
    arr = Array(c_char_p, 1)
    p = Process(target=worker, args=(arr,))
    p.daemon = True
    p.start()
    
    while True:
        print(arr[:])
        try:
            print(arr[:].decode('utf-8'))
        except :
            pass
        # try:
        #     val = arr[:]
        #     val = val.decode('utf-8')
        #     print(f'main {val}')
        # except:
        #     pass
        time.sleep(1)

if __name__ == "__main__":
    main()
    
    
'''
from multiprocessing import Process, Array
from ctypes import c_char_p
import time

def worker(arr):
    count = 0
    while True:
        count += 1
        val = 'val' + str(count)
        arr[0] = bytes(val, 'utf-8')
        print(arr[:])
        time.sleep(2)

def main():
    arr = Array(c_char_p, 1)
    p = Process(target=worker, args=(arr,))
    p.daemon = True
    p.start()
    
    while True:
        print(arr[:])
        try:
            print(arr[:].decode('utf-8'))
        except :
            pass

        time.sleep(1)

if __name__ == "__main__":
    main()

if __name__ == "__main__":
    main()
'''

EDIT2: Благодаря @RolandSmith, я выстоял с очередями, и я думаю, что у меня есть шаблон о том, как я могу двигаться вперед. Смотрите код ниже. Если мне не удастся заставить это работать в программе, я вернусь сюда. 

from multiprocessing import Process, Queue
import time
import datetime

def worker(camera, q):
    val = ''
    while True:    
        if q.full() == True:
            val = q.get()
        else:
            val = val
        print(f'WORKER{camera} {val}')
        time.sleep(0.2)

def main():
    
    cameras = 2
    
    processes = []
    queues = []
    
    for camera in range(cameras):
        queues.append(Queue(maxsize=1))
        processes.append(Process(target=worker, args=(camera, queues[camera])))
        
    for process in processes:                     
        process.daemon = True
        process.start()

    while True:
        for q in queues:
            if not q.empty():
                try:
                    _ = q.get()
                except:
                    pass
            else:
                q.put(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        time.sleep(.5)

if __name__ == "__main__":
    main()

1 Ответ

1 голос
/ 07 августа 2020

На мой взгляд, использование Queue является менее подверженным ошибкам решением, чем использование Array.

Вот ваш второй пример, преобразованный в использование Queue: 

from multiprocessing import Process, Queue
import time


def worker(q):
    count = 0 
    while True:
        count += 1
        val = 'val' + str(count)
        q.put(val)
        print('worker:', val)
        time.sleep(2)


def main():
    q = Queue()

    p = Process(target=worker, args=(q, ))
    p.daemon = True
    p.start()

    while True:
        if not q.empty():
            print('main:', q.get())
        time.sleep(1)


if __name__ == "__main__":
    main()

Это дает: 

> python3 test3.py
worker: val1
main: val1
worker: val2
main: val2
worker: val3
main: val3
worker: val4
main: val4
worker: val5

Вот тот же пример с использованием Pipe: 

from multiprocessing import Process, Pipe
import time


def worker(p):
    count = 0 
    while True:
        count += 1
        val = 'val' + str(count)
        p.send(val)
        print('worker:', val)
        time.sleep(2)


def main():
    child, parent = Pipe()

    p = Process(target=worker, args=(child, ))
    p.daemon = True
    p.start()

    while True:
        if parent.poll():
            print('main:', parent.recv())
        time.sleep(1)


if __name__ == "__main__":
    main()

Результат тот же, что и в предыдущем примере.

Кроме того, по умолчанию канал является двунаправленным. Таким образом, вы также можете отправить данные от рабочих к родительскому.

...