nextmsg: optionally reuse given carray objects

Author: osch

.github/workflows/test.yml

@@ -24,6 +24,7 @@ jobs:
     - name: setup 
       run: |
         luarocks install .github/workflows/lua-llthreads2-0.1.6-1.rockspec
+        luarocks --server=https://luarocks.org/dev install carray
         luarocks make rockspecs/mtmsg-scm-0.rockspec
 
     - name: test
@@ -43,6 +44,7 @@ jobs:
         lua test10.lua
         lua test11.lua
         lua test12.lua
+        lua test13.lua
         cd ../examples
         lua example01.lua
         lua example02.lua

README.md

@@ -359,7 +359,7 @@ assert(lst:nextmsg(0) == nil)
   Possible errors: *mtmsg.error.object_closed*
 
 
-* **`buffer:nextmsg([timeout])`**
+* **`buffer:nextmsg([timeout][, carray]*)`**
 
   Returns all the arguments that were given as one message by invoking the method
   *buffer:addmsg()* or *buffer:setmsg()*. The returned message is removed
@@ -368,10 +368,15 @@ assert(lst:nextmsg(0) == nil)
   * *timeout* optional float, maximal time in seconds for waiting for the next 
     message. The method returns without result if timeout is reached.
 
-  If no timeout is given and *buffer:isnonblock() == false* then this methods waits
-  without timeout limit until a next message becomes available.
+  * *carray* optional one or more [carray] objects. These objects are reused 
+             if the message contains an array of the corresponding data type 
+             of the given *carray*. Unused *carray* objects are reset to length 0
+             after this call.
+  
+  If timeout is not given or *nil* and *buffer:isnonblock() == false* then this 
+  methods waits without timeout limit forever until a next message becomes available.
 
-  If no timeout is given and *buffer:isnonblock() == true* then this method
+  If timeout is not given or *nil* and *buffer:isnonblock() == true* then this method
   returns immediately without result if no next message is available or if the
   buffer is concurrently accessed from another thread.
 
@@ -492,7 +497,7 @@ assert(lst:nextmsg(0) == nil)
   Possible errors: *mtmsg.error.operation_aborted*
 
 
-* **`listener:nextmsg([timeout])`**
+* **`listener:nextmsg([timeout][, carray]*)`**
 
   Returns all the arguments that were given as one message by invoking the method
   *buffer:addmsg()* or *buffer:setmsg()* to one of the buffers that are
@@ -502,10 +507,15 @@ assert(lst:nextmsg(0) == nil)
   * *timeout* optional float, maximal time in seconds for waiting for the next 
     message. The method returns without result if timeout is reached.
 
-  If no timeout is given and *listener:isnonblock() == false* then this methods waits
-  without timeout limit until a next message becomes available.
+  * *carray* optional one or more [carray] objects. These objects are reused 
+             if the message contains an array of the corresponding data type 
+             of the given *carray*. Unused *carray* objects are reset to length 0
+             after this call.
+  
+  If timeout is not given or *nil* and *listener:isnonblock() == false* then this 
+  methods waits without timeout limit forever until a next message becomes available.
 
-  If no timeout is given and *listener:isnonblock() == true* then this methods
+  If timeout is not given or *nil* and *listener:isnonblock() == true* then this methods
   returns immediately without result if no next message is available or if the
   listener is concurrently accessed from another thread.
 
