1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 */
19
20 #define BOOST_TEST_MODULE TSocketInterruptTest
21 #include <boost/test/auto_unit_test.hpp>
22
23 #include <boost/chrono/duration.hpp>
24 #include <boost/date_time/posix_time/posix_time_duration.hpp>
25 #include <boost/thread/thread.hpp>
26 #include <thrift/transport/TSocket.h>
27 #include <thrift/transport/TServerSocket.h>
28 #include <thrift/stdcxx.h>
29
30 using apache::thrift::transport::TServerSocket;
31 using apache::thrift::transport::TSocket;
32 using apache::thrift::transport::TTransport;
33 using apache::thrift::transport::TTransportException;
34 using namespace apache::thrift;
35
BOOST_AUTO_TEST_SUITE(TSocketInterruptTest)36 BOOST_AUTO_TEST_SUITE(TSocketInterruptTest)
37
38 void readerWorker(stdcxx::shared_ptr<TTransport> tt, uint32_t expectedResult) {
39 uint8_t buf[4];
40 BOOST_CHECK_EQUAL(expectedResult, tt->read(buf, 4));
41 }
42
readerWorkerMustThrow(stdcxx::shared_ptr<TTransport> tt)43 void readerWorkerMustThrow(stdcxx::shared_ptr<TTransport> tt) {
44 try {
45 uint8_t buf[4];
46 tt->read(buf, 4);
47 BOOST_ERROR("should not have gotten here");
48 } catch (const TTransportException& tx) {
49 BOOST_CHECK_EQUAL(TTransportException::INTERRUPTED, tx.getType());
50 }
51 }
52
BOOST_AUTO_TEST_CASE(test_interruptable_child_read)53 BOOST_AUTO_TEST_CASE(test_interruptable_child_read) {
54 TServerSocket sock1("localhost", 0);
55 sock1.listen();
56 int port = sock1.getPort();
57 TSocket clientSock("localhost", port);
58 clientSock.open();
59 stdcxx::shared_ptr<TTransport> accepted = sock1.accept();
60 boost::thread readThread(stdcxx::bind(readerWorkerMustThrow, accepted));
61 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
62 // readThread is practically guaranteed to be blocking now
63 sock1.interruptChildren();
64 BOOST_CHECK_MESSAGE(readThread.try_join_for(boost::chrono::milliseconds(200)),
65 "server socket interruptChildren did not interrupt child read");
66 clientSock.close();
67 accepted->close();
68 sock1.close();
69 }
70
BOOST_AUTO_TEST_CASE(test_non_interruptable_child_read)71 BOOST_AUTO_TEST_CASE(test_non_interruptable_child_read) {
72 TServerSocket sock1("localhost", 0);
73 sock1.setInterruptableChildren(false); // returns to pre-THRIFT-2441 behavior
74 sock1.listen();
75 int port = sock1.getPort();
76 TSocket clientSock("localhost", port);
77 clientSock.open();
78 stdcxx::shared_ptr<TTransport> accepted = sock1.accept();
79 boost::thread readThread(stdcxx::bind(readerWorker, accepted, 0));
80 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
81 // readThread is practically guaranteed to be blocking here
82 sock1.interruptChildren();
83 BOOST_CHECK_MESSAGE(!readThread.try_join_for(boost::chrono::milliseconds(200)),
84 "server socket interruptChildren interrupted child read");
85
86 // only way to proceed is to have the client disconnect
87 clientSock.close();
88 readThread.join();
89 accepted->close();
90 sock1.close();
91 }
92
BOOST_AUTO_TEST_CASE(test_cannot_change_after_listen)93 BOOST_AUTO_TEST_CASE(test_cannot_change_after_listen) {
94 TServerSocket sock1("localhost", 0);
95 sock1.listen();
96 BOOST_CHECK_THROW(sock1.setInterruptableChildren(false), std::logic_error);
97 sock1.close();
98 }
99
peekerWorker(stdcxx::shared_ptr<TTransport> tt,bool expectedResult)100 void peekerWorker(stdcxx::shared_ptr<TTransport> tt, bool expectedResult) {
101 BOOST_CHECK_EQUAL(expectedResult, tt->peek());
102 }
103
BOOST_AUTO_TEST_CASE(test_interruptable_child_peek)104 BOOST_AUTO_TEST_CASE(test_interruptable_child_peek) {
105 TServerSocket sock1("localhost", 0);
106 sock1.listen();
107 int port = sock1.getPort();
108 TSocket clientSock("localhost", port);
109 clientSock.open();
110 stdcxx::shared_ptr<TTransport> accepted = sock1.accept();
111 // peek() will return false if child is interrupted
112 boost::thread peekThread(stdcxx::bind(peekerWorker, accepted, false));
113 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
114 // peekThread is practically guaranteed to be blocking now
115 sock1.interruptChildren();
116 BOOST_CHECK_MESSAGE(peekThread.try_join_for(boost::chrono::milliseconds(200)),
117 "server socket interruptChildren did not interrupt child peek");
118 clientSock.close();
119 accepted->close();
120 sock1.close();
121 }
122
BOOST_AUTO_TEST_CASE(test_non_interruptable_child_peek)123 BOOST_AUTO_TEST_CASE(test_non_interruptable_child_peek) {
124 TServerSocket sock1("localhost", 0);
125 sock1.setInterruptableChildren(false); // returns to pre-THRIFT-2441 behavior
126 sock1.listen();
127 int port = sock1.getPort();
128 TSocket clientSock("localhost", port);
129 clientSock.open();
130 stdcxx::shared_ptr<TTransport> accepted = sock1.accept();
131 // peek() will return false when remote side is closed
132 boost::thread peekThread(stdcxx::bind(peekerWorker, accepted, false));
133 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
134 // peekThread is practically guaranteed to be blocking now
135 sock1.interruptChildren();
136 BOOST_CHECK_MESSAGE(!peekThread.try_join_for(boost::chrono::milliseconds(200)),
137 "server socket interruptChildren interrupted child peek");
138
139 // only way to proceed is to have the client disconnect
140 clientSock.close();
141 peekThread.join();
142 accepted->close();
143 sock1.close();
144 }
145
146 BOOST_AUTO_TEST_SUITE_END()
147