[Tarantool-patches] [PATCH] Release all taken tasks on start

[Leonid Vasiliev]

Add some stylistic edits

---
diff --git a/queue/abstract.lua b/queue/abstract.lua
index ed07d85..3c26d35 100644
--- a/queue/abstract.lua
+++ b/queue/abstract.lua
@@ -516,10 +516,12 @@ function method.start()
          tube = recreate_tube(tube_tuple)
          local space = box.space[space_name]
          for _, task_tuple in space:pairs() do
-            -- Release all taken tasks
-            local tid, task_state = task_tuple:unpack()
+            -- Release all taken tasks on start
+            -- See https://github.com/tarantool/queue/issues/66
+            -- for more information
+            local tid, task_state = task_tuple[1], task_tuple[2]
              if task_state == state.TAKEN then
-                tube.raw:release(tid, {})
+                tube.raw:release(tid)
              end
          end
      end


On 12/18/19 11:52 AM, Leonid Vasiliev wrote:
> https://github.com/tarantool/queue/issues/66
> https://github.com/tarantool/queue/tree/lvasiliev/gh-66-release-taken-tasks-after-reboot
> 
> If some tasks have been taken and don't released before shutdown
> of the tarantool instance (for example: tarantool instance has been killed)
> such task go to 'hung' state (noone can take the task now).
> So, we must release all taken tasks on start of the queue module.
> 
> ---
>   queue/abstract.lua | 13 ++++++++++++-
>   1 file changed, 12 insertions(+), 1 deletion(-)
> 
> diff --git a/queue/abstract.lua b/queue/abstract.lua
> index ad8817d..ed07d85 100644
> --- a/queue/abstract.lua
> +++ b/queue/abstract.lua
> @@ -511,7 +511,18 @@ function method.start()
>           })
>       end
>   
> -    _queue:pairs():each(recreate_tube)
> +    for _, tube_tuple in _queue:pairs() do
> +        local space_name = tube_tuple[3]
> +        tube = recreate_tube(tube_tuple)
> +        local space = box.space[space_name]
> +        for _, task_tuple in space:pairs() do
> +            -- Release all taken tasks
> +            local tid, task_state = task_tuple:unpack()
> +            if task_state == state.TAKEN then
> +                tube.raw:release(tid, {})
> +            end
> +        end
> +    end
>   
>       session.on_disconnect(queue._on_consumer_disconnect)
>       return queue
>