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)
--- /dev/null
+#include "test/lib/common.hpp"
+#include "test/lib/MemcachedCluster.hpp"
+
+#include "libmemcachedutil-1.0/pool.h"
+
+#include <atomic>
+#include <sstream>
+
+static atomic<bool> 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<stringstream> 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<worker_ctx *>(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<pthread_t, NUM_THREADS> 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);
+ }
+}
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<memcached_st *, POOL_MAX> 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<uintptr_t>(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<uintptr_t>(h));
+ Malloced val(memcached_get(h, s.c_str(), s.length(), nullptr, nullptr, &rc));
+ REQUIRE_SUCCESS(rc);
+ REQUIRE(*val);
+ REQUIRE(h == reinterpret_cast<memcached_st *>(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));
+ }
+ }
}
--- /dev/null
+#include "test/lib/common.hpp"
+
+#include "libmemcachedutil-1.0/pool.h"
+
+#include <cassert>
+
+#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<test_pool_context_st *>(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
projects / m6w6 / libmemcached / commitdiff
