api.cpp.svn-base

UDT 4.0 based on the UDP.

/*****************************************************************************
Copyright (c) 2001 - 2008, The Board of Trustees of the University of Illinois.
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:

* Redistributions of source code must retain the above
  copyright notice, this list of conditions and the
  following disclaimer.

* Redistributions in binary form must reproduce the
  above copyright notice, this list of conditions
  and the following disclaimer in the documentation
  and/or other materials provided with the distribution.

* Neither the name of the University of Illinois
  nor the names of its contributors may be used to
  endorse or promote products derived from this
  software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*****************************************************************************/

/*****************************************************************************
written by
   Yunhong Gu, last updated 07/25/2008
*****************************************************************************/

#ifdef WIN32
   #include <winsock2.h>
   #include <ws2tcpip.h>
   #include <wspiapi.h>
#else
   #include <unistd.h>
#endif
#include <cassert>
#include <cstring>
#include "api.h"
#include "core.h"

#define GC_EXIT_TIMEOUT 5*1000

#ifdef USING_USTL
using namespace ustl;
#else
using namespace std;
#endif

CUDTSocket::CUDTSocket() throw():
m_pSelfAddr(NULL),
m_pPeerAddr(NULL),
m_pUDT(NULL),
m_pQueuedSockets(NULL),
m_pAcceptSockets(NULL)
{
   #ifndef WIN32
      pthread_mutex_init(&m_AcceptLock, NULL);
      pthread_cond_init(&m_AcceptCond, NULL);
   #else
      m_AcceptLock = CreateMutex(NULL, false, NULL);
      m_AcceptCond = CreateEvent(NULL, false, false, NULL);
   #endif
}

bool CUDTSocket::construct() throw()
{
   return true;
}

CUDTSocket::~CUDTSocket()
{
   if (AF_INET == m_iIPversion)
   {
      if (m_pSelfAddr)
         delete (sockaddr_in*)m_pSelfAddr;
      if (m_pPeerAddr)
         delete (sockaddr_in*)m_pPeerAddr;
   }
   else
   {
      if (m_pSelfAddr)
         delete (sockaddr_in6*)m_pSelfAddr;
      if (m_pPeerAddr)
         delete (sockaddr_in6*)m_pPeerAddr;
   }

   if (m_pUDT)
      delete m_pUDT;

   if (m_pQueuedSockets)
      delete m_pQueuedSockets;
   if (m_pAcceptSockets)
      delete m_pAcceptSockets;

   #ifndef WIN32
      pthread_mutex_destroy(&m_AcceptLock);
      pthread_cond_destroy(&m_AcceptCond);
   #else
      CloseHandle(m_AcceptLock);
      CloseHandle(m_AcceptCond);
   #endif
}

////////////////////////////////////////////////////////////////////////////////

CUDTUnited::CUDTUnited() throw()
:m_pController(NULL)
{
   // Global initialization code
#ifdef WIN32
   WORD wVersionRequested;
   WSADATA wsaData;
   wVersionRequested = MAKEWORD(2, 2);

   if (0 != WSAStartup(wVersionRequested, &wsaData))
      return;
#endif

   srand((unsigned int)CTimer::getTime());
   m_SocketID = 1 + (int)((1 << 30) * (rand()/(RAND_MAX + 1.0)));

#ifndef WIN32
   pthread_mutex_init(&m_ControlLock, NULL);
   pthread_mutex_init(&m_IDLock, NULL);
#else
   m_ControlLock = CreateMutex(NULL, false, NULL);
   m_IDLock = CreateMutex(NULL, false, NULL);
#endif

#ifndef WIN32
   pthread_key_create(&m_TLSError, TLSDestroy);
#else
   m_TLSError = TlsAlloc();
#endif

#ifndef WIN32
   pthread_mutex_init(&m_GCStopLock, NULL);
   pthread_cond_init(&m_GCStopCond, NULL);
   pthread_create(&m_GCThread, NULL, garbageCollect, this);
#else
   m_GCStopCond = CreateEvent(NULL, false, false, NULL);
   m_GCExitCond = CreateEvent(NULL, false, false, NULL);
   DWORD ThreadID;
   m_GCThread = CreateThread(NULL, 0, garbageCollect, this, NULL, &ThreadID);
#endif

   m_bClosing = false;
}

CUDTUnited::~CUDTUnited()
{
   m_bClosing = true;
   #ifndef WIN32
      pthread_cond_signal(&m_GCStopCond);
      pthread_join(m_GCThread, NULL);
      pthread_mutex_destroy(&m_GCStopLock);
      pthread_cond_destroy(&m_GCStopCond);
   #else
      SetEvent(m_GCStopCond);
      WaitForSingleObject(m_GCExitCond, GC_EXIT_TIMEOUT);//INFINITE;!LW!
      TerminateThread(m_GCThread, 0);
      CloseHandle(m_GCThread);
      CloseHandle(m_GCStopCond);
      CloseHandle(m_GCExitCond);
   #endif

   #ifndef WIN32
      pthread_mutex_destroy(&m_ControlLock);
      pthread_mutex_destroy(&m_IDLock);
   #else
      CloseHandle(m_ControlLock);
      CloseHandle(m_IDLock);
   #endif

   #ifndef WIN32
      pthread_key_delete(m_TLSError);
   #else
      TlsFree(m_TLSError);
   #endif

   m_vMultiplexer.clear();
   delete m_pController;

   // Global destruction code
   #ifdef WIN32
      WSACleanup();
   #endif
}

