mirror of
https://github.com/fnproject/fn.git
synced 2022-10-28 21:29:17 +03:00
add jaeger support, link hot container & req span (#840)
* add jaeger support, link hot container & req span * adds jaeger support now with FN_JAEGER_URL, there's a simple tutorial in the operating/metrics.md file now and it's pretty easy to get up and running. * links a hot request span to a hot container span. when we change this to sample at a lower ratio we'll need to finagle the hot container span to always sample or something, otherwise we'll hide that info. at least, since we're sampling at 100% for now if this is flipped on, can see freeze/unfreeze etc. if they hit. this is useful for debugging. note that zipkin's exporter does not follow the link at all, hence jaeger... and they're backed by the Cloud Empire now (CNCF) so we'll probably use it anyway. * vendor: add thrift for jaeger
This commit is contained in:
Reed Allman
536
vendor/git.apache.org/thrift.git/lib/cpp/test/TServerIntegrationTest.cpp
generated
vendored
Normal file
536
vendor/git.apache.org/thrift.git/lib/cpp/test/TServerIntegrationTest.cpp
generated
vendored
Normal file
@@ -0,0 +1,536 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
#define BOOST_TEST_MODULE TServerIntegrationTest
|
||||
#include <boost/test/auto_unit_test.hpp>
|
||||
#include <boost/atomic.hpp>
|
||||
#include <boost/date_time/posix_time/ptime.hpp>
|
||||
#include <boost/foreach.hpp>
|
||||
#include <boost/format.hpp>
|
||||
#include <boost/thread.hpp>
|
||||
#include <thrift/server/TSimpleServer.h>
|
||||
#include <thrift/server/TThreadPoolServer.h>
|
||||
#include <thrift/server/TThreadedServer.h>
|
||||
#include <thrift/stdcxx.h>
|
||||
#include <thrift/protocol/TBinaryProtocol.h>
|
||||
#include <thrift/transport/TServerSocket.h>
|
||||
#include <thrift/transport/TSocket.h>
|
||||
#include <thrift/transport/TTransport.h>
|
||||
#include "gen-cpp/ParentService.h"
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
using apache::thrift::concurrency::Guard;
|
||||
using apache::thrift::concurrency::Monitor;
|
||||
using apache::thrift::concurrency::Mutex;
|
||||
using apache::thrift::concurrency::Synchronized;
|
||||
using apache::thrift::protocol::TBinaryProtocol;
|
||||
using apache::thrift::protocol::TBinaryProtocolFactory;
|
||||
using apache::thrift::protocol::TProtocol;
|
||||
using apache::thrift::protocol::TProtocolFactory;
|
||||
using apache::thrift::transport::TServerSocket;
|
||||
using apache::thrift::transport::TServerTransport;
|
||||
using apache::thrift::transport::TSocket;
|
||||
using apache::thrift::transport::TTransport;
|
||||
using apache::thrift::transport::TTransportException;
|
||||
using apache::thrift::transport::TTransportFactory;
|
||||
using apache::thrift::server::TServer;
|
||||
using apache::thrift::server::TServerEventHandler;
|
||||
using apache::thrift::server::TSimpleServer;
|
||||
using apache::thrift::server::TThreadPoolServer;
|
||||
using apache::thrift::server::TThreadedServer;
|
||||
using apache::thrift::stdcxx::dynamic_pointer_cast;
|
||||
using apache::thrift::stdcxx::make_shared;
|
||||
using apache::thrift::stdcxx::shared_ptr;
|
||||
using apache::thrift::test::ParentServiceClient;
|
||||
using apache::thrift::test::ParentServiceIf;
|
||||
using apache::thrift::test::ParentServiceIfFactory;
|
||||
using apache::thrift::test::ParentServiceIfSingletonFactory;
|
||||
using apache::thrift::test::ParentServiceProcessor;
|
||||
using apache::thrift::test::ParentServiceProcessorFactory;
|
||||
using apache::thrift::TProcessor;
|
||||
using apache::thrift::TProcessorFactory;
|
||||
using boost::posix_time::milliseconds;
|
||||
|
||||
/**
|
||||
* preServe runs after listen() is successful, when we can connect
|
||||
*/
|
||||
class TServerReadyEventHandler : public TServerEventHandler, public Monitor {
|
||||
public:
|
||||
TServerReadyEventHandler() : isListening_(false), accepted_(0) {}
|
||||
virtual ~TServerReadyEventHandler() {}
|
||||
virtual void preServe() {
|
||||
Synchronized sync(*this);
|
||||
isListening_ = true;
|
||||
notify();
|
||||
}
|
||||
virtual void* createContext(shared_ptr<TProtocol> input,
|
||||
shared_ptr<TProtocol> output) {
|
||||
Synchronized sync(*this);
|
||||
++accepted_;
|
||||
notify();
|
||||
|
||||
(void)input;
|
||||
(void)output;
|
||||
return NULL;
|
||||
}
|
||||
bool isListening() const { return isListening_; }
|
||||
uint64_t acceptedCount() const { return accepted_; }
|
||||
|
||||
private:
|
||||
bool isListening_;
|
||||
uint64_t accepted_;
|
||||
};
|
||||
|
||||
/**
|
||||
* Reusing another generated test, just something to serve up
|
||||
*/
|
||||
class ParentHandler : public ParentServiceIf {
|
||||
public:
|
||||
ParentHandler() : generation_(0) {}
|
||||
|
||||
int32_t incrementGeneration() {
|
||||
Guard g(mutex_);
|
||||
return ++generation_;
|
||||
}
|
||||
|
||||
int32_t getGeneration() {
|
||||
Guard g(mutex_);
|
||||
return generation_;
|
||||
}
|
||||
|
||||
void addString(const std::string& s) {
|
||||
Guard g(mutex_);
|
||||
strings_.push_back(s);
|
||||
}
|
||||
|
||||
void getStrings(std::vector<std::string>& _return) {
|
||||
Guard g(mutex_);
|
||||
_return = strings_;
|
||||
}
|
||||
|
||||
void getDataWait(std::string& _return, const int32_t length) {
|
||||
THRIFT_UNUSED_VARIABLE(_return);
|
||||
THRIFT_UNUSED_VARIABLE(length);
|
||||
}
|
||||
|
||||
void onewayWait() {}
|
||||
|
||||
void exceptionWait(const std::string& message) { THRIFT_UNUSED_VARIABLE(message); }
|
||||
|
||||
void unexpectedExceptionWait(const std::string& message) { THRIFT_UNUSED_VARIABLE(message); }
|
||||
|
||||
protected:
|
||||
Mutex mutex_;
|
||||
int32_t generation_;
|
||||
std::vector<std::string> strings_;
|
||||
};
|
||||
|
||||
void autoSocketCloser(TSocket* pSock) {
|
||||
pSock->close();
|
||||
delete pSock;
|
||||
}
|
||||
|
||||
template <class TServerType>
|
||||
class TServerIntegrationTestFixture {
|
||||
public:
|
||||
TServerIntegrationTestFixture(const shared_ptr<TProcessorFactory>& _processorFactory)
|
||||
: pServer(new TServerType(_processorFactory,
|
||||
shared_ptr<TServerTransport>(
|
||||
new TServerSocket("localhost", 0)),
|
||||
shared_ptr<TTransportFactory>(new TTransportFactory),
|
||||
shared_ptr<TProtocolFactory>(new TBinaryProtocolFactory))),
|
||||
pEventHandler(shared_ptr<TServerReadyEventHandler>(new TServerReadyEventHandler)),
|
||||
bStressDone(false),
|
||||
bStressConnectionCount(0),
|
||||
bStressRequestCount(0) {
|
||||
pServer->setServerEventHandler(pEventHandler);
|
||||
}
|
||||
|
||||
TServerIntegrationTestFixture(const shared_ptr<TProcessor>& _processor)
|
||||
: pServer(
|
||||
new TServerType(_processor,
|
||||
shared_ptr<TServerTransport>(new TServerSocket("localhost", 0)),
|
||||
shared_ptr<TTransportFactory>(new TTransportFactory),
|
||||
shared_ptr<TProtocolFactory>(new TBinaryProtocolFactory))),
|
||||
pEventHandler(shared_ptr<TServerReadyEventHandler>(new TServerReadyEventHandler)),
|
||||
bStressDone(false),
|
||||
bStressConnectionCount(0),
|
||||
bStressRequestCount(0) {
|
||||
pServer->setServerEventHandler(pEventHandler);
|
||||
}
|
||||
|
||||
void startServer() {
|
||||
pServerThread.reset(new boost::thread(apache::thrift::stdcxx::bind(&TServerType::serve, pServer.get())));
|
||||
|
||||
// block until listen() completes so clients will be able to connect
|
||||
Synchronized sync(*(pEventHandler.get()));
|
||||
while (!pEventHandler->isListening()) {
|
||||
pEventHandler->wait();
|
||||
}
|
||||
|
||||
BOOST_TEST_MESSAGE(" server is listening");
|
||||
}
|
||||
|
||||
void blockUntilAccepted(uint64_t numAccepted) {
|
||||
Synchronized sync(*(pEventHandler.get()));
|
||||
while (pEventHandler->acceptedCount() < numAccepted) {
|
||||
pEventHandler->wait();
|
||||
}
|
||||
|
||||
BOOST_TEST_MESSAGE(boost::format(" server has accepted %1%") % numAccepted);
|
||||
}
|
||||
|
||||
void stopServer() {
|
||||
if (pServerThread) {
|
||||
pServer->stop();
|
||||
BOOST_TEST_MESSAGE(" server stop completed");
|
||||
|
||||
pServerThread->join();
|
||||
BOOST_TEST_MESSAGE(" server thread joined");
|
||||
pServerThread.reset();
|
||||
}
|
||||
}
|
||||
|
||||
~TServerIntegrationTestFixture() { stopServer(); }
|
||||
|
||||
/**
|
||||
* Performs a baseline test where some clients are opened and issue a single operation
|
||||
* and then disconnect at different intervals.
|
||||
* \param[in] numToMake the number of concurrent clients
|
||||
* \param[in] expectedHWM the high water mark we expect of concurrency
|
||||
* \param[in] purpose a description of the test for logging purposes
|
||||
*/
|
||||
void baseline(int64_t numToMake, int64_t expectedHWM, const std::string& purpose) {
|
||||
BOOST_TEST_MESSAGE(boost::format("Testing %1%: %2% with %3% clients, expect %4% HWM")
|
||||
% typeid(TServerType).name() % purpose % numToMake % expectedHWM);
|
||||
|
||||
startServer();
|
||||
|
||||
std::vector<shared_ptr<TSocket> > holdSockets;
|
||||
std::vector<shared_ptr<boost::thread> > holdThreads;
|
||||
|
||||
for (int64_t i = 0; i < numToMake; ++i) {
|
||||
shared_ptr<TSocket> pClientSock(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
holdSockets.push_back(pClientSock);
|
||||
shared_ptr<TProtocol> pClientProtocol(new TBinaryProtocol(pClientSock));
|
||||
ParentServiceClient client(pClientProtocol);
|
||||
pClientSock->open();
|
||||
client.incrementGeneration();
|
||||
holdThreads.push_back(shared_ptr<boost::thread>(
|
||||
new boost::thread(apache::thrift::stdcxx::bind(&TServerIntegrationTestFixture::delayClose,
|
||||
this,
|
||||
pClientSock,
|
||||
milliseconds(10 * numToMake)))));
|
||||
}
|
||||
|
||||
BOOST_CHECK_EQUAL(expectedHWM, pServer->getConcurrentClientCountHWM());
|
||||
|
||||
BOOST_FOREACH (shared_ptr<boost::thread> pThread, holdThreads) { pThread->join(); }
|
||||
holdThreads.clear();
|
||||
holdSockets.clear();
|
||||
|
||||
stopServer();
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper method used to close a connection after a delay.
|
||||
* \param[in] toClose the connection to close
|
||||
* \param[in] after the delay to impose
|
||||
*/
|
||||
void delayClose(shared_ptr<TTransport> toClose, boost::posix_time::time_duration after) {
|
||||
boost::this_thread::sleep(after);
|
||||
toClose->close();
|
||||
}
|
||||
|
||||
/**
|
||||
* \returns the server port number
|
||||
*/
|
||||
int getServerPort() {
|
||||
TServerSocket* pSock = dynamic_cast<TServerSocket*>(pServer->getServerTransport().get());
|
||||
if (!pSock) { throw std::logic_error("how come?"); }
|
||||
return pSock->getPort();
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs a stress test by spawning threads that connect, do a number of operations
|
||||
* and disconnect, then a random delay, then do it over again. This is done for a fixed
|
||||
* period of time to test for concurrency correctness.
|
||||
* \param[in] numToMake the number of concurrent clients
|
||||
*/
|
||||
void stress(int64_t numToMake, const boost::posix_time::time_duration& duration) {
|
||||
BOOST_TEST_MESSAGE(boost::format("Stress testing %1% with %2% clients for %3% seconds")
|
||||
% typeid(TServerType).name() % numToMake % duration.total_seconds());
|
||||
|
||||
startServer();
|
||||
|
||||
std::vector<shared_ptr<boost::thread> > holdThreads;
|
||||
for (int64_t i = 0; i < numToMake; ++i) {
|
||||
holdThreads.push_back(shared_ptr<boost::thread>(
|
||||
new boost::thread(apache::thrift::stdcxx::bind(&TServerIntegrationTestFixture::stressor, this))));
|
||||
}
|
||||
|
||||
boost::this_thread::sleep(duration);
|
||||
bStressDone = true;
|
||||
|
||||
BOOST_TEST_MESSAGE(boost::format(" serviced %1% connections (HWM %2%) totaling %3% requests")
|
||||
% bStressConnectionCount % pServer->getConcurrentClientCountHWM() % bStressRequestCount);
|
||||
|
||||
BOOST_FOREACH (shared_ptr<boost::thread> pThread, holdThreads) { pThread->join(); }
|
||||
holdThreads.clear();
|
||||
|
||||
BOOST_CHECK(bStressRequestCount > 0);
|
||||
|
||||
stopServer();
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper method to stress the system
|
||||
*/
|
||||
void stressor() {
|
||||
while (!bStressDone) {
|
||||
shared_ptr<TSocket> pSocket(new TSocket("localhost", getServerPort()), autoSocketCloser);
|
||||
shared_ptr<TProtocol> pProtocol(new TBinaryProtocol(pSocket));
|
||||
ParentServiceClient client(pProtocol);
|
||||
pSocket->open();
|
||||
bStressConnectionCount.fetch_add(1, boost::memory_order_relaxed);
|
||||
for (int i = 0; i < rand() % 1000; ++i) {
|
||||
client.incrementGeneration();
|
||||
bStressRequestCount.fetch_add(1, boost::memory_order_relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
shared_ptr<TServerType> pServer;
|
||||
shared_ptr<TServerReadyEventHandler> pEventHandler;
|
||||
shared_ptr<boost::thread> pServerThread;
|
||||
boost::atomic<bool> bStressDone;
|
||||
boost::atomic_int64_t bStressConnectionCount;
|
||||
boost::atomic_int64_t bStressRequestCount;
|
||||
};
|
||||
|
||||
template <class TServerType>
|
||||
class TServerIntegrationProcessorFactoryTestFixture
|
||||
: public TServerIntegrationTestFixture<TServerType> {
|
||||
public:
|
||||
TServerIntegrationProcessorFactoryTestFixture()
|
||||
: TServerIntegrationTestFixture<TServerType>(make_shared<ParentServiceProcessorFactory>(
|
||||
make_shared<ParentServiceIfSingletonFactory>(
|
||||
make_shared<ParentHandler>()))) {}
|
||||
};
|
||||
|
||||
template <class TServerType>
|
||||
class TServerIntegrationProcessorTestFixture : public TServerIntegrationTestFixture<TServerType> {
|
||||
public:
|
||||
TServerIntegrationProcessorTestFixture()
|
||||
: TServerIntegrationTestFixture<TServerType>(
|
||||
make_shared<ParentServiceProcessor>(make_shared<ParentHandler>())) {}
|
||||
};
|
||||
|
||||
BOOST_AUTO_TEST_SUITE(constructors)
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_simple_factory,
|
||||
TServerIntegrationProcessorFactoryTestFixture<TSimpleServer>) {
|
||||
baseline(3, 1, "factory");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_simple, TServerIntegrationProcessorTestFixture<TSimpleServer>) {
|
||||
baseline(3, 1, "processor");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threaded_factory,
|
||||
TServerIntegrationProcessorFactoryTestFixture<TThreadedServer>) {
|
||||
baseline(10, 10, "factory");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threaded, TServerIntegrationProcessorTestFixture<TThreadedServer>) {
|
||||
baseline(10, 10, "processor");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threaded_bound,
|
||||
TServerIntegrationProcessorTestFixture<TThreadedServer>) {
|
||||
pServer->setConcurrentClientLimit(4);
|
||||
baseline(10, 4, "limit by server framework");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threaded_stress,
|
||||
TServerIntegrationProcessorFactoryTestFixture<TThreadedServer>) {
|
||||
stress(10, boost::posix_time::seconds(3));
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threadpool_factory,
|
||||
TServerIntegrationProcessorFactoryTestFixture<TThreadPoolServer>) {
|
||||
pServer->getThreadManager()->threadFactory(
|
||||
shared_ptr<apache::thrift::concurrency::ThreadFactory>(
|
||||
new apache::thrift::concurrency::PlatformThreadFactory));
|
||||
pServer->getThreadManager()->start();
|
||||
|
||||
// thread factory has 4 threads as a default
|
||||
// thread factory however is a bad way to limit concurrent clients
|
||||
// as accept() will be called to grab a 5th client socket, in this case
|
||||
// and then the thread factory will block adding the thread to manage
|
||||
// that client.
|
||||
baseline(10, 5, "limit by thread manager");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threadpool,
|
||||
TServerIntegrationProcessorTestFixture<TThreadPoolServer>) {
|
||||
pServer->getThreadManager()->threadFactory(
|
||||
shared_ptr<apache::thrift::concurrency::ThreadFactory>(
|
||||
new apache::thrift::concurrency::PlatformThreadFactory));
|
||||
pServer->getThreadManager()->start();
|
||||
|
||||
// thread factory has 4 threads as a default
|
||||
// thread factory however is a bad way to limit concurrent clients
|
||||
// as accept() will be called to grab a 5th client socket, in this case
|
||||
// and then the thread factory will block adding the thread to manage
|
||||
// that client.
|
||||
baseline(10, 5, "limit by thread manager");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threadpool_bound,
|
||||
TServerIntegrationProcessorTestFixture<TThreadPoolServer>) {
|
||||
pServer->getThreadManager()->threadFactory(
|
||||
shared_ptr<apache::thrift::concurrency::ThreadFactory>(
|
||||
new apache::thrift::concurrency::PlatformThreadFactory));
|
||||
pServer->getThreadManager()->start();
|
||||
pServer->setConcurrentClientLimit(4);
|
||||
|
||||
baseline(10, 4, "server framework connection limit");
|
||||
}
|
||||
|
||||
BOOST_FIXTURE_TEST_CASE(test_threadpool_stress,
|
||||
TServerIntegrationProcessorTestFixture<TThreadPoolServer>) {
|
||||
pServer->getThreadManager()->threadFactory(
|
||||
shared_ptr<apache::thrift::concurrency::ThreadFactory>(
|
||||
new apache::thrift::concurrency::PlatformThreadFactory));
|
||||
pServer->getThreadManager()->start();
|
||||
|
||||
stress(10, boost::posix_time::seconds(3));
|
||||
}
|
||||
|
||||
BOOST_AUTO_TEST_SUITE_END()
|
||||
|
||||
BOOST_FIXTURE_TEST_SUITE(TServerIntegrationTest,
|
||||
TServerIntegrationProcessorTestFixture<TThreadedServer>)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(test_stop_with_interruptable_clients_connected) {
|
||||
// This tests THRIFT-2441 new behavior: stopping the server disconnects clients
|
||||
BOOST_TEST_MESSAGE("Testing stop with interruptable clients");
|
||||
|
||||
startServer();
|
||||
|
||||
shared_ptr<TSocket> pClientSock1(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock1->open();
|
||||
|
||||
shared_ptr<TSocket> pClientSock2(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock2->open();
|
||||
|
||||
// Ensure they have been accepted
|
||||
blockUntilAccepted(2);
|
||||
|
||||
// The test fixture destructor will force the sockets to disconnect
|
||||
// Prior to THRIFT-2441, pServer->stop() would hang until clients disconnected
|
||||
stopServer();
|
||||
|
||||
// extra proof the server end disconnected the clients
|
||||
uint8_t buf[1];
|
||||
BOOST_CHECK_EQUAL(0, pClientSock1->read(&buf[0], 1)); // 0 = disconnected
|
||||
BOOST_CHECK_EQUAL(0, pClientSock2->read(&buf[0], 1)); // 0 = disconnected
|
||||
}
|
||||
|
||||
BOOST_AUTO_TEST_CASE(test_stop_with_uninterruptable_clients_connected) {
|
||||
// This tests pre-THRIFT-2441 behavior: stopping the server blocks until clients
|
||||
// disconnect.
|
||||
BOOST_TEST_MESSAGE("Testing stop with uninterruptable clients");
|
||||
|
||||
dynamic_pointer_cast<TServerSocket>(pServer->getServerTransport())
|
||||
->setInterruptableChildren(false); // returns to pre-THRIFT-2441 behavior
|
||||
|
||||
startServer();
|
||||
|
||||
shared_ptr<TSocket> pClientSock1(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock1->open();
|
||||
|
||||
shared_ptr<TSocket> pClientSock2(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock2->open();
|
||||
|
||||
// Ensure they have been accepted
|
||||
blockUntilAccepted(2);
|
||||
|
||||
boost::thread t1(apache::thrift::stdcxx::bind(&TServerIntegrationTestFixture::delayClose,
|
||||
this,
|
||||
pClientSock1,
|
||||
milliseconds(250)));
|
||||
boost::thread t2(apache::thrift::stdcxx::bind(&TServerIntegrationTestFixture::delayClose,
|
||||
this,
|
||||
pClientSock2,
|
||||
milliseconds(250)));
|
||||
|
||||
// Once the clients disconnect the server will stop
|
||||
stopServer();
|
||||
BOOST_CHECK(pServer->getConcurrentClientCountHWM() > 0);
|
||||
t1.join();
|
||||
t2.join();
|
||||
}
|
||||
|
||||
BOOST_AUTO_TEST_CASE(test_concurrent_client_limit) {
|
||||
startServer();
|
||||
BOOST_TEST_MESSAGE("Testing the concurrent client limit");
|
||||
|
||||
BOOST_CHECK_EQUAL(INT64_MAX, pServer->getConcurrentClientLimit());
|
||||
pServer->setConcurrentClientLimit(2);
|
||||
BOOST_CHECK_EQUAL(0, pServer->getConcurrentClientCount());
|
||||
BOOST_CHECK_EQUAL(2, pServer->getConcurrentClientLimit());
|
||||
|
||||
shared_ptr<TSocket> pClientSock1(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock1->open();
|
||||
blockUntilAccepted(1);
|
||||
BOOST_CHECK_EQUAL(1, pServer->getConcurrentClientCount());
|
||||
|
||||
shared_ptr<TSocket> pClientSock2(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock2->open();
|
||||
blockUntilAccepted(2);
|
||||
BOOST_CHECK_EQUAL(2, pServer->getConcurrentClientCount());
|
||||
|
||||
// a third client cannot connect until one of the other two closes
|
||||
boost::thread t2(apache::thrift::stdcxx::bind(&TServerIntegrationTestFixture::delayClose,
|
||||
this,
|
||||
pClientSock2,
|
||||
milliseconds(250)));
|
||||
shared_ptr<TSocket> pClientSock3(new TSocket("localhost", getServerPort()),
|
||||
autoSocketCloser);
|
||||
pClientSock2->open();
|
||||
blockUntilAccepted(2);
|
||||
BOOST_CHECK_EQUAL(2, pServer->getConcurrentClientCount());
|
||||
BOOST_CHECK_EQUAL(2, pServer->getConcurrentClientCountHWM());
|
||||
|
||||
stopServer();
|
||||
BOOST_CHECK(pServer->getConcurrentClientCountHWM() > 0);
|
||||
t2.join();
|
||||
}
|
||||
|
||||
BOOST_AUTO_TEST_SUITE_END()
|
||||
Reference in New Issue
Block a user