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

added stream.cpp

Browse source 
git-svn-id: svn://elaine/murooma/trunk@72 d8a302eb-03bc-478d-80e4-98257eca68ef
This commit is contained in:
(no author) 2014-07-05 21:29:51 +00:00
parent 18b3901c57
commit ff2e24e219
5 changed files with 176 additions and 414 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Makefile
View file

@ -5,7 +5,7 @@ LDFLAGS = -lzmq -lpthread -lportaudio
OBJ_SERVER = server.o OBJ_SERVER = server.o
BIN_SERVER = server BIN_SERVER = server
OBJ_CLIENT = client.o OBJ_CLIENT = client.o stream.o
BIN_CLIENT = client BIN_CLIENT = client
OBJ = $(OBJ_SERVER) $(OBJ_CLIENT) OBJ = $(OBJ_SERVER) $(OBJ_CLIENT)

7
client.cpp
View file

@ -16,12 +16,13 @@
#include <thread> #include <thread>
#include <portaudio.h> #include <portaudio.h>
#include "chunk.h" #include "chunk.h"
#include "timeUtils.h"
#include "stream.h" #include "stream.h"
std::deque<int> timeDiffs; std::deque<int> timeDiffs;
int bufferMs; int bufferMs;
std::mutex mutex;
void player() void player()
@ -50,7 +51,9 @@ static int patestCallback( const void *inputBuffer, void *outputBuffer,
(void) statusFlags; (void) statusFlags;
(void) inputBuffer; (void) inputBuffer;
mutex.lock();
PlayerChunk* playerChunk = stream->getChunk(timeInfo->outputBufferDacTime, framesPerBuffer); PlayerChunk* playerChunk = stream->getChunk(timeInfo->outputBufferDacTime, framesPerBuffer);
mutex.unlock();
for (size_t n=0; n<framesPerBuffer; n++) for (size_t n=0; n<framesPerBuffer; n++)
{ {
@ -180,7 +183,9 @@ int main (int argc, char *argv[])
// memcpy(chunk, update.data(), sizeof(Chunk)); // memcpy(chunk, update.data(), sizeof(Chunk));
chunk = (Chunk*)(update.data()); chunk = (Chunk*)(update.data());
// std::cerr << "New chunk: " << chunkTime(*chunk) << "\t" << timeToStr(now) << "\t" << getAge(*chunk) << "\n"; // std::cerr << "New chunk: " << chunkTime(*chunk) << "\t" << timeToStr(now) << "\t" << getAge(*chunk) << "\n";
mutex.lock();
stream->addChunk(chunk); stream->addChunk(chunk);
mutex.unlock();
} }
return 0; return 0;
} }

162
stream.cpp Normal file
View file

