api.cpp.svn-base

UDT 4.0 based on the UDP.

      }
      else
      {
         m_Sockets.erase(ns->m_SocketID);
         delete ns;
      }

      return -1;
   }

   // wake up a waiting accept() call
   #ifndef WIN32
      pthread_cond_signal(&(ls->m_AcceptCond));
   #else
      SetEvent(ls->m_AcceptCond);
   #endif

   return 1;
}

CUDTException CUDTUnited::lookup(const UDTSOCKET u, CUDT*& outUdt)
{
   // protects the m_Sockets structure
   CGuard cg(m_ControlLock);

   map<UDTSOCKET, CUDTSocket*>::iterator i = m_Sockets.find(u);

   if ((i == m_Sockets.end()) || (i->second->m_Status == CUDTSocket::CLOSED))
      return CUDTException(5, 4, 0);

   outUdt = i->second->m_pUDT;
   return CUDTException();
}

CUDTSocket::UDTSTATUS CUDTUnited::getStatus(const UDTSOCKET u)
{
   // protects the m_Sockets structure
   CGuard cg(m_ControlLock);

   map<UDTSOCKET, CUDTSocket*>::iterator i = m_Sockets.find(u);

   if (i == m_Sockets.end())
      return CUDTSocket::INIT;

   if (i->second->m_pUDT->m_bBroken)
      return CUDTSocket::BROKEN;

   return i->second->m_Status;   
}

