expire: fix expire fiber termination in ro mode

The instance can return from read-only mode,
so the fiber cannot be terminated.

Closes #119

Author: 0x501D

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Don't terminate the memcached expire fiber in read-only mode.
+
 ## [1.1.0] - 2022-12-07
 
 ### Added

memcached/internal/expiration.c

@@ -10,6 +10,7 @@
 #include "memcached_layer.h"
 
 #include "error.h"
+#include "utils.h"
 
 int
 memcached_expire_process(struct memcached_service *p, box_iterator_t **iterp)
@@ -58,16 +59,31 @@ memcached_expire_process(struct memcached_service *p, box_iterator_t **iterp)
 int
 memcached_expire_loop(va_list ap)
 {
+	static int box_ro;
 	struct memcached_service *p = va_arg(ap, struct memcached_service *);
 	char key[2], *key_end = mp_encode_array(key, 0);
 	box_iterator_t *iter = NULL;
 	int rv = 0;
 	say_info("Memcached expire fiber started");
-restart:
-	if (iter == NULL) {
-		iter = box_index_iterator(p->space_id, 0, ITER_ALL, key, key_end);
+restart: ;
+	int old_state = box_ro;
+
+	if (memcached_box_is_ro(&box_ro) != true) {
+		say_error("Cannot get box.info.ro value");
+		goto finish;
+	}
+
+	if (box_ro) {
+		if (old_state != box_ro) {
+			say_info("Expire: the instance has been moved to a read-only mode");
+		}
+		goto delay;
 	}
+
 	if (rv == -1 || iter == NULL) {
+		iter = box_index_iterator(p->space_id, 0, ITER_ALL, key, key_end);
+	}
+	if (iter == NULL) {
 		const box_error_t *err = box_error_last();
 		say_error("Unexpected error %u: %s",
 				box_error_code(err),
@@ -82,7 +98,7 @@ memcached_expire_loop(va_list ap)
 				box_error_message(err));
 		goto finish;
 	}
-
+delay: ;
 	/* This part is where we rest after all deletes */
 	double delay = ((double )p->expire_count * p->expire_time) /
 			(box_index_len(p->space_id, 0) + 1);

memcached/internal/utils.c

@@ -5,6 +5,7 @@
 #include <string.h>
 #include <inttypes.h>
 
+#include <tarantool/lua.h>
 #include <tarantool/module.h>
 #include <msgpuck.h>
 
@@ -101,3 +102,30 @@ strindex(const char **haystack, const char *needle, uint32_t hmax)
 			return index;
 	return hmax;
 }
+
+/** Get box.info.ro value.
+ *
+ * @param result pointer to where the value will be stored.
+ * @return true if the value was retrieved successfully.
+*/
+bool
+memcached_box_is_ro(int *result)
+{
+	lua_State *L = luaT_state();
+	int top = lua_gettop(L);
+	int ro;
+
+	lua_getfield(L, LUA_GLOBALSINDEX, "box");
+	lua_getfield(L, -1, "info");
+	lua_getfield(L, -1, "ro");
+
+	if (!lua_isboolean(L, -1)) {
+		return false;
+	}
+
+	ro = lua_toboolean(L, -1);
+	lua_settop(L, top);
+	*result = ro;
+
+	return true;
+}

memcached/internal/utils.h

@@ -18,6 +18,9 @@ safe_strtoull(const char *begin, const char *end, uint64_t *out);
 void
 memcached_binary_header_dump(struct memcached_hdr *hdr);
 
+bool
+memcached_box_is_ro(int *result);
+
 /* Macros to define enum and corresponding strings. */
 #define ENUM0_MEMBER(s, ...) s,
 #define ENUM_MEMBER(s, v, ...) s = v,

test/text/expirations.result

@@ -64,6 +64,18 @@ add add 0 1 7
 addval2
 >>--------------------------------------------------
 STORED
+# return from RO - key should be removed
+<<--------------------------------------------------
+add key 0 1 3
+foo
+>>--------------------------------------------------
+STORED
+# set RO
+# set RW
+box.space.__mc_memcached:len()
+---
+- 0
+...
 <<--------------------------------------------------
 flush_all
 >>--------------------------------------------------

test/text/expirations.test.py

@@ -98,6 +98,20 @@
 print("""# readd add - shoud be success""")
 mc_client("add add 0 1 7\r\naddval2\r\n")
 
+server.wait_lsn(master_id, lsn + 3)
+print("""# return from RO - key should be removed""")
+mc_client("add key 0 1 3\r\nfoo\r\n")
+print("""# set RO""")
+server.admin("box.cfg({read_only = true})", silent=True)
+log = server.get_log()
+while log.seek_once("Expire: the instance has been moved to a read-only mode") < 0:
+    time.sleep(0.01)
+time.sleep(1)
+print("""# set RW""")
+server.admin("box.cfg({read_only = false})", silent=True)
+server.wait_lsn(master_id, lsn + 6)
+server.admin("box.space.__mc_memcached:len()", silent=False)
+
 mc_client("flush_all\r\n")
 
 sys.path = saved_path