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/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 <memory>
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(std::shared_ptr<TTransport> tt, uint32_t expectedResult) {
39 uint8_t buf[4];
40 BOOST_CHECK_EQUAL(expectedResult, tt->read(buf, 4));
41 }
42
readerWorkerMustThrow(std::shared_ptr<TTransport> tt)43 void readerWorkerMustThrow(std::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 BOOST_CHECK(sock1.isOpen());
57 int port = sock1.getPort();
58 TSocket clientSock("localhost", port);
59 clientSock.open();
60 std::shared_ptr<TTransport> accepted = sock1.accept();
61 boost::thread readThread(std::bind(readerWorkerMustThrow, accepted));
62 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
63 // readThread is practically guaranteed to be blocking now
64 sock1.interruptChildren();
65 BOOST_CHECK_MESSAGE(readThread.try_join_for(boost::chrono::milliseconds(200)),
66 "server socket interruptChildren did not interrupt child read");
67 clientSock.close();
68 accepted->close();
69 sock1.close();
70 }
71
BOOST_AUTO_TEST_CASE(test_non_interruptable_child_read)72 BOOST_AUTO_TEST_CASE(test_non_interruptable_child_read) {
73 TServerSocket sock1("localhost", 0);
74 sock1.setInterruptableChildren(false); // returns to pre-THRIFT-2441 behavior
75 sock1.listen();
76 int port = sock1.getPort();
77 TSocket clientSock("localhost", port);
78 clientSock.open();
79 std::shared_ptr<TTransport> accepted = sock1.accept();
80 boost::thread readThread(std::bind(readerWorker, accepted, 0));
81 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
82 // readThread is practically guaranteed to be blocking here
83 sock1.interruptChildren();
84 BOOST_CHECK_MESSAGE(!readThread.try_join_for(boost::chrono::milliseconds(200)),
85 "server socket interruptChildren interrupted child read");
86
87 // only way to proceed is to have the client disconnect
88 clientSock.close();
89 readThread.join();
90 accepted->close();
91 sock1.close();
92 }
93
BOOST_AUTO_TEST_CASE(test_cannot_change_after_listen)94 BOOST_AUTO_TEST_CASE(test_cannot_change_after_listen) {
95 TServerSocket sock1("localhost", 0);
96 sock1.listen();
97 BOOST_CHECK_THROW(sock1.setInterruptableChildren(false), std::logic_error);
98 sock1.close();
99 }
100
peekerWorker(std::shared_ptr<TTransport> tt,bool expectedResult)101 void peekerWorker(std::shared_ptr<TTransport> tt, bool expectedResult) {
102 BOOST_CHECK_EQUAL(expectedResult, tt->peek());
103 }
104
BOOST_AUTO_TEST_CASE(test_interruptable_child_peek)105 BOOST_AUTO_TEST_CASE(test_interruptable_child_peek) {
106 TServerSocket sock1("localhost", 0);
107 sock1.listen();
108 int port = sock1.getPort();
109 TSocket clientSock("localhost", port);
110 clientSock.open();
111 std::shared_ptr<TTransport> accepted = sock1.accept();
112 // peek() will return false if child is interrupted
113 boost::thread peekThread(std::bind(peekerWorker, accepted, false));
114 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
115 // peekThread is practically guaranteed to be blocking now
116 sock1.interruptChildren();
117 BOOST_CHECK_MESSAGE(peekThread.try_join_for(boost::chrono::milliseconds(200)),
118 "server socket interruptChildren did not interrupt child peek");
119 clientSock.close();
120 accepted->close();
121 sock1.close();
122 }
123
BOOST_AUTO_TEST_CASE(test_non_interruptable_child_peek)124 BOOST_AUTO_TEST_CASE(test_non_interruptable_child_peek) {
125 TServerSocket sock1("localhost", 0);
126 sock1.setInterruptableChildren(false); // returns to pre-THRIFT-2441 behavior
127 sock1.listen();
128 int port = sock1.getPort();
129 TSocket clientSock("localhost", port);
130 clientSock.open();
131 std::shared_ptr<TTransport> accepted = sock1.accept();
132 // peek() will return false when remote side is closed
133 boost::thread peekThread(std::bind(peekerWorker, accepted, false));
134 boost::this_thread::sleep(boost::posix_time::milliseconds(50));
135 // peekThread is practically guaranteed to be blocking now
136 sock1.interruptChildren();
137 BOOST_CHECK_MESSAGE(!peekThread.try_join_for(boost::chrono::milliseconds(200)),
138 "server socket interruptChildren interrupted child peek");
139
140 // only way to proceed is to have the client disconnect
141 clientSock.close();
142 peekThread.join();
143 accepted->close();
144 sock1.close();
145 }
146
147 BOOST_AUTO_TEST_SUITE_END()
148
