Loading...
Searching...
No Matches
stream_expand_support.h
Go to the documentation of this file.
//***************************************************************************/
// This software is released under the 2-Clause BSD license, included
// below.
//
// Copyright (c) 2024, Aous Naman
// Copyright (c) 2024, Kakadu Software Pty Ltd, Australia
// Copyright (c) 2024, The University of New South Wales, Australia
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
// IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
// TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
// PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
// TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//***************************************************************************/
// This file is part of the OpenJPH software implementation.
// File: stream_expand_support.cpp
// Author: Aous Naman
// Date: 18 April 2024
//***************************************************************************/
#ifndef OJPH_STR_EX_SUPPORT_H
#define OJPH_STR_EX_SUPPORT_H
#include <atomic>
#include <cassert>
#include "ojph_base.h"
#include "ojph_file.h"
#include "ojph_sockets.h"
namespace ojph
{
namespace thds
{ class thread_pool; }
namespace stex // stream expand
{
// defined here
class packets_handler;
class frames_handler;
// defined elsewhere
struct j2k_frame_storer;
//
//
//
//
//
/*****************************************************************************/
struct rtp_packet
{
enum packet_type : ui32
{
PT_BODY = 0, // this is body packet
PT_MAIN_FOLLOWED_BY_MAIN = 1,
PT_MAIN_FOLLOWED_BY_BODY = 2,
PT_MAIN = 3, // frame has only one main packet
};
public:
rtp_packet() { num_bytes = 0; next = NULL; }
void init(rtp_packet* next) { this->next = next; }
public:
// RTP header
ui32 get_rtp_version() { return ((ui32)data[0]) >> 6; }
bool is_padded() { return (data[0] & 0x20) != 0; }
bool is_extended() { return (data[0] & 0x10) != 0; }
ui32 get_csrc_count() { return (ui32)(data[0]) & 0xF; }
bool is_marked() { return (data[1] & 0x80) != 0; }
ui32 get_payload_type() { return (ui32)(data[1]) & 0x7F; }
ui32 get_seq_num() {
ui32 result = ntohs(*(ui16*)(data + 2));
result |= ((ui32)data[15]) << 16; // extended sequence (ESEQ)
return result;
}
ui32 get_time_stamp()
{ return ntohl(*(ui32*)(data + 4)); }
ui32 get_ssrc() // not used for the time being
{ return ntohl(*(ui32*)(data + 8)); }
// common in main and body payload headers
ui32 get_packet_type()
{ return ((ui32)data[12]) >> 6; }
ui32 get_TP()
{ return (((ui32)data[12]) >> 3) & 0x7; }
ui32 get_ORDH() {
if (get_packet_type() != PT_BODY) return ((ui32)data[12]) & 0x7;
else return (((ui32)data[13]) >> 7) & 0x1;
}
ui32 get_PTSTAMP() {
ui32 result = (((ui32)data[13]) & 0xF) << 8;
result |= (ui32)data[14];
return result;
}
ui8* get_data()
{ return data + 20; }
ui32 get_data_size()
{ return (ui32)num_bytes - 20; }
// only in main payload header
bool is_PTSTAMP_used() {
assert(get_packet_type() != PT_BODY);
return (((ui32)data[13]) & 0x80) != 0;
}
ui32 get_XTRAC() {
assert(get_packet_type() != PT_BODY);
return (((ui32)data[13]) >> 4) & 0x7;
}
bool is_codestream_header_reusable() {
assert(get_packet_type() != PT_BODY);
return (((ui32)data[16]) & 0x80) != 0;
}
bool is_component_colorimetry_used() {
assert(get_packet_type() != PT_BODY);
return (((ui32)data[16]) & 0x40) != 0;
}
bool is_codeblock_caching_used() {
assert(get_packet_type() != PT_BODY);
return (((ui32)data[16]) & 0x20) != 0;
}
bool is_RANGE() {
assert(get_packet_type() != PT_BODY);
return ((ui32)data[16] & 1) != 0;
}
ui32 get_PRIMS(){
assert(get_packet_type() != PT_BODY);
return (ui32)data[17];
}
ui32 get_TRANS() {
assert(get_packet_type() != PT_BODY);
return (ui32)data[18];
}
ui32 get_MAT() {
assert(get_packet_type() != PT_BODY);
return (ui32)data[19];
}
// only in body payload header
ui32 get_RES() {
assert(get_packet_type() == PT_BODY);
return ((ui32)data[12]) & 0x7;
}
ui32 get_QUAL() {
assert(get_packet_type() == PT_BODY);
return (((ui32)data[13]) >> 4) & 0x7;
}
ui32 get_data_pos() {
ui32 result = 0;
if (get_packet_type() == PT_BODY) {
result = ((ui32)data[16]) << 4;
result |= (((ui32)data[17]) >> 4) & 0xF;
}
return result;
}
ui32 get_PID() {
assert(get_packet_type() == PT_BODY);
ui32 result = (((ui32)data[17]) & 0xF) << 16;
result |= ((ui32)data[18]) << 8;
result |= ((ui32)data[19]);
return result;
}
public:
static constexpr int max_size = 2048;
// ethernet packet are only 1500
ui8 data[max_size];
ui32 num_bytes;
rtp_packet* next;
};
//
//
//
//
//
/*****************************************************************************/
class packets_handler
{
public:
packets_handler()
{
quiet = false;
avail = in_use = NULL;
last_seq_num = lost_packets = 0;
frames = NULL;
num_packets = 0;
packet_store = NULL;
}
~packets_handler()
{ if (packet_store) delete[] packet_store; }
public:
void init(bool quiet, ui32 num_packets, frames_handler* frames);
rtp_packet* exchange(rtp_packet* p);
ui32 get_num_lost_packets() const { return lost_packets; }
void flush();
private:
void consume_packet();
private:
bool quiet;
rtp_packet* avail;
rtp_packet* in_use;
ui32 last_seq_num;
ui32 lost_packets;
frames_handler* frames;
ui32 num_packets;
rtp_packet* packet_store;
};
//
//
//
//
//
/*****************************************************************************/
struct stex_file {
public:
stex_file()
{
time_stamp = last_seen_seq = 0;
done.store(0, std::memory_order_relaxed);
frame_idx = 0;
parent = NULL;
name_template = NULL;
storer = NULL;
next = NULL;
}
public:
void init(frames_handler* parent, stex_file* next, j2k_frame_storer *storer,
const char *name_template)
{
this->parent = parent;
this->name_template = name_template;
this->next = next;
this->storer = storer;
}
void notify_file_completion();
public:
ojph::mem_outfile f;
ui32 time_stamp;
ui32 last_seen_seq;
std::atomic_int done;
ui32 frame_idx;
frames_handler* parent;
const char *name_template;
j2k_frame_storer* storer;
stex_file* next;
};
//
//
//
//
//
/*****************************************************************************/
class frames_handler
{
public:
frames_handler()
{
quiet = false;
num_threads = 0;
target_name = NULL;
num_files = 0;
last_seq_number = last_time_stamp = 0;
total_frames = trunc_frames = lost_frames = 0;
files_store = in_use = avail = processing = NULL;
num_complete_files.store(0);
thread_pool = NULL;
storers_store = NULL;
}
~frames_handler();
public:
void init(bool quiet, const char *target_name,
thds::thread_pool* thread_pool);
void push(rtp_packet* p);
void get_stats(ui32& total_frames, ui32& trunc_frames, ui32& lost_frames);
bool flush();
void increment_num_complete_files()
{ num_complete_files.fetch_add(1, std::memory_order_release); }
private:
void check_files_in_processing();
void send_to_processing();
private:
bool quiet;
ui32 num_threads;
const char *target_name;
ui32 num_files;
ui32 last_seq_number;
ui32 last_time_stamp;
ui32 total_frames;
ui32 trunc_frames;
ui32 lost_frames;
stex_file* files_store;
stex_file* in_use;
stex_file* avail;
stex_file* processing;
std::atomic_int32_t
num_complete_files; //<!num. of files for which processing is complete
thds::thread_pool*
thread_pool;
j2k_frame_storer*
storers_store;
};
} // !stex namespace
} // !ojph namespace
#endif
ojph::mem_outfile
mem_outfile stores encoded j2k codestreams in memory
Definition ojph_file.h:127
ojph::stex::frames_handler
Assumes packets arrive in order.
Definition stream_expand_support.h:429
ojph::stex::frames_handler::num_threads
ui32 num_threads
number of threads used for saving
Definition stream_expand_support.h:531
ojph::stex::frames_handler::storers_store
j2k_frame_storer * storers_store
address for allocated frame storers
Definition stream_expand_support.h:548
ojph::stex::frames_handler::total_frames
ui32 total_frames
total number of frames that were observed
Definition stream_expand_support.h:536
ojph::stex::frames_handler::in_use
stex_file * in_use
the frame that is being filled with data
Definition stream_expand_support.h:540
ojph::stex::frames_handler::push
void push(rtp_packet *p)
call this function to push rtp_packets to this object
Definition stream_expand_support.cpp:291
ojph::stex::frames_handler::frames_handler
frames_handler()
default construction
Definition stream_expand_support.h:434
ojph::stex::frames_handler::flush
bool flush()
This function is not used, and therefore it is not clear how to use it.
Definition stream_expand_support.cpp:386
ojph::stex::frames_handler::get_stats
void get_stats(ui32 &total_frames, ui32 &trunc_frames, ui32 &lost_frames)
call this function to collect statistics about frames
Definition stream_expand_support.cpp:377
ojph::stex::frames_handler::processing
stex_file * processing
frames that are being saved
Definition stream_expand_support.h:542
ojph::stex::frames_handler::avail
stex_file * avail
available frames structures
Definition stream_expand_support.h:541
ojph::stex::frames_handler::last_time_stamp
ui32 last_time_stamp
last observed time stamp
Definition stream_expand_support.h:535
ojph::stex::frames_handler::last_seq_number
ui32 last_seq_number
last observed sequence number
Definition stream_expand_support.h:534
ojph::stex::frames_handler::trunc_frames
ui32 trunc_frames
truncated frames (because of a packet lostt)
Definition stream_expand_support.h:537
ojph::stex::frames_handler::check_files_in_processing
void check_files_in_processing()
call this function to process stex_file for which processing is complete
Definition stream_expand_support.cpp:405
ojph::stex::frames_handler::increment_num_complete_files
void increment_num_complete_files()
other threads call this function to let frames_handler know that processing is done.
Definition stream_expand_support.h:506
ojph::stex::frames_handler::~frames_handler
~frames_handler()
default destructor
Definition stream_expand_support.cpp:257
ojph::stex::frames_handler::send_to_processing
void send_to_processing()
Handles complete/truncated files and send them for storing.
Definition stream_expand_support.cpp:447
ojph::stex::frames_handler::lost_frames
ui32 lost_frames
frames for which main header was not received
Definition stream_expand_support.h:538
ojph::stex::frames_handler::init
void init(bool quiet, const char *target_name, thds::thread_pool *thread_pool)
call this function to initialize this object
Definition stream_expand_support.cpp:266
ojph::stex::frames_handler::target_name
const char * target_name
target file name template
Definition stream_expand_support.h:532
ojph::stex::frames_handler::thread_pool
thds::thread_pool * thread_pool
thread pool for processing frames
Definition stream_expand_support.h:546
ojph::stex::frames_handler::quiet
bool quiet
no informational info is printed when true
Definition stream_expand_support.h:530
ojph::stex::frames_handler::num_complete_files
std::atomic_int32_t num_complete_files
Definition stream_expand_support.h:544
ojph::stex::frames_handler::num_files
ui32 num_files
maximum number of in-flight files.
Definition stream_expand_support.h:533
ojph::stex::frames_handler::files_store
stex_file * files_store
address for allocated files
Definition stream_expand_support.h:539
ojph::stex::packets_handler
Interprets new packets, buffers them if needed.
Definition stream_expand_support.h:248
ojph::stex::packets_handler::packet_store
rtp_packet * packet_store
address of packet memory allocation
Definition stream_expand_support.h:327
ojph::stex::packets_handler::~packets_handler
~packets_handler()
default destructor
Definition stream_expand_support.h:265
ojph::stex::packets_handler::exchange
rtp_packet * exchange(rtp_packet *p)
Call this function to get a packet from the packet chain.
Definition stream_expand_support.cpp:120
ojph::stex::packets_handler::frames
frames_handler * frames
frames object
Definition stream_expand_support.h:324
ojph::stex::packets_handler::consume_packet
void consume_packet()
This function sends the packet in in_use (oldest) to frames handler object.
Definition stream_expand_support.cpp:221
ojph::stex::packets_handler::get_num_lost_packets
ui32 get_num_lost_packets() const
This function provides information about the observed number of lost packets.
Definition stream_expand_support.h:302
ojph::stex::packets_handler::quiet
bool quiet
no informational info is printed when true
Definition stream_expand_support.h:319
ojph::stex::packets_handler::num_packets
ui32 num_packets
maximum number of packets in packet_store
Definition stream_expand_support.h:326
ojph::stex::packets_handler::in_use
rtp_packet * in_use
start of used packet chain
Definition stream_expand_support.h:321
ojph::stex::packets_handler::last_seq_num
ui32 last_seq_num
the last observed sequence number
Definition stream_expand_support.h:322
ojph::stex::packets_handler::lost_packets
ui32 lost_packets
number of lost packets – just statistics
Definition stream_expand_support.h:323
ojph::stex::packets_handler::packets_handler
packets_handler()
default constructor
Definition stream_expand_support.h:253
ojph::stex::packets_handler::flush
void flush()
This function is not used, and therefore it is not clear how to use it.
Definition stream_expand_support.cpp:208
ojph::stex::packets_handler::avail
rtp_packet * avail
start of available packets chain
Definition stream_expand_support.h:320
ojph::stex::packets_handler::init
void init(bool quiet, ui32 num_packets, frames_handler *frames)
call this to initialize packets_handler
Definition stream_expand_support.cpp:105
ojph::thds::thread_pool
Implements a pool of threads, and can queue tasks.
Definition ojph_threads.h:98
ojph::stex
Definition stream_expand_support.cpp:47
ojph::thds
Definition ojph_threads.h:51
ojph
Definition ojph_img_io.h:52
ojph::ui16
uint16_t ui16
Definition ojph_defs.h:52
ojph::ui32
uint32_t ui32
Definition ojph_defs.h:54
ojph::ui8
uint8_t ui8
Definition ojph_defs.h:50
ojph_base.h
ojph_file.h
ojph_sockets.h
ojph::stex::j2k_frame_storer
Saves a j2k frame to disk without decoding.
Definition threaded_frame_processors.h:65
ojph::stex::rtp_packet
interprets RTP header and payload, and holds received packets.
Definition stream_expand_support.h:79
ojph::stex::rtp_packet::num_bytes
ui32 num_bytes
number of bytes
Definition stream_expand_support.h:209
ojph::stex::rtp_packet::get_XTRAC
ui32 get_XTRAC()
Definition stream_expand_support.h:146
ojph::stex::rtp_packet::get_data_pos
ui32 get_data_pos()
Definition stream_expand_support.h:188
ojph::stex::rtp_packet::get_payload_type
ui32 get_payload_type()
Definition stream_expand_support.h:111
ojph::stex::rtp_packet::get_time_stamp
ui32 get_time_stamp()
Definition stream_expand_support.h:117
ojph::stex::rtp_packet::get_TP
ui32 get_TP()
Definition stream_expand_support.h:125
ojph::stex::rtp_packet::is_codestream_header_reusable
bool is_codestream_header_reusable()
Definition stream_expand_support.h:150
ojph::stex::rtp_packet::rtp_packet
rtp_packet()
default constructor
Definition stream_expand_support.h:95
ojph::stex::rtp_packet::get_PRIMS
ui32 get_PRIMS()
Definition stream_expand_support.h:166
ojph::stex::rtp_packet::get_seq_num
ui32 get_seq_num()
Definition stream_expand_support.h:112
ojph::stex::rtp_packet::get_PID
ui32 get_PID()
Definition stream_expand_support.h:196
ojph::stex::rtp_packet::get_TRANS
ui32 get_TRANS()
Definition stream_expand_support.h:170
ojph::stex::rtp_packet::get_QUAL
ui32 get_QUAL()
Definition stream_expand_support.h:184
ojph::stex::rtp_packet::get_MAT
ui32 get_MAT()
Definition stream_expand_support.h:174
ojph::stex::rtp_packet::get_ssrc
ui32 get_ssrc()
Definition stream_expand_support.h:119
ojph::stex::rtp_packet::get_RES
ui32 get_RES()
Definition stream_expand_support.h:180
ojph::stex::rtp_packet::is_extended
bool is_extended()
Definition stream_expand_support.h:108
ojph::stex::rtp_packet::get_data_size
ui32 get_data_size()
Definition stream_expand_support.h:138
ojph::stex::rtp_packet::get_csrc_count
ui32 get_csrc_count()
Definition stream_expand_support.h:109
ojph::stex::rtp_packet::get_data
ui8 * get_data()
Definition stream_expand_support.h:136
ojph::stex::rtp_packet::data
ui8 data[max_size]
data in the packet
Definition stream_expand_support.h:208
ojph::stex::rtp_packet::max_size
static constexpr int max_size
maximum packet size
Definition stream_expand_support.h:206
ojph::stex::rtp_packet::is_PTSTAMP_used
bool is_PTSTAMP_used()
Definition stream_expand_support.h:142
ojph::stex::rtp_packet::next
rtp_packet * next
used for linking packets
Definition stream_expand_support.h:210
ojph::stex::rtp_packet::is_padded
bool is_padded()
Definition stream_expand_support.h:107
ojph::stex::rtp_packet::init
void init(rtp_packet *next)
Call this to link packets.
Definition stream_expand_support.h:102
ojph::stex::rtp_packet::is_component_colorimetry_used
bool is_component_colorimetry_used()
Definition stream_expand_support.h:154
ojph::stex::rtp_packet::get_packet_type
ui32 get_packet_type()
Definition stream_expand_support.h:123
ojph::stex::rtp_packet::get_PTSTAMP
ui32 get_PTSTAMP()
Definition stream_expand_support.h:131
ojph::stex::rtp_packet::get_rtp_version
ui32 get_rtp_version()
Definition stream_expand_support.h:106
ojph::stex::rtp_packet::is_codeblock_caching_used
bool is_codeblock_caching_used()
Definition stream_expand_support.h:158
ojph::stex::rtp_packet::packet_type
packet_type
packet types based on the main header of draft-ietf-avtcore-rtp-j2k-scl-00
Definition stream_expand_support.h:85
ojph::stex::rtp_packet::PT_MAIN_FOLLOWED_BY_MAIN
@ PT_MAIN_FOLLOWED_BY_MAIN
Definition stream_expand_support.h:87
ojph::stex::rtp_packet::PT_MAIN
@ PT_MAIN
Definition stream_expand_support.h:89
ojph::stex::rtp_packet::PT_BODY
@ PT_BODY
Definition stream_expand_support.h:86
ojph::stex::rtp_packet::PT_MAIN_FOLLOWED_BY_BODY
@ PT_MAIN_FOLLOWED_BY_BODY
Definition stream_expand_support.h:88
ojph::stex::rtp_packet::is_RANGE
bool is_RANGE()
Definition stream_expand_support.h:162
ojph::stex::rtp_packet::get_ORDH
ui32 get_ORDH()
Definition stream_expand_support.h:127
ojph::stex::rtp_packet::is_marked
bool is_marked()
Definition stream_expand_support.h:110
ojph::stex::stex_file
holds in memory j2k codestream together with other info
Definition stream_expand_support.h:353
ojph::stex::stex_file::parent
frames_handler * parent
the object holding this frame
Definition stream_expand_support.h:406
ojph::stex::stex_file::name_template
const char * name_template
name template for saved files
Definition stream_expand_support.h:408
ojph::stex::stex_file::f
ojph::mem_outfile f
holds in-memory j2k codestream
Definition stream_expand_support.h:401
ojph::stex::stex_file::frame_idx
ui32 frame_idx
frame number in the sequence
Definition stream_expand_support.h:405
ojph::stex::stex_file::init
void init(frames_handler *parent, stex_file *next, j2k_frame_storer *storer, const char *name_template)
call this function to initialize stex_file
Definition stream_expand_support.h:381
ojph::stex::stex_file::notify_file_completion
void notify_file_completion()
other threads can call this function to signal completion of processing.
Definition stream_expand_support.cpp:241
ojph::stex::stex_file::last_seen_seq
ui32 last_seen_seq
the last seen RTP sequence number
Definition stream_expand_support.h:403
ojph::stex::stex_file::next
stex_file * next
used to create files chain
Definition stream_expand_support.h:411
ojph::stex::stex_file::stex_file
stex_file()
default constructor
Definition stream_expand_support.h:358
ojph::stex::stex_file::time_stamp
ui32 time_stamp
time stamp at which this file must be displayed
Definition stream_expand_support.h:402
ojph::stex::stex_file::done
std::atomic_int done
saving is completed when 0 is reached
Definition stream_expand_support.h:404
ojph::stex::stex_file::storer
j2k_frame_storer * storer
stores a j2k frame using another thread
Definition stream_expand_support.h:409