@ -0,0 +1,162 @@
#include "stream.h"
#include "timeUtils.h"
#include <iostream>
#include <string.h>
Stream::Stream() : lastPlayerChunk(NULL), median(0), shortMedian(0), lastUpdate(0), skip(0), idx(0)
{
silentPlayerChunk = new PlayerChunk();
pBuffer = new DoubleBuffer<int>(30000 / PLAYER_CHUNK_MS);
pShortBuffer = new DoubleBuffer<int>(5000 / PLAYER_CHUNK_MS);
pLock = new std::unique_lock<std::mutex>(mtx);
bufferMs = 500;
}
void Stream::addChunk(Chunk* chunk)
{
Chunk* c = new Chunk(*chunk);
chunks.push_back(c);
cv.notify_all();
}
Chunk* Stream::getNextChunk()
{
Chunk* chunk = NULL;
if (chunks.empty())
cv.wait(*pLock);
chunk = chunks.front();
return chunk;
}
PlayerChunk* Stream::getNextPlayerChunk(int correction)
{
Chunk* chunk = getNextChunk();
if (correction > PLAYER_CHUNK_MS / 2)
correction = PLAYER_CHUNK_MS/2;
else if (correction < -PLAYER_CHUNK_MS/2)
correction = -PLAYER_CHUNK_MS/2;
//std::cerr << "GetNextPlayerChunk: " << correction << "\n";
// int age(0);
// age = getAge(*chunk) + outputBufferDacTime*1000 - bufferMs;
// std::cerr << "age: " << age << " \tidx: " << chunk->idx << "\n";
PlayerChunk* playerChunk = new PlayerChunk();
playerChunk->tv_sec = chunk->tv_sec;
playerChunk->tv_usec = chunk->tv_usec;
playerChunk->idx = 0;
size_t missing = PLAYER_CHUNK_SIZE;// + correction*PLAYER_CHUNK_MS_SIZE;
/* double factor = (double)PLAYER_CHUNK_MS / (double)(PLAYER_CHUNK_MS + correction);
size_t idx(0);
size_t idxCorrection(0);
for (size_t n=0; n<PLAYER_CHUNK_SIZE/2; ++n)
{
idx = chunk->idx + 2*floor(n*factor) - idxCorrection;
//std::cerr << factor << "\t" << n << "\t" << idx << "\n";
if (idx >= WIRE_CHUNK_SIZE)
{
idxCorrection = 2*floor(n*factor);
idx = 0;
chunks.pop_front();
delete chunk;
chunk = getNextChunk();
}
playerChunk->payload[2*n] = chunk->payload[idx];
playerChunk->payload[2*n+1] = chunk->payload[idx + 1];
}
addMs(chunk, -PLAYER_CHUNK_MS - correction);
chunk->idx = idx;
if (idx >= WIRE_CHUNK_SIZE)
{
chunks.pop_front();
delete chunk;
}
*/
if (chunk->idx + PLAYER_CHUNK_SIZE > WIRE_CHUNK_SIZE)
{
//std::cerr << "chunk->idx + PLAYER_CHUNK_SIZE >= WIRE_CHUNK_SIZE: " << chunk->idx + PLAYER_CHUNK_SIZE << " >= " << WIRE_CHUNK_SIZE << "\n";
memcpy(&(playerChunk->payload[0]), &chunk->payload[chunk->idx], sizeof(int16_t)*(WIRE_CHUNK_SIZE - chunk->idx));
missing = chunk->idx + PLAYER_CHUNK_SIZE - WIRE_CHUNK_SIZE;
chunks.pop_front();
delete chunk;
chunk = getNextChunk();
}
memcpy(&(playerChunk->payload[0]), &chunk->payload[chunk->idx], sizeof(int16_t)*missing);
addMs(chunk, -PLAYER_CHUNK_MS);
chunk->idx += missing;
if (chunk->idx >= WIRE_CHUNK_SIZE)
{
chunks.pop_front();
delete chunk;
}
return playerChunk;
}
PlayerChunk* Stream::getChunk(double outputBufferDacTime, unsigned long framesPerBuffer)
{
int correction(0);
if (pBuffer->full() && (abs(median) <= PLAYER_CHUNK_MS))
correction = median;
PlayerChunk* playerChunk = getNextPlayerChunk(correction);
int age = getAge(playerChunk) - bufferMs + outputBufferDacTime*1000;
pBuffer->add(age);
pShortBuffer->add(age);
// std::cerr << "Chunk: " << age << "\t" << outputBufferDacTime*1000 << "\n";
int sleep(0);
time_t now = time(NULL);
if (now != lastUpdate)
{
lastUpdate = now;
median = pBuffer->median();
shortMedian = pShortBuffer->median();
if (abs(age) > 300)
sleep = age;
if (pShortBuffer->full() && (abs(shortMedian) > WIRE_CHUNK_MS))
sleep = shortMedian;
if (pBuffer->full() && (abs(median) > PLAYER_CHUNK_MS))
sleep = median;
std::cerr << "Chunk: " << age << "\t" << shortMedian << "\t" << median << "\t" << pBuffer->size() << "\t" << outputBufferDacTime*1000 << "\n";
}
if (sleep != 0)
{
std::cerr << "Sleep: " << sleep << "\n";
pBuffer->clear();
pShortBuffer->clear();
if (sleep < 0)
{
sleepMs(100-sleep);
}
else
{
for (size_t n=0; n<(size_t)(sleep / PLAYER_CHUNK_MS); ++n)
{
delete playerChunk;
playerChunk = getNextPlayerChunk();
}
}
}
// int age = getAge(*lastPlayerChunk) + outputBufferDacTime*1000 - bufferMs;
return playerChunk;
}
void Stream::sleepMs(int ms)
{
if (ms > 0)
usleep(ms * 1000);
}

