8 files changed, 310 insertions, 162 deletions

diff --git a/CHANGES b/CHANGES
index 31ca342..f4ff88a 100644
--- a/CHANGES
+++ b/CHANGES
@@ -3,6 +3,14 @@ CHANGES:
 
 CHANGE X:
 
+CHANGE 31 BGe 17-Apr-2011
+* linda uses a fast FIFO implementation to speed up data exchanges
+* new linda:count() method
+* new linda batched data read mode
+* proper key type check in all linda methods
+* fix setup-vc.cmd to support Visual Studio 2010 and Windows 7 64 bits
+* bugfix: release keeper state mutex at desinit
+
 CHANGE 30 BGe 30-Mar-2011
 * linda honors __tostring and __concat
 * new accessor linda:keys(), to retrieve the list of keys with pending data inside a linda
diff --git a/Makefile b/Makefile
index ddd8675..23bec03 100644
--- a/Makefile
+++ b/Makefile
@@ -85,11 +85,12 @@ test:
         $(MAKE) recursive
         $(MAKE) func_is_string
         $(MAKE) atexit
+        $(MAKE) linda_perf
 
 basic: tests/basic.lua $(_TARGET_SO)
         $(_PREFIX) $(LUA) $<
 
-# 
+#
 # This tries to show out a bug which happens in lane cleanup (multicore CPU's only)
 #
 REP_ARGS=-llanes -e "print'say aaa'; for i=1,10 do print(i) end"
@@ -155,6 +156,9 @@ appendud: tests/appendud.lua $(_TARGET_SO)
 func_is_string: tests/func_is_string.lua $(_TARGET_SO)
         $(_PREFIX) $(LUA) $<
 
+linda_perf: tests/linda_perf.lua $(_TARGET_SO)
+        $(_PREFIX) $(LUA) $<
+
 atexit: tests/atexit.lua $(_TARGET_SO)
         $(_PREFIX) $(LUA) $<
 
@@ -190,7 +194,7 @@ LUA_LIBDIR=$(DESTDIR)/lib/lua/5.1
 LUA_SHAREDIR=$(DESTDIR)/share/lua/5.1
 
 #
-# AKa 17-Oct: changed to use 'install -m 644' and 'cp -p' 
+# AKa 17-Oct: changed to use 'install -m 644' and 'cp -p'
 #
 install: $(_TARGET_SO) src/lanes.lua
         mkdir -p $(LUA_LIBDIR) $(LUA_SHAREDIR)
@@ -211,7 +215,7 @@ tar tgz:
 ifeq "$(VERSION)" ""
         echo "Usage: make tar VERSION=x.x"; false
 else
-        $(MAKE) clean 
+        $(MAKE) clean
         -rm -rf $(MODULE)-$(VERSION)
         mkdir $(MODULE)-$(VERSION)
         tar c * --exclude=.svn --exclude=.DS_Store --exclude="_*" \
@@ -223,8 +227,8 @@ else
         rm -rf $(MODULE)-$(VERSION)
         md5sum $(MODULE)-$(VERSION).tgz
 endif
-        
-        
+
+
 #--- Undocumented ---
 #
 
diff --git a/docs/index.html b/docs/index.html
index 7d61d96..3c4fe59 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -483,7 +483,7 @@ level locking is required; each Linda operation is atomic.
 <p>Characteristics of the Lanes implementation of Lindas are:
 
 <ul>
-    <li>keys can be of number, string or boolean type
+    <li>keys can be of boolean, number, string or light userdata type
     </li>
     <li>values can be any type supported by inter-state copying (same limits
     as for function parameters and upvalues)
@@ -560,17 +560,23 @@ They can be used for making shared tables of storage among the lanes.
 Writing to a slot overwrites existing value, and clears any possible queued
 entries. Table access and <tt>send</tt>/<tt>receive</tt> can be used together;
 reading a slot essentially peeks the next outcoming value of a queue.
-</p><p>
+</p>
+  <p>
 
     <table border="1" bgcolor="#E0E0FF" cellpadding="10">
       <tr>
         <td>
