Skip to content

refactor Streaming strategy to allow MP4 preview and rename legacy Streaming strategy to Sequential #249

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jolavillette wants to merge 2 commits into
base: master
Choose a base branch
from
Open

refactor Streaming strategy to allow MP4 preview and rename legacy Streaming strategy to Sequential #249

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aadd7a7
refactor Streaming strategy to allow MP4 preview and rename legacy St…
jolavillette Jan 23, 2026
0ac4b92
fix windows compilation: replace non-portable be32toh with rs_endian_fix
jolavillette Jan 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 40 additions & 1 deletion src/ft/ftchunkmap.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
#include "retroshare/rspeers.h"
#include "ftchunkmap.h"
#include "util/rstime.h"
#include "util/rsdebug.h"

static const uint32_t SOURCE_CHUNK_MAP_UPDATE_PERIOD = 60 ; //! TTL for chunkmap info
static const uint32_t INACTIVE_CHUNK_TIME_LAPSE = 3600 ; //! TTL for an inactive chunk
Expand Down Expand Up @@ -599,14 +600,41 @@ uint32_t ChunkMap::getAvailableChunk(const RsPeerId& peer_id,bool& map_is_too_ol
{
uint32_t chosen_chunk_number ;

// Check high priority chunks first (regardless of strategy)
if (!_high_priority_chunks.empty())
{
uint32_t j = 0;
for(uint32_t i=0; i<_map.size(); ++i)
{
if(_map[i] == FileChunksInfo::CHUNK_OUTSTANDING && (peer_chunks->is_full || peer_chunks->cmap[i]))
{
if (i < _high_priority_chunks.size() && _high_priority_chunks[i])
{
RsDbg() << "STREAMING: ChunkMap::getAvailableChunk: PRIORITY returning chunk " << i << " for peer " << peer_id;
return i;
}
else
j++;
}
}
}

switch(_strategy)
{
case FileChunksInfo::CHUNK_STRATEGY_STREAMING: chosen_chunk_number = 0 ;
case FileChunksInfo::CHUNK_STRATEGY_SEQUENTIAL: chosen_chunk_number = 0 ;
break ;
case FileChunksInfo::CHUNK_STRATEGY_RANDOM: chosen_chunk_number = rand() % available_chunks ;
break ;
case FileChunksInfo::CHUNK_STRATEGY_PROGRESSIVE: chosen_chunk_number = rand() % std::min(available_chunks, available_chunks_before_max_dist+FT_CHUNKMAP_MAX_CHUNK_JUMP) ;
break ;
case FileChunksInfo::CHUNK_STRATEGY_STREAMING:
{
// Legacy heuristic removed.
// Now relies on _high_priority_chunks (checked above)
// or falls back to standard streaming (0).
chosen_chunk_number = 0 ;
}
break ;
default:
chosen_chunk_number = 0 ;
}
Expand Down Expand Up @@ -705,4 +733,15 @@ void ChunkMap::buildPlainMap(uint64_t size, CompressedChunkMap& map)
map = CompressedChunkMap(n,~uint32_t(0)) ;
}

void ChunkMap::setHighPriorityRange(uint32_t startChunk, uint32_t endChunk)
{
RsDbg() << "STREAMING: ChunkMap::setHighPriorityRange " << startChunk << " -> " << endChunk;

if (_high_priority_chunks.size() != _map.size())
_high_priority_chunks.resize(_map.size(), false);

for (uint32_t i = startChunk; i <= endChunk && i < _high_priority_chunks.size(); ++i)
_high_priority_chunks[i] = true;
}


4 changes: 4 additions & 0 deletions src/ft/ftchunkmap.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,9 @@ class ChunkMap
virtual void getAvailabilityMap(CompressedChunkMap& cmap) const ;
void setAvailabilityMap(const CompressedChunkMap& cmap) ;

/// Sets a range of chunks to be downloaded with high priority (checked before strategy)
void setHighPriorityRange(uint32_t startChunk, uint32_t endChunk);

/// Removes the source availability map. The map
void removeFileSource(const RsPeerId& peer_id) ;

Expand Down Expand Up @@ -250,6 +253,7 @@ class ChunkMap
std::map<RsPeerId,Chunk> _active_chunks_feed ; //! vector of chunks being downloaded. Exactly 1 chunk per peer.
std::map<ChunkNumber,ChunkDownloadInfo> _slices_to_download ; //! list of (slice offset,slice size) currently being downloaded
std::vector<FileChunksInfo::ChunkState> _map ; //! vector of chunk state over the whole file
std::vector<bool> _high_priority_chunks; //! boolean mask for high priority chunks
std::map<RsPeerId,SourceChunksInfo> _peers_chunks_availability ; //! what does each source peer have
uint64_t _total_downloaded ; //! completion for the file
bool _file_is_complete ; //! set to true when the file is complete.
Expand Down
9 changes: 8 additions & 1 deletion src/ft/ftcontroller.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1877,10 +1877,12 @@ bool ftController::saveList(bool &cleanup, std::list<RsItem *>& saveData)

switch(mDefaultChunkStrategy)
{
case FileChunksInfo::CHUNK_STRATEGY_STREAMING: configMap[default_chunk_strategy_ss] = "STREAMING" ;
case FileChunksInfo::CHUNK_STRATEGY_SEQUENTIAL: configMap[default_chunk_strategy_ss] = "SEQUENTIAL" ;
break ;
case FileChunksInfo::CHUNK_STRATEGY_RANDOM: configMap[default_chunk_strategy_ss] = "RANDOM" ;
break ;
case FileChunksInfo::CHUNK_STRATEGY_STREAMING: configMap[default_chunk_strategy_ss] = "STREAMING" ;
break ;

default:
case FileChunksInfo::CHUNK_STRATEGY_PROGRESSIVE:configMap[default_chunk_strategy_ss] = "PROGRESSIVE" ;
Expand Down Expand Up @@ -2165,6 +2167,11 @@ bool ftController::loadConfigMap(std::map<std::string, std::string> &configMap)
setDefaultChunkStrategy(FileChunksInfo::CHUNK_STRATEGY_STREAMING) ;
std::cerr << "Note: loading default value for chunk strategy: streaming" << std::endl;
}
else if(mit->second == "SEQUENTIAL")
{
setDefaultChunkStrategy(FileChunksInfo::CHUNK_STRATEGY_SEQUENTIAL) ;
std::cerr << "Note: loading default value for chunk strategy: sequential" << std::endl;
}
else if(mit->second == "RANDOM")
{
setDefaultChunkStrategy(FileChunksInfo::CHUNK_STRATEGY_RANDOM) ;
Expand Down
107 changes: 104 additions & 3 deletions src/ft/ftfilecreator.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
#include <cerrno>
#include <cstdio>
#include <sys/stat.h>
#include "util/rsdebug.h"
#include "util/rsendian.h"

#include "ftfilecreator.h"
#include "util/rstime.h"
Expand All @@ -49,8 +51,9 @@
***********************************************************/

ftFileCreator::ftFileCreator(const std::string& path, uint64_t size, const RsFileHash& hash,bool assume_availability)
: ftFileProvider(path,size,hash), chunkMap(size,assume_availability)
: ftFileProvider(path,size,hash), chunkMap(size,assume_availability), _mp4_index_found(false), _mp4_index_offset(0)
{
RsDbg() << "STREAMING: ftFileCreator CONSTRUCTOR for " << path;
/*
* FIXME any inits to do?
*/
Expand Down Expand Up @@ -239,6 +242,13 @@ bool ftFileCreator::addFileData(uint64_t offset, uint32_t chunk_size, void *data
* Notify ftFileChunker about chunks received
*/
locked_notifyReceived(offset,chunk_size);

// MP4 Smart Preview hook
RsDbg() << "STREAMING: addFileData offset=" << offset << " strat=" << chunkMap.getStrategy();
if (!_mp4_index_found)
{
checkForMp4Index();
}

complete = chunkMap.isComplete();
}
Expand Down Expand Up @@ -497,15 +507,16 @@ void ftFileCreator::setChunkStrategy(FileChunksInfo::ChunkStrategy s)
RsStackMutex stack(ftcMutex); /********** STACK LOCKED MTX ******/

// Let's check, for safety.
if(s != FileChunksInfo::CHUNK_STRATEGY_STREAMING && s != FileChunksInfo::CHUNK_STRATEGY_RANDOM && s != FileChunksInfo::CHUNK_STRATEGY_PROGRESSIVE)
if(s != FileChunksInfo::CHUNK_STRATEGY_SEQUENTIAL && s != FileChunksInfo::CHUNK_STRATEGY_RANDOM && s != FileChunksInfo::CHUNK_STRATEGY_PROGRESSIVE && s != FileChunksInfo::CHUNK_STRATEGY_STREAMING)
{
std::cerr << "ftFileCreator::ERROR: invalid chunk strategy " << s << "!" << " setting default value " << FileChunksInfo::CHUNK_STRATEGY_STREAMING << std::endl ;
std::cerr << "ftFileCreator::ERROR: invalid chunk strategy " << s << "!" << " setting default value " << FileChunksInfo::CHUNK_STRATEGY_SEQUENTIAL << std::endl ;
s = FileChunksInfo::CHUNK_STRATEGY_PROGRESSIVE ;
}

#ifdef FILE_DEBUG
std::cerr << "ftFileCtreator: setting chunk strategy to " << s << std::endl ;
#endif
RsDbg() << "STREAMING: ftFileCreator::setChunkStrategy " << s;
chunkMap.setStrategy(s) ;
}

Expand Down Expand Up @@ -761,5 +772,95 @@ bool ftFileCreator::verifyChunk(uint32_t chunk_number,const Sha1CheckSum& sum)
return true ;
}

bool ftFileCreator::checkForMp4Index()
{
int strat = (int)chunkMap.getStrategy();

// STRICT RULE: Only active in STREAMING_PRIO_END
if (strat != FileChunksInfo::CHUNK_STRATEGY_STREAMING)
{
// RsDbg() << "STREAMING: ftFileCreator::checkForMp4Index ignored. Strategy=" << strat;
return false;
}

// Phase 1: Just log that we passed the guard
RsDbg() << "STREAMING: ftFileCreator::checkForMp4Index RUNNING. Strategy=" << strat << ". Parsing loop start.";

// Phase 2: Atom Parsing (Observer Mode)
if (_mp4_index_found) return true;

// Open file to read atoms
FILE* f = fopen(file_name.c_str(), "rb");
if (!f)
{
RsDbg() << "STREAMING: Failed to open file " << file_name;
return false;
}

uint64_t currentPos = 0;

// MP4 Atom Header
struct {
uint32_t size;
char type[4];
} header;

int safe_loop_count = 0;

while (true)
{
if (fseek(f, currentPos, SEEK_SET) != 0 || fread(&header, 1, 8, f) != 8)
break;

uint32_t atomSize = rs_endian_fix(header.size);
uint64_t realAtomSize = atomSize;

if (atomSize == 1) {
uint64_t bigSize;
if (fread(&bigSize, 1, 8, f) == 8) realAtomSize = rs_endian_fix(bigSize);
}

// Create a null-terminated string for logging safely
char typeStr[5] = {0};
memcpy(typeStr, header.type, 4);

RsDbg() << "STREAMING: Found atom '" << typeStr << "' at " << currentPos << " size " << realAtomSize;

if (strncmp(header.type, "moov", 4) == 0) {
RsDbg() << "STREAMING: MOOV atom found at " << currentPos << " (Beginning of file?). Stop.";
_mp4_index_found = true;
break;
}

if (strncmp(header.type, "mdat", 4) == 0) {
uint64_t predictedMoov = currentPos + realAtomSize;
RsDbg() << "STREAMING: MDAT found. Size: " << realAtomSize << ". Predicted MOOV at: " << predictedMoov;

// Phase 3: Actuation
// Calculate chunks covering the MOOV index (from predictedMoov to end of file)
uint32_t chunkSize = ChunkMap::CHUNKMAP_FIXED_CHUNK_SIZE;
uint32_t startChunk = predictedMoov / chunkSize;
uint32_t endChunk = mSize / chunkSize;

RsDbg() << "STREAMING: Setting High Priority Range: " << startChunk << " -> " << endChunk;
chunkMap.setHighPriorityRange(startChunk, endChunk);

_mp4_index_found = true;
break;
}

currentPos += realAtomSize;
if (realAtomSize == 0 || currentPos >= mSize) break;

if (++safe_loop_count > 50) {
RsDbg() << "STREAMING: Safety break (too many atoms)";
break;
}
}

fclose(f);
return false;
}



6 changes: 6 additions & 0 deletions src/ft/ftfilecreator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,8 +144,14 @@ class ftFileCreator: public ftFileProvider

ChunkMap chunkMap ;


rstime_t _last_recv_time_t ; /// last time stamp when data was received. Used for queue control.
rstime_t _creation_time ; /// time at which the file creator was created. Used to spot long-inactive transfers.

// MP4 Smart Preview
bool _mp4_index_found;
uint64_t _mp4_index_offset;
bool checkForMp4Index();
};

#endif // FT_FILE_CREATOR_HEADER
5 changes: 3 additions & 2 deletions src/retroshare/rstypes.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,9 +353,10 @@ struct FileChunksInfo : RsSerializable

enum ChunkStrategy : uint8_t
{
CHUNK_STRATEGY_STREAMING,
CHUNK_STRATEGY_SEQUENTIAL,
CHUNK_STRATEGY_RANDOM,
CHUNK_STRATEGY_PROGRESSIVE
CHUNK_STRATEGY_PROGRESSIVE,
CHUNK_STRATEGY_STREAMING
};

struct SliceInfo : RsSerializable
Expand Down
Loading