CConnectionSocket.cpp

/*
 Copyright [2010] [Richard Bross]

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
*/
// CConnectionSocket.cpp file
//
//	Created: Richard A. Bross
//
#if defined(_WIN32)
#include "stdafx.h"
#endif
#include "XPlat.h"
#include "CBaseSocket.h"
#include "CConnectionSocket.h"
#include <syslog.h>
#include "CProfileTimer.h"
#include <poll.h>

// Constructor

CConnectionSocket::CConnectionSocket(CBaseSocket *pOwner, struct sockaddr *pAddr) : CBaseSocket()
{
 memset(&hThread, 0, sizeof(hThread));
 cOwner = pOwner;
 memset(&sAddr, 0, sizeof(sAddr));
 memset(&sUAddr, 0, sizeof(sUAddr));
 if (pAddr)
 memcpy(&sAddr, pAddr, sizeof(sAddr));
 eRequestClose = CLOSE_ACTIVE;
 if (cOwner)
 iAF = pOwner->GetSocketType();
}

CConnectionSocket::~CConnectionSocket()
{
 TerminateConnection();
}


// Requests that the handleconenction thread close
// NOTES:
// We used to just close the socket. However, data waiting in the socket buffer was lost.
// We still want the thread to exit on an invalid socket, since the other side may close the connection.
// Now we set a flag and let the HandleConnection thread get remaining data.
// If a user is calling this and they are stupid enough to call SendDataOut at the same time,
// tough luck, the connection will be closed after 50 ms max.

void CConnectionSocket::TerminateConnection(ECLOSE_TYPE eRequest)
{
 // Ask the connection thread to close
 CProfileTimer::SetTimestamp("CCS-CLOSE");
 eRequestClose = eRequest;
 if (hThread)
 {
#if defined(_WIN32)
 ::WaitForSingleObject(hThread, INFINITE);
 ::CloseHandle(hThread); // _beginthreadex doesn't auto close the handle
#else
 // Make sure thread finishes
 int iRet;
 if ((iRet = ::pthread_join(hThread, NULL)) != 0)
 { // EDEADLK (35) will occur if some bozo calls this routine from the thread itself
 ::openlog("connectionsocket", LOG_PID, LOG_LOCAL0);
 char strLogEntry[1024];
 ::sprintf(strLogEntry, "pthread_join error could cause memory leak. Error number: %d", iRet);
 ::syslog(LOG_CRIT, "%s", strLogEntry);
 ::closelog();
 }
 else
 { // Only on successs
 ::memset(&hThread, 0, sizeof(hThread));
 }

#endif
 };
 // Close the socket
 CloseSocket();
};


// Starts thread to service connection. Returns thread handle.

XPLAT_HANDLE CConnectionSocket::ServiceConnection(SOCKET sNewSocket)
{
 // Timestamp
 tConnected = time(NULL);

 // Already servicing a connection?
 if (hThread)
 {
 if (CXPlatThread::IsThreadActive(hThread))
 return(hThread);
 else
 {
#if defined(_WIN32)
 ::CloseHandle(hThread);
#endif
 ::memset(&hThread, 0, sizeof(hThread));
 };
 };

 // Assign socket
 sSocket = sNewSocket;


#if defined(_WIN32)
 // Begin the servicing thread
 unsigned pThread;

 hThread = (HANDLE) ::_beginthreadex(NULL,
 0,
 ServiceConnectionThread,
 (void *) this,
 1,
 &pThread);
#else
 int iReturn;
 iReturn = ::pthread_create(&hThread, pthread_attr_default, (START_ROUTINE) ServiceConnectionThread, (void *) this);
#endif

 return(hThread);
};




