messenger should use zmq syncronously and block

messenger/make.py

@@ -7,6 +7,15 @@
 import shutil
 import subprocess
 
+use_shell = True if sys.platform.startswith("win32") else False
+
+
+def esc(path):
+    if ' ' in path:
+        return '"' + path + '"'
+    else:
+        return path
+
 
 def get_messenger_dir():
     # Check the system platform first
@@ -50,7 +59,7 @@ def do_build(make_cmd, messenger_exe):
     print('Building %s...' % messenger_exe)
     print(make_cmd)
     messenger_dir = get_messenger_dir()
-    subprocess.check_output(shlex.split(make_cmd), shell=True)
+    subprocess.check_output(shlex.split(make_cmd), shell=use_shell)
 
     messenger_loc = os.path.join(messenger_dir, messenger_exe)
 
@@ -69,7 +78,7 @@ def build_octave():
 
 def build_matlab(static=False):
     """build the messenger mex for MATLAB
-    
+
     static : bool
         Determines if the zmq library has been statically linked.
         If so, it will append the command line option -DZMQ_STATIC
@@ -78,14 +87,21 @@ def build_matlab(static=False):
     cfg = get_config()
     # To deal with spaces, remove quotes now, and add
     # to the full commands themselves.
-    matlab_bin = cfg['matlab_bin'].strip('"')
+    if 'matlab_bin' in cfg and cfg['matlab_bin'] != '.':
+        matlab_bin = cfg['matlab_bin'].strip('"')
+    else:  # attempt to autodetect MATLAB filepath
+        try:
+            matlab_path = subprocess.check_output(['which', 'matlab']).strip()
+            matlab_bin = os.path.dirname(os.path.realpath(matlab_path))
+        except (OSError, subprocess.CalledProcessError):
+            raise ValueError("specify 'matlab_bin' in cfg file")
     # Get the extension
-    extcmd = '"' + os.path.join(matlab_bin, "mexext") + '"'
-    extension = subprocess.check_output(extcmd, shell=True)
+    extcmd = esc(os.path.join(matlab_bin, "mexext"))
+    extension = subprocess.check_output(extcmd, shell=use_shell)
     extension = extension.decode('utf-8').rstrip('\r\n')
 
     # Build the mex file
-    mex = '"' + os.path.join(matlab_bin, "mex") + '"'
+    mex = esc(os.path.join(matlab_bin, "mex"))
     paths = "-L%(zmq_lib)s -I%(zmq_inc)s" % cfg
     make_cmd = '%s -O %s -lzmq ./src/messenger.c' % (mex, paths)
     if static:
@@ -107,6 +123,8 @@ def build_matlab(static=False):
     if args.target == "matlab":
         build_matlab(static=args.static)
     elif args.target == "octave":
+        if args.static:
+            raise ValueError("static building not yet supported for octave")
         build_octave()
     else:
         raise ValueError()

messenger/mexa64/local.cfg

@@ -1,4 +1,4 @@
-MATLAB_BIN=/home/silvester/matlab2014b/bin/
+MATLAB_BIN=
 OCTAVE_INC=/usr/include
 OCTAVE_LIB=/usr/lib/x86_64-linux-gnu/
 ZMQ_INC=

messenger/mexw64/local.cfg

@@ -1,4 +1,4 @@
-MATLAB_BIN="C:\Program Files\MATLAB\2014b\bin"
+MATLAB_BIN="C:\Program Files\MATLAB\2013a\bin"
 OCTAVE_INC="C:\Octave\Octave-3.8.2\include\octave-3.8.2\octave"
 OCTAVE_LIB="C:\Octave\Octave-3.8.2\lib\octave\3.8.2"
 ZMQ_INC="C:\zeromq-4.0.5\include"

messenger/src/messenger.c

@@ -7,6 +7,7 @@
 /* Set a 200MB receiver buffer size */
 #define BUFLEN 200000000
 
+/* The variable cannot be named socket on windows */
 void *ctx, *socket_ptr;
 static int initialized = 0;
 
@@ -26,28 +27,18 @@ int initialize(char *socket_addr) {
     }
 }
 
-/* Listen over an existing socket
- * Now the receiver buffer is pre-allocated
- * In the future we can possibly use multi-part messaging
- */
-int listen_zmq(char *buffer, int buflen) {
+/* Check if the ZMQ server is intialized and print an error if not */
+int checkInitialized(void) {
     if (!initialized) {
         mexErrMsgTxt("Error: ZMQ session not initialized");
+        return 0;
     }
-
-    return zmq_recv(socket_ptr, buffer, buflen, ZMQ_NOBLOCK);
-}
-
-/* Sending out a message */
-int respond(char *msg_out, int len) {
-    if (!initialized) {
-        mexErrMsgTxt("Error: ZMQ session not initialized");
+    else {
+        return 1;
     }
-
-    return zmq_send(socket_ptr, msg_out, len, 0);
-
 }
 
+
 /* Cleaning up after session finished */
 void cleanup (void) {
     /* Send a confirmation message to the client */
@@ -106,14 +97,18 @@ void mexFunction(int nlhs, mxArray *plhs[],
 
     /* Listen over an existing socket */
     } else if (strcmp(cmd, "listen") == 0) {
+        int byte_recvd;
         char *recv_buffer = mxCalloc(BUFLEN, sizeof(char));
+        zmq_pollitem_t polls[] = {{socket_ptr, 0, ZMQ_POLLIN, 0}};
+
+        if (!checkInitialized()) return;
+
+        /* allow MATLAB to draw its graphics every 20ms */
+        while (zmq_poll(polls, 1, 20000) == 0) {
+            mexEvalString("drawnow");
+        }
 
-        int byte_recvd = listen_zmq(recv_buffer, BUFLEN);
-
-        while (byte_recvd == -1 && errno == EAGAIN) {
-        	mexCallMATLAB(0, NULL, 0, NULL, "drawnow");
-        	byte_recvd = listen_zmq(recv_buffer, BUFLEN);
-       	}
+        byte_recvd = zmq_recv(socket_ptr, recv_buffer, BUFLEN, 0);
 
         /* Check if the received data is complete and correct */
         if ((byte_recvd > -1) && (byte_recvd <= BUFLEN)) {
@@ -138,13 +133,15 @@ void mexFunction(int nlhs, mxArray *plhs[],
             mexErrMsgTxt("Please provide the message to send");
         }
 
+        if (!checkInitialized()) return;
+
         msglen = mxGetNumberOfElements(prhs[1]);
         msg_out = mxArrayToString(prhs[1]);
 
         plhs[0] = mxCreateLogicalMatrix(1, 1);
         p = mxGetLogicals(plhs[0]);
 
-        if (msglen == respond(msg_out, msglen)) {
+        if (msglen == zmq_send(socket_ptr, msg_out, msglen, 0)) {
             p[0] = 1;
         } else {
             p[0] = 0;