@@ -653,15 +663,20 @@ is equivalent to
     local a, b, c = buffer:nextmsg()
 ```
 
-* **`reader:next([count])`**
+* **`reader:next([count][, carray]*)`**
 
-  Returns next message elements from the reader. Returns nothing no more message elements 
+  Returns next message elements from the reader. Returns nothing if no more message elements 
   are available. The returned elements are removed from the reader's internal list of message
   elements.
 
   * *count* - optional integer, the maximaum number of message elements to be returned
               (defaults to 1).
 
+  * *carray* optional one or more [carray] objects. These objects are reused 
+             if the message contains an array of the corresponding data type 
+             of the given *carray*. Unused *carray* objects are reset to length 0
+             after this call.
+
 * **`reader:nextmsg(buffer|listener, [timeout])`**
 
   Gets the next message of a buffer or listener into the reader in one atomic step. The

appveyor.yml

@@ -1,7 +1,6 @@
 version: 0.3.2.{build}
 
-os:
-- Windows Server 2012 R2
+image: Windows Server 2012 R2
 
 shallow_clone: true
 
@@ -12,6 +11,7 @@ environment:
   - LUA: "lua 5.1"
   - LUA: "lua 5.2"
   - LUA: "lua 5.3"
+  - LUA: "lua 5.4"
 
 platform:
   - x64
@@ -24,7 +24,13 @@ cache:
 
 install:
   - echo "Setup..."
-  - set PATH=C:\Python27\Scripts;%LR_EXTERNAL%;%PATH%
+  - set PATH=C:\Python38\;C:\Python38\Scripts;%LR_EXTERNAL%;%PATH%
+  - where py
+  - py --version
+  - where python
+  - python --version
+  - where pip
+  - pip --version
   - if /I "%platform%"=="x86" set HR_TARGET=vs_32
   - if /I "%platform%"=="x64" set HR_TARGET=vs_64
   - if /I "%platform%"=="mingw" set HR_TARGET=mingw
@@ -34,10 +40,9 @@ install:
       mkdir "%LR_EXTERNAL%\lib" &&
       mkdir "%LR_EXTERNAL%\include"
     )
-  - if not exist c:\hererocks (
-      pip install hererocks &&
-      hererocks c:\hererocks --%LUA% --target %HR_TARGET% -rlatest
-    )
+  - python -m pip install --upgrade pip
+  - pip install git+https://github.com/luarocks/hererocks@0.25.0
+  - hererocks c:\hererocks --%LUA% --target %HR_TARGET% -rlatest
   - call c:\hererocks\bin\activate
 
 before_build:
@@ -46,6 +51,7 @@ before_build:
 build_script:
   - echo "Making..."
   - luarocks install lua-llthreads2
+  - luarocks --server=https://luarocks.org/dev install carray
   - luarocks make rockspecs/mtmsg-scm-0.rockspec
 
 before_test:
@@ -74,6 +80,7 @@ test_script:
   - lua test10.lua
   - lua test11.lua
   - lua test12.lua
+  - lua test13.lua
 
 after_test:
   # deploy

src/Makefile

@@ -11,7 +11,7 @@ LNX_COPTS   :=
 WIN_COPTS   := -I/mingw64/include/lua5.1 
 MAC_COPTS   := -I/usr/local/opt/lua/include/lua5.3 
 
-LNX_LOPTS   := -lpthread
+LNX_LOPTS   := -g -lpthread
 WIN_LOPTS   := -lkernel32
 MAC_LOPTS   := -lpthread
 

src/buffer.c

@@ -4,6 +4,8 @@
 #define RECEIVER_CAPI_IMPLEMENT_SET_CAPI 1
 #define SENDER_CAPI_IMPLEMENT_SET_CAPI   1
 
+#define CARRAY_CAPI_IMPLEMENT_GET_CAPI   1
+
 #include "buffer.h"
 #include "listener.h"
 #include "serialize.h"
@@ -12,6 +14,7 @@
 #include "receiver_capi_impl.h"
 #include "sender_capi_impl.h"
 #include "notify_capi_impl.h"
+#include "carray_capi.h"
 
 const char* const MTMSG_BUFFER_CLASS_NAME = "mtmsg.buffer";
 
@@ -793,16 +796,41 @@ int mtmsg_buffer_next_msg(lua_State* L, BufferUserData* udata,
                           MsgBuffer* b, bool nonblock, int arg, double timeoutSeconds , MemBuffer* resultBuffer, size_t* argsSize,
                           sender_error_handler sender_eh, void* sender_ehdata)
 {
-    lua_Number endTime   = 0; /* 0 = no timeout */
-
+    lua_Number endTime   = -1; /* -1 = no timeout, wait forever */
+    int argTop = 0;
     if (L) {
-        if (arg && !lua_isnoneornil(L, arg)) {
-            lua_Number waitSeconds = luaL_checknumber(L, arg);
-            endTime = mtmsg_current_time_seconds() + waitSeconds;
+        argTop = lua_gettop(L);
+        if (arg && arg <= argTop) {
+            int t = lua_type(L, arg);
+            if (t == LUA_TNIL || t == LUA_TNUMBER) {
+                if (t == LUA_TNUMBER) {
+                    lua_Number waitSeconds = lua_tonumber(L, arg);
+                    if (waitSeconds < 0) waitSeconds = 0;
+                    endTime = mtmsg_current_time_seconds() + waitSeconds;
+                    if (waitSeconds == 0) {
+                        nonblock = true;
+                    }
+                }
+                arg += 1;
+            } else {
+                int reason = 0;
+                if (t != LUA_TUSERDATA || !carray_get_capi(L, arg, &reason)) {
+                    if (reason == 1) {
+                        return luaL_argerror(L, arg, "incompatible carray capi version number");
+                    } else if (!resultBuffer) {
+                        return luaL_argerror(L, arg, "timeout seconds or carray expected");
+                    } else {
+                        return luaL_argerror(L, arg, "timeout seconds expected");
+                    }
+                }
+            }
         }
     } else {
-        if (timeoutSeconds >= 0) {
+        if (timeoutSeconds >= 0) { /* timeoutSeconds < 0 -> no timeout, wait forever */
             endTime = mtmsg_current_time_seconds() + timeoutSeconds;
+            if (timeoutSeconds == 0) {
+                nonblock = true;
+            }
         }
     }
 
@@ -848,13 +876,21 @@ int mtmsg_buffer_next_msg(lua_State* L, BufferUserData* udata,
                                par.inMaxArgCount  = -1;
                                par.parsedLength   = 0;
                                par.parsedArgCount = 0;
-                               par.carrayCapi     = udata->carrayCapi;
+                               par.carrayCapi     = udata->carrayCapi; 
+                               par.errorArg       = 0;
             lua_pushcfunction(L, mtmsg_serialize_get_msg_args);
+            lua_insert(L, arg);
             lua_pushlightuserdata(L, &par);
-            int rc = lua_pcall(L, 1, LUA_MULTRET, 0);
+            lua_insert(L, arg + 1);
+            int nargs = argTop - arg + 1;
+            int rc = lua_pcall(L, nargs + 1, LUA_MULTRET, 0);
             if (rc != LUA_OK) {
                 async_mutex_unlock(b->sharedMutex);
-                return lua_error(L);
+                if (par.errorArg) {
+                    return luaL_argerror(L, arg + par.errorArg - 2, lua_tostring(L, -1));
+                } else {
+                    return lua_error(L);
+                }
             }
             rslt = par.parsedArgCount;
             udata->carrayCapi = par.carrayCapi;
@@ -912,7 +948,7 @@ int mtmsg_buffer_next_msg(lua_State* L, BufferUserData* udata,
         }
         return rslt;
     } else {
-        if (endTime > 0) {
+        if (endTime >= 0) {
             lua_Number now = mtmsg_current_time_seconds();
             if (now < endTime) {
                 async_mutex_wait_millis(b->sharedMutex, (int)((endTime - now) * 1000 + 0.5));

src/carray_capi.h

@@ -11,9 +11,9 @@
 #endif
 
 #define CARRAY_CAPI_ID_STRING     "_capi_carray"
-#define CARRAY_CAPI_VERSION_MAJOR  -1
+#define CARRAY_CAPI_VERSION_MAJOR  -2
 #define CARRAY_CAPI_VERSION_MINOR   0
-#define CARRAY_CAPI_VERSION_PATCH   1
+#define CARRAY_CAPI_VERSION_PATCH   0
 
 typedef struct carray_capi     carray_capi;
 typedef struct carray_info     carray_info;
@@ -69,6 +69,7 @@ struct carray_info
     carray_attr attr;
     size_t      elementSize;
     size_t      elementCount;
+    size_t      elementCapacity;
 };
 
 /**
@@ -186,6 +187,20 @@ struct carray_capi
      * otherwise behaviour may be undefined.
      */
     const void* (*getReadableElementPtr)(const carray* a, size_t offset, size_t count);
+    
+    /**
+     * Resizes the array.
+     *
+     * newElementCount - new number of elements. If this is larger than the current 
+     *                   element count, the new elements are uninitialized.
+     *
+     * shrinkCapacity  - flag, if true the capacity is set to the new size.
+     *
+     * Returns pointer to the first element in the array. The caller may read
+     * or write newElementCount bytes at this pointer.
+     * Returns NULL on failure or if newElementCount == 0.
+     */
+    void* (*resizeCarray)(carray* a, size_t newElementCount, int shrinkCapacity);
 };
 
 #if CARRAY_CAPI_IMPLEMENT_SET_CAPI

src/listener.c

@@ -1,8 +1,11 @@
+#define CARRAY_CAPI_IMPLEMENT_GET_CAPI   1
+
 #include "listener.h"
 #include "buffer.h"
 #include "serialize.h"
 #include "main.h"
 #include "error.h"
+#include "carray_capi.h"
 
 const char* const MTMSG_LISTENER_CLASS_NAME = "mtmsg.listener";
 
@@ -407,11 +410,35 @@ int mtmsg_listener_next_msg(lua_State* L, ListenerUserData* udata,
                             MsgListener* listener, bool nonblock, int arg, 
                             MemBuffer* resultBuffer, size_t* argsSize)
 {
-    lua_Number endTime   = 0; /* 0 = no timeout */
+    lua_Number endTime = -1; /* -1 = no timeout, wait forever */
+
+    int argTop = lua_gettop(L);
 
-    if (!lua_isnoneornil(L, arg)) {
-        lua_Number waitSeconds = luaL_checknumber(L, arg);
-        endTime = mtmsg_current_time_seconds() + waitSeconds;
+    if (arg && arg <= argTop) {
+        int t = lua_type(L, arg);
+        if (t == LUA_TNIL || t == LUA_TNUMBER) {
+            if (t == LUA_TNUMBER) {
+                lua_Number waitSeconds = lua_tonumber(L, arg);
+                if (waitSeconds < 0) waitSeconds = 0;
+                endTime = mtmsg_current_time_seconds() + waitSeconds;
+                if (waitSeconds == 0) {
+                    nonblock = true;
+                }
+            }
+            arg += 1;
+        }
+        else {
+            int reason = 0;
+            if (t != LUA_TUSERDATA || !carray_get_capi(L, arg, &reason)) {
+                if (reason == 1) {
+                    return luaL_argerror(L, arg, "incompatible carray capi version number");
+                } else if (!resultBuffer) {
+                    return luaL_argerror(L, arg, "timeout seconds or carray expected");
+                } else {
+                    return luaL_argerror(L, arg, "timeout seconds expected");
+                }
+            }
+        }
     }
 
     if (nonblock) {
@@ -457,12 +484,20 @@ int mtmsg_listener_next_msg(lua_State* L, ListenerUserData* udata,
                                        par.parsedLength   = 0;
                                        par.parsedArgCount = 0;
                                        par.carrayCapi     = udata->carrayCapi;
+                                       par.errorArg       = 0;
                     lua_pushcfunction(L, mtmsg_serialize_get_msg_args);
+                    lua_insert(L, arg);
                     lua_pushlightuserdata(L, &par);
-                    int rc = lua_pcall(L, 1, LUA_MULTRET, 0);
+                    lua_insert(L, arg + 1);
+                    int nargs = argTop - arg + 1;
+                    int rc = lua_pcall(L, nargs + 1, LUA_MULTRET, 0);
                     if (rc != LUA_OK) {
                         async_mutex_unlock(&listener->listenerMutex);
-                        return lua_error(L);
+                        if (par.errorArg) {
+                            return luaL_argerror(L, par.errorArg, lua_tostring(L, -1));
+                        } else {
+                            return lua_error(L);
+                        }
                     }
                     rslt = par.parsedArgCount;
                     udata->carrayCapi = par.carrayCapi;
@@ -525,7 +560,7 @@ int mtmsg_listener_next_msg(lua_State* L, ListenerUserData* udata,
             }
         }
     }
-    if (endTime > 0) {
+    if (endTime >= 0) {
         lua_Number now = mtmsg_current_time_seconds();
         if (now < endTime) {
             async_mutex_wait_millis(&listener->listenerMutex, (int)((endTime - now) * 1000 + 0.5));