X-Git-Url: https://git.netwichtig.de/gitweb/?a=blobdiff_plain;f=src%2Finspsocket.cpp;h=fc8bc142f9df2dfeb7ef54f4294e9fdb9dc7b8a9;hb=9db7af579c46a9f0379fdf71fb773a0a76a94846;hp=eb92030c248284f0d726acfe4c946bfe2a7faf23;hpb=03a3ca90198b1e61c72c817c58b68d55b8115d22;p=user%2Fhenk%2Fcode%2Finspircd.git diff --git a/src/inspsocket.cpp b/src/inspsocket.cpp index eb92030c2..fc8bc142f 100644 --- a/src/inspsocket.cpp +++ b/src/inspsocket.cpp @@ -2,8 +2,8 @@ * | Inspire Internet Relay Chat Daemon | * +------------------------------------+ * - * InspIRCd: (C) 2002-2007 InspIRCd Development Team - * See: http://www.inspircd.org/wiki/index.php/Credits + * InspIRCd: (C) 2002-2009 InspIRCd Development Team + * See: http://wiki.inspircd.org/Credits * * This program is free but copyrighted software; see * the file COPYING for details. @@ -11,337 +11,112 @@ * --------------------------------------------------- */ +#include "inspircd.h" #include "socket.h" -#include "configreader.h" #include "inspstring.h" #include "socketengine.h" -#include "inspircd.h" - -using irc::sockets::OpenTCPSocket; -bool InspSocket::Readable() -{ - return ((this->state != I_CONNECTING) && (this->WaitingForWriteEvent == false)); -} +#ifndef DISABLE_WRITEV +#include +#ifndef IOV_MAX +#define IOV_MAX 1024 +#endif +#endif -InspSocket::InspSocket(InspIRCd* SI) +BufferedSocket::BufferedSocket() { - this->Timeout = NULL; - this->state = I_DISCONNECTED; - this->fd = -1; - this->WaitingForWriteEvent = false; - this->Instance = SI; - this->IsIOHooked = false; + Timeout = NULL; + state = I_ERROR; } -InspSocket::InspSocket(InspIRCd* SI, int newfd, const char* ip) +BufferedSocket::BufferedSocket(int newfd) { - this->Timeout = NULL; + Timeout = NULL; this->fd = newfd; this->state = I_CONNECTED; - strlcpy(this->IP,ip,MAXBUF); - this->WaitingForWriteEvent = false; - this->Instance = SI; - this->IsIOHooked = false; - if (this->fd > -1) - this->Instance->SE->AddFd(this); + if (fd > -1) + ServerInstance->SE->AddFd(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE); } -InspSocket::InspSocket(InspIRCd* SI, const std::string &ipaddr, int aport, bool listening, unsigned long maxtime, const std::string &connectbindip) +void BufferedSocket::DoConnect(const std::string &ipaddr, int aport, unsigned long maxtime, const std::string &connectbindip) { - this->cbindip = connectbindip; - this->fd = -1; - this->Instance = SI; - strlcpy(host,ipaddr.c_str(),MAXBUF); - this->WaitingForWriteEvent = false; - this->IsIOHooked = false; - this->Timeout = NULL; - if (listening) + BufferedSocketError err = BeginConnect(ipaddr, aport, maxtime, connectbindip); + if (err != I_ERR_NONE) { - if ((this->fd = OpenTCPSocket(host)) == ERROR) - { - this->fd = -1; - this->state = I_ERROR; - this->OnError(I_ERR_SOCKET); - return; - } - else - { - if (!SI->BindSocket(this->fd,aport,(char*)ipaddr.c_str())) - { - this->Close(); - this->fd = -1; - this->state = I_ERROR; - this->OnError(I_ERR_BIND); - this->ClosePending = true; - return; - } - else - { - this->state = I_LISTENING; - this->port = aport; - if (this->fd > -1) - { - if (!this->Instance->SE->AddFd(this)) - { - this->Close(); - this->state = I_ERROR; - this->OnError(I_ERR_NOMOREFDS); - } - } - return; - } - } + state = I_ERROR; + SetError(strerror(errno)); + OnError(err); } - else +} + +BufferedSocketError BufferedSocket::BeginConnect(const std::string &ipaddr, int aport, unsigned long maxtime, const std::string &connectbindip) +{ + irc::sockets::sockaddrs addr, bind; + if (!irc::sockets::aptosa(ipaddr, aport, &addr)) { - strlcpy(this->host,ipaddr.c_str(),MAXBUF); - this->port = aport; + ServerInstance->Logs->Log("SOCKET", DEBUG, "BUG: Hostname passed to BufferedSocket, rather than an IP address!"); + return I_ERR_CONNECT; + } - bool ipvalid = true; -#ifdef IPV6 - if (strchr(host,':')) - { - in6_addr n; - if (inet_pton(AF_INET6, host, &n) < 1) - ipvalid = false; - } - else -#endif - { - in_addr n; - if (inet_aton(host,&n) < 1) - ipvalid = false; - } - if (!ipvalid) - { - this->Instance->Log(DEBUG,"BUG: Hostname passed to InspSocket, rather than an IP address!"); - this->OnError(I_ERR_CONNECT); - this->Close(); - this->fd = -1; - this->state = I_ERROR; - return; - } - else + bind.sa.sa_family = 0; + if (!connectbindip.empty()) + { + if (!irc::sockets::aptosa(connectbindip, 0, &bind)) { - strlcpy(this->IP,host,MAXBUF); - timeout_val = maxtime; - if (!this->DoConnect()) - { - this->OnError(I_ERR_CONNECT); - this->Close(); - this->fd = -1; - this->state = I_ERROR; - return; - } + return I_ERR_BIND; } } -} -void InspSocket::WantWrite() -{ - this->Instance->SE->WantWrite(this); - this->WaitingForWriteEvent = true; + return BeginConnect(addr, bind, maxtime); } -void InspSocket::SetQueues(int nfd) +static void IncreaseOSBuffers(int fd) { // attempt to increase socket sendq and recvq as high as its possible int sendbuf = 32768; int recvbuf = 32768; - if(setsockopt(nfd,SOL_SOCKET,SO_SNDBUF,(const char *)&sendbuf,sizeof(sendbuf)) || setsockopt(nfd,SOL_SOCKET,SO_RCVBUF,(const char *)&recvbuf,sizeof(sendbuf))) - this->Instance->Log(DEFAULT, "Could not increase SO_SNDBUF/SO_RCVBUF for socket %u", GetFd()); + setsockopt(fd,SOL_SOCKET,SO_SNDBUF,(const char *)&sendbuf,sizeof(sendbuf)); + setsockopt(fd,SOL_SOCKET,SO_RCVBUF,(const char *)&recvbuf,sizeof(recvbuf)); + // on failure, do nothing. I'm a little sick of people trying to interpret this message as a result of why their incorrect setups don't work. } -/* Most irc servers require you to specify the ip you want to bind to. - * If you dont specify an IP, they rather dumbly bind to the first IP - * of the box (e.g. INADDR_ANY). In InspIRCd, we scan thought the IP - * addresses we've bound server ports to, and we try and bind our outbound - * connections to the first usable non-loopback and non-any IP we find. - * This is easier to configure when you have a lot of links and a lot - * of servers to configure. - */ -bool InspSocket::BindAddr(const std::string &ip) +BufferedSocketError BufferedSocket::BeginConnect(const irc::sockets::sockaddrs& dest, const irc::sockets::sockaddrs& bind, unsigned long timeout) { - ConfigReader Conf(this->Instance); - socklen_t size = sizeof(sockaddr_in); -#ifdef IPV6 - bool v6 = false; - /* Are we looking for a binding to fit an ipv6 host? */ - if ((ip.empty()) || (ip.find(':') != std::string::npos)) - v6 = true; -#endif - int j = 0; - while (j < Conf.Enumerate("bind") || (!ip.empty())) - { - std::string IP = ip.empty() ? Conf.ReadValue("bind","address",j) : ip; - if (!ip.empty() || Conf.ReadValue("bind","type",j) == "servers") - { - if (!ip.empty() || ((IP != "*") && (IP != "127.0.0.1") && (!IP.empty()) && (IP != "::1"))) - { - sockaddr* s = new sockaddr[2]; -#ifdef IPV6 - if (v6) - { - in6_addr n; - if (inet_pton(AF_INET6, IP.c_str(), &n) > 0) - { - memcpy(&((sockaddr_in6*)s)->sin6_addr, &n, sizeof(sockaddr_in6)); - ((sockaddr_in6*)s)->sin6_port = 0; - ((sockaddr_in6*)s)->sin6_family = AF_INET6; - size = sizeof(sockaddr_in6); - } - else - { - delete[] s; - j++; - continue; - } - } - else -#endif - { - in_addr n; - if (inet_aton(IP.c_str(), &n) > 0) - { - ((sockaddr_in*)s)->sin_addr = n; - ((sockaddr_in*)s)->sin_port = 0; - ((sockaddr_in*)s)->sin_family = AF_INET; - } - else - { - delete[] s; - j++; - continue; - } - } + if (fd < 0) + fd = socket(dest.sa.sa_family, SOCK_STREAM, 0); - if (bind(this->fd, s, size) < 0) - { - this->state = I_ERROR; - this->OnError(I_ERR_BIND); - this->fd = -1; - delete[] s; - return false; - } - - delete[] s; - return true; - } - } - j++; - } - return true; -} + if (fd < 0) + return I_ERR_SOCKET; -bool InspSocket::DoConnect() -{ - sockaddr* addr = new sockaddr[2]; - socklen_t size = sizeof(sockaddr_in); -#ifdef IPV6 - bool v6 = false; - if ((!*this->host) || strchr(this->host, ':')) - v6 = true; - - if (v6) - { - this->fd = socket(AF_INET6, SOCK_STREAM, 0); - if ((this->fd > -1) && ((strstr(this->IP,"::ffff:") != (char*)&this->IP) && (strstr(this->IP,"::FFFF:") != (char*)&this->IP))) - { - if (!this->BindAddr(this->cbindip)) - { - delete[] addr; - return false; - } - } - } - else -#endif + if (bind.sa.sa_family != 0) { - this->fd = socket(AF_INET, SOCK_STREAM, 0); - if (this->fd > -1) - { - if (!this->BindAddr(this->cbindip)) - { - delete[] addr; - return false; - } - } + if (ServerInstance->SE->Bind(fd, &bind.sa, sa_size(bind)) < 0) + return I_ERR_BIND; } - if (this->fd == -1) - { - this->state = I_ERROR; - this->OnError(I_ERR_SOCKET); - delete[] addr; - return false; - } + ServerInstance->SE->NonBlocking(fd); -#ifdef IPV6 - if (v6) - { - in6_addr addy; - if (inet_pton(AF_INET6, this->host, &addy) > 0) - { - ((sockaddr_in6*)addr)->sin6_family = AF_INET6; - memcpy(&((sockaddr_in6*)addr)->sin6_addr, &addy, sizeof(addy)); - ((sockaddr_in6*)addr)->sin6_port = htons(this->port); - size = sizeof(sockaddr_in6); - } - } - else -#endif - { - in_addr addy; - if (inet_aton(this->host, &addy) > 0) - { - ((sockaddr_in*)addr)->sin_family = AF_INET; - ((sockaddr_in*)addr)->sin_addr = addy; - ((sockaddr_in*)addr)->sin_port = htons(this->port); - } - } -#ifndef WIN32 - int flags = fcntl(this->fd, F_GETFL, 0); - fcntl(this->fd, F_SETFL, flags | O_NONBLOCK); -#else - unsigned long flags = 0; - ioctlsocket(this->fd, FIONBIO, &flags); -#endif - if (connect(this->fd, (sockaddr*)addr, size) == -1) + if (ServerInstance->SE->Connect(this, &dest.sa, sa_size(dest)) == -1) { if (errno != EINPROGRESS) - { - this->OnError(I_ERR_CONNECT); - this->Close(); - this->state = I_ERROR; - return false; - } - - this->Timeout = new SocketTimeout(this->GetFd(), this->Instance, this, timeout_val, this->Instance->Time()); - this->Instance->Timers->AddTimer(this->Timeout); + return I_ERR_CONNECT; } -#ifdef WIN32 - /* Set nonblocking mode after the connect() call */ - flags = 0; - ioctlsocket(this->fd, FIONBIO, &flags); -#endif + this->state = I_CONNECTING; - if (this->fd > -1) - { - if (!this->Instance->SE->AddFd(this)) - { - this->OnError(I_ERR_NOMOREFDS); - this->Close(); - this->state = I_ERROR; - return false; - } - this->SetQueues(this->fd); - } - return true; -} + if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_READ | FD_WANT_SINGLE_WRITE)) + return I_ERR_NOMOREFDS; + + this->Timeout = new SocketTimeout(this->GetFd(), this, timeout, ServerInstance->Time()); + ServerInstance->Timers->AddTimer(this->Timeout); + + IncreaseOSBuffers(fd); -void InspSocket::Close() + ServerInstance->Logs->Log("SOCKET", DEBUG,"BufferedSocket::DoConnect success"); + return I_ERR_NONE; +} + +void StreamSocket::Close() { /* Save this, so we dont lose it, * otherise on failure, error messages @@ -350,192 +125,326 @@ void InspSocket::Close() int save = errno; if (this->fd > -1) { - if (this->IsIOHooked && Instance->Config->GetIOHook(this)) + if (IOHook) { try { - Instance->Config->GetIOHook(this)->OnRawSocketClose(this->fd); + IOHook->OnStreamSocketClose(this); } catch (CoreException& modexcept) { - Instance->Log(DEFAULT,"%s threw an exception: %s", modexcept.GetSource(), modexcept.GetReason()); + ServerInstance->Logs->Log("SOCKET", DEFAULT,"%s threw an exception: %s", + modexcept.GetSource(), modexcept.GetReason()); } } - shutdown(this->fd,2); - if (close(this->fd) != -1) - this->OnClose(); - - if (Instance->SocketCull.find(this) == Instance->SocketCull.end()) - Instance->SocketCull[this] = this; + ServerInstance->SE->Shutdown(this, 2); + ServerInstance->SE->DelFd(this); + ServerInstance->SE->Close(this); + fd = -1; } errno = save; } -std::string InspSocket::GetIP() +CullResult StreamSocket::cull() { - return this->IP; + Close(); + return EventHandler::cull(); } -char* InspSocket::Read() +bool StreamSocket::GetNextLine(std::string& line, char delim) { - if (!Instance->SE->BoundsCheckFd(this)) - return NULL; - - int n = 0; + std::string::size_type i = recvq.find(delim); + if (i == std::string::npos) + return false; + line = recvq.substr(0, i - 1); + // TODO is this the most efficient way to split? + recvq = recvq.substr(i + 1); + return true; +} - if (this->IsIOHooked) +void StreamSocket::DoRead() +{ + if (IOHook) { - int result2 = 0; - int MOD_RESULT = 0; + int rv = -1; try { - MOD_RESULT = Instance->Config->GetIOHook(this)->OnRawSocketRead(this->fd,this->ibuf,sizeof(this->ibuf),result2); + rv = IOHook->OnStreamSocketRead(this, recvq); } catch (CoreException& modexcept) { - Instance->Log(DEFAULT,"%s threw an exception: %s", modexcept.GetSource(), modexcept.GetReason()); + ServerInstance->Logs->Log("SOCKET", DEFAULT, "%s threw an exception: %s", + modexcept.GetSource(), modexcept.GetReason()); + return; } - if (MOD_RESULT < 0) + if (rv > 0) + OnDataReady(); + if (rv < 0) + SetError("Read Error"); // will not overwrite a better error message + } + else + { + char* ReadBuffer = ServerInstance->GetReadBuffer(); + int n = recv(fd, ReadBuffer, ServerInstance->Config->NetBufferSize, 0); + if (n == ServerInstance->Config->NetBufferSize) { - n = -1; - errno = EAGAIN; + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ); + recvq.append(ReadBuffer, n); + OnDataReady(); } - else + else if (n > 0) { - n = result2; + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ); + recvq.append(ReadBuffer, n); + OnDataReady(); + } + else if (n == 0) + { + error = "Connection closed"; + ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); + } + else if (errno == EAGAIN) + { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_READ_WILL_BLOCK); + } + else if (errno == EINTR) + { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ); } - } - else - { - n = recv(this->fd,this->ibuf,sizeof(this->ibuf),0); - } - - if ((n > 0) && (n <= (int)sizeof(this->ibuf))) - { - ibuf[n] = 0; - return ibuf; - } - else - { - int err = errno; - if (err == EAGAIN) - return ""; else - return NULL; + { + error = strerror(errno); + ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); + } } } -void InspSocket::MarkAsClosed() +void StreamSocket::DoWrite() { -} - -// There are two possible outcomes to this function. -// It will either write all of the data, or an undefined amount. -// If an undefined amount is written the connection has failed -// and should be aborted. -int InspSocket::Write(const std::string &data) -{ - /* Try and append the data to the back of the queue, and send it on its way - */ - outbuffer.push_back(data); - this->Instance->SE->WantWrite(this); - return (!this->FlushWriteBuffer()); -} + if (sendq.empty()) + return; + if (!error.empty() || fd < 0 || fd == INT_MAX) + { + ServerInstance->Logs->Log("SOCKET", DEBUG, "DoWrite on errored or closed socket"); + return; + } -bool InspSocket::FlushWriteBuffer() -{ - errno = 0; - if ((this->fd > -1) && (this->state == I_CONNECTED)) +#ifndef DISABLE_WRITEV + if (IOHook) +#endif { - if (this->IsIOHooked) + int rv = -1; + try { - while (outbuffer.size() && (errno != EAGAIN)) + while (!sendq.empty()) { - try + if (sendq.size() > 1 && sendq[0].length() < 1024) + { + // Avoid multiple repeated SSL encryption invocations + // This adds a single copy of the queue, but avoids + // much more overhead in terms of system calls invoked + // by the IOHook. + // + // The length limit of 1024 is to prevent merging strings + // more than once when writes begin to block. + std::string tmp; + tmp.reserve(sendq_len); + for(unsigned int i=0; i < sendq.size(); i++) + tmp.append(sendq[i]); + sendq.clear(); + sendq.push_back(tmp); + } + std::string& front = sendq.front(); + int itemlen = front.length(); + if (IOHook) { - /* XXX: The lack of buffering here is NOT a bug, modules implementing this interface have to - * implement their own buffering mechanisms - */ - Instance->Config->GetIOHook(this)->OnRawSocketWrite(this->fd, outbuffer[0].c_str(), outbuffer[0].length()); - outbuffer.pop_front(); + rv = IOHook->OnStreamSocketWrite(this, front); + if (rv > 0) + { + // consumed the entire string, and is ready for more + sendq_len -= itemlen; + sendq.pop_front(); + } + else if (rv == 0) + { + // socket has blocked. Stop trying to send data. + // IOHook has requested unblock notification from the socketengine + + // Since it is possible that a partial write took place, adjust sendq_len + sendq_len = sendq_len - itemlen + front.length(); + return; + } + else + { + SetError("Write Error"); // will not overwrite a better error message + return; + } } - catch (CoreException& modexcept) +#ifdef DISABLE_WRITEV + else { - Instance->Log(DEBUG,"%s threw an exception: %s", modexcept.GetSource(), modexcept.GetReason()); - return true; + rv = ServerInstance->SE->Send(this, front.data(), itemlen, 0); + if (rv == 0) + { + SetError("Connection closed"); + return; + } + else if (rv < 0) + { + if (errno == EAGAIN || errno == EINTR) + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK); + else + SetError(strerror(errno)); + return; + } + else if (rv < itemlen) + { + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK); + front = front.substr(itemlen - rv); + sendq_len -= rv; + return; + } + else + { + sendq_len -= itemlen; + sendq.pop_front(); + if (sendq.empty()) + ServerInstance->SE->ChangeEventMask(this, FD_WANT_EDGE_WRITE); + } } +#endif } } - else + catch (CoreException& modexcept) + { + ServerInstance->Logs->Log("SOCKET", DEBUG,"%s threw an exception: %s", + modexcept.GetSource(), modexcept.GetReason()); + } + } +#ifndef DISABLE_WRITEV + else + { + // don't even try if we are known to be blocking + if (GetEventMask() & FD_WRITE_WILL_BLOCK) + return; + // start out optimistic - we won't need to write any more + int eventChange = FD_WANT_EDGE_WRITE; + while (sendq_len && eventChange == FD_WANT_EDGE_WRITE) { - /* If we have multiple lines, try to send them all, - * not just the first one -- Brain - */ - while (outbuffer.size() && (errno != EAGAIN)) + // Prepare a writev() call to write all buffers efficiently + int bufcount = sendq.size(); + + // cap the number of buffers at IOV_MAX + if (bufcount > IOV_MAX) { - /* Send a line */ -#ifndef WIN32 - int result = write(this->fd,outbuffer[0].c_str(),outbuffer[0].length()); -#else - int result = send(this->fd,outbuffer[0].c_str(),outbuffer[0].length(), 0); -#endif - if (result > 0) + bufcount = IOV_MAX; + } + + int rv_max = 0; + iovec* iovecs = new iovec[bufcount]; + for(int i=0; i < bufcount; i++) + { + iovecs[i].iov_base = const_cast(sendq[i].data()); + iovecs[i].iov_len = sendq[i].length(); + rv_max += sendq[i].length(); + } + int rv = writev(fd, iovecs, bufcount); + delete[] iovecs; + + if (rv == (int)sendq_len) + { + // it's our lucky day, everything got written out. Fast cleanup. + // This won't ever happen if the number of buffers got capped. + sendq_len = 0; + sendq.clear(); + } + else if (rv > 0) + { + // Partial write. Clean out strings from the sendq + if (rv < rv_max) { - if ((unsigned int)result >= outbuffer[0].length()) + // it's going to block now + eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK; + } + sendq_len -= rv; + while (rv > 0 && !sendq.empty()) + { + std::string& front = sendq.front(); + if (front.length() <= (size_t)rv) { - /* The whole block was written (usually a line) - * Pop the block off the front of the queue, - * dont set errno, because we are clear of errors - * and want to try and write the next block too. - */ - outbuffer.pop_front(); + // this string got fully written out + rv -= front.length(); + sendq.pop_front(); } else { - std::string temp = outbuffer[0].substr(result); - outbuffer[0] = temp; - /* We didnt get the whole line out. arses. - * Try again next time, i guess. Set errno, - * because we shouldnt be writing any more now, - * until the socketengine says its safe to do so. - */ - errno = EAGAIN; + // stopped in the middle of this string + front = front.substr(rv); + rv = 0; } } - else if ((result == -1) && (errno != EAGAIN)) - { - this->OnError(I_ERR_WRITE); - this->state = I_ERROR; - this->Instance->SE->DelFd(this); - this->Close(); - return true; - } + } + else if (rv == 0) + { + error = "Connection closed"; + } + else if (errno == EAGAIN) + { + eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK; + } + else if (errno == EINTR) + { + // restart interrupted syscall + } + else + { + error = strerror(errno); } } + if (!error.empty()) + { + // error - kill all events + ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); + } + else + { + ServerInstance->SE->ChangeEventMask(this, eventChange); + } } +#endif +} - if ((errno == EAGAIN) && (fd > -1)) +void StreamSocket::WriteData(const std::string &data) +{ + if (fd < 0) { - this->Instance->SE->WantWrite(this); + ServerInstance->Logs->Log("SOCKET", DEBUG, "Attempt to write data to dead socket: %s", + data.c_str()); + return; } - return (fd < 0); + /* Append the data to the back of the queue ready for writing */ + sendq.push_back(data); + sendq_len += data.length(); + + ServerInstance->SE->ChangeEventMask(this, FD_ADD_TRIAL_WRITE); } -void SocketTimeout::Tick(time_t now) +void SocketTimeout::Tick(time_t) { + ServerInstance->Logs->Log("SOCKET", DEBUG,"SocketTimeout::Tick"); + if (ServerInstance->SE->GetRef(this->sfd) != this->sock) return; if (this->sock->state == I_CONNECTING) { - // for non-listening sockets, the timeout can occur + // for connecting sockets, the timeout can occur // which causes termination of the connection after // the given number of seconds without a successful // connection. this->sock->OnTimeout(); this->sock->OnError(I_ERR_TIMEOUT); - this->sock->timeout = true; /* NOTE: We must set this AFTER DelFd, as we added * this socket whilst writeable. This means that we @@ -543,206 +452,87 @@ void SocketTimeout::Tick(time_t now) */ this->sock->state = I_ERROR; - if (ServerInstance->SocketCull.find(this->sock) == ServerInstance->SocketCull.end()) - ServerInstance->SocketCull[this->sock] = this->sock; + ServerInstance->GlobalCulls.AddItem(sock); } this->sock->Timeout = NULL; } -bool InspSocket::Poll() -{ - int incoming = -1; +void BufferedSocket::OnConnected() { } +void BufferedSocket::OnTimeout() { return; } -#ifndef WINDOWS - if (!Instance->SE->BoundsCheckFd(this)) - return false; -#endif - - if (Instance->SE->GetRef(this->fd) != this) - return false; - - switch (this->state) +void BufferedSocket::DoWrite() +{ + if (state == I_CONNECTING) { - case I_CONNECTING: - /* Our socket was in write-state, so delete it and re-add it - * in read-state. - */ -#ifndef WINDOWS - if (this->fd > -1) - { - this->Instance->SE->DelFd(this); - if (!this->Instance->SE->AddFd(this)) - return false; - } -#endif - this->SetState(I_CONNECTED); - - if (Instance->Config->GetIOHook(this)) - { - Instance->Log(DEBUG,"Hook for raw connect"); - try - { - Instance->Config->GetIOHook(this)->OnRawSocketConnect(this->fd); - } - catch (CoreException& modexcept) - { - Instance->Log(DEBUG,"%s threw an exception: %s", modexcept.GetSource(), modexcept.GetReason()); - } - } - return this->OnConnected(); - break; - case I_LISTENING: - { - sockaddr* client = new sockaddr[2]; - length = sizeof (sockaddr_in); - std::string recvip; -#ifdef IPV6 - if ((!*this->host) || strchr(this->host, ':')) - length = sizeof(sockaddr_in6); -#endif - void* m_acceptEvent = NULL; - GetExt("windows_acceptevent", m_acceptEvent); - incoming = _accept (this->fd, client, &length); -#ifdef IPV6 - if ((!*this->host) || strchr(this->host, ':')) - { - char buf[1024]; - recvip = inet_ntop(AF_INET6, &((sockaddr_in6*)client)->sin6_addr, buf, sizeof(buf)); - } - else -#endif - recvip = inet_ntoa(((sockaddr_in*)client)->sin_addr); - this->OnIncomingConnection(incoming, (char*)recvip.c_str()); - - if (this->IsIOHooked) - { - try - { - Instance->Config->GetIOHook(this)->OnRawSocketAccept(incoming, recvip.c_str(), this->port); - } - catch (CoreException& modexcept) - { - Instance->Log(DEBUG,"%s threw an exception: %s", modexcept.GetSource(), modexcept.GetReason()); - } - } - - this->SetQueues(incoming); - - delete[] client; - return true; - } - break; - case I_CONNECTED: - /* Process the read event */ - return this->OnDataReady(); - break; - default: - break; + state = I_CONNECTED; + this->OnConnected(); + if (GetIOHook()) + GetIOHook()->OnStreamSocketConnect(this); + else + ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE); } - return true; -} - -void InspSocket::SetState(InspSocketState s) -{ - this->state = s; -} - -InspSocketState InspSocket::GetState() -{ - return this->state; -} - -int InspSocket::GetFd() -{ - return this->fd; + this->StreamSocket::DoWrite(); } -bool InspSocket::OnConnected() { return true; } -void InspSocket::OnError(InspSocketError e) { return; } -int InspSocket::OnDisconnect() { return 0; } -int InspSocket::OnIncomingConnection(int newfd, char* ip) { return 0; } -bool InspSocket::OnDataReady() { return true; } -bool InspSocket::OnWriteReady() { return true; } -void InspSocket::OnTimeout() { return; } -void InspSocket::OnClose() { return; } - -InspSocket::~InspSocket() +BufferedSocket::~BufferedSocket() { this->Close(); if (Timeout) { - Instance->Timers->DelTimer(Timeout); + ServerInstance->Timers->DelTimer(Timeout); Timeout = NULL; } } -void InspSocket::HandleEvent(EventType et, int errornum) +void StreamSocket::HandleEvent(EventType et, int errornum) { + if (!error.empty()) + return; + BufferedSocketError errcode = I_ERR_OTHER; switch (et) { case EVENT_ERROR: + { + if (errornum == 0) + SetError("Connection closed"); + else + SetError(strerror(errornum)); switch (errornum) { case ETIMEDOUT: - this->OnError(I_ERR_TIMEOUT); - break; + errcode = I_ERR_TIMEOUT; + break; case ECONNREFUSED: case 0: - this->OnError(this->state == I_CONNECTING ? I_ERR_CONNECT : I_ERR_WRITE); - break; + errcode = I_ERR_CONNECT; + break; case EADDRINUSE: - this->OnError(I_ERR_BIND); - break; + errcode = I_ERR_BIND; + break; case EPIPE: case EIO: - this->OnError(I_ERR_WRITE); - break; + errcode = I_ERR_WRITE; + break; } - if (this->Instance->SocketCull.find(this) == this->Instance->SocketCull.end()) - this->Instance->SocketCull[this] = this; - return; - break; + break; + } case EVENT_READ: - if (!this->Poll()) - { - if (this->Instance->SocketCull.find(this) == this->Instance->SocketCull.end()) - this->Instance->SocketCull[this] = this; - return; - } - break; + { + DoRead(); + break; + } case EVENT_WRITE: - if (this->WaitingForWriteEvent) - { - this->WaitingForWriteEvent = false; - if (!this->OnWriteReady()) - { - if (this->Instance->SocketCull.find(this) == this->Instance->SocketCull.end()) - this->Instance->SocketCull[this] = this; - return; - } - } - if (this->state == I_CONNECTING) - { - /* This might look wrong as if we should be actually calling - * with EVENT_WRITE, but trust me it is correct. There are some - * writeability-state things in the read code, because of how - * InspSocket used to work regarding write buffering in previous - * versions of InspIRCd. - Brain - */ - this->HandleEvent(EVENT_READ); - return; - } - else - { - if (this->FlushWriteBuffer()) - { - if (this->Instance->SocketCull.find(this) == this->Instance->SocketCull.end()) - this->Instance->SocketCull[this] = this; - return; - } - } - break; + { + DoWrite(); + break; + } + } + if (!error.empty()) + { + ServerInstance->Logs->Log("SOCKET", DEBUG, "Error on FD %d - '%s'", fd, error.c_str()); + OnError(errcode); + ServerInstance->GlobalCulls.AddItem(this); } }