bool CUDTUnited::construct() throw()
{
   //m_vMultiplexer.clear();
   m_pController = UDT::construct(new (nothrow) CControl);
   if (!m_pController)
      return false;
   return true;
}

CUDTException CUDTUnited::newSocket(const int af, const int type, UDTSOCKET& outUdtSocket)
{
   outUdtSocket = CUDT::INVALID_SOCK;

   if ((type != SOCK_STREAM) && (type != SOCK_DGRAM))
      return CUDTException(5, 3, 0);

   CUDTSocket* ns = NULL;

   ns = UDT::construct(new (nothrow) CUDTSocket);
   if (ns)
   {
      if (AF_INET == af)
      {
         ns->m_pSelfAddr = (sockaddr*)(new (nothrow) sockaddr_in);
         if (ns->m_pSelfAddr)
         {
            ((sockaddr_in*)(ns->m_pSelfAddr))->sin_port = 0;
         }
         else
         {
ERR_ALLOC_FAILED_newSocket:
            delete ns;
            return CUDTException(3, 2, 0);
         }
      }
      else
      {
         ns->m_pSelfAddr = (sockaddr*)(new (nothrow) sockaddr_in6);
         if (ns->m_pSelfAddr)
         {
            ((sockaddr_in6*)(ns->m_pSelfAddr))->sin6_port = 0;
         }
         else
         {
            goto ERR_ALLOC_FAILED_newSocket;
         }
      }
      ns->m_pUDT = UDT::construct(new (nothrow) CUDT);
      if (!(ns->m_pUDT))
      {
         goto ERR_ALLOC_FAILED_newSocket;
      }
   }
   else
   {
      return CUDTException(3, 2, 0);
   }

   #ifndef WIN32
      pthread_mutex_lock(&m_IDLock);
   #else
      WaitForSingleObject(m_IDLock, INFINITE);
   #endif
   ns->m_SocketID = -- m_SocketID;
   #ifndef WIN32
      pthread_mutex_unlock(&m_IDLock);
   #else
      ReleaseMutex(m_IDLock);
   #endif

   ns->m_Status = CUDTSocket::INIT;
   ns->m_ListenSocket = 0;
   ns->m_pUDT->m_SocketID = ns->m_SocketID;
   ns->m_pUDT->m_iSockType = (SOCK_STREAM == type) ? UDT_STREAM : UDT_DGRAM;
   ns->m_pUDT->m_iIPversion = ns->m_iIPversion = af;
   ns->m_pUDT->m_pController = m_pController;

   // protect the m_Sockets structure.
   #ifndef WIN32
      pthread_mutex_lock(&m_ControlLock);
   #else
      WaitForSingleObject(m_ControlLock, INFINITE);
   #endif
   
   if (!m_Sockets.insert(std::make_pair(ns->m_SocketID, ns)).second)
   {
      //failure and rollback
      delete ns;
      ns = NULL;
   }
   #ifndef WIN32
      pthread_mutex_unlock(&m_ControlLock);
   #else
      ReleaseMutex(m_ControlLock);
   #endif

   if (NULL == ns)
      return CUDTException(3, 2, 0);

   outUdtSocket = ns->m_SocketID;

   return CUDTException();
}