-          <code>[val]= linda_h:keys()</code>
+          <code>[val]= linda_h:count( [key[,...]])</code>
         </table>
 
   </p>
   <p>
-    A way to examine the slots in which a linda contains some pending data to be read.
+    Returns some information about the contents of the linda.
+    <br/>If no key is specified, and the linda is empty, returns nothing.
+    <br/>If no key is specified, and the linda is not empty, returns a table of key/count pairs that counts the number of items in each
+    of the exiting keys of the linda. This count can be 0 if the key has been used but is empty.
+    <br/>If a single key is specified, returns the number of pending items, or nothing if the key is unknown.
+    <br/>If more than one key is specified, return a table of key/count pairs for the known keys.
   </p>
 
 <!--
diff --git a/setup-vc.cmd b/setup-vc.cmd
index 247459c..e623495 100644
--- a/setup-vc.cmd
+++ b/setup-vc.cmd
@@ -30,10 +30,19 @@ goto FOUND_VC
 

 :TRY_VC9

 set VSINSTALLDIR=%ProgramFiles%\Microsoft Visual Studio 9.0

-if not exist "%VSINSTALLDIR%\VC\vcvarsall.bat" goto ERR_NOVC

+if exist "%VSINSTALLDIR%\VC\vcvarsall.bat" goto WARN_VC

+set VSINSTALLDIR=%ProgramFiles(x86)%\Microsoft Visual Studio 9.0

+if exist "%VSINSTALLDIR%\VC\vcvarsall.bat" goto WARN_VC

 

+:TRY_VC10

+set VSINSTALLDIR=%ProgramFiles%\Microsoft Visual Studio 10.0

+if exist "%VSINSTALLDIR%\VC\vcvarsall.bat" goto WARN_VC

+set VSINSTALLDIR=%ProgramFiles(x86)%\Microsoft Visual Studio 10.0

+if exist "%VSINSTALLDIR%\VC\vcvarsall.bat" goto WARN_VC

+

+:WARN_VC

 echo.

-echo *** Warning: Visual C++ 2008 in use ***

+echo *** Warning: Visual C++ 2008/2010 in use ***

 echo.

 echo Using VC++2005 is recommended for runtime compatibility issues

 echo (LuaBinaries and LfW use it; if you compile everything from

@@ -70,7 +79,7 @@ goto EXIT
 REM ---

 :ERR_NOVC

 echo.

-echo ** ERROR: Visual C++ 2005/08 Express - not detected

+echo ** ERROR: Visual C++ 2005/08/10 Express - not detected

 echo           You can set the environment variables separately, and run 'make-vc.cmd'

 echo           or download the compiler from:

 echo           http://msdn.microsoft.com/vstudio/express/downloads/

diff --git a/src/keeper.c b/src/keeper.c
index 01e8880..5b355cb 100644
--- a/src/keeper.c
+++ b/src/keeper.c
@@ -103,15 +103,10 @@ char const *init_keepers( int const _nbKeepers)
                 luaG_openlibs( L, "io,table,package" );     // 'io' for debugging messages, package because we need to require modules exporting idfuncs
                 serialize_require( L);
 
-                /* We could use an empty table in 'keeper.lua' as the sentinel, but maybe
-                * checking for a lightuserdata is faster. (any unique value will do -> take the address of some global of ours)
-                */
-                lua_pushlightuserdata( L, &GNbKeepers);
-                lua_setglobal( L, "nil_sentinel");
 
                 // Read in the preloaded chunk (and run it)
                 //
-                if (luaL_loadbuffer( L, keeper_chunk, sizeof(keeper_chunk), "=lanes_keeper" ))
+                if (luaL_loadbuffer( L, keeper_chunk, sizeof(keeper_chunk), "@keeper.lua"))
                         return "luaL_loadbuffer() failed";   // LUA_ERRMEM
 
                 if (lua_pcall( L, 0 /*args*/, 0 /*results*/, 0 /*errfunc*/ ))
@@ -152,6 +147,34 @@ void keeper_release( struct s_Keeper *K)
         MUTEX_UNLOCK( &K->lock_);
 }
 
