/root/doris/be/src/util/thrift_server.h
Line | Count | Source (jump to first uncovered line) |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #pragma once |
19 | | |
20 | | #include <thrift/server/TServer.h> |
21 | | |
22 | | #include <memory> |
23 | | #include <mutex> |
24 | | #include <string> |
25 | | #include <thread> |
26 | | #include <unordered_map> |
27 | | |
28 | | #include "common/status.h" |
29 | | #include "util/metrics.h" |
30 | | |
31 | | namespace apache { |
32 | | namespace thrift { |
33 | | class TProcessor; |
34 | | } // namespace thrift |
35 | | } // namespace apache |
36 | | |
37 | | namespace doris { |
38 | | // Utility class for all Thrift servers. Runs a TNonblockingServer(default) or a |
39 | | // TThreadPoolServer with, by default, 2 worker threads, that exposes the interface |
40 | | // described by a user-supplied TProcessor object. |
41 | | // If TNonblockingServer is used, client must use TFramedTransport. |
42 | | // If TThreadPoolServer is used, client must use TSocket as transport. |
43 | | class ThriftServer { |
44 | | public: |
45 | | // An opaque identifier for the current session, which identifies a client connection. |
46 | | using SessionKey = std::string; |
47 | | |
48 | | // Interface class for receiving session creation / termination events. |
49 | | class SessionHandlerIf { |
50 | | public: |
51 | | virtual ~SessionHandlerIf() = default; |
52 | | // Called when a session is established (when a client connects). |
53 | | virtual void session_start(const SessionKey& session_key) = 0; |
54 | | |
55 | | // Called when a session is terminated (when a client closes the connection). |
56 | | // After this callback returns, the memory session_key references is no longer valid |
57 | | // and clients must not refer to it again. |
58 | | virtual void session_end(const SessionKey& session_key) = 0; |
59 | | }; |
60 | | |
61 | | static const int DEFAULT_WORKER_THREADS = 2; |
62 | | |
63 | | // There are 3 servers supported by Thrift with different threading models. |
64 | | // THREAD_POOL -- Allocates a fixed number of threads. A thread is used by a |
65 | | // connection until it closes. |
66 | | // THREADED -- Allocates 1 thread per connection, as needed. |
67 | | // NON_BLOCKING -- Threads are allocated to a connection only when the server |
68 | | // is working on behalf of the connection. |
69 | | enum ServerType { THREAD_POOL = 0, THREADED, NON_BLOCKING }; |
70 | | |
71 | | // Creates, but does not start, a new server on the specified port |
72 | | // that exports the supplied interface. |
73 | | // - name: human-readable name of this server. Should not contain spaces |
74 | | // - processor: Thrift processor to handle RPCs |
75 | | // - port: The port the server will listen for connections on |
76 | | // - num_worker_threads: the number of worker threads to use in any thread pool |
77 | | // - server_type: the type of IO strategy this server should employ |
78 | | ThriftServer(const std::string& name, |
79 | | const std::shared_ptr<apache::thrift::TProcessor>& processor, int port, |
80 | | int num_worker_threads = DEFAULT_WORKER_THREADS, |
81 | | ServerType server_type = THREADED); |
82 | | |
83 | 0 | ~ThriftServer() {} |
84 | | |
85 | 0 | int port() const { return _port; } |
86 | | |
87 | | void stop(); |
88 | | // Blocks until the server stops and exits its main thread. |
89 | | void join(); |
90 | | |
91 | | // FOR TESTING ONLY; stop the server and block until the server is stopped; use it |
92 | | // only if it is a Threaded server. |
93 | | void stop_for_testing(); |
94 | | |
95 | | // Starts the main server thread. Once this call returns, clients |
96 | | // may connect to this server and issue RPCs. May not be called more |
97 | | // than once. |
98 | | Status start(); |
99 | | |
100 | | // Sets the session handler which receives events when sessions are created or closed. |
101 | 0 | void set_session_handler(SessionHandlerIf* session) { _session_handler = session; } |
102 | | |
103 | | // Returns a unique identifier for the current session. A session is |
104 | | // identified with the lifetime of a socket connection to this server. |
105 | | // It is only safe to call this method during a Thrift processor RPC |
106 | | // implementation. Otherwise, the result of calling this method is undefined. |
107 | | // It is also only safe to reference the returned value during an RPC method. |
108 | | static SessionKey* get_thread_session_key(); |
109 | | |
110 | | private: |
111 | | // True if the server has been successfully started, for internal use only |
112 | | bool _started; |
113 | | |
114 | | // The port on which the server interface is exposed |
115 | | int _port; |
116 | | |
117 | | // How many worker threads to use to serve incoming requests |
118 | | // (requests are queued if no thread is immediately available) |
119 | | int _num_worker_threads; |
120 | | |
121 | | // ThreadPool or NonBlocking server |
122 | | ServerType _server_type; |
123 | | |
124 | | // User-specified identifier that shows up in logs |
125 | | const std::string _name; |
126 | | |
127 | | // Thread that runs the TNonblockingServer::serve loop |
128 | | std::unique_ptr<std::thread> _server_thread; |
129 | | |
130 | | // Thrift housekeeping |
131 | | std::unique_ptr<apache::thrift::server::TServer> _server; |
132 | | std::shared_ptr<apache::thrift::TProcessor> _processor; |
133 | | |
134 | | // If not nullptr, called when session events happen. Not owned by us. |
135 | | SessionHandlerIf* _session_handler; |
136 | | |
137 | | // Protects _session_keys |
138 | | std::mutex _session_keys_lock; |
139 | | |
140 | | // Map of active session keys to shared_ptr containing that key; when a key is |
141 | | // removed it is automatically freed. |
142 | | typedef std::unordered_map<SessionKey*, std::shared_ptr<SessionKey>> SessionKeySet; |
143 | | SessionKeySet _session_keys; |
144 | | |
145 | | // Helper class which monitors starting servers. Needs access to internal members, and |
146 | | // is not used outside of this class. |
147 | | class ThriftServerEventProcessor; |
148 | | |
149 | | friend class ThriftServerEventProcessor; |
150 | | |
151 | | std::shared_ptr<MetricEntity> _thrift_server_metric_entity; |
152 | | // Number of currently active connections |
153 | | IntGauge* thrift_current_connections; |
154 | | // Total connections made over the lifetime of this server |
155 | | IntCounter* thrift_connections_total; |
156 | | }; |
157 | | |
158 | | } // namespace doris |