latency

git-svn-id: svn://elaine/murooma/trunk@257 d8a302eb-03bc-478d-80e4-98257eca68ef

client/controller.cpp

@@ -6,6 +6,7 @@
 #include "pcmDecoder.h"
 #include "player.h"
 #include "common/serverSettings.h"
+#include "common/timeMsg.h"
 
 using namespace std;
 
@@ -18,7 +19,6 @@ Controller::Controller() : MessageReceiver(), active_(false), streamClient(NULL)
 
 void Controller::onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer)
 {
-//cout << "onMessageReceived: " << baseMessage.type << ", size: " << baseMessage.size << ", sent: " << baseMessage.sent.sec << "," << baseMessage.sent.usec << ", recv: " << baseMessage.received.sec << "," << baseMessage.received.usec << "\n"; 
 	if (baseMessage.type == message_type::payload)
 	{
 		if ((stream != NULL) && (decoder != NULL))
@@ -98,9 +98,19 @@ void Controller::worker()
 
 	while (active_)
 	{
-		usleep(100000);
+		usleep(1000000);
-//		BaseMessage msg;
+		TimeMsg timeMsg;
-//		controlConnection->send(&msg);
+		BaseMessage* reply = controlConnection->sendRequest(&timeMsg, 2000);
+		if (reply != NULL)
+		{
+cout << "Reply: " << reply->type << ", size: " << reply->size << ", sent: " << reply->sent.sec << "," << reply->sent.usec << ", recv: " << reply->received.sec << "," << reply->received.usec << "\n"; 
+			if (reply->type == message_type::timemsg)
+			{
+				TimeMsg* timeMsg = (TimeMsg*)(reply);
+				long latency = (timeMsg->received.sec - timeMsg->sent.sec) * 1000000 + (timeMsg->received.usec - timeMsg->sent.usec);
+				cout << "C2S: " << timeMsg->latency << ", S2C: " << latency << "\n";
+			}
+		}
 	}
 }
 

common/message.h

@@ -34,7 +34,8 @@ enum message_type
 	header = 1,
 	payload = 2,
 	sampleformat = 3,
-	serversettings = 4
+	serversettings = 4,
+	timemsg = 5
 };
 
 

common/socketConnection.cpp

@@ -69,27 +69,32 @@ BaseMessage* SocketConnection::sendRequest(BaseMessage* message, size_t timeout)
 	BaseMessage* response(NULL);
 	if (++reqId == 0)
 		++reqId;
+	message->id = reqId;
 	shared_ptr<PendingRequest> pendingRequest(new PendingRequest(reqId));
 	pendingRequests.insert(pendingRequest);
 	std::mutex mtx;
 	std::unique_lock<std::mutex> lck(mtx);
 	send(message);
 	if (pendingRequest->cv.wait_for(lck,std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
+	{
+cout << "response received\n";
 		response = pendingRequest->response;
+	}
 	pendingRequests.erase(pendingRequest);
+cout << "response != NULL: " << (response != NULL) << "\n";
 	return response;
 }
 
 
 void SocketConnection::getNextMessage()
 {
-//cout << "getNextMessage\n";
+cout << "getNextMessage\n";
 	BaseMessage baseMessage;
 	size_t baseMsgSize = baseMessage.getSize();
 	vector<char> buffer(baseMsgSize);
 	socketRead(&buffer[0], baseMsgSize);
 	baseMessage.deserialize(&buffer[0]);
-//cout << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << "\n";
+cout << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n";
 	if (baseMessage.size > buffer.size())
 		buffer.resize(baseMessage.size);
 	socketRead(&buffer[0], baseMessage.size);
@@ -100,6 +105,10 @@ void SocketConnection::getNextMessage()
 	{
 		if (req->id == baseMessage.refersTo)
 		{
+cout << "getNextMessage response: " << baseMessage.type << ", size: " << baseMessage.size << "\n";
+long latency = (baseMessage.received.sec - baseMessage.sent.sec) * 1000000 + (baseMessage.received.usec - baseMessage.sent.usec);
+cout << "latency: " << latency << "\n";
+			req->response = new BaseMessage(baseMessage);
 			req->cv.notify_one();
 			return;
 		}

common/timeUtils.h

@@ -77,6 +77,12 @@ static void addUs(timeval& tv, int us)
 }
 
 
+/*static long diffMs(const timeval& t1, const timeval& t2)
+{
+    return (((t1.tv_sec - t2.tv_sec) * 1000000) + 
+            (t1.tv_usec - t2.tv_usec))/1000;
+}*/
+
 
 static long getTickCount()
 {

server/controlServer.cpp

@@ -1,4 +1,5 @@
 #include "controlServer.h"
+#include "common/timeMsg.h"
 #include <iostream>
 
 
@@ -9,7 +10,16 @@ ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NUL
 
 void ControlServer::onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer)
 {
-	cout << "onMessageReceived: " << baseMessage.type << ", " << baseMessage.size << "\n";
+	cout << "onMessageReceived: " << baseMessage.type << ", size: " << baseMessage.size << ", sent: " << baseMessage.sent.sec << "," << baseMessage.sent.usec << ", recv: " << baseMessage.received.sec << "," << baseMessage.received.usec << "\n"; 
+	if (baseMessage.type == message_type::timemsg)
+	{	
+		TimeMsg* timeMsg = new TimeMsg();
+		timeMsg->deserialize(baseMessage, buffer);
+		timeMsg->latency = (timeMsg->received.sec - timeMsg->sent.sec) * 1000000 + (timeMsg->received.usec - timeMsg->sent.usec);
+		cout << "Latency: " << timeMsg->latency << "\n";
+		timeMsg->refersTo = timeMsg->id;
+		connection->send(timeMsg);
+	}
 }