+void keeper_toggle_nil_sentinels( lua_State *L, int _val_i, int _nil_to_sentinel)
+{
+        int i, n = lua_gettop( L);
+        /* We could use an empty table in 'keeper.lua' as the sentinel, but maybe
+        * checking for a lightuserdata is faster. (any unique value will do -> take the address of some global of ours)
+        */
+        void *nil_sentinel = &GNbKeepers;
+        for( i = _val_i; i <= n; ++ i)
+        {
+                if( _nil_to_sentinel)
+                {
+                        if( lua_isnil( L, i))
+                        {
+                                lua_pushlightuserdata( L, nil_sentinel);
+                                lua_replace( L, i);
+                        }
+                }
+                else
+                {
+                        if( lua_touserdata( L, i) == nil_sentinel)
+                        {
+                                lua_pushnil( L);
+                                lua_replace( L, i);
+                        }
+                }
+        }
+}
+
 /*
 * Call a function ('func_name') in the keeper state, and pass on the returned
 * values to 'L'.
@@ -197,6 +220,7 @@ void close_keepers(void)
                 lua_close( GKeepers[i].L);
                 GKeepers[i].L = 0;
                 //assert( GKeepers[i].count == 0);
+                MUTEX_FREE( &GKeepers[i].lock_);
         }
         if( GKeepers) free( GKeepers);
         GKeepers = NULL;
diff --git a/src/keeper.h b/src/keeper.h
index e959c7c..27a0f68 100644
--- a/src/keeper.h
+++ b/src/keeper.h
@@ -11,6 +11,7 @@ struct s_Keeper
 const char *init_keepers( int const _nbKeepers);
 struct s_Keeper *keeper_acquire( const void *ptr);
 void keeper_release( struct s_Keeper *K);
+void keeper_toggle_nil_sentinels( lua_State *L, int _val_i, int _nil_to_sentinel);
 int keeper_call( lua_State *K, char const *func_name, lua_State *L, void *linda, uint_t starting_index);
 void close_keepers(void);
 
diff --git a/src/keeper.lua b/src/keeper.lua
index 4bb181a..828932e 100644
--- a/src/keeper.lua
+++ b/src/keeper.lua
@@ -34,30 +34,27 @@ THE SOFTWARE.
 ===============================================================================
 ]]--
 
--- unique key instead of 'nil' in queues
---
-assert( nil_sentinel )
-
 -- We only need to have base and table libraries (and io for debugging)
 --
 local table_concat = assert( table.concat)
 local table_insert = assert( table.insert)
 local table_remove = assert( table.remove)
+local select, unpack = assert( select), assert( unpack)
 
 --[[
 local function WR(...)
-    if io then 
-        io.stderr:write( table_concat({...},'\t').."\n" ) 
+    if io then
+        io.stderr:write( table_concat({...},'\t').."\n" )
     end
 end
 
 local function DEBUG(title,ud,key)
     assert( title and ud and key )
 
-    local data,incoming,_= tables(ud)
+    local data,_= tables(ud)
 
     local s= tostring(data[key])
-    for _,v in ipairs( incoming[key] or {} ) do
+    for _,v in ipairs( data[key] or {} ) do
         s= s..", "..tostring(v)
     end
     WR( "*** "..title.." ("..tostring(key).."): ", s )
@@ -65,34 +62,64 @@ end
 --]]
 
 -----
--- Actual data store
---
--- { [linda_deep_ud]= { key= val [, ...] }
---      ...
--- }
+-- FIFO for a key
 --
-local _data= {}
+
+local fifo_new = function()
+    return { first = 1, count = 0}
+end
+
+local fifo_push = function( fifo, ...)
+    local first, count, added = fifo.first, fifo.count, select( '#', ...)
+    local start = first + count - 1
+    for i = 1, added do
+        fifo[start + i] = select( i, ...)
+    end
+    fifo.count = count + added
+end
+
+local fifo_peek = function( fifo, count)
+    if count <= fifo.count then
+                        local first = fifo.first
+                        local last = first + count - 1
+                        return unpack( fifo, first, last)
+    end
+end
+
+local fifo_pop = function( fifo, count)
+    if count > fifo.count then error("list is too short") end
+    local first = fifo.first
+    local last = first + count - 1
+    local out = { unpack( fifo, first, last)}
+    for i = first, last do
+        fifo[i] = nil
+    end
+    fifo.first = first + count
+    fifo.count = fifo.count - count
+    return unpack( out)
+end
+
 
 -----
--- Entries queued for use when the existing 'data[ud][key]' entry is consumed.
+-- Actual data store
 --
--- { [linda_deep_ud]= { key= { val [, ... } [, ...] }
+-- { [linda_deep_ud]= { key= { val [, ... ] } [, ...] }
 --      ...
 -- }
 --
-local _incoming= {}
+local _data= {}
 
 -----
 -- Length limits (if any) for queues
 --
 -- 0:   don't queue values at all; ':send()' waits if the slot is not vacant
 -- N:   allow N values to be queued (slot itself + N-1); wait if full
--- nil: no limits, '_incoming' may grow endlessly
+-- nil: no limits, '_data' may grow endlessly
 --
 local _limits= {}
 
 -----
--- data_tbl, incoming_tbl, limits_tbl = tables( linda_deep_ud )
+-- data_tbl, limits_tbl = tables( linda_deep_ud )
 --
 -- Gives appropriate tables for a certain Linda (creates them if needed)
 --
@@ -101,14 +128,13 @@ local function tables( ud )
     --
     if not _data[ud] then
         _data[ud]= {}
-        _incoming[ud]= {}
         _limits[ud]= {}
     end
-    return _data[ud], _incoming[ud], _limits[ud]
+    return _data[ud], _limits[ud]
 end
 
 -----
--- bool= send( linda_deep_ud, key, ... )
+-- bool= send( linda_deep_ud, key, ...)
 --
 -- Send new data (1..N) to 'key' slot. This send is atomic; all the values
 -- end up one after each other (this is why having possibility for sending
@@ -119,42 +145,28 @@ end
 -- Returns: 'true' if all the values were placed
 --          'false' if sending would exceed the queue limit (wait & retry)
 --
-function send( ud, key, ... )
+function send( ud, key, ...)
 
-    local data,incoming,limits= tables(ud)
+    local data, limits = tables( ud)
 
-    local n= select('#',...)
-    if n==0 then return true end    -- nothing to send
+    local n = select( '#', ...)
+    if n == 0 then return true end    -- nothing to send
 
     -- Initialize queue for all keys that have been used with ':send()'
     --
-    if incoming[key]==nil then
-        incoming[key]= {}
+    if data[key] == nil then
+        data[key] = fifo_new()
     end
+    local fifo = data[key]
 
-    local len= data[key] and 1+#incoming[key] or 0
-    local m= limits[key]
+    local len = fifo.count
+    local m = limits[key]
 
     if m and len+n > m then
         return false    -- would exceed the limit; try again later
     end
 
-    for i=1,n do
-        local val= select(i,...)
-
-        -- 'nil' in the data replaced by sentinel
-        if val==nil then
-            val= nil_sentinel
-        end
-
-        if len==0 then
-            data[key]= val
-            len= 1
-        else
-            incoming[key][len]= val
-            len= len+1
-        end
-    end
+    fifo_push( fifo, ...)
     return true
 end
 
@@ -165,94 +177,125 @@ end
 -- Read any of the given keys, consuming the data found. Keys are read in
 -- order.
 --
-function receive( ud, ... )
+function receive( ud, ...)
 
-    local data,incoming,_= tables(ud)
+    local data, _ = tables( ud)
 
-    for i=1,select('#',...) do
-        local key= select(i,...)
-        local val= data[key]
-
-        if val~=nil then
-            if incoming[key] and incoming[key][1]~=nil then
-                -- pop [1] from 'incoming[key]' into the actual slot
-                data[key]= table_remove( incoming[key], 1 )
-            else
-                data[key]= nil  -- empty the slot
-            end
-            if val==nil_sentinel then
-                val= nil
+    for i = 1, select( '#', ...) do
+        local key = select( i, ...)
+        local fifo = data[key]
+        if fifo and fifo.count > 0 then
+            local val = fifo_pop( fifo, 1)
+            if val ~= nil then
+                    return val, key
             end
-            return val, key
         end
     end
-    --return nil
+end
+
+
+-----
+-- [val1, ... valCOUNT]= receive_batched( linda_deep_ud, batch_sentinel, key , COUNT)
+--
+-- Read any of the given keys, consuming the data found. Keys are read in
+-- order.
+--
+receive_batched = function( ud, batch_sentinel, key, count)
+    if count > 0 then
+        local data, _ = tables( ud)
+        local fifo = data[key]
+        if fifo and fifo.count >= count then
+            return fifo_pop( fifo, count)
+        end
+    end
 end
 
 
 -----
 -- = limit( linda_deep_ud, key, uint )
 --
-function limit( ud, key, n )
+function limit( ud, key, n)
 
-    local _,_,limits= tables(ud)
+    local _, limits = tables( ud)
 
-    limits[key]= n
+    limits[key] = n
 end
 
 
 -----
 -- void= set( linda_deep_ud, key, [val] )
 --
-function set( ud, key, val )
+function set( ud, key, val)
 
-    local data,incoming,_= tables(ud)
+    local data, _ = tables( ud)
 
     -- Setting a key to 'nil' really clears it; only queing uses sentinels.
     --
-    data[key]= val
-    incoming[key]= nil
+    if val ~= nil then
+        local fifo = fifo_new()
+        fifo_push( fifo, val)
+        data[key] = fifo
+    else
+        data[key] = nil
+    end
 end
 
 
 -----
 -- [val]= get( linda_deep_ud, key )
 --
-function get( ud, key )
-
-    local data,_,_= tables(ud)
-
-    local val= data[key]
-    if val==nil_sentinel then
-        val= nil
-    end
-    return val
+function get( ud, key)
+    local data, _ = tables( ud)
+    local fifo = data[key]
+    return fifo and fifo_peek( fifo, 1)
 end
 
 
 -----
--- [val]= keys( linda_deep_ud)
+-- [val]= count( linda_deep_ud, ...)
 --
-function keys( ud)
-
-    local data,_,_= tables(ud)
-
-    local out = {}
-    for key, v in pairs( data) do
-        table_insert( out, key)
+-- 3 modes of operation
+-- linda:count() -> returns a table of key/count pairs
+-- linda:count(key) returns the number of items waiting in the key
+-- linda:count(key,...) -> returns a table telling, for each key, the number of items
+function count( ud, ...)
+    local data, _ = tables( ud)
+    local n = select( '#', ...)
+    if n == 0 then
+        local out
+        for key, _ in pairs( data) do
+                        local fifo = data[key]
+                        local count = fifo and fifo.count or 0
+                        out = out or {}
+            out[key] = count
+            found = true
+        end
+        return out
+    elseif n == 1 then
+        local key = ...
+        local fifo = data[key]
+                return fifo and fifo.count or nil
+    else -- more than 1 key
+        local out
+        for i = 1, n do
+            local key = select( i, ...)
+                        local fifo = data[key]
+                        local count = fifo and fifo.count or nil
+                        out = out or {}
+            out[key] = count
+                end
+        return out
     end
-    return (#out > 0) and out or nil
 end
 
 
 -----
--- void= clear( linda_deep_ud )
+-- void= clear( linda_deep_ud)
 --
 -- Clear the data structures used for a Linda (at its destructor)
 --
-function clear( ud )
+function clear( ud)
 
     _data[ud]= nil
-    _incoming[ud]= nil
     _limits[ud]= nil
 end
diff --git a/src/lanes.c b/src/lanes.c
index e06d367..e514e6b 100644
--- a/src/lanes.c
+++ b/src/lanes.c
@@ -51,7 +51,7 @@
  *      ...
  */
 
