/* * InspIRCd -- Internet Relay Chat Daemon * * Copyright (C) 2020 Matt Schatz * Copyright (C) 2019 linuxdaemon * Copyright (C) 2018 Dylan Frank * Copyright (C) 2013-2016 Attila Molnar * Copyright (C) 2013, 2017-2020 Sadie Powell * Copyright (C) 2013 Adam * Copyright (C) 2012 Robby * Copyright (C) 2009-2010 Daniel De Graaf * Copyright (C) 2007-2008 Robin Burchell * Copyright (C) 2007 John Brooks * Copyright (C) 2007 Dennis Friis * Copyright (C) 2006-2007 Craig Edwards * Copyright (C) 2006 Oliver Lupton * * This file is part of InspIRCd. InspIRCd is free software: you can * redistribute it and/or modify it under the terms of the GNU General Public * License as published by the Free Software Foundation, version 2. * * This program is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more * details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ #include "inspircd.h" #include "iohook.h" static IOHook* GetNextHook(IOHook* hook) { IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(hook); if (iohm) return iohm->GetNextHook(); return NULL; } BufferedSocket::BufferedSocket() { Timeout = NULL; state = I_ERROR; } BufferedSocket::BufferedSocket(int newfd) { Timeout = NULL; this->fd = newfd; this->state = I_CONNECTED; if (HasFd()) SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE); } void BufferedSocket::DoConnect(const irc::sockets::sockaddrs& dest, const irc::sockets::sockaddrs& bind, unsigned int maxtime) { BufferedSocketError err = BeginConnect(dest, bind, maxtime); if (err != I_ERR_NONE) { state = I_ERROR; SetError(SocketEngine::LastError()); OnError(err); } } BufferedSocketError BufferedSocket::BeginConnect(const irc::sockets::sockaddrs& dest, const irc::sockets::sockaddrs& bind, unsigned int timeout) { if (!HasFd()) fd = socket(dest.family(), SOCK_STREAM, 0); if (!HasFd()) return I_ERR_SOCKET; if (bind.family() != 0) { if (SocketEngine::Bind(fd, bind) < 0) return I_ERR_BIND; } SocketEngine::NonBlocking(fd); if (SocketEngine::Connect(this, dest) == -1) { if (errno != EINPROGRESS) return I_ERR_CONNECT; } this->state = I_CONNECTING; if (!SocketEngine::AddFd(this, FD_WANT_NO_READ | FD_WANT_SINGLE_WRITE | FD_WRITE_WILL_BLOCK)) return I_ERR_NOMOREFDS; this->Timeout = new SocketTimeout(this->GetFd(), this, timeout); ServerInstance->Timers.AddTimer(this->Timeout); ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "BufferedSocket::DoConnect success"); return I_ERR_NONE; } void StreamSocket::Close() { if (closing) return; closing = true; if (HasFd()) { // final chance, dump as much of the sendq as we can DoWrite(); IOHook* hook = GetIOHook(); DelIOHook(); while (hook) { hook->OnStreamSocketClose(this); IOHook* const nexthook = GetNextHook(hook); delete hook; hook = nexthook; } SocketEngine::Shutdown(this, 2); SocketEngine::Close(this); } } void StreamSocket::Close(bool writeblock) { if (getSendQSize() != 0 && writeblock) closeonempty = true; else Close(); } CullResult StreamSocket::cull() { Close(); return EventHandler::cull(); } bool StreamSocket::GetNextLine(std::string& line, char delim) { std::string::size_type i = recvq.find(delim); if (i == std::string::npos) return false; line.assign(recvq, 0, i); recvq.erase(0, i + 1); return true; } int StreamSocket::HookChainRead(IOHook* hook, std::string& rq) { if (!hook) return ReadToRecvQ(rq); IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(hook); if (iohm) { // Call the next hook to put data into the recvq of the current hook const int ret = HookChainRead(iohm->GetNextHook(), iohm->GetRecvQ()); if (ret <= 0) return ret; } return hook->OnStreamSocketRead(this, rq); } void StreamSocket::DoRead() { const std::string::size_type prevrecvqsize = recvq.size(); const int result = HookChainRead(GetIOHook(), recvq); if (result < 0) { SetError("Read Error"); // will not overwrite a better error message return; } if (recvq.size() > prevrecvqsize) OnDataReady(); } int StreamSocket::ReadToRecvQ(std::string& rq) { char* ReadBuffer = ServerInstance->GetReadBuffer(); int n = SocketEngine::Recv(this, ReadBuffer, ServerInstance->Config->NetBufferSize, 0); if (n == ServerInstance->Config->NetBufferSize) { SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ); rq.append(ReadBuffer, n); } else if (n > 0) { SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ); rq.append(ReadBuffer, n); } else if (n == 0) { error = "Connection closed"; SocketEngine::ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); return -1; } else if (SocketEngine::IgnoreError()) { SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_READ_WILL_BLOCK); return 0; } else if (errno == EINTR) { SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ); return 0; } else { error = SocketEngine::LastError(); SocketEngine::ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); return -1; } return n; } /* Don't try to prepare huge blobs of data to send to a blocked socket */ static const int MYIOV_MAX = IOV_MAX < 128 ? IOV_MAX : 128; void StreamSocket::DoWrite() { if (getSendQSize() == 0) { if (closeonempty) Close(); return; } if (!error.empty() || !HasFd()) { ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "DoWrite on errored or closed socket"); return; } SendQueue* psendq = &sendq; IOHook* hook = GetIOHook(); while (hook) { int rv = hook->OnStreamSocketWrite(this, *psendq); psendq = NULL; // rv == 0 means the socket has blocked. Stop trying to send data. // IOHook has requested unblock notification from the socketengine. if (rv == 0) break; if (rv < 0) { SetError("Write Error"); // will not overwrite a better error message break; } IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(hook); hook = NULL; if (iohm) { psendq = &iohm->GetSendQ(); hook = iohm->GetNextHook(); } } if (psendq) FlushSendQ(*psendq); if (getSendQSize() == 0 && closeonempty) Close(); } void StreamSocket::FlushSendQ(SendQueue& sq) { // 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 (error.empty() && !sq.empty() && eventChange == FD_WANT_EDGE_WRITE) { // Prepare a writev() call to write all buffers efficiently int bufcount = sq.size(); // cap the number of buffers at MYIOV_MAX if (bufcount > MYIOV_MAX) { bufcount = MYIOV_MAX; } int rv_max = 0; int rv; { SocketEngine::IOVector iovecs[MYIOV_MAX]; size_t j = 0; for (SendQueue::const_iterator i = sq.begin(), end = i+bufcount; i != end; ++i, j++) { const SendQueue::Element& elem = *i; iovecs[j].iov_base = const_cast(elem.data()); iovecs[j].iov_len = elem.length(); rv_max += iovecs[j].iov_len; } rv = SocketEngine::WriteV(this, iovecs, bufcount); } if (rv == (int)sq.bytes()) { // it's our lucky day, everything got written out. Fast cleanup. // This won't ever happen if the number of buffers got capped. sq.clear(); } else if (rv > 0) { // Partial write. Clean out strings from the sendq if (rv < rv_max) { // it's going to block now eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK; } while (rv > 0 && !sq.empty()) { const SendQueue::Element& front = sq.front(); if (front.length() <= (size_t)rv) { // this string got fully written out rv -= front.length(); sq.pop_front(); } else { // stopped in the middle of this string sq.erase_front(rv); rv = 0; } } } else if (rv == 0) { error = "Connection closed"; } else if (SocketEngine::IgnoreError()) { eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK; } else if (errno == EINTR) { // restart interrupted syscall errno = 0; } else { error = SocketEngine::LastError(); } } if (!error.empty()) { // error - kill all events SocketEngine::ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE); } else { SocketEngine::ChangeEventMask(this, eventChange); } } bool StreamSocket::OnSetEndPoint(const irc::sockets::sockaddrs& local, const irc::sockets::sockaddrs& remote) { return false; } void StreamSocket::WriteData(const std::string &data) { if (!HasFd()) { ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "Attempt to write data to dead socket: %s", data.c_str()); return; } /* Append the data to the back of the queue ready for writing */ sendq.push_back(data); SocketEngine::ChangeEventMask(this, FD_ADD_TRIAL_WRITE); } bool SocketTimeout::Tick(time_t) { ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "SocketTimeout::Tick"); if (SocketEngine::GetRef(this->sfd) != this->sock) { delete this; return false; } if (this->sock->state == I_CONNECTING) { // 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->state = I_ERROR; ServerInstance->GlobalCulls.AddItem(sock); } this->sock->Timeout = NULL; delete this; return false; } void BufferedSocket::OnConnected() { } void BufferedSocket::OnTimeout() { return; } void BufferedSocket::OnEventHandlerWrite() { if (state == I_CONNECTING) { state = I_CONNECTED; this->OnConnected(); if (!GetIOHook()) SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE); } this->StreamSocket::OnEventHandlerWrite(); } BufferedSocket::~BufferedSocket() { this->Close(); // The timer is removed from the TimerManager in Timer::~Timer() delete Timeout; } void StreamSocket::OnEventHandlerError(int errornum) { if (!error.empty()) return; if (errornum == 0) SetError("Connection closed"); else SetError(SocketEngine::GetError(errornum)); BufferedSocketError errcode = I_ERR_OTHER; switch (errornum) { case ETIMEDOUT: errcode = I_ERR_TIMEOUT; break; case ECONNREFUSED: case 0: errcode = I_ERR_CONNECT; break; case EADDRINUSE: errcode = I_ERR_BIND; break; case EPIPE: case EIO: errcode = I_ERR_WRITE; break; } // Log and call OnError() CheckError(errcode); } void StreamSocket::OnEventHandlerRead() { if (!error.empty()) return; try { DoRead(); } catch (CoreException& ex) { ServerInstance->Logs->Log("SOCKET", LOG_DEFAULT, "Caught exception in socket processing on FD %d - '%s'", fd, ex.GetReason().c_str()); SetError(ex.GetReason()); } CheckError(I_ERR_OTHER); } void StreamSocket::OnEventHandlerWrite() { if (!error.empty()) return; DoWrite(); CheckError(I_ERR_OTHER); } void StreamSocket::CheckError(BufferedSocketError errcode) { if (!error.empty()) { ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "Error on FD %d - '%s'", fd, error.c_str()); OnError(errcode); } } IOHook* StreamSocket::GetModHook(Module* mod) const { for (IOHook* curr = GetIOHook(); curr; curr = GetNextHook(curr)) { if (curr->prov->creator == mod) return curr; } return NULL; } IOHook* StreamSocket::GetLastHook() const { IOHook* curr = GetIOHook(); IOHook* last = curr; for (; curr; curr = GetNextHook(curr)) last = curr; return last; } void StreamSocket::AddIOHook(IOHook* newhook) { IOHook* curr = GetIOHook(); if (!curr) { iohook = newhook; return; } IOHookMiddle* lasthook; while (curr) { lasthook = IOHookMiddle::ToMiddleHook(curr); if (!lasthook) return; curr = lasthook->GetNextHook(); } lasthook->SetNextHook(newhook); } size_t StreamSocket::getSendQSize() const { size_t ret = sendq.bytes(); IOHook* curr = GetIOHook(); while (curr) { const IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(curr); if (!iohm) break; ret += iohm->GetSendQ().bytes(); curr = iohm->GetNextHook(); } return ret; } void StreamSocket::SwapInternals(StreamSocket& other) { if (type != other.type) return; EventHandler::SwapInternals(other); std::swap(closeonempty, other.closeonempty); std::swap(closing, other.closing); std::swap(error, other.error); std::swap(iohook, other.iohook); std::swap(recvq, other.recvq); std::swap(sendq, other.sendq); }