7 files changed, 66 insertions, 44 deletions

diff --git a/CHANGES b/CHANGES
index dd0917f..e7a08f7 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,5 +1,8 @@
 CHANGES:
 
+CHANGE 100: BGe 17-Feb-14
+   * lanes.linda() accepts an optional integer group to give control on keeper state repartition
+
 CHANGE 99: BGe 17-Feb-14
    * version 3.9.0
    * keepers now require "package", receive package.path & package.cpath, and call on_state_create() if it is a C function
diff --git a/docs/index.html b/docs/index.html
index 051f9d0..c6d612f 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -70,7 +70,7 @@
                         </p>
 
                         <p>
-                                This document was revised on 17-Feb-14, and applies to version <tt>3.9.0</tt>.
+                                This document was revised on 17-Feb-14, and applies to version <tt>3.9.1</tt>.
                         </p>
                 </font>
         </center>
@@ -1029,11 +1029,12 @@
                 <li><tt>receive</tt> has a batched mode to consume more than one value from a single key, as in <tt>linda:receive( 1.0, linda.batched, "key", 3, 6).</tt></li>
                 <li>individual keys' queue length can be limited, balancing speed differences in a producer/consumer scenario (making <tt>:send</tt> wait).</li>
                 <li><tt>tostring( linda)</tt> returns a string of the form <tt>"Linda: &lt;opt_name&gt;"</tt></li>
+                <li>several lindas may share the same keeper state. Since version 3.9.1, state assignation can be controlled with the linda's group (an integer). All lindas belonging to the same group will share the same keeper state. One keeper state may be shared by several groups.</li>
         </ul>
 </p>
 
 <table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
-        h = lanes.linda( [opt_name])
+        h = lanes.linda( [opt_name, [opt_group]])
 
         [true|lanes.cancel_error] = h:send( [timeout_secs,] key, ...)
 
@@ -1171,7 +1172,7 @@
         <br/>
         If the thread is woken but the condition is not yet fulfilled, it goes back to sleep, until the timeout expires.
         <br/>
-        When a lane is cancelled, the signal it is waiting on (if any) is signalled. In that case, the linda operation will return no data.
+        When a lane is cancelled, the signal it is waiting on (if any) is signalled. In that case, the linda operation will return <tt>lanes.cancel_error</tt>.
 </p>
 
 <p>
diff --git a/src/keeper.c b/src/keeper.c
index 4eb8bb3..9b77c75 100644
--- a/src/keeper.c
+++ b/src/keeper.c
@@ -186,9 +186,9 @@ static void push_table( lua_State* L, int idx)
         STACK_END( L, 1);
 }
 
-int keeper_push_linda_storage( lua_State* L, void* ptr)
+int keeper_push_linda_storage( lua_State* L, void* ptr, unsigned long magic_)
 {
-        struct s_Keeper* K = keeper_acquire( ptr);
+        struct s_Keeper* K = keeper_acquire( magic_);
         lua_State* KL = K ? K->L : NULL;
         if( KL == NULL) return 0;
         STACK_GROW( KL, 4);
@@ -685,7 +685,7 @@ char const* init_keepers( lua_State* L)
         return NULL; // ok
 }
 
