14 files changed, 174 insertions, 174 deletions

diff --git a/CHANGES b/CHANGES
index 7cf28b2..ccfa923 100644
--- a/CHANGES
+++ b/CHANGES
@@ -45,6 +45,7 @@ CHANGE 2: BGe 27-Nov-24
                 - Providing "auto" as name when constructing a Linda cause Lanes to provide a name built from the source location of the construction.
                 - Specifying a group to lanes.linda() is mandatory when Lanes is configured with user Keepers.
             - linda:deep() result no longer contains the raw C pointer of the Linda object.
+            - new function linda:receive_batched() to replace linda:receive(linda.batched). linda.batched special value is removed.
             - linda :receive(), :send(), :get(), :set(), :limit() return nil, error in case of problem. Returned values in case of success change too.
             - linda:limit() can be used to read the value if no new limit is provided.
             - linda:restrict() can restrain the use of send/receive or set/get on any key.
diff --git a/docs/index.html b/docs/index.html
index 8826b57..4dd5848 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -1107,7 +1107,7 @@
         b = f()
         c = f()
 
-        sync_linda:receive(nil, sync_linda.batched, "done", 3) -- wait for 3 lanes to write something in "done" slot of sync_linda
+        sync_linda:receive_batched(nil, "done", 3) -- wait for 3 lanes to write something in "done" slot of sync_linda
 </pre></td></tr></table>
 
 <!-- cancelling +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ -->
@@ -1260,7 +1260,7 @@
                 <li>Two producer-side methods: <tt>:send</tt> and <tt>:set</tt> (not out).</li>
                 <li><tt>send</tt> allows for sending multiple values -atomically- to a given slot.</li>
                 <li><tt>receive</tt> can wait for multiple slots at once.</li>
-                <li><tt>receive</tt> has a batched mode to consume more than one value from a single slot, as in <tt>linda:receive(1.0, linda.batched, "slot", 3, 6).</tt></li>
+                <li><tt>receive_batched</tt> can be used to consume more than one value from a single slot, as in <tt>linda:receive_batched(1.0, "slot", 3, 6).</tt></li>
                 <li><tt>restrict</tt> can restrain a particular slot to function either with <tt>send/receive</tt> or <tt>set/get</tt>.</li>
                 <li>Individual slots' 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>
@@ -1349,12 +1349,12 @@
 <table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
         slot, val = h:receive([timeout_secs,] slot [, slot...])
 
-        slot, val [, val...] = h:receive([timeout,] h.batched, slot, n_uint_min[, n_uint_max])
+        slot, val [, val...] = h:receive_batched([timeout,] slot, n_uint_min[, n_uint_max])
 </pre></td></tr></table>
 
 <p>
-        <tt>receive()</tt> raises an error if called when a restriction forbids its use on any provided slot.<br />
-        In batched mode, <tt>receive()</tt> will raise an error if <tt>min_count < 1</tt> or <tt>max_count < min_count</tt>.
+        <tt>receive()</tt> and <tt>receive_batched()</tt> raise an error if called when a restriction forbids their use on any provided slot.<br />
+        <tt>receive_batched()</tt> will raise an error if <tt>min_count < 1</tt> or <tt>max_count < min_count</tt>.
 </p>
 
 <p>
