3pp-mirror/snapcast
1
0
Fork 0
mirror of https://github.com/badaix/snapcast.git synced 2025-06-04 03:41:43 +02:00
Code Issues Projects Releases Packages Wiki Activity

fixed request timeout

Browse source
This commit is contained in:
badaix 2015-07-11 00:24:04 +02:00
parent aba04a0675
commit 8f21948a4d
7 changed files with 24 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

17
client/clientConnection.cpp
View file

@ -72,6 +72,8 @@ void ClientConnection::start()
// cout << "socket: " << socket->native() << "\n"; // cout << "socket: " << socket->native() << "\n";
// setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); // setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
// setsockopt(socket->native(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); // setsockopt(socket->native(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
//boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string("127.0.0.1"), port_);
//socket_->connect(endpoint);//*iterator);
socket_->connect(*iterator); socket_->connect(*iterator);
logO << "MAC: \"" << getMacAddress(socket_->native()) << "\"\n"; logO << "MAC: \"" << getMacAddress(socket_->native()) << "\"\n";
connected_ = true; connected_ = true;
@ -130,24 +132,23 @@ bool ClientConnection::send(msg::BaseMessage* message)
shared_ptr<msg::SerializedMessage> ClientConnection::sendRequest(msg::BaseMessage* message, const chronos::msec& timeout) shared_ptr<msg::SerializedMessage> ClientConnection::sendRequest(msg::BaseMessage* message, const chronos::msec& timeout)
{ {
shared_ptr<msg::SerializedMessage> response(NULL); shared_ptr<msg::SerializedMessage> response(NULL);
if (++reqId_ == 10000) if (++reqId_ >= 10000)
reqId_ = 1; reqId_ = 1;
message->id = reqId_; message->id = reqId_;
//logD << "Req: " << reqId << "\n"; // logD << "Req: " << reqId_ << "\n";
shared_ptr<PendingRequest> pendingRequest(new PendingRequest(reqId_)); shared_ptr<PendingRequest> pendingRequest(new PendingRequest(reqId_));
{ {
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
pendingRequests_.insert(pendingRequest); pendingRequests_.insert(pendingRequest);
} }
std::mutex m; std::unique_lock<std::mutex> lck(requestMutex_);
std::unique_lock<std::mutex> lck(m);
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)
{ {
response = pendingRequest->response; response = pendingRequest->response;
sumTimeout_ = chronos::msec(0); sumTimeout_ = chronos::msec(0);
//logD << "Resp: " << pendingRequest->id << "\n"; // logD << "Resp: " << pendingRequest->id << "\n";
} }
else else
{ {
@ -171,7 +172,7 @@ void ClientConnection::getNextMessage()
vector<char> buffer(baseMsgSize); vector<char> buffer(baseMsgSize);
socketRead(&buffer[0], baseMsgSize); socketRead(&buffer[0], baseMsgSize);
baseMessage.deserialize(&buffer[0]); baseMessage.deserialize(&buffer[0]);
//logD << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n"; // logD << "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);
@ -180,6 +181,7 @@ void ClientConnection::getNextMessage()
{ {
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
// logD << "got lock - getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n";
{ {
for (auto req: pendingRequests_) for (auto req: pendingRequests_)
{ {
@ -189,6 +191,7 @@ void ClientConnection::getNextMessage()
req->response->message = baseMessage; req->response->message = baseMessage;
req->response->buffer = (char*)malloc(baseMessage.size); req->response->buffer = (char*)malloc(baseMessage.size);
memcpy(req->response->buffer, &buffer[0], baseMessage.size); memcpy(req->response->buffer, &buffer[0], baseMessage.size);
std::unique_lock<std::mutex> lck(requestMutex_);
req->cv.notify_one(); req->cv.notify_one();
return; return;
} }

1
client/clientConnection.h
View file

@ -99,6 +99,7 @@ protected:
std::atomic<bool> connected_; std::atomic<bool> connected_;
MessageReceiver* messageReceiver_; MessageReceiver* messageReceiver_;
mutable std::mutex mutex_; mutable std::mutex mutex_;
mutable std::mutex requestMutex_;
std::set<std::shared_ptr<PendingRequest>> pendingRequests_; std::set<std::shared_ptr<PendingRequest>> pendingRequests_;
uint16_t reqId_; uint16_t reqId_;
std::string ip_; std::string ip_;

7
client/controller.cpp
View file

@ -123,17 +123,17 @@ void Controller::worker()
decoder_->setHeader(headerChunk.get()); decoder_->setHeader(headerChunk.get());
msg::Request timeReq(kTime); msg::Request timeReq(kTime);
for (size_t n=0; n<50 && active_; ++n) for (size_t n=0; n<100 && active_; ++n)
{ {
shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq, chronos::msec(2000)); shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq, chronos::msec(2000));
if (reply) if (reply)
{ {
double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.; double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.;
TimeProvider::getInstance().setDiffToServer((reply->latency - latency) * 1000 / 2); TimeProvider::getInstance().setDiffToServer((reply->latency - latency) * 1000 / 2);
usleep(1000); usleep(100);
} }
} }
logO << "diff to server [ms]: " << TimeProvider::getInstance().getDiffToServer<chronos::msec>().count() << "\n"; logO << "diff to server [ms]: " << (float)TimeProvider::getInstance().getDiffToServer<chronos::usec>().count() / 1000.f << "\n";
stream_ = new Stream(*sampleFormat_); stream_ = new Stream(*sampleFormat_);
stream_->setBufferLen(serverSettings->bufferMs - latency_); stream_->setBufferLen(serverSettings->bufferMs - latency_);
@ -148,7 +148,6 @@ void Controller::worker()
while (active_) while (active_)
{ {
usleep(500*1000); usleep(500*1000);
//throw SnapException("timeout");
shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq); shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq);
if (reply) if (reply)
{ {

4
server/controlServer.cpp
View file

@ -57,7 +57,7 @@ void ControlServer::send(shared_ptr<msg::BaseMessage> message)
void ControlServer::onMessageReceived(ServerSession* connection, const msg::BaseMessage& baseMessage, char* buffer) void ControlServer::onMessageReceived(ServerSession* connection, const msg::BaseMessage& baseMessage, char* buffer)
{ {
// logD << "onMessageReceived: " << baseMessage.type << ", size: " << baseMessage.size << ", sent: " << baseMessage.sent.sec << "," << baseMessage.sent.usec << ", recv: " << baseMessage.received.sec << "," << baseMessage.received.usec << "\n"; // logD << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << ", sent: " << baseMessage.sent.sec << "," << baseMessage.sent.usec << ", recv: " << baseMessage.received.sec << "," << baseMessage.received.usec << "\n";
if (baseMessage.type == message_type::kRequest) if (baseMessage.type == message_type::kRequest)
{ {
msg::Request requestMsg; msg::Request requestMsg;
@ -70,7 +70,7 @@ void ControlServer::onMessageReceived(ServerSession* connection, const msg::Base
timeMsg.refersTo = requestMsg.id; timeMsg.refersTo = requestMsg.id;
timeMsg.latency = (requestMsg.received.sec - requestMsg.sent.sec) + (requestMsg.received.usec - requestMsg.sent.usec) / 1000000.; timeMsg.latency = (requestMsg.received.sec - requestMsg.sent.sec) + (requestMsg.received.usec - requestMsg.sent.usec) / 1000000.;
// tv diff = timeMsg.received - timeMsg.sent; // tv diff = timeMsg.received - timeMsg.sent;
// logD << "Latency: " << diff.sec << "." << diff.usec << "\n"; // logD << "Latency: " << timeMsg.latency << ", refers to: " << timeMsg.refersTo << "\n";
connection->send(&timeMsg); connection->send(&timeMsg);
} }
else if (requestMsg.request == kServerSettings) else if (requestMsg.request == kServerSettings)

2
server/flacEncoder.cpp
View file

@ -51,7 +51,7 @@ double FlacEncoder::encode(msg::PcmChunk* chunk)
{ {
int samples = chunk->getSampleCount(); int samples = chunk->getSampleCount();
int frames = chunk->getFrameCount(); int frames = chunk->getFrameCount();
logO << "payload: " << chunk->payloadSize << "\tframes: " << frames << "\tsamples: " << samples << "\tduration: " << chunk->duration<chronos::msec>().count() << "\n"; logD << "payload: " << chunk->payloadSize << "\tframes: " << frames << "\tsamples: " << samples << "\tduration: " << chunk->duration<chronos::msec>().count() << "\n";
if (pcmBufferSize_ < samples) if (pcmBufferSize_ < samples)
{ {

4
server/serverSession.cpp
View file

@ -109,6 +109,7 @@ void ServerSession::add(shared_ptr<msg::BaseMessage> message)
bool ServerSession::send(msg::BaseMessage* message) bool ServerSession::send(msg::BaseMessage* message)
{ {
// logD << "send: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
if (!socket_) if (!socket_)
return false; return false;
@ -124,13 +125,12 @@ bool ServerSession::send(msg::BaseMessage* message)
void ServerSession::getNextMessage() void ServerSession::getNextMessage()
{ {
//logD << "getNextMessage\n";
msg::BaseMessage baseMessage; msg::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]);
//logD << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n"; // logD << "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);

5
todo.txt Normal file
View file

@ -0,0 +1,5 @@
client: cannot connect to server when started first?
server: throw exception in encode init
server: put only one encoded frame into ota chunks
client: time sync timeouts over wired LAN
test 24, 32bit samples