int CUDTUnited::newConnection(const UDTSOCKET listen, const sockaddr* peer, CHandShake* hs)
{
   CUDTSocket* ns = NULL;
   CUDTSocket* ls = locate(listen);

   // if this connection has already been processed
   if (NULL != (ns = locate(listen, peer, hs->m_iID, hs->m_iISN)))
   {
      if (ns->m_pUDT->m_bBroken)
      {
         // last connection from the "peer" address has been broken
         ns->m_Status = CUDTSocket::CLOSED;
         ns->m_TimeStamp = CTimer::getTime();

         #ifndef WIN32
            pthread_mutex_lock(&(ls->m_AcceptLock));
         #else
            WaitForSingleObject(ls->m_AcceptLock, INFINITE);
         #endif
         ls->m_pQueuedSockets->erase(ns->m_SocketID);
         ls->m_pAcceptSockets->erase(ns->m_SocketID);
         #ifndef WIN32
            pthread_mutex_unlock(&(ls->m_AcceptLock));
         #else
            ReleaseMutex(ls->m_AcceptLock);
         #endif
      }
      else
      {
         // connection already exist, this is a repeated connection request
         // respond with existing HS information

         hs->m_iISN = ns->m_pUDT->m_iISN;
         hs->m_iMSS = ns->m_pUDT->m_iMSS;
         hs->m_iFlightFlagSize = ns->m_pUDT->m_iFlightFlagSize;
         hs->m_iReqType = -1;
         hs->m_iID = ns->m_SocketID;

         return 0;

         //except for this situation a new connection should be started
      }
   }

   // exceeding backlog, refuse the connection request
   if (ls->m_pQueuedSockets->size() >= ls->m_uiBackLog)
      return -1;

   ns = UDT::construct(new (nothrow) CUDTSocket);
   if (ns)
   {
      ns->m_pUDT = UDT::construct(new (nothrow) CUDT(*(ls->m_pUDT)));
      if (ns->m_pUDT)
      {
         if (AF_INET == ls->m_iIPversion)
         {
            ns->m_pSelfAddr = (sockaddr*)(new (nothrow) sockaddr_in);
            if (ns->m_pSelfAddr)
            {
               ((sockaddr_in*)(ns->m_pSelfAddr))->sin_port = 0;
               ns->m_pPeerAddr = (sockaddr*)(new (nothrow) sockaddr_in);
               if (ns->m_pPeerAddr)
                  memcpy(ns->m_pPeerAddr, peer, sizeof(sockaddr_in));
               else
                  goto ERR_ALLOC_FAILED_newConnection;
            }
            else
            {
               goto ERR_ALLOC_FAILED_newConnection;
            }
         }
         else
         {
            ns->m_pSelfAddr = (sockaddr*)(new (nothrow) sockaddr_in6);
            if (ns->m_pSelfAddr)
            {
               ((sockaddr_in6*)(ns->m_pSelfAddr))->sin6_port = 0;
               ns->m_pPeerAddr = (sockaddr*)(new (nothrow) sockaddr_in6);
               if (ns->m_pPeerAddr)
                  memcpy(ns->m_pPeerAddr, peer, sizeof(sockaddr_in6));
               else
                  goto ERR_ALLOC_FAILED_newConnection;
            }
            else
            {
               goto ERR_ALLOC_FAILED_newConnection;
            }
         }
      }
      else
      {
ERR_ALLOC_FAILED_newConnection:
         delete ns;
         return -1;
      }
   }
   else
   {
      return -1;
   }

   #ifndef WIN32
      pthread_mutex_lock(&m_IDLock);
   #else
      WaitForSingleObject(m_IDLock, INFINITE);
   #endif
   ns->m_SocketID = -- m_SocketID;
   #ifndef WIN32
      pthread_mutex_unlock(&m_IDLock);
   #else
      ReleaseMutex(m_IDLock);
   #endif

   ns->m_ListenSocket = listen;
   ns->m_iIPversion = ls->m_iIPversion;
   ns->m_pUDT->m_SocketID = ns->m_SocketID;
   ns->m_PeerID = hs->m_iID;
   ns->m_iISN = hs->m_iISN;

   int error = 0;

   {
      // bind to the same addr of listening socket
      CUDTException e = ns->m_pUDT->open();
      if (e.getErrorCode())
      {
         error = 1;
         goto ERR_ROLLBACK;
      }
      else
      {
         updateMux(ns->m_pUDT, ls);
         e = ns->m_pUDT->connect(peer, hs);
         if (e.getErrorCode())
         {
            error = 1;
            goto ERR_ROLLBACK;
         }
      }
   }

   ns->m_Status = CUDTSocket::CONNECTED;

   // copy address information of local node
   ns->m_pUDT->m_pSndQueue->m_pChannel->getSockAddr(ns->m_pSelfAddr);

   // protect the m_Sockets structure.
   #ifndef WIN32
      pthread_mutex_lock(&m_ControlLock);
   #else
      WaitForSingleObject(m_ControlLock, INFINITE);
   #endif
   if (!m_Sockets.insert(make_pair(ns->m_SocketID, ns)).second)
   {
      error = 2;
   }
   #ifndef WIN32
      pthread_mutex_unlock(&m_ControlLock);
   #else
      ReleaseMutex(m_ControlLock);
   #endif

   #ifndef WIN32
      pthread_mutex_lock(&(ls->m_AcceptLock));
   #else
      WaitForSingleObject(ls->m_AcceptLock, INFINITE);
   #endif
   if (!ls->m_pQueuedSockets->insert(ns->m_SocketID).second)
   {
      error = 3;
   }
   #ifndef WIN32
      pthread_mutex_unlock(&(ls->m_AcceptLock));
   #else
      ReleaseMutex(ls->m_AcceptLock);
   #endif

   CTimer::triggerEvent();

ERR_ROLLBACK:
   if (error > 0)
   {
      ns->m_pUDT->close();
      ns->m_Status = CUDTSocket::CLOSED;
      ns->m_TimeStamp = CTimer::getTime();
      // !nash! there was a memory leak in original implementation
      if (error < 3) 
      {
         delete ns;