From d738d179e4c040e62438516bc0992736d00b902f Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Fri, 4 Jul 2008 19:07:33 +0000 Subject: Cluster prototype: handles client-initiated commands (not dequeues) Details - Cluster.cpp: serializes all frames thru cluster (see below) - broker/ConnectionManager: Added handler chain in front of Connection::received. - sys::Fork and ForkWithMessage - abstractions for forking with posix impl. - tests/ForkedBroker.h: test utility to fork a broker process. - broker/SignalHandler: Encapsulated signal handling from qpidd.cpp - Various minor logging & error message improvements to aid debugging. NB: current impl will not scale. It is functional working starting point so we can start testing & profiling to find the right optimizations. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@674107 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/qpid/broker/Broker.h | 3 ++ cpp/src/qpid/broker/Connection.cpp | 8 +++- cpp/src/qpid/broker/Connection.h | 9 ++++ cpp/src/qpid/broker/ConnectionManager.cpp | 41 ++++++++++++++++++ cpp/src/qpid/broker/ConnectionManager.h | 70 +++++++++++++++++++++++++++++++ cpp/src/qpid/broker/SignalHandler.cpp | 51 ++++++++++++++++++++++ cpp/src/qpid/broker/SignalHandler.h | 47 +++++++++++++++++++++ 7 files changed, 227 insertions(+), 2 deletions(-) create mode 100644 cpp/src/qpid/broker/ConnectionManager.cpp create mode 100644 cpp/src/qpid/broker/ConnectionManager.h create mode 100644 cpp/src/qpid/broker/SignalHandler.cpp create mode 100644 cpp/src/qpid/broker/SignalHandler.h (limited to 'cpp/src/qpid/broker') diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h index 531817db83..00fb4b9995 100644 --- a/cpp/src/qpid/broker/Broker.h +++ b/cpp/src/qpid/broker/Broker.h @@ -23,6 +23,7 @@ */ #include "ConnectionFactory.h" +#include "ConnectionManager.h" #include "ConnectionToken.h" #include "DirectExchange.h" #include "DtxManager.h" @@ -120,6 +121,7 @@ class Broker : public sys::Runnable, public Plugin::Target, Options& getOptions() { return config; } SessionManager& getSessionManager() { return sessionManager; } + ConnectionManager& getConnectionManager() { return connectionManager; } management::ManagementObject::shared_ptr GetManagementObject (void) const; management::Manageable* GetVhostObject (void) const; @@ -158,6 +160,7 @@ class Broker : public sys::Runnable, public Plugin::Target, ConnectionFactory factory; DtxManager dtxManager; SessionManager sessionManager; + ConnectionManager connectionManager; management::ManagementAgent::shared_ptr managementAgent; management::Broker::shared_ptr mgmtObject; Vhost::shared_ptr vhostObject; diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp index 9e763f6775..f6d35ff6ca 100644 --- a/cpp/src/qpid/broker/Connection.cpp +++ b/cpp/src/qpid/broker/Connection.cpp @@ -53,7 +53,9 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std isLink(isLink_), mgmtClosing(false), mgmtId(mgmtId_), - links(broker_.getLinks()) + links(broker_.getLinks()), + lastInHandler(*this), + inChain(lastInHandler) { Manageable* parent = broker.GetVhostObject(); @@ -86,7 +88,9 @@ Connection::~Connection() links.notifyClosed(mgmtId); } -void Connection::received(framing::AMQFrame& frame){ +void Connection::received(framing::AMQFrame& frame){ inChain(frame); } + +void Connection::receivedLast(framing::AMQFrame& frame){ if (frame.getChannel() == 0 && frame.getMethod()) { adapter.handle(frame); } else { diff --git a/cpp/src/qpid/broker/Connection.h b/cpp/src/qpid/broker/Connection.h index 9e713140dd..6b3530366d 100644 --- a/cpp/src/qpid/broker/Connection.h +++ b/cpp/src/qpid/broker/Connection.h @@ -56,6 +56,7 @@ class Connection : public sys::ConnectionInputHandler, { public: typedef boost::shared_ptr shared_ptr; + Connection(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isLink = false); ~Connection (); @@ -90,10 +91,15 @@ class Connection : public sys::ConnectionInputHandler, void notifyConnectionForced(const std::string& text); void setUserId(const string& uid); + framing::FrameHandler::Chain& getInChain() { return inChain; } + private: typedef boost::ptr_map ChannelMap; typedef std::vector::iterator queue_iterator; + // End of the received handler chain. + void receivedLast(framing::AMQFrame& frame); + ChannelMap channels; framing::AMQP_ClientProxy::Connection* client; ConnectionHandler adapter; @@ -103,6 +109,9 @@ class Connection : public sys::ConnectionInputHandler, boost::function0 ioCallback; management::Connection::shared_ptr mgmtObject; LinkRegistry& links; + framing::FrameHandler::MemFunRef lastInHandler; + framing::FrameHandler::Chain inChain; + }; }} diff --git a/cpp/src/qpid/broker/ConnectionManager.cpp b/cpp/src/qpid/broker/ConnectionManager.cpp new file mode 100644 index 0000000000..165de5220e --- /dev/null +++ b/cpp/src/qpid/broker/ConnectionManager.cpp @@ -0,0 +1,41 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +#include "ConnectionManager.h" +#include "Connection.h" + +namespace qpid { +namespace broker { + +std::auto_ptr +ConnectionManager::create(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isClient) { + std::auto_ptr c(new Connection(out, broker, mgmtId, isClient)); + sys::Mutex::ScopedLock l(lock); + std::for_each(observers.begin(), observers.end(), + boost::bind(&Observer::created, _1, boost::ref(*c))); + return c; +} + +void ConnectionManager::add(const boost::intrusive_ptr& observer) { + sys::Mutex::ScopedLock l(lock); + observers.push_back(observer); +} + +}} // namespace qpid::broker diff --git a/cpp/src/qpid/broker/ConnectionManager.h b/cpp/src/qpid/broker/ConnectionManager.h new file mode 100644 index 0000000000..a999523d0d --- /dev/null +++ b/cpp/src/qpid/broker/ConnectionManager.h @@ -0,0 +1,70 @@ +#ifndef QPID_BROKER_CONNECTIONMANAGER_H +#define QPID_BROKER_CONNECTIONMANAGER_H + +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#include "qpid/RefCounted.h" +#include "qpid/sys/Mutex.h" +#include +#include +#include + +namespace qpid { + +namespace sys { +class ConnectionOutputHandler; +} + +namespace broker { + +class Broker; +class Connection; + +/** + * Manages connections and observers. + */ +class ConnectionManager { + public: + + /** + * Observer notified of ConnectionManager events. + */ + struct Observer : public RefCounted { + /** Called when a connection is attached. */ + virtual void created(Connection&) {} + }; + + /** Called to create a new Connection, applies observers. */ + std::auto_ptr create(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isClient = false); + + /** Add an observer */ + void add(const boost::intrusive_ptr&); + + private: + typedef std::vector > Observers; + + sys::Mutex lock; + Observers observers; +}; +}} // namespace qpid::broker + +#endif /*!QPID_BROKER_CONNECTIONMANAGER_H*/ diff --git a/cpp/src/qpid/broker/SignalHandler.cpp b/cpp/src/qpid/broker/SignalHandler.cpp new file mode 100644 index 0000000000..c6d7b10218 --- /dev/null +++ b/cpp/src/qpid/broker/SignalHandler.cpp @@ -0,0 +1,51 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +#include "SignalHandler.h" +#include "Broker.h" +#include + +namespace qpid { +namespace broker { + +boost::shared_ptr SignalHandler::broker; + +void SignalHandler::setBroker(const boost::shared_ptr& b) { + broker = b; + + signal(SIGINT,shutdownHandler); + signal(SIGTERM, shutdownHandler); + + signal(SIGHUP,SIG_IGN); // TODO aconway 2007-07-18: reload config. + + signal(SIGCHLD,SIG_IGN); + signal(SIGTSTP,SIG_IGN); + signal(SIGTTOU,SIG_IGN); + signal(SIGTTIN,SIG_IGN); +} + +void SignalHandler::shutdownHandler(int) { + if (broker.get()) { + broker->shutdown(); + broker.reset(); + } +} + +}} // namespace qpid::broker diff --git a/cpp/src/qpid/broker/SignalHandler.h b/cpp/src/qpid/broker/SignalHandler.h new file mode 100644 index 0000000000..3da7d09756 --- /dev/null +++ b/cpp/src/qpid/broker/SignalHandler.h @@ -0,0 +1,47 @@ +#ifndef QPID_BROKER_SIGNALHANDLER_H +#define QPID_BROKER_SIGNALHANDLER_H + +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#include + +namespace qpid { +namespace broker { + +class Broker; + +/** + * Handle signals e.g. to shut-down a broker. + */ +class SignalHandler +{ + public: + /** Set the broker to be shutdown on signals */ + static void setBroker(const boost::shared_ptr& broker); + + private: + static void shutdownHandler(int); + static boost::shared_ptr broker; +}; +}} // namespace qpid::broker + +#endif /*!QPID_BROKER_SIGNALHANDLER_H*/ -- cgit v1.2.1