tarantool · mkostoevr · Mar 24, 2025 · Mar 25, 2025 · ligurio · Mar 28, 2025
diff --git a/perf/lua/1mops_write.lua b/perf/lua/1mops_write.lua
@@ -10,6 +10,7 @@ local fiber = require('fiber')
 local math = require('math')
 local json = require('json')
 local fio = require('fio')
+local tarantool = require('tarantool')
 
 -- XXX: This benchmark should be able to work standalone without
 -- any provided libraries or set LUA_PATH. Add stubs for that
@@ -43,6 +44,7 @@ local HELP = [[
 local parsed_params = {
     {'engine', 'string'},
     {'fibers', 'number'},
+    {'columns', 'number'},
     {'index', 'string'},
     {'nodes', 'number'},
     {'nohint', 'boolean'},
@@ -53,6 +55,12 @@ local parsed_params = {
     {'warmup', 'number'},
 }
 
+local BUILDDIR = fio.abspath(fio.pathjoin(os.getenv('BUILDDIR') or '.'))
+local MODULEPATH = fio.pathjoin(BUILDDIR, 'perf', 'lua',
+                                '?.' .. tarantool.build.mod_format)
+package.cpath = MODULEPATH .. ';' .. package.cpath
+local module_is_available, module = pcall(require, '1mops_write_module')
+
 local params
 local bench
 
@@ -117,14 +125,18 @@ local nodes = params.nodes or 1
 
 -- engine to use
 local engine = params.engine or 'memtx'
-assert(engine == 'memtx' or engine == 'vinyl')
+assert(engine == 'memtx' or engine == 'vinyl' or engine == 'memcs')
 
 -- WAL mode
 local wal_mode = params.wal_mode or 'write'
 assert(wal_mode == 'write'
        or wal_mode == 'none'
        or wal_mode == 'fsync')
 
+-- number of columns in the table
+local num_columns = params.columns or 1
+assert(num_columns >= 1)
+
 -- type of index to use
 local index_config = {}
 index_config.type = params.index or 'HASH'
@@ -157,14 +169,14 @@ local std_redirect = {
 }
 
 print(string.format([[
-# making %d REPLACE operations,
+# making %d REPLACE operations in %s engine,
 # %d operations per txn,
 # using %d fibers,
 # in a replicaset of %d nodes,
 # using %s index type%s
 # with WAL mode %s
 # ]],
-      num_ops, ops_per_txn, num_fibers, nodes,
+      num_ops, engine, ops_per_txn, num_fibers, nodes,
       index_config.type, hints_msg, wal_mode))
 
 box.cfg{
@@ -223,21 +235,25 @@ if (nodes > 1) then
     end
 end
 
-local space
-local done = false
-local err
-
 if (test_sync) then
     box.cfg{replication_synchro_quorum = nodes}
     print('# promoting')
     box.ctl.promote()
     print('# done')
-    space, err = box.schema.create_space('test',
-                                         {engine = engine, is_sync = true})
-else
-    space, err = box.schema.create_space('test',
-                                         {engine = engine})
 end
+
+local done = false
+local space_format = {}
+for i = 1, num_columns do
+    table.insert(space_format, {'field_' .. tostring(i), 'unsigned'})
+end
+local create_space_opts = {
+    engine = engine,
+    field_count = num_columns,
+    format = space_format,
+    is_sync = test_sync or nil
+}
+local space, err = box.schema.create_space('test', create_space_opts)
 if space == false then
     exit(1, 'error creating space ' .. json.encode(err))
 end
@@ -247,17 +263,28 @@ if (res ~= true) then
          json.encode(index_config) .. ' :' .. json.encode(err))
 end
 
+-- Preallocate the test tuple
+local tuple = {}
+for i = 1, num_columns do -- luacheck: no unused
+    table.insert(tuple, 0)
+end
+
 -- THE load fiber
 local function fiber_load(start, s)
     start = start % 1000000 -- limit the size of space to 1M elements
-    for _ = 1, trans_per_fiber do
-        box.begin()
-        for _ = 1, ops_per_txn do
-            s:replace{start}
-            start = start + 1
+    if module_is_available then
+        module.fiber(s.id, trans_per_fiber, ops_per_txn, num_columns, start)
+    else
+        for _ = 1, trans_per_fiber do
+            box.begin()
+            for _ = 1, ops_per_txn do
+                tuple[1] = start
+                s:replace(tuple)
+                start = start + 1
+            end
+            box.commit()
+            fiber.yield()
         end
-        box.commit()
-        fiber.yield()
     end
 end
 
@@ -307,7 +334,7 @@ end)
 
 -- start fibers for the main load
 for i = 1, num_fibers do
-    fibers_storage[i] = fiber.create(fiber_load, i*num_ops, space)
+    fibers_storage[i] = fiber.create(fiber_load, i*(num_ops/num_fibers), space)
     if (fibers_storage[i]:status() ~= 'dead') then
         fibers_storage[i]:wakeup() -- needed for backward compatibility with 1.7
     end

diff --git a/perf/lua/1mops_write_module.c b/perf/lua/1mops_write_module.c
@@ -0,0 +1,173 @@
+#include <lua.h>
+#include <lauxlib.h>
+#include <module.h>
+#include <msgpuck.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include "trivia/config.h"
+#include "trivia/util.h"
+
+/** Template tuples to update and insert: one per PK msgpack field size. */
+static char *base_tuple_pk1;
+static char *base_tuple_pk1_end;
+static char *base_tuple_pk2;
+static char *base_tuple_pk2_end;
+static char *base_tuple_pk3;
+static char *base_tuple_pk3_end;
+static char *base_tuple_pk5;
+static char *base_tuple_pk5_end;
+
+static bool
+find_base_tuple(uint32_t pk_value, char **tuple, char **tuple_end)
+{
+	size_t sizeof_pk_value = mp_sizeof_uint(pk_value);
+	switch (sizeof_pk_value) {
+	case 1:
+		*tuple = base_tuple_pk1;
+		*tuple_end = base_tuple_pk1_end;
+		return true;
+	case 2:
+		*tuple = base_tuple_pk2;
+		*tuple_end = base_tuple_pk2_end;
+		return true;
+	case 3:
+		*tuple = base_tuple_pk3;
+		*tuple_end = base_tuple_pk3_end;
+		return true;
+	case 5:
+		*tuple = base_tuple_pk5;
+		*tuple_end = base_tuple_pk5_end;
+		return true;
+	}
+	box_error_raise(ER_UNSUPPORTED, "No tuple for PK value of size %lu, "
+			"value: %u", sizeof_pk_value, pk_value);
+	return false;
+}
+
+static bool
+test_tuple(uint32_t pk_value, char **tuple, char **tuple_end)
+{
+	/* Get the tuple to update. */
+	if (!find_base_tuple(pk_value, tuple, tuple_end))
+		return false;
+
+	/* Check if the new PK value is of exact size. */
+	const char *data = *tuple;
+	mp_decode_array(&data);
+	char *pk_value_ptr = (char *)data;
+	uint32_t old_pk_value = mp_decode_uint(&data);
+	if (mp_sizeof_uint(pk_value) != mp_sizeof_uint(old_pk_value)) {
+		box_error_raise(ER_UNSUPPORTED, "Wrong base tuple, "
+				"PK value %u", pk_value);
+		return false;
+	}
+
+	/* Write the new PK value. */
+	mp_encode_uint(pk_value_ptr, pk_value);
+	return true;
+}
+
+static char *
+encode_base_tuple(char *data, ptrdiff_t *data_sz,
+		  uint32_t pk_value, uint32_t num_columns)
+{
+	data = mp_encode_array_safe(data, data_sz, num_columns);
+	data = mp_encode_uint_safe(data, data_sz, pk_value);
+	for (uint32_t i = 1; i < num_columns; i++)
+		data = mp_encode_uint_safe(data, data_sz, 0);
+	return data;
+}
+
+static bool
+create_base_tuples(uint32_t num_columns)
+{
+	uint32_t uint_1 = 0;
+	uint32_t uint_2 = UINT8_MAX;
+	uint32_t uint_3 = UINT16_MAX;
+	uint32_t uint_5 = UINT32_MAX;
+	if (mp_sizeof_uint(uint_1) != 1 ||
+	    mp_sizeof_uint(uint_2) != 2 ||
+	    mp_sizeof_uint(uint_3) != 3 ||
+	    mp_sizeof_uint(uint_5) != 5) {
+		box_error_raise(ER_UNKNOWN, "PK value size assertion failed");
+		return false;
+	}
+
+	struct {
+		uint32_t pk_value;
+		char **ptr;
+		char **end;
+	} base_tuples[] = {
+		{ uint_1, &base_tuple_pk1, &base_tuple_pk1_end },
+		{ uint_2, &base_tuple_pk2, &base_tuple_pk2_end },
+		{ uint_3, &base_tuple_pk3, &base_tuple_pk3_end },
+		{ uint_5, &base_tuple_pk5, &base_tuple_pk5_end },
+	};
+
+	for (size_t i = 0; i < lengthof(base_tuples); i++) {
+		uint32_t pk_value = base_tuples[i].pk_value;
+		char **base_tuple_ptr = base_tuples[i].ptr;
+		char **base_tuple_end_ptr = base_tuples[i].end;
+
+		ptrdiff_t sizeof_tuple = 0;
+		encode_base_tuple(NULL, &sizeof_tuple, pk_value, num_columns);
+		*base_tuple_ptr = xcalloc((uint32_t)sizeof_tuple, 1);
+		*base_tuple_end_ptr = encode_base_tuple(*base_tuple_ptr, NULL,
+							pk_value, num_columns);
+	}
+	return true;
+}
+
+static bool
+do_transaction(uint32_t space_id, uint32_t ops_per_txn, uint32_t *start)
+{
+	if (box_txn_begin() != 0)
+		return false;
+	for (uint32_t j = 0; j < ops_per_txn; j++) {
+		char *tuple, *tuple_end;
+		if (!test_tuple(*start, &tuple, &tuple_end))
+			goto fail;
+		box_tuple_t *result;
+		if (box_replace(space_id, tuple,
+				tuple_end, &result) != 0)
+			goto fail;
+		++*start;
+	}
+	return box_txn_commit() == 0;
+
+fail:
+	box_txn_commit();
+	return false;
+}
+
+static int
+fiber_lua_func(struct lua_State *L)
+{
+	uint32_t space_id = luaL_checkinteger(L, 1);
+	uint32_t trans_per_fiber = luaL_checkinteger(L, 2);
+	uint32_t ops_per_txn = luaL_checkinteger(L, 3);
+	uint32_t num_columns = luaL_checkinteger(L, 4);
+	uint32_t start = luaL_checkinteger(L, 5);
+	if (!create_base_tuples(num_columns))
+		return luaT_error(L);
+	bool success = true;
+	for (uint32_t i = 0; success && i < trans_per_fiber; i++) {
+		success = do_transaction(space_id, ops_per_txn, &start);
+		fiber_sleep(0);
+	}
+	if (!success)
+		return luaT_error(L);
+	return 0;
+}
+
+LUA_API int
+luaopen_1mops_write_module(struct lua_State *L)
+{
+	static const struct luaL_Reg lib[] = {
+		{"fiber", fiber_lua_func},
+		{NULL, NULL},
+	};
+	luaL_register(L, "1mops_write_module", lib);
+	return 1;
+}
diff --git a/perf/lua/CMakeLists.txt b/perf/lua/CMakeLists.txt
@@ -64,7 +64,6 @@ function(create_perf_lua_test)
     DEPENDS ${BENCH_TARGET}-deps)
 endfunction()
 
-create_perf_lua_test(NAME 1mops_write)
 create_perf_lua_test(NAME box_select)
 create_perf_lua_test(NAME gh-7089-vclock-copy)
 create_perf_lua_test(NAME luafun)
@@ -73,6 +72,12 @@ create_perf_lua_test(NAME uri_escape_unescape)
 
 include_directories(${MSGPUCK_INCLUDE_DIRS})
 
+build_module(1mops_write_module 1mops_write_module.c)
+target_link_libraries(1mops_write_module msgpuck)
+create_perf_lua_test(NAME 1mops_write
+                     DEPENDS column_scan_module
+)
+
 build_module(column_scan_module column_scan_module.c)
 target_link_libraries(column_scan_module msgpuck)
 create_perf_lua_test(NAME column_scan