// Send data
BOOL CConnectionSocket::SendDataOut(LPSTR lpData, int nPacketSize, int nTotalLength)
{
 int nDataSent = 0;
 int nReturn = 0;
#if defined(__linux)
 int nFlags = 0x4000; // MSG_NOSIGNAL
#else
 int nFlags = 0;
#endif
// fd_set writesets;
 struct pollfd fds[1];
// int nReturnSet = SOCKET_ERROR;
#if defined(__linux)	
 //	In Linux, doing a close	on a socket in a thread doesn't 
 // seem to always unblock select or listen in another
 // thread. So for Linux, we'll timeout after .5 seconds and loop
// struct timeval sTv;
 // 3 ms. This is the time that we wait for the output buffer to have some space. Probably almost never have to wait.
// long lUSec = 3000;
 long lUSec = 3;
#endif

 // Make sure pointer isn't NULL
 if (!lpData)
 return FALSE;

 // Send data
// FD_ZERO(&writesets);
 fds[0].fd = sSocket;
 fds[0].events = POLLOUT | POLLWRBAND; // POLLOUT == POLLWRNORM 

 while(TRUE)
 {
 cCriticalSocket.Lock();
 if (sSocket == SOCKET_ERROR)
 {
 cCriticalSocket.Unlock();
 return(FALSE);
 };
// FD_SET(sSocket, &writesets);
 cCriticalSocket.Unlock();

 // Anything waiting to be sent?
#if defined(__linux)
// sTv.tv_sec = 0;
// sTv.tv_usec = lUSec;
 try
 {
// nReturnSet = ::select(sSocket + 1, NULL, &writesets, NULL, &sTv);
 nReturn = ::poll(fds, 1, lUSec);
 }
 catch(...)
 {
 nReturn = SOCKET_ERROR;
 }
#else
 nReturnSet = ::select((int) sSocket + 1, NULL, (XPLAT_FD_SET_PTR) & writesets, (XPLAT_FD_SET_PTR) NULL, NULL);
#endif
 if (nReturn == SOCKET_ERROR)
 {
 CloseSocket();
 return(FALSE);
 };

 cCriticalSocket.Lock();
 if (sSocket == SOCKET_ERROR)
 {
 cCriticalSocket.Unlock();
 return(FALSE);
 };
// nReturn = FD_ISSET(sSocket, &writesets);
 cCriticalSocket.Unlock();
 if (!nReturn && (fds[0].revents & POLLHUP || fds[0].revents & POLLERR)) 
 return(FALSE);
 if (nReturn)
 { // Other end closed
 // If not one of the "ok to write" events, loop. Is this possible at this point
 if (!fds[0].revents & POLLWRBAND && !fds[0].revents & POLLOUT) 
 continue;
 // Send the whole packet to the client if the packet is big enough
 cCriticalSocket.Lock();
 if (sSocket == SOCKET_ERROR)
 {
 cCriticalSocket.Unlock();
 return(FALSE);
 };
 if (nDataSent + nPacketSize >= nTotalLength)
 {
 try
 {
 nReturn = ::send(sSocket, (LPSTR) lpData + nDataSent, nTotalLength - nDataSent, nFlags);
 }
 catch(...)
 {
 nReturn = SOCKET_ERROR;
 }
 cCriticalSocket.Unlock();
 if (nReturn == SOCKET_ERROR)
 {
 nReturn = XPLAT_WSAGETLASTERR;
 if (nReturn != XEWOULDBLOCK && // would block
 nReturn != XEINPROGRESS) // in progress
 { // error return
 CloseSocket();
 return(FALSE);
 }
 };
 nDataSent += (nTotalLength - nDataSent);
 break;
 }
 else
 {
 try
 {
 nReturn = ::send(sSocket, lpData + nDataSent, nPacketSize, nFlags);
 }
 catch(...)
 {
 nReturn = SOCKET_ERROR;
 }
 cCriticalSocket.Unlock();
 if (nReturn == SOCKET_ERROR)
 {
 nReturn = XPLAT_WSAGETLASTERR;
 if (nReturn != XEWOULDBLOCK && // would block
 nReturn != XEINPROGRESS) // in progress
 { // error return
 CloseSocket();
 return(FALSE);
 }
 };
 nDataSent += nPacketSize;
 }
 // It was all sent
 if (nDataSent >= nTotalLength)
 break;
 // We only have to worry about this for partial data
 CProfileTimer::SetTimestamp("CCS-SEND");
 }
 else
 { // If nothing to write, it's been 6ms since the last write or 50 ms since we were asked to close
 if (eRequestClose == CLOSE_IMMEDIATE || // Close right now
 (eRequestClose == CLOSE_WAIT && // Wait for data. 1ms (6 with the select call) since last send or 50ms for any send
 (CProfileTimer::GetElapsedMS("CCS-SEND", "NowSend", true) > 1 || CProfileTimer::GetElapsedMS("CCS-CLOSE", "NowSend", false) > 50)))
 {
 break;
 }
 }
 // Clean up
 cCriticalSocket.Lock();
 if (sSocket == SOCKET_ERROR)
 {
 cCriticalSocket.Unlock();
 return(FALSE);
 };
// FD_ZERO(&writesets);
 cCriticalSocket.Unlock();
 }

 return(nDataSent == nTotalLength);
}



