Thrift-2029:Port C++ tests to Windows
Client: cpp
Patch: Ben Craig
Updates cpp tests to work with windows and c++11
diff --git a/test/cpp/src/StressTest.cpp b/test/cpp/src/StressTest.cpp
index 7da3db0..dfe8350 100644
--- a/test/cpp/src/StressTest.cpp
+++ b/test/cpp/src/StressTest.cpp
@@ -39,9 +39,11 @@
#include <stdexcept>
#include <sstream>
#include <map>
+#if _WIN32
+ #include <thrift/windows/TWinsockSingleton.h>
+#endif
using namespace std;
-using namespace boost;
using namespace apache::thrift;
using namespace apache::thrift::protocol;
@@ -169,6 +171,7 @@
int8_t arg = 1;
int8_t result;
result =_client->echoByte(arg);
+ (void)result;
assert(result == arg);
}
}
@@ -178,6 +181,7 @@
int32_t arg = 1;
int32_t result;
result =_client->echoI32(arg);
+ (void)result;
assert(result == arg);
}
}
@@ -187,6 +191,7 @@
int64_t arg = 1;
int64_t result;
result =_client->echoI64(arg);
+ (void)result;
assert(result == arg);
}
}
@@ -212,8 +217,31 @@
Monitor _sleep;
};
+class TStartObserver : public apache::thrift::server::TServerEventHandler
+{
+public:
+ TStartObserver() : awake_(false) {}
+ virtual void preServe()
+ {
+ apache::thrift::concurrency::Synchronized s(m_);
+ awake_ = true;
+ m_.notifyAll();
+ }
+ void waitForService()
+ {
+ apache::thrift::concurrency::Synchronized s(m_);
+ while(!awake_)
+ m_.waitForever();
+ }
+ private:
+ apache::thrift::concurrency::Monitor m_;
+ bool awake_;
+};
int main(int argc, char **argv) {
+#if _WIN32
+ transport::TWinsockSingleton::create();
+#endif
int port = 9091;
string serverType = "thread-pool";
@@ -323,7 +351,7 @@
} catch(std::exception& e) {
cerr << e.what() << endl;
- cerr << usage;
+ cerr << usage.str();
}
boost::shared_ptr<PlatformThreadFactory> threadFactory = boost::shared_ptr<PlatformThreadFactory>(new PlatformThreadFactory());
@@ -376,15 +404,15 @@
boost::shared_ptr<TTransportFactory>(new TPipedTransportFactory(fileTransport));
}
- boost::shared_ptr<Thread> serverThread;
+ boost::shared_ptr<TServer> server;
if (serverType == "simple") {
- serverThread = threadFactory->newThread(boost::shared_ptr<TServer>(new TSimpleServer(serviceProcessor, serverSocket, transportFactory, protocolFactory)));
+ server.reset(new TSimpleServer(serviceProcessor, serverSocket, transportFactory, protocolFactory));
} else if (serverType == "threaded") {
- serverThread = threadFactory->newThread(boost::shared_ptr<TServer>(new TThreadedServer(serviceProcessor, serverSocket, transportFactory, protocolFactory)));
+ server.reset(new TThreadedServer(serviceProcessor, serverSocket, transportFactory, protocolFactory));
} else if (serverType == "thread-pool") {
@@ -392,15 +420,19 @@
threadManager->threadFactory(threadFactory);
threadManager->start();
- serverThread = threadFactory->newThread(boost::shared_ptr<TServer>(new TThreadPoolServer(serviceProcessor, serverSocket, transportFactory, protocolFactory, threadManager)));
+ server.reset(new TThreadPoolServer(serviceProcessor, serverSocket, transportFactory, protocolFactory, threadManager));
}
+ boost::shared_ptr<TStartObserver> observer(new TStartObserver);
+ server->setServerEventHandler(observer);
+ boost::shared_ptr<Thread> serverThread = threadFactory->newThread(server);
+
cerr << "Starting the server on port " << port << endl;
serverThread->start();
+ observer->waitForService();
// If we aren't running clients, just wait forever for external clients
-
if (clientCount == 0) {
serverThread->join();
}
@@ -463,7 +495,7 @@
for (set<boost::shared_ptr<Thread> >::iterator ix = clientThreads.begin(); ix != clientThreads.end(); ix++) {
- boost::shared_ptr<ClientThread> client = dynamic_pointer_cast<ClientThread>((*ix)->runnable());
+ boost::shared_ptr<ClientThread> client = boost::dynamic_pointer_cast<ClientThread>((*ix)->runnable());
int64_t delta = client->_endTime - client->_startTime;
diff --git a/test/cpp/src/StressTestNonBlocking.cpp b/test/cpp/src/StressTestNonBlocking.cpp
index c230c84..20320c7 100644
--- a/test/cpp/src/StressTestNonBlocking.cpp
+++ b/test/cpp/src/StressTestNonBlocking.cpp
@@ -35,7 +35,6 @@
#include "Service.h"
-#include <unistd.h>
#include <boost/shared_ptr.hpp>
#include <iostream>
@@ -43,9 +42,11 @@
#include <stdexcept>
#include <sstream>
#include <map>
+#if _WIN32
+ #include <thrift/windows/TWinsockSingleton.h>
+#endif
using namespace std;
-using namespace boost;
using namespace apache::thrift;
using namespace apache::thrift::protocol;
@@ -84,7 +85,7 @@
void echoVoid() {
count("echoVoid");
// Sleep to simulate work
- usleep(5000);
+ THRIFT_SLEEP_USEC(1);
return;
}
@@ -175,6 +176,7 @@
int8_t arg = 1;
int8_t result;
result =_client->echoByte(arg);
+ (void)result;
assert(result == arg);
}
}
@@ -184,6 +186,7 @@
int32_t arg = 1;
int32_t result;
result =_client->echoI32(arg);
+ (void)result;
assert(result == arg);
}
}
@@ -193,6 +196,7 @@
int64_t arg = 1;
int64_t result;
result =_client->echoI64(arg);
+ (void)result;
assert(result == arg);
}
}
@@ -220,13 +224,16 @@
int main(int argc, char **argv) {
+#if _WIN32
+ transport::TWinsockSingleton::create();
+#endif
int port = 9091;
string serverType = "simple";
string protocolType = "binary";
- size_t workerCount = 4;
- size_t clientCount = 20;
- size_t loopCount = 50000;
+ uint32_t workerCount = 4;
+ uint32_t clientCount = 20;
+ uint32_t loopCount = 1000;
TType loopType = T_VOID;
string callName = "echoVoid";
bool runServer = true;
@@ -318,7 +325,7 @@
} catch(std::exception& e) {
cerr << e.what() << endl;
- cerr << usage;
+ cerr << usage.str();
}
boost::shared_ptr<PlatformThreadFactory> threadFactory = boost::shared_ptr<PlatformThreadFactory>(new PlatformThreadFactory());
@@ -397,7 +404,7 @@
serverThread2->join();
}
}
- sleep(1);
+ THRIFT_SLEEP_SEC(1);
if (clientCount > 0) {
@@ -414,7 +421,7 @@
else if (callName == "echoString") { loopType = T_STRING;}
else {throw invalid_argument("Unknown service call "+callName);}
- for (size_t ix = 0; ix < clientCount; ix++) {
+ for (uint32_t ix = 0; ix < clientCount; ix++) {
boost::shared_ptr<TSocket> socket(new TSocket("127.0.0.1", port + (ix % 2)));
boost::shared_ptr<TFramedTransport> framedSocket(new TFramedTransport(socket));
@@ -456,7 +463,7 @@
for (set<boost::shared_ptr<Thread> >::iterator ix = clientThreads.begin(); ix != clientThreads.end(); ix++) {
- boost::shared_ptr<ClientThread> client = dynamic_pointer_cast<ClientThread>((*ix)->runnable());
+ boost::shared_ptr<ClientThread> client = boost::dynamic_pointer_cast<ClientThread>((*ix)->runnable());
int64_t delta = client->_endTime - client->_startTime;
diff --git a/test/cpp/src/TestClient.cpp b/test/cpp/src/TestClient.cpp
index fbf04f0..cd78505 100644
--- a/test/cpp/src/TestClient.cpp
+++ b/test/cpp/src/TestClient.cpp
@@ -21,8 +21,6 @@
#include <inttypes.h>
#include <iostream>
-#include <unistd.h>
-#include <sys/time.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
#include <thrift/transport/THttpClient.h>
@@ -34,11 +32,13 @@
#include <boost/shared_ptr.hpp>
#include <boost/program_options.hpp>
-#include <tr1/functional>
+#include <thrift/cxxfunctional.h>
+#if _WIN32
+ #include <thrift/windows/TWinsockSingleton.h>
+#endif
#include "ThriftTest.h"
-using namespace boost;
using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::protocol;
@@ -54,7 +54,7 @@
int64_t ret;
struct timeval tv;
- gettimeofday(&tv, NULL);
+ THRIFT_GETTIMEOFDAY(&tv, NULL);
ret = tv.tv_sec;
ret = ret*1000*1000 + tv.tv_usec;
return ret;
@@ -69,7 +69,7 @@
client->recv_testString(s);
cout << "testString: " << s << endl;
} catch (TException& exn) {
- cout << "Error: " << exn.what() << endl;
+ cout << "Error: " << exn.what() << endl;
}
event_base_loopbreak(base); // end test
@@ -86,11 +86,14 @@
client = new ThriftTestCobClient(channel, protocolFactory);
client->testString(tr1::bind(testString_clientReturn, host, port, base, protocolFactory, std::tr1::placeholders::_1), "Test");
} catch (TException& exn) {
- cout << "Error: " << exn.what() << endl;
+ cout << "Error: " << exn.what() << endl;
}
}
int main(int argc, char** argv) {
+#if _WIN32
+ transport::TWinsockSingleton::create();
+#endif
string host = "localhost";
int port = 9090;
int numTests = 1;
@@ -99,28 +102,28 @@
string protocol_type = "binary";
string domain_socket = "";
- program_options::options_description desc("Allowed options");
+ boost::program_options::options_description desc("Allowed options");
desc.add_options()
("help,h", "produce help message")
- ("host", program_options::value<string>(&host)->default_value(host), "Host to connect")
- ("port", program_options::value<int>(&port)->default_value(port), "Port number to connect")
- ("domain-socket", program_options::value<string>(&domain_socket)->default_value(domain_socket), "Domain Socket (e.g. /tmp/ThriftTest.thrift), instead of host and port")
- ("transport", program_options::value<string>(&transport_type)->default_value(transport_type), "Transport: buffered, framed, http, evhttp")
- ("protocol", program_options::value<string>(&protocol_type)->default_value(protocol_type), "Protocol: binary, json")
+ ("host", boost::program_options::value<string>(&host)->default_value(host), "Host to connect")
+ ("port", boost::program_options::value<int>(&port)->default_value(port), "Port number to connect")
+ ("domain-socket", boost::program_options::value<string>(&domain_socket)->default_value(domain_socket), "Domain Socket (e.g. /tmp/ThriftTest.thrift), instead of host and port")
+ ("transport", boost::program_options::value<string>(&transport_type)->default_value(transport_type), "Transport: buffered, framed, http, evhttp")
+ ("protocol", boost::program_options::value<string>(&protocol_type)->default_value(protocol_type), "Protocol: binary, json")
("ssl", "Encrypted Transport using SSL")
- ("testloops,n", program_options::value<int>(&numTests)->default_value(numTests), "Number of Tests")
+ ("testloops,n", boost::program_options::value<int>(&numTests)->default_value(numTests), "Number of Tests")
;
- program_options::variables_map vm;
- program_options::store(program_options::parse_command_line(argc, argv, desc), vm);
- program_options::notify(vm);
+ boost::program_options::variables_map vm;
+ boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
+ boost::program_options::notify(vm);
if (vm.count("help")) {
cout << desc << "\n";
return 1;
}
- try {
+ try {
if (!protocol_type.empty()) {
if (protocol_type == "binary") {
} else if (protocol_type == "json") {
@@ -210,7 +213,7 @@
boost::shared_ptr<TAsyncChannel> channel(new TEvhttpClientChannel(host.c_str(), "/", host.c_str(), port, base));
ThriftTestCobClient* client = new ThriftTestCobClient(channel, protocolFactory.get());
client->testVoid(tr1::bind(testVoid_clientReturn, host.c_str(), port, base, protocolFactory.get(), std::tr1::placeholders::_1));
-
+
event_base_loop(base, 0);
return 0;
}
@@ -574,7 +577,7 @@
printf(" void\nFAILURE\n");
failCount++;
- } catch(TException& e) {
+ } catch(const TException&) {
printf(" Caught TException\n");
}
@@ -622,9 +625,9 @@
/* test oneway void */
{
- printf("testClient.testOneway(3) =>");
+ printf("testClient.testOneway(1) =>");
uint64_t startOneway = now();
- testClient.testOneway(3);
+ testClient.testOneway(1);
uint64_t elapsed = now() - startOneway;
if (elapsed > 200 * 1000) { // 0.2 seconds
printf(" FAILURE - took %.2f ms\n", (double)elapsed/1000.0);
diff --git a/test/cpp/src/TestServer.cpp b/test/cpp/src/TestServer.cpp
index c99fbac..adb8fd1 100644
--- a/test/cpp/src/TestServer.cpp
+++ b/test/cpp/src/TestServer.cpp
@@ -46,9 +46,11 @@
#include <boost/program_options.hpp>
#include <signal.h>
+#if _WIN32
+ #include <thrift/windows/TWinsockSingleton.h>
+#endif
using namespace std;
-using namespace boost;
using namespace apache::thrift;
using namespace apache::thrift::concurrency;
@@ -266,7 +268,7 @@
(void) arg3;
(void) arg4;
(void) arg5;
-
+
printf("testMulti()\n");
hello.string_thing = "Hello2";
@@ -314,9 +316,9 @@
}
}
- void testOneway(int sleepFor) {
+ void testOneway(const int32_t sleepFor) {
printf("testOneway(%d): Sleeping...\n", sleepFor);
- sleep(sleepFor);
+ THRIFT_SLEEP_SEC(sleepFor);
printf("testOneway(%d): done sleeping!\n", sleepFor);
}
};
@@ -447,7 +449,7 @@
}
virtual void testInsanity(std::tr1::function<void(std::map<UserId, std::map<Numberz::type, Insanity> > const& _return)> cob, const Insanity& argument) {
- std::map<UserId, std::map<Numberz::type, Insanity> > res;
+ std::map<UserId, std::map<Numberz::type, Insanity> > res;
_delegate->testInsanity(res, argument);
cob(res);
}
@@ -490,6 +492,9 @@
int main(int argc, char **argv) {
+#if _WIN32
+ transport::TWinsockSingleton::create();
+#endif
int port = 9090;
bool ssl = false;
string transport_type = "buffered";
@@ -498,34 +503,34 @@
string domain_socket = "";
size_t workers = 4;
-
- program_options::options_description desc("Allowed options");
+
+ boost::program_options::options_description desc("Allowed options");
desc.add_options()
("help,h", "produce help message")
- ("port", program_options::value<int>(&port)->default_value(port), "Port number to listen")
- ("domain-socket", program_options::value<string>(&domain_socket)->default_value(domain_socket),
+ ("port", boost::program_options::value<int>(&port)->default_value(port), "Port number to listen")
+ ("domain-socket", boost::program_options::value<string>(&domain_socket)->default_value(domain_socket),
"Unix Domain Socket (e.g. /tmp/ThriftTest.thrift)")
- ("server-type", program_options::value<string>(&server_type)->default_value(server_type),
+ ("server-type", boost::program_options::value<string>(&server_type)->default_value(server_type),
"type of server, \"simple\", \"thread-pool\", \"threaded\", or \"nonblocking\"")
- ("transport", program_options::value<string>(&transport_type)->default_value(transport_type),
+ ("transport", boost::program_options::value<string>(&transport_type)->default_value(transport_type),
"transport: buffered, framed, http")
- ("protocol", program_options::value<string>(&protocol_type)->default_value(protocol_type),
+ ("protocol", boost::program_options::value<string>(&protocol_type)->default_value(protocol_type),
"protocol: binary, json")
("ssl", "Encrypted Transport using SSL")
("processor-events", "processor-events")
- ("workers,n", program_options::value<size_t>(&workers)->default_value(workers),
+ ("workers,n", boost::program_options::value<size_t>(&workers)->default_value(workers),
"Number of thread pools workers. Only valid for thread-pool server type")
;
- program_options::variables_map vm;
- program_options::store(program_options::parse_command_line(argc, argv, desc), vm);
- program_options::notify(vm);
+ boost::program_options::variables_map vm;
+ boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
+ boost::program_options::notify(vm);
if (vm.count("help")) {
cout << desc << "\n";
return 1;
}
-
+
try {
if (!server_type.empty()) {
if (server_type == "simple") {
@@ -536,7 +541,7 @@
throw invalid_argument("Unknown server type "+server_type);
}
}
-
+
if (!protocol_type.empty()) {
if (protocol_type == "binary") {
} else if (protocol_type == "json") {
@@ -562,7 +567,6 @@
if (vm.count("ssl")) {
ssl = true;
- signal(SIGPIPE, SIG_IGN);
}
// Dispatcher
@@ -578,12 +582,12 @@
// Processor
boost::shared_ptr<TestHandler> testHandler(new TestHandler());
boost::shared_ptr<ThriftTestProcessor> testProcessor(new ThriftTestProcessor(testHandler));
-
+
if (vm.count("processor-events")) {
testProcessor->setEventHandler(boost::shared_ptr<TProcessorEventHandler>(
new TestProcessorEventHandler()));
}
-
+
// Transport
boost::shared_ptr<TSSLSocketFactory> sslSocketFactory;
boost::shared_ptr<TServerSocket> serverSocket;
@@ -607,15 +611,15 @@
// Factory
boost::shared_ptr<TTransportFactory> transportFactory;
-
+
if (transport_type == "http" && server_type != "nonblocking") {
- boost::shared_ptr<TTransportFactory> httpTransportFactory(new THttpServerTransportFactory());
+ boost::shared_ptr<TTransportFactory> httpTransportFactory(new THttpServerTransportFactory());
transportFactory = httpTransportFactory;
} else if (transport_type == "framed") {
- boost::shared_ptr<TTransportFactory> framedTransportFactory(new TFramedTransportFactory());
+ boost::shared_ptr<TTransportFactory> framedTransportFactory(new TFramedTransportFactory());
transportFactory = framedTransportFactory;
} else {
- boost::shared_ptr<TTransportFactory> bufferedTransportFactory(new TBufferedTransportFactory());
+ boost::shared_ptr<TTransportFactory> bufferedTransportFactory(new TBufferedTransportFactory());
transportFactory = bufferedTransportFactory;
}
@@ -628,14 +632,13 @@
cout << endl;
// Server
+ boost::shared_ptr<apache::thrift::server::TServer> server;
+
if (server_type == "simple") {
- TSimpleServer simpleServer(testProcessor,
+ server.reset(new TSimpleServer(testProcessor,
serverSocket,
transportFactory,
- protocolFactory);
-
- simpleServer.serve();
-
+ protocolFactory));
} else if (server_type == "thread-pool") {
boost::shared_ptr<ThreadManager> threadManager =
@@ -648,37 +651,49 @@
threadManager->start();
- TThreadPoolServer threadPoolServer(testProcessor,
+ server.reset(new TThreadPoolServer(testProcessor,
serverSocket,
transportFactory,
protocolFactory,
- threadManager);
-
- threadPoolServer.serve();
-
+ threadManager));
} else if (server_type == "threaded") {
- TThreadedServer threadedServer(testProcessor,
+ server.reset(new TThreadedServer(testProcessor,
serverSocket,
transportFactory,
- protocolFactory);
-
- threadedServer.serve();
-
+ protocolFactory));
} else if (server_type == "nonblocking") {
if(transport_type == "http") {
boost::shared_ptr<TestHandlerAsync> testHandlerAsync(new TestHandlerAsync(testHandler));
boost::shared_ptr<TAsyncProcessor> testProcessorAsync(new ThriftTestAsyncProcessor(testHandlerAsync));
boost::shared_ptr<TAsyncBufferProcessor> testBufferProcessor(new TAsyncProtocolProcessor(testProcessorAsync, protocolFactory));
-
+
+ // not loading nonblockingServer into "server" because
+ // TEvhttpServer doesn't inherit from TServer, and doesn't
+ // provide a stop method.
TEvhttpServer nonblockingServer(testBufferProcessor, port);
nonblockingServer.serve();
-} else {
- TNonblockingServer nonblockingServer(testProcessor, port);
- nonblockingServer.serve();
+ } else {
+ server.reset(new TNonblockingServer(testProcessor, port));
}
}
+ if(server.get() != NULL)
+ {
+ apache::thrift::concurrency::PlatformThreadFactory factory;
+ factory.setDetached(false);
+ boost::shared_ptr<apache::thrift::concurrency::Runnable> serverThreadRunner(server);
+ boost::shared_ptr<apache::thrift::concurrency::Thread> thread = factory.newThread(serverThreadRunner);
+ thread->start();
+
+ cout<<"Press enter to stop the server."<<endl;
+ cin.ignore(); //wait until a key is pressed
+
+ server->stop();
+ thread->join();
+ server.reset();
+ }
+
cout << "done." << endl;
return 0;
}