CUDTException CUDTUnited::bind(const UDTSOCKET u, const sockaddr* name, const int namelen)
{
   CUDTSocket* s = locate(u);

   if (NULL == s)
      return CUDTException(5, 4, 0);

   // cannot bind a socket more than once
   if (CUDTSocket::INIT != s->m_Status)
      return CUDTException(5, 0, 0);

   // check the size of SOCKADDR structure
   if (AF_INET == s->m_iIPversion)
   {
      if (namelen != sizeof(sockaddr_in))
         return CUDTException(5, 3, 0);
   }
   else
   {
      if (namelen != sizeof(sockaddr_in6))
         return CUDTException(5, 3, 0);
   }

   CUDTException e = s->m_pUDT->open();
   if (!e.getErrorCode())
   {
      e = updateMux(s->m_pUDT, name);
      if (!e.getErrorCode())
      {
         s->m_Status = CUDTSocket::OPENED;

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

   return e;
}

CUDTException CUDTUnited::bind(UDTSOCKET u, UDPSOCKET udpsock)
{
   CUDTSocket* s = locate(u);

   if (NULL == s)
      return CUDTException(5, 4, 0);

   // cannot bind a socket more than once
   if (CUDTSocket::INIT != s->m_Status)
      return CUDTException(5, 0, 0);

   sockaddr_in name4;
   sockaddr_in6 name6;
   sockaddr* name;
   socklen_t namelen;

   if (AF_INET == s->m_iIPversion)
   {
      namelen = sizeof(sockaddr_in);
      name = (sockaddr*)&name4;
   }
   else
   {
      namelen = sizeof(sockaddr_in6);
      name = (sockaddr*)&name6;
   }

   if (-1 == ::getsockname(udpsock, name, &namelen))
      return CUDTException(5, 3);

   CUDTException e = s->m_pUDT->open();
   if (!e.getErrorCode())
   {
      e = updateMux(s->m_pUDT, name, &udpsock);
      if (!e.getErrorCode())
      {
         s->m_Status = CUDTSocket::OPENED;

         // copy address information of local node
         s->m_pUDT->m_pSndQueue->m_pChannel->getSockAddr(s->m_pSelfAddr);
      }
   }
   return e;
}

CUDTException CUDTUnited::listen(const UDTSOCKET u, const int backlog)
{
   CUDTSocket* s = locate(u);

   if (NULL == s)
      return CUDTException(5, 4, 0);

   // do nothing if the socket is already listening
   if (CUDTSocket::LISTENING == s->m_Status)
      return CUDTException();

   // a socket can listen only if is in OPENED status
   if (CUDTSocket::OPENED != s->m_Status)
      return CUDTException(5, 5, 0);

   // listen is not supported in rendezvous connection setup
   if (s->m_pUDT->m_bRendezvous)
      return CUDTException(5, 7, 0);

   if (backlog <= 0)
      return CUDTException(5, 3, 0);

   s->m_uiBackLog = backlog;

   s->m_pQueuedSockets = new (nothrow) set<UDTSOCKET>;
   if (s->m_pQueuedSockets)
   {
      s->m_pAcceptSockets = new (nothrow) set<UDTSOCKET>;
      if (!(s->m_pAcceptSockets))
      {
         delete s->m_pQueuedSockets;
         return CUDTException(3, 2, 0);
      }
   }
   else   
   {
      return CUDTException(3, 2, 0);
   }

   s->m_pUDT->listen();

   s->m_Status = CUDTSocket::LISTENING;

   return CUDTException();
}

CUDTException CUDTUnited::accept(const UDTSOCKET listen, sockaddr* addr, int* addrlen, UDTSOCKET& outUdtSocket)
{
   outUdtSocket = CUDT::INVALID_SOCK;
   if ((NULL != addr) && (NULL == addrlen))
      return CUDTException(5, 3, 0);

   CUDTSocket* ls = locate(listen);

   if (ls == NULL)
      return CUDTException(5, 4, 0);

   // the "listen" socket must be in LISTENING status
   if (CUDTSocket::LISTENING != ls->m_Status)
      return CUDTException(5, 6, 0);

   // no "accept" in rendezvous connection setup
   if (ls->m_pUDT->m_bRendezvous)
      return CUDTException(5, 7, 0);

   UDTSOCKET u = CUDT::INVALID_SOCK;
   bool accepted = false;

   // !!only one conection can be set up each time!!
   #ifndef WIN32
      while (!accepted)
      {
         pthread_mutex_lock(&(ls->m_AcceptLock));

         if (ls->m_pQueuedSockets->size() > 0)
         {
            u = *(ls->m_pQueuedSockets->begin());
            ls->m_pAcceptSockets->insert(u);
            ls->m_pQueuedSockets->erase(ls->m_pQueuedSockets->begin());

            accepted = true;
         }
         else if (!ls->m_pUDT->m_bSynRecving)
            accepted = true;
         else if (CUDTSocket::LISTENING == ls->m_Status)
            pthread_cond_wait(&(ls->m_AcceptCond), &(ls->m_AcceptLock));

         if (CUDTSocket::LISTENING != ls->m_Status)
            accepted = true;

         pthread_mutex_unlock(&(ls->m_AcceptLock));
      }
   #else
      while (!accepted)
      {
         WaitForSingleObject(ls->m_AcceptLock, INFINITE);

         if (ls->m_pQueuedSockets->size() > 0)
         {
            u = *(ls->m_pQueuedSockets->begin());
            ls->m_pAcceptSockets->insert(ls->m_pAcceptSockets->end(), u);
            ls->m_pQueuedSockets->erase(ls->m_pQueuedSockets->begin());

            accepted = true;
         }
         else if (!ls->m_pUDT->m_bSynRecving)
            accepted = true;

         ReleaseMutex(ls->m_AcceptLock);

         if  (!accepted & (CUDTSocket::LISTENING == ls->m_Status))
            WaitForSingleObject(ls->m_AcceptCond, INFINITE);

         if (CUDTSocket::LISTENING != ls->m_Status)
         {
            SetEvent(ls->m_AcceptCond);
            accepted = true;
         }
      }
   #endif

   if (u == CUDT::INVALID_SOCK)
   {
      // non-blocking receiving, no connection available
      if (!ls->m_pUDT->m_bSynRecving)
         return CUDTException(6, 2, 0);

      // listening socket is closed
      return CUDTException(5, 6, 0);
   }

   if (AF_INET == locate(u)->m_iIPversion)
      *addrlen = sizeof(sockaddr_in);
   else
      *addrlen = sizeof(sockaddr_in6);

   // copy address information of peer node
   memcpy(addr, locate(u)->m_pPeerAddr, *addrlen);

   outUdtSocket = u;
   return CUDTException();
}

CUDTException CUDTUnited::connect(const UDTSOCKET u, const sockaddr* name, const int namelen)
{
   CUDTSocket* s = locate(u);

   if (NULL == s)
      return CUDTException(5, 4, 0);

   // check the size of SOCKADDR structure
   if (AF_INET == s->m_iIPversion)
   {
      if (namelen != sizeof(sockaddr_in))
         return CUDTException(5, 3, 0);
   }
   else
   {
      if (namelen != sizeof(sockaddr_in6))
         return CUDTException(5, 3, 0);
   }

   // a socket can "connect" only if it is in INIT or OPENED status
   if (CUDTSocket::INIT == s->m_Status)
   {
      if (!s->m_pUDT->m_bRendezvous)
      {
         CUDTException e = s->m_pUDT->open();
         if (e.getErrorCode())
            return e;
         e = updateMux(s->m_pUDT);
         if (e.getErrorCode())
            return e;
         s->m_Status = CUDTSocket::OPENED;
      }
      else
         return CUDTException(5, 8, 0);
   }
   else if (CUDTSocket::OPENED != s->m_Status)
   {
      return CUDTException(5, 2, 0);
   }

   s->m_pUDT->connect(name);
   s->m_Status = CUDTSocket::CONNECTED;

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

   // record peer address
   if (AF_INET == s->m_iIPversion)
   {
      s->m_pPeerAddr = (sockaddr*)(new (nothrow) sockaddr_in);
      if (s->m_pPeerAddr)
         memcpy(s->m_pPeerAddr, name, sizeof(sockaddr_in));
      else
         return CUDTException(3, 2, 0);
   }
   else
   {
      s->m_pPeerAddr = (sockaddr*)(new (nothrow) sockaddr_in6);
      if (s->m_pPeerAddr)
         memcpy(s->m_pPeerAddr, name, sizeof(sockaddr_in6));
      else
         return CUDTException(3, 2, 0);
   }

   return CUDTException();
}

int CUDTUnited::close(const UDTSOCKET u)
{
   CUDTSocket* s = locate(u);

   // silently drop a request to close an invalid ID, rather than return error   
   if (NULL == s)
      return 0;

   s->m_pUDT->close();

   // a socket will not be immediated removed when it is closed
   // in order to prevent other methods from accessing invalid address
   // a timer is started and the socket will be removed after approximately 1 second
   s->m_TimeStamp = CTimer::getTime();

   CUDTSocket::UDTSTATUS os = s->m_Status;

   // synchronize with garbage collection.
   #ifndef WIN32
      pthread_mutex_lock(&m_ControlLock);
   #else
      WaitForSingleObject(m_ControlLock, INFINITE);
   #endif
   s->m_Status = CUDTSocket::CLOSED;
   #ifndef WIN32
      pthread_mutex_unlock(&m_ControlLock);
   #else
      ReleaseMutex(m_ControlLock);
   #endif

   // broadcast all "accept" waiting
   if (CUDTSocket::LISTENING == os)
   {
      #ifndef WIN32
         pthread_mutex_lock(&(s->m_AcceptLock));
         pthread_mutex_unlock(&(s->m_AcceptLock));
         pthread_cond_broadcast(&(s->m_AcceptCond));
      #else
         SetEvent(s->m_AcceptCond);
      #endif
   }

   CTimer::triggerEvent();

   return 0;
}

CUDTException CUDTUnited::getpeername(const UDTSOCKET u, sockaddr* name, int* namelen)
{
   if (CUDTSocket::CONNECTED != getStatus(u))
      return CUDTException(2, 2, 0);

   CUDTSocket* s = locate(u);

   if (NULL == s)
      return CUDTException(5, 4, 0);

   if (!s->m_pUDT->m_bConnected || s->m_pUDT->m_bBroken)
      return CUDTException(2, 2, 0);

   if (AF_INET == s->m_iIPversion)
      *namelen = sizeof(sockaddr_in);
   else
      *namelen = sizeof(sockaddr_in6);

   // copy address information of peer node
   memcpy(name, s->m_pPeerAddr, *namelen);

   return CUDTException();
}

CUDTException CUDTUnited::getsockname(const UDTSOCKET u, sockaddr* name, int* namelen)
{
   CUDTSocket* s = locate(u);

   if (NULL == s)
      return CUDTException(5, 4, 0);

   if (CUDTSocket::INIT == s->m_Status)
      return CUDTException(2, 2, 0);

   if (AF_INET == s->m_iIPversion)
      *namelen = sizeof(sockaddr_in);
   else
      *namelen = sizeof(sockaddr_in6);

   // copy address information of local node
   memcpy(name, s->m_pSelfAddr, *namelen);

   return CUDTException();
}

CUDTException CUDTUnited::select(ud_set* readfds, ud_set* writefds, ud_set* exceptfds, const timeval* timeout, int& count)
{
   uint64_t entertime = CTimer::getTime();

   uint64_t to;
   if (NULL == timeout)
      to = 0xFFFFFFFFFFFFFFFFULL;
   else
      to = timeout->tv_sec * 1000000 + timeout->tv_usec;

   // initialize results
   count = 0;
   set<UDTSOCKET> rs, ws, es;

   // retrieve related UDT sockets
   vector<CUDTSocket*> ru, wu;//, eu;
   CUDTSocket* s;
   if (NULL != readfds)
   {
      for (set<UDTSOCKET>::iterator i1 = readfds->begin(); i1 != readfds->end(); ++i1)
      {
         if (CUDTSocket::BROKEN == getStatus(*i1))
         {
            rs.insert(*i1);
            ++count;
         }
         else if (NULL == (s = locate(*i1)))
         {
            //return CUDTException(5, 4, 0);
         }
         else
         {
            ru.insert(ru.end(), s);
         }
      }
   }
   if (NULL != writefds)
   {
      for (set<UDTSOCKET>::iterator i2 = writefds->begin(); i2 != writefds->end(); ++i2)
      {
         if (CUDTSocket::BROKEN == getStatus(*i2))
         {
            ws.insert(*i2);
            ++count;
         }
         else if (NULL == (s = locate(*i2)))
         {
            //return CUDTException(5, 4, 0);
         }
         else
         {
            wu.insert(wu.end(), s);
         }
      }
   }
   if (NULL != exceptfds)
   {
      for (set<UDTSOCKET>::iterator i3 = exceptfds->begin(); i3 != exceptfds->end(); ++i3)