Connect multiple Python processes to one JVM?

[Ján Petrák]

Hello,

is it possible to use Py4J to connect multiple python processes to a single running JVM?

I.e., each python process would be using a unique port number, whereas the java side would contain a single GatewayServer using, e.g., the default port?

My use case: The python side keeps creating (and then deleting) processes running a certain task. Each task needs to run something in java and progressively receive some data from it (via proxy callback).

Thanks in advance.

Best Regards,
Jan

[Johann Petrak]

My understanding would have been that for this there also have to be the same number of GatewayServer instances on the Java side

as port numbers must match between python and java.

Which should not be a problem if each GatewayServer runs in its own Process.

I would assume the best way to do this is have one process be responsible for spawning additional GateServers, so Python would

ask on the spawning port to get a new server on port nnnn then connect to that server etc.

I have never tried this though so no idea if there is anything in GatewayServer that would prevent it from running several instances in parallel.

Cheers,

  Johann

[Jonah Graham]

The GatewayServer can run multiple instances. The Eclipse EASE project uses it that way.

Jonah

~~~
Jonah Graham
Kichwa Coders
www.kichwacoders.com

--
You received this message because you are subscribed to the Google Groups "Py4J Support and Comments" group.
To unsubscribe from this group and stop receiving emails from it, send an email to py4j+uns...@py4j.org.
To view this discussion on the web visit https://groups.google.com/a/py4j.org/d/msgid/py4j/157e80bf-6252-41ab-a332-b827e42ea492n%40py4j.org.

[Barthelemy Dagenais]

Hi!

> is it possible to use Py4J to connect multiple python processes to a single running JVM?

Yes it is possible. To test this, start a GatewayServer in Java. Then open two shells in Python, create a gateway and call something on the Java side, e.g.,

from py4j.java_gateway import JavaGateway

gateway = JavaGateway()

gateway.jvm.System.out.println("Hello World from shell 1")

> My use case: The python side keeps creating (and then deleting) processes running a certain task. Each task needs to run something in java and progressively receive some data from it (via proxy callback).

This is different then.

If you want multiple python processes to access Java and receive callbacks **concurrently**, you need to create multiple GatewayServer instances because one GatewayServer instance is tied to a specific callback port.

If only one python process accesses the Java side and receives callbacks at a time, you can reuse the same GatewayServer instance. Just call GatewayServer.resetCallbackClient() to pass the new callback port if it changes.

https://www.py4j.org/_static/javadoc/py4j/GatewayServer.html#resetCallbackClient-java.net.InetAddress-int-

To view this discussion on the web visit https://groups.google.com/a/py4j.org/d/msgid/py4j/CAPmGMvh-C2hKbG13b-C_w6c_iLmF62AZ7b9bU4s4FmDZrAKieQ%40mail.gmail.com.

[Ján Petrák]

Hello, thank you, that is precisely what I needed to know!

I need to receive callbacks concurrently. However, in a minimal example, I am running into a problem -- when I initialize two java side instances of a python class, the first one through 25333 <-> 25334 gateway, the second one through 25335 <-> 25336 gateway, and then call the first instance's method, it seems like there is some issue. Specifically:

Exception in thread "Thread-3" py4j.Py4JException: Cannot obtain a new communication channel

at py4j.CallbackClient.sendCommand(CallbackClient.java:380)

at py4j.CallbackClient.sendCommand(CallbackClient.java:356)

at py4j.reflection.PythonProxyHandler.invoke(PythonProxyHandler.java:106)

at com.sun.proxy.$Proxy0.sendData(Unknown Source)

at Main.lambda$start$0(Main.java:40)

at java.base/java.lang.Thread.run(Thread.java:832)

These are the codes used, the interface Sender has only one method void sendData(String data).

Python:

from py4j.java_gateway import JavaGateway, CallbackServerParameters, GatewayParameters

from py4j.protocol import Py4JNetworkError, Py4JError

class C:

    def sendData(self, data):

        print("Received: " + data)

    class Java:

        implements = ["Sender"]

try:

    gateway = JavaGateway(callback_server_parameters=CallbackServerParameters())

    gateway.entry_point.start("abc")

    gateway.entry_point.initSender(C())

except (Py4JNetworkError, Py4JError) as e:

    print("25333 occupied, will connect to 25335")

    gateway = JavaGateway(gateway_parameters = GatewayParameters(port=25335),

                          callback_server_parameters=CallbackServerParameters(port=25336))

    gateway.entry_point.initSndSender(C())

Java:

import py4j.GatewayServer;

public class Main {

    private Thread myThread;

    private Sender mySender;

    private Sender mySender2;

    private int x;

    public int getX() {

        return x;

    }

    public void initSender(Sender fst) {

        mySender = fst;

    }

    public void initSndSender(Sender snd) {

        mySender2 = snd;

    }