diff --git a/src/intercopycontext.cpp b/src/intercopycontext.cpp
index 93a8160..a93615b 100644
--- a/src/intercopycontext.cpp
+++ b/src/intercopycontext.cpp
@@ -994,7 +994,7 @@ bool InterCopyContext::interCopyLightuserdata() const
     // recognize and print known UniqueKey names here
     if constexpr (USE_DEBUG_SPEW()) {
         bool _found{ false };
-        static constexpr std::array<std::reference_wrapper<UniqueKey const>, 3> kKeysToCheck{ kLindaBatched, kCancelError, kNilSentinel };
+        static constexpr std::array<std::reference_wrapper<UniqueKey const>, 2> kKeysToCheck{ kCancelError, kNilSentinel };
         for (UniqueKey const& _key : kKeysToCheck) {
             if (_key.equals(L1, L1_i)) {
                 DEBUGSPEW_CODE(DebugSpew(nullptr) << _key.debugName);
diff --git a/src/lanes.lua b/src/lanes.lua
index 98f8c20..bd94a14 100644
--- a/src/lanes.lua
+++ b/src/lanes.lua
@@ -603,7 +603,6 @@ local configure_timers = function()
                 return next_wakeup  -- may be 'nil'
             end -- check_timers()
 
-            local timer_gateway_batched = timerLinda.batched
             set_finalizer(function(err, stk)
                 if err and type(err) ~= "userdata" then
                     error("LanesTimer error: "..tostring(err))
@@ -628,7 +627,7 @@ local configure_timers = function()
 
                 if _timerKey == TGW_KEY then
                     assert(getmetatable(_what) == "Linda") -- '_what' should be a linda on which the client sets a timer
-                    local _, key, wakeup_at, period = timerLinda:receive(0, timer_gateway_batched, TGW_KEY, 3)
+                    local _, key, wakeup_at, period = timerLinda:receive_batched(0, TGW_KEY, 3)
                     assert(key)
                     set_timer(_what, key, wakeup_at, period and period > 0 and period or nil)
                 elseif _timerKey == TGW_QUERY then
@@ -758,7 +757,7 @@ local genlock = function(linda_, key_, N)
             -- 'nil' timeout allows 'key_' to be numeric
             return linda_:send(timeout, key_, trues(M_))    -- suspends until been able to push them
         else
-            local _k, _v = linda_:receive(nil, linda_.batched, key_, -M_)
+            local _k, _v = linda_:receive_batched(nil, key_, -M_)
             -- propagate cancel_error if we got it, else return true or false
             return (_v == cancel_error and _v) or (_k and true or false)
         end
diff --git a/src/linda.cpp b/src/linda.cpp
index 1119d71..0cdacfa 100644
--- a/src/linda.cpp
+++ b/src/linda.cpp
@@ -62,7 +62,7 @@ namespace {
 
             case LuaType::LIGHTUSERDATA:
                 {
-                    static constexpr std::array<std::reference_wrapper<UniqueKey const>, 3> kKeysToCheck{ kLindaBatched, kCancelError, kNilSentinel };
+                    static constexpr std::array<std::reference_wrapper<UniqueKey const>, 2> kKeysToCheck{ kCancelError, kNilSentinel };
                     for (UniqueKey const& _key : kKeysToCheck) {
                         if (_key.equals(L_, _i)) {
                             raise_luaL_error(L_, "argument #%d: can't use %s as a slot", _i, _key.debugName.data());
@@ -123,6 +123,8 @@ namespace {
         return 0;
     }
 
+    // #############################################################################################
+
     // a helper to process the timeout argument of linda:send() and linda:receive()
     [[nodiscard]]
     static auto ProcessTimeoutArg(lua_State* const L_)
@@ -145,6 +147,142 @@ namespace {
     }
 
     // #############################################################################################
+
+    // the implementation for linda:receive() and linda:receive_batched()
+    static int ReceiveInternal(lua_State* const L_, bool const batched_)
+    {
+        Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
+
+        auto const [_key_i, _until] = ProcessTimeoutArg(L_);
+
+        keeper_api_t _selected_keeper_receive{ nullptr };
+        int _expected_pushed_min{ 0 }, _expected_pushed_max{ 0 };
+        // are we in batched mode?
+        if (batched_) {
+            // make sure the keys are of a valid type
+            CheckKeyTypes(L_, _key_i, _key_i);
+            // receive multiple values from a single slot
+            _selected_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);
+            if (_expected_pushed_min < 1) {
+                raise_luaL_argerror(L_, StackIndex{ _key_i + 1 }, "bad min count");
+            }
+            _expected_pushed_max = (int) luaL_optinteger(L_, _key_i + 2, _expected_pushed_min);
+            // don't forget to count the slot in addition to the values
+            ++_expected_pushed_min;
+            ++_expected_pushed_max;
+            if (_expected_pushed_min > _expected_pushed_max) {
+                raise_luaL_argerror(L_, StackIndex{ _key_i + 2 }, "batched min/max error");
+            }
+        } else {
+            // make sure the keys are of a valid type
+            CheckKeyTypes(L_, _key_i, StackIndex{ lua_gettop(L_) });
+            // receive a single value, checking multiple slots
+            _selected_keeper_receive = KEEPER_API(receive);
+            // we expect a single (value, slot) pair of returned values
+            _expected_pushed_min = _expected_pushed_max = 2;
+        }
+
+        Lane* const _lane{ kLanePointerRegKey.readLightUserDataValue<Lane>(L_) };
+        Keeper* const _keeper{ _linda->whichKeeper() };
+        KeeperState const _K{ _keeper ? _keeper->K : KeeperState{ static_cast<lua_State*>(nullptr) } };
+        if (_K == nullptr)
+            return 0;
+
+        CancelRequest _cancel{ CancelRequest::None };
+        KeeperCallResult _pushed{};
+
+        STACK_CHECK_START_REL(_K, 0);
+        for (bool _try_again{ true };;) {
+            if (_lane != nullptr) {
+                _cancel = _lane->cancelRequest.load(std::memory_order_relaxed);
+            }
+            _cancel = (_cancel != CancelRequest::None)
+                ? _cancel
+                : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
+            // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
+            if (!_try_again || _cancel != CancelRequest::None) {
+                _pushed.emplace(0);
+                break;
+            }
+
+            // all arguments of receive() but the first are passed to the keeper's receive function
+            STACK_CHECK(_K, 0);
+            _pushed = keeper_call(_K, _selected_keeper_receive, L_, _linda, _key_i);
+            if (!_pushed.has_value()) {
+                break;
+            }
+            if (_pushed.value() > 0) {
+                LUA_ASSERT(L_, _pushed.value() >= _expected_pushed_min && _pushed.value() <= _expected_pushed_max);
+                if (kRestrictedChannel.equals(L_, StackIndex{ kIdxTop })) {
+                    raise_luaL_error(L_, "Key is restricted");
+                }
+                _linda->readHappened.notify_all();
+                break;
+            }
+
+            if (std::chrono::steady_clock::now() >= _until) {
+                break; /* instant timeout */
+            }
+
+            // nothing received, wait until timeout or signalled that we should try again
+            {
+                Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
+                if (_lane != nullptr) {
+                    // change status of lane to "waiting"
+                    _prev_status = _lane->status.load(std::memory_order_acquire); // Running, most likely
+                    LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
+                    LUA_ASSERT(L_, _lane->waiting_on == nullptr);
+                    _lane->waiting_on = &_linda->writeHappened;
+                    _lane->status.store(Lane::Waiting, std::memory_order_release);
+                }
+                // not enough data to read: wakeup when data was sent, or when timeout is reached
+                std::unique_lock<std::mutex> _guard{ _keeper->mutex, std::adopt_lock };
+                std::cv_status const _status{ _linda->writeHappened.wait_until(_guard, _until) };
+                _guard.release(); // we don't want to unlock the mutex on exit!
+                _try_again = (_status == std::cv_status::no_timeout); // detect spurious wakeups
+                if (_lane != nullptr) {
+                    _lane->waiting_on = nullptr;
+                    _lane->status.store(_prev_status, std::memory_order_release);
+                }
+            }
+        }
+        STACK_CHECK(_K, 0);
+
+        if (!_pushed.has_value()) {
+            raise_luaL_error(L_, "tried to copy unsupported types");
+        }
+
+        switch (_cancel) {
+        case CancelRequest::None:
+            {
+                int const _nbPushed{ _pushed.value() };
+                if (_nbPushed == 0) {
+                    // not enough data in the linda slot to fulfill the request, return nil, "timeout"
+                    lua_pushnil(L_);
+                    luaG_pushstring(L_, "timeout");
+                    return 2;
+                }
+                return _nbPushed;
+            }
+
+        case CancelRequest::Soft:
+            // if user wants to soft-cancel, the call returns nil, kCancelError
+            lua_pushnil(L_);
+            kCancelError.pushKey(L_);
+            return 2;
+
+        case CancelRequest::Hard:
+            // raise an error interrupting execution only in case of hard cancel
+            raise_cancel_error(L_); // raises an error and doesn't return
+
+        default:
+            raise_luaL_error(L_, "internal error: unknown cancel request");
+        }
+    }
+
+    // #############################################################################################
     // #############################################################################################
 } // namespace
 // #################################################################################################
@@ -650,153 +788,25 @@ LUAG_FUNC(linda_limit)
 // #################################################################################################
 
 /*
- * 2 modes of operation
- * [val, slot]= linda:receive([timeout_secs_num=nil], key_num|str|bool|lightuserdata [, ...] )
+ * [val, slot] = linda:receive([timeout_secs_num=nil], key_num|str|bool|lightuserdata [, ...] )
  * Consumes a single value from the Linda, in any slot.
  * Returns: received value (which is consumed from the slot), and the slot 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 slot.
- * returns the actual consumed values, or nil if there weren't enough values to consume
  */
 LUAG_FUNC(linda_receive)
 {
-    static constexpr lua_CFunction _receive{
-        +[](lua_State* const L_) {
-            Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
-
-            auto [_key_i, _until] = ProcessTimeoutArg(L_);
-
-            keeper_api_t _selected_keeper_receive{ nullptr };
-            int _expected_pushed_min{ 0 }, _expected_pushed_max{ 0 };
-            // are we in batched mode?
-            if (kLindaBatched.equals(L_, _key_i)) {
-                // no need to pass linda.batched in the keeper state
-                ++_key_i;
-                // make sure the keys are of a valid type
-                CheckKeyTypes(L_, _key_i, _key_i);
-                // receive multiple values from a single slot
-                _selected_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);
-                if (_expected_pushed_min < 1) {
-                    raise_luaL_argerror(L_, StackIndex{ _key_i + 1 }, "bad min count");
-                }
-                _expected_pushed_max = (int) luaL_optinteger(L_, _key_i + 2, _expected_pushed_min);
-                // don't forget to count the slot in addition to the values
-                ++_expected_pushed_min;
-                ++_expected_pushed_max;
-                if (_expected_pushed_min > _expected_pushed_max) {
-                    raise_luaL_argerror(L_, StackIndex{ _key_i + 2 }, "batched min/max error");
-                }
-            } else {
-                // make sure the keys are of a valid type
-                CheckKeyTypes(L_, _key_i, StackIndex{ lua_gettop(L_) });
-                // receive a single value, checking multiple slots
-                _selected_keeper_receive = KEEPER_API(receive);
-                // we expect a single (value, slot) pair of returned values
-                _expected_pushed_min = _expected_pushed_max = 2;
-            }
-
-            Lane* const _lane{ kLanePointerRegKey.readLightUserDataValue<Lane>(L_) };
-            Keeper* const _keeper{ _linda->whichKeeper() };
-            KeeperState const _K{ _keeper ? _keeper->K : KeeperState{ static_cast<lua_State*>(nullptr) } };
-            if (_K == nullptr)
-                return 0;
-
-            CancelRequest _cancel{ CancelRequest::None };
-            KeeperCallResult _pushed{};
-
-            STACK_CHECK_START_REL(_K, 0);
-            for (bool _try_again{ true };;) {
-                if (_lane != nullptr) {
-                    _cancel = _lane->cancelRequest.load(std::memory_order_relaxed);
-                }
-                _cancel = (_cancel != CancelRequest::None)
-                    ? _cancel
-                    : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
-                // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
-                if (!_try_again || _cancel != CancelRequest::None) {
-                    _pushed.emplace(0);
-                    break;
-                }
-
-                // all arguments of receive() but the first are passed to the keeper's receive function
-                STACK_CHECK(_K, 0);
-                _pushed = keeper_call(_K, _selected_keeper_receive, L_, _linda, _key_i);
-                if (!_pushed.has_value()) {
-                    break;
-                }
-                if (_pushed.value() > 0) {
-                    LUA_ASSERT(L_, _pushed.value() >= _expected_pushed_min && _pushed.value() <= _expected_pushed_max);
-                    if (kRestrictedChannel.equals(L_, StackIndex{ kIdxTop })) {
-                        raise_luaL_error(L_, "Key is restricted");
-                    }
-                    _linda->readHappened.notify_all();
-                    break;
-                }
-
-                if (std::chrono::steady_clock::now() >= _until) {
-                    break; /* instant timeout */
-                }
-
-                // nothing received, wait until timeout or signalled that we should try again
-                {
-                    Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
-                    if (_lane != nullptr) {
-                        // change status of lane to "waiting"
-                        _prev_status = _lane->status.load(std::memory_order_acquire); // Running, most likely
-                        LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
-                        LUA_ASSERT(L_, _lane->waiting_on == nullptr);
-                        _lane->waiting_on = &_linda->writeHappened;
-                        _lane->status.store(Lane::Waiting, std::memory_order_release);
-                    }
-                    // not enough data to read: wakeup when data was sent, or when timeout is reached
-                    std::unique_lock<std::mutex> _guard{ _keeper->mutex, std::adopt_lock };
-                    std::cv_status const _status{ _linda->writeHappened.wait_until(_guard, _until) };
-                    _guard.release(); // we don't want to unlock the mutex on exit!
-                    _try_again = (_status == std::cv_status::no_timeout); // detect spurious wakeups
-                    if (_lane != nullptr) {
-                        _lane->waiting_on = nullptr;
-                        _lane->status.store(_prev_status, std::memory_order_release);
-                    }
-                }
-            }
-            STACK_CHECK(_K, 0);
-
-            if (!_pushed.has_value()) {
-                raise_luaL_error(L_, "tried to copy unsupported types");
-            }
-
-            switch (_cancel) {
-            case CancelRequest::None:
-                {
-                    int const _nbPushed{ _pushed.value() };
-                    if (_nbPushed == 0) {
-                        // not enough data in the linda slot to fulfill the request, return nil, "timeout"
-                        lua_pushnil(L_);
-                        luaG_pushstring(L_, "timeout");
-                        return 2;
-                    }
-                    return _nbPushed;
-                }
-
-            case CancelRequest::Soft:
-                // if user wants to soft-cancel, the call returns nil, kCancelError
-                lua_pushnil(L_);
-                kCancelError.pushKey(L_);
-                return 2;
+    return Linda::ProtectedCall(L_, [](lua_State* const L_) { return ReceiveInternal(L_, false); });
+}
 
-            case CancelRequest::Hard:
-                // raise an error interrupting execution only in case of hard cancel
-                raise_cancel_error(L_); // raises an error and doesn't return
+// #################################################################################################
 
-            default:
-                raise_luaL_error(L_, "internal error: unknown cancel request");
-            }
-        }
-    };
-    return Linda::ProtectedCall(L_, _receive);
+/*
+ * [val1, ... valCOUNT] = linda_receive_batched( linda_ud, [timeout_secs_num=-1], key_num|str|bool|lightuserdata, min_COUNT[, max_COUNT])
+ * Consumes between min_COUNT and max_COUNT values from the linda, from a single slot.
+ * returns the actual consumed values, or nil if there weren't enough values to consume
+ */
+LUAG_FUNC(linda_receive_batched)
+{
+    return Linda::ProtectedCall(L_, [](lua_State* const L_) { return ReceiveInternal(L_, true); });
 }
 
 // #################################################################################################
@@ -1102,6 +1112,7 @@ namespace {
             { "get", LG_linda_get },
             { "limit", LG_linda_limit },
             { "receive", LG_linda_receive },
+            { "receive_batched", LG_linda_receive_batched },
             { "restrict", LG_linda_restrict },
             { "send", LG_linda_send },
             { "set", LG_linda_set },
diff --git a/src/linda.hpp b/src/linda.hpp
index 01ca7e1..2d5c9dc 100644
--- a/src/linda.hpp
+++ b/src/linda.hpp
@@ -8,11 +8,6 @@ struct Keeper;
 
 // #################################################################################################
 
-// xxh64 of string "kLindaBatched" generated at https://www.pelock.com/products/hash-calculator
-static constexpr UniqueKey kLindaBatched{ 0xB8234DF772646567ull, "linda.batched" };
-
-// #################################################################################################
-
 DECLARE_UNIQUE_TYPE(LindaGroup, int);
 
 class Linda final
diff --git a/src/lindafactory.cpp b/src/lindafactory.cpp
index 6e3f759..42d0984 100644
--- a/src/lindafactory.cpp
+++ b/src/lindafactory.cpp
@@ -53,10 +53,6 @@ void LindaFactory::createMetatable(lua_State* L_) const
     // the linda functions
     luaG_registerlibfuncs(L_, mLindaMT);
 
-    // some constants
-    kLindaBatched.pushKey(L_);                                                                     // L_: mt kLindaBatched
-    lua_setfield(L_, -2, "batched");                                                               // L_: mt
-
     kNilSentinel.pushKey(L_);                                                                      // L_: mt kNilSentinel
     lua_setfield(L_, -2, "null");                                                                  // L_: mt
 
diff --git a/tests/basic.lua b/tests/basic.lua
index 170e821..a9c85cc 100644
--- a/tests/basic.lua
+++ b/tests/basic.lua
@@ -209,7 +209,7 @@ PRINT "wait_receive_lane is cancelled"
 local wait_receive_batched = function()
     local k, v1, v2
     set_finalizer(function() print("wait_receive_batched", k, v1, v2) end)
-    k, v1, v2 = limited:receive(limited.batched, "dummy", 2) -- infinite timeout, returns only when lane is cancelled
+    k, v1, v2 = limited:receive_batched("dummy", 2) -- infinite timeout, returns only when lane is cancelled
 end
 
 local wait_receive_batched_lane = lanes.gen("*", { name = 'auto' }, wait_receive_batched)()
@@ -273,7 +273,7 @@ local b,x,y,z,w = linda:get("<->", 4)
 assert(b == 3 and x == "x" and y == "y" and z == "z" and w == nil)
 local k, x = linda:receive("<->")
 assert(k == "<->" and x == "x")
-local k,y,z = linda:receive(linda.batched, "<->", 2)
+local k,y,z = linda:receive_batched("<->", 2)
 assert(k == "<->" and y == "y" and z == "z")
 linda:set("<->")
 local b,x,y,z,w = linda:get("<->", 4)
diff --git a/tests/errhangtest.lua b/tests/errhangtest.lua
index fff0dee..5b3f0c0 100644
--- a/tests/errhangtest.lua
+++ b/tests/errhangtest.lua
@@ -19,7 +19,6 @@ end
 if true then
         print "\n#### reserved sentinels"
         print(pcall(linda.set, linda, lanes.cancel_error))
-        print(pcall(linda.set, linda, linda.batched))
         local _count, _val = linda:get("test")
         assert(_count == 0 and _val == nil)
         print "OK"
@@ -28,13 +27,13 @@ end
 -- get/set a few values
 if true then
         print "\n#### set 3 -> receive batched"
-        assert.fails(function() linda:receive(linda.batched, "some key", -1, 1) end)
-        assert.fails(function() linda:receive(linda.batched, "some key", 2, 1) end)
-        assert.failsnot(function() linda:receive(0, linda.batched, "some key", 1, 3) end)
+        assert.fails(function() linda:receive_batched("some key", -1, 1) end)
+        assert.fails(function() linda:receive_batched("some key", 2, 1) end)
+        assert.failsnot(function() linda:receive_batched(0, "some key", 1, 3) end)
         local fun = function() print "function test ok" end
         print(pcall(linda.set, linda, 'test', true, nil, fun))
         -- read back the contents
-        local k,b,n,f = linda:receive(linda.batched, 'test', 3)
+        local k,b,n,f = linda:receive_batched('test', 3)
         local _count, _val = linda:get("test")
         assert(_count == 0 and _val == nil)
         -- check they are ok
diff --git a/tests/linda_perf.lua b/tests/linda_perf.lua
index bba1408..107fd25 100644
--- a/tests/linda_perf.lua
+++ b/tests/linda_perf.lua
@@ -68,7 +68,7 @@ local gobbler = function( l, loop, batch)
         l:receive( "go")
         -- eat data in batches
         for i = 1, loop/batch do
-                l:receive( l.batched, "key", batch)
+                l:receive_batched("key", batch)
                 -- print("gobbler:", batch)
         end
         print "loop is over"
@@ -168,9 +168,9 @@ local function ziva2( preloop, loop, batch)
         local l = lanes.linda("ziva2("..preloop..":"..loop..":"..tostring(batch)..")", group_uid)
         group_uid = (group_uid % config.nb_user_keepers) + 1
         -- prefill the linda a bit to increase fifo stress
-        local top, step = math.max( preloop, loop), (l.batched and batch) and batch or 1
+        local top, step = math.max( preloop, loop), batch or 1
         local batch_send, batch_read
-        if l.batched and batch then
+        if batch then
                 local batch_values = {}
                 for i = 1, batch do
                         table.insert( batch_values, i)
@@ -180,7 +180,7 @@ local function ziva2( preloop, loop, batch)
                         l:send( "key", table_unpack( batch_values))
                 end
                 batch_read = function()
-                        l:receive( l.batched, "key", batch)
+                        l:receive_batched("key", batch)
                 end
         else -- not batched
                 batch_send = function()
diff --git a/tests/protect_allocator.lua b/tests/protect_allocator.lua
index e13a57c..325726a 100644
--- a/tests/protect_allocator.lua
+++ b/tests/protect_allocator.lua
@@ -52,7 +52,7 @@ end
 
 -- wait for completion
 print "wait for completion"
-linda:receive( linda.batched, "key", COUNT)
+linda:receive_batched("key", COUNT)
 print "waiting a bit more ..."
 SLEEP(1)
 print "SUCCESS"
diff --git a/unit_tests/linda_tests.cpp b/unit_tests/linda_tests.cpp
index 2308737..f2934eb 100644
--- a/unit_tests/linda_tests.cpp
+++ b/unit_tests/linda_tests.cpp
@@ -106,7 +106,6 @@ TEST_CASE("linda.single Keeper")
             S.requireFailure("lanes.linda():send(0, io.stdin, 'v')");
             S.requireFailure("lanes.linda():send(0, lanes.null, 'v')");
             S.requireFailure("lanes.linda():send(0, lanes.cancel_error, 'v')");
-            S.requireFailure("local l = lanes.linda(); l:send(0, l.batched, 'v')");
         }
 
         // -----------------------------------------------------------------------------------------
diff --git a/unit_tests/scripts/lane/tasking_cancelling.lua b/unit_tests/scripts/lane/tasking_cancelling.lua
index 8bee3a6..ea4516e 100644
--- a/unit_tests/scripts/lane/tasking_cancelling.lua
+++ b/unit_tests/scripts/lane/tasking_cancelling.lua
@@ -61,7 +61,7 @@ PRINT "wait_receive_lane is cancelled"
 local wait_receive_batched = function()
     local k, v1, v2
     set_finalizer(function() print("wait_receive_batched", k, v1, v2) end)
-    k, v1, v2 = limited:receive(limited.batched, "dummy", 2) -- infinite timeout, returns only when lane is cancelled
+    k, v1, v2 = limited:receive_batched("dummy", 2) -- infinite timeout, returns only when lane is cancelled
 end
 
 local wait_receive_batched_lane = lanes_gen("*", { name = 'auto' }, wait_receive_batched)()
diff --git a/unit_tests/scripts/lane/tasking_communications.lua b/unit_tests/scripts/lane/tasking_communications.lua
index 1fd43b0..631d105 100644
--- a/unit_tests/scripts/lane/tasking_communications.lua
+++ b/unit_tests/scripts/lane/tasking_communications.lua
@@ -90,7 +90,7 @@ local b,x,y,z,w = linda:get("<->", 4)
 assert(b == 3 and x == "x" and y == "y" and z == "z" and w == nil)
 local k, x = linda:receive("<->")
 assert(k == "<->" and x == "x")
-local k,y,z = linda:receive(linda.batched, "<->", 2)
+local k,y,z = linda:receive_batched("<->", 2)
 assert(k == "<->" and y == "y" and z == "z")
 linda:set("<->")
 local b,x,y,z,w = linda:get("<->", 4)