-const char *VERSION= "2.1.0";
+const char *VERSION= "2.2.0";
 
 /*
 ===============================================================================
@@ -249,6 +249,20 @@ static void linda_id( lua_State*, char const * const which);
 #define lua_toLinda(L,n) ((struct s_Linda *)luaG_todeep( L, linda_id, n ))
 
 
+static void check_key_types( lua_State *L, int _start, int _end)
+{
+        int i;
+        for( i = _start; i <= _end; ++ i)
+        {
+                int t = lua_type( L, i);
+                if( t == LUA_TBOOLEAN || t == LUA_TNUMBER || t == LUA_TSTRING || t == LUA_TLIGHTUSERDATA)
+                {
+                        continue;
+                }
+                luaL_error( L, "argument #%d: invalid key type (not a boolean, string, number or light userdata)", i);
+        }
+}
+
 /*
 * bool= linda_send( linda_ud, [timeout_secs=-1,] key_num|str|bool|lightuserdata, ... )
 *
@@ -271,15 +285,18 @@ LUAG_FUNC( linda_send)
         if( lua_isnumber(L, 2))
         {
                 timeout= SIGNAL_TIMEOUT_PREPARE( lua_tonumber(L,2) );
-                key_i++;
+                ++ key_i;
         }
-        else if( lua_isnil( L, 2))
+        else if( lua_isnil( L, 2)) // alternate explicit "no timeout" by passing nil before the key
         {
-                key_i++;
+                ++ key_i;
         }
 
-        if( lua_isnil( L, key_i))
-                luaL_error( L, "nil key" );
+        // make sure the keys are of a valid type
+        check_key_types( L, key_i, key_i);
+
+        // convert nils to some special non-nil sentinel in sent values
+        keeper_toggle_nil_sentinels( L, key_i + 1, 1);
 
         STACK_GROW(L, 1);
         {
@@ -315,11 +332,11 @@ LUAG_FUNC( linda_send)
 
                         cancel = cancel_test( L);   // testing here causes no delays
                         if (cancel)
+                        {
                                 break;
+                        }
 
-                        // Bugfix by Benoit Germain Dec-2009: change status of lane to "waiting"
-                        //
-#if 1
+                        // change status of lane to "waiting"
                         {
                                 struct s_lane *s;
                                 enum e_status prev_status = ERROR_ST; // prevent 'might be used uninitialized' warnings
@@ -338,6 +355,7 @@ LUAG_FUNC( linda_send)
                                         ASSERT_L( s->waiting_on == NULL);
                                         s->waiting_on = &linda->read_happened;
                                 }
+                                // could not send because no room: wait until some data was read before trying again, or until timeout is reached
                                 if( !SIGNAL_WAIT( &linda->read_happened, &K->lock_, timeout))
                                 {
                                         if( s)
@@ -353,12 +371,6 @@ LUAG_FUNC( linda_send)
                                         s->status = prev_status;
                                 }
                         }
-#else
-                        // K lock will be released for the duration of wait and re-acquired
-                        //
-                        if( !SIGNAL_WAIT( &linda->read_happened, &K->lock_, timeout))
-                                break;  // timeout
-#endif
                 }
                 STACK_END( KL, 0)
                 keeper_release( K);
@@ -379,18 +391,23 @@ LUAG_FUNC( linda_send)
 
 
 /*
-* [val, key]= linda_receive( linda_ud, [timeout_secs_num=-1], key_num|str|bool|lightuserdata [, ...] )
-*
-* Receive a value from Linda, consuming it.
-*
-* Returns:  value received (which is consumed from the slot)
-*           key which had it
-*/
+ * 2 modes of operation
+ * [val, key]= linda_receive( linda_ud, [timeout_secs_num=-1], key_num|str|bool|lightuserdata [, ...] )
+ * Consumes a single value from the Linda, in any key.
+ * Returns: received value (which is consumed from the slot), and the key which had it
+
+ * [val1, ... valCOUNT]= linda_receive( linda_ud, [timeout_secs_num=-1], linda.batched, key_num|str|bool|lightuserdata, COUNT)
+ * Consumes COUNT values from the linda, from a single key.
+ * returns the COUNT consumed values, or nil if there weren't enough values to consume
+ *
+ */
+#define BATCH_SENTINEL "270e6c9d-280f-4983-8fee-a7ecdda01475"
 LUAG_FUNC( linda_receive)
 {
         struct s_Linda *linda = lua_toLinda( L, 1);
-        int pushed;
+        int pushed, expected_pushed;
         bool_t cancel = FALSE;
+        char *keeper_receive;
         
         time_d timeout = -1.0;
         uint_t key_i = 2;
@@ -400,26 +417,44 @@ LUAG_FUNC( linda_receive)
         if( lua_isnumber( L, 2))
         {
                 timeout = SIGNAL_TIMEOUT_PREPARE( lua_tonumber( L, 2));
-                key_i++;
+                ++ key_i;
         }
-        else if( lua_isnil( L, 2))
+        else if( lua_isnil( L, 2)) // alternate explicit "no timeout" by passing nil before the key
         {
-                key_i++;
+                ++ key_i;
         }
 
+        // make sure the keys are of a valid type
+        check_key_types( L, key_i, lua_gettop( L));
+
+        // are we in batched mode?
+        lua_pushliteral( L, BATCH_SENTINEL);
+        if( lua_equal( L, key_i, -1))
+        {
+                keeper_receive = "receive_batched";
+                expected_pushed = (int)luaL_checkinteger( L, key_i + 2);
+        }
+        else
+        {
+                keeper_receive = "receive";
+                expected_pushed = 2;
+        }
+        lua_pop( L, 1);
+
         {
                 struct s_Keeper *K = keeper_acquire( linda);
                 for( ;;)
                 {
-                        pushed = keeper_call( K->L, "receive", L, linda, key_i);
+                        pushed = keeper_call( K->L, keeper_receive, L, linda, key_i);
                         if( pushed < 0)
                         {
                                 break;
                         }
                         if( pushed > 0)
                         {
-                                ASSERT_L( pushed == 2);
-
+                                ASSERT_L( pushed == expected_pushed);
+                                // replace sentinels with real nils
+                                keeper_toggle_nil_sentinels( L, lua_gettop( L) - pushed, 0);
                                 // To be done from within the 'K' locking area
                                 //
                                 SIGNAL_ALL( &linda->read_happened);
@@ -438,9 +473,7 @@ LUAG_FUNC( linda_receive)
                                 break;
                         }
 
-                        // Bugfix by Benoit Germain Dec-2009: change status of lane to "waiting"
-                        //
-#if 1
+                        // change status of lane to "waiting"
                         {
                                 struct s_lane *s;
                                 enum e_status prev_status = ERROR_ST; // prevent 'might be used uninitialized' warnings
@@ -459,6 +492,7 @@ LUAG_FUNC( linda_receive)
                                         ASSERT_L( s->waiting_on == NULL);
                                         s->waiting_on = &linda->write_happened;
                                 }
+                                // not enough data to read: wakeup when data was sent, or when timeout is reached
                                 if( !SIGNAL_WAIT( &linda->write_happened, &K->lock_, timeout))
                                 {
                                         if( s)
@@ -474,12 +508,6 @@ LUAG_FUNC( linda_receive)
                                         s->status = prev_status;
                                 }
                         }
-#else
-                        // Release the K lock for the duration of wait, and re-acquire
-                        //
-                        if( !SIGNAL_WAIT( &linda->write_happened, &K->lock_, timeout))
-                                break;
-#endif
                 }
                 keeper_release( K);
         }
