0
我真的不明白這裏出了什麼問題，所以我希望有人會發現我錯過的東西。套接字在HPUX上沒有收到完整的數據
我正在寫一個用戶守護進程，它接受我在java中開發的客戶端。目前此客戶端只連接併發送用戶名密碼。我開發了cygwin下的代碼，它在那裏工作。守護進程發送它的介紹，然後客戶端發送用戶名和密碼，守護進程響應斷開客戶端或發送確定（尚未完成）。
當我使用cygwin測試它時，它在localhost上工作。我將代碼移植到HPUX，客戶端可以連接並從守護進程接收引入。現在，當客戶端發送它的用戶名和密碼時，它不再起作用。守護進程只接收一個字節，當它試圖再次讀取時，我得到-1作爲EAGAIN的結果，沒有別的。客戶端不顯示任何錯誤，並且在守護程序端也沒有。當我用gdb執行代碼時，消息完全恢復。 :(
，我用的是這樣的代碼，如果需要更多的信息，我可以添加：
int TCPSocket::receive(SocketMessage &oMessage, int nBytes)
{
int max = getSendBufferSize();
if(nBytes != -1 && nBytes < max)
max = nBytes;
SocketMessage sb(max);
int rcv_bytes = 0;
int total = 0;
int error = 0;
while(1)
{
rcv_bytes = ::recv(getSocketId(), &sb[0], sb.size(), 0);
error = errno;
FILE_LOG(logDEBUG4) << "Received on socket: " << mSocketId << " bytes: " << rcv_bytes << " expected:" << sb.size() << " total: " << total << " errno: " << error;
if(rcv_bytes == -1)
{
if(error == EAGAIN || error == EWOULDBLOCK)
return total;
throw SocketException(this, SocketException::RECEIVE, error, "Socket", "Client connection error!", __FILE__, __LINE__);
}
//if(rcv_bytes == 0)
// throw SocketException(this, SocketException::RECEIVE, error, "Socket", "Client connection has been closed!");
total += rcv_bytes;
oMessage.insert(oMessage.end(), sb.begin(), sb.begin()+rcv_bytes);
}
return total;
}
日誌的輸出是這樣的：
16:16:04.391 DEBUG4: Received on socket: 4 bytes: 1 expected:32768 total: 0 errno: 2
16:16:04.391 DEBUG4: Received on socket: 4 bytes: -1 expected:32768 total: 1 errno: 11
那麼，是在30個字節的休息，爲什麼不回來了？
UPDATE
這只是實際接收數據的代碼的一部分。套接字類本身只處理沒有任何協議的原始套接字。該協議在一個單獨的類中實現。這個接收函數應該獲取網絡上可用的字節數，並將其放入緩衝區（SocketMessage）。不管是字節數是多個消息還是隻有一個消息的一部分，因爲控制類將構建（可能）部分消息流中的實際消息塊。因此，第一次調用只接收一個字節不是問題，因爲如果消息未完成，調用程序將一直等待，直到更多數據到達並且消息完成。因爲可以有多個客戶端使用非阻塞套接字。我不想處理單獨的線程，所以我的服務器複用了連接。 這裏的問題是，接收只接收一個字節，而我知道應該有更多。處理錯誤代碼EAGAIN，並在接下來輸入接收時，它應該得到更多的字節。即使網絡只傳輸一個字節，其餘的信息仍然會到達下一個，但事實並非如此。在socket上等待接收數據塊的select，就好像這裏沒有任何東西。當我在dbg中運行相同的代碼並逐步完成工作。 當我再次連接到同一個客戶端時，突然更多的字節被恢復。 當我使用本地代碼與cygwin相同的代碼，它工作正常。
UPDATE
下面是完整的代碼。
Main.cpp
mServerSocket = new TCPSocket(getListeningPort());
mServerSocket->bindSocket();
mServerSocket->setSocketBlocking(false);
mServerSocket->listenToClient(0);
setupSignals();
while(isSIGTERM() == false)
{
try
{
max = prepareListening();
//memset(&ts, 0, sizeof(struct timespec));
pret = pselect(max+1, &mReaders, &mWriters, &mExceptions, NULL, &mSignalMask);
error_code = errno;
if(pret == 0)
{
// Timeout occured, but we are not interested in that for now.
// Currently this shouldn't happen anyway.
continue;
}
processRequest(pret, error_code);
}
catch (SocketException &excp)
{
removeClientConnection(findClientConnection(excp.getTCPSocket()));
}
} // while sigTERM
BaseSocket.cpp:
#ifdef UNIX
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#endif
#include "support/exceptions/socket_exception.h"
#include "support/logging/simple_log.h"
#include "support/network/base_socket.h"
using namespace std;
BaseSocket::BaseSocket(void)
{
mSocketId = -1;
mSendBufferSize = MAX_SEND_LEN;
}
BaseSocket::BaseSocket(int pNumber)
{
mSocketId = -1;
mPortNumber = pNumber;
mBlocking = 1;
mSendBufferSize = MAX_SEND_LEN;
try
{
if ((mSocketId = ::socket(AF_INET, SOCK_STREAM, 0)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::CONSTRUCTOR, errno, "Socket", "unix: error in socket constructor", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
/*
set the initial address of client that shall be communicated with to
any address as long as they are using the same port number.
The clientAddr structure is used in the future for storing the actual
address of client applications with which communication is going
to start
*/
mClientAddr.sin_family = AF_INET;
mClientAddr.sin_addr.s_addr = htonl(INADDR_ANY);
mClientAddr.sin_port = htons(mPortNumber);
updateSendBufferSize(MAX_SEND_LEN);
}
void BaseSocket::updateSendBufferSize(int nNewSize)
{
mSendBufferSize = getSendBufferSize();
if(mSendBufferSize > nNewSize)
mSendBufferSize = nNewSize;
}
BaseSocket::~BaseSocket(void)
{
close();
}
void BaseSocket::setSocketId(int socketFd)
{
mSocketId = socketFd;
}
int BaseSocket::getSocketId()
{
return mSocketId;
}
// returns the port number
int BaseSocket::getPortNumber()
{
return mPortNumber;
}
void BaseSocket::setDebug(int debugToggle)
{
try
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_DEBUG, (char *) &debugToggle, sizeof(debugToggle)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set debug", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
void BaseSocket::setReuseAddr(int reuseToggle)
{
try
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_REUSEADDR, (char *) &reuseToggle,
sizeof(reuseToggle)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set reuse address", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
void BaseSocket::setKeepAlive(int aliveToggle)
{
try
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_KEEPALIVE, (char *) &aliveToggle,
sizeof(aliveToggle)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set keep alive", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
void BaseSocket::setLingerSeconds(int seconds)
{
struct linger lingerOption;
if (seconds > 0)
{
lingerOption.l_linger = seconds;
lingerOption.l_onoff = 1;
}
else
lingerOption.l_onoff = 0;
try
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_LINGER, (char *) &lingerOption,
sizeof(struct linger)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set linger seconds", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
void BaseSocket::setLingerOnOff(bool lingerOn)
{
struct linger lingerOption;
if (lingerOn)
lingerOption.l_onoff = 1;
else
lingerOption.l_onoff = 0;
try
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_LINGER, (char *) &lingerOption,
sizeof(struct linger)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set linger on/off", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
void BaseSocket::setSendBufferSize(int sendBufSize)
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_SNDBUF, (char *) &sendBufSize, sizeof(sendBufSize)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error send buffer size", __FILE__, __LINE__);
#endif
}
updateSendBufferSize(sendBufSize);
}
void BaseSocket::setReceiveBufferSize(int receiveBufSize)
{
if (setsockopt(mSocketId, SOL_SOCKET, SO_RCVBUF, (char *) &receiveBufSize, sizeof(receiveBufSize)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set receive buffer size", __FILE__, __LINE__);
#endif
}
}
int BaseSocket::isSocketBlocking()
{
return mBlocking;
}
void BaseSocket::setSocketBlocking(int blockingToggle)
{
if (blockingToggle)
{
if (isSocketBlocking())
return;
else
mBlocking = 1;
}
else
{
if (!isSocketBlocking())
return;
else
mBlocking = 0;
}
try
{
#ifdef UNIX
int flags;
if (-1 == (flags = fcntl(mSocketId, F_GETFL, 0)))
flags = 0;
if(mBlocking)
fcntl(mSocketId, F_SETFL, flags & (~O_NONBLOCK));
else
fcntl(mSocketId, F_SETFL, flags | O_NONBLOCK);
/*if (ioctl(socketId, FIONBIO, (char *) &blocking) == -1)
{
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error set socke blocking");
}*/
#endif
}
catch (SocketException& excp)
{
excp.response();
}
}
int BaseSocket::getDebug()
{
int myOption;
int myOptionLen = sizeof(myOption);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_DEBUG, (void *) &myOption, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get debug", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return -1;
}
return myOption;
}
int BaseSocket::getReuseAddr()
{
int myOption;
int myOptionLen = sizeof(myOption);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_REUSEADDR, (void *) &myOption, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get reuse address", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return -1;
}
return myOption;
}
int BaseSocket::getKeepAlive()
{
int myOption;
int myOptionLen = sizeof(myOption);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_KEEPALIVE, (void *) &myOption, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get keep alive", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return -1;
}
return myOption;
}
int BaseSocket::getLingerSeconds()
{
struct linger lingerOption;
int myOptionLen = sizeof(struct linger);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_LINGER, (void *) &lingerOption, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get linger seconds", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return -1;
}
return lingerOption.l_linger;
}
bool BaseSocket::getLingerOnOff()
{
struct linger lingerOption;
int myOptionLen = sizeof(struct linger);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_LINGER, (void *) &lingerOption, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get linger on/off", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
if (lingerOption.l_onoff == 1)
return true;
else
return false;
}
int BaseSocket::getSendBufferSize()
{
int sendBuf;
int myOptionLen = sizeof(sendBuf);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_SNDBUF, (void *)&sendBuf, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get send buffer size", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return -1;
}
return sendBuf;
}
int BaseSocket::getReceiveBufferSize()
{
int rcvBuf;
int myOptionLen = sizeof(rcvBuf);
try
{
if (getsockopt(mSocketId, SOL_SOCKET, SO_RCVBUF, (void *) &rcvBuf, &myOptionLen) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::OPTION, errno, "Socket", "unix: error get receive buffer size", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return -1;
}
return rcvBuf;
}
ostream &operator<<(ostream& io, BaseSocket& s)
{
string flagStr = "";
io << endl;
io << "Summary of socket settings:" << endl;
io << " Socket Id: " << s.getSocketId() << endl;
io << " port #: " << s.getPortNumber() << endl;
io << " debug: " << (flagStr = s.getDebug() ? "true" : "false")
<< endl;
io << " reuse addr: " << (flagStr = s.getReuseAddr() ? "true" : "false")
<< endl;
io << " keep alive: " << (flagStr = s.getKeepAlive() ? "true" : "false")
<< endl;
io << " send buf size: " << s.getSendBufferSize() << endl;
io << " recv bug size: " << s.getReceiveBufferSize() << endl;
io << " blocking: "
<< (flagStr = s.isSocketBlocking() ? "true" : "false") << endl;
io << " linger on: "
<< (flagStr = s.getLingerOnOff() ? "true" : "false") << endl;
io << " linger seconds: " << s.getLingerSeconds() << endl;
io << endl;
return io;
}
void BaseSocket::close(void)
{
::close(mSocketId);
}
TCPSocket.cpp:
#ifdef UNIX
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#endif
#include <sstream>
#include "support/logging/log.h"
#include "support/exceptions/socket_exception.h"
#include "support/logging/simple_log.h"
#include "support/network/tcp_socket.h"
using namespace std;
const int MSG_HEADER_LEN = 6;
TCPSocket::TCPSocket()
: BaseSocket()
{
}
TCPSocket::TCPSocket(int portId)
: BaseSocket(portId)
{
}
TCPSocket::~TCPSocket()
{
}
void TCPSocket::initialize()
{
}
void TCPSocket::bindSocket()
{
try
{
if (bind(mSocketId, (struct sockaddr *) &mClientAddr, sizeof(struct sockaddr_in)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::BIND, 0, "Socket", "unix: error calling bind()", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
void TCPSocket::connectToServer(string& serverNameOrAddr, hostType hType)
{
/*
when this method is called, a client socket has been built already,
so we have the socketId and portNumber ready.
a HostInfo instance is created, no matter how the server's name is
given (such as www.yuchen.net) or the server's address is given (such
as 169.56.32.35), we can use this HostInfo instance to get the
IP address of the server
*/
HostInfo serverInfo(serverNameOrAddr, hType);
// Store the IP address and socket port number
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = inet_addr(
serverInfo.getHostIPAddress().c_str());
serverAddress.sin_port = htons(mPortNumber);
// Connect to the given address
try
{
if (connect(mSocketId, (struct sockaddr *) &serverAddress, sizeof(serverAddress)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::CONNECT, 0, "Socket", "unix: error calling connect()", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
TCPSocket *TCPSocket::acceptClient(string& clientHost)
{
int newSocket; // the new socket file descriptor returned by the accept system call
// the length of the client's address
int clientAddressLen = sizeof(struct sockaddr_in);
struct sockaddr_in clientAddress; // Address of the client that sent data
// Accepts a new client connection and stores its socket file descriptor
try
{
if ((newSocket = accept(mSocketId, (struct sockaddr *) &clientAddress, &clientAddressLen)) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::ACCEPT, 0, "Socket", "unix: error calling accept()", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
return NULL;
}
// Get the host name given the address
char *sAddress = inet_ntoa((struct in_addr) clientAddress.sin_addr);
HostInfo clientInfo(sAddress, ADDRESS);
clientHost += clientInfo.getHostName();
// Create and return the new TCPSocket object
TCPSocket* retSocket = new TCPSocket();
retSocket->setSocketId(newSocket);
return retSocket;
}
void TCPSocket::listenToClient(int totalNumPorts)
{
try
{
if (listen(mSocketId, totalNumPorts) == -1)
{
#ifdef UNIX
throw SocketException(this, SocketException::LISTEN, 0, "Socket", "unix: error calling listen()", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
}
ostream &operator<<(ostream &oStream, const TCPSocket &oSocket)
{
oStream << oSocket.mSocketId;
return oStream;
}
int TCPSocket::send(SocketMessage const &oBuffer, int nSize)
{
int numBytes; // the number of bytes sent
int error = errno;
if(nSize == -1)
nSize = oBuffer.size();
if((unsigned int)nSize > oBuffer.size())
{
std::stringstream ss;
ss << "Invalid Buffersize! Requested: " << (unsigned int)nSize << " Provided: " << oBuffer.size();
std::string s;
ss >> s;
FILE_LOG(logERROR) << s;
throw SocketException(this, SocketException::SEND, 0, "Socket", s, __FILE__, __LINE__);
}
// Sends the message to the connected host
try
{
FILE_LOG(logDEBUG4) << "Sending on socket: "<< mSocketId << " bytes:" << nSize;
numBytes = ::send(mSocketId, &oBuffer[0], nSize, 0);
error = errno;
FILE_LOG(logDEBUG4) << "Sent on socket: "<< mSocketId << " bytes:" << nSize << " errno: " << error;
if(numBytes == -1)
{
#ifdef UNIX
if(error == EAGAIN || error == EWOULDBLOCK)
{
return -1;
}
else
throw SocketException(this, SocketException::SEND, error, "Socket", "unix: error calling send()", __FILE__, __LINE__);
#endif
}
}
catch (SocketException& excp)
{
excp.response();
}
return numBytes;
}
int TCPSocket::receive(SocketMessage &oMessage, int nBytes)
{
int max = getSendBufferSize();
if(nBytes != -1 && nBytes < max)
max = nBytes;
SocketMessage sb(max);
int rcv_bytes = 0;
int total = 0;
int error = 0;
while(1)
{
rcv_bytes = ::recv(getSocketId(), &sb[0], sb.size(), 0);
error = errno;
FILE_LOG(logDEBUG4) << "Received on socket: " << getSocketId() << " bytes: " << rcv_bytes << " expected:" << sb.size() << " total: " << total << " errno: " << error;
if(rcv_bytes == -1)
{
if(error == EAGAIN || error == EWOULDBLOCK)
return total;
throw SocketException(this, SocketException::RECEIVE, error, "Socket", "Client connection error!", __FILE__, __LINE__);
}
// Socket has been closed.
if(rcv_bytes == 0)
return total;
total += rcv_bytes;
oMessage.insert(oMessage.end(), sb.begin(), sb.begin()+rcv_bytes);
}
return total;
}
void TCPSocket::close(void)
{
BaseSocket::close();
}
psst，那不是c – 2013-05-14 14:28:22
問題是關於底層的C函數，即在套接字上接收，而不是關於C++特性。 – Devolus 2013-05-14 14:30:16
當然，但如果將其標記爲C++，則可以獲得更多幫助。 – 2013-05-14 14:32:06