Merge pull request #206 in LCL/wolframclientforpython from feature/better-read-write-performances to master

* commit '2c9366cf91fee5c84650f3bf0ab1368291553906':
  Increment version.
  Update kernel controller and session class __repr__.
  Update Read message in WL initkernel.m
  ExternalEvaluate use Frame.
  Wrapping evaluation data into a Frame
  Update code with recent changes in ZeroMQLink.

Author: Dorian Birraux

PacletInfo.m

@@ -1,6 +1,6 @@
 Paclet[
     Name -> "WolframClientForPython",
-    Version -> "1.1.0",
+    Version -> "1.1.1",
     MathematicaVersion -> "11.3+",
     Loading -> Automatic,
     Extensions -> {}

wolframclient/about.py

@@ -2,6 +2,6 @@
 
 __name__ = "wolframclient"
 __description__ = "A Python library with various tools to interact with the Wolfram Language and the Wolfram Cloud."
-__version__ = "1.1.0"
+__version__ = "1.1.1"
 __author__ = "Wolfram Research"
 __author_email__ = "support@wolfram.com, dorianb@wolfram.com, riccardod@wolfram.com"

wolframclient/evaluation/kernel/initkernel.m

@@ -21,13 +21,24 @@ no matter the total size (tested with 80MB) *)
 
 Begin["`Private`"];
 
-{SocketWriteByteArrayFunc, SocketReadByteArrayFunc} = If[
+(*Define the most efficient pair of write bytearray and non-blocking read, for various WL versions. *)
+{SocketWriteByteArrayFunc, SocketReadByteArrayFuncNoWait} = If[
 	$VersionNumber < 12,
-	iSocketWriteByteArray[socket_,ba_ByteArray] := ZeroMQLink`Private`ZMQWriteInternal[socket, Normal[ba]];
-	iSocketReadByteArray[uuid_String, flags_Integer]:= ByteArray@iRecvSingleMultipartMessageSocket[uuid, flags];
-	{iSocketWriteByteArray, iSocketReadByteArray}
+	{
+		Function[{socketOut, ba}, ZeroMQLink`Private`ZMQWriteInternal[socketOut, Normal[ba]]],
+		Function[{socketIn},
+			Block[
+				{data=ByteArray@iRecvSingleMultipartMessageSocket[First@socketIn, 1(*Flag NOWAIT*)]},
+				If[Length[data] >= 3, Part[data,4;;], {}]
+			]
+		]
+	}
 	,
-	{ZMQSocketWriteMessage, iRecvSingleMultipartBinaryMessageSocket}
+	{
+		ZMQSocketWriteMessage,
+		SocketReadMessage[#, "Blocking"->False] &
+	}
+
 ];
 
 $DEBUG=1;
@@ -169,14 +180,13 @@ no matter the total size (tested with 80MB) *)
 	$VersionNumber < $TaskSupportMinVersion,
 	(* Low CPU wait but need synchronous loop. *)
 	evaluationLoop[socketIn_SocketObject]:= With[
-		{maxPause=$MaxIdlePause, minPause=$MinIdlePause, incr=$PauseIncrement, 
-		uuidIn=First@socketIn, poller={socketIn}},
+		{maxPause=$MaxIdlePause, minPause=$MinIdlePause, incr=$PauseIncrement, poller={socketIn}},
 		Block[{msg},
 			SendAck[];
 			While[True,
-				msg = SocketReadByteArrayFunc[uuidIn, 1 (* NOWAIT *)];
-				If[Length[msg]>3, 
-					socketEventHandler[msg[[4;;]]];
+				msg = SocketReadByteArrayFuncNoWait[socketIn];
+				If[Length[msg]>0,
+					socketEventHandler[msg];
 					,
 					SocketWaitNext[poller];
 				]
@@ -186,14 +196,13 @@ no matter the total size (tested with 80MB) *)
 	True,
 	(* Version with fixed asynchronous tasks *)
 	evaluationLoop[socketIn_SocketObject]:= With[
-		{maxPause=$MaxIdlePause, minPause=$MinIdlePause, incr=$PauseIncrement, 
-		uuidIn=First@socketIn, poller={socketIn}}, 
+		{maxPause=$MaxIdlePause, minPause=$MinIdlePause, incr=$PauseIncrement, poller={socketIn}},
 		$Task = SessionSubmit[
 			ScheduledTask[
 			(
-				msg = SocketReadByteArrayFunc[uuidIn, 1 (* NOWAIT *)];
-				If[Length[msg]>3, 
-					socketEventHandler[msg[[4;;]]];
+				msg = SocketReadByteArrayFuncNoWait[socketIn];
+				If[Length[msg]>0,
+					socketEventHandler[msg];
 					,
 					SocketWaitNext[poller];
 				]

wolframclient/evaluation/kernel/kernelcontroller.py

@@ -498,7 +498,7 @@ def evaluate_future(self, wxf, future, result_update_callback=None, **kwargs):
 
     def _do_evaluate(self, wxf, future, result_update_callback):
         start = time.perf_counter()
-        self.kernel_socket_out.send(wxf)
+        self.kernel_socket_out.send(zmq.Frame(wxf))
         if logger.isEnabledFor(logging.DEBUG):
             logger.debug("Expression sent to kernel in %.06fsec", time.perf_counter() - start)
             start = time.perf_counter()
@@ -592,14 +592,17 @@ def _cancel_tasks(self):
 
     def __repr__(self):
         if self.started:
-            return "<%s: pid:%i, kernel sockets: (in:%s, out:%s)>" % (
+            return '<%s[%s🔵], "%s", pid:%i, kernel sockets: (in:%s, out:%s)>'\
+                   % (
                 self.__class__.__name__,
+                self.name,
+                self.kernel,
                 self.kernel_proc.pid,
                 self.kernel_socket_in.uri,
                 self.kernel_socket_out.uri,
             )
         else:
-            return "<%s: %s>" % (self.__class__.__name__, self.name)
+            return '<%s[%s🔴], "%s">' % (self.__class__.__name__, self.name, self.kernel)
 
 
 class _StartEvent(object):

wolframclient/evaluation/kernel/localsession.py

@@ -283,10 +283,7 @@ def set_parameter(self, parameter_name, parameter_value):
     set_parameter.__doc__ = WolframKernelController.set_parameter.__doc__
 
     def __repr__(self):
-        if self.started:
-            return "<%s: kernel controller=%s>" % (
-                self.__class__.__name__,
-                self.kernel_controller,
-            )
-        else:
-            return "<%s: not started>" % self.__class__.__name__
+        return "<%s: %s>" % (
+            self.__class__.__name__,
+            self.kernel_controller,
+        )

wolframclient/utils/api.py

@@ -64,6 +64,7 @@
 
 zmq = API(
     Context="zmq.Context",
+    Frame="zmq.Frame",
     PUSH="zmq.PUSH",
     PULL="zmq.PULL",
     PAIR="zmq.PAIR",

wolframclient/utils/externalevaluate.py

@@ -90,8 +90,7 @@ def __init__(self, socket):
         self.socket = socket
 
     def write(self, bytes):
-        self.socket.send(bytes)
-
+        self.socket.send(zmq.Frame(bytes))
 
 class StdoutProxy:
 
@@ -159,7 +158,7 @@ def handle_message(socket, evaluate_message=evaluate_message, consumer=None):
 
     __traceback_hidden_variables__ = True
 
-    message = binary_deserialize(socket.recv(), consumer=consumer)
+    message = binary_deserialize(socket.recv(copy=False).buffer, consumer=consumer)
     result = evaluate_message(**message)
 
     sys.stdout.flush()