Bart/bullet3
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

use TCP in main thread, accumulate bytes until size matches.

Browse Source
This commit is contained in:
Erwin Coumans
2017-02-20 18:54:12 -08:00
parent 3d73a9d788
commit 942015df9d
3 changed files with 194 additions and 413 deletions
Download Patch File Download Diff File Expand all files Collapse all files

405
examples/SharedMemory/PhysicsClientTCP.cpp
View File

@@ -10,45 +10,7 @@
#include "SharedMemoryCommands.h" #include "SharedMemoryCommands.h"
#include <string> #include <string>
#include "Bullet3Common/b3Logging.h" #include "Bullet3Common/b3Logging.h"
#include "../MultiThreading/b3ThreadSupportInterface.h" #include "Bullet3Common/b3AlignedObjectArray.h"
void TCPThreadFunc(void* userPtr, void* lsMemory);
void* TCPlsMemoryFunc();
bool gVerboseNetworkMessagesClient2 = false;
#ifndef _WIN32
#include "../MultiThreading/b3PosixThreadSupport.h"
b3ThreadSupportInterface* createTCPThreadSupport(int numThreads)
{
b3PosixThreadSupport::ThreadConstructionInfo constructionInfo("TCPThread",
TCPThreadFunc,
TCPlsMemoryFunc,
numThreads);
b3ThreadSupportInterface* threadSupport = new b3PosixThreadSupport(constructionInfo);
return threadSupport;
}
#elif defined( _WIN32)
#include "../MultiThreading/b3Win32ThreadSupport.h"
b3ThreadSupportInterface* createTCPThreadSupport(int numThreads)
{
b3Win32ThreadSupport::Win32ThreadConstructionInfo threadConstructionInfo("TCPThread", TCPThreadFunc, TCPlsMemoryFunc, numThreads);
b3Win32ThreadSupport* threadSupport = new b3Win32ThreadSupport(threadConstructionInfo);
return threadSupport;
}
#endif
struct TCPThreadLocalStorage
{
int threadId;
};
@@ -57,6 +19,8 @@ unsigned int b3DeserializeInt2(const unsigned char* input)
unsigned int tmp = (input[3] << 24) + (input[2] << 16) + (input[1] << 8) + input[0]; unsigned int tmp = (input[3] << 24) + (input[2] << 16) + (input[1] << 8) + input[0];
return tmp; return tmp;
} }
bool gVerboseNetworkMessagesClient2 = false;
struct TcpNetworkedInternalData struct TcpNetworkedInternalData
{ {
@@ -70,29 +34,24 @@ struct TcpNetworkedInternalData
bool m_isConnected; bool m_isConnected;
b3ThreadSupportInterface* m_threadSupport;
b3CriticalSection* m_cs;
TcpNetworkedInternalData* m_tcpInternalData; TcpNetworkedInternalData* m_tcpInternalData;
SharedMemoryCommand m_clientCmd; SharedMemoryCommand m_clientCmd;
bool m_hasCommand; bool m_hasCommand;
bool m_hasStatus;
SharedMemoryStatus m_lastStatus; SharedMemoryStatus m_lastStatus;
b3AlignedObjectArray<char> m_stream; b3AlignedObjectArray<char> m_stream;
std::string m_hostName; std::string m_hostName;
int m_port; int m_port;
b3AlignedObjectArray<unsigned char> m_tempBuffer;
TcpNetworkedInternalData() TcpNetworkedInternalData()
: :
m_isConnected(false), m_isConnected(false),
m_threadSupport(0), m_hasCommand(false)
m_hasCommand(false),
m_hasStatus(false)
{ {
} }
@@ -105,8 +64,7 @@ struct TcpNetworkedInternalData
m_tcpSocket.Initialize(); m_tcpSocket.Initialize();
m_isConnected = m_tcpSocket.Open(m_hostName.c_str(),m_port); m_isConnected = m_tcpSocket.Open(m_hostName.c_str(),m_port);
m_isConnected = true;
return m_isConnected; return m_isConnected;
} }
@@ -117,203 +75,66 @@ struct TcpNetworkedInternalData
//int serviceResult = enet_host_service(m_client, &m_event, 0); //int serviceResult = enet_host_service(m_client, &m_event, 0);
int maxLen = 4 + sizeof(SharedMemoryStatus)+SHARED_MEMORY_MAX_STREAM_CHUNK_SIZE; int maxLen = 4 + sizeof(SharedMemoryStatus)+SHARED_MEMORY_MAX_STREAM_CHUNK_SIZE;
int recBytes = m_tcpSocket.Receive(maxLen); int rBytes = m_tcpSocket.Receive(maxLen);
if (rBytes<=0)
return false;
if (gVerboseNetworkMessagesClient2) //append to tmp buffer
{ //recBytes
printf("A packet of length %d bytes received\n", recBytes);
}
unsigned char* data = (unsigned char*)m_tcpSocket.GetData();
int packetSizeInBytes = b3DeserializeInt2(data);
if (packetSizeInBytes == recBytes) unsigned char* d2 = (unsigned char*)m_tcpSocket.GetData();
{
SharedMemoryStatus* statPtr = (SharedMemoryStatus*)&data[4];
if (statPtr->m_type == CMD_STEP_FORWARD_SIMULATION_COMPLETED)
{
SharedMemoryStatus dummy;
dummy.m_type = CMD_STEP_FORWARD_SIMULATION_COMPLETED;
m_lastStatus = dummy;
m_stream.resize(0);
}
else
{
m_lastStatus = *statPtr; int curSize = m_tempBuffer.size();
int streamOffsetInBytes = 4 + sizeof(SharedMemoryStatus); m_tempBuffer.resize(curSize+rBytes);
int numStreamBytes = packetSizeInBytes - streamOffsetInBytes; for (int i=0;i<rBytes;i++)
m_stream.resize(numStreamBytes); {
for (int i = 0; i < numStreamBytes; i++) m_tempBuffer[curSize+i] = d2[i];
{ }
m_stream[i] = data[i + streamOffsetInBytes];
} int packetSizeInBytes = -1;
}
} if (m_tempBuffer.size()>=4)
else {
{ packetSizeInBytes = b3DeserializeInt2(&m_tempBuffer[0]);
printf("unknown status message received\n"); }
}
if (m_tempBuffer.size() == packetSizeInBytes)
{
unsigned char* data = &m_tempBuffer[0];
if (gVerboseNetworkMessagesClient2)
{
printf("A packet of length %d bytes received\n", m_tempBuffer.size());
}
hasStatus = true;
SharedMemoryStatus* statPtr = (SharedMemoryStatus*)&data[4];
if (statPtr->m_type == CMD_STEP_FORWARD_SIMULATION_COMPLETED)
{
SharedMemoryStatus dummy;
dummy.m_type = CMD_STEP_FORWARD_SIMULATION_COMPLETED;
m_lastStatus = dummy;
m_stream.resize(0);
}
else
{
m_lastStatus = *statPtr;
int streamOffsetInBytes = 4 + sizeof(SharedMemoryStatus);
int numStreamBytes = packetSizeInBytes - streamOffsetInBytes;
m_stream.resize(numStreamBytes);
for (int i = 0; i < numStreamBytes; i++)
{
m_stream[i] = data[i + streamOffsetInBytes];
}
}
m_tempBuffer.clear();
}
return hasStatus; return hasStatus;
} }
}; };
enum TCPThreadEnums
{
eTCPRequestTerminate = 13,
eTCPIsUnInitialized,
eTCPIsInitialized,
eTCPInitializationFailed,
eTCPHasTerminated
};
enum TCPCommandEnums
{
eTCPIdle = 13,
eTCP_ConnectRequest,
eTCP_Connected,
eTCP_ConnectionFailed,
eTCP_DisconnectRequest,
eTCP_Disconnected,
};
void TCPThreadFunc(void* userPtr, void* lsMemory)
{
printf("TCPThreadFunc thread started\n");
TcpNetworkedInternalData* args = (TcpNetworkedInternalData*)userPtr;
// int workLeft = true;
b3Clock clock;
clock.reset();
bool init = true;
if (init)
{
args->m_cs->lock();
args->m_cs->setSharedParam(0, eTCPIsInitialized);
args->m_cs->unlock();
double deltaTimeInSeconds = 0;
do
{
b3Clock::usleep(0);
deltaTimeInSeconds += double(clock.getTimeMicroseconds()) / 1000000.;
{
clock.reset();
deltaTimeInSeconds = 0.f;
switch (args->m_cs->getSharedParam(1))
{
case eTCP_ConnectRequest:
{
bool connected = args->connectTCP();
if (connected)
{
args->m_cs->setSharedParam(1, eTCP_Connected);
}
else
{
args->m_cs->setSharedParam(1, eTCP_ConnectionFailed);
}
break;
}
default:
{
}
};
if (args->m_isConnected)
{
args->m_cs->lock();
bool hasCommand = args->m_hasCommand;
args->m_cs->unlock();
if (hasCommand)
{
int sz = 0;
unsigned char* data = 0;
if (args->m_clientCmd.m_type == CMD_STEP_FORWARD_SIMULATION)
{
sz = sizeof(int);
data = (unsigned char*) &args->m_clientCmd.m_type;
}
else
{
sz = sizeof(SharedMemoryCommand);
data = (unsigned char*)&args->m_clientCmd;
}
int res;
args->m_tcpSocket.Send((const uint8 *)data,sz);
args->m_cs->lock();
args->m_hasCommand = false;
args->m_cs->unlock();
}
bool hasNewStatus = args->checkData();
if (hasNewStatus)
{
if (args->m_hasStatus)
{
//overflow: last status hasn't been processed yet
b3Assert(0);
printf("Error: received new status but previous status not processed yet");
}
else
{
args->m_cs->lock();
args->m_hasStatus = hasNewStatus;
args->m_cs->unlock();
}
}
}
}
} while (args->m_cs->getSharedParam(0) != eTCPRequestTerminate);
}
else
{
args->m_cs->lock();
args->m_cs->setSharedParam(0, eTCPInitializationFailed);
args->m_cs->unlock();
}
printf("finished\n");
}
void* TCPlsMemoryFunc()
{
//don't create local store memory, just return 0
return new TCPThreadLocalStorage;
}
TcpNetworkedPhysicsProcessor::TcpNetworkedPhysicsProcessor(const char* hostName, int port) TcpNetworkedPhysicsProcessor::TcpNetworkedPhysicsProcessor(const char* hostName, int port)
{ {
@@ -338,49 +159,42 @@ bool TcpNetworkedPhysicsProcessor::processCommand(const struct SharedMemoryComma
{ {
printf("PhysicsClientTCP::processCommand\n"); printf("PhysicsClientTCP::processCommand\n");
} }
// int sz = sizeof(SharedMemoryCommand);
int timeout = 1024 * 1024 * 1024;
m_data->m_cs->lock();
m_data->m_clientCmd = clientCmd;
m_data->m_hasCommand = true;
m_data->m_cs->unlock();
while (m_data->m_hasCommand && (timeout-- > 0))
{ {
b3Clock::usleep(0); int sz = 0;
} unsigned char* data = 0;
m_data->m_tempBuffer.clear();
#if 0 if (clientCmd.m_type == CMD_STEP_FORWARD_SIMULATION)
{
sz = sizeof(int);
data = (unsigned char*) &clientCmd.m_type;
}
else
{
sz = sizeof(SharedMemoryCommand);
data = (unsigned char*)&clientCmd;
}
int res;
m_data->m_tcpSocket.Send((const uint8 *)data,sz);
timeout = 1024 * 1024 * 1024; }
bool hasStatus = false;
const SharedMemoryStatus* stat = 0;
while ((!hasStatus) && (timeout-- > 0))
{
hasStatus = receiveStatus(serverStatusOut, bufferServerToClient, bufferSizeInBytes);
b3Clock::usleep(100);
}
return hasStatus;
#endif
return false; return false;
} }
bool TcpNetworkedPhysicsProcessor::receiveStatus(struct SharedMemoryStatus& serverStatusOut, char* bufferServerToClient, int bufferSizeInBytes) bool TcpNetworkedPhysicsProcessor::receiveStatus(struct SharedMemoryStatus& serverStatusOut, char* bufferServerToClient, int bufferSizeInBytes)
{ {
bool hasStatus = false; bool hasStatus = m_data->checkData();
if (m_data->m_hasStatus)
if (hasStatus)
{ {
if (gVerboseNetworkMessagesClient2) if (gVerboseNetworkMessagesClient2)
{ {
printf("TcpNetworkedPhysicsProcessor::receiveStatus\n"); printf("TcpNetworkedPhysicsProcessor::receiveStatus\n");
} }
hasStatus = true;
serverStatusOut = m_data->m_lastStatus; serverStatusOut = m_data->m_lastStatus;
int numStreamBytes = m_data->m_stream.size(); int numStreamBytes = m_data->m_stream.size();
@@ -396,9 +210,6 @@ bool TcpNetworkedPhysicsProcessor::receiveStatus(struct SharedMemoryStatus& serv
printf("Error: steam buffer overflow\n"); printf("Error: steam buffer overflow\n");
} }
m_data->m_cs->lock();
m_data->m_hasStatus = false;
m_data->m_cs->unlock();
} }
@@ -427,70 +238,18 @@ bool TcpNetworkedPhysicsProcessor::isConnected() const
bool TcpNetworkedPhysicsProcessor::connect() bool TcpNetworkedPhysicsProcessor::connect()
{ {
if (m_data->m_threadSupport==0) bool isConnected = m_data->connectTCP();
{
m_data->m_threadSupport = createTCPThreadSupport(1);
m_data->m_cs = m_data->m_threadSupport->createCriticalSection();
m_data->m_cs->setSharedParam(0, eTCPIsUnInitialized);
m_data->m_threadSupport->runTask(B3_THREAD_SCHEDULE_TASK, (void*) m_data, 0);
while (m_data->m_cs->getSharedParam(0) == eTCPIsUnInitialized)
{
b3Clock::usleep(1000);
}
m_data->m_cs->lock();
m_data->m_cs->setSharedParam(1, eTCP_ConnectRequest);
m_data->m_cs->unlock();
while (m_data->m_cs->getSharedParam(1) == eTCP_ConnectRequest)
{
b3Clock::usleep(1000);
}
}
unsigned int sharedParam = m_data->m_cs->getSharedParam(1);
bool isConnected = (sharedParam == eTCP_Connected);
return isConnected; return isConnected;
} }
void TcpNetworkedPhysicsProcessor::disconnect() void TcpNetworkedPhysicsProcessor::disconnect()
{ {
if (m_data->m_threadSupport) m_data->m_tcpSocket.Close();
{ m_data->m_isConnected = false;
m_data->m_cs->lock();
m_data->m_cs->setSharedParam(0, eTCPRequestTerminate);
m_data->m_cs->unlock();
int numActiveThreads = 1;
while (numActiveThreads)
{
int arg0, arg1;
if (m_data->m_threadSupport->isTaskCompleted(&arg0, &arg1, 0))
{
numActiveThreads--;
printf("numActiveThreads = %d\n", numActiveThreads);
}
else
{
b3Clock::usleep(1000);
}
};
printf("stopping threads\n");
delete m_data->m_threadSupport;
m_data->m_threadSupport = 0;
m_data->m_isConnected = false;
}
} }

196
examples/SharedMemory/tcp/main.cpp
View File

@@ -17,7 +17,6 @@ typedef SharedMemoryCommandProcessor MyCommandProcessor;
#include "PhysicsServerCommandProcessor.h" #include "PhysicsServerCommandProcessor.h"
#include "../Utils/b3Clock.h" #include "../Utils/b3Clock.h"
#define MAX_PACKET 4096
bool gVerboseNetworkMessagesServer = true; bool gVerboseNetworkMessagesServer = true;
@@ -75,6 +74,8 @@ int main(int argc, char *argv[])
socket.Initialize(); socket.Initialize();
socket.Listen("localhost", port); socket.Listen("localhost", port);
b3AlignedObjectArray<char> bytesReceived;
while (!exitRequested) while (!exitRequested)
{ {
@@ -93,111 +94,130 @@ int main(int argc, char *argv[])
//printf("try receive\n"); //printf("try receive\n");
bool receivedData = false; bool receivedData = false;
if (pClient->Receive(MAX_PACKET)) int maxLen = 4 + sizeof(SharedMemoryStatus)+SHARED_MEMORY_MAX_STREAM_CHUNK_SIZE;
if (pClient->Receive(maxLen))
{ {
char* msg = (char*) pClient->GetData();
int numBytesRec = pClient->GetBytesReceived(); char* msg2 = (char*) pClient->GetData();
if (gVerboseNetworkMessagesServer) int numBytesRec2 = pClient->GetBytesReceived();
{
printf("received message length [%d]\n",numBytesRec); int curSize = bytesReceived.size();
} bytesReceived.resize(bytesReceived.size()+numBytesRec2);
for (int i=0;i<numBytesRec2;i++)
{
bytesReceived[curSize+i] = msg2[i];
}
if (bytesReceived.size() == 4 || bytesReceived.size()==sizeof(SharedMemoryCommand))
{
int numBytesRec = bytesReceived.size();
if (gVerboseNetworkMessagesServer)
{
printf("received message length [%d]\n",numBytesRec);
}
receivedData = true; receivedData = true;
if (strncmp(msg,"stop",4)==0) if (strncmp(&bytesReceived[0],"stop",4)==0)
{ {
printf("Stop request received\n"); printf("Stop request received\n");
exitRequested = true; exitRequested = true;
break; bytesReceived.clear();
} break;
}
SharedMemoryCommand cmd; SharedMemoryCommand cmd;
SharedMemoryCommand* cmdPtr = 0; SharedMemoryCommand* cmdPtr = 0;
//performance test //performance test
if (numBytesRec == sizeof(int)) if (numBytesRec == sizeof(int))
{ {
cmdPtr = &cmd; cmdPtr = &cmd;
cmd.m_type = *(int*)pClient->GetData(); cmd.m_type = *(int*)&bytesReceived[0];
} }
if (numBytesRec == sizeof(SharedMemoryCommand)) if (numBytesRec == sizeof(SharedMemoryCommand))
{ {
cmdPtr = (SharedMemoryCommand*)pClient->GetData(); cmdPtr = (SharedMemoryCommand*)&bytesReceived[0];
} }
if (cmdPtr) if (cmdPtr)
{ {
SharedMemoryStatus serverStatus; SharedMemoryStatus serverStatus;
b3AlignedObjectArray<char> buffer; b3AlignedObjectArray<char> buffer;
buffer.resize(SHARED_MEMORY_MAX_STREAM_CHUNK_SIZE); buffer.resize(SHARED_MEMORY_MAX_STREAM_CHUNK_SIZE);
bool hasStatus = sm->processCommand(*cmdPtr,serverStatus, &buffer[0], buffer.size()); bool hasStatus = sm->processCommand(*cmdPtr,serverStatus, &buffer[0], buffer.size());
double startTimeSeconds = clock.getTimeInSeconds(); double startTimeSeconds = clock.getTimeInSeconds();
double curTimeSeconds = clock.getTimeInSeconds(); double curTimeSeconds = clock.getTimeInSeconds();
while ((!hasStatus) && ((curTimeSeconds - startTimeSeconds) <timeOutInSeconds)) while ((!hasStatus) && ((curTimeSeconds - startTimeSeconds) <timeOutInSeconds))
{ {
hasStatus = sm->receiveStatus(serverStatus, &buffer[0], buffer.size()); hasStatus = sm->receiveStatus(serverStatus, &buffer[0], buffer.size());
curTimeSeconds = clock.getTimeInSeconds(); curTimeSeconds = clock.getTimeInSeconds();
} }
if (gVerboseNetworkMessagesServer) if (gVerboseNetworkMessagesServer)
{ {
printf("buffer.size = %d\n", buffer.size()); printf("buffer.size = %d\n", buffer.size());
printf("serverStatus.m_numDataStreamBytes = %d\n", serverStatus.m_numDataStreamBytes); printf("serverStatus.m_numDataStreamBytes = %d\n", serverStatus.m_numDataStreamBytes);
} }
if (hasStatus) if (hasStatus)
{ {
b3AlignedObjectArray<unsigned char> packetData; b3AlignedObjectArray<unsigned char> packetData;
unsigned char* statBytes = (unsigned char*)&serverStatus; unsigned char* statBytes = (unsigned char*)&serverStatus;
if (cmdPtr->m_type == CMD_STEP_FORWARD_SIMULATION) if (cmdPtr->m_type == CMD_STEP_FORWARD_SIMULATION)
{ {
packetData.resize(4 + sizeof(int)); packetData.resize(4 + sizeof(int));
int sz = packetData.size(); int sz = packetData.size();
int curPos = 0; int curPos = 0;
MySerializeInt(sz, &packetData[curPos]); MySerializeInt(sz, &packetData[curPos]);
curPos += 4; curPos += 4;
for (int i = 0; i < sizeof(int); i++) for (int i = 0; i < sizeof(int); i++)
{ {
packetData[i + curPos] = statBytes[i]; packetData[i + curPos] = statBytes[i];
} }
curPos += sizeof(int); curPos += sizeof(int);
pClient->Send( &packetData[0], packetData.size() ); pClient->Send( &packetData[0], packetData.size() );
} }
else else
{ {
//create packetData with [int packetSizeInBytes, status, streamBytes) //create packetData with [int packetSizeInBytes, status, streamBytes)
packetData.resize(4 + sizeof(SharedMemoryStatus) + serverStatus.m_numDataStreamBytes); packetData.resize(4 + sizeof(SharedMemoryStatus) + serverStatus.m_numDataStreamBytes);
int sz = packetData.size(); int sz = packetData.size();
int curPos = 0; int curPos = 0;
MySerializeInt(sz, &packetData[curPos]); MySerializeInt(sz, &packetData[curPos]);
curPos += 4; curPos += 4;
for (int i = 0; i < sizeof(SharedMemoryStatus); i++) for (int i = 0; i < sizeof(SharedMemoryStatus); i++)
{ {
packetData[i + curPos] = statBytes[i]; packetData[i + curPos] = statBytes[i];
} }
curPos += sizeof(SharedMemoryStatus); curPos += sizeof(SharedMemoryStatus);
for (int i = 0; i < serverStatus.m_numDataStreamBytes; i++) for (int i = 0; i < serverStatus.m_numDataStreamBytes; i++)
{ {
packetData[i + curPos] = buffer[i]; packetData[i + curPos] = buffer[i];
} }
pClient->Send( &packetData[0], packetData.size() ); pClient->Send( &packetData[0], packetData.size() );
} }
} }
} }
else else
{ {
printf("received packet with unknown contents\n"); printf("received packet with unknown contents\n");
} }
bytesReceived.clear();
}
} }
if (!receivedData) if (!receivedData)
{ {

6
test/clsocket/EchoServer.cpp
View File

@@ -28,10 +28,12 @@ int main(int argc, char **argv)
{ {
//printf("try receive\n"); //printf("try receive\n");
bool receivedData = false; bool receivedData = false;
int recBytes = 0;
if (pClient->Receive(MAX_PACKET)) recBytes = pClient->Receive(MAX_PACKET);
if (recBytes)
{ {
char* msg = (char*) pClient->GetData(); char* msg = (char*) pClient->GetData();
msg[recBytes]=0;
printf("received message [%s]\n",msg); printf("received message [%s]\n",msg);
//------------------------------------------------------------------ //------------------------------------------------------------------
// Send response to client and close connection to the client. // Send response to client and close connection to the client.