@@ -501,6 +529,7 @@ LUAG_FUNC( linda_receive)
 * = linda_set( linda_ud, key_num|str|bool|lightuserdata [,value] )
 *
 * Set a value to Linda.
+* TODO: what do we do if we set to non-nil and limit is 0?
 *
 * Existing slot value is replaced, and possible queue entries removed.
 */
@@ -510,9 +539,14 @@ LUAG_FUNC( linda_set)
         bool_t has_value = !lua_isnil( L, 3);
         luaL_argcheck( L, linda, 1, "expected a linda object!");
 
+        // make sure the key is of a valid type
+        check_key_types( L, 2, 2);
+
         {
+                int pushed;
                 struct s_Keeper *K = keeper_acquire( linda);
-                int pushed = keeper_call( K->L, "set", L, linda, 2);
+                // no nil->sentinel toggling, we really clear the linda contents for the given key with a set()
+                pushed = keeper_call( K->L, "set", L, linda, 2);
                 if( pushed >= 0) // no error?
                 {
                         ASSERT_L( pushed == 0);
@@ -537,28 +571,28 @@ LUAG_FUNC( linda_set)
 
 
 /*
-* [val]= linda_keys( linda_ud)
-*
-* Get the list of keys with pending data in the linda
-*/
-LUAG_FUNC( linda_keys)
+ * [val] = linda_count( linda_ud, [key [, ...]])
+ *
+ * Get a count of the pending elements in the specified keys
+ */
+LUAG_FUNC( linda_count)
 {
         struct s_Linda *linda= lua_toLinda( L, 1);
         int pushed;
+
         luaL_argcheck( L, linda, 1, "expected a linda object!");
+        // make sure the keys are of a valid type
+        check_key_types( L, 2, lua_gettop( L));
 
         {
                 struct s_Keeper *K = keeper_acquire( linda);
-                pushed = keeper_call( K->L, "keys", L, linda, 2);
-                ASSERT_L( pushed==0 || pushed==1 );
-                keeper_release(K);
-                // must trigger error after keeper state has been released
+                pushed = keeper_call( K->L, "count", L, linda, 2);
+                keeper_release( K);
                 if( pushed < 0)
                 {
-                        luaL_error( L, "tried to copy unsupported types");
+                        luaL_error( L, "tried to count an invalid key");
                 }
         }
-
         return pushed;
 }
 
@@ -567,17 +601,25 @@ LUAG_FUNC( linda_keys)
 * [val]= linda_get( linda_ud, key_num|str|bool|lightuserdata )
 *
 * Get a value from Linda.
+* TODO: add support to get multiple values?
 */
 LUAG_FUNC( linda_get)
 {
         struct s_Linda *linda= lua_toLinda( L, 1);
         int pushed;
+
         luaL_argcheck( L, linda, 1, "expected a linda object!");
+        // make sure the key is of a valid type
+        check_key_types( L, 2, 2);
 
         {
                 struct s_Keeper *K = keeper_acquire( linda);
                 pushed = keeper_call( K->L, "get", L, linda, 2);
                 ASSERT_L( pushed==0 || pushed==1 );
+                if( pushed > 0)
+                {
+                        keeper_toggle_nil_sentinels( L, lua_gettop( L) - pushed, 0);
+                }
                 keeper_release(K);
                 // must trigger error after keeper state has been released
                 if( pushed < 0)
@@ -598,7 +640,10 @@ LUAG_FUNC( linda_get)
 LUAG_FUNC( linda_limit)
 {
         struct s_Linda *linda= lua_toLinda( L, 1 );
+
         luaL_argcheck( L, linda, 1, "expected a linda object!");
+        // make sure the key is of a valid type
+        check_key_types( L, 2, 2);
 
         {
                 struct s_Keeper *K = keeper_acquire( linda);
@@ -775,21 +820,24 @@ static void linda_id( lua_State *L, char const * const which)
         lua_pushcfunction( L, LG_linda_receive );
         lua_setfield( L, -2, "receive" );
 
-        lua_pushcfunction( L, LG_linda_keys );
-        lua_setfield( L, -2, "keys" );
-
         lua_pushcfunction( L, LG_linda_limit );
         lua_setfield( L, -2, "limit" );
 
         lua_pushcfunction( L, LG_linda_set );
         lua_setfield( L, -2, "set" );
     
+        lua_pushcfunction( L, LG_linda_count );
+        lua_setfield( L, -2, "count" );
+    
         lua_pushcfunction( L, LG_linda_get );
         lua_setfield( L, -2, "get" );
 
         lua_pushcfunction( L, LG_linda_deep );
         lua_setfield( L, -2, "deep" );
 
+        lua_pushliteral( L, BATCH_SENTINEL);
+        lua_setfield(L, -2, "batched");
+
         STACK_END(L,1)
     }
     else if( strcmp( which, "module") == 0)
@@ -803,6 +851,11 @@ static void linda_id( lua_State *L, char const * const which)
     }
 }
 
+/*
+ * ud = lanes.linda()
+ *
+ * returns a linda object
+ */
 LUAG_FUNC( linda)
 {
         return luaG_deep_userdata( L, linda_id);