    public void start(String str) {

        System.out.println("Starting with " + str);

        new Thread(() -> {

            myThread = Thread.currentThread();

            while (true) {

                try {

                    x++;

                    Thread.sleep(1000);

                    System.out.println(x);

                    if (mySender != null && x % 10 == 6) {

                        mySender.sendData("Sending x=" + x + " to the first");

                    }

                    if (mySender2 != null && x % 10 == 7) {

                        mySender2.sendData("Sending x=" + x + " to the second");

                    }

                } catch (InterruptedException e) {

                    System.out.println("Interrupted and ending with x: " + x);

                    break;

                }

            }

        }).start();

    }

    public void stop(String str) {

        myThread.interrupt();

    }

    public static void main(String[] args) {

        Main app = new Main();

        GatewayServer server = new GatewayServer(app);

        GatewayServer server2 = new GatewayServer(app, 25335);

        server.start();

        server2.start();

        System.out.println("Server running");

    }

}

I've also tried calling .resetCallbackClient() on server2 right after the initialization, but it did not help.

Would you be able to tell where I'm making a mistake?
Thank you a lot!

[Barthelemy Dagenais]

Hi,

Did you try running your program with (1) only one gateway server and callback server on default ports and then (2) only one gateway server and callback server on custom ports?

Btw, the second constructor call, GatewayServer server2 = new GatewayServer(app, 25335);, binds the GatewayServer to the port 25335 but does set the Python port to 25336

[Ján Petrák]

Hello,

(1) Yes, and it works correctly.

(2) I just did that and it did not work. Then, I've changed the server2 initialization to: `GatewayServer server2 = new GatewayServer(app, 25335, 25336, InetAddress.getLoopbackAddress(), InetAddress.getLoopbackAddress(), 0, 0, null);` and now it works.

However, when I try to run both, I get the same error. Specifically I:

1. run the java code with both GatewayServers initialized;

2. run the first terminal with the python code, connecting via default ports; via these ports I run the method entry_point.start(), creating a new thread increasing "x", and I am successfully receiving messages on terminal via callback, when x % 10 == 6;

3. run the second terminal with the python code, connecting via custom ports; the thread increasing "x" keeps running until it is supposed to sendData via default ports callback (i.e., when x % 10 == 6 again), throwing an exception posted above --  saying the sendData source is unknown.

It is as if setting up the callback for the `server2` somehow messed up the callback for the `server`.

[Ján Petrák]

Found out, the problem is with the try/except block.

When I do something like "if first terminal, then run via default ports, if second terminal, use custom ports", it all works.

However, when I use the approach "if connect using default ports throws an exception, use custom ports", I receive the aforementioned exception "unknown source".

Seems like attempting to connect to the ports that are already being used causes some sort of callback reset(?).

[Ján Petrák]

I turned on logging, and the moment I attempt to connect from the second python process using the ports the first process uses, the first process terminal window shows: Closing down callback connection

And the terminal of the second process shows:

An error occurred while trying to start the callback server (127.0.0.1:25334)

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2207, in start

    self.server_socket.bind((self.address, self.port))

OSError: [Errno 98] Address already in use

Exception while shutting down a socket

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2207, in start

    self.server_socket.bind((self.address, self.port))

OSError: [Errno 98] Address already in use

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 1894, in start_callback_server

    self._callback_server.start()

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2216, in start

    raise Py4JNetworkError(msg, e)

py4j.protocol.Py4JNetworkError: An error occurred while trying to start the callback server (127.0.0.1:25334)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 498, in quiet_shutdown

    socket_instance.shutdown(socket.SHUT_RDWR)

OSError: [Errno 9] Bad file descriptor

Callback Server Shutting Down

Exception while shutting down a socket

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2207, in start

    self.server_socket.bind((self.address, self.port))

OSError: [Errno 98] Address already in use

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 1894, in start_callback_server

    self._callback_server.start()

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2216, in start

    raise Py4JNetworkError(msg, e)

py4j.protocol.Py4JNetworkError: An error occurred while trying to start the callback server (127.0.0.1:25334)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 498, in quiet_shutdown

    socket_instance.shutdown(socket.SHUT_RDWR)

OSError: [Errno 107] Transport endpoint is not connected

Exception while shutting down callback server

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2207, in start

    self.server_socket.bind((self.address, self.port))

OSError: [Errno 98] Address already in use

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 1894, in start_callback_server

    self._callback_server.start()

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2216, in start

    raise Py4JNetworkError(msg, e)

py4j.protocol.Py4JNetworkError: An error occurred while trying to start the callback server (127.0.0.1:25334)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 1987, in shutdown_callback_server

    self._callback_server.shutdown()

  File "/home/jan/.local/lib/python3.8/site-packages/py4j/java_gateway.py", line 2340, in shutdown

    self.thread.join()

AttributeError: 'CallbackServer' object has no attribute 'thread'

Right after these messages the terminal continues showing output from the "except" of my python script:

25333 occupied, will connect to 25335

Callback Server Starting

Command to send: c

t

initSndSender

fp0;Sender

e

Socket listening on ('127.0.0.1', 25336)

Answer received: !yv

Is that the intended behavior?