3pp-mirror/snapcast
1
0
Fork 0
mirror of https://github.com/badaix/snapcast.git synced 2025-05-21 13:06:15 +02:00
Code Issues Projects Releases Packages Wiki Activity

timesync

Browse source 
git-svn-id: svn://elaine/murooma/trunk@258 d8a302eb-03bc-478d-80e4-98257eca68ef
This commit is contained in:
(no author) 2014-09-14 16:40:19 +00:00
parent 6d6272d23a
commit a16aa82deb
4 changed files with 50 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

25
client/controller.cpp
View file

@ -1,6 +1,7 @@
#include "controller.h" #include "controller.h"
#include <iostream> #include <iostream>
#include <string> #include <string>
#include <memory>
#include <unistd.h> #include <unistd.h>
#include "oggDecoder.h" #include "oggDecoder.h"
#include "pcmDecoder.h" #include "pcmDecoder.h"
@ -98,19 +99,27 @@ void Controller::worker()
while (active_) while (active_)
{ {
usleep(1000000); usleep(1000000);//1000000);
TimeMsg timeMsg; TimeMsg timeMsg;
BaseMessage* reply = controlConnection->sendRequest(&timeMsg, 2000);
if (reply != NULL) try
{ {
cout << "Reply: " << reply->type << ", size: " << reply->size << ", sent: " << reply->sent.sec << "," << reply->sent.usec << ", recv: " << reply->received.sec << "," << reply->received.usec << "\n"; shared_ptr<PendingRequest> reply = controlConnection->sendRequest(&timeMsg, 2000);
if (reply->type == message_type::timemsg) if (reply)
{ {
TimeMsg* timeMsg = (TimeMsg*)(reply); if (reply->response->type == message_type::timemsg)
long latency = (timeMsg->received.sec - timeMsg->sent.sec) * 1000000 + (timeMsg->received.usec - timeMsg->sent.usec); {
cout << "C2S: " << timeMsg->latency << ", S2C: " << latency << "\n"; //cout << "Reply: " << reply->response->type << ", size: " << reply->response->size << ", sent: " << reply->response->sent.sec << "," << reply->response->sent.usec << ", recv: " << reply->response->received.sec << "," << reply->response->received.usec << "\n";
TimeMsg timeMsg;
timeMsg.deserialize(*reply->response, reply->buffer);
long latency = (timeMsg.received.sec - timeMsg.sent.sec) * 1000000 + (timeMsg.received.usec - timeMsg.sent.usec);
cout << "C2S: " << timeMsg.latency << ", S2C: " << latency << ", diff: " << (timeMsg.latency - latency) / 2 << endl;
}
} }
} }
catch (const std::exception& e)
{
}
} }
} }

24
common/socketConnection.cpp
View file

