Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include "xmpp/xmpp_server.h"
7 :
8 : #include <boost/foreach.hpp>
9 : #include <boost/tuple/tuple.hpp>
10 :
11 : #include "base/task_annotations.h"
12 : #include "xmpp/xmpp_connection.h"
13 : #include "xmpp/xmpp_factory.h"
14 : #include "xmpp/xmpp_lifetime.h"
15 : #include "xmpp/xmpp_log.h"
16 : #include "xmpp/xmpp_sandesh.h"
17 : #include "xmpp/xmpp_session.h"
18 :
19 : #include "sandesh/request_pipeline.h"
20 : #include "sandesh/common/vns_types.h"
21 : #include "sandesh/common/vns_constants.h"
22 : #include "sandesh/xmpp_server_types.h"
23 : #include "sandesh/xmpp_trace_sandesh_types.h"
24 : #include "sandesh/xmpp_client_server_sandesh_types.h"
25 :
26 : using namespace std;
27 : using namespace boost::asio;
28 : using boost::tie;
29 :
30 : #define DEFAULT_XMPP_HOLD_TIME 90
31 : class XmppServer::DeleteActor : public LifetimeActor {
32 : public:
33 2158 : DeleteActor(XmppServer *server)
34 2158 : : LifetimeActor(server->lifetime_manager()), server_(server) {
35 2158 : }
36 2156 : virtual bool MayDelete() const {
37 2156 : CHECK_CONCURRENCY("bgp::Config");
38 2156 : return server_->MayDelete();
39 : }
40 2156 : virtual void Shutdown() {
41 2156 : CHECK_CONCURRENCY("bgp::Config");
42 2156 : server_->SessionShutdown();
43 2156 : }
44 2156 : virtual void Destroy() {
45 2156 : CHECK_CONCURRENCY("bgp::Config");
46 2156 : server_->Terminate();
47 2156 : }
48 :
49 : private:
50 : XmppServer *server_;
51 : };
52 :
53 :
54 414 : XmppServer::XmppServer(EventManager *evm, const string &server_addr,
55 414 : const XmppChannelConfig *config)
56 : : XmppConnectionManager(
57 414 : evm, ssl::context::sslv23_server, config->auth_enabled, true),
58 414 : max_connections_(0),
59 414 : lifetime_manager_(XmppStaticObjectFactory::Create<XmppLifetimeManager>(
60 : TaskScheduler::GetInstance()->GetTaskId("bgp::Config"))),
61 414 : deleter_(new DeleteActor(this)),
62 414 : server_addr_(server_addr),
63 414 : log_uve_(false),
64 414 : auth_enabled_(config->auth_enabled),
65 414 : tcp_hold_time_(config->tcp_hold_time),
66 414 : gr_helper_disable_(config->gr_helper_disable),
67 414 : dscp_value_(0),
68 414 : connection_queue_(TaskScheduler::GetInstance()->GetTaskId("bgp::Config"),
69 1242 : 0, boost::bind(&XmppServer::DequeueConnection, this, _1)) {
70 :
71 414 : if (config->auth_enabled) {
72 :
73 : // Get SSL context from base class and update
74 284 : boost::asio::ssl::context *ctx = context();
75 284 : boost::system::error_code ec;
76 :
77 : // set mode
78 284 : ctx->set_options(ssl::context::default_workarounds |
79 : ssl::context::no_sslv3 | ssl::context::no_sslv2, ec);
80 284 : if (ec.value() != 0) {
81 0 : LOG(ERROR, "Error : " << ec.message() << ", setting ssl options");
82 0 : exit(EINVAL);
83 : }
84 :
85 : // CA certificate, used to verify if the peer certificate
86 : // is signed by a trusted CA
87 284 : std::string ca_cert_filename = config->path_to_ca_cert;
88 284 : if (!ca_cert_filename.empty()) {
89 :
90 : // Verify peer has CA signed certificate
91 0 : ctx->set_verify_mode(boost::asio::ssl::verify_peer, ec);
92 0 : if (ec.value() != 0) {
93 0 : LOG(ERROR, "Error : " << ec.message()
94 : << ", while setting ssl verification mode");
95 0 : exit(EINVAL);
96 : }
97 :
98 0 : ctx->load_verify_file(config->path_to_ca_cert, ec);
99 0 : if (ec.value() != 0) {
100 0 : LOG(ERROR, "Error : " << ec.message()
101 : << ", while using cacert file : "
102 : << config->path_to_ca_cert);
103 0 : exit(EINVAL);
104 : }
105 : }
106 :
107 : // server certificate
108 284 : ctx->use_certificate_file(config->path_to_server_cert,
109 : boost::asio::ssl::context::pem, ec);
110 284 : if (ec.value() != 0) {
111 0 : LOG(ERROR, "Error : " << ec.message()
112 : << ", while using server cert file : "
113 : << config->path_to_server_cert);
114 0 : exit(EINVAL);
115 : }
116 :
117 : // server private key
118 284 : ctx->use_private_key_file(config->path_to_server_priv_key,
119 : boost::asio::ssl::context::pem, ec);
120 284 : if (ec.value() != 0) {
121 0 : LOG(ERROR, "Error : " << ec.message()
122 : << ", while using privkey file : "
123 : << config->path_to_server_priv_key);
124 0 : exit(EINVAL);
125 : }
126 284 : }
127 414 : }
128 :
129 : class XmppConfigUpdater {
130 : public:
131 2078 : explicit XmppConfigUpdater(XmppServer *server,
132 2078 : BgpConfigManager *config_manager) :
133 2078 : server_(server) {
134 2078 : BgpConfigManager::Observers obs;
135 : obs.system= boost::bind(&XmppConfigUpdater::ProcessGlobalSystemConfig,
136 2078 : this, _1, _2);
137 : obs.protocol = boost::bind(&XmppConfigUpdater::ProcessProtocolConfig,
138 2078 : this, _1, _2);
139 2078 : config_manager->RegisterObservers(obs);
140 2078 : }
141 :
142 28650 : const BgpGlobalSystemConfig &config() const { return config_; }
143 :
144 2686 : void ProcessProtocolConfig(const BgpProtocolConfig *protocol_config,
145 : BgpConfigManager::EventType event) {
146 :
147 2686 : if (server_->subcluster_name() != protocol_config->subcluster_name()) {
148 8 : server_->set_subcluster_name(protocol_config->subcluster_name());
149 8 : server_->ClearAllConnections();
150 : }
151 2686 : }
152 :
153 467 : void ProcessGlobalSystemConfig(const BgpGlobalSystemConfig *system,
154 : BgpConfigManager::EventType event) {
155 : // Clear peers only if GR is or was enabled.
156 467 : bool clear_peers = config_.gr_enable() || system->gr_enable();
157 467 : bool update_peers = false;
158 467 : config_.set_gr_enable(system->gr_enable());
159 467 : config_.set_gr_time(system->gr_time());
160 467 : config_.set_llgr_time(system->llgr_time());
161 467 : config_.set_end_of_rib_timeout(system->end_of_rib_timeout());
162 467 : config_.set_gr_xmpp_helper(system->gr_xmpp_helper());
163 :
164 : // Process any change in xmpp-hold-time
165 926 : if (config_.fc_enabled() != system->fc_enabled() ||
166 459 : config_.xmpp_hold_time() != system->xmpp_hold_time()) {
167 16 : config_.set_xmpp_hold_time(system->xmpp_hold_time());
168 16 : config_.set_fc_enabled(system->fc_enabled());
169 : // if fc_enabled is not set, use default hold time
170 16 : if (!system->fc_enabled())
171 0 : config_.set_xmpp_hold_time(DEFAULT_XMPP_HOLD_TIME);
172 16 : update_peers = true;
173 : }
174 :
175 : // Process any change in rd-cluster-seed knob
176 467 : if (config_.rd_cluster_seed() != system->rd_cluster_seed()) {
177 22 : config_.set_rd_cluster_seed(system->rd_cluster_seed());
178 22 : clear_peers = true;
179 : }
180 :
181 467 : if (clear_peers)
182 447 : server_->ClearAllConnections();
183 20 : else if (update_peers)
184 16 : server_->UpdateAllConnections(config_.xmpp_hold_time());
185 467 : }
186 :
187 : const string subcluster_name() const { return server_->subcluster_name(); }
188 : void set_subcluster_name(const string& name) {
189 : server_->set_subcluster_name(name);
190 : }
191 :
192 6531 : uint8_t xmpp_hold_time() const { return config_.xmpp_hold_time(); }
193 : void set_xmpp_hold_time(int hold_time) {
194 : config_.set_xmpp_hold_time(hold_time);
195 : }
196 :
197 : private:
198 : XmppServer *server_;
199 : BgpGlobalSystemConfig config_;
200 : };
201 :
202 1737 : XmppServer::XmppServer(EventManager *evm, const string &server_addr)
203 : : XmppConnectionManager(evm, ssl::context::sslv23_server, false, false),
204 1737 : max_connections_(0),
205 1737 : lifetime_manager_(XmppStaticObjectFactory::Create<XmppLifetimeManager>(
206 : TaskScheduler::GetInstance()->GetTaskId("bgp::Config"))),
207 1737 : deleter_(new DeleteActor(this)),
208 1737 : server_addr_(server_addr),
209 1737 : log_uve_(false),
210 1737 : auth_enabled_(false),
211 1737 : tcp_hold_time_(XmppChannelConfig::kTcpHoldTime),
212 1737 : gr_helper_disable_(false),
213 1737 : xmpp_config_updater_(NULL),
214 1737 : dscp_value_(0),
215 1737 : connection_queue_(TaskScheduler::GetInstance()->GetTaskId("bgp::Config"),
216 5211 : 0, boost::bind(&XmppServer::DequeueConnection, this, _1)) {
217 1737 : }
218 :
219 :
220 7 : XmppServer::XmppServer(EventManager *evm)
221 : : XmppConnectionManager(evm, ssl::context::sslv23_server, false, false),
222 7 : max_connections_(0),
223 7 : lifetime_manager_(new LifetimeManager(
224 14 : TaskScheduler::GetInstance()->GetTaskId("bgp::Config"))),
225 7 : deleter_(new DeleteActor(this)),
226 7 : log_uve_(false),
227 7 : auth_enabled_(false),
228 7 : tcp_hold_time_(XmppChannelConfig::kTcpHoldTime),
229 7 : gr_helper_disable_(false),
230 7 : dscp_value_(0),
231 7 : connection_queue_(TaskScheduler::GetInstance()->GetTaskId("bgp::Config"),
232 14 : 0, boost::bind(&XmppServer::DequeueConnection, this, _1)) {
233 7 : }
234 :
235 2078 : void XmppServer::CreateConfigUpdater(BgpConfigManager *config_manager) {
236 2078 : xmpp_config_updater_.reset(new XmppConfigUpdater(this, config_manager));
237 2078 : }
238 :
239 195 : uint16_t XmppServer::GetGracefulRestartTime() const {
240 : // Check if GR is disabled..
241 195 : if (!xmpp_config_updater_ || !xmpp_config_updater_->config().gr_enable())
242 0 : return 0;
243 195 : return xmpp_config_updater_->config().gr_time();
244 : }
245 :
246 58 : uint32_t XmppServer::GetLongLivedGracefulRestartTime() const {
247 : // Check if GR is disabled..
248 58 : if (!xmpp_config_updater_ || !xmpp_config_updater_->config().gr_enable())
249 0 : return 0;
250 58 : return xmpp_config_updater_->config().llgr_time();
251 : }
252 :
253 6745 : uint32_t XmppServer::GetEndOfRibReceiveTime() const {
254 6745 : if (xmpp_config_updater_)
255 6745 : return xmpp_config_updater_->config().end_of_rib_timeout();
256 0 : return BgpGlobalSystemConfig::kEndOfRibTime;
257 : }
258 :
259 9599 : uint32_t XmppServer::GetEndOfRibSendTime() const {
260 9599 : if (xmpp_config_updater_)
261 9599 : xmpp_config_updater_->config().end_of_rib_timeout();
262 9599 : return BgpGlobalSystemConfig::kEndOfRibTime;
263 : }
264 :
265 10604 : bool XmppServer::IsGRHelperModeEnabled() const {
266 : // Check if disabled in .conf file.
267 10604 : if (gr_helper_disable_)
268 832 : return false;
269 :
270 : // Check from configuration.
271 9772 : if (!xmpp_config_updater_)
272 44 : return false;
273 :
274 : // Check if GR is disabled..
275 9728 : if (!xmpp_config_updater_->config().gr_enable())
276 8564 : return false;
277 :
278 1163 : return xmpp_config_updater_->config().gr_xmpp_helper();
279 : }
280 :
281 12087 : bool XmppServer::IsPeerCloseGraceful() const {
282 :
283 : // If the server is deleted, do not do graceful restart
284 12087 : if (deleter()->IsDeleted())
285 1753 : return false;
286 :
287 : // Check if GR helper mode is disabled.
288 10334 : if (!IsGRHelperModeEnabled())
289 9423 : return false;
290 :
291 : // Enable GR if either gr-time or llgr-time is configured.
292 910 : return (xmpp_config_updater_->config().gr_time() ||
293 910 : xmpp_config_updater_->config().llgr_time());
294 : }
295 :
296 3526 : XmppServer::~XmppServer() {
297 2158 : STLDeleteElements(&connection_endpoint_map_);
298 2158 : TcpServer::ClearSessions();
299 3526 : }
300 :
301 0 : bool XmppServer::Initialize(short port) {
302 0 : log_uve_ = false;
303 0 : return TcpServer::Initialize(port);
304 : }
305 :
306 2036 : bool XmppServer::Initialize(short port, bool logUVE) {
307 2036 : log_uve_ = logUVE;
308 2036 : return TcpServer::Initialize(port);
309 : }
310 :
311 120 : bool XmppServer::Initialize(short port, bool logUVE, const IpAddress& ip) {
312 120 : log_uve_ = logUVE;
313 120 : return TcpServer::Initialize(port, ip);
314 : }
315 :
316 : //
317 : // Can be removed after Shutdown is renamed to ManagedDelete.
318 : //
319 2156 : void XmppServer::SessionShutdown() {
320 2156 : XmppConnectionManager::Shutdown();
321 2156 : }
322 :
323 : //
324 : // Return true if it's possible to delete the XmppServer.
325 : //
326 : // No need to check the connection WorkQueue since XmppServerConnections on it
327 : // are dependents of the XmppServer.
328 : //
329 2156 : bool XmppServer::MayDelete() const {
330 2156 : return (GetSessionQueueSize() == 0);
331 : }
332 :
333 : //
334 : // Trigger deletion of the XmppServer.
335 : //
336 : // A mutex is used to ensure that we do not create new XmppServerConnections
337 : // after this point. Note that this routine and AcceptSession may be called
338 : // concurrently from 2 different threads in tests.
339 : //
340 2223 : void XmppServer::Shutdown() {
341 2223 : std::scoped_lock lock(deletion_mutex_);
342 2223 : deleter_->Delete();
343 2223 : }
344 :
345 : //
346 : // Called when the XmppServer delete actor is being destroyed.
347 : //
348 2156 : void XmppServer::Terminate() {
349 2156 : ClearSessions();
350 2156 : connection_queue_.Shutdown();
351 2156 : }
352 :
353 6617 : LifetimeActor *XmppServer::deleter() {
354 6617 : return deleter_.get();
355 : }
356 :
357 12087 : LifetimeActor *XmppServer::deleter() const {
358 12087 : return deleter_.get();
359 : }
360 :
361 8968 : LifetimeManager *XmppServer::lifetime_manager() {
362 8968 : return lifetime_manager_.get();
363 : }
364 :
365 22 : TcpSession *XmppServer::CreateSession() {
366 : typedef boost::asio::detail::socket_option::boolean<
367 : SOL_SOCKET, SO_REUSEADDR> reuse_addr_t;
368 22 : TcpSession *session = TcpServer::CreateSession();
369 22 : Socket *socket = session->socket();
370 :
371 22 : boost::system::error_code err;
372 22 : socket->open(ip::tcp::v4(), err);
373 22 : if (err) {
374 0 : XMPP_WARNING(ServerOpenFail, err.message());
375 : }
376 :
377 22 : socket->set_option(reuse_addr_t(true), err);
378 22 : if (err) {
379 0 : XMPP_WARNING(SetSockOptFail, "", XMPP_PEER_DIR_OUT, err.message());
380 : }
381 :
382 22 : socket->bind(LocalEndpoint(), err);
383 22 : if (err) {
384 22 : XMPP_WARNING(ServerBindFailure, err.message());
385 : }
386 :
387 22 : XmppSession *xmpps = static_cast<XmppSession *>(session);
388 22 : err = xmpps->EnableTcpKeepalive(tcp_hold_time_);
389 22 : if (err) {
390 0 : XMPP_WARNING(ServerKeepAliveFailure, session->ToUVEKey(),
391 : XMPP_PEER_DIR_OUT, err.message());
392 : }
393 :
394 22 : return session;
395 : }
396 :
397 0 : size_t XmppServer::ConnectionEventCount() const {
398 0 : return connection_event_map_.size();
399 : }
400 :
401 854 : size_t XmppServer::ConnectionMapSize() const {
402 854 : ReadLock lock(connection_map_mutex_);
403 1708 : return connection_map_.size();
404 854 : }
405 :
406 853 : size_t XmppServer::ConnectionCount() const {
407 853 : return ConnectionMapSize() + deleted_connection_set_.size();
408 : }
409 :
410 6520 : XmppServerConnection *XmppServer::FindConnection(Endpoint remote_endpoint) {
411 6520 : ReadLock lock(connection_map_mutex_);
412 6520 : ConnectionMap::iterator loc = connection_map_.find(remote_endpoint);
413 6520 : if (loc != connection_map_.end()) {
414 0 : return loc->second;
415 : }
416 6520 : return NULL;
417 6520 : }
418 :
419 658 : XmppServerConnection *XmppServer::FindConnection(const string &address) {
420 658 : ReadLock lock(connection_map_mutex_);
421 669 : for (auto& value : connection_map_) {
422 38 : if (value.second->ToString() == address)
423 27 : return value.second;
424 : }
425 631 : return NULL;
426 658 : }
427 :
428 56 : bool XmppServer::ClearConnection(const string &hostname) {
429 56 : ReadLock lock(connection_map_mutex_);
430 197 : for (auto& value : connection_map_) {
431 157 : if (value.second->GetComputeHostName() == hostname) {
432 16 : value.second->Clear();
433 16 : return true;
434 : }
435 : }
436 40 : return false;
437 56 : }
438 :
439 16 : void XmppServer::UpdateAllConnections(uint8_t time_out) {
440 16 : ReadLock lock(connection_map_mutex_);
441 40 : for (auto& value : connection_map_) {
442 24 : value.second->UpdateKeepAliveTimer(time_out);
443 : }
444 16 : }
445 :
446 5301 : void XmppServer::ClearAllConnections() {
447 5301 : ReadLock lock(connection_map_mutex_);
448 7295 : for (auto& value : connection_map_) {
449 1994 : value.second->Clear();
450 : }
451 5301 : }
452 :
453 2090 : void XmppServer::RegisterConnectionEvent(xmps::PeerId id,
454 : ConnectionEventCb cb) {
455 2090 : connection_event_map_.insert(make_pair(id, cb));
456 2090 : }
457 :
458 2078 : void XmppServer::UnRegisterConnectionEvent(xmps::PeerId id) {
459 2078 : ConnectionEventCbMap::iterator it = connection_event_map_.find(id);
460 2078 : if (it != connection_event_map_.end())
461 2078 : connection_event_map_.erase(it);
462 2078 : }
463 :
464 12897 : void XmppServer::NotifyConnectionEvent(XmppChannelMux *mux,
465 : xmps::PeerState state) {
466 12897 : ConnectionEventCbMap::iterator iter = connection_event_map_.begin();
467 25770 : for (; iter != connection_event_map_.end(); ++iter) {
468 12870 : ConnectionEventCb cb = iter->second;
469 12869 : cb(mux, state);
470 12873 : }
471 12901 : }
472 :
473 6566 : SslSession *XmppServer::AllocSession(SslSocket *socket) {
474 6566 : SslSession *session = new XmppSession(this, socket);
475 6566 : boost::system::error_code err;
476 6566 : XmppSession *xmpp_session = static_cast<XmppSession *>(session);
477 6566 : err = xmpp_session->EnableTcpKeepalive(tcp_hold_time_);
478 6566 : if (err) {
479 22 : XMPP_WARNING(ServerKeepAliveFailure, xmpp_session->ToUVEKey(),
480 : XMPP_PEER_DIR_OUT, err.message());
481 : }
482 6566 : return session;
483 : }
484 :
485 : //
486 : // Accept newly formed passive tcp session by creating necessary xmpp data
487 : // structures. We do so to make sure that if there is any error reported
488 : // over this tcp session, it can still be correctly handled, even though
489 : // the allocated xmpp data structures are not fully processed yet.
490 : //
491 6544 : bool XmppServer::AcceptSession(TcpSession *tcp_session) {
492 6544 : std::scoped_lock lock(deletion_mutex_);
493 6544 : if (deleter_->IsDeleted())
494 0 : return false;
495 :
496 6544 : XmppSession *session = dynamic_cast<XmppSession *>(tcp_session);
497 6544 : XmppServerConnection *connection = CreateConnection(session);
498 :
499 6544 : if (xmpp_config_updater_) {
500 13062 : connection->state_machine()->set_hold_time(
501 6531 : xmpp_config_updater_->xmpp_hold_time());
502 : }
503 : // Register event handler.
504 6544 : tcp_session->set_observer(boost::bind(&XmppStateMachine::OnSessionEvent,
505 : connection->state_machine(), _1, _2));
506 :
507 : // set async_read_ready as false
508 6544 : session->set_read_on_connect(false);
509 6544 : connection->set_session(session);
510 6544 : connection->state_machine()->set_session(session);
511 6544 : connection->set_on_work_queue();
512 6544 : connection_queue_.Enqueue(connection);
513 6544 : return true;
514 6544 : }
515 :
516 : //
517 : // Remove the given XmppServerConnection from the ConnectionMap.
518 : //
519 6544 : void XmppServer::RemoveConnection(XmppServerConnection *connection) {
520 6544 : CHECK_CONCURRENCY("bgp::Config");
521 :
522 6544 : assert(connection->IsDeleted());
523 6544 : Endpoint endpoint = connection->endpoint();
524 6544 : WriteLock lock(connection_map_mutex_);
525 6544 : ConnectionMap::iterator loc = connection_map_.find(endpoint);
526 6544 : assert(loc != connection_map_.end() && loc->second == connection);
527 6544 : connection_map_.erase(loc);
528 6544 : }
529 :
530 104 : void XmppServer::SwapXmppConnectionMapEntries(
531 : XmppConnection *connection1, XmppConnection *connection2) {
532 104 : WriteLock lock(connection_map_mutex_);
533 : ConnectionMap::iterator loc1 =
534 104 : connection_map_.find(connection1->endpoint());
535 104 : assert(loc1 != connection_map_.end());
536 : ConnectionMap::iterator loc2 =
537 104 : connection_map_.find(connection2->endpoint());
538 104 : assert(loc2 != connection_map_.end());
539 104 : swap(loc1->second, loc2->second);
540 104 : swap(loc1->second->endpoint(), loc2->second->endpoint());
541 104 : }
542 :
543 : //
544 : // Insert the given XmppServerConnection into the ConnectionMap.
545 : //
546 6544 : void XmppServer::InsertConnection(XmppServerConnection *connection) {
547 6544 : CHECK_CONCURRENCY("bgp::Config");
548 :
549 6544 : assert(!connection->IsDeleted());
550 6544 : connection->Initialize();
551 6544 : Endpoint endpoint = connection->endpoint();
552 6544 : ConnectionMap::iterator loc;
553 : bool result;
554 6544 : WriteLock lock(connection_map_mutex_);
555 6544 : tie(loc, result) = connection_map_.insert(make_pair(endpoint, connection));
556 6544 : assert(result);
557 6544 : max_connections_ = max(max_connections_, connection_map_.size());
558 6544 : }
559 :
560 : //
561 : // Create XmppConnnection and its associated data structures. This API is
562 : // only used to allocate data structures and initialize necessary fields.
563 : // The data structures are not populated to any maps in the XmppServer at
564 : // this point. However, the newly created XmppServerConnection does add
565 : // itself as a dependent of the XmppServer via LifetimeManager linkage.
566 : //
567 6544 : XmppServerConnection *XmppServer::CreateConnection(XmppSession *session) {
568 : XmppServerConnection *connection;
569 :
570 6544 : XmppChannelConfig cfg(false);
571 6544 : cfg.endpoint = session->remote_endpoint();
572 6544 : cfg.local_endpoint = session->local_endpoint();
573 6544 : cfg.FromAddr = server_addr_;
574 6544 : cfg.logUVE = log_uve_;
575 6544 : cfg.auth_enabled = auth_enabled_;
576 6544 : cfg.dscp_value = dscp_value_;
577 :
578 6544 : XMPP_DEBUG(XmppCreateConnection, session->ToUVEKey(), XMPP_PEER_DIR_OUT,
579 : session->ToString());
580 : connection = XmppStaticObjectFactory::Create<XmppServerConnection>
581 6544 : (this, static_cast<const XmppChannelConfig*>(&cfg));
582 :
583 6544 : return connection;
584 6544 : }
585 :
586 7 : void XmppServer::SetDscpValue(uint8_t value) {
587 7 : ReadLock lock(connection_map_mutex_);
588 7 : dscp_value_ = value;
589 18 : for (auto& value : connection_map_) {
590 11 : XmppServerConnection *connection = value.second;
591 11 : connection->SetDscpValue(dscp_value_);
592 : }
593 7 : }
594 : //
595 : // Handler for XmppServerConnections that are dequeued from the WorkQueue.
596 : //
597 : // Since the XmppServerConnections on the WorkQueue are dependents of the
598 : // XmppServer, we are guaranteed that the XmppServer won't get destroyed
599 : // before the connection WorkQueue is drained.
600 : //
601 6544 : bool XmppServer::DequeueConnection(XmppServerConnection *connection) {
602 6544 : CHECK_CONCURRENCY("bgp::Config");
603 6544 : connection->clear_on_work_queue();
604 :
605 : // This happens if the XmppServer got deleted while the XmppConnnection
606 : // was on the WorkQueue.
607 6544 : if (connection->IsDeleted()) {
608 24 : connection->RetryDelete();
609 24 : return true;
610 : }
611 :
612 6520 : XmppSession *session = connection->session();
613 6520 : connection->clear_session();
614 6520 : Endpoint remote_endpoint = session->remote_endpoint();
615 6520 : XmppServerConnection *old_connection = FindConnection(remote_endpoint);
616 :
617 : // Close as duplicate if we have a connection from the same Endpoint.
618 : // Otherwise go ahead and insert into the ConnectionMap. We may find
619 : // it has a conflicting Endpoint name and decide to terminate it when
620 : // we process the Open message.
621 6520 : if (old_connection) {
622 0 : XMPP_DEBUG(XmppCreateConnection, session->ToUVEKey(), XMPP_PEER_DIR_IN,
623 : "Close duplicate connection " + session->ToString());
624 : // Remove reference to the session from StateMachine directly. We don't
625 : // go through the entire normal cleanup pipeline for cleaning these
626 : // duplicate connections.
627 0 : assert(connection->state_machine()->session());
628 0 : assert(connection->state_machine()->session() == session);
629 0 : connection->state_machine()->RemoveSession();
630 0 : DeleteSession(session);
631 0 : connection->set_duplicate();
632 0 : connection->ManagedDelete();
633 0 : InsertDeletedConnection(connection);
634 : } else {
635 6520 : InsertConnection(connection);
636 6520 : connection->AcceptSession(session);
637 : }
638 :
639 6520 : return true;
640 : }
641 :
642 674 : size_t XmppServer::GetConnectionQueueSize() const {
643 674 : return connection_queue_.Length();
644 : }
645 :
646 48 : void XmppServer::SetConnectionQueueDisable(bool disabled) {
647 48 : connection_queue_.set_disable(disabled);
648 48 : }
649 :
650 : //
651 : // Connection is marked deleted, add it to the ConnectionSet.
652 : //
653 6568 : void XmppServer::InsertDeletedConnection(XmppServerConnection *connection) {
654 6568 : CHECK_CONCURRENCY("bgp::Config");
655 :
656 6568 : assert(connection->IsDeleted());
657 6568 : ConnectionSet::iterator it;
658 : bool result;
659 6568 : tie(it, result) = deleted_connection_set_.insert(connection);
660 6568 : assert(result);
661 6568 : }
662 :
663 : //
664 : // Connection is being destroyed, remove it from the ConnectionSet.
665 : //
666 6568 : void XmppServer::RemoveDeletedConnection(XmppServerConnection *connection) {
667 6568 : CHECK_CONCURRENCY("bgp::Config");
668 :
669 6568 : assert(connection->IsDeleted());
670 6568 : ConnectionSet::iterator it = deleted_connection_set_.find(connection);
671 6568 : assert(it != deleted_connection_set_.end());
672 6568 : deleted_connection_set_.erase(it);
673 6568 : ReleaseConnectionEndpoint(connection);
674 6568 : }
675 :
676 11744 : XmppConnectionEndpoint *XmppServer::FindConnectionEndpoint(
677 : const string &endpoint_name) {
678 11744 : std::scoped_lock lock(endpoint_map_mutex_);
679 : ConnectionEndpointMap::const_iterator loc =
680 11744 : connection_endpoint_map_.find(endpoint_name);
681 23488 : return (loc != connection_endpoint_map_.end() ? loc->second : NULL);
682 11744 : }
683 :
684 6305 : XmppConnectionEndpoint *XmppServer::LocateConnectionEndpoint(
685 : XmppServerConnection *connection, bool &created) {
686 6305 : created = false;
687 6305 : if (!connection)
688 0 : return NULL;
689 :
690 6305 : std::scoped_lock lock(endpoint_map_mutex_);
691 :
692 : ConnectionEndpointMap::const_iterator loc =
693 6305 : connection_endpoint_map_.find(connection->ToString());
694 : XmppConnectionEndpoint *conn_endpoint;
695 :
696 6304 : if (loc != connection_endpoint_map_.end()) {
697 1429 : conn_endpoint = loc->second;
698 1429 : if (!conn_endpoint->connection()) {
699 1325 : created = true;
700 1325 : conn_endpoint->set_connection(connection);
701 1325 : connection->set_conn_endpoint(conn_endpoint);
702 : }
703 1429 : return conn_endpoint;
704 : }
705 :
706 4875 : created = true;
707 4875 : conn_endpoint = new XmppConnectionEndpoint(connection->ToString());
708 : bool result;
709 4875 : tie(loc, result) = connection_endpoint_map_.insert(
710 9751 : make_pair(connection->ToString(), conn_endpoint));
711 4876 : assert(result);
712 4876 : conn_endpoint->set_connection(connection);
713 4876 : connection->set_conn_endpoint(conn_endpoint);
714 4876 : return conn_endpoint;
715 6305 : }
716 :
717 : //
718 : // Remove association of the given XmppConnectionEndpoint with XmppConnection.
719 : // This method is provided just to make things symmetrical - the caller could
720 : // simply have called XmppConnectionEndpoint::reset_connection directly.
721 : //
722 13112 : void XmppServer::ReleaseConnectionEndpoint(XmppServerConnection *connection) {
723 13112 : std::scoped_lock lock(endpoint_map_mutex_);
724 :
725 13112 : if (!connection->conn_endpoint())
726 6911 : return;
727 6201 : assert(connection->conn_endpoint()->connection() == connection);
728 6201 : connection->conn_endpoint()->reset_connection();
729 6201 : connection->set_conn_endpoint(NULL);
730 13112 : }
731 :
732 40 : void XmppServer::FillShowConnections(
733 : vector<ShowXmppConnection> *show_connection_list) const {
734 40 : ReadLock lock(connection_map_mutex_);
735 88 : for (const auto& value : connection_map_) {
736 48 : const XmppServerConnection *connection = value.second;
737 48 : ShowXmppConnection show_connection;
738 48 : connection->FillShowInfo(&show_connection);
739 48 : show_connection_list->push_back(show_connection);
740 48 : }
741 64 : for (const auto& connection : deleted_connection_set_) {
742 24 : ShowXmppConnection show_connection;
743 24 : connection->FillShowInfo(&show_connection);
744 24 : show_connection_list->push_back(show_connection);
745 24 : }
746 40 : }
747 :
748 1 : void XmppServer::FillShowServer(ShowXmppServerResp *resp) const {
749 1 : SocketIOStats peer_socket_stats;
750 1 : GetRxSocketStats(&peer_socket_stats);
751 1 : resp->set_rx_socket_stats(peer_socket_stats);
752 1 : GetTxSocketStats(&peer_socket_stats);
753 1 : resp->set_tx_socket_stats(peer_socket_stats);
754 1 : resp->set_current_connections(ConnectionMapSize());
755 1 : resp->set_max_connections(max_connections_);
756 1 : }
757 :
758 : class ShowXmppConnectionHandler {
759 : public:
760 40 : static bool CallbackS1(const Sandesh *sr,
761 : const RequestPipeline::PipeSpec ps, int stage, int instNum,
762 : RequestPipeline::InstData *data) {
763 : const ShowXmppConnectionReq *req =
764 40 : static_cast<const ShowXmppConnectionReq *>(ps.snhRequest_.get());
765 : XmppSandeshContext *xsc =
766 40 : dynamic_cast<XmppSandeshContext *>(req->module_context("XMPP"));
767 :
768 40 : ShowXmppConnectionResp *resp = new ShowXmppConnectionResp;
769 40 : vector<ShowXmppConnection> connections;
770 40 : if (xsc)
771 40 : xsc->xmpp_server->FillShowConnections(&connections);
772 40 : resp->set_connections(connections);
773 40 : resp->set_context(req->context());
774 40 : resp->Response();
775 40 : return true;
776 40 : }
777 : };
778 :
779 40 : void ShowXmppConnectionReq::HandleRequest() const {
780 40 : RequestPipeline::PipeSpec ps(this);
781 :
782 : // Request pipeline has single stage to collect connection info and
783 : // respond to the request.
784 40 : RequestPipeline::StageSpec s1;
785 40 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
786 40 : s1.taskId_ = scheduler->GetTaskId("bgp::ShowCommand");
787 40 : s1.cbFn_ = ShowXmppConnectionHandler::CallbackS1;
788 40 : s1.instances_.push_back(0);
789 40 : ps.stages_.push_back(s1);
790 40 : RequestPipeline rp(ps);
791 40 : }
792 :
793 : class ClearXmppConnectionHandler {
794 : public:
795 48 : static bool CallbackS1(const Sandesh *sr,
796 : const RequestPipeline::PipeSpec ps, int stage, int instNum,
797 : RequestPipeline::InstData *data) {
798 :
799 : const ClearXmppConnectionReq *req =
800 48 : static_cast<const ClearXmppConnectionReq *>(ps.snhRequest_.get());
801 : XmppSandeshContext *xsc =
802 48 : dynamic_cast<XmppSandeshContext *>(req->module_context("XMPP"));
803 :
804 48 : ClearXmppConnectionResp *resp = new ClearXmppConnectionResp;
805 48 : if (!xsc || !xsc->test_mode) {
806 24 : resp->set_success(false);
807 24 : } else if (req->get_hostname_or_all() != "all") {
808 16 : if (xsc->xmpp_server->ClearConnection(req->get_hostname_or_all())) {
809 8 : resp->set_success(true);
810 : } else {
811 8 : resp->set_success(false);
812 : }
813 : } else {
814 8 : if (xsc->xmpp_server->ConnectionCount()) {
815 8 : xsc->xmpp_server->ClearAllConnections();
816 8 : resp->set_success(true);
817 : } else {
818 0 : resp->set_success(false);
819 : }
820 : }
821 :
822 48 : resp->set_context(req->context());
823 48 : resp->Response();
824 48 : return true;
825 : }
826 : };
827 :
828 48 : void ClearXmppConnectionReq::HandleRequest() const {
829 :
830 : // config task is used to create and delete connection objects.
831 : // hence use the same task to find the connection
832 48 : RequestPipeline::StageSpec s1;
833 48 : s1.taskId_ = TaskScheduler::GetInstance()->GetTaskId("bgp::Config");
834 48 : s1.instances_.push_back(0);
835 48 : s1.cbFn_ = ClearXmppConnectionHandler::CallbackS1;
836 :
837 48 : RequestPipeline::PipeSpec ps(this);
838 48 : ps.stages_.push_back(s1);
839 48 : RequestPipeline rp(ps);
840 48 : }
841 :
842 : class ShowXmppServerHandler {
843 : public:
844 1 : static bool CallbackS1(const Sandesh *sr,
845 : const RequestPipeline::PipeSpec ps, int stage, int instNum,
846 : RequestPipeline::InstData *data) {
847 : const ShowXmppServerReq *req =
848 1 : static_cast<const ShowXmppServerReq *>(ps.snhRequest_.get());
849 : XmppSandeshContext *xsc =
850 1 : dynamic_cast<XmppSandeshContext *>(req->module_context("XMPP"));
851 :
852 1 : ShowXmppServerResp *resp = new ShowXmppServerResp;
853 1 : if (xsc)
854 1 : xsc->xmpp_server->FillShowServer(resp);
855 1 : resp->set_context(req->context());
856 1 : resp->Response();
857 1 : return true;
858 : }
859 : };
860 :
861 1 : void ShowXmppServerReq::HandleRequest() const {
862 1 : RequestPipeline::PipeSpec ps(this);
863 1 : RequestPipeline::StageSpec s1;
864 1 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
865 1 : s1.taskId_ = scheduler->GetTaskId("bgp::ShowCommand");
866 1 : s1.cbFn_ = ShowXmppServerHandler::CallbackS1;
867 1 : s1.instances_.push_back(0);
868 1 : ps.stages_.push_back(s1);
869 1 : RequestPipeline rp(ps);
870 1 : }
|