418
stream.h
View file

@ -7,414 +7,24 @@
#include <condition_variable> #include <condition_variable>
#include <vector> #include <vector>
#include "doubleBuffer.h" #include "doubleBuffer.h"
#include "timeUtils.h"
#include "chunk.h" #include "chunk.h"
#include "timeUtils.h"
class Stream class Stream
{ {
public: public:
Stream() : lastPlayerChunk(NULL), median(0), shortMedian(0), lastUpdate(0), skip(0), idx(0) Stream();
{ void addChunk(Chunk* chunk);
silentPlayerChunk = new PlayerChunk(); Chunk* getNextChunk();
pBuffer = new DoubleBuffer<int>(30000 / PLAYER_CHUNK_MS); PlayerChunk* getNextPlayerChunk(int correction = 0);
pShortBuffer = new DoubleBuffer<int>(5000 / PLAYER_CHUNK_MS); PlayerChunk* getChunk(double outputBufferDacTime, unsigned long framesPerBuffer);
pLock = new std::unique_lock<std::mutex>(mtx);
bufferMs = 500;
}
void addChunk(Chunk* chunk)
{
Chunk* c = new Chunk(*chunk);
mutex.lock();
chunks.push_back(c);
mutex.unlock();
cv.notify_all();
}
Chunk* getNextChunk()
{
Chunk* chunk = NULL;
if (chunks.empty())
cv.wait(*pLock);
mutex.lock();
chunk = chunks.front();
mutex.unlock();
return chunk;
}
PlayerChunk* getNextPlayerChunk(int correction = 0)
{
Chunk* chunk = getNextChunk();
if (correction > PLAYER_CHUNK_MS / 2)
correction = PLAYER_CHUNK_MS/2;
else if (correction < -PLAYER_CHUNK_MS/2)
correction = -PLAYER_CHUNK_MS/2;
//std::cerr << "GetNextPlayerChunk: " << correction << "\n";
// int age(0);
// age = getAge(*chunk) + outputBufferDacTime*1000 - bufferMs;
// std::cerr << "age: " << age << " \tidx: " << chunk->idx << "\n";
PlayerChunk* playerChunk = new PlayerChunk();
playerChunk->tv_sec = chunk->tv_sec;
playerChunk->tv_usec = chunk->tv_usec;
playerChunk->idx = 0;
size_t missing = PLAYER_CHUNK_SIZE;// + correction*PLAYER_CHUNK_MS_SIZE;
/* double factor = (double)PLAYER_CHUNK_MS / (double)(PLAYER_CHUNK_MS + correction);
size_t idx(0);
size_t idxCorrection(0);
for (size_t n=0; n<PLAYER_CHUNK_SIZE/2; ++n)
{
idx = chunk->idx + 2*floor(n*factor) - idxCorrection;
//std::cerr << factor << "\t" << n << "\t" << idx << "\n";
if (idx >= WIRE_CHUNK_SIZE)
{
idxCorrection = 2*floor(n*factor);
idx = 0;
chunks.pop_front();
delete chunk;
chunk = getNextChunk();
}
playerChunk->payload[2*n] = chunk->payload[idx];
playerChunk->payload[2*n+1] = chunk->payload[idx + 1];
}
addMs(chunk, -PLAYER_CHUNK_MS - correction);
chunk->idx = idx;
if (idx >= WIRE_CHUNK_SIZE)
{
chunks.pop_front();
delete chunk;
}
*/
if (chunk->idx + PLAYER_CHUNK_SIZE > WIRE_CHUNK_SIZE)
{
//std::cerr << "chunk->idx + PLAYER_CHUNK_SIZE >= WIRE_CHUNK_SIZE: " << chunk->idx + PLAYER_CHUNK_SIZE << " >= " << WIRE_CHUNK_SIZE << "\n";
memcpy(&(playerChunk->payload[0]), &chunk->payload[chunk->idx], sizeof(int16_t)*(WIRE_CHUNK_SIZE - chunk->idx));
missing = chunk->idx + PLAYER_CHUNK_SIZE - WIRE_CHUNK_SIZE;
chunks.pop_front();
delete chunk;
chunk = getNextChunk();
}
memcpy(&(playerChunk->payload[0]), &chunk->payload[chunk->idx], sizeof(int16_t)*missing);
addMs(chunk, -PLAYER_CHUNK_MS);
chunk->idx += missing;
if (chunk->idx >= WIRE_CHUNK_SIZE)
{
chunks.pop_front();
delete chunk;
}
return playerChunk;
}
PlayerChunk* getChunk(double outputBufferDacTime, unsigned long framesPerBuffer)
{
int correction(0);
if (pBuffer->full() && (abs(median) <= PLAYER_CHUNK_MS))
correction = median;
PlayerChunk* playerChunk = getNextPlayerChunk(correction);
int age = getAge(playerChunk) - bufferMs + outputBufferDacTime*1000;
pBuffer->add(age);
pShortBuffer->add(age);
// std::cerr << "Chunk: " << age << "\t" << outputBufferDacTime*1000 << "\n";
int sleep(0);
time_t now = time(NULL);
if (now != lastUpdate)
{
lastUpdate = now;
median = pBuffer->median();
shortMedian = pShortBuffer->median();
if (abs(age) > 300)
sleep = age;
if (pShortBuffer->full() && (abs(shortMedian) > WIRE_CHUNK_MS))
sleep = shortMedian;
if (pBuffer->full() && (abs(median) > PLAYER_CHUNK_MS))
sleep = median;
std::cerr << "Chunk: " << age << "\t" << shortMedian << "\t" << median << "\t" << pBuffer->size() << "\t" << outputBufferDacTime*1000 << "\n";
}
if (sleep != 0)
{
std::cerr << "Sleep: " << sleep << "\n";
pBuffer->clear();
pShortBuffer->clear();
if (sleep < 0)
{
sleepMs(100-sleep);
}
else
{
for (size_t n=0; n<(size_t)(sleep / PLAYER_CHUNK_MS); ++n)
{
delete playerChunk;
playerChunk = getNextPlayerChunk();
}
}
}
// int age = getAge(*lastPlayerChunk) + outputBufferDacTime*1000 - bufferMs;
return playerChunk;
}
/* if (age < -500)
{
std::vector<short> v;
if (skip < 0)
{
++skip;
std::cerr << "chunk too new, sleeping\n";//age > WIRE_CHUNK_MS (" << age << " ms)\n";
for (size_t n=0; n<(size_t)PLAYER_CHUNK_SIZE; ++n)
v.push_back(chunk->payload[n]);
return v;
}
}
pBuffer->add(age);
pShortBuffer->add(age);
time_t now = time(NULL);
if (skip == 0)
{
if (now != lastUpdate)
{
lastUpdate = now;
median = pBuffer->median();
shortMedian = pShortBuffer->median();
if (abs(age) > 500)
skip = age / PLAYER_CHUNK_SIZE;
if (pShortBuffer->full() && (abs(shortMedian) > WIRE_CHUNK_MS))
skip = shortMedian / PLAYER_CHUNK_SIZE;
if (pBuffer->full() && (abs(median) > WIRE_CHUNK_MS))
skip = median / PLAYER_CHUNK_SIZE;
}
std::cerr << "Chunk: " << age << "\t" << shortMedian << "\t" << median << "\t" << pBuffer->size() << "\t" << chunkCount << "\t" << outputBufferDacTime*1000 << "\n";
}
std::vector<short> v;
if (skip < 0)
{
++skip;
std::cerr << "chunk too new, sleeping\n";//age > WIRE_CHUNK_MS (" << age << " ms)\n";
for (size_t n=0; n<(size_t)PLAYER_CHUNK_SIZE; ++n)
v.push_back(chunk->payload[n]);
return v;
}
for (size_t n=chunk->idx; n<chunk->idx + (size_t)PLAYER_CHUNK_SIZE; ++n)
{
v.push_back(chunk->payload[n]);
}
//std::cerr << "before: " << chunkTime(*chunk) << ", after: ";
addMs(*chunk, -PLAYER_CHUNK_MS);
//std::cerr << chunkTime(*chunk) << "\n";
chunk->idx += PLAYER_CHUNK_SIZE;
if (chunk->idx >= WIRE_CHUNK_SIZE)
{
//std::cerr << "Chunk played out, deleting\n";
chunks.pop_front();
delete chunk;
}
return v;
}
*/
/*
std::vector<short> getChunk(double outputBufferDacTime, unsigned long framesPerBuffer)
{
while (1)
{
Chunk* chunk = NULL;
if (chunks.empty())
cv.wait(*pLock);
int age(0);
int chunkCount(0);
mutex.lock();
chunk = chunks.front();
while (!chunks.empty())
{
chunk = chunks.front();
if (skip != 0)
{
pBuffer->clear();
pShortBuffer->clear();
shortMedian = 0;
median = 0;
}
if (skip > 0)
{
--skip;
std::cerr << "chunk too old, skipping\n";//age > WIRE_CHUNK_MS (" << age << " ms)\n";
delete chunk;
chunk = NULL;
chunks.pop_front();
}
else
break;
}
chunkCount = chunks.size();
mutex.unlock();
if (chunk == NULL)
{
std::cerr << "no chunks available\n";
continue;
}
age = getAge(*chunk) + outputBufferDacTime*1000 - bufferMs;
pBuffer->add(age);
pShortBuffer->add(age);
time_t now = time(NULL);
if (skip == 0)
{
if (now != lastUpdate)
{
lastUpdate = now;
median = pBuffer->median();
shortMedian = pShortBuffer->median();
if (abs(age) > 500)
skip = age / PLAYER_CHUNK_SIZE;
if (pShortBuffer->full() && (abs(shortMedian) > WIRE_CHUNK_MS))
skip = shortMedian / PLAYER_CHUNK_SIZE;
if (pBuffer->full() && (abs(median) > WIRE_CHUNK_MS))
skip = median / PLAYER_CHUNK_SIZE;
}
std::cerr << "Chunk: " << age << "\t" << shortMedian << "\t" << median << "\t" << pBuffer->size() << "\t" << chunkCount << "\t" << outputBufferDacTime*1000 << "\n";
}
std::vector<short> v;
if (skip < 0)
{
++skip;
std::cerr << "chunk too new, sleeping\n";//age > WIRE_CHUNK_MS (" << age << " ms)\n";
for (size_t n=0; n<(size_t)PLAYER_CHUNK_SIZE; ++n)
v.push_back(chunk->payload[n]);
return v;
}
for (size_t n=chunk->idx; n<chunk->idx + (size_t)PLAYER_CHUNK_SIZE; ++n)
{
v.push_back(chunk->payload[n]);
}
//std::cerr << "before: " << chunkTime(*chunk) << ", after: ";
addMs(*chunk, -PLAYER_CHUNK_MS);
//std::cerr << chunkTime(*chunk) << "\n";
chunk->idx += PLAYER_CHUNK_SIZE;
if (chunk->idx >= WIRE_CHUNK_SIZE)
{
//std::cerr << "Chunk played out, deleting\n";
chunks.pop_front();
delete chunk;
}
return v;
}
}
*/
/*
std::cerr << "age: " << getAge(*chunk) << "\t" << age << "\t" << pBuffer->size() << "\t" << chunkCount << "\n";
pBuffer->add(age);
pShortBuffer->add(age);
time_t now = time(NULL);
if (skip == 0)
{
int age = 0;
int median = 0;
int shortMedian = 0;
if (now != lastUpdate)
{
lastUpdate = now;
median = pBuffer->median();
shortMedian = pShortBuffer->median();
std::cerr << "age: " << getAge(*chunk) << "\t" << age << "\t" << shortMedian << "\t" << median << "\t" << pBuffer->size() << "\t" << chunkCount << "\t" << outputBufferDacTime*1000 << "\n";
}
if ((age > 500) || (age < -500))
skip = age / PLAYER_CHUNK_MS;
else if (pShortBuffer->full() && ((shortMedian > 100) || (shortMedian < -100)))
skip = shortMedian / PLAYER_CHUNK_MS;
else if (pBuffer->full() && ((median > 10) || (median < -10)))
skip = median / PLAYER_CHUNK_MS;
}
if (skip != 0)
{
// std::cerr << "age: " << getAge(*chunk) << "\t" << age << "\t" << shortMedian << "\t" << median << "\t" << buffer->size() << "\t" << outputBufferDacTime*1000 << "\n";
}
// bool silence = (age < -500) || (shortBuffer.full() && (shortMedian < -100)) || (buffer.full() && (median < -15));
if (skip > 0)
{
skip--;
chunks.pop_front();
delete chunk;
std::cerr << "packe too old, dropping\n";
pBuffer->clear();
pShortBuffer->clear();
usleep(100);
}
else if (skip < 0)
{
skip++;
chunk = new Chunk();//PlayerChunk();
memset(&(chunk->payload[0]), 0, sizeof(int16_t)*PLAYER_CHUNK_SIZE);
// std::cerr << "age < bufferMs (" << age << " < " << bufferMs << "), playing silence\n";
pBuffer->clear();
pShortBuffer->clear();
usleep(100);
break;
}
else
{
chunks.pop_front();
break;
}
}
std::vector<short> v;
for (size_t n=0; n<framesPerBuffer; ++n)
{
v.push_back(0);
v.push_back(0);
}
return v;
// return chunk;
}
*/
private: private:
void sleepMs(int ms) void sleepMs(int ms);
{
if (ms > 0)
usleep(ms * 1000);
}
std::deque<Chunk*> chunks; std::deque<Chunk*> chunks;
std::mutex mtx; std::mutex mtx;
std::mutex mutex;
std::unique_lock<std::mutex>* pLock; std::unique_lock<std::mutex>* pLock;
std::condition_variable cv; std::condition_variable cv;
DoubleBuffer<int>* pBuffer; DoubleBuffer<int>* pBuffer;
@ -432,22 +42,6 @@ private:
}; };
// std::cerr << chunk->tv_sec << "\t" << now.tv_sec << "\n";
/* for (size_t n=0; n<WIRE_CHUNK_MS/PLAYER_CHUNK_MS; ++n)
{
PlayerChunk* playerChunk = new PlayerChunk();
playerChunk->tv_sec = chunk->tv_sec;
playerChunk->tv_usec = chunk->tv_usec;
addMs(*playerChunk, n*PLAYER_CHUNK_MS);
memcpy(&(playerChunk->payload[0]), &chunk->payload[n*PLAYER_CHUNK_SIZE], sizeof(int16_t)*PLAYER_CHUNK_SIZE);
mutex.lock();
chunks.push_back(playerChunk);
mutex.unlock();
cv.notify_all();
}
*/
#endif #endif

1
timeUtils.h
View file

@ -2,6 +2,7 @@
#define TIME_UTILS_H #define TIME_UTILS_H
#include "chunk.h" #include "chunk.h"
#include <sys/time.h>
std::string timeToStr(const timeval& timestamp) std::string timeToStr(const timeval& timestamp)
{ {