From: Michael Wallner Date: Tue, 6 Oct 2020 07:43:17 +0000 (+0200) Subject: testing: pools X-Git-Tag: 1.1.0-beta1~236^2~19 X-Git-Url: https://git.m6w6.name/?a=commitdiff_plain;h=64f8472e82873d5f7e9f169bdf160b0f4f8ffb5c;p=m6w6%2Flibmemcached testing: pools --- diff --git a/CMake/_Include.cmake b/CMake/_Include.cmake index a6b5463c..3c00c79c 100644 --- a/CMake/_Include.cmake +++ b/CMake/_Include.cmake @@ -78,6 +78,7 @@ check_header(io.h) check_header(limits.h) check_header(netdb.h) check_header(poll.h) +check_header(semaphore.h) check_header(stddef.h) check_header(stdlib.h) check_header(strings.h) diff --git a/test/tests/memcached/regression/lp_000-962-815.cpp b/test/tests/memcached/regression/lp_000-962-815.cpp new file mode 100644 index 00000000..658c339e --- /dev/null +++ b/test/tests/memcached/regression/lp_000-962-815.cpp @@ -0,0 +1,92 @@ +#include "test/lib/common.hpp" +#include "test/lib/MemcachedCluster.hpp" + +#include "libmemcachedutil-1.0/pool.h" + +#include +#include + +static atomic running = false; + +static inline void set_running(bool is) { + running.store(is, memory_order_release); +} +static inline bool is_running() { + return running.load(memory_order_consume); +} + +struct worker_ctx { + memcached_pool_st *pool; + vector errors; + + explicit worker_ctx(memcached_st *memc) + : pool{memcached_pool_create(memc, 5, 10)} + , errors{} + { + } + + ~worker_ctx() { + memcached_pool_destroy(pool); + for (const auto &err : errors) { + cerr << err.str() << endl; + } + } + + stringstream &err() { + return errors[errors.size()]; + } +}; + +static void *worker(void *arg) { + auto *ctx = static_cast(arg); + + while (is_running()) { + memcached_return_t rc; + timespec block{5, 0}; + auto *mc = memcached_pool_fetch(ctx->pool, &block, &rc); + + if (!mc || memcached_failed(rc)) { + ctx->err() << "failed to fetch connection from pool: " + << memcached_strerror(nullptr, rc); + this_thread::sleep_for(100ms); + } + + auto rs = random_ascii_string(12); + rc = memcached_set(mc, rs.c_str(), rs.length(), rs.c_str(), rs.length(), 0, 0); + if (memcached_failed(rc)) { + ctx->err() << "failed to memcached_set() " + << memcached_last_error_message(mc); + } + rc = memcached_pool_release(ctx->pool, mc); + if (memcached_failed(rc)) { + ctx->err() << "failed to release connection to pool: " + << memcached_strerror(nullptr, rc); + } + } + + return ctx; +} + +TEST_CASE("memcached_regression_lp962815") { + auto test = MemcachedCluster::mixed(); + auto memc = &test.memc; + + constexpr auto NUM_THREADS = 20; + array tid; + worker_ctx ctx{memc}; + + set_running(true); + + for (auto &t : tid) { + REQUIRE(0 == pthread_create(&t, nullptr, worker, &ctx)); + } + + this_thread::sleep_for(5s); + set_running(false); + + for (auto t : tid) { + void *ret = nullptr; + REQUIRE(0 == pthread_join(t, &ret)); + REQUIRE(ret == &ctx); + } +} diff --git a/test/tests/memcached/util.cpp b/test/tests/memcached/util.cpp index 9612b1d0..3fb018a3 100644 --- a/test/tests/memcached/util.cpp +++ b/test/tests/memcached/util.cpp @@ -73,4 +73,94 @@ TEST_CASE("memcached_util") { REQUIRE_SUCCESS(rc); } } + + SECTION("pool") { + SECTION("deprecated") { + auto conf = "--SERVER=host10.example.com --SERVER=host11.example.com --SERVER=host10.example.com --POOL-MIN=10 --POOL-MAX=32"; + auto pool = memcached_pool(S(conf)); + REQUIRE(pool); + + memcached_return_t rc; + auto memc = memcached_pool_pop(pool, false, &rc); + REQUIRE(memc); + REQUIRE(MEMCACHED_SUCCESS == rc); + + REQUIRE(MEMCACHED_SUCCESS == memcached_pool_push(pool, memc)); + REQUIRE(nullptr == memcached_pool_destroy(pool)); + } + + SECTION("basic") { + auto test = MemcachedCluster::mixed(); + auto memc = &test.memc; + memcached_return_t rc; + + constexpr auto POOL_MIN = 5; + constexpr auto POOL_MAX = 10; + + auto pool = memcached_pool_create(memc, POOL_MIN, POOL_MAX); + REQUIRE(pool); + + array hold{}; + for (auto &h : hold) { + h = memcached_pool_fetch(pool, nullptr, &rc); + REQUIRE_SUCCESS(rc); + REQUIRE(h); + } + + SECTION("depleted") { + REQUIRE(nullptr == memcached_pool_fetch(pool, nullptr, &rc)); + REQUIRE_RC(MEMCACHED_NOTFOUND, rc); + } + + SECTION("usable") { + for (auto h : hold) { + auto s = to_string(reinterpret_cast(h)); + REQUIRE_SUCCESS(memcached_set(h, s.c_str(), s.length(), s.c_str(), s.length(), 0, 0)); + } + for (auto h : hold) { + auto s = to_string(reinterpret_cast(h)); + Malloced val(memcached_get(h, s.c_str(), s.length(), nullptr, nullptr, &rc)); + REQUIRE_SUCCESS(rc); + REQUIRE(*val); + REQUIRE(h == reinterpret_cast(stoul(*val))); + } + + REQUIRE_SUCCESS(memcached_set(hold[0], S(__func__), "0", 1, 0, 0)); + uint64_t inc = 0; + for (auto h : hold) { + uint64_t val; + REQUIRE_SUCCESS(memcached_increment(h, S(__func__), 1, &val)); + CHECK(++inc == val); + } + } + + SECTION("behaviors") { + uint64_t val; + REQUIRE_SUCCESS(memcached_pool_behavior_get(pool, MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK, &val)); + REQUIRE_FALSE(val == 9999); + REQUIRE_SUCCESS(memcached_pool_behavior_set(pool, MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK, 9999)); + REQUIRE_SUCCESS(memcached_pool_behavior_get(pool, MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK, &val)); + REQUIRE(val == 9999); + + for (auto &h : hold) { + REQUIRE_FALSE(9999 == memcached_behavior_get(h, MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK)); + REQUIRE_SUCCESS(memcached_pool_release(pool, h)); + h = memcached_pool_fetch(pool, nullptr, &rc); + REQUIRE_SUCCESS(rc); + REQUIRE(h); + REQUIRE(9999 == memcached_behavior_get(h, MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK)); + } + } + + for (auto h : hold) { + REQUIRE_SUCCESS(memcached_pool_release(pool, h)); + auto again = memcached_pool_fetch(pool, nullptr, &rc); + REQUIRE_SUCCESS(rc); + REQUIRE(again); + REQUIRE_SUCCESS(memcached_pool_release(pool, again)); + } + + REQUIRE(memc == memcached_pool_destroy(pool)); + } + } } diff --git a/test/tests/memcached/util_pool_thread.cpp b/test/tests/memcached/util_pool_thread.cpp new file mode 100644 index 00000000..88330284 --- /dev/null +++ b/test/tests/memcached/util_pool_thread.cpp @@ -0,0 +1,104 @@ +#include "test/lib/common.hpp" + +#include "libmemcachedutil-1.0/pool.h" + +#include + +#if HAVE_SEMAPHORE_H + +#include "semaphore.h" + +#ifndef __APPLE__ +struct test_pool_context_st { + volatile memcached_return_t rc; + memcached_pool_st *pool; + memcached_st *memc; + sem_t _lock; + + test_pool_context_st(memcached_pool_st *pool_arg, memcached_st *memc_arg) : + rc(MEMCACHED_FAILURE), + pool(pool_arg), + memc(memc_arg) { + sem_init(&_lock, 0, 0); + } + + void wait() { + sem_wait(&_lock); + } + + void release() { + sem_post(&_lock); + } + + ~test_pool_context_st() { + sem_destroy(&_lock); + } +}; + +static void *connection_release(void *arg) { + assert(arg != nullptr); + + this_thread::sleep_for(2s); + auto res = static_cast(arg); + res->rc = memcached_pool_release(res->pool, res->memc); + res->release(); + + pthread_exit(arg); +} +#endif + +TEST_CASE("memcached_util_pool_thread") { +#ifdef __APPLE__ + SUCCEED("skip: pthreads"); +#else + MemcachedPtr memc; + auto pool = memcached_pool_create(*memc, 1, 1); + REQUIRE(pool); + + memcached_return_t rc; + auto pool_memc = memcached_pool_fetch(pool, nullptr, &rc); + REQUIRE(MEMCACHED_SUCCESS == rc); + REQUIRE(pool_memc); + + /* + @note This comment was written to describe what was believed to be the original authors intent. + + This portion of the test creates a thread that will wait until told to free a memcached_st + that will be grabbed by the main thread. + + It is believed that this tests whether or not we are handling ownership correctly. + */ + pthread_t tid; + test_pool_context_st item(pool, pool_memc); + + REQUIRE(0 == pthread_create(&tid, nullptr, connection_release, &item)); + item.wait(); + + memcached_st *pop_memc; + // We do a hard loop, and try N times + int counter = 5; + do { + struct timespec relative_time = {0, 0}; + pop_memc = memcached_pool_fetch(pool, &relative_time, &rc); + + if (memcached_success(rc)) { + break; + } +cerr << "rc == " << memcached_strerror(nullptr, rc); + if (memcached_failed(rc)) { + REQUIRE_FALSE(pop_memc); + REQUIRE(rc != MEMCACHED_TIMEOUT); // As long as relative_time is zero, MEMCACHED_TIMEOUT is invalid + } + } while (--counter); + + // Cleanup thread since we will exit once we test. + REQUIRE(0 == pthread_join(tid, nullptr)); + REQUIRE(MEMCACHED_SUCCESS == rc); + REQUIRE(pool_memc == pop_memc); + REQUIRE(MEMCACHED_SUCCESS == memcached_pool_release(pool, pop_memc)); + REQUIRE(memcached_pool_destroy(pool) == *memc); + +#endif __APPLE__ +} + +#endif // HAVE_SEMAPHORE_H