diff --git a/client/clientConnection.cpp b/client/clientConnection.cpp index 74473414..83a5c10c 100644 --- a/client/clientConnection.cpp +++ b/client/clientConnection.cpp @@ -72,6 +72,8 @@ void ClientConnection::start() // cout << "socket: " << socket->native() << "\n"; // setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &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); logO << "MAC: \"" << getMacAddress(socket_->native()) << "\"\n"; connected_ = true; @@ -130,24 +132,23 @@ bool ClientConnection::send(msg::BaseMessage* message) shared_ptr ClientConnection::sendRequest(msg::BaseMessage* message, const chronos::msec& timeout) { shared_ptr response(NULL); - if (++reqId_ == 10000) + if (++reqId_ >= 10000) reqId_ = 1; message->id = reqId_; -//logD << "Req: " << reqId << "\n"; +// logD << "Req: " << reqId_ << "\n"; shared_ptr pendingRequest(new PendingRequest(reqId_)); { std::unique_lock mlock(mutex_); pendingRequests_.insert(pendingRequest); } - std::mutex m; - std::unique_lock lck(m); + std::unique_lock lck(requestMutex_); 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; sumTimeout_ = chronos::msec(0); -//logD << "Resp: " << pendingRequest->id << "\n"; +// logD << "Resp: " << pendingRequest->id << "\n"; } else { @@ -171,7 +172,7 @@ void ClientConnection::getNextMessage() vector buffer(baseMsgSize); socketRead(&buffer[0], baseMsgSize); 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()) buffer.resize(baseMessage.size); socketRead(&buffer[0], baseMessage.size); @@ -180,6 +181,7 @@ void ClientConnection::getNextMessage() { std::unique_lock mlock(mutex_); +// logD << "got lock - getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n"; { for (auto req: pendingRequests_) { @@ -189,6 +191,7 @@ void ClientConnection::getNextMessage() req->response->message = baseMessage; req->response->buffer = (char*)malloc(baseMessage.size); memcpy(req->response->buffer, &buffer[0], baseMessage.size); + std::unique_lock lck(requestMutex_); req->cv.notify_one(); return; } diff --git a/client/clientConnection.h b/client/clientConnection.h index 853bf0c1..b9aa5c74 100644 --- a/client/clientConnection.h +++ b/client/clientConnection.h @@ -99,6 +99,7 @@ protected: std::atomic connected_; MessageReceiver* messageReceiver_; mutable std::mutex mutex_; + mutable std::mutex requestMutex_; std::set> pendingRequests_; uint16_t reqId_; std::string ip_; diff --git a/client/controller.cpp b/client/controller.cpp index 9b89d079..692c8c10 100644 --- a/client/controller.cpp +++ b/client/controller.cpp @@ -123,17 +123,17 @@ void Controller::worker() decoder_->setHeader(headerChunk.get()); msg::Request timeReq(kTime); - for (size_t n=0; n<50 && active_; ++n) + for (size_t n=0; n<100 && active_; ++n) { shared_ptr reply = clientConnection_->sendReq(&timeReq, chronos::msec(2000)); if (reply) { double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.; TimeProvider::getInstance().setDiffToServer((reply->latency - latency) * 1000 / 2); - usleep(1000); + usleep(100); } } - logO << "diff to server [ms]: " << TimeProvider::getInstance().getDiffToServer().count() << "\n"; + logO << "diff to server [ms]: " << (float)TimeProvider::getInstance().getDiffToServer().count() / 1000.f << "\n"; stream_ = new Stream(*sampleFormat_); stream_->setBufferLen(serverSettings->bufferMs - latency_); @@ -148,7 +148,6 @@ void Controller::worker() while (active_) { usleep(500*1000); -//throw SnapException("timeout"); shared_ptr reply = clientConnection_->sendReq(&timeReq); if (reply) { diff --git a/server/controlServer.cpp b/server/controlServer.cpp index 4264887d..5ccbe556 100644 --- a/server/controlServer.cpp +++ b/server/controlServer.cpp @@ -57,7 +57,7 @@ void ControlServer::send(shared_ptr message) 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) { msg::Request requestMsg; @@ -70,7 +70,7 @@ void ControlServer::onMessageReceived(ServerSession* connection, const msg::Base timeMsg.refersTo = requestMsg.id; timeMsg.latency = (requestMsg.received.sec - requestMsg.sent.sec) + (requestMsg.received.usec - requestMsg.sent.usec) / 1000000.; // 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); } else if (requestMsg.request == kServerSettings) diff --git a/server/flacEncoder.cpp b/server/flacEncoder.cpp index dce5e469..acff7bee 100644 --- a/server/flacEncoder.cpp +++ b/server/flacEncoder.cpp @@ -51,7 +51,7 @@ double FlacEncoder::encode(msg::PcmChunk* chunk) { int samples = chunk->getSampleCount(); int frames = chunk->getFrameCount(); - logO << "payload: " << chunk->payloadSize << "\tframes: " << frames << "\tsamples: " << samples << "\tduration: " << chunk->duration().count() << "\n"; + logD << "payload: " << chunk->payloadSize << "\tframes: " << frames << "\tsamples: " << samples << "\tduration: " << chunk->duration().count() << "\n"; if (pcmBufferSize_ < samples) { diff --git a/server/serverSession.cpp b/server/serverSession.cpp index 05ac032d..d976892f 100644 --- a/server/serverSession.cpp +++ b/server/serverSession.cpp @@ -109,6 +109,7 @@ void ServerSession::add(shared_ptr message) bool ServerSession::send(msg::BaseMessage* message) { +// logD << "send: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n"; std::unique_lock mlock(mutex_); if (!socket_) return false; @@ -124,13 +125,12 @@ bool ServerSession::send(msg::BaseMessage* message) void ServerSession::getNextMessage() { -//logD << "getNextMessage\n"; msg::BaseMessage baseMessage; size_t baseMsgSize = baseMessage.getSize(); vector buffer(baseMsgSize); socketRead(&buffer[0], baseMsgSize); 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()) buffer.resize(baseMessage.size); socketRead(&buffer[0], baseMessage.size); diff --git a/todo.txt b/todo.txt new file mode 100644 index 00000000..9120a11a --- /dev/null +++ b/todo.txt @@ -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