// Receive the data from socket, external buffer

int CConnectionSocket::ReceiveData(void *lpBuf, int nBufLen)
{
 int iReturn;
#if defined(__linux)
 int nFlags = 0x4000; // MSG_NOSIGNAL
#else
 int nFlags = 0;
#endif

 // Get any data sent over the connection
 cCriticalSocket.Lock();
 try
 {
 iReturn = ::recv(sSocket, (LPSTR) lpBuf, nBufLen, nFlags);
 }
 catch(...)
 {
 iReturn = SOCKET_ERROR;
 }
 cCriticalSocket.Unlock();

 return(iReturn);
}

// Receive the data from socket, internal buffer

int CConnectionSocket::ReceiveData()
{
 return(ReceiveData(ucBuffer, CS_BUFFER_SIZE));
}


// Called by ServiceConnectionThread to actually do the work

int CConnectionSocket::HandleConnection()
{
// fd_set readsets;
 struct pollfd fds[1];
 int nReturn;
 int nRevLength;
 int nPacket = 2048;
 unsigned char *ucHS;
#if defined(__linux)
 // Linux "select()" does not unblock when the socket is closed. So we set a timeout.
// struct timeval sTv;
// long lUSec = 3000; // If the socket gets closed, this will shut it down within 3ms
 long lUSec = 3; // If the socket gets closed, this will shut it down within 3ms
#endif
 try
 {
 // Send handshake, if derived class wants us to
 if (GetHandshake(&ucHS, nReturn))
 SendDataOut((char *) ucHS, nPacket, nReturn);

 // Clear sets
// FD_ZERO(&readsets);
 fds[0].fd = sSocket;
 fds[0].events = POLLIN | POLLRDBAND; // POLLRDNORM == POLLIN. POLLRDHUP is only available if _GNU_SOURCE is defined before headers

 // Read the data
 while(TRUE)
 {
 cCriticalSocket.Lock(); // Lock
 // Someone close it?
 if (sSocket == SOCKET_ERROR)
 {
 cCriticalSocket.Unlock(); // if Unlock
 Disconnected();
 return(0);
 };

 // Init sets
// FD_SET(sSocket, &readsets);
// cCriticalSocket.Unlock(); // Unlock

 // Blocking call
#if defined(__linux)
 // Reset every time because the Linux "select()" modifies the value
// sTv.tv_sec = 0;
// sTv.tv_usec = lUSec;
 try
 {
// nReturn = ::select(sSocket + 1, &readsets, NULL, NULL, &sTv);
 nReturn = ::poll(fds, 1, lUSec);
 }
 catch(...)
 {
 nReturn = SOCKET_ERROR;
 }
#else
 nReturn = ::select((int) sSocket + 1, (XPLAT_FD_SET_PTR) & readsets, NULL, (XPLAT_FD_SET_PTR) NULL, NULL);
#endif
 // Someone close it or an error
// cCriticalSocket.Lock();
 cCriticalSocket.Unlock();
 if (nReturn == SOCKET_ERROR)
 {
 if (sSocket != SOCKET_ERROR)
 CloseSocket();
 Disconnected();
 return(0);
 };

// nReturn = FD_ISSET(sSocket, &readsets);
// cCriticalSocket.Unlock();
 if (!nReturn && (fds[0].revents & POLLHUP || fds[0].revents & POLLERR)) 
 {
 Disconnected();
 return(0);
 }
 // We have data
 if (nReturn)
 { // If not one of the "ok to read" events, loop. Is this possible at this point
 if (!fds[0].revents & POLLRDBAND && !fds[0].revents & POLLIN) 
 continue;
 nRevLength = ReceiveData();
 if (nRevLength == SOCKET_ERROR || nRevLength == 0)
 {
 nReturn = XPLAT_WSAGETLASTERR;
 if (nReturn != XEWOULDBLOCK && // would block
 nReturn != XEINPROGRESS) // in progress
 { // error return, probably closed and ended nornally
 if (sSocket != SOCKET_ERROR)
 CloseSocket();
 Disconnected();
 return(0);
 }
 }
 // This should be impossible
// if (nRevLength > CConnectionSocket::CS_BUFFER_SIZE)
// {
// Disconnected();
// return(0);
// };
 // Echo data
 if (nRevLength > 0)
 { // Make sure our socket isn't closed. If we're in our destructor, the derived class
 // is already destroyed and ProcessData will be calling a pure virtual function.
// if (sSocket == SOCKET_ERROR)
// {
// Disconnected();
// return(0);
// };
 // Process the data
 ProcessData(ucBuffer, nRevLength);
 CProfileTimer::SetTimestamp("CCS-RECV");
 }
 }
 else
 { // If nothing to read, it's been 6ms since the last read or 50 ms since we were asked to close
 if (eRequestClose == CLOSE_IMMEDIATE || // Close right now
 (eRequestClose == CLOSE_WAIT && // Wait for data. 1ms (6 with the select call) since last receive or 50ms for any receive
 (CProfileTimer::GetElapsedMS("CCS-RECV", "NowRecv", true) > 1 || CProfileTimer::GetElapsedMS("CCS-CLOSE", "NowRecv", false) > 50)))
 {
 Disconnected();
 return(0);
 }
 }

// // clear up
// cCriticalSocket.Lock();
// if (sSocket == SOCKET_ERROR)
// {
// Disconnected();
// return(0);
// };
 
// FD_ZERO(&readsets);
 };
 }
 catch(...)
 { // If we're not locked, no harm. But if we are it prevents a deadlock.
 cCriticalSocket.Unlock();
 Disconnected();
 return(-1);
 }

 Disconnected();
 return(0);
};

// This thread waits for data

unsigned CConnectionSocket::ServiceConnectionThread(void* pVoid)
{
 CConnectionSocket *cConnectionSocket;
 unsigned uReturn;

 // Assign our "this" pointer
 if (!pVoid)
 return((unsigned) - 1);

 cConnectionSocket = (CConnectionSocket *) pVoid;
 uReturn = cConnectionSocket->HandleConnection();

 return(uReturn);
}


// Return the time the server was started

time_t CConnectionSocket::GetTimeConnected()
{
 return(tConnected);
};




// Connected?

BOOL CConnectionSocket::IsConnected()
{
 if ((int) sSocket == SOCKET_ERROR || !hThread)
 return(FALSE);

 return(CXPlatThread::IsThreadActive(hThread));
};


// Return the socket address

struct sockaddr_in CConnectionSocket::GetSocketAddress()
{
 return(sAddr);
};