@ -64,9 +64,9 @@ bool SocketConnection::send(BaseMessage* message)
} }
BaseMessage* SocketConnection::sendRequest(BaseMessage* message, size_t timeout) shared_ptr<PendingRequest> SocketConnection::sendRequest(BaseMessage* message, size_t timeout)
{ {
BaseMessage* response(NULL); shared_ptr<PendingRequest> response(NULL);
if (++reqId == 0) if (++reqId == 0)
++reqId; ++reqId;
message->id = reqId; message->id = reqId;
@ -77,24 +77,26 @@ BaseMessage* SocketConnection::sendRequest(BaseMessage* message, size_t timeout)
send(message); send(message);
if (pendingRequest->cv.wait_for(lck,std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout) if (pendingRequest->cv.wait_for(lck,std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
{ {
cout << "response received\n"; response = pendingRequest;
response = pendingRequest->response; }
else
{
cout << "timeout while waiting for response to: " << reqId << "\n";
} }
pendingRequests.erase(pendingRequest); pendingRequests.erase(pendingRequest);
cout << "response != NULL: " << (response != NULL) << "\n";
return response; return response;
} }
void SocketConnection::getNextMessage() void SocketConnection::getNextMessage()
{ {
cout << "getNextMessage\n"; //cout << "getNextMessage\n";
BaseMessage baseMessage; BaseMessage baseMessage;
size_t baseMsgSize = baseMessage.getSize(); size_t baseMsgSize = baseMessage.getSize();
vector<char> buffer(baseMsgSize); vector<char> buffer(baseMsgSize);
socketRead(&buffer[0], baseMsgSize); socketRead(&buffer[0], baseMsgSize);
baseMessage.deserialize(&buffer[0]); baseMessage.deserialize(&buffer[0]);
cout << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n"; //cout << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n";
if (baseMessage.size > buffer.size()) if (baseMessage.size > buffer.size())
buffer.resize(baseMessage.size); buffer.resize(baseMessage.size);
socketRead(&buffer[0], baseMessage.size); socketRead(&buffer[0], baseMessage.size);
@ -105,10 +107,12 @@ cout << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size
{ {
if (req->id == baseMessage.refersTo) if (req->id == baseMessage.refersTo)
{ {
cout << "getNextMessage response: " << baseMessage.type << ", size: " << baseMessage.size << "\n"; //cout << "getNextMessage response: " << baseMessage.type << ", size: " << baseMessage.size << "\n";
long latency = (baseMessage.received.sec - baseMessage.sent.sec) * 1000000 + (baseMessage.received.usec - baseMessage.sent.usec); //long latency = (baseMessage.received.sec - baseMessage.sent.sec) * 1000000 + (baseMessage.received.usec - baseMessage.sent.usec);
cout << "latency: " << latency << "\n"; //cout << "latency: " << latency << "\n";
req->response = new BaseMessage(baseMessage); req->response = new BaseMessage(baseMessage);
req->buffer = (char*)malloc(baseMessage.size);
memcpy(req->buffer, &buffer[0], baseMessage.size);
req->cv.notify_one(); req->cv.notify_one();
return; return;
} }

13
common/socketConnection.h
View file

@ -5,6 +5,7 @@
#include <thread> #include <thread>
#include <atomic> #include <atomic>
#include <mutex> #include <mutex>
#include <memory>
#include <boost/asio.hpp> #include <boost/asio.hpp>
#include <condition_variable> #include <condition_variable>
#include <set> #include <set>
@ -19,10 +20,18 @@ class SocketConnection;
struct PendingRequest struct PendingRequest
{ {
PendingRequest(uint16_t reqId) : id(reqId), response(NULL) {}; PendingRequest(uint16_t reqId) : id(reqId), response(NULL), buffer(NULL) {};
~PendingRequest()
{
if (response != NULL);
delete response;
if (buffer != NULL)
free(buffer);
}
uint16_t id; uint16_t id;
BaseMessage* response; BaseMessage* response;
char* buffer;
std::condition_variable cv; std::condition_variable cv;
}; };
@ -42,7 +51,7 @@ public:
virtual void start(); virtual void start();
virtual void stop(); virtual void stop();
virtual bool send(BaseMessage* _message); virtual bool send(BaseMessage* _message);
virtual BaseMessage* sendRequest(BaseMessage* message, size_t timeout); virtual std::shared_ptr<PendingRequest> sendRequest(BaseMessage* message, size_t timeout);
virtual bool active() virtual bool active()
{ {

14
server/controlServer.cpp
View file

@ -10,15 +10,15 @@ ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NUL
void ControlServer::onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer) void ControlServer::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"; // 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) if (baseMessage.type == message_type::timemsg)
{ {
TimeMsg* timeMsg = new TimeMsg(); TimeMsg timeMsg;
timeMsg->deserialize(baseMessage, buffer); timeMsg.deserialize(baseMessage, buffer);
timeMsg->latency = (timeMsg->received.sec - timeMsg->sent.sec) * 1000000 + (timeMsg->received.usec - timeMsg->sent.usec); timeMsg.latency = (timeMsg.received.sec - timeMsg.sent.sec) * 1000000 + (timeMsg.received.usec - timeMsg.sent.usec);
cout << "Latency: " << timeMsg->latency << "\n"; // cout << "Latency: " << timeMsg.latency << "\n";
timeMsg->refersTo = timeMsg->id; timeMsg.refersTo = timeMsg.id;
connection->send(timeMsg); connection->send(&timeMsg);
} }
} }