What's wrong in this multiprocessing python script?

Question

I am writting a simple messaging program (for 2 users) that works on the terminal.

For its implementation, I decided to create 2 processes, one for the server (waits for the message to arrive from the other user), and another one for the client (just sends the message to the server process of the other user)

The fact is that when I run it, I get the following error:

C:\>python message.py

> Process Process-2:
Traceback (most recent call last):
File "C:\Python27\lib\multiprocessing\process.py", line 258, in_bootstrap
self.run()
File "C:\Python27\lib\multiprocessing\process.py", line 114, in run
self._target(*self._args, **self._kwargs)
File "C:\message.py", line 12, in send_messages
message = raw_input('> ')
EOFError

Process Process-1:
Traceback (most recent call last):
File "C:\Python27\lib\multiprocessing\process.py", line 258, in 
_bootstrap
self.run()
File "C:\Python27\lib\multiprocessing\process.py", line 114, in run
self._target(*self._args, **self._kwargs)
File "C:\message.py", line 25, in receive_messages
message = sc.recv(1024)
error: [Errno 10054] An existing connection was forcibly closed by the 
remote host

This is my Python code

from multiprocessing import Process
import socket

direction = "localhost"

global s
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
def send_messages():
   s.connect((direction, 5124))
   while True:
       message = raw_input('> ')
       s.send(message)
       if message == 'close':
           break
   print 'Bye'
   s.close()

def receive_messages():
   s.bind(("localhost",5124))
   s.listen(2)
   sc, addr = s.accept()

   while True:
       message = sc.recv(1024)
       print message
   sc.close()
   s.close()

if __name__ == '__main__':

   p1 = Process(target = receive_messages)
   p1.start()
   p2 = Process(target = send_messages)
   p2.start()
   p1.join()
   p2.join()

NOTE 1: There can be some indentation errors due to cut and paste from my text editor to stackoverflow.

NOTE 2: I am working on Windows 10

score 2 · Accepted Answer · edited Jun 20 '20 at 09:12

From the documentation:

Beware of replacing sys.stdin with a “file like object”

multiprocessing originally unconditionally called:

os.close(sys.stdin.fileno())

in the multiprocessing.Process._bootstrap() method — this resulted in issues with processes-in-processes. This has been changed to:

sys.stdin.close()
sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)

Which solves the fundamental issue of processes colliding with each other resulting in a bad file descriptor error, but introduces a potential danger to applications which replace sys.stdin() with a “file-like object” with output buffering. This danger is that if multiple processes call close() on this file-like object, it could result in the same data being flushed to the object multiple times, resulting in corruption.

The bottom line is that your Processes are closing stdin unless you subclass them and avoid doing so.

You might consider using one (capital-P) Process to handle communications, and then doing the input/output in your original (lowercase) process:

if __name__ == '__main__':

   p1 = Process(target = receive_messages)
   p1.start()

   send_messages()

   p1.join()

hxysayhi · Answer 2 · 2017-04-14T07:14:26.140

See this question.

When you spawn a thread in Python, it closes stdin. You can't use a subprocess to collect standard input. Use the main thread to collect input instead and post them to the Queue from the main thread. It may be possible to pass the stdin to another thread, but you likely need to close it in your main thread.

Maybe you can work around this by using fdopen() to reopen stdin in the subprocess. See this answer.

Here is an example from your code:

from multiprocessing import Process
import socket
import sys
import os

direction = "localhost"

# I got this error
# error: [Errno 106] Transport endpoint is already connected
# when I run your code ,so I made some change
# You can ignore it if everything is ok when your code run

global s1
s1 = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

global s2
s2 = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

def send_messages(fileno):
    s1.connect((direction, 5104))
    sys.stdin = os.fdopen(fileno)  # open stdin in this process
    while True:
        message = ''
        message = raw_input('> ')
        s1.send(message)
        if message == 'close':
            break

    print 'Bye'
    s1.close()

def receive_messages():
    s2.bind(("localhost",5104))
    s2.listen(2)
    sc, addr = s2.accept()

    while True:
        message = sc.recv(1024)
        if message == 'close':
            print 'Bye!'
            break
        print message

    sc.close()
    s2.close()

if __name__ == '__main__':
    fn = sys.stdin.fileno()  # get original file descriptor
    p1 = Process(target = receive_messages)
    p1.start()
    p2 = Process(target = send_messages, args=(fn,))
    p2.start()
    p1.join()
    p2.join()

I tested it,and it worked.

score 0 · Answer 3 · answered Apr 14 '17 at 03:11

The error you are receiving basically means your raw_input is receiving an empty input. That condition raises the EOFError you can read about it in the built-in-exceptions section of the documentation.

I've never tried something like this with multiprocessing before but i would imagine that is where your problem lies. Perhaps make sure your logic is working as expected in a single process before moving to multiprocess, but i still feel like trying to start multiple processes to receive user input is gonna be a headache.

What's wrong in this multiprocessing python script?

3 Answers3