barrier/net/CTCPSocket.cpp

322 lines
6.3 KiB
C++
Raw Normal View History

2001-10-06 14:13:28 +00:00
#include "CTCPSocket.h"
#include "CBufferedInputStream.h"
#include "CBufferedOutputStream.h"
#include "CNetworkAddress.h"
#include "XIO.h"
#include "XSocket.h"
#include "CCondVar.h"
2001-10-06 14:13:28 +00:00
#include "CLock.h"
#include "CMutex.h"
#include "CThread.h"
#include "CStopwatch.h"
#include "TMethodJob.h"
2001-10-06 14:13:28 +00:00
//
// CTCPSocket
//
CTCPSocket::CTCPSocket()
2001-10-06 14:13:28 +00:00
{
m_fd = CNetwork::socket(PF_INET, SOCK_STREAM, 0);
if (m_fd == CNetwork::Null) {
2001-10-06 14:13:28 +00:00
throw XSocketCreate();
}
init();
}
CTCPSocket::CTCPSocket(CNetwork::Socket fd) :
m_fd(fd)
2001-10-06 14:13:28 +00:00
{
assert(m_fd != CNetwork::Null);
2001-10-06 14:13:28 +00:00
init();
// socket starts in connected state
m_connected = kReadWrite;
// start handling socket
m_thread = new CThread(new TMethodJob<CTCPSocket>(
2001-10-21 00:21:02 +00:00
this, &CTCPSocket::ioThread));
2001-10-06 14:13:28 +00:00
}
CTCPSocket::~CTCPSocket()
{
try {
close();
}
catch (...) {
// ignore failures
}
// clean up
delete m_input;
delete m_output;
2001-10-21 00:21:02 +00:00
delete m_mutex;
2001-10-06 14:13:28 +00:00
}
void
2002-06-17 13:31:21 +00:00
CTCPSocket::bind(const CNetworkAddress& addr)
2001-10-06 14:13:28 +00:00
{
if (CNetwork::bind(m_fd, addr.getAddress(),
addr.getAddressLength()) == CNetwork::Error) {
if (errno == CNetwork::kEADDRINUSE) {
2001-10-06 14:13:28 +00:00
throw XSocketAddressInUse();
}
throw XSocketBind();
}
}
void
CTCPSocket::close()
2001-10-06 14:13:28 +00:00
{
// see if buffers should be flushed
bool doFlush = false;
{
CLock lock(m_mutex);
doFlush = (m_thread != NULL && (m_connected & kWrite) != 0);
}
// flush buffers
if (doFlush) {
m_output->flush();
}
// cause ioThread to exit
{
CLock lock(m_mutex);
if (m_fd != CNetwork::Null) {
CNetwork::shutdown(m_fd, 2);
m_connected = kClosed;
2001-10-06 14:13:28 +00:00
}
}
2001-10-06 14:13:28 +00:00
// wait for thread
if (m_thread != NULL) {
2001-10-06 14:13:28 +00:00
m_thread->wait();
delete m_thread;
m_thread = NULL;
}
// close socket
if (m_fd != CNetwork::Null) {
if (CNetwork::close(m_fd) == CNetwork::Error) {
2001-10-06 14:13:28 +00:00
throw XIOClose();
}
m_fd = CNetwork::Null;
2001-10-06 14:13:28 +00:00
}
}
void
2002-06-17 13:31:21 +00:00
CTCPSocket::connect(const CNetworkAddress& addr)
{
// connect asynchronously so we can check for cancellation
CNetwork::setblocking(m_fd, false);
if (CNetwork::connect(m_fd, addr.getAddress(),
addr.getAddressLength()) == CNetwork::Error) {
// check for failure
if (CNetwork::getsockerror() != CNetwork::kECONNECTING) {
XSocketConnect e;
CNetwork::setblocking(m_fd, true);
throw e;
}
// wait for connection or failure
CNetwork::PollEntry pfds[1];
pfds[0].fd = m_fd;
pfds[0].events = CNetwork::kPOLLOUT;
for (;;) {
CThread::testCancel();
const int status = CNetwork::poll(pfds, 1, 10);
if (status > 0) {
if ((pfds[0].revents & (CNetwork::kPOLLERR |
CNetwork::kPOLLNVAL)) != 0) {
// connection failed
int error = 0;
CNetwork::AddressLength size = sizeof(error);
CNetwork::setblocking(m_fd, true);
CNetwork::getsockopt(m_fd, SOL_SOCKET, SO_ERROR,
reinterpret_cast<char*>(&error), &size);
throw XSocketConnect(error);
}
if ((pfds[0].revents & CNetwork::kPOLLOUT) != 0) {
int error;
CNetwork::AddressLength size = sizeof(error);
if (CNetwork::getsockopt(m_fd, SOL_SOCKET, SO_ERROR,
reinterpret_cast<char*>(&error),
&size) == CNetwork::Error ||
error != 0) {
// connection failed
CNetwork::setblocking(m_fd, true);
throw XSocketConnect(error);
}
// connected!
break;
}
}
}
}
// back to blocking
CNetwork::setblocking(m_fd, true);
// start servicing the socket
m_connected = kReadWrite;
m_thread = new CThread(new TMethodJob<CTCPSocket>(
this, &CTCPSocket::ioThread));
}
IInputStream*
CTCPSocket::getInputStream()
2001-10-06 14:13:28 +00:00
{
return m_input;
}
IOutputStream*
CTCPSocket::getOutputStream()
2001-10-06 14:13:28 +00:00
{
return m_output;
}
void
CTCPSocket::init()
2001-10-06 14:13:28 +00:00
{
m_mutex = new CMutex;
m_thread = NULL;
m_connected = kClosed;
m_input = new CBufferedInputStream(m_mutex,
new TMethodJob<CTCPSocket>(
this, &CTCPSocket::closeInput));
m_output = new CBufferedOutputStream(m_mutex,
new TMethodJob<CTCPSocket>(
this, &CTCPSocket::closeOutput));
// turn off Nagle algorithm. we send lots of very short messages
// that should be sent without (much) delay. for example, the
// mouse motion messages are much less useful if they're delayed.
CNetwork::TCPNoDelayType flag = 1;
CNetwork::setsockopt(m_fd, SOL_TCP, TCP_NODELAY, &flag, sizeof(flag));
2001-10-06 14:13:28 +00:00
}
void
CTCPSocket::ioThread(void*)
2001-10-21 00:21:02 +00:00
{
try {
ioService();
2002-05-30 16:11:59 +00:00
ioCleanup();
2001-10-21 00:21:02 +00:00
}
catch (...) {
2002-05-30 16:11:59 +00:00
ioCleanup();
throw;
}
}
void
CTCPSocket::ioCleanup()
2002-05-30 16:11:59 +00:00
{
try {
2001-10-21 00:21:02 +00:00
m_input->close();
2002-05-30 16:11:59 +00:00
}
catch (...) {
// ignore
}
try {
2001-10-21 00:21:02 +00:00
m_output->close();
2002-05-30 16:11:59 +00:00
}
catch (...) {
// ignore
2001-10-21 00:21:02 +00:00
}
}
void
CTCPSocket::ioService()
2001-10-06 14:13:28 +00:00
{
assert(m_fd != CNetwork::Null);
2001-10-06 14:13:28 +00:00
// now service the connection
CNetwork::PollEntry pfds[1];
2001-10-06 14:13:28 +00:00
pfds[0].fd = m_fd;
for (;;) {
{
// choose events to poll for
CLock lock(m_mutex);
pfds[0].events = 0;
2002-05-30 16:11:59 +00:00
if (m_connected == 0) {
return;
}
2001-10-06 14:13:28 +00:00
if ((m_connected & kRead) != 0) {
// still open for reading
pfds[0].events |= CNetwork::kPOLLIN;
2001-10-06 14:13:28 +00:00
}
if ((m_connected & kWrite) != 0 && m_output->getSize() > 0) {
// data queued for writing
pfds[0].events |= CNetwork::kPOLLOUT;
2001-10-06 14:13:28 +00:00
}
}
// check for status
const int status = CNetwork::poll(pfds, 1, 10);
2001-10-06 14:13:28 +00:00
// transfer data and handle errors
if (status == 1) {
if ((pfds[0].revents & (CNetwork::kPOLLERR |
CNetwork::kPOLLNVAL)) != 0) {
2001-10-06 14:13:28 +00:00
// stream is no good anymore so bail
m_input->hangup();
return;
}
// read some data
if (pfds[0].revents & CNetwork::kPOLLIN) {
2001-10-06 14:13:28 +00:00
UInt8 buffer[4096];
ssize_t n = CNetwork::read(m_fd, buffer, sizeof(buffer));
2001-10-06 14:13:28 +00:00
if (n > 0) {
CLock lock(m_mutex);
m_input->write(buffer, n);
}
else if (n == 0) {
// stream hungup
m_input->hangup();
2002-05-30 16:11:59 +00:00
m_connected &= ~kRead;
2001-10-06 14:13:28 +00:00
}
}
// write some data
if (pfds[0].revents & CNetwork::kPOLLOUT) {
2001-10-06 14:13:28 +00:00
CLock lock(m_mutex);
// get amount of data to write
UInt32 n = m_output->getSize();
// write data
const void* buffer = m_output->peek(n);
n = (UInt32)CNetwork::write(m_fd, buffer, n);
2001-10-06 14:13:28 +00:00
// discard written data
if (n > 0) {
m_output->pop(n);
}
else if (n == (UInt32)-1 && CNetwork::getsockerror() == EPIPE) {
2001-10-21 00:21:02 +00:00
return;
}
2001-10-06 14:13:28 +00:00
}
}
}
}
void
CTCPSocket::closeInput(void*)
2001-10-06 14:13:28 +00:00
{
// note -- m_mutex should already be locked
CNetwork::shutdown(m_fd, 0);
2001-10-06 14:13:28 +00:00
m_connected &= ~kRead;
}
void
CTCPSocket::closeOutput(void*)
2001-10-06 14:13:28 +00:00
{
// note -- m_mutex should already be locked
CNetwork::shutdown(m_fd, 1);
2001-10-06 14:13:28 +00:00
m_connected &= ~kWrite;
}