src/inspsocket.cpp

   1 /*
   2  * InspIRCd -- Internet Relay Chat Daemon
   3  *
   4  *   Copyright (C) 2009 Daniel De Graaf <danieldg@inspircd.org>
   5  *   Copyright (C) 2007-2009 Robin Burchell <robin+git@viroteck.net>
   6  *   Copyright (C) 2008 Thomas Stagner <aquanight@inspircd.org>
   7  *   Copyright (C) 2006-2007 Craig Edwards <craigedwards@brainbox.cc>
   8  *   Copyright (C) 2007 Dennis Friis <peavey@inspircd.org>
   9  *   Copyright (C) 2006 Oliver Lupton <oliverlupton@gmail.com>
  10  *
  11  * This file is part of InspIRCd.  InspIRCd is free software: you can
  12  * redistribute it and/or modify it under the terms of the GNU General Public
  13  * License as published by the Free Software Foundation, version 2.
  14  *
  15  * This program is distributed in the hope that it will be useful, but WITHOUT
  16  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  17  * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
  18  * details.
  19  *
  20  * You should have received a copy of the GNU General Public License
  21  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  22  */
  23
  24
  25 #include "inspircd.h"
  26 #include "socket.h"
  27 #include "inspstring.h"
  28 #include "socketengine.h"
  29
  30 #ifndef DISABLE_WRITEV
  31 #include <sys/uio.h>
  32 #endif
  33
  34 #ifndef IOV_MAX
  35 #define IOV_MAX 1024
  36 #endif
  37
  38 BufferedSocket::BufferedSocket()
  39 {
  40         Timeout = NULL;
  41         state = I_ERROR;
  42 }
  43
  44 BufferedSocket::BufferedSocket(int newfd)
  45 {
  46         Timeout = NULL;
  47         this->fd = newfd;
  48         this->state = I_CONNECTED;
  49         if (fd > -1)
  50                 ServerInstance->SE->AddFd(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE);
  51 }
  52
  53 void BufferedSocket::DoConnect(const std::string &ipaddr, int aport, unsigned long maxtime, const std::string &connectbindip)
  54 {
  55         BufferedSocketError err = BeginConnect(ipaddr, aport, maxtime, connectbindip);
  56         if (err != I_ERR_NONE)
  57         {
  58                 state = I_ERROR;
  59                 SetError(strerror(errno));
  60                 OnError(err);
  61         }
  62 }
  63
  64 BufferedSocketError BufferedSocket::BeginConnect(const std::string &ipaddr, int aport, unsigned long maxtime, const std::string &connectbindip)
  65 {
  66         irc::sockets::sockaddrs addr, bind;
  67         if (!irc::sockets::aptosa(ipaddr, aport, addr))
  68         {
  69                 ServerInstance->Logs->Log("SOCKET", DEBUG, "BUG: Hostname passed to BufferedSocket, rather than an IP address!");
  70                 return I_ERR_CONNECT;
  71         }
  72
  73         bind.sa.sa_family = 0;
  74         if (!connectbindip.empty())
  75         {
  76                 if (!irc::sockets::aptosa(connectbindip, 0, bind))
  77                 {
  78                         return I_ERR_BIND;
  79                 }
  80         }
  81
  82         return BeginConnect(addr, bind, maxtime);
  83 }
  84
  85 static void IncreaseOSBuffers(int fd)
  86 {
  87         // attempt to increase socket sendq and recvq as high as its possible
  88         int sendbuf = 32768;
  89         int recvbuf = 32768;
  90         setsockopt(fd,SOL_SOCKET,SO_SNDBUF,(const char *)&sendbuf,sizeof(sendbuf));
  91         setsockopt(fd,SOL_SOCKET,SO_RCVBUF,(const char *)&recvbuf,sizeof(recvbuf));
  92         // 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.
  93 }
  94
  95 BufferedSocketError BufferedSocket::BeginConnect(const irc::sockets::sockaddrs& dest, const irc::sockets::sockaddrs& bind, unsigned long timeout)
  96 {
  97         if (fd < 0)
  98                 fd = socket(dest.sa.sa_family, SOCK_STREAM, 0);
  99
 100         if (fd < 0)
 101                 return I_ERR_SOCKET;
 102
 103         if (bind.sa.sa_family != 0)
 104         {
 105                 if (ServerInstance->SE->Bind(fd, bind) < 0)
 106                         return I_ERR_BIND;
 107         }
 108
 109         ServerInstance->SE->NonBlocking(fd);
 110
 111         if (ServerInstance->SE->Connect(this, &dest.sa, sa_size(dest)) == -1)
 112         {
 113                 if (errno != EINPROGRESS)
 114                         return I_ERR_CONNECT;
 115         }
 116
 117         this->state = I_CONNECTING;
 118
 119         if (!ServerInstance->SE->AddFd(this, FD_WANT_NO_READ | FD_WANT_SINGLE_WRITE | FD_WRITE_WILL_BLOCK))
 120                 return I_ERR_NOMOREFDS;
 121
 122         this->Timeout = new SocketTimeout(this->GetFd(), this, timeout, ServerInstance->Time());
 123         ServerInstance->Timers->AddTimer(this->Timeout);
 124
 125         IncreaseOSBuffers(fd);
 126
 127         ServerInstance->Logs->Log("SOCKET", DEBUG,"BufferedSocket::DoConnect success");
 128         return I_ERR_NONE;
 129 }
 130
 131 void StreamSocket::Close()
 132 {
 133         if (this->fd > -1)
 134         {
 135                 // final chance, dump as much of the sendq as we can
 136                 DoWrite();
 137                 if (IOHook)
 138                 {
 139                         try
 140                         {
 141                                 IOHook->OnStreamSocketClose(this);
 142                         }
 143                         catch (CoreException& modexcept)
 144                         {
 145                                 ServerInstance->Logs->Log("SOCKET", DEFAULT,"%s threw an exception: %s",
 146                                         modexcept.GetSource(), modexcept.GetReason());
 147                         }
 148                         IOHook = NULL;
 149                 }
 150                 ServerInstance->SE->Shutdown(this, 2);
 151                 ServerInstance->SE->DelFd(this);
 152                 ServerInstance->SE->Close(this);
 153                 fd = -1;
 154         }
 155 }
 156
 157 CullResult StreamSocket::cull()
 158 {
 159         Close();
 160         return EventHandler::cull();
 161 }
 162
 163 bool StreamSocket::GetNextLine(std::string& line, char delim)
 164 {
 165         std::string::size_type i = recvq.find(delim);
 166         if (i == std::string::npos)
 167                 return false;
 168         line = recvq.substr(0, i);
 169         // TODO is this the most efficient way to split?
 170         recvq = recvq.substr(i + 1);
 171         return true;
 172 }
 173
 174 void StreamSocket::DoRead()
 175 {
 176         if (IOHook)
 177         {
 178                 int rv = -1;
 179                 try
 180                 {
 181                         rv = IOHook->OnStreamSocketRead(this, recvq);
 182                 }
 183                 catch (CoreException& modexcept)
 184                 {
 185                         ServerInstance->Logs->Log("SOCKET", DEFAULT, "%s threw an exception: %s",
 186                                 modexcept.GetSource(), modexcept.GetReason());
 187                         return;
 188                 }
 189                 if (rv > 0)
 190                         OnDataReady();
 191                 if (rv < 0)
 192                         SetError("Read Error"); // will not overwrite a better error message
 193         }
 194         else
 195         {
 196                 char* ReadBuffer = ServerInstance->GetReadBuffer();
 197                 int n = ServerInstance->SE->Recv(this, ReadBuffer, ServerInstance->Config->NetBufferSize, 0);
 198                 if (n == ServerInstance->Config->NetBufferSize)
 199                 {
 200                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ);
 201                         recvq.append(ReadBuffer, n);
 202                         OnDataReady();
 203                 }
 204                 else if (n > 0)
 205                 {
 206                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ);
 207                         recvq.append(ReadBuffer, n);
 208                         OnDataReady();
 209                 }
 210                 else if (n == 0)
 211                 {
 212                         error = "Connection closed";
 213                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE);
 214                 }
 215                 else if (errno == EAGAIN)
 216                 {
 217                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_READ_WILL_BLOCK);
 218                 }
 219                 else if (errno == EINTR)
 220                 {
 221                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ);
 222                 }
 223                 else
 224                 {
 225                         error = strerror(errno);
 226                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE);
 227                 }
 228         }
 229 }
 230
 231 /* Don't try to prepare huge blobs of data to send to a blocked socket */
 232 static const int MYIOV_MAX = IOV_MAX < 128 ? IOV_MAX : 128;
 233
 234 void StreamSocket::DoWrite()
 235 {
 236         if (sendq.empty())
 237                 return;
 238         if (!error.empty() || fd < 0 || fd == INT_MAX)
 239         {
 240                 ServerInstance->Logs->Log("SOCKET", DEBUG, "DoWrite on errored or closed socket");
 241                 return;
 242         }
 243
 244 #ifndef DISABLE_WRITEV
 245         if (IOHook)
 246 #endif
 247         {
 248                 int rv = -1;
 249                 try
 250                 {
 251                         while (error.empty() && !sendq.empty())
 252                         {
 253                                 if (sendq.size() > 1 && sendq[0].length() < 1024)
 254                                 {
 255                                         // Avoid multiple repeated SSL encryption invocations
 256                                         // This adds a single copy of the queue, but avoids
 257                                         // much more overhead in terms of system calls invoked
 258                                         // by the IOHook.
 259                                         //
 260                                         // The length limit of 1024 is to prevent merging strings
 261                                         // more than once when writes begin to block.
 262                                         std::string tmp;
 263                                         tmp.reserve(sendq_len);
 264                                         for(unsigned int i=0; i < sendq.size(); i++)
 265                                                 tmp.append(sendq[i]);
 266                                         sendq.clear();
 267                                         sendq.push_back(tmp);
 268                                 }
 269                                 std::string& front = sendq.front();
 270                                 int itemlen = front.length();
 271                                 if (IOHook)
 272                                 {
 273                                         rv = IOHook->OnStreamSocketWrite(this, front);
 274                                         if (rv > 0)
 275                                         {
 276                                                 // consumed the entire string, and is ready for more
 277                                                 sendq_len -= itemlen;
 278                                                 sendq.pop_front();
 279                                         }
 280                                         else if (rv == 0)
 281                                         {
 282                                                 // socket has blocked. Stop trying to send data.
 283                                                 // IOHook has requested unblock notification from the socketengine
 284
 285                                                 // Since it is possible that a partial write took place, adjust sendq_len
 286                                                 sendq_len = sendq_len - itemlen + front.length();
 287                                                 return;
 288                                         }
 289                                         else
 290                                         {
 291                                                 SetError("Write Error"); // will not overwrite a better error message
 292                                                 return;
 293                                         }
 294                                 }
 295 #ifdef DISABLE_WRITEV
 296                                 else
 297                                 {
 298                                         rv = ServerInstance->SE->Send(this, front.data(), itemlen, 0);
 299                                         if (rv == 0)
 300                                         {
 301                                                 SetError("Connection closed");
 302                                                 return;
 303                                         }
 304                                         else if (rv < 0)
 305                                         {
 306                                                 if (errno == EAGAIN || errno == EINTR)
 307                                                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK);
 308                                                 else
 309                                                         SetError(strerror(errno));
 310                                                 return;
 311                                         }
 312                                         else if (rv < itemlen)
 313                                         {
 314                                                 ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK);
 315                                                 front = front.substr(itemlen - rv);
 316                                                 sendq_len -= rv;
 317                                                 return;
 318                                         }
 319                                         else
 320                                         {
 321                                                 sendq_len -= itemlen;
 322                                                 sendq.pop_front();
 323                                                 if (sendq.empty())
 324                                                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_EDGE_WRITE);
 325                                         }
 326                                 }
 327 #endif
 328                         }
 329                 }
 330                 catch (CoreException& modexcept)
 331                 {
 332                         ServerInstance->Logs->Log("SOCKET", DEBUG,"%s threw an exception: %s",
 333                                 modexcept.GetSource(), modexcept.GetReason());
 334                 }
 335         }
 336 #ifndef DISABLE_WRITEV
 337         else
 338         {
 339                 // don't even try if we are known to be blocking
 340                 if (GetEventMask() & FD_WRITE_WILL_BLOCK)
 341                         return;
 342                 // start out optimistic - we won't need to write any more
 343                 int eventChange = FD_WANT_EDGE_WRITE;
 344                 while (error.empty() && sendq_len && eventChange == FD_WANT_EDGE_WRITE)
 345                 {
 346                         // Prepare a writev() call to write all buffers efficiently
 347                         int bufcount = sendq.size();
 348
 349                         // cap the number of buffers at MYIOV_MAX
 350                         if (bufcount > MYIOV_MAX)
 351                         {
 352                                 bufcount = MYIOV_MAX;
 353                         }
 354
 355                         int rv_max = 0;
 356                         iovec* iovecs = new iovec[bufcount];
 357                         for(int i=0; i < bufcount; i++)
 358                         {
 359                                 iovecs[i].iov_base = const_cast<char*>(sendq[i].data());
 360                                 iovecs[i].iov_len = sendq[i].length();
 361                                 rv_max += sendq[i].length();
 362                         }
 363                         int rv = writev(fd, iovecs, bufcount);
 364                         delete[] iovecs;
 365
 366                         if (rv == (int)sendq_len)
 367                         {
 368                                 // it's our lucky day, everything got written out. Fast cleanup.
 369                                 // This won't ever happen if the number of buffers got capped.
 370                                 sendq_len = 0;
 371                                 sendq.clear();
 372                         }
 373                         else if (rv > 0)
 374                         {
 375                                 // Partial write. Clean out strings from the sendq
 376                                 if (rv < rv_max)
 377                                 {
 378                                         // it's going to block now
 379                                         eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK;
 380                                 }
 381                                 sendq_len -= rv;
 382                                 while (rv > 0 && !sendq.empty())
 383                                 {
 384                                         std::string& front = sendq.front();
 385                                         if (front.length() <= (size_t)rv)
 386                                         {
 387                                                 // this string got fully written out
 388                                                 rv -= front.length();
 389                                                 sendq.pop_front();
 390                                         }
 391                                         else
 392                                         {
 393                                                 // stopped in the middle of this string
 394                                                 front = front.substr(rv);
 395                                                 rv = 0;
 396                                         }
 397                                 }
 398                         }
 399                         else if (rv == 0)
 400                         {
 401                                 error = "Connection closed";
 402                         }
 403                         else if (errno == EAGAIN)
 404                         {
 405                                 eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK;
 406                         }
 407                         else if (errno == EINTR)
 408                         {
 409                                 // restart interrupted syscall
 410                                 errno = 0;
 411                         }
 412                         else
 413                         {
 414                                 error = strerror(errno);
 415                         }
 416                 }
 417                 if (!error.empty())
 418                 {
 419                         // error - kill all events
 420                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE);
 421                 }
 422                 else
 423                 {
 424                         ServerInstance->SE->ChangeEventMask(this, eventChange);
 425                 }
 426         }
 427 #endif
 428 }
 429
 430 void StreamSocket::WriteData(const std::string &data)
 431 {
 432         if (fd < 0)
 433         {
 434                 ServerInstance->Logs->Log("SOCKET", DEBUG, "Attempt to write data to dead socket: %s",
 435                         data.c_str());
 436                 return;
 437         }
 438
 439         /* Append the data to the back of the queue ready for writing */
 440         sendq.push_back(data);
 441         sendq_len += data.length();
 442
 443         ServerInstance->SE->ChangeEventMask(this, FD_ADD_TRIAL_WRITE);
 444 }
 445
 446 void SocketTimeout::Tick(time_t)
 447 {
 448         ServerInstance->Logs->Log("SOCKET", DEBUG,"SocketTimeout::Tick");
 449
 450         if (ServerInstance->SE->GetRef(this->sfd) != this->sock)
 451                 return;
 452
 453         if (this->sock->state == I_CONNECTING)
 454         {
 455                 // for connecting sockets, the timeout can occur
 456                 // which causes termination of the connection after
 457                 // the given number of seconds without a successful
 458                 // connection.
 459                 this->sock->OnTimeout();
 460                 this->sock->OnError(I_ERR_TIMEOUT);
 461                 this->sock->state = I_ERROR;
 462
 463                 ServerInstance->GlobalCulls.AddItem(sock);
 464         }
 465
 466         this->sock->Timeout = NULL;
 467 }
 468
 469 void BufferedSocket::OnConnected() { }
 470 void BufferedSocket::OnTimeout() { return; }
 471
 472 void BufferedSocket::DoWrite()
 473 {
 474         if (state == I_CONNECTING)
 475         {
 476                 state = I_CONNECTED;
 477                 this->OnConnected();
 478                 if (GetIOHook())
 479                         GetIOHook()->OnStreamSocketConnect(this);
 480                 else
 481                         ServerInstance->SE->ChangeEventMask(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE);
 482         }
 483         this->StreamSocket::DoWrite();
 484 }
 485
 486 BufferedSocket::~BufferedSocket()
 487 {
 488         this->Close();
 489         if (Timeout)
 490         {
 491                 ServerInstance->Timers->DelTimer(Timeout);
 492                 Timeout = NULL;
 493         }
 494 }
 495
 496 void StreamSocket::HandleEvent(EventType et, int errornum)
 497 {
 498         if (!error.empty())
 499                 return;
 500         BufferedSocketError errcode = I_ERR_OTHER;
 501         try {
 502                 switch (et)
 503                 {
 504                         case EVENT_ERROR:
 505                         {
 506                                 if (errornum == 0)
 507                                         SetError("Connection closed");
 508                                 else
 509                                         SetError(strerror(errornum));
 510                                 switch (errornum)
 511                                 {
 512                                         case ETIMEDOUT:
 513                                                 errcode = I_ERR_TIMEOUT;
 514                                                 break;
 515                                         case ECONNREFUSED:
 516                                         case 0:
 517                                                 errcode = I_ERR_CONNECT;
 518                                                 break;
 519                                         case EADDRINUSE:
 520                                                 errcode = I_ERR_BIND;
 521                                                 break;
 522                                         case EPIPE:
 523                                         case EIO:
 524                                                 errcode = I_ERR_WRITE;
 525                                                 break;
 526                                 }
 527                                 break;
 528                         }
 529                         case EVENT_READ:
 530                         {
 531                                 DoRead();
 532                                 break;
 533                         }
 534                         case EVENT_WRITE:
 535                         {
 536                                 DoWrite();
 537                                 break;
 538                         }
 539                 }
 540         }
 541         catch (CoreException& ex)
 542         {
 543                 ServerInstance->Logs->Log("SOCKET", DEFAULT, "Caught exception in socket processing on FD %d - '%s'",
 544                         fd, ex.GetReason());
 545                 SetError(ex.GetReason());
 546         }
 547         if (!error.empty())
 548         {
 549                 ServerInstance->Logs->Log("SOCKET", DEBUG, "Error on FD %d - '%s'", fd, error.c_str());
 550                 OnError(errcode);
 551         }
 552 }
 553