Use stream compression.

Previously each data packet was compressed independently. After this
change all new packets reference the previously sent data, which
achieves better compression.
This commit is contained in:
Bartosz Taudul 2017-09-17 13:10:42 +02:00
parent 4c2bd7d9df
commit d7914439e9
4 changed files with 39 additions and 9 deletions

View File

@ -8,7 +8,6 @@
#include <memory>
#include <limits>
#include "../common/tracy_lz4.hpp"
#include "../common/TracyProtocol.hpp"
#include "../common/TracySocket.hpp"
#include "../common/TracySystem.hpp"
@ -31,6 +30,9 @@ Profiler::Profiler()
: m_timeBegin( GetTime() )
, m_shutdown( false )
, m_id( 0 )
, m_stream( LZ4_createStream() )
, m_buffer( new char[TargetFrameSize*3] )
, m_bufferOffset( 0 )
{
assert( PointerCheckA == PointerCheckB );
assert( !s_instance );
@ -45,9 +47,11 @@ Profiler::~Profiler()
m_shutdown.store( true, std::memory_order_relaxed );
m_thread.join();
delete[] m_buffer;
LZ4_freeStream( m_stream );
assert( s_instance );
s_instance = nullptr;
}
uint64_t Profiler::GetNewId()
@ -125,6 +129,8 @@ void Profiler::Worker()
m_sock->Send( &val, 1 );
#endif
LZ4_resetStream( m_stream );
for(;;)
{
if( m_shutdown.load( std::memory_order_relaxed ) ) return;
@ -133,8 +139,8 @@ void Profiler::Worker()
const auto sz = m_queue.try_dequeue_bulk( token, item, BulkSize );
if( sz > 0 )
{
char buf[TargetFrameSize];
char* ptr = buf;
auto buf = m_buffer + m_bufferOffset;
auto ptr = buf;
for( int i=0; i<sz; i++ )
{
const auto dsz = QueueDataSize[item[i].hdr.idx];
@ -142,6 +148,8 @@ void Profiler::Worker()
ptr += dsz;
}
if( !SendData( buf, ptr - buf ) ) break;
m_bufferOffset += ptr - buf;
if( m_bufferOffset > TargetFrameSize * 2 ) m_bufferOffset = 0;
}
else
{
@ -164,7 +172,7 @@ bool Profiler::SendData( const char* data, size_t len )
if( m_sock->Send( data, len ) == -1 ) return false;
#else
char lz4[LZ4Size + sizeof( lz4sz_t )];
const lz4sz_t lz4sz = LZ4_compress_default( data, lz4 + sizeof( lz4sz_t ), len, LZ4Size );
const lz4sz_t lz4sz = LZ4_compress_fast_continue( m_stream, data, lz4 + sizeof( lz4sz_t ), len, LZ4Size, 1 );
memcpy( lz4, &lz4sz, sizeof( lz4sz ) );
if( m_sock->Send( lz4, lz4sz + sizeof( lz4sz_t ) ) == -1 ) return false;
#endif
@ -179,7 +187,7 @@ bool Profiler::SendString( uint64_t str )
hdr.type = QueueType::StringData;
hdr.id = str;
char buf[TargetFrameSize];
auto buf = m_buffer + m_bufferOffset;
memcpy( buf, &hdr, sizeof( hdr ) );
auto len = strlen( ptr );
@ -189,6 +197,9 @@ bool Profiler::SendString( uint64_t str )
memcpy( buf + sizeof( hdr ), &l16, sizeof( l16 ) );
memcpy( buf + sizeof( hdr ) + sizeof( l16 ), ptr, l16 );
m_bufferOffset += sizeof( hdr ) + sizeof( l16 ) + l16;
if( m_bufferOffset > TargetFrameSize * 2 ) m_bufferOffset = 0;
return SendData( buf, sizeof( hdr ) + sizeof( l16 ) + l16 );
}

View File

@ -7,6 +7,7 @@
#include <thread>
#include "concurrentqueue.h"
#include "../common/tracy_lz4.hpp"
#include "../common/TracyQueue.hpp"
namespace tracy
@ -52,6 +53,10 @@ private:
moodycamel::ConcurrentQueue<QueueItem> m_queue;
std::atomic<uint64_t> m_id;
std::unique_ptr<Socket> m_sock;
LZ4_stream_t* m_stream;
char* m_buffer;
int m_bufferOffset;
};
};

View File

@ -8,7 +8,6 @@
#include <assert.h>
#include <limits>
#include "../common/tracy_lz4.hpp"
#include "../common/TracyProtocol.hpp"
#include "../common/TracySystem.hpp"
#include "../common/TracyQueue.hpp"
@ -24,6 +23,9 @@ View::View( const char* addr )
: m_addr( addr )
, m_shutdown( false )
, m_mbps( 64 )
, m_stream( LZ4_createStreamDecode() )
, m_buffer( new char[TargetFrameSize*3] )
, m_bufferOffset( 0 )
{
assert( s_instance == nullptr );
s_instance = this;
@ -37,6 +39,9 @@ View::~View()
m_shutdown.store( true, std::memory_order_relaxed );
m_thread.join();
delete[] m_buffer;
LZ4_freeStreamDecode( m_stream );
assert( s_instance != nullptr );
s_instance = nullptr;
}
@ -67,6 +72,7 @@ void View::Worker()
if( !m_sock.Read( &lz4, sizeof( lz4 ), &tv, ShouldExit ) ) goto close;
m_frames.push_back( timeStart );
LZ4_setStreamDecode( m_stream, nullptr, 0 );
t0 = std::chrono::high_resolution_clock::now();
@ -76,14 +82,14 @@ void View::Worker()
if( lz4 )
{
char buf[TargetFrameSize];
auto buf = m_buffer + m_bufferOffset;
char lz4buf[LZ4Size];
lz4sz_t lz4sz;
if( !m_sock.Read( &lz4sz, sizeof( lz4sz ), &tv, ShouldExit ) ) goto close;
if( !m_sock.Read( lz4buf, lz4sz, &tv, ShouldExit ) ) goto close;
bytes += sizeof( lz4sz ) + lz4sz;
auto sz = LZ4_decompress_safe( lz4buf, buf, lz4sz, TargetFrameSize );
auto sz = LZ4_decompress_safe_continue( m_stream, lz4buf, buf, lz4sz, TargetFrameSize );
assert( sz >= 0 );
const char* ptr = buf;
@ -93,6 +99,9 @@ void View::Worker()
auto ev = (QueueItem*)ptr;
DispatchProcess( *ev, ptr );
}
m_bufferOffset += sz;
if( m_bufferOffset > TargetFrameSize * 2 ) m_bufferOffset = 0;
}
else
{

View File

@ -9,6 +9,7 @@
#include <unordered_set>
#include <vector>
#include "../common/tracy_lz4.hpp"
#include "../common/TracySocket.hpp"
#include "../common/TracyQueue.hpp"
#include "TracyEvent.hpp"
@ -70,6 +71,10 @@ private:
std::unordered_set<uint64_t> m_pendingStrings;
Slab<EventSize*1024*1024> m_slab;
LZ4_streamDecode_t* m_stream;
char* m_buffer;
int m_bufferOffset;
};
}