- Main Page
- Modules
- Namespaces
- Classes
- Files

- File List

# EventedServer.h

    00001 /*
    00002  *  Phusion Passenger - http://www.modrails.com/
    00003  *  Copyright (c) 2010 Phusion
    00004  *
    00005  *  "Phusion Passenger" is a trademark of Hongli Lai & Ninh Bui.
    00006  *
    00007  *  Permission is hereby granted, free of charge, to any person obtaining a copy
    00008  *  of this software and associated documentation files (the "Software"), to deal
    00009  *  in the Software without restriction, including without limitation the rights
    00010  *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
    00011  *  copies of the Software, and to permit persons to whom the Software is
    00012  *  furnished to do so, subject to the following conditions:
    00013  *
    00014  *  The above copyright notice and this permission notice shall be included in
    00015  *  all copies or substantial portions of the Software.
    00016  *
    00017  *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    00018  *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    00019  *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
    00020  *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    00021  *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
    00022  *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
    00023  *  THE SOFTWARE.
    00024  */
    00025 #ifndef _PASSENGER_EVENTED_SERVER_H_
    00026 #define _PASSENGER_EVENTED_SERVER_H_
    00027
    00028 #include <ev++.h>
    00029 #include <boost/shared_ptr.hpp>
    00030 #include <boost/enable_shared_from_this.hpp>
    00031 #include <oxt/system_calls.hpp>
    00032 #include <oxt/macros.hpp>
    00033 #include <algorithm>
    00034 #include <string>
    00035 #include <set>
    00036 #include <sys/types.h>
    00037 #include <sys/un.h>
    00038 #include <sys/uio.h>
    00039 #include <netinet/in.h>
    00040 #include <unistd.h>
    00041 #include <cerrno>
    00042 #include "EventedClient.h"
    00043 #include "FileDescriptor.h"
    00044 #include "StaticString.h"
    00045 #include "Logging.h"
    00046 #include "Utils/ScopeGuard.h"
    00047 #include "Utils/StrIntUtils.h"
    00048
    00049 namespace Passenger {
    00050
    00051 using namespace std;
    00052 using namespace boost;
    00053 using namespace oxt;
    00054
    00055
    00056 /**
    00057  * A base class for writing single-threaded, evented servers that use non-blocking I/O.
    00058  * It uses libev for its event loop. EventedServer handles much of the situps regarding
    00059  * client connection management and output buffering and tries to make it easy to
    00060  * implement a zero-copy architecture.
    00061  *
    00062  * <h2>Basic usage</h2>
    00063  * Derived classes can override the onClientReadable() method, which is called every time
    00064  * a specific client becomes readable. It is passed a Client object which contains information
    00065  * about the client, such as its file descriptor. One can use the read() system call in
    00066  * that method to receive data from the client. Please note that client file descriptors
    00067  * are always set to non-blocking mode so you need to handle this gracefully.
    00068  *
    00069  * EventedServer provides the write() method for sending data to the client. This method
    00070  * will attempt to send the data to the client immediately; if it fails with EAGAIN then
    00071  * EventedServer will take care of scheduling the send at a later time when the client
    00072  * is ready again to receive data.
    00073  *
    00074  * To disconnect the client, call disconnect(). The connection might not be actually
    00075  * closed until all pending outgoing data have been sent out, but all the gory details
    00076  * is taken care of for you.
    00077  *
    00078  * <h2>Keeping per-client information</h2>
    00079  * If you need to keep per-client information then you can override the createClient()
    00080  * method and make it return an object that's a subclass of EventedServer::Client. This
    00081  * object is passed to onClientReadable(), so in there you can just cast the client object
    00082  * to your subclass.
    00083  *
    00084  * <h2>Client object life times</h2>
    00085  * All client objects are destroyed along with the EventedServer. Disconnecting
    00086  * a client also causes it to be destroyed after the current event handler has
    00087  * run or a short period of time after that.
    00088  */
    00089 class EventedServer {
    00090 protected:
    00091         typedef set<EventedClient *> ClientSet;
    00092
    00093         const ClientSet &getClients() const {
    00094                 return clients;
    00095         }
    00096
    00097         string getClientName(const EventedClient *client) const {
    00098                 return toString(client);
    00099         }
    00100
    00101         void logError(const EventedClient *client, const string &message) {
    00102                 P_ERROR("Error in client " << getClientName(client) << ": " << message);
    00103         }
    00104
    00105         void logSystemError(const EventedClient *client, const string &message, int errorCode) {
    00106                 P_ERROR("Error in client " << getClientName(client) << ": " <<
    00107                         message << ": " << strerror(errorCode) << " (" << errorCode << ")");
    00108         }
    00109
    00110         void logSystemError(const string &message, int errorCode) {
    00111                 P_ERROR(message << ": " << strerror(errorCode) << " (" << errorCode << ")");
    00112         }
    00113
    00114         virtual EventedClient *createClient(const FileDescriptor &fd) {
    00115                 return new EventedClient(loop, fd);
    00116         }
    00117
    00118         virtual void onNewClient(EventedClient *client) { }
    00119         virtual void onClientReadable(EventedClient *client) { }
    00120
    00121         /**
    00122          * Called when a client has been disconnected. This may either be triggered
    00123          * immediately by disconnect() or triggered after pending data has been sent
    00124          * out. This means that if you call disconnect() from onClientReadable() you
    00125          * need take care of the possibility that control returns to onClientReadable()
    00126          * after this method is done.
    00127          *
    00128          * Please note that when EventedServer is being destroyed,
    00129          * onClientDisconnected() is *not* triggered.
    00130          */
    00131         virtual void onClientDisconnected(EventedClient *client) { }
    00132
    00133 private:
    00134         struct ev_loop *loop;
    00135         FileDescriptor fd;
    00136         ev::io acceptWatcher;
    00137         ClientSet clients;
    00138
    00139         void removeClient(EventedClient *client) {
    00140                 clients.erase(client);
    00141         }
    00142
    00143         void freeAllClients() {
    00144                 ClientSet::iterator it;
    00145                 ClientSet::iterator end = clients.end();
    00146
    00147                 for (it = clients.begin(); it != clients.end(); it++) {
    00148                         (*it)->unref();
    00149                 }
    00150                 clients.clear();
    00151         }
    00152
    00153         static void _onReadable(EventedClient *client) {
    00154                 EventedServer *server = (EventedServer *) client->userData;
    00155                 client->ref();
    00156                 ScopeGuard guard(boost::bind(&EventedClient::unref, client));
    00157                 server->onClientReadable((EventedClient *) client);
    00158         }
    00159
    00160         static void _onDisconnect(EventedClient *client) {
    00161                 EventedServer *server = (EventedServer *) client->userData;
    00162                 ScopeGuard guard1(boost::bind(&EventedClient::unref, client));
    00163
    00164                 client->ref();
    00165                 ScopeGuard guard2(boost::bind(&EventedClient::unref, client));
    00166
    00167                 server->removeClient(client);
    00168                 server->onClientDisconnected((EventedClient *) client);
    00169         }
    00170
    00171         static void _onDetach(EventedClient *client) {
    00172                 EventedServer *server = (EventedServer *) client->userData;
    00173                 ScopeGuard guard1(boost::bind(&EventedClient::unref, client));
    00174
    00175                 client->ref();
    00176                 ScopeGuard guard2(boost::bind(&EventedClient::unref, client));
    00177
    00178                 server->removeClient(client);
    00179         }
    00180
    00181         static void _onSystemError(EventedClient *client, const string &message, int code) {
    00182                 EventedServer *server = (EventedServer *) client->userData;
    00183                 server->logSystemError(client, message, code);
    00184         }
    00185
    00186         void exceptionThrownWhileInitializingClient(EventedClient *client, ClientSet::iterator it) {
    00187                 if (!client->ioAllowed()) {
    00188                         // onNewClient() disconnected or detached the
    00189                         // client, so client refcount == 1
    00190                         client->unref();
    00191                 } else {
    00192                         // client refcount == 2
    00193                         client->unref();
    00194                         client->unref();
    00195                         clients.erase(it);
    00196                 }
    00197                 // Now client refcount == 0
    00198         }
    00199
    00200         void onAcceptable(ev::io &w, int revents) {
    00201                 this_thread::disable_syscall_interruption dsi;
    00202                 int i = 0;
    00203                 bool done = false;
    00204
    00205                 // Accept at most 10 connections on every accept readiness event
    00206                 // in order to give other events the chance to be processed.
    00207                 while (i < 10 && !done) {
    00208                         // Reserve enough space to hold both a Unix domain socket
    00209                         // address and an IP socket address.
    00210                         union {
    00211                                 struct sockaddr_un local;
    00212                                 struct sockaddr_in inet;
    00213                         } addr;
    00214                         socklen_t len = sizeof(addr);
    00215
    00216                         int clientfd = syscalls::accept(fd, (struct sockaddr *) &addr, &len);
    00217                         if (clientfd == -1) {
    00218                                 if (errno != EAGAIN && errno != EWOULDBLOCK) {
    00219                                         int e = errno;
    00220                                         logSystemError("Cannot accept new client", e);
    00221                                 }
    00222                                 done = true;
    00223                         } else {
    00224                                 FileDescriptor clientfdGuard = clientfd;
    00225                                 int optval = 1;
    00226
    00227                                 setNonBlocking(clientfdGuard);
    00228                                 syscalls::setsockopt(clientfd, SOL_SOCKET, SO_KEEPALIVE,
    00229                                         &optval, sizeof(optval));
    00230
    00231                                 EventedClient *client = createClient(clientfdGuard);
    00232                                 client->onReadable    = _onReadable;
    00233                                 client->onDisconnect  = _onDisconnect;
    00234                                 client->onDetach      = _onDetach;
    00235                                 client->onSystemError = _onSystemError;
    00236                                 client->userData      = this;
    00237                                 client->notifyReads(true);
    00238
    00239                                 pair<ClientSet::iterator, bool> p = clients.insert(client);
    00240
    00241                                 client->ref();
    00242                                 // client refcount == 2
    00243                                 {
    00244                                         ScopeGuard g(boost::bind(
    00245                                                 &EventedServer::exceptionThrownWhileInitializingClient,
    00246                                                 this,
    00247                                                 client,
    00248                                                 p.first));
    00249                                         onNewClient(client);
    00250                                         g.clear();
    00251                                         // If exception occurred: client refcount == 0
    00252                                 }
    00253
    00254                                 /* No exception occured.
    00255                                  * If onNewClient() disconnected or detached the client:
    00256                                  *     client refcount == 1
    00257                                  *     We want the refcount to become 0.
    00258                                  * If not:
    00259                                  *     client refcount == 2
    00260                                  *     We want the refcount to become 1.
    00261                                  */
    00262                                 client->unref();
    00263                         }
    00264                         i++;
    00265                 }
    00266         }
    00267
    00268 public:
    00269         EventedServer(struct ev_loop *_loop, FileDescriptor serverFd)
    00270                 : loop(_loop),
    00271                   acceptWatcher(_loop)
    00272         {
    00273                 fd = serverFd;
    00274                 setNonBlocking(serverFd);
    00275                 acceptWatcher.set<EventedServer, &EventedServer::onAcceptable>(this);
    00276                 acceptWatcher.start(fd, ev::READ);
    00277         }
    00278
    00279         virtual ~EventedServer() {
    00280                 freeAllClients();
    00281         }
    00282
    00283         struct ev_loop *getLoop() const {
    00284                 return loop;
    00285         }
    00286
    00287         FileDescriptor getServerFd() const {
    00288                 return fd;
    00289         }
    00290 };
    00291
    00292
    00293 } // namespace Passenger
    00294
    00295 #endif /* _PASSENGER_EVENTED_SERVER_H_ */