#include "inspircd.h"
#include "iohook.h"
-#ifndef DISABLE_WRITEV
-#include <sys/uio.h>
-#endif
-
-#ifndef IOV_MAX
-#define IOV_MAX 1024
-#endif
-
BufferedSocket::BufferedSocket()
{
Timeout = NULL;
DoWrite();
if (GetIOHook())
{
- try
- {
- GetIOHook()->OnStreamSocketClose(this);
- }
- catch (CoreException& modexcept)
- {
- ServerInstance->Logs->Log("SOCKET", LOG_DEFAULT, "%s threw an exception: %s",
- modexcept.GetSource().c_str(), modexcept.GetReason().c_str());
- }
+ GetIOHook()->OnStreamSocketClose(this);
delete iohook;
DelIOHook();
}
{
if (GetIOHook())
{
- int rv = -1;
- try
- {
- rv = GetIOHook()->OnStreamSocketRead(this, recvq);
- }
- catch (CoreException& modexcept)
- {
- ServerInstance->Logs->Log("SOCKET", LOG_DEFAULT, "%s threw an exception: %s",
- modexcept.GetSource().c_str(), modexcept.GetReason().c_str());
- return;
- }
+ int rv = GetIOHook()->OnStreamSocketRead(this, recvq);
if (rv > 0)
OnDataReady();
if (rv < 0)
{
if (sendq.empty())
return;
- if (!error.empty() || fd < 0 || fd == INT_MAX)
+ if (!error.empty() || fd < 0)
{
ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "DoWrite on errored or closed socket");
return;
}
-#ifndef DISABLE_WRITEV
if (GetIOHook())
-#endif
{
- int rv = -1;
- try
- {
- while (error.empty() && !sendq.empty())
- {
- 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(1280);
- while (!sendq.empty() && tmp.length() < 1024)
- {
- tmp.append(sendq.front());
- sendq.pop_front();
- }
- sendq.push_front(tmp);
- }
- std::string& front = sendq.front();
- int itemlen = front.length();
- if (GetIOHook())
- {
- rv = GetIOHook()->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
+ int rv = GetIOHook()->OnStreamSocketWrite(this);
+ if (rv < 0)
+ SetError("Write Error"); // will not overwrite a better error message
- // 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;
- }
- }
-#ifdef DISABLE_WRITEV
- else
- {
- rv = SocketEngine::Send(this, front.data(), itemlen, 0);
- if (rv == 0)
- {
- SetError("Connection closed");
- return;
- }
- else if (rv < 0)
- {
- if (errno == EINTR || SocketEngine::IgnoreError())
- SocketEngine::ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK);
- else
- SetError(SocketEngine::LastError());
- return;
- }
- else if (rv < itemlen)
- {
- SocketEngine::ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK);
- front.erase(0, rv);
- sendq_len -= rv;
- return;
- }
- else
- {
- sendq_len -= itemlen;
- sendq.pop_front();
- if (sendq.empty())
- SocketEngine::ChangeEventMask(this, FD_WANT_EDGE_WRITE);
- }
- }
-#endif
- }
- }
- catch (CoreException& modexcept)
- {
- ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "%s threw an exception: %s",
- modexcept.GetSource().c_str(), modexcept.GetReason().c_str());
- }
+ // rv == 0 means the socket has blocked. Stop trying to send data.
+ // IOHook has requested unblock notification from the socketengine.
}
-#ifndef DISABLE_WRITEV
else
{
// don't even try if we are known to be blocking
return;
// start out optimistic - we won't need to write any more
int eventChange = FD_WANT_EDGE_WRITE;
- while (error.empty() && sendq_len && eventChange == FD_WANT_EDGE_WRITE)
+ while (error.empty() && !sendq.empty() && eventChange == FD_WANT_EDGE_WRITE)
{
// Prepare a writev() call to write all buffers efficiently
int bufcount = sendq.size();
}
int rv_max = 0;
- iovec* iovecs = new iovec[bufcount];
- for(int i=0; i < bufcount; i++)
+ int rv;
{
- iovecs[i].iov_base = const_cast<char*>(sendq[i].data());
- iovecs[i].iov_len = sendq[i].length();
- rv_max += sendq[i].length();
+ SocketEngine::IOVector iovecs[MYIOV_MAX];
+ size_t j = 0;
+ for (SendQueue::const_iterator i = sendq.begin(), end = i+bufcount; i != end; ++i, j++)
+ {
+ const SendQueue::Element& elem = *i;
+ iovecs[j].iov_base = const_cast<char*>(elem.data());
+ iovecs[j].iov_len = elem.length();
+ rv_max += elem.length();
+ }
+ rv = SocketEngine::WriteV(this, iovecs, bufcount);
}
- int rv = writev(fd, iovecs, bufcount);
- delete[] iovecs;
- if (rv == (int)sendq_len)
+ if (rv == (int)sendq.bytes())
{
// 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)
// 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();
+ const SendQueue::Element& front = sendq.front();
if (front.length() <= (size_t)rv)
{
// this string got fully written out
else
{
// stopped in the middle of this string
- front.erase(0, rv);
+ sendq.erase_front(rv);
rv = 0;
}
}
SocketEngine::ChangeEventMask(this, eventChange);
}
}
-#endif
}
void StreamSocket::WriteData(const std::string &data)
/* Append the data to the back of the queue ready for writing */
sendq.push_back(data);
- sendq_len += data.length();
SocketEngine::ChangeEventMask(this, FD_ADD_TRIAL_WRITE);
}
void BufferedSocket::OnConnected() { }
void BufferedSocket::OnTimeout() { return; }
-void BufferedSocket::DoWrite()
+void BufferedSocket::OnEventHandlerWrite()
{
if (state == I_CONNECTING)
{
if (!GetIOHook())
SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE);
}
- this->StreamSocket::DoWrite();
+ this->StreamSocket::OnEventHandlerWrite();
}
BufferedSocket::~BufferedSocket()
delete Timeout;
}
-void StreamSocket::HandleEvent(EventType et, int errornum)
+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;
- try {
- switch (et)
- {
- case EVENT_ERROR:
- {
- if (errornum == 0)
- SetError("Connection closed");
- else
- SetError(SocketEngine::GetError(errornum));
- 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;
- }
- break;
- }
- case EVENT_READ:
- {
- DoRead();
- break;
- }
- case EVENT_WRITE:
- {
- DoWrite();
- break;
- }
- }
+ 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());
+ 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());
projects / user / henk / code / inspircd.git / blobdiff