1 files changed, 945 insertions, 0 deletions

diff --git a/src/linda.cpp b/src/linda.cpp
new file mode 100644
index 0000000..eac6458
--- /dev/null
+++ b/src/linda.cpp
@@ -0,0 +1,945 @@
+/*
+ * LINDA.C                          Copyright (c) 2018, Benoit Germain
+ *
+ * Linda deep userdata.
+*/
+
+/*
+===============================================================================
+
+Copyright (C) 2018 benoit Germain <bnt.germain@gmail.com>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+
+===============================================================================
+*/
+
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+
+#include "threading.h"
+#include "compat.h"
+#include "tools.h"
+#include "universe.h"
+#include "keeper.h"
+#include "deep.h"
+#include "lanes_private.h"
+
+/*
+* Actual data is kept within a keeper state, which is hashed by the 's_Linda'
+* pointer (which is same to all userdatas pointing to it).
+*/
+struct s_Linda
+{
+    DeepPrelude prelude; // Deep userdata MUST start with this header
+    SIGNAL_T read_happened;
+    SIGNAL_T write_happened;
+    Universe* U; // the universe this linda belongs to
+    ptrdiff_t group; // a group to control keeper allocation between lindas
+    enum e_cancel_request simulate_cancel;
+    char name[1];
+};
+#define LINDA_KEEPER_HASHSEED( linda) (linda->group ? linda->group : (ptrdiff_t)linda)
+
+static void* linda_id( lua_State*, DeepOp);
+
+static inline struct s_Linda* lua_toLinda( lua_State* L, int idx_)
+{
+    struct s_Linda* linda = (struct s_Linda*) luaG_todeep( L, linda_id, idx_);
+    luaL_argcheck( L, linda != NULL, idx_, "expecting a linda object");
+    return linda;
+}
+
+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;
+        }
+        (void) luaL_error( L, "argument #%d: invalid key type (not a boolean, string, number or light userdata)", i);
+    }
+}
+
+LUAG_FUNC( linda_protected_call)
+{
+    int rc = LUA_OK;
+    struct s_Linda* linda = lua_toLinda( L, 1);
+
+    // acquire the keeper
+    Keeper* K = keeper_acquire( linda->U->keepers, LINDA_KEEPER_HASHSEED(linda));
+    lua_State* KL = K ? K->L : NULL; // need to do this for 'STACK_CHECK'
+    if( KL == NULL) return 0;
+
+    // retrieve the actual function to be called and move it before the arguments
+    lua_pushvalue( L, lua_upvalueindex( 1));
+    lua_insert( L, 1);
+    // do a protected call
+    rc = lua_pcall( L, lua_gettop( L) - 1, LUA_MULTRET, 0);
+
+    // release the keeper
+    keeper_release( K);
+
+    // if there was an error, forward it
+    if( rc != LUA_OK)
+    {
+        return lua_error( L);
+    }
+    // return whatever the actual operation provided
+    return lua_gettop( L);
+}
+
+/*
+* bool= linda_send( linda_ud, [timeout_secs=-1,] [linda.null,] key_num|str|bool|lightuserdata, ... )
+*
+* Send one or more values to a Linda. If there is a limit, all values must fit.
+*
+* Returns:  'true' if the value was queued
+*           'false' for timeout (only happens when the queue size is limited)
+*           nil, CANCEL_ERROR if cancelled
+*/
+LUAG_FUNC( linda_send)
+{
+    struct s_Linda* linda = lua_toLinda( L, 1);
+    bool_t ret = FALSE;
+    enum e_cancel_request cancel = CANCEL_NONE;
+    int pushed;
+    time_d timeout = -1.0;
+    uint_t key_i = 2; // index of first key, if timeout not there
+    bool_t as_nil_sentinel; // if not NULL, send() will silently send a single nil if nothing is provided
+
+    if( lua_type( L, 2) == LUA_TNUMBER) // we don't want to use lua_isnumber() because of autocoercion
+    {
+        timeout = SIGNAL_TIMEOUT_PREPARE( lua_tonumber( L, 2));
+        ++ key_i;
+    }
+    else if( lua_isnil( L, 2)) // alternate explicit "no timeout" by passing nil before the key
+    {
+        ++ key_i;
+    }
+
+    as_nil_sentinel = equal_unique_key( L, key_i, NIL_SENTINEL);
+    if( as_nil_sentinel)
+    {
+        // the real key to send data to is after the NIL_SENTINEL marker
+        ++ key_i;
+    }
+
+    // make sure the key is of a valid type
+    check_key_types( L, key_i, key_i);
+
+    STACK_GROW( L, 1);
+
+    // make sure there is something to send
+    if( (uint_t)lua_gettop( L) == key_i)
+    {
+        if( as_nil_sentinel)
+        {
+            // send a single nil if nothing is provided
+            push_unique_key( L, NIL_SENTINEL);
+        }
+        else
+        {
+            return luaL_error( L, "no data to send");
+        }
+    }
+
+    // convert nils to some special non-nil sentinel in sent values
+    keeper_toggle_nil_sentinels( L, key_i + 1, eLM_ToKeeper);
+
+    {
+        bool_t try_again = TRUE;
+        Lane* const s = get_lane_from_registry( L);
+        Keeper* K = which_keeper( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+        lua_State* KL = K ? K->L : NULL; // need to do this for 'STACK_CHECK'
+        if( KL == NULL) return 0;
+        STACK_CHECK( KL, 0);
+        for( ;;)
+        {
+            if( s != NULL)
+            {
+                cancel = s->cancel_request;
+            }
+            cancel = (cancel != CANCEL_NONE) ? cancel : linda->simulate_cancel;
+            // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
+            if( !try_again || cancel != CANCEL_NONE)
+            {
+                pushed = 0;
+                break;
+            }
+
+            STACK_MID( KL, 0);
+            pushed = keeper_call( linda->U, KL, KEEPER_API( send), L, linda, key_i);
+            if( pushed < 0)
+            {
+                break;
+            }
+            ASSERT_L( pushed == 1);
+
+            ret = lua_toboolean( L, -1);
+            lua_pop( L, 1);
+
+            if( ret)
+            {
+                // Wake up ALL waiting threads
+                SIGNAL_ALL( &linda->write_happened);
+                break;
+            }
+
+            // instant timout to bypass the wait syscall
+            if( timeout == 0.0)
+            {
+                break;  /* no wait; instant timeout */
+            }
+
+            // storage limit hit, wait until timeout or signalled that we should try again
+            {
+                enum e_status prev_status = ERROR_ST; // prevent 'might be used uninitialized' warnings
+                if( s != NULL)
+                {
+                    // change status of lane to "waiting"
+                    prev_status = s->status; // RUNNING, most likely
+                    ASSERT_L( prev_status == RUNNING); // but check, just in case
+                    s->status = WAITING;
+                    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
+                try_again = SIGNAL_WAIT( &linda->read_happened, &K->keeper_cs, timeout);
+                if( s != NULL)
+                {
+                    s->waiting_on = NULL;
+                    s->status = prev_status;
+                }
+            }
+        }
+        STACK_END( KL, 0);
+    }
+
+    if( pushed < 0)
+    {
+        return luaL_error( L, "tried to copy unsupported types");
+    }
+
+    switch( cancel)
+    {
+        case CANCEL_SOFT:
+        // if user wants to soft-cancel, the call returns lanes.cancel_error
+        push_unique_key( L, CANCEL_ERROR);
+        return 1;
+
+        case CANCEL_HARD:
+        // raise an error interrupting execution only in case of hard cancel
+        return cancel_error( L); // raises an error and doesn't return
+
+        default:
+        lua_pushboolean( L, ret); // true (success) or false (timeout)
+        return 1;
+    }
+}
+
+
+/*
+ * 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, min_COUNT[, max_COUNT])
+ * Consumes between min_COUNT and max_COUNT values from the linda, from a single key.
+ * returns the actual 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, expected_pushed_min, expected_pushed_max;
+    enum e_cancel_request cancel = CANCEL_NONE;
+    keeper_api_t keeper_receive;
+    
+    time_d timeout = -1.0;
+    uint_t key_i = 2;
+
+    if( lua_type( L, 2) == LUA_TNUMBER) // we don't want to use lua_isnumber() because of autocoercion
+    {
+        timeout = SIGNAL_TIMEOUT_PREPARE( lua_tonumber( L, 2));
+        ++ key_i;
+    }
+    else if( lua_isnil( L, 2)) // alternate explicit "no timeout" by passing nil before the key
+    {
+        ++ key_i;
+    }
+
+    // are we in batched mode?
+    {
+        int is_batched;
+        lua_pushliteral( L, BATCH_SENTINEL);
+        is_batched = lua501_equal( L, key_i, -1);
+        lua_pop( L, 1);
+        if( is_batched)
+        {
+            // no need to pass linda.batched in the keeper state
+            ++ key_i;
+            // make sure the keys are of a valid type
+            check_key_types( L, key_i, key_i);
+            // receive multiple values from a single slot
+            keeper_receive = KEEPER_API( receive_batched);
+            // we expect a user-defined amount of return value
+            expected_pushed_min = (int)luaL_checkinteger( L, key_i + 1);
+            expected_pushed_max = (int)luaL_optinteger( L, key_i + 2, expected_pushed_min);
+            // don't forget to count the key in addition to the values
+            ++ expected_pushed_min;
+            ++ expected_pushed_max;
+            if( expected_pushed_min > expected_pushed_max)
+            {
+                return luaL_error( L, "batched min/max error");
+            }
+        }
+        else
+        {
+            // make sure the keys are of a valid type
+            check_key_types( L, key_i, lua_gettop( L));
+            // receive a single value, checking multiple slots
+            keeper_receive = KEEPER_API( receive);
+            // we expect a single (value, key) pair of returned values
+            expected_pushed_min = expected_pushed_max = 2;
+        }
+    }
+
+    {
+        bool_t try_again = TRUE;
+        Lane* const s = get_lane_from_registry( L);
+        Keeper* K = which_keeper( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+        if( K == NULL) return 0;
+        for( ;;)
+        {
+            if( s != NULL)
+            {
+                cancel = s->cancel_request;
+            }
+            cancel = (cancel != CANCEL_NONE) ? cancel : linda->simulate_cancel;
+            // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
+            if( !try_again || cancel != CANCEL_NONE)
+            {
+                pushed = 0;
+                break;
+            }
+
+            // all arguments of receive() but the first are passed to the keeper's receive function
+            pushed = keeper_call( linda->U, K->L, keeper_receive, L, linda, key_i);
+            if( pushed < 0)
+            {
+                break;
+            }
+            if( pushed > 0)
+            {
+                ASSERT_L( pushed >= expected_pushed_min && pushed <= expected_pushed_max);
+                // replace sentinels with real nils
+                keeper_toggle_nil_sentinels( L, lua_gettop( L) - pushed, eLM_FromKeeper);
+                // To be done from within the 'K' locking area
+                //
+                SIGNAL_ALL( &linda->read_happened);
+                break;
+            }
+
+            if( timeout == 0.0)
+            {
+                break;  /* instant timeout */
+            }
+
+            // nothing received, wait until timeout or signalled that we should try again
+            {
+                enum e_status prev_status = ERROR_ST; // prevent 'might be used uninitialized' warnings
+                if( s != NULL)
+                {
+                    // change status of lane to "waiting"
+                    prev_status = s->status; // RUNNING, most likely
+                    ASSERT_L( prev_status == RUNNING); // but check, just in case
+                    s->status = WAITING;
+                    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
+                try_again = SIGNAL_WAIT( &linda->write_happened, &K->keeper_cs, timeout);
+                if( s != NULL)
+                {
+                    s->waiting_on = NULL;
+                    s->status = prev_status;
+                }
+            }
+        }
+    }
+
+    if( pushed < 0)
+    {
+        return luaL_error( L, "tried to copy unsupported types");
+    }
+
+    switch( cancel)
+    {
+        case CANCEL_SOFT:
+        // if user wants to soft-cancel, the call returns CANCEL_ERROR
+        push_unique_key( L, CANCEL_ERROR);
+        return 1;
+
+        case CANCEL_HARD:
+        // raise an error interrupting execution only in case of hard cancel
+        return cancel_error( L); // raises an error and doesn't return
+
+        default:
+        return pushed;
+    }
+}
+
+
+/*
+* [true|lanes.cancel_error] = linda_set( linda_ud, key_num|str|bool|lightuserdata [, value [, ...]])
+*
+* Set one or more 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 queued entries removed.
+*/
+LUAG_FUNC( linda_set)
+{
+    struct s_Linda* const linda = lua_toLinda( L, 1);
+    int pushed;
+    bool_t has_value = lua_gettop( L) > 2;
+
+    // make sure the key is of a valid type (throws an error if not the case)
+    check_key_types( L, 2, 2);
+
+    {
+        Keeper* K = which_keeper( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+
+        if( linda->simulate_cancel == CANCEL_NONE)
+        {
+            if( has_value)
+            {
+                // convert nils to some special non-nil sentinel in sent values
+                keeper_toggle_nil_sentinels( L, 3, eLM_ToKeeper);
+            }
+            pushed = keeper_call( linda->U, K->L, KEEPER_API( set), L, linda, 2);
+            if( pushed >= 0) // no error?
+            {
+                ASSERT_L( pushed == 0 || pushed == 1);
+
+                if( has_value)
+                {
+                    // we put some data in the slot, tell readers that they should wake
+                    SIGNAL_ALL( &linda->write_happened); // To be done from within the 'K' locking area
+                }
+                if( pushed == 1)
+                {
+                    // the key was full, but it is no longer the case, tell writers they should wake
+                    ASSERT_L( lua_type( L, -1) == LUA_TBOOLEAN && lua_toboolean( L, -1) == 1);
+                    SIGNAL_ALL( &linda->read_happened); // To be done from within the 'K' locking area
+                }
+            }
+        }
+        else // linda is cancelled
+        {
+            // do nothing and return lanes.cancel_error
+            push_unique_key( L, CANCEL_ERROR);
+            pushed = 1;
+        }
+    }
+
+    // must trigger any error after keeper state has been released
+    return (pushed < 0) ? luaL_error( L, "tried to copy unsupported types") : pushed;
+}
+
+
+/*
+ * [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;
+
+    // make sure the keys are of a valid type
+    check_key_types( L, 2, lua_gettop( L));
+
+    {
+        Keeper* K = which_keeper( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+        pushed = keeper_call( linda->U, K->L, KEEPER_API( count), L, linda, 2);
+        if( pushed < 0)
+        {
+            return luaL_error( L, "tried to count an invalid key");
+        }
+    }
+    return pushed;
+}
+
+
+/*
+* [val [, ...]] = linda_get( linda_ud, key_num|str|bool|lightuserdata [, count = 1])
+*
+* Get one or more values from Linda.
+*/
+LUAG_FUNC( linda_get)
+{
+    struct s_Linda* const linda = lua_toLinda( L, 1);
+    int pushed;
+    lua_Integer count = luaL_optinteger( L, 3, 1);
+    luaL_argcheck( L, count >= 1, 3, "count should be >= 1");
+    luaL_argcheck( L, lua_gettop( L) <= 3, 4, "too many arguments");
+
+    // make sure the key is of a valid type (throws an error if not the case)
+    check_key_types( L, 2, 2);
+    {
+        Keeper* K = which_keeper( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+
+        if( linda->simulate_cancel == CANCEL_NONE)
+        {
+            pushed = keeper_call( linda->U, K->L, KEEPER_API( get), L, linda, 2);
+            if( pushed > 0)
+            {
+                keeper_toggle_nil_sentinels( L, lua_gettop( L) - pushed, eLM_FromKeeper);
+            }
+        }
+        else // linda is cancelled
+        {
+            // do nothing and return lanes.cancel_error
+            push_unique_key( L, CANCEL_ERROR);
+            pushed = 1;
+        }
+        // an error can be raised if we attempt to read an unregistered function
+        if( pushed < 0)
+        {
+            return luaL_error( L, "tried to copy unsupported types");
+        }
+    }
+
+    return pushed;
+}
+
+
+/*
+* [true] = linda_limit( linda_ud, key_num|str|bool|lightuserdata, int)
+*
+* Set limit to 1 Linda keys.
+* Optionally wake threads waiting to write on the linda, in case the limit enables them to do so
+*/
+LUAG_FUNC( linda_limit)
+{
+    struct s_Linda* linda = lua_toLinda( L, 1);
+    int pushed;
+
+    // make sure we got 3 arguments: the linda, a key and a limit
+    luaL_argcheck( L, lua_gettop( L) == 3, 2, "wrong number of arguments");
+    // make sure we got a numeric limit
+    luaL_checknumber( L, 3);
+    // make sure the key is of a valid type
+    check_key_types( L, 2, 2);
+
+    {
+        Keeper* K = which_keeper( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+
+        if( linda->simulate_cancel == CANCEL_NONE)
+        {
+            pushed = keeper_call( linda->U, K->L, KEEPER_API( limit), L, linda, 2);
+            ASSERT_L( pushed == 0 || pushed == 1); // no error, optional boolean value saying if we should wake blocked writer threads
+            if( pushed == 1)
+            {
+                ASSERT_L( lua_type( L, -1) == LUA_TBOOLEAN && lua_toboolean( L, -1) == 1);
+                SIGNAL_ALL( &linda->read_happened); // To be done from within the 'K' locking area
+            }
+        }
+        else // linda is cancelled
+        {
+            // do nothing and return lanes.cancel_error
+            push_unique_key( L, CANCEL_ERROR);
+            pushed = 1;
+        }
+    }
+    // propagate pushed boolean if any
+    return pushed;
+}
+
+
+/*
+* (void) = linda_cancel( linda_ud, "read"|"write"|"both"|"none")
+*
+* Signal linda so that waiting threads wake up as if their own lane was cancelled
+*/
+LUAG_FUNC( linda_cancel)
+{
+    struct s_Linda* linda = lua_toLinda( L, 1);
+    char const* who = luaL_optstring( L, 2, "both");
+
+    // make sure we got 3 arguments: the linda, a key and a limit
+    luaL_argcheck( L, lua_gettop( L) <= 2, 2, "wrong number of arguments");
+
+    linda->simulate_cancel = CANCEL_SOFT;
+    if( strcmp( who, "both") == 0) // tell everyone writers to wake up
+    {
+        SIGNAL_ALL( &linda->write_happened);
+        SIGNAL_ALL( &linda->read_happened);
+    }
+    else if( strcmp( who, "none") == 0) // reset flag
+    {
+        linda->simulate_cancel = CANCEL_NONE;
+    }
+    else if( strcmp( who, "read") == 0) // tell blocked readers to wake up
+    {
+        SIGNAL_ALL( &linda->write_happened);
+    }
+    else if( strcmp( who, "write") == 0) // tell blocked writers to wake up
+    {
+        SIGNAL_ALL( &linda->read_happened);
+    }
+    else
+    {
+        return luaL_error( L, "unknown wake hint '%s'", who);
+    }
+    return 0;
+}
+
+
+/*
+* lightuserdata= linda_deep( linda_ud )
+*
+* Return the 'deep' userdata pointer, identifying the Linda.
+*
+* This is needed for using Lindas as key indices (timer system needs it);
+* separately created proxies of the same underlying deep object will have
+* different userdata and won't be known to be essentially the same deep one
+* without this.
+*/
+LUAG_FUNC( linda_deep)
+{
+    struct s_Linda* linda= lua_toLinda( L, 1);
+    lua_pushlightuserdata( L, linda); // just the address
+    return 1;
+}
+
+
+/*
+* string = linda:__tostring( linda_ud)
+*
+* Return the stringification of a linda
+*
+* Useful for concatenation or debugging purposes
+*/
+
+static int linda_tostring( lua_State* L, int idx_, bool_t opt_)
+{
+    struct s_Linda* linda = (struct s_Linda*) luaG_todeep( L, linda_id, idx_);
+    if( !opt_)
+    {
+        luaL_argcheck( L, linda, idx_, "expecting a linda object");
+    }
+    if( linda != NULL)
+    {
+        char text[128];
+        int len;
+        if( linda->name[0])
+            len = sprintf( text, "Linda: %.*s", (int)sizeof(text) - 8, linda->name);
+        else
+            len = sprintf( text, "Linda: %p", linda);
+        lua_pushlstring( L, text, len);
+        return 1;
+    }
+    return 0;
+}
+
+LUAG_FUNC( linda_tostring)
+{
+    return linda_tostring( L, 1, FALSE);
+}
+
+
+/*
+* string = linda:__concat( a, b)
+*
+* Return the concatenation of a pair of items, one of them being a linda
+*
+* Useful for concatenation or debugging purposes
+*/
+LUAG_FUNC( linda_concat)
+{                                   // linda1? linda2?
+    bool_t atLeastOneLinda = FALSE;
+    // Lua semantics enforce that one of the 2 arguments is a Linda, but not necessarily both.
+    if( linda_tostring( L, 1, TRUE))
+    {
+        atLeastOneLinda = TRUE;
+        lua_replace( L, 1);
+    }
+    if( linda_tostring( L, 2, TRUE))
+    {
+        atLeastOneLinda = TRUE;
+        lua_replace( L, 2);
+    }
+    if( !atLeastOneLinda) // should not be possible
+    {
+        return luaL_error( L, "internal error: linda_concat called on non-Linda");
+    }
+    lua_concat( L, 2);
+    return 1;
+}
+
+/*
+ * table = linda:dump()
+ * return a table listing all pending data inside the linda
+ */
+LUAG_FUNC( linda_dump)
+{
+    struct s_Linda* linda = lua_toLinda( L, 1);
+    ASSERT_L( linda->U == universe_get( L));
+    return keeper_push_linda_storage( linda->U, L, linda, LINDA_KEEPER_HASHSEED( linda));
+}
+
+/*
+ * table = linda:dump()
+ * return a table listing all pending data inside the linda
+ */
+LUAG_FUNC( linda_towatch)
+{
+    struct s_Linda* linda = lua_toLinda( L, 1);
+    int pushed;
+    ASSERT_L( linda->U == universe_get( L));
+    pushed = keeper_push_linda_storage( linda->U, L, linda, LINDA_KEEPER_HASHSEED( linda));
+    if( pushed == 0)
+    {
+        // if the linda is empty, don't return nil
+        pushed = linda_tostring( L, 1, FALSE);
+    }
+    return pushed;
+}
+
+/*
+* Identity function of a shared userdata object.
+* 
+*   lightuserdata= linda_id( "new" [, ...] )
+*   = linda_id( "delete", lightuserdata )
+*
+* Creation and cleanup of actual 'deep' objects. 'luaG_...' will wrap them into
+* regular userdata proxies, per each state using the deep data.
+*
+*   tbl= linda_id( "metatable" )
+*
+* Returns a metatable for the proxy objects ('__gc' method not needed; will
+* be added by 'luaG_...')
+*
+*   string= linda_id( "module")
+*
+* Returns the name of the module that a state should require
+* in order to keep a handle on the shared library that exported the idfunc
+*
+*   = linda_id( str, ... )
+*
+* For any other strings, the ID function must not react at all. This allows
+* future extensions of the system. 
+*/
+static void* linda_id( lua_State* L, DeepOp op_)
+{
+    switch( op_)
+    {
+        case eDO_new:
+        {
+            struct s_Linda* s;
+            size_t name_len = 0;
+            char const* linda_name = NULL;
+            unsigned long linda_group = 0;
+            // should have a string and/or a number of the stack as parameters (name and group)
+            switch( lua_gettop( L))
+            {
+                default: // 0
+                break;
+
+                case 1: // 1 parameter, either a name or a group
+                if( lua_type( L, -1) == LUA_TSTRING)
+                {
+                    linda_name = lua_tolstring( L, -1, &name_len);
+                }
+                else
+                {
+                    linda_group = (unsigned long) lua_tointeger( L, -1);
+                }
+                break;
+
+                case 2: // 2 parameters, a name and group, in that order
+                linda_name = lua_tolstring( L, -2, &name_len);
+                linda_group = (unsigned long) lua_tointeger( L, -1);
+                break;
+            }
+
+            /* The deep data is allocated separately of Lua stack; we might no
+            * longer be around when last reference to it is being released.
+            * One can use any memory allocation scheme.
+            * just don't use L's allocF because we don't know which state will get the honor of GCing the linda
+            */
+            {
+                Universe* const U = universe_get(L);
+                AllocatorDefinition* const allocD = &U->internal_allocator;
+                s = (struct s_Linda*) allocD->allocF(allocD->allocUD, NULL, 0, sizeof(struct s_Linda) + name_len); // terminating 0 is already included
+            }
+            if( s)
+            {
+                s->prelude.magic.value = DEEP_VERSION.value;
+                SIGNAL_INIT( &s->read_happened);
+                SIGNAL_INIT( &s->write_happened);
+                s->U = universe_get( L);
+                s->simulate_cancel = CANCEL_NONE;
+                s->group = linda_group << KEEPER_MAGIC_SHIFT;
+                s->name[0] = 0;
+                memcpy( s->name, linda_name, name_len ? name_len + 1 : 0);
+            }
+            return s;
+        }
+
+        case eDO_delete:
+        {
+            Keeper* K;
+            struct s_Linda* linda = (struct s_Linda*) lua_touserdata( L, 1);
+            ASSERT_L( linda);
+
+            // Clean associated structures in the keeper state.
+            K = keeper_acquire( linda->U->keepers, LINDA_KEEPER_HASHSEED( linda));
+            if( K && K->L) // can be NULL if this happens during main state shutdown (lanes is GC'ed -> no keepers -> no need to cleanup)
+            {
+                // hopefully this won't ever raise an error as we would jump to the closest pcall site while forgetting to release the keeper mutex...
+                keeper_call( linda->U, K->L, KEEPER_API( clear), L, linda, 0);
+            }
+            keeper_release( K);
+
+            // There aren't any lanes waiting on these lindas, since all proxies have been gc'ed. Right?
+            SIGNAL_FREE( &linda->read_happened);
+            SIGNAL_FREE( &linda->write_happened);
+            {
+                Universe* const U = universe_get(L);
+                AllocatorDefinition* const allocD = &U->internal_allocator;
+                (void) allocD->allocF(allocD->allocUD, linda, sizeof(struct s_Linda) + strlen(linda->name), 0);
+            }
+            return NULL;
+        }
+
+        case eDO_metatable:
+        {
+
+            STACK_CHECK( L, 0);
+            lua_newtable( L);
+            // metatable is its own index
+            lua_pushvalue( L, -1);
+            lua_setfield( L, -2, "__index");
+
+            // protect metatable from external access
+            lua_pushliteral( L, "Linda");
+            lua_setfield( L, -2, "__metatable");
+
+            lua_pushcfunction( L, LG_linda_tostring);
+            lua_setfield( L, -2, "__tostring");
+
+            // Decoda __towatch support
+            lua_pushcfunction( L, LG_linda_towatch);
+            lua_setfield( L, -2, "__towatch");
+
+            lua_pushcfunction( L, LG_linda_concat);
+            lua_setfield( L, -2, "__concat");
+
+            // protected calls, to ensure associated keeper is always released even in case of error
+            // all function are the protected call wrapper, where the actual operation is provided as upvalue
+            // note that this kind of thing can break function lookup as we use the function pointer here and there
+
+            lua_pushcfunction( L, LG_linda_send);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "send");
+
+            lua_pushcfunction( L, LG_linda_receive);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "receive");
+
+            lua_pushcfunction( L, LG_linda_limit);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "limit");
+
+            lua_pushcfunction( L, LG_linda_set);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "set");
+
+            lua_pushcfunction( L, LG_linda_count);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "count");
+
+            lua_pushcfunction( L, LG_linda_get);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "get");
+
+            lua_pushcfunction( L, LG_linda_cancel);
+            lua_setfield( L, -2, "cancel");
+
+            lua_pushcfunction( L, LG_linda_deep);
+            lua_setfield( L, -2, "deep");
+
+            lua_pushcfunction( L, LG_linda_dump);
+            lua_pushcclosure( L, LG_linda_protected_call, 1);
+            lua_setfield( L, -2, "dump");
+
+            // some constants
+            lua_pushliteral( L, BATCH_SENTINEL);
+            lua_setfield( L, -2, "batched");
+
+            push_unique_key( L, NIL_SENTINEL);
+            lua_setfield( L, -2, "null");
+
+            STACK_END( L, 1);
+            return NULL;
+        }
+
+        case eDO_module:
+        // linda is a special case because we know lanes must be loaded from the main lua state
+        // to be able to ever get here, so we know it will remain loaded as long a the main state is around
+        // in other words, forever.
+        default:
+        {
+            return NULL;
+        }
+    }
+}
+
+/*
+ * ud = lanes.linda( [name[,group]])
+ *
+ * returns a linda object, or raises an error if creation failed
+ */
+LUAG_FUNC( linda)
+{
+    int const top = lua_gettop( L);
+    luaL_argcheck( L, top <= 2, top, "too many arguments");
+    if( top == 1)
+    {
+        int const t = lua_type( L, 1);
+        luaL_argcheck( L, t == LUA_TSTRING || t == LUA_TNUMBER, 1, "wrong parameter (should be a string or a number)");
+    }
+    else if( top == 2)
+    {
+        luaL_checktype( L, 1, LUA_TSTRING);
+        luaL_checktype( L, 2, LUA_TNUMBER);
+    }
+    return luaG_newdeepuserdata( L, linda_id, 0);
+}