From fb1e56cd6550a72b8a4ec8bcd50eb7faeda3eccc Mon Sep 17 00:00:00 2001 From: edam Date: Mon, 20 Jul 2020 13:50:18 +0100 Subject: [PATCH 1/4] added pause/unpause and watchdog for box ro/rw state --- indexpiration.lua | 44 +++++++++++++++++++++++++++++++++----------- 1 file changed, 33 insertions(+), 11 deletions(-) diff --git a/indexpiration.lua b/indexpiration.lua index ff4dc29..a35cbcf 100644 --- a/indexpiration.lua +++ b/indexpiration.lua @@ -60,22 +60,43 @@ end local F = {} -function F:stop() - self.stopped = true +function F:terminate() + self._terminate = true + self:stop_worker() +end + +function F:start_watchdog() + self._watcher = fiber.create(function(expiration) + while not expiration._terminate do + if box.info.ro then + expiration:stop_worker() + pcall(box.ctl.wait_rw) + else + expiration:start_worker() + pcall(box.ctl.wait_ro) + end + end + end, self) +end + +function F:stop_worker() + if not self.running then return end + self.running = false self._wait:put(true,0) end function F:start_worker() + if self._terminate or self.running then return end + self.running = true self._worker = fiber.create(function(space,expiration,expire_index) local fname = space.name .. '.xpr' if package.reload then fname = fname .. '.' .. package.reload.count end fiber.name(string.sub(fname,1,32)) repeat fiber.sleep(0.001) until space.expiration - local chan = expiration._wait - log.info("Worker started") + log.info("Worker started: %s",space.name) local curwait local collect = {} - while box.space[space.name] and space.expiration == expiration and not expiration.stopped do + while box.space[space.name] and space.expiration == expiration and expiration.running do local r,e = pcall(function() -- print("runat loop 2 ",box.time64()) local remaining @@ -139,12 +160,12 @@ function F:start_worker() end -- log.info("Wait %0.2fs",curwait) if curwait == 0 then fiber.sleep(0) end - chan:get(curwait) + expiration._wait:get(curwait) end - if expiration.stopped then - log.info("Expiration worker was stopped") + if expiration.running then + log.info("Worker finished: %s",space.name) else - log.info("Worker finished") + log.info("Worker stopped: %s",space.name) end end,box.space[self.space],self,self.expire_index) end @@ -257,8 +278,9 @@ function M.upgrade(space,opts,depth) self.space = space.id self.expire_index = expire_index - self:start_worker() - + self._terminate = false + self.running = false + self:start_watchdog() if self.precise then self._on_repl = space:on_replace(function(old, new) From 30a951b92737e8138aceac8adcc8c0b3ac7d9a60 Mon Sep 17 00:00:00 2001 From: edam Date: Mon, 20 Jul 2020 19:32:50 +0100 Subject: [PATCH 2/4] removed duplicate space name from logs --- indexpiration.lua | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/indexpiration.lua b/indexpiration.lua index a35cbcf..ebbb63b 100644 --- a/indexpiration.lua +++ b/indexpiration.lua @@ -93,7 +93,7 @@ function F:start_worker() if package.reload then fname = fname .. '.' .. package.reload.count end fiber.name(string.sub(fname,1,32)) repeat fiber.sleep(0.001) until space.expiration - log.info("Worker started: %s",space.name) + log.info("Worker started") local curwait local collect = {} while box.space[space.name] and space.expiration == expiration and expiration.running do @@ -163,9 +163,9 @@ function F:start_worker() expiration._wait:get(curwait) end if expiration.running then - log.info("Worker finished: %s",space.name) + log.info("Worker finished") else - log.info("Worker stopped: %s",space.name) + log.info("Worker stopped") end end,box.space[self.space],self,self.expire_index) end From 9d416d58832c6a6de907faa5ea1b9658001195d6 Mon Sep 17 00:00:00 2001 From: edam Date: Tue, 11 Aug 2020 13:47:23 +0100 Subject: [PATCH 3/4] name watchdog fiber --- indexpiration.lua | 1 + 1 file changed, 1 insertion(+) diff --git a/indexpiration.lua b/indexpiration.lua index ebbb63b..f34b3c9 100644 --- a/indexpiration.lua +++ b/indexpiration.lua @@ -67,6 +67,7 @@ end function F:start_watchdog() self._watcher = fiber.create(function(expiration) + fiber.name('indexpiration-watchdog') while not expiration._terminate do if box.info.ro then expiration:stop_worker() From 7ef680d66baccdb2c1e9b52a2002ca47cb3ab1c0 Mon Sep 17 00:00:00 2001 From: edam Date: Thu, 5 Nov 2020 11:28:35 +0000 Subject: [PATCH 4/4] check if stopped running prior to delete --- indexpiration.lua | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/indexpiration.lua b/indexpiration.lua index f34b3c9..6bf27f8 100644 --- a/indexpiration.lua +++ b/indexpiration.lua @@ -125,6 +125,10 @@ function F:start_worker() end for _,t in pairs(collect) do if not expiration.txn then + if not expiration.running then + remaining = 0 + break + end t = box.space[space.name]:get(expiration._pk(t)) if expiration.check( t ) > 0 then t = nil