-struct s_Keeper* keeper_acquire( void const* ptr)
+struct s_Keeper* keeper_acquire( unsigned long magic_)
 {
         // can be 0 if this happens during main state shutdown (lanes is being GC'ed -> no keepers)
         if( GNbKeepers == 0)
@@ -701,7 +701,7 @@ struct s_Keeper* keeper_acquire( void const* ptr)
                 * Pointers are often aligned by 8 or so - ignore the low order bits
                 * have to cast to unsigned long to avoid compilation warnings about loss of data when converting pointer-to-integer
                 */
-                unsigned int i = (unsigned int)(((unsigned long)(ptr) >> 3) % GNbKeepers);
+                unsigned int i = (unsigned int)((magic_ >> KEEPER_MAGIC_SHIFT) % GNbKeepers);
                 struct s_Keeper* K= &GKeepers[i];
 
                 MUTEX_LOCK( &K->lock_);
diff --git a/src/keeper.h b/src/keeper.h
index c0fd5d0..ed11ec4 100644
--- a/src/keeper.h
+++ b/src/keeper.h
@@ -18,10 +18,11 @@ char const* init_keepers( lua_State* L);
 void close_keepers( void);
 #endif // HAVE_KEEPER_ATEXIT_DESINIT
 
-struct s_Keeper *keeper_acquire( const void *ptr);
-void keeper_release( struct s_Keeper *K);
+struct s_Keeper *keeper_acquire( unsigned long magic_);
+#define KEEPER_MAGIC_SHIFT 3
+void keeper_release( struct s_Keeper* K);
 void keeper_toggle_nil_sentinels( lua_State *L, int _val_i, enum eLookupMode const mode_);
-int keeper_push_linda_storage( lua_State* L, void* ptr);
+int keeper_push_linda_storage( lua_State* L, void* ptr, unsigned long magic_);
 
 typedef lua_CFunction keeper_api_t;
 #define KEEPER_API( _op) keepercall_ ## _op
diff --git a/src/lanes.c b/src/lanes.c
index 597ac4b..71a9000 100644
--- a/src/lanes.c
+++ b/src/lanes.c
@@ -419,8 +419,10 @@ struct s_Linda
         SIGNAL_T read_happened;
         SIGNAL_T write_happened;
         enum e_cancel_request simulate_cancel;
+        unsigned long group;
         char name[1];
 };
+#define LINDA_KEEPER_HASHSEED( linda) (linda->group ? linda->group : (unsigned long)linda)
 
 static void* linda_id( lua_State*, enum eDeepOp);
 
@@ -489,7 +491,7 @@ LUAG_FUNC( linda_send)
         {
                 bool_t try_again = TRUE;
                 struct s_lane* const s = get_lane_from_registry( L);
-                struct s_Keeper* K = keeper_acquire( linda);
+                struct s_Keeper* K = keeper_acquire( 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);
@@ -652,7 +654,7 @@ LUAG_FUNC( linda_receive)
         {
                 bool_t try_again = TRUE;
                 struct s_lane* const s = get_lane_from_registry( L);
-                struct s_Keeper* K = keeper_acquire( linda);
+                struct s_Keeper* K = keeper_acquire( LINDA_KEEPER_HASHSEED( linda));
                 if( K == NULL) return 0;
                 for( ;;)
                 {
@@ -755,7 +757,7 @@ LUAG_FUNC( linda_set)
         check_key_types( L, 2, 2);
 
         {
-                struct s_Keeper* K = keeper_acquire( linda);
+                struct s_Keeper* K = keeper_acquire( LINDA_KEEPER_HASHSEED( linda));
                 if( K == NULL) return 0;
 
                 if( linda->simulate_cancel == CANCEL_NONE)
@@ -811,7 +813,7 @@ LUAG_FUNC( linda_count)
         check_key_types( L, 2, lua_gettop( L));
 
         {
-                struct s_Keeper* K = keeper_acquire( linda);
+                struct s_Keeper* K = keeper_acquire( LINDA_KEEPER_HASHSEED( linda));
                 if( K == NULL) return 0;
                 pushed = keeper_call( K->L, KEEPER_API( count), L, linda, 2);
                 keeper_release( K);
@@ -840,7 +842,7 @@ LUAG_FUNC( linda_get)
         // make sure the key is of a valid type (throws an error if not the case)
         check_key_types( L, 2, 2);
         {
-                struct s_Keeper* K = keeper_acquire( linda);
+                struct s_Keeper* K = keeper_acquire( LINDA_KEEPER_HASHSEED( linda));
                 if( K == NULL) return 0;
 
                 if( linda->simulate_cancel == CANCEL_NONE)
@@ -890,7 +892,7 @@ LUAG_FUNC( linda_limit)
         check_key_types( L, 2, 2);
 
         {
-                struct s_Keeper* K = keeper_acquire( linda);
+                struct s_Keeper* K = keeper_acquire( LINDA_KEEPER_HASHSEED( linda));
                 if( K == NULL) return 0;
 
                 if( linda->simulate_cancel == CANCEL_NONE)
@@ -931,7 +933,7 @@ LUAG_FUNC( linda_cancel)
         luaL_argcheck( L, lua_gettop( L) <= 2, 2, "wrong number of arguments");
 
         // signalling must be done from inside the K locking area
-        K = keeper_acquire( linda);
+        K = keeper_acquire( LINDA_KEEPER_HASHSEED( linda));
         if( K == NULL) return 0;
 
         linda->simulate_cancel = CANCEL_SOFT;
@@ -1057,7 +1059,7 @@ LUAG_FUNC( linda_concat)
 LUAG_FUNC( linda_dump)
 {
         struct s_Linda* linda = lua_toLinda( L, 1);
-        return keeper_push_linda_storage( L, linda);
+        return keeper_push_linda_storage( L, linda, LINDA_KEEPER_HASHSEED( linda));
 }
 
 /*
@@ -1093,11 +1095,28 @@ static void* linda_id( lua_State* L, enum eDeepOp op_)
                         struct s_Linda* s;
                         size_t name_len = 0;
                         char const* linda_name = NULL;
-                        int const top = lua_gettop( L);
-
-                        if( top > 0 && lua_type( L, top) == LUA_TSTRING)
+                        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))
                         {
-                                linda_name = lua_tolstring( L, top, &name_len);
+                                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 = lua_tointeger( L, -1);
+                                break;
                         }
 
                         /* The deep data is allocated separately of Lua stack; we might no
@@ -1110,6 +1129,7 @@ static void* linda_id( lua_State* L, enum eDeepOp op_)
                         SIGNAL_INIT( &s->read_happened);
                         SIGNAL_INIT( &s->write_happened);
                         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;
@@ -1118,24 +1138,24 @@ static void* linda_id( lua_State* L, enum eDeepOp op_)
                 case eDO_delete:
                 {
                         struct s_Keeper* K;
-                        struct s_Linda* l = lua_touserdata( L, 1);
-                        ASSERT_L( l);
+                        struct s_Linda* linda = lua_touserdata( L, 1);
+                        ASSERT_L( linda);
 
                         /* Clean associated structures in the keeper state.
                         */
-                        K = keeper_acquire( l);
+                        K = keeper_acquire( 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)
                         {
-                                keeper_call( K->L, KEEPER_API( clear), L, l, 0);
+                                keeper_call( 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( &l->read_happened);
-                        SIGNAL_FREE( &l->write_happened);
-                        free( l);
+                        SIGNAL_FREE( &linda->read_happened);
+                        SIGNAL_FREE( &linda->write_happened);
+                        free( linda);
                         return NULL;
                 }
 
@@ -1209,16 +1229,24 @@ static void* linda_id( lua_State* L, enum eDeepOp op_)
 }
 
 /*
- * ud = lanes.linda()
+ * ud = lanes.linda( [name[,group]])
  *
  * returns a linda object
  */
 LUAG_FUNC( linda)
 {
         int const top = lua_gettop( L);
-        luaL_argcheck( L, top <= 1, top, "too many arguments");
+        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);
 }
 
diff --git a/src/lanes.lua b/src/lanes.lua
index 86dbe47..5dfe41d 100644
--- a/src/lanes.lua
+++ b/src/lanes.lua
@@ -306,17 +306,6 @@ local function gen( ... )
     end
 end
 
----=== Lindas ===---
-
--- We let the C code attach methods to userdata directly
-
------
--- lanes.linda(["name"]) -> linda_ud
---
--- PUBLIC LANES API
-local linda = core.linda
-
-
 ---=== Timers ===---
 
 -- PUBLIC LANES API
diff --git a/tests/keeper.lua b/tests/keeper.lua
index 73ed3cf..4aff51c 100644
--- a/tests/keeper.lua
+++ b/tests/keeper.lua
@@ -19,10 +19,10 @@ local function keeper(linda)
 end
 
 --
-local lindaA= lanes.linda()
+local lindaA= lanes.linda( "A", 1)
 local A= keeper( lindaA )
 
-local lindaB= lanes.linda()
+local lindaB= lanes.linda( "B", 2)
 local B= keeper( lindaB )
 
 A.some= 1