diff --git a/examples/SharedMemory/PhysicsClientTCP.cpp b/examples/SharedMemory/PhysicsClientTCP.cpp index 35fc08964..443fc3bef 100644 --- a/examples/SharedMemory/PhysicsClientTCP.cpp +++ b/examples/SharedMemory/PhysicsClientTCP.cpp @@ -10,45 +10,7 @@ #include "SharedMemoryCommands.h" #include #include "Bullet3Common/b3Logging.h" -#include "../MultiThreading/b3ThreadSupportInterface.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; -}; +#include "Bullet3Common/b3AlignedObjectArray.h" @@ -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]; return tmp; } +bool gVerboseNetworkMessagesClient2 = false; + struct TcpNetworkedInternalData { @@ -70,29 +34,24 @@ struct TcpNetworkedInternalData bool m_isConnected; - b3ThreadSupportInterface* m_threadSupport; - - b3CriticalSection* m_cs; - TcpNetworkedInternalData* m_tcpInternalData; SharedMemoryCommand m_clientCmd; bool m_hasCommand; - bool m_hasStatus; SharedMemoryStatus m_lastStatus; b3AlignedObjectArray m_stream; std::string m_hostName; int m_port; + b3AlignedObjectArray m_tempBuffer; + TcpNetworkedInternalData() : m_isConnected(false), - m_threadSupport(0), - m_hasCommand(false), - m_hasStatus(false) + m_hasCommand(false) { } @@ -105,8 +64,7 @@ struct TcpNetworkedInternalData m_tcpSocket.Initialize(); m_isConnected = m_tcpSocket.Open(m_hostName.c_str(),m_port); - - m_isConnected = true; + return m_isConnected; } @@ -117,203 +75,66 @@ struct TcpNetworkedInternalData //int serviceResult = enet_host_service(m_client, &m_event, 0); 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) - { - printf("A packet of length %d bytes received\n", recBytes); - } - - unsigned char* data = (unsigned char*)m_tcpSocket.GetData(); - - int packetSizeInBytes = b3DeserializeInt2(data); + //append to tmp buffer + //recBytes - 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 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]; - } - } - } - else - { - printf("unknown status message received\n"); - } - + int curSize = m_tempBuffer.size(); + m_tempBuffer.resize(curSize+rBytes); + for (int i=0;i=4) + { + packetSizeInBytes = b3DeserializeInt2(&m_tempBuffer[0]); + } + + 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; } }; -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) { @@ -338,49 +159,42 @@ bool TcpNetworkedPhysicsProcessor::processCommand(const struct SharedMemoryComma { 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; } bool TcpNetworkedPhysicsProcessor::receiveStatus(struct SharedMemoryStatus& serverStatusOut, char* bufferServerToClient, int bufferSizeInBytes) { - bool hasStatus = false; - if (m_data->m_hasStatus) + bool hasStatus = m_data->checkData(); + + if (hasStatus) { if (gVerboseNetworkMessagesClient2) { printf("TcpNetworkedPhysicsProcessor::receiveStatus\n"); } - hasStatus = true; serverStatusOut = m_data->m_lastStatus; int numStreamBytes = m_data->m_stream.size(); @@ -396,9 +210,6 @@ bool TcpNetworkedPhysicsProcessor::receiveStatus(struct SharedMemoryStatus& serv 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() { - if (m_data->m_threadSupport==0) - { - 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); + bool isConnected = m_data->connectTCP(); return isConnected; } void TcpNetworkedPhysicsProcessor::disconnect() { - if (m_data->m_threadSupport) - { - 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; - } - - - + m_data->m_tcpSocket.Close(); + m_data->m_isConnected = false; } + diff --git a/examples/SharedMemory/tcp/main.cpp b/examples/SharedMemory/tcp/main.cpp index 497766f08..efd667ed9 100644 --- a/examples/SharedMemory/tcp/main.cpp +++ b/examples/SharedMemory/tcp/main.cpp @@ -17,7 +17,6 @@ typedef SharedMemoryCommandProcessor MyCommandProcessor; #include "PhysicsServerCommandProcessor.h" #include "../Utils/b3Clock.h" -#define MAX_PACKET 4096 bool gVerboseNetworkMessagesServer = true; @@ -75,6 +74,8 @@ int main(int argc, char *argv[]) socket.Initialize(); socket.Listen("localhost", port); + + b3AlignedObjectArray bytesReceived; while (!exitRequested) { @@ -93,111 +94,130 @@ int main(int argc, char *argv[]) //printf("try receive\n"); 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(); - if (gVerboseNetworkMessagesServer) - { - printf("received message length [%d]\n",numBytesRec); - } + + char* msg2 = (char*) pClient->GetData(); + int numBytesRec2 = pClient->GetBytesReceived(); + + int curSize = bytesReceived.size(); + bytesReceived.resize(bytesReceived.size()+numBytesRec2); + for (int i=0;iGetData(); - } + //performance test + if (numBytesRec == sizeof(int)) + { + cmdPtr = &cmd; + cmd.m_type = *(int*)&bytesReceived[0]; + } - if (numBytesRec == sizeof(SharedMemoryCommand)) - { - cmdPtr = (SharedMemoryCommand*)pClient->GetData(); - } - if (cmdPtr) - { - SharedMemoryStatus serverStatus; - b3AlignedObjectArray buffer; - buffer.resize(SHARED_MEMORY_MAX_STREAM_CHUNK_SIZE); + if (numBytesRec == sizeof(SharedMemoryCommand)) + { + cmdPtr = (SharedMemoryCommand*)&bytesReceived[0]; + } + if (cmdPtr) + { + SharedMemoryStatus serverStatus; + b3AlignedObjectArray buffer; + 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 curTimeSeconds = clock.getTimeInSeconds(); + double startTimeSeconds = clock.getTimeInSeconds(); + double curTimeSeconds = clock.getTimeInSeconds(); - while ((!hasStatus) && ((curTimeSeconds - startTimeSeconds) receiveStatus(serverStatus, &buffer[0], buffer.size()); - curTimeSeconds = clock.getTimeInSeconds(); - } - if (gVerboseNetworkMessagesServer) - { - printf("buffer.size = %d\n", buffer.size()); - printf("serverStatus.m_numDataStreamBytes = %d\n", serverStatus.m_numDataStreamBytes); - } - if (hasStatus) - { - b3AlignedObjectArray packetData; - unsigned char* statBytes = (unsigned char*)&serverStatus; + while ((!hasStatus) && ((curTimeSeconds - startTimeSeconds) receiveStatus(serverStatus, &buffer[0], buffer.size()); + curTimeSeconds = clock.getTimeInSeconds(); + } + if (gVerboseNetworkMessagesServer) + { + printf("buffer.size = %d\n", buffer.size()); + printf("serverStatus.m_numDataStreamBytes = %d\n", serverStatus.m_numDataStreamBytes); + } + if (hasStatus) + { + b3AlignedObjectArray packetData; + unsigned char* statBytes = (unsigned char*)&serverStatus; - if (cmdPtr->m_type == CMD_STEP_FORWARD_SIMULATION) - { - packetData.resize(4 + sizeof(int)); - int sz = packetData.size(); - int curPos = 0; + if (cmdPtr->m_type == CMD_STEP_FORWARD_SIMULATION) + { + packetData.resize(4 + sizeof(int)); + int sz = packetData.size(); + int curPos = 0; - MySerializeInt(sz, &packetData[curPos]); - curPos += 4; - for (int i = 0; i < sizeof(int); i++) - { - packetData[i + curPos] = statBytes[i]; - } - curPos += sizeof(int); + MySerializeInt(sz, &packetData[curPos]); + curPos += 4; + for (int i = 0; i < sizeof(int); i++) + { + packetData[i + curPos] = statBytes[i]; + } + curPos += sizeof(int); - pClient->Send( &packetData[0], packetData.size() ); + pClient->Send( &packetData[0], packetData.size() ); - } - else - { - //create packetData with [int packetSizeInBytes, status, streamBytes) - packetData.resize(4 + sizeof(SharedMemoryStatus) + serverStatus.m_numDataStreamBytes); - int sz = packetData.size(); - int curPos = 0; + } + else + { + //create packetData with [int packetSizeInBytes, status, streamBytes) + packetData.resize(4 + sizeof(SharedMemoryStatus) + serverStatus.m_numDataStreamBytes); + int sz = packetData.size(); + int curPos = 0; - MySerializeInt(sz, &packetData[curPos]); - curPos += 4; - for (int i = 0; i < sizeof(SharedMemoryStatus); i++) - { - packetData[i + curPos] = statBytes[i]; - } - curPos += sizeof(SharedMemoryStatus); + MySerializeInt(sz, &packetData[curPos]); + curPos += 4; + for (int i = 0; i < sizeof(SharedMemoryStatus); i++) + { + packetData[i + curPos] = statBytes[i]; + } + curPos += sizeof(SharedMemoryStatus); - for (int i = 0; i < serverStatus.m_numDataStreamBytes; i++) - { - packetData[i + curPos] = buffer[i]; - } + for (int i = 0; i < serverStatus.m_numDataStreamBytes; i++) + { + packetData[i + curPos] = buffer[i]; + } - pClient->Send( &packetData[0], packetData.size() ); - } - } - } - else - { - printf("received packet with unknown contents\n"); - } + pClient->Send( &packetData[0], packetData.size() ); + } + } + } + else + { + printf("received packet with unknown contents\n"); + } + bytesReceived.clear(); + + } } if (!receivedData) { diff --git a/test/clsocket/EchoServer.cpp b/test/clsocket/EchoServer.cpp index 74c03ac7c..4388fc2f2 100644 --- a/test/clsocket/EchoServer.cpp +++ b/test/clsocket/EchoServer.cpp @@ -28,10 +28,12 @@ int main(int argc, char **argv) { //printf("try receive\n"); bool receivedData = false; - - if (pClient->Receive(MAX_PACKET)) + int recBytes = 0; + recBytes = pClient->Receive(MAX_PACKET); + if (recBytes) { char* msg = (char*) pClient->GetData(); + msg[recBytes]=0; printf("received message [%s]\n",msg); //------------------------------------------------------------------ // Send response to client and close connection to the client.