blob: 11718cac3d5cf23bdb4ac15b788e9de6bf749fc1 [file] [log] [blame]
David Reissea2cba82009-03-30 21:35:00 +00001/*
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 */
Mark Slee9f0c6512007-02-28 23:58:26 +000019
Mark Sleeb3cb6292007-02-01 22:55:00 +000020#include "server/TThreadedServer.h"
21#include "transport/TTransportException.h"
22#include "concurrency/PosixThreadFactory.h"
23
24#include <string>
25#include <iostream>
26#include <pthread.h>
27#include <unistd.h>
28
T Jake Lucianib5e62212009-01-31 22:36:20 +000029namespace apache { namespace thrift { namespace server {
Mark Sleeb3cb6292007-02-01 22:55:00 +000030
Mark Slee5ea15f92007-03-05 22:55:59 +000031using boost::shared_ptr;
Mark Sleeb3cb6292007-02-01 22:55:00 +000032using namespace std;
T Jake Lucianib5e62212009-01-31 22:36:20 +000033using namespace apache::thrift;
34using namespace apache::thrift::protocol;
35using namespace apache::thrift::transport;
36using namespace apache::thrift::concurrency;
Mark Sleeb3cb6292007-02-01 22:55:00 +000037
38class TThreadedServer::Task: public Runnable {
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000039
Mark Sleeb3cb6292007-02-01 22:55:00 +000040public:
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000041
42 Task(TThreadedServer& server,
Mark Slee1d4ce802007-03-07 05:16:16 +000043 shared_ptr<TProcessor> processor,
Mark Sleeb3cb6292007-02-01 22:55:00 +000044 shared_ptr<TProtocol> input,
David Reiss23248712010-10-06 17:10:08 +000045 shared_ptr<TProtocol> output,
46 shared_ptr<TTransport> transport) :
Mark Slee1d4ce802007-03-07 05:16:16 +000047 server_(server),
Mark Sleeb3cb6292007-02-01 22:55:00 +000048 processor_(processor),
49 input_(input),
David Reiss23248712010-10-06 17:10:08 +000050 output_(output),
51 transport_(transport) {
Mark Sleeb3cb6292007-02-01 22:55:00 +000052 }
53
54 ~Task() {}
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000055
Mark Sleeb3cb6292007-02-01 22:55:00 +000056 void run() {
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000057 boost::shared_ptr<TServerEventHandler> eventHandler =
58 server_.getEventHandler();
David Reiss23248712010-10-06 17:10:08 +000059 void* connectionContext = NULL;
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000060 if (eventHandler != NULL) {
David Reiss23248712010-10-06 17:10:08 +000061 connectionContext = eventHandler->createContext(input_, output_);
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000062 }
Mark Sleeb3cb6292007-02-01 22:55:00 +000063 try {
David Reiss23248712010-10-06 17:10:08 +000064 for (;;) {
65 if (eventHandler != NULL) {
66 eventHandler->processContext(connectionContext, transport_);
67 }
68 if (!processor_->process(input_, output_, connectionContext) ||
69 !input_->getTransport()->peek()) {
Mark Sleeb3cb6292007-02-01 22:55:00 +000070 break;
71 }
72 }
73 } catch (TTransportException& ttx) {
Mark Slee2e8a8d42008-01-16 00:38:20 +000074 string errStr = string("TThreadedServer client died: ") + ttx.what();
75 GlobalOutput(errStr.c_str());
Mark Sleeb3cb6292007-02-01 22:55:00 +000076 } catch (TException& x) {
Mark Slee2e8a8d42008-01-16 00:38:20 +000077 string errStr = string("TThreadedServer exception: ") + x.what();
78 GlobalOutput(errStr.c_str());
Mark Sleeb3cb6292007-02-01 22:55:00 +000079 } catch (...) {
Mark Slee2e8a8d42008-01-16 00:38:20 +000080 GlobalOutput("TThreadedServer uncaught exception.");
Mark Sleeb3cb6292007-02-01 22:55:00 +000081 }
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000082 if (eventHandler != NULL) {
David Reiss23248712010-10-06 17:10:08 +000083 eventHandler->deleteContext(connectionContext, input_, output_);
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000084 }
Mark Slee2e8a8d42008-01-16 00:38:20 +000085
86 try {
87 input_->getTransport()->close();
88 } catch (TTransportException& ttx) {
89 string errStr = string("TThreadedServer input close failed: ") + ttx.what();
90 GlobalOutput(errStr.c_str());
91 }
92 try {
93 output_->getTransport()->close();
94 } catch (TTransportException& ttx) {
95 string errStr = string("TThreadedServer output close failed: ") + ttx.what();
96 GlobalOutput(errStr.c_str());
97 }
Mark Sleeb4d3e7b2007-11-28 01:51:43 +000098
Mark Slee1d4ce802007-03-07 05:16:16 +000099 // Remove this task from parent bookkeeping
100 {
Mark Sleeb4d3e7b2007-11-28 01:51:43 +0000101 Synchronized s(server_.tasksMonitor_);
102 server_.tasks_.erase(this);
103 if (server_.tasks_.empty()) {
104 server_.tasksMonitor_.notify();
Mark Slee1d4ce802007-03-07 05:16:16 +0000105 }
106 }
107
Mark Sleeb3cb6292007-02-01 22:55:00 +0000108 }
109
110 private:
Mark Sleeb4d3e7b2007-11-28 01:51:43 +0000111 TThreadedServer& server_;
Mark Slee1d4ce802007-03-07 05:16:16 +0000112 friend class TThreadedServer;
113
Mark Sleeb3cb6292007-02-01 22:55:00 +0000114 shared_ptr<TProcessor> processor_;
115 shared_ptr<TProtocol> input_;
116 shared_ptr<TProtocol> output_;
David Reiss23248712010-10-06 17:10:08 +0000117 shared_ptr<TTransport> transport_;
Mark Sleeb3cb6292007-02-01 22:55:00 +0000118};
119
120
121TThreadedServer::TThreadedServer(shared_ptr<TProcessor> processor,
122 shared_ptr<TServerTransport> serverTransport,
123 shared_ptr<TTransportFactory> transportFactory,
124 shared_ptr<TProtocolFactory> protocolFactory):
Mark Slee1d4ce802007-03-07 05:16:16 +0000125 TServer(processor, serverTransport, transportFactory, protocolFactory),
126 stop_(false) {
Mark Sleeb3cb6292007-02-01 22:55:00 +0000127 threadFactory_ = shared_ptr<PosixThreadFactory>(new PosixThreadFactory());
128}
129
David Reiss45d56962009-03-14 23:35:16 +0000130TThreadedServer::TThreadedServer(boost::shared_ptr<TProcessor> processor,
131 boost::shared_ptr<TServerTransport> serverTransport,
132 boost::shared_ptr<TTransportFactory> transportFactory,
133 boost::shared_ptr<TProtocolFactory> protocolFactory,
134 boost::shared_ptr<ThreadFactory> threadFactory):
135 TServer(processor, serverTransport, transportFactory, protocolFactory),
136 threadFactory_(threadFactory),
137 stop_(false) {
138}
139
Mark Sleeb3cb6292007-02-01 22:55:00 +0000140TThreadedServer::~TThreadedServer() {}
141
142void TThreadedServer::serve() {
143
144 shared_ptr<TTransport> client;
145 shared_ptr<TTransport> inputTransport;
146 shared_ptr<TTransport> outputTransport;
147 shared_ptr<TProtocol> inputProtocol;
148 shared_ptr<TProtocol> outputProtocol;
149
150 try {
151 // Start the server listening
152 serverTransport_->listen();
153 } catch (TTransportException& ttx) {
Mark Slee2e8a8d42008-01-16 00:38:20 +0000154 string errStr = string("TThreadedServer::run() listen(): ") +ttx.what();
155 GlobalOutput(errStr.c_str());
Mark Sleeb3cb6292007-02-01 22:55:00 +0000156 return;
157 }
158
Mark Sleeb4d3e7b2007-11-28 01:51:43 +0000159 // Run the preServe event
160 if (eventHandler_ != NULL) {
161 eventHandler_->preServe();
162 }
163
164 while (!stop_) {
Mark Sleeb3cb6292007-02-01 22:55:00 +0000165 try {
Mark Slee1d4ce802007-03-07 05:16:16 +0000166 client.reset();
167 inputTransport.reset();
168 outputTransport.reset();
169 inputProtocol.reset();
170 outputProtocol.reset();
171
Mark Sleeb3cb6292007-02-01 22:55:00 +0000172 // Fetch client from server
173 client = serverTransport_->accept();
Mark Slee1d4ce802007-03-07 05:16:16 +0000174
Mark Sleeb3cb6292007-02-01 22:55:00 +0000175 // Make IO transports
176 inputTransport = inputTransportFactory_->getTransport(client);
177 outputTransport = outputTransportFactory_->getTransport(client);
178 inputProtocol = inputProtocolFactory_->getProtocol(inputTransport);
179 outputProtocol = outputProtocolFactory_->getProtocol(outputTransport);
180
Mark Sleeb4d3e7b2007-11-28 01:51:43 +0000181 TThreadedServer::Task* task = new TThreadedServer::Task(*this,
182 processor_,
Mark Slee1d4ce802007-03-07 05:16:16 +0000183 inputProtocol,
David Reiss23248712010-10-06 17:10:08 +0000184 outputProtocol,
185 client);
Mark Sleeb4d3e7b2007-11-28 01:51:43 +0000186
Mark Slee1d4ce802007-03-07 05:16:16 +0000187 // Create a task
188 shared_ptr<Runnable> runnable =
189 shared_ptr<Runnable>(task);
Mark Sleeb3cb6292007-02-01 22:55:00 +0000190
191 // Create a thread for this task
192 shared_ptr<Thread> thread =
Mark Slee1d4ce802007-03-07 05:16:16 +0000193 shared_ptr<Thread>(threadFactory_->newThread(runnable));
Mark Sleeb4d3e7b2007-11-28 01:51:43 +0000194
Mark Slee1d4ce802007-03-07 05:16:16 +0000195 // Insert thread into the set of threads
196 {
197 Synchronized s(tasksMonitor_);
198 tasks_.insert(task);
199 }
200
Mark Sleeb3cb6292007-02-01 22:55:00 +0000201 // Start the thread!
202 thread->start();
203
204 } catch (TTransportException& ttx) {
Mark Slee1d4ce802007-03-07 05:16:16 +0000205 if (inputTransport != NULL) { inputTransport->close(); }
206 if (outputTransport != NULL) { outputTransport->close(); }
207 if (client != NULL) { client->close(); }
208 if (!stop_ || ttx.getType() != TTransportException::INTERRUPTED) {
Mark Slee2e8a8d42008-01-16 00:38:20 +0000209 string errStr = string("TThreadedServer: TServerTransport died on accept: ") + ttx.what();
210 GlobalOutput(errStr.c_str());
Mark Slee1d4ce802007-03-07 05:16:16 +0000211 }
Mark Slee907e3d62007-02-08 22:29:24 +0000212 continue;
213 } catch (TException& tx) {
Mark Slee1d4ce802007-03-07 05:16:16 +0000214 if (inputTransport != NULL) { inputTransport->close(); }
215 if (outputTransport != NULL) { outputTransport->close(); }
216 if (client != NULL) { client->close(); }
Mark Slee2e8a8d42008-01-16 00:38:20 +0000217 string errStr = string("TThreadedServer: Caught TException: ") + tx.what();
218 GlobalOutput(errStr.c_str());
Mark Slee907e3d62007-02-08 22:29:24 +0000219 continue;
220 } catch (string s) {
Mark Slee1d4ce802007-03-07 05:16:16 +0000221 if (inputTransport != NULL) { inputTransport->close(); }
222 if (outputTransport != NULL) { outputTransport->close(); }
223 if (client != NULL) { client->close(); }
Mark Slee2e8a8d42008-01-16 00:38:20 +0000224 string errStr = "TThreadedServer: Unknown exception: " + s;
225 GlobalOutput(errStr.c_str());
Mark Sleeb3cb6292007-02-01 22:55:00 +0000226 break;
227 }
228 }
Mark Slee1d4ce802007-03-07 05:16:16 +0000229
230 // If stopped manually, make sure to close server transport
231 if (stop_) {
232 try {
233 serverTransport_->close();
234 } catch (TException &tx) {
Mark Slee2e8a8d42008-01-16 00:38:20 +0000235 string errStr = string("TThreadedServer: Exception shutting down: ") + tx.what();
236 GlobalOutput(errStr.c_str());
Mark Slee1d4ce802007-03-07 05:16:16 +0000237 }
238 try {
239 Synchronized s(tasksMonitor_);
240 while (!tasks_.empty()) {
241 tasksMonitor_.wait();
242 }
243 } catch (TException &tx) {
Mark Slee2e8a8d42008-01-16 00:38:20 +0000244 string errStr = string("TThreadedServer: Exception joining workers: ") + tx.what();
245 GlobalOutput(errStr.c_str());
Mark Slee1d4ce802007-03-07 05:16:16 +0000246 }
247 stop_ = false;
248 }
249
Mark Sleeb3cb6292007-02-01 22:55:00 +0000250}
251
T Jake Lucianib5e62212009-01-31 22:36:20 +0000